Multiple subscriptions to Observable

Question

I create my own Observable and subscribed two functions to it. I would expect to have both functions executed for each element in the sequence but only the last one is.

let observer = null
const notificationArrayStream = Rx.Observable.create(function (obs) {
  observer = obs;
  return () => {}
})

function trigger(something) {
  observer.next(something)
}

notificationArrayStream.subscribe((x) => console.log('a: ' + x))
notificationArrayStream.subscribe((x) => console.log('b: ' + x))

trigger('TEST')

Expected output

a: TEST
b: TEST

Actual output

b: TEST

Here's the JSBin: http://jsbin.com/cahoyey/edit?js,console

Why is that? How can I have multiple functions subscribed to a single Observable?

Answer 1

Subject

In your case, you could simply use a Subject. A subject allows you to share a single execution with multiple observers when using it as a proxy for a group of subscribers and a source.

In essence, here's your example using a subject:

const subject = new Subject();

function trigger(something) {
    subject.next(something);
}

subject.subscribe((x) => console.log('a: ' + x));
subject.subscribe((x) => console.log('b: ' + x));

trigger('TEST');

Result:

a: TEST
b: TEST

---

Pitfall: Observers arriving too late

Note that the timing of when you subscribe and when you broadcast the data is relevant. If you send a broadcast before subscribing, you're not getting notified by this broadcast:

function trigger(something) {
    subject.next(something);
}

trigger('TEST');

subject.subscribe((x) => console.log('a: ' + x));
subject.subscribe((x) => console.log('b: ' + x));

Result: (empty)

---

ReplaySubject & BehaviorSubject

If you want to ensure that even future subscribers get notified, you can use a ReplaySubject or a BehaviorSubject instead.

Here's an example using a ReplaySubject (with a cache-size of 5, meaning up to 5 values from the past will be remembered, as opposed to a BehaviorSubject which can remember only the last value):

const subject = new ReplaySubject(5); // buffer size is 5

function trigger(something) {
    subject.next(something);
}

trigger('TEST');

subject.subscribe((x) => console.log('a: ' + x));
subject.subscribe((x) => console.log('b: ' + x));

Result:

a: TEST
b: TEST