如何等待 IObservable<T> 完成而不用 Reactive Extensions 阻塞 UI？

Question

我正在尝试更新一段代码，以便它模拟模式对话框，而 Reactive Extensions 感觉是一个合适的工具，但我无法让它工作。

目前，代码如下所示：

public bool ShowConfirmationDialogs(IEnumerable<ItemType> items)
{
    bool canContinue = true;

    foreach (var item in items)
    {
        Dialog dialog = new Dialog();
        dialog.Prepare(item);   // Prepares a "dialog" specific to each item

        IObservable<bool> o = Service.ShowDialog(dialog, result =>
        {
             // do stuff with result that may impact next iteration, e.g.
             canContinue = !result.Condition;
        });

        // tried using the following line to wait for observable to complete
        // but it blocks the UI thread
        o.FirstOrDefault();

        if (!canContinue)
            break;
    }

    if (!canContinue)
    {
        // do something that changes current object's state
    }

    return canContinue;
}

到目前为止，当ShowDialog 显示的“对话框”关闭时，来自 lambda 表达式的代码被用来做一些事情。对ShowDialog 的调用是非阻塞的，用于返回void。

ShowDialog 的幕后发生的事情是一个对象被添加到ObservableCollection 以便它显示在屏幕上。

我修改了ShowDialog，使其在对话框关闭时返回一个IObservable<bool>，该OnCompleted 的订阅者调用它。这行得通。我用以下代码对其进行了测试：

o.Subscribe(b => Console.WriteLine(b), () => Console.WriteLine("Completed"));

当对话框关闭时，我可以看到字符串"Completed"。我的问题是上面的行是非阻塞的，所以我可能会显示几个我不想做的对话框。

我尝试了以下方法：

o.FirstOrDefault();

假设程序会在那里等到 observable 发送一些东西或完成。程序可以阻塞，但它也会冻结 UI，这意味着我永远看不到我的对话框，我永远无法关闭，所以 observable 永远不会完成。

我使用ObserveOn 和SubscribeOn 尝试了几种变体，试图让UI 线程继续工作，但没有成功。任何想法都将不胜感激，我的主要目标是保持代码看起来连续，有点像使用 Window.ShowDialog 时。

总结一下：（并在 cmets 中回答 Chris）

问题在于ShowDialog 是非阻塞的，如上所述，预期的行为与使用Window.ShowDialog 时的行为相同。现在，我不能阻止——但循环继续，我得到几个对话框——或者我可以阻止（使用FirstOrDefault），但它也阻止了 UI，这阻止我关闭对话框以完成可观察到。

更多解释：（用于谜题）

当我调用ShowDialog 时，显示的控件是模态的——从某种意义上说，它阻止用户访问应用程序的其余部分——但对该方法的调用是非阻塞的，因此会立即继续执行。在我的示例中，由于循环，这可能会显示多个对话框。该方法是非阻塞的，因为它所做的只是将一个对象添加到集合中，而我无法更改此行为。

但是，为了使用 Rx，我做到了，ShowDialog 将返回一个 IObservable。所以现在该方法立即返回，但我有一个对象，一旦ShowDialog 的操作显示的控件关闭，我将调用任何观察者的OnCompleted。我为此使用Subject，以防万一。

我现在想要的是等待返回的IObservable 在继续之前完成，因此模拟阻塞调用。 FirstOrDefault 成功完成了等待部分，但不幸的是，它也阻塞了 UI 线程，阻止控件实际显示，从而阻止用户关闭它，从而阻止 IObservable 完成。

我知道我的想法不会太远，因为我可以通过在 x 秒后自动关闭对话框来让事情顺利进行。我现在需要的只是“等待”部分不要阻塞 UI，以便用户可以关闭控件而不是计时器。

Answer 1

我找到了解决问题的方法，所以如果你有兴趣，我会分享它。

经过一些重构，我重命名了问题中使用的服务方法并创建了一个新方法。它的界面是这样的：

public interface IDialogService
{
    /// <summary>
    /// Displays the specified dialog.
    /// </summary>
    /// <remarks>This method is non-blocking. If you need to access the return value of the dialog, you can either
    /// provide a callback method or subscribe to the <see cref="T:System.IObservable{bool?}" /> that is returned.</remarks>
    /// <param name="dialog">The dialog to display.</param>
    /// <param name="callback">The callback to be called when the dialog closes.</param>
    /// <returns>An <see cref="T:System.IObservable{bool?}" /> that broadcasts the value returned by the dialog
    /// to any observers.</returns>
    IObservable<bool?> Show(Dialog dialog, Action<bool?> callback = null);

    /// <summary>
    /// Displays the specified dialog. This method waits for the dialog to close before it returns.
    /// </summary>
    /// <remarks>This method waits for the dialog to close before it returns. If you need to show a dialog and
    /// return immediately, use <see cref="M:Show"/>.</remarks>
    /// <param name="dialog">The dialog to display.</param>
    /// <returns>The value returned by the dialog.</returns>
    bool? ShowDialog(Dialog dialog);
}

解决我问题的部分是ShowDialog的实现：

public bool? ShowDialog(Dialog dialog)
{
    // This will hold the result returned by the dialog
    bool? result = null;

    // We show a dialog using the method that returns an IObservable
    var subject = this.Show(dialog);

    // but we have to wait for it to close on another thread, otherwise we'll block the UI
    // we do this by preparing  a new DispatcherFrame that exits when we get a value
    // back from the dialog
    DispatcherFrame frame = new DispatcherFrame();

    // So start observing on a new thread. The Start method will return immediately.
    new Thread((ThreadStart)(() =>
    {
        // This line will block on the new thread until the subject sends an OnNext or an OnComplete
        result = subject.FirstOrDefault();

        // once we get the result from the dialog, we can tell the frame to stop
        frame.Continue = false;
    })).Start();

    // This gets executed immediately after Thread.Start
    // The Dispatcher will now wait for the frame to stop before continuing
    // but since we are not blocking the current frame, the UI is still responsive
    Dispatcher.PushFrame(frame);

    return result;
}

我认为 cmets 应该足以理解代码，我现在可以这样使用：

public bool? ShowConfirmationDialogs(IEnumerable<ItemType> items)
{
    bool canContinue = true;

    foreach (var item in items)
    {
        Dialog dialog = new Dialog();
        dialog.Prepare(item);   // Prepares a "dialog" specific to each item

        bool? result = Service.ShowDialog(dialog);

        canContinue = result.HasValue && result.Value;

        if (!canContinue)
            break;
    }

    if (!canContinue)
    {
        // do something that changes current object's state
    }

    return canContinue;
}

我很想听听其他用户可能有的任何 cmets 或替代方案。

Answer 2

您的代码是可观察和可枚举序列的奇怪混合。您还试图从使用 observables 的函数返回 bool，因此您强制要求阻塞操作的不良做法。

您最好尝试让一切都保持可观察性。这是最佳实践方法。

在ShowDialog 函数中也没有明确使用每个item。

这是我目前所能提供的最好的，我不知道更多。试试这个：

public IObservable<bool> ShowConfirmationDialogs(IEnumerable items)
{
    var query =
        from item in items.OfType<SOMEBASETYPE>().ToObservable()
        from result in Service.ShowDialog(item =>
        {
            // do stuff with result that may impact next iteration of foreach
        })
        select new
        {
            Item = item,
            Result = result,
        };

    return query.TakeWhile(x => x.Result == true);
}

调用代码应该在 UI 线程上观察。

如果这有帮助，请告诉我。