无法从串联的 Observable 中提取/解包值

Question

我有这个链，它应该将 10 个 Observable 连接成 1 个 Observable，其中 10 个 Observable 中的每一个基本上都应该被解包为一个整数：

const Rx = require('rxjs');

var i = 0;
const obs = Rx.Observable.interval(10)
  .map(() => i++)
  .map(val => Rx.Observable.create(obs => {
          obs.next(val)
  }))
  .take(10)
  .reduce((prev, curr) => {
      return prev.concat(curr);  // concat all observables
  })
  .last(val => val.flatMap(inner => inner));

// subscribe to Observable
obs.subscribe(v => {
  console.log('\n next (and only) result => \n', v);
});

发生的事情是所有 10 个 Observables 应该连接在一起，但我无法从这 10 个 Observables（它们已成为 1 个 Observable）中提取值。所以我的问题是，我怎样才能解开最终的 observable 并提取值？

有人知道我在说什么吗？

Answer 1

这个问题与Observable.prototype.concatAll does not seem to yield expected result中的问题完全相同

.map(function(val){  // note: map *not* flatMap
  return Rx.Observable.create(obs => {
      obs.next(val)
  });
})

使用 Rx.Observable.create 创建自己的 observable 时，您需要自己 .complete()。因为您忘记了这样做，所以像 .reduce() 这样的运算符将无法工作，因为它们需要等待完成才能运行。

此外，您对.last() 运算符的使用不正确；它需要一个谓词，您的流将在该谓词上被过滤，并且将发出与谓词匹配的最后一个发射。这也有点多余，因为您的 .reduce() 只会发出一个值。清理它会导致：

const obs = Rx.Observable.interval(10)
  .map(() => i++)
  .map(val => Rx.Observable.of(val))
  .take(10)
  .reduce((acc, curr) => acc.concat(curr))
  .flatMap(v => v)
  .toArray()

但是你可以通过直接使用.concatMap() 操作符而不是map+take+reduce+flatMap 来缩短它：

const obs = Rx.Observable.interval(10)
  .map(() => i++)
  .concatMap(val => Rx.Observable.of(val))
  .take(10)
  .toArray()

Answer 2

您可以使用 concatAll（或 concatMap）来连接内部序列并保持顺序，而不是 reduce + last：

Observable
  .interval(0)
  .map(i => Observable.of(i))
  .take(10)
  .concatAll()
  .takeLast(1)
  .subscribe(v => console.log('\n next (and only) result => \n', v);)

编辑：takeLast(1) 而不是 finalValue()

Answer 3

使用非原型方法。

let range = [...Array(10).keys()];

//array of 10 observables of 10 values each
let ten = range.map(i => Rx.Observable.interval(1000).map(_ => i).take(10));     
let sequence = Rx.Observable.concat(ten) // all of them in sequence