Save allocations in Zip (#635)

* Let ZipObserver inherit from SafeObserver which inlines the SingleAssignmentDisposables currently used. We save n allocations for the n-ary Zip.

* Override Dispose in ZipObserver so it can clear its own queue, that saves the allocation of an anonymous disposable, a closure and a delegate.

Rx.NET/Source/src/System.Reactive/Linq/Observable/Zip.Generated.tt

@@ -76,15 +76,14 @@ for (var j = 1; j <= i; j++)
 
             public void Run(<#=os#>)
             {
-                var disposables = new IDisposable[<#=i + 1#>];
+                var disposables = new IDisposable[<#=i#>];
 
 <#
 for (var j = 1; j <= i; j++)
 {
 #>
-                var d<#=j#> = new SingleAssignmentDisposable();
-                disposables[<#=j - 1#>] = d<#=j#>;
-                _observer<#=j#> = new ZipObserver<T<#=j#>>(_gate, this, <#=j - 1#>, d<#=j#>);
+                _observer<#=j#> = new ZipObserver<T<#=j#>>(_gate, this, <#=j - 1#>);
+                disposables[<#=j - 1#>] = _observer<#=j#>;
                 base.Queues[<#= j - 1#>] = _observer<#=j#>.Values;
 
 <#
@@ -93,23 +92,11 @@ for (var j = 1; j <= i; j++)
 for (var j = 1; j <= i; j++)
 {
 #>
-                d<#=j#>.Disposable = source<#=j#>.SubscribeSafe(_observer<#=j#>);
+                _observer<#=j#>.SetResource(source<#=j#>.SubscribeSafe(_observer<#=j#>));
 <#
 }
 #>
 
-                disposables[<#=i#>] = Disposable.Create(() =>
-                {
-<#
-for (var j = 1; j <= i; j++)
-{
-#>
-                    _observer<#=j#>.Values.Clear();
-<#
-}
-#>
-                });
-
                 SetUpstream(StableCompositeDisposable.Create(disposables));
             }
 

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

@@ -476,26 +476,37 @@ namespace System.Reactive.Linq.ObservableImpl
         }
     }
 
-    internal sealed class ZipObserver<T> : IObserver<T>
+    internal sealed class ZipObserver<T> : SafeObserver<T>
     {
         private readonly object _gate;
         private readonly IZip _parent;
         private readonly int _index;
-        private readonly IDisposable _self;
         private readonly Queue<T> _values;
 
-        public ZipObserver(object gate, IZip parent, int index, IDisposable self)
+        public ZipObserver(object gate, IZip parent, int index)
         {
             _gate = gate;
             _parent = parent;
             _index = index;
-            _self = self;
             _values = new Queue<T>();
         }
 
         public Queue<T> Values => _values;
 
-        public void OnNext(T value)
+        protected override void Dispose(bool disposing)
+        {
+            base.Dispose(disposing);
+
+            if (disposing)
+            {
+                lock (_gate)
+                {
+                    _values.Clear();
+                }
+            }
+        }
+
+        public override void OnNext(T value)
         {
             lock (_gate)
             {
@@ -504,9 +515,9 @@ namespace System.Reactive.Linq.ObservableImpl
             }
         }
 
-        public void OnError(Exception error)
+        public override void OnError(Exception error)
         {
-            _self.Dispose();
+            Dispose();
 
             lock (_gate)
             {
@@ -514,9 +525,9 @@ namespace System.Reactive.Linq.ObservableImpl
             }
         }
 
-        public void OnCompleted()
+        public override void OnCompleted()
         {
-            _self.Dispose();
+            Dispose();
 
             lock (_gate)
             {