IObservable<T>.ToTask<T> 方法返回等待激活的任务答案

【问题标题】：IObservable<T>.ToTask<T> method returns Task awaiting activationIObservable<T>.ToTask<T> 方法返回等待激活的任务
【发布时间】：2014-06-08 21:58:44
【问题描述】：

为什么task 永远等待？：

var task = Observable
    .FromEventPattern<MessageResponseEventArgs>(communicator, "PushMessageRecieved")
    .Where(i => i.EventArgs.GetRequestFromReceivedMessage().Name == requestName)
    .Select(i => i.EventArgs)
    .RunAsync(System.Threading.CancellationToken.None)
    .ToTask();

task.Wait();

我知道"PushMessageRecieved" 被解雇了；我可以在 Select lambda 上设置一个断点并点击它。但是task.Wait() 永远不动。

更好的更新： FirstAsync() 是我正在寻找的：

    public static Task<MessageResponseEventArgs> HandlePushMessageRecievedAsync(this ICommunicator communicator, RequestName requestName)
    {
        if (communicator == null) return Task.FromResult<MessageResponseEventArgs>(null);

        var observable = GetCommunicatorObservableForPushMessageReceived(communicator);
        return observable
            .Where(i => i.GetRequestFromReceivedMessage().Name == requestName)
            .Select(i => i)
            .FirstAsync()
            .ToTask();
    }

GetCommunicatorObservableForPushMessageReceived() 在哪里：

    static IObservable<MessageResponseEventArgs> GetCommunicatorObservableForPushMessageReceived(ICommunicator communicator)
    {
        if (communicatorObservableForPushMessageReceived == null)
        {
            communicatorObservableForPushMessageReceived = Observable
                .FromEventPattern<MessageResponseEventArgs>(communicator, "PushMessageRecieved")
                .Where(i => !IsPreviousMessage(i.EventArgs.GetRequestFromReceivedMessage().EventId))
                .Select(i => i.EventArgs);
        }

        return communicatorObservableForPushMessageReceived;
    }

更新：这有点可怕（但它有效）：

public static Task<MessageResponseEventArgs> HandlePushMessageRecievedAsync(this ICommunicator communicator, RequestName requestName)
{
    if (communicator == null) return Task.FromResult<MessageResponseEventArgs>(null);

    var completionSource = new TaskCompletionSource<MessageResponseEventArgs>();

    Observable
        .FromEventPattern<MessageResponseEventArgs>(communicator, "PushMessageRecieved")
        .Where(i => i.EventArgs.GetRequestFromReceivedMessage().Name == requestName)
        .Select(i => i.EventArgs)
        .ToEvent().OnNext += (args) =>
        {
            if (args.Response.Errors != null && args.Response.Errors.Any())
            {
                completionSource.TrySetException(args.Response.Errors.Select(j => new Exception(j.ErrorMessage)));
            }
            else
            {
                completionSource.TrySetResult(args);
            }
        };

    return completionSource.Task;
}

【问题讨论】：

Name 属性是否匹配 requestName？
@Lee：如果它达到了Select lambda，正如 OP 所说，那么是的，它必须匹配 Where 子句。
ToTask 不应该在 RunAsync 之前发生吗？
@travis-j RunAsync 返回AsyncSubject<TSource>
FWIW @JamesWorld 提出了很好的观点。 ToTask 仅适用于将新的 Rx 代码与现有的基于任务的代码集成的目的。如果您不尝试将新的 Rx 事件与现有的任务代码集成，最好不要调用ToTask。您仍然可以使用 Rx（等待、等待、延续）更好地完成所有类似任务的事情。

标签： c# task-parallel-library system.reactive

【解决方案1】：

RunAsync 和 ToTask 在 observable 中产生最后一个值。因此在可观察的完成之前不会产生任何值。但是使用 FromEventPattern 创建的 observables 通常不会完成。您需要强制他们完成类似Take 或Until 的内容。

我还要注意，RunAsync 和 ToTask 本质上是多余的，不需要两者都做。

在你的情况下，我假设你真的对通过你的过滤器的 first 值感兴趣：

var task = Observable
    .FromEventPattern<MessageResponseEventArgs>(communicator, "PushMessageRecieved")
    .FirstAsync(i => i.EventArgs.GetRequestFromReceivedMessage().Name == requestName)
    .Select(i => i.EventArgs)
    .ToTask();

task.Wait();

【讨论】：

非常感谢！这有效：Select(i => i).Take(1).ToTask()
这样更好：Select(i => i).FirstAsync().ToTask()
你为什么一直加Select(i=>i)？这没有任何意义。
@Brandon 这不是我的代码，但我经常在调试时将select(x => x) 添加到我的代码中。它让我可以添加断点，并查看显示的值。在此示例中，Task 永远不会返回，因此我添加了 select 以确保某些值至少达到了这个程度。这些已被删除。

【解决方案2】：

您正在观察的PushMessageRecieved 事件的处理程序需要在当前同步上下文中的 UI 线程上运行。在等待此任务时，您正在阻塞 UI 线程（代表当前上下文）。任务无法完成，因为您正在等待它，您永远无法完成等待，因为它无法运行。死锁。

您不应该同步阻塞任务，而是异步执行代码作为该任务的延续。

【讨论】：

我很欣赏你所说的，但必须将此建议应用于单元测试的情况：没有 UI 线程。
@rasx 首先，您应该确保单元测试中没有同步上下文；很可能有一个。如果没有那么应该有，否则你就不能正确地模拟你的真实环境。