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.
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)