Update client-libraries.md (#19952)

* Update client-libraries.md

Fixes a few issues in the `Observable` code sample, there was a missing generic definition in the `GrpcStreamSubscription`. Removes the `Debug.Assert` and use argument exceptions instead. Few other cosmetic changes.

* Update client-libraries.md

Fix linting issues

* Update client-libraries.md

Forgot a `_` before the observer

* Update client-libraries.md

Fix mix up with the cancellation token

* Remove redundant code

Co-authored-by: David Pine <david.pine@microsoft.com>

Co-authored-by: David Pine <david.pine@microsoft.com>

Author: Gimly

docs/architecture/grpc-for-wcf-developers/client-libraries.md

@@ -29,9 +29,7 @@ This code is longer than the `IAsyncEnumerable<T>` code, because C# doesn't have
 
 ```csharp
 using System;
-using System.Diagnostics;
 using System.Threading;
-using System.Threading.Tasks;
 
 namespace Grpc.Core
 {
@@ -43,14 +41,14 @@ namespace Grpc.Core
 
         public GrpcStreamObservable(IAsyncStreamReader<T> reader, CancellationToken token = default)
         {
-            _reader = reader;
+            _reader = reader ?? throw new ArgumentNullException(nameof(reader));
             _token = token;
             _used = 0;
         }
 
         public IDisposable Subscribe(IObserver<T> observer) =>
             Interlocked.Exchange(ref _used, 1) == 0
-                ? new GrpcStreamSubscription(_reader, observer, _token)
+                ? new GrpcStreamSubscription<T>(_reader, observer, _token)
                 : throw new InvalidOperationException("Subscribe can only be called once.");
 
     }
@@ -63,28 +61,35 @@ namespace Grpc.Core
 The `GrpcStreamSubscription` class handles the enumeration of the `IAsyncStreamReader`:
 
 ```csharp
-public class GrpcStreamSubscription : IDisposable
+public class GrpcStreamSubscription<T> : IDisposable
 {
-    private readonly Task _task;
+    private readonly IAsyncStreamReader<T> _reader;
+    private readonly IObserver<T> _observer;
+
     private readonly CancellationTokenSource _tokenSource;
+
+    private readonly Task _task;
+
     private bool _completed;
 
-    public GrpcStreamSubscription(IAsyncStreamReader<T> reader, IObserver<T> observer, CancellationToken token)
+    public GrpcStreamSubscription(IAsyncStreamReader<T> reader, IObserver<T> observer, CancellationToken token = default)
     {
-        Debug.Assert(reader != null);
-        Debug.Assert(observer != null);
+        _reader = reader ?? throw new ArgumentNullException(nameof(reader));
+        _observer = observer ?? throw new ArgumentNullException(nameof(observer));
+
         _tokenSource = new CancellationTokenSource();
         token.Register(_tokenSource.Cancel);
-        _task = Run(reader, observer, _tokenSource.Token);
+
+        _task = Run(_tokenSource.Token);
     }
 
-    private async Task Run(IAsyncStreamReader<T> reader, IObserver<T> observer, CancellationToken token)
+    private async Task Run(CancellationToken token)
     {
         while (!token.IsCancellationRequested)
         {
             try
             {
-                if (!await reader.MoveNext(token)) break;
+                if (!await _reader.MoveNext(token)) break;
             }
             catch (RpcException e) when (e.StatusCode == Grpc.Core.StatusCode.NotFound)
             {
@@ -96,16 +101,16 @@ public class GrpcStreamSubscription : IDisposable
             }
             catch (Exception e)
             {
-                observer.OnError(e);
+                _observer.OnError(e);
                 _completed = true;
                 return;
             }
 
-            observer.OnNext(reader.Current);
+            _observer.OnNext(_reader.Current);
         }
 
         _completed = true;
-        observer.OnCompleted();
+        _observer.OnCompleted();
     }
 
     public void Dispose()
@@ -126,16 +131,16 @@ All that is required now is a simple extension method to create the observable f
 
 ```csharp
 using System;
-using System.Diagnostics;
 using System.Threading;
-using System.Threading.Tasks;
 
 namespace Grpc.Core
 {
     public static class AsyncStreamReaderObservableExtensions
     {
-        public static IObservable<T> AsObservable<T>(this IAsyncStreamReader<T> reader) =>
-            new GrpcStreamObservable<T>(reader);
+        public static IObservable<T> AsObservable<T>(
+            this IAsyncStreamReader<T> reader,
+            CancellationToken cancellationToken = default) =>
+            new GrpcStreamObservable<T>(reader, cancellationToken);
     }
 }
 ```