如何在响应式扩展中实现动态时间间隔

本文关键字：动态 实现 时间 扩展 响应 | 更新日期: 2023-09-27 18:08:51

我有一个场景，其中计时器间隔在每个tick事件上更改。如下面的代码所示:

    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中实现定时器。我可以实现使用定时器功能。但是，我如何像上面的代码所示的那样在事件tick上操作Interval呢?当前定时器间隔实现如下:

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方法:

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,
  _ => _);
相关文章：