如何組合多個IObservable <bool>以形成複合bool訂閱值

Question

我想爲WPF創建一種基於Rx的ICommand。我想要做的是能夠通過組合任意數量的IObservable流來控制CanExecute。

我想工作的方式是我想使用所有謂詞的最新組合邏輯與值，並使用它來控制bool ICommand.CanExecute（object parameter）方法實現。我不想等所有的謂詞產生，它應該使用任何一個源謂詞流OnNexts（產生一個值）。

我試圖弄清楚如何連接它，以致任何謂詞都應該導致ICommand.CanExecute產生一個新的值。

忘記實際的ICommand實現一分鐘（因爲我的問題是更多的東西RX端），任何人都可以建議我怎麼能連線了一堆謂詞（的IObservable <布爾>），其產量每當底層的東西創造他們的流改變，但也將一致工作，以創建一個我可以訂閱的整體結束價值。結束值將是當前謂詞流值的邏輯與。

我希望我不需要訂閱所有的謂詞流，並且希望在RX內有一個很酷的運算符，我可能忽略了它。

我知道我可以合併流，但這並不完全是我之後的行爲，因爲這僅僅是來自已合併的輸入流的最新值，我也知道我可以CombineLatest，但這不是非常正確，因爲只有當所有的合併流產生價值時它纔會產生。

我想要的是要組合的流，所以任何更改都會通知訂閱者，但我也想知道什麼邏輯與操作組合的謂詞IObservable流現在是正確的，這樣我就可以驅動ICommand.CanExecute從這個整體組合值。

我希望這是有道理的。

下面是一些骨架代碼（我已經留下了一些註釋掉的代碼，這說明背後說我的Rx指揮理念的思考，因爲它可能有助於說明什麼，我想工作）

public class ViewModel : INPCBase 
{ 
    private string title; 
    private bool hasStuff; 

    public ViewModel() 
    { 
     //Initialise some command with 1st predicate, and 
     // initial CanExecute value 
     //SomeCommand = new ReactiveCommand(
     // this.ObserveProperty(x => x.Title) 
     //  .Select(x => !string.IsNullOrEmpty(x)), false); 
     //SomeCommand.AddPredicate(this.ObserveProperty(x => x.HasStuff)); 
     //SomeCommand.CommandExecutedStream.Subscribe(x => 
     // { 
     //  MessageBox.Show("Command Running"); 
     // }); 

     IObservable<bool> obsPred = this.ObserveProperty(x => x.Title) 
      .Select(x => !string.IsNullOrEmpty(x)) 
      .StartWith(!string.IsNullOrEmpty(this.Title)); 
     IObservable<bool> obsPred2 = this.ObserveProperty(x => 
      x.HasStuff).StartWith(this.HasStuff); 


     obsPred.Merge(obsPred2).Subscribe(x => 
      { 
       //How do I get this to fire whenever obsPred OR 
       //obsPred2 fire OnNext, but also get a combined value (bool) 
       //of the AND of obsPred & obsPred2 (bearing in mind I may 
       //want more than 2 predicates, it should cope with any number of 
       //IObservable<bool> predicates 
      }); 
    } 

    public string Title 
    { 
     get 
     { 
      return this.title; 
     } 
     set 
     { 
      RaiseAndSetIfChanged(ref this.title, value,() => Title); 
     } 
    } 


    public bool HasStuff 
    { 
     get 
     { 
      return this.hasStuff; 
     } 
     set 
     { 
      RaiseAndSetIfChanged(ref this.hasStuff, value,() => HasStuff); 
     } 
    } 

} 

Answer 1

您正在尋找對於CombineLatest操作

ISubject<bool> obsPred = new BehaviorSubject<bool>(false); 
ISubject<bool> obsPred2 = new BehaviorSubject<bool>(false); 

Observable.CombineLatest(obsPred, obsPred2, (a, b)=>a&&b) 
     .DistinctUntilChanged() 
     .Dump(); 

obsPred.OnNext(true); 
obsPred2.OnNext(true); 
obsPred2.OnNext(true); 
obsPred.OnNext(true); 

obsPred.OnNext(false); 

這將輸出

False 
True 
False 

使用DistinctUntilChanged()將停止重複連續的值被返回。

顯然換掉BehaviorSubject s爲您的財產observables。

Answer 2

好了，所以這就是我設法得到這個工作

public interface IReactiveCommand : ICommand 
{ 
    IObservable<object> CommandExecutedStream { get; } 
    IObservable<Exception> CommandExeceptionsStream { get; } 
    void AddPredicate(IObservable<bool> predicate); 
} 

然後實際的命令執行

public class ReactiveCommand : IReactiveCommand, IDisposable 
{ 
    private Subject<object> commandExecutedSubject = new Subject<object>(); 
    private Subject<Exception> commandExeceptionsSubjectStream = new Subject<Exception>(); 
    private List<IObservable<bool>> predicates = new List<IObservable<bool>>(); 
    private IObservable<bool> canExecuteObs; 
    private bool canExecuteLatest = true; 
    private CompositeDisposable disposables = new CompositeDisposable(); 

    public ReactiveCommand(IObservable<bool> initPredicate, bool initialCondition) 
    { 
     if (initPredicate != null) 
     { 
      canExecuteObs = initPredicate; 
      SetupSubscriptions(); 
     } 
     RaiseCanExecute(initialCondition); 
    } 


    private void RaiseCanExecute(bool value) 
    { 
     canExecuteLatest = value; 
     this.raiseCanExecuteChanged(EventArgs.Empty); 
    } 


    public ReactiveCommand() 
    { 
     RaiseCanExecute(true); 
    } 


    private void SetupSubscriptions() 
    { 

     disposables = new CompositeDisposable(); 
     disposables.Add(this.canExecuteObs.Subscribe(
      //OnNext 
      x => 
      { 
       RaiseCanExecute(x); 
      }, 
      //onError 
      commandExeceptionsSubjectStream.OnNext 
     )); 
    } 



    public void AddPredicate(IObservable<bool> predicate) 
    { 
     disposables.Dispose(); 
     predicates.Add(predicate); 
     this.canExecuteObs = this.canExecuteObs.CombineLatest(predicates.Last(), (a, b) => a && b).DistinctUntilChanged(); 
     SetupSubscriptions(); 
    } 

    bool ICommand.CanExecute(object parameter) 
    { 
     return canExecuteLatest; 
    } 

    public event EventHandler CanExecuteChanged; 

    public void Execute(object parameter) 
    { 
     commandExecutedSubject.OnNext(parameter); 
    } 


    public IObservable<object> CommandExecutedStream 
    { 
     get { return this.commandExecutedSubject.AsObservable(); } 
    } 

    public IObservable<Exception> CommandExeceptionsStream 
    { 
     get { return this.commandExeceptionsSubjectStream.AsObservable(); } 
    } 


    protected virtual void raiseCanExecuteChanged(EventArgs e) 
    { 
     var handler = this.CanExecuteChanged; 

     if (handler != null) 
     { 
      handler(this, e); 
     } 
    } 

    public void Dispose() 
    { 
     disposables.Dispose(); 
    } 
} 

當我使用下面的幫助

public static class ObservableExtensions 
{ 
    public static IObservable<TValue> ObserveProperty<T, TValue>(
     this T source, 
     Expression<Func<T, TValue>> propertyExpression 
    ) 
     where T : INotifyPropertyChanged 
    { 
     return source.ObserveProperty(propertyExpression, false); 
    } 

    public static IObservable<TValue> ObserveProperty<T, TValue>(
     this T source, 
     Expression<Func<T, TValue>> propertyExpression, 
     bool observeInitialValue 
    ) 
     where T : INotifyPropertyChanged 
    { 
     var memberExpression = (MemberExpression)propertyExpression.Body; 

     var getter = propertyExpression.Compile(); 

     var observable = Observable 
      .FromEvent<PropertyChangedEventHandler, PropertyChangedEventArgs>(
       h => new PropertyChangedEventHandler(h), 
       h => source.PropertyChanged += h, 
       h => source.PropertyChanged -= h) 
      .Where(x => x.EventArgs.PropertyName == memberExpression.Member.Name) 
      .Select(_ => getter(source)); 

     if (observeInitialValue) 
      return observable.Merge(Observable.Return(getter(source))); 

     return observable; 
    } 


    public static IObservable<string> ObservePropertyChanged<T>(this T source) 
     where T : INotifyPropertyChanged 
    { 
     var observable = Observable 
      .FromEvent<PropertyChangedEventHandler, PropertyChangedEventArgs>(
       h => new PropertyChangedEventHandler(h), 
       h => source.PropertyChanged += h, 
       h => source.PropertyChanged -= h) 
      .Select(x => x.EventArgs.PropertyName); 

     return observable; 
    } 
} 

這裏是如何接線的例子

下面是一個例子視圖模型

public class ViewModel : INPCBase 
{ 
    private string title; 
    private bool hasStuff; 

    public ViewModel() 
    { 
     IObservable<bool> initPredicate = this.ObserveProperty(x => x.Title).Select(x => !string.IsNullOrEmpty(x)).StartWith(!string.IsNullOrEmpty(this.Title)); 
     IObservable<bool> predicate = this.ObserveProperty(x => x.HasStuff).StartWith(this.HasStuff); 
     SomeCommand = new ReactiveCommand(initPredicate, false); 
     SomeCommand.AddPredicate(predicate); 
     SomeCommand.CommandExecutedStream.Subscribe(x => 
      { 
       MessageBox.Show("Command Running"); 
      }); 
    } 

    public ReactiveCommand SomeCommand { get; set; } 



    public string Title 
    { 
     get 
     { 
      return this.title; 
     } 
     set 
     { 
      RaiseAndSetIfChanged(ref this.title, value,() => Title); 
     } 
    } 


    public bool HasStuff 
    { 
     get 
     { 
      return this.hasStuff; 
     } 
     set 
     { 
      RaiseAndSetIfChanged(ref this.hasStuff, value,() => HasStuff); 
     } 
    } 

} 

下面是一個例子查看

<Window x:Class="RxCommand.MainWindow" 
     xmlns="http://schemas.microsoft.com/winfx/2006/xaml/presentation" 
     xmlns:x="http://schemas.microsoft.com/winfx/2006/xaml" 
     Title="MainWindow" Height="350" Width="525"> 
    <Grid> 
     <StackPanel Orientation="Horizontal" Height="60" VerticalAlignment="Top"> 
      <CheckBox IsChecked="{Binding HasStuff, Mode=TwoWay}" Margin="10"></CheckBox> 
      <TextBox Text="{Binding Title, Mode=TwoWay, UpdateSourceTrigger=PropertyChanged}" Width="150" Margin="10"></TextBox> 
      <Button Command="{Binding SomeCommand}" Width="150" Margin="10"></Button> 


     </StackPanel> 
    </Grid> 
</Window>