javascript - 如何使用 rxjs 根据时间从数组中过滤项目？

Question

我有一个 Observable，它发出的值是对象数组。发出的值如下所示:

[
    {
        id: 1,
        title: 'Foo',
        datetime: 'Thu Jun 29 2017 15:14:05 GMT-0400 (Eastern Daylight Time)'
    },
    {
        id: 2,
        title: 'Bar',
        datetime: 'Wed Jun 28 2017 12:10:42 GMT-0400 (Eastern Daylight Time)'
    }
]

请注意，datetime 值是 Date() 对象。

我想要做的是创建一个“即将发生的事件”列表，显示当前时间之后的所有事件。为了实现这一点，我做了以下工作:

events$.subscribe(data => {
    const upcomingEvents = data.filter(item => item.datetime > new Date());
})

同样可以通过以下方式完成:

events$.map(data => {
    return data.filter(item => item.datetime > new Date());
})
.subscribe(data => {
    const upcomingEvents = data;
});

这种方法的问题在于，虽然它确实添加了通过过滤器的任何新项目，但当当前时间超过项目时间时，它不会删除项目。

我知道这种行为是由于过滤器仅在 Observable 发出新值时运行，但我无法找出解决方案。

如有任何帮助，我们将不胜感激。

Answer 1

这不是一个非常优雅的解决方案，但我能想到解决这个问题的唯一方法是合并两个流:第一个是您的可观察流，然后以任意间隔创建第二个时间戳流以“垃圾收集” “你的第一个字符串。您需要向该合并流添加一个 scan ，以便您可以在流末尾维护一个具有已过滤日期状态的累加器。所以像这样(警告，未经测试!):

// modeling your observable, not sure how you're creating it
const events$ = Rx.Observable.interval(3000)
  .concatMap(function (x) {
    return Rx.Observable
      .of([{title: "bar", datetime: new Date()}, {title: "foo", datetime: new Date()}])
  });

// Infinite timestamp stream
// this one checks every second, you may want to 
// check every minute, or every hour
const timestamps$ = Rx.Observable.interval(1000)
  .timestamp()
  .delay(0)
  .repeat(-1);

const merged$ = events$.merge(timestamps$);
const filtered$ = merged$
  // Starts with empty array until your first upcoming events array
  .startWith([])
  .scan((acc, val) => {
    if (val.timestamp) {
      acc = acc.filter(item => item.datetime.getTime() < val.timestamp);
      return acc
    } else {
      acc = val
      return acc
    }
  });

有点笨拙，但目前我想不出更好的解决方案来将所有状态保留在 rxjs 中。