将`IObservable `转换为`IEnumerable >`？

Question

我有几个使用回调或事件而不是async的异步API。我成功地使用TaskCompletionSource来包装它们as described here。

现在，我想使用一个返回IObservable<T>并产生多个对象的API。我读到有关Rx for .NET的信息，这似乎是可行的方法。但是，我不愿意包含另一个依赖关系和另一个新范例，因为我已经在此应用程序中使用了很多对我来说是新的东西（例如XAML，MVVM，C＃的async / await）。

是否有类似于包裹单个回调API的包裹IObservable<T>的方法？我想这样调用API：

foreach (var t in GetMultipleInstancesAsync()) {
    var res = await t;
    Console.WriteLine("Received item:", res);
}

Answer 1

如果观察对象发出多个值，则可以将它们转换为Task<T>，然后将它们添加到任何IEnumberable结构中。

检查IObservable ToTask。如此处所述，可观察对象必须在等待之前完成，否则可能会出现更多的值。

This guide here might do the trick for you too

public static Task<IList<T>> BufferAllAsync<T>(this IObservable<T> observable)
{
    List<T> result = new List<T>();
    object gate = new object();
    TaskCompletionSource<IList<T>> finalTask = new TaskCompletionSource<IList<T>>();

    observable.Subscribe(
        value =>
        {
            lock (gate)
            {
                result.Add(value);
            }
        },
        exception => finalTask.TrySetException(exception),
        () => finalTask.SetResult(result.AsReadOnly())
    );

    return finalTask.Task;
}

Answer 2

如果您想使用Rx，则可以使用返回列表：

GetMultipleInstancesAsync().ToObservable().Subscribe(...);

您可以订阅OnCompleted / OnError处理程序。

也可以将任务列表包装在其中：

var result = await Task.WhenAll(GetMultipleInstancesAsync().ToArray());

所以您得到了一系列的结果，您就完成了。