diff options
Diffstat (limited to 'src')
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); } }); } |