Merge pull request #1648 from jtattermusch/csharp_disposable_calls

Incorporate feedback from API review: Disposable calls

17 files changed, 210 insertions, 144 deletions

diff --git a/src/csharp/Grpc.Core/AsyncClientStreamingCall.cs b/src/csharp/Grpc.Core/AsyncClientStreamingCall.cs
index 8cdc1c895b..d66b0d4974 100644
--- a/src/csharp/Grpc.Core/AsyncClientStreamingCall.cs
+++ b/src/csharp/Grpc.Core/AsyncClientStreamingCall.cs
@@ -40,15 +40,17 @@ namespace Grpc.Core
     /// <summary>
     /// Return type for client streaming calls.
     /// </summary>
-    public sealed class AsyncClientStreamingCall<TRequest, TResponse>
+    public sealed class AsyncClientStreamingCall<TRequest, TResponse> : IDisposable
     {
         readonly IClientStreamWriter<TRequest> requestStream;
         readonly Task<TResponse> result;
+        readonly Action disposeAction;
 
-        public AsyncClientStreamingCall(IClientStreamWriter<TRequest> requestStream, Task<TResponse> result)
+        public AsyncClientStreamingCall(IClientStreamWriter<TRequest> requestStream, Task<TResponse> result, Action disposeAction)
         {
             this.requestStream = requestStream;
             this.result = result;
+            this.disposeAction = disposeAction;
         }
 
         /// <summary>
@@ -81,5 +83,16 @@ namespace Grpc.Core
         {
             return result.GetAwaiter();
         }
+
+        /// <summary>
+        /// Provides means to provide after the call.
+        /// If the call has already finished normally (request stream has been completed and call result has been received), doesn't do anything.
+        /// Otherwise, requests cancellation of the call which should terminate all pending async operations associated with the call.
+        /// As a result, all resources being used by the call should be released eventually.
+        /// </summary>
+        public void Dispose()
+        {
+            disposeAction.Invoke();
+        }
     }
 }
diff --git a/src/csharp/Grpc.Core/AsyncDuplexStreamingCall.cs b/src/csharp/Grpc.Core/AsyncDuplexStreamingCall.cs
index 0d13a3d052..4c0d5936ac 100644
--- a/src/csharp/Grpc.Core/AsyncDuplexStreamingCall.cs
+++ b/src/csharp/Grpc.Core/AsyncDuplexStreamingCall.cs
@@ -40,15 +40,17 @@ namespace Grpc.Core
     /// <summary>
     /// Return type for bidirectional streaming calls.
     /// </summary>
-    public sealed class AsyncDuplexStreamingCall<TRequest, TResponse>
+    public sealed class AsyncDuplexStreamingCall<TRequest, TResponse> : IDisposable
     {
         readonly IClientStreamWriter<TRequest> requestStream;
         readonly IAsyncStreamReader<TResponse> responseStream;
+        readonly Action disposeAction;
 
-        public AsyncDuplexStreamingCall(IClientStreamWriter<TRequest> requestStream, IAsyncStreamReader<TResponse> responseStream)
+        public AsyncDuplexStreamingCall(IClientStreamWriter<TRequest> requestStream, IAsyncStreamReader<TResponse> responseStream, Action disposeAction)
         {
             this.requestStream = requestStream;
             this.responseStream = responseStream;
+            this.disposeAction = disposeAction;
         }
 
         /// <summary>
@@ -72,5 +74,16 @@ namespace Grpc.Core
                 return requestStream;
             }
         }
+
+        /// <summary>
+        /// Provides means to cleanup after the call.
+        /// If the call has already finished normally (request stream has been completed and response stream has been fully read), doesn't do anything.
+        /// Otherwise, requests cancellation of the call which should terminate all pending async operations associated with the call.
+        /// As a result, all resources being used by the call should be released eventually.
+        /// </summary>
+        public void Dispose()
+        {
+            disposeAction.Invoke();
+        }
     }
 }
diff --git a/src/csharp/Grpc.Core/AsyncServerStreamingCall.cs b/src/csharp/Grpc.Core/AsyncServerStreamingCall.cs
index 6a258d132c..7a479b9a23 100644
--- a/src/csharp/Grpc.Core/AsyncServerStreamingCall.cs
+++ b/src/csharp/Grpc.Core/AsyncServerStreamingCall.cs
@@ -40,13 +40,15 @@ namespace Grpc.Core
     /// <summary>
     /// Return type for server streaming calls.
     /// </summary>
-    public sealed class AsyncServerStreamingCall<TResponse>
+    public sealed class AsyncServerStreamingCall<TResponse> : IDisposable
     {
         readonly IAsyncStreamReader<TResponse> responseStream;
+        readonly Action disposeAction;
 
-        public AsyncServerStreamingCall(IAsyncStreamReader<TResponse> responseStream)
+        public AsyncServerStreamingCall(IAsyncStreamReader<TResponse> responseStream, Action disposeAction)
         {
             this.responseStream = responseStream;
+            this.disposeAction = disposeAction;
         }
 
         /// <summary>
@@ -59,5 +61,16 @@ namespace Grpc.Core
                 return responseStream;
             }
         }
+
+        /// <summary>
+        /// Provides means to cleanup after the call.
+        /// If the call has already finished normally (response stream has been fully read), doesn't do anything.
+        /// Otherwise, requests cancellation of the call which should terminate all pending async operations associated with the call.
+        /// As a result, all resources being used by the call should be released eventually.
+        /// </summary>
+        public void Dispose()
+        {
+            disposeAction.Invoke();
+        }
     }
 }
diff --git a/src/csharp/Grpc.Core/Calls.cs b/src/csharp/Grpc.Core/Calls.cs
index ba42a2d4f8..9f8baac684 100644
--- a/src/csharp/Grpc.Core/Calls.cs
+++ b/src/csharp/Grpc.Core/Calls.cs
@@ -73,7 +73,7 @@ namespace Grpc.Core
             asyncCall.StartServerStreamingCall(req, call.Headers);
             RegisterCancellationCallback(asyncCall, token);
             var responseStream = new ClientResponseStream<TRequest, TResponse>(asyncCall);
-            return new AsyncServerStreamingCall<TResponse>(responseStream);
+            return new AsyncServerStreamingCall<TResponse>(responseStream, asyncCall.Cancel);
         }
 
         public static AsyncClientStreamingCall<TRequest, TResponse> AsyncClientStreamingCall<TRequest, TResponse>(Call<TRequest, TResponse> call, CancellationToken token)
@@ -85,7 +85,7 @@ namespace Grpc.Core
             var resultTask = asyncCall.ClientStreamingCallAsync(call.Headers);
             RegisterCancellationCallback(asyncCall, token);
             var requestStream = new ClientRequestStream<TRequest, TResponse>(asyncCall);
-            return new AsyncClientStreamingCall<TRequest, TResponse>(requestStream, resultTask);
+            return new AsyncClientStreamingCall<TRequest, TResponse>(requestStream, resultTask, asyncCall.Cancel);
         }
 
         public static AsyncDuplexStreamingCall<TRequest, TResponse> AsyncDuplexStreamingCall<TRequest, TResponse>(Call<TRequest, TResponse> call, CancellationToken token)
@@ -98,7 +98,7 @@ namespace Grpc.Core
             RegisterCancellationCallback(asyncCall, token);
             var requestStream = new ClientRequestStream<TRequest, TResponse>(asyncCall);
             var responseStream = new ClientResponseStream<TRequest, TResponse>(asyncCall);
-            return new AsyncDuplexStreamingCall<TRequest, TResponse>(requestStream, responseStream);
+            return new AsyncDuplexStreamingCall<TRequest, TResponse>(requestStream, responseStream, asyncCall.Cancel);
         }
 
         private static void RegisterCancellationCallback<TRequest, TResponse>(AsyncCall<TRequest, TResponse> asyncCall, CancellationToken token)
diff --git a/src/csharp/Grpc.Core/IAsyncStreamReader.cs b/src/csharp/Grpc.Core/IAsyncStreamReader.cs
index 95b674c018..371fbf27ce 100644
--- a/src/csharp/Grpc.Core/IAsyncStreamReader.cs
+++ b/src/csharp/Grpc.Core/IAsyncStreamReader.cs
@@ -45,5 +45,6 @@ namespace Grpc.Core
     /// <typeparam name="T"></typeparam>
     public interface IAsyncStreamReader<TResponse> : IAsyncEnumerator<TResponse>
     {
+        // TODO(jtattermusch): consider just using IAsyncEnumerator instead of this interface.
     }
 }
diff --git a/src/csharp/Grpc.Core/IAsyncStreamWriter.cs b/src/csharp/Grpc.Core/IAsyncStreamWriter.cs
index 644f445401..2000210252 100644
--- a/src/csharp/Grpc.Core/IAsyncStreamWriter.cs
+++ b/src/csharp/Grpc.Core/IAsyncStreamWriter.cs
@@ -49,6 +49,6 @@ namespace Grpc.Core
         /// Writes a single asynchronously. Only one write can be pending at a time.
         /// </summary>
         /// <param name="message">the message to be written. Cannot be null.</param>
-        Task Write(T message);
+        Task WriteAsync(T message);
     }
 }
diff --git a/src/csharp/Grpc.Core/IClientStreamWriter.cs b/src/csharp/Grpc.Core/IClientStreamWriter.cs
index cc76d1369d..a3028bc374 100644
--- a/src/csharp/Grpc.Core/IClientStreamWriter.cs
+++ b/src/csharp/Grpc.Core/IClientStreamWriter.cs
@@ -48,6 +48,6 @@ namespace Grpc.Core
         /// <summary>
         /// Completes/closes the stream. Can only be called once there is no pending write. No writes should follow calling this.
         /// </summary>
-        Task Complete();
+        Task CompleteAsync();
     }
 }
diff --git a/src/csharp/Grpc.Core/IServerStreamWriter.cs b/src/csharp/Grpc.Core/IServerStreamWriter.cs
index 199a585a3f..9f3af59109 100644
--- a/src/csharp/Grpc.Core/IServerStreamWriter.cs
+++ b/src/csharp/Grpc.Core/IServerStreamWriter.cs
@@ -43,7 +43,7 @@ namespace Grpc.Core
     /// A writable stream of messages that is used in server-side handlers.
     /// </summary>
     public interface IServerStreamWriter<T> : IAsyncStreamWriter<T>
-        where T : class
     {
+        // TODO(jtattermusch): consider just using IAsyncStreamWriter instead of this interface.
     }
 }
diff --git a/src/csharp/Grpc.Core/Internal/ClientRequestStream.cs b/src/csharp/Grpc.Core/Internal/ClientRequestStream.cs
index b9fc10cd16..58f493463b 100644
--- a/src/csharp/Grpc.Core/Internal/ClientRequestStream.cs
+++ b/src/csharp/Grpc.Core/Internal/ClientRequestStream.cs
@@ -46,14 +46,14 @@ namespace Grpc.Core.Internal
             this.call = call;
         }
 
-        public Task Write(TRequest message)
+        public Task WriteAsync(TRequest message)
         {
             var taskSource = new AsyncCompletionTaskSource<object>();
             call.StartSendMessage(message, taskSource.CompletionDelegate);
             return taskSource.Task;
         }
 
-        public Task Complete()
+        public Task CompleteAsync()
         {
             var taskSource = new AsyncCompletionTaskSource<object>();
             call.StartSendCloseFromClient(taskSource.CompletionDelegate);
diff --git a/src/csharp/Grpc.Core/Internal/ServerCallHandler.cs b/src/csharp/Grpc.Core/Internal/ServerCallHandler.cs
index 20ac46c234..f494d9e0ff 100644
--- a/src/csharp/Grpc.Core/Internal/ServerCallHandler.cs
+++ b/src/csharp/Grpc.Core/Internal/ServerCallHandler.cs
@@ -78,7 +78,7 @@ namespace Grpc.Core.Internal
                 Preconditions.CheckArgument(!await requestStream.MoveNext());
                 var context = new ServerCallContext();  // TODO(jtattermusch): initialize the context
                 var result = await handler(context, request);
-                await responseStream.Write(result);
+                await responseStream.WriteAsync(result);
             } 
             catch (Exception e)
             {
@@ -87,7 +87,7 @@ namespace Grpc.Core.Internal
             }
             try
             {
-                await responseStream.WriteStatus(status);
+                await responseStream.WriteStatusAsync(status);
             }
             catch (OperationCanceledException)
             {
@@ -140,7 +140,7 @@ namespace Grpc.Core.Internal
 
             try
             {
-                await responseStream.WriteStatus(status);
+                await responseStream.WriteStatusAsync(status);
             }
             catch (OperationCanceledException)
             {
@@ -181,7 +181,7 @@ namespace Grpc.Core.Internal
                 var result = await handler(context, requestStream);
                 try
                 {
-                    await responseStream.Write(result);
+                    await responseStream.WriteAsync(result);
                 }
                 catch (OperationCanceledException)
                 {
@@ -196,7 +196,7 @@ namespace Grpc.Core.Internal
 
             try
             {
-                await responseStream.WriteStatus(status);
+                await responseStream.WriteStatusAsync(status);
             }
             catch (OperationCanceledException)
             {
@@ -243,7 +243,7 @@ namespace Grpc.Core.Internal
             }
             try
             {
-                await responseStream.WriteStatus(status);
+                await responseStream.WriteStatusAsync(status);
             }
             catch (OperationCanceledException)
             {
@@ -266,7 +266,7 @@ namespace Grpc.Core.Internal
             var requestStream = new ServerRequestStream<byte[], byte[]>(asyncCall);
             var responseStream = new ServerResponseStream<byte[], byte[]>(asyncCall);
 
-            await responseStream.WriteStatus(new Status(StatusCode.Unimplemented, "No such method."));
+            await responseStream.WriteStatusAsync(new Status(StatusCode.Unimplemented, "No such method."));
             // TODO(jtattermusch): if we don't read what client has sent, the server call never gets disposed.
             await requestStream.ToList();
             await finishedTask;
diff --git a/src/csharp/Grpc.Core/Internal/ServerResponseStream.cs b/src/csharp/Grpc.Core/Internal/ServerResponseStream.cs
index da688d504f..a2d77dd5b7 100644
--- a/src/csharp/Grpc.Core/Internal/ServerResponseStream.cs
+++ b/src/csharp/Grpc.Core/Internal/ServerResponseStream.cs
@@ -49,14 +49,14 @@ namespace Grpc.Core.Internal
             this.call = call;
         }
 
-        public Task Write(TResponse message)
+        public Task WriteAsync(TResponse message)
         {
             var taskSource = new AsyncCompletionTaskSource<object>();
             call.StartSendMessage(message, taskSource.CompletionDelegate);
             return taskSource.Task;
         }
 
-        public Task WriteStatus(Status status)
+        public Task WriteStatusAsync(Status status)
         {
             var taskSource = new AsyncCompletionTaskSource<object>();
             call.StartSendStatusFromServer(status, taskSource.CompletionDelegate);
diff --git a/src/csharp/Grpc.Core/Utils/AsyncStreamExtensions.cs b/src/csharp/Grpc.Core/Utils/AsyncStreamExtensions.cs
index a4f8989b30..8a748b45a8 100644
--- a/src/csharp/Grpc.Core/Utils/AsyncStreamExtensions.cs
+++ b/src/csharp/Grpc.Core/Utils/AsyncStreamExtensions.cs
@@ -78,11 +78,11 @@ namespace Grpc.Core.Utils
         {
             foreach (var element in elements)
             {
-                await streamWriter.Write(element);
+                await streamWriter.WriteAsync(element);
             }
             if (complete)
             {
-                await streamWriter.Complete();
+                await streamWriter.CompleteAsync();
             }
         }
 
@@ -94,7 +94,7 @@ namespace Grpc.Core.Utils
         {
             foreach (var element in elements)
             {
-                await streamWriter.Write(element);
+                await streamWriter.WriteAsync(element);
             }
         }
     }
diff --git a/src/csharp/Grpc.Examples.Tests/MathClientServerTests.cs b/src/csharp/Grpc.Examples.Tests/MathClientServerTests.cs
index 4997d3aa42..5aa6f4162d 100644
--- a/src/csharp/Grpc.Examples.Tests/MathClientServerTests.cs
+++ b/src/csharp/Grpc.Examples.Tests/MathClientServerTests.cs
@@ -96,7 +96,19 @@ namespace math.Tests
             Assert.AreEqual(0, response.Remainder);
         }
 
-        // TODO(jtattermusch): test division by zero
+        [Test]
+        public void DivByZero()
+        {
+            try
+            {
+                DivReply response = client.Div(new DivArgs.Builder { Dividend = 0, Divisor = 0 }.Build());
+                Assert.Fail();
+            }
+            catch (RpcException e)
+            {
+                Assert.AreEqual(StatusCode.Unknown, e.Status.StatusCode);
+            }   
+        }
 
         [Test]
         public void DivAsync()
@@ -114,11 +126,12 @@ namespace math.Tests
         {
             Task.Run(async () =>
             {
-                var call = client.Fib(new FibArgs.Builder { Limit = 6 }.Build());
-
-                var responses = await call.ResponseStream.ToList();
-                CollectionAssert.AreEqual(new List<long> { 1, 1, 2, 3, 5, 8 },
-                    responses.ConvertAll((n) => n.Num_));
+                using (var call = client.Fib(new FibArgs.Builder { Limit = 6 }.Build()))
+                {
+                    var responses = await call.ResponseStream.ToList();
+                    CollectionAssert.AreEqual(new List<long> { 1, 1, 2, 3, 5, 8 },
+                        responses.ConvertAll((n) => n.Num_));
+                }
             }).Wait();
         }
 
@@ -128,13 +141,15 @@ namespace math.Tests
         {
             Task.Run(async () =>
             {
-                var call = client.Sum();
-                var numbers = new List<long> { 10, 20, 30 }.ConvertAll(
-                         n => Num.CreateBuilder().SetNum_(n).Build());
+                using (var call = client.Sum())
+                {
+                    var numbers = new List<long> { 10, 20, 30 }.ConvertAll(
+                             n => Num.CreateBuilder().SetNum_(n).Build());
 
-                await call.RequestStream.WriteAll(numbers);
-                var result = await call.Result;
-                Assert.AreEqual(60, result.Num_);
+                    await call.RequestStream.WriteAll(numbers);
+                    var result = await call.Result;
+                    Assert.AreEqual(60, result.Num_);
+                }
             }).Wait();
         }
 
@@ -150,12 +165,14 @@ namespace math.Tests
                     new DivArgs.Builder { Dividend = 7, Divisor = 2 }.Build()
                 };
 
-                var call = client.DivMany();
-                await call.RequestStream.WriteAll(divArgsList);
-                var result = await call.ResponseStream.ToList();
+                using (var call = client.DivMany())
+                {
+                    await call.RequestStream.WriteAll(divArgsList);
+                    var result = await call.ResponseStream.ToList();
 
-                CollectionAssert.AreEqual(new long[] { 3, 4, 3 }, result.ConvertAll((divReply) => divReply.Quotient));
-                CollectionAssert.AreEqual(new long[] { 1, 16, 1 }, result.ConvertAll((divReply) => divReply.Remainder));
+                    CollectionAssert.AreEqual(new long[] { 3, 4, 3 }, result.ConvertAll((divReply) => divReply.Quotient));
+                    CollectionAssert.AreEqual(new long[] { 1, 16, 1 }, result.ConvertAll((divReply) => divReply.Remainder));
+                }
             }).Wait();
         }
     }
diff --git a/src/csharp/Grpc.Examples/MathExamples.cs b/src/csharp/Grpc.Examples/MathExamples.cs
index ab06a44c0d..d2cfbee18f 100644
--- a/src/csharp/Grpc.Examples/MathExamples.cs
+++ b/src/csharp/Grpc.Examples/MathExamples.cs
@@ -51,18 +51,13 @@ namespace math
             Console.WriteLine("DivAsync Result: " + result);
         }
 
-        public static async Task DivAsyncWithCancellationExample(Math.IMathClient stub)
-        {
-            Task<DivReply> resultTask = stub.DivAsync(new DivArgs.Builder { Dividend = 4, Divisor = 5 }.Build());
-            DivReply result = await resultTask;
-            Console.WriteLine(result);
-        }
-
         public static async Task FibExample(Math.IMathClient stub)
         {
-            var call = stub.Fib(new FibArgs.Builder { Limit = 5 }.Build());
-            List<Num> result = await call.ResponseStream.ToList();
-            Console.WriteLine("Fib Result: " + string.Join("|", result));
+            using (var call = stub.Fib(new FibArgs.Builder { Limit = 5 }.Build()))
+            {
+                List<Num> result = await call.ResponseStream.ToList();
+                Console.WriteLine("Fib Result: " + string.Join("|", result));
+            }
         }
 
         public static async Task SumExample(Math.IMathClient stub)
@@ -74,9 +69,11 @@ namespace math
                 new Num.Builder { Num_ = 3 }.Build()
             };
 
-            var call = stub.Sum();
-            await call.RequestStream.WriteAll(numbers);
-            Console.WriteLine("Sum Result: " + await call.Result);
+            using (var call = stub.Sum())
+            {
+                await call.RequestStream.WriteAll(numbers);
+                Console.WriteLine("Sum Result: " + await call.Result);
+            }
         }
 
         public static async Task DivManyExample(Math.IMathClient stub)
@@ -87,9 +84,11 @@ namespace math
                 new DivArgs.Builder { Dividend = 100, Divisor = 21 }.Build(),
                 new DivArgs.Builder { Dividend = 7, Divisor = 2 }.Build()
             };
-            var call = stub.DivMany();
-            await call.RequestStream.WriteAll(divArgsList);
-            Console.WriteLine("DivMany Result: " + string.Join("|", await call.ResponseStream.ToList()));
+            using (var call = stub.DivMany())
+            { 
+                await call.RequestStream.WriteAll(divArgsList);
+                Console.WriteLine("DivMany Result: " + string.Join("|", await call.ResponseStream.ToList()));
+            }
         }
 
         public static async Task DependendRequestsExample(Math.IMathClient stub)
@@ -101,9 +100,12 @@ namespace math
                 new Num.Builder { Num_ = 3 }.Build()
             };
 
-            var sumCall = stub.Sum();
-            await sumCall.RequestStream.WriteAll(numbers);
-            Num sum = await sumCall.Result;
+            Num sum;
+            using (var sumCall = stub.Sum())
+            {
+                await sumCall.RequestStream.WriteAll(numbers);
+                sum = await sumCall.Result;
+            }
 
             DivReply result = await stub.DivAsync(new DivArgs.Builder { Dividend = sum.Num_, Divisor = numbers.Count }.Build());
             Console.WriteLine("Avg Result: " + result);
diff --git a/src/csharp/Grpc.Examples/MathServiceImpl.cs b/src/csharp/Grpc.Examples/MathServiceImpl.cs
index 3b33b09bbd..e247ac9d73 100644
--- a/src/csharp/Grpc.Examples/MathServiceImpl.cs
+++ b/src/csharp/Grpc.Examples/MathServiceImpl.cs
@@ -62,7 +62,7 @@ namespace math
             {
                 foreach (var num in FibInternal(request.Limit))
                 {
-                    await responseStream.Write(num);
+                    await responseStream.WriteAsync(num);
                 }
             }
         }
@@ -81,7 +81,7 @@ namespace math
         {
             await requestStream.ForEach(async divArgs =>
             {
-                await responseStream.Write(DivInternal(divArgs));
+                await responseStream.WriteAsync(DivInternal(divArgs));
             });
         }
 
diff --git a/src/csharp/Grpc.IntegrationTesting/InteropClient.cs b/src/csharp/Grpc.IntegrationTesting/InteropClient.cs
index d907699698..dfaf18cae1 100644
--- a/src/csharp/Grpc.IntegrationTesting/InteropClient.cs
+++ b/src/csharp/Grpc.IntegrationTesting/InteropClient.cs
@@ -213,11 +213,13 @@ namespace Grpc.IntegrationTesting
 
                 var bodySizes = new List<int> { 27182, 8, 1828, 45904 }.ConvertAll((size) => StreamingInputCallRequest.CreateBuilder().SetPayload(CreateZerosPayload(size)).Build());
 
-                var call = client.StreamingInputCall();
-                await call.RequestStream.WriteAll(bodySizes);
+                using (var call = client.StreamingInputCall())
+                {
+                    await call.RequestStream.WriteAll(bodySizes);
 
-                var response = await call.Result;
-                Assert.AreEqual(74922, response.AggregatedPayloadSize);
+                    var response = await call.Result;
+                    Assert.AreEqual(74922, response.AggregatedPayloadSize);
+                }
                 Console.WriteLine("Passed!");
             }).Wait();
         }
@@ -236,14 +238,15 @@ namespace Grpc.IntegrationTesting
                     (size) => ResponseParameters.CreateBuilder().SetSize(size).Build()))
                 .Build();
 
-                var call = client.StreamingOutputCall(request);
-
-                var responseList = await call.ResponseStream.ToList();
-                foreach (var res in responseList)
+                using (var call = client.StreamingOutputCall(request))
                 {
-                    Assert.AreEqual(PayloadType.COMPRESSABLE, res.Payload.Type);
+                    var responseList = await call.ResponseStream.ToList();
+                    foreach (var res in responseList)
+                    {
+                        Assert.AreEqual(PayloadType.COMPRESSABLE, res.Payload.Type);
+                    }
+                    CollectionAssert.AreEqual(bodySizes, responseList.ConvertAll((item) => item.Payload.Body.Length));
                 }
-                CollectionAssert.AreEqual(bodySizes, responseList.ConvertAll((item) => item.Payload.Body.Length));
                 Console.WriteLine("Passed!");
             }).Wait();
         }
@@ -254,48 +257,48 @@ namespace Grpc.IntegrationTesting
             {
                 Console.WriteLine("running ping_pong");
 
-                var call = client.FullDuplexCall();
-
-                await call.RequestStream.Write(StreamingOutputCallRequest.CreateBuilder()
-                .SetResponseType(PayloadType.COMPRESSABLE)
-                .AddResponseParameters(ResponseParameters.CreateBuilder().SetSize(31415))
-                .SetPayload(CreateZerosPayload(27182)).Build());
-
-                Assert.IsTrue(await call.ResponseStream.MoveNext());
-                Assert.AreEqual(PayloadType.COMPRESSABLE, call.ResponseStream.Current.Payload.Type);
-                Assert.AreEqual(31415, call.ResponseStream.Current.Payload.Body.Length);
+                using (var call = client.FullDuplexCall())
+                {
+                    await call.RequestStream.WriteAsync(StreamingOutputCallRequest.CreateBuilder()
+                    .SetResponseType(PayloadType.COMPRESSABLE)
+                    .AddResponseParameters(ResponseParameters.CreateBuilder().SetSize(31415))
+                    .SetPayload(CreateZerosPayload(27182)).Build());
 
-                await call.RequestStream.Write(StreamingOutputCallRequest.CreateBuilder()
-                          .SetResponseType(PayloadType.COMPRESSABLE)
-                          .AddResponseParameters(ResponseParameters.CreateBuilder().SetSize(9))
-                          .SetPayload(CreateZerosPayload(8)).Build());
+                    Assert.IsTrue(await call.ResponseStream.MoveNext());
+                    Assert.AreEqual(PayloadType.COMPRESSABLE, call.ResponseStream.Current.Payload.Type);
+                    Assert.AreEqual(31415, call.ResponseStream.Current.Payload.Body.Length);
 
-                Assert.IsTrue(await call.ResponseStream.MoveNext());
-                Assert.AreEqual(PayloadType.COMPRESSABLE, call.ResponseStream.Current.Payload.Type);
-                Assert.AreEqual(9, call.ResponseStream.Current.Payload.Body.Length);
+                    await call.RequestStream.WriteAsync(StreamingOutputCallRequest.CreateBuilder()
+                              .SetResponseType(PayloadType.COMPRESSABLE)
+                              .AddResponseParameters(ResponseParameters.CreateBuilder().SetSize(9))
+                              .SetPayload(CreateZerosPayload(8)).Build());
 
-                await call.RequestStream.Write(StreamingOutputCallRequest.CreateBuilder()
-                          .SetResponseType(PayloadType.COMPRESSABLE)
-                          .AddResponseParameters(ResponseParameters.CreateBuilder().SetSize(2653))
-                          .SetPayload(CreateZerosPayload(1828)).Build());
+                    Assert.IsTrue(await call.ResponseStream.MoveNext());
+                    Assert.AreEqual(PayloadType.COMPRESSABLE, call.ResponseStream.Current.Payload.Type);
+                    Assert.AreEqual(9, call.ResponseStream.Current.Payload.Body.Length);
 
-                Assert.IsTrue(await call.ResponseStream.MoveNext());
-                Assert.AreEqual(PayloadType.COMPRESSABLE, call.ResponseStream.Current.Payload.Type);
-                Assert.AreEqual(2653, call.ResponseStream.Current.Payload.Body.Length);
+                    await call.RequestStream.WriteAsync(StreamingOutputCallRequest.CreateBuilder()
+                              .SetResponseType(PayloadType.COMPRESSABLE)
+                              .AddResponseParameters(ResponseParameters.CreateBuilder().SetSize(2653))
+                              .SetPayload(CreateZerosPayload(1828)).Build());
 
-                await call.RequestStream.Write(StreamingOutputCallRequest.CreateBuilder()
-                          .SetResponseType(PayloadType.COMPRESSABLE)
-                          .AddResponseParameters(ResponseParameters.CreateBuilder().SetSize(58979))
-                          .SetPayload(CreateZerosPayload(45904)).Build());
+                    Assert.IsTrue(await call.ResponseStream.MoveNext());
+                    Assert.AreEqual(PayloadType.COMPRESSABLE, call.ResponseStream.Current.Payload.Type);
+                    Assert.AreEqual(2653, call.ResponseStream.Current.Payload.Body.Length);
 
-                Assert.IsTrue(await call.ResponseStream.MoveNext());
-                Assert.AreEqual(PayloadType.COMPRESSABLE, call.ResponseStream.Current.Payload.Type);
-                Assert.AreEqual(58979, call.ResponseStream.Current.Payload.Body.Length);
+                    await call.RequestStream.WriteAsync(StreamingOutputCallRequest.CreateBuilder()
+                              .SetResponseType(PayloadType.COMPRESSABLE)
+                              .AddResponseParameters(ResponseParameters.CreateBuilder().SetSize(58979))
+                              .SetPayload(CreateZerosPayload(45904)).Build());
 
-                await call.RequestStream.Complete();
+                    Assert.IsTrue(await call.ResponseStream.MoveNext());
+                    Assert.AreEqual(PayloadType.COMPRESSABLE, call.ResponseStream.Current.Payload.Type);
+                    Assert.AreEqual(58979, call.ResponseStream.Current.Payload.Body.Length);
 
-                Assert.IsFalse(await call.ResponseStream.MoveNext());
+                    await call.RequestStream.CompleteAsync();
 
+                    Assert.IsFalse(await call.ResponseStream.MoveNext());
+                }
                 Console.WriteLine("Passed!");
             }).Wait();
         }
@@ -305,12 +308,13 @@ namespace Grpc.IntegrationTesting
             Task.Run(async () =>
             {
                 Console.WriteLine("running empty_stream");
-                var call = client.FullDuplexCall();
-                await call.RequestStream.Complete();
-
-                var responseList = await call.ResponseStream.ToList();
-                Assert.AreEqual(0, responseList.Count);
+                using (var call = client.FullDuplexCall())
+                {
+                    await call.RequestStream.CompleteAsync();
 
+                    var responseList = await call.ResponseStream.ToList();
+                    Assert.AreEqual(0, responseList.Count);
+                }
                 Console.WriteLine("Passed!");
             }).Wait();
         }
@@ -362,19 +366,21 @@ namespace Grpc.IntegrationTesting
                 Console.WriteLine("running cancel_after_begin");
 
                 var cts = new CancellationTokenSource();
-                var call = client.StreamingInputCall(cts.Token);
-                // TODO(jtattermusch): we need this to ensure call has been initiated once we cancel it.
-                await Task.Delay(1000);
-                cts.Cancel();
-
-                try
+                using (var call = client.StreamingInputCall(cts.Token))
                 {
-                    var response = await call.Result;
-                    Assert.Fail();
-                } 
-                catch (RpcException e)
-                {
-                    Assert.AreEqual(StatusCode.Cancelled, e.Status.StatusCode);
+                    // TODO(jtattermusch): we need this to ensure call has been initiated once we cancel it.
+                    await Task.Delay(1000);
+                    cts.Cancel();
+
+                    try
+                    {
+                        var response = await call.Result;
+                        Assert.Fail();
+                    }
+                    catch (RpcException e)
+                    {
+                        Assert.AreEqual(StatusCode.Cancelled, e.Status.StatusCode);
+                    }
                 }
                 Console.WriteLine("Passed!");
             }).Wait();
@@ -387,27 +393,28 @@ namespace Grpc.IntegrationTesting
                 Console.WriteLine("running cancel_after_first_response");
 
                 var cts = new CancellationTokenSource();
-                var call = client.FullDuplexCall(cts.Token);
-
-                await call.RequestStream.Write(StreamingOutputCallRequest.CreateBuilder()
-                    .SetResponseType(PayloadType.COMPRESSABLE)
-                    .AddResponseParameters(ResponseParameters.CreateBuilder().SetSize(31415))
-                    .SetPayload(CreateZerosPayload(27182)).Build());
+                using (var call = client.FullDuplexCall(cts.Token))
+                {
+                    await call.RequestStream.WriteAsync(StreamingOutputCallRequest.CreateBuilder()
+                        .SetResponseType(PayloadType.COMPRESSABLE)
+                        .AddResponseParameters(ResponseParameters.CreateBuilder().SetSize(31415))
+                        .SetPayload(CreateZerosPayload(27182)).Build());
 
-                Assert.IsTrue(await call.ResponseStream.MoveNext());
-                Assert.AreEqual(PayloadType.COMPRESSABLE, call.ResponseStream.Current.Payload.Type);
-                Assert.AreEqual(31415, call.ResponseStream.Current.Payload.Body.Length);
+                    Assert.IsTrue(await call.ResponseStream.MoveNext());
+                    Assert.AreEqual(PayloadType.COMPRESSABLE, call.ResponseStream.Current.Payload.Type);
+                    Assert.AreEqual(31415, call.ResponseStream.Current.Payload.Body.Length);
 
-                cts.Cancel();
+                    cts.Cancel();
 
-                try
-                {
-                    await call.ResponseStream.MoveNext();
-                    Assert.Fail();
-                }
-                catch (RpcException e)
-                {
-                    Assert.AreEqual(StatusCode.Cancelled, e.Status.StatusCode);
+                    try
+                    {
+                        await call.ResponseStream.MoveNext();
+                        Assert.Fail();
+                    }
+                    catch (RpcException e)
+                    {
+                        Assert.AreEqual(StatusCode.Cancelled, e.Status.StatusCode);
+                    }
                 }
                 Console.WriteLine("Passed!");
             }).Wait();
diff --git a/src/csharp/Grpc.IntegrationTesting/TestServiceImpl.cs b/src/csharp/Grpc.IntegrationTesting/TestServiceImpl.cs
index d6ba61ef82..6bd997d1f4 100644
--- a/src/csharp/Grpc.IntegrationTesting/TestServiceImpl.cs
+++ b/src/csharp/Grpc.IntegrationTesting/TestServiceImpl.cs
@@ -64,7 +64,7 @@ namespace grpc.testing
             {
                 var response = StreamingOutputCallResponse.CreateBuilder()
                     .SetPayload(CreateZerosPayload(responseParam.Size)).Build();
-                await responseStream.Write(response);
+                await responseStream.WriteAsync(response);
             }
         }
 
@@ -86,7 +86,7 @@ namespace grpc.testing
                 {
                     var response = StreamingOutputCallResponse.CreateBuilder()
                         .SetPayload(CreateZerosPayload(responseParam.Size)).Build();
-                    await responseStream.Write(response);
+                    await responseStream.WriteAsync(response);
                 }
             });
         }