Replace System.Reactive with internal extensions

build/Base.props

@@ -1,6 +1,7 @@
 <Project DefaultTargets="Build" xmlns="http://schemas.microsoft.com/developer/msbuild/2003">
   <ItemGroup>
     <PackageReference Include="System.ValueTuple" Version="4.5.0" />
+    <PackageReference Include="System.Threading.Tasks.Extensions" Version="4.5.4" />
     <PackageReference Include="System.Runtime.CompilerServices.Unsafe" Version="4.6.0" />
   </ItemGroup>
 </Project>

src/Avalonia.Base/Animation/Animation.cs

@@ -2,8 +2,7 @@ using System;
 using System.Collections.Generic;
 using System.Diagnostics.CodeAnalysis;
 using System.Linq;
-using System.Reactive.Disposables;
-using System.Reactive.Linq;
+using Avalonia.Reactive;
 using System.Threading;
 using System.Threading.Tasks;
 

src/Avalonia.Base/Animation/AnimationInstance`1.cs

@@ -1,11 +1,8 @@
 using System;
 using System.Linq;
-using System.Reactive.Linq;
+using Avalonia.Reactive;
 using Avalonia.Animation.Animators;
-using Avalonia.Animation.Utils;
 using Avalonia.Data;
-using Avalonia.Reactive;
-using JetBrains.Annotations;
 
 namespace Avalonia.Animation
 {

src/Avalonia.Base/Animation/AnimatorKeyFrame.cs

@@ -63,7 +63,7 @@ namespace Avalonia.Animation
             }
             else
             {
-                return this.Bind(ValueProperty, ObservableEx.SingleValue(value).ToBinding(), targetControl);
+                return this.Bind(ValueProperty, Observable.SingleValue(value).ToBinding(), targetControl);
             }
         }
 

src/Avalonia.Base/Animation/Animators/Animator`1.cs

@@ -1,8 +1,6 @@
 using System;
 using System.Collections.Generic;
 using System.Linq;
-using System.Reactive.Disposables;
-using System.Reactive.Linq;
 using Avalonia.Animation.Utils;
 using Avalonia.Collections;
 using Avalonia.Data;

src/Avalonia.Base/Animation/Animators/ColorAnimator.cs

@@ -2,7 +2,7 @@
 // and adopted from LottieSharp Project (https://github.com/ascora/LottieSharp).
 
 using System;
-using System.Reactive.Disposables;
+using Avalonia.Reactive;
 using Avalonia.Logging;
 using Avalonia.Media;
 

src/Avalonia.Base/Animation/Animators/TransformAnimator.cs

@@ -1,5 +1,5 @@
 using System;
-using System.Reactive.Disposables;
+using Avalonia.Reactive;
 using Avalonia.Logging;
 using Avalonia.Media;
 using Avalonia.Media.Transformation;

src/Avalonia.Base/Animation/Clock.cs

@@ -1,4 +1,5 @@
 using System;
+using Avalonia.Reactive;
 
 namespace Avalonia.Animation
 {

src/Avalonia.Base/Animation/CrossFade.cs

@@ -1,6 +1,6 @@
 using System;
 using System.Collections.Generic;
-using System.Reactive.Disposables;
+using Avalonia.Reactive;
 using System.Threading;
 using System.Threading.Tasks;
 using Avalonia.Animation.Easings;
@@ -108,7 +108,7 @@ namespace Avalonia.Animation
             }
 
             var tasks = new List<Task>();
-            using (var disposables = new CompositeDisposable())
+            using (var disposables = new CompositeDisposable(1))
             {
                 if (to != null)
                 {

src/Avalonia.Base/Avalonia.Base.csproj

@@ -14,7 +14,6 @@
   </ItemGroup>
   <Import Project="..\..\build\Base.props" />
   <Import Project="..\..\build\Binding.props" />
-  <Import Project="..\..\build\Rx.props" />
   <Import Project="..\..\build\JetBrains.Annotations.props" />
   <Import Project="..\..\build\System.Memory.props" />
   <Import Project="..\..\build\ApiDiff.props" />
@@ -35,6 +34,12 @@
     <InternalsVisibleTo Include="Avalonia.Markup.Xaml, PublicKey=$(AvaloniaPublicKey)" />
     <InternalsVisibleTo Include="Avalonia.Controls.ColorPicker, PublicKey=$(AvaloniaPublicKey)" />
     <InternalsVisibleTo Include="Avalonia.Controls.DataGrid, PublicKey=$(AvaloniaPublicKey)" />
+    <InternalsVisibleTo Include="Avalonia.Headless, PublicKey=$(AvaloniaPublicKey)" />
+    <InternalsVisibleTo Include="Avalonia.Native, PublicKey=$(AvaloniaPublicKey)" />
+    <InternalsVisibleTo Include="Avalonia.FreeDesktop, PublicKey=$(AvaloniaPublicKey)" />
+    <InternalsVisibleTo Include="Avalonia.X11, PublicKey=$(AvaloniaPublicKey)" />
+    <InternalsVisibleTo Include="Avalonia.OpenGL, PublicKey=$(AvaloniaPublicKey)" />
+    <InternalsVisibleTo Include="Avalonia.Skia, PublicKey=$(AvaloniaPublicKey)" />
     <InternalsVisibleTo Include="Avalonia.Controls.UnitTests, PublicKey=$(AvaloniaPublicKey)" />
     <InternalsVisibleTo Include="Avalonia.DesignerSupport, PublicKey=$(AvaloniaPublicKey)" />
     <InternalsVisibleTo Include="Avalonia.Direct2D1.RenderTests, PublicKey=$(AvaloniaPublicKey)" />

src/Avalonia.Base/AvaloniaObjectExtensions.cs

@@ -1,10 +1,6 @@
 using System;
-using System.Reactive;
-using System.Reactive.Disposables;
-using System.Reactive.Linq;
-using System.Reactive.Subjects;
-using Avalonia.Data;
 using Avalonia.Reactive;
+using Avalonia.Data;
 
 namespace Avalonia
 {
@@ -127,108 +123,6 @@ namespace Avalonia
                 property ?? throw new ArgumentNullException(nameof(property)));
         }
 
-        /// <summary>
-        /// Gets a subject for an <see cref="AvaloniaProperty"/>.
-        /// </summary>
-        /// <param name="o">The object.</param>
-        /// <param name="property">The property.</param>
-        /// <param name="priority">
-        /// The priority with which binding values are written to the object.
-        /// </param>
-        /// <returns>
-        /// An <see cref="ISubject{Object}"/> which can be used for two-way binding to/from the 
-        /// property.
-        /// </returns>
-        public static ISubject<object?> GetSubject(
-            this AvaloniaObject o,
-            AvaloniaProperty property,
-            BindingPriority priority = BindingPriority.LocalValue)
-        {
-            return Subject.Create<object?>(
-                Observer.Create<object?>(x => o.SetValue(property, x, priority)),
-                o.GetObservable(property));
-        }
-
-        /// <summary>
-        /// Gets a subject for an <see cref="AvaloniaProperty"/>.
-        /// </summary>
-        /// <typeparam name="T">The property type.</typeparam>
-        /// <param name="o">The object.</param>
-        /// <param name="property">The property.</param>
-        /// <param name="priority">
-        /// The priority with which binding values are written to the object.
-        /// </param>
-        /// <returns>
-        /// An <see cref="ISubject{T}"/> which can be used for two-way binding to/from the 
-        /// property.
-        /// </returns>
-        public static ISubject<T> GetSubject<T>(
-            this AvaloniaObject o,
-            AvaloniaProperty<T> property,
-            BindingPriority priority = BindingPriority.LocalValue)
-        {
-            return Subject.Create<T>(
-                Observer.Create<T>(x => o.SetValue(property, x, priority)),
-                o.GetObservable(property));
-        }
-
-        /// <summary>
-        /// Gets a subject for a <see cref="AvaloniaProperty"/>.
-        /// </summary>
-        /// <param name="o">The object.</param>
-        /// <param name="property">The property.</param>
-        /// <param name="priority">
-        /// The priority with which binding values are written to the object.
-        /// </param>
-        /// <returns>
-        /// An <see cref="ISubject{Object}"/> which can be used for two-way binding to/from the 
-        /// property.
-        /// </returns>
-        public static ISubject<BindingValue<object?>> GetBindingSubject(
-            this AvaloniaObject o,
-            AvaloniaProperty property,
-            BindingPriority priority = BindingPriority.LocalValue)
-        {
-            return Subject.Create<BindingValue<object?>>(
-                Observer.Create<BindingValue<object?>>(x =>
-                {
-                    if (x.HasValue)
-                    {
-                        o.SetValue(property, x.Value, priority);
-                    }
-                }),
-                o.GetBindingObservable(property));
-        }
-
-        /// <summary>
-        /// Gets a subject for a <see cref="AvaloniaProperty"/>.
-        /// </summary>
-        /// <typeparam name="T">The property type.</typeparam>
-        /// <param name="o">The object.</param>
-        /// <param name="property">The property.</param>
-        /// <param name="priority">
-        /// The priority with which binding values are written to the object.
-        /// </param>
-        /// <returns>
-        /// An <see cref="ISubject{T}"/> which can be used for two-way binding to/from the 
-        /// property.
-        /// </returns>
-        public static ISubject<BindingValue<T>> GetBindingSubject<T>(
-            this AvaloniaObject o,
-            AvaloniaProperty<T> property,
-            BindingPriority priority = BindingPriority.LocalValue)
-        {
-            return Subject.Create<BindingValue<T>>(
-                Observer.Create<BindingValue<T>>(x =>
-                {
-                    if (x.HasValue)
-                    {
-                        o.SetValue(property, x.Value, priority);
-                    }
-                }),
-                o.GetBindingObservable(property));
-        }
-
         /// <summary>
         /// Binds an <see cref="AvaloniaProperty"/> to an observable.
         /// </summary>
@@ -407,13 +301,7 @@ namespace Avalonia
             Action<TTarget, AvaloniaPropertyChangedEventArgs> action)
             where TTarget : AvaloniaObject
         {
-            return observable.Subscribe(e =>
-            {
-                if (e.Sender is TTarget target)
-                {
-                    action(target, e);
-                }
-            });
+            return observable.Subscribe(new ClassHandlerObserver<TTarget>(action));
         }
 
         /// <summary>
@@ -431,13 +319,7 @@ namespace Avalonia
             this IObservable<AvaloniaPropertyChangedEventArgs<TValue>> observable,
             Action<TTarget, AvaloniaPropertyChangedEventArgs<TValue>> action) where TTarget : AvaloniaObject
         {
-            return observable.Subscribe(e =>
-            {
-                if (e.Sender is TTarget target)
-                {
-                    action(target, e);
-                }
-            });
+            return observable.Subscribe(new ClassHandlerObserver<TTarget, TValue>(action));
         }
 
         private class BindingAdaptor : IBinding
@@ -458,5 +340,57 @@ namespace Avalonia
                 return InstancedBinding.OneWay(_source);
             }
         }
+        
+        private class ClassHandlerObserver<TTarget, TValue> : IObserver<AvaloniaPropertyChangedEventArgs<TValue>>
+        {
+            private readonly Action<TTarget, AvaloniaPropertyChangedEventArgs<TValue>> _action;
+
+            public ClassHandlerObserver(Action<TTarget, AvaloniaPropertyChangedEventArgs<TValue>> action)
+            {
+                _action = action;
+            }
+
+            public void OnCompleted()
+            {
+            }
+
+            public void OnError(Exception error)
+            {
+            }
+
+            public void OnNext(AvaloniaPropertyChangedEventArgs<TValue> value)
+            {
+                if (value.Sender is TTarget target)
+                {
+                    _action(target, value);
+                }
+            }
+        }
+
+        private class ClassHandlerObserver<TTarget> : IObserver<AvaloniaPropertyChangedEventArgs>
+        {
+            private readonly Action<TTarget, AvaloniaPropertyChangedEventArgs> _action;
+
+            public ClassHandlerObserver(Action<TTarget, AvaloniaPropertyChangedEventArgs> action)
+            {
+                _action = action;
+            }
+
+            public void OnCompleted()
+            {
+            }
+
+            public void OnError(Exception error)
+            {
+            }
+
+            public void OnNext(AvaloniaPropertyChangedEventArgs value)
+            {
+                if (value.Sender is TTarget target)
+                {
+                    _action(target, value);
+                }
+            }
+        }
     }
 }

src/Avalonia.Base/AvaloniaProperty`1.cs

@@ -1,7 +1,7 @@
 using System;
 using System.Diagnostics.CodeAnalysis;
-using System.Reactive.Subjects;
 using Avalonia.Data;
+using Avalonia.Reactive;
 using Avalonia.Utilities;
 
 namespace Avalonia
@@ -12,7 +12,7 @@ namespace Avalonia
     /// <typeparam name="TValue">The value type of the property.</typeparam>
     public abstract class AvaloniaProperty<TValue> : AvaloniaProperty
     {
-        private readonly Subject<AvaloniaPropertyChangedEventArgs<TValue>> _changed;
+        private readonly LightweightSubject<AvaloniaPropertyChangedEventArgs<TValue>> _changed;
 
         /// <summary>
         /// Initializes a new instance of the <see cref="AvaloniaProperty{TValue}"/> class.
@@ -28,7 +28,7 @@ namespace Avalonia
             Action<AvaloniaObject, bool>? notifying = null)
             : base(name, typeof(TValue), ownerType, metadata, notifying)
         {
-            _changed = new Subject<AvaloniaPropertyChangedEventArgs<TValue>>();
+            _changed = new LightweightSubject<AvaloniaPropertyChangedEventArgs<TValue>>();
         }
 
         /// <summary>

src/Avalonia.Base/ClassBindingManager.cs

@@ -1,6 +1,7 @@
 using System;
 using System.Collections.Generic;
 using Avalonia.Data;
+using Avalonia.Reactive;
 
 namespace Avalonia
 {

src/Avalonia.Base/Collections/AvaloniaListExtensions.cs

@@ -3,7 +3,7 @@ using System.Collections;
 using System.Collections.Generic;
 using System.Collections.Specialized;
 using System.ComponentModel;
-using System.Reactive.Disposables;
+using Avalonia.Reactive;
 
 namespace Avalonia.Collections
 {

src/Avalonia.Base/Collections/NotifyCollectionChangedExtensions.cs

@@ -1,6 +1,5 @@
 using System;
 using System.Collections.Specialized;
-using System.Reactive.Linq;
 using Avalonia.Reactive;
 using Avalonia.Utilities;
 

src/Avalonia.Base/Controls/NameScopeLocator.cs

@@ -1,5 +1,5 @@
 using System;
-using System.Reactive.Disposables;
+using Avalonia.Reactive;
 using Avalonia.Utilities;
 
 namespace Avalonia.Controls

src/Avalonia.Base/Data/BindingOperations.cs

@@ -1,6 +1,5 @@
 using System;
-using System.Reactive.Disposables;
-using System.Reactive.Linq;
+using Avalonia.Reactive;
 
 namespace Avalonia.Data
 {
@@ -46,15 +45,15 @@ namespace Avalonia.Data
                         throw new InvalidOperationException("InstancedBinding does not contain an observable.");
                     return target.Bind(property, binding.Observable, binding.Priority);
                 case BindingMode.TwoWay:
+                    if (binding.Observable is null)
+                        throw new InvalidOperationException("InstancedBinding does not contain an observable.");
                     if (binding.Subject is null)
                         throw new InvalidOperationException("InstancedBinding does not contain a subject.");
                     return new TwoWayBindingDisposable(
-                        target.Bind(property, binding.Subject, binding.Priority),
+                        target.Bind(property, binding.Observable, binding.Priority),
                         target.GetObservable(property).Subscribe(binding.Subject));
                 case BindingMode.OneTime:
-                    var source = binding.Subject ?? binding.Observable;
-
-                    if (source != null)
+                    if (binding.Observable is {} source)
                     {
                         // Perf: Avoid allocating closure in the outer scope.
                         var targetCopy = target;

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

@@ -1,5 +1,4 @@
 using System;
-using System.Reactive.Linq;
 using Avalonia.Reactive;
 
 namespace Avalonia.Data.Core

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

@@ -1,11 +1,9 @@
 using System;
 using System.Diagnostics.CodeAnalysis;
 using System.Globalization;
-using System.Reactive.Linq;
-using System.Reactive.Subjects;
+using Avalonia.Reactive;
 using Avalonia.Data.Converters;
 using Avalonia.Logging;
-using Avalonia.Reactive;
 using Avalonia.Utilities;
 
 namespace Avalonia.Data.Core
@@ -15,7 +13,7 @@ namespace Avalonia.Data.Core
     /// that are sent and received.
     /// </summary>
     [RequiresUnreferencedCode(TrimmingMessages.TypeConvertionRequiresUnreferencedCodeMessage)]
-    public class BindingExpression : LightweightObservableBase<object?>, ISubject<object?>, IDescription
+    public class BindingExpression : LightweightObservableBase<object?>, IAvaloniaSubject<object?>, IDescription
     {
         private readonly ExpressionObserver _inner;
         private readonly Type _targetType;

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

@@ -2,8 +2,6 @@ using System;
 using System.Collections.Generic;
 using System.Diagnostics.CodeAnalysis;
 using System.Linq.Expressions;
-using System.Reactive;
-using System.Reactive.Linq;
 using Avalonia.Data.Core.Parsers;
 using Avalonia.Data.Core.Plugins;
 using Avalonia.Reactive;
@@ -99,14 +97,14 @@ namespace Avalonia.Data.Core
         /// </summary>
         /// <param name="rootGetter">A function which gets the root object.</param>
         /// <param name="node">The expression.</param>
-        /// <param name="update">An observable which triggers a re-read of the getter.</param>
+        /// <param name="update">An observable which triggers a re-read of the getter. Generic argument value is not used.</param>
         /// <param name="description">
         /// A description of the expression.
         /// </param>
         public ExpressionObserver(
             Func<object?> rootGetter,
             ExpressionNode node,
-            IObservable<Unit> update,
+            IObservable<object> update,
             string? description)
         {
             Description = description;
@@ -164,7 +162,7 @@ namespace Avalonia.Data.Core
         /// </summary>
         /// <param name="rootGetter">A function which gets the root object.</param>
         /// <param name="expression">The expression.</param>
-        /// <param name="update">An observable which triggers a re-read of the getter.</param>
+        /// <param name="update">An observable which triggers a re-read of the getter. Generic argument value is not used.</param>
         /// <param name="enableDataValidation">Whether or not to track data validation</param>
         /// <param name="description">
         /// A description of the expression. If null, <paramref name="expression"/>'s string representation will be used.
@@ -173,7 +171,7 @@ namespace Avalonia.Data.Core
         public static ExpressionObserver Create<T, U>(
             Func<T> rootGetter,
             Expression<Func<T, U>> expression,
-            IObservable<Unit> update,
+            IObservable<object> update,
             bool enableDataValidation = false,
             string? description = null)
         {
@@ -296,9 +294,10 @@ namespace Avalonia.Data.Core
             if (_root is IObservable<object> observable)
             {
                 _rootSubscription = observable.Subscribe(
-                    x => _node.Target = new WeakReference<object?>(x != AvaloniaProperty.UnsetValue ? x : null),
-                    x => PublishCompleted(),
-                    () => PublishCompleted());
+                    new AnonymousObserver<object>(
+                        x => _node.Target = new WeakReference<object?>(x != AvaloniaProperty.UnsetValue ? x : null),
+                        x => PublishCompleted(),
+                        PublishCompleted));
             }
             else
             {

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

@@ -1,48 +1,47 @@
 using System;
 using System.Collections;
-using System.Collections.Generic;
 using System.Collections.Specialized;
 using System.ComponentModel;
-using System.Linq;
-using System.Reactive.Linq;
+using Avalonia.Reactive;
 using Avalonia.Utilities;
 
 namespace Avalonia.Data.Core
 {
-    public abstract class IndexerNodeBase : SettableNode
+    public abstract class IndexerNodeBase : SettableNode,
+        IWeakEventSubscriber<NotifyCollectionChangedEventArgs>,
+        IWeakEventSubscriber<PropertyChangedEventArgs>
     {
-        private IDisposable? _subscription;
-        
         protected override void StartListeningCore(WeakReference<object?> reference)
         {
             reference.TryGetTarget(out var target);
 
-            var incc = target as INotifyCollectionChanged;
-            var inpc = target as INotifyPropertyChanged;
-            var inputs = new List<IObservable<object?>>();
-
-            if (incc != null)
+            if (target is INotifyCollectionChanged incc)
             {
-                inputs.Add(WeakObservable.FromEventPattern(
-                    incc, WeakEvents.CollectionChanged)
-                    .Where(x => ShouldUpdate(x.Sender, x.EventArgs))
-                    .Select(_ => GetValue(target)));
+                WeakEvents.CollectionChanged.Subscribe(incc, this);
             }
 
-            if (inpc != null)
+            if (target is INotifyPropertyChanged inpc)
             {
-                inputs.Add(WeakObservable.FromEventPattern(
-                    inpc, WeakEvents.PropertyChanged)
-                    .Where(x => ShouldUpdate(x.Sender, x.EventArgs))
-                    .Select(_ => GetValue(target)));
+                WeakEvents.PropertyChanged.Subscribe(inpc, this);
             }
-
-            _subscription = Observable.Merge(inputs).StartWith(GetValue(target)).Subscribe(ValueChanged);
+            
+            ValueChanged(GetValue(target));
         }
 
         protected override void StopListeningCore()
         {
-            _subscription?.Dispose();
+            if (Target.TryGetTarget(out var target))
+            {
+                if (target is INotifyCollectionChanged incc)
+                {
+                    WeakEvents.CollectionChanged.Unsubscribe(incc, this);
+                }
+
+                if (target is INotifyPropertyChanged inpc)
+                {
+                    WeakEvents.PropertyChanged.Unsubscribe(inpc, this);
+                }
+            }
         }
 
         protected abstract object? GetValue(object? target);
@@ -83,5 +82,21 @@ namespace Avalonia.Data.Core
         }
 
         protected abstract bool ShouldUpdate(object? sender, PropertyChangedEventArgs e);
+
+        void IWeakEventSubscriber<NotifyCollectionChangedEventArgs>.OnEvent(object? sender, WeakEvent ev, NotifyCollectionChangedEventArgs e)
+        {
+            if (ShouldUpdate(sender, e))
+            {
+                ValueChanged(GetValue(sender));
+            }
+        }
+
+        void IWeakEventSubscriber<PropertyChangedEventArgs>.OnEvent(object? sender, WeakEvent ev, PropertyChangedEventArgs e)
+        {
+            if (ShouldUpdate(sender, e))
+            {
+                ValueChanged(GetValue(sender));
+            }
+        }
     }
 }

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

@@ -1,7 +1,7 @@
 using System;
 using System.Diagnostics.CodeAnalysis;
 using System.Linq;
-using System.Reactive.Linq;
+using Avalonia.Reactive;
 using System.Reflection;
 
 namespace Avalonia.Data.Core.Plugins
@@ -12,8 +12,15 @@ namespace Avalonia.Data.Core.Plugins
     [UnconditionalSuppressMessage("Trimming", "IL3050", Justification = TrimmingMessages.IgnoreNativeAotSupressWarningMessage)]
     public class ObservableStreamPlugin : IStreamPlugin
     {
-        static MethodInfo? observableSelect;
+        private static MethodInfo? s_observableGeneric;
+        private static MethodInfo? s_observableSelect;
 
+        [DynamicDependency(DynamicallyAccessedMemberTypes.NonPublicProperties, "Avalonia.Data.Core.Plugins.ObservableStreamPlugin", "Avalonia.Base")]
+        public ObservableStreamPlugin()
+        {
+            
+        }
+        
         /// <summary>
         /// Checks whether this plugin handles the specified value.
         /// </summary>
@@ -54,56 +61,32 @@ namespace Avalonia.Data.Core.Plugins
                   x.IsGenericType &&
                   x.GetGenericTypeDefinition() == typeof(IObservable<>)).GetGenericArguments()[0];
 
-            // Get the Observable.Select method.
-            var select = GetObservableSelect(sourceType);
-
-            // Make a Box<> delegate of the correct type.
-            var funcType = typeof(Func<,>).MakeGenericType(sourceType, typeof(object));
-            var box = GetType().GetMethod(nameof(Box), BindingFlags.Static | BindingFlags.NonPublic)!
-                .MakeGenericMethod(sourceType)
-                .CreateDelegate(funcType);
+            // Get the BoxObservable<T> method.
+            var select = GetBoxObservable(sourceType);
 
-            // Call Observable.Select(target, box);
+            // Call BoxObservable(target);
             return (IObservable<object?>)select.Invoke(
                 null,
-                new object[] { target, box })!;
+                new[] { target })!;
         }
 
         [RequiresUnreferencedCode(TrimmingMessages.StreamPluginRequiresUnreferencedCodeMessage)]
-        private static MethodInfo GetObservableSelect(Type source)
+        private static MethodInfo GetBoxObservable(Type source)
         {
-            return GetObservableSelect().MakeGenericMethod(source, typeof(object));
+            return (s_observableGeneric ??= GetBoxObservable()).MakeGenericMethod(source);
         }
 
-        private static MethodInfo GetObservableSelect()
+        [RequiresUnreferencedCode(TrimmingMessages.StreamPluginRequiresUnreferencedCodeMessage)]
+        private static MethodInfo GetBoxObservable()
         {
-            if (observableSelect == null)
-            {
-                observableSelect = typeof(Observable).GetRuntimeMethods().First(x =>
-                {
-                    if (x.Name == nameof(Observable.Select) &&
-                        x.ContainsGenericParameters &&
-                        x.GetGenericArguments().Length == 2)
-                    {
-                        var parameters = x.GetParameters();
-
-                        if (parameters.Length == 2 &&
-                            parameters[0].ParameterType.IsConstructedGenericType &&
-                            parameters[0].ParameterType.GetGenericTypeDefinition() == typeof(IObservable<>) &&
-                            parameters[1].ParameterType.IsConstructedGenericType &&
-                            parameters[1].ParameterType.GetGenericTypeDefinition() == typeof(Func<,>))
-                        {
-                            return true;
-                        }
-                    }
-
-                    return false;
-                });
-            }
-
-            return observableSelect;
+            return s_observableSelect
+               ??= typeof(ObservableStreamPlugin).GetMethod(nameof(BoxObservable), BindingFlags.Static | BindingFlags.NonPublic)
+               ?? throw new InvalidOperationException("BoxObservable method was not found.");
         }
 
-        private static object? Box<T>(T value) => (object?)value;
+        private static IObservable<object?> BoxObservable<T>(IObservable<T> source)
+        {
+            return source.Select(v => (object?)v);
+        }
     }
 }

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

@@ -1,9 +1,8 @@
 using System;
 using System.Diagnostics.CodeAnalysis;
-using System.Reactive.Linq;
-using System.Reactive.Subjects;
 using System.Reflection;
 using System.Threading.Tasks;
+using Avalonia.Reactive;
 
 namespace Avalonia.Data.Core.Plugins
 {
@@ -50,7 +49,7 @@ namespace Avalonia.Data.Core.Plugins
                         case TaskStatus.Faulted:
                             return HandleCompleted(task);
                         default:
-                            var subject = new Subject<object?>();
+                            var subject = new LightweightSubject<object?>();
                             task.ContinueWith(
                                     x => HandleCompleted(task).Subscribe(subject),
                                     TaskScheduler.FromCurrentSynchronizationContext())

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

@@ -1,7 +1,7 @@
 using System;
 using System.Diagnostics.CodeAnalysis;
-using System.Reactive.Linq;
 using Avalonia.Data.Core.Plugins;
+using Avalonia.Reactive;
 
 namespace Avalonia.Data.Core
 {

src/Avalonia.Base/Data/IndexerBinding.cs

@@ -1,4 +1,6 @@
-namespace Avalonia.Data
+using Avalonia.Reactive;
+
+namespace Avalonia.Data
 {
     public class IndexerBinding : IBinding
     {
@@ -22,7 +24,10 @@
             object? anchor = null,
             bool enableDataValidation = false)
         {
-            return new InstancedBinding(Source.GetSubject(Property), Mode, BindingPriority.LocalValue);
+            var subject = new CombinedSubject<object?>(
+                new AnonymousObserver<object?>(x => Source.SetValue(Property, x, BindingPriority.LocalValue)),
+                Source.GetObservable(Property));
+            return new InstancedBinding(subject, Mode, BindingPriority.LocalValue);
         }
     }
 }

src/Avalonia.Base/Data/IndexerDescriptor.cs

@@ -1,12 +1,11 @@
 using System;
-using System.Reactive;
 
 namespace Avalonia.Data
 {
     /// <summary>
     /// Holds a description of a binding for <see cref="AvaloniaObject"/>'s [] operator.
     /// </summary>
-    public class IndexerDescriptor : ObservableBase<object?>, IDescription
+    public class IndexerDescriptor : IObservable<object?>, IDescription
     {
         /// <summary>
         /// Gets or sets the binding mode.
@@ -104,7 +103,7 @@ namespace Avalonia.Data
         }
 
         /// <inheritdoc/>
-        protected override IDisposable SubscribeCore(IObserver<object?> observer)
+        public IDisposable Subscribe(IObserver<object?> observer)
         {
             if (SourceObservable is null && Source is null)
                 throw new InvalidOperationException("Cannot subscribe to IndexerDescriptor.");

src/Avalonia.Base/Data/InstancedBinding.cs

@@ -1,5 +1,5 @@
 using System;
-using System.Reactive.Subjects;
+using Avalonia.Reactive;
 
 namespace Avalonia.Data
 {
@@ -14,28 +14,7 @@ namespace Avalonia.Data
     /// </remarks>
     public class InstancedBinding
     {
-        /// <summary>
-        /// Initializes a new instance of the <see cref="InstancedBinding"/> class.
-        /// </summary>
-        /// <param name="subject">The binding source.</param>
-        /// <param name="mode">The binding mode.</param>
-        /// <param name="priority">The priority of the binding.</param>
-        /// <remarks>
-        /// This constructor can be used to create any type of binding and as such requires an
-        /// <see cref="ISubject{Object}"/> as the binding source because this is the only binding
-        /// source which can be used for all binding modes. If you wish to create an instance with
-        /// something other than a subject, use one of the static creation methods on this class.
-        /// </remarks>
-        public InstancedBinding(ISubject<object?> subject, BindingMode mode, BindingPriority priority)
-        {
-            Contract.Requires<ArgumentNullException>(subject != null);
-
-            Mode = mode;
-            Priority = priority;
-            Value = subject;
-        }
-
-        private InstancedBinding(object? value, BindingMode mode, BindingPriority priority)
+        internal InstancedBinding(object? value, BindingMode mode, BindingPriority priority)
         {
             Mode = mode;
             Priority = priority;
@@ -63,9 +42,14 @@ namespace Avalonia.Data
         public IObservable<object?>? Observable => Value as IObservable<object?>;
 
         /// <summary>
-        /// Gets the <see cref="Value"/> as a subject.
+        /// Gets the <see cref="Value"/> as an observer.
         /// </summary>
-        public ISubject<object?>? Subject => Value as ISubject<object?>;
+        public IObserver<object?>? Observer => Value as IObserver<object?>;
+
+        /// <summary>
+        /// Gets the <see cref="Subject"/> as an subject.
+        /// </summary>
+        internal IAvaloniaSubject<object?>? Subject => Value as IAvaloniaSubject<object?>;
 
         /// <summary>
         /// Creates a new one-time binding with a fixed value.
@@ -113,30 +97,34 @@ namespace Avalonia.Data
         /// <summary>
         /// Creates a new one-way to source binding.
         /// </summary>
-        /// <param name="subject">The binding source.</param>
+        /// <param name="observer">The binding source.</param>
         /// <param name="priority">The priority of the binding.</param>
         /// <returns>An <see cref="InstancedBinding"/> instance.</returns>
         public static InstancedBinding OneWayToSource(
-            ISubject<object?> subject,
+            IObserver<object?> observer,
             BindingPriority priority = BindingPriority.LocalValue)
         {
-            _ = subject ?? throw new ArgumentNullException(nameof(subject));
+            _ = observer ?? throw new ArgumentNullException(nameof(observer));
 
-            return new InstancedBinding(subject, BindingMode.OneWayToSource, priority);
+            return new InstancedBinding(observer, BindingMode.OneWayToSource, priority);
         }
 
         /// <summary>
         /// Creates a new two-way binding.
         /// </summary>
-        /// <param name="subject">The binding source.</param>
+        /// <param name="observable">The binding source.</param>
+        /// <param name="observer">The binding source.</param>
         /// <param name="priority">The priority of the binding.</param>
         /// <returns>An <see cref="InstancedBinding"/> instance.</returns>
         public static InstancedBinding TwoWay(
-            ISubject<object?> subject,
+            IObservable<object?> observable,
+            IObserver<object?> observer,
             BindingPriority priority = BindingPriority.LocalValue)
         {
-            _ = subject ?? throw new ArgumentNullException(nameof(subject));
+            _ = observable ?? throw new ArgumentNullException(nameof(observable));
+            _ = observer ?? throw new ArgumentNullException(nameof(observer));
 
+            var subject = new CombinedSubject<object?>(observer, observable);
             return new InstancedBinding(subject, BindingMode.TwoWay, priority);
         }
 

src/Avalonia.Base/Input/Gestures.cs

@@ -1,6 +1,7 @@
 using System;
 using Avalonia.Interactivity;
 using Avalonia.Platform;
+using Avalonia.Reactive;
 using Avalonia.VisualTree;
 
 namespace Avalonia.Input

src/Avalonia.Base/Input/InputElement.cs

@@ -7,6 +7,7 @@ using Avalonia.Data;
 using Avalonia.Input.GestureRecognizers;
 using Avalonia.Input.TextInput;
 using Avalonia.Interactivity;
+using Avalonia.Reactive;
 using Avalonia.VisualTree;
 
 #nullable enable

src/Avalonia.Base/Input/InputManager.cs

@@ -1,6 +1,6 @@
 using System;
-using System.Reactive.Subjects;
 using Avalonia.Input.Raw;
+using Avalonia.Reactive;
 
 namespace Avalonia.Input
 {
@@ -10,9 +10,9 @@ namespace Avalonia.Input
     /// </summary>
     public class InputManager : IInputManager
     {
-        private readonly Subject<RawInputEventArgs> _preProcess = new Subject<RawInputEventArgs>();
-        private readonly Subject<RawInputEventArgs> _process = new Subject<RawInputEventArgs>();
-        private readonly Subject<RawInputEventArgs> _postProcess = new Subject<RawInputEventArgs>();
+        private readonly LightweightSubject<RawInputEventArgs> _preProcess = new();
+        private readonly LightweightSubject<RawInputEventArgs> _process = new();
+        private readonly LightweightSubject<RawInputEventArgs> _postProcess = new();
 
         /// <summary>
         /// Gets the global instance of the input manager.

src/Avalonia.Base/Input/MouseDevice.cs

@@ -1,5 +1,6 @@
 using System;
 using System.Collections.Generic;
+using Avalonia.Reactive;
 using Avalonia.Input.Raw;
 using Avalonia.Platform;
 using Avalonia.Utilities;

src/Avalonia.Base/Input/TextInput/InputMethodManager.cs

@@ -1,5 +1,5 @@
 using System;
-using Avalonia.VisualTree;
+using Avalonia.Reactive;
 
 namespace Avalonia.Input.TextInput
 {

src/Avalonia.Base/Interactivity/InteractiveExtensions.cs

@@ -1,6 +1,5 @@
 using System;
-using System.Reactive.Disposables;
-using System.Reactive.Linq;
+using Avalonia.Reactive;
 
 namespace Avalonia.Interactivity
 {

src/Avalonia.Base/Interactivity/RoutedEvent.cs

@@ -1,5 +1,5 @@
 using System;
-using System.Reactive.Subjects;
+using Avalonia.Reactive;
 
 namespace Avalonia.Interactivity
 {
@@ -13,8 +13,8 @@ namespace Avalonia.Interactivity
 
     public class RoutedEvent
     {
-        private readonly Subject<(object, RoutedEventArgs)> _raised = new Subject<(object, RoutedEventArgs)>();
-        private readonly Subject<RoutedEventArgs> _routeFinished = new Subject<RoutedEventArgs>();
+        private readonly LightweightSubject<(object, RoutedEventArgs)> _raised = new();
+        private readonly LightweightSubject<RoutedEventArgs> _routeFinished = new();
 
         public RoutedEvent(
             string name,

src/Avalonia.Base/Layout/Layoutable.cs

@@ -1,6 +1,6 @@
 using System;
 using Avalonia.Logging;
-using Avalonia.Styling;
+using Avalonia.Reactive;
 using Avalonia.VisualTree;
 
 #nullable enable

src/Avalonia.Base/Media/Brush.cs

@@ -3,6 +3,7 @@ using System.ComponentModel;
 using Avalonia.Animation;
 using Avalonia.Animation.Animators;
 using Avalonia.Media.Immutable;
+using Avalonia.Reactive;
 
 namespace Avalonia.Media
 {

src/Avalonia.Base/Media/DashStyle.cs

@@ -4,6 +4,7 @@ using System.Collections.Specialized;
 using Avalonia.Animation;
 using Avalonia.Collections;
 using Avalonia.Media.Immutable;
+using Avalonia.Reactive;
 
 #nullable enable
 

src/Avalonia.Base/Media/ExperimentalAcrylicMaterial.cs

@@ -1,4 +1,5 @@
 using System;
+using Avalonia.Reactive;
 
 namespace Avalonia.Media
 {

src/Avalonia.Base/Media/Geometry.cs

@@ -1,5 +1,6 @@
 using System;
 using Avalonia.Platform;
+using Avalonia.Reactive;
 
 namespace Avalonia.Media
 {

src/Avalonia.Base/Media/GradientBrush.cs

@@ -6,6 +6,7 @@ using System.ComponentModel;
 using Avalonia.Animation.Animators;
 using Avalonia.Collections;
 using Avalonia.Metadata;
+using Avalonia.Reactive;
 
 namespace Avalonia.Media
 {

src/Avalonia.Base/Media/MatrixTransform.cs

@@ -1,4 +1,5 @@
 using System;
+using Avalonia.Reactive;
 using Avalonia.VisualTree;
 
 namespace Avalonia.Media

src/Avalonia.Base/Media/RotateTransform.cs

@@ -1,4 +1,5 @@
 using System;
+using Avalonia.Reactive;
 using Avalonia.VisualTree;
 
 namespace Avalonia.Media

src/Avalonia.Base/Media/ScaleTransform.cs

@@ -1,4 +1,5 @@
 using System;
+using Avalonia.Reactive;
 using Avalonia.VisualTree;
 
 namespace Avalonia.Media

src/Avalonia.Base/Media/SkewTransform.cs

@@ -1,4 +1,5 @@
 using System;
+using Avalonia.Reactive;
 using Avalonia.VisualTree;
 
 namespace Avalonia.Media

src/Avalonia.Base/Media/TranslateTransform.cs

@@ -1,4 +1,5 @@
 using System;
+using Avalonia.Reactive;
 using Avalonia.VisualTree;
 
 namespace Avalonia.Media

src/Avalonia.Base/PropertyStore/BindingEntryBase.cs

@@ -1,6 +1,6 @@
 using System;
 using System.Collections.Generic;
-using System.Reactive.Disposables;
+using Avalonia.Reactive;
 using Avalonia.Data;
 using Avalonia.Threading;
 

src/Avalonia.Base/Reactive/AnonymousObserver.cs

@@ -0,0 +1,62 @@
+using System;
+using System.Threading.Tasks;
+
+namespace Avalonia.Reactive;
+
+internal class AnonymousObserver<T> : IObserver<T>
+{
+    private static readonly Action<Exception> ThrowsOnError = ex => throw ex;
+    private static readonly Action NoOpCompleted = () => { };  
+    private readonly Action<T> _onNext;
+    private readonly Action<Exception> _onError;
+    private readonly Action _onCompleted;
+
+    public AnonymousObserver(TaskCompletionSource<T> tcs)
+    {
+        if (tcs is null)
+        {
+            throw new ArgumentNullException(nameof(tcs));
+        }
+
+        _onNext = tcs.SetResult;
+        _onError = tcs.SetException;
+        _onCompleted = NoOpCompleted;
+    }
+    
+    public AnonymousObserver(Action<T> onNext, Action<Exception> onError, Action onCompleted)
+    {
+        _onNext = onNext ?? throw new ArgumentNullException(nameof(onNext));
+        _onError = onError ?? throw new ArgumentNullException(nameof(onError));
+        _onCompleted = onCompleted ?? throw new ArgumentNullException(nameof(onCompleted));
+    }
+
+    public AnonymousObserver(Action<T> onNext)
+        : this(onNext, ThrowsOnError, NoOpCompleted)
+    {
+    }
+
+    public AnonymousObserver(Action<T> onNext, Action<Exception> onError)
+        : this(onNext, onError, NoOpCompleted)
+    {
+    }
+
+    public AnonymousObserver(Action<T> onNext, Action onCompleted)
+        : this(onNext, ThrowsOnError, onCompleted)
+    {
+    }
+
+    public void OnCompleted()
+    {
+        _onCompleted.Invoke();
+    }
+
+    public void OnError(Exception error)
+    {
+        _onError.Invoke(error);
+    }
+
+    public void OnNext(T value)
+    {
+        _onNext.Invoke(value);
+    }
+}

src/Avalonia.Base/Reactive/CombinedSubject.cs

@@ -0,0 +1,23 @@
+using System;
+
+namespace Avalonia.Reactive;
+
+internal class CombinedSubject<T> : IAvaloniaSubject<T>
+{
+    private readonly IObserver<T> _observer;
+    private readonly IObservable<T> _observable;
+
+    public CombinedSubject(IObserver<T> observer, IObservable<T> observable)
+    {
+        _observer = observer;
+        _observable = observable;
+    }
+
+    public void OnCompleted() => _observer.OnCompleted();
+
+    public void OnError(Exception error) => _observer.OnError(error);
+
+    public void OnNext(T value) => _observer.OnNext(value);
+
+    public IDisposable Subscribe(IObserver<T> observer) => _observable.Subscribe(observer);
+}

src/Avalonia.Base/Reactive/CompositeDisposable.cs

@@ -0,0 +1,427 @@
+using System;
+using System.Collections;
+using System.Collections.Generic;
+using System.Threading;
+
+namespace Avalonia.Reactive;
+
+internal sealed class CompositeDisposable : ICollection<IDisposable>, IDisposable
+{
+    private readonly object _gate = new object();
+    private bool _disposed;
+    private List<IDisposable?> _disposables;
+    private int _count;
+    private const int ShrinkThreshold = 64;
+
+    /// <summary>
+    /// Initializes a new instance of the <see cref="CompositeDisposable"/> class with the specified number of disposables.
+    /// </summary>
+    /// <param name="capacity">The number of disposables that the new CompositeDisposable can initially store.</param>
+    /// <exception cref="ArgumentOutOfRangeException"><paramref name="capacity"/> is less than zero.</exception>
+    public CompositeDisposable(int capacity)
+    {
+        if (capacity < 0)
+        {
+            throw new ArgumentOutOfRangeException(nameof(capacity));
+        }
+
+        _disposables = new List<IDisposable?>(capacity);
+    }
+
+    /// <summary>
+    /// Initializes a new instance of the <see cref="CompositeDisposable"/> class from a group of disposables.
+    /// </summary>
+    /// <param name="disposables">Disposables that will be disposed together.</param>
+    /// <exception cref="ArgumentNullException"><paramref name="disposables"/> is <c>null</c>.</exception>
+    /// <exception cref="ArgumentException">Any of the disposables in the <paramref name="disposables"/> collection is <c>null</c>.</exception>
+    public CompositeDisposable(params IDisposable[] disposables)
+    {
+        if (disposables == null)
+        {
+            throw new ArgumentNullException(nameof(disposables));
+        }
+
+        _disposables = ToList(disposables);
+
+        // _count can be read by other threads and thus should be properly visible
+        // also releases the _disposables contents so it becomes thread-safe
+        Volatile.Write(ref _count, _disposables.Count);
+    }
+
+    /// <summary>
+    /// Initializes a new instance of the <see cref="CompositeDisposable"/> class from a group of disposables.
+    /// </summary>
+    /// <param name="disposables">Disposables that will be disposed together.</param>
+    /// <exception cref="ArgumentNullException"><paramref name="disposables"/> is <c>null</c>.</exception>
+    /// <exception cref="ArgumentException">Any of the disposables in the <paramref name="disposables"/> collection is <c>null</c>.</exception>
+    public CompositeDisposable(IList<IDisposable> disposables)
+    {
+        if (disposables == null)
+        {
+            throw new ArgumentNullException(nameof(disposables));
+        }
+
+        _disposables = ToList(disposables);
+
+        // _count can be read by other threads and thus should be properly visible
+        // also releases the _disposables contents so it becomes thread-safe
+        Volatile.Write(ref _count, _disposables.Count);
+    }
+
+    private static List<IDisposable?> ToList(IEnumerable<IDisposable> disposables)
+    {
+        var capacity = disposables switch
+        {
+            IDisposable[] a => a.Length,
+            ICollection<IDisposable> c => c.Count,
+            _ => 12
+        };
+
+        var list = new List<IDisposable?>(capacity);
+
+        // do the copy and null-check in one step to avoid a
+        // second loop for just checking for null items
+        foreach (var d in disposables)
+        {
+            if (d == null)
+            {
+                throw new ArgumentException("Disposables can't contain null", nameof(disposables));
+            }
+
+            list.Add(d);
+        }
+
+        return list;
+    }
+
+    /// <summary>
+    /// Gets the number of disposables contained in the <see cref="CompositeDisposable"/>.
+    /// </summary>
+    public int Count => Volatile.Read(ref _count);
+
+    /// <summary>
+    /// Adds a disposable to the <see cref="CompositeDisposable"/> or disposes the disposable if the <see cref="CompositeDisposable"/> is disposed.
+    /// </summary>
+    /// <param name="item">Disposable to add.</param>
+    /// <exception cref="ArgumentNullException"><paramref name="item"/> is <c>null</c>.</exception>
+    public void Add(IDisposable item)
+    {
+        if (item == null)
+        {
+            throw new ArgumentNullException(nameof(item));
+        }
+
+        lock (_gate)
+        {
+            if (!_disposed)
+            {
+                _disposables.Add(item);
+
+                // If read atomically outside the lock, it should be written atomically inside
+                // the plain read on _count is fine here because manipulation always happens
+                // from inside a lock.
+                Volatile.Write(ref _count, _count + 1);
+                return;
+            }
+        }
+
+        item.Dispose();
+    }
+
+    /// <summary>
+    /// Removes and disposes the first occurrence of a disposable from the <see cref="CompositeDisposable"/>.
+    /// </summary>
+    /// <param name="item">Disposable to remove.</param>
+    /// <returns>true if found; false otherwise.</returns>
+    /// <exception cref="ArgumentNullException"><paramref name="item"/> is <c>null</c>.</exception>
+    public bool Remove(IDisposable item)
+    {
+        if (item == null)
+        {
+            throw new ArgumentNullException(nameof(item));
+        }
+
+        lock (_gate)
+        {
+            // this composite was already disposed and if the item was in there
+            // it has been already removed/disposed
+            if (_disposed)
+            {
+                return false;
+            }
+
+            //
+            // List<T> doesn't shrink the size of the underlying array but does collapse the array
+            // by copying the tail one position to the left of the removal index. We don't need
+            // index-based lookup but only ordering for sequential disposal. So, instead of spending
+            // cycles on the Array.Copy imposed by Remove, we use a null sentinel value. We also
+            // do manual Swiss cheese detection to shrink the list if there's a lot of holes in it.
+            //
+
+            // read fields as infrequently as possible
+            var current = _disposables;
+
+            var i = current.IndexOf(item);
+            if (i < 0)
+            {
+                // not found, just return
+                return false;
+            }
+
+            current[i] = null;
+
+            if (current.Capacity > ShrinkThreshold && _count < current.Capacity / 2)
+            {
+                var fresh = new List<IDisposable?>(current.Capacity / 2);
+
+                foreach (var d in current)
+                {
+                    if (d != null)
+                    {
+                        fresh.Add(d);
+                    }
+                }
+
+                _disposables = fresh;
+            }
+
+            // make sure the Count property sees an atomic update
+            Volatile.Write(ref _count, _count - 1);
+        }
+
+        // if we get here, the item was found and removed from the list
+        // just dispose it and report success
+
+        item.Dispose();
+
+        return true;
+    }
+
+    /// <summary>
+    /// Disposes all disposables in the group and removes them from the group.
+    /// </summary>
+    public void Dispose()
+    {
+        List<IDisposable?>? currentDisposables = null;
+
+        lock (_gate)
+        {
+            if (!_disposed)
+            {
+                currentDisposables = _disposables;
+
+                // nulling out the reference is faster no risk to
+                // future Add/Remove because _disposed will be true
+                // and thus _disposables won't be touched again.
+                _disposables = null!; // NB: All accesses are guarded by _disposed checks.
+
+                Volatile.Write(ref _count, 0);
+                Volatile.Write(ref _disposed, true);
+            }
+        }
+
+        if (currentDisposables != null)
+        {
+            foreach (var d in currentDisposables)
+            {
+                d?.Dispose();
+            }
+        }
+    }
+
+    /// <summary>
+    /// Removes and disposes all disposables from the <see cref="CompositeDisposable"/>, but does not dispose the <see cref="CompositeDisposable"/>.
+    /// </summary>
+    public void Clear()
+    {
+        IDisposable?[] previousDisposables;
+
+        lock (_gate)
+        {
+            // disposed composites are always clear
+            if (_disposed)
+            {
+                return;
+            }
+
+            var current = _disposables;
+
+            previousDisposables = current.ToArray();
+            current.Clear();
+
+            Volatile.Write(ref _count, 0);
+        }
+
+        foreach (var d in previousDisposables)
+        {
+            d?.Dispose();
+        }
+    }
+
+    /// <summary>
+    /// Determines whether the <see cref="CompositeDisposable"/> contains a specific disposable.
+    /// </summary>
+    /// <param name="item">Disposable to search for.</param>
+    /// <returns>true if the disposable was found; otherwise, false.</returns>
+    /// <exception cref="ArgumentNullException"><paramref name="item"/> is <c>null</c>.</exception>
+    public bool Contains(IDisposable item)
+    {
+        if (item == null)
+        {
+            throw new ArgumentNullException(nameof(item));
+        }
+
+        lock (_gate)
+        {
+            if (_disposed)
+            {
+                return false;
+            }
+
+            return _disposables.Contains(item);
+        }
+    }
+
+    /// <summary>
+    /// Copies the disposables contained in the <see cref="CompositeDisposable"/> to an array, starting at a particular array index.
+    /// </summary>
+    /// <param name="array">Array to copy the contained disposables to.</param>
+    /// <param name="arrayIndex">Target index at which to copy the first disposable of the group.</param>
+    /// <exception cref="ArgumentNullException"><paramref name="array"/> is <c>null</c>.</exception>
+    /// <exception cref="ArgumentOutOfRangeException"><paramref name="arrayIndex"/> is less than zero. -or - <paramref name="arrayIndex"/> is larger than or equal to the array length.</exception>
+    public void CopyTo(IDisposable[] array, int arrayIndex)
+    {
+        if (array == null)
+        {
+            throw new ArgumentNullException(nameof(array));
+        }
+
+        if (arrayIndex < 0 || arrayIndex >= array.Length)
+        {
+            throw new ArgumentOutOfRangeException(nameof(arrayIndex));
+        }
+
+        lock (_gate)
+        {
+            // disposed composites are always empty
+            if (_disposed)
+            {
+                return;
+            }
+
+            if (arrayIndex + _count > array.Length)
+            {
+                // there is not enough space beyond arrayIndex 
+                // to accommodate all _count disposables in this composite
+                throw new ArgumentOutOfRangeException(nameof(arrayIndex));
+            }
+
+            var i = arrayIndex;
+
+            foreach (var d in _disposables)
+            {
+                if (d != null)
+                {
+                    array[i++] = d;
+                }
+            }
+        }
+    }
+
+    /// <summary>
+    /// Always returns false.
+    /// </summary>
+    public bool IsReadOnly => false;
+
+    /// <summary>
+    /// Returns an enumerator that iterates through the <see cref="CompositeDisposable"/>.
+    /// </summary>
+    /// <returns>An enumerator to iterate over the disposables.</returns>
+    public IEnumerator<IDisposable> GetEnumerator()
+    {
+        lock (_gate)
+        {
+            if (_disposed || _count == 0)
+            {
+                return EmptyEnumerator;
+            }
+
+            // the copy is unavoidable but the creation
+            // of an outer IEnumerable is avoidable
+            return new CompositeEnumerator(_disposables.ToArray());
+        }
+    }
+
+    /// <summary>
+    /// Returns an enumerator that iterates through the <see cref="CompositeDisposable"/>.
+    /// </summary>
+    /// <returns>An enumerator to iterate over the disposables.</returns>
+    IEnumerator IEnumerable.GetEnumerator() => GetEnumerator();
+
+    /// <summary>
+    /// Gets a value that indicates whether the object is disposed.
+    /// </summary>
+    public bool IsDisposed => Volatile.Read(ref _disposed);
+
+    /// <summary>
+    /// An empty enumerator for the <see cref="GetEnumerator"/>
+    /// method to avoid allocation on disposed or empty composites.
+    /// </summary>
+    private static readonly CompositeEnumerator EmptyEnumerator =
+        new CompositeEnumerator(Array.Empty<IDisposable?>());
+
+    /// <summary>
+    /// An enumerator for an array of disposables.
+    /// </summary>
+    private sealed class CompositeEnumerator : IEnumerator<IDisposable>
+    {
+        private readonly IDisposable?[] _disposables;
+        private int _index;
+
+        public CompositeEnumerator(IDisposable?[] disposables)
+        {
+            _disposables = disposables;
+            _index = -1;
+        }
+
+        public IDisposable Current => _disposables[_index]!; // NB: _index is only advanced to non-null positions.
+
+        object IEnumerator.Current => _disposables[_index]!;
+
+        public void Dispose()
+        {
+            // Avoid retention of the referenced disposables
+            // beyond the lifecycle of the enumerator.
+            // Not sure if this happens by default to
+            // generic array enumerators though.
+            var disposables = _disposables;
+            Array.Clear(disposables, 0, disposables.Length);
+        }
+
+        public bool MoveNext()
+        {
+            var disposables = _disposables;
+
+            for (;;)
+            {
+                var idx = ++_index;
+
+                if (idx >= disposables.Length)
+                {
+                    return false;
+                }
+
+                // inlined that filter for null elements
+                if (disposables[idx] != null)
+                {
+                    return true;
+                }
+            }
+        }
+
+        public void Reset()
+        {
+            _index = -1;
+        }
+    }
+}

src/Avalonia.Base/Reactive/Disposable.cs

@@ -0,0 +1,98 @@
+using System;
+using System.Threading;
+
+namespace Avalonia.Reactive;
+
+/// <summary>
+/// Provides a set of static methods for creating <see cref="IDisposable"/> objects.
+/// </summary>
+internal static class Disposable
+{
+    /// <summary>
+    /// Represents a disposable that does nothing on disposal.
+    /// </summary>
+    private sealed class EmptyDisposable : IDisposable
+    {
+        public static readonly EmptyDisposable Instance = new();
+
+        private EmptyDisposable()
+        {
+        }
+
+        public void Dispose()
+        {
+            // no op
+        }
+    }
+    
+    internal sealed class AnonymousDisposable : IDisposable
+    {
+        private volatile Action? _dispose;
+        public AnonymousDisposable(Action dispose)
+        {
+            _dispose = dispose;
+        }
+        public bool IsDisposed => _dispose == null;
+        public void Dispose()
+        {
+            Interlocked.Exchange(ref _dispose, null)?.Invoke();
+        }
+    }
+
+    internal sealed class AnonymousDisposable<TState> : IDisposable
+    {
+        private TState _state;
+        private volatile Action<TState>? _dispose;
+
+        public AnonymousDisposable(TState state, Action<TState> dispose)
+        {
+            _state = state;
+            _dispose = dispose;
+        }
+
+        public bool IsDisposed => _dispose == null;
+        public void Dispose()
+        {
+            Interlocked.Exchange(ref _dispose, null)?.Invoke(_state);
+            _state = default!;
+        }
+    }
+
+    /// <summary>
+    /// Gets the disposable that does nothing when disposed.
+    /// </summary>
+    public static IDisposable Empty => EmptyDisposable.Instance;
+
+    /// <summary>
+    /// Creates a disposable object that invokes the specified action when disposed.
+    /// </summary>
+    /// <param name="dispose">Action to run during the first call to <see cref="IDisposable.Dispose"/>. The action is guaranteed to be run at most once.</param>
+    /// <returns>The disposable object that runs the given action upon disposal.</returns>
+    /// <exception cref="ArgumentNullException"><paramref name="dispose"/> is <c>null</c>.</exception>
+    public static IDisposable Create(Action dispose)
+    {
+        if (dispose == null)
+        {
+            throw new ArgumentNullException(nameof(dispose));
+        }
+
+        return new AnonymousDisposable(dispose);
+    }
+
+    /// <summary>
+    /// Creates a disposable object that invokes the specified action when disposed.
+    /// </summary>
+    /// <param name="state">The state to be passed to the action.</param>
+    /// <param name="dispose">Action to run during the first call to <see cref="IDisposable.Dispose"/>. The action is guaranteed to be run at most once.</param>
+    /// <returns>The disposable object that runs the given action upon disposal.</returns>
+    /// <exception cref="ArgumentNullException"><paramref name="dispose"/> is <c>null</c>.</exception>
+    public static IDisposable Create<TState>(TState state, Action<TState> dispose)
+    {
+        if (dispose == null)
+        {
+            throw new ArgumentNullException(nameof(dispose));
+        }
+
+        return new AnonymousDisposable<TState>(state, dispose);
+    }
+}

src/Avalonia.Base/Reactive/DisposableMixin.cs

@@ -0,0 +1,37 @@
+using System;
+using Avalonia.Reactive;
+
+namespace Avalonia.Reactive;
+
+/// <summary>
+/// Extension methods associated with the IDisposable interface.
+/// </summary>
+internal static class DisposableMixin
+{
+    /// <summary>
+    /// Ensures the provided disposable is disposed with the specified <see cref="CompositeDisposable"/>.
+    /// </summary>
+    /// <typeparam name="T">
+    /// The type of the disposable.
+    /// </typeparam>
+    /// <param name="item">
+    /// The disposable we are going to want to be disposed by the CompositeDisposable.
+    /// </param>
+    /// <param name="compositeDisposable">
+    /// The <see cref="CompositeDisposable"/> to which <paramref name="item"/> will be added.
+    /// </param>
+    /// <returns>
+    /// The disposable.
+    /// </returns>
+    public static T DisposeWith<T>(this T item, CompositeDisposable compositeDisposable)
+        where T : IDisposable
+    {
+        if (compositeDisposable is null)
+        {
+            throw new ArgumentNullException(nameof(compositeDisposable));
+        }
+
+        compositeDisposable.Add(item);
+        return item;
+    }
+}

src/Avalonia.Base/Reactive/IAvaloniaSubject.cs

@@ -0,0 +1,8 @@
+using System;
+
+namespace Avalonia.Reactive;
+
+internal interface IAvaloniaSubject<T> : IObserver<T>, IObservable<T> /*, ISubject<T> */
+{
+    
+}

src/Avalonia.Base/Reactive/LightweightObservableBase.cs

@@ -1,7 +1,5 @@
 using System;
 using System.Collections.Generic;
-using System.Reactive;
-using System.Reactive.Disposables;
 using System.Threading;
 using Avalonia.Threading;
 
@@ -12,7 +10,7 @@ namespace Avalonia.Reactive
     /// </summary>
     /// <typeparam name="T">The observable type.</typeparam>
     /// <remarks>
-    /// <see cref="ObservableBase{T}"/> is rather heavyweight in terms of allocations and memory
+    /// 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>
@@ -21,11 +19,13 @@ namespace Avalonia.Reactive
         private Exception? _error;
         private List<IObserver<T>>? _observers = new List<IObserver<T>>();
 
+        public bool HasObservers => _observers?.Count > 0;
+        
         public IDisposable Subscribe(IObserver<T> observer)
         {
             _ = observer ?? throw new ArgumentNullException(nameof(observer));
 
-            Dispatcher.UIThread.VerifyAccess();
+            //Dispatcher.UIThread.VerifyAccess();
 
             var first = false;
 

src/Avalonia.Base/Reactive/LightweightSubject.cs

@@ -0,0 +1,30 @@
+using System;
+using System.Collections.Generic;
+using System.Diagnostics;
+using System.Runtime.InteropServices;
+using System.Threading;
+using Avalonia.Threading;
+
+namespace Avalonia.Reactive;
+
+internal class LightweightSubject<T> : LightweightObservableBase<T>, IAvaloniaSubject<T>
+{
+    public void OnCompleted()
+    {
+        PublishCompleted();
+    }
+
+    public void OnError(Exception error)
+    {
+        PublishError(error);
+    }
+
+    public void OnNext(T value)
+    {
+        PublishNext(value);
+    }
+
+    protected override void Initialize() { }
+
+    protected override void Deinitialize() { }
+}

src/Avalonia.Base/Reactive/Observable.cs

@@ -0,0 +1,238 @@
+using System;
+using System.Collections;
+using System.Collections.Generic;
+using System.Linq;
+using Avalonia.Reactive.Operators;
+using Avalonia.Threading;
+
+namespace Avalonia.Reactive;
+
+/// <summary>
+/// Provides common observable methods as a replacement for the Rx framework.
+/// </summary>
+internal static class Observable
+{
+    public static IObservable<TSource> Create<TSource>(Func<IObserver<TSource>, IDisposable> subscribe)
+    {
+        return new CreateWithDisposableObservable<TSource>(subscribe);
+    }
+
+    public static IDisposable Subscribe<T>(this IObservable<T> source, Action<T> action)
+    {
+        return source.Subscribe(new AnonymousObserver<T>(action));
+    }
+
+    public static IObservable<TResult> Select<TSource, TResult>(this IObservable<TSource> source, Func<TSource, TResult> selector)
+    {
+        return Create<TResult>(obs =>
+        {
+            return source.Subscribe(new AnonymousObserver<TSource>(
+                input =>
+                {
+                    TResult value;
+                    try
+                    {
+                        value = selector(input);
+                    }
+                    catch (Exception ex)
+                    {
+                        obs.OnError(ex);
+                        return;
+                    }
+
+                    obs.OnNext(value);
+                }, obs.OnError, obs.OnCompleted));
+        });
+    }
+    
+    public static IObservable<TSource> Where<TSource>(this IObservable<TSource> source, Func<TSource, bool> predicate)
+    {
+        return Create<TSource>(obs =>
+        {
+            return source.Subscribe(new AnonymousObserver<TSource>(
+                input =>
+                {
+                    bool shouldRun;
+                    try
+                    {
+                        shouldRun = predicate(input);
+                    }
+                    catch (Exception ex)
+                    {
+                        obs.OnError(ex);
+                        return;
+                    }
+                    if (shouldRun)
+                    {
+                        obs.OnNext(input);
+                    }
+                }, obs.OnError, obs.OnCompleted));
+        });
+    }
+
+    public static IObservable<TSource> Switch<TSource>(
+        this IObservable<IObservable<TSource>> sources)
+    {
+        return new Switch<TSource>(sources);
+    }
+
+    public static IObservable<TResult> CombineLatest<TFirst, TSecond, TResult>(
+        this IObservable<TFirst> first, IObservable<TSecond> second,
+        Func<TFirst, TSecond, TResult> resultSelector)
+    {
+        return new CombineLatest<TFirst, TSecond, TResult>(first, second, resultSelector);
+    }
+    
+    public static IObservable<TInput[]> CombineLatest<TInput>(
+        this IEnumerable<IObservable<TInput>> inputs)
+    {
+        return new CombineLatest<TInput, TInput[]>(inputs, items => items);
+    }
+
+    public static IObservable<T> Skip<T>(this IObservable<T> source, int skipCount)
+    {
+        if (skipCount <= 0)
+        {
+            throw new ArgumentException("Skip count must be bigger than zero", nameof(skipCount));
+        }
+
+        return Create<T>(obs =>
+        {
+            var remaining = skipCount;
+            return source.Subscribe(new AnonymousObserver<T>(
+                input =>
+                {
+                    if (remaining <= 0)
+                    {
+                        obs.OnNext(input);
+                    }
+                    else
+                    {
+                        remaining--;
+                    }
+                }, obs.OnError, obs.OnCompleted));
+        });
+    }
+    
+    public static IObservable<T> Take<T>(this IObservable<T> source, int takeCount)
+    {
+        if (takeCount <= 0)
+        {
+            return Empty<T>();
+        }
+
+        return Create<T>(obs =>
+        {
+            var remaining = takeCount;
+            IDisposable? sub = null;
+            sub = source.Subscribe(new AnonymousObserver<T>(
+                input =>
+                {
+                    if (remaining > 0)
+                    {
+                        --remaining;
+                        obs.OnNext(input);
+
+                        if (remaining == 0)
+                        {
+                            sub?.Dispose();
+                            obs.OnCompleted();
+                        }
+                    }
+                }, obs.OnError, obs.OnCompleted));
+            return sub;
+        });
+    }
+
+    public static IObservable<EventArgs> FromEventPattern(Action<EventHandler> addHandler, Action<EventHandler> removeHandler)
+    {
+        return Create<EventArgs>(observer =>
+        {
+            var handler = new Action<EventArgs>(observer.OnNext);
+            var converted = new EventHandler((_, args) => handler(args));
+            addHandler(converted);
+
+            return Disposable.Create(() => removeHandler(converted));
+        });
+    }
+    
+    public static IObservable<T> Return<T>(T value)
+    {
+        return new ReturnImpl<T>(value);
+    }
+    
+    public static IObservable<T> Empty<T>()
+    {
+        return EmptyImpl<T>.Instance;
+    }
+        
+    /// <summary>
+    /// Returns an observable that fires once with the specified value and never completes.
+    /// </summary>
+    /// <typeparam name="T">The type of the value.</typeparam>
+    /// <param name="value">The value.</param>
+    /// <returns>The observable.</returns>
+    public static IObservable<T> SingleValue<T>(T value)
+    {
+        return new SingleValueImpl<T>(value);
+    }
+ 
+    private sealed class SingleValueImpl<T> : IObservable<T>
+    {
+        private readonly T _value;
+
+        public SingleValueImpl(T value)
+        {
+            _value = value;
+        }
+        public IDisposable Subscribe(IObserver<T> observer)
+        {
+            observer.OnNext(_value);
+            return Disposable.Empty;
+        }
+    }
+    
+    private sealed class ReturnImpl<T> : IObservable<T>
+    {
+        private readonly T _value;
+
+        public ReturnImpl(T value)
+        {
+            _value = value;
+        }
+        public IDisposable Subscribe(IObserver<T> observer)
+        {
+            observer.OnNext(_value);
+            observer.OnCompleted();
+            return Disposable.Empty;
+        }
+    }
+    
+    internal sealed class EmptyImpl<TResult> : IObservable<TResult>
+    {
+        internal static readonly IObservable<TResult> Instance = new EmptyImpl<TResult>();
+
+        private EmptyImpl() { }
+
+        public IDisposable Subscribe(IObserver<TResult> observer)
+        {
+            observer.OnCompleted();
+            return Disposable.Empty;
+        }
+    }
+    
+    private sealed class CreateWithDisposableObservable<TSource> : IObservable<TSource>
+    {
+        private readonly Func<IObserver<TSource>, IDisposable> _subscribe;
+
+        public CreateWithDisposableObservable(Func<IObserver<TSource>, IDisposable> subscribe)
+        {
+            _subscribe = subscribe;
+        }
+
+        public IDisposable Subscribe(IObserver<TSource> observer)
+        {
+            return _subscribe(observer);
+        }
+    }
+}

src/Avalonia.Base/Reactive/ObservableEx.cs

@@ -1,37 +0,0 @@
-using System;
-using System.Reactive.Disposables;
-
-namespace Avalonia.Reactive
-{
-    /// <summary>
-    /// Provides common observable methods not found in standard Rx framework.
-    /// </summary>
-    public static class ObservableEx
-    {
-        /// <summary>
-        /// Returns an observable that fires once with the specified value and never completes.
-        /// </summary>
-        /// <typeparam name="T">The type of the value.</typeparam>
-        /// <param name="value">The value.</param>
-        /// <returns>The observable.</returns>
-        public static IObservable<T> SingleValue<T>(T value)
-        {
-            return new SingleValueImpl<T>(value);
-        }
- 
-        private class SingleValueImpl<T> : IObservable<T>
-        {
-            private T _value;
-
-            public SingleValueImpl(T value)
-            {
-                _value = value;
-            }
-            public IDisposable Subscribe(IObserver<T> observer)
-            {
-                observer.OnNext(_value);
-                return Disposable.Empty;
-            }
-        }
-    }
-}

src/Avalonia.Base/Reactive/Operators/CombineLatest.cs

@@ -0,0 +1,374 @@
+using System;
+using System.Collections.Generic;
+using System.Collections.ObjectModel;
+using System.Linq;
+
+namespace Avalonia.Reactive.Operators;
+
+// Code based on https://github.com/dotnet/reactive/blob/main/Rx.NET/Source/src/System.Reactive/Linq/Observable/CombineLatest.cs
+
+internal sealed class CombineLatest<TFirst, TSecond, TResult> : IObservable<TResult>
+{
+    private readonly IObservable<TFirst> _first;
+    private readonly IObservable<TSecond> _second;
+    private readonly Func<TFirst, TSecond, TResult> _resultSelector;
+
+    public CombineLatest(IObservable<TFirst> first, IObservable<TSecond> second,
+        Func<TFirst, TSecond, TResult> resultSelector)
+    {
+        _first = first;
+        _second = second;
+        _resultSelector = resultSelector;
+    }
+
+    public IDisposable Subscribe(IObserver<TResult> observer)
+    {
+        var sink = new _(_resultSelector, observer);
+        sink.Run(_first, _second);
+        return sink;
+    }
+
+    internal sealed class _ : IdentitySink<TResult>
+    {
+        private readonly Func<TFirst, TSecond, TResult> _resultSelector;
+        private readonly object _gate = new object();
+
+        public _(Func<TFirst, TSecond, TResult> resultSelector, IObserver<TResult> observer)
+            : base(observer)
+        {
+            _resultSelector = resultSelector;
+            _firstDisposable = null!;
+            _secondDisposable = null!;
+        }
+
+        private IDisposable _firstDisposable;
+        private IDisposable _secondDisposable;
+
+        public void Run(IObservable<TFirst> first, IObservable<TSecond> second)
+        {
+            var fstO = new FirstObserver(this);
+            var sndO = new SecondObserver(this);
+
+            fstO.SetOther(sndO);
+            sndO.SetOther(fstO);
+
+            _firstDisposable = first.Subscribe(fstO);
+            _secondDisposable = second.Subscribe(sndO);
+        }
+
+        protected override void Dispose(bool disposing)
+        {
+            if (disposing)
+            {
+                _firstDisposable.Dispose();
+                _secondDisposable.Dispose();
+            }
+
+            base.Dispose(disposing);
+        }
+
+        private sealed class FirstObserver : IObserver<TFirst>
+        {
+            private readonly _ _parent;
+            private SecondObserver _other;
+
+            public FirstObserver(_ parent)
+            {
+                _parent = parent;
+                _other = default!; // NB: Will be set by SetOther.
+            }
+
+            public void SetOther(SecondObserver other) { _other = other; }
+
+            public bool HasValue { get; private set; }
+            public TFirst? Value { get; private set; }
+            public bool Done { get; private set; }
+
+            public void OnNext(TFirst value)
+            {
+                lock (_parent._gate)
+                {
+                    HasValue = true;
+                    Value = value;
+
+                    if (_other.HasValue)
+                    {
+                        TResult res;
+                        try
+                        {
+                            res = _parent._resultSelector(value, _other.Value!);
+                        }
+                        catch (Exception ex)
+                        {
+                            _parent.ForwardOnError(ex);
+                            return;
+                        }
+
+                        _parent.ForwardOnNext(res);
+                    }
+                    else if (_other.Done)
+                    {
+                        _parent.ForwardOnCompleted();
+                    }
+                }
+            }
+
+            public void OnError(Exception error)
+            {
+                lock (_parent._gate)
+                {
+                    _parent.ForwardOnError(error);
+                }
+            }
+
+            public void OnCompleted()
+            {
+                lock (_parent._gate)
+                {
+                    Done = true;
+
+                    if (_other.Done)
+                    {
+                        _parent.ForwardOnCompleted();
+                    }
+                    else
+                    {
+                        _parent._firstDisposable.Dispose();
+                    }
+                }
+            }
+        }
+
+        private sealed class SecondObserver : IObserver<TSecond>
+        {
+            private readonly _ _parent;
+            private FirstObserver _other;
+
+            public SecondObserver(_ parent)
+            {
+                _parent = parent;
+                _other = default!; // NB: Will be set by SetOther.
+            }
+
+            public void SetOther(FirstObserver other) { _other = other; }
+
+            public bool HasValue { get; private set; }
+            public TSecond? Value { get; private set; }
+            public bool Done { get; private set; }
+
+            public void OnNext(TSecond value)
+            {
+                lock (_parent._gate)
+                {
+                    HasValue = true;
+                    Value = value;
+
+                    if (_other.HasValue)
+                    {
+                        TResult res;
+                        try
+                        {
+                            res = _parent._resultSelector(_other.Value!, value);
+                        }
+                        catch (Exception ex)
+                        {
+                            _parent.ForwardOnError(ex);
+                            return;
+                        }
+
+                        _parent.ForwardOnNext(res);
+                    }
+                    else if (_other.Done)
+                    {
+                        _parent.ForwardOnCompleted();
+                    }
+                }
+            }
+
+            public void OnError(Exception error)
+            {
+                lock (_parent._gate)
+                {
+                    _parent.ForwardOnError(error);
+                }
+            }
+
+            public void OnCompleted()
+            {
+                lock (_parent._gate)
+                {
+                    Done = true;
+
+                    if (_other.Done)
+                    {
+                        _parent.ForwardOnCompleted();
+                    }
+                    else
+                    {
+                        _parent._secondDisposable.Dispose();
+                    }
+                }
+            }
+        }
+    }
+}
+
+internal sealed class CombineLatest<TSource, TResult> : IObservable<TResult>
+{
+    private readonly IEnumerable<IObservable<TSource>> _sources;
+    private readonly Func<TSource[], TResult> _resultSelector;
+
+    public CombineLatest(IEnumerable<IObservable<TSource>> sources, Func<TSource[], TResult> resultSelector)
+    {
+        _sources = sources;
+        _resultSelector = resultSelector;
+    }
+
+    public IDisposable Subscribe(IObserver<TResult> observer)
+    {
+        var sink = new _(_resultSelector, observer);
+        sink.Run(_sources);
+        return sink;
+    }
+
+    internal sealed class _ : IdentitySink<TResult>
+    {
+        private readonly object _gate = new object();
+        private readonly Func<TSource[], TResult> _resultSelector;
+
+        public _(Func<TSource[], TResult> resultSelector, IObserver<TResult> observer)
+            : base(observer)
+        {
+            _resultSelector = resultSelector;
+
+            // NB: These will be set in Run before getting used.
+            _hasValue = null!;
+            _values = null!;
+            _isDone = null!;
+            _subscriptions = null!;
+        }
+
+        private bool[] _hasValue;
+        private bool _hasValueAll;
+        private TSource[] _values;
+        private bool[] _isDone;
+        private IDisposable[] _subscriptions;
+
+        public void Run(IEnumerable<IObservable<TSource>> sources)
+        {
+            var srcs = sources.ToArray();
+
+            var N = srcs.Length;
+
+            _hasValue = new bool[N];
+            _hasValueAll = false;
+
+            _values = new TSource[N];
+
+            _isDone = new bool[N];
+
+            _subscriptions = new IDisposable[N];
+
+            for (var i = 0; i < N; i++)
+            {
+                var j = i;
+
+                var o = new SourceObserver(this, j);
+                _subscriptions[j] = o;
+
+                o.Disposable = srcs[j].Subscribe(o);
+            }
+
+            SetUpstream(new CompositeDisposable(_subscriptions));
+        }
+
+        private void OnNext(int index, TSource value)
+        {
+            lock (_gate)
+            {
+                _values[index] = value;
+
+                _hasValue[index] = true;
+
+                if (_hasValueAll || (_hasValueAll = _hasValue.All(v => v)))
+                {
+                    TResult res;
+                    try
+                    {
+                        res = _resultSelector(_values);
+                    }
+                    catch (Exception ex)
+                    {
+                        ForwardOnError(ex);
+                        return;
+                    }
+
+                    ForwardOnNext(res);
+                }
+                else if (_isDone.Where((_, i) => i != index).All(d => d))
+                {
+                    ForwardOnCompleted();
+                }
+            }
+        }
+
+        private new void OnError(Exception error)
+        {
+            lock (_gate)
+            {
+                ForwardOnError(error);
+            }
+        }
+
+        private void OnCompleted(int index)
+        {
+            lock (_gate)
+            {
+                _isDone[index] = true;
+
+                if (_isDone.All(d => d))
+                {
+                    ForwardOnCompleted();
+                }
+                else
+                {
+                    _subscriptions[index].Dispose();
+                }
+            }
+        }
+
+        private sealed class SourceObserver : IObserver<TSource>, IDisposable
+        {
+            private readonly _ _parent;
+            private readonly int _index;
+
+            public SourceObserver(_ parent, int index)
+            {
+                _parent = parent;
+                _index = index;
+            }
+
+            public IDisposable? Disposable { get; set; }
+
+            public void OnNext(TSource value)
+            {
+                _parent.OnNext(_index, value);
+            }
+
+            public void OnError(Exception error)
+            {
+                _parent.OnError(error);
+            }
+
+            public void OnCompleted()
+            {
+                _parent.OnCompleted(_index);
+            }
+
+            public void Dispose()
+            {
+                Disposable?.Dispose();
+            }
+        }
+    }
+}

src/Avalonia.Base/Reactive/Operators/Sink.cs

@@ -0,0 +1,109 @@
+using System;
+using System.Threading;
+
+namespace Avalonia.Reactive.Operators;
+
+internal abstract class Sink<TTarget> : IDisposable
+{
+    private IDisposable? _upstream;
+    private volatile IObserver<TTarget> _observer;
+
+    protected Sink(IObserver<TTarget> observer)
+    {
+        _observer = observer;
+    }
+
+    public void Dispose()
+    {
+        Dispose(true);
+    }
+
+    /// <summary>
+    /// Override this method to dispose additional resources.
+    /// The method is guaranteed to be called at most once.
+    /// </summary>
+    /// <param name="disposing">If true, the method was called from <see cref="Dispose()"/>.</param>
+    protected virtual void Dispose(bool disposing)
+    {
+        //Calling base.Dispose(true) is not a proper disposal, so we can omit the assignment here.
+        //Sink is internal so this can pretty much be enforced.
+        //_observer = NopObserver<TTarget>.Instance;
+
+        _upstream?.Dispose();
+    }
+
+    public void ForwardOnNext(TTarget value)
+    {
+        _observer.OnNext(value);
+    }
+
+    public void ForwardOnCompleted()
+    {
+        _observer.OnCompleted();
+        Dispose();
+    }
+
+    public void ForwardOnError(Exception error)
+    {
+        _observer.OnError(error);
+        Dispose();
+    }
+
+    protected void SetUpstream(IDisposable upstream)
+    {
+        _upstream = upstream;
+    }
+
+    protected void DisposeUpstream()
+    {
+        _upstream?.Dispose();
+    }
+}
+
+internal abstract class Sink<TSource, TTarget> : Sink<TTarget>, IObserver<TSource>
+{
+    protected Sink(IObserver<TTarget> observer) : base(observer)
+    {
+    }
+
+    public virtual void Run(IObservable<TSource> source)
+    {
+        SetUpstream(source.Subscribe(this));
+    }
+
+    public abstract void OnNext(TSource value);
+
+    public virtual void OnError(Exception error) => ForwardOnError(error);
+
+    public virtual void OnCompleted() => ForwardOnCompleted();
+
+    public IObserver<TTarget> GetForwarder() => new _(this);
+
+    private sealed class _ : IObserver<TTarget>
+    {
+        private readonly Sink<TSource, TTarget> _forward;
+
+        public _(Sink<TSource, TTarget> forward)
+        {
+            _forward = forward;
+        }
+
+        public void OnNext(TTarget value) => _forward.ForwardOnNext(value);
+
+        public void OnError(Exception error) => _forward.ForwardOnError(error);
+
+        public void OnCompleted() => _forward.ForwardOnCompleted();
+    }
+}
+
+internal abstract class IdentitySink<T> : Sink<T, T>
+{
+    protected IdentitySink(IObserver<T> observer) : base(observer)
+    {
+    }
+
+    public override void OnNext(T value)
+    {
+        ForwardOnNext(value);
+    }
+}

src/Avalonia.Base/Reactive/Operators/Switch.cs

@@ -0,0 +1,144 @@
+using System;
+
+namespace Avalonia.Reactive.Operators;
+
+// Code based on https://github.com/dotnet/reactive/blob/main/Rx.NET/Source/src/System.Reactive/Linq/Observable/Switch.cs
+
+internal sealed class Switch<TSource> : IObservable<TSource>
+{
+    private readonly IObservable<IObservable<TSource>> _sources;
+
+    public Switch(IObservable<IObservable<TSource>> sources)
+    {
+        _sources = sources;
+    }
+
+    public IDisposable Subscribe(IObserver<TSource> observer)
+    {
+        return _sources.Subscribe(new _(observer));
+    }
+
+    internal sealed class _ : Sink<IObservable<TSource>, TSource>
+    {
+        private readonly object _gate = new object();
+
+        public _(IObserver<TSource> observer)
+            : base(observer)
+        {
+        }
+
+        private IDisposable? _innerSerialDisposable;
+        private bool _isStopped;
+        private ulong _latest;
+        private bool _hasLatest;
+
+        protected override void Dispose(bool disposing)
+        {
+            if (disposing)
+            {
+                _innerSerialDisposable?.Dispose();
+            }
+
+            base.Dispose(disposing);
+        }
+
+        public override void OnNext(IObservable<TSource> value)
+        {
+            ulong id;
+
+            lock (_gate)
+            {
+                id = unchecked(++_latest);
+                _hasLatest = true;
+            }
+
+            var innerObserver = new InnerObserver(this, id);
+
+            _innerSerialDisposable = innerObserver;
+            innerObserver.Disposable = value.Subscribe(innerObserver);
+        }
+
+        public override void OnError(Exception error)
+        {
+            lock (_gate)
+            {
+                ForwardOnError(error);
+            }
+        }
+
+        public override void OnCompleted()
+        {
+            lock (_gate)
+            {
+                DisposeUpstream();
+
+                _isStopped = true;
+                if (!_hasLatest)
+                {
+                    ForwardOnCompleted();
+                }
+            }
+        }
+
+        private sealed class InnerObserver : IObserver<TSource>, IDisposable
+        {
+            private readonly _ _parent;
+            private readonly ulong _id;
+
+            public InnerObserver(_ parent, ulong id)
+            {
+                _parent = parent;
+                _id = id;
+            }
+
+            public IDisposable? Disposable { get; set; }
+
+            public void OnNext(TSource value)
+            {
+                lock (_parent._gate)
+                {
+                    if (_parent._latest == _id)
+                    {
+                        _parent.ForwardOnNext(value);
+                    }
+                }
+            }
+
+            public void OnError(Exception error)
+            {
+                lock (_parent._gate)
+                {
+                    Dispose();
+
+                    if (_parent._latest == _id)
+                    {
+                        _parent.ForwardOnError(error);
+                    }
+                }
+            }
+
+            public void OnCompleted()
+            {
+                lock (_parent._gate)
+                {
+                    Dispose();
+
+                    if (_parent._latest == _id)
+                    {
+                        _parent._hasLatest = false;
+
+                        if (_parent._isStopped)
+                        {
+                            _parent.ForwardOnCompleted();
+                        }
+                    }
+                }
+            }
+
+            public void Dispose()
+            {
+                Disposable?.Dispose();
+            }
+        }
+    }
+}

src/Avalonia.Base/Rendering/PlatformRenderInterfaceContextManager.cs

@@ -1,8 +1,8 @@
 using System;
 using System.Collections.Generic;
-using System.Reactive.Disposables;
 using Avalonia.Metadata;
 using Avalonia.Platform;
+using Avalonia.Reactive;
 
 namespace Avalonia.Rendering;
 

src/Avalonia.Base/Rendering/SceneGraph/VisualNode.cs

@@ -1,7 +1,7 @@
 using System;
 using System.Collections.Generic;
 using System.Diagnostics.CodeAnalysis;
-using System.Reactive.Disposables;
+using Avalonia.Reactive;
 using Avalonia.Media;
 using Avalonia.Platform;
 using Avalonia.Utilities;

src/Avalonia.Base/Rendering/UiThreadRenderTimer.cs

@@ -1,6 +1,6 @@
 using System;
 using System.Diagnostics;
-using System.Reactive.Disposables;
+using Avalonia.Reactive;
 using Avalonia.Threading;
 
 namespace Avalonia.Rendering

src/Avalonia.Base/Styling/StyleInstance.cs

@@ -1,9 +1,9 @@
 using System;
 using System.Collections.Generic;
-using System.Reactive.Subjects;
 using Avalonia.Animation;
 using Avalonia.Data;
 using Avalonia.PropertyStore;
+using Avalonia.Reactive;
 using Avalonia.Styling.Activators;
 
 namespace Avalonia.Styling
@@ -24,7 +24,7 @@ namespace Avalonia.Styling
         private bool _isActive;
         private List<ISetterInstance>? _setters;
         private List<IAnimation>? _animations;
-        private Subject<bool>? _animationTrigger;
+        private LightweightSubject<bool>? _animationTrigger;
 
         public StyleInstance(
             IStyle style,
@@ -67,7 +67,7 @@ namespace Avalonia.Styling
         {
             if (_animations is not null && control is Animatable animatable)
             {
-                _animationTrigger ??= new Subject<bool>();
+                _animationTrigger ??= new LightweightSubject<bool>();
                 foreach (var animation in _animations)
                     animation.Apply(animatable, null, _animationTrigger);
 

src/Avalonia.Base/Threading/DispatcherTimer.cs

@@ -1,5 +1,5 @@
 using System;
-using System.Reactive.Disposables;
+using Avalonia.Reactive;
 using Avalonia.Platform;
 
 namespace Avalonia.Threading

src/Avalonia.Base/Utilities/IWeakSubscriber.cs

@@ -1,18 +0,0 @@
-using System;
-
-namespace Avalonia.Utilities
-{
-    /// <summary>
-    /// Defines a listener to a event subscribed vis the <see cref="WeakObservable"/>.
-    /// </summary>
-    /// <typeparam name="T">The type of the event arguments.</typeparam>
-    public interface IWeakSubscriber<T> where T : EventArgs
-    {
-        /// <summary>
-        /// Invoked when the subscribed event is raised.
-        /// </summary>
-        /// <param name="sender">The event sender.</param>
-        /// <param name="e">The event arguments.</param>
-        void OnEvent(object? sender, T e);
-    }
-}

src/Avalonia.Base/Utilities/WeakObservable.cs

@@ -1,60 +0,0 @@
-using System;
-using System.Reactive;
-using System.Reactive.Linq;
-
-namespace Avalonia.Utilities
-{
-    /// <summary>
-    /// Provides extension methods for working with weak event handlers.
-    /// </summary>
-    public static class WeakObservable
-    {
-
-        private class Handler<TEventArgs> 
-            : IWeakSubscriber<TEventArgs>,
-                IWeakEventSubscriber<TEventArgs> where TEventArgs : EventArgs
-        {
-            private IObserver<EventPattern<object, TEventArgs>> _observer;
-
-            public Handler(IObserver<EventPattern<object, TEventArgs>> observer)
-            {
-                _observer = observer;
-            }
-
-            public void OnEvent(object? sender, TEventArgs e)
-            {
-                _observer.OnNext(new EventPattern<object, TEventArgs>(sender, e));
-            }
-
-            public void OnEvent(object? sender, WeakEvent ev, TEventArgs e)
-            {
-                _observer.OnNext(new EventPattern<object, TEventArgs>(sender, e));
-            }
-        }
-        
-        /// <summary>
-        /// Converts a WeakEvent conforming to the standard .NET event pattern into an observable
-        /// sequence, subscribing weakly.
-        /// </summary>
-        /// <typeparam name="TTarget">The type of target.</typeparam>
-        /// <typeparam name="TEventArgs">The type of the event args.</typeparam>
-        /// <param name="target">Object instance that exposes the event to convert.</param>
-        /// <param name="ev">The weak event to convert.</param>
-        /// <returns></returns>
-        public static IObservable<EventPattern<object, TEventArgs>> FromEventPattern<TTarget, TEventArgs>(
-            TTarget target, WeakEvent<TTarget, TEventArgs> ev)
-            where TEventArgs : EventArgs where TTarget : class
-        {
-            _ = target ?? throw new ArgumentNullException(nameof(target));
-            _ = ev ?? throw new ArgumentNullException(nameof(ev));
-
-            return Observable.Create<EventPattern<object, TEventArgs>>(observer =>
-            {
-                var handler = new Handler<TEventArgs>(observer);
-                ev.Subscribe(target, handler);
-                return () => ev.Unsubscribe(target, handler);
-            }).Publish().RefCount();
-        }
-
-    }
-}

src/Avalonia.Base/Visual.cs

@@ -11,6 +11,7 @@ using Avalonia.Logging;
 using Avalonia.LogicalTree;
 using Avalonia.Media;
 using Avalonia.Metadata;
+using Avalonia.Reactive;
 using Avalonia.Rendering;
 using Avalonia.Rendering.Composition;
 using Avalonia.Rendering.Composition.Server;
@@ -384,52 +385,55 @@ namespace Avalonia
         protected static void AffectsRender<T>(params AvaloniaProperty[] properties)
             where T : Visual
         {
-            static void Invalidate(AvaloniaPropertyChangedEventArgs e)
-            {
-                if (e.Sender is T sender)
+            var invalidateObserver = new AnonymousObserver<AvaloniaPropertyChangedEventArgs>(
+                static e =>
                 {
-                    sender.InvalidateVisual();
-                }
-            }
-
-            static void InvalidateAndSubscribe(AvaloniaPropertyChangedEventArgs e)
-            {
-                if (e.Sender is T sender)
+                    if (e.Sender is T sender)
+                    {
+                        sender.InvalidateVisual();
+                    }
+                });
+            
+            
+            var invalidateAndSubscribeObserver = new AnonymousObserver<AvaloniaPropertyChangedEventArgs>(
+                static e =>
                 {
-                    if (e.OldValue is IAffectsRender oldValue)
+                    if (e.Sender is T sender)
                     {
-                        if (sender._affectsRenderWeakSubscriber != null)
+                        if (e.OldValue is IAffectsRender oldValue)
                         {
-                            InvalidatedWeakEvent.Unsubscribe(oldValue, sender._affectsRenderWeakSubscriber);
+                            if (sender._affectsRenderWeakSubscriber != null)
+                            {
+                                InvalidatedWeakEvent.Unsubscribe(oldValue, sender._affectsRenderWeakSubscriber);
+                            }
                         }
-                    }
 
-                    if (e.NewValue is IAffectsRender newValue)
-                    {
-                        if (sender._affectsRenderWeakSubscriber == null)
+                        if (e.NewValue is IAffectsRender newValue)
                         {
-                            sender._affectsRenderWeakSubscriber = new TargetWeakEventSubscriber<Visual, EventArgs>(
-                                sender, static (target, _, _, _) =>
-                                {
-                                    target.InvalidateVisual();
-                                });
+                            if (sender._affectsRenderWeakSubscriber == null)
+                            {
+                                sender._affectsRenderWeakSubscriber = new TargetWeakEventSubscriber<Visual, EventArgs>(
+                                    sender, static (target, _, _, _) =>
+                                    {
+                                        target.InvalidateVisual();
+                                    });
+                            }
+                            InvalidatedWeakEvent.Subscribe(newValue, sender._affectsRenderWeakSubscriber);
                         }
-                        InvalidatedWeakEvent.Subscribe(newValue, sender._affectsRenderWeakSubscriber);
-                    }
 
-                    sender.InvalidateVisual();
-                }
-            }
+                        sender.InvalidateVisual();
+                    }
+                });
 
             foreach (var property in properties)
             {
                 if (property.CanValueAffectRender())
                 {
-                    property.Changed.Subscribe(e => InvalidateAndSubscribe(e));
+                    property.Changed.Subscribe(invalidateAndSubscribeObserver);
                 }
                 else
                 {
-                    property.Changed.Subscribe(e => Invalidate(e));
+                    property.Changed.Subscribe(invalidateObserver);
                 }
             }
         }
@@ -616,23 +620,22 @@ namespace Avalonia
         /// Called when a visual's <see cref="RenderTransform"/> changes.
         /// </summary>
         /// <param name="e">The event args.</param>
-        private static void RenderTransformChanged(AvaloniaPropertyChangedEventArgs e)
+        private static void RenderTransformChanged(AvaloniaPropertyChangedEventArgs<ITransform?> e)
         {
             var sender = e.Sender as Visual;
 
             if (sender?.VisualRoot != null)
             {
-                var oldValue = e.OldValue as Transform;
-                var newValue = e.NewValue as Transform;
+                var (oldValue, newValue) = e.GetOldAndNewValue<ITransform?>();
 
-                if (oldValue != null)
+                if (oldValue is Transform oldTransform)
                 {
-                    oldValue.Changed -= sender.RenderTransformChanged;
+                    oldTransform.Changed -= sender.RenderTransformChanged;
                 }
 
-                if (newValue != null)
+                if (newValue is Transform newTransform)
                 {
-                    newValue.Changed += sender.RenderTransformChanged;
+                    newTransform.Changed += sender.RenderTransformChanged;
                 }
                 
                 sender.InvalidateVisual();

src/Avalonia.Controls.ColorPicker/Avalonia.Controls.ColorPicker.csproj

@@ -15,7 +15,6 @@
     <!-- Compatibility with old apps -->
     <EmbeddedResource Include="Themes\**\*.xaml" />
   </ItemGroup>
-  <Import Project="..\..\build\Rx.props" />
   <Import Project="..\..\build\EmbedXaml.props" />
   <Import Project="..\..\build\JetBrains.Annotations.props" />
   <Import Project="..\..\build\BuildTargets.targets" />

src/Avalonia.Controls.ColorPicker/ColorSpectrum/ColorSpectrum.cs

@@ -15,6 +15,7 @@ using Avalonia.Media;
 using Avalonia.Media.Imaging;
 using Avalonia.Threading;
 using Avalonia.Utilities;
+using Avalonia.Reactive;
 
 namespace Avalonia.Controls.Primitives
 {

src/Avalonia.Controls.DataGrid/Avalonia.Controls.DataGrid.csproj

@@ -12,7 +12,6 @@
     <!-- Compatibility with old apps -->
     <EmbeddedResource Include="Themes\**\*.xaml" />
   </ItemGroup>
-  <Import Project="..\..\build\Rx.props" />
   <Import Project="..\..\build\EmbedXaml.props" />
   <Import Project="..\..\build\JetBrains.Annotations.props" />
   <Import Project="..\..\build\BuildTargets.targets" />

src/Avalonia.Controls.DataGrid/DataGrid.cs

@@ -27,6 +27,7 @@ using Avalonia.Layout;
 using Avalonia.Controls.Metadata;
 using Avalonia.Input.GestureRecognizers;
 using Avalonia.Styling;
+using Avalonia.Reactive;
 
 namespace Avalonia.Controls
 {

src/Avalonia.Controls.DataGrid/DataGridBoundColumn.cs

@@ -4,12 +4,7 @@
 // All other rights reserved. 
 
 using Avalonia.Data;
-using Avalonia.Utilities;
 using System;
-using System.Reactive.Disposables;
-using System.Reactive.Subjects;
-using Avalonia.Reactive;
-using System.Diagnostics;
 using Avalonia.Controls.Utils;
 using Avalonia.Markup.Xaml.MarkupExtensions;
 

src/Avalonia.Controls.DataGrid/DataGridRow.cs

@@ -15,6 +15,7 @@ using Avalonia.Utilities;
 using Avalonia.VisualTree;
 using System;
 using System.Diagnostics;
+using Avalonia.Reactive;
 
 namespace Avalonia.Controls
 {
@@ -1021,11 +1022,11 @@ namespace Avalonia.Controls
                         {
                             layoutableContent.LayoutUpdated += DetailsContent_LayoutUpdated;
 
-                            _detailsContentSizeSubscription =
-                                System.Reactive.Disposables.StableCompositeDisposable.Create(
-                                    System.Reactive.Disposables.Disposable.Create(() => layoutableContent.LayoutUpdated -= DetailsContent_LayoutUpdated),
-                                    _detailsContent.GetObservable(MarginProperty)
-                                                   .Subscribe(DetailsContent_MarginChanged));
+                            _detailsContentSizeSubscription = new CompositeDisposable(2)
+                            {
+                                Disposable.Create(() => layoutableContent.LayoutUpdated -= DetailsContent_LayoutUpdated),
+                                _detailsContent.GetObservable(MarginProperty).Subscribe(DetailsContent_MarginChanged)
+                            };
 
 
                         }

src/Avalonia.Controls.DataGrid/DataGridRowGroupHeader.cs

@@ -10,7 +10,7 @@ using Avalonia.Input;
 using Avalonia.Media;
 using System;
 using System.Diagnostics;
-using System.Reactive.Linq;
+using Avalonia.Reactive;
 
 namespace Avalonia.Controls
 {

src/Avalonia.Controls.DataGrid/Utils/CellEditBinding.cs

@@ -2,7 +2,7 @@
 using Avalonia.Reactive;
 using System;
 using System.Collections.Generic;
-using System.Reactive.Subjects;
+using Avalonia.Reactive;
 
 namespace Avalonia.Controls.Utils
 {
@@ -16,16 +16,16 @@ namespace Avalonia.Controls.Utils
 
     internal class CellEditBinding : ICellEditBinding
     {
-        private readonly Subject<bool> _changedSubject = new Subject<bool>();
+        private readonly LightweightSubject<bool> _changedSubject = new();
         private readonly List<Exception> _validationErrors = new List<Exception>();
         private readonly SubjectWrapper _inner;
 
         public bool IsValid => _validationErrors.Count <= 0;
         public IEnumerable<Exception> ValidationErrors => _validationErrors;
         public IObservable<bool> ValidationChanged => _changedSubject;
-        public ISubject<object> InternalSubject => _inner;
+        public IAvaloniaSubject<object> InternalSubject => _inner;
 
-        public CellEditBinding(ISubject<object> bindingSourceSubject)
+        public CellEditBinding(IAvaloniaSubject<object> bindingSourceSubject)
         {
             _inner = new SubjectWrapper(bindingSourceSubject, this);
         }
@@ -48,16 +48,16 @@ namespace Avalonia.Controls.Utils
             return IsValid;
         }
 
-        class SubjectWrapper : LightweightObservableBase<object>, ISubject<object>, IDisposable
+        class SubjectWrapper : LightweightObservableBase<object>, IAvaloniaSubject<object>, IDisposable
         {
-            private readonly ISubject<object> _sourceSubject;
+            private readonly IAvaloniaSubject<object> _sourceSubject;
             private readonly CellEditBinding _editBinding;
             private IDisposable _subscription;
             private object _controlValue;
             private bool _isControlValueSet = false;
             private bool _settingSourceValue = false;
 
-            public SubjectWrapper(ISubject<object> bindingSourceSubject, CellEditBinding editBinding)
+            public SubjectWrapper(IAvaloniaSubject<object> bindingSourceSubject, CellEditBinding editBinding)
             {
                 _sourceSubject = bindingSourceSubject;
                 _editBinding = editBinding;

src/Avalonia.Controls/Application.cs

@@ -1,7 +1,5 @@
 using System;
 using System.Collections.Generic;
-using System.Reactive.Concurrency;
-using System.Threading;
 using Avalonia.Animation;
 using Avalonia.Controls;
 using Avalonia.Controls.ApplicationLifetimes;
@@ -231,7 +229,6 @@ namespace Avalonia
                 .Bind<IFocusManager>().ToConstant(FocusManager)
                 .Bind<IInputManager>().ToConstant(InputManager)
                 .Bind<IKeyboardNavigationHandler>().ToTransient<KeyboardNavigationHandler>()
-                .Bind<IScheduler>().ToConstant(AvaloniaScheduler.Instance)
                 .Bind<IDragDropDevice>().ToConstant(DragDropDevice.Instance);
             
             // TODO: Fix this, for now we keep this behavior since someone might be relying on it in 0.9.x

src/Avalonia.Controls/AutoCompleteBox/AutoCompleteBox.cs

@@ -10,7 +10,7 @@ using System.Collections.ObjectModel;
 using System.Collections.Specialized;
 using System.ComponentModel;
 using System.Linq;
-using System.Reactive.Linq;
+using Avalonia.Reactive;
 using System.Threading;
 using System.Threading.Tasks;
 using Avalonia.Collections;

src/Avalonia.Controls/Avalonia.Controls.csproj

@@ -6,7 +6,6 @@
     <ProjectReference Include="..\Avalonia.Base\Avalonia.Base.csproj" />
     <ProjectReference Include="..\Avalonia.Remote.Protocol\Avalonia.Remote.Protocol.csproj" />
   </ItemGroup>
-  <Import Project="..\..\build\Rx.props" />
   <Import Project="..\..\build\JetBrains.Annotations.props" />
   <Import Project="..\..\build\ApiDiff.props" />
   <Import Project="..\..\build\NullableEnable.props" />

src/Avalonia.Controls/Button.cs

@@ -5,6 +5,7 @@ using System.Windows.Input;
 using Avalonia.Automation.Peers;
 using Avalonia.Controls.Metadata;
 using Avalonia.Controls.Primitives;
+using Avalonia.Controls.Templates;
 using Avalonia.Data;
 using Avalonia.Input;
 using Avalonia.Interactivity;

src/Avalonia.Controls/ButtonSpinner.cs

@@ -1,7 +1,6 @@
-using System;
 using Avalonia.Controls.Metadata;
 using Avalonia.Controls.Primitives;
-using Avalonia.Data;
+using Avalonia.Reactive;
 using Avalonia.Input;
 using Avalonia.Interactivity;
 

src/Avalonia.Controls/CalendarDatePicker/CalendarDatePicker.cs

@@ -7,7 +7,7 @@ using System;
 using System.Collections.ObjectModel;
 using System.Diagnostics;
 using System.Globalization;
-using System.Reactive.Disposables;
+using Avalonia.Reactive;
 using Avalonia.Controls.Metadata;
 using Avalonia.Controls.Primitives;
 using Avalonia.Data;

src/Avalonia.Controls/Canvas.cs

@@ -1,7 +1,4 @@
-using System;
-using System.Reactive.Concurrency;
 using Avalonia.Input;
-using Avalonia.Layout;
 
 namespace Avalonia.Controls
 {

src/Avalonia.Controls/Chrome/CaptionButtons.cs

@@ -1,5 +1,5 @@
 using System;
-using System.Reactive.Disposables;
+using Avalonia.Reactive;
 using Avalonia.Controls.Metadata;
 using Avalonia.Controls.Primitives;
 
@@ -15,7 +15,7 @@ namespace Avalonia.Controls.Chrome
     [PseudoClasses(":minimized", ":normal", ":maximized", ":fullscreen")]
     public class CaptionButtons : TemplatedControl
     {
-        private CompositeDisposable? _disposables;
+        private IDisposable? _disposables;
 
         /// <summary>
         /// Currently attached window.
@@ -28,17 +28,14 @@ namespace Avalonia.Controls.Chrome
             {
                 HostWindow = hostWindow;
 
-                _disposables = new CompositeDisposable
-                {
-                    HostWindow.GetObservable(Window.WindowStateProperty)
+                _disposables = HostWindow.GetObservable(Window.WindowStateProperty)
                     .Subscribe(x =>
                     {
                         PseudoClasses.Set(":minimized", x == WindowState.Minimized);
                         PseudoClasses.Set(":normal", x == WindowState.Normal);
                         PseudoClasses.Set(":maximized", x == WindowState.Maximized);
                         PseudoClasses.Set(":fullscreen", x == WindowState.FullScreen);
-                    })
-                };
+                    });
             }
         }
 

src/Avalonia.Controls/Chrome/TitleBar.cs

@@ -1,5 +1,5 @@
 using System;
-using System.Reactive.Disposables;
+using Avalonia.Reactive;
 using Avalonia.Controls.Metadata;
 using Avalonia.Controls.Primitives;
 
@@ -61,7 +61,7 @@ namespace Avalonia.Controls.Chrome
 
             if (VisualRoot is Window window)
             {
-                _disposables = new CompositeDisposable
+                _disposables = new CompositeDisposable(6)
                 {
                     window.GetObservable(Window.WindowDecorationMarginProperty)
                         .Subscribe(x => UpdateSize(window)),

src/Avalonia.Controls/ComboBox.cs

@@ -1,7 +1,7 @@
 using System;
 using System.Linq;
 using Avalonia.Automation.Peers;
-using System.Reactive.Disposables;
+using Avalonia.Reactive;
 using Avalonia.Controls.Generators;
 using Avalonia.Controls.Mixins;
 using Avalonia.Controls.Presenters;

src/Avalonia.Controls/ComboBoxItem.cs

@@ -1,5 +1,4 @@
-using System;
-using System.Reactive.Linq;
+using Avalonia.Reactive;
 using Avalonia.Automation;
 using Avalonia.Automation.Peers;
 
@@ -12,8 +11,14 @@ namespace Avalonia.Controls
     {
         public ComboBoxItem()
         {
-            this.GetObservable(ComboBoxItem.IsFocusedProperty).Where(focused => focused)
-                .Subscribe(_ => (Parent as ComboBox)?.ItemFocused(this));
+            this.GetObservable(ComboBoxItem.IsFocusedProperty)
+                .Subscribe(focused =>
+                {
+                    if (focused)
+                    {
+                        (Parent as ComboBox)?.ItemFocused(this);
+                    }
+                });
         }
 
         static ComboBoxItem()

src/Avalonia.Controls/ContextMenu.cs

@@ -15,6 +15,7 @@ using Avalonia.Interactivity;
 using Avalonia.Layout;
 using Avalonia.Styling;
 using Avalonia.Automation;
+using Avalonia.Reactive;
 
 namespace Avalonia.Controls
 {

src/Avalonia.Controls/ControlExtensions.cs

@@ -1,8 +1,5 @@
 using System;
-using System.Linq;
-using Avalonia.Data;
-using Avalonia.LogicalTree;
-using Avalonia.Styling;
+using Avalonia.Reactive;
 
 namespace Avalonia.Controls
 {

src/Avalonia.Controls/DataValidationErrors.cs

@@ -1,7 +1,7 @@
 using System;
 using System.Collections.Generic;
 using System.Linq;
-using System.Reactive.Linq;
+using Avalonia.Reactive;
 using Avalonia.Controls.Metadata;
 using Avalonia.Controls.Templates;
 using Avalonia.Data;

src/Avalonia.Controls/DefinitionBase.cs

@@ -7,6 +7,7 @@ using System;
 using System.Collections;
 using System.Collections.Generic;
 using System.Diagnostics;
+using Avalonia.Reactive;
 using Avalonia.Utilities;
 
 namespace Avalonia.Controls

src/Avalonia.Controls/ExperimentalAcrylicBorder.cs

@@ -3,6 +3,7 @@ using Avalonia.Layout;
 using Avalonia.Media;
 using Avalonia.Platform;
 using System;
+using Avalonia.Reactive;
 using Avalonia.Media.Immutable;
 
 namespace Avalonia.Controls

src/Avalonia.Controls/Flyouts/FlyoutBase.cs

@@ -7,6 +7,7 @@ using Avalonia.Input.Platform;
 using Avalonia.Input.Raw;
 using Avalonia.Layout;
 using Avalonia.Logging;
+using Avalonia.Reactive;
 
 namespace Avalonia.Controls.Primitives
 {

src/Avalonia.Controls/HotkeyManager.cs

@@ -2,6 +2,7 @@ using System;
 using System.Windows.Input;
 using Avalonia.Controls.Utils;
 using Avalonia.Input;
+using Avalonia.Reactive;
 
 namespace Avalonia.Controls
 {

src/Avalonia.Controls/LayoutTransformControl.cs

@@ -4,7 +4,7 @@
 
 using System;
 using System.Diagnostics.CodeAnalysis;
-using System.Reactive.Linq;
+using Avalonia.Reactive;
 using Avalonia.Media;
 
 namespace Avalonia.Controls
@@ -424,9 +424,9 @@ namespace Avalonia.Controls
 
             if (newTransform != null)
             {
-                _transformChangedEvent = Observable.FromEventPattern<EventHandler, EventArgs>(
+                _transformChangedEvent = Observable.FromEventPattern(
                                         v => newTransform.Changed += v, v => newTransform.Changed -= v)
-                                        .Subscribe(onNext: v => ApplyLayoutTransform());
+                                        .Subscribe(_ => ApplyLayoutTransform());
             }
 
             ApplyLayoutTransform();

src/Avalonia.Controls/MenuItem.cs

@@ -1,7 +1,7 @@
 using System;
 using System.Collections.Generic;
 using System.Linq;
-using System.Reactive.Linq;
+using Avalonia.Reactive;
 using System.Windows.Input;
 using Avalonia.Automation.Peers;
 using Avalonia.Controls.Generators;
@@ -159,12 +159,13 @@ namespace Avalonia.Controls
             // menu layout.
 
             var parentSharedSizeScope = this.GetObservable(VisualParentProperty)
-                .SelectMany(x =>
+                .Select(x =>
                 {
                     var parent = x as Control;
                     return parent?.GetObservable(DefinitionBase.PrivateSharedSizeScopeProperty) ??
                            Observable.Return<DefinitionBase.SharedSizeScope?>(null);
-                });
+                })
+                .Switch();
 
             this.Bind(DefinitionBase.PrivateSharedSizeScopeProperty, parentSharedSizeScope);
         }

src/Avalonia.Controls/Mixins/DisposableMixin.cs

@@ -1,38 +0,0 @@
-using System;
-using System.Reactive.Disposables;
-
-namespace Avalonia.Controls.Mixins
-{
-    /// <summary>
-    /// Extension methods associated with the IDisposable interface.
-    /// </summary>
-    public static class DisposableMixin
-    {
-        /// <summary>
-        /// Ensures the provided disposable is disposed with the specified <see cref="CompositeDisposable"/>.
-        /// </summary>
-        /// <typeparam name="T">
-        /// The type of the disposable.
-        /// </typeparam>
-        /// <param name="item">
-        /// The disposable we are going to want to be disposed by the CompositeDisposable.
-        /// </param>
-        /// <param name="compositeDisposable">
-        /// The <see cref="CompositeDisposable"/> to which <paramref name="item"/> will be added.
-        /// </param>
-        /// <returns>
-        /// The disposable.
-        /// </returns>
-        public static T DisposeWith<T>(this T item, CompositeDisposable compositeDisposable)
-            where T : IDisposable
-        {
-            if (compositeDisposable is null)
-            {
-                throw new ArgumentNullException(nameof(compositeDisposable));
-            }
-
-            compositeDisposable.Add(item);
-            return item;
-        }
-    }
-}

src/Avalonia.Controls/Mixins/SelectableMixin.cs

@@ -1,7 +1,7 @@
 using System;
 using Avalonia.Interactivity;
 using Avalonia.Controls.Primitives;
-using Avalonia.VisualTree;
+using Avalonia.Reactive;
 
 namespace Avalonia.Controls.Mixins
 {

src/Avalonia.Controls/NativeMenu.Export.cs

@@ -1,7 +1,6 @@
 using System;
-using System.Collections.Generic;
 using Avalonia.Controls.Platform;
-using Avalonia.Data;
+using Avalonia.Reactive;
 
 namespace Avalonia.Controls
 {

src/Avalonia.Controls/NativeMenuBar.cs

@@ -2,6 +2,7 @@ using System;
 using System.Diagnostics.CodeAnalysis;
 using Avalonia.Controls.Primitives;
 using Avalonia.Interactivity;
+using Avalonia.Reactive;
 
 namespace Avalonia.Controls
 {

src/Avalonia.Controls/NativeMenuItem.cs

@@ -4,6 +4,7 @@ using Avalonia.Input;
 using Avalonia.Media.Imaging;
 using Avalonia.Metadata;
 using Avalonia.Utilities;
+using Avalonia.Reactive;
 
 namespace Avalonia.Controls
 {