具有一些顺序订阅的 Angular2 自定义 observables

Question

您知道我的问题的解决方案吗？ 我需要一个灵活的订阅序列，封装在一个 observable 中，如下所示：

saveData() {
    return new Observable((observer) => {
      let success = true;

      if(example === true) {
        this.saveCallbarrings().subscribe((response) => {
          // this response was ignored from angular
          success = (response === true);
        });
      }

      if(example2 === true) {
        this.saveCallbarrings().subscribe((response) => {
          // this response was ignored from angular too
           success = (response === true);
        });
      }

      // and so on ... in the end I need a result of all responses
      observer.next(success);
    });
  }

最后我在提交方法中调用了这个“响应集合”的结果：

onSubmit() {
// Validations and others
...
if(this.isNew) {
        observable = this.create();
      } else {
        observable = this.update();
      }

      return observable.subscribe(success => {
        if(success == true) {
          let subscription = this.saveData().subscribe(successFinished => {
            // And here is the problem, because this var doesnt have the correct response
            if(successFinished === true) {
              this.alertService.success('ALERT.success_saved', {value: 'ALERT.success_edit_user', param: {user: this.user.username}});
            }
          });

          subscription.unsubscribe();
        }
      });

主要问题是 Angular 不会等到第一个代码块中订阅了“成功”变量。 为什么以及对我来说更好的解决方案是什么？

Answer 1

第一个问题：为什么它不起作用？

因为每个订阅都是异步的。当您执行this.saveCallbarrings().subscribe(...) 时，subscribe 内部的事情可能随时发生（也许永远不会发生！），因此程序继续执行下一条指令，即observer.next(success);，其初始值为success。

第二个问题：对我来说最好的解决方案是什么？

Rx.Observables 有so many operators 来处理这个异步的东西。在您的情况下，您需要的运算符是forkJoin。该运算符允许您向他传递一个流数组，它将订阅所有流，当所有流完成时，它将为您提供一个数组，其中包含每个流的每个结果。所以你的代码会变成：

saveData() {
    return Rx.Observable.defer(() => {
        let streams = [];
        if(example === true) {
            streams.push(this.saveCallbarrings());
        }
        if(example2 === true) {
            streams.push(this.saveCallbarrings());
        }

        // And so on

        return Rx.Observable.forkJoin(streams);
    });
}

话虽如此，我不知道你为什么要多次订阅同一个this.saveCallbarrings()，我想这只是为了让问题更简单，例如。

另外，这里我使用.defer() 而不是创建。有了这个，你可以返回另一个流，它会订阅它并将它传递给观察者。做defer 和什么都不做（即设置流并只返回forkJoin）之间的区别在于defer 不会执行任何代码，直到有人订阅它，所以你得到的副作用更少。