Immediate observable not firing through Observable.defer

Question

I'm trying to use RxJS caching to avoid unnecessary repetion of certain HTTP calls. While experimenting with publishReplay, I ended up with the following snippet (inspired by this blog post):

let counter = 1;
const updateRequest = Observable.defer(() => mockDataFetch())
  .publishReplay(1, 1000)
  .refCount();

function mockDataFetch() {
  return Observable.of(counter++)
    .delay(0); // <-- delay by 0 milliseconds
}

function mockHttpCache() {
  return updateRequest
    .take(1);
}

setTimeout(() => mockHttpCache().subscribe(val => console.log("Response 50:", val)), 50);
setTimeout(() => mockHttpCache().subscribe(val => console.log("Response 500:", val)), 500);
setTimeout(() => mockHttpCache().subscribe(val => console.log("Response 1500:", val)), 1500);

This works as expected and yields the output:

'Response 50:', 1
'Response 500:', 1
'Response 1500:', 2

However, when removing .delay(0) from the inner observable, making it immediate, the wrapper yields no more results after the caching duration has passed. The output is then:

'Response 50:', 1
'Response 500:', 1

It looks like mockDataFetch doesn't get called to gather new data even if there's no cached item anymore. Is this intended behaviour, and if so, what's the rationale behind it?

user_9632621 · Answer

Here's your code translated into RxJs 6.5.5, along with some other small modifications:

let counter = 1;
const updateRequest = defer(() => mockDataFetch())
  .pipe(
    publishReplay(1, 1000),
    refCount()
  );

function mockDataFetch() {
  console.log('RESUBSCRIBING');

  return of(counter++)
    .pipe(
      // delay(0), // <-- delay by 0 milliseconds
    );
}

function mockHttpCache() {
  return updateRequest
    .pipe(
      take(1),
    );
}

setTimeout(
  () => mockHttpCache().subscribe(val => console.log("Response 50:", val), null, () => console.warn('complete[1]')
), 50);
setTimeout(
  () => mockHttpCache().subscribe(val => console.log("Response 500:", val), null, () => console.warn('complete[2]')
), 500);
setTimeout(
  () => mockHttpCache().subscribe(val => console.log("Response 1500:", val), null, () => console.warn('complete[3]')
), 1500);

StackBlitz.

Without `delay(0)`

Let's first have a look at how publishReplay is implemented:

const selector = typeof selectorOrScheduler === 'function' ? selectorOrScheduler : undefined;
const subject = new ReplaySubject(bufferSize, windowTime, scheduler);

return (source: Observable) => multicast(() => subject, selector!)(source) as ConnectableObservable;

As we can see, it returns a ConnectableObservable, due to multicast:

const connectable: any = Object.create(source, connectableObservableDescriptor);
connectable.source = source;
connectable.subjectFactory = subjectFactory;

return > connectable;

And this is how refCount looks like:

// `connectable` - the `ConnectableObservable` from above
constructor(private connectable: ConnectableObservable) { }

// `call` - called when the source is subscribed
// `source` - the `ConnectableObservable` from above
call(subscriber: Subscriber, source: any): TeardownLogic {
  const { connectable } = this;
  ( connectable)._refCount++;

  const refCounter = new RefCountSubscriber(subscriber, connectable);
  const subscription = source.subscribe(refCounter);

  if (!refCounter.closed) {
    ( refCounter).connection = connectable.connect();
  }

  return subscription;
}

Now, let's have a closer look at ConnectableObservable, especially at the subscribe method:

// Invoked as a result of `const subscription = source.subscribe(refCounter);` from `refCount`
_subscribe(subscriber: Subscriber) {
  return this.getSubject().subscribe(subscriber);
}

protected getSubject(): Subject {
  const subject = this._subject;
  if (!subject || subject.isStopped) {
    this._subject = this.subjectFactory();
  }
  return this._subject!;
}

Where subjectFactory returns a ReplaySubject instance. What basically happens on const subscription = source.subscribe(refCounter); is that a RefCounterSubscriber will be added to the ReplaySubject's list of active subscribers. A RefCounterSubscriber keeps track of the number of subscribers and when the are no more subscribers, it will automatically subscribe to the source when a new subscriber is registered(while using the same ReplaySubject) instance.

Next, ( refCounter).connection = connectable.connect(); will be invoked.

connectable.connect() does the following:

  connect(): Subscription {
  let connection = this._connection;
  if (!connection) {
    this._isComplete = false;
    connection = this._connection = new Subscription();
    connection.add(this.source
      .subscribe(new ConnectableSubscriber(this.getSubject(), this)));
    if (connection.closed) {
      this._connection = null;
      connection = Subscription.EMPTY;
    }
  }
  return connection;
}

When these lines are reached:

connection.add(this.source
    .subscribe(new ConnectableSubscriber(this.getSubject(), this)));

the source(e.g mockDataFetch()) will actually be subscribed.

Now, of(counter) is implemented like this(roughly):

// In this case, `arr = [counter]`
new Observable(subscriber => {
  for (let i = 0; i < arr.length; i++) {
    subscriber.next(arr[i]);
  }

  subscriber.complete();
});

This means that take(1) will be reached first and when it happens, it will emit that value and then it will send a complete notification(by invoking Subscriber._complete() eventually):

protected _complete(): void {
  this.destination.complete();
  this.unsubscribe();
}

So, besides sending the complete notification further in the chain, it will also unsubscribe. It will eventually reach the RefCounterSubscriber's unsubscription logic, but it won't work as expected, because everything is happening synchronously. In normal circumstances, if the ReplaySubject was left without any subscribers, the source would be unsubscribed.

But since it's being left without subscribers while subscribing to the source, there will be a slightly different behavior. The ReplaySubject's subscribers list will be empty, but the source won't be unsubscribed, because, as mentioned above, it is still in the subscription process.

What it means in the end is that subscriber.complete(); will be called, which in turn will cause the ReplaySubject to receive a complete notification. But remember that the same ReplaySubject will be used when the source would be re-subscribed.

Next time when it would subscribe to the source again, these lines would be reached:

const refCounter = new RefCountSubscriber(subscriber, connectable);
// Subscribing to a **completed** Subject
// If the Subject is completed, an EMPTY subscription will be reached
const subscription = source.subscribe(refCounter);

if (!refCounter.closed) { // Since `closed === true`, this block won't be reached
  ( refCounter).connection = connectable.connect();
}

// Returning the EMPTY subscription
return subscription;

EMPTY implementation.

This would be the flow of the program, without delay(0)

setTimeout(
  // Source emits and the value is cached by the subject for 1 second
  // `take(1)` is reached
  // Send the value, then a `complete` notif.
  // But since sending a `complete` notif involves unsubscribing as well
  // The current subscriber will be removed from the `ReplaySubject`'s subscribers list
  // Then, the `ReplaySubject` will receive the `complete` notification and the subject becomes **completed**
  () => mockHttpCache().subscribe(val => console.log("Response 50:", val), null, () => console.warn('complete[1]')
), 50);
setTimeout(
  // Subscribing to a **completed** subject, but because it's a `ReplaySubject`
  // We'd still be getting the cached values, along with a `complete` notification
  () => mockHttpCache().subscribe(val => console.log("Response 500:", val), null, () => console.warn('complete[2]')
), 500);
setTimeout(
  // Since `1`'s time expired at 1 second, the `ReplaySubject` will only send a complete notification
  () => mockHttpCache().subscribe(val => console.log("Response 1500:", val), null, () => console.warn('complete[3]')
), 1500);

This is what will get logged:

RESUBSCRIBING
Response 50:
1
complete[1]
Response 500:
1
complete[2]
complete[3]

With `delay(0)`

This relies on some details mentioned in the previous section.

delay(0) will schedule an action in the AsyncScheduler(by default) on every nexted notification. The action's task is to emit that received value after 0 ms have passed. It's essentially the same as using setTimeout, which means it will not be something synchronous.

However, when using of(), the complete notification will be sent synchronously. This is how delay deals with it:

protected _complete() {
  // `this.queue` is populated when a `nexted` value arrives
  if (this.queue.length === 0) {
    this.destination.complete();
  }

  // Unsubscribe from the previous items from the chain
  // What's further will **not** be affected
  this.unsubscribe();
}

The complete notification will eventually be sent when the queue is empty. But keep in mind that this is all asynchronous, meaning that the RefCountSubscriber will behave normally.

This would be the flow of the program, with delay(0):

setTimeout(
  // Subscribing to the source, which emits a value and a complete notif, synchronously
  // `delay` schedules an action that will do its job in 0ms(still asynchronously)
  // The value is emitted by the `delay`'s scheduled action
  // `take(1)` is reached
  // The value will be passed along then a `complete` notif will be sent
  // Then, the source will be unsubscribed
  // Due to `refCount`, the complete notif that came from the source
  // Won't reach the `ReplaySubject`. as it will already be unsubscribed from the source
  () => mockHttpCache().subscribe(val => console.log("Response 50:", val), null, () => console.warn('complete[1]')
), 50);
setTimeout(
  // Since only `500ms` have passed, this subscriber will receive the cached value (`1`)
  // and a `complete` notification, due to `take(1)`
  // But since `take(1)` operates synchronously, the `RefCountSubscriber` would be closed already, so the source won't be re-subscribed (//1)
  () => mockHttpCache().subscribe(val => console.log("Response 500:", val), null, () => console.warn('complete[2]')
), 500);
setTimeout(
  // `1500ms` passed, since `1000ms` the cache is empty
  // So the `take(1)` operator will receive nothing, meaning that the source
  // will be re-subscribed
  () => mockHttpCache().subscribe(val => console.log("Response 1500:", val), null, () => console.warn('complete[3]')
), 1500);

Output:

RESUBSCRIBING
Response 50:
1
complete[1]
Response 500:
1
complete[2]
RESUBSCRIBING
Response 1500:
2
complete[3]

//1 in order to see that the RefCountSubscriber is closed, you can open the dev tools in your SB project, press CTRL + P, type refCount.ts and place a logpoint on line 78(e.g: 'refCounter.closed', refCounter.closed):

if (!refCounter.closed) { /* ... */ }

and if you comment out the last setTimeout(() => {}, 1500), you should see something like this:

refCounter.closed
false
RESUBSCRIBING
Response 50:
1
complete[1]
Response 500:
1
complete[2]
refCounter.closed // Closed due to `take(1)`
true

Immediate observable not firing through Observable.defer

1 Answers

Without delay(0)

With delay(0)

Without `delay(0)`

With `delay(0)`