The replacement of publishReplay + refCount seems broken

[JounQin]

We've written a custom operator which wraps publishReplay + refCount, and it was just working perfectly.

export const publishRef = <T>(bufferSize = 1, windowTime?: number) =>
  pipe(publishReplay<T>(bufferSize, windowTime), refCount());

But after v7, it is recommended to migrate to share: https://rxjs.dev/deprecations/multicasting#publishreplay

So we changed the implementation following the recommendation:

export type PublishRefConfig<T> =
  | number
  | (import('rxjs/internal/operators/share').ShareConfig<T> & {
      bufferSize?: number;
      windowTime?: number;
      timestampProvider?: TimestampProvider;
    });

export const publishRef = <T>(bufferSizeOrConfig: PublishRefConfig<T> = {}) => {
  const {
    bufferSize = 1,
    windowTime,
    timestampProvider,
    connector = () =>
      new ReplaySubject(bufferSize, windowTime, timestampProvider),
    resetOnError = false,
    resetOnComplete = false,
    resetOnRefCountZero = false,
  } = typeof bufferSizeOrConfig === 'number'
    ? ({
        bufferSize: bufferSizeOrConfig,
      } as Exclude<PublishRefConfig<T>, number>)
    : bufferSizeOrConfig;
  return share<T>({
    connector,
    resetOnError,
    resetOnComplete,
    resetOnRefCountZero,
  });
};

But in our Angular application, it seems working differently:

When we are consuming a watching stream only on Angular template via async pipe, and after navigating to other rules, the stream should be completed automatically, but the current implementation does not work as before. I don't know is our new implementation correct? Or it's just incorrect documentation.

Sorry I can't reproduce it easily because it's in our own business codes, I'll try to reproduce further, but I still want to let you know it firstly.

[voliva]

I think your example is a bit hard to follow because you have changed the signature of publishRef, which now can have completely different behaviours based on the parameters you pass in... why don't you try to just change the inner part of publishRef so that it behaves exactly as it did in v6?

Wouldn't something like this work?

export const publishRef = <T>(bufferSize = 1, windowTime?: number) =>
  share<T>({
    connector: () => new ReplaySubject(bufferSize, windowTime),
    resetOnComplete: false,
  });

What this does is multicast the subscription by using a replay subject, which will get reset when everyone unsubscribes, but not when the source completes. The only problem this code has, if I'm not wrong, is that windowTime won't work in some cases, because share will reset the subject... this could probably be solved with #6430

Again, I think it will be easier to reason about if you keep the signature of publishRef the same as before.

[backbone87]

you could try the following, which should do what you want:

share({
  connector: () => new ReplaySubject(1),
  resetOnComplete: () => timer(cacheDuration),
  resetOnError: true, // can be left out since its the default, could also something like () => timer(cooldown)
  resetOnRefCountZero: false,
})

[sherlock1982]

I followed the migration guide and now ended up with a strange-working app untils I found this discussion.
I used publishReplay(1)+refCount for a simple share of the last value.
When refCount zero it should reset.

For me the good replacement will be:

export const publishRef = <T>() => (source: Observable<T>) => source.pipe(share({
    connector: () => new ReplaySubject(1),
    resetOnError: false,
    resetOnComplete: false,
}));

Well resetOnError & resetOnComplete I don't think I really care (my observables theoretically doesn't error or complete) but for the sake of purity I left it.

But resetOnRefCountZero: false mentioned in migration guide is absolutely wrong.

[ntziolis]

Same use case as @sherlock1982, when looking at the migration guide and saw resetOnRefCountZero: false I got suspicious and found this thread.

The entire reason we were using publishReplay(1) followed by refCount() was to:

• ensure that the base cold observable would only get subscribed once
• and last result being 'cached' for additional subscribers
• All this should happen only WHILE and only while ANY subscribers are still subscribed
• Once all subscribers are unsubscribed the base observable should be unsubscribed from (route change etc.)
• Note:
  • As long as the underlying observables are things like simple web request that complete after the first value is emitted you will not run into this issue
  • However in our case the base observables are long/forever-living observables that would never complete (graphql subscriptions)

We actually migrated away from shareReplay for this exact reason years ago. From what I can tell, the migration guides recommendation resembles the behavior of shareReplay instead of publishReplay(1) + refCount().

Since we are using this EVERYWHERE I'm keen to understand what the correct settings are for the other two parameters in order to match the previous behavior 100%:

• resetOnError
• resetOnComplete

The behavior we want to achieve is:

• After all previous subscribers are unsubscribed
• The state should be as if the observable had never been subscribed to
• The next subscriber should see the exact same behavior as the initial subscriber

Any input would be highly appreciated.

I have attached a repro that illustrate that the current recommendation does not achieve the same effect as publishReplay(1) + refCount():
https://stackblitz.com/edit/rxjs-1rnn3g?file=index.ts

[ntziolis]

In addition the signature of share is misleading:

• When not setting resetOnRefCountZeroat all it behaves the same way as resetOnRefCountZero: true.
• The property is optional + boolean and hence I would expect not specifying it to have the same effect as false

I would suggest to either

• change the behavior to default to the same as when false is specified explicitly
• or extend the signature to call out that true is the default when not specified

I'm not sure if this also applies to resetOnError and resetOnComplete, but it would make sense to have a look at the implementation to ensure consistent behavior.

[NoMoreViolence]

import { interval, Observable, ReplaySubject, share, tap } from "rxjs";

// Replacement of publishReplay(1) + refCount()
function publishRef<T>(source$: Observable<T>) {
  const replaySubject = new ReplaySubject<T>(1);
  return source$.pipe(
    share({
      connector: () => replaySubject,
      resetOnError: false,
      resetOnComplete: false,
      resetOnRefCountZero: true
    })
  );
}

const pipe$ = interval(1000).pipe(
  tap(() => {
    console.log("share pipe runed");
  }),
  publishRef
);

const { unsubscribe } = pipe$.subscribe((value) => {
  console.log(value, "1");
});

setTimeout(() => {
  unsubscribe(); 
}, 4000);

In that case, If the pipe$ will not be used anymore(remove or anything), the replaySubject (in publishRef function) will be garbage collectible (will complete method be called to replaySubject?) ??

[mfp22]

It's been a year, but I'm finding that what I really want is not possible in a simple way and probably never has been. It's like a simple version of what TanStack Query does.

I used to think publishReplay(1), refCount() was exactly what I wanted, because shareReplay(1) is like an immovable object once a single subscriber has existed. publishReplay(1), refCount(), however, will unsubscribe from the source if refCount goes to zero. So this would cancel an Angular HTTP request. This is great, and that's why I believed it was good enough.

However, what I really want is to abstract away the fact that there are multiple subscribers, and not have multiple network requests triggered. And the behavior of a single self-completing observable is that if you subscribe again, it runs again. Obviously this is desirable, because why wouldn't you want fresh data when subscribing to an observable that everybody stopped caring about a long time ago?

So I've been playing around with the new API, which I was reluctant to try for a long time, because it's so verbose. But it seems like I am forced to choose between 2 suboptimal behaviors: After source completes, either I have resetOnComplete: true where every new subscription will cause a new source subscription, and then it doesn't even update old subscribers with the results, so I get inconsistent state in my UI; OR, when all my components unsubscribe, the shared state doesn't reset even with resetOnRefCountZero: true, because it already completed and that forced the refCount to go to 0 already.

So this is my problem: When the source completes, I don't want the destination observers to complete. I want the refCount to stay where it is until they unsubscribe. I really do want resetOnRefCountZero: true, but I don't want refCount to go to 0 when the source completes.

But there are no options for this in share. In the source code (v7+), when source completes, it automatically calls dest.complete():

          complete: () => {
            hasCompleted = true;
            cancelReset();
            resetConnection = handleReset(reset, resetOnComplete);
            dest.complete();
          },

I'm writing an article about this right now, and this isn't a very satisfying conclusion :/

I want there to be an option like propagateCompletion = true that I can pass false into to keep the observers around even if the source completes. Using that in conjunction with an observable returned from resetOnRefCountZero would be so nice. Almost as nice as TanStack Query.

[mfp22]

This gets me exactly what I want :)

function shareState<T>(source$: Observable<T>) {
  const replaySubject = new ReplaySubject<T>(1);
  const sharedSource$ = merge(source$, NEVER).pipe(
    tap((result) => replaySubject.next(result)),
    share()
  );
  return using(
    () => sharedSource$.subscribe(),
    () => replaySubject
  );
}

I realized I had already figured these tricks out while working on StateAdapt :)

[voliva]

Yep, that's exactly it.... I was going to comment that you could just make the observable not emit a complete by concatting NEVER.

I guess this is fine, but I think that you can achieve the same just by using share with the new API instead of tap + using... wouldn't this work?:

function shareState<T>(source$: Observable<T>) {
  return merge(source$, NEVER).pipe(
    share({
      resetOnRefCountZero: true,
      connector: () => new ReplaySubject<T>(1)
    })
  );
}

Actually, I think your solution won't work when refCount goes to zero, because the ReplaySubject is not getting re-created, so it will keep whatever value it had on the previous call ⚠️

You will also need to think what happens in case of error.

[mfp22]

Yeah that's perfect. I always go for using when I want to do something hacky, but this is very simple and I think it's better not to replay the previous value. Thank you!