ReactiveX：Group和Buffer仅为每个组中的最后一项 [英] ReactiveX: Group and Buffer only last item in each group

问题描述

如何对一个Observable进行分组，并从每个GroupedObservable只保留内存中最后一个发出的项目？
这样每个组的行为就像BehaviorSubject一样。

How to group an Observable, and from each GroupedObservable keep in memory only the last emitted item? So that each group would behave just like BehaviorSubject.

这样的事情：

{user: 1, msg: "Anyone here?"}
{user: 2, msg: "Hi"}
{user: 2, msg: "How are you?"}
{user: 1, msg: "Hello"}
{user: 1, msg: "Good"}

所以在内存中我们只有每个用户的最后一项：

So in memory we'd have only have the last item for each user:

{user: 2, msg: "How are you?"}
{user: 1, msg: "Good"}

当订阅者订阅时，这两个项目立即发布（每个项目都在其自己的排放中）。就像我们为每个用户设置了BehaviorSubject。

And when a subscriber subscribes, these two items were issued right away (each in it's own emission). Like we had BehaviorSubject for each user.

onCompleted（）永远不会被激发，因为人们可能会永远聊天。

onCompleted() is never expected to fire, as people may chat forever.

我事先并不知道可以有什么用户值。

I don't know in advance what user values there can be.

推荐答案

我认为你的chatlog可观察性很热。 #groupBy发出的groupObservable因此也会很热，并且不会自行保留内存中的任何内容。

I assume your chatlog observable is hot. The groupObservables emitted by #groupBy will consequently also be hot and won't keep anything in memory by themselves.

获取所需的行为（丢弃除最后一个值之外的所有内容）从订阅之前并继续从那里开始）你可以使用ReplaySubject（1）。

To get the behavior you want (discard everything but the last value from before subscription and continue from there) you could use a ReplaySubject(1).

如果我错了请纠正我

请参阅 jsbin

var groups = chatlog
      .groupBy(message => message.user)
      .map(groupObservable => {
        var subject = new Rx.ReplaySubject(1);
        groupObservable.subscribe(value => subject.onNext(value));
        return subject;
      });

这篇关于ReactiveX：Group和Buffer仅为每个组中的最后一项的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！