Пользовательские наблюдаемые Angular2 с некоторыми последовательными подписками

Question

Ты знаешь решение моей проблемы? Мне нужна гибкая последовательность подписок в виде наблюдаемой, как здесь:

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 не ждет, пока в первом блоке кода будет подписана переменная "success". Почему и какое решение лучше для меня?

Answer 1

1-й вопрос: почему это не работает?

Потому что каждая подписка асинхронная. Когда вы делаете this.saveCallbarrings().subscribe(...) что-то внутри подписки может произойти в любое время (может быть, никогда!), поэтому программа переходит к следующей инструкции, то есть observer.next(success);, который имеет начальное значение success,

2-й вопрос: какое решение лучше для меня?

Rx.Observables имеет так много операторов, чтобы иметь дело с этим асинхронным материалом. В вашем случае вам нужен оператор 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 не будет выполнять какой-либо код, пока кто-то не подпишется на него, так что вы получите меньше побочных эффектов.