c# - RX : how to pass the latest value from observable into 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 执行

我怎样才能实现它？

更新 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");
    }