How do I reset an observable in Angular?

Question

I have a service which caches a bunch of autocomplete phrases:

constructor(private http: HttpClient) { }

private phrases$: Observable<string[]>;

public getPhrases(): Observable<string[]> {
    if (!this.phrases$) {
        this.phrases$ = this.http.get<string[]>(this.baseUrl + '/getAutocompletePhrasesList')
            .pipe(
                shareReplay(1)
            );;
    }
    return this.phrases$;
}

I have a directive that calls this service, and the directive is added throughout the application for most text boxes:

ngOnInit() {
  this.mentionService.getPhrases()
    .subscribe(
      result => {
        this.phrases = result;
        if (this.phrases && this.phrases.length > 0) {

          // This basically processes autocomplete, removed the code for this question
        }
      },
      () => { });

}

Works really well, however how do I 'reset' my observable when a new phrases is added. I've tried setting phrases$ to null, but that doesn't work.

Answer 1

So as far as I understand from your comments, you want to have the following functionality:

1. Any time your autocomplete directive needs access to the list of phrases$, the MentionService should either call the API, or return the cached list it stores.
2. This cached list should be reset every time some specific component/service does an API call, so that MentionService knows it needs to fetch the new list, should it be asked for it by the directive.
3. Moreover, you want the directive to be able to subscribe to this service only once in its ngOnInit, and then never worry about it anymore, even if internally the MentionService has to temporarily delete or re-fetch the list.

I hope I got it right. Let's get to the solution!

---

@Injectable({ providedIn: 'root' })
export class MentionService implements OnInit {
    private baseUrl: 'some-url';

    // a private Subject that MentionService fully controls.
    // we will consider the latest value that this Subject has emitted as our "cache".
    private phrasesSubject$: BehaviorSubject<string[]>;

    constructor(private http: HttpClient) {
      // let's initialize the Subject's "cache" with null value, since we haven't fetched anything yet.
      this.phrasesSubject$ = new BehaviorSubject(null);
    }

    ngOnInit() {
        // optionally, load up and cache the phrases upon app init, if you want to.
        this.loadAndCachePhrases();
    }

    // not exposed outside - no external class needs to know how exactly
    // MentionService sources the list of phrases it provides.
    private loadAndCachePhrases(): void {
        this.http.get<string[]>(this.baseUrl + '/getAutocompletePhrasesList').pipe(
          // every time we load the list, we immediately cache it in our Subject.
          tap(list => this.phrasesSubject$.next(list))
        )
    }

    // the exposed way of accessing the phrases stream from outside, e.g. from AutocompleteDirective
    public getPhrases$(): Observable<string[]> {
        // we beed to decide whether or not we should reload data,
        // based on what we already have cached in our subject:
        this.phrasesSubject$.pipe(
          take(1), // just do this action once with the *current* value, then ignore next incoming values
          tap((currentCache) => {
              // if cache is present, no need to do anything, but if it is not, we need to reload data.
              if (!currentCache) {
                this.loadAndCachePhrases();
              }
          })
        )

        // don't return the actual Subject, because then external classes could alter it -
        // return it as a (read-only) Observable instead.
        return this.phrasesSubject$.asObservable();
    }

    // the other service can call resetPhrasesCache() to inform MentionService
    // that the phrases it might have had cached are no longer up to date,
    // and need to be re-fetched the next time it's asked for them.
    public resetPhrasesCache(): void {
        // as we know, the latest value from this Subject *is* the cache, so 
        // setting it to null means that the cache is gone.
        this.phrasesSubject$.next(null);
    }
}

Let me know if that solves your problem!

---

(As a minor sidenote, you might want to consider that the directive class is instantiated for every bound input, and the service only once for the entire app. This means it's always the most efficient to store the cached list just in the service, and never copy it to the directive instances at all. You're currently copying it with this.phrases = result, which theoretically is fine as it's only by reference, but this reference can desynchronize. Some performance wins might be possible here, depending on the number of inputs and the length of your list of course.)

Answer 2

For the problem I've understood the solution would be to also unsubscribe and not just setting it to null.

constructor(private http: HttpClient) { }

private phrases$: Observable<string[]>;
private reset$ = new Subject<any>();

public getPhrases(): Observable<string[]> {
    if (!this.phrases$) {
        this.phrases$ = this.http.get<string[]>(this.baseUrl + '/getAutocompletePhrasesList')
            .pipe(
                takeUntil(this.reset$), // <--- Unsubscribe when called
                shareReplay(1)
            );;
    }
    return this.phrases$;
}

public reset(): void {
    this.reset$.next(null);
    this.phrases$ = null;
}

Having a tough time time trying to understand what you need as all the other answers and comments does solve your question. From my understanding, Charlie's answer should be the best approach to solve your issue.

My guess is that what you really want is to "reset" the subscription you've already made the first time you called the service. That'll be the reason of "why" setting phrases$ = null doesn't work. As you would be mistakenly deleting the reference inside the service and not the subscription you've previously made.

Answer 3

For "resetting" you could do the following

#reset$ = new Subject<null>();

loadPhrases() {
  merge(              // Merge makes two observables emit values to one subscriber
                      // so the subscribe will receive `http.get` result, then `reset`
    this.http.get(),  // This will get the data
    this.#reset$      // This will emit `null` when `reset()` is called
  ).subscribe()
}

reset(): void {
  this.#reset$.next(null);
}

Also if the directive itself does the caching you could have a subject. In the directive have a subscription to it and when it emits reset the stored value.

However, your use case isn't very clear on what it should do. Maybe there is another better option rather than "reset".

Answer 4

One way would be to use a private Subject. And use the Observable to inform the outside world of updates.

private phrases: BehaviorSubject<string[]> = new BehaviorSubject<string[]>([]);
phrases$:Observable<string[]> = this.phrases.asObservable();

constructor(private http: HttpClient) { }

ngOnInit(){
  this.loadPhrases();
}

loadPhrases(){
  this.http.get<string[]>(this.baseUrl + '/getAutocompletePhrasesList')
  .subscribe(res=>this.phrases.next(res))
}

getPhrases(): Observable<string[]> {
  return this.phrases$;
}

Whenever you call loadPhrases() the Subject will emit the new array and everybody subscribed to the Observable phrases$ will receive the updated list.