如何在评估缓冲区时根据缓冲区中的值暂停 RxJS 缓冲的 observable？

Question

我有一个 observable，它封装了来自服务器的 socket.io 事件流（称为 source）。来自 socket.io 的每条消息都会在 observable 上发出。这个 observable 会根据 socket.io 消息的内容过滤并映射到多个订阅。

例如：

var filtered = source.filter(function(msg) { return msg.name === 'EventName' });
filtered.subscribe(function handeEventName(msg) { /* do something with msg */ });

有时这些订阅会在我的应用中触发较长的动画。发生这种情况时，我想暂停源 observable，缓冲新事件直到动画播放完毕，然后恢复 observable。

使用pausableBuffered，我已经按预期完成了所有这些工作：

var pausable = source.pausableBuffered();
var filtered = pausable.filter(function(msg) { return msg.name === 'EventName' });
filtered.subscribe(function handeEventName(msg) {
    pausable.pause();
    /**
     * do something async, like animation, then when done call
     * pausable.resume();
     */
});

到目前为止一切顺利。

但是，我们假设当 observable 暂停时，缓冲了 5 条消息。第三条消息是需要再次暂停流的消息。它有一个订阅设置来这样做。然而，一旦源 observable 未暂停，它会立即清空所有五个事件的缓冲区，所有这些事件都得到处理并传递给所有五个订阅，此时第三条消息的订阅最终暂停原始流。

我明白为什么会这样，但我真正想要的是：

1. 来源已暂停
2. 缓冲了五个事件，第三个事件应在处理订阅时暂停源。
3. 来源已恢复。
4. 事件 #1 和 #2 由它们的订阅处理，
5. 事件 #3 的订阅暂停源。
6. 可能有更多的事件被缓冲在 #4 和 #5 后面，它们仍在缓冲区中等待。
7. 事件 #3 的订阅在短时间内恢复源
8. 事件 #4 和 #5 以及任何其他事件开始传播，直到源发出另一个应该暂停的事件。

似乎我使用pausableBuffered 的每一种方式最终都会将整个缓冲区转储到他们的所有订阅中。我怎样才能实现我正在寻找的东西？

Answer 1

你可以试试controlled observable。为您提供几乎完全的控制权。 例如：

var source = Rx.Observable.interval(300).take(10);
var controlled = source.controlled();

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

var controlledSub = controlled.subscribe(
    function (x) {
        console.log('Next controlled: ' + x.toString());
        if (x === 3) {
            setTimeout(function(){
                controlled.request(1)
            }, 2000)
        } else {
            controlled.request(1);
        }
    },
    function (err) {
        console.log('Error: ' + err);
    },
    function () {
        console.log('Completed');
    });

controlled.request(1);

plunker