使用异步扩展方法时订阅 observable 未触发

Question

对于一个新项目，我正在使用 ReactiveUI。我开始真正喜欢反应式编程背后的想法，但在理解一些概念和编写惯用代码时仍然有些困难。

在此示例中，我有一个非常基本的 MainWindow，其中包含一个名为“UserId”的文本框：

<Window
    x:Class="ReactiveUiDemo.MainWindow"
    xmlns="http://schemas.microsoft.com/winfx/2006/xaml/presentation"
    xmlns:x="http://schemas.microsoft.com/winfx/2006/xaml"
    Title="MainWindow"
    Width="800"
    Height="450">
    <StackPanel>
        <TextBox x:Name="UserId" />
    </StackPanel>
</Window>

TextBox 绑定到 MainWindow 的构造函数中相应视图模型上的一个属性：

using System.Reactive.Disposables;
using System.Windows;
using ReactiveUI;

namespace ReactiveUiDemo
{
    public partial class MainWindow : Window, IViewFor<MainWindowViewModel>
    {
        public MainWindowViewModel ViewModel { get; set; }

        object IViewFor.ViewModel
        {
            get => ViewModel;
            set => ViewModel = (MainWindowViewModel)value;
        }

        public MainWindow()
        {
            InitializeComponent();
            ViewModel = new MainWindowViewModel();

            this.WhenActivated(disposables =>
            {
                this
                    .Bind(ViewModel, vm => vm.UserId, v => v.UserId.Text)
                    .DisposeWith(disposables);
            });
        }
    }
}

这背后的 ViewModel 然后观察这个属性（应用了 ReactiveUI.Fody [Reactive] 属性）。当输入 4 位数字时，它会尝试查找用户，在 UserId 1234 的情况下查找成功或失败。然后使用 MessageBox 显示此结果。

using System;
using System.Reactive;
using System.Reactive.Linq;
using System.Threading.Tasks;
using System.Windows;
using ReactiveUI;
using ReactiveUI.Fody.Helpers;

namespace ReactiveUiDemo
{
    public sealed class MainWindowViewModel : ReactiveObject
    {
        [Reactive]
        public string UserId { get; private set; }

        public MainWindowViewModel()
        {
            this
                .WhenAnyValue(t => t.UserId)
                .Where(u => (u?.Length ?? 0) == 4)
                .Where(u => int.TryParse(u, out _))
                .Select(i => int.Parse(i))
                .Select(i => GetUserName(i))
                .Match(
                    userName => DisplaySuccess(userName),
                    failure => DisplayError(failure))
                .Do(_ => UserId = string.Empty)
                .Subscribe(
                    _ => MessageBox.Show("OnNext"),
                    _ => MessageBox.Show("OnError"),
                    () => MessageBox.Show("OnCompleted"));
        }

        private enum Failure { UserNotFound }

        private Result<string, Failure> GetUserName(int userId)
        {
            if (userId == 1234)
                return "Waldo";

            return Failure.UserNotFound;
        }

        private async Task<Unit> DisplayError(Failure failure)
        {
            MessageBox.Show($"Error: {failure}.");
            await Task.CompletedTask;

            return Unit.Default;
        }

        private async Task<Unit> DisplaySuccess(string userName)
        {
            MessageBox.Show($"Found {userName}!");
            await Task.CompletedTask;

            return Unit.Default;
        }
    }
}

“结果”类（或其剥离版本）包含 TSuccess 或 TFailure：

using System;

namespace ReactiveUiDemo
{
    public sealed class Result<TSuccess, TFailure>
    {
        private readonly bool _isSuccess;
        private readonly TSuccess _success;
        private readonly TFailure _failure;

        private Result(TSuccess value)
        {
            _isSuccess = true;
            _success = value;
            _failure = default;
        }

        private Result(TFailure value)
        {
            _isSuccess = false;
            _success = default;
            _failure = value;
        }

        public TResult Match<TResult>(Func<TSuccess, TResult> successFunc, Func<TFailure, TResult> failureFunc)
            => _isSuccess ? successFunc(_success) : failureFunc(_failure);

        public static implicit operator Result<TSuccess, TFailure>(TSuccess value)
            => new Result<TSuccess, TFailure>(value);

        public static implicit operator Result<TSuccess, TFailure>(TFailure value)
            => new Result<TSuccess, TFailure>(value);
    }
}

让我头疼的扩展方法是Match方法，定义如下：

using System;
using System.Reactive.Linq;
using System.Reactive.Threading.Tasks;
using System.Threading.Tasks;

namespace ReactiveUiDemo
{
    public static class ObservableExtensions
    {
        public static IObservable<TResult> Match<TSuccess, TFailure, TResult>(
            this IObservable<Result<TSuccess, TFailure>> source,
            Func<TSuccess, Task<TResult>> success,
            Func<TFailure, Task<TResult>> failure)
            => Observable.FromAsync(async () => await source.SelectMany(result => result.Match(success, failure).ToObservable()));
    }
}

在 Match 扩展方法之前，代码按预期工作。 DisplayError 或 DisplaySuccess 正在被调用，但这就是它的结束； Do 和 Subscribe 中的操作不执行。我认为我的 Match 扩展方法存在问题，但我不知道如何解决。

顺便说一句，我想有一个更好的方法来写这个位：

                .Where(u => (u?.Length ?? 0) == 4)
                .Where(u => int.TryParse(u, out _))
                .Select(int.Parse)

我可以想象一个 TryParseInt 扩展方法，但也许不需要它？

编辑

根据@GlennWatson 的回答更新了扩展方法，它现在可以正常工作了：

public static IObservable<TResult> Match<TSuccess, TFailure, TResult>(
    this IObservable<Result<TSuccess, TFailure>> source,
    Func<TSuccess, Task<TResult>> success,
    Func<TFailure, Task<TResult>> failure)
    => source.SelectMany(r => r.Match(success, failure).ToObservable());

Answer 1

FromAsync() 方法仅用于基于任务的系统。

在使用 Observable 和后续 Linq 风格的方法时，尽量保持 Observable 形式。

在您的示例中，您正在等待 Observable 并将其包装在 FromAsync 中。 SelectMany 具有了解基于任务的操作的重载。