c# - 可从链式任务中观察到

Question

我正在尝试创建一个 Observable，其中每个项目都是通过异步任务生成的。下一个项目应该通过对前一个项目(共同递归)的结果的异步调用来产生。用“生成”的说法，这看起来像这样 - 除了 生成不支持异步(也不支持初始状态的委托(delegate)。

var ob = Observable.Generate(
   async () => await ProduceFirst(),        // Task<T> ProduceFirst()
   prev => Continue(prev)                   // bool Continue(T);
   async prev => await ProduceNext(prev)    // Task<T> ProduceNext(T)
   item => item
);

---

作为一个更具体的示例，要通过一次获取 100 条消息来查看 ServiceBus 队列中的所有消息，请按如下方式实现 ProduceFirst、Continue 和 ProduceNext:

Task<IEnumerable<BrokeredMessage>> ProduceFirst() 
{
    const int batchSize = 100;
    return _serviceBusReceiver.PeekBatchAsync(batchSize);
}

bool Continue(IEnumerable<BrokeredMessage> prev)
{
    return prev.Any();
}

async Task<IEnumerable<BrokeredMessage>> ProduceNext(IEnumerable<BrokeredMessage> prev) 
{
    const int batchSize = 100;
    return (await _serviceBusReceiver.PeekBatchAsync(prev.Last().SequenceNumber, batchSize + 1)).Skip(1)
}

然后调用.SelectMany(i => i)在 IObservable<IEnumerable<BrokeredMessage>> 上把它变成一个IObservable<BrokeredMessage>

其中 _serviceBusReceiver 是接口(interface)的实例，如下所示:

public interface IServiceBusReceiver
{
    Task<IEnumerable<BrokeredMessage>> PeekBatchAsync(int batchSize);
    Task<IEnumerable<BrokeredMessage>> PeekBatchAsync(long fromSequenceNumber, int batchSize);
}

BrokeredMessage 来自https://msdn.microsoft.com/en-us/library/microsoft.servicebus.messaging.brokeredmessage.aspx

Answer 1

如果您打算推出自己的异步 Generate 函数，我建议您使用递归调度而不是包装一个 while 循环。

public static IObservable<TResult> Generate<TResult>(
    Func<Task<TResult>> initialState,
    Func<TResult, bool> condition,
    Func<TResult, Task<TResult>> iterate,
    Func<TResult, TResult> resultSelector,
    IScheduler scheduler = null) 
{
  var s = scheduler ?? Scheduler.Default;

  return Observable.Create<TResult>(async obs => {
    return s.Schedule(await initialState(), async (state, self) => 
    {
      if (!condition(state))
      {
        obs.OnCompleted();
        return;
      }

      obs.OnNext(resultSelector(state));

      self(await iterate(state));

    });
  });
}

这有两个优点。首先，您可以取消它，使用一个简单的 while 循环是无法直接取消它的，事实上，您甚至不会在 observable 完成之前返回订阅函数。其次，这使您可以控制每个项目的调度/异步(这使测试变得轻而易举)，这也使它更适合图书馆