动态延迟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"));

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

• 在响应式扩展中如何使用动态延迟进行订阅？

- Cel

Answer 1

看起来有一个新的 .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)));

根据用户输入，它会延迟与输入长度相等的时间。换句话说，每个元素的延迟现在可以依赖于数据本身。

- Cel

---

这应该被接受为答案，因为代码清晰，而且这是实现动态延迟的 Rx 方式。 - marsop

您确定在样本中正确使用了 Delay 过载吗？ - SuperJMN

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

- Andrey the Autobot

Answer 3

我认为这个可以运行：

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

如果您的延迟时间减少，生成的流可能与原始流的顺序不同。

- Matthew Finlay

---

根据我的初步测试，你的代码完美地运行了（带有一个递增时间函数）。 - Cel

Answer 4

如果你想要延迟一次，可以使用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));

- Richard Anthony Freeman-Hein

---

你的代码实现了一个固定的延迟，这正是我已经成功实现的（我的第一个代码示例，其中Result确实是一个IObservable）- 但是这个问题不是关于固定时间延迟的，而是延迟应该是一个具有可变返回值的函数（GetAsymptotingTime(); 应该在每个IObservable值上被多次调用）。 - Cel