Pause RxJS stream based on a value in the stream

Here's a custom pause operator that will just accumulate values in a buffer when the pause signal is true, and emit them one by one when it is false.

Combine it with a simple tap operator to toggle the behavior subject pause signal when the value hits a specific condition, and you have something will pause on button click and also pause when the value meets a condition (multiple of 12 in this case):

Here is the pause operator:

function pause<T>(pauseSignal: Observable<boolean>) {
  return (source: Observable<T>) => Observable.create(observer => {
    const buffer = [];
    let paused = false;
    let error;
    let isComplete = false;

    function notify() {
      while (!paused && buffer.length) {
        const value = buffer.shift();
        observer.next(value);
      }

      if (!buffer.length && error) {
        observer.error(error);
      }

      if (!buffer.length && isComplete) {
        observer.complete();
      }
    }

    const subscription = pauseSignal.subscribe(
      p => {
        paused = !p;
        setTimeout(notify, 0);
      },
      e => {
        error = e;
        setTimeout(notify, 0);
      },
      () => {});

    subscription.add(source.subscribe(
      v => {
        buffer.push(v);
        notify();
      },
      e => {
        error = e;
        notify();
      },
      () => {
        isComplete = true;
        notify();
      }
    ));

    return subscription;
  });
}

Here is the usage of it:

const CONDITION = x => (x > 0) && ((x % 12) === 0); // is multiple
this.active$ = new BehaviorSubject<boolean>(true);
const stream$ = timer(500, 500);
const out$ = stream$.pipe(
  pause(this.active$),
  tap(value => {
    if (CONDITION(value)) {
      this.active$.next(false);
    }
  }));

this.d = out$.subscribe(v => console.log(v));

And a working example: https://stackblitz.com/edit/angular-bvxnbf

I'm assuming that the desired behaviour is not related to getting the values that the timer emits per se, and that instead of pausing notifications to an ongoing stream (in your example, the timer continues even if we don't see the values being printed), it's okay to actually stop emitting when paused.

My solution is inspired by the Stopwatch recipe

The solution below uses two separate buttons for play and pause, but you can adjust this to taste. We pass the (ViewChild) buttons to the service in the ngAfterViewInit hook of the component, then we subscribe to the stream.

// pausable.component.ts
  ngAfterViewInit() {
    this.pausableService.initPausableStream(this.start.nativeElement, this.pause.nativeElement);

    this.pausableService.counter$
      .pipe(takeUntil(this.unsubscribe$)) // don't forget to unsubscribe :)
      .subscribe((state: State) => {
        console.log(state.value); // whatever you need
    });
  }

// pausable.service.ts
import { Injectable } from '@angular/core';

import { merge, fromEvent, Subject, interval, NEVER } from 'rxjs';
import { mapTo, startWith, scan, switchMap, tap, map } from 'rxjs/operators';

export interface State {
  active: boolean;
  value: number;
}

@Injectable({
  providedIn: 'root'
})
export class PausableService {

  public counter$;

  constructor() { }

  initPausableStream(start: HTMLElement, pause: HTMLElement) {

    // convenience functions to map an element click to a result
    const fromClick = (el: HTMLElement) => fromEvent(el, 'click');
    const clickMapTo = (el: HTMLElement, obj: {}) => fromClick(el).pipe(mapTo(obj));

    const pauseByCondition$ = new Subject();
    const pauseCondition = (state: State): boolean => state.value % 5 === 0 && state.value !== 0;

    // define the events that may trigger a change
    const events$ = merge(
      clickMapTo(start, { active: true }),
      clickMapTo(pause, { active: false }),
      pauseByCondition$.pipe(mapTo({ active: false }))
    );

    // switch the counter stream based on events
    this.counter$ = events$.pipe(
      startWith({ active: true, value: 0 }),
      scan((state: State, curr) => ({ ...state, ...curr }), {}),
      switchMap((state: State) => state.active
        ? interval(500).pipe(
          tap(_ => ++state.value),
          map(_ => state))
        : NEVER),
      tap((state: State) => {
        if (pauseCondition(state)) {
          pauseByCondition$.next(); // trigger pause
        }
      })
    );
  }

}

It's possible to either (1) expand your current bufferToggle / windowToggle approach or to (2) use a custom buffer implementation.

1. Expanding the bufferToggle / windowToggle approach

You can add an array to the operator queue after bufferToggle.

When bufferToggle emits append those values to the array.
Take values from the array until a certain element in the array matches a halt condition.
Emit those values and pause your stream.

pausable (Demo)

The pausable operator will emit values that match the halt condition and then stop the stream immediately.

export function pausable<T, O>(
  on$: Observable<any>, // when on$ emits 'pausable' will emit values from the buffer and all incoming values 
  off$: Observable<O>, // when off$ emits 'pausable' will stop emitting and buffer incoming values
  haltCondition: (value: T) => boolean, // if 'haltCondition' returns true for a value in the stream the stream will be paused
  pause: () => void, // pauses the stream by triggering the given on$ and off$ observables
  spread: boolean = true // if true values from the buffer will be emitted separately, if 'false' values from the buffer will be emitted in an array
) {
  return (source: Observable<T>) => defer(() => { // defer is used so that each subscription gets its own buffer
    let buffer: T[] = [];
    return merge(
      source.pipe(
        bufferToggle(off$, () => on$),
        tap(values => buffer = buffer.concat(values)), // append values to your custom buffer
        map(_ => buffer.findIndex(haltCondition)), // find the index of the first element that matches the halt condition
        tap(haltIndex => haltIndex >= 0 ? pause() : null), // pause the stream when a value matching the halt condition was found
        map(haltIndex => buffer.splice(0, haltIndex === -1 ? customBuffer.length : haltIndex + 1)), // get all values from your custom buffer until a haltCondition is met
        mergeMap(toEmit => spread ? from(toEmit) : toEmit.length > 0 ? of(toEmit) : EMPTY) // optional value spread (what your mergeAll did)
      ),
      source.pipe(
        windowToggle(on$, () => off$),
        mergeMap(x => x),
        tap(value => haltCondition(value) ? pause() : null), // pause the stream when an unbuffered value matches the halt condition
      ),
    );
  });
}

You can adjust this operator to your specific needs e.g. use less input parameters and incorporate share into it, see this version with less parameters.

Usage

active$ = new BehaviorSubject<boolean>(true);
on$ = this.active$.pipe(filter(v => v));
off$ = this.active$.pipe(filter(v => !v));

interval(500).pipe(
  share(),
  pausable(on$, off$, v => this.active$.value && this.pauseOn(v), () => this.active$.next(false))
).subscribe(console.log);

pauseOn = (value: number) => value > 0 && value % 10 === 0

2. A fully custom buffer

You can go with a fully custom approach using only one input observable similar to Brandon's approach.

bufferIf (Demo)

bufferIf will buffer incoming values when the given condition emits true and emits all values from the buffer or passes new ones through when the condition is false.

export function bufferIf<T>(condition: Observable<boolean>) {
  return (source: Observable<T>) => defer(() => {
    const buffer: T[] = [];
    let paused = false;
    let sourceTerminated = false;
    return merge( // add a custon streamId to values from the source and the condition so that they can be differentiated later on
      source.pipe(map(v => [v, 0]), finalize(() => sourceTerminated = true)),
      condition.pipe(map(v => [v, 1]))
    ).pipe( // add values from the source to the buffer or set the paused variable
      tap(([value, streamId]) => streamId === 0 ? buffer.push(value as T) : paused = value as boolean), 
      switchMap(_ => new Observable<T>(s => {
        setTimeout(() => { // map to a stream of values taken from the buffer, setTimeout is used so that a subscriber to the condition outside of this function gets the values in the correct order (also see Brandons answer & comments)
          while (buffer.length > 0 && !paused) s.next(buffer.shift())
        }, 0)
      })), // complete the stream when the source terminated and the buffer is empty
      takeWhile(_ => !sourceTerminated || buffer.length > 0, true) 
    );
  })
}

Usage

pause$ = new BehaviorSubject<boolean>(false);

interval(500).pipe(
  bufferIf(this.pause$),
  tap(value => this.pauseOn(value) ? this.pause$.next(true) : null)
).subscribe(console.log);

pauseOn = (value: number) => value > 0 && value % 10 === 0

Here is a simple way to do it. Use the timer() just as an emitter, and increment a count separately. This gives you a little more direct control.

export class AppComponent implements OnInit {
  active = true;
  out$: Observable<number>;

  count = 0;

  ngOnInit(): void {

    const stream$ = timer(500, 500);

    this.out$ = stream$.pipe(
      filter(v => this.active),
      map(v => {
        this.count += 1;
        return this.count;
      }),
      tap(v => {
        if (this.count % 5 === 0) {
          this.active = false;
        }
      })
    )
  }

}

https://stackblitz.com/edit/angular-nzs7zh

Pause RxJS stream based on a value in the stream

1. Expanding the bufferToggle / windowToggle approach

pausable (Demo)

2. A fully custom buffer

bufferIf (Demo)

Tags:

Javascript

Angular

Rxjs

Related

Recent Posts