Using Share with observables derived from each other

[jarkkoskyttala]

Hi,

The problem is that when I subscribe multiple times, only the first subscribtion is going through as expected. The second subscription triggers with a previous value from interval$.
If I don't call share() on trigger$ then behavior is as expected and all triggers have correct values. Obviously in this case it doesn't matter if trigger$ is multicasted but in application code it is necessary.

This is a piece of code extracted and simplified from our application.

let count = 0;

const interval$ = interval(100).pipe(
  map(() => count++),
  take(10),
  share()
);

const trigger$ = interval$.pipe(
  filter((val) => val === 5),
  map((val) => {
    return `trigger_${val}`;
  }),
  share()
);

const test$ = trigger$.pipe(withLatestFrom(interval$)));

test$.subscribe((val) => console.log('test$', val));
test$.subscribe((val) => console.log('test$', val));

/**
 * console:
 * test$ [ 'trigger_5', 5 ]
 * test$ [ 'trigger_5', 4 ] <--- This should be identical to above
 */

Here is another example where if I subscribe to test$ first I get expected behaviour but if I subscribe to test$ last I get previous value from interval$ in my test$ observable

let count = 0;

const interval$ = interval(100).pipe(
  map(() => count++),
  take(10),
  share()
);

const trigger$ = interval$.pipe(
  filter((val) => val === 5),
  map((val) => {
    return `trigger_${val}`;
  }),
  share()
);

const test$ = trigger$.pipe(withLatestFrom(interval$));

interval$.subscribe((val) => console.log('interval$', val));
trigger$.subscribe((val) => console.log('trigger$', val));
test$.subscribe((val) => console.log('test$', val));

/**
 * console:
 * interval$ 0
 * interval$ 1
 * interval$ 2
 * interval$ 3
 * interval$ 4
 * interval$ 5
 * trigger$ trigger_5
 * test$ [ 'trigger_5', 4 ] <--- value at index 1 should be 5. This is correct if I subscribe to test$ first
 * interval$ 6
 * interval$ 7
 * interval$ 8
 * interval$ 9
 */

[voliva]

This is something that often catches many people off-guard. There's a related discussion here: #6605

It has to do with the order of subscriptions.

In your first case, as soon as you subscribe to test$, the list of subscribers for each shared observable would be:

interval$: [withLatestFrom_0, trigger$]
trigger$: [withLatestFrom_0]

When the second subscription to test$ happens, the list of subscriptions becomes:

interval$: [withLatestFrom_0, trigger$, withLatestFrom_1]
trigger$: [withLatestFrom_0, withLatestFrom_1]

Now, when interval$ hits the value 5, it will notify each of their subscribers in order:

1. interval$ notifies withLatestFrom_0: The first console.log subscription gets that the latest value from interval is 5.
2. interval$ notifies trigger$
3. trigger$ notifies withLatestFrom_0 as its source:
4. withLatestFrom_0 emits [trigger_5, 5] (the value from the source + the latest value from the parameter)
5. the first subscription logs [trigger_5, 5]
6. trigger$ notifies withLatestFrom_1 as its source
7. withLatestFrom_1 emits [trigger_5, 4]. That's now not the desired behaviour, but it's just following the synchronous chain of subscriptions.
8. the second subscription logs [trigger_5, 4]
9. interval$ notifies withLatestFrom_1, and that retains the value but it's too late

Something similar is happening on your second case.

I think if you also were to share test$ it would work. Often subscribeOn or observeOn with another scheduler (e.g. asapScheduler or asyncScheduler) could also help, but the solution depends quite a lot on your specific case.

[jarkkoskyttala]

Hi @voliva,

Thanks for the explanation, it is good to know that it's working as intended and there is a logic. I think in our application we will try to solve these cases by using observeOn(asapScheduler) which I think is a reasonable solution. For us the trickiest part is identifying these cases since this kind of problem is usually pretty hard to detect. (We just migrated our app from another reactive stream library to rxjs)

Regards,

Jarkko