Use lightweight observable for ExpressionObserver.

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/Reactive/LightweightObservableBase.cs

@@ -39,26 +39,34 @@ namespace Avalonia.Reactive
                 return Disposable.Empty;
             }
 
+            var first = _observers.Count == 0;
+
             lock (_observers)
             {
                 _observers.Add(observer);
             }
 
-            if (_observers.Count == 1)
+            if (first)
             {
                 Initialize();
             }
 
-            Subscribed(observer);
+            Subscribed(observer, first);
 
             return Disposable.Create(() =>
             {
-                _observers?.Remove(observer);
-
-                if (_observers?.Count == 0)
+                if (_observers != null)
                 {
-                    Deinitialize();
-                    _observers.TrimExcess();
+                    lock (_observers)
+                    {
+                        _observers?.Remove(observer);
+
+                        if (_observers?.Count == 0)
+                        {
+                            Deinitialize();
+                            _observers.TrimExcess();
+                        }
+                    }
                 }
             });
         }
@@ -68,9 +76,16 @@ namespace Avalonia.Reactive
 
         protected void PublishNext(T value)
         {
-            lock (_observers)
+            if (_observers != null)
             {
-                foreach (var observer in _observers)
+                IObserver<T>[] observers;
+
+                lock (_observers)
+                {
+                    observers = _observers.ToArray();
+                }
+
+                foreach (var observer in observers)
                 {
                     observer.OnNext(value);
                 }
@@ -79,36 +94,48 @@ namespace Avalonia.Reactive
 
         protected void PublishCompleted()
         {
-            lock (_observers)
+            if (_observers != null)
             {
-                foreach (var observer in _observers)
+                IObserver<T>[] observers;
+
+                lock (_observers)
+                {
+                    observers = _observers.ToArray();
+                    _observers = null;
+                }
+
+                foreach (var observer in observers)
                 {
                     observer.OnCompleted();
                 }
 
-                _observers = null;
+                Deinitialize();
             }
-
-            Deinitialize();
         }
 
         protected void PublishError(Exception error)
         {
-            lock (_observers)
+            if (_observers != null)
             {
-                foreach (var observer in _observers)
+                IObserver<T>[] observers;
+
+                lock (_observers)
+                {
+                    observers = _observers.ToArray();
+                    _observers = null;
+                }
+
+                foreach (var observer in observers)
                 {
                     observer.OnError(error);
                 }
 
-                _observers = null;
+                _error = error;
+                Deinitialize();
             }
-
-            _error = error;
-            Deinitialize();
         }
 
-        protected virtual void Subscribed(IObserver<T> observer)
+        protected virtual void Subscribed(IObserver<T> observer, bool first)
         {
         }
     }

src/Avalonia.Styling/Styling/ActivatedValue.cs

@@ -94,9 +94,9 @@ namespace Avalonia.Styling
             _activatorSubscription = Activator.Subscribe(Listener);
         }
 
-        protected override void Subscribed(IObserver<object> observer)
+        protected override void Subscribed(IObserver<object> observer, bool first)
         {
-            if (IsActive == true)
+            if (IsActive == true && !first)
             {
                 observer.OnNext(Value);
             }