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

Question

我正在尝试构建一个倒数计时器应用程序。自然，我需要一个Observable<Int> 序列，它每秒给我一个元素。不同的是，我需要这个序列来响应对其他两个序列的更改：

• 暂停：Observable<Bool>。这模拟了用户按下暂停/恢复按钮
• 重置：Observable<Void>。这模拟了用户按下重置按钮。

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

用户可以随时按下重置按钮：

1. 定时器未启动时重置
2. 在定时器运行、未暂停、未结束时重置
3. 定时器暂停时重置，未结束
4. 定时器结束后重置

结合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 在之前的值完成之前被第二次分配，但我认为没有办法只是“完成”一个未完成的 observable，是吗？

我什至无法用语言描述情况 2 和 3 中发生的情况。

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

Answer 1

更新

在 cmets 中，我被要求解释为什么我建议对“新代码”进行测试。部分答案是你永远不应该接受你的代码的初稿。正如任何作文老师都会告诉你的那样，不要交出你的初稿，花一些时间来完善你写的东西（如果你能得到的话，请进行同行评审。）所以鉴于这一点以及我的测试错过了其中一个规范的事实，我打算用这个更精致的版本替换我最初的答案，但我认为保留原来的答案是有启发性的，因此可以将其与精致的答案进行比较。

在下面，您将看到我更新了测试以适应新规范并改进了代码。

函数中有一个 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 代码。