[RxJS] ReplaySubject with buffer

A BehaviorSubject can remember the latest value emitted, but what if we wanted Observer B to see all the previous values emitted in the past? We can't do that with BehaviorSubject, but there is ReplaySubject, which allows us to do that. This lessons teaches you everything you need to know about ReplaySubjects.

It is not good to cache all the value by using ReplaySubject, so we need to add cache logic for this.

The first parameter to the constructor of ReplaySubject takes a number that represents how many values we want to buffer:

var subject = new Rx.ReplaySubject(); // Buffer size of 2

  subject.next();

  subject.next();

  subject.next();

subject.subscribe(function(n) {

  console.log('Received value:', n);

});

/*

"Received value:"

3

"Received value:"

2

*/

The second parameter takes a number that represents the time in miliseconds during which we want to buffer values:

var subject = new Rx.ReplaySubject(null, ); // Buffer size of 200ms

  setTimeout(function() { subject.next(); }, );  //350(subscribe time) - 200 (buffer time) > 100 --> not replay

  setTimeout(function() { subject.next(); }, );  //350 - 200 < 200 --> replay

  setTimeout(function() { subject.next(); }, );  //350 - 200 < 300 --> replay

  setTimeout(function() {

subject.subscribe(function(n) {

  console.log('Received value:', n);

});

subject.onNext();

}, );

/*

"Received value:"

2

"Received value:"

3

"Received value:"

4

*/

var subject = new Rx.ReplaySubject();

           // new Rx.BehaviorSubject(0);

var observerA = {

  next: function (x) { console.log('A next ' + x); },

  error: function (err) { console.log('A error ' + err); },

  complete: function () { console.log('A done'); },

};

subject.subscribe(observerA);

console.log('observerA subscribed');

var observerB = {

  next: function (x) { console.log('B next ' + x); },

  error: function (err) { console.log('B error ' + err); },

  complete: function () { console.log('B done'); },

};

setTimeout(() => subject.next(), );

setTimeout(() => subject.next(), );

setTimeout(() => subject.next(), );

setTimeout(() => subject.complete(), );

/*

----1---2---3--|

  ..1...2...3...

                 1,2,3|

*/

setTimeout(function () {

  subject.subscribe(observerB);

  console.log('observerB subscribed');

}, );

/*

"observerA subscribed"

"A next 1"

"A next 2"

"A next 3"

"A done"

"B next 1"

"B next 2"

"B next 3"

"B done"

"observerB subscribed"

*/

巴特西

[RxJS] ReplaySubject with buffer

最新文章

热门文章