RX：如何将 observable 中的最新值传递给 ReactiveCommand

Question

我想获取IdStream 的最新值并在命令Execute 操作中使用它。

public IObservable<Option<Guid>> IdStream { get; }

IdStream = documentStream.OfType<DocumentOpened().Select(x => x.Document.Id.Some())
.Merge(documentStream.OfType<DocumentClosed().Select(x => Option<Guid>.None()));

var saveCommand = ReactiveCommand.Create(() => Save(id), CanExecute);

我曾尝试使用答案 https://stackoverflow.com/a/31168822/7779560 并得到这样的结果：

var saveCommand = ReactiveCommand.Create(() => { }, CanExecute);
saveCommand.WithLatestFrom(IdStream, (_, id) => id)
            .Subscribe(id => Save(id));

它可以工作，但在这种情况下我不能使用 IsExecuting 和 ThrownExceptions 命令的功能（它们仅适用于我在命令创建期间作为 Execute 传递的空操作）。

更新：

执行顺序：

1. IdStream正在创建
2. 命令创建
3. documentStream 处理 DocumentOpened 事件（获取一些 Id 值 - 我检查过）
4. saveCommand执行

我怎样才能实现它？

UPD 2：我还需要等待命令体内的方法（例如SaveAsync）。

Answer 1

这对你有用吗？重播将保留最新发布的值。执行该命令时，它将获取最新的值，然后 Take(1) 取消订阅，因为您只需要一个值，然后将其推送到 Save；

    [Test]
    public void ExecuteUsingLastProducedValue()
    {
        Subject<string> producer = new Subject<string>();
        IObservable<bool> CanExecute = Observable.Return(true);
        IObservable<string> IdStream = producer;
        string SaveCalledWith = String.Empty;

        Func<string, Task> SaveAsync = (id) =>
        {
            SaveCalledWith = id;
            return Task.Delay(0);
        };

        // IdStream creating
         var connectedIdStream =
            IdStream
            .Replay(1);

        connectedIdStream
            .Connect();

        //Command creating
        var command = ReactiveCommand.CreateFromObservable(() =>
        {
            return connectedIdStream
                .Take(1)
                .Do(async id =>
                {
                    await SaveAsync(id);
                });
        }
        , CanExecute);


        //Alternate way
        //Command creating
        //var command = ReactiveCommand.CreateFromObservable(() =>
        //{
        //    return connectedIdStream
        //        .Take(1)
        //        .SelectMany(id => SaveAsync(id).ToObservable());
        //}
        //, CanExecute);


        //documentStream processes DocumentOpened event (get some Id value - I checked it)
        producer.OnNext("something random");
        producer.OnNext("working");

        //At this point Save still hasen't been called so just verifiyng it's still empty
        Assert.AreEqual(String.Empty, SaveCalledWith);

        //trigger execution of command
        command.Execute(Unit.Default).Subscribe();

        //Verified Saved Called With is called
        Assert.AreEqual(SaveCalledWith, "working");
    }

Answer 2

你想用Observable.Sample

    [Fact]
    public void ExecuteUsingLastProducedValue()
    {
        Subject<string> producer = new Subject<string>();
        IObservable<bool> CanExecute = Observable.Return(true);
        IObservable<string> IdStream = producer;
        string SaveCalledWith = String.Empty;

        Func<string, Task> SaveAsync = (id) =>
        {
            SaveCalledWith = id;
            return Task.Delay(0);
        };

        // IdStream creating
        var connectedIdStream =
            IdStream
                .Replay(1);

        connectedIdStream
            .Connect();

        //Command creating
        var command = ReactiveCommand.Create(() => { } , CanExecute);
        connectedIdStream.Sample( command )
                         .Subscribe( id => SaveAsync(id) );

        //documentStream processes DocumentOpened event (get some Id value - I checked it)
        producer.OnNext("something random");
        producer.OnNext("working");

        //At this point Save still hasen't been called so just verifiyng it's still empty
        SaveCalledWith.Should().Be( String.Empty );

        //trigger execution of command
        command.Execute(Unit.Default).Subscribe();

        //Verified Saved Called With is called
        SaveCalledWith.Should().Be( "working");
    }

（我用 XUnit 重写了，因为我手头有它）

这是一个稍微简化和扩展的测试用例，您的一些代码替换为我推荐的代码。

    [Fact]
    public void ExecuteUsingLastProducedValue()
    {
        var producer = new Subject<string>();
        var canExecute = Observable.Return(true);
        var saveCalledWith = String.Empty;

        void Save(string id) => saveCalledWith = id;

        var rcommand = ReactiveCommand.Create(() => { } , canExecute);

        // When cast to ICommand ReactiveCommand has a
        // more convienient Execute method. No need
        // to Subscribe.
        var command = (ICommand) rcommand; 


        producer
            .Sample( rcommand )
            .Subscribe( Save );

        //documentStream processes DocumentOpened event (get some Id value - I checked it)
        producer.OnNext("something random");
        producer.OnNext("working");

        //At this point Save still hasen't been called so just verifiyng it's still empty
        saveCalledWith.Should().Be( String.Empty );

        //trigger execution of command
        command.Execute( Unit.Default );

        //Verified Saved Called With is called
        saveCalledWith.Should().Be( "working");

        command.Execute( Unit.Default );

        saveCalledWith.Should().Be( "working");

        producer.OnNext("cat");
        saveCalledWith.Should().Be( "working");
        command.Execute( Unit.Default );
        saveCalledWith.Should().Be( "cat");
        producer.OnNext("dog");
        saveCalledWith.Should().Be( "cat");
        command.Execute( Unit.Default );
        saveCalledWith.Should().Be( "dog");
    }

Answer 3

您可以将流的最新值存储在 ObservableAsPropertyHelper<> 属性中，并在您的命令中使用它。

您的类级别属性如下所示：

public IObservable<Option<Guid>> IdStream { get; }

private ObservableAsPropertyHelper<Option<Guid>> _currentId;
public Option<Guid> CurrentId => _currentId.Value;

你的构造函数会像这样连接起来：

IdStream.ToProperty(this, x => x.CurrentId, out _currentId);
var saveCommand = ReactiveCommand.Create(() => Save(CurrentId), CanExecute);

您可能希望为CurrentId 属性提供默认值。您可以在ToProperty() 电话中执行此操作。