如何在流 B 触发时抑制流 A 的下一个事件

Question

我想每当流 B 触发时就停止流 A 以获得一个通知。两个流都将保持在线状态并且永远不会完成。

A: o--o--o--o--o--o--o--o--o  
B: --o-----o--------o-------  
R: o-----o-----o--o-----o--o  

或者

A: o--o--o--o--o--o--o--o--o  
B: -oo----oo-------oo-------  
R: o-----o-----o--o-----o--o  

Answer 1

这是我为 类似问题所做的 SkipWhen 运算符的版本 （区别在于，在原来的情况下，多个“B”会跳过多个“A”）：

public static IObservable<TSource> SkipWhen<TSource, TOther>(this IObservable<TSource> source, 
    IObservable<TOther> other)
{
    return Observable.Create<TSource>(observer =>
    {
        object lockObject = new object();
        bool shouldSkip = false;

        var otherSubscription = new MutableDisposable();
        var sourceSubscription = new MutableDisposable();

        otherSubscription.Disposable = other.Subscribe(
            x => { lock(lockObject) { shouldSkip = true; } });

        sourceSubscription.Disposable = source.Where(_ =>
        {
            lock(lockObject)
            {
                if (shouldSkip)
                {
                    shouldSkip = false;
                    return false;
                }
                else
                {
                    return true;
                }
            }
        }).Subscribe(observer);

        return new CompositeDisposable(
            sourceSubscription, otherSubscription);
    });
}

如果当前实现成为瓶颈，请考虑更改锁实现以使用 ReaderWriterLockSlim。

Answer 2

当可观察量很热时（并且没有 refCount），此解决方案将起作用：

streamA
    .takeUntil(streamB)
    .skip(1)
    .repeat()
    .merge(streamA.take(1))
    .subscribe(console.log);

1. .takeUntil(streamB)：使流 A 在流 B 产生一个值。
2. .skip(1)：使流 A 在启动时跳过一个值（或作为 .repeat() 的结果）。
3. .repeat()：使流A无限期地重复（重新连接）。
4. .merge(streamA.take(1))：在流的开头偏移 .skip(1) 的效果。

让 A 流每 5 秒跳过一次的示例：

var streamA,
    streamB;

streamA = Rx.Observable
    .interval(1000)
    .map(function (x) {
        return 'A:' + x;
}).publish();

streamB = Rx.Observable
    .interval(5000);

streamA
    .takeUntil(streamB)
    .skip(1)
    .repeat()
    .merge(streamA.take(1))
    .subscribe(console.log);

streamA.connect();

您还可以使用此沙箱 http://jsbin.com/gijorid/4/edit?js ,console 在运行代码时在控制台日志中执行BACTION()，手动推送一个值到streamB（这有助于分析代码）。