Опубликуйте последнее значение наблюдаемого

У меня есть горячая наблюдаемая (тема в этом случае):

var subject = new Rx.Subject();

Я хочу создать еще одно наблюдаемое, что каждый раз, когда производится новая подписка, сразу возникает последнее значение, которое было создано.

Итак, в псевдокоде:

var myObservableWithLastValue = subject.publishLast();

subject.onNext(3);

myObservableWithLastValue.subscribe(function(x){
    console.log(x); //should write 3
});

myObservableWithLastValue.subscribe(function(x){
    console.log(x); //should write 3, too
});

subject.onNext(4);

myObservableWithLastValue.subscribe(function(x){
    console.log(x); //should write 4
});

Это примерно то, что я хочу, и, похоже, это работает. Тем не менее, я предполагаю, что для достижения того же

Rx.Observable.prototype.keepLatest = function () {
    var latestValue;

    var disposable = this.subscribe(function (value) {
        latestValue = value;
    });

    return Rx.Observable.create(function (observer) {
        observer.onNext(latestValue);
        return disposable.dispose;
    });
};

Ответ 1

RxJs теперь имеет ReplaySubject. Инициализируйте его с помощью 1 буфера, и у вас есть BehaviorSubject.

// as an example, use buffer size of 2
var subject = new Rx.ReplaySubject(2 /* buffer size */);

subject.onNext('a');
subject.onNext('b');
subject.onNext('c');

subject.subscribe(function (x) { document.write('x1:' + x + '<br>'); });

subject.onNext('d');

subject.subscribe(function (x) { document.write('x2:' + x + '<br>'); });

<script src='https://rawgit.com/Reactive-Extensions/RxJS/v.2.5.3/dist/rx.all.js'></script>

Ответ 2

BehaviorSubject:

Инициализирует новый экземпляр класса Rx.BehaviorSubject, который создает объект, который кэширует свое последнее значение и начинает с указанного значения.

var subject = new Rx.BehaviorSubject('a' /* initial value */);

subject.subscribe(function (x) {
    console.log('x1:' + x);
});

subject.onNext('d');

// Will produce the last value.

subject.subscribe(function (x) {
    console.log('x2:' + x);
});

<script src='https://rawgit.com/Reactive-Extensions/RxJS/v.2.5.3/dist/rx.all.js'></script>

Ответ 3

Я использовал .cache(1) в горячем Observable. Кажется, чтобы дать поведение, которое вы просили (но я новичок).

видно, что каждый раз, когда производится новая подписка, происходит мгновенное получение последнего значения, которое было создано.