RxJS：在组合流中识别流源的正确模式？

Question

我正在寻找一些关于如何识别哪个流进入merge 或combineLatest 函数的最佳实践建议，以便只对新流进行操作。

在 TODO 应用程序的上下文中，我有传入的添加和删除流，我想将它们组合起来，以便我的列表编辑可以在单个流中以无状态的方式进行。输出是一个集成了 add(concat) 和 remove(filter) 事件的列表，否则您似乎最终会得到包含所有 add 或 remove 事件的其他流，从而无限增长。

遇到的问题包括：

• 使用merge 并不表示传入的是哪个流；
• 使用combineLatest 并不表示是哪个流触发了订阅的流，因此您不能只执行与该流相关的操作。
• 使用withLatestFrom 会生成一个不会更新withLatestFrom 输入源的新列表，因此除非也对此列表感兴趣的下一个流订阅该列表，否则该列表将不同步（或者您拥有每个新列表基于之前的列表，这会导致不必要地重新执行之前发生的所有转换...）。

目前发现的似乎不太理想的方法包括：

• Cycle JS TODO 应用程序明确地将type 属性直接分配给早期流的创建对象以进行识别，我认为应该避免使用直接识别流来自哪里的 RxJS 方法？
• 接受http://www.jisaacks.com/manipulating-rxjs-streams/ 的建议，并将添加流和删除流的输出拆分为[null, {addItem}] 和[removeItem, {null}]，这样当将merge 用于添加和删除事件时，我仍然可以识别传入流正在更新，以便我可以在单个流中执行添加和删除（但随后我想添加切换事件等，但这似乎也不正确，因为我需要创建知道的流输出所有其他潜在流（以 [null, null, myOutput, null, null 等结束）。

非常欢迎任何最佳实践建议。

Answer 1

您可以在合并之前操作精确的流，以避免不必要的 id 处理。一般来说，对于流和函数式编程，尽量避免 if-else 结构，因为有一些工具可以帮助你。通常，您可以通过查看呼叫的位置并提前采取行动来避免它。

这是一个例子。列表操作只是为了让最重要的部分可见。

let listSubject: Rx.Subject<Item[]> = new Rx.BehaviourSubject<Item[]>([]);
let addObs: Rx.Observable<Item[]> = initAddObs();
let removeObs: Rx.Observable<Item[]> = initRemoveObs();

Rx.Observable
  .merge([
    addObs.map(items => (list: Item[]) => list.concat(items)),
    removeObs.map(items => (list: Item[]) => list.filter(items))
  ])
  .withLatestFrom(listSubject, (operation, list) => operation(list))
  .subscribe(listSubject)

对于列表，使用了 BehaviourSubject，因此无论何时订阅，任何人都可以获得最新情况。 AddObs 和 removeObs 流映射到具有相同签名并且可以合并的操作。在对当前列表执行操作后，我们必须将结果回收回 listSubject。现在您可以订阅 listSubject 并且您将获得从那里发出的最新列表。

Answer 2

没有允许识别流来自何处的 Rxjs 方法。溪流上没有名字。如果你想要一个，你需要自己放。因此，如果您不喜欢这两种方式中的任何一种，请找到另一种方式来为它们命名。如果您的问题是在对象上放置标识符的最佳做法是什么，那么答案就是在该死的对象上放置标识符。

我个人喜欢的方式是通过柯里化函数：

function label(identifier){return function (x){var obj={};obj[identifier]=x;return obj;};}

我使用如下：

source1.map(label('remove')).merge(source2.map(label('add')))

但实际上，按照你的意愿去做，在我看来，你如何做到这一点是一个相当小的问题。