Why does the rxjs share operator not work as expected in this setTimeout() example?

user_9332260 0
Viewed 326

I don't understand why the rxjs share operator does not work with setTimeout().

I'm trying to understand this blogpost. In this example, the concept of "shared subscription" does not seem to work as expected.

const observable1 = Observable.create(observer => {
  observer.next(`I am alive.`);
  setTimeout(() => {
    observer.next(`I am alive again.`);
  }, 1000);
}).pipe(share());

observable1.subscribe(x => console.log(x));
observable1.subscribe(x => console.log(x));

Expected:

I am alive.
I am alive again.

Actual:

I am alive.
I am alive again.
I am alive again.

Reproducable stackblitz.

rxjs typescript javascript
2 Answers

That is the expected output.

From official docs on share() operator:

Returns a new Observable that multicasts (shares) the original Observable. As long as there is at least one Subscriber this Observable will be subscribed and emitting data.

That means as soon as an observer subscribes, the observable starts emitting data.

So when the first subscribe statement observable1.subscribe(x => console.log(x)); executes, an observer subscribes and data is emitted by observer.next('I am alive.); statement.

When second subscribe statement executes, another observer subscribes and it receives only the data emitted from that point of time. This is the data emitted by observer.next('I am alive again.'); in setTimeout() method.

We can see this clearly in this StackBlitz demo where we are logging Observer1 and Observer2 text along with the received data.

I think the point of confusion is seeing two I am alive again. statements. It is logged twice because we are logging it in each subscriber. Move these log statements to the observable and they will only be logged once. This makes it more evident that the observable is executed only once.

user_29245770

This is the supposed behaviour of share(). It monitores and shares only one action. Here is an example taken from learnrxjs.com. As you can see only the tap()-operator is monitored. The mapTo()-operator is ignored.

// RxJS v6+
import { timer } from 'rxjs';
import { tap, mapTo, share } from 'rxjs/operators';

//emit value in 1s
const source = timer(1000);
//log side effect, emit result
const example = source.pipe(
   tap(() => console.log('***SIDE EFFECT***')),
   mapTo('***RESULT***')
);

/*
  ***NOT SHARED, SIDE EFFECT WILL BE EXECUTED        
 TWICE***
  output:
  "***SIDE EFFECT***"
  "***RESULT***"
  "***SIDE EFFECT***"
  "***RESULT***"
*/
const subscribe = example.subscribe(val => console.log(val));
const subscribeTwo = example.subscribe(val => console.log(val));

//share observable among subscribers
const sharedExample = example.pipe(share());
/*
  ***SHARED, SIDE EFFECT EXECUTED ONCE***
  output:
  "***SIDE EFFECT***"
  "***RESULT***"
  "***RESULT***"
*/
const subscribeThree = sharedExample.subscribe(val => console.log(val));
const subscribeFour = sharedExample.subscribe(val => console.log(val));
Admin1
Related