BehaviorSubject“分组”

Question

我刚刚开始使用RXJS，看看它是否可以取代我目前的手动数据流。我试图移植的一件事是这样一种情况，即记住流中的最后一个值，这样将来的观察者将始终获得“当前”值以及后续的值。这似乎是由BehaviorSubject完成的。

但是，我需要为一组实体执行此操作。例如，我可能有表示来自用户的消息的数据：

{ userId: 1, message: "Hello!" }

我想要一个类似BehaviorSubject的对象来存储所有用户的最后一条消息。这是我可以使用RXJS开箱即用来完成的事情，还是需要我自己来做？(如果是这样的话，如果有任何指示将不胜感激)。

编辑:经过进一步的思考，似乎应该有一个“传入”的主题，一个更新Map的观察者，然后是一个我可以调用的函数，该函数从映射值初始化一个可观察对象，并与传入的流合并...？

Answer 1

我使用的是带有类似redux的状态设置的RxJS。我有一个保存当前状态的BehaviorSubject，每次触发事件/动作时，当前状态都会通过函数传递，这些函数会生成主体订阅的新状态。

下面是我使用的简化版本：

export default class Flux {

  constructor(state) {
    //all resources are saved here for disposal (for example, when hot loading)
    //this is just a flux dispatcher
    this.dispatcher = new rx.Subject(),
    // BehaviorSuject constructed with initial state
    this.state = new Rx.BehaviorSubject(state),
  }

  addStore(store, initialState, feature = store.feature) {
      this.dispatcher
        .share()
        // this is where the "reduction" happens. store is a reducer that 
        // takes an existing state and returns the new state
        .flatMap(({action, payload}) =>
            store(this.state.getValue(), action, payload))
        .startWith(initialState || {})
        .subscribe(this.state)
    );

    return this;
  }

  addActions(actions: rx.Subject) {
    // actions are fed to the dispatcher
    this.resources.push(actions.share().subscribe(this.dispatcher));
    return this;
  }

}

我创建了一个带有管理状态的全局Flux对象。然后，对于每个“功能”或“页面”或任何我想要的东西，我都会添加操作和存储。它使得管理状态变得非常容易，像时间旅行这样的事情是Rx给出的。