Use Disposable helper methods were applicable (#568)

Rx.NET/Source/src/System.Reactive/Disposables/StableCompositeDisposable.cs

@@ -69,21 +69,21 @@ namespace System.Reactive.Disposables
 
         private sealed class Binary : StableCompositeDisposable
         {
-            private volatile IDisposable _disposable1;
-            private volatile IDisposable _disposable2;
+            private IDisposable _disposable1;
+            private IDisposable _disposable2;
 
             public Binary(IDisposable disposable1, IDisposable disposable2)
             {
-                _disposable1 = disposable1;
-                _disposable2 = disposable2;
+                Volatile.Write(ref _disposable1, disposable1);
+                Volatile.Write(ref _disposable2, disposable2);
             }
 
-            public override bool IsDisposed => _disposable1 == null;
+            public override bool IsDisposed => Disposable.GetIsDisposed(ref _disposable1);
 
             public override void Dispose()
             {
-                Interlocked.Exchange(ref _disposable1, null)?.Dispose();
-                Interlocked.Exchange(ref _disposable2, null)?.Dispose();
+                Disposable.TryDispose(ref _disposable1);
+                Disposable.TryDispose(ref _disposable2);
             }
         }
 

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

@@ -328,7 +328,7 @@ namespace System.Reactive
 
             if (disposing)
             {
-                Interlocked.Exchange(ref _cancel, null)?.Dispose();
+                Disposable.TryDispose(ref _cancel);
             }
         }
     }

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

@@ -2,6 +2,7 @@
 // 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.Reactive.Disposables;
 using System.Threading;
 
 namespace System.Reactive
@@ -25,8 +26,7 @@ namespace System.Reactive
         protected virtual void Dispose(bool disposing)
         {
             _observer = NopObserver<TTarget>.Instance;
-
-            Interlocked.Exchange(ref _cancel, null)?.Dispose();
+            Disposable.TryDispose(ref _cancel);
         }
 
         protected void ForwardOnNext(TTarget value)

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

@@ -95,10 +95,7 @@ namespace System.Reactive.Linq.ObservableImpl
 
                 internal void OnSubscribe(IDisposable d)
                 {
-                    if (Interlocked.CompareExchange(ref upstream, d, null) != null)
-                    {
-                        d?.Dispose();
-                    }
+                    Disposable.SetSingle(ref upstream, d);
                 }
 
                 public void Dispose()

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

@@ -205,10 +205,7 @@ namespace System.Reactive.Linq.ObservableImpl
 
             internal void OnSubscribe(IDisposable d)
             {
-                if (Interlocked.CompareExchange(ref upstream, d, null) != null)
-                {
-                    d?.Dispose();
-                }
+                Disposable.SetSingle(ref upstream, d);
             }
         }
 

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

@@ -52,10 +52,7 @@ namespace System.Reactive.Linq.ObservableImpl
 
             internal void OnSubscribe(IDisposable d)
             {
-                if (Interlocked.CompareExchange(ref upstream, d, null) != null)
-                {
-                    d?.Dispose();
-                }
+                Disposable.SetSingle(ref upstream, d);
             }
 
             public void Dispose()
@@ -178,15 +175,7 @@ namespace System.Reactive.Linq.ObservableImpl
             {
                 readonly ConcatManyOuterObserver parent;
 
-                internal SingleAssignmentDisposable upstream;
-
-                static readonly SingleAssignmentDisposable DISPOSED;
-
-                static InnerObserver()
-                {
-                    DISPOSED = new SingleAssignmentDisposable();
-                    DISPOSED.Dispose();
-                }
+                internal IDisposable upstream;
 
                 internal InnerObserver(ConcatManyOuterObserver parent)
                 {
@@ -195,13 +184,13 @@ namespace System.Reactive.Linq.ObservableImpl
 
                 internal bool SetDisposable(SingleAssignmentDisposable sad)
                 {
-                    return Interlocked.CompareExchange(ref upstream, sad, null) == null;
+                    return Disposable.TrySetSingle(ref upstream, sad) == TrySetSingleResult.Success;
                 }
 
                 internal bool Finish()
                 {
                     var sad = Volatile.Read(ref upstream);
-                    if (sad != DISPOSED)
+                    if (sad != BooleanDisposable.True)
                     {
                         if (Interlocked.CompareExchange(ref upstream, null, sad) == sad)
                         {
@@ -214,7 +203,7 @@ namespace System.Reactive.Linq.ObservableImpl
 
                 public void Dispose()
                 {
-                    Interlocked.Exchange(ref upstream, DISPOSED)?.Dispose();
+                    Disposable.TryDispose(ref upstream);
                 }
 
                 public void OnCompleted()

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

@@ -61,10 +61,10 @@ namespace System.Reactive.Linq.ObservableImpl
             readonly IObserver<Exception> errorSignal;
 
             internal readonly HandlerObserver handlerObserver;
-
+            
             readonly IObservable<T> source;
 
-            SingleAssignmentDisposable upstream;
+            IDisposable upstream;
 
             int trampoline;
 
@@ -72,14 +72,6 @@ namespace System.Reactive.Linq.ObservableImpl
 
             Exception error;
 
-            static readonly SingleAssignmentDisposable DISPOSED;
-
-            static MainObserver()
-            {
-                DISPOSED = new SingleAssignmentDisposable();
-                DISPOSED.Dispose();
-            }
-
             internal MainObserver(IObserver<T> downstream, IObservable<T> source, IObserver<Exception> errorSignal)
             {
                 this.downstream = downstream;
@@ -90,7 +82,7 @@ namespace System.Reactive.Linq.ObservableImpl
 
             public void Dispose()
             {
-                Interlocked.Exchange(ref upstream, DISPOSED)?.Dispose();
+                Disposable.TryDispose(ref upstream);
                 handlerObserver.Dispose();
             }
 
@@ -108,7 +100,7 @@ namespace System.Reactive.Linq.ObservableImpl
                 for (; ; )
                 {
                     var d = Volatile.Read(ref upstream);
-                    if (d == DISPOSED)
+                    if (d == BooleanDisposable.True)
                     {
                         break;
                     }
@@ -168,7 +160,7 @@ namespace System.Reactive.Linq.ObservableImpl
                     do
                     {
                         var sad = new SingleAssignmentDisposable();
-                        if (Interlocked.CompareExchange(ref upstream, sad, null) != null)
+                        if (Disposable.TrySetSingle(ref upstream, sad) != TrySetSingleResult.Success)
                         {
                             return;
                         }
@@ -192,10 +184,7 @@ namespace System.Reactive.Linq.ObservableImpl
 
                 internal void OnSubscribe(IDisposable d)
                 {
-                    if (Interlocked.CompareExchange(ref upstream, d, null) != null)
-                    {
-                        d?.Dispose();
-                    }
+                    Disposable.SetSingle(ref upstream, d);
                 }
 
                 public void Dispose()

Rx.NET/Source/src/System.Reactive/Linq/QueryLanguage.Creation.cs

@@ -178,10 +178,7 @@ namespace System.Reactive.Linq
 
                 public void OnNext(IDisposable value)
                 {
-                    if (Interlocked.CompareExchange(ref disposable, value, null) != null)
-                    {
-                        value?.Dispose();
-                    }
+                    Disposable.SetSingle(ref disposable, value);
                 }
             }
         }