Rxjs 合并（）不工作答案

【问题标题】：Rxjs merge() not workingRxjs 合并（）不工作
【发布时间】：2016-01-19 09:24:26
【问题描述】：

上下文：为了使用 JSON 树生成 swift 代码，我以预排序格式遍历树，向下到达所有叶节点，有效地将树扁平化为一个数组。

描述： 使用 Rx.Observable.generate() 从数组元素创建一个可观察的流，但是在过滤时我得到了一些奇怪的结果。我在下面提供了一个简化的示例：

示例：

var Rx = require('/usr/local/lib/node_modules/rx') // 4.0.7

// source 1,3,5,7,9 (WAT)
var source = Rx.Observable.generate(
    0,
    function (x) { return x < 10; },
    function (x) { return x + 1; },
    function (x) { return x; }
)

// filter & merge
var a = source.filter(x => x % 2 == 0)
var b = source.filter(x => x % 2 != 0)
var source = a.merge(b)

// subscribe & output
var subscription = source.subscribe(
    x => console.log(x)
)

问题：为什么我得到的输出结果是 1,3,5,7,9 而不是 0,1,2,3,4,5,6,7,8,9预期？

我应用合并的方式似乎并不重要......当反转时我也会得到输出 0,2,4,6,8。

编辑、npm 安装和节点版本 user3743222：感谢您的反馈，项目信息如下：

$ npm install

chai@3.4.1 node_modules/chai
├── assertion-error@1.0.1
├── type-detect@1.0.0
└── deep-eql@0.1.3 (type-detect@0.1.1)

moment@2.11.1 node_modules/moment

mocha@2.3.4 node_modules/mocha
├── escape-string-regexp@1.0.2
├── diff@1.4.0
├── commander@2.3.0
├── supports-color@1.2.0
├── growl@1.8.1
├── debug@2.2.0 (ms@0.7.1)
├── jade@0.26.3 (commander@0.6.1, mkdirp@0.3.0)
├── mkdirp@0.5.0 (minimist@0.0.8)
└── glob@3.2.3 (inherits@2.0.1, graceful-fs@2.0.3, minimatch@0.2.14)

rx@4.0.7 node_modules/rx


$ node --version

v4.2.1

【问题讨论】：

迈克，我运行你的代码，我确实在控制台中得到了 [0,1,2,3,4,5,6,7,8,9]。 jsfiddle.net/ewqpcjum/1。我正在使用 RxJS v4。你用的是哪个版本？
顺便说一句，我刚刚注意到您正在重新分配source。除非这是故意的，否则您应该将source=a.merge(b) 替换为mergedSource=a.merge(b)，以便能够更轻松地推断您的程序。刚刚也看到您正在使用 4.07，这也是我正在使用的。这真的是给你错误结果的代码吗？
感谢您的反馈已更新更多信息。旁注：这个项目是从命令行上的节点运行的。
我认为问题不在于您将真实示例归结为示例。通过这种沸腾，您可能已经精确地抑制了错误的来源。这个想法是能够重现错误，以便对其进行调查。我看不出你的配置或代码有什么问题，所以如果你有一个不起作用的正确代码，请在此处发布问题：github.com/Reactive-Extensions/RxJS/…
好的，在 kakigoori 回答后更新，我用 Rxjs 2.2 运行 jsfiddle，我用 Rxjs4 运行它，我重现了你的错误结果。见这里：jsfiddle.net/fz3LL7e5。您绝对应该在上面的链接中发布问题，其中两个 jsbin 链接都表现出这两种行为。这是generate 的问题，而不是merge 的问题

标签： javascript node.js rxjs reactivex

【解决方案1】：

我认为您的问题是 generate 旨在创建一个有状态的 Observable 它将为第一个订阅者提供您生成的项目，但下一个订阅者将不会得到任何东西。 See this JSBin

例如，如果您使用通过range 创建的普通 Observable，您将获得您正在寻找的合并结果。 See this JSBin

【讨论】：

在你的第一个jsbin 中发生的事情是generate 同步发出它的所有值（默认情况下是立即调度程序），这就是为什么在你完成第一个subscribe 时它们都被消耗掉了在你的 jsbin 中。我不明白的是为什么下一个subscribe 不重启数据流。 generate 返回行为类似于 hot observable 的东西，这肯定是无处记录的。除了之前版本的 Rxjs 之外，它还表现得像一个冷的 observable。 range 肯定会返回一个冷的 observable，所以它按预期工作。
+1 无论如何都可以找出答案。我无法自己弄清楚。请注意，尽管二分法是 cold vs hot observables，而不是 stateful 和 normal observables。我不太了解generate 源代码（他们使用对我来说是中文的递归调度）知道发生了什么，但最终的行为确实相当违反直觉。

【解决方案2】：

正如 cmets 中提到的，对于 generate 运算符，RxjsV4 似乎表现出与 Rxjsv2 不同的行为。

周转时间：

降级版本
使用defer 重新启动源Cf。 http://jsfiddle.net/fz3LL7e5/1/ 在这种情况下会起作用，但不一定适用于所有情况。
实际上我认为您想要的是使用share，因为您希望您的下游 observable 看到相同的值，对吗？而不是重新开始生成。参照。 http://jsfiddle.net/fz3LL7e5/2/ 这按预期工作。

关于热与冷here (official documentation)和here (illustrated dataflows)的解释

【讨论】：

【解决方案3】：

我做了一些摆弄，尝试不同的方式来创建事件流

我喜欢使用类似 from 数组的选项，其中传递给 from() 的对象必须具有公共 length 属性。不确定这是否是因为它已编入索引。

var E = Rx.Observable.from(arrayLike,
    function (_, i) {
        return i
    })

显示在这里工作：http://jsfiddle.net/th3caaaz/

【讨论】：