延迟功能可动态延迟IObable值

Question

[这个问题是关于IObservable / Rx]

工作固定延迟

var frequency = TimeSpan.FromMinutes(5);
Result.Delay(frequency).Subscribe(i => Debug.WriteLine("After Fixed Delay"));

可变延迟的伪代码

Result.Delay(GetAsymptotingTime()).Subscribe(i => Debug.WriteLine("After Changing Delay"));

虽然可变延迟的代码仅被编译一次，仅提供第一个值（本质上是固定值）。

• 您如何在Reactive Extensions中动态延迟订阅？

Answer 1

我认为这可行：

source
  .SelectMany(i => Observable.Timer(GetAsymptotingTime()).Select(_=>i))

如果延迟减少，则结果流可能与原始流的顺序不同。

Answer 2

好像有一个新的.Delay重载，它允许RX本身具有此功能：

来自http://blogs.msdn.com/b/rxteam/archive/2012/03/12/reactive-extensions-v2-0-beta-available-now.aspx：

var res = input.Delay(x => Observable.Timer(TimeSpan.FromSeconds(x.Length)));

给定用户输入，它获得的延迟时间等于输入长度（以秒为单位）。换句话说，每个元素的延迟现在可以取决于数据本身。

Answer 3

如果要延迟一次，则可以使用Observable.Delay。我不知道Result的类型是什么，所以我假设它已经是IObservable了，因此，您可以执行以下操作：

    var Result = Observable.Range(0, 10);
var frequency = TimeSpan.FromMilliseconds(1500);
var delay = Result.Delay(frequency);
delay.Subscribe(x => Debug.WriteLine(x));

Answer 4

可能是我没有正确回答问题，但是当我寻找一种方法时我已经介入了在项目之间设置相同的固定延迟，这些延迟太快了。先前的解决方案对我不起作用，实际上（RX 4.0）完全没有延迟。

我的解决方法很简单：

myObservable.Zip(Observable.Interval(TimeSpan.FromSeconds(1)), (a, _) => a)

我在测试动态序列时使用过它，例如

      using var d2 = new int[] { 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17 }
        .ToObservable()
        .Zip(Observable.Interval(TimeSpan.FromSeconds(1)), (a, _) => a)
        .Do(a => Log.Logger.Debug("b {a}", a))
        .Subscribe();