从内部 flatMap 通知答案

【问题标题】：Notify from inner flatMap从内部 flatMap 通知
【发布时间】：2016-12-06 10:09:06
【问题描述】：

这是一个相当复杂的示例：

主要：

this.runInstructionAndGetResult().subscribe({
      next: val => console.log(`NEXT VALUE: ${val}`),
      error: val => console.log(`ERROR VALUE: ${val}`),
      complete: val => console.log(`COMPLETE`)
    });

可观察的：

public runInstructionAndGetResult(): Observable<string> {
    return this.runAnInstruction()
        .flatMap((data) => {
            console.info("flatMap of runAnInstruction:", data);
            return this.getInstructionExecutionStatusInPolling()
                .filter(data => data != "Polling")
                .take(1)
                .flatMap((data) => {
                    console.info("flatMap of getInstructionExecutionStatusInPolling:", data);
                    return this.getInstructionResult();
                }).map((data) => {
                    console.info("Map of getInstructionResult:", data);
                    return data;
                });
        });
  }

  public runAnInstruction(): Observable<string> {
    return Observable.of("StartRun");
  }

  public getInstructionResult(): Observable<string> {
    return Observable.of("FinalResult");
  }

  public getInstructionExecutionStatusInPolling(): Observable<string> {
    return Observable.interval(1000)
        .concatMap(data => {
            return this.getInstructionExecutionStatus();
        });
  }

  public getInstructionExecutionStatus(): Observable<string> {
    return Observable.of("Polling", "Terminate");
  }

这里插话： https://plnkr.co/edit/c1cahMtVARQnLgnHWlEe?p=preview

主要问题是我只想了解外部内部流的“演变”。

现在我们只有在所有内部 flatMap 都完成后才会在 main 上设置“next”事件。

如何获得通知？如何在轮询期间向主流发出显式值？

谢谢。

【问题讨论】：

你能更明确地说你想得到什么结果吗？或者你想在控制台中看到什么输出。我不明白这一切应该做什么。
@martin 我只想在 main 上订阅下一个呼叫，例如“StartRun”数据或“Terminate”数据。
所以如果我理解这个正确性：你想从一个只有 1 个数据开始的流中接收 2 个 next-calls - 在这种情况下你应该看看 concat-operators，或者尝试类似this.runInstructionAndGetResult().startWith("StartRun").subscribe...

标签： rxjs rxjs5

【解决方案1】：

我找到了一个解决方案来分享。

这里 plunker 更新了：

https://plnkr.co/edit/c1cahMtVARQnLgnHWlEe?p=preview

基本上我创建了一个简单的 observable 使用：https://github.com/Reactive-Extensions/RxJS/blob/master/doc/api/core/operators/create.md

然后我以编程方式调用下一个方法并最终完成：

 public runInstructionAndGetResult(): Observable<string> {
    return Observable.create((ops)=> {
      ops.next(1);
      this.runAnInstruction()
        .concatMap((data) => {
            ops.next(2);
            console.info("flatMap of runAnInstruction:", data);
            return this.getInstructionExecutionStatusInPolling()
                .filter(data => data != "Polling")
                .take(1)
                .concatMap((data) => {
                    ops.next(3);
                    console.info("flatMap of getInstructionExecutionStatusInPolling:", data);
                    return this.getInstructionResult();
                }).map((data) => {
                    console.info("Map of getInstructionResult:", data);
                    ops.next(4);
                    ops.complete();
                    return data;
                });
        }).subscribe();
        });
  }

【讨论】：