How can I poll status using Reactive Extensions?

Question

There is already a good question about polling a database using Reactive ( Polling a database with reactive extensions )

I have a similar question, but with a twist: I need to pass the value from the previous result to the next query. Basically, I would like to poll this:

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

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

The idea is that it returns all new items from a previous request

---

• every minute I would like to call GetNewResultsAsync
• I would like to pass CurrentAsOffrom the previous call as an argument to the parameterpreviousRequest
• the next call GetNewResultsAsyncshould happen a minute after the previous

---

Basically, is there a better way than:

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 )

Answer 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:

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 , , . , . - , .

Answer 2

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