4

已经有一个关于使用 Reactive 进行数据库轮询的好问题（Database polling with Reactive Extensions）

我有一个类似的问题，但有一个转折：我需要将上一个结果中的值提供给下一个请求。基本上，我想对此进行投票：

interface ResultSet<T>
{
   int? CurrentAsOfHandle {get;}
   IList<T> Results {get;}
}

Task<ResultSet<T>> GetNewResultsAsync<T>(int? previousRequestHandle);

这个想法是返回自上次请求以来的所有新项目

  1. 每分钟我都想打电话GetNewResultsAsync
  2. 我想将CurrentAsOf上一次调用中的作为参数传递给previousRequest参数
  3. 下一个呼叫GetNewResultsAsync实际上应该在上一个呼叫之后一分钟发生

基本上，有没有比以下更好的方法：

return Observable.Create<IMessage>(async (observer, cancellationToken) =>
{
    int? currentVersion = null;

    while (!cancellationToken.IsCancellationRequested)
    {
        MessageResultSet messageResultSet = await ReadLatestMessagesAsync(currentVersion);

        currentVersion = messageResultSet.CurrentVersionHandle;

        foreach (IMessage resultMessage in messageResultSet.Messages)
            observer.OnNext(resultMessage);

        await Task.Delay(TimeSpan.FromMinutes(1), cancellationToken);
    }
});

另请注意，此版本允许messageResultSet在等待下一次迭代时收集 （例如，我想也许我可以使用Scan将先前的结果集对象传递给下一次迭代）

4

2 回答 2

1

您的问题基本上可以简化为：有一个Scan带有签名的函数：

IObservable<TAccumulate> Scan<TSource, TAccumulate>(this IObservable<TSource> source, 
     TAccumulate initialValue, Func<TAccumulate, TSource, TAccumulate> accumulator)

但是你需要类似的东西

IObservable<TAccumulate> Scan<TSource, TAccumulate>(this IObservable<TSource> source, 
     TAccumulate initialValue, Func<TAccumulate, TSource, IObservable<TAccumulate>> accumulator)

...累加器函数返回一个可观察对象，而 Scan 函数自动将其减少以传递给下一次调用。

这是一个穷人的功能实现Scan：

public static IObservable<TAccumulate> MyScan<TSource, TAccumulate>(this IObservable<TSource> source, 
    TAccumulate initialValue, Func<TAccumulate, TSource, TAccumulate> accumulator)
{
    return source
        .Publish(_source => _source
            .Take(1)
            .Select(s => accumulator(initialValue, s))
            .SelectMany(m => _source.MyScan(m, accumulator).StartWith(m))
        );
}

鉴于此，我们可以对其进行一些更改以包含缩减功能：

public static IObservable<TAccumulate> MyObservableScan<TSource, TAccumulate>(this IObservable<TSource> source,
    TAccumulate initialValue, Func<TAccumulate, TSource, IObservable<TAccumulate>> accumulator)
{
    return source
        .Publish(_source => _source
            .Take(1)
            .Select(s => accumulator(initialValue, s))
            .SelectMany(async o => (await o.LastOrDefaultAsync())
                .Let(m => _source
                    .MyObservableScan(m, accumulator)
                    .StartWith(m)
                )
            )
            .Merge()
        );
}

//Wrapper to accommodate easy Task -> Observable transformations
public static IObservable<TAccumulate> MyObservableScan<TSource, TAccumulate>(this IObservable<TSource> source,
    TAccumulate initialValue, Func<TAccumulate, TSource, Task<TAccumulate>> accumulator)
{
    return source.MyObservableScan(initialValue, (a, s) => Observable.FromAsync(() => accumulator(a, s)));  
}

//Function to prevent re-evaluation in functional scenarios
public static U Let<T, U>(this T t, Func<T, U> selector)
{
    return selector(t);
}

现在我们有了这个花哨MyObservableScan的运算符，我们可以相对容易地解决您的问题：

var o = Observable.Interval(TimeSpan.FromMinutes(1))
    .MyObservableScan<long, ResultSet<string>>(null, (r, _) => Methods.GetNewResultsAsync<string>(r?.CurrentAsOfHandle))

请注意，在测试中我注意到如果累加器Task/Observable 函数花费的时间比源中的时间间隔长，则可观察对象将终止。我不确定为什么。如果有人可以纠正，非常有义务。

于 2017-01-29T04:29:33.487 回答
0

从那以后，我发现这Observable.Generate几乎可以解决问题。主要缺点是它不适用于async.

public static IObservable<TResult> Generate<TState, TResult>(TState initialState, Func<TState, bool> condition, Func<TState, TState> iterate, Func<TState, TResult> resultSelector, Func<TState, TimeSpan> timeSelector, IScheduler scheduler);

null我作为我的初始状态传入。作为我的条件传入（无休止x => true地轮询）。在 内部iterate，我根据传入的状态进行实际轮询。然后timeSelector返回轮询间隔。

所以：

var resultSets = Observable.Generate<ResultSet<IMessage>, IEnumerable<IMessage>>(
   //initial (empty) result
   new ResultSet<IMessage>(),

   //poll endlessly (until unsubscription)
   rs => true,

   //each polling iteration
   rs => 
   {
      //get the version from the previous result (which could be that initial result)
      int? previousVersion = rs.CurrentVersionHandle;

      //here's the problem, though: it won't work with async methods :(
      MessageResultSet messageResultSet = ReadLatestMessagesAsync(currentVersion).Result;

      return messageResultSet;
   },

   //we only care about spitting out the messages in a result set
   rs => rs.Messages, 

   //polling interval
   TimeSpan.FromMinutes(1),

   //which scheduler to run each iteration 
   TaskPoolScheduler.Default);

return resultSets
  //project the list of messages into a sequence
  .SelectMany(messageList => messageList);
于 2017-02-24T15:22:57.200 回答