Capture the concept of a "safe observer" (#597)

danielcweber · web-flow · commit da201773e960 · 2018-06-18T10:23:18.000+02:00
Capture the concept of a "safe observer" since it has been used and copied multiple times in the repo. ISafeObserver implements IDisposable and IObserver and can hold onto a resource. Implementors must at least dispose the resource on disposal of the SafeObserver, inheriting classes may choose to dispose the resource on other occasions as well. So far, AnonymousSafeObserver, SafeObserver and ObserverWithToken work like that are have been changed to implement ISafeObserver.
diff --git a/Rx.NET/Source/src/System.Reactive/AnonymousObserver.cs b/Rx.NET/Source/src/System.Reactive/AnonymousObserver.cs
@@ -84,6 +84,6 @@ public AnonymousObserver(Action<T> onNext, Action onCompleted)
         /// </summary>
         protected override void OnCompletedCore() => _onCompleted();
 
-        internal IObserver<T> MakeSafe(IDisposable disposable) => new AnonymousSafeObserver<T>(_onNext, _onError, _onCompleted, disposable);
+        internal ISafeObserver<T> MakeSafe() => new AnonymousSafeObserver<T>(_onNext, _onError, _onCompleted);
     }
 }
diff --git a/Rx.NET/Source/src/System.Reactive/AnonymousSafeObserver.cs b/Rx.NET/Source/src/System.Reactive/AnonymousSafeObserver.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
@@ -18,21 +19,20 @@ namespace System.Reactive
     /// that accept delegates for the On* handlers. By doing the fusion, we make the call stack depth shorter which
     /// helps debugging and some performance.
     /// </summary>
-    internal sealed class AnonymousSafeObserver<T> : IObserver<T>
+    internal sealed class AnonymousSafeObserver<T> : ISafeObserver<T>
     {
         private readonly Action<T> _onNext;
         private readonly Action<Exception> _onError;
         private readonly Action _onCompleted;
-        private readonly IDisposable _disposable;
+        private IDisposable _disposable;
 
         private int isStopped;
 
-        public AnonymousSafeObserver(Action<T> onNext, Action<Exception> onError, Action onCompleted, IDisposable disposable)
+        public AnonymousSafeObserver(Action<T> onNext, Action<Exception> onError, Action onCompleted)
         {
             _onNext = onNext;
             _onError = onError;
             _onCompleted = onCompleted;
-            _disposable = disposable;
         }
 
         public void OnNext(T value)
@@ -48,7 +48,7 @@ public void OnNext(T value)
                 finally
                 {
                     if (!__noError)
-                        _disposable.Dispose();
+                        Dispose();
                 }
             }
         }
@@ -57,30 +57,32 @@ public void OnError(Exception error)
         {
             if (Interlocked.Exchange(ref isStopped, 1) == 0)
             {
-                try
+                using (this)
                 {
                     _onError(error);
                 }
-                finally
-                {
-                    _disposable.Dispose();
-                }
             }
         }
 
         public void OnCompleted()
         {
             if (Interlocked.Exchange(ref isStopped, 1) == 0)
             {
-                try
+                using (this)
                 {
                     _onCompleted();
                 }
-                finally
-                {
-                    _disposable.Dispose();
-                }
             }
         }
+
+        public void SetResource(IDisposable resource)
+        {
+            Disposable.SetSingle(ref _disposable, resource);
+        }
+
+        public void Dispose()
+        {
+            Disposable.TryDispose(ref _disposable);
+        }
     }
 }
diff --git a/Rx.NET/Source/src/System.Reactive/Internal/ISafeObserver.cs b/Rx.NET/Source/src/System.Reactive/Internal/ISafeObserver.cs
@@ -0,0 +1,16 @@
+﻿// 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. 
+
+namespace System.Reactive
+{
+    /// <summary>
+    /// Base interface for observers that can dispose of a resource on a terminal notification
+    /// or when disposed itself.
+    /// </summary>
+    /// <typeparam name="T"></typeparam>
+    internal interface ISafeObserver<in T> : IObserver<T>, IDisposable
+    {
+        void SetResource(IDisposable resource);
+    }
+}
diff --git a/Rx.NET/Source/src/System.Reactive/Internal/ObserverWithToken.cs b/Rx.NET/Source/src/System.Reactive/Internal/ObserverWithToken.cs
@@ -15,49 +15,46 @@ namespace System.Reactive
     /// to be disposed upon termination.
     /// </summary>
     /// <typeparam name="T">The element type of the sequence.</typeparam>
-    internal sealed class ObserverWithToken<T> : IObserver<T>
+    internal sealed class ObserverWithToken<T> : ISafeObserver<T>
     {
         readonly IObserver<T> _downstream;
 
         IDisposable _tokenDisposable;
 
         public ObserverWithToken(IObserver<T> downstream)
         {
-            this._downstream = downstream;
-        }
-
-        internal void SetTokenDisposable(IDisposable d)
-        {
-            Disposable.SetSingle(ref _tokenDisposable, d);
+            _downstream = downstream;
         }
 
         public void OnCompleted()
         {
-            try
+            using (this)
             {
                 _downstream.OnCompleted();
             }
-            finally
-            {
-                Disposable.TryDispose(ref _tokenDisposable);    
-            }
         }
 
         public void OnError(Exception error)
         {
-            try
+            using (this)
             {
                 _downstream.OnError(error);
             }
-            finally
-            {
-                Disposable.TryDispose(ref _tokenDisposable);
-            }
         }
 
         public void OnNext(T value)
         {
             _downstream.OnNext(value);
         }
+
+        public void SetResource(IDisposable resource)
+        {
+            Disposable.SetSingle(ref _tokenDisposable, resource);
+        }
+
+        public void Dispose()
+        {
+            Disposable.TryDispose(ref _tokenDisposable);
+        }
     }
 }
diff --git a/Rx.NET/Source/src/System.Reactive/Internal/Producer.cs b/Rx.NET/Source/src/System.Reactive/Internal/Producer.cs
@@ -46,7 +46,9 @@ public IDisposable SubscribeRaw(IObserver<TSource> observer, bool enableSafeguar
             //
             if (enableSafeguard)
             {
-                observer = SafeObserver<TSource>.Create(observer, subscription);
+                var safeObserver = SafeObserver<TSource>.Create(observer);
+                safeObserver.SetResource(subscription);
+                observer = safeObserver;
             }
 
             if (CurrentThreadScheduler.IsScheduleRequired)
@@ -90,22 +92,20 @@ public IDisposable Subscribe(IObserver<TTarget> observer)
 
         public IDisposable SubscribeRaw(IObserver<TTarget> observer, bool enableSafeguard)
         {
-            SingleAssignmentDisposable subscription = null;
+            ISafeObserver<TTarget> safeObserver = null;
 
             //
             // See AutoDetachObserver.cs for more information on the safeguarding requirement and
             // its implementation aspects.
             //
             if (enableSafeguard)
             {
-                subscription = new SingleAssignmentDisposable();
-                observer = SafeObserver<TTarget>.Create(observer, subscription);
+                observer = safeObserver = SafeObserver<TTarget>.Create(observer);
             }
 
             var sink = CreateSink(observer);
 
-            if (subscription != null)
-                subscription.Disposable = sink;
+            safeObserver?.SetResource(sink);
 
             if (CurrentThreadScheduler.IsScheduleRequired)
             {
@@ -118,7 +118,7 @@ public IDisposable SubscribeRaw(IObserver<TTarget> observer, bool enableSafeguar
                 Run(sink);
             }
 
-            return (IDisposable)subscription ?? sink;
+            return sink;
         }
 
         /// <summary>
diff --git a/Rx.NET/Source/src/System.Reactive/Internal/SafeObserver.cs b/Rx.NET/Source/src/System.Reactive/Internal/SafeObserver.cs
@@ -2,34 +2,36 @@
 // 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;
+
 namespace System.Reactive
 {
     //
     // See AutoDetachObserver.cs for more information on the safeguarding requirement and
     // its implementation aspects.
     //
 
-    internal sealed class SafeObserver<TSource> : IObserver<TSource>
+    internal sealed class SafeObserver<TSource> : ISafeObserver<TSource>
     {
         private readonly IObserver<TSource> _observer;
-        private readonly IDisposable _disposable;
 
-        public static IObserver<TSource> Create(IObserver<TSource> observer, IDisposable disposable)
+        private IDisposable _disposable;
+
+        public static ISafeObserver<TSource> Create(IObserver<TSource> observer)
         {
             if (observer is AnonymousObserver<TSource> a)
             {
-                return a.MakeSafe(disposable);
+                return a.MakeSafe();
             }
             else
             {
-                return new SafeObserver<TSource>(observer, disposable);
+                return new SafeObserver<TSource>(observer);
             }
         }
 
-        private SafeObserver(IObserver<TSource> observer, IDisposable disposable)
+        private SafeObserver(IObserver<TSource> observer)
         {
             _observer = observer;
-            _disposable = disposable;
         }
 
         public void OnNext(TSource value)
@@ -44,33 +46,35 @@ public void OnNext(TSource value)
             {
                 if (!__noError)
                 {
-                    _disposable.Dispose();
+                    Dispose();
                 }
             }
         }
 
         public void OnError(Exception error)
         {
-            try
+            using (this)
             {
                 _observer.OnError(error);
             }
-            finally
-            {
-                _disposable.Dispose();
-            }
         }
 
         public void OnCompleted()
         {
-            try
+            using (this)
             {
                 _observer.OnCompleted();
             }
-            finally
-            {
-                _disposable.Dispose();
-            }
+        }
+
+        public void SetResource(IDisposable resource)
+        {
+            Disposable.SetSingle(ref _disposable, resource);
+        }
+
+        public void Dispose()
+        {
+            Disposable.TryDispose(ref _disposable);
         }
     }
 }
diff --git a/Rx.NET/Source/src/System.Reactive/Observable.Extensions.cs b/Rx.NET/Source/src/System.Reactive/Observable.Extensions.cs
@@ -266,7 +266,7 @@ private static void Subscribe_<T>(this IObservable<T> source, IObserver<T> obser
                     //
                     var d = source.Subscribe/*Unsafe*/(consumer);
 
-                    consumer.SetTokenDisposable(token.Register(state => ((IDisposable)state).Dispose(), d));
+                    consumer.SetResource(token.Register(state => ((IDisposable)state).Dispose(), d));
                 }
             }
             else
diff --git a/Rx.NET/Source/src/System.Reactive/Platforms/Desktop/Linq/QueryLanguage.Remoting.cs b/Rx.NET/Source/src/System.Reactive/Platforms/Desktop/Linq/QueryLanguage.Remoting.cs
@@ -92,7 +92,7 @@ public IDisposable Subscribe(IObserver<T> observer)
                 //
                 var d = remotableObservable.Subscribe/*Unsafe*/(new RemotableObserver<T>(consumer));
 
-                consumer.SetTokenDisposable(d);
+                consumer.SetResource(d);
 
                 return d;
             }

Original file line number	Diff line number	Diff line change
`@@ -84,6 +84,6 @@ public AnonymousObserver(Action<T> onNext, Action onCompleted)`
`84`	`84`	`/// </summary>`
`85`	`85`	`protected override void OnCompletedCore() => _onCompleted();`
`86`	`86`
`87`		`- internal IObserver<T> MakeSafe(IDisposable disposable) => new AnonymousSafeObserver<T>(_onNext, _onError, _onCompleted, disposable);`
	`87`	`+ internal ISafeObserver<T> MakeSafe() => new AnonymousSafeObserver<T>(_onNext, _onError, _onCompleted);`
`88`	`88`	`}`
`89`	`89`	`}`
Original file line number	Diff line number	Diff line change
`@@ -2,6 +2,7 @@`
`2`	`2`	`// The .NET Foundation licenses this file to you under the Apache 2.0 License.`
`3`	`3`	`// See the LICENSE file in the project root for more information.`
`4`	`4`
	`5`	`+using System.Reactive.Disposables;`
`5`	`6`	`using System.Threading;`
`6`	`7`
`7`	`8`	`namespace System.Reactive`
`@@ -18,21 +19,20 @@ namespace System.Reactive`
`18`	`19`	`/// that accept delegates for the On* handlers. By doing the fusion, we make the call stack depth shorter which`
`19`	`20`	`/// helps debugging and some performance.`
`20`	`21`	`/// </summary>`
`21`		`- internal sealed class AnonymousSafeObserver<T> : IObserver<T>`
	`22`	`+ internal sealed class AnonymousSafeObserver<T> : ISafeObserver<T>`
`22`	`23`	`{`
`23`	`24`	`private readonly Action<T> _onNext;`
`24`	`25`	`private readonly Action<Exception> _onError;`
`25`	`26`	`private readonly Action _onCompleted;`
`26`		`- private readonly IDisposable _disposable;`
	`27`	`+ private IDisposable _disposable;`
`27`	`28`
`28`	`29`	`private int isStopped;`
`29`	`30`
`30`		`- public AnonymousSafeObserver(Action<T> onNext, Action<Exception> onError, Action onCompleted, IDisposable disposable)`
	`31`	`+ public AnonymousSafeObserver(Action<T> onNext, Action<Exception> onError, Action onCompleted)`
`31`	`32`	`{`
`32`	`33`	`_onNext = onNext;`
`33`	`34`	`_onError = onError;`
`34`	`35`	`_onCompleted = onCompleted;`
`35`		`- _disposable = disposable;`
`36`	`36`	`}`
`37`	`37`
`38`	`38`	`public void OnNext(T value)`
`@@ -48,7 +48,7 @@ public void OnNext(T value)`
`48`	`48`	`finally`
`49`	`49`	`{`
`50`	`50`	`if (!__noError)`
`51`		`- _disposable.Dispose();`
	`51`	`+ Dispose();`
`52`	`52`	`}`
`53`	`53`	`}`
`54`	`54`	`}`
`@@ -57,30 +57,32 @@ public void OnError(Exception error)`
`57`	`57`	`{`
`58`	`58`	`if (Interlocked.Exchange(ref isStopped, 1) == 0)`
`59`	`59`	`{`
`60`		`- try`
	`60`	`+ using (this)`
`61`	`61`	`{`
`62`	`62`	`_onError(error);`
`63`	`63`	`}`
`64`		`- finally`
`65`		`- {`
`66`		`- _disposable.Dispose();`
`67`		`- }`
`68`	`64`	`}`
`69`	`65`	`}`
`70`	`66`
`71`	`67`	`public void OnCompleted()`
`72`	`68`	`{`
`73`	`69`	`if (Interlocked.Exchange(ref isStopped, 1) == 0)`
`74`	`70`	`{`
`75`		`- try`
	`71`	`+ using (this)`
`76`	`72`	`{`
`77`	`73`	`_onCompleted();`
`78`	`74`	`}`
`79`		`- finally`
`80`		`- {`
`81`		`- _disposable.Dispose();`
`82`		`- }`
`83`	`75`	`}`
`84`	`76`	`}`
	`77`	`+`
	`78`	`+ public void SetResource(IDisposable resource)`
	`79`	`+ {`
	`80`	`+ Disposable.SetSingle(ref _disposable, resource);`
	`81`	`+ }`
	`82`	`+`
	`83`	`+ public void Dispose()`
	`84`	`+ {`
	`85`	`+ Disposable.TryDispose(ref _disposable);`
	`86`	`+ }`
`85`	`87`	`}`
`86`	`88`	`}`
Original file line number	Diff line number	Diff line change
`@@ -15,49 +15,46 @@ namespace System.Reactive`
`15`	`15`	`/// to be disposed upon termination.`
`16`	`16`	`/// </summary>`
`17`	`17`	`/// <typeparam name="T">The element type of the sequence.</typeparam>`
`18`		`- internal sealed class ObserverWithToken<T> : IObserver<T>`
	`18`	`+ internal sealed class ObserverWithToken<T> : ISafeObserver<T>`
`19`	`19`	`{`
`20`	`20`	`readonly IObserver<T> _downstream;`
`21`	`21`
`22`	`22`	`IDisposable _tokenDisposable;`
`23`	`23`
`24`	`24`	`public ObserverWithToken(IObserver<T> downstream)`
`25`	`25`	`{`
`26`		`- this._downstream = downstream;`
`27`		`- }`
`28`		`-`
`29`		`- internal void SetTokenDisposable(IDisposable d)`
`30`		`- {`
`31`		`- Disposable.SetSingle(ref _tokenDisposable, d);`
	`26`	`+ _downstream = downstream;`
`32`	`27`	`}`
`33`	`28`
`34`	`29`	`public void OnCompleted()`
`35`	`30`	`{`
`36`		`- try`
	`31`	`+ using (this)`
`37`	`32`	`{`
`38`	`33`	`_downstream.OnCompleted();`
`39`	`34`	`}`
`40`		`- finally`
`41`		`- {`
`42`		`- Disposable.TryDispose(ref _tokenDisposable);`
`43`		`- }`
`44`	`35`	`}`
`45`	`36`
`46`	`37`	`public void OnError(Exception error)`
`47`	`38`	`{`
`48`		`- try`
	`39`	`+ using (this)`
`49`	`40`	`{`
`50`	`41`	`_downstream.OnError(error);`
`51`	`42`	`}`
`52`		`- finally`
`53`		`- {`
`54`		`- Disposable.TryDispose(ref _tokenDisposable);`
`55`		`- }`
`56`	`43`	`}`
`57`	`44`
`58`	`45`	`public void OnNext(T value)`
`59`	`46`	`{`
`60`	`47`	`_downstream.OnNext(value);`
`61`	`48`	`}`
	`49`	`+`
	`50`	`+ public void SetResource(IDisposable resource)`
	`51`	`+ {`
	`52`	`+ Disposable.SetSingle(ref _tokenDisposable, resource);`
	`53`	`+ }`
	`54`	`+`
	`55`	`+ public void Dispose()`
	`56`	`+ {`
	`57`	`+ Disposable.TryDispose(ref _tokenDisposable);`
	`58`	`+ }`
`62`	`59`	`}`
`63`	`60`	`}`
Original file line number	Diff line number	Diff line change
`@@ -46,7 +46,9 @@ public IDisposable SubscribeRaw(IObserver<TSource> observer, bool enableSafeguar`
`46`	`46`	`//`
`47`	`47`	`if (enableSafeguard)`
`48`	`48`	`{`
`49`		`- observer = SafeObserver<TSource>.Create(observer, subscription);`
	`49`	`+ var safeObserver = SafeObserver<TSource>.Create(observer);`
	`50`	`+ safeObserver.SetResource(subscription);`
	`51`	`+ observer = safeObserver;`
`50`	`52`	`}`
`51`	`53`
`52`	`54`	`if (CurrentThreadScheduler.IsScheduleRequired)`
`@@ -90,22 +92,20 @@ public IDisposable Subscribe(IObserver<TTarget> observer)`
`90`	`92`
`91`	`93`	`public IDisposable SubscribeRaw(IObserver<TTarget> observer, bool enableSafeguard)`
`92`	`94`	`{`
`93`		`- SingleAssignmentDisposable subscription = null;`
	`95`	`+ ISafeObserver<TTarget> safeObserver = null;`
`94`	`96`
`95`	`97`	`//`
`96`	`98`	`// See AutoDetachObserver.cs for more information on the safeguarding requirement and`
`97`	`99`	`// its implementation aspects.`
`98`	`100`	`//`
`99`	`101`	`if (enableSafeguard)`
`100`	`102`	`{`
`101`		`- subscription = new SingleAssignmentDisposable();`
`102`		`- observer = SafeObserver<TTarget>.Create(observer, subscription);`
	`103`	`+ observer = safeObserver = SafeObserver<TTarget>.Create(observer);`
`103`	`104`	`}`
`104`	`105`
`105`	`106`	`var sink = CreateSink(observer);`
`106`	`107`
`107`		`- if (subscription != null)`
`108`		`- subscription.Disposable = sink;`
	`108`	`+ safeObserver?.SetResource(sink);`
`109`	`109`
`110`	`110`	`if (CurrentThreadScheduler.IsScheduleRequired)`
`111`	`111`	`{`
`@@ -118,7 +118,7 @@ public IDisposable SubscribeRaw(IObserver<TTarget> observer, bool enableSafeguar`
`118`	`118`	`Run(sink);`
`119`	`119`	`}`
`120`	`120`
`121`		`- return (IDisposable)subscription ?? sink;`
	`121`	`+ return sink;`
`122`	`122`	`}`
`123`	`123`
`124`	`124`	`/// <summary>`
Original file line number	Diff line number	Diff line change
`@@ -2,34 +2,36 @@`
`2`	`2`	`// The .NET Foundation licenses this file to you under the Apache 2.0 License.`
`3`	`3`	`// See the LICENSE file in the project root for more information.`
`4`	`4`
	`5`	`+using System.Reactive.Disposables;`
	`6`	`+`
`5`	`7`	`namespace System.Reactive`
`6`	`8`	`{`
`7`	`9`	`//`
`8`	`10`	`// See AutoDetachObserver.cs for more information on the safeguarding requirement and`
`9`	`11`	`// its implementation aspects.`
`10`	`12`	`//`
`11`	`13`
`12`		`- internal sealed class SafeObserver<TSource> : IObserver<TSource>`
	`14`	`+ internal sealed class SafeObserver<TSource> : ISafeObserver<TSource>`
`13`	`15`	`{`
`14`	`16`	`private readonly IObserver<TSource> _observer;`
`15`		`- private readonly IDisposable _disposable;`
`16`	`17`
`17`		`- public static IObserver<TSource> Create(IObserver<TSource> observer, IDisposable disposable)`
	`18`	`+ private IDisposable _disposable;`
	`19`	`+`
	`20`	`+ public static ISafeObserver<TSource> Create(IObserver<TSource> observer)`
`18`	`21`	`{`
`19`	`22`	`if (observer is AnonymousObserver<TSource> a)`
`20`	`23`	`{`
`21`		`- return a.MakeSafe(disposable);`
	`24`	`+ return a.MakeSafe();`
`22`	`25`	`}`
`23`	`26`	`else`
`24`	`27`	`{`
`25`		`- return new SafeObserver<TSource>(observer, disposable);`
	`28`	`+ return new SafeObserver<TSource>(observer);`
`26`	`29`	`}`
`27`	`30`	`}`
`28`	`31`
`29`		`- private SafeObserver(IObserver<TSource> observer, IDisposable disposable)`
	`32`	`+ private SafeObserver(IObserver<TSource> observer)`
`30`	`33`	`{`
`31`	`34`	`_observer = observer;`
`32`		`- _disposable = disposable;`
`33`	`35`	`}`
`34`	`36`
`35`	`37`	`public void OnNext(TSource value)`
`@@ -44,33 +46,35 @@ public void OnNext(TSource value)`
`44`	`46`	`{`
`45`	`47`	`if (!__noError)`
`46`	`48`	`{`
`47`		`- _disposable.Dispose();`
	`49`	`+ Dispose();`
`48`	`50`	`}`
`49`	`51`	`}`
`50`	`52`	`}`
`51`	`53`
`52`	`54`	`public void OnError(Exception error)`
`53`	`55`	`{`
`54`		`- try`
	`56`	`+ using (this)`
`55`	`57`	`{`
`56`	`58`	`_observer.OnError(error);`
`57`	`59`	`}`
`58`		`- finally`
`59`		`- {`
`60`		`- _disposable.Dispose();`
`61`		`- }`
`62`	`60`	`}`
`63`	`61`
`64`	`62`	`public void OnCompleted()`
`65`	`63`	`{`
`66`		`- try`
	`64`	`+ using (this)`
`67`	`65`	`{`
`68`	`66`	`_observer.OnCompleted();`
`69`	`67`	`}`
`70`		`- finally`
`71`		`- {`
`72`		`- _disposable.Dispose();`
`73`		`- }`
	`68`	`+ }`
	`69`	`+`
	`70`	`+ public void SetResource(IDisposable resource)`
	`71`	`+ {`
	`72`	`+ Disposable.SetSingle(ref _disposable, resource);`
	`73`	`+ }`
	`74`	`+`
	`75`	`+ public void Dispose()`
	`76`	`+ {`
	`77`	`+ Disposable.TryDispose(ref _disposable);`
`74`	`78`	`}`
`75`	`79`	`}`
`76`	`80`	`}`
Original file line number	Diff line number	Diff line change
`@@ -266,7 +266,7 @@ private static void Subscribe_<T>(this IObservable<T> source, IObserver<T> obser`
`266`	`266`	`//`
`267`	`267`	`var d = source.Subscribe/Unsafe/(consumer);`
`268`	`268`
`269`		`- consumer.SetTokenDisposable(token.Register(state => ((IDisposable)state).Dispose(), d));`
	`269`	`+ consumer.SetResource(token.Register(state => ((IDisposable)state).Dispose(), d));`
`270`	`270`	`}`
`271`	`271`	`}`
`272`	`272`	`else`
Original file line number	Diff line number	Diff line change
`@@ -92,7 +92,7 @@ public IDisposable Subscribe(IObserver<T> observer)`
`92`	`92`	`//`
`93`	`93`	`var d = remotableObservable.Subscribe/Unsafe/(new RemotableObserver<T>(consumer));`
`94`	`94`
`95`		`- consumer.SetTokenDisposable(d);`
	`95`	`+ consumer.SetResource(d);`
`96`	`96`
`97`	`97`	`return d;`
`98`	`98`	`}`