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.

1. When bufferToggle emits append those values to the array.
2. Take values from the array until a certain element in the array matches a halt condition.
3. 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'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 multiplethis.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

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