Naïve attempt at fixing some unobserved task exceptions (#175)

I am 100% positive that this is a bad approach and I'm not sure exactly how to go about testing this... but figured giving you some insight into where the problem is might help understand what those things should look like!

Author: deanward81

src/protobuf-net.Grpc/CallContext.Streaming.cs

@@ -31,21 +31,26 @@ private static async IAsyncEnumerable<TResponse> FullDuplexImpl<TRequest, TRespo
             [EnumeratorCancellation] CancellationToken cancellationToken)
         {
             using var allDone = CancellationTokenSource.CreateLinkedTokenSource(context.CancellationToken, cancellationToken);
+            Task? consumed = null;
             try
             {
                 context = new CallContext(context, allDone.Token);
-                var consumed = Task.Run(() => consumer(source, context).AsTask(), allDone.Token); // note this shares a capture scope
+                consumed = Task.Run(() => consumer(source, context).AsTask(), allDone.Token); // note this shares a capture scope
 
                 await foreach (var value in producer(context).WithCancellation(cancellationToken).ConfigureAwait(false))
                 {
                     yield return value;
                 }
-                await consumed.ConfigureAwait(false);
             }
             finally
             {
                 // stop the producer, in any exit scenario
                 allDone.Cancel();
+                
+                if (consumed != null)
+                {
+                    await consumed.ConfigureAwait(false);
+                }
             }
         }
 
@@ -68,10 +73,11 @@ private static async IAsyncEnumerable<TResponse> FullDuplexImpl<TRequest, TRespo
             [EnumeratorCancellation] CancellationToken cancellationToken)
         {
             using var allDone = CancellationTokenSource.CreateLinkedTokenSource(context.CancellationToken, cancellationToken);
+            Task? consumed = null;
             try
             {
                 context = new CallContext(context, allDone.Token);
-                var consumed = Task.Run(async () =>
+                consumed = Task.Run(async () =>
                 {// note this shares a capture scope
                     await foreach (var value in source.WithCancellation(allDone.Token).ConfigureAwait(false))
                     {
@@ -82,12 +88,16 @@ private static async IAsyncEnumerable<TResponse> FullDuplexImpl<TRequest, TRespo
                 {
                     yield return value;
                 }
-                await consumed.ConfigureAwait(false);
             }
             finally
             {
                 // stop the producer, in any exit scenario
                 allDone.Cancel();
+
+                if (consumed != null)
+                {
+                    await consumed.ConfigureAwait(false);
+                }
             }
         }