Resetting ReplaySubject in RxJS 6

The problem becomes easier if you can use the fact that the buffer consumes data from the original source, and that subscribers to buffered data can switch to the original source after receiving all the old values.

Eg.

let data$ = new Subject<any>() // Data source

let buffer$ = new ReplaySubject<any>() 
let bs = data$.subscribe(buffer$)  // Buffer subscribes to data

// Observable that returns values until nearest reset
let getRepeater = () => {
   return concat(buffer$.pipe(
      takeUntil(data$), // Switch from buffer to original source when data comes in
    ), data$)
}

To clear, replace the buffer

// Begin Buffer Clear Sequence
bs.unsubscribe()
buffer$.complete()

buffer$ = new ReplaySubject()
bs = data$.subscribe(buffer$)
buffObs.next(buffer$)

To make the code more functional, you can replace the function getRepeater() with a subject that reflects the latest reference

let buffObs = new ReplaySubject<ReplaySubject<any>>(1)
buffObs.next(buffer$)        

let repeater$ = concat(buffObs.pipe(
   takeUntil(data$),
   switchMap((e) => e),                    
), data$)

The following

    let data$ = new Subject<any>()

    let buffer$ = new ReplaySubject<any>()
    let bs = data$.subscribe(buffer$)         

    let buffObs = new ReplaySubject<ReplaySubject<any>>(1)
    buffObs.next(buffer$)        

    let repeater$ = concat(buffObs.pipe(
      takeUntil(data$),
      switchMap((e) => e),                    
    ), data$)

    // Begin Test

    data$.next(1)
    data$.next(2)
    data$.next(3)

    console.log('rep1 sub')
    let r1 = repeater$.subscribe((e) => {          
      console.log('rep1 ' + e)
    })

    // Begin Buffer Clear Sequence
    bs.unsubscribe()
    buffer$.complete()

    buffer$ = new ReplaySubject()
    bs = data$.subscribe(buffer$)
    buffObs.next(buffer$)
    // End Buffer Clear Sequence

    console.log('rep2 sub')
    let r2 = repeater$.subscribe((e) => {
      console.log('rep2 ' + e)
    })

    data$.next(4)
    data$.next(5)
    data$.next(6)

    r1.unsubscribe()
    r2.unsubscribe()

    data$.next(7)
    data$.next(8)
    data$.next(9)        

    console.log('rep3 sub')
    let r3 = repeater$.subscribe((e) => {
      console.log('rep3 ' + e)
    })

Outputs

rep1 sub

rep1 1

rep1 2

rep1 3

rep2 sub

rep1 4

rep2 4

rep1 5

rep2 5

rep1 6

rep2 6

rep3 sub

rep3 4

rep3 5

rep3 6

rep3 7

rep3 8

rep3 9

If you want to be able to reset a subject without having its subscribers explicitly unsubscribe and resubscribe, you could do something like this:

import { Observable, Subject } from "rxjs";
import { startWith, switchMap } from "rxjs/operators";

function resettable<T>(factory: () => Subject<T>): {
  observable: Observable<T>,
  reset(): void,
  subject: Subject<T>
} {
  const resetter = new Subject<any>();
  const source = new Subject<T>();
  let destination = factory();
  let subscription = source.subscribe(destination);
  return {
    observable: resetter.asObservable().pipe(
      startWith(null),
      switchMap(() => destination)
    ),
    reset: () => {
      subscription.unsubscribe();
      destination = factory();
      subscription = source.subscribe(destination);
      resetter.next();
    },
    subject: source
  };
}

resettable will return an object containing:

• an observable to which subscribers to the re-settable subject should subscribe;
• a subject upon which you'd call next, error or complete; and
• a reset function that will reset the (inner) subject.

You'd use it like this:

import { ReplaySubject } from "rxjs";
const { observable, reset, subject } = resettable(() => new ReplaySubject(3));
observable.subscribe(value => console.log(`a${value}`)); // a1, a2, a3, a4, a5, a6
subject.next(1);
subject.next(2);
subject.next(3);
subject.next(4);
observable.subscribe(value => console.log(`b${value}`)); // b2, b3, b4, b5, b6
reset();
observable.subscribe(value => console.log(`c${value}`)); // c5, c6
subject.next(5);
subject.next(6);

I had kind of same problem: One of my components subscribed to an ReplaySubject of a shared service. Once navigated away and coming back the former values where still delivered to the component. Just completing the subject was not enough.

The solutions above seemed to complicated for this purpose but I found another real simple solution in just completing the subject and assigning a newly created one in the shared service like so:

constructor() {
    this.selectedFeatures = new ReplaySubject()
    this.selectedFeaturesObservable$ = this.selectedFeatures.asObservable()
}

completeSelectedFeatures() {
    this.selectedFeatures.complete()
    this.selectedFeatures = new ReplaySubject()
    this.selectedFeaturesObservable$ = this.selectedFeatures.asObservable()

}

I also printed the constructor of the shared service to show the types I used. That way any time I move away from my component I just call that method on my shared service and hence get a new fresh and empty ReplaySubject anytime I navigate back to my component thats consuming the shared services observable. I call that method inside ngOnDestroy Angular lifecycle hook:

ngOnDestroy() {
    console.log('unsubscribe')
    this.featureSub.unsubscribe()
    this.sharedDataService.completeSelectedFeatures()
}

Here is a class that is using the resettable factory posted here before, so you can use const myReplaySubject = new ResettableReplaySubject<myType>()

import { ReplaySubject, Subject, Observable, SchedulerLike } from "rxjs";
import { startWith, switchMap } from "rxjs/operators";

export class ResettableReplaySubject<T> extends ReplaySubject<T> {

reset: () => void;

constructor(bufferSize?: number, windowTime?: number, scheduler?: SchedulerLike) {
    super(bufferSize, windowTime, scheduler);
    const resetable = this.resettable(() => new ReplaySubject<T>(bufferSize, windowTime, scheduler));

    Object.keys(resetable.subject).forEach(key => {
        this[key] = resetable.subject[key];
    })

    Object.keys(resetable.observable).forEach(key => {
        this[key] = resetable.observable[key];
    })

    this.reset = resetable.reset;
}


private resettable<T>(factory: () => Subject<T>): {
    observable: Observable<T>,
    reset(): void,
    subject: Subject<T>,
} {
    const resetter = new Subject<any>();
    const source = new Subject<T>();
    let destination = factory();
    let subscription = source.subscribe(destination);
    return {
        observable: resetter.asObservable().pipe(
            startWith(null),
            switchMap(() => destination)
        ) as Observable<T>,
        reset: () => {
            subscription.unsubscribe();
            destination = factory();
            subscription = source.subscribe(destination);
            resetter.next();
        },
        subject: source,
    };
}
}