[RxJS] Subject: an Observable and Observer hybrid

Posted 2020-08-12

This lesson teaches you how a Subject is simply a hybrid of Observable and Observer which can act as a bridge between the source Observable and multiple observers, effectively making it possible for multiple observers to share the same Observable execution.

 

var observable = Rx.Observable.interval(1000).take(5);

var observerA = {
  next: function (x) { console.log(‘A next ‘ + x); },
  error: function (err) { console.log(‘A error ‘ + err); },
  complete: function () { console.log(‘A done‘); },
};


var observerB = {
  next: function (x) { console.log(‘B next ‘ + x); },
  error: function (err) { console.log(‘B error ‘ + err); },
  complete: function () { console.log(‘B done‘); },
};

observable.subscribe(observerA);

setTimeout(
  () => {
    observable.subscribe(observerB);
  },2000
)

In the code above, we have two ‘observers‘, because we call subscribe twice:

observable.scbscribe(ObserverA);
observable.scbscribe(ObserverB);

 

If we want to have one observer, so we need to call subscribe only once.

For that we can build a bridgeObservers, which will loop though the observers:

const observable = Rx.Observable.interval(1000).take(5);

const ObserverA = {
  next: function(x){
    console.log("A next " + x)
  },
  error: function(x){
    console.error("A error " + x)
  },
  complete: function(){
    console.log("A Done")
  },
};


const ObserverB = {
  next: function(x){
    console.log("B next " + x)
  },
  error: function(x){
    console.error("B error " + x)
  },
  complete: function(){
    console.log("B Done")
  },
};


const BridgeObservers = {
  next: function(x){
    this.observers.forEach(
      o => o.next(x)
    )
  },
  error: function(x){
    this.observers.forEach(
      o => o.error(x)
    )
  },
  complete: function(){
    this.observers.forEach(
      o => o.complete()
    )
  },
  observers: [],
  addObserver: function(observer){
    this.observers.push(observer)
  }
};

observable.subscribe(BridgeObservers);
BridgeObservers.addObserver(ObserverA);

setTimeout(function(){
  BridgeObservers.addObserver(ObserverB);
}, 2000)

 

And this partten:

observable.subscribe(BridgeObservers);
BridgeObservers.addObserver(ObserverA); // BirdegeObservers.subscribe(ObserverA)

is actually ‘subject‘ partten, works both as Observer and Observable.

 

Subject:

const observable = Rx.Observable.interval(1000).take(5);

const ObserverA = {
  next: function(x){
    console.log("A next " + x)
  },
  error: function(x){
    console.error("A error " + x)
  },
  complete: function(){
    console.log("A Done")
  },
};


const ObserverB = {
  next: function(x){
    console.log("B next " + x)
  },
  error: function(x){
    console.error("B error " + x)
  },
  complete: function(){
    console.log("B Done")
  },
};

const subject = new Rx.Subject();
/*const BridgeObservers = {
  next: function(x){
    this.observers.forEach(
      o => o.next(x)
    )
  },
  error: function(x){
    this.observers.forEach(
      o => o.error(x)
    )
  },
  complete: function(){
    this.observers.forEach(
      o => o.complete()
    )
  },
  observers: [],
  subscribe: function(observer){
    this.observers.push(observer)
  }
};*/

observable.subscribe(subject);
subject.subscribe(ObserverA);
//BridgeObservers.subscribe(ObserverA);

setTimeout(function(){
  subject.subscribe(ObserverB);
 // BridgeObservers.subscribe(ObserverB);
}, 2000)

 In the end, ObserverA and ObserverB share one single observer.