当另一个Observable提供值时，暂时忽略Observable中的值

Question

当另一个Observable提供值时，我需要在一段时间内忽略Observable值。

目前，我的实现使用一个变量来控制阻塞（或忽略）。

bool block = false;
var blocker = observable1.Do(_ => block = true )
                         .Throttle( _ => Observable.Timer(_timeToBlock)
                         .Subscribe( _ => block = false ));
var receiver = observable2.Where( i => !block && SomeCondition(i) )
                          .Subscribe( i=> EvenMoreStuff(i) );

有没有一种更Rx的方法可以做到这一点，通过结合这两个可观察性？

编辑：拦截订阅的小更改

Answer 1

第一个任务是将block变量表示为可观察变量。

IObservable<bool> blockObservable = observable1
    .Select(x => Observable.Concat(
        Observable.Return(true),
        Observable.Return(false).Delay(_timeToBlock)))
    .Switch()
    .DistinctUntilChanged();

每次observable1发出一个值时，我们都会选择一个可观察到的值，该值会发出true，等待_timeToBlock，然后发出false。Switch总是切换到这些可观测值中的最新可观测值。

这是一张大理石图。假设_timeToBlock有3个字符长。

observable1      -----x--xx-x-----x-----
select0               T--F
select1                  T--F
select2                   T--F
select3                     T--F
select4                           T--F
switch           -----T--TT-T--F--T--F--
blockObservable  -----T--------F--T--F--

现在我们可以用blockObservable的MostRecent值来Zip您的值序列。

var receiverObservable = observable2
    .Zip(blockObservable.MostRecent(false), (value, block) => new { value, block })
    .Where(x => !x.block)
    .Select(x => x.value);

Answer 2

作为使用一次性用品的替代方案，您可以创建一个小的扩展方法：

public static IObservable<TResult> Suppress<TResult, TOther>(
                                    this IObservable<TResult> source, 
                                         IObservable<TOther> other,
                                         TimeSpan delayFor)
{
  return Observable.Create<TResult>(observer => {
    var published = source.Publish();
    var connected = new SerialDisposable();
    Func<IDisposable> connect = () => published.Subscribe(observer);
    var suppressor = other.Select(_ => Observable.Timer(delayFor)
                                .Select(_2 => connect())
                                .StartWith(Disposable.Empty))
    .Switch();
    return new CompositeDisposable(
                connected,
                suppressor.StartWith(connect())
                          .Subscribe(d => connected.Disposable = d),
                published.Connect());
  });
}

这将把可观察的源转换为ConnectableObservable，然后每次other源发出时，它都会处理其订阅，然后在计时器到期时重新连接。