javascript - 如何连接到源？

Question

我有以下代码:

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，但我只得到最后一行。看来，driverC 只收到最后一条消息，即使 B$.tap(console.log) 输出所有三个 ("a"、"b"、"c");

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

版本:

• @cycle/core@6.0.3
• rx@4.1.0

Answer 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这样当数据流动时，所有源都已经准备好接收它。