swift - 如何制作可重置的 RxSwift 计时器？

Question

我正在尝试构建一个倒数计时器应用程序。当然，我需要一个 Observable<Int>每秒给我一个元素的序列。扭曲的是我需要这个序列来响应其他两个序列的变化:

暂停:Observable<Bool> .这模拟用户按下暂停/恢复按钮

重置:Observable<Void> .这模拟了用户按下重置按钮。

重置按钮会将计时器重置回其起始值 ( resetState )，并暂停计时器。

用户可以随时按下复位按钮:

计时器未启动时重置

计时器运行时重置，未暂停，未结束

计时器暂停时重置，未结束

计时器结束后重置

结合回答 this question和 this question ，我想出了一个 Timer像这样的类:

class Timer {
    var paused = true
    {
        didSet {
            rxPaused.accept(paused)
        }
    }
    var ended = false

    let rxPaused = BehaviorRelay(value: true)
    let disposeBag = DisposeBag()

    // timerEvents is the observable that client code should subscribe to
    var timerEvents: Observable<Int>!

    var currentState: Int
    let resetState: Int

    init(resetState: Int) {
        self.currentState = resetState
        self.resetState = resetState
        reset()
    }

    func start() {
        if !ended {
            paused = false
        }
    }

    func pause() {
        paused = true
    }

    func reset() {
        ended = false
        currentState = resetState
        pause()
        timerEvents = rxPaused.asObservable()
            .flatMapLatest {  isRunning in
                isRunning ? .empty() : Observable<Int>.interval(.seconds(1), scheduler: MainScheduler.instance)
            }
            .enumerated().flatMap { (index, int) in Observable.just(index) }
        .map { [weak self] x in (self?.resetState ?? x) - x }
            .take(self.resetState)
        timerEvents.subscribe(onNext: { [weak self]
            timerEvent in
            self?.currentState -= 1
            }, onCompleted: {
                [weak self] in
                self?.ended = true
        }).disposed(by: disposeBag)
    }
}

用法:

let timer = Timer(resetState: 20)
timer.timerEvents.subscribe(
    onNext: { [unowned self] (timerEvent) in
        print(timerEvent.state)
        self.updateTimerLabelText()
}).disposed(by: disposeBag)

这仅在上述情况 4 中按预期工作。

如果你试图在它结束之前重置这个计时器，事情就会变得有趣。例如，在情况 1(甚至在启动之前重置)中， timerEvents每秒产生两个相同的元素。这不仅不正确，还会导致 currentState以两倍的速度减少。我怀疑这是因为 timerEvents在其先前的值完成之前被第二次分配，但我认为没有办法“完成”未完成的可观察，是吗？

我什至无法用语言表达在情况 2 和 3 中会发生什么。

我怎样才能使这个重置计时器工作？

Answer 1

更新
在评论中，我被要求证明为什么我建议对“新代码”进行测试。部分答案是你永远不应该接受你的代码的初稿。正如任何作文老师都会告诉你的，不要交出你的初稿，花一些时间完善你写的东西(如果你能得到同行评审)。鉴于这一点以及我的测试错过了其中一个规范的事实，我打算用这个更精致的版本替换我的初始答案，但我认为保留原始答案是有益的，因此可以将其与精炼的答案进行比较。
在下面，您将看到我更新了测试以适应新规范并改进了代码。
函数中有一个 flatMap 的事实意味着这里有两个抽象。所以我把它分解成一个单独的函数。
我有两个案例的枚举的事实意味着我可以使用 Bool 代替并删除开关。

class rx_sandboxTests: XCTestCase {

    func testPause() {
        let scheduler = TestScheduler(initialClock: 0)
        let pause = scheduler.createColdObservable([.next(10, ()), .next(20, ())])
        let reset = scheduler.createColdObservable([.next(30, ())])
        let result = scheduler.start {
            isPaused(pause: pause.asObservable(), reset: reset.asObservable())
        }
        XCTAssertEqual(result.events, [.next(200, true), .next(210, false), .next(220, true)])
    }

    func testTimerStart() {
        let scheduler = TestScheduler(initialClock: 0)
        let pause = scheduler.createColdObservable([.next(10, ())])
        let reset = scheduler.createColdObservable([Recorded<Event<Void>>]())

        let result = scheduler.start {
            timer(initial: 10, pause: pause.asObservable(), reset: reset.asObservable(), scheduler: scheduler)
        }

        XCTAssertEqual(result.events, [.next(200, 10), .next(211, 9), .next(212, 8), .next(213, 7), .next(214, 6), .next(215, 5), .next(216, 4), .next(217, 3), .next(218, 2), .next(219, 1), .next(220, 0)])
    }

    func testPausedTimer() {
        let scheduler = TestScheduler(initialClock: 0)
        let pause = scheduler.createColdObservable([.next(10, ()), .next(13, ()), .next(20, ())])
        let reset = scheduler.createColdObservable([Recorded<Event<Void>>]())

        let result = scheduler.start {
            timer(initial: 4, pause: pause.asObservable(), reset: reset.asObservable(), scheduler: scheduler)
        }

        XCTAssertEqual(result.events, [.next(200, 4), .next(211, 3), .next(212, 2), .next(221, 1), .next(222, 0)])
    }

    func testResetBeforeStarting() {
        let scheduler = TestScheduler(initialClock: 0)
        let pause = scheduler.createColdObservable([.next(20, ())])
        let reset = scheduler.createColdObservable([.next(10, ())])

        let result = scheduler.start {
            timer(initial: 3, pause: pause.asObservable(), reset: reset.asObservable(), scheduler: scheduler)
        }

        XCTAssertEqual(result.events, [.next(200, 3), .next(221, 2), .next(222, 1), .next(223, 0)])
    }

    func testResetWhileRunning() {
        let scheduler = TestScheduler(initialClock: 0)
        let pause = scheduler.createColdObservable([.next(10, ()), .next(20, ())])
        let reset = scheduler.createColdObservable([.next(13, ())])

        let result = scheduler.start {
            timer(initial: 4, pause: pause.asObservable(), reset: reset.asObservable(), scheduler: scheduler)
        }

        XCTAssertEqual(result.events, [.next(200, 4), .next(211, 3), .next(212, 2), .next(213, 4), .next(221, 3), .next(222, 2), .next(223, 1), .next(224, 0)])
    }

    func testResetWhilePaused() {
        let scheduler = TestScheduler(initialClock: 0)
        let pause = scheduler.createColdObservable([.next(10, ()), .next(13, ()), .next(20, ())])
        let reset = scheduler.createColdObservable([.next(15, ())])

        let result = scheduler.start {
            timer(initial: 4, pause: pause.asObservable(), reset: reset.asObservable(), scheduler: scheduler)
        }

        XCTAssertEqual(result.events, [.next(200, 4), .next(211, 3), .next(212, 2), .next(215, 4), .next(221, 3), .next(222, 2), .next(223, 1), .next(224, 0)])
    }

    func testResetWhenEnded() {
        let scheduler = TestScheduler(initialClock: 0)
        let pause = scheduler.createColdObservable([.next(10, ()), .next(20, ())])
        let reset = scheduler.createColdObservable([.next(15, ())])

        let result = scheduler.start {
            timer(initial: 4, pause: pause.asObservable(), reset: reset.asObservable(), scheduler: scheduler)
        }

        XCTAssertEqual(result.events, [.next(200, 4), .next(211, 3), .next(212, 2), .next(213, 1), .next(214, 0), .next(215, 4), .next(221, 3), .next(222, 2), .next(223, 1), .next(224, 0)])
    }
}

func timer(initial: Int, pause: Observable<Void>, reset: Observable<Void>, scheduler: SchedulerType) -> Observable<Int> {
    let tick = isPaused(pause: pause, reset: reset)
        .flatMapLatest { $0 ? .empty() : Observable<Int>.interval(.seconds(1), scheduler: scheduler).take(initial) }

    return ticker(initial: initial, tick: tick, reset: reset)
}

func isPaused(pause: Observable<Void>, reset: Observable<Void>) -> Observable<Bool> {
    Observable.merge(pause.map { false }, reset.map { true })
        .scan(true) { $1 || !$0 }
        .startWith(true)
        .distinctUntilChanged()
}

func ticker<T>(initial: Int, tick: Observable<T>, reset: Observable<Void>) -> Observable<Int> {
    return Observable.merge(tick.map { _ in false }, reset.map { true })
        .scan(initial) { $1 ? initial : $0 - 1 }
        .startWith(initial)
        .filter { 0 <= $0 }
        .distinctUntilChanged()
}

原答案如下:
我把你的停顿从 Observable<Bool> 改了至 Observable<Void> . Bool 没有任何意义，因为重置也可能导致暂停，这会与其他 observable 发生冲突。
这是完整的代码，包括一个测试工具:

class rx_sandboxTests: XCTestCase {

    func testTimerStart() {
        let scheduler = TestScheduler(initialClock: 0)
        let pause = scheduler.createColdObservable([.next(10, ())])
        let reset = scheduler.createColdObservable([Recorded<Event<Void>>]())

        let result = scheduler.start {
            timer(initial: 10, pause: pause.asObservable(), reset: reset.asObservable(), scheduler: scheduler)
        }

        XCTAssertEqual(result.events, [.next(211, 9), .next(212, 8), .next(213, 7), .next(214, 6), .next(215, 5), .next(216, 4), .next(217, 3), .next(218, 2), .next(219, 1), .next(220, 0)])
    }

    func testPause() {
        let scheduler = TestScheduler(initialClock: 0)
        let pause = scheduler.createColdObservable([.next(10, ()), .next(13, ()), .next(20, ())])
        let reset = scheduler.createColdObservable([Recorded<Event<Void>>]())

        let result = scheduler.start {
            timer(initial: 4, pause: pause.asObservable(), reset: reset.asObservable(), scheduler: scheduler)
        }

        XCTAssertEqual(result.events, [.next(211, 3), .next(212, 2), .next(221, 1), .next(222, 0)])
    }

    func testResetBeforeStarting() {
        let scheduler = TestScheduler(initialClock: 0)
        let pause = scheduler.createColdObservable([.next(20, ())])
        let reset = scheduler.createColdObservable([.next(10, ())])

        let result = scheduler.start {
            timer(initial: 3, pause: pause.asObservable(), reset: reset.asObservable(), scheduler: scheduler)
        }

        XCTAssertEqual(result.events, [.next(221, 2), .next(222, 1), .next(223, 0)])
    }

    func testResetWhileRunning() {
        let scheduler = TestScheduler(initialClock: 0)
        let pause = scheduler.createColdObservable([.next(10, ()), .next(20, ())])
        let reset = scheduler.createColdObservable([.next(13, ())])

        let result = scheduler.start {
            timer(initial: 4, pause: pause.asObservable(), reset: reset.asObservable(), scheduler: scheduler)
        }

        XCTAssertEqual(result.events, [.next(211, 3), .next(212, 2), .next(221, 3), .next(222, 2), .next(223, 1), .next(224, 0)])
    }

    func testResetWhilePaused() {
        let scheduler = TestScheduler(initialClock: 0)
        let pause = scheduler.createColdObservable([.next(10, ()), .next(13, ()), .next(20, ())])
        let reset = scheduler.createColdObservable([.next(15, ())])

        let result = scheduler.start {
            timer(initial: 4, pause: pause.asObservable(), reset: reset.asObservable(), scheduler: scheduler)
        }

        XCTAssertEqual(result.events, [.next(211, 3), .next(212, 2), .next(221, 3), .next(222, 2), .next(223, 1), .next(224, 0)])
    }

    func testResetWhenEnded() {
        let scheduler = TestScheduler(initialClock: 0)
        let pause = scheduler.createColdObservable([.next(10, ()), .next(20, ())])
        let reset = scheduler.createColdObservable([.next(15, ())])

        let result = scheduler.start {
            timer(initial: 4, pause: pause.asObservable(), reset: reset.asObservable(), scheduler: scheduler)
        }

        XCTAssertEqual(result.events, [.next(211, 3), .next(212, 2), .next(213, 1), .next(214, 0), .next(221, 3), .next(222, 2), .next(223, 1), .next(224, 0)])
    }
}

func timer(initial: Int, pause: Observable<Void>, reset: Observable<Void>, scheduler: SchedulerType) -> Observable<Int> {
    enum Action { case pause, reset, tick }
    let intent = Observable.merge(
        pause.map { Action.pause },
        reset.map { Action.reset }
    )

    let isPaused = intent
        .scan(true) { isPaused, action in
            switch action {
            case .pause:
                return !isPaused
            case .reset:
                return true
            case .tick:
                fatalError()
            }
        }
        .startWith(true)

    let tick = isPaused
        .flatMapLatest { $0 ? .empty() : Observable<Int>.interval(.seconds(1), scheduler: scheduler) }

    return Observable.merge(tick.map { _ in Action.tick }, reset.map { Action.reset })
        .scan(initial) { (current, action) -> Int in
            switch action {
            case .pause:
                fatalError()
            case .reset:
                return initial
            case .tick:
                return current == -1 ? -1 : current - 1
            }

        }
        .filter { 0 <= $0 && $0 < initial }
}

很高兴知道如何测试 Rx 代码。