有没有办法解开IObservable<Task<T>>
进入IObservable<T>
保持相同的事件顺序,就像这样?
Tasks: ----a-------b--c----------d------e---f---->
Values: -------A-----------B--C------D-----E---F-->
假设我有一个使用消息流的桌面应用程序,其中一些消息需要大量的后期处理:
IObservable<Message> streamOfMessages = ...;
IObservable<Task<Result>> streamOfTasks = streamOfMessages
.Select(async msg => await PostprocessAsync(msg));
IObservable<Result> streamOfResults = ???; // unwrap streamOfTasks
我设想了两种处理方式。
首先,我可以订阅streamOfTasks
使用异步事件处理程序:
streamOfTasks.Subscribe(async task =>
{
var result = await task;
Display(result);
});
其次,我可以转换streamOfTasks
使用 Observable.Create
,像这样:
var streamOfResults =
from task in streamOfTasks
from value in Observable.Create<T>(async (obs, cancel) =>
{
var v = await task;
obs.OnNext(v);
// TODO: don't know when to call obs.OnComplete()
})
select value;
streamOfResults.Subscribe(result => Display(result));
无论哪种方式,消息的顺序都不会保留:一些较晚的消息 不需要任何后期处理比早期的消息更快 需要后期处理。我的两个解决方案都处理传入的消息 并行处理,但我希望它们按顺序、一个接一个地处理。
我可以编写一个简单的任务队列来一次只处理一个任务, 但也许这是一个矫枉过正。在我看来,我遗漏了一些明显的东西。
更新。我编写了一个示例控制台程序来演示我的方法。到目前为止,所有解决方案都不会保留事件的原始顺序。这是程序的输出:
Timer: 0
Timer: 1
Async handler: 1
Observable.Create: 1
Observable.FromAsync: 1
Timer: 2
Async handler: 2
Observable.Create: 2
Observable.FromAsync: 2
Observable.Create: 0
Async handler: 0
Observable.FromAsync: 0
完整的源代码如下:
// "C:\Program Files (x86)\MSBuild\14.0\Bin\csc.exe" test.cs /r:System.Reactive.Core.dll /r:System.Reactive.Linq.dll /r:System.Reactive.Interfaces.dll
using System;
using System.Reactive;
using System.Reactive.Concurrency;
using System.Reactive.Linq;
using System.Threading.Tasks;
class Program
{
static void Main()
{
Console.WriteLine("Press ENTER to exit.");
// the source stream
var timerEvents = Observable.Timer(TimeSpan.Zero, TimeSpan.FromSeconds(1));
timerEvents.Subscribe(x => Console.WriteLine($"Timer: {x}"));
// solution #1: using async event handler
timerEvents.Subscribe(async x =>
{
var result = await PostprocessAsync(x);
Console.WriteLine($"Async handler: {x}");
});
// solution #2: using Observable.Create
var processedEventsV2 =
from task in timerEvents.Select(async x => await PostprocessAsync(x))
from value in Observable.Create<long>(async (obs, cancel) =>
{
var v = await task;
obs.OnNext(v);
})
select value;
processedEventsV2.Subscribe(x => Console.WriteLine($"Observable.Create: {x}"));
// solution #3: using FromAsync, as answered by @Enigmativity
var processedEventsV3 =
from msg in timerEvents
from result in Observable.FromAsync(() => PostprocessAsync(msg))
select result;
processedEventsV3.Subscribe(x => Console.WriteLine($"Observable.FromAsync: {x}"));
Console.ReadLine();
}
static async Task<long> PostprocessAsync(long x)
{
// some messages require long post-processing
if (x % 3 == 0)
{
await Task.Delay(TimeSpan.FromSeconds(2.5));
}
// and some don't
return x;
}
}
最佳答案
结合@Enigmativity 的 simple approach与@VMAtm 的想法 attaching the counter和来自 this SO question 的一些代码片段,我想出了这个解决方案:
// usage
var processedStream = timerEvents.SelectAsync(async t => await PostprocessAsync(t));
processedStream.Subscribe(x => Console.WriteLine($"Processed: {x}"));
// my sample console program prints the events ordered properly:
Timer: 0
Timer: 1
Timer: 2
Processed: 0
Processed: 1
Processed: 2
Timer: 3
Timer: 4
Timer: 5
Processed: 3
Processed: 4
Processed: 5
....
这是我的 SelectAsync
转换扩展方法IObservable<Task<TSource>>
进入IObservable<TResult>
保持事件的原始顺序:
public static IObservable<TResult> SelectAsync<TSource, TResult>(
this IObservable<TSource> src,
Func<TSource, Task<TResult>> selectorAsync)
{
// using local variable for counter is easier than src.Scan(...)
var counter = 0;
var streamOfTasks =
from source in src
from result in Observable.FromAsync(async () => new
{
Index = Interlocked.Increment(ref counter) - 1,
Result = await selectorAsync(source)
})
select result;
// buffer the results coming out of order
return Observable.Create<TResult>(observer =>
{
var index = 0;
var buffer = new Dictionary<int, TResult>();
return streamOfTasks.Subscribe(item =>
{
buffer.Add(item.Index, item.Result);
TResult result;
while (buffer.TryGetValue(index, out result))
{
buffer.Remove(index);
observer.OnNext(result);
index++;
}
});
});
}
我对我的解决方案不是特别满意,因为它对我来说太复杂了,但至少它不需要任何外部依赖项。我在这里使用一个简单的字典来缓冲和重新排序任务结果,因为订阅者 need not to be thread-safe (订阅不需要同时调用)。
欢迎提出任何意见或建议。我仍然希望找到无需自定义缓冲扩展方法的原生 RX 方式。
关于c# - 将 IObservable<Task<T>> 解包为 IObservable<T> 并保留顺序,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/43314307/