Less allocations in CombineLatest (#637)

* Add a benchmark runner with some initial benchmarks for the Zip and CombineLatest-operators.

* Save the allocations of SingleAssignmentDisposables in the n-ary CombineLatest-operators (for n >= 3).

Rx.NET/Source/System.Reactive.sln

@@ -56,6 +56,10 @@ Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "System.Reactive.Interfaces"
 EndProject
 Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Tests.System.Reactive.ApiApprovals", "tests\Tests.System.Reactive.ApiApprovals\Tests.System.Reactive.ApiApprovals.csproj", "{01CCDA6D-4D00-4DF2-82B0-359FD5E0CDC6}"
 EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Benchmarks.System.Reactive", "benchmarks\Benchmarks.System.Reactive\Benchmarks.System.Reactive.csproj", "{5C7906F6-232E-455C-9269-68EF84F393C9}"
+EndProject
+Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Benchmarks", "Benchmarks", "{C8E480ED-B592-4341-A0C9-183E822EB6B9}"
+EndProject
 Global
 	GlobalSection(SolutionConfigurationPlatforms) = preSolution
 		Debug|Any CPU = Debug|Any CPU
@@ -330,6 +334,22 @@ Global
 		{01CCDA6D-4D00-4DF2-82B0-359FD5E0CDC6}.Release|x64.Build.0 = Release|Any CPU
 		{01CCDA6D-4D00-4DF2-82B0-359FD5E0CDC6}.Release|x86.ActiveCfg = Release|Any CPU
 		{01CCDA6D-4D00-4DF2-82B0-359FD5E0CDC6}.Release|x86.Build.0 = Release|Any CPU
+		{5C7906F6-232E-455C-9269-68EF84F393C9}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{5C7906F6-232E-455C-9269-68EF84F393C9}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{5C7906F6-232E-455C-9269-68EF84F393C9}.Debug|ARM.ActiveCfg = Debug|Any CPU
+		{5C7906F6-232E-455C-9269-68EF84F393C9}.Debug|ARM.Build.0 = Debug|Any CPU
+		{5C7906F6-232E-455C-9269-68EF84F393C9}.Debug|x64.ActiveCfg = Debug|Any CPU
+		{5C7906F6-232E-455C-9269-68EF84F393C9}.Debug|x64.Build.0 = Debug|Any CPU
+		{5C7906F6-232E-455C-9269-68EF84F393C9}.Debug|x86.ActiveCfg = Debug|Any CPU
+		{5C7906F6-232E-455C-9269-68EF84F393C9}.Debug|x86.Build.0 = Debug|Any CPU
+		{5C7906F6-232E-455C-9269-68EF84F393C9}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{5C7906F6-232E-455C-9269-68EF84F393C9}.Release|Any CPU.Build.0 = Release|Any CPU
+		{5C7906F6-232E-455C-9269-68EF84F393C9}.Release|ARM.ActiveCfg = Release|Any CPU
+		{5C7906F6-232E-455C-9269-68EF84F393C9}.Release|ARM.Build.0 = Release|Any CPU
+		{5C7906F6-232E-455C-9269-68EF84F393C9}.Release|x64.ActiveCfg = Release|Any CPU
+		{5C7906F6-232E-455C-9269-68EF84F393C9}.Release|x64.Build.0 = Release|Any CPU
+		{5C7906F6-232E-455C-9269-68EF84F393C9}.Release|x86.ActiveCfg = Release|Any CPU
+		{5C7906F6-232E-455C-9269-68EF84F393C9}.Release|x86.Build.0 = Release|Any CPU
 	EndGlobalSection
 	GlobalSection(SolutionProperties) = preSolution
 		HideSolutionNode = FALSE
@@ -348,6 +368,7 @@ Global
 		{E9BB350B-D4EF-42E1-B4E2-14058AC6809B} = {A0F39260-B8F8-4FCB-9679-0ED917A22BDF}
 		{A31DDC7E-E883-4DBD-8FB8-D7DFC56059F6} = {A0F39260-B8F8-4FCB-9679-0ED917A22BDF}
 		{01CCDA6D-4D00-4DF2-82B0-359FD5E0CDC6} = {D324579D-CBE6-4867-8980-D7842C7C45A2}
+		{5C7906F6-232E-455C-9269-68EF84F393C9} = {C8E480ED-B592-4341-A0C9-183E822EB6B9}
 	EndGlobalSection
 	GlobalSection(ExtensibilityGlobals) = postSolution
 		SolutionGuid = {2483F58F-A8D6-4FFE-A3C1-10F3A36DBE69}

Rx.NET/Source/benchmarks/Benchmarks.System.Reactive/Benchmarks.System.Reactive.csproj

@@ -0,0 +1,16 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+  <PropertyGroup>
+    <OutputType>Exe</OutputType>
+    <TargetFramework>net46</TargetFramework>
+  </PropertyGroup>
+
+  <ItemGroup>
+    <PackageReference Include="BenchmarkDotNet" Version="0.10.14" />
+  </ItemGroup>
+
+  <ItemGroup>
+    <ProjectReference Include="..\..\tests\Tests.System.Reactive\Tests.System.Reactive.csproj" />
+  </ItemGroup>
+
+</Project>

Rx.NET/Source/benchmarks/Benchmarks.System.Reactive/CombineLatestBenchmark.cs

@@ -0,0 +1,105 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the Apache 2.0 License.
+// See the LICENSE file in the project root for more information.
+
+using BenchmarkDotNet.Attributes;
+using ReactiveTests.Tests;
+
+namespace Benchmarks.System.Reactive
+{
+    [MemoryDiagnoser]
+    public class CombineLatestBenchmark
+    {
+        private CombineLatestTest _zipTest = new CombineLatestTest();
+
+        [Benchmark]
+        public void CombineLatest_Typical2()
+        {
+            _zipTest.CombineLatest_Typical2();
+        }
+
+        [Benchmark]
+        public void CombineLatest_Typical3()
+        {
+            _zipTest.CombineLatest_Typical3();
+        }
+
+        [Benchmark]
+        public void CombineLatest_Typical4()
+        {
+            _zipTest.CombineLatest_Typical4();
+        }
+
+        [Benchmark]
+        public void CombineLatest_Typical5()
+        {
+            _zipTest.CombineLatest_Typical5();
+        }
+
+        [Benchmark]
+        public void CombineLatest_Typical6()
+        {
+            _zipTest.CombineLatest_Typical6();
+        }
+
+        [Benchmark]
+        public void CombineLatest_Typical7()
+        {
+            _zipTest.CombineLatest_Typical7();
+        }
+
+        [Benchmark]
+        public void CombineLatest_Typical8()
+        {
+            _zipTest.CombineLatest_Typical8();
+        }
+
+        [Benchmark]
+        public void CombineLatest_Typical9()
+        {
+            _zipTest.CombineLatest_Typical9();
+        }
+
+        [Benchmark]
+        public void CombineLatest_Typical10()
+        {
+            _zipTest.CombineLatest_Typical10();
+        }
+
+        [Benchmark]
+        public void CombineLatest_Typical11()
+        {
+            _zipTest.CombineLatest_Typical11();
+        }
+
+        [Benchmark]
+        public void CombineLatest_Typical12()
+        {
+            _zipTest.CombineLatest_Typical12();
+        }
+        
+        [Benchmark]
+        public void CombineLatest_Typical13()
+        {
+            _zipTest.CombineLatest_Typical13();
+        }
+
+        [Benchmark]
+        public void CombineLatest_Typical14()
+        {
+            _zipTest.CombineLatest_Typical14();
+        }
+
+        [Benchmark]
+        public void CombineLatest_Typical15()
+        {
+            _zipTest.CombineLatest_Typical15();
+        }
+
+        [Benchmark]
+        public void CombineLatest_Typical16()
+        {
+            _zipTest.CombineLatest_Typical16();
+        }
+    }
+}

Rx.NET/Source/benchmarks/Benchmarks.System.Reactive/Program.cs

@@ -0,0 +1,23 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the Apache 2.0 License.
+// See the LICENSE file in the project root for more information.
+
+using System;
+using BenchmarkDotNet.Running;
+
+namespace Benchmarks.System.Reactive
+{
+    class Program
+    {
+        static void Main()
+        {
+            var switcher = new BenchmarkSwitcher(new[] {
+                typeof(ZipBenchmark),
+                typeof(CombineLatestBenchmark)
+            });
+
+            switcher.Run();
+            Console.ReadLine();
+        }
+    }
+}

Rx.NET/Source/benchmarks/Benchmarks.System.Reactive/ZipBenchmark.cs

@@ -0,0 +1,135 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the Apache 2.0 License.
+// See the LICENSE file in the project root for more information.
+
+using BenchmarkDotNet.Attributes;
+using ReactiveTests.Tests;
+
+namespace Benchmarks.System.Reactive
+{
+    [MemoryDiagnoser]
+    public class ZipBenchmark
+    {
+        private ZipTest _zipTest = new ZipTest();
+
+        [Benchmark]
+        public void Zip_NAry_Asymmetric()
+        {
+            _zipTest.Zip_NAry_Asymmetric();
+        }
+
+        [Benchmark]
+        public void Zip_NAry_Asymmetric_Selector()
+        {
+            _zipTest.Zip_NAry_Asymmetric_Selector();
+        }
+
+        [Benchmark]
+        public void Zip_NAry_Symmetric()
+        {
+            _zipTest.Zip_NAry_Symmetric();
+        }
+
+        [Benchmark]
+        public void Zip_NAry_Symmetric_Selector()
+        {
+            _zipTest.Zip_NAry_Symmetric_Selector();
+        }
+
+        [Benchmark]
+        public void Zip_NAry_Enumerable_Simple()
+        {
+            _zipTest.Zip_NAry_Enumerable_Simple();
+        }
+
+        [Benchmark]
+        public void Zip_AllCompleted2()
+        {
+            _zipTest.Zip_AllCompleted2();
+        }
+
+        [Benchmark]
+        public void Zip_AllCompleted3()
+        {
+            _zipTest.Zip_AllCompleted3();
+        }
+
+        [Benchmark]
+        public void Zip_AllCompleted4()
+        {
+            _zipTest.Zip_AllCompleted4();
+        }
+
+        [Benchmark]
+        public void Zip_AllCompleted5()
+        {
+            _zipTest.Zip_AllCompleted5();
+        }
+
+        [Benchmark]
+        public void Zip_AllCompleted6()
+        {
+            _zipTest.Zip_AllCompleted6();
+        }
+
+        [Benchmark]
+        public void Zip_AllCompleted7()
+        {
+            _zipTest.Zip_AllCompleted7();
+        }
+        
+        [Benchmark]
+        public void Zip_AllCompleted8()
+        {
+            _zipTest.Zip_AllCompleted8();
+        }
+
+        [Benchmark]
+        public void Zip_AllCompleted9()
+        {
+            _zipTest.Zip_AllCompleted9();
+        }
+
+        [Benchmark]
+        public void Zip_AllCompleted10()
+        {
+            _zipTest.Zip_AllCompleted10();
+        }
+
+        [Benchmark]
+        public void Zip_AllCompleted11()
+        {
+            _zipTest.Zip_AllCompleted11();
+        }
+
+        [Benchmark]
+        public void Zip_AllCompleted12()
+        {
+            _zipTest.Zip_AllCompleted12();
+        }
+
+        [Benchmark]
+        public void Zip_AllCompleted13()
+        {
+            _zipTest.Zip_AllCompleted13();
+        }
+
+        [Benchmark]
+        public void Zip_AllCompleted14()
+        {
+            _zipTest.Zip_AllCompleted14();
+        }
+
+        [Benchmark]
+        public void Zip_AllCompleted15()
+        {
+            _zipTest.Zip_AllCompleted15();
+        }
+
+        [Benchmark]
+        public void Zip_AllCompleted16()
+        {
+            _zipTest.Zip_AllCompleted16();
+        }
+    }
+}

Rx.NET/Source/src/System.Reactive/Linq/Observable/CombineLatest.Generated.cs

@@ -10,7 +10,7 @@ namespace System.Reactive.Linq.ObservableImpl
 
     /* The following code is generated by a T4 template. */
 
-    #region CombineLatest auto-generated code (06.06.2018 11:24:41)
+    #region CombineLatest auto-generated code (21.06.2018 10:59:51)
 
     internal sealed class CombineLatest<T1, T2, T3, TResult> : Producer<TResult, CombineLatest<T1, T2, T3, TResult>._>
     {
@@ -47,17 +47,15 @@ namespace System.Reactive.Linq.ObservableImpl
 
             public void Run(IObservable<T1> source1, IObservable<T2> source2, IObservable<T3> source3)
             {
-                var subscriptions = new SingleAssignmentDisposable[3];
-                for (int i = 0; i < 3; i++)
-                    subscriptions[i] = new SingleAssignmentDisposable();
+                var subscriptions = new IDisposable[3];
 
-                _observer1 = new CombineLatestObserver<T1>(_gate, this, 0, subscriptions[0]);
-                _observer2 = new CombineLatestObserver<T2>(_gate, this, 1, subscriptions[1]);
-                _observer3 = new CombineLatestObserver<T3>(_gate, this, 2, subscriptions[2]);
+                subscriptions[0] = _observer1 = new CombineLatestObserver<T1>(_gate, this, 0);
+                subscriptions[1] = _observer2 = new CombineLatestObserver<T2>(_gate, this, 1);
+                subscriptions[2] = _observer3 = new CombineLatestObserver<T3>(_gate, this, 2);
 
-                subscriptions[0].Disposable = source1.SubscribeSafe(_observer1);
-                subscriptions[1].Disposable = source2.SubscribeSafe(_observer2);
-                subscriptions[2].Disposable = source3.SubscribeSafe(_observer3);
+                _observer1.SetResource(source1.SubscribeSafe(_observer1));
+                _observer2.SetResource(source2.SubscribeSafe(_observer2));
+                _observer3.SetResource(source3.SubscribeSafe(_observer3));
 
                 SetUpstream(StableCompositeDisposable.Create(subscriptions));
             }
@@ -104,19 +102,17 @@ namespace System.Reactive.Linq.ObservableImpl
 
             public void Run(IObservable<T1> source1, IObservable<T2> source2, IObservable<T3> source3, IObservable<T4> source4)
             {
-                var subscriptions = new SingleAssignmentDisposable[4];
-                for (int i = 0; i < 4; i++)
-                    subscriptions[i] = new SingleAssignmentDisposable();
+                var subscriptions = new IDisposable[4];
 
-                _observer1 = new CombineLatestObserver<T1>(_gate, this, 0, subscriptions[0]);
-                _observer2 = new CombineLatestObserver<T2>(_gate, this, 1, subscriptions[1]);
-                _observer3 = new CombineLatestObserver<T3>(_gate, this, 2, subscriptions[2]);
-                _observer4 = new CombineLatestObserver<T4>(_gate, this, 3, subscriptions[3]);
+                subscriptions[0] = _observer1 = new CombineLatestObserver<T1>(_gate, this, 0);
+                subscriptions[1] = _observer2 = new CombineLatestObserver<T2>(_gate, this, 1);
+                subscriptions[2] = _observer3 = new CombineLatestObserver<T3>(_gate, this, 2);
+                subscriptions[3] = _observer4 = new CombineLatestObserver<T4>(_gate, this, 3);
 
-                subscriptions[0].Disposable = source1.SubscribeSafe(_observer1);
-                subscriptions[1].Disposable = source2.SubscribeSafe(_observer2);
-                subscriptions[2].Disposable = source3.SubscribeSafe(_observer3);
-                subscriptions[3].Disposable = source4.SubscribeSafe(_observer4);
+                _observer1.SetResource(source1.SubscribeSafe(_observer1));
+                _observer2.SetResource(source2.SubscribeSafe(_observer2));
+                _observer3.SetResource(source3.SubscribeSafe(_observer3));
+                _observer4.SetResource(source4.SubscribeSafe(_observer4));
 
                 SetUpstream(StableCompositeDisposable.Create(subscriptions));
             }
@@ -166,21 +162,19 @@ namespace System.Reactive.Linq.ObservableImpl
 
             public void Run(IObservable<T1> source1, IObservable<T2> source2, IObservable<T3> source3, IObservable<T4> source4, IObservable<T5> source5)
             {
-                var subscriptions = new SingleAssignmentDisposable[5];
-                for (int i = 0; i < 5; i++)
-                    subscriptions[i] = new SingleAssignmentDisposable();
-
-                _observer1 = new CombineLatestObserver<T1>(_gate, this, 0, subscriptions[0]);
-                _observer2 = new CombineLatestObserver<T2>(_gate, this, 1, subscriptions[1]);
-                _observer3 = new CombineLatestObserver<T3>(_gate, this, 2, subscriptions[2]);
-                _observer4 = new CombineLatestObserver<T4>(_gate, this, 3, subscriptions[3]);
-                _observer5 = new CombineLatestObserver<T5>(_gate, this, 4, subscriptions[4]);
-
-                subscriptions[0].Disposable = source1.SubscribeSafe(_observer1);
-                subscriptions[1].Disposable = source2.SubscribeSafe(_observer2);
-                subscriptions[2].Disposable = source3.SubscribeSafe(_observer3);
-                subscriptions[3].Disposable = source4.SubscribeSafe(_observer4);
-                subscriptions[4].Disposable = source5.SubscribeSafe(_observer5);
+                var subscriptions = new IDisposable[5];
+
+                subscriptions[0] = _observer1 = new CombineLatestObserver<T1>(_gate, this, 0);
+                subscriptions[1] = _observer2 = new CombineLatestObserver<T2>(_gate, this, 1);
+                subscriptions[2] = _observer3 = new CombineLatestObserver<T3>(_gate, this, 2);
+                subscriptions[3] = _observer4 = new CombineLatestObserver<T4>(_gate, this, 3);
+                subscriptions[4] = _observer5 = new CombineLatestObserver<T5>(_gate, this, 4);
+
+                _observer1.SetResource(source1.SubscribeSafe(_observer1));
+                _observer2.SetResource(source2.SubscribeSafe(_observer2));
+                _observer3.SetResource(source3.SubscribeSafe(_observer3));
+                _observer4.SetResource(source4.SubscribeSafe(_observer4));
+                _observer5.SetResource(source5.SubscribeSafe(_observer5));
 
                 SetUpstream(StableCompositeDisposable.Create(subscriptions));
             }
@@ -233,23 +227,21 @@ namespace System.Reactive.Linq.ObservableImpl
 
             public void Run(IObservable<T1> source1, IObservable<T2> source2, IObservable<T3> source3, IObservable<T4> source4, IObservable<T5> source5, IObservable<T6> source6)
             {
-                var subscriptions = new SingleAssignmentDisposable[6];
-                for (int i = 0; i < 6; i++)
-                    subscriptions[i] = new SingleAssignmentDisposable();
-
-                _observer1 = new CombineLatestObserver<T1>(_gate, this, 0, subscriptions[0]);
-                _observer2 = new CombineLatestObserver<T2>(_gate, this, 1, subscriptions[1]);
-                _observer3 = new CombineLatestObserver<T3>(_gate, this, 2, subscriptions[2]);
-                _observer4 = new CombineLatestObserver<T4>(_gate, this, 3, subscriptions[3]);
-                _observer5 = new CombineLatestObserver<T5>(_gate, this, 4, subscriptions[4]);
-                _observer6 = new CombineLatestObserver<T6>(_gate, this, 5, subscriptions[5]);
-
-                subscriptions[0].Disposable = source1.SubscribeSafe(_observer1);
-                subscriptions[1].Disposable = source2.SubscribeSafe(_observer2);
-                subscriptions[2].Disposable = source3.SubscribeSafe(_observer3);
-                subscriptions[3].Disposable = source4.SubscribeSafe(_observer4);
-                subscriptions[4].Disposable = source5.SubscribeSafe(_observer5);
-                subscriptions[5].Disposable = source6.SubscribeSafe(_observer6);
+                var subscriptions = new IDisposable[6];
+
+                subscriptions[0] = _observer1 = new CombineLatestObserver<T1>(_gate, this, 0);
+                subscriptions[1] = _observer2 = new CombineLatestObserver<T2>(_gate, this, 1);
+                subscriptions[2] = _observer3 = new CombineLatestObserver<T3>(_gate, this, 2);
+                subscriptions[3] = _observer4 = new CombineLatestObserver<T4>(_gate, this, 3);
+                subscriptions[4] = _observer5 = new CombineLatestObserver<T5>(_gate, this, 4);
+                subscriptions[5] = _observer6 = new CombineLatestObserver<T6>(_gate, this, 5);
+
+                _observer1.SetResource(source1.SubscribeSafe(_observer1));
+                _observer2.SetResource(source2.SubscribeSafe(_observer2));
+                _observer3.SetResource(source3.SubscribeSafe(_observer3));
+                _observer4.SetResource(source4.SubscribeSafe(_observer4));
+                _observer5.SetResource(source5.SubscribeSafe(_observer5));
+                _observer6.SetResource(source6.SubscribeSafe(_observer6));
 
                 SetUpstream(StableCompositeDisposable.Create(subscriptions));
             }
@@ -305,25 +297,23 @@ namespace System.Reactive.Linq.ObservableImpl
 
             public void Run(IObservable<T1> source1, IObservable<T2> source2, IObservable<T3> source3, IObservable<T4> source4, IObservable<T5> source5, IObservable<T6> source6, IObservable<T7> source7)
             {
-                var subscriptions = new SingleAssignmentDisposable[7];
-                for (int i = 0; i < 7; i++)
-                    subscriptions[i] = new SingleAssignmentDisposable();
-
-                _observer1 = new CombineLatestObserver<T1>(_gate, this, 0, subscriptions[0]);
-                _observer2 = new CombineLatestObserver<T2>(_gate, this, 1, subscriptions[1]);
-                _observer3 = new CombineLatestObserver<T3>(_gate, this, 2, subscriptions[2]);
-                _observer4 = new CombineLatestObserver<T4>(_gate, this, 3, subscriptions[3]);
-                _observer5 = new CombineLatestObserver<T5>(_gate, this, 4, subscriptions[4]);
-                _observer6 = new CombineLatestObserver<T6>(_gate, this, 5, subscriptions[5]);
-                _observer7 = new CombineLatestObserver<T7>(_gate, this, 6, subscriptions[6]);
-
-                subscriptions[0].Disposable = source1.SubscribeSafe(_observer1);
-                subscriptions[1].Disposable = source2.SubscribeSafe(_observer2);
-                subscriptions[2].Disposable = source3.SubscribeSafe(_observer3);
-                subscriptions[3].Disposable = source4.SubscribeSafe(_observer4);
-                subscriptions[4].Disposable = source5.SubscribeSafe(_observer5);
-                subscriptions[5].Disposable = source6.SubscribeSafe(_observer6);
-                subscriptions[6].Disposable = source7.SubscribeSafe(_observer7);
+                var subscriptions = new IDisposable[7];
+
+                subscriptions[0] = _observer1 = new CombineLatestObserver<T1>(_gate, this, 0);
+                subscriptions[1] = _observer2 = new CombineLatestObserver<T2>(_gate, this, 1);
+                subscriptions[2] = _observer3 = new CombineLatestObserver<T3>(_gate, this, 2);
+                subscriptions[3] = _observer4 = new CombineLatestObserver<T4>(_gate, this, 3);
+                subscriptions[4] = _observer5 = new CombineLatestObserver<T5>(_gate, this, 4);
+                subscriptions[5] = _observer6 = new CombineLatestObserver<T6>(_gate, this, 5);
+                subscriptions[6] = _observer7 = new CombineLatestObserver<T7>(_gate, this, 6);
+
+                _observer1.SetResource(source1.SubscribeSafe(_observer1));
+                _observer2.SetResource(source2.SubscribeSafe(_observer2));
+                _observer3.SetResource(source3.SubscribeSafe(_observer3));
+                _observer4.SetResource(source4.SubscribeSafe(_observer4));
+                _observer5.SetResource(source5.SubscribeSafe(_observer5));
+                _observer6.SetResource(source6.SubscribeSafe(_observer6));
+                _observer7.SetResource(source7.SubscribeSafe(_observer7));
 
                 SetUpstream(StableCompositeDisposable.Create(subscriptions));
             }
@@ -382,27 +372,25 @@ namespace System.Reactive.Linq.ObservableImpl
 
             public void Run(IObservable<T1> source1, IObservable<T2> source2, IObservable<T3> source3, IObservable<T4> source4, IObservable<T5> source5, IObservable<T6> source6, IObservable<T7> source7, IObservable<T8> source8)
             {
-                var subscriptions = new SingleAssignmentDisposable[8];
-                for (int i = 0; i < 8; i++)
-                    subscriptions[i] = new SingleAssignmentDisposable();
-
-                _observer1 = new CombineLatestObserver<T1>(_gate, this, 0, subscriptions[0]);
-                _observer2 = new CombineLatestObserver<T2>(_gate, this, 1, subscriptions[1]);
-                _observer3 = new CombineLatestObserver<T3>(_gate, this, 2, subscriptions[2]);
-                _observer4 = new CombineLatestObserver<T4>(_gate, this, 3, subscriptions[3]);
-                _observer5 = new CombineLatestObserver<T5>(_gate, this, 4, subscriptions[4]);
-                _observer6 = new CombineLatestObserver<T6>(_gate, this, 5, subscriptions[5]);
-                _observer7 = new CombineLatestObserver<T7>(_gate, this, 6, subscriptions[6]);
-                _observer8 = new CombineLatestObserver<T8>(_gate, this, 7, subscriptions[7]);
-
-                subscriptions[0].Disposable = source1.SubscribeSafe(_observer1);
-                subscriptions[1].Disposable = source2.SubscribeSafe(_observer2);
-                subscriptions[2].Disposable = source3.SubscribeSafe(_observer3);
-                subscriptions[3].Disposable = source4.SubscribeSafe(_observer4);
-                subscriptions[4].Disposable = source5.SubscribeSafe(_observer5);
-                subscriptions[5].Disposable = source6.SubscribeSafe(_observer6);
-                subscriptions[6].Disposable = source7.SubscribeSafe(_observer7);
-                subscriptions[7].Disposable = source8.SubscribeSafe(_observer8);
+                var subscriptions = new IDisposable[8];
+
+                subscriptions[0] = _observer1 = new CombineLatestObserver<T1>(_gate, this, 0);
+                subscriptions[1] = _observer2 = new CombineLatestObserver<T2>(_gate, this, 1);
+                subscriptions[2] = _observer3 = new CombineLatestObserver<T3>(_gate, this, 2);
+                subscriptions[3] = _observer4 = new CombineLatestObserver<T4>(_gate, this, 3);
+                subscriptions[4] = _observer5 = new CombineLatestObserver<T5>(_gate, this, 4);
+                subscriptions[5] = _observer6 = new CombineLatestObserver<T6>(_gate, this, 5);
+                subscriptions[6] = _observer7 = new CombineLatestObserver<T7>(_gate, this, 6);
+                subscriptions[7] = _observer8 = new CombineLatestObserver<T8>(_gate, this, 7);
+
+                _observer1.SetResource(source1.SubscribeSafe(_observer1));
+                _observer2.SetResource(source2.SubscribeSafe(_observer2));
+                _observer3.SetResource(source3.SubscribeSafe(_observer3));
+                _observer4.SetResource(source4.SubscribeSafe(_observer4));
+                _observer5.SetResource(source5.SubscribeSafe(_observer5));
+                _observer6.SetResource(source6.SubscribeSafe(_observer6));
+                _observer7.SetResource(source7.SubscribeSafe(_observer7));
+                _observer8.SetResource(source8.SubscribeSafe(_observer8));
 
                 SetUpstream(StableCompositeDisposable.Create(subscriptions));
             }
@@ -464,29 +452,27 @@ namespace System.Reactive.Linq.ObservableImpl
 
             public void Run(IObservable<T1> source1, IObservable<T2> source2, IObservable<T3> source3, IObservable<T4> source4, IObservable<T5> source5, IObservable<T6> source6, IObservable<T7> source7, IObservable<T8> source8, IObservable<T9> source9)
             {
-                var subscriptions = new SingleAssignmentDisposable[9];
-                for (int i = 0; i < 9; i++)
-                    subscriptions[i] = new SingleAssignmentDisposable();
-
-                _observer1 = new CombineLatestObserver<T1>(_gate, this, 0, subscriptions[0]);
-                _observer2 = new CombineLatestObserver<T2>(_gate, this, 1, subscriptions[1]);
-                _observer3 = new CombineLatestObserver<T3>(_gate, this, 2, subscriptions[2]);
-                _observer4 = new CombineLatestObserver<T4>(_gate, this, 3, subscriptions[3]);
-                _observer5 = new CombineLatestObserver<T5>(_gate, this, 4, subscriptions[4]);
-                _observer6 = new CombineLatestObserver<T6>(_gate, this, 5, subscriptions[5]);
-                _observer7 = new CombineLatestObserver<T7>(_gate, this, 6, subscriptions[6]);
-                _observer8 = new CombineLatestObserver<T8>(_gate, this, 7, subscriptions[7]);
-                _observer9 = new CombineLatestObserver<T9>(_gate, this, 8, subscriptions[8]);
-
-                subscriptions[0].Disposable = source1.SubscribeSafe(_observer1);
-                subscriptions[1].Disposable = source2.SubscribeSafe(_observer2);
-                subscriptions[2].Disposable = source3.SubscribeSafe(_observer3);
-                subscriptions[3].Disposable = source4.SubscribeSafe(_observer4);
-                subscriptions[4].Disposable = source5.SubscribeSafe(_observer5);
-                subscriptions[5].Disposable = source6.SubscribeSafe(_observer6);
-                subscriptions[6].Disposable = source7.SubscribeSafe(_observer7);
-                subscriptions[7].Disposable = source8.SubscribeSafe(_observer8);
-                subscriptions[8].Disposable = source9.SubscribeSafe(_observer9);
+                var subscriptions = new IDisposable[9];
+
+                subscriptions[0] = _observer1 = new CombineLatestObserver<T1>(_gate, this, 0);
+                subscriptions[1] = _observer2 = new CombineLatestObserver<T2>(_gate, this, 1);
+                subscriptions[2] = _observer3 = new CombineLatestObserver<T3>(_gate, this, 2);
+                subscriptions[3] = _observer4 = new CombineLatestObserver<T4>(_gate, this, 3);
+                subscriptions[4] = _observer5 = new CombineLatestObserver<T5>(_gate, this, 4);
+                subscriptions[5] = _observer6 = new CombineLatestObserver<T6>(_gate, this, 5);
+                subscriptions[6] = _observer7 = new CombineLatestObserver<T7>(_gate, this, 6);
+                subscriptions[7] = _observer8 = new CombineLatestObserver<T8>(_gate, this, 7);
+                subscriptions[8] = _observer9 = new CombineLatestObserver<T9>(_gate, this, 8);
+
+                _observer1.SetResource(source1.SubscribeSafe(_observer1));
+                _observer2.SetResource(source2.SubscribeSafe(_observer2));
+                _observer3.SetResource(source3.SubscribeSafe(_observer3));
+                _observer4.SetResource(source4.SubscribeSafe(_observer4));
+                _observer5.SetResource(source5.SubscribeSafe(_observer5));
+                _observer6.SetResource(source6.SubscribeSafe(_observer6));
+                _observer7.SetResource(source7.SubscribeSafe(_observer7));
+                _observer8.SetResource(source8.SubscribeSafe(_observer8));
+                _observer9.SetResource(source9.SubscribeSafe(_observer9));
 
                 SetUpstream(StableCompositeDisposable.Create(subscriptions));
             }
@@ -551,31 +537,29 @@ namespace System.Reactive.Linq.ObservableImpl
 
             public void Run(IObservable<T1> source1, IObservable<T2> source2, IObservable<T3> source3, IObservable<T4> source4, IObservable<T5> source5, IObservable<T6> source6, IObservable<T7> source7, IObservable<T8> source8, IObservable<T9> source9, IObservable<T10> source10)
             {
-                var subscriptions = new SingleAssignmentDisposable[10];
-                for (int i = 0; i < 10; i++)
-                    subscriptions[i] = new SingleAssignmentDisposable();
-
-                _observer1 = new CombineLatestObserver<T1>(_gate, this, 0, subscriptions[0]);
-                _observer2 = new CombineLatestObserver<T2>(_gate, this, 1, subscriptions[1]);
-                _observer3 = new CombineLatestObserver<T3>(_gate, this, 2, subscriptions[2]);
-                _observer4 = new CombineLatestObserver<T4>(_gate, this, 3, subscriptions[3]);
-                _observer5 = new CombineLatestObserver<T5>(_gate, this, 4, subscriptions[4]);
-                _observer6 = new CombineLatestObserver<T6>(_gate, this, 5, subscriptions[5]);
-                _observer7 = new CombineLatestObserver<T7>(_gate, this, 6, subscriptions[6]);
-                _observer8 = new CombineLatestObserver<T8>(_gate, this, 7, subscriptions[7]);
-                _observer9 = new CombineLatestObserver<T9>(_gate, this, 8, subscriptions[8]);
-                _observer10 = new CombineLatestObserver<T10>(_gate, this, 9, subscriptions[9]);
-
-                subscriptions[0].Disposable = source1.SubscribeSafe(_observer1);
-                subscriptions[1].Disposable = source2.SubscribeSafe(_observer2);
-                subscriptions[2].Disposable = source3.SubscribeSafe(_observer3);
-                subscriptions[3].Disposable = source4.SubscribeSafe(_observer4);
-                subscriptions[4].Disposable = source5.SubscribeSafe(_observer5);
-                subscriptions[5].Disposable = source6.SubscribeSafe(_observer6);
-                subscriptions[6].Disposable = source7.SubscribeSafe(_observer7);
-                subscriptions[7].Disposable = source8.SubscribeSafe(_observer8);
-                subscriptions[8].Disposable = source9.SubscribeSafe(_observer9);
-                subscriptions[9].Disposable = source10.SubscribeSafe(_observer10);
+                var subscriptions = new IDisposable[10];
+
+                subscriptions[0] = _observer1 = new CombineLatestObserver<T1>(_gate, this, 0);
+                subscriptions[1] = _observer2 = new CombineLatestObserver<T2>(_gate, this, 1);
+                subscriptions[2] = _observer3 = new CombineLatestObserver<T3>(_gate, this, 2);
+                subscriptions[3] = _observer4 = new CombineLatestObserver<T4>(_gate, this, 3);
+                subscriptions[4] = _observer5 = new CombineLatestObserver<T5>(_gate, this, 4);
+                subscriptions[5] = _observer6 = new CombineLatestObserver<T6>(_gate, this, 5);
+                subscriptions[6] = _observer7 = new CombineLatestObserver<T7>(_gate, this, 6);
+                subscriptions[7] = _observer8 = new CombineLatestObserver<T8>(_gate, this, 7);
+                subscriptions[8] = _observer9 = new CombineLatestObserver<T9>(_gate, this, 8);
+                subscriptions[9] = _observer10 = new CombineLatestObserver<T10>(_gate, this, 9);
+
+                _observer1.SetResource(source1.SubscribeSafe(_observer1));
+                _observer2.SetResource(source2.SubscribeSafe(_observer2));
+                _observer3.SetResource(source3.SubscribeSafe(_observer3));
+                _observer4.SetResource(source4.SubscribeSafe(_observer4));
+                _observer5.SetResource(source5.SubscribeSafe(_observer5));
+                _observer6.SetResource(source6.SubscribeSafe(_observer6));
+                _observer7.SetResource(source7.SubscribeSafe(_observer7));
+                _observer8.SetResource(source8.SubscribeSafe(_observer8));
+                _observer9.SetResource(source9.SubscribeSafe(_observer9));
+                _observer10.SetResource(source10.SubscribeSafe(_observer10));
 
                 SetUpstream(StableCompositeDisposable.Create(subscriptions));
             }
@@ -643,33 +627,31 @@ namespace System.Reactive.Linq.ObservableImpl
 
             public void Run(IObservable<T1> source1, IObservable<T2> source2, IObservable<T3> source3, IObservable<T4> source4, IObservable<T5> source5, IObservable<T6> source6, IObservable<T7> source7, IObservable<T8> source8, IObservable<T9> source9, IObservable<T10> source10, IObservable<T11> source11)
             {
-                var subscriptions = new SingleAssignmentDisposable[11];
-                for (int i = 0; i < 11; i++)
-                    subscriptions[i] = new SingleAssignmentDisposable();
-
-                _observer1 = new CombineLatestObserver<T1>(_gate, this, 0, subscriptions[0]);
-                _observer2 = new CombineLatestObserver<T2>(_gate, this, 1, subscriptions[1]);
-                _observer3 = new CombineLatestObserver<T3>(_gate, this, 2, subscriptions[2]);
-                _observer4 = new CombineLatestObserver<T4>(_gate, this, 3, subscriptions[3]);
-                _observer5 = new CombineLatestObserver<T5>(_gate, this, 4, subscriptions[4]);
-                _observer6 = new CombineLatestObserver<T6>(_gate, this, 5, subscriptions[5]);
-                _observer7 = new CombineLatestObserver<T7>(_gate, this, 6, subscriptions[6]);
-                _observer8 = new CombineLatestObserver<T8>(_gate, this, 7, subscriptions[7]);
-                _observer9 = new CombineLatestObserver<T9>(_gate, this, 8, subscriptions[8]);
-                _observer10 = new CombineLatestObserver<T10>(_gate, this, 9, subscriptions[9]);
-                _observer11 = new CombineLatestObserver<T11>(_gate, this, 10, subscriptions[10]);
-
-                subscriptions[0].Disposable = source1.SubscribeSafe(_observer1);
-                subscriptions[1].Disposable = source2.SubscribeSafe(_observer2);
-                subscriptions[2].Disposable = source3.SubscribeSafe(_observer3);
-                subscriptions[3].Disposable = source4.SubscribeSafe(_observer4);
-                subscriptions[4].Disposable = source5.SubscribeSafe(_observer5);
-                subscriptions[5].Disposable = source6.SubscribeSafe(_observer6);
-                subscriptions[6].Disposable = source7.SubscribeSafe(_observer7);
-                subscriptions[7].Disposable = source8.SubscribeSafe(_observer8);
-                subscriptions[8].Disposable = source9.SubscribeSafe(_observer9);
-                subscriptions[9].Disposable = source10.SubscribeSafe(_observer10);
-                subscriptions[10].Disposable = source11.SubscribeSafe(_observer11);
+                var subscriptions = new IDisposable[11];
+
+                subscriptions[0] = _observer1 = new CombineLatestObserver<T1>(_gate, this, 0);
+                subscriptions[1] = _observer2 = new CombineLatestObserver<T2>(_gate, this, 1);
+                subscriptions[2] = _observer3 = new CombineLatestObserver<T3>(_gate, this, 2);
+                subscriptions[3] = _observer4 = new CombineLatestObserver<T4>(_gate, this, 3);
+                subscriptions[4] = _observer5 = new CombineLatestObserver<T5>(_gate, this, 4);
+                subscriptions[5] = _observer6 = new CombineLatestObserver<T6>(_gate, this, 5);
+                subscriptions[6] = _observer7 = new CombineLatestObserver<T7>(_gate, this, 6);
+                subscriptions[7] = _observer8 = new CombineLatestObserver<T8>(_gate, this, 7);
+                subscriptions[8] = _observer9 = new CombineLatestObserver<T9>(_gate, this, 8);
+                subscriptions[9] = _observer10 = new CombineLatestObserver<T10>(_gate, this, 9);
+                subscriptions[10] = _observer11 = new CombineLatestObserver<T11>(_gate, this, 10);
+
+                _observer1.SetResource(source1.SubscribeSafe(_observer1));
+                _observer2.SetResource(source2.SubscribeSafe(_observer2));
+                _observer3.SetResource(source3.SubscribeSafe(_observer3));
+                _observer4.SetResource(source4.SubscribeSafe(_observer4));
+                _observer5.SetResource(source5.SubscribeSafe(_observer5));
+                _observer6.SetResource(source6.SubscribeSafe(_observer6));
+                _observer7.SetResource(source7.SubscribeSafe(_observer7));
+                _observer8.SetResource(source8.SubscribeSafe(_observer8));
+                _observer9.SetResource(source9.SubscribeSafe(_observer9));
+                _observer10.SetResource(source10.SubscribeSafe(_observer10));
+                _observer11.SetResource(source11.SubscribeSafe(_observer11));
 
                 SetUpstream(StableCompositeDisposable.Create(subscriptions));
             }
@@ -740,35 +722,33 @@ namespace System.Reactive.Linq.ObservableImpl
 
             public void Run(IObservable<T1> source1, IObservable<T2> source2, IObservable<T3> source3, IObservable<T4> source4, IObservable<T5> source5, IObservable<T6> source6, IObservable<T7> source7, IObservable<T8> source8, IObservable<T9> source9, IObservable<T10> source10, IObservable<T11> source11, IObservable<T12> source12)
             {
-                var subscriptions = new SingleAssignmentDisposable[12];
-                for (int i = 0; i < 12; i++)
-                    subscriptions[i] = new SingleAssignmentDisposable();
-
-                _observer1 = new CombineLatestObserver<T1>(_gate, this, 0, subscriptions[0]);
-                _observer2 = new CombineLatestObserver<T2>(_gate, this, 1, subscriptions[1]);
-                _observer3 = new CombineLatestObserver<T3>(_gate, this, 2, subscriptions[2]);
-                _observer4 = new CombineLatestObserver<T4>(_gate, this, 3, subscriptions[3]);
-                _observer5 = new CombineLatestObserver<T5>(_gate, this, 4, subscriptions[4]);
-                _observer6 = new CombineLatestObserver<T6>(_gate, this, 5, subscriptions[5]);
-                _observer7 = new CombineLatestObserver<T7>(_gate, this, 6, subscriptions[6]);
-                _observer8 = new CombineLatestObserver<T8>(_gate, this, 7, subscriptions[7]);
-                _observer9 = new CombineLatestObserver<T9>(_gate, this, 8, subscriptions[8]);
-                _observer10 = new CombineLatestObserver<T10>(_gate, this, 9, subscriptions[9]);
-                _observer11 = new CombineLatestObserver<T11>(_gate, this, 10, subscriptions[10]);
-                _observer12 = new CombineLatestObserver<T12>(_gate, this, 11, subscriptions[11]);
-
-                subscriptions[0].Disposable = source1.SubscribeSafe(_observer1);
-                subscriptions[1].Disposable = source2.SubscribeSafe(_observer2);
-                subscriptions[2].Disposable = source3.SubscribeSafe(_observer3);
-                subscriptions[3].Disposable = source4.SubscribeSafe(_observer4);
-                subscriptions[4].Disposable = source5.SubscribeSafe(_observer5);
-                subscriptions[5].Disposable = source6.SubscribeSafe(_observer6);
-                subscriptions[6].Disposable = source7.SubscribeSafe(_observer7);
-                subscriptions[7].Disposable = source8.SubscribeSafe(_observer8);
-                subscriptions[8].Disposable = source9.SubscribeSafe(_observer9);
-                subscriptions[9].Disposable = source10.SubscribeSafe(_observer10);
-                subscriptions[10].Disposable = source11.SubscribeSafe(_observer11);
-                subscriptions[11].Disposable = source12.SubscribeSafe(_observer12);
+                var subscriptions = new IDisposable[12];
+
+                subscriptions[0] = _observer1 = new CombineLatestObserver<T1>(_gate, this, 0);
+                subscriptions[1] = _observer2 = new CombineLatestObserver<T2>(_gate, this, 1);
+                subscriptions[2] = _observer3 = new CombineLatestObserver<T3>(_gate, this, 2);
+                subscriptions[3] = _observer4 = new CombineLatestObserver<T4>(_gate, this, 3);
+                subscriptions[4] = _observer5 = new CombineLatestObserver<T5>(_gate, this, 4);
+                subscriptions[5] = _observer6 = new CombineLatestObserver<T6>(_gate, this, 5);
+                subscriptions[6] = _observer7 = new CombineLatestObserver<T7>(_gate, this, 6);
+                subscriptions[7] = _observer8 = new CombineLatestObserver<T8>(_gate, this, 7);
+                subscriptions[8] = _observer9 = new CombineLatestObserver<T9>(_gate, this, 8);
+                subscriptions[9] = _observer10 = new CombineLatestObserver<T10>(_gate, this, 9);
+                subscriptions[10] = _observer11 = new CombineLatestObserver<T11>(_gate, this, 10);
+                subscriptions[11] = _observer12 = new CombineLatestObserver<T12>(_gate, this, 11);
+
+                _observer1.SetResource(source1.SubscribeSafe(_observer1));
+                _observer2.SetResource(source2.SubscribeSafe(_observer2));
+                _observer3.SetResource(source3.SubscribeSafe(_observer3));
+                _observer4.SetResource(source4.SubscribeSafe(_observer4));
+                _observer5.SetResource(source5.SubscribeSafe(_observer5));
+                _observer6.SetResource(source6.SubscribeSafe(_observer6));
+                _observer7.SetResource(source7.SubscribeSafe(_observer7));
+                _observer8.SetResource(source8.SubscribeSafe(_observer8));
+                _observer9.SetResource(source9.SubscribeSafe(_observer9));
+                _observer10.SetResource(source10.SubscribeSafe(_observer10));
+                _observer11.SetResource(source11.SubscribeSafe(_observer11));
+                _observer12.SetResource(source12.SubscribeSafe(_observer12));
 
                 SetUpstream(StableCompositeDisposable.Create(subscriptions));
             }
@@ -842,37 +822,35 @@ namespace System.Reactive.Linq.ObservableImpl
 
             public void Run(IObservable<T1> source1, IObservable<T2> source2, IObservable<T3> source3, IObservable<T4> source4, IObservable<T5> source5, IObservable<T6> source6, IObservable<T7> source7, IObservable<T8> source8, IObservable<T9> source9, IObservable<T10> source10, IObservable<T11> source11, IObservable<T12> source12, IObservable<T13> source13)
             {
-                var subscriptions = new SingleAssignmentDisposable[13];
-                for (int i = 0; i < 13; i++)
-                    subscriptions[i] = new SingleAssignmentDisposable();
-
-                _observer1 = new CombineLatestObserver<T1>(_gate, this, 0, subscriptions[0]);
-                _observer2 = new CombineLatestObserver<T2>(_gate, this, 1, subscriptions[1]);
-                _observer3 = new CombineLatestObserver<T3>(_gate, this, 2, subscriptions[2]);
-                _observer4 = new CombineLatestObserver<T4>(_gate, this, 3, subscriptions[3]);
-                _observer5 = new CombineLatestObserver<T5>(_gate, this, 4, subscriptions[4]);
-                _observer6 = new CombineLatestObserver<T6>(_gate, this, 5, subscriptions[5]);
-                _observer7 = new CombineLatestObserver<T7>(_gate, this, 6, subscriptions[6]);
-                _observer8 = new CombineLatestObserver<T8>(_gate, this, 7, subscriptions[7]);
-                _observer9 = new CombineLatestObserver<T9>(_gate, this, 8, subscriptions[8]);
-                _observer10 = new CombineLatestObserver<T10>(_gate, this, 9, subscriptions[9]);
-                _observer11 = new CombineLatestObserver<T11>(_gate, this, 10, subscriptions[10]);
-                _observer12 = new CombineLatestObserver<T12>(_gate, this, 11, subscriptions[11]);
-                _observer13 = new CombineLatestObserver<T13>(_gate, this, 12, subscriptions[12]);
-
-                subscriptions[0].Disposable = source1.SubscribeSafe(_observer1);
-                subscriptions[1].Disposable = source2.SubscribeSafe(_observer2);
-                subscriptions[2].Disposable = source3.SubscribeSafe(_observer3);
-                subscriptions[3].Disposable = source4.SubscribeSafe(_observer4);
-                subscriptions[4].Disposable = source5.SubscribeSafe(_observer5);
-                subscriptions[5].Disposable = source6.SubscribeSafe(_observer6);
-                subscriptions[6].Disposable = source7.SubscribeSafe(_observer7);
-                subscriptions[7].Disposable = source8.SubscribeSafe(_observer8);
-                subscriptions[8].Disposable = source9.SubscribeSafe(_observer9);
-                subscriptions[9].Disposable = source10.SubscribeSafe(_observer10);
-                subscriptions[10].Disposable = source11.SubscribeSafe(_observer11);
-                subscriptions[11].Disposable = source12.SubscribeSafe(_observer12);
-                subscriptions[12].Disposable = source13.SubscribeSafe(_observer13);
+                var subscriptions = new IDisposable[13];
+
+                subscriptions[0] = _observer1 = new CombineLatestObserver<T1>(_gate, this, 0);
+                subscriptions[1] = _observer2 = new CombineLatestObserver<T2>(_gate, this, 1);
+                subscriptions[2] = _observer3 = new CombineLatestObserver<T3>(_gate, this, 2);
+                subscriptions[3] = _observer4 = new CombineLatestObserver<T4>(_gate, this, 3);
+                subscriptions[4] = _observer5 = new CombineLatestObserver<T5>(_gate, this, 4);
+                subscriptions[5] = _observer6 = new CombineLatestObserver<T6>(_gate, this, 5);
+                subscriptions[6] = _observer7 = new CombineLatestObserver<T7>(_gate, this, 6);
+                subscriptions[7] = _observer8 = new CombineLatestObserver<T8>(_gate, this, 7);
+                subscriptions[8] = _observer9 = new CombineLatestObserver<T9>(_gate, this, 8);
+                subscriptions[9] = _observer10 = new CombineLatestObserver<T10>(_gate, this, 9);
+                subscriptions[10] = _observer11 = new CombineLatestObserver<T11>(_gate, this, 10);
+                subscriptions[11] = _observer12 = new CombineLatestObserver<T12>(_gate, this, 11);
+                subscriptions[12] = _observer13 = new CombineLatestObserver<T13>(_gate, this, 12);
+
+                _observer1.SetResource(source1.SubscribeSafe(_observer1));
+                _observer2.SetResource(source2.SubscribeSafe(_observer2));
+                _observer3.SetResource(source3.SubscribeSafe(_observer3));
+                _observer4.SetResource(source4.SubscribeSafe(_observer4));
+                _observer5.SetResource(source5.SubscribeSafe(_observer5));
+                _observer6.SetResource(source6.SubscribeSafe(_observer6));
+                _observer7.SetResource(source7.SubscribeSafe(_observer7));
+                _observer8.SetResource(source8.SubscribeSafe(_observer8));
+                _observer9.SetResource(source9.SubscribeSafe(_observer9));
+                _observer10.SetResource(source10.SubscribeSafe(_observer10));
+                _observer11.SetResource(source11.SubscribeSafe(_observer11));
+                _observer12.SetResource(source12.SubscribeSafe(_observer12));
+                _observer13.SetResource(source13.SubscribeSafe(_observer13));
 
                 SetUpstream(StableCompositeDisposable.Create(subscriptions));
             }
@@ -949,39 +927,37 @@ namespace System.Reactive.Linq.ObservableImpl
 
             public void Run(IObservable<T1> source1, IObservable<T2> source2, IObservable<T3> source3, IObservable<T4> source4, IObservable<T5> source5, IObservable<T6> source6, IObservable<T7> source7, IObservable<T8> source8, IObservable<T9> source9, IObservable<T10> source10, IObservable<T11> source11, IObservable<T12> source12, IObservable<T13> source13, IObservable<T14> source14)
             {
-                var subscriptions = new SingleAssignmentDisposable[14];
-                for (int i = 0; i < 14; i++)
-                    subscriptions[i] = new SingleAssignmentDisposable();
-
-                _observer1 = new CombineLatestObserver<T1>(_gate, this, 0, subscriptions[0]);
-                _observer2 = new CombineLatestObserver<T2>(_gate, this, 1, subscriptions[1]);
-                _observer3 = new CombineLatestObserver<T3>(_gate, this, 2, subscriptions[2]);
-                _observer4 = new CombineLatestObserver<T4>(_gate, this, 3, subscriptions[3]);
-                _observer5 = new CombineLatestObserver<T5>(_gate, this, 4, subscriptions[4]);
-                _observer6 = new CombineLatestObserver<T6>(_gate, this, 5, subscriptions[5]);
-                _observer7 = new CombineLatestObserver<T7>(_gate, this, 6, subscriptions[6]);
-                _observer8 = new CombineLatestObserver<T8>(_gate, this, 7, subscriptions[7]);
-                _observer9 = new CombineLatestObserver<T9>(_gate, this, 8, subscriptions[8]);
-                _observer10 = new CombineLatestObserver<T10>(_gate, this, 9, subscriptions[9]);
-                _observer11 = new CombineLatestObserver<T11>(_gate, this, 10, subscriptions[10]);
-                _observer12 = new CombineLatestObserver<T12>(_gate, this, 11, subscriptions[11]);
-                _observer13 = new CombineLatestObserver<T13>(_gate, this, 12, subscriptions[12]);
-                _observer14 = new CombineLatestObserver<T14>(_gate, this, 13, subscriptions[13]);
-
-                subscriptions[0].Disposable = source1.SubscribeSafe(_observer1);
-                subscriptions[1].Disposable = source2.SubscribeSafe(_observer2);
-                subscriptions[2].Disposable = source3.SubscribeSafe(_observer3);
-                subscriptions[3].Disposable = source4.SubscribeSafe(_observer4);
-                subscriptions[4].Disposable = source5.SubscribeSafe(_observer5);
-                subscriptions[5].Disposable = source6.SubscribeSafe(_observer6);
-                subscriptions[6].Disposable = source7.SubscribeSafe(_observer7);
-                subscriptions[7].Disposable = source8.SubscribeSafe(_observer8);
-                subscriptions[8].Disposable = source9.SubscribeSafe(_observer9);
-                subscriptions[9].Disposable = source10.SubscribeSafe(_observer10);
-                subscriptions[10].Disposable = source11.SubscribeSafe(_observer11);
-                subscriptions[11].Disposable = source12.SubscribeSafe(_observer12);
-                subscriptions[12].Disposable = source13.SubscribeSafe(_observer13);
-                subscriptions[13].Disposable = source14.SubscribeSafe(_observer14);
+                var subscriptions = new IDisposable[14];
+
+                subscriptions[0] = _observer1 = new CombineLatestObserver<T1>(_gate, this, 0);
+                subscriptions[1] = _observer2 = new CombineLatestObserver<T2>(_gate, this, 1);
+                subscriptions[2] = _observer3 = new CombineLatestObserver<T3>(_gate, this, 2);
+                subscriptions[3] = _observer4 = new CombineLatestObserver<T4>(_gate, this, 3);
+                subscriptions[4] = _observer5 = new CombineLatestObserver<T5>(_gate, this, 4);
+                subscriptions[5] = _observer6 = new CombineLatestObserver<T6>(_gate, this, 5);
+                subscriptions[6] = _observer7 = new CombineLatestObserver<T7>(_gate, this, 6);
+                subscriptions[7] = _observer8 = new CombineLatestObserver<T8>(_gate, this, 7);
+                subscriptions[8] = _observer9 = new CombineLatestObserver<T9>(_gate, this, 8);
+                subscriptions[9] = _observer10 = new CombineLatestObserver<T10>(_gate, this, 9);
+                subscriptions[10] = _observer11 = new CombineLatestObserver<T11>(_gate, this, 10);
+                subscriptions[11] = _observer12 = new CombineLatestObserver<T12>(_gate, this, 11);
+                subscriptions[12] = _observer13 = new CombineLatestObserver<T13>(_gate, this, 12);
+                subscriptions[13] = _observer14 = new CombineLatestObserver<T14>(_gate, this, 13);
+
+                _observer1.SetResource(source1.SubscribeSafe(_observer1));
+                _observer2.SetResource(source2.SubscribeSafe(_observer2));
+                _observer3.SetResource(source3.SubscribeSafe(_observer3));
+                _observer4.SetResource(source4.SubscribeSafe(_observer4));
+                _observer5.SetResource(source5.SubscribeSafe(_observer5));
+                _observer6.SetResource(source6.SubscribeSafe(_observer6));
+                _observer7.SetResource(source7.SubscribeSafe(_observer7));
+                _observer8.SetResource(source8.SubscribeSafe(_observer8));
+                _observer9.SetResource(source9.SubscribeSafe(_observer9));
+                _observer10.SetResource(source10.SubscribeSafe(_observer10));
+                _observer11.SetResource(source11.SubscribeSafe(_observer11));
+                _observer12.SetResource(source12.SubscribeSafe(_observer12));
+                _observer13.SetResource(source13.SubscribeSafe(_observer13));
+                _observer14.SetResource(source14.SubscribeSafe(_observer14));
 
                 SetUpstream(StableCompositeDisposable.Create(subscriptions));
             }
@@ -1061,41 +1037,39 @@ namespace System.Reactive.Linq.ObservableImpl
 
             public void Run(IObservable<T1> source1, IObservable<T2> source2, IObservable<T3> source3, IObservable<T4> source4, IObservable<T5> source5, IObservable<T6> source6, IObservable<T7> source7, IObservable<T8> source8, IObservable<T9> source9, IObservable<T10> source10, IObservable<T11> source11, IObservable<T12> source12, IObservable<T13> source13, IObservable<T14> source14, IObservable<T15> source15)
             {
-                var subscriptions = new SingleAssignmentDisposable[15];
-                for (int i = 0; i < 15; i++)
-                    subscriptions[i] = new SingleAssignmentDisposable();
-
-                _observer1 = new CombineLatestObserver<T1>(_gate, this, 0, subscriptions[0]);
-                _observer2 = new CombineLatestObserver<T2>(_gate, this, 1, subscriptions[1]);
-                _observer3 = new CombineLatestObserver<T3>(_gate, this, 2, subscriptions[2]);
-                _observer4 = new CombineLatestObserver<T4>(_gate, this, 3, subscriptions[3]);
-                _observer5 = new CombineLatestObserver<T5>(_gate, this, 4, subscriptions[4]);
-                _observer6 = new CombineLatestObserver<T6>(_gate, this, 5, subscriptions[5]);
-                _observer7 = new CombineLatestObserver<T7>(_gate, this, 6, subscriptions[6]);
-                _observer8 = new CombineLatestObserver<T8>(_gate, this, 7, subscriptions[7]);
-                _observer9 = new CombineLatestObserver<T9>(_gate, this, 8, subscriptions[8]);
-                _observer10 = new CombineLatestObserver<T10>(_gate, this, 9, subscriptions[9]);
-                _observer11 = new CombineLatestObserver<T11>(_gate, this, 10, subscriptions[10]);
-                _observer12 = new CombineLatestObserver<T12>(_gate, this, 11, subscriptions[11]);
-                _observer13 = new CombineLatestObserver<T13>(_gate, this, 12, subscriptions[12]);
-                _observer14 = new CombineLatestObserver<T14>(_gate, this, 13, subscriptions[13]);
-                _observer15 = new CombineLatestObserver<T15>(_gate, this, 14, subscriptions[14]);
-
-                subscriptions[0].Disposable = source1.SubscribeSafe(_observer1);
-                subscriptions[1].Disposable = source2.SubscribeSafe(_observer2);
-                subscriptions[2].Disposable = source3.SubscribeSafe(_observer3);
-                subscriptions[3].Disposable = source4.SubscribeSafe(_observer4);
-                subscriptions[4].Disposable = source5.SubscribeSafe(_observer5);
-                subscriptions[5].Disposable = source6.SubscribeSafe(_observer6);
-                subscriptions[6].Disposable = source7.SubscribeSafe(_observer7);
-                subscriptions[7].Disposable = source8.SubscribeSafe(_observer8);
-                subscriptions[8].Disposable = source9.SubscribeSafe(_observer9);
-                subscriptions[9].Disposable = source10.SubscribeSafe(_observer10);
-                subscriptions[10].Disposable = source11.SubscribeSafe(_observer11);
-                subscriptions[11].Disposable = source12.SubscribeSafe(_observer12);
-                subscriptions[12].Disposable = source13.SubscribeSafe(_observer13);
-                subscriptions[13].Disposable = source14.SubscribeSafe(_observer14);
-                subscriptions[14].Disposable = source15.SubscribeSafe(_observer15);
+                var subscriptions = new IDisposable[15];
+
+                subscriptions[0] = _observer1 = new CombineLatestObserver<T1>(_gate, this, 0);
+                subscriptions[1] = _observer2 = new CombineLatestObserver<T2>(_gate, this, 1);
+                subscriptions[2] = _observer3 = new CombineLatestObserver<T3>(_gate, this, 2);
+                subscriptions[3] = _observer4 = new CombineLatestObserver<T4>(_gate, this, 3);
+                subscriptions[4] = _observer5 = new CombineLatestObserver<T5>(_gate, this, 4);
+                subscriptions[5] = _observer6 = new CombineLatestObserver<T6>(_gate, this, 5);
+                subscriptions[6] = _observer7 = new CombineLatestObserver<T7>(_gate, this, 6);
+                subscriptions[7] = _observer8 = new CombineLatestObserver<T8>(_gate, this, 7);
+                subscriptions[8] = _observer9 = new CombineLatestObserver<T9>(_gate, this, 8);
+                subscriptions[9] = _observer10 = new CombineLatestObserver<T10>(_gate, this, 9);
+                subscriptions[10] = _observer11 = new CombineLatestObserver<T11>(_gate, this, 10);
+                subscriptions[11] = _observer12 = new CombineLatestObserver<T12>(_gate, this, 11);
+                subscriptions[12] = _observer13 = new CombineLatestObserver<T13>(_gate, this, 12);
+                subscriptions[13] = _observer14 = new CombineLatestObserver<T14>(_gate, this, 13);
+                subscriptions[14] = _observer15 = new CombineLatestObserver<T15>(_gate, this, 14);
+
+                _observer1.SetResource(source1.SubscribeSafe(_observer1));
+                _observer2.SetResource(source2.SubscribeSafe(_observer2));
+                _observer3.SetResource(source3.SubscribeSafe(_observer3));
+                _observer4.SetResource(source4.SubscribeSafe(_observer4));
+                _observer5.SetResource(source5.SubscribeSafe(_observer5));
+                _observer6.SetResource(source6.SubscribeSafe(_observer6));
+                _observer7.SetResource(source7.SubscribeSafe(_observer7));
+                _observer8.SetResource(source8.SubscribeSafe(_observer8));
+                _observer9.SetResource(source9.SubscribeSafe(_observer9));
+                _observer10.SetResource(source10.SubscribeSafe(_observer10));
+                _observer11.SetResource(source11.SubscribeSafe(_observer11));
+                _observer12.SetResource(source12.SubscribeSafe(_observer12));
+                _observer13.SetResource(source13.SubscribeSafe(_observer13));
+                _observer14.SetResource(source14.SubscribeSafe(_observer14));
+                _observer15.SetResource(source15.SubscribeSafe(_observer15));
 
                 SetUpstream(StableCompositeDisposable.Create(subscriptions));
             }
@@ -1178,43 +1152,41 @@ namespace System.Reactive.Linq.ObservableImpl
 
             public void Run(IObservable<T1> source1, IObservable<T2> source2, IObservable<T3> source3, IObservable<T4> source4, IObservable<T5> source5, IObservable<T6> source6, IObservable<T7> source7, IObservable<T8> source8, IObservable<T9> source9, IObservable<T10> source10, IObservable<T11> source11, IObservable<T12> source12, IObservable<T13> source13, IObservable<T14> source14, IObservable<T15> source15, IObservable<T16> source16)
             {
-                var subscriptions = new SingleAssignmentDisposable[16];
-                for (int i = 0; i < 16; i++)
-                    subscriptions[i] = new SingleAssignmentDisposable();
-
-                _observer1 = new CombineLatestObserver<T1>(_gate, this, 0, subscriptions[0]);
-                _observer2 = new CombineLatestObserver<T2>(_gate, this, 1, subscriptions[1]);
-                _observer3 = new CombineLatestObserver<T3>(_gate, this, 2, subscriptions[2]);
-                _observer4 = new CombineLatestObserver<T4>(_gate, this, 3, subscriptions[3]);
-                _observer5 = new CombineLatestObserver<T5>(_gate, this, 4, subscriptions[4]);
-                _observer6 = new CombineLatestObserver<T6>(_gate, this, 5, subscriptions[5]);
-                _observer7 = new CombineLatestObserver<T7>(_gate, this, 6, subscriptions[6]);
-                _observer8 = new CombineLatestObserver<T8>(_gate, this, 7, subscriptions[7]);
-                _observer9 = new CombineLatestObserver<T9>(_gate, this, 8, subscriptions[8]);
-                _observer10 = new CombineLatestObserver<T10>(_gate, this, 9, subscriptions[9]);
-                _observer11 = new CombineLatestObserver<T11>(_gate, this, 10, subscriptions[10]);
-                _observer12 = new CombineLatestObserver<T12>(_gate, this, 11, subscriptions[11]);
-                _observer13 = new CombineLatestObserver<T13>(_gate, this, 12, subscriptions[12]);
-                _observer14 = new CombineLatestObserver<T14>(_gate, this, 13, subscriptions[13]);
-                _observer15 = new CombineLatestObserver<T15>(_gate, this, 14, subscriptions[14]);
-                _observer16 = new CombineLatestObserver<T16>(_gate, this, 15, subscriptions[15]);
-
-                subscriptions[0].Disposable = source1.SubscribeSafe(_observer1);
-                subscriptions[1].Disposable = source2.SubscribeSafe(_observer2);
-                subscriptions[2].Disposable = source3.SubscribeSafe(_observer3);
-                subscriptions[3].Disposable = source4.SubscribeSafe(_observer4);
-                subscriptions[4].Disposable = source5.SubscribeSafe(_observer5);
-                subscriptions[5].Disposable = source6.SubscribeSafe(_observer6);
-                subscriptions[6].Disposable = source7.SubscribeSafe(_observer7);
-                subscriptions[7].Disposable = source8.SubscribeSafe(_observer8);
-                subscriptions[8].Disposable = source9.SubscribeSafe(_observer9);
-                subscriptions[9].Disposable = source10.SubscribeSafe(_observer10);
-                subscriptions[10].Disposable = source11.SubscribeSafe(_observer11);
-                subscriptions[11].Disposable = source12.SubscribeSafe(_observer12);
-                subscriptions[12].Disposable = source13.SubscribeSafe(_observer13);
-                subscriptions[13].Disposable = source14.SubscribeSafe(_observer14);
-                subscriptions[14].Disposable = source15.SubscribeSafe(_observer15);
-                subscriptions[15].Disposable = source16.SubscribeSafe(_observer16);
+                var subscriptions = new IDisposable[16];
+
+                subscriptions[0] = _observer1 = new CombineLatestObserver<T1>(_gate, this, 0);
+                subscriptions[1] = _observer2 = new CombineLatestObserver<T2>(_gate, this, 1);
+                subscriptions[2] = _observer3 = new CombineLatestObserver<T3>(_gate, this, 2);
+                subscriptions[3] = _observer4 = new CombineLatestObserver<T4>(_gate, this, 3);
+                subscriptions[4] = _observer5 = new CombineLatestObserver<T5>(_gate, this, 4);
+                subscriptions[5] = _observer6 = new CombineLatestObserver<T6>(_gate, this, 5);
+                subscriptions[6] = _observer7 = new CombineLatestObserver<T7>(_gate, this, 6);
+                subscriptions[7] = _observer8 = new CombineLatestObserver<T8>(_gate, this, 7);
+                subscriptions[8] = _observer9 = new CombineLatestObserver<T9>(_gate, this, 8);
+                subscriptions[9] = _observer10 = new CombineLatestObserver<T10>(_gate, this, 9);
+                subscriptions[10] = _observer11 = new CombineLatestObserver<T11>(_gate, this, 10);
+                subscriptions[11] = _observer12 = new CombineLatestObserver<T12>(_gate, this, 11);
+                subscriptions[12] = _observer13 = new CombineLatestObserver<T13>(_gate, this, 12);
+                subscriptions[13] = _observer14 = new CombineLatestObserver<T14>(_gate, this, 13);
+                subscriptions[14] = _observer15 = new CombineLatestObserver<T15>(_gate, this, 14);
+                subscriptions[15] = _observer16 = new CombineLatestObserver<T16>(_gate, this, 15);
+
+                _observer1.SetResource(source1.SubscribeSafe(_observer1));
+                _observer2.SetResource(source2.SubscribeSafe(_observer2));
+                _observer3.SetResource(source3.SubscribeSafe(_observer3));
+                _observer4.SetResource(source4.SubscribeSafe(_observer4));
+                _observer5.SetResource(source5.SubscribeSafe(_observer5));
+                _observer6.SetResource(source6.SubscribeSafe(_observer6));
+                _observer7.SetResource(source7.SubscribeSafe(_observer7));
+                _observer8.SetResource(source8.SubscribeSafe(_observer8));
+                _observer9.SetResource(source9.SubscribeSafe(_observer9));
+                _observer10.SetResource(source10.SubscribeSafe(_observer10));
+                _observer11.SetResource(source11.SubscribeSafe(_observer11));
+                _observer12.SetResource(source12.SubscribeSafe(_observer12));
+                _observer13.SetResource(source13.SubscribeSafe(_observer13));
+                _observer14.SetResource(source14.SubscribeSafe(_observer14));
+                _observer15.SetResource(source15.SubscribeSafe(_observer15));
+                _observer16.SetResource(source16.SubscribeSafe(_observer16));
 
                 SetUpstream(StableCompositeDisposable.Create(subscriptions));
             }

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

@@ -76,15 +76,13 @@ for (var j = 1; j <= i; j++)
 
             public void Run(<#=os#>)
             {
-                var subscriptions = new SingleAssignmentDisposable[<#=i#>];
-                for (int i = 0; i < <#=i#>; i++)
-                    subscriptions[i] = new SingleAssignmentDisposable();
+                var subscriptions = new IDisposable[<#=i#>];
 
 <#
 for (var j = 1; j <= i; j++)
 {
 #>
-                _observer<#=j#> = new CombineLatestObserver<T<#=j#>>(_gate, this, <#=j - 1#>, subscriptions[<#=j - 1#>]);
+                subscriptions[<#=j - 1#>] = _observer<#=j#> = new CombineLatestObserver<T<#=j#>>(_gate, this, <#=j - 1#>);
 <#
 }
 #>
@@ -93,7 +91,7 @@ for (var j = 1; j <= i; j++)
 for (var j = 1; j <= i; j++)
 {
 #>
-                subscriptions[<#=j - 1#>].Disposable = source<#=j#>.SubscribeSafe(_observer<#=j#>);
+                _observer<#=j#>.SetResource(source<#=j#>.SubscribeSafe(_observer<#=j#>));
 <#
 }
 #>

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

@@ -328,25 +328,23 @@ namespace System.Reactive.Linq.ObservableImpl
         }
     }
 
-    internal sealed class CombineLatestObserver<T> : IObserver<T>
+    internal sealed class CombineLatestObserver<T> : SafeObserver<T>
     {
         private readonly object _gate;
         private readonly ICombineLatest _parent;
         private readonly int _index;
-        private readonly IDisposable _self;
         private T _value;
 
-        public CombineLatestObserver(object gate, ICombineLatest parent, int index, IDisposable self)
+        public CombineLatestObserver(object gate, ICombineLatest parent, int index)
         {
             _gate = gate;
             _parent = parent;
             _index = index;
-            _self = self;
         }
 
         public T Value => _value;
 
-        public void OnNext(T value)
+        public override void OnNext(T value)
         {
             lock (_gate)
             {
@@ -355,9 +353,9 @@ namespace System.Reactive.Linq.ObservableImpl
             }
         }
 
-        public void OnError(Exception error)
+        public override void OnError(Exception error)
         {
-            _self.Dispose();
+            Dispose();
 
             lock (_gate)
             {
@@ -365,10 +363,9 @@ namespace System.Reactive.Linq.ObservableImpl
             }
         }
 
-        public void OnCompleted()
+        public override void OnCompleted()
         {
-            _self.Dispose();
-
+            Dispose();
             lock (_gate)
             {
                 _parent.Done(_index);