有条件延迟

Question

有一个代码：

someObservable.Select(x => getY(x));    
Y getY(X x)
{
  if (x.Value == X.ABC)
    return new Y(1);
  else
    return new Y(2);
}

在某些情况下，我需要在一段时间后仔细检查 x.Value。 最简单但最糟糕的解决方案是使用 Thread.Sleep：

Y getY(X x)
{
  if (x.Value == X.ABC)
    return new Y(1);
  else
  if (x.SomethingElse == true)
  {
     Thread.Sleep(timeout);
     if (x.Value == X.ABC)
       return new Y(1);
     else
       return new Y(2);
  }
}

这里的正确代码是什么？ 我需要以与接收相同的方式订购事件。 这意味着如果我有一个延迟并且我得到一个新值，它必须等待处理。

Answer 1

解决方案（来自https://rsdn.org/forum/dotnet/6629370.1）是在 getY 中返回 IObservable 而不是 Y，并使用 Concat 生成 Observable.Delay。

IObservable<Y> getY(X x)
{
  if (x.Value == X.ABC)
    return Observable.Return(new Y(1));
  else
  if (x.SomethingElse == true)
  {
     return Observable.Delay(Observable.Return(x), timeout).Select(xx => xx.Value == X.ABC ? new Y(1) : new Y(2));
  }
}

或

IObservable<Y> getY(X x)
{
  return Observable.Create<Y>(async (obs, token) =>
  {
    if (x.Value == X.ABC)
        obs.OnNext(new Y(1));
    else
    if (x.SomethingElse == true)
    {
        await Task.Delay(timeout, token);
        if (x.Value == X.ABC)
           obs.OnNext(new Y(1));
        else
           obs.OnNext(new Y(2));
     }
   }
 }

然后：someObservable.Select(x => getY(x)).Concat();

Answer 2

如果这是一个普遍的问题，我建议你实现一个自定义操作符。 delay 方法有一个重载，它接受另一个 observable 来控制延迟时间。示例中偶数立即推送，奇数延迟；

Observable.Interval(TimeSpan.FromSeconds(1))
          .Delay(i => (i % 2 == 0) ? Observable.Return(0L) : Observable.Timer(TimeSpan.FromSeconds(0.9)))
          .Subscribe(Console.WriteLine);

编辑： 下面是一个更简单的、保留顺序的延迟运算符：

  static IObservable<T> DelayOrdered<T>(this IObservable<T> observable, Func<T, TimeSpan> delaySelector, IScheduler scheduler = default(IScheduler))
        {
            scheduler = scheduler ?? DefaultScheduler.Instance;
            return Observable.Create<T>(observer =>
            {
                var now = scheduler.Now;

                return observable
                       .Subscribe(value =>
                {
                    now = now.Add(delaySelector(value));
                    scheduler.Schedule(now, () => observer.OnNext(value));
                });
            });
        }

用法：

  Observable.Interval(TimeSpan.FromSeconds(0.2))
                      .DelayOrdered(i => (i % 2 == 0) ? TimeSpan.Zero : TimeSpan.FromSeconds(1))
                      .Subscribe(Console.WriteLine);

订单将被保留，因为绝对预定时间只能增加。