Как опросить состояние с помощью реактивных расширений?

Ваш вопрос в основном сводится к следующему: есть функция 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 занимает больше времени, чем интервалы в источнике, наблюдаемое прервется. Я не уверен, почему. Если кто-то может исправить, очень обязан.

С тех пор я обнаружил, что существует перегрузка 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 я возвращаю интервал опроса.

So:

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