将数据从一个可观察对象传递到另一个对象的更好方法

Question

我创建了两个函数来测试Observables，每个函数都返回一个Observable：

foo() {
  return new Observable(observer => {
    let i = 0;
    setInterval( () => {
      if(i === 10) {
        observer.complete();
      } else {
        observer.next(i);
        i++;
      }
    }, 1000);
    // If I call observer.complete() here then it never completes
  });
}

bar(fooResult) {
  return new Observable(observer => {
    let j = 0;
    setInterval( () => {
      if(fooResult) {
        observer.next('j -> '+j+' plus fooResult '+JSON.stringify(fooResult));
        observer.complete();
      } else {
        observer.next(j);
        j++;
      }
    }, 2000);
  });
}

并像这样使用它们：

    let fooResult = [];

    // Testing observables...
    this.exampleProductService.foo().subscribe(
      (res) => { 
        console.log('foo next() -> '+res);
        fooResult.push(res);
      },
      (err) => {
        console.error('foo error: '+JSON.stringify(err));
      },
      () => { 
        console.log('foo finished');
        this.exampleProductService.bar(fooResult).subscribe(
          (res) => {
            console.log('bar next() -> '+res);
          },
           (err) => {
            console.error('bar error: '+JSON.stringify(err));
          },
          () => {
            console.log('bar finished');
          }
        );
      }
    );

提出的问题是：

1. 有没有更好的方法将数据从 Observable 的完成传递到另一个也返回 Observable 的函数？建立一个数组似乎很麻烦，我不能做以下事情，因为Observable 的complete callback 部分没有传递像progressUpdate 和onError 这样的参数：

   (complete) => { this.exampleProductService.bar(complete).// rest of code } 
   

2. 我尝试将第一个函数的结果分配给一个变量，然后传递该变量，但正如预期的那样，我得到了一个 Observable 而不是我想要的结果。

3. 我上面的做法有什么不对吗？

谢谢

附：这是一个 Angular 2 应用程序！

Answer 1

我认为您的函数有点过于复杂。一方面，当工厂函数已经可用时不要使用构造函数，在这种情况下interval 或timer 正如@Meir 指出的那样，尽管在这种情况下它会更冗长。

其次，bar 函数实际上并没有多大意义，因为您似乎在等待完成您已经知道已经完成的事情（因为您直到完成块才订阅它由foo 生成的 Observable）。

我根据您声明的目标进行了重构，即等待一个 Observable 完成，然后再开始第二个，同时在第二个中使用第一个的结果。

// Factory function to emit 10 items, 1 every second
function foo() {
  return Observable.interval(1000)
    .take(10);
}

// Lifts the passed in value into an Observable and stringfys it
function bar(fooResult) {
  return Rx.Observable.of(fooResult)
    .map(res => JSON.stringify(fooResult))
}

现在当使用它们时，你会这样做：

foo()
  // Log the side effects of foo
  .do(
    x => console.log(`foo next() -> ${x}`),
   err => console.error(`foo error: ${JSON.stringify(err)}`),
   () => console.log('foo finished')
  )
  // Collect the results from foo and only emit when foo completes
  .reduce((total, diff) => [...total, diff], [])

  // Pass the result from the reduce on to bar
  .concatMap(fooResult => bar(fooResult))

  //Subscribe to the results of bar
  .subscribe(
    res => console.log(`bar next() -> ${res}`),
    err => console.error(`bar error: ${JSON.stringify(err)}`),
    ()  => console.log('bar finished')
  );

注意上面我也摆脱了全局状态，这是函数式编程的诅咒。您的状态应尽可能本地化到流中。

在此处查看工作示例：http://jsbin.com/pexayohoho/1/edit?js,console

Answer 2

我不确定您想要实现什么，但这里有一个 jsbin that 尝试复制您的代码。

注意事项：

您的 foo() 使用 Observable.timer 和 .take() 运算符更容易创建。

你的 bar() 可以被另一个带有地图和 .takeWhile() 运算符的计时器使用。

至于最后一次订阅（完整部分），它只会打印 'foo finished' 而不会打印其他内容，因为它订阅了一个已经终止的非缓冲序列。