How to add a stop and start feature for an RxJS timer?

Question

I added a start, stop, pause button. Start will start a count down timer which will start from a value, keep decrementing until value reaches 0. We can pause the timer on clicking the pause button. On click of Stop also timer observable completes.

• However, once the timer is completed ( either when value reaches 0 or when clicked on stop button ), I am not able to start properly. I tried adding repeatWhen operator. It starts on clicking twice. Not at the first time.

• Also, at stop, value is not resetting back to the initial value.

const subscription = merge(
  startClick$.pipe(mapTo(true)),
  pauseBtn$.pipe(mapTo(false))
)
  .pipe(
    tap(val => {
      console.log(val);
    }),
    switchMap(val => (val ? interval(10).pipe(takeUntil(stopClick$)) : EMPTY)),
    mapTo(-1),
    scan((acc: number, curr: number) => acc + curr, startValue),
    takeWhile(val => val >= 0),
    repeatWhen(() => startClick$),
    startWith(startValue)
  )
  .subscribe(val => {
    counterDisplayHeader.innerHTML = val.toString();
  });

Stackblitz Code link is available here

Answer 1

reset$
  .pipe(
    startWith(undefined),
    switchMap(() =>
      pause$.pipe(
        map(() => false),
        scan((acc) => !acc),
        startWith(true)
      ).pipe(
        switchMap((toggle) => (toggle ? interval(1000) : NEVER)),
        scan((acc) => acc + 1)
      )
    )
  )
  .subscribe(console.log);

In this code snippet we have a reset$ observable and a pause$ observable. The reset resets the count, and pause is a toggle for pausing and unpausing.

1. We begin with startWith undefined, so you don't need an emit of reset$ for the timer to start running.

2. Then we switchMap to the pause$ stream, and we map it to false, scan it to the negation of the latest value to act as a pause/unpause toggle, and provide a startWith value so again there is no pause$ emit required for the timer to start.

3. Finally we switchMap based on the toggle value between NEVER which never emits a value, and an interval. This means when this switchMap receives true, the timer starts, and false it wil emit nothing anymore. Then we scan the value because when switchMapping to NEVER, the interval has no subscribers anymore and the count resets, this way we prevent the counter from resetting.

See the stackblitz url for a working solution with two buttons, open the console in the right pane to see the timer output. https://stackblitz.com/edit/rxjs-9ywbxe?file=index.ts

Answer 2

You can achieve that in another way without completing the main observable or resubscribing to it using takeUntil, repeatWhen, or other operators, like the following:

• create a simple state to handle the counter changes (count, isTicking)
• merge all the observables that affecting the counter within one observable.
• create intermediate observable to interact with the main merge observable (start/stop counting).

interface CounterStateModel {
  count: number;
  isTicking: boolean;
}

// Setup counter state
const initialCounterState: CounterStateModel = {
  count: startValue,
  isTicking: false
};

const patchCounterState = new Subject<Partial<CounterStateModel>>();
const counterCommands$ = merge(
  startClick$.pipe(mapTo({ isTicking: true })),
  pauseBtn$.pipe(mapTo({ isTicking: false })),
  stopClick$.pipe(mapTo({ ...initialCounterState })),
  patchCounterState.asObservable()
);

const counterState$: Observable<CounterStateModel> = counterCommands$.pipe(
  startWith(initialCounterState),
  scan(
    (counterState: CounterStateModel, command): CounterStateModel => ({
      ...counterState,
      ...command
    })
  ),
  shareReplay(1)
);

const isTicking$ = counterState$.pipe(
  map(state => state.isTicking),
  distinctUntilChanged()
);

const commandFromTick$ = isTicking$.pipe(
  switchMap(isTicking => (isTicking ? timer(0, 10) : NEVER)),
  withLatestFrom(counterState$, (_, counterState) => ({
    count: counterState.count
  })),
  tap(({ count }) => {
    if (count) {
      patchCounterState.next({ count: count - 1 });
    } else {
      patchCounterState.next({ ...initialCounterState });
    }
  })
);

const commandFromReset$ = stopClick$.pipe(mapTo({ ...initialCounterState }));

merge(commandFromTick$, commandFromReset$)
  .pipe(startWith(initialCounterState))
  .subscribe(
    state => (counterDisplayHeader.innerHTML = state.count.toString())
  );

Also here is the working version: https://stackblitz.com/edit/rxjs-o86zg5

Answer 3

This is a pretty complicated usecase. There are two issues I think:

• You have two subscriptions to startClick$ and the order of subscriptions matters in this case. When the chain completes repeatWhen is waiting for startClick$ to emit. However, when you click the button the emission is first propagated into the first subscription inside merge(...) and does nothing because the chain has already completed. Only after that it resubscribes thanks to repeatWhen but you have to press the button again to trigger the switchMap() operator.

• When you use repeatWhen() it'll resubscribe every time the inner Observable emits so you want it to emit on startClick$ but only once. At the same time you don't want it to complete so you need to use something like this:

  repeatWhen(notifier$ => notifier$.pipe(
    switchMap(() => startClick$.pipe(take(1))),
  )),
  

So to avoid all that I think you can just complete the chain using takeUntil(stopClick$) and then immediatelly resubscribe with repeat() to start over.

merge(
  startClick$.pipe(mapTo(true)),
  pauseBtn$.pipe(mapTo(false))
)
  .pipe(
    switchMap(val => (val ? interval(10) : EMPTY)),
    mapTo(-1),
    scan((acc: number, curr: number) => acc + curr, startValue),
    takeWhile(val => val >= 0),
    startWith(startValue),
    takeUntil(stopClick$),
    repeat(),
  )
  .subscribe(val => {
    counterDisplayHeader.innerHTML = val.toString();
  });

Your updated demo: https://stackblitz.com/edit/rxjs-tum4xq?file=index.ts

Answer 4

Here's an example stopwatch that counts up instead of down. Perhaps you can re-tool it.

type StopwatchAction = "START" | "STOP" | "RESET" | "END";

function createStopwatch(
  control$: Observable<StopwatchAction>, 
  interval = 1000
): Observable<number>{

  return defer(() => {
    let toggle: boolean = false;
    let count: number = 0;

    const ticker = timer(0, interval).pipe(
      map(x => count++)
    );
    const end$ = of("END");

    return concat(
      control$,
      end$
    ).pipe(
      catchError(_ => end$),
      switchMap(control => {
        if(control === "START" && !toggle){
          toggle = true;
          return ticker;
        }else if(control === "STOP" && toggle){
          toggle = false;
          return EMPTY;
        }else if(control === "RESET"){
          count = 0;
          if(toggle){
            return ticker;
          }
        }
        return EMPTY;
      })
    );
  });
}

Here's an example of this in use:

const start$: Observable<StopwatchAction> = fromEvent(startBtn, 'click').pipe(mapTo("START"));
const reset$: Observable<StopwatchAction> = fromEvent(resetBtn, 'click').pipe(mapTo("RESET"));

createStopwatch(merge(start$,reset$)).subscribe(seconds => {
  secondsField.innerHTML  = seconds % 60;
  minuitesField.innerHTML = Math.floor(seconds / 60) % 60;
  hoursField.innerHTML    = Math.floor(seconds / 3600);
});