RxJS 中的 extent / manySelect 方法实际上是做什么的？

Question

根据https://github.com/Reactive-Extensions/RxJS/blob/master/doc/api/core/operators/manyselect.md，extent/manySelect方法是一个comonadic转换器，其形式为Observable a -> (Observable a -> b) -> Observable b，是selectMany/flatMat的“反转”版本。

文档中给出的示例：

var source = Rx.Observable.range(0, 3)
    .manySelect(function (ys) { return ys.first(); })
    .mergeAll();

var subscription = source.subscribe(
    function (x) {
        console.log('Next: ' + x);
    },
    function (err) {
        console.log('Error: ' + err);
    },
    function () {
        console.log('Completed');
    });

// => Next: 0
// => Next: 1
// => Next: 2
// => Completed

我的理解是manySelect将每个观察值包装成Observable（通过Rx.Observable.just(x)之类的东西），然后将其放入选择器函数中，最后将所有返回值合并成一个新的Observable。

但是，如果我将ys.first() 更改为ys.count() 或ys.last()，程序会崩溃并显示Error: TypeError: innerSource.subscribe is not a function。

发生了什么事？ manySelect 到底是做什么的，我们应该在什么时候使用它？

JSBin

还有一个小问题： 在测试 manySelect 时，我发现 `Rx.Observable.range(0, 5).flatMap(() => [1, 2, 3, 4])` 的结果如下所示： onNext: 1 onNext: 2 onNext: 1 onNext: 3 onNext: 2 onNext: 1 onNext: 4 onNext: 3 onNext: 2 onNext: 1 onNext: 4 onNext: 3 onNext: 2 onNext: 1 onNext: 4 onNext: 3 onNext: 2 onNext: 4 onNext: 3 onNext: 4 完全的 而 `Rx.Observable.interval(0).take(10).flatMap(() => [1, 2, 3, 4])` 给出了正确序列的结果。 这是错误还是功能？

感谢@user3743222，我找到了flatMap行为的答案：

请注意，FlatMap 会合并这些 Observable 的发射，以便它们可以交错。

Answer 1

两个额外的理解资源manySelect：

• the unit tests!
• the marbles。这应该清楚为什么您的最新示例的行为如此。对于您的第一个问题，您应该提供一个 jsfiddle 以便可以复制它。
• the actual implementation

manySelect 运算符通常被描述为“comonadic 绑定”。如果这可以为​​您解决问题，不客气。否则， 这是一个解释：

manySelect 在内部转换源发出的每个项目 将 Observable 转换为发出该项目和所有项目的 Observable 随后由源 Observable 以相同的顺序发出。所以， 例如，它在内部转换了一个 Observable，该 Observable 发出 将数字 1,2,3 分成三个 Observable：一个发出 1,2,3，一个发出 发射 2,3，一个发射 3。

然后manySelect 将这些 Observable 中的每一个传递给一个函数 你提供和发出的，作为来自 Observable 的发射 manySelect 返回这些函数调用的返回值。

这样，生成的 Observable 发出的每个项目都是一个 源 Observable 中对应项的函数和所有的 源 Observable 在它之后发出的项目。