Add concept of "value plugins".

These are used to extensibly handle special values like `Task` and
`IObservable<>`. Previously this was baked into the expression observer
architecture with a TODO comment saying that it needs to be extensible.

src/Markup/Avalonia.Markup/Avalonia.Markup.csproj

@@ -60,6 +60,9 @@
     <Compile Include="Data\Parsers\IdentifierParser.cs" />
     <Compile Include="Data\Parsers\ExpressionParser.cs" />
     <Compile Include="Data\Parsers\Reader.cs" />
+    <Compile Include="Data\Plugins\ObservableValuePlugin.cs" />
+    <Compile Include="Data\Plugins\TaskValuePlugin.cs" />
+    <Compile Include="Data\Plugins\IValuePlugin.cs" />
     <Compile Include="Data\Plugins\PropertyAccessorBase.cs" />
     <Compile Include="Data\Plugins\PropertyError.cs" />
     <Compile Include="Data\Plugins\DataValidatiorBase.cs" />

src/Markup/Avalonia.Markup/Data/ExpressionNode.cs

@@ -17,6 +17,7 @@ namespace Avalonia.Markup.Data
         private WeakReference _target = UnsetReference;
         private IDisposable _valueSubscription;
         private IObserver<object> _observer;
+        private IDisposable _valuePluginSubscription;
 
         public ExpressionNode Next { get; set; }
 
@@ -35,6 +36,7 @@ namespace Avalonia.Markup.Data
                 {
                     _valueSubscription?.Dispose();
                     _valueSubscription = null;
+                    _valuePluginSubscription?.Dispose();
                     _target = value;
 
                     if (running)
@@ -60,6 +62,8 @@ namespace Avalonia.Markup.Data
             {
                 _valueSubscription?.Dispose();
                 _valueSubscription = null;
+                _valuePluginSubscription?.Dispose();
+                _valuePluginSubscription = null;
                 nextSubscription?.Dispose();
                 _observer = null;
             });
@@ -117,13 +121,16 @@ namespace Avalonia.Markup.Data
 
             if (notification == null)
             {
-                if (Next != null)
+                if (!HandleSpecialValue(value))
                 {
-                    Next.Target = new WeakReference(value);
-                }
-                else
-                {
-                    _observer.OnNext(value);
+                    if (Next != null)
+                    {
+                        Next.Target = new WeakReference(value);
+                    }
+                    else
+                    {
+                        _observer.OnNext(value);
+                    }
                 }
             }
             else
@@ -134,16 +141,38 @@ namespace Avalonia.Markup.Data
                 }
                 else if (notification.HasValue)
                 {
-                    if (Next != null)
+                    if (!HandleSpecialValue(notification.Value))
                     {
-                        Next.Target = new WeakReference(notification.Value);
+                        if (Next != null)
+                        {
+                            Next.Target = new WeakReference(notification.Value);
+                        }
+                        else
+                        {
+                            _observer.OnNext(value);
+                        }
                     }
-                    else
+                }
+            }
+        }
+
+        private bool HandleSpecialValue(object value)
+        {
+            if (_valuePluginSubscription == null)
+            {
+                var reference = new WeakReference(value);
+
+                foreach (var plugin in ExpressionObserver.ValueHandlers)
+                {
+                    if (plugin.Match(reference))
                     {
-                        _observer.OnNext(value);
+                        _valuePluginSubscription = plugin.Start(reference)?.Subscribe(TargetValueChanged);
+                        return true;
                     }
                 }
             }
+
+            return false;
         }
 
         private BindingNotification TargetNullNotification()

src/Markup/Avalonia.Markup/Data/ExpressionObserver.cs

@@ -36,7 +36,18 @@ namespace Avalonia.Markup.Data
             new List<IDataValidationPlugin>
             {
                 new IndeiValidationPlugin(),
-                ExceptionValidationPlugin.Instance,
+                new ExceptionValidationPlugin(),
+            };
+
+        /// <summary>
+        /// An ordered collection of value handlers that can be used to customize the handling
+        /// of certain values.
+        /// </summary>
+        public static readonly IList<IValuePlugin> ValueHandlers =
+            new List<IValuePlugin>
+            {
+                new TaskValuePlugin(),
+                new ObservableValuePlugin(),
             };
 
         private static readonly object UninitializedValue = new object();

src/Markup/Avalonia.Markup/Data/Plugins/ExceptionValidationPlugin.cs

@@ -12,11 +12,6 @@ namespace Avalonia.Markup.Data.Plugins
     /// </summary>
     public class ExceptionValidationPlugin : IDataValidationPlugin
     {
-        /// <summary>
-        /// Gets the default instance of the <see cref="ExceptionValidationPlugin"/>/
-        /// </summary>
-        public static ExceptionValidationPlugin Instance { get; } = new ExceptionValidationPlugin();
-
         /// <inheritdoc/>
         public bool Match(WeakReference reference) => true;
 

src/Markup/Avalonia.Markup/Data/Plugins/IValuePlugin.cs

@@ -0,0 +1,29 @@
+// Copyright (c) The Avalonia Project. All rights reserved.
+// Licensed under the MIT license. See licence.md file in the project root for full license information.
+
+using System;
+
+namespace Avalonia.Markup.Data.Plugins
+{
+    /// <summary>
+    /// Defines how values are observed by an <see cref="ExpressionObserver"/>.
+    /// </summary>
+    public interface IValuePlugin
+    {
+        /// <summary>
+        /// Checks whether this plugin handles the specified value.
+        /// </summary>
+        /// <param name="reference">A weak reference to the value.</param>
+        /// <returns>True if the plugin can handle the value; otherwise false.</returns>
+        bool Match(WeakReference reference);
+
+        /// <summary>
+        /// Starts producing output based on the specified value.
+        /// </summary>
+        /// <param name="reference">A weak reference to the object.</param>
+        /// <returns>
+        /// An observable that produces the output for the value.
+        /// </returns>
+        IObservable<object> Start(WeakReference reference);
+    }
+}

src/Markup/Avalonia.Markup/Data/Plugins/ObservableValuePlugin.cs

@@ -0,0 +1,44 @@
+// Copyright (c) The Avalonia Project. All rights reserved.
+// Licensed under the MIT license. See licence.md file in the project root for full license information.
+
+using System;
+using System.Reactive.Linq;
+using System.Reactive.Subjects;
+using System.Reflection;
+using System.Threading.Tasks;
+using System.Windows.Input;
+using Avalonia.Data;
+
+namespace Avalonia.Markup.Data.Plugins
+{
+    /// <summary>
+    /// Handles binding to <see cref="IObservable{T}"/>s in an <see cref="ExpressionObserver"/>.
+    /// </summary>
+    public class ObservableValuePlugin : IValuePlugin
+    {
+        /// <summary>
+        /// Checks whether this plugin handles the specified value.
+        /// </summary>
+        /// <param name="reference">A weak reference to the value.</param>
+        /// <returns>True if the plugin can handle the value; otherwise false.</returns>
+        public virtual bool Match(WeakReference reference)
+        {
+            var target = reference.Target;
+
+            // ReactiveCommand is an IObservable but we want to bind to it, not its value.
+            return target is IObservable<object> && !(target is ICommand);
+        }
+
+        /// <summary>
+        /// Starts producing output based on the specified value.
+        /// </summary>
+        /// <param name="reference">A weak reference to the object.</param>
+        /// <returns>
+        /// An observable that produces the output for the value.
+        /// </returns>
+        public virtual IObservable<object> Start(WeakReference reference)
+        {
+            return reference.Target as IObservable<object>;
+        }
+    }
+}

src/Markup/Avalonia.Markup/Data/Plugins/TaskValuePlugin.cs

@@ -0,0 +1,82 @@
+// Copyright (c) The Avalonia Project. All rights reserved.
+// Licensed under the MIT license. See licence.md file in the project root for full license information.
+
+using System;
+using System.Reactive.Concurrency;
+using System.Reactive.Linq;
+using System.Reactive.Subjects;
+using System.Reflection;
+using System.Threading.Tasks;
+using Avalonia.Data;
+
+namespace Avalonia.Markup.Data.Plugins
+{
+    /// <summary>
+    /// Handles binding to <see cref="Task"/>s in an <see cref="ExpressionObserver"/>.
+    /// </summary>
+    public class TaskValuePlugin : IValuePlugin
+    {
+        /// <summary>
+        /// Checks whether this plugin handles the specified value.
+        /// </summary>
+        /// <param name="reference">A weak reference to the value.</param>
+        /// <returns>True if the plugin can handle the value; otherwise false.</returns>
+        public virtual bool Match(WeakReference reference) => reference.Target is Task;
+
+        /// <summary>
+        /// Starts producing output based on the specified value.
+        /// </summary>
+        /// <param name="reference">A weak reference to the object.</param>
+        /// <returns>
+        /// An observable that produces the output for the value.
+        /// </returns>
+        public virtual IObservable<object> Start(WeakReference reference)
+        {
+            var task = reference.Target as Task;
+
+            if (task != null)
+            {
+                var resultProperty = task.GetType().GetTypeInfo().GetDeclaredProperty("Result");
+
+                if (resultProperty != null)
+                {
+                    switch (task.Status)
+                    {
+                        case TaskStatus.RanToCompletion:
+                        case TaskStatus.Faulted:
+                            return HandleCompleted(task);
+                        default:
+                            var subject = new Subject<object>();
+                            task.ContinueWith(
+                                    x => HandleCompleted(task).Subscribe(subject),
+                                    TaskScheduler.FromCurrentSynchronizationContext())
+                                .ConfigureAwait(false);
+                            return subject;
+                    }
+                }
+            }
+
+            return Observable.Empty<object>();
+        }
+
+        protected IObservable<object> HandleCompleted(Task task)
+        {
+            var resultProperty = task.GetType().GetTypeInfo().GetDeclaredProperty("Result");
+            
+            if (resultProperty != null)
+            {
+                switch (task.Status)
+                {
+                    case TaskStatus.RanToCompletion:
+                        return Observable.Return(resultProperty.GetValue(task));
+                    case TaskStatus.Faulted:
+                        return Observable.Return(new BindingNotification(task.Exception, BindingErrorType.Error));
+                    default:
+                        throw new AvaloniaInternalException("HandleCompleted called for non-completed Task.");
+                }
+            }
+
+            return Observable.Empty<object>();
+        }
+    }
+}

tests/Avalonia.Markup.UnitTests/Data/ExpressionObserverTests_Observable.cs

@@ -5,6 +5,7 @@ using System;
 using System.Collections.Generic;
 using System.Reactive.Linq;
 using System.Reactive.Subjects;
+using Avalonia.Data;
 using Avalonia.Markup.Data;
 using Avalonia.UnitTests;
 using Xunit;
@@ -27,7 +28,7 @@ namespace Avalonia.Markup.UnitTests.Data
                 source.OnNext("bar");
                 sync.ExecutePostedCallbacks();
 
-                Assert.Equal(new[] { AvaloniaProperty.UnsetValue, "foo", "bar" }, result);
+                Assert.Equal(new[] { "foo", "bar" }, result);
             }
         }
 
@@ -44,7 +45,47 @@ namespace Avalonia.Markup.UnitTests.Data
                 data.Next.OnNext(new Class2("foo"));
                 sync.ExecutePostedCallbacks();
 
-                Assert.Equal(new[] { AvaloniaProperty.UnsetValue, "foo" }, result);
+                Assert.Equal(new[] { "foo" }, result);
+
+                sub.Dispose();
+                Assert.Equal(0, data.PropertyChangedSubscriptionCount);
+            }
+        }
+
+        [Fact]
+        public void Should_Get_Simple_Observable_Value_With_DataValidation_Enabled()
+        {
+            using (var sync = UnitTestSynchronizationContext.Begin())
+            {
+                var source = new BehaviorSubject<string>("foo");
+                var data = new { Foo = source };
+                var target = new ExpressionObserver(data, "Foo", true);
+                var result = new List<object>();
+
+                var sub = target.Subscribe(x => result.Add(x));
+                source.OnNext("bar");
+                sync.ExecutePostedCallbacks();
+
+                // What does it mean to have data validation on an observable? Without a use-case
+                // it's hard to know what to do here so for the moment the value is returned.
+                Assert.Equal(new[] { "foo", "bar" }, result);
+            }
+        }
+
+        [Fact]
+        public void Should_Get_Property_Value_From_Observable_With_DataValidation_Enabled()
+        {
+            using (var sync = UnitTestSynchronizationContext.Begin())
+            {
+                var data = new Class1();
+                var target = new ExpressionObserver(data, "Next.Foo", true);
+                var result = new List<object>();
+
+                var sub = target.Subscribe(x => result.Add(x));
+                data.Next.OnNext(new Class2("foo"));
+                sync.ExecutePostedCallbacks();
+
+                Assert.Equal(new[] { new BindingNotification("foo") }, result);
 
                 sub.Dispose();
                 Assert.Equal(0, data.PropertyChangedSubscriptionCount);

tests/Avalonia.Markup.UnitTests/Data/ExpressionObserverTests_Task.cs

@@ -4,8 +4,8 @@
 using System;
 using System.Collections.Generic;
 using System.Reactive.Linq;
-using System.Threading;
 using System.Threading.Tasks;
+using Avalonia.Data;
 using Avalonia.Markup.Data;
 using Avalonia.UnitTests;
 using Xunit;
@@ -28,7 +28,7 @@ namespace Avalonia.Markup.UnitTests.Data
                 tcs.SetResult("foo");
                 sync.ExecutePostedCallbacks();
 
-                Assert.Equal(new object[] { AvaloniaProperty.UnsetValue, "foo" }, result.ToArray());
+                Assert.Equal(new[] { "foo" }, result);
             }
         }
 
@@ -43,7 +43,7 @@ namespace Avalonia.Markup.UnitTests.Data
 
                 var sub = target.Subscribe(x => result.Add(x));
 
-                Assert.Equal(new object[] { "foo" }, result.ToArray());
+                Assert.Equal(new[] { "foo" }, result);
             }
         }
 
@@ -61,7 +61,74 @@ namespace Avalonia.Markup.UnitTests.Data
                 tcs.SetResult(new Class2("foo"));
                 sync.ExecutePostedCallbacks();
 
-                Assert.Equal(new object[] { AvaloniaProperty.UnsetValue, "foo" }, result.ToArray());
+                Assert.Equal(new[] { "foo" }, result);
+            }
+        }
+
+        [Fact]
+        public void Should_Return_BindingNotification_Error_On_Task_Exception()
+        {
+            using (var sync = UnitTestSynchronizationContext.Begin())
+            {
+                var tcs = new TaskCompletionSource<string>();
+                var data = new { Foo = tcs.Task };
+                var target = new ExpressionObserver(data, "Foo");
+                var result = new List<object>();
+
+                var sub = target.Subscribe(x => result.Add(x));
+                tcs.SetException(new NotSupportedException());
+                sync.ExecutePostedCallbacks();
+
+                Assert.Equal(
+                    new[] 
+                    {
+                        new BindingNotification(
+                            new AggregateException(new NotSupportedException()),
+                            BindingErrorType.Error)
+                    }, 
+                    result);
+            }
+        }
+
+        [Fact]
+        public void Should_Return_BindingNotification_Error_For_Faulted_Task()
+        {
+            using (var sync = UnitTestSynchronizationContext.Begin())
+            {
+                var data = new { Foo = Task.FromException(new NotSupportedException()) };
+                var target = new ExpressionObserver(data, "Foo");
+                var result = new List<object>();
+
+                var sub = target.Subscribe(x => result.Add(x));
+
+                Assert.Equal(
+                    new[]
+                    {
+                        new BindingNotification(
+                            new AggregateException(new NotSupportedException()),
+                            BindingErrorType.Error)
+                    },
+                    result);
+            }
+        }
+
+        [Fact]
+        public void Should_Get_Simple_Task_Value_With_Data_DataValidation_Enabled()
+        {
+            using (var sync = UnitTestSynchronizationContext.Begin())
+            {
+                var tcs = new TaskCompletionSource<string>();
+                var data = new { Foo = tcs.Task };
+                var target = new ExpressionObserver(data, "Foo", true);
+                var result = new List<object>();
+
+                var sub = target.Subscribe(x => result.Add(x));
+                tcs.SetResult("foo");
+                sync.ExecutePostedCallbacks();
+
+                // What does it mean to have data validation on a Task? Without a use-case it's
+                // hard to know what to do here so for the moment the value is returned.
+                Assert.Equal(new [] { "foo" }, result);
             }
         }