javascript - 如何从无限的RxJs流中获得不是初始值的单个最新值？

这个概念

这是一个模拟的angular2项目。

当使用来自redux存储的可观察流时，我尝试先进行过滤，然后接受/获取最后/最后一个最新值。
之后，我想在流完成时解决 promise ，但在使用takeLast运算符时不解决 promise 。

因此，问题是:我可以使用哪种运算符设置从流中获取最新值？

设置

我将Angular 2的安装简化为RxJs使用的要点。

源可观察到是由redux库管理的，尚未完成

服务提供了一些逻辑以从流

中检索最新值

组件正在使用值(value) promise 样式

这是一个工作示例:https://fiddle.jshell.net/markus_falk/an41z6g9/

Redux商店模拟:

var latestTime$ = new Rx.Subject();
setInterval(function(){
     latestTime$.onNext(Date.now());
}, 2000);

服务可注入(inject)模拟:

var timeStore = null;
var getLatestTime = function() {

  return new Promise((resolve, reject) => {

     latestTime$

     /*
        filter out 'null' for when the button is clicked
        before the store updates the first time
      */
     .filter(function(x) {
        console.log('filter: ', x);
        return x === typeof('number');
     })

     // try to end to stream by taking the last from the stream ?!?!?!?
     .takeLast(1)

     // handle promise
     .subscribe(

       function (x) {
         console.log('Next: ' + x);
         // store latest stream value
         timeStore = x;
       },
       function (err) {
         console.log('Error: ' + err);
         reject(err)
       },
       function () {
         console.log('Completed');
         // pass on latest value of endless when stream completes
         resolve(timeStore);
       }

    );

  });

};

以及一个消耗大量的模拟组件:

document.querySelector("#foo").addEventListener("click", function(event) {

  var time = getLatestTime();

  time.then((latestTime) => {
    console.log('latestTime: ', latestTime);
  });

  time.catch((err) => {
    console.log('oh oh: ', err);
  });

}, false);

最佳答案

这应该可以模拟您的情况。

观看现场演示:https://jsfiddle.net/usualcarrot/zh07hfrc/1/

var subject = new Rx.Subject();

subject.skip(1).last().subscribe(function(val) {
  console.log('next:', val);
}, function(val) {
  console.log('error:', val);
}, function() {
  console.log('completed');
});

subject.onNext(1);
subject.onNext(2);
subject.onNext(3);
subject.onNext(4);
subject.onNext(5);
subject.onCompleted();

打印到控制台:

next: 5
completed

您可以将console.log('completed');代替resolve(...)。也许这甚至是没有必要的，您可以根据使用情况只返回Subject并订阅它(？)。在这种情况下，请使用 asObservable() to hide the fact you're using a Subject 。参见similar use-case with asObservable() 。