rxjs 7: Redo requests after a certain time / cache expiration

[StefanKern]

Initially I reported this as a bug, as it turned out, it previous was a bug and is now fixed in rxjs 7. So I'm just reiterating my question here, hoping that someone has a good solution :)

In rxjs v7 publishReplay, refCount, etc where deprecated and share and shareReplay are now the 'only' options when wanting use the result of an observable in multiple parts of the code, without redoing the complete 'observable'. An example for this is an API call, where the result is used in two places of the application, but you don't want to hit the network each time.

This was usually done with publishReplay and refCount. If you wanted to limit the time these requests where valid, you could set a windowTime in publishReplay and after "X"ms, the request repeated. (you also had to add a take(1) at the end, but that would be another question/issue). This solution was taken from SO: https://stackoverflow.com/a/54957061

With share or shareReplay, you cannot reproduce the same behaviour.

You can pass a windowTime to shareReplay, however after the time is exceeded, no more values are emitted at all!

Example how it was working before:

export class AppComponent {
  name = "Angular " + VERSION.major;

  urlofApi = "https://api.github.com/search/repositories?q=helloWorld";

  testX = this.http.get(this.urlofApi).pipe(
    tap(() =>
      console.log(
        `++ api was newly called at ${new Date().toLocaleTimeString()} ++`
      )
    ),
    publishReplay(1, 5000),
    refCount(),
    take(1)
  );

  constructor(private http: HttpClient) {}

  test() {
    console.log("Requesting data from API");
    this.testX.pipe(take(1)).subscribe(() => console.log("completed"));
  }
}

rxjs: 6.6.3
Stackblitz: https://stackblitz.com/edit/angular-ivy-tyrbdw?file=src%2Fapp%2Fapp.component.ts
Result:

Example with shareReplay

testX = this.http.get(this.urlofApi).pipe(
  tap(() =>
    console.log(
      `++ api was newly called at ${new Date().toLocaleTimeString()} ++`
    )
  ),
  shareReplay(1, 5000)
);

rxjs": "^7.0.0-rc.2
Stackblitz: https://stackblitz.com/edit/angular-ivy-ggkvpw?file=src%2Fapp%2Fapp.component.ts
Result;

Note how the Observable doesn't emit anything anymore, after 5sec.

Expected behaviour:

Requests are repeated once the windowTime is exceeded.

[josepot]

Hi @StefanKern

One thing that can be a bit surprising about shareReplay with refCount is that if the the source completes, then that's it: the resulting observable will never re-subscribe to the source, and every time that a new subscriber comes: it will replay the latest value(s) and then complete. I once opened an issue because I thought that was a bug 😅.

So, since your source is an Observable that emits once and then it completes, then I would suggest enhancing it so that it doesn't complete, i.e: concat(source, NEVER)

Ok, sure, but then how do we make sure that the cached value stays alive for a certain amount of time 🤔?

I suggest using a pipeable operator like this:

import { Observable, noop } from 'rxjs'

const delayLatestUnsubscription = (ms: number) => <T>(source$: Observable<T>) => {
  let token = 0
  let cleanupFn: (() => void) = noop
  return new Observable<T>(observer => {
    // Notice that it's necessary to create a new Observer, otherwise `take` would
    // close the original one and immediately kill the subscription
    const subscription = source$.subscribe(
      (v) => {
        observer.next(v)
      },
      (e) => {
        observer.error(e)
      },
      () => {
        observer.complete()
      }
    )
    cleanupFn()

    return () => {
      cleanupFn = () => {
        clearTimeout(token)
        cleanupFn = noop
        subscription.unsubscribe()
      }
      token = setTimeout(cleanupFn, ms)
    }
  })
}

Let's put it together:

@Component({
  selector: "my-app",
  templateUrl: "./app.component.html",
  styleUrls: ["./app.component.css"]
})
export class AppComponent {
  name = "Angular " + VERSION.major;

  urlofApi = "https://api.github.com/search/repositories?q=helloWorld";

  testX = concat(
    this.http.get(this.urlofApi),
    NEVER
  ).pipe(
    tap(() =>
      console.log(
        `++ api was newly called at ${new Date().toLocaleTimeString()} ++`
      )
    ),
    shareReplay({
      bufferSize: 1,
      refCount: true
    }),
    delayLatestUnsubscription(5000),
  );

  constructor(private http: HttpClient) {}

  test() {
    console.log("Requesting data from API");
    this.testX.pipe(take(1)).subscribe(
      (value) => {
        console.log("value received", value)
      }
    );
  }
}

I think that this solution has the behavior that you were expecting and it works on both RxJS 6 and 7. I'm sure that there are other ways to get the same behavior, but this is the first one that came to mind. I hope this helps.

[josepot]

It returns the same value because the source is emitting the same data this.http.get(this.urlofApi)

If you replace that request with a deferred value that increases every time that there is a new subscription, then you can see that it works properly 🙂

[josepot]

Oh, I think that I misunderstood the problem: you want for the cache to live 5 seconds after it was created, not to stay around for 5 seconds after the last time that it was used... I see... My bad!

[StefanKern]

Haha yeah. Should time out after 5 seconds, then refech the API. Just like the browser cache would behave

[josepot]

Should time out after 5 seconds, then refech the API. Just like the browser cache would behave

Then, with RxJS 7 this does it:

@Component({
  selector: "my-app",
  templateUrl: "./app.component.html",
  styleUrls: ["./app.component.css"]
})
export class AppComponent {
  name = "Angular " + VERSION.major;

  urlofApi = "https://api.github.com/search/repositories?q=helloWorld";

  testX = concat(
    this.http.get(this.urlofApi),
    timer(5000).pipe(ignoreElements()),
  ).pipe(
    tap(() =>
      console.log(
        `++ api was newly called at ${new Date().toLocaleTimeString()} ++`
      )
    ),
    share({
      connector: () => new ReplaySubject(1),
      resetOnError: true,
      resetOnComplete: true,
      resetOnRefCountZero: false
    }),
    take(1),
  );

  constructor(private http: HttpClient) {}

  test() {
    console.log("Requesting data from API");
    this.testX.subscribe((x) => console.log("completed", x));
  }
}

[josepot]

And with RxJS 6 (if you don't want to rely on a bug 😅) you could do this:

const DONE = {}

@Component({
  selector: "my-app",
  templateUrl: "./app.component.html",
  styleUrls: ["./app.component.css"]
})
export class AppComponent {
  name = "Angular " + VERSION.major;

  urlofApi = "https://api.github.com/search/repositories?q=helloWorld";

  testX = concat(
    this.http.get(this.urlofApi),
    timer(5000).pipe(ignoreElements()),
    throwError(() => DONE),
  ).pipe(
    tap(() =>
      console.log(
        `++ api was newly called at ${new Date().toLocaleTimeString()} ++`
      )
    ),
    shareReplay(1),
    catchError(e => e === DONE ? [] : throwError(() => e)),
    take(1)
  );

  constructor(private http: HttpClient) {}

  test() {
    console.log("Requesting data from API");
    this.testX.subscribe(() => console.log("completed"));
  }
}

[cartant]

FWIW - as I mentioned in the issue you opened, I don't really have the time to give you a proper answer, ATM - this refCountDelay implementation might be close to what you are looking for, but I think you'll have to use something other than publish - as it won't reset the subject. Probably multicast with a factory to create the subject, instead. And when I get around to reviewing it - I have higher priorities, right now - something similar will be added to v7 - see #6169.

[StefanKern]

@jsayol that looks very cool. Unless something changes I think we will go with it when upgrading to v7 :)

@cartant if you find the time, I would be very happy if you have a look too :)

Thanks for the great comunity support :D