First steps to reduce operator object layouts

Rx.NET/Source/src/System.Reactive/Concurrency/Synchronization.ObserveOn.cs

@@ -7,101 +7,105 @@
 
 namespace System.Reactive.Concurrency
 {
-    internal sealed class ObserveOn<TSource> : Producer<TSource>
+    internal static class ObserveOn<TSource>
     {
-        private readonly IObservable<TSource> _source;
-        private readonly IScheduler _scheduler;
-        private readonly SynchronizationContext _context;
-
-        public ObserveOn(IObservable<TSource> source, IScheduler scheduler)
-        {
-            _source = source;
-            _scheduler = scheduler;
-        }
-
-        public ObserveOn(IObservable<TSource> source, SynchronizationContext context)
+        internal sealed class Scheduler : Producer<TSource>
         {
-            _source = source;
-            _context = context;
-        }
+            private readonly IObservable<TSource> _source;
+            private readonly IScheduler _scheduler;
 
-        [System.Diagnostics.CodeAnalysis.SuppressMessage("Microsoft.Design", "CA1062:Validate arguments of public methods", MessageId = "2", Justification = "Visibility restricted to friend assemblies. Those should be correct by inspection.")]
-        protected override IDisposable Run(IObserver<TSource> observer, IDisposable cancel, Action<IDisposable> setSink)
-        {
-            if (_context != null)
+            public Scheduler(IObservable<TSource> source, IScheduler scheduler)
             {
-                var sink = new ObserveOnSink(this, observer, cancel);
-                setSink(sink);
-                return sink.Run();
+                _source = source;
+                _scheduler = scheduler;
             }
-            else
+
+            [System.Diagnostics.CodeAnalysis.SuppressMessage("Microsoft.Design", "CA1062:Validate arguments of public methods", MessageId = "2", Justification = "Visibility restricted to friend assemblies. Those should be correct by inspection.")]
+            protected override IDisposable Run(IObserver<TSource> observer, IDisposable cancel, Action<IDisposable> setSink)
             {
                 var sink = new ObserveOnObserver<TSource>(_scheduler, observer, cancel);
                 setSink(sink);
                 return _source.SubscribeSafe(sink);
             }
         }
 
-        private sealed class ObserveOnSink : Sink<TSource>, IObserver<TSource>
+        internal sealed class Context : Producer<TSource>
         {
-            private readonly ObserveOn<TSource> _parent;
+            private readonly IObservable<TSource> _source;
+            private readonly SynchronizationContext _context;
 
-            public ObserveOnSink(ObserveOn<TSource> parent, IObserver<TSource> observer, IDisposable cancel)
-                : base(observer, cancel)
+            public Context(IObservable<TSource> source, SynchronizationContext context)
             {
-                _parent = parent;
+                _source = source;
+                _context = context;
             }
 
-            public IDisposable Run()
+            [System.Diagnostics.CodeAnalysis.SuppressMessage("Microsoft.Design", "CA1062:Validate arguments of public methods", MessageId = "2", Justification = "Visibility restricted to friend assemblies. Those should be correct by inspection.")]
+            protected override IDisposable Run(IObserver<TSource> observer, IDisposable cancel, Action<IDisposable> setSink)
             {
-                //
-                // The interactions with OperationStarted/OperationCompleted below allow
-                // for test frameworks to wait until a whole sequence is observed, running
-                // asserts on a per-message level. Also, for ASP.NET pages, the use of the
-                // built-in synchronization context would allow processing to finished in
-                // its entirety before moving on with the page lifecycle.
-                //
-                _parent._context.OperationStarted();
-
-                var d = _parent._source.SubscribeSafe(this);
-                var c = Disposable.Create(() =>
-                {
-                    _parent._context.OperationCompleted();
-                });
-
-                return StableCompositeDisposable.Create(d, c);
+                var sink = new _(_context, observer, cancel);
+                setSink(sink);
+                return sink.Run(_source);
             }
 
-            public void OnNext(TSource value)
+            private sealed class _ : Sink<TSource>, IObserver<TSource>
             {
-                _parent._context.Post(OnNextPosted, value);
-            }
+                private readonly SynchronizationContext _context;
 
-            public void OnError(Exception error)
-            {
-                _parent._context.Post(OnErrorPosted, error);
-            }
+                public _(SynchronizationContext context, IObserver<TSource> observer, IDisposable cancel)
+                    : base(observer, cancel)
+                {
+                    _context = context;
+                }
 
-            public void OnCompleted()
-            {
-                _parent._context.Post(OnCompletedPosted, state: null);
-            }
+                public IDisposable Run(IObservable<TSource> source)
+                {
+                    //
+                    // The interactions with OperationStarted/OperationCompleted below allow
+                    // for test frameworks to wait until a whole sequence is observed, running
+                    // asserts on a per-message level. Also, for ASP.NET pages, the use of the
+                    // built-in synchronization context would allow processing to finished in
+                    // its entirety before moving on with the page lifecycle.
+                    //
+                    _context.OperationStarted();
+
+                    var d = source.SubscribeSafe(this);
+                    var c = Disposable.Create(_context.OperationCompleted);
+
+                    return StableCompositeDisposable.Create(d, c);
+                }
+
+                public void OnNext(TSource value)
+                {
+                    _context.Post(OnNextPosted, value);
+                }
 
-            private void OnNextPosted(object value)
-            {
-                _observer.OnNext((TSource)value);
-            }
+                public void OnError(Exception error)
+                {
+                    _context.Post(OnErrorPosted, error);
+                }
 
-            private void OnErrorPosted(object error)
-            {
-                _observer.OnError((Exception)error);
-                Dispose();
-            }
+                public void OnCompleted()
+                {
+                    _context.Post(OnCompletedPosted, state: null);
+                }
 
-            private void OnCompletedPosted(object ignored)
-            {
-                _observer.OnCompleted();
-                Dispose();
+                private void OnNextPosted(object value)
+                {
+                    _observer.OnNext((TSource)value);
+                }
+
+                private void OnErrorPosted(object error)
+                {
+                    _observer.OnError((Exception)error);
+                    Dispose();
+                }
+
+                private void OnCompletedPosted(object ignored)
+                {
+                    _observer.OnCompleted();
+                    Dispose();
+                }
             }
         }
     }

Rx.NET/Source/src/System.Reactive/Concurrency/Synchronization.cs

@@ -102,7 +102,7 @@ public static IObservable<TSource> ObserveOn<TSource>(IObservable<TSource> sourc
             if (scheduler == null)
                 throw new ArgumentNullException(nameof(scheduler));
 
-            return new ObserveOn<TSource>(source, scheduler);
+            return new ObserveOn<TSource>.Scheduler(source, scheduler);
         }
 
         /// <summary>
@@ -120,7 +120,7 @@ public static IObservable<TSource> ObserveOn<TSource>(IObservable<TSource> sourc
             if (context == null)
                 throw new ArgumentNullException(nameof(context));
 
-            return new ObserveOn<TSource>(source, context);
+            return new ObserveOn<TSource>.Context(source, context);
         }
 
         #endregion

Rx.NET/Source/src/System.Reactive/Internal/Helpers.cs

@@ -39,5 +39,34 @@ public static IObservable<T> Unpack<T>(IObservable<T> source)
 
             return source;
         }
+
+        public static bool All(this bool[] values)
+        {
+            foreach (var value in values)
+            {
+                if (!value)
+                {
+                    return false;
+                }
+            }
+
+            return true;
+        }
+
+        public static bool AllExcept(this bool[] values, int index)
+        {
+            for (var i = 0; i < values.Length; i++)
+            {
+                if (i != index)
+                {
+                    if (!values[i])
+                    {
+                        return false;
+                    }
+                }
+            }
+
+            return true;
+        }
     }
 }

Rx.NET/Source/src/System.Reactive/Linq/GroupedObservable.cs

@@ -7,29 +7,25 @@
 
 namespace System.Reactive.Linq
 {
-    class GroupedObservable<TKey, TElement> : ObservableBase<TElement>, IGroupedObservable<TKey, TElement>
+    internal sealed class GroupedObservable<TKey, TElement> : ObservableBase<TElement>, IGroupedObservable<TKey, TElement>
     {
-        private readonly TKey _key;
         private readonly IObservable<TElement> _subject;
         private readonly RefCountDisposable _refCount;
 
         public GroupedObservable(TKey key, ISubject<TElement> subject, RefCountDisposable refCount)
         {
-            _key = key;
+            Key = key;
             _subject = subject;
             _refCount = refCount;
         }
 
         public GroupedObservable(TKey key, ISubject<TElement> subject)
         {
-            _key = key;
+            Key = key;
             _subject = subject;
         }
 
-        public TKey Key
-        {
-            get { return _key; }
-        }
+        public TKey Key { get; }
 
         protected override IDisposable SubscribeCore(IObserver<TElement> observer)
         {

Rx.NET/Source/src/System.Reactive/Linq/Observable/AddRef.cs

@@ -26,7 +26,7 @@ protected override IDisposable Run(IObserver<TSource> observer, IDisposable canc
             return _source.SubscribeSafe(sink);
         }
 
-        class _ : Sink<TSource>, IObserver<TSource>
+        private sealed class _ : Sink<TSource>, IObserver<TSource>
         {
             public _(IObserver<TSource> observer, IDisposable cancel)
                 : base(observer, cancel)