Merge pull request #645 from reactiveui/command-improvements

Finish up ReactiveCommand

Author: Paul Betts

ReactiveUI/Legacy/ReactiveCommand.cs

@@ -0,0 +1,341 @@
+using System;
+using System.Collections.Generic;
+using System.Diagnostics.Contracts;
+using System.Linq;
+using System.Reactive;
+using System.Reactive.Concurrency;
+using System.Reactive.Threading.Tasks;
+using System.Reactive.Linq;
+using System.Reactive.Subjects;
+using System.Text;
+using System.Threading;
+using System.Threading.Tasks;
+using System.Windows.Input;
+using System.Linq.Expressions;
+using ReactiveUI;
+
+using LegacyRxCmd = ReactiveUI.Legacy.ReactiveCommand;
+
+namespace ReactiveUI.Legacy
+{
+ /// <summary>
+ /// ReactiveCommand is the default Command implementation in ReactiveUI, which
+ /// conforms to the spec described in IReactiveCommand. 
+ /// </summary>
+ public class ReactiveCommand : IReactiveCommand, IObservable<object>
+ {
+ IDisposable innerDisp;
+
+ readonly Subject<bool> inflight = new Subject<bool>();
+ readonly ScheduledSubject<Exception> exceptions;
+ readonly Subject<object> executed = new Subject<object>();
+ readonly IScheduler defaultScheduler;
+
+ public ReactiveCommand() : this(null, false, null) { }
+ public ReactiveCommand(IObservable<bool> canExecute, bool initialCondition = true) : this(canExecute, false, null, initialCondition) { }
+
+ public ReactiveCommand(IObservable<bool> canExecute, bool allowsConcurrentExecution, IScheduler scheduler, bool initialCondition = true)
+ {
+ canExecute = canExecute ?? Observable.Return(true);
+ defaultScheduler = scheduler ?? RxApp.MainThreadScheduler;
+ AllowsConcurrentExecution = allowsConcurrentExecution;
+
+ canExecute = canExecute.Catch<bool, Exception>(ex => {
+ exceptions.OnNext(ex);
+ return Observable.Empty<bool>();
+ });
+
+ ThrownExceptions = exceptions = new ScheduledSubject<Exception>(defaultScheduler, RxApp.DefaultExceptionHandler);
+
+ var isExecuting = inflight
+ .Scan(0, (acc, x) => acc + (x ? 1 : -1))
+ .Select(x => x > 0)
+ .Publish(false)
+ .PermaRef()
+ .DistinctUntilChanged();
+
+ IsExecuting = isExecuting.ObserveOn(defaultScheduler);
+
+ var isBusy = allowsConcurrentExecution ? Observable.Return(false) : isExecuting;
+ var canExecuteAndNotBusy = Observable.CombineLatest(canExecute, isBusy, (ce, b) => ce && !b);
+
+ var canExecuteObs = canExecuteAndNotBusy
+ .Publish(initialCondition)
+ .RefCount();
+
+ CanExecuteObservable = canExecuteObs
+ .DistinctUntilChanged()
+ .ObserveOn(defaultScheduler);
+
+ innerDisp = canExecuteObs.Subscribe(x => {
+ if (canExecuteLatest == x) return;
+
+ canExecuteLatest = x;
+ defaultScheduler.Schedule(() => this.raiseCanExecuteChanged(EventArgs.Empty));
+ }, exceptions.OnNext);
+ }
+
+ /// <summary>
+ /// This creates a ReactiveCommand that calls several child 
+ /// ReactiveCommands when invoked. Its CanExecute will match the
+ /// combined result of the child CanExecutes (i.e. if any child
+ /// commands cannot execute, neither can the parent)
+ /// </summary>
+ /// <param name="canExecute">An Observable that determines whether the 
+ /// parent command can execute</param>
+ /// <param name="commands">The commands to combine.</param>
+ public static LegacyRxCmd CreateCombined(IObservable<bool> canExecute, params ReactiveCommand[] commands)
+ {
+ var childrenCanExecute = commands
+ .Select(x => x.CanExecuteObservable)
+ .CombineLatest(latestCanExecute => latestCanExecute.All(x => x != false));
+
+ var canExecuteSum = Observable.CombineLatest(
+ canExecute.StartWith(true),
+ childrenCanExecute,
+ (parent, child) => parent && child);
+
+ var ret = new LegacyRxCmd(canExecuteSum);
+ ret.Subscribe(x => {
+ foreach (var cmd in commands) cmd.Execute(x);
+ });
+
+ return ret;
+ }
+
+ public static ReactiveCommand CreateCombined(params ReactiveCommand[] commands)
+ {
+ return CreateCombined(Observable.Return(true), commands);
+ }
+
+ /// <summary>
+ /// Registers an asynchronous method to be called whenever the command
+ /// is Executed. This method returns an IObservable representing the
+ /// asynchronous operation, and is allowed to OnError / should OnComplete.
+ /// </summary>
+ /// <returns>A filtered version of the Observable which is marshaled 
+ /// to the UI thread. This Observable should only report successes and
+ /// instead send OnError messages to the ThrownExceptions property.</returns>
+ /// <param name="asyncBlock">The asynchronous method to call.</param>
+ /// <typeparam name="T">The 1st type parameter.</typeparam>
+ public IObservable<T> RegisterAsync<T>(Func<object, IObservable<T>> asyncBlock)
+ {
+ var ret = executed.Select(x => {
+ return asyncBlock(x)
+ .Catch<T, Exception>(ex => {
+ exceptions.OnNext(ex);
+ return Observable.Empty<T>();
+ })
+ .Finally(() => { lock (inflight) { inflight.OnNext(false); } });
+ });
+
+ return ret
+ .Do(_ => { lock (inflight) { inflight.OnNext(true); } })
+ .Merge()
+ .ObserveOn(defaultScheduler)
+ .Publish().RefCount();
+ }
+
+ /// <summary>
+ /// Gets a value indicating whether this instance is executing. This 
+ /// Observable is guaranteed to always return a value immediately (i.e.
+ /// it is backed by a BehaviorSubject), meaning it is safe to determine
+ /// the current state of the command via IsExecuting.First()
+ /// </summary>
+ /// <value>true</value>
+ /// <c>false</c>
+ public IObservable<bool> IsExecuting { get; protected set; }
+
+ public bool AllowsConcurrentExecution { get; protected set; }
+
+ /// <summary>
+ /// Fires whenever an exception would normally terminate ReactiveUI 
+ /// internal state.
+ /// </summary>
+ /// <value>The thrown exceptions.</value>
+ public IObservable<Exception> ThrownExceptions { get; protected set; }
+
+ public IDisposable Subscribe(IObserver<object> observer)
+ {
+ return executed.Subscribe(
+ Observer.Create<object>(
+ x => marshalFailures(observer.OnNext, x),
+ ex => marshalFailures(observer.OnError, ex),
+ () => marshalFailures(observer.OnCompleted)));
+ }
+
+ bool canExecuteLatest;
+ public bool CanExecute(object parameter)
+ {
+ return canExecuteLatest;
+ }
+
+ public event EventHandler CanExecuteChanged;
+
+ public void Execute(object parameter)
+ {
+ lock(inflight) { inflight.OnNext(true); }
+ executed.OnNext(parameter);
+ lock(inflight) { inflight.OnNext(false); }
+ }
+
+ public IObservable<bool> CanExecuteObservable { get; protected set; }
+ 
+ public void Dispose()
+ {
+ var disp = Interlocked.Exchange(ref innerDisp, null);
+ if (disp != null) disp.Dispose();
+ }
+
+ void marshalFailures<T>(Action<T> block, T param)
+ {
+ try {
+ block(param);
+ } catch (Exception ex) {
+ exceptions.OnNext(ex);
+ }
+ }
+
+ void marshalFailures(Action block)
+ {
+ marshalFailures(_ => block(), Unit.Default);
+ }
+
+ protected virtual void raiseCanExecuteChanged(EventArgs e)
+ {
+ var handler = this.CanExecuteChanged;
+
+ if (handler != null) {
+ handler(this, e);
+ }
+ }
+ }
+
+ public static class ReactiveCommandMixins
+ {
+ /// <summary>
+ /// ToCommand is a convenience method for returning a new
+ /// ReactiveCommand based on an existing Observable chain.
+ /// </summary>
+ /// <param name="scheduler">The scheduler to publish events on - default
+ /// is RxApp.MainThreadScheduler.</param>
+ /// <returns>A new ReactiveCommand whose CanExecute Observable is the
+ /// current object.</returns>
+ public static ReactiveCommand ToCommand(this IObservable<bool> This, bool allowsConcurrentExecution = false, IScheduler scheduler = null)
+ {
+ return new ReactiveCommand(This, allowsConcurrentExecution, scheduler);
+ }
+
+ /// <summary>
+ /// A utility method that will pipe an Observable to an ICommand (i.e.
+ /// it will first call its CanExecute with the provided value, then if
+ /// the command can be executed, Execute() will be called)
+ /// </summary>
+ /// <param name="command">The command to be executed.</param>
+ /// <returns>An object that when disposes, disconnects the Observable
+ /// from the command.</returns>
+ public static IDisposable InvokeCommand<T>(this IObservable<T> This, ICommand command)
+ {
+ return This.ObserveOn(RxApp.MainThreadScheduler).Subscribe(x => {
+ if (!command.CanExecute(x)) {
+ return;
+ }
+ command.Execute(x);
+ });
+ }
+
+ /// <summary>
+ /// A utility method that will pipe an Observable to an ICommand (i.e.
+ /// it will first call its CanExecute with the provided value, then if
+ /// the command can be executed, Execute() will be called)
+ /// </summary>
+ /// <param name="target">The root object which has the Command.</param>
+ /// <param name="commandProperty">The expression to reference the Command.</param>
+ /// <returns>An object that when disposes, disconnects the Observable
+ /// from the command.</returns>
+ public static IDisposable InvokeCommand<T, TTarget>(this IObservable<T> This, TTarget target, Expression<Func<TTarget, ICommand>> commandProperty)
+ {
+ return This.CombineLatest(target.WhenAnyValue(commandProperty), (val, cmd) => new { val, cmd })
+ .ObserveOn(RxApp.MainThreadScheduler)
+ .Subscribe(x => {
+ if (!x.cmd.CanExecute(x.val)) {
+ return;
+ }
+
+ x.cmd.Execute(x.val);
+ });
+ }
+
+ /// <summary>
+ /// RegisterAsyncFunction registers an asynchronous method that returns a result
+ /// to be called whenever the Command's Execute method is called.
+ /// </summary>
+ /// <param name="calculationFunc">The function to be run in the
+ /// background.</param>
+ /// <param name="scheduler"></param>
+ /// <returns>An Observable that will fire on the UI thread once per
+ /// invoecation of Execute, once the async method completes. Subscribe to
+ /// this to retrieve the result of the calculationFunc.</returns>
+ public static IObservable<TResult> RegisterAsyncFunction<TResult>(this LegacyRxCmd This,
+ Func<object, TResult> calculationFunc,
+ IScheduler scheduler = null)
+ {
+ Contract.Requires(calculationFunc != null);
+
+ var asyncFunc = calculationFunc.ToAsync(scheduler ?? RxApp.TaskpoolScheduler);
+ return This.RegisterAsync(asyncFunc);
+ }
+
+ /// <summary>
+ /// RegisterAsyncAction registers an asynchronous method that runs
+ /// whenever the Command's Execute method is called and doesn't return a
+ /// result.
+ /// </summary>
+ /// <param name="calculationFunc">The function to be run in the
+ /// background.</param>
+ public static IObservable<Unit> RegisterAsyncAction(this LegacyRxCmd This, 
+ Action<object> calculationFunc,
+ IScheduler scheduler = null)
+ {
+ Contract.Requires(calculationFunc != null);
+
+ // NB: This PermaRef isn't exactly correct, but the people using
+ // this method probably are Doing It Wrong, so let's let them
+ // continue to do so.
+ return This.RegisterAsyncFunction(x => { calculationFunc(x); return new Unit(); }, scheduler)
+ .Publish().PermaRef();
+ }
+
+ /// <summary>
+ /// RegisterAsyncTask registers an TPL/Async method that runs when a 
+ /// Command gets executed and returns the result
+ /// </summary>
+ /// <returns>An Observable that will fire on the UI thread once per
+ /// invoecation of Execute, once the async method completes. Subscribe to
+ /// this to retrieve the result of the calculationFunc.</returns>
+ public static IObservable<TResult> RegisterAsyncTask<TResult>(this LegacyRxCmd This, Func<object, Task<TResult>> calculationFunc)
+ {
+ Contract.Requires(calculationFunc != null);
+ return This.RegisterAsync(x => calculationFunc(x).ToObservable());
+ }
+
+ /// <summary>
+ /// RegisterAsyncTask registers an TPL/Async method that runs when a 
+ /// Command gets executed and returns no result. 
+ /// </summary>
+ /// <param name="calculationFunc">The function to be run in the
+ /// background.</param>
+ /// <returns>An Observable that signals when the Task completes, on
+ /// the UI thread.</returns>
+ public static IObservable<Unit> RegisterAsyncTask(this LegacyRxCmd This, Func<object, Task> calculationFunc)
+ {
+ Contract.Requires(calculationFunc != null);
+
+ // NB: This PermaRef isn't exactly correct, but the people using
+ // this method probably are Doing It Wrong, so let's let them
+ // continue to do so.
+ return This.RegisterAsync(x => calculationFunc(x).ToObservable())
+ .Publish().PermaRef();
+ }
+ }
+}