【问题标题】：How to concat to sources?如何连接到来源？
【发布时间】：2016-10-03 23:28:06
【问题描述】：

我有以下代码：

const Cycle = require('@cycle/core');
const {Observable} = require('rx');

function main(sources) {
    const B$ = sources.driverA
        .concat(Observable.just('b'))
        .concat(Observable.just('c'));

    const C$ = sources.driverB.map(x => x.toUpperCase());

    return {
        driverA: Observable.just('a'),
        driverB: B$,
        driverC: C$
    }
}

Cycle.run(main, {
    driverA: (A$) => A$,
    driverB: (B$) => B$,
    driverC: msg$ => { msg$.subscribe(msg => console.log(msg)) }
});

我希望在控制台上得到三行：A、B 和 C，但我只得到最后一行。看来，即使 B$.tap(console.log) 输出所有三个 ("a", "b", "c");

对此行为的解释是什么？如何将所有三个消息传播到 driverC？

版本：

@cycle/core@6.0.3
rx@4.1.0

【问题讨论】：

标签： javascript rxjs cyclejs

【解决方案1】：

行为说明

其实这并不容易解释。这是由于cycle.run 如何连接其周期。以下代码在 trycicle 中运行：

const Cycle = require('@cycle/core');
const {Observable} = require('rx');

function main(sources) {
    const B$ = sources.driverA
        .concat(Observable.just('b'))
        .concat(Observable.just('c'))
        .concat(Observable.just('d'));

    const C$ = sources.driverB.map(x => x.toUpperCase());

    return {
        driverA: Observable.just('a'),
        driverB: B$,
        driverC: C$
    }
}

Cycle.run(main, {
    driverA: (A$) => A$.tap(msg => console.log(msg)),
    driverB: (B$) => B$.tap(msg => console.log(msg)),
    driverC: msg$ => { msg$.subscribe(msg => console.log(msg)) }
});

并且只显示a d D。所以它实际上是显示的最后一个字母。

现在如果你运行这个：

const Cycle = require('@cycle/core');
const {Observable} = require('rx');

function main(sources) {
    const B$ = sources.driverA
        .concat(Observable.just('b').delay(1))
        .concat(Observable.just('c'))
        .concat(Observable.just('d'));

    const C$ = sources.driverB.map(x => x.toUpperCase());

    return {
        driverA: Observable.just('a'),
        driverB: B$,
        driverC: C$
    }
}

Cycle.run(main, {
    driverA: (A$) => A$.tap(msg => console.log(msg)),
    driverB: (B$) => B$.tap(msg => console.log(msg)),
    driverC: msg$ => { msg$.subscribe(msg => console.log(msg)) }
});

你会得到a a A b B c C d D，这是你所期望的。

发生的情况是run 通过主题将驱动程序连接到源，并按顺序执行。哪个顺序？ var x in obj 中的属性枚举顺序未指定，因此不能依赖 - 可能取决于浏览器（参见 Does ES6 introduce a well-defined order of enumeration for object properties?））。现在chrome 和firefox 最新版本似乎按字母数字属性的定义顺序枚举属性，但数字属性的数字顺序（following ES2015 spec）。

所以这里，driverA 首先连接到源，它启动相应的数据流。当driverB 连接到源时，同样的事情。由于您编写B$ 的方式，该数据流是同步的。所以当subscribe即连线时，所有数据a b c d从B$同步流出，当driverC连线时，B$已经完成。鉴于接线是使用replaySubject(1) 进行的，该接线会为您提供完成前最后发出的值，即d。

因此，由于同步性，顺序很重要：如果 B 和 C 先连接，那就没问题了。不幸的是，您的执行顺序不充分。

为了让您相信这一点，我按拓扑顺序对您的流进行排序的代码按预期工作：

const Cycle = require('@cycle/core');
const {Observable} = require('rx');

function main(sources) {
    const B$ = sources.driverA
        .concat(Observable.just('b'))
        .concat(Observable.just('c'))
        .concat(Observable.just('d'));

    const C$ = sources.driverB.map(x => x.toUpperCase());

    return {
        driverC: C$,
        driverB: B$,
        driverA: Observable.just('a'),
    }
}

Cycle.run(main, {
    driverA: (A$) => A$.tap(msg => console.log(msg)),
    driverB: (B$) => B$.tap(msg => console.log(msg)),
    driverC: msg$ => { msg$.subscribe(msg => console.log(msg)) }})

如何传播所有三个消息

好吧，要么按拓扑顺序排列您的接收器，要么删除同步性。我添加了一个delay(1) 以使数据流在下一个滴答声中继续，此时driverC 已经连接以接收下一个值。这可能是最强大的选项，因为拓扑顺序可能并不总是像这里那样明显计算，可能会随着源的交错而改变，并且依赖于浏览器相关的对象属性枚举（！）。

另外说明，当数据流的同步性无法避免时，您通常使用publish先连接所有源，然后connect处理连接问题，这样当数据流时，所有来源已经准备好接收它。

【讨论】：

【解决方案2】：

只需在driverA 之后添加delay：

const B$ = sources.driverA.delay(1)

WebpackBin example.

或者，您可以调用concat 一次并列出所有可观察对象并延迟。

const B$ = Observable.concat(
    sources.driverA,
    Observable.just('b'),
    Observable.just('c'),
    Observable.just('d')
).delay(1);

WebpackBin example #2.

需要记住的是，main 函数只是连接管道。 run 开水。

【讨论】：