如何在 Reactive Extension 中实现动态时间间隔 [英] How to impelment dynamic time interval in Reactive Extension

问题描述

我有一个场景，每个滴答事件的计时器间隔都会发生变化.如下代码所示:

I have a scenario where timer interval changes on every tick event. As shown in below code:

    Timer tmrObj = new Timer();
    tmrObj.Interval = TimeSpan.FromSeconds(11);
    tmrObj.Tick += TimerTickHandler;

   public void TimerTickHandler(EventArg arg)
   {
     tmrObj.pause();

     var response = MakeSomeServiceCall();
     tmr.Interval = response.Interval;

     tmrObj.resume();
   }

如果我需要在 Rx 中实现相同的计时器.我可以使用定时器功能来实现.但是如何在事件刻度上操作 Interval ，如上面的代码所示.当前定时器间隔实现如下:

If I need to implement Timers in Rx for the same. I can achieve using Timer function. But how can I manipulate Interval on event tick as shown in the above code. The current timer interval implementation is as below:

var serviceCall = Observable.FromAsync<DataResponse>(MakeServiceCall);
var timerCall = Observable.Timer(TimeSpan.FromSeconds(100));

var response = from timer in timerCall
               from reponse in serviceCall.TakeUntil(timerCall)
               .Select(result => result); 

推荐答案

如果是非异步生成，您可以使用 Generate 来处理数据生成.如果您的方法将使用异步，但您可以推出自己的 GenerateAsync 方法:

You can use Generate to handle the data generation if it is a non-async generation. If your method is going to being using async though you can roll your own GenerateAsync method:

public static IObservable<TOut> GenerateAsync<TResult, TOut>(
    Func<Task<TResult>> initialState,
    Func<TResult, bool> condition,
    Func<TResult, Task<TResult>> iterate,
    Func<TResult, TimeSpan> timeSelector,
    Func<TResult, TOut> resultSelector,
    IScheduler scheduler = null) 
{
  var s = scheduler ?? Scheduler.Default;

  return Observable.Create<TOut>(async obs => {

    //You have to do your initial time delay here.
    var init = await initialState();
    return s.Schedule(init, timeSelector(init), async (state, recurse) => 
    {
      //Check if we are done
      if (!condition(state))
      {
        obs.OnCompleted();
        return;
      }

      //Process the result
      obs.OnNext(resultSelector(state));

      //Initiate the next request
      state = await iterate(state);

      //Recursively schedule again
      recurse(state, timeSelector(state));

    });
  });
}

请参阅原始答案

你可以这样使用它:

var timeStream = ObservableStatic.GenerateAsync(
  () => MakeServiceCall(),
  _ => true,
  _ => MakeServiceCall(),
  result => result.Interval,
  _ => _);

这篇关于如何在 Reactive Extension 中实现动态时间间隔的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！