How do I make a resettable RxSwift Timer?

Question

I'm trying to build a countdown timer app. Naturally, I need an Observable<Int> sequence that gives me an element each second. The twist is that I need this sequence to respond to changes to two other sequences:

• Paused: Observable<Bool>. This models the user pressing the pause/resume button
• Reset: Observable<Void>. This models the user pressing the reset button.

The reset button will reset the timer back to its starting value (resetState), and also pause the timer.

The user can press the reset button at any time:

1. reset when the timer has not been started
2. reset when the timer is running, not paused, not ended
3. reset when the timer is paused, not ended
4. reset when the timer has ended

Combining the answers to this question and this question, I was able to came up with a Timer class like this:

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)
    }
}

Usage:

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

This only works as expected in situation 4 mentioned above.

If you try to reset this timer before it ends, things get funny. For example, in situation 1 (resetting before even starting), timerEvents produces two identical elements each seconds. Not only is this incorrect, it also causes currentState to decrease twice as quickly. I suspect that this is because timerEvents is assigned to a second time, before its previous value is completed, but I don't think there is a way to just "complete" an uncompleted observable, is there?

And I can't even put in words what happens in situations 2 and 3.

How can I make this resetting timer work?

Answer 1

Update

In the comments, I was asked to justify why I suggested making a test for "new code". Part of the answer was that you should never accept the first draft of your code. As any composition teacher would tell you, don't turn in your first draft, spend some time refining what you write (with peer review if you can get it.) So given that and the fact that my tests missed one of the specifications, I was going to replace my initial answer with this more refined version, but I think it is instructive to keep the original so it can be compared to the refined answer.

In the below, you will see that I have updated the tests to accommodate the new specification and refined the code.

The fact that there is a flatMap in the function implies that there are two abstractions here. So I broke that out into a separate function.

The fact that I have enums with two case implies that I could use a Bool instead and remove the switches.

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()
}

Original Answer Follows:

I changed your pause from an Observable<Bool> to Observable<Void>. The Bool didn't make any sense because the reset can also cause a pause and that would conflict with the other observable.

Here's the complete code, including a test harness:

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 }
}

It's good to know how to test Rx code.