Merge branch 'master' into fixes/direct2d1-image-brush-null-source

.ncrunch/Avalonia.Designer.HostApp.NetFX.v3.ncrunchproject

@@ -0,0 +1,5 @@
+<ProjectConfiguration>
+  <Settings>
+    <IgnoreThisComponentCompletely>True</IgnoreThisComponentCompletely>
+  </Settings>
+</ProjectConfiguration>

.ncrunch/BindingDemo.net461.v3.ncrunchproject

@@ -0,0 +1,5 @@
+<ProjectConfiguration>
+  <Settings>
+    <IgnoreThisComponentCompletely>True</IgnoreThisComponentCompletely>
+  </Settings>
+</ProjectConfiguration>

.ncrunch/BindingDemo.netcoreapp2.0.v3.ncrunchproject

@@ -0,0 +1,5 @@
+<ProjectConfiguration>
+  <Settings>
+    <IgnoreThisComponentCompletely>True</IgnoreThisComponentCompletely>
+  </Settings>
+</ProjectConfiguration>

.ncrunch/Previewer.v3.ncrunchproject

@@ -0,0 +1,5 @@
+<ProjectConfiguration>
+  <Settings>
+    <IgnoreThisComponentCompletely>True</IgnoreThisComponentCompletely>
+  </Settings>
+</ProjectConfiguration>

.ncrunch/RemoteDemo.v3.ncrunchproject

@@ -0,0 +1,5 @@
+<ProjectConfiguration>
+  <Settings>
+    <IgnoreThisComponentCompletely>True</IgnoreThisComponentCompletely>
+  </Settings>
+</ProjectConfiguration>

.ncrunch/RenderDemo.net461.v3.ncrunchproject

@@ -0,0 +1,5 @@
+<ProjectConfiguration>
+  <Settings>
+    <IgnoreThisComponentCompletely>True</IgnoreThisComponentCompletely>
+  </Settings>
+</ProjectConfiguration>

.ncrunch/RenderDemo.netcoreapp2.0.v3.ncrunchproject

@@ -0,0 +1,5 @@
+<ProjectConfiguration>
+  <Settings>
+    <IgnoreThisComponentCompletely>True</IgnoreThisComponentCompletely>
+  </Settings>
+</ProjectConfiguration>

.ncrunch/VirtualizationDemo.net461.v3.ncrunchproject

@@ -0,0 +1,5 @@
+<ProjectConfiguration>
+  <Settings>
+    <IgnoreThisComponentCompletely>True</IgnoreThisComponentCompletely>
+  </Settings>
+</ProjectConfiguration>

.ncrunch/VirtualizationDemo.netcoreapp2.0.v3.ncrunchproject

@@ -0,0 +1,5 @@
+<ProjectConfiguration>
+  <Settings>
+    <IgnoreThisComponentCompletely>True</IgnoreThisComponentCompletely>
+  </Settings>
+</ProjectConfiguration>

src/Avalonia.Base/AvaloniaObject.cs

@@ -10,6 +10,7 @@ using System.Reactive.Linq;
 using Avalonia.Data;
 using Avalonia.Diagnostics;
 using Avalonia.Logging;
+using Avalonia.Reactive;
 using Avalonia.Threading;
 using Avalonia.Utilities;
 
@@ -38,7 +39,7 @@ namespace Avalonia
         /// Maintains a list of direct property binding subscriptions so that the binding source
         /// doesn't get collected.
         /// </summary>
-        private List<IDisposable> _directBindings;
+        private List<DirectBindingSubscription> _directBindings;
 
         /// <summary>
         /// Event handler for <see cref="INotifyPropertyChanged"/> implementation.
@@ -359,25 +360,12 @@ namespace Avalonia
                     property, 
                     description);
 
-                IDisposable subscription = null;
-
                 if (_directBindings == null)
                 {
-                    _directBindings = new List<IDisposable>();
+                    _directBindings = new List<DirectBindingSubscription>();
                 }
 
-                subscription = source
-                    .Select(x => CastOrDefault(x, property.PropertyType))
-                    .Do(_ => { }, () => _directBindings.Remove(subscription))
-                    .Subscribe(x => SetDirectValue(property, x));
-
-                _directBindings.Add(subscription);
-
-                return Disposable.Create(() =>
-                {
-                    subscription.Dispose();
-                    _directBindings.Remove(subscription);
-                });
+                return new DirectBindingSubscription(this, property, source);
             }
             else
             {
@@ -908,5 +896,38 @@ namespace Avalonia
                 value,
                 priority);
         }
+
+        private class DirectBindingSubscription : IObserver<object>, IDisposable
+        {
+            readonly AvaloniaObject _owner;
+            readonly AvaloniaProperty _property;
+            IDisposable _subscription;
+
+            public DirectBindingSubscription(
+                AvaloniaObject owner,
+                AvaloniaProperty property,
+                IObservable<object> source)
+            {
+                _owner = owner;
+                _property = property;
+                _owner._directBindings.Add(this);
+                _subscription = source.Subscribe(this);
+            }
+
+            public void Dispose()
+            {
+                _subscription.Dispose();
+                _owner._directBindings.Remove(this);
+            }
+
+            public void OnCompleted() => Dispose();
+            public void OnError(Exception error) => Dispose();
+
+            public void OnNext(object value)
+            {
+                var castValue = CastOrDefault(value, _property.PropertyType);
+                _owner.SetDirectValue(_property, castValue);
+            }
+        }
     }
 }

src/Avalonia.Base/AvaloniaObjectExtensions.cs

@@ -36,32 +36,15 @@ namespace Avalonia
         /// An observable which fires immediately with the current value of the property on the
         /// object and subsequently each time the property value changes.
         /// </returns>
+        /// <remarks>
+        /// The subscription to <paramref name="o"/> is created using a weak reference.
+        /// </remarks>
         public static IObservable<object> GetObservable(this IAvaloniaObject o, AvaloniaProperty property)
         {
             Contract.Requires<ArgumentNullException>(o != null);
             Contract.Requires<ArgumentNullException>(property != null);
 
-            return new AvaloniaObservable<object>(
-                observer =>
-                {
-                    EventHandler<AvaloniaPropertyChangedEventArgs> handler = (s, e) =>
-                    {
-                        if (e.Property == property)
-                        {
-                            observer.OnNext(e.NewValue);
-                        }
-                    };
-
-                    observer.OnNext(o.GetValue(property));
-
-                    o.PropertyChanged += handler;
-
-                    return Disposable.Create(() =>
-                    {
-                        o.PropertyChanged -= handler;
-                    });
-                },
-                GetDescription(o, property));
+            return new AvaloniaPropertyObservable<object>(o, property);
         }
 
         /// <summary>
@@ -74,51 +57,36 @@ namespace Avalonia
         /// An observable which fires immediately with the current value of the property on the
         /// object and subsequently each time the property value changes.
         /// </returns>
+        /// <remarks>
+        /// The subscription to <paramref name="o"/> is created using a weak reference.
+        /// </remarks>
         public static IObservable<T> GetObservable<T>(this IAvaloniaObject o, AvaloniaProperty<T> property)
         {
             Contract.Requires<ArgumentNullException>(o != null);
             Contract.Requires<ArgumentNullException>(property != null);
 
-            return o.GetObservable((AvaloniaProperty)property).Cast<T>();
+            return new AvaloniaPropertyObservable<T>(o, property);
         }
 
         /// <summary>
-        /// Gets an observable for a <see cref="AvaloniaProperty"/>.
+        /// Gets an observable that listens for property changed events for an
+        /// <see cref="AvaloniaProperty"/>.
         /// </summary>
         /// <param name="o">The object.</param>
-        /// <typeparam name="T">The type of the property.</typeparam>
         /// <param name="property">The property.</param>
         /// <returns>
-        /// An observable which when subscribed pushes the old and new values of the property each
-        /// time it is changed. Note that the observable returned from this method does not fire
-        /// with the current value of the property immediately.
+        /// An observable which when subscribed pushes the property changed event args
+        /// each time a <see cref="IAvaloniaObject.PropertyChanged"/> event is raised
+        /// for the specified property.
         /// </returns>
-        public static IObservable<Tuple<T, T>> GetObservableWithHistory<T>(
+        public static IObservable<AvaloniaPropertyChangedEventArgs> GetPropertyChangedObservable(
             this IAvaloniaObject o, 
-            AvaloniaProperty<T> property)
+            AvaloniaProperty property)
         {
             Contract.Requires<ArgumentNullException>(o != null);
             Contract.Requires<ArgumentNullException>(property != null);
 
-            return new AvaloniaObservable<Tuple<T, T>>(
-                observer =>
-                {
-                    EventHandler<AvaloniaPropertyChangedEventArgs> handler = (s, e) =>
-                    {
-                        if (e.Property == property)
-                        {
-                            observer.OnNext(Tuple.Create((T)e.OldValue, (T)e.NewValue));
-                        }
-                    };
-
-                    o.PropertyChanged += handler;
-
-                    return Disposable.Create(() =>
-                    {
-                        o.PropertyChanged -= handler;
-                    });
-                },
-                GetDescription(o, property));
+            return new AvaloniaPropertyChangedObservable(o, property);
         }
 
         /// <summary>
@@ -166,23 +134,6 @@ namespace Avalonia
                 o.GetObservable(property));
         }
 
-        /// <summary>
-        /// Gets a weak observable for a <see cref="AvaloniaProperty"/>.
-        /// </summary>
-        /// <param name="o">The object.</param>
-        /// <param name="property">The property.</param>
-        /// <returns>An observable.</returns>
-        public static IObservable<object> GetWeakObservable(this IAvaloniaObject o, AvaloniaProperty property)
-        {
-            Contract.Requires<ArgumentNullException>(o != null);
-            Contract.Requires<ArgumentNullException>(property != null);
-
-            return new WeakPropertyChangedObservable(
-                new WeakReference<IAvaloniaObject>(o), 
-                property, 
-                GetDescription(o, property));
-        }
-
         /// <summary>
         /// Binds a property on an <see cref="IAvaloniaObject"/> to an <see cref="IBinding"/>.
         /// </summary>

src/Avalonia.Base/Collections/NotifyCollectionChangedExtensions.cs

@@ -2,13 +2,9 @@
 // Licensed under the MIT license. See licence.md file in the project root for full license information.
 
 using System;
-using System.Collections;
-using System.Collections.Generic;
 using System.Collections.Specialized;
-using System.Reactive;
-using System.Reactive.Disposables;
 using System.Reactive.Linq;
-using System.Reactive.Subjects;
+using Avalonia.Reactive;
 using Avalonia.Utilities;
 
 namespace Avalonia.Collections
@@ -43,9 +39,8 @@ namespace Avalonia.Collections
             Contract.Requires<ArgumentNullException>(collection != null);
             Contract.Requires<ArgumentNullException>(handler != null);
 
-            return
-                collection.GetWeakCollectionChangedObservable()
-                          .Subscribe(e => handler.Invoke(collection, e));
+            return collection.GetWeakCollectionChangedObservable()
+                .Subscribe(e => handler(collection, e));
         }
 
         /// <summary>
@@ -63,18 +58,13 @@ namespace Avalonia.Collections
             Contract.Requires<ArgumentNullException>(collection != null);
             Contract.Requires<ArgumentNullException>(handler != null);
 
-            return
-                collection.GetWeakCollectionChangedObservable()
-                          .Subscribe(handler);
+            return collection.GetWeakCollectionChangedObservable().Subscribe(handler);
         }
 
-        private class WeakCollectionChangedObservable : ObservableBase<NotifyCollectionChangedEventArgs>,
+        private class WeakCollectionChangedObservable : LightweightObservableBase<NotifyCollectionChangedEventArgs>,
             IWeakSubscriber<NotifyCollectionChangedEventArgs>
         {
             private WeakReference<INotifyCollectionChanged> _sourceReference;
-            private readonly Subject<NotifyCollectionChangedEventArgs> _changed = new Subject<NotifyCollectionChangedEventArgs>();
-
-            private int _count;
 
             public WeakCollectionChangedObservable(WeakReference<INotifyCollectionChanged> source)
             {
@@ -83,43 +73,28 @@ namespace Avalonia.Collections
 
             public void OnEvent(object sender, NotifyCollectionChangedEventArgs e)
             {
-                _changed.OnNext(e);
+                PublishNext(e);
             }
 
-            protected override IDisposable SubscribeCore(IObserver<NotifyCollectionChangedEventArgs> observer)
+            protected override void Initialize()
             {
                 if (_sourceReference.TryGetTarget(out INotifyCollectionChanged instance))
                 {
-                    if (_count++ == 0)
-                    {
-                        WeakSubscriptionManager.Subscribe(
-                            instance,
-                            nameof(instance.CollectionChanged),
-                            this);
-                    }
-
-                    return Observable.Using(() => Disposable.Create(DecrementCount), _ => _changed)
-                        .Subscribe(observer);
-                }
-                else
-                {
-                    _changed.OnCompleted();
-                    observer.OnCompleted();
-                    return Disposable.Empty;
+                    WeakSubscriptionManager.Subscribe(
+                    instance,
+                    nameof(instance.CollectionChanged),
+                    this);
                 }
             }
 
-            private void DecrementCount()
+            protected override void Deinitialize()
             {
-                if (--_count == 0)
+                if (_sourceReference.TryGetTarget(out INotifyCollectionChanged instance))
                 {
-                    if (_sourceReference.TryGetTarget(out INotifyCollectionChanged instance))
-                    {
-                        WeakSubscriptionManager.Unsubscribe(
-                            instance,
-                            nameof(instance.CollectionChanged),
-                            this);
-                    }
+                    WeakSubscriptionManager.Unsubscribe(
+                        instance,
+                        nameof(instance.CollectionChanged),
+                        this);
                 }
             }
         }

src/Avalonia.Base/Data/Core/BindingExpression.cs

@@ -7,21 +7,23 @@ using System.Reactive.Linq;
 using System.Reactive.Subjects;
 using Avalonia.Data.Converters;
 using Avalonia.Logging;
+using Avalonia.Reactive;
 using Avalonia.Utilities;
 
 namespace Avalonia.Data.Core
 {
     /// <summary>
     /// Binds to an expression on an object using a type value converter to convert the values
-    /// that are send and received.
+    /// that are sent and received.
     /// </summary>
-    public class BindingExpression : ISubject<object>, IDescription
+    public class BindingExpression : LightweightObservableBase<object>, ISubject<object>, IDescription
     {
         private readonly ExpressionObserver _inner;
         private readonly Type _targetType;
         private readonly object _fallbackValue;
         private readonly BindingPriority _priority;
-        private readonly Subject<object> _errors = new Subject<object>();
+        InnerListener _innerListener;
+        WeakReference<object> _value;
 
         /// <summary>
         /// Initializes a new instance of the <see cref="ExpressionObserver"/> class.
@@ -139,7 +141,7 @@ namespace Avalonia.Data.Core
                                 "IValueConverter should not return non-errored BindingNotification.");
                         }
 
-                        _errors.OnNext(notification);
+                        PublishNext(notification);
 
                         if (_fallbackValue != AvaloniaProperty.UnsetValue)
                         {
@@ -170,12 +172,18 @@ namespace Avalonia.Data.Core
             }
         }
 
-        /// <inheritdoc/>
-        public IDisposable Subscribe(IObserver<object> observer)
+        protected override void Initialize() => _innerListener = new InnerListener(this);
+        protected override void Deinitialize() => _innerListener.Dispose();
+
+        protected override void Subscribed(IObserver<object> observer, bool first)
         {
-            return _inner.Select(ConvertValue).Merge(_errors).Subscribe(observer);
+            if (!first && _value != null && _value.TryGetTarget(out var val) == true)
+            {
+                observer.OnNext(val);
+            }
         }
 
+        /// <inheritdoc/>
         private object ConvertValue(object value)
         {
             var notification = value as BindingNotification;
@@ -301,5 +309,28 @@ namespace Avalonia.Data.Core
 
             return a;
         }
+
+        public class InnerListener : IObserver<object>, IDisposable
+        {
+            private readonly BindingExpression _owner;
+            private readonly IDisposable _dispose;
+
+            public InnerListener(BindingExpression owner)
+            {
+                _owner = owner;
+                _dispose = owner._inner.Subscribe(this);
+            }
+
+            public void Dispose() => _dispose.Dispose();
+            public void OnCompleted() => _owner.PublishCompleted();
+            public void OnError(Exception error) => _owner.PublishError(error);
+
+            public void OnNext(object value)
+            {
+                var converted = _owner.ConvertValue(value);
+                _owner._value = new WeakReference<object>(converted);
+                _owner.PublishNext(converted);
+            }
+        }
     }
 }

src/Avalonia.Base/Data/Core/ExpressionObserver.cs

@@ -4,18 +4,19 @@
 using System;
 using System.Collections.Generic;
 using System.Reactive;
-using System.Reactive.Disposables;
 using System.Reactive.Linq;
-using System.Reactive.Subjects;
 using Avalonia.Data;
 using Avalonia.Data.Core.Plugins;
+using Avalonia.Reactive;
 
 namespace Avalonia.Data.Core
 {
     /// <summary>
     /// Observes and sets the value of an expression on an object.
     /// </summary>
-    public class ExpressionObserver : ObservableBase<object>, IDescription
+    public class ExpressionObserver : LightweightObservableBase<object>,
+        IDescription,
+        IObserver<object>
     {
         /// <summary>
         /// An ordered collection of property accessor plugins that can be used to customize
@@ -54,9 +55,10 @@ namespace Avalonia.Data.Core
 
         private static readonly object UninitializedValue = new object();
         private readonly ExpressionNode _node;
-        private readonly Subject<Unit> _finished;
-        private readonly object _root;
-        private IObservable<object> _result;
+        private IDisposable _nodeSubscription;
+        private object _root;
+        private IDisposable _rootSubscription;
+        private WeakReference<object> _value;
 
         /// <summary>
         /// Initializes a new instance of the <see cref="ExpressionObserver"/> class.
@@ -107,7 +109,6 @@ namespace Avalonia.Data.Core
             Expression = expression;
             Description = description ?? expression;
             _node = Parse(expression, enableDataValidation);
-            _finished = new Subject<Unit>();
             _root = rootObservable;
         }
 
@@ -135,8 +136,6 @@ namespace Avalonia.Data.Core
             Expression = expression;
             Description = description ?? expression;
             _node = Parse(expression, enableDataValidation);
-            _finished = new Subject<Unit>();
-
             _node.Target = new WeakReference(rootGetter());
             _root = update.Select(x => rootGetter());
         }
@@ -203,27 +202,42 @@ namespace Avalonia.Data.Core
             }
         }
 
-        /// <inheritdoc/>
-        protected override IDisposable SubscribeCore(IObserver<object> observer)
+        void IObserver<object>.OnNext(object value)
         {
-            if (_result == null)
-            {
-                var source = (IObservable<object>)_node;
+            var broken = BindingNotification.ExtractError(value) as MarkupBindingChainException;
+            broken?.Commit(Description);
+            _value = new WeakReference<object>(value);
+            PublishNext(value);
+        }
 
-                if (_finished != null)
-                {
-                    source = source.TakeUntil(_finished);
-                }
+        void IObserver<object>.OnCompleted()
+        {
+        }
 
-                _result = Observable.Using(StartRoot, _ => source)
-                    .Select(ToWeakReference)
-                    .Publish(UninitializedValue)
-                    .RefCount()
-                    .Where(x => x != UninitializedValue)
-                    .Select(Translate);
-            }
+        void IObserver<object>.OnError(Exception error)
+        {
+        }
+
+        protected override void Initialize()
+        {
+            _value = null;
+            _nodeSubscription = _node.Subscribe(this);
+            StartRoot();
+        }
+
+        protected override void Deinitialize()
+        {
+            _rootSubscription?.Dispose();
+            _nodeSubscription?.Dispose();
+            _rootSubscription = _nodeSubscription = null;
+        }
 
-            return _result.Subscribe(observer);
+        protected override void Subscribed(IObserver<object> observer, bool first)
+        {
+            if (!first && _value != null && _value.TryGetTarget(out var value))
+            {
+                observer.OnNext(value);
+            }
         }
 
         private static ExpressionNode Parse(string expression, bool enableDataValidation)
@@ -238,42 +252,19 @@ namespace Avalonia.Data.Core
             }
         }
 
-        private static object ToWeakReference(object o)
+        private void StartRoot()
         {
-            return o is BindingNotification ? o : new WeakReference(o);
-        }
-
-        private object Translate(object o)
-        {
-            if (o is WeakReference weak)
+            if (_root is IObservable<object> observable)
             {
-                return weak.Target;
+                _rootSubscription = observable.Subscribe(
+                    x => _node.Target = new WeakReference(x != AvaloniaProperty.UnsetValue ? x : null),
+                    x => PublishCompleted(),
+                    () => PublishCompleted());
             }
-            else if (BindingNotification.ExtractError(o) is MarkupBindingChainException broken)
-            {
-                broken.Commit(Description);
-            }
-
-            return o;
-        }
-
-        private IDisposable StartRoot()
-        {
-            switch (_root)
+            else
             {
-                case IObservable<object> observable:
-                    return observable.Subscribe(
-                        x => _node.Target = new WeakReference(x != AvaloniaProperty.UnsetValue ? x : null),
-                        _ => _finished.OnNext(Unit.Default),
-                        () => _finished.OnNext(Unit.Default));
-                case WeakReference weak:
-                    _node.Target = weak;
-                    break;
-                default:
-                    throw new AvaloniaInternalException("The ExpressionObserver._root member should only be either an observable or WeakReference.");
+                _node.Target = (WeakReference)_root;
             }
-
-            return Disposable.Empty;
         }
     }
 }

src/Avalonia.Base/Data/Core/Plugins/AvaloniaPropertyAccessorPlugin.cs

@@ -153,7 +153,7 @@ namespace Avalonia.Data.Core.Plugins
 
             protected override void SubscribeCore(IObserver<object> observer)
             {
-                _subscription = Instance?.GetWeakObservable(_property).Subscribe(observer);
+                _subscription = Instance?.GetObservable(_property).Subscribe(observer);
             }
         }
     }

src/Avalonia.Base/Reactive/AvaloniaObservable.cs

@@ -1,42 +0,0 @@
-// 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;
-using System.Reactive.Disposables;
-
-namespace Avalonia.Reactive
-{
-    /// <summary>
-    /// An <see cref="IObservable{T}"/> with an additional description.
-    /// </summary>
-    /// <typeparam name="T">The type of the elements in the sequence.</typeparam>
-    public class AvaloniaObservable<T> : ObservableBase<T>, IDescription
-    {
-        private readonly Func<IObserver<T>, IDisposable> _subscribe;
-
-        /// <summary>
-        /// Initializes a new instance of the <see cref="AvaloniaObservable{T}"/> class.
-        /// </summary>
-        /// <param name="subscribe">The subscribe function.</param>
-        /// <param name="description">The description of the observable.</param>
-        public AvaloniaObservable(Func<IObserver<T>, IDisposable> subscribe, string description)
-        {
-            Contract.Requires<ArgumentNullException>(subscribe != null);            
-
-            _subscribe = subscribe;
-            Description = description;
-        }
-
-        /// <summary>
-        /// Gets the description of the observable.
-        /// </summary>
-        public string Description { get; }
-
-        /// <inheritdoc/>
-        protected override IDisposable SubscribeCore(IObserver<T> observer)
-        {
-            return _subscribe(observer) ?? Disposable.Empty;
-        }
-    }
-}

src/Avalonia.Base/Reactive/AvaloniaPropertyChangedObservable.cs

@@ -0,0 +1,46 @@
+using System;
+
+namespace Avalonia.Reactive
+{
+    internal class AvaloniaPropertyChangedObservable : 
+        LightweightObservableBase<AvaloniaPropertyChangedEventArgs>,
+        IDescription
+    {
+        private readonly WeakReference<IAvaloniaObject> _target;
+        private readonly AvaloniaProperty _property;
+
+        public AvaloniaPropertyChangedObservable(
+            IAvaloniaObject target,
+            AvaloniaProperty property)
+        {
+            _target = new WeakReference<IAvaloniaObject>(target);
+            _property = property;
+        }
+
+        public string Description => $"{_target.GetType().Name}.{_property.Name}";
+
+        protected override void Initialize()
+        {
+            if (_target.TryGetTarget(out var target))
+            {
+                target.PropertyChanged += PropertyChanged;
+            }
+        }
+
+        protected override void Deinitialize()
+        {
+            if (_target.TryGetTarget(out var target))
+            {
+                target.PropertyChanged -= PropertyChanged;
+            }
+        }
+
+        private void PropertyChanged(object sender, AvaloniaPropertyChangedEventArgs e)
+        {
+            if (e.Property == _property)
+            {
+                PublishNext(e);
+            }
+        }
+    }
+}

src/Avalonia.Base/Reactive/AvaloniaPropertyObservable.cs

@@ -0,0 +1,52 @@
+using System;
+
+namespace Avalonia.Reactive
+{
+    internal class AvaloniaPropertyObservable<T> : LightweightObservableBase<T>, IDescription
+    {
+        private readonly WeakReference<IAvaloniaObject> _target;
+        private readonly AvaloniaProperty _property;
+        private T _value;
+
+        public AvaloniaPropertyObservable(
+            IAvaloniaObject target,
+            AvaloniaProperty property)
+        {
+            _target = new WeakReference<IAvaloniaObject>(target);
+            _property = property;
+        }
+
+        public string Description => $"{_target.GetType().Name}.{_property.Name}";
+
+        protected override void Initialize()
+        {
+            if (_target.TryGetTarget(out var target))
+            {
+                _value = (T)target.GetValue(_property);
+                target.PropertyChanged += PropertyChanged;
+            }
+        }
+
+        protected override void Deinitialize()
+        {
+            if (_target.TryGetTarget(out var target))
+            {
+                target.PropertyChanged -= PropertyChanged;
+            }
+        }
+
+        protected override void Subscribed(IObserver<T> observer, bool first)
+        {
+            observer.OnNext(_value);
+        }
+
+        private void PropertyChanged(object sender, AvaloniaPropertyChangedEventArgs e)
+        {
+            if (e.Property == _property)
+            {
+                _value = (T)e.NewValue;
+                PublishNext(_value);
+            }
+        }
+    }
+}

src/Avalonia.Base/Reactive/LightweightObservableBase.cs

@@ -0,0 +1,202 @@
+using System;
+using System.Collections.Generic;
+using System.Reactive;
+using System.Reactive.Disposables;
+using System.Threading;
+using Avalonia.Threading;
+
+namespace Avalonia.Reactive
+{
+    /// <summary>
+    /// Lightweight base class for observable implementations.
+    /// </summary>
+    /// <typeparam name="T">The observable type.</typeparam>
+    /// <remarks>
+    /// <see cref="ObservableBase{T}"/> is rather heavyweight in terms of allocations and memory
+    /// usage. This class provides a more lightweight base for some internal observable types
+    /// in the Avalonia framework.
+    /// </remarks>
+    public abstract class LightweightObservableBase<T> : IObservable<T>
+    {
+        private Exception _error;
+        private List<IObserver<T>> _observers = new List<IObserver<T>>();
+
+        public IDisposable Subscribe(IObserver<T> observer)
+        {
+            Contract.Requires<ArgumentNullException>(observer != null);
+            Dispatcher.UIThread.VerifyAccess();
+
+            var first = false;
+
+            for (; ; )
+            {
+                if (Volatile.Read(ref _observers) == null)
+                {
+                    if (_error != null)
+                    {
+                        observer.OnError(_error);
+                    }
+                    else
+                    {
+                        observer.OnCompleted();
+                    }
+
+                    return Disposable.Empty;
+                }
+
+                lock (this)
+                {
+                    if (_observers == null)
+                    {
+                        continue;
+                    }
+
+                    first = _observers.Count == 0;
+                    _observers.Add(observer);
+                    break;
+                }
+            }
+
+            if (first)
+            {
+                Initialize();
+            }
+
+            Subscribed(observer, first);
+
+            return new RemoveObserver(this, observer);
+        }
+
+        void Remove(IObserver<T> observer)
+        {
+            if (Volatile.Read(ref _observers) != null)
+            {
+                lock (this)
+                {
+                    var observers = _observers;
+
+                    if (observers != null)
+                    {
+                        observers.Remove(observer);
+
+                        if (observers.Count == 0)
+                        {
+                            observers.TrimExcess();
+                        }
+                        else
+                        {
+                            return;
+                        }
+                    } else
+                    {
+                        return;
+                    }
+                }
+
+                Deinitialize();
+            }
+        }
+
+        sealed class RemoveObserver : IDisposable
+        {
+            LightweightObservableBase<T> _parent;
+
+            IObserver<T> _observer;
+
+            public RemoveObserver(LightweightObservableBase<T> parent, IObserver<T> observer)
+            {
+                _parent = parent;
+                Volatile.Write(ref _observer, observer);
+            }
+
+            public void Dispose()
+            {
+                var observer = _observer;
+                Interlocked.Exchange(ref _parent, null)?.Remove(observer);
+                _observer = null;
+            }
+        }
+
+        protected abstract void Initialize();
+        protected abstract void Deinitialize();
+
+        protected void PublishNext(T value)
+        {
+            if (Volatile.Read(ref _observers) != null)
+            {
+                IObserver<T>[] observers;
+
+                lock (this)
+                {
+                    if (_observers == null)
+                    {
+                        return;
+                    }
+                    observers = _observers.ToArray();
+                }
+
+                foreach (var observer in observers)
+                {
+                    observer.OnNext(value);
+                }
+            }
+        }
+
+        protected void PublishCompleted()
+        {
+            if (Volatile.Read(ref _observers) != null)
+            {
+                IObserver<T>[] observers;
+
+                lock (this)
+                {
+                    if (_observers == null)
+                    {
+                        return;
+                    }
+                    observers = _observers.ToArray();
+                    Volatile.Write(ref _observers, null);
+                }
+
+                foreach (var observer in observers)
+                {
+                    observer.OnCompleted();
+                }
+
+                Deinitialize();
+            }
+        }
+
+        protected void PublishError(Exception error)
+        {
+            if (Volatile.Read(ref _observers) != null)
+            {
+
+                IObserver<T>[] observers;
+
+                lock (this)
+                {
+                    if (_observers == null)
+                    {
+                        return;
+                    }
+
+                    _error = error;
+                    observers = _observers.ToArray();
+                    Volatile.Write(ref _observers, null);
+                }
+
+                foreach (var observer in observers)
+                {
+                    observer.OnError(error);
+                }
+
+                Deinitialize();
+            }
+        }
+
+        protected virtual void Subscribed(IObserver<T> observer, bool first)
+        {
+        }
+    }
+}

src/Avalonia.Base/Reactive/SingleSubscriberObservableBase.cs

@@ -0,0 +1,76 @@
+using System;
+using Avalonia.Threading;
+
+namespace Avalonia.Reactive
+{
+    public abstract class SingleSubscriberObservableBase<T> : IObservable<T>, IDisposable
+    {
+        private Exception _error;
+        private IObserver<T> _observer;
+        private bool _completed;
+
+        public IDisposable Subscribe(IObserver<T> observer)
+        {
+            Contract.Requires<ArgumentNullException>(observer != null);
+            Dispatcher.UIThread.VerifyAccess();
+
+            if (_observer != null)
+            {
+                throw new InvalidOperationException("The observable can only be subscribed once.");
+            }
+
+            if (_error != null)
+            {
+                observer.OnError(_error);
+            }
+            else if (_completed)
+            {
+                observer.OnCompleted();
+            }
+            else
+            {
+                _observer = observer;
+                Subscribed();
+            }
+
+            return this;
+        }
+
+        void IDisposable.Dispose()
+        {
+            Unsubscribed();
+            _observer = null;
+        }
+
+        protected abstract void Unsubscribed();
+
+        protected void PublishNext(T value)
+        {
+            _observer?.OnNext(value);
+        }
+
+        protected void PublishCompleted()
+        {
+            if (_observer != null)
+            {
+                _observer.OnCompleted();
+                _completed = true;
+                Unsubscribed();
+                _observer = null;
+            }
+        }
+
+        protected void PublishError(Exception error)
+        {
+            if (_observer != null)
+            {
+                _observer.OnError(error);
+                _error = error;
+                Unsubscribed();
+                _observer = null;
+            }
+        }
+
+        protected abstract void Subscribed();
+    }
+}

src/Avalonia.Base/Reactive/WeakPropertyChangedObservable.cs

@@ -1,85 +0,0 @@
-// 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;
-using System.Reactive.Disposables;
-using System.Reactive.Linq;
-using System.Reactive.Subjects;
-using Avalonia.Utilities;
-
-namespace Avalonia.Reactive
-{
-    internal class WeakPropertyChangedObservable : ObservableBase<object>, 
-        IWeakSubscriber<AvaloniaPropertyChangedEventArgs>, IDescription
-    {
-        private WeakReference<IAvaloniaObject> _sourceReference;
-        private readonly AvaloniaProperty _property;
-        private readonly Subject<object> _changed = new Subject<object>();
-
-        private int _count;
-
-        public WeakPropertyChangedObservable(
-            WeakReference<IAvaloniaObject> source, 
-            AvaloniaProperty property, 
-            string description)
-        {
-            _sourceReference = source;
-            _property = property;
-            Description = description;
-        }
-
-        public string Description { get; }
-
-        public void OnEvent(object sender, AvaloniaPropertyChangedEventArgs e)
-        {
-            if (e.Property == _property)
-            {
-                _changed.OnNext(e.NewValue);
-            }
-        }
-
-        protected override IDisposable SubscribeCore(IObserver<object> observer)
-        {
-            IAvaloniaObject instance;
-
-            if (_sourceReference.TryGetTarget(out instance))
-            {
-                if (_count++ == 0)
-                {
-                    WeakSubscriptionManager.Subscribe(
-                        instance, 
-                        nameof(instance.PropertyChanged), 
-                        this);
-                }
-
-                observer.OnNext(instance.GetValue(_property));
-
-                return Observable.Using(() => Disposable.Create(DecrementCount), _ => _changed)
-                    .Subscribe(observer);
-            }
-            else
-            {
-                _changed.OnCompleted();
-                observer.OnCompleted();
-                return Disposable.Empty;
-            }
-        }
-
-        private void DecrementCount()
-        {
-            if (--_count == 0)
-            {
-                IAvaloniaObject instance;
-
-                if (_sourceReference.TryGetTarget(out instance))
-                {
-                    WeakSubscriptionManager.Unsubscribe(
-                    instance,
-                    nameof(instance.PropertyChanged),
-                    this);
-                }
-            }
-        }
-    }
-}

src/Avalonia.Controls/Mixins/ContentControlMixin.cs

@@ -49,11 +49,9 @@ namespace Avalonia.Controls.Mixins
             Contract.Requires<ArgumentNullException>(content != null);
             Contract.Requires<ArgumentNullException>(logicalChildrenSelector != null);
 
-            EventHandler<RoutedEventArgs> templateApplied = (s, ev) =>
+            void TemplateApplied(object s, RoutedEventArgs ev)
             {
-                var sender = s as TControl;
-
-                if (sender != null)
+                if (s is TControl sender)
                 {
                     var e = (TemplateAppliedEventArgs)ev;
                     var presenter = (IControl)e.NameScope.Find(presenterName);
@@ -64,12 +62,12 @@ namespace Avalonia.Controls.Mixins
 
                         var logicalChildren = logicalChildrenSelector(sender);
                         var subscription = presenter
-                            .GetObservableWithHistory(ContentPresenter.ChildProperty)
-                            .Subscribe(child => UpdateLogicalChild(
+                            .GetPropertyChangedObservable(ContentPresenter.ChildProperty)
+                            .Subscribe(c => UpdateLogicalChild(
                                 sender,
-                                logicalChildren, 
-                                child.Item1, 
-                                child.Item2));
+                                logicalChildren,
+                                c.OldValue,
+                                c.NewValue));
 
                         UpdateLogicalChild(
                             sender,
@@ -80,18 +78,16 @@ namespace Avalonia.Controls.Mixins
                         subscriptions.Value.Add(sender, subscription);
                     }
                 }
-            };
+            }
 
             TemplatedControl.TemplateAppliedEvent.AddClassHandler(
                 typeof(TControl),
-                templateApplied,
+                TemplateApplied,
                 RoutingStrategies.Direct);
 
             content.Changed.Subscribe(e =>
             {
-                var sender = e.Sender as TControl;
-
-                if (sender != null)
+                if (e.Sender is TControl sender)
                 {
                     var logicalChildren = logicalChildrenSelector(sender);
                     UpdateLogicalChild(sender, logicalChildren, e.OldValue, e.NewValue);
@@ -100,9 +96,7 @@ namespace Avalonia.Controls.Mixins
 
             Control.TemplatedParentProperty.Changed.Subscribe(e =>
             {
-                var sender = e.Sender as TControl;
-
-                if (sender != null)
+                if (e.Sender is TControl sender)
                 {
                     var logicalChild = logicalChildrenSelector(sender).FirstOrDefault() as IControl;
                     logicalChild?.SetValue(Control.TemplatedParentProperty, sender.TemplatedParent);
@@ -111,13 +105,9 @@ namespace Avalonia.Controls.Mixins
 
             TemplatedControl.TemplateProperty.Changed.Subscribe(e =>
             {
-                var sender = e.Sender as TControl;
-
-                if (sender != null)
+                if (e.Sender is TControl sender)
                 {
-                    IDisposable subscription;
-
-                    if (subscriptions.Value.TryGetValue(sender, out subscription))
+                    if (subscriptions.Value.TryGetValue(sender, out IDisposable subscription))
                     {
                         subscription.Dispose();
                         subscriptions.Value.Remove(sender);
@@ -134,9 +124,7 @@ namespace Avalonia.Controls.Mixins
         {
             if (oldValue != newValue)
             {
-                var child = oldValue as IControl;
-
-                if (child != null)
+                if (oldValue is IControl child)
                 {
                     logicalChildren.Remove(child);
                 }

src/Avalonia.Styling/Controls/ResourceProviderExtensions.cs

@@ -1,6 +1,7 @@
 using System;
 using System.Reactive;
 using System.Reactive.Linq;
+using Avalonia.Reactive;
 
 namespace Avalonia.Controls
 {
@@ -55,11 +56,39 @@ namespace Avalonia.Controls
 
         public static IObservable<object> GetResourceObservable(this IResourceNode target, string key)
         {
-            return Observable.FromEventPattern<ResourcesChangedEventArgs>(
-                x => target.ResourcesChanged += x,
-                x => target.ResourcesChanged -= x)
-                .StartWith((EventPattern<ResourcesChangedEventArgs>)null)
-                .Select(x => target.FindResource(key));
+            return new ResourceObservable(target, key);
+        }
+
+        private class ResourceObservable : LightweightObservableBase<object>
+        {
+            private readonly IResourceNode _target;
+            private readonly string _key;
+
+            public ResourceObservable(IResourceNode target, string key)
+            {
+                _target = target;
+                _key = key;
+            }
+
+            protected override void Initialize()
+            {
+                _target.ResourcesChanged += ResourcesChanged;
+            }
+
+            protected override void Deinitialize()
+            {
+                _target.ResourcesChanged -= ResourcesChanged;
+            }
+
+            protected override void Subscribed(IObserver<object> observer, bool first)
+            {
+                observer.OnNext(_target.FindResource(_key));
+            }
+
+            private void ResourcesChanged(object sender, ResourcesChangedEventArgs e)
+            {
+                PublishNext(_target.FindResource(_key));
+            }
         }
     }
 }

src/Avalonia.Styling/LogicalTree/ControlLocator.cs

@@ -6,6 +6,7 @@ using System.Linq;
 using System.Reactive.Linq;
 using System.Reflection;
 using Avalonia.Controls;
+using Avalonia.Reactive;
 
 namespace Avalonia.LogicalTree
 {
@@ -23,75 +24,122 @@ namespace Avalonia.LogicalTree
         /// <param name="name">The name of the control to find.</param>
         public static IObservable<ILogical> Track(ILogical relativeTo, string name)
         {
-            var attached = Observable.FromEventPattern<LogicalTreeAttachmentEventArgs>(
-                x => relativeTo.AttachedToLogicalTree += x,
-                x => relativeTo.AttachedToLogicalTree -= x)
-                .Select(x => ((ILogical)x.Sender).FindNameScope())
-                .StartWith(relativeTo.FindNameScope());
-
-            var detached = Observable.FromEventPattern<LogicalTreeAttachmentEventArgs>(
-                x => relativeTo.DetachedFromLogicalTree += x,
-                x => relativeTo.DetachedFromLogicalTree -= x)
-                .Select(x => (INameScope)null);
-
-            return attached.Merge(detached).Select(nameScope =>
+            return new ControlTracker(relativeTo, name);
+        }
+
+        public static IObservable<ILogical> Track(ILogical relativeTo, int ancestorLevel, Type ancestorType = null)
+        {
+            return new ControlTracker(relativeTo, ancestorLevel, ancestorType);
+        }
+
+        private class ControlTracker : LightweightObservableBase<ILogical>
+        {
+            private readonly ILogical _relativeTo;
+            private readonly string _name;
+            private readonly int _ancestorLevel;
+            private readonly Type _ancestorType;
+            INameScope _nameScope;
+            ILogical _value;
+
+            public ControlTracker(ILogical relativeTo, string name)
+            {
+                _relativeTo = relativeTo;
+                _name = name;
+            }
+
+            public ControlTracker(ILogical relativeTo, int ancestorLevel, Type ancestorType)
             {
-                if (nameScope != null)
+                _relativeTo = relativeTo;
+                _ancestorLevel = ancestorLevel;
+                _ancestorType = ancestorType;
+            }
+
+            protected override void Initialize()
+            {
+                Update();
+                _relativeTo.AttachedToLogicalTree += Attached;
+                _relativeTo.DetachedFromLogicalTree += Detached;
+            }
+
+            protected override void Deinitialize()
+            {
+                _relativeTo.AttachedToLogicalTree -= Attached;
+                _relativeTo.DetachedFromLogicalTree -= Detached;
+
+                if (_nameScope != null)
                 {
-                    var registered = Observable.FromEventPattern<NameScopeEventArgs>(
-                        x => nameScope.Registered += x,
-                        x => nameScope.Registered -= x)
-                        .Where(x => x.EventArgs.Name == name)
-                        .Select(x => x.EventArgs.Element)
-                        .OfType<ILogical>();
-                    var unregistered = Observable.FromEventPattern<NameScopeEventArgs>(
-                        x => nameScope.Unregistered += x,
-                        x => nameScope.Unregistered -= x)
-                        .Where(x => x.EventArgs.Name == name)
-                        .Select(_ => (ILogical)null);
-                    return registered
-                        .StartWith(nameScope.Find<ILogical>(name))
-                        .Merge(unregistered);
+                    _nameScope.Registered -= Registered;
+                    _nameScope.Unregistered -= Unregistered;
                 }
-                else
+
+                _value = null;
+            }
+
+            protected override void Subscribed(IObserver<ILogical> observer, bool first)
+            {
+                observer.OnNext(_value);
+            }
+
+            private void Attached(object sender, LogicalTreeAttachmentEventArgs e)
+            {
+                Update();
+                PublishNext(_value);
+            }
+
+            private void Detached(object sender, LogicalTreeAttachmentEventArgs e)
+            {
+                if (_nameScope != null)
                 {
-                    return Observable.Return<ILogical>(null);
+                    _nameScope.Registered -= Registered;
+                    _nameScope.Unregistered -= Unregistered;
                 }
-            }).Switch();
-        }
 
-        public static IObservable<ILogical> Track(ILogical relativeTo, int ancestorLevel, Type ancestorType = null)
-        {
-            return TrackAttachmentToTree(relativeTo).Select(isAttachedToTree =>
+                _value = null;
+                PublishNext(null);
+            }
+
+            private void Registered(object sender, NameScopeEventArgs e)
             {
-                if (isAttachedToTree)
+                if (e.Name == _name && e.Element is ILogical logical)
                 {
-                    return relativeTo.GetLogicalAncestors()
-                        .Where(x => ancestorType?.GetTypeInfo().IsAssignableFrom(x.GetType().GetTypeInfo()) ?? true)
-                        .ElementAtOrDefault(ancestorLevel);
+                    _value = logical;
+                    PublishNext(logical);
                 }
-                else
+            }
+
+            private void Unregistered(object sender, NameScopeEventArgs e)
+            {
+                if (e.Name == _name)
                 {
-                    return null;
+                    _value = null;
+                    PublishNext(null);
                 }
-            });
-        }
+            }
 
-        private static IObservable<bool> TrackAttachmentToTree(ILogical relativeTo)
-        {
-            var attached = Observable.FromEventPattern<LogicalTreeAttachmentEventArgs>(
-                x => relativeTo.AttachedToLogicalTree += x,
-                x => relativeTo.AttachedToLogicalTree -= x)
-                .Select(x => true)
-                .StartWith(relativeTo.IsAttachedToLogicalTree);
-
-            var detached = Observable.FromEventPattern<LogicalTreeAttachmentEventArgs>(
-                x => relativeTo.DetachedFromLogicalTree += x,
-                x => relativeTo.DetachedFromLogicalTree -= x)
-                .Select(x => false);
-
-            var attachmentStatus = attached.Merge(detached);
-            return attachmentStatus;
+            private void Update()
+            {
+                if (_name != null)
+                {
+                    _nameScope = _relativeTo.FindNameScope();
+
+                    if (_nameScope != null)
+                    {
+                        _nameScope.Registered += Registered;
+                        _nameScope.Unregistered += Unregistered;
+                        _value = _nameScope.Find<ILogical>(_name);
+                    }
+                    else
+                    {
+                        _value = null;
+                    }
+                }
+                else
+                {
+                    _value = _relativeTo.GetLogicalAncestors()
+                        .Where(x => _ancestorType?.GetTypeInfo().IsAssignableFrom(x.GetType().GetTypeInfo()) ?? true)
+                        .ElementAtOrDefault(_ancestorLevel);
+                }
+            }
         }
     }
 }

src/Avalonia.Styling/Styling/ActivatedObservable.cs

@@ -2,8 +2,6 @@
 // Licensed under the MIT license. See licence.md file in the project root for full license information.
 
 using System;
-using System.Reactive;
-using System.Reactive.Linq;
 
 namespace Avalonia.Styling
 {
@@ -11,14 +9,16 @@ namespace Avalonia.Styling
     /// An observable which is switched on or off according to an activator observable.
     /// </summary>
     /// <remarks>
-    /// An <see cref="ActivatedObservable"/> has two inputs: an activator observable a 
+    /// An <see cref="ActivatedObservable"/> has two inputs: an activator observable and a 
     /// <see cref="Source"/> observable which produces the activated value. When the activator 
     /// produces true, the <see cref="ActivatedObservable"/> will produce the current activated 
     /// value. When the activator produces false it will produce
     /// <see cref="AvaloniaProperty.UnsetValue"/>.
     /// </remarks>
-    internal class ActivatedObservable : ObservableBase<object>, IDescription
+    internal class ActivatedObservable : ActivatedValue, IDescription
     {
+        private IDisposable _sourceSubscription;
+
         /// <summary>
         /// Initializes a new instance of the <see cref="ActivatedObservable"/> class.
         /// </summary>
@@ -29,49 +29,49 @@ namespace Avalonia.Styling
             IObservable<bool> activator,
             IObservable<object> source,
             string description)
+            : base(activator, AvaloniaProperty.UnsetValue, description)
         {
-            Contract.Requires<ArgumentNullException>(activator != null);
             Contract.Requires<ArgumentNullException>(source != null);
 
-            Activator = activator;
-            Description = description;
             Source = source;
         }
 
-        /// <summary>
-        /// Gets the activator observable.
-        /// </summary>
-        public IObservable<bool> Activator { get; }
-
-        /// <summary>
-        /// Gets a description of the binding.
-        /// </summary>
-        public string Description { get; }
-
         /// <summary>
         /// Gets an observable which produces the <see cref="ActivatedValue"/>.
         /// </summary>
         public IObservable<object> Source { get; }
 
-        /// <summary>
-        /// Notifies the provider that an observer is to receive notifications.
-        /// </summary>
-        /// <param name="observer">The observer.</param>
-        /// <returns>IDisposable object used to unsubscribe from the observable sequence.</returns>
-        protected override IDisposable SubscribeCore(IObserver<object> observer)
+        protected override ActivatorListener CreateListener() => new ValueListener(this);
+
+        protected override void Deinitialize()
         {
-            Contract.Requires<ArgumentNullException>(observer != null);
+            base.Deinitialize();
+            _sourceSubscription.Dispose();
+            _sourceSubscription = null;
+        }
+
+        protected override void Initialize()
+        {
+            base.Initialize();
+            _sourceSubscription = Source.Subscribe((ValueListener)Listener);
+        }
 
-            var sourceCompleted = Source.LastOrDefaultAsync().Select(_ => Unit.Default);
-            var activatorCompleted = Activator.LastOrDefaultAsync().Select(_ => Unit.Default);
-            var completed = sourceCompleted.Merge(activatorCompleted);
+        protected virtual void NotifyValue(object value)
+        {
+            Value = value;
+        }
+
+        private class ValueListener : ActivatorListener, IObserver<object>
+        {
+            public ValueListener(ActivatedObservable parent)
+                : base(parent)
+            {
+            }
+            protected new ActivatedObservable Parent => (ActivatedObservable)base.Parent;
 
-            return Activator
-                .CombineLatest(Source, (x, y) => new { Active = x, Value = y })
-                .Select(x => x.Active ? x.Value : AvaloniaProperty.UnsetValue)
-                .DistinctUntilChanged()
-                .TakeUntil(completed)
-                .Subscribe(observer);
+            void IObserver<object>.OnCompleted() => Parent.CompletedReceived();
+            void IObserver<object>.OnError(Exception error) => Parent.ErrorReceived(error);
+            void IObserver<object>.OnNext(object value) => Parent.NotifyValue(value);
         }
     }
 }

src/Avalonia.Styling/Styling/ActivatedSubject.cs

@@ -2,7 +2,6 @@
 // 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;
 
 namespace Avalonia.Styling
@@ -11,17 +10,14 @@ namespace Avalonia.Styling
     /// A subject which is switched on or off according to an activator observable.
     /// </summary>
     /// <remarks>
-    /// An <see cref="ActivatedSubject"/> has two inputs: an activator observable and either an
-    /// <see cref="ActivatedValue"/> or a <see cref="Source"/> observable which produces the
-    /// activated value. When the activator produces true, the <see cref="ActivatedObservable"/> will
-    /// produce the current activated value. When the activator produces false it will produce
-    /// <see cref="AvaloniaProperty.UnsetValue"/>.
+    /// An <see cref="ActivatedSubject"/> extends <see cref="ActivatedObservable"/> to
+    /// be an <see cref="ISubject{Object}"/>. When the object is active then values
+    /// received via <see cref="OnNext(object)"/> will be passed to the source subject.
     /// </remarks>
     internal class ActivatedSubject : ActivatedObservable, ISubject<object>, IDescription
     {
-        private bool? _active;
         private bool _completed;
-        private object _value;
+        private object _pushValue;
 
         /// <summary>
         /// Initializes a new instance of the <see cref="ActivatedSubject"/> class.
@@ -35,7 +31,6 @@ namespace Avalonia.Styling
             string description)
             : base(activator, source, description)
         {
-            Activator.Subscribe(ActivatorChanged, ActivatorError, ActivatorCompleted);
         }
 
         /// <summary>
@@ -46,53 +41,57 @@ namespace Avalonia.Styling
             get { return (ISubject<object>)base.Source; }
         }
 
-        /// <summary>
-        /// Notifies all subscribed observers about the end of the sequence.
-        /// </summary>
         public void OnCompleted()
         {
-            if (_active.Value && !_completed)
-            {
-                Source.OnCompleted();
-            }
+            Source.OnCompleted();
         }
 
-        /// <summary>
-        /// Notifies all subscribed observers with the exception.
-        /// </summary>
-        /// <param name="error">The exception to send to all subscribed observers.</param>
-        /// <exception cref="ArgumentNullException"><paramref name="error"/> is null.</exception>
         public void OnError(Exception error)
         {
-            if (_active.Value && !_completed)
-            {
-                Source.OnError(error);
-            }
+            Source.OnError(error);
         }
 
-        /// <summary>
-        /// Notifies all subscribed observers with the value.
-        /// </summary>
-        /// <param name="value">The value to send to all subscribed observers.</param>        
         public void OnNext(object value)
         {
-            _value = value;
+            _pushValue = value;
 
-            if (_active.Value && !_completed)
+            if (IsActive == true && !_completed)
             {
-                Source.OnNext(value);
+                Source.OnNext(_pushValue);
             }
         }
 
-        private void ActivatorChanged(bool active)
+        protected override void ActiveChanged(bool active)
         {
-            bool first = !_active.HasValue;
+            bool first = !IsActive.HasValue;
 
-            _active = active;
+            base.ActiveChanged(active);
 
             if (!first)
             {
-                Source.OnNext(active ? _value : AvaloniaProperty.UnsetValue);
+                Source.OnNext(active ? _pushValue : AvaloniaProperty.UnsetValue);
+            }
+        }
+
+        protected override void CompletedReceived()
+        {
+            base.CompletedReceived();
+
+            if (!_completed)
+            {
+                Source.OnCompleted();
+                _completed = true;
+            }
+        }
+
+        protected override void ErrorReceived(Exception error)
+        {
+            base.ErrorReceived(error);
+
+            if (!_completed)
+            {
+                Source.OnError(error);
+                _completed = true;
             }
         }
 

src/Avalonia.Styling/Styling/ActivatedValue.cs

@@ -2,8 +2,7 @@
 // Licensed under the MIT license. See licence.md file in the project root for full license information.
 
 using System;
-using System.Reactive;
-using System.Reactive.Linq;
+using Avalonia.Reactive;
 
 namespace Avalonia.Styling
 {
@@ -16,12 +15,12 @@ namespace Avalonia.Styling
     /// <see cref="ActivatedValue"/> will produce the current value. When the activator 
     /// produces false it will produce <see cref="AvaloniaProperty.UnsetValue"/>.
     /// </remarks>
-    internal class ActivatedValue : ObservableBase<object>, IDescription
+    internal class ActivatedValue : LightweightObservableBase<object>, IDescription
     {
-        /// <summary>
-        /// The activator.
-        /// </summary>
-        private readonly IObservable<bool> _activator;
+        private static readonly object NotSent = new object();
+        private IDisposable _activatorSubscription;
+        private object _value;
+        private object _last = NotSent;
 
         /// <summary>
         /// Initializes a new instance of the <see cref="ActivatedObservable"/> class.
@@ -34,39 +33,101 @@ namespace Avalonia.Styling
             object value,
             string description)
         {
-            _activator = activator;
+            Contract.Requires<ArgumentNullException>(activator != null);
+
+            Activator = activator;
             Value = value;
             Description = description;
+            Listener = CreateListener();
         }
 
         /// <summary>
-        /// Gets the activated value.
+        /// Gets the activator observable.
         /// </summary>
-        public object Value
-        {
-            get;
-        }
+        public IObservable<bool> Activator { get; }
 
         /// <summary>
         /// Gets a description of the binding.
         /// </summary>
-        public string Description
-        {
-            get;
-        }
+        public string Description { get; }
+
+        /// <summary>
+        /// Gets a value indicating whether the activator is active.
+        /// </summary>
+        public bool? IsActive { get; private set; }
 
         /// <summary>
-        /// Notifies the provider that an observer is to receive notifications.
+        /// Gets the value that will be produced when <see cref="IsActive"/> is true.
         /// </summary>
-        /// <param name="observer">The observer.</param>
-        /// <returns>IDisposable object used to unsubscribe from the observable sequence.</returns>
-        protected override IDisposable SubscribeCore(IObserver<object> observer)
+        public object Value
+        {
+            get => _value;
+            protected set
+            {
+                _value = value;
+                PublishValue();
+            }
+        }
+
+        protected ActivatorListener Listener { get; }
+
+        protected virtual void ActiveChanged(bool active)
+        {
+            IsActive = active;
+            PublishValue();
+        }
+
+        protected virtual void CompletedReceived() => PublishCompleted();
+
+        protected virtual ActivatorListener CreateListener() => new ActivatorListener(this);
+
+        protected override void Deinitialize()
+        {
+            _activatorSubscription.Dispose();
+            _activatorSubscription = null;
+        }
+
+        protected virtual void ErrorReceived(Exception error) => PublishError(error);
+
+        protected override void Initialize()
+        {
+            _activatorSubscription = Activator.Subscribe(Listener);
+        }
+
+        protected override void Subscribed(IObserver<object> observer, bool first)
         {
-            Contract.Requires<ArgumentNullException>(observer != null);
+            if (IsActive == true && !first)
+            {
+                observer.OnNext(Value);
+            }
+        }
+
+        private void PublishValue()
+        {
+            if (IsActive.HasValue)
+            {
+                var v = IsActive.Value ? Value : AvaloniaProperty.UnsetValue;
+
+                if (!Equals(v, _last))
+                {
+                    PublishNext(v);
+                    _last = v;
+                }
+            }
+        }
+
+        protected class ActivatorListener : IObserver<bool>
+        {
+            public ActivatorListener(ActivatedValue parent)
+            {
+                Parent = parent;
+            }
+
+            protected ActivatedValue Parent { get; }
 
-            return _activator
-                .Select(active => active ? Value : AvaloniaProperty.UnsetValue)
-                .Subscribe(observer);
+            void IObserver<bool>.OnCompleted() => Parent.CompletedReceived();
+            void IObserver<bool>.OnError(Exception error) => Parent.ErrorReceived(error);
+            void IObserver<bool>.OnNext(bool value) => Parent.ActiveChanged(value);
         }
     }
 }

src/Avalonia.Styling/Styling/StyleActivator.cs

@@ -48,8 +48,8 @@ namespace Avalonia.Styling
             else
             {
                 return inputs.CombineLatest()
-                .Select(values => values.Any(x => x))
-                .DistinctUntilChanged();
+                    .Select(values => values.Any(x => x))
+                    .DistinctUntilChanged();
             }
         }
     }

src/Avalonia.Styling/Styling/TypeNameAndClassSelector.cs

@@ -4,10 +4,10 @@
 using System;
 using System.Collections.Generic;
 using System.Collections.Specialized;
-using System.Reactive;
-using System.Reactive.Linq;
 using System.Reflection;
 using System.Text;
+using Avalonia.Collections;
+using Avalonia.Reactive;
 
 namespace Avalonia.Styling
 {
@@ -122,14 +122,7 @@ namespace Avalonia.Styling
             {
                 if (subscribe)
                 {
-                    var observable = Observable.FromEventPattern<
-                            NotifyCollectionChangedEventHandler,
-                            NotifyCollectionChangedEventArgs>(
-                        x => control.Classes.CollectionChanged += x,
-                        x => control.Classes.CollectionChanged -= x)
-                        .StartWith((EventPattern<NotifyCollectionChangedEventArgs>)null)
-                        .Select(_ => Matches(control.Classes))
-                        .DistinctUntilChanged();
+                    var observable = new ClassObserver(control.Classes, _classes.Value);
                     return new SelectorMatch(observable);
                 }
                 else
@@ -204,5 +197,60 @@ namespace Avalonia.Styling
 
             return builder.ToString();
         }
+
+        private class ClassObserver : LightweightObservableBase<bool>
+        {
+            readonly IList<string> _match;
+            IAvaloniaReadOnlyList<string> _classes;
+            bool _value;
+
+            public ClassObserver(IAvaloniaReadOnlyList<string> classes, IList<string> match)
+            {
+                _classes = classes;
+                _match = match;
+            }
+
+            protected override void Deinitialize() => _classes.CollectionChanged -= ClassesChanged;
+
+            protected override void Initialize()
+            {
+                _value = GetResult();
+                _classes.CollectionChanged += ClassesChanged;
+            }
+
+            protected override void Subscribed(IObserver<bool> observer, bool first)
+            {
+                observer.OnNext(_value);
+            }
+
+            private void ClassesChanged(object sender, NotifyCollectionChangedEventArgs e)
+            {
+                if (e.Action != NotifyCollectionChangedAction.Move)
+                {
+                    var value = GetResult();
+
+                    if (value != _value)
+                    {
+                        PublishNext(GetResult());
+                        _value = value;
+                    }
+                }
+            }
+
+            private bool GetResult()
+            {
+                int remaining = _match.Count;
+
+                foreach (var c in _classes)
+                {
+                    if (_match.Contains(c))
+                    {
+                        --remaining;
+                    }
+                }
+
+                return remaining == 0;
+            }
+        }
     }
 }

src/Avalonia.Visuals/VisualTree/VisualLocator.cs

@@ -1,9 +1,8 @@
 using System;
-using System.Collections.Generic;
 using System.Linq;
 using System.Reactive.Linq;
 using System.Reflection;
-using System.Text;
+using Avalonia.Reactive;
 
 namespace Avalonia.VisualTree
 {
@@ -11,36 +10,54 @@ namespace Avalonia.VisualTree
     {
         public static IObservable<IVisual> Track(IVisual relativeTo, int ancestorLevel, Type ancestorType = null)
         {
-            return TrackAttachmentToTree(relativeTo).Select(isAttachedToTree =>
+            return new VisualTracker(relativeTo, ancestorLevel, ancestorType);
+        }
+
+        private class VisualTracker : LightweightObservableBase<IVisual>
+        {
+            private readonly IVisual _relativeTo;
+            private readonly int _ancestorLevel;
+            private readonly Type _ancestorType;
+
+            public VisualTracker(IVisual relativeTo, int ancestorLevel, Type ancestorType)
+            {
+                _relativeTo = relativeTo;
+                _ancestorLevel = ancestorLevel;
+                _ancestorType = ancestorType;
+            }
+
+            protected override void Initialize()
+            {
+                _relativeTo.AttachedToVisualTree += AttachedDetached;
+                _relativeTo.DetachedFromVisualTree += AttachedDetached;
+            }
+
+            protected override void Deinitialize()
             {
-                if (isAttachedToTree)
+                _relativeTo.AttachedToVisualTree -= AttachedDetached;
+                _relativeTo.DetachedFromVisualTree -= AttachedDetached;
+            }
+
+            protected override void Subscribed(IObserver<IVisual> observer, bool first)
+            {
+                observer.OnNext(GetResult());
+            }
+
+            private void AttachedDetached(object sender, VisualTreeAttachmentEventArgs e) => PublishNext(GetResult());
+
+            private IVisual GetResult()
+            {
+                if (_relativeTo.IsAttachedToVisualTree)
                 {
-                    return relativeTo.GetVisualAncestors()
-                        .Where(x => ancestorType?.GetTypeInfo().IsAssignableFrom(x.GetType().GetTypeInfo()) ?? true)
-                        .ElementAtOrDefault(ancestorLevel);
+                    return _relativeTo.GetVisualAncestors()
+                        .Where(x => _ancestorType?.GetTypeInfo().IsAssignableFrom(x.GetType().GetTypeInfo()) ?? true)
+                        .ElementAtOrDefault(_ancestorLevel);
                 }
                 else
                 {
                     return null;
                 }
-            });
-        }
-
-        private static IObservable<bool> TrackAttachmentToTree(IVisual relativeTo)
-        {
-            var attached = Observable.FromEventPattern<VisualTreeAttachmentEventArgs>(
-                x => relativeTo.AttachedToVisualTree += x,
-                x => relativeTo.AttachedToVisualTree -= x)
-                .Select(x => true)
-                .StartWith(relativeTo.IsAttachedToVisualTree);
-
-            var detached = Observable.FromEventPattern<VisualTreeAttachmentEventArgs>(
-                x => relativeTo.DetachedFromVisualTree += x,
-                x => relativeTo.DetachedFromVisualTree -= x)
-                .Select(x => false);
-
-            var attachmentStatus = attached.Merge(detached);
-            return attachmentStatus;
+            }
         }
     }
 }

src/Markup/Avalonia.Markup/Data/Binding.cs

@@ -5,11 +5,10 @@ using System;
 using System.Linq;
 using System.Reactive;
 using System.Reactive.Linq;
-using System.Reflection;
-using Avalonia.Controls;
 using Avalonia.Data.Converters;
 using Avalonia.Data.Core;
 using Avalonia.LogicalTree;
+using Avalonia.Reactive;
 using Avalonia.VisualTree;
 
 namespace Avalonia.Data
@@ -190,13 +189,10 @@ namespace Avalonia.Data
 
             if (!targetIsDataContext)
             {
-                var update = target.GetObservable(StyledElement.DataContextProperty)
-                    .Skip(1)
-                    .Select(_ => Unit.Default);
                 var result = new ExpressionObserver(
                     () => target.GetValue(StyledElement.DataContextProperty),
                     path,
-                    update,
+                    new UpdateSignal(target, StyledElement.DataContextProperty),
                     enableDataValidation);
 
                 return result;
@@ -278,14 +274,10 @@ namespace Avalonia.Data
         {
             Contract.Requires<ArgumentNullException>(target != null);
 
-            var update = target.GetObservable(StyledElement.TemplatedParentProperty)
-                .Skip(1)
-                .Select(_ => Unit.Default);
-
             var result = new ExpressionObserver(
                 () => target.GetValue(StyledElement.TemplatedParentProperty),
                 path,
-                update,
+                new UpdateSignal(target, StyledElement.TemplatedParentProperty),
                 enableDataValidation);
 
             return result;
@@ -306,5 +298,35 @@ namespace Avalonia.Data
                            Observable.Return((object)null);
                 }).Switch();
         }
+
+        private class UpdateSignal : SingleSubscriberObservableBase<Unit>
+        {
+            private readonly IAvaloniaObject _target;
+            private readonly AvaloniaProperty _property;
+
+            public UpdateSignal(IAvaloniaObject target, AvaloniaProperty property)
+            {
+                _target = target;
+                _property = property;
+            }
+
+            protected override void Subscribed()
+            {
+                _target.PropertyChanged += PropertyChanged;
+            }
+
+            protected override void Unsubscribed()
+            {
+                _target.PropertyChanged -= PropertyChanged;
+            }
+
+            private void PropertyChanged(object sender, AvaloniaPropertyChangedEventArgs e)
+            {
+                if (e.Property == _property)
+                {
+                    PublishNext(Unit.Default);
+                }
+            }
+        }
     }
 }

tests/Avalonia.Base.UnitTests/Data/Core/BindingExpressionTests.cs

@@ -337,6 +337,21 @@ namespace Avalonia.Base.UnitTests.Data.Core
             GC.KeepAlive(data);
         }
 
+        [Fact]
+        public void Second_Subscription_Should_Fire_Immediately()
+        {
+            var data = new Class1 { StringValue = "foo" };
+            var target = new BindingExpression(new ExpressionObserver(data, "StringValue"), typeof(string));
+            object result = null;
+
+            target.Subscribe();
+            target.Subscribe(x => result = x);
+
+            Assert.Equal("foo", result);
+
+            GC.KeepAlive(data);
+        }
+
         private class Class1 : NotifyingBase
         {
             private string _stringValue;

tests/Avalonia.Styling.UnitTests/ActivatedObservableTests.cs

@@ -54,15 +54,16 @@ namespace Avalonia.Styling.UnitTests
         }
 
         [Fact]
-        public void Should_Complete_When_Activator_Completes()
+        public void Should_Error_When_Source_Errors()
         {
             var activator = new BehaviorSubject<bool>(false);
             var source = new BehaviorSubject<object>(1);
             var target = new ActivatedObservable(activator, source, string.Empty);
+            var error = new Exception();
             var completed = false;
 
-            target.Subscribe(_ => { }, () => completed = true);
-            activator.OnCompleted();
+            target.Subscribe(_ => { }, x => completed = true);
+            source.OnError(error);
 
             Assert.True(completed);
         }

tests/Avalonia.Styling.UnitTests/ActivatedSubjectTests.cs

@@ -17,6 +17,7 @@ namespace Avalonia.Styling.UnitTests
             var source = new TestSubject();
             var target = new ActivatedSubject(activator, source, string.Empty);
 
+            target.Subscribe();
             target.OnNext("bar");
             Assert.Equal(AvaloniaProperty.UnsetValue, source.Value);
             activator.OnNext(true);
@@ -36,6 +37,7 @@ namespace Avalonia.Styling.UnitTests
             var source = new TestSubject();
             var target = new ActivatedSubject(activator, source, string.Empty);
 
+            target.Subscribe();
             activator.OnCompleted();
 
             Assert.True(source.Completed);
@@ -47,10 +49,14 @@ namespace Avalonia.Styling.UnitTests
             var activator = new BehaviorSubject<bool>(false);
             var source = new TestSubject();
             var target = new ActivatedSubject(activator, source, string.Empty);
+            var targetError = default(Exception);
+            var error = new Exception();
 
-            activator.OnError(new Exception());
+            target.Subscribe(_ => { }, e => targetError = e);
+            activator.OnError(error);
 
-            Assert.NotNull(source.Error);
+            Assert.Same(error, source.Error);
+            Assert.Same(error, targetError);
         }
 
         private class TestSubject : ISubject<object>

tests/Avalonia.Styling.UnitTests/ActivatedValueTests.cs

@@ -40,6 +40,20 @@ namespace Avalonia.Styling.UnitTests
             Assert.True(completed);
         }
 
+        [Fact]
+        public void Should_Error_When_Activator_Errors()
+        {
+            var activator = new BehaviorSubject<bool>(false);
+            var target = new ActivatedValue(activator, 1, string.Empty);
+            var error = new Exception();
+            var completed = false;
+
+            target.Subscribe(_ => { }, x => completed = true);
+            activator.OnError(error);
+
+            Assert.True(completed);
+        }
+
         [Fact]
         public void Should_Unsubscribe_From_Activator_When_All_Subscriptions_Disposed()
         {

tests/Avalonia.Styling.UnitTests/SelectorTests_Class.cs

@@ -1,6 +1,7 @@
 // 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.Linq;
 using System.Reactive.Linq;
 using System.Threading.Tasks;
@@ -8,6 +9,7 @@ using Moq;
 using Avalonia.Controls;
 using Avalonia.Styling;
 using Xunit;
+using System.Collections.Generic;
 
 namespace Avalonia.Styling.UnitTests
 {
@@ -117,6 +119,28 @@ namespace Avalonia.Styling.UnitTests
             Assert.False(await activator.Take(1));
         }
 
+        [Fact]
+        public void Only_Notifies_When_Result_Changes()
+        {
+            // Test for #1698
+            var control = new Control1
+            {
+                Classes = new Classes { "foo" },
+            };
+
+            var target = default(Selector).Class("foo");
+            var activator = target.Match(control).ObservableResult;
+            var result = new List<bool>();
+
+            using (activator.Subscribe(x => result.Add(x)))
+            {
+                control.Classes.Add("bar");
+                control.Classes.Remove("foo");
+            }
+
+            Assert.Equal(new[] { true, false }, result);
+        }
+
         public class Control1 : TestControlBase
         {
         }