RX：如何將可觀察的最新值傳遞給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傳遞的空行爲）。

UPD：

執行順序：

1. IdStream創建
2. 命令創建
3. documentStream過程DocumentOpened事件（得到一些Id值 - 我檢查）
4. saveCommand執行

我該如何實現它？

UPD 2：我還需要等待命令體內的方法（例如，SaveAsync）。

Answer 1

這是否適合您？重播將保留已發佈的最新值。執行該命令後，它將獲取最新值，Take（1）之後取消訂閱，因爲您只需要一個值，然後將其推送至保存;

[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()電話中這樣做。