XStreams 中忽略的第二个合并分支

Question

我正在使用 storage 和 state 插件开发 Cycle JS 应用程序。

该应用使用 xstream 作为响应式库。

我的行为很奇怪。

我的流就是这个图

在第一次会话 var 更新时，我得到了这个调试结果：

• 调试 0
• 调试 1
• 调试 3

问题：“Debug 2”分支未执行

如果我更新会话项（以便产生新的存储事件），则两个分支都将按预期执行

• 调试 0
• 调试 1
• 调试 2
• 调试 3

如果在“debug 0”处添加 .remember() 也会发生同样的良好行为

• 调试 0
• 调试 1
• 调试 2
• 调试 3

更奇怪的是，如果我删除过滤器，流程会按预期工作

没有过滤器（并且没有记住），自第一个事件以来，流程给出了这个结果

• 调试 0
• 调试 1
• 调试 2
• 调试 3

我的怀疑是在附加第二个分支之前在“调试 0”处观察到的东西，因此第一个事件已经被消耗。 但是如果这两个分支是 xs.merged 在一起怎么会发生这种情况呢？如何执行一个分支而第二个不执行？这两个分支没有过滤器或其他处理，它们只是映射到一个 reducer 函数。 有关如何调试和解决这种情况的任何建议？

Answer 1

问题发生是因为storage.session.getItem 在订阅后发出，而merge(a$, b$) 在订阅a$ 之前订阅a$，所以a$ 得到了事件，但是当b$ 被订阅时，它来得太晚了。这个问题是众所周知的，称为glitch（在反应式编程中），通常发生在菱形流图时，这正是你的情况。

我有两篇关于故障的博客文章可以为您提供更多背景信息：Primer on RxJS schedulers 和 Rx glitches aren't actually a problem。它提到了 RxJS，但 xstream 在实现方面非常接近 RxJS。 xstream 和 RxJS 的区别在于 xstream 流总是多播（“共享”），而 RxJS 有很多调度器类型，但 xstream 只有一种。 RxJS 中的默认调度程序与 xstream 具有相同的行为。

解决方案是在到达钻石之前申请.remember()。这是因为发出的值需要为该流的其他消费者缓存。 .remember() 只是将 Stream 转换为 MemoryStream。我认为源流最初是一个 MemoryStream，映射一个 MemoryStream 会创建其他 MemoryStream，但 filter 是一个打破这一点的运算符。 filter 总是返回一个 Stream，原因是 MemoryStreams 应该总是有一个当前值，但 filter 可以删除值，所以过滤后的流可能不会 有任何当前值。

作为 Cycle.js 的作者，我认为 cyclejs/storage 的设计方式并不是最好的，我认为我们会找到设计这些 API 的方法，以便最大限度地减少 MemoryStream 与 Stream 的混淆。但就目前而言，重要的是要了解这两者之间的区别并规划您的应用以避免出现问题（和故障），或者在正确的位置使用.remember()。