正在保存订阅状态以便稍后恢复

更新-已解决

最终的解决方案与布兰登的建议有点不同，但他的回答让我走上了正轨。

class State
{
  public int Offset { get; set; }
  public HashSet<string> UniqueImageUrls = new HashSet<string>();
}
public IObservable<TPicture> GetPictures(ref object _state)
{
  var localState = (State) _state ?? new State();
  _state = localState;
  return Observable.Defer(()=>
  {
    return Observable.Defer(() => Observable.Return(GetPage(localState.Offset)))
      .SubscribeOn(TaskPoolScheduler.Default)
      .Do(x=> localState.Offset += 20)
      .Repeat()
      .TakeWhile(x=> x.Count > 0)
      .SelectMany(x=> x)
      .Where(x=> !localState.UniqueImageUrls.Contains(x.ImageUrl))
      .Do(x=> localState.UniqueImageUrls.Add(x.ImageUrl));
  });
}
IList<TPicture> GetPage(int offset)
{
  ... 
  return result;
}

原始问题

我目前正在努力解决以下问题。下面显示的PictureProvider实现使用一个偏移量变量，该变量用于分页提供实际数据的后端服务的结果。我想实现的是一个优雅的解决方案，使当前偏移量可用于可观察的消费者，以允许在稍后以正确的偏移量恢复可观察的序列。GetPictures（）的initialState参数已经说明了恢复。

以更像RX的方式改进代码的建议也将受到欢迎。实际上，我不太确定Task.Run（）的内容在这里是否合适。

  public class PictureProvider :
    IPictureProvider<Picture>
  {
    #region IPictureProvider implementation
    public IObservable<Picture> GetPictures(object initialState)
    {
      return Observable.Create<Picture>((IObserver<Picture> observer) =>
      {
        var state = new ProducerState(initialState);
        ProducePictures(observer, state);
        return state;
      });
    }
    #endregion
    void ProducePictures(IObserver<Picture> observer, ProducerState state)
    {
      Task.Run(() =>
      {
        try
        {
          while(!state.Terminate.WaitOne(0))
          {
            var page = GetPage(state.Offset);
            if(page.Count == 0)
            {
              observer.OnCompleted();
              break;
            }
            else
            {
              foreach(var picture in page)
                observer.OnNext(picture);

              state.Offset += page.Count;
            }
          }
        }
        catch (Exception ex)
        {
          observer.OnError(ex);
        }
        state.TerminateAck.Set();
      });
    }
    IList<Picture> GetPage(int offset)
    {
      var result = new List<Picture>();
      ... boring web service call here
      return result;
    }
    public class ProducerState :
      IDisposable
    {
      public ProducerState(object initialState)
      {
        Terminate = new ManualResetEvent(false);
        TerminateAck = new ManualResetEvent(false);
        if(initialState != null)
          Offset = (int) initialState;
      }
      public ManualResetEvent Terminate { get; private set; }
      public ManualResetEvent TerminateAck { get; private set; }
      public int Offset { get; set; }
      #region IDisposable implementation
      public void Dispose()
      {
        Terminate.Set();
        TerminateAck.WaitOne();
        Terminate.Dispose();
        TerminateAck.Dispose();
      }
      #endregion
    }
  }