Use custom Rx classes

src/Avalonia.Base/AvaloniaObject.cs

@@ -10,6 +10,7 @@
 using Avalonia.Data;
 using Avalonia.Diagnostics;
 using Avalonia.Logging;
+using Avalonia.Reactive;
 using Avalonia.Threading;
 using Avalonia.Utilities;
 
@@ -38,7 +39,7 @@ public class AvaloniaObject : IAvaloniaObject, IAvaloniaObjectDebug, INotifyProp
         /// 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 @@ public IDisposable Bind(
                     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 @@ private void LogPropertySet(AvaloniaProperty property, object value, BindingPrio
                 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 @@ public static IBinding ToBinding<T>(this IObservable<T> source)
         /// 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 @@ public static IObservable<object> GetObservable(this IAvaloniaObject o, 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 @@ public static ISubject<T> GetSubject<T>(
                 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 @@ public static IDisposable WeakSubscribe(
             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 @@ public static IDisposable WeakSubscribe(
             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 @@ public WeakCollectionChangedObservable(WeakReference<INotifyCollectionChanged> s
 
             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.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 @@ public void OnNext(object value)
                                 "IValueConverter should not return non-errored BindingNotification.");
                         }
 
-                        _errors.OnNext(notification);
+                        PublishNext(notification);
 
                         if (_fallbackValue != AvaloniaProperty.UnsetValue)
                         {
@@ -170,12 +172,18 @@ public void OnNext(object value)
             }
         }
 
-        /// <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 @@ private static BindingNotification Merge(BindingNotification a, BindingNotificat
 
             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);
+            }
+        }
     }
 }