如何将 SelectMany 用于 Observable.FromEventPattern 序列上的异步逻辑？答案

【问题标题】：How to use SelectMany for async logic on Observable.FromEventPattern sequence?如何将 SelectMany 用于 Observable.FromEventPattern 序列上的异步逻辑？
【发布时间】：2019-06-12 19:35:19
【问题描述】：

我有一个公开以下可观察序列的类：

internal IObservable<TaskDoneEventArgs> WhenTaskDone => Observable
    .FromEventPattern<TaskDoneEventHandler, TaskDoneEventArgs>(
            handler => NiTask.Done += handler,
            handler => NiTask.Done -= handler)
    .Select(x => x.EventArgs);

每当从序列中观察到新项目时，我想执行一些异步 TPL 逻辑。据我了解，SelectMany() 是处理异步逻辑的好方法。但是，我无法正确获取语法。

以下是我的尝试，但它不会编译：

_output.WhenTaskDone
    .SelectMany(async _ => await StopDelivery())
    .Subscribe(_ => Debug.WriteLine("Delivery stopped"));

我收到以下与SelectMany() 相关的错误：

  Error CS0411: The type arguments for method 'Observable.SelectMany<TSource, TOther>(IObservable<TSource>, IObservable<TOther>)' cannot be inferred from the usage. Try specifying the type arguments explicitly. (85, 18)

StopDelivery()函数的签名如下：

internal Task StopDelivery()

我在这里错过了什么和做错了什么？

【问题讨论】：

标签： c# observable reactive-programming system.reactive

【解决方案1】：

如果您想基于返回Task 的方法将单个操作插入到管道中，以便在将原始值返回给订阅者之前完成Task，那么您需要执行这种操作其中：

.SelectMany(_ => Observable.FromAsync(() => StopDelivery()), (x, y) => x)

这是一个完整的工作示例：

void Main()
{
    WhenTaskDone
        .SelectMany(_ => Observable.FromAsync(() => StopDelivery()), (x, y) => x)
        .Subscribe(_ => Debug.WriteLine("Delivery stopped"));

    NiTask.OnDone();
}

private NiTaskClass NiTask = new NiTaskClass();

internal IObservable<TaskDoneEventArgs> WhenTaskDone =>
    Observable
        .FromEventPattern<TaskDoneEventHandler, TaskDoneEventArgs>(
            handler => NiTask.Done += handler,
            handler => NiTask.Done -= handler)
        .Select(x => x.EventArgs);


internal Task StopDelivery() => Task.Run(() => Console.WriteLine("StopDelivery"));

public delegate void TaskDoneEventHandler(object sender, TaskDoneEventArgs e);

public class TaskDoneEventArgs : EventArgs { }

public class NiTaskClass
{
    public event TaskDoneEventHandler Done;
    public void OnDone()
    {
        this.Done?.Invoke(this, new TaskDoneEventArgs());
    }
}

这是放在LINQPad 中的 - 通过 NuGet 添加 System.Reactive 后，您可以进行复制和粘贴，这应该可以正常运行。

【讨论】：