Skip to content
Snippets Groups Projects
Commit a96cdee8 authored by Tim Emiola's avatar Tim Emiola
Browse files

Merge pull request #1648 from jtattermusch/csharp_disposable_calls

Incorporate feedback from API review: Disposable calls
parents 2c523e5f 2d2652d6
No related branches found
No related tags found
No related merge requests found
Showing
with 210 additions and 144 deletions
......@@ -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();
}
}
}
......@@ -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();
}
}
}
......@@ -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();
}
}
}
......@@ -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)
......
......@@ -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.
}
}
......@@ -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);
}
}
......@@ -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();
}
}
......@@ -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.
}
}
......@@ -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);
......
......@@ -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;
......
......@@ -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);
......
......@@ -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);
}
}
}
......
......@@ -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();
}
}
......
......@@ -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);
......
......@@ -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));
});
}
......
......@@ -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();
......
......@@ -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);
}
});
}
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment