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

Question

这个概念

这是一个模拟的angular2项目.

当从redux存储中使用可观察流时,我尝试先过滤,然后取/ takeLast/last最新值.之后,我希望在流完成时解析promise,但在使用takeLast运算符时则不会.

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

设置

我将Angular 2设置简化为RxJs用法的要点.

• source observable由redux库管理,未完成
• service提供了一些逻辑来从流中检索最新值
• 组件正在消耗价值承诺风格

这是一个有效的例子:https://fiddle.jshell.net/markus_falk/an41z6g9/

redux商店模拟:

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

服务注入模拟:

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

Answer 1

这应该模拟您的情况。

查看现场演示：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()来隐藏您正在使用Subject. 请参阅的类似用例asObservable()。