[Rx.js] Zip运算符何时发出错误？

Question

我正在学习Rx.js，zip操作员有一个问题：

    var error =Rx.Observable.throw('Oop!');
    var age$ = Rx.Observable.concat(Rx.Observable.of(21,22,23),error);
    var sex$ = Rx.Observable.of("male","male","female","female");
    var name$ = Rx.Observable.of("jack","john","james","lucy");
    var example = Rx.Observable.zip(age$,sex$,name$,(age,sex,name)=>{ return {age,sex,name} });

我订阅了example源并打印了一些消息：

    example.subscribe({
        next: (value) => { console.log(value); },
        error: (err) => { console.log('Error: ' + err); },
        complete: () => { console.log('complete'); }
    });

输出不是我期望的：

{age:21,sex:"male",name:"jack"}
{age:22,sex:"male",name:"john"}
{age:23,sex:"female",name:"james"}
error

但只有一行with no value output：

    error

阅读官方文档，但在zip操作人员退出时没有解释任何章节error。

任何人都可以帮忙吗？

Answer 1

您会立即看到错误，因为您传递的第一个 observable 同步发出其值。（其他 observable 也同步发出它们的值，但这在这种情况下无关紧要。）

zip一一订阅传递的 observables 并按照它们传递的顺序。在订阅第一个 observable 时，zip同步接收所有 observable 的值和串联错误。然后它发出自己的错误并完成。

如果您指定可选的调度程序参数 - 以便 observables 异步发出它们的值 - 您将看到您期望的行为：

var age$ = Rx.Observable.concat(
  Rx.Observable.of(21, 22, 23, Rx.Scheduler.async),
  Rx.Observable.throw("Oop!", Rx.Scheduler.async)
);

var sex$ = Rx.Observable.of(
  "male", "male", "female", "female",
  Rx.Scheduler.async
);

var name$ = Rx.Observable.of(
  "jack", "john", "james", "lucy",
  Rx.Scheduler.async
);

var zipped$ = Rx.Observable.zip(
  age$, sex$, name$,
  (age, sex, name) => ({ age, sex, name })
);
zipped$.subscribe(
  (value) => console.log(value),
  (error) => console.log(error),
  () => console.log("complete")
);

.as-console-wrapper { max-height: 100% !important; top: 0; }

<script src="https://unpkg.com/rxjs@5/bundles/Rx.min.js"></script>