'等待'一个可观察的

Question

我有一个正在处理的任务列表（启用驱动器、更改位置、等待停止、禁用）。

“等待”监控我想要等待的IObservable<Status>（这样我就可以将它通过ContinueWith 和其他任务线程化）。

我开始在订阅者的 OnNext 处理内执行以下任务，但这很丑陋。我现在想出的是这个扩展方法：

public static Task<T> WaitFor<T>(this IObservable<T> source, Func<T, bool> pred)
{
    var tcs = new TaskCompletionSource<T>();
    source
        .Where(pred)
        .DistinctUntilChanged()
        .Take(1)  //OnCompletes the observable, subscription will self-dispose
        .Subscribe(val => tcs.TrySetResult(val),
                    ex => tcs.TrySetException(ex),
                    () => tcs.TrySetCanceled());

    return tcs.Task;
}

（已更新 svick 建议处理OnCompleted 和OnError）

问题：

• 这是好事、坏事还是丑陋？
• 我是否错过了可以做到这一点的现有扩展？
• Where 和 DistinctUntilChanged 的顺序是否正确？ （我认为他们是）

Answer 1

对此不是 100% 确定，但通过阅读 Rx 2.0 beta 博客文章，我认为如果您可以使用 async/await，您可以“返回 await source.FirstAsync(pred)”或不使用 async，“return source.FirstAsync(pred).ToTask()"

http://blogs.msdn.com/b/rxteam/archive/2012/03/12/reactive-extensions-v2-0-beta-available-now.aspx

Answer 2

至少我会把这个扩展方法改成这样：

public static Task<T> WaitFor<T>(this IObservable<T> source, Func<T, bool> pred)
{
    return
        source
            .Where(pred)
            .DistinctUntilChanged()
            .Take(1)
            .ToTask();
}

使用.ToTask() 比引入TaskCompletionSource 要好得多。您需要引用 System.Reactive.Threading.Tasks 命名空间来获取 .ToTask() 扩展方法。

此外，DistinctUntilChanged 在此代码中是多余的。您只能获得一个值，因此默认情况下它必须是不同的。

现在，我的下一个建议可能有点争议。这个扩展是个坏主意，因为它隐藏了正在发生的事情的真正语义。

如果我有这两个代码片段：

var t = xs.WaitFor(x => x > 10);

或者：

var t = xs.Where(x => x > 10).Take(1).ToTask();

我通常更喜欢第二个片段，因为它清楚地向我展示了正在发生的事情 - 我不需要记住 WaitFor 的语义。

除非您使 WaitFor 的名称更具描述性 - 也许是 TakeOneAsTaskWhere - 那么您将在使用它的代码中明确使用运算符并使代码更难管理。

以下内容不是更容易记住语义吗？

var t = xs.TakeOneAsTaskWhere(x => x > 10);

对我来说，底线是 Rx 运算符是要组合的，而不是封装的，但如果你要封装它们，那么它们的含义必须清楚。

我希望这会有所帮助。