延迟功能可动态推迟 IObservable 值

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"));

当变量延迟的代码编译时，它只被调用一次，只提供第一个值（本质上是一个固定值）。

• 如何在反应式扩展中订阅动态延迟？

Answer 1

看起来有一个新的重载。允许在 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 2

可能是我没有把问题做对，但是当我在寻找方法时，我已经踏入了它在项目之间放置相同的固定延迟，这些延迟来得太快了。以前的解决方案对我不起作用，（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();

Answer 3

我认为这有效：

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

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

Answer 4

如果你想延迟一次，那么你可以使用 Observable.Delay。 我不知道结果的类型是什么，所以我假设它已经是IObservable，所以，你可以做这样的事情：

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