Added a "stream" binding expression operator.

Which is now required when wanting to get the value produced by a
Task/IObservable rather than the Task/IObservable itself. Fixes #711.

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

@@ -42,6 +42,7 @@
     <Compile Include="..\..\Shared\SharedAssemblyInfo.cs">
       <Link>Properties\SharedAssemblyInfo.cs</Link>
     </Compile>
+    <Compile Include="Data\StreamNode.cs" />
     <Compile Include="Data\MarkupBindingChainNullException.cs" />
     <Compile Include="Data\CommonPropertyNames.cs" />
     <Compile Include="Data\EmptyExpressionNode.cs" />

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

@@ -17,7 +17,6 @@ namespace Avalonia.Markup.Data
         private WeakReference _target = UnsetReference;
         private IDisposable _valueSubscription;
         private IObserver<object> _observer;
-        private IDisposable _valuePluginSubscription;
 
         public abstract string Description { get; }
         public ExpressionNode Next { get; set; }
@@ -37,7 +36,6 @@ namespace Avalonia.Markup.Data
                 {
                     _valueSubscription?.Dispose();
                     _valueSubscription = null;
-                    _valuePluginSubscription?.Dispose();
                     _target = value;
 
                     if (running)
@@ -63,8 +61,6 @@ namespace Avalonia.Markup.Data
             {
                 _valueSubscription?.Dispose();
                 _valueSubscription = null;
-                _valuePluginSubscription?.Dispose();
-                _valuePluginSubscription = null;
                 nextSubscription?.Dispose();
                 _observer = null;
             });
@@ -115,25 +111,22 @@ namespace Avalonia.Markup.Data
                 source = StartListeningCore(_target);
             }
 
-            return source.Subscribe(TargetValueChanged);
+            return source.Subscribe(ValueChanged);
         }
 
-        private void TargetValueChanged(object value)
+        private void ValueChanged(object value)
         {
             var notification = value as BindingNotification;
 
             if (notification == null)
             {
-                if (!HandleSpecialValue(value))
+                if (Next != null)
                 {
-                    if (Next != null)
-                    {
-                        Next.Target = new WeakReference(value);
-                    }
-                    else
-                    {
-                        _observer.OnNext(value);
-                    }
+                    Next.Target = new WeakReference(value);
+                }
+                else
+                {
+                    _observer.OnNext(value);
                 }
             }
             else
@@ -144,38 +137,16 @@ namespace Avalonia.Markup.Data
                 }
                 else if (notification.HasValue)
                 {
-                    if (!HandleSpecialValue(notification.Value))
+                    if (Next != null)
                     {
-                        if (Next != null)
-                        {
-                            Next.Target = new WeakReference(notification.Value);
-                        }
-                        else
-                        {
-                            _observer.OnNext(value);
-                        }
+                        Next.Target = new WeakReference(notification.Value);
                     }
-                }
-            }
-        }
-
-        private bool HandleSpecialValue(object value)
-        {
-            if (_valuePluginSubscription == null)
-            {
-                var reference = new WeakReference(value);
-
-                foreach (var plugin in ExpressionObserver.ValueHandlers)
-                {
-                    if (plugin.Match(reference))
+                    else
                     {
-                        _valuePluginSubscription = plugin.Start(reference)?.Subscribe(TargetValueChanged);
-                        return true;
+                        _observer.OnNext(value);
                     }
                 }
             }
-
-            return false;
         }
 
         private BindingNotification TargetNullNotification()

src/Markup/Avalonia.Markup/Data/Parsers/ExpressionParser.cs

@@ -87,6 +87,11 @@ namespace Avalonia.Markup.Data.Parsers
             {
                 return State.BeforeMember;
             }
+            else if (ParseStreamOperator(r))
+            {
+                nodes.Add(new StreamNode());
+                return State.AfterMember;
+            }
             else
             {
                 var args = ArgumentListParser.Parse(r, '[', ']');
@@ -161,6 +166,11 @@ namespace Avalonia.Markup.Data.Parsers
             return !r.End && r.TakeIf('(');
         }
 
+        private static bool ParseStreamOperator(Reader r)
+        {
+            return !r.End && r.TakeIf('^');
+        }
+
         private enum State
         {
             Start,

src/Markup/Avalonia.Markup/Data/StreamNode.cs

@@ -0,0 +1,31 @@
+// 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.Globalization;
+using Avalonia.Data;
+using System.Reactive.Linq;
+
+namespace Avalonia.Markup.Data
+{
+    internal class StreamNode : ExpressionNode
+    {
+        public override string Description => "^";
+
+        protected override IObservable<object> StartListeningCore(WeakReference reference)
+        {
+            foreach (var plugin in ExpressionObserver.ValueHandlers)
+            {
+                if (plugin.Match(reference))
+                {
+                    return plugin.Start(reference);
+                }
+            }
+
+            // TODO: Improve error.
+            return Observable.Return(new BindingNotification(
+                new InvalidCastException("Value could not be streamed."),
+                BindingErrorType.Error));
+        }
+    }
+}

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

@@ -15,7 +15,7 @@ namespace Avalonia.Markup.UnitTests.Data
     public class ExpressionObserverTests_Observable
     {
         [Fact]
-        public void Should_Get_Simple_Observable_Value()
+        public void Should_Not_Get_Observable_Value_Without_Modifier_Char()
         {
             using (var sync = UnitTestSynchronizationContext.Begin())
             {
@@ -28,6 +28,24 @@ namespace Avalonia.Markup.UnitTests.Data
                 source.OnNext("bar");
                 sync.ExecutePostedCallbacks();
 
+                Assert.Equal(new[] { source }, result);
+            }
+        }
+
+        [Fact]
+        public void Should_Get_Simple_Observable_Value()
+        {
+            using (var sync = UnitTestSynchronizationContext.Begin())
+            {
+                var source = new BehaviorSubject<string>("foo");
+                var data = new { Foo = source };
+                var target = new ExpressionObserver(data, "Foo^");
+                var result = new List<object>();
+
+                var sub = target.Subscribe(x => result.Add(x));
+                source.OnNext("bar");
+                sync.ExecutePostedCallbacks();
+
                 Assert.Equal(new[] { "foo", "bar" }, result);
             }
         }
@@ -38,7 +56,7 @@ namespace Avalonia.Markup.UnitTests.Data
             using (var sync = UnitTestSynchronizationContext.Begin())
             {
                 var data = new Class1();
-                var target = new ExpressionObserver(data, "Next.Foo");
+                var target = new ExpressionObserver(data, "Next^.Foo");
                 var result = new List<object>();
 
                 var sub = target.Subscribe(x => result.Add(x));
@@ -59,7 +77,7 @@ namespace Avalonia.Markup.UnitTests.Data
             {
                 var source = new BehaviorSubject<string>("foo");
                 var data = new { Foo = source };
-                var target = new ExpressionObserver(data, "Foo", true);
+                var target = new ExpressionObserver(data, "Foo^", true);
                 var result = new List<object>();
 
                 var sub = target.Subscribe(x => result.Add(x));
@@ -78,7 +96,7 @@ namespace Avalonia.Markup.UnitTests.Data
             using (var sync = UnitTestSynchronizationContext.Begin())
             {
                 var data = new Class1();
-                var target = new ExpressionObserver(data, "Next.Foo", true);
+                var target = new ExpressionObserver(data, "Next^.Foo", true);
                 var result = new List<object>();
 
                 var sub = target.Subscribe(x => result.Add(x));

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

@@ -15,7 +15,7 @@ namespace Avalonia.Markup.UnitTests.Data
     public class ExpressionObserverTests_Task
     {
         [Fact]
-        public void Should_Get_Simple_Task_Value()
+        public void Should_Not_Get_Task_Result_Without_Modifier_Char()
         {
             using (var sync = UnitTestSynchronizationContext.Begin())
             {
@@ -28,7 +28,8 @@ namespace Avalonia.Markup.UnitTests.Data
                 tcs.SetResult("foo");
                 sync.ExecutePostedCallbacks();
 
-                Assert.Equal(new[] { "foo" }, result);
+                Assert.Equal(1, result.Count);
+                Assert.IsType<Task<string>>(result[0]);
             }
         }
 
@@ -38,7 +39,7 @@ namespace Avalonia.Markup.UnitTests.Data
             using (var sync = UnitTestSynchronizationContext.Begin())
             {
                 var data = new { Foo = Task.FromResult("foo") };
-                var target = new ExpressionObserver(data, "Foo");
+                var target = new ExpressionObserver(data, "Foo^");
                 var result = new List<object>();
 
                 var sub = target.Subscribe(x => result.Add(x));
@@ -54,7 +55,7 @@ namespace Avalonia.Markup.UnitTests.Data
             {
                 var tcs = new TaskCompletionSource<Class2>();
                 var data = new Class1(tcs.Task);
-                var target = new ExpressionObserver(data, "Next.Foo");
+                var target = new ExpressionObserver(data, "Next^.Foo");
                 var result = new List<object>();
 
                 var sub = target.Subscribe(x => result.Add(x));
@@ -72,7 +73,7 @@ namespace Avalonia.Markup.UnitTests.Data
             {
                 var tcs = new TaskCompletionSource<string>();
                 var data = new { Foo = tcs.Task };
-                var target = new ExpressionObserver(data, "Foo");
+                var target = new ExpressionObserver(data, "Foo^");
                 var result = new List<object>();
 
                 var sub = target.Subscribe(x => result.Add(x));
@@ -96,7 +97,7 @@ namespace Avalonia.Markup.UnitTests.Data
             using (var sync = UnitTestSynchronizationContext.Begin())
             {
                 var data = new { Foo = TaskFromException(new NotSupportedException()) };
-                var target = new ExpressionObserver(data, "Foo");
+                var target = new ExpressionObserver(data, "Foo^");
                 var result = new List<object>();
 
                 var sub = target.Subscribe(x => result.Add(x));
@@ -119,7 +120,7 @@ namespace Avalonia.Markup.UnitTests.Data
             {
                 var tcs = new TaskCompletionSource<string>();
                 var data = new { Foo = tcs.Task };
-                var target = new ExpressionObserver(data, "Foo", true);
+                var target = new ExpressionObserver(data, "Foo^", true);
                 var result = new List<object>();
 
                 var sub = target.Subscribe(x => result.Add(x));