Rxjs observables 减少

Question

我正在处理字符串，但有些函数是可观察的，所以我试图找出一种简单的方法来做到这一点。

假设我有以下课程：

class Something {
  doSomething(input: string): Observable<string> {
    return of('123 - ' + input);
  }
}

of('123 - ' + input) 是被调用并返回字符串的服务（如 http.get()）。

我想将服务的每个结果附加到下一个服务的结果，直到达到最后一个。 但我不知道如何做到这一点。

我得到的最新信息如下：

return of(
  new Something(),
  new Something(),
  new Something(),
)
  .pipe(
    reduce(
      (acc, curr, i) => {
        console.log('NEXT', acc);
        if(!(curr instanceof Observable)){
          acc = of(acc);
        }
        return acc.pipe(map(c => curr.doSomething(c)));
      },
      'Start string: '
  );

这将导致Start string: 123 - 123 - 123 -，但事实并非如此。我已经没有办法让这个工作了

如果您想尝试一些事情，这里有一个基本的 stackblitz 入门： https://stackblitz.com/edit/typescript-htcna8?file=index.ts

Answer 1

发现您可以执行以下操作：

const source = [
  new Something(),
  new Something(),
  new Something(),
  new Something()
];

let chain = source[0].doSomething('Initial string');
source.splice(0, 1);

source.forEach(r => {
  chain = chain.pipe(
    switchMap(s => r.doSomething(s))
  );
});

Answer 2

好像你想要mergeScan:

const source = [
  new Something(),
  new Something(),
  new Something(),
  new Something()
];

from(source).pipe(
  // Scan over all the inner streams passing each result to the next call
  mergeScan((acc, s) => s.doSomething(acc), '', 1),
  // The above emits partial results, so ignore all except the last one.
  last()
).subscribe(console.log);

// Expected output: 123 - 123 - 123 - 123

文档here

在源 Observable 上应用一个累加器函数，其中累加器函数本身返回一个 Observable，然后返回的每个中间 Observable 合并到输出 Observable 中。

和scan类似，只是累加器返回的Observables被合并到外层的Observable中。

Answer 3

因为我不确定您的期望，所以我为您准备了 3 个案例：

// Expected output: One next with ["123 - 1", "123 - 2", "123 - 3", "123 - 4"]
forkJoin(source).subscribe(val => console.log(val));

// Expected output: Many next with : 
// * 123 - 1
// * 123 - 2
// * 123 - 3
// * 123 - 4
forkJoin(source).pipe(concatAll()).subscribe(val => console.log(val));

// Expected output : One next with string "123 - 1, 123 - 2, 123 - 3, 123 - 4"
forkJoin(source).pipe(map((e) => {
  return e.join(', ');
})).subscribe(val => console.log(val));

示例：https://stackblitz.com/edit/typescript-s3yr3h?file=index.ts