IObservable<T>.ToTask<T>方法返回等待激活的任务 [英] IObservable&lt;T&gt;.ToTask&lt;T&gt; method returns Task awaiting activation

问题描述

为什么 task 永远等待?:

Why does task await forever?:

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() 永远不会移动.

I know "PushMessageRecieved" is fired; I can set a break point on the Select lambda and hit it. But task.Wait() never moves.

更好的更新: 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;
    }

更新:有点可怕的是(但它有效):

Update: what is somewhat horrific is this (but it works):

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;
}

推荐答案

RunAsync 和 ToTask 都产生 observable 中的最后一个值.因此，在 observable 完成之前不会产生任何值.但是用 FromEventPattern 创建的 observables 通常不会完成.您需要使用诸如 Take 或 Until 之类的东西来强制它们完成.

Both RunAsync and ToTask yield the very last value in the observable. Thus no value will be produced until the observable completes. But observables created with FromEventPattern do not usually complete. You need to force them to complete with something like Take or Until.

我还要指出，RunAsync 和 ToTask 本质上是多余的，不需要两者都做.

I'll also note that RunAsync and ToTask are essentially redundant and there isn't need to do both.

就您而言，我假设您对通过过滤器的第一个值很感兴趣:

In your case, I'll assume you are really interested in the first value that makes it through your filter:

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

task.Wait();

这篇关于IObservable<T>.ToTask<T>方法返回等待激活的任务的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！