RxJS share vs shareReplay differences

user_106376 0
Viewed 2147

There seems to be an odd discrepancy with how share and shareReplay (with refcount:true) unsubscribe.

Consider the following (can paste into rxviz.com):

const { interval } = Rx;
const { take,  shareReplay, share, timeoutWith, startWith , finalize} = RxOperators;

const shareReplay$ = interval(2000).pipe(
  finalize(() => console.log('[finalize] Called shareReplay$')), 
  take(1), 
  shareReplay({refcount:true, bufferSize: 0}));

shareReplay$.pipe(
  timeoutWith(1000, shareReplay$.pipe(startWith('X'))), 
)



const share$ = interval(2000).pipe(
  finalize(() => console.log('[finalize] Called on share$')),
  take(1), 
  share());

share$.pipe(
  timeoutWith(1000, share$.pipe(startWith('X'))), 
)

The output from streamReplay$ will be -X-0- while the output from shareStream$ will be -X--0. It appears as though share unsubscribes from the source before timeoutWith can re-subscribe, while shareReplay manages to keep the shared subscription around for long enough for it to be re-used.

I want to use this to add a local timeout on an RPC (while keeping the call open), and any re-subscribe would be disastrous, so want to avoid the risk one of these behaviors is a mistake and gets changed in the future.

I could use race(), and merge the rpc call with a delayed startsWith (so they both subscribe at the same time) but it would be more code and operators.

EDIT: One possible solution is to merge two subscriptions one to a shared request and one to a delayed stream that takes until the shared stream emits:

merge(share$, of('still working...').pipe( delay(1000), takeUntil(share$)));

This way the shated stream is subscribed to at the same time, so there is no "grey area" when one operator unsubscribes as a child subscribes. (Will turn this into an answer unless someone comes up with a better suggestion) or can explain the intentions/differences between share and shareReplay

rxjs javascript
2 Answers

Timing Guarantees

Javascript's runtime doesn't come with baked in timing guarantees of any sort. In a single threaded environment, that makes sense (and you can start to do a bit better with web workers and such). If something compute-heavy happens, everything covered in that timing window just waits. Mostly it happens in the expected order though.

Regardless,

const hello = (name: string) => () => console.log(`Hello ${name}`);

setTimeout(hello("first"), 2000);
setTimeout(hello("second"), 2000);

In Node, V8, or SpiderMonkey you do preserve the order here. But what about this?

const hello = (name: string) => () => console.log(`Hello ${name}`);

setTimeout(hello("first"), 1);
setTimeout(hello("second"), 0);

Here you would assume second always comes first because it's supposed to happen a millisecond earlier. If you run this using SpiderMonkey, the order depends on how busy the event loop is. They bucket shorter timeouts since a 0ms timeout takes about 8ms on average anyway.

Always make asynchronous dependencies explicit

In JavaScript, it's best practice to never ever make any timing dependencies implicit.

In the code below, we can reasonably know that that data will not be undefined when we call data.value. This implicitly relies on asynchronous interleaving:

let data;
setTimeout(() => {data = {value: 5};}, 1000);
setTimeout(() => console.log(data.value), 2000);

We really should make this dependency explicit. Either by checking if data is undefined or by restructuring our calls

setTimeout(() => {
  const data = {value: 5};

  setTimeout(() => console.log(data.value), 1000);
}, 1000);

Share vs ShareReplay

want to avoid the risk one of these behaviors is a mistake and gets changed in the future.

The real risk here isn't even in how the library implements the difference. It's a language-level risk as well. You're depending on asynchronous interleaving either way. You run the risk of having a bug that only appears once in a blue moon and can't be easily re-created/tested etc.

The share operator has a ShareConfig (source)

export interface ShareConfig<T> {
  connector?: () => SubjectLike<T>;
  resetOnError?: boolean | ((error: any) => Observable<any>);
  resetOnComplete?: boolean | (() => Observable<any>);
  resetOnRefCountZero?: boolean | (() => Observable<any>);
}

If you use vanilla shareReplay(1) or replay({resetOnRefCountZero: false}) then you aren't relying on how events are ordered in the JS Event Loop.

user_135009860

Quick Aside:

I suspect you're not going to get many answers until you describe the behavior you're after.

use this to add a local timeout on an RPC (while keeping the call open)

I'm not sure what keeping an RPC call opens entails. It's sounds like you want something that cancels the request within RxJS but doesn't cancel the inflight RPC. I feel like that's a domain mismatch. Why not keep your observables aligned with the semantics of the calls they represent?

Toward a solution

Having seen your update, I suspect you don't need a timeout. It looks like the behavior you're after can be achieved without unsubscribing to the observable representing your RPC. I'd argue that doing so simplifies your logic and makes it easier to maintain/extend in the future.

It also sidesteps all the async interleaving concerns you had earlier.

Here I'll take an educated guess at the behaviour you're after. It appears like you want to following:

  1. Take one value from the source (happens to be an RPC in this case).
  2. If that value takes more than 1s to arrive, emit `"still working..." and continue to wait for one value from the source.

If that's the case, you don't need share at all. Normally a timeout means to cancel an inflight request if it takes too long. Maybe you don't want a timeout at all, you want to stay subscribed to the source...

If that's the case, here's a way to do this:

const source$ = rpc(arg1 arg2);
// create a unique token (its memory address).
// We'll embed a message inside so it's doing double duty
const token = {a: "still working..."};

merge(
  source$,
  timer(1000).pipe(mapTo(token))
).pipe(
  // take(1), but ignoring the token
  takeWhile(v => v === token, true),
  // Unwrap the token, emit the contained message
  map(v => v === token ? v.a : v)
);

On the other hand, if you know that observable wrapping your RPC will never emit "still working..." then you don't need a unique token and you can simplify this to check the value directly.

merge(
  rpc(arg1 arg2),
  timer(1000).pipe(mapTo("still working..."))
).pipe(
  takeWhile(v => v === "still working...", true)
);
user_135009860
Related