From a5272b6adc5fb7e8c71b7216b0f5e690980a79b2 Mon Sep 17 00:00:00 2001 From: Jan Tattermusch Date: Thu, 30 Apr 2015 11:56:46 -0700 Subject: A new version C# API based on async/await --- src/csharp/.gitignore | 1 + src/csharp/Grpc.Core.Tests/ClientServerTest.cs | 84 ++++++++------ src/csharp/Grpc.Core/AsyncClientStreamingCall.cs | 101 ++++++++++++++++ src/csharp/Grpc.Core/AsyncDuplexStreamingCall.cs | 101 ++++++++++++++++ src/csharp/Grpc.Core/AsyncServerStreamingCall.cs | 72 ++++++++++++ src/csharp/Grpc.Core/Call.cs | 3 + src/csharp/Grpc.Core/Calls.cs | 32 +++--- src/csharp/Grpc.Core/Channel.cs | 16 +-- src/csharp/Grpc.Core/ClientStreamingAsyncResult.cs | 69 ----------- src/csharp/Grpc.Core/Credentials.cs | 2 +- src/csharp/Grpc.Core/Grpc.Core.csproj | 20 +++- src/csharp/Grpc.Core/IAsyncStreamReader.cs | 54 +++++++++ src/csharp/Grpc.Core/IAsyncStreamWriter.cs | 54 +++++++++ src/csharp/Grpc.Core/IClientStreamWriter.cs | 53 +++++++++ src/csharp/Grpc.Core/IServerStreamWriter.cs | 48 ++++++++ src/csharp/Grpc.Core/Internal/AsyncCall.cs | 44 +++---- src/csharp/Grpc.Core/Internal/AsyncCallBase.cs | 116 ++++++++----------- src/csharp/Grpc.Core/Internal/AsyncCallServer.cs | 22 ++-- src/csharp/Grpc.Core/Internal/AsyncCompletion.cs | 18 +-- .../Grpc.Core/Internal/ClientRequestStream.cs | 63 ++++++++++ .../Grpc.Core/Internal/ClientResponseStream.cs | 56 +++++++++ .../Internal/ClientStreamingInputObserver.cs | 66 ----------- src/csharp/Grpc.Core/Internal/ServerCallHandler.cs | 128 +++++++++++++++------ src/csharp/Grpc.Core/Internal/ServerCalls.cs | 63 ++++++++++ .../Grpc.Core/Internal/ServerRequestStream.cs | 56 +++++++++ .../Grpc.Core/Internal/ServerResponseStream.cs | 64 +++++++++++ .../Internal/ServerStreamingOutputObserver.cs | 71 ------------ src/csharp/Grpc.Core/Method.cs | 11 +- src/csharp/Grpc.Core/Server.cs | 6 +- src/csharp/Grpc.Core/ServerCalls.cs | 57 --------- src/csharp/Grpc.Core/ServerMethods.cs | 61 ++++++++++ src/csharp/Grpc.Core/ServerServiceDefinition.cs | 24 +++- src/csharp/Grpc.Core/Status.cs | 5 + .../Grpc.Core/Utils/AsyncStreamExtensions.cs | 111 ++++++++++++++++++ src/csharp/Grpc.Core/Utils/RecordingObserver.cs | 65 ----------- src/csharp/Grpc.Core/Utils/RecordingQueue.cs | 83 ------------- .../Grpc.Examples.Tests/MathClientServerTests.cs | 68 ++++++----- src/csharp/Grpc.Examples/MathExamples.cs | 26 ++--- src/csharp/Grpc.Examples/MathGrpc.cs | 24 ++-- src/csharp/Grpc.Examples/MathServiceImpl.cs | 67 +++-------- .../Grpc.IntegrationTesting/InteropClient.cs | 119 +++++++++---------- .../InteropClientServerTest.cs | 2 +- .../Grpc.IntegrationTesting/TestServiceGrpc.cs | 34 +++--- .../Grpc.IntegrationTesting/TestServiceImpl.cs | 77 ++++--------- 44 files changed, 1449 insertions(+), 868 deletions(-) create mode 100644 src/csharp/Grpc.Core/AsyncClientStreamingCall.cs create mode 100644 src/csharp/Grpc.Core/AsyncDuplexStreamingCall.cs create mode 100644 src/csharp/Grpc.Core/AsyncServerStreamingCall.cs delete mode 100644 src/csharp/Grpc.Core/ClientStreamingAsyncResult.cs create mode 100644 src/csharp/Grpc.Core/IAsyncStreamReader.cs create mode 100644 src/csharp/Grpc.Core/IAsyncStreamWriter.cs create mode 100644 src/csharp/Grpc.Core/IClientStreamWriter.cs create mode 100644 src/csharp/Grpc.Core/IServerStreamWriter.cs create mode 100644 src/csharp/Grpc.Core/Internal/ClientRequestStream.cs create mode 100644 src/csharp/Grpc.Core/Internal/ClientResponseStream.cs delete mode 100644 src/csharp/Grpc.Core/Internal/ClientStreamingInputObserver.cs create mode 100644 src/csharp/Grpc.Core/Internal/ServerCalls.cs create mode 100644 src/csharp/Grpc.Core/Internal/ServerRequestStream.cs create mode 100644 src/csharp/Grpc.Core/Internal/ServerResponseStream.cs delete mode 100644 src/csharp/Grpc.Core/Internal/ServerStreamingOutputObserver.cs delete mode 100644 src/csharp/Grpc.Core/ServerCalls.cs create mode 100644 src/csharp/Grpc.Core/ServerMethods.cs create mode 100644 src/csharp/Grpc.Core/Utils/AsyncStreamExtensions.cs delete mode 100644 src/csharp/Grpc.Core/Utils/RecordingObserver.cs delete mode 100644 src/csharp/Grpc.Core/Utils/RecordingQueue.cs diff --git a/src/csharp/.gitignore b/src/csharp/.gitignore index dbaf60de0c..c064ff1fa6 100644 --- a/src/csharp/.gitignore +++ b/src/csharp/.gitignore @@ -1,4 +1,5 @@ *.userprefs +*.csproj.user StyleCop.Cache test-results packages diff --git a/src/csharp/Grpc.Core.Tests/ClientServerTest.cs b/src/csharp/Grpc.Core.Tests/ClientServerTest.cs index 3da9e33e53..9e510e82a6 100644 --- a/src/csharp/Grpc.Core.Tests/ClientServerTest.cs +++ b/src/csharp/Grpc.Core.Tests/ClientServerTest.cs @@ -44,16 +44,18 @@ namespace Grpc.Core.Tests { public class ClientServerTest { - string host = "localhost"; + const string Host = "localhost"; + const string ServiceName = "/tests.Test"; - string serviceName = "/tests.Test"; - - Method unaryEchoStringMethod = new Method( + static readonly Method UnaryEchoStringMethod = new Method( MethodType.Unary, "/tests.Test/UnaryEchoString", Marshallers.StringMarshaller, Marshallers.StringMarshaller); + static readonly ServerServiceDefinition ServiceDefinition = ServerServiceDefinition.CreateBuilder(ServiceName) + .AddMethod(UnaryEchoStringMethod, HandleUnaryEchoString).Build(); + [TestFixtureSetUp] public void Init() { @@ -69,21 +71,39 @@ namespace Grpc.Core.Tests [Test] public void UnaryCall() { - Server server = new Server(); - server.AddServiceDefinition( - ServerServiceDefinition.CreateBuilder(serviceName) - .AddMethod(unaryEchoStringMethod, HandleUnaryEchoString).Build()); - - int port = server.AddListeningPort(host + ":0"); + var server = new Server(); + server.AddServiceDefinition(ServiceDefinition); + int port = server.AddListeningPort(Host + ":0"); server.Start(); - using (Channel channel = new Channel(host + ":" + port)) + using (Channel channel = new Channel(Host + ":" + port)) { - var call = new Call(serviceName, unaryEchoStringMethod, channel, Metadata.Empty); - + var call = new Call(ServiceName, UnaryEchoStringMethod, channel, Metadata.Empty); Assert.AreEqual("ABC", Calls.BlockingUnaryCall(call, "ABC", default(CancellationToken))); + } - Assert.AreEqual("abcdef", Calls.BlockingUnaryCall(call, "abcdef", default(CancellationToken))); + server.ShutdownAsync().Wait(); + } + + [Test] + public void CallOnDisposedChannel() + { + var server = new Server(); + server.AddServiceDefinition(ServiceDefinition); + int port = server.AddListeningPort(Host + ":0"); + server.Start(); + + Channel channel = new Channel(Host + ":" + port); + channel.Dispose(); + + var call = new Call(ServiceName, UnaryEchoStringMethod, channel, Metadata.Empty); + try + { + Calls.BlockingUnaryCall(call, "ABC", default(CancellationToken)); + Assert.Fail(); + } + catch (ObjectDisposedException e) + { } server.ShutdownAsync().Wait(); @@ -92,18 +112,15 @@ namespace Grpc.Core.Tests [Test] public void UnaryCallPerformance() { - Server server = new Server(); - server.AddServiceDefinition( - ServerServiceDefinition.CreateBuilder(serviceName) - .AddMethod(unaryEchoStringMethod, HandleUnaryEchoString).Build()); - - int port = server.AddListeningPort(host + ":0"); + var server = new Server(); + server.AddServiceDefinition(ServiceDefinition); + int port = server.AddListeningPort(Host + ":0"); server.Start(); - using (Channel channel = new Channel(host + ":" + port)) + using (Channel channel = new Channel(Host + ":" + port)) { - var call = new Call(serviceName, unaryEchoStringMethod, channel, Metadata.Empty); - BenchmarkUtil.RunBenchmark(100, 1000, + var call = new Call(ServiceName, UnaryEchoStringMethod, channel, Metadata.Empty); + BenchmarkUtil.RunBenchmark(100, 100, () => { Calls.BlockingUnaryCall(call, "ABC", default(CancellationToken)); }); } @@ -113,17 +130,14 @@ namespace Grpc.Core.Tests [Test] public void UnknownMethodHandler() { - Server server = new Server(); - server.AddServiceDefinition( - ServerServiceDefinition.CreateBuilder(serviceName).Build()); - - int port = server.AddListeningPort(host + ":0"); + var server = new Server(); + server.AddServiceDefinition(ServerServiceDefinition.CreateBuilder(ServiceName).Build()); + int port = server.AddListeningPort(Host + ":0"); server.Start(); - using (Channel channel = new Channel(host + ":" + port)) + using (Channel channel = new Channel(Host + ":" + port)) { - var call = new Call(serviceName, unaryEchoStringMethod, channel, Metadata.Empty); - + var call = new Call(ServiceName, UnaryEchoStringMethod, channel, Metadata.Empty); try { Calls.BlockingUnaryCall(call, "ABC", default(CancellationToken)); @@ -138,10 +152,12 @@ namespace Grpc.Core.Tests server.ShutdownAsync().Wait(); } - private void HandleUnaryEchoString(string request, IObserver responseObserver) + /// + /// Handler for unaryEchoString method. + /// + private static Task HandleUnaryEchoString(string request) { - responseObserver.OnNext(request); - responseObserver.OnCompleted(); + return Task.FromResult(request); } } } diff --git a/src/csharp/Grpc.Core/AsyncClientStreamingCall.cs b/src/csharp/Grpc.Core/AsyncClientStreamingCall.cs new file mode 100644 index 0000000000..e81ce01ebb --- /dev/null +++ b/src/csharp/Grpc.Core/AsyncClientStreamingCall.cs @@ -0,0 +1,101 @@ +#region Copyright notice and license + +// Copyright 2015, Google Inc. +// All rights reserved. +// +// Redistribution and use in source and binary forms, with or without +// modification, are permitted provided that the following conditions are +// met: +// +// * Redistributions of source code must retain the above copyright +// notice, this list of conditions and the following disclaimer. +// * Redistributions in binary form must reproduce the above +// copyright notice, this list of conditions and the following disclaimer +// in the documentation and/or other materials provided with the +// distribution. +// * Neither the name of Google Inc. nor the names of its +// contributors may be used to endorse or promote products derived from +// this software without specific prior written permission. +// +// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS +// "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT +// LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR +// A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT +// OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, +// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT +// LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, +// DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY +// THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT +// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE +// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + +#endregion + +using System; +using System.Runtime.CompilerServices; +using System.Threading.Tasks; + +namespace Grpc.Core +{ + /// + /// Return type for client streaming calls. + /// + public struct AsyncClientStreamingCall + { + readonly IClientStreamWriter requestStream; + readonly Task result; + + public AsyncClientStreamingCall(IClientStreamWriter requestStream, Task result) + { + this.requestStream = requestStream; + this.result = result; + } + + /// + /// Writes a request to RequestStream. + /// + public Task Write(TRequest message) + { + return requestStream.Write(message); + } + + /// + /// Closes the RequestStream. + /// + public Task Close() + { + return requestStream.Close(); + } + + /// + /// Asynchronous call result. + /// + public Task Result + { + get + { + return this.result; + } + } + + /// + /// Async stream to send streaming requests. + /// + public IClientStreamWriter RequestStream + { + get + { + return requestStream; + } + } + + /// + /// Allows awaiting this object directly. + /// + /// + public TaskAwaiter GetAwaiter() + { + return result.GetAwaiter(); + } + } +} diff --git a/src/csharp/Grpc.Core/AsyncDuplexStreamingCall.cs b/src/csharp/Grpc.Core/AsyncDuplexStreamingCall.cs new file mode 100644 index 0000000000..1cb30f4779 --- /dev/null +++ b/src/csharp/Grpc.Core/AsyncDuplexStreamingCall.cs @@ -0,0 +1,101 @@ +#region Copyright notice and license + +// Copyright 2015, Google Inc. +// All rights reserved. +// +// Redistribution and use in source and binary forms, with or without +// modification, are permitted provided that the following conditions are +// met: +// +// * Redistributions of source code must retain the above copyright +// notice, this list of conditions and the following disclaimer. +// * Redistributions in binary form must reproduce the above +// copyright notice, this list of conditions and the following disclaimer +// in the documentation and/or other materials provided with the +// distribution. +// * Neither the name of Google Inc. nor the names of its +// contributors may be used to endorse or promote products derived from +// this software without specific prior written permission. +// +// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS +// "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT +// LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR +// A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT +// OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, +// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT +// LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, +// DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY +// THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT +// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE +// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + +#endregion + +using System; +using System.Runtime.CompilerServices; +using System.Threading.Tasks; + +namespace Grpc.Core +{ + /// + /// Return type for bidirectional streaming calls. + /// + public struct AsyncDuplexStreamingCall + { + readonly IClientStreamWriter requestStream; + readonly IAsyncStreamReader responseStream; + + public AsyncDuplexStreamingCall(IClientStreamWriter requestStream, IAsyncStreamReader responseStream) + { + this.requestStream = requestStream; + this.responseStream = responseStream; + } + + /// + /// Writes a request to RequestStream. + /// + public Task Write(TRequest message) + { + return requestStream.Write(message); + } + + /// + /// Closes the RequestStream. + /// + public Task Close() + { + return requestStream.Close(); + } + + /// + /// Reads a response from ResponseStream. + /// + /// + public Task ReadNext() + { + return responseStream.ReadNext(); + } + + /// + /// Async stream to read streaming responses. + /// + public IAsyncStreamReader ResponseStream + { + get + { + return responseStream; + } + } + + /// + /// Async stream to send streaming requests. + /// + public IClientStreamWriter RequestStream + { + get + { + return requestStream; + } + } + } +} diff --git a/src/csharp/Grpc.Core/AsyncServerStreamingCall.cs b/src/csharp/Grpc.Core/AsyncServerStreamingCall.cs new file mode 100644 index 0000000000..d614916fb7 --- /dev/null +++ b/src/csharp/Grpc.Core/AsyncServerStreamingCall.cs @@ -0,0 +1,72 @@ +#region Copyright notice and license + +// Copyright 2015, Google Inc. +// All rights reserved. +// +// Redistribution and use in source and binary forms, with or without +// modification, are permitted provided that the following conditions are +// met: +// +// * Redistributions of source code must retain the above copyright +// notice, this list of conditions and the following disclaimer. +// * Redistributions in binary form must reproduce the above +// copyright notice, this list of conditions and the following disclaimer +// in the documentation and/or other materials provided with the +// distribution. +// * Neither the name of Google Inc. nor the names of its +// contributors may be used to endorse or promote products derived from +// this software without specific prior written permission. +// +// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS +// "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT +// LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR +// A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT +// OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, +// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT +// LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, +// DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY +// THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT +// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE +// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + +#endregion + +using System; +using System.Runtime.CompilerServices; +using System.Threading.Tasks; + +namespace Grpc.Core +{ + /// + /// Return type for server streaming calls. + /// + public struct AsyncServerStreamingCall + { + readonly IAsyncStreamReader responseStream; + + public AsyncServerStreamingCall(IAsyncStreamReader responseStream) + { + this.responseStream = responseStream; + } + + /// + /// Reads the next response from ResponseStream + /// + /// + public Task ReadNext() + { + return responseStream.ReadNext(); + } + + /// + /// Async stream to read streaming responses. + /// + public IAsyncStreamReader ResponseStream + { + get + { + return responseStream; + } + } + } +} diff --git a/src/csharp/Grpc.Core/Call.cs b/src/csharp/Grpc.Core/Call.cs index fe5f40f5e9..070dfb569d 100644 --- a/src/csharp/Grpc.Core/Call.cs +++ b/src/csharp/Grpc.Core/Call.cs @@ -37,6 +37,9 @@ using Grpc.Core.Utils; namespace Grpc.Core { + /// + /// Abstraction of a call to be invoked on a client. + /// public class Call { readonly string name; diff --git a/src/csharp/Grpc.Core/Calls.cs b/src/csharp/Grpc.Core/Calls.cs index 280387b323..9365ccd9fb 100644 --- a/src/csharp/Grpc.Core/Calls.cs +++ b/src/csharp/Grpc.Core/Calls.cs @@ -39,7 +39,7 @@ using Grpc.Core.Internal; namespace Grpc.Core { /// - /// Helper methods for generated stubs to make RPC calls. + /// Helper methods for generated client stubs to make RPC calls. /// public static class Calls { @@ -56,35 +56,37 @@ namespace Grpc.Core return await asyncCall.UnaryCallAsync(req, call.Headers); } - public static void AsyncServerStreamingCall(Call call, TRequest req, IObserver outputs, CancellationToken token) + public static AsyncServerStreamingCall AsyncServerStreamingCall(Call call, TRequest req, CancellationToken token) { var asyncCall = new AsyncCall(call.RequestMarshaller.Serializer, call.ResponseMarshaller.Deserializer); asyncCall.Initialize(call.Channel, GetCompletionQueue(), call.Name); - asyncCall.StartServerStreamingCall(req, outputs, call.Headers); + asyncCall.StartServerStreamingCall(req, call.Headers); + var responseStream = new ClientResponseStream(asyncCall); + return new AsyncServerStreamingCall(responseStream); } - public static ClientStreamingAsyncResult AsyncClientStreamingCall(Call call, CancellationToken token) + public static AsyncClientStreamingCall AsyncClientStreamingCall(Call call, CancellationToken token) { var asyncCall = new AsyncCall(call.RequestMarshaller.Serializer, call.ResponseMarshaller.Deserializer); asyncCall.Initialize(call.Channel, GetCompletionQueue(), call.Name); - var task = asyncCall.ClientStreamingCallAsync(call.Headers); - var inputs = new ClientStreamingInputObserver(asyncCall); - return new ClientStreamingAsyncResult(task, inputs); + var resultTask = asyncCall.ClientStreamingCallAsync(call.Headers); + var requestStream = new ClientRequestStream(asyncCall); + return new AsyncClientStreamingCall(requestStream, resultTask); } - public static TResponse BlockingClientStreamingCall(Call call, IObservable inputs, CancellationToken token) - { - throw new NotImplementedException(); - } - - public static IObserver DuplexStreamingCall(Call call, IObserver outputs, CancellationToken token) + public static AsyncDuplexStreamingCall AsyncDuplexStreamingCall(Call call, CancellationToken token) { var asyncCall = new AsyncCall(call.RequestMarshaller.Serializer, call.ResponseMarshaller.Deserializer); asyncCall.Initialize(call.Channel, GetCompletionQueue(), call.Name); - asyncCall.StartDuplexStreamingCall(outputs, call.Headers); - return new ClientStreamingInputObserver(asyncCall); + asyncCall.StartDuplexStreamingCall(call.Headers); + var requestStream = new ClientRequestStream(asyncCall); + var responseStream = new ClientResponseStream(asyncCall); + return new AsyncDuplexStreamingCall(requestStream, responseStream); } + /// + /// Gets shared completion queue used for async calls. + /// private static CompletionQueueSafeHandle GetCompletionQueue() { return GrpcEnvironment.ThreadPool.CompletionQueue; diff --git a/src/csharp/Grpc.Core/Channel.cs b/src/csharp/Grpc.Core/Channel.cs index 3a42dac1d7..b47d810672 100644 --- a/src/csharp/Grpc.Core/Channel.cs +++ b/src/csharp/Grpc.Core/Channel.cs @@ -66,14 +66,6 @@ namespace Grpc.Core this.target = GetOverridenTarget(target, channelArgs); } - internal ChannelSafeHandle Handle - { - get - { - return this.handle; - } - } - public string Target { get @@ -88,6 +80,14 @@ namespace Grpc.Core GC.SuppressFinalize(this); } + internal ChannelSafeHandle Handle + { + get + { + return this.handle; + } + } + protected virtual void Dispose(bool disposing) { if (handle != null && !handle.IsInvalid) diff --git a/src/csharp/Grpc.Core/ClientStreamingAsyncResult.cs b/src/csharp/Grpc.Core/ClientStreamingAsyncResult.cs deleted file mode 100644 index 65bedb0a33..0000000000 --- a/src/csharp/Grpc.Core/ClientStreamingAsyncResult.cs +++ /dev/null @@ -1,69 +0,0 @@ -#region Copyright notice and license - -// Copyright 2015, Google Inc. -// All rights reserved. -// -// Redistribution and use in source and binary forms, with or without -// modification, are permitted provided that the following conditions are -// met: -// -// * Redistributions of source code must retain the above copyright -// notice, this list of conditions and the following disclaimer. -// * Redistributions in binary form must reproduce the above -// copyright notice, this list of conditions and the following disclaimer -// in the documentation and/or other materials provided with the -// distribution. -// * Neither the name of Google Inc. nor the names of its -// contributors may be used to endorse or promote products derived from -// this software without specific prior written permission. -// -// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS -// "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT -// LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR -// A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT -// OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, -// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT -// LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, -// DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY -// THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT -// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE -// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. - -#endregion - -using System; -using System.Threading.Tasks; - -namespace Grpc.Core -{ - /// - /// Return type for client streaming async method. - /// - public struct ClientStreamingAsyncResult - { - readonly Task task; - readonly IObserver inputs; - - public ClientStreamingAsyncResult(Task task, IObserver inputs) - { - this.task = task; - this.inputs = inputs; - } - - public Task Task - { - get - { - return this.task; - } - } - - public IObserver Inputs - { - get - { - return this.inputs; - } - } - } -} diff --git a/src/csharp/Grpc.Core/Credentials.cs b/src/csharp/Grpc.Core/Credentials.cs index 15dd3ef321..e64c1e3dc1 100644 --- a/src/csharp/Grpc.Core/Credentials.cs +++ b/src/csharp/Grpc.Core/Credentials.cs @@ -37,7 +37,7 @@ using Grpc.Core.Internal; namespace Grpc.Core { /// - /// Client-side credentials. + /// Client-side credentials. Used for creation of a secure channel. /// public abstract class Credentials { diff --git a/src/csharp/Grpc.Core/Grpc.Core.csproj b/src/csharp/Grpc.Core/Grpc.Core.csproj index 0b85392e15..fee742b220 100644 --- a/src/csharp/Grpc.Core/Grpc.Core.csproj +++ b/src/csharp/Grpc.Core/Grpc.Core.csproj @@ -39,12 +39,18 @@ + + + + + + - + @@ -59,14 +65,10 @@ - - - - - + @@ -86,6 +88,12 @@ + + + + + + diff --git a/src/csharp/Grpc.Core/IAsyncStreamReader.cs b/src/csharp/Grpc.Core/IAsyncStreamReader.cs new file mode 100644 index 0000000000..61cf57f7e0 --- /dev/null +++ b/src/csharp/Grpc.Core/IAsyncStreamReader.cs @@ -0,0 +1,54 @@ +#region Copyright notice and license + +// Copyright 2015, Google Inc. +// All rights reserved. +// +// Redistribution and use in source and binary forms, with or without +// modification, are permitted provided that the following conditions are +// met: +// +// * Redistributions of source code must retain the above copyright +// notice, this list of conditions and the following disclaimer. +// * Redistributions in binary form must reproduce the above +// copyright notice, this list of conditions and the following disclaimer +// in the documentation and/or other materials provided with the +// distribution. +// * Neither the name of Google Inc. nor the names of its +// contributors may be used to endorse or promote products derived from +// this software without specific prior written permission. +// +// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS +// "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT +// LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR +// A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT +// OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, +// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT +// LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, +// DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY +// THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT +// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE +// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + +#endregion + +using System; +using System.Collections.Generic; +using System.Linq; +using System.Text; +using System.Threading.Tasks; + +namespace Grpc.Core +{ + /// + /// A stream of messages to be read. + /// + /// + public interface IAsyncStreamReader + { + /// + /// Reads a single message. Returns default(T) if the last message was already read. + /// A following read can only be started when the previous one finishes. + /// + Task ReadNext(); + } +} diff --git a/src/csharp/Grpc.Core/IAsyncStreamWriter.cs b/src/csharp/Grpc.Core/IAsyncStreamWriter.cs new file mode 100644 index 0000000000..724bae8f31 --- /dev/null +++ b/src/csharp/Grpc.Core/IAsyncStreamWriter.cs @@ -0,0 +1,54 @@ +#region Copyright notice and license + +// Copyright 2015, Google Inc. +// All rights reserved. +// +// Redistribution and use in source and binary forms, with or without +// modification, are permitted provided that the following conditions are +// met: +// +// * Redistributions of source code must retain the above copyright +// notice, this list of conditions and the following disclaimer. +// * Redistributions in binary form must reproduce the above +// copyright notice, this list of conditions and the following disclaimer +// in the documentation and/or other materials provided with the +// distribution. +// * Neither the name of Google Inc. nor the names of its +// contributors may be used to endorse or promote products derived from +// this software without specific prior written permission. +// +// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS +// "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT +// LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR +// A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT +// OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, +// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT +// LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, +// DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY +// THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT +// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE +// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + +#endregion + +using System; +using System.Collections.Generic; +using System.Linq; +using System.Text; +using System.Threading.Tasks; + +namespace Grpc.Core +{ + /// + /// A writable stream of messages. + /// + /// + public interface IAsyncStreamWriter + { + /// + /// Writes a single message. Only one write can be pending at a time. + /// + /// the message to be written. Cannot be null. + Task Write(T message); + } +} diff --git a/src/csharp/Grpc.Core/IClientStreamWriter.cs b/src/csharp/Grpc.Core/IClientStreamWriter.cs new file mode 100644 index 0000000000..6da42e9ccc --- /dev/null +++ b/src/csharp/Grpc.Core/IClientStreamWriter.cs @@ -0,0 +1,53 @@ +#region Copyright notice and license + +// Copyright 2015, Google Inc. +// All rights reserved. +// +// Redistribution and use in source and binary forms, with or without +// modification, are permitted provided that the following conditions are +// met: +// +// * Redistributions of source code must retain the above copyright +// notice, this list of conditions and the following disclaimer. +// * Redistributions in binary form must reproduce the above +// copyright notice, this list of conditions and the following disclaimer +// in the documentation and/or other materials provided with the +// distribution. +// * Neither the name of Google Inc. nor the names of its +// contributors may be used to endorse or promote products derived from +// this software without specific prior written permission. +// +// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS +// "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT +// LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR +// A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT +// OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, +// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT +// LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, +// DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY +// THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT +// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE +// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + +#endregion + +using System; +using System.Collections.Generic; +using System.Linq; +using System.Text; +using System.Threading.Tasks; + +namespace Grpc.Core +{ + /// + /// Client-side writable stream of messages with Close capability. + /// + /// + public interface IClientStreamWriter : IAsyncStreamWriter + { + /// + /// Closes the stream. Can only be called once there is no pending write. No writes should follow calling this. + /// + Task Close(); + } +} diff --git a/src/csharp/Grpc.Core/IServerStreamWriter.cs b/src/csharp/Grpc.Core/IServerStreamWriter.cs new file mode 100644 index 0000000000..e76397d8a0 --- /dev/null +++ b/src/csharp/Grpc.Core/IServerStreamWriter.cs @@ -0,0 +1,48 @@ +#region Copyright notice and license + +// Copyright 2015, Google Inc. +// All rights reserved. +// +// Redistribution and use in source and binary forms, with or without +// modification, are permitted provided that the following conditions are +// met: +// +// * Redistributions of source code must retain the above copyright +// notice, this list of conditions and the following disclaimer. +// * Redistributions in binary form must reproduce the above +// copyright notice, this list of conditions and the following disclaimer +// in the documentation and/or other materials provided with the +// distribution. +// * Neither the name of Google Inc. nor the names of its +// contributors may be used to endorse or promote products derived from +// this software without specific prior written permission. +// +// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS +// "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT +// LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR +// A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT +// OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, +// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT +// LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, +// DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY +// THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT +// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE +// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + +#endregion + +using System; +using System.Collections.Generic; +using System.Linq; +using System.Text; +using System.Threading.Tasks; + +namespace Grpc.Core +{ + /// + /// A writable stream of messages that is used in server-side handlers. + /// + public interface IServerStreamWriter : IAsyncStreamWriter + { + } +} diff --git a/src/csharp/Grpc.Core/Internal/AsyncCall.cs b/src/csharp/Grpc.Core/Internal/AsyncCall.cs index bc72cb78de..fd94771ddd 100644 --- a/src/csharp/Grpc.Core/Internal/AsyncCall.cs +++ b/src/csharp/Grpc.Core/Internal/AsyncCall.cs @@ -43,7 +43,7 @@ using Grpc.Core.Utils; namespace Grpc.Core.Internal { /// - /// Handles client side native call lifecycle. + /// Manages client side native call lifecycle. /// internal class AsyncCall : AsyncCallBase { @@ -160,7 +160,7 @@ namespace Grpc.Core.Internal /// /// Starts a unary request - streamed response call. /// - public void StartServerStreamingCall(TRequest msg, IObserver readObserver, Metadata headers) + public void StartServerStreamingCall(TRequest msg, Metadata headers) { lock (myLock) { @@ -169,17 +169,13 @@ namespace Grpc.Core.Internal started = true; halfcloseRequested = true; halfclosed = true; // halfclose not confirmed yet, but it will be once finishedHandler is called. - - this.readObserver = readObserver; byte[] payload = UnsafeSerialize(msg); - + using (var metadataArray = MetadataArraySafeHandle.Create(headers)) { call.StartServerStreaming(payload, finishedHandler, metadataArray); } - - StartReceiveMessage(); } } @@ -187,7 +183,7 @@ namespace Grpc.Core.Internal /// Starts a streaming request - streaming response call. /// Use StartSendMessage and StartSendCloseFromClient to stream requests. /// - public void StartDuplexStreamingCall(IObserver readObserver, Metadata headers) + public void StartDuplexStreamingCall(Metadata headers) { lock (myLock) { @@ -195,14 +191,10 @@ namespace Grpc.Core.Internal started = true; - this.readObserver = readObserver; - using (var metadataArray = MetadataArraySafeHandle.Create(headers)) { call.StartDuplexStreaming(finishedHandler, metadataArray); } - - StartReceiveMessage(); } } @@ -210,17 +202,26 @@ namespace Grpc.Core.Internal /// Sends a streaming request. Only one pending send action is allowed at any given time. /// completionDelegate is called when the operation finishes. /// - public void StartSendMessage(TRequest msg, AsyncCompletionDelegate completionDelegate) + public void StartSendMessage(TRequest msg, AsyncCompletionDelegate completionDelegate) { StartSendMessageInternal(msg, completionDelegate); } + /// + /// Receives a streaming response. Only one pending read action is allowed at any given time. + /// completionDelegate is called when the operation finishes. + /// + public void StartReadMessage(AsyncCompletionDelegate completionDelegate) + { + StartReadMessageInternal(completionDelegate); + } + /// /// Sends halfclose, indicating client is done with streaming requests. /// Only one pending send action is allowed at any given time. /// completionDelegate is called when the operation finishes. /// - public void StartSendCloseFromClient(AsyncCompletionDelegate completionDelegate) + public void StartSendCloseFromClient(AsyncCompletionDelegate completionDelegate) { lock (myLock) { @@ -235,12 +236,12 @@ namespace Grpc.Core.Internal } /// - /// On client-side, we only fire readObserver.OnCompleted once all messages have been read + /// On client-side, we only fire readCompletionDelegate once all messages have been read /// and status has been received. /// - protected override void CompleteReadObserver() + protected override void ProcessLastRead(AsyncCompletionDelegate completionDelegate) { - if (readingDone && finishedStatus.HasValue) + if (completionDelegate != null && readingDone && finishedStatus.HasValue) { bool shouldComplete; lock (myLock) @@ -254,11 +255,11 @@ namespace Grpc.Core.Internal var status = finishedStatus.Value; if (status.StatusCode != StatusCode.OK) { - FireReadObserverOnError(new RpcException(status)); + FireCompletion(completionDelegate, default(TResponse), new RpcException(status)); } else { - FireReadObserverOnCompleted(); + FireCompletion(completionDelegate, default(TResponse), null); } } } @@ -304,15 +305,18 @@ namespace Grpc.Core.Internal { var status = ctx.GetReceivedStatus(); + AsyncCompletionDelegate origReadCompletionDelegate = null; lock (myLock) { finished = true; finishedStatus = status; + origReadCompletionDelegate = readCompletionDelegate; + ReleaseResourcesIfPossible(); } - CompleteReadObserver(); + ProcessLastRead(origReadCompletionDelegate); } } } \ No newline at end of file diff --git a/src/csharp/Grpc.Core/Internal/AsyncCallBase.cs b/src/csharp/Grpc.Core/Internal/AsyncCallBase.cs index 15b0cfe249..2bde4b3720 100644 --- a/src/csharp/Grpc.Core/Internal/AsyncCallBase.cs +++ b/src/csharp/Grpc.Core/Internal/AsyncCallBase.cs @@ -44,7 +44,7 @@ namespace Grpc.Core.Internal { /// /// Base for handling both client side and server side calls. - /// Handles native call lifecycle and provides convenience methods. + /// Manages native call lifecycle and provides convenience methods. /// internal abstract class AsyncCallBase { @@ -65,16 +65,14 @@ namespace Grpc.Core.Internal protected bool errorOccured; protected bool cancelRequested; - protected AsyncCompletionDelegate sendCompletionDelegate; // Completion of a pending send or sendclose if not null. - protected bool readPending; // True if there is a read in progress. + protected AsyncCompletionDelegate sendCompletionDelegate; // Completion of a pending send or sendclose if not null. + protected AsyncCompletionDelegate readCompletionDelegate; // Completion of a pending send or sendclose if not null. + protected bool readingDone; protected bool halfcloseRequested; protected bool halfclosed; protected bool finished; // True if close has been received from the peer. - // Streaming reads will be delivered to this observer. For a call that only does unary read it may remain null. - protected IObserver readObserver; - public AsyncCallBase(Func serializer, Func deserializer) { this.serializer = Preconditions.CheckNotNull(serializer); @@ -131,10 +129,10 @@ namespace Grpc.Core.Internal } /// - /// Initiates sending a message. Only once send operation can be active at a time. + /// Initiates sending a message. Only one send operation can be active at a time. /// completionDelegate is invoked upon completion. /// - protected void StartSendMessageInternal(TWrite msg, AsyncCompletionDelegate completionDelegate) + protected void StartSendMessageInternal(TWrite msg, AsyncCompletionDelegate completionDelegate) { byte[] payload = UnsafeSerialize(msg); @@ -149,31 +147,29 @@ namespace Grpc.Core.Internal } /// - /// Requests receiving a next message. + /// Initiates reading a message. Only one read operation can be active at a time. + /// completionDelegate is invoked upon completion. /// - protected void StartReceiveMessage() + protected void StartReadMessageInternal(AsyncCompletionDelegate completionDelegate) { lock (myLock) { - Preconditions.CheckState(started); - Preconditions.CheckState(!disposed); - Preconditions.CheckState(!errorOccured); - - Preconditions.CheckState(!readingDone); - Preconditions.CheckState(!readPending); + Preconditions.CheckNotNull(completionDelegate, "Completion delegate cannot be null"); + CheckReadingAllowed(); call.StartReceiveMessage(readFinishedHandler); - readPending = true; + readCompletionDelegate = completionDelegate; } } + // TODO(jtattermusch): find more fitting name for this method. /// /// Default behavior just completes the read observer, but more sofisticated behavior might be required /// by subclasses. /// - protected virtual void CompleteReadObserver() + protected virtual void ProcessLastRead(AsyncCompletionDelegate completionDelegate) { - FireReadObserverOnCompleted(); + FireCompletion(completionDelegate, default(TRead), null); } /// @@ -213,6 +209,16 @@ namespace Grpc.Core.Internal Preconditions.CheckState(sendCompletionDelegate == null, "Only one write can be pending at a time"); } + protected void CheckReadingAllowed() + { + Preconditions.CheckState(started); + Preconditions.CheckState(!disposed); + Preconditions.CheckState(!errorOccured); + + Preconditions.CheckState(!readingDone, "Stream has already been closed."); + Preconditions.CheckState(readCompletionDelegate == null, "Only one write can be pending at a time"); + } + protected byte[] UnsafeSerialize(TWrite msg) { return serializer(msg); @@ -248,47 +254,11 @@ namespace Grpc.Core.Internal } } - protected void FireReadObserverOnNext(TRead value) - { - try - { - readObserver.OnNext(value); - } - catch (Exception e) - { - Console.WriteLine("Exception occured while invoking readObserver.OnNext: " + e); - } - } - - protected void FireReadObserverOnCompleted() + protected void FireCompletion(AsyncCompletionDelegate completionDelegate, T value, Exception error) { try { - readObserver.OnCompleted(); - } - catch (Exception e) - { - Console.WriteLine("Exception occured while invoking readObserver.OnCompleted: " + e); - } - } - - protected void FireReadObserverOnError(Exception error) - { - try - { - readObserver.OnError(error); - } - catch (Exception e) - { - Console.WriteLine("Exception occured while invoking readObserver.OnError: " + e); - } - } - - protected void FireCompletion(AsyncCompletionDelegate completionDelegate, Exception error) - { - try - { - completionDelegate(error); + completionDelegate(value, error); } catch (Exception e) { @@ -322,7 +292,7 @@ namespace Grpc.Core.Internal /// private void HandleSendFinished(bool wasError, BatchContextSafeHandleNotOwned ctx) { - AsyncCompletionDelegate origCompletionDelegate = null; + AsyncCompletionDelegate origCompletionDelegate = null; lock (myLock) { origCompletionDelegate = sendCompletionDelegate; @@ -333,11 +303,11 @@ namespace Grpc.Core.Internal if (wasError) { - FireCompletion(origCompletionDelegate, new OperationFailedException("Send failed")); + FireCompletion(origCompletionDelegate, null, new OperationFailedException("Send failed")); } else { - FireCompletion(origCompletionDelegate, null); + FireCompletion(origCompletionDelegate, null, null); } } @@ -346,7 +316,7 @@ namespace Grpc.Core.Internal /// private void HandleHalfclosed(bool wasError, BatchContextSafeHandleNotOwned ctx) { - AsyncCompletionDelegate origCompletionDelegate = null; + AsyncCompletionDelegate origCompletionDelegate = null; lock (myLock) { halfclosed = true; @@ -358,11 +328,11 @@ namespace Grpc.Core.Internal if (wasError) { - FireCompletion(origCompletionDelegate, new OperationFailedException("Halfclose failed")); + FireCompletion(origCompletionDelegate, null, new OperationFailedException("Halfclose failed")); } else { - FireCompletion(origCompletionDelegate, null); + FireCompletion(origCompletionDelegate, null, null); } } @@ -373,11 +343,19 @@ namespace Grpc.Core.Internal { var payload = ctx.GetReceivedMessage(); + AsyncCompletionDelegate origCompletionDelegate = null; lock (myLock) { - readPending = false; - if (payload == null) + origCompletionDelegate = readCompletionDelegate; + if (payload != null) { + readCompletionDelegate = null; + } + else + { + // This was the last read. Keeping the readCompletionDelegate + // to be either fired by this handler or by client-side finished + // handler. readingDone = true; } @@ -392,15 +370,11 @@ namespace Grpc.Core.Internal TRead msg; TryDeserialize(payload, out msg); - FireReadObserverOnNext(msg); - - // Start a new read. The current one has already been delivered, - // so correct ordering of reads is assured. - StartReceiveMessage(); + FireCompletion(origCompletionDelegate, msg, null); } else { - CompleteReadObserver(); + ProcessLastRead(origCompletionDelegate); } } } diff --git a/src/csharp/Grpc.Core/Internal/AsyncCallServer.cs b/src/csharp/Grpc.Core/Internal/AsyncCallServer.cs index d3a2be553f..25dc15bbc0 100644 --- a/src/csharp/Grpc.Core/Internal/AsyncCallServer.cs +++ b/src/csharp/Grpc.Core/Internal/AsyncCallServer.cs @@ -43,7 +43,7 @@ using Grpc.Core.Utils; namespace Grpc.Core.Internal { /// - /// Handles server side native call lifecycle. + /// Manages server side native call lifecycle. /// internal class AsyncCallServer : AsyncCallBase { @@ -61,20 +61,17 @@ namespace Grpc.Core.Internal } /// - /// Starts a server side call. Currently, all server side calls are implemented as duplex - /// streaming call and they are adapted to the appropriate streaming arity. + /// Starts a server side call. /// - public Task ServerSideCallAsync(IObserver readObserver) + public Task ServerSideCallAsync() { lock (myLock) { Preconditions.CheckNotNull(call); started = true; - this.readObserver = readObserver; call.StartServerSide(finishedServersideHandler); - StartReceiveMessage(); return finishedServersideTcs.Task; } } @@ -83,17 +80,26 @@ namespace Grpc.Core.Internal /// Sends a streaming response. Only one pending send action is allowed at any given time. /// completionDelegate is called when the operation finishes. /// - public void StartSendMessage(TResponse msg, AsyncCompletionDelegate completionDelegate) + public void StartSendMessage(TResponse msg, AsyncCompletionDelegate completionDelegate) { StartSendMessageInternal(msg, completionDelegate); } + /// + /// Receives a streaming request. Only one pending read action is allowed at any given time. + /// completionDelegate is called when the operation finishes. + /// + public void StartReadMessage(AsyncCompletionDelegate completionDelegate) + { + StartReadMessageInternal(completionDelegate); + } + /// /// Sends call result status, also indicating server is done with streaming responses. /// Only one pending send action is allowed at any given time. /// completionDelegate is called when the operation finishes. /// - public void StartSendStatusFromServer(Status status, AsyncCompletionDelegate completionDelegate) + public void StartSendStatusFromServer(Status status, AsyncCompletionDelegate completionDelegate) { lock (myLock) { diff --git a/src/csharp/Grpc.Core/Internal/AsyncCompletion.cs b/src/csharp/Grpc.Core/Internal/AsyncCompletion.cs index 673b527fb2..c88cae98fe 100644 --- a/src/csharp/Grpc.Core/Internal/AsyncCompletion.cs +++ b/src/csharp/Grpc.Core/Internal/AsyncCompletion.cs @@ -45,22 +45,22 @@ namespace Grpc.Core.Internal /// /// If error != null, there's been an error or operation has been cancelled. /// - internal delegate void AsyncCompletionDelegate(Exception error); + internal delegate void AsyncCompletionDelegate(T result, Exception error); /// /// Helper for transforming AsyncCompletionDelegate into full-fledged Task. /// - internal class AsyncCompletionTaskSource + internal class AsyncCompletionTaskSource { - readonly TaskCompletionSource tcs = new TaskCompletionSource(); - readonly AsyncCompletionDelegate completionDelegate; + readonly TaskCompletionSource tcs = new TaskCompletionSource(); + readonly AsyncCompletionDelegate completionDelegate; public AsyncCompletionTaskSource() { - completionDelegate = new AsyncCompletionDelegate(HandleCompletion); + completionDelegate = new AsyncCompletionDelegate(HandleCompletion); } - public Task Task + public Task Task { get { @@ -68,7 +68,7 @@ namespace Grpc.Core.Internal } } - public AsyncCompletionDelegate CompletionDelegate + public AsyncCompletionDelegate CompletionDelegate { get { @@ -76,11 +76,11 @@ namespace Grpc.Core.Internal } } - private void HandleCompletion(Exception error) + private void HandleCompletion(T value, Exception error) { if (error == null) { - tcs.SetResult(null); + tcs.SetResult(value); return; } if (error is OperationCanceledException) diff --git a/src/csharp/Grpc.Core/Internal/ClientRequestStream.cs b/src/csharp/Grpc.Core/Internal/ClientRequestStream.cs new file mode 100644 index 0000000000..6854922a6f --- /dev/null +++ b/src/csharp/Grpc.Core/Internal/ClientRequestStream.cs @@ -0,0 +1,63 @@ +#region Copyright notice and license +// Copyright 2015, Google Inc. +// All rights reserved. +// +// Redistribution and use in source and binary forms, with or without +// modification, are permitted provided that the following conditions are +// met: +// +// * Redistributions of source code must retain the above copyright +// notice, this list of conditions and the following disclaimer. +// * Redistributions in binary form must reproduce the above +// copyright notice, this list of conditions and the following disclaimer +// in the documentation and/or other materials provided with the +// distribution. +// * Neither the name of Google Inc. nor the names of its +// contributors may be used to endorse or promote products derived from +// this software without specific prior written permission. +// +// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS +// "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT +// LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR +// A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT +// OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, +// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT +// LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, +// DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY +// THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT +// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE +// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. +#endregion +using System; +using System.Threading.Tasks; +using Grpc.Core.Internal; + +namespace Grpc.Core.Internal +{ + /// + /// Writes requests asynchronously to an underlying AsyncCall object. + /// + internal class ClientRequestStream : IClientStreamWriter + { + readonly AsyncCall call; + + public ClientRequestStream(AsyncCall call) + { + this.call = call; + } + + public Task Write(TRequest message) + { + var taskSource = new AsyncCompletionTaskSource(); + call.StartSendMessage(message, taskSource.CompletionDelegate); + return taskSource.Task; + } + + public Task Close() + { + var taskSource = new AsyncCompletionTaskSource(); + call.StartSendCloseFromClient(taskSource.CompletionDelegate); + return taskSource.Task; + } + } +} diff --git a/src/csharp/Grpc.Core/Internal/ClientResponseStream.cs b/src/csharp/Grpc.Core/Internal/ClientResponseStream.cs new file mode 100644 index 0000000000..7fa511faa8 --- /dev/null +++ b/src/csharp/Grpc.Core/Internal/ClientResponseStream.cs @@ -0,0 +1,56 @@ +#region Copyright notice and license + +// Copyright 2015, Google Inc. +// All rights reserved. +// +// Redistribution and use in source and binary forms, with or without +// modification, are permitted provided that the following conditions are +// met: +// +// * Redistributions of source code must retain the above copyright +// notice, this list of conditions and the following disclaimer. +// * Redistributions in binary form must reproduce the above +// copyright notice, this list of conditions and the following disclaimer +// in the documentation and/or other materials provided with the +// distribution. +// * Neither the name of Google Inc. nor the names of its +// contributors may be used to endorse or promote products derived from +// this software without specific prior written permission. +// +// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS +// "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT +// LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR +// A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT +// OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, +// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT +// LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, +// DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY +// THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT +// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE +// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + +#endregion + +using System; +using System.Collections.Generic; +using System.Threading.Tasks; + +namespace Grpc.Core.Internal +{ + internal class ClientResponseStream : IAsyncStreamReader + { + readonly AsyncCall call; + + public ClientResponseStream(AsyncCall call) + { + this.call = call; + } + + public Task ReadNext() + { + var taskSource = new AsyncCompletionTaskSource(); + call.StartReadMessage(taskSource.CompletionDelegate); + return taskSource.Task; + } + } +} diff --git a/src/csharp/Grpc.Core/Internal/ClientStreamingInputObserver.cs b/src/csharp/Grpc.Core/Internal/ClientStreamingInputObserver.cs deleted file mode 100644 index 286c54f2c4..0000000000 --- a/src/csharp/Grpc.Core/Internal/ClientStreamingInputObserver.cs +++ /dev/null @@ -1,66 +0,0 @@ -#region Copyright notice and license -// Copyright 2015, Google Inc. -// All rights reserved. -// -// Redistribution and use in source and binary forms, with or without -// modification, are permitted provided that the following conditions are -// met: -// -// * Redistributions of source code must retain the above copyright -// notice, this list of conditions and the following disclaimer. -// * Redistributions in binary form must reproduce the above -// copyright notice, this list of conditions and the following disclaimer -// in the documentation and/or other materials provided with the -// distribution. -// * Neither the name of Google Inc. nor the names of its -// contributors may be used to endorse or promote products derived from -// this software without specific prior written permission. -// -// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS -// "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT -// LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR -// A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT -// OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, -// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT -// LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, -// DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY -// THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT -// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE -// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. -#endregion -using System; -using Grpc.Core.Internal; - -namespace Grpc.Core.Internal -{ - internal class ClientStreamingInputObserver : IObserver - { - readonly AsyncCall call; - - public ClientStreamingInputObserver(AsyncCall call) - { - this.call = call; - } - - public void OnCompleted() - { - var taskSource = new AsyncCompletionTaskSource(); - call.StartSendCloseFromClient(taskSource.CompletionDelegate); - // TODO: how bad is the Wait here? - taskSource.Task.Wait(); - } - - public void OnError(Exception error) - { - throw new InvalidOperationException("This should never be called."); - } - - public void OnNext(TWrite value) - { - var taskSource = new AsyncCompletionTaskSource(); - call.StartSendMessage(value, taskSource.CompletionDelegate); - // TODO: how bad is the Wait here? - taskSource.Task.Wait(); - } - } -} diff --git a/src/csharp/Grpc.Core/Internal/ServerCallHandler.cs b/src/csharp/Grpc.Core/Internal/ServerCallHandler.cs index 25fd4fab8f..2ae924b4a8 100644 --- a/src/csharp/Grpc.Core/Internal/ServerCallHandler.cs +++ b/src/csharp/Grpc.Core/Internal/ServerCallHandler.cs @@ -33,6 +33,7 @@ using System; using System.Linq; +using System.Threading.Tasks; using Grpc.Core.Internal; using Grpc.Core.Utils; @@ -40,96 +41,147 @@ namespace Grpc.Core.Internal { internal interface IServerCallHandler { - void StartCall(string methodName, CallSafeHandle call, CompletionQueueSafeHandle cq); + Task HandleCall(string methodName, CallSafeHandle call, CompletionQueueSafeHandle cq); } - internal class UnaryRequestServerCallHandler : IServerCallHandler + internal class UnaryServerCallHandler : IServerCallHandler { readonly Method method; - readonly UnaryRequestServerMethod handler; + readonly UnaryServerMethod handler; - public UnaryRequestServerCallHandler(Method method, UnaryRequestServerMethod handler) + public UnaryServerCallHandler(Method method, UnaryServerMethod handler) { this.method = method; this.handler = handler; } - public void StartCall(string methodName, CallSafeHandle call, CompletionQueueSafeHandle cq) + public async Task HandleCall(string methodName, CallSafeHandle call, CompletionQueueSafeHandle cq) { var asyncCall = new AsyncCallServer( method.ResponseMarshaller.Serializer, method.RequestMarshaller.Deserializer); asyncCall.Initialize(call); - - var requestObserver = new RecordingObserver(); - var finishedTask = asyncCall.ServerSideCallAsync(requestObserver); - - var request = requestObserver.ToList().Result.Single(); - var responseObserver = new ServerStreamingOutputObserver(asyncCall); - handler(request, responseObserver); - - finishedTask.Wait(); + var finishedTask = asyncCall.ServerSideCallAsync(); + var requestStream = new ServerRequestStream(asyncCall); + var responseStream = new ServerResponseStream(asyncCall); + + var request = await requestStream.ReadNext(); + // TODO(jtattermusch): we need to read the full stream so that native callhandle gets deallocated. + Preconditions.CheckArgument(await requestStream.ReadNext() == null); + + var result = await handler(request); + await responseStream.Write(result); + await responseStream.WriteStatus(Status.DefaultSuccess); + await finishedTask; } } - internal class StreamingRequestServerCallHandler : IServerCallHandler + internal class ServerStreamingServerCallHandler : IServerCallHandler { readonly Method method; - readonly StreamingRequestServerMethod handler; + readonly ServerStreamingServerMethod handler; - public StreamingRequestServerCallHandler(Method method, StreamingRequestServerMethod handler) + public ServerStreamingServerCallHandler(Method method, ServerStreamingServerMethod handler) { this.method = method; this.handler = handler; } - public void StartCall(string methodName, CallSafeHandle call, CompletionQueueSafeHandle cq) + public async Task HandleCall(string methodName, CallSafeHandle call, CompletionQueueSafeHandle cq) { var asyncCall = new AsyncCallServer( method.ResponseMarshaller.Serializer, method.RequestMarshaller.Deserializer); asyncCall.Initialize(call); + var finishedTask = asyncCall.ServerSideCallAsync(); + var requestStream = new ServerRequestStream(asyncCall); + var responseStream = new ServerResponseStream(asyncCall); + + var request = await requestStream.ReadNext(); + // TODO(jtattermusch): we need to read the full stream so that native callhandle gets deallocated. + Preconditions.CheckArgument(await requestStream.ReadNext() == null); - var responseObserver = new ServerStreamingOutputObserver(asyncCall); - var requestObserver = handler(responseObserver); - var finishedTask = asyncCall.ServerSideCallAsync(requestObserver); - finishedTask.Wait(); + await handler(request, responseStream); + await responseStream.WriteStatus(Status.DefaultSuccess); + await finishedTask; } } - internal class NoSuchMethodCallHandler : IServerCallHandler + internal class ClientStreamingServerCallHandler : IServerCallHandler { - public void StartCall(string methodName, CallSafeHandle call, CompletionQueueSafeHandle cq) - { - // We don't care about the payload type here. - var asyncCall = new AsyncCallServer( - (payload) => payload, (payload) => payload); - - asyncCall.Initialize(call); + readonly Method method; + readonly ClientStreamingServerMethod handler; - var finishedTask = asyncCall.ServerSideCallAsync(new NullObserver()); + public ClientStreamingServerCallHandler(Method method, ClientStreamingServerMethod handler) + { + this.method = method; + this.handler = handler; + } - // TODO: check result of the completion status. - asyncCall.StartSendStatusFromServer(new Status(StatusCode.Unimplemented, "No such method."), new AsyncCompletionDelegate((error) => { })); + public async Task HandleCall(string methodName, CallSafeHandle call, CompletionQueueSafeHandle cq) + { + var asyncCall = new AsyncCallServer( + method.ResponseMarshaller.Serializer, + method.RequestMarshaller.Deserializer); - finishedTask.Wait(); + asyncCall.Initialize(call); + var finishedTask = asyncCall.ServerSideCallAsync(); + var requestStream = new ServerRequestStream(asyncCall); + var responseStream = new ServerResponseStream(asyncCall); + + var result = await handler(requestStream); + await responseStream.Write(result); + await responseStream.WriteStatus(Status.DefaultSuccess); + await finishedTask; } } - internal class NullObserver : IObserver + internal class DuplexStreamingServerCallHandler : IServerCallHandler { - public void OnCompleted() + readonly Method method; + readonly DuplexStreamingServerMethod handler; + + public DuplexStreamingServerCallHandler(Method method, DuplexStreamingServerMethod handler) { + this.method = method; + this.handler = handler; } - public void OnError(Exception error) + public async Task HandleCall(string methodName, CallSafeHandle call, CompletionQueueSafeHandle cq) { + var asyncCall = new AsyncCallServer( + method.ResponseMarshaller.Serializer, + method.RequestMarshaller.Deserializer); + + asyncCall.Initialize(call); + var finishedTask = asyncCall.ServerSideCallAsync(); + var requestStream = new ServerRequestStream(asyncCall); + var responseStream = new ServerResponseStream(asyncCall); + + await handler(requestStream, responseStream); + await responseStream.WriteStatus(Status.DefaultSuccess); + await finishedTask; } + } - public void OnNext(T value) + internal class NoSuchMethodCallHandler : IServerCallHandler + { + public async Task HandleCall(string methodName, CallSafeHandle call, CompletionQueueSafeHandle cq) { + // We don't care about the payload type here. + var asyncCall = new AsyncCallServer( + (payload) => payload, (payload) => payload); + + asyncCall.Initialize(call); + var finishedTask = asyncCall.ServerSideCallAsync(); + var requestStream = new ServerRequestStream(asyncCall); + var responseStream = new ServerResponseStream(asyncCall); + await responseStream.WriteStatus(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/ServerCalls.cs b/src/csharp/Grpc.Core/Internal/ServerCalls.cs new file mode 100644 index 0000000000..5c6b335c7f --- /dev/null +++ b/src/csharp/Grpc.Core/Internal/ServerCalls.cs @@ -0,0 +1,63 @@ +#region Copyright notice and license + +// Copyright 2015, Google Inc. +// All rights reserved. +// +// Redistribution and use in source and binary forms, with or without +// modification, are permitted provided that the following conditions are +// met: +// +// * Redistributions of source code must retain the above copyright +// notice, this list of conditions and the following disclaimer. +// * Redistributions in binary form must reproduce the above +// copyright notice, this list of conditions and the following disclaimer +// in the documentation and/or other materials provided with the +// distribution. +// * Neither the name of Google Inc. nor the names of its +// contributors may be used to endorse or promote products derived from +// this software without specific prior written permission. +// +// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS +// "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT +// LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR +// A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT +// OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, +// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT +// LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, +// DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY +// THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT +// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE +// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + +#endregion + +using System; +using System.Threading; +using System.Threading.Tasks; +using Grpc.Core; + +namespace Grpc.Core.Internal +{ + internal static class ServerCalls + { + public static IServerCallHandler UnaryCall(Method method, UnaryServerMethod handler) + { + return new UnaryServerCallHandler(method, handler); + } + + public static IServerCallHandler ClientStreamingCall(Method method, ClientStreamingServerMethod handler) + { + return new ClientStreamingServerCallHandler(method, handler); + } + + public static IServerCallHandler ServerStreamingCall(Method method, ServerStreamingServerMethod handler) + { + return new ServerStreamingServerCallHandler(method, handler); + } + + public static IServerCallHandler DuplexStreamingCall(Method method, DuplexStreamingServerMethod handler) + { + return new DuplexStreamingServerCallHandler(method, handler); + } + } +} diff --git a/src/csharp/Grpc.Core/Internal/ServerRequestStream.cs b/src/csharp/Grpc.Core/Internal/ServerRequestStream.cs new file mode 100644 index 0000000000..aa311059c3 --- /dev/null +++ b/src/csharp/Grpc.Core/Internal/ServerRequestStream.cs @@ -0,0 +1,56 @@ +#region Copyright notice and license + +// Copyright 2015, Google Inc. +// All rights reserved. +// +// Redistribution and use in source and binary forms, with or without +// modification, are permitted provided that the following conditions are +// met: +// +// * Redistributions of source code must retain the above copyright +// notice, this list of conditions and the following disclaimer. +// * Redistributions in binary form must reproduce the above +// copyright notice, this list of conditions and the following disclaimer +// in the documentation and/or other materials provided with the +// distribution. +// * Neither the name of Google Inc. nor the names of its +// contributors may be used to endorse or promote products derived from +// this software without specific prior written permission. +// +// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS +// "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT +// LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR +// A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT +// OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, +// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT +// LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, +// DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY +// THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT +// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE +// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + +#endregion + +using System; +using System.Collections.Generic; +using System.Threading.Tasks; + +namespace Grpc.Core.Internal +{ + internal class ServerRequestStream : IAsyncStreamReader + { + readonly AsyncCallServer call; + + public ServerRequestStream(AsyncCallServer call) + { + this.call = call; + } + + public Task ReadNext() + { + var taskSource = new AsyncCompletionTaskSource(); + call.StartReadMessage(taskSource.CompletionDelegate); + return taskSource.Task; + } + } +} diff --git a/src/csharp/Grpc.Core/Internal/ServerResponseStream.cs b/src/csharp/Grpc.Core/Internal/ServerResponseStream.cs new file mode 100644 index 0000000000..686017c048 --- /dev/null +++ b/src/csharp/Grpc.Core/Internal/ServerResponseStream.cs @@ -0,0 +1,64 @@ +#region Copyright notice and license +// Copyright 2015, Google Inc. +// All rights reserved. +// +// Redistribution and use in source and binary forms, with or without +// modification, are permitted provided that the following conditions are +// met: +// +// * Redistributions of source code must retain the above copyright +// notice, this list of conditions and the following disclaimer. +// * Redistributions in binary form must reproduce the above +// copyright notice, this list of conditions and the following disclaimer +// in the documentation and/or other materials provided with the +// distribution. +// * Neither the name of Google Inc. nor the names of its +// contributors may be used to endorse or promote products derived from +// this software without specific prior written permission. +// +// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS +// "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT +// LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR +// A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT +// OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, +// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT +// LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, +// DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY +// THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT +// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE +// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. +#endregion + +using System; +using System.Threading.Tasks; +using Grpc.Core.Internal; + +namespace Grpc.Core.Internal +{ + /// + /// Writes responses asynchronously to an underlying AsyncCallServer object. + /// + internal class ServerResponseStream : IServerStreamWriter + { + readonly AsyncCallServer call; + + public ServerResponseStream(AsyncCallServer call) + { + this.call = call; + } + + public Task Write(TResponse message) + { + var taskSource = new AsyncCompletionTaskSource(); + call.StartSendMessage(message, taskSource.CompletionDelegate); + return taskSource.Task; + } + + public Task WriteStatus(Status status) + { + var taskSource = new AsyncCompletionTaskSource(); + call.StartSendStatusFromServer(status, taskSource.CompletionDelegate); + return taskSource.Task; + } + } +} diff --git a/src/csharp/Grpc.Core/Internal/ServerStreamingOutputObserver.cs b/src/csharp/Grpc.Core/Internal/ServerStreamingOutputObserver.cs deleted file mode 100644 index 97b62d0569..0000000000 --- a/src/csharp/Grpc.Core/Internal/ServerStreamingOutputObserver.cs +++ /dev/null @@ -1,71 +0,0 @@ -#region Copyright notice and license -// Copyright 2015, Google Inc. -// All rights reserved. -// -// Redistribution and use in source and binary forms, with or without -// modification, are permitted provided that the following conditions are -// met: -// -// * Redistributions of source code must retain the above copyright -// notice, this list of conditions and the following disclaimer. -// * Redistributions in binary form must reproduce the above -// copyright notice, this list of conditions and the following disclaimer -// in the documentation and/or other materials provided with the -// distribution. -// * Neither the name of Google Inc. nor the names of its -// contributors may be used to endorse or promote products derived from -// this software without specific prior written permission. -// -// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS -// "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT -// LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR -// A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT -// OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, -// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT -// LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, -// DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY -// THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT -// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE -// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. -#endregion -using System; -using Grpc.Core.Internal; - -namespace Grpc.Core.Internal -{ - /// - /// Observer that writes all arriving messages to a call abstraction (in blocking fashion) - /// and then halfcloses the call. Used for server-side call handling. - /// - internal class ServerStreamingOutputObserver : IObserver - { - readonly AsyncCallServer call; - - public ServerStreamingOutputObserver(AsyncCallServer call) - { - this.call = call; - } - - public void OnCompleted() - { - var taskSource = new AsyncCompletionTaskSource(); - call.StartSendStatusFromServer(new Status(StatusCode.OK, ""), taskSource.CompletionDelegate); - // TODO: how bad is the Wait here? - taskSource.Task.Wait(); - } - - public void OnError(Exception error) - { - // TODO: implement this... - throw new InvalidOperationException("This should never be called."); - } - - public void OnNext(TResponse value) - { - var taskSource = new AsyncCompletionTaskSource(); - call.StartSendMessage(value, taskSource.CompletionDelegate); - // TODO: how bad is the Wait here? - taskSource.Task.Wait(); - } - } -} diff --git a/src/csharp/Grpc.Core/Method.cs b/src/csharp/Grpc.Core/Method.cs index 4f97eeef37..156e780c7d 100644 --- a/src/csharp/Grpc.Core/Method.cs +++ b/src/csharp/Grpc.Core/Method.cs @@ -35,12 +35,15 @@ using System; namespace Grpc.Core { + /// + /// Method types supported by gRPC. + /// public enum MethodType { - Unary, - ClientStreaming, - ServerStreaming, - DuplexStreaming + Unary, // Unary request, unary response. + ClientStreaming, // Streaming request, unary response. + ServerStreaming, // Unary request, streaming response. + DuplexStreaming // Streaming request, streaming response. } /// diff --git a/src/csharp/Grpc.Core/Server.cs b/src/csharp/Grpc.Core/Server.cs index e686cdddef..a3000cee46 100644 --- a/src/csharp/Grpc.Core/Server.cs +++ b/src/csharp/Grpc.Core/Server.cs @@ -181,7 +181,7 @@ namespace Grpc.Core /// /// Selects corresponding handler for given call and handles the call. /// - private void InvokeCallHandler(CallSafeHandle call, string method) + private async Task InvokeCallHandler(CallSafeHandle call, string method) { try { @@ -190,7 +190,7 @@ namespace Grpc.Core { callHandler = new NoSuchMethodCallHandler(); } - callHandler.StartCall(method, call, GetCompletionQueue()); + await callHandler.HandleCall(method, call, GetCompletionQueue()); } catch (Exception e) { @@ -218,7 +218,7 @@ namespace Grpc.Core // after server shutdown, the callback returns with null call if (!call.IsInvalid) { - Task.Run(() => InvokeCallHandler(call, method)); + Task.Run(async () => await InvokeCallHandler(call, method)); } AllowOneRpc(); diff --git a/src/csharp/Grpc.Core/ServerCalls.cs b/src/csharp/Grpc.Core/ServerCalls.cs deleted file mode 100644 index dcae99446f..0000000000 --- a/src/csharp/Grpc.Core/ServerCalls.cs +++ /dev/null @@ -1,57 +0,0 @@ -#region Copyright notice and license - -// Copyright 2015, Google Inc. -// All rights reserved. -// -// Redistribution and use in source and binary forms, with or without -// modification, are permitted provided that the following conditions are -// met: -// -// * Redistributions of source code must retain the above copyright -// notice, this list of conditions and the following disclaimer. -// * Redistributions in binary form must reproduce the above -// copyright notice, this list of conditions and the following disclaimer -// in the documentation and/or other materials provided with the -// distribution. -// * Neither the name of Google Inc. nor the names of its -// contributors may be used to endorse or promote products derived from -// this software without specific prior written permission. -// -// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS -// "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT -// LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR -// A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT -// OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, -// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT -// LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, -// DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY -// THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT -// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE -// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. - -#endregion - -using System; -using Grpc.Core.Internal; - -namespace Grpc.Core -{ - // TODO: perhaps add also serverSideStreaming and clientSideStreaming - - public delegate void UnaryRequestServerMethod(TRequest request, IObserver responseObserver); - - public delegate IObserver StreamingRequestServerMethod(IObserver responseObserver); - - internal static class ServerCalls - { - public static IServerCallHandler UnaryRequestCall(Method method, UnaryRequestServerMethod handler) - { - return new UnaryRequestServerCallHandler(method, handler); - } - - public static IServerCallHandler StreamingRequestCall(Method method, StreamingRequestServerMethod handler) - { - return new StreamingRequestServerCallHandler(method, handler); - } - } -} diff --git a/src/csharp/Grpc.Core/ServerMethods.cs b/src/csharp/Grpc.Core/ServerMethods.cs new file mode 100644 index 0000000000..6646bb5a89 --- /dev/null +++ b/src/csharp/Grpc.Core/ServerMethods.cs @@ -0,0 +1,61 @@ +#region Copyright notice and license + +// Copyright 2015, Google Inc. +// All rights reserved. +// +// Redistribution and use in source and binary forms, with or without +// modification, are permitted provided that the following conditions are +// met: +// +// * Redistributions of source code must retain the above copyright +// notice, this list of conditions and the following disclaimer. +// * Redistributions in binary form must reproduce the above +// copyright notice, this list of conditions and the following disclaimer +// in the documentation and/or other materials provided with the +// distribution. +// * Neither the name of Google Inc. nor the names of its +// contributors may be used to endorse or promote products derived from +// this software without specific prior written permission. +// +// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS +// "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT +// LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR +// A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT +// OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, +// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT +// LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, +// DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY +// THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT +// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE +// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + +#endregion + +using System; +using System.Threading; +using System.Threading.Tasks; + +using Grpc.Core.Internal; + +namespace Grpc.Core +{ + /// + /// Server-side handler for unary call. + /// + public delegate Task UnaryServerMethod(TRequest request); + + /// + /// Server-side handler for client streaming call. + /// + public delegate Task ClientStreamingServerMethod(IAsyncStreamReader requestStream); + + /// + /// Server-side handler for server streaming call. + /// + public delegate Task ServerStreamingServerMethod(TRequest request, IServerStreamWriter responseStream); + + /// + /// Server-side handler for bidi streaming call. + /// + public delegate Task DuplexStreamingServerMethod(IAsyncStreamReader requestStream, IServerStreamWriter responseStream); +} diff --git a/src/csharp/Grpc.Core/ServerServiceDefinition.cs b/src/csharp/Grpc.Core/ServerServiceDefinition.cs index f08c7d88f3..01b1dc8f7b 100644 --- a/src/csharp/Grpc.Core/ServerServiceDefinition.cs +++ b/src/csharp/Grpc.Core/ServerServiceDefinition.cs @@ -75,17 +75,33 @@ namespace Grpc.Core public Builder AddMethod( Method method, - UnaryRequestServerMethod handler) + UnaryServerMethod handler) { - callHandlers.Add(GetFullMethodName(serviceName, method.Name), ServerCalls.UnaryRequestCall(method, handler)); + callHandlers.Add(GetFullMethodName(serviceName, method.Name), ServerCalls.UnaryCall(method, handler)); return this; } public Builder AddMethod( Method method, - StreamingRequestServerMethod handler) + ClientStreamingServerMethod handler) { - callHandlers.Add(GetFullMethodName(serviceName, method.Name), ServerCalls.StreamingRequestCall(method, handler)); + callHandlers.Add(GetFullMethodName(serviceName, method.Name), ServerCalls.ClientStreamingCall(method, handler)); + return this; + } + + public Builder AddMethod( + Method method, + ServerStreamingServerMethod handler) + { + callHandlers.Add(GetFullMethodName(serviceName, method.Name), ServerCalls.ServerStreamingCall(method, handler)); + return this; + } + + public Builder AddMethod( + Method method, + DuplexStreamingServerMethod handler) + { + callHandlers.Add(GetFullMethodName(serviceName, method.Name), ServerCalls.DuplexStreamingCall(method, handler)); return this; } diff --git a/src/csharp/Grpc.Core/Status.cs b/src/csharp/Grpc.Core/Status.cs index 7d76aec4d1..b588170694 100644 --- a/src/csharp/Grpc.Core/Status.cs +++ b/src/csharp/Grpc.Core/Status.cs @@ -39,6 +39,11 @@ namespace Grpc.Core /// public struct Status { + /// + /// Default result of a successful RPC. StatusCode=OK, empty details message. + /// + public static readonly Status DefaultSuccess = new Status(StatusCode.OK, ""); + readonly StatusCode statusCode; readonly string detail; diff --git a/src/csharp/Grpc.Core/Utils/AsyncStreamExtensions.cs b/src/csharp/Grpc.Core/Utils/AsyncStreamExtensions.cs new file mode 100644 index 0000000000..f915155f8a --- /dev/null +++ b/src/csharp/Grpc.Core/Utils/AsyncStreamExtensions.cs @@ -0,0 +1,111 @@ +#region Copyright notice and license + +// Copyright 2015, Google Inc. +// All rights reserved. +// +// Redistribution and use in source and binary forms, with or without +// modification, are permitted provided that the following conditions are +// met: +// +// * Redistributions of source code must retain the above copyright +// notice, this list of conditions and the following disclaimer. +// * Redistributions in binary form must reproduce the above +// copyright notice, this list of conditions and the following disclaimer +// in the documentation and/or other materials provided with the +// distribution. +// * Neither the name of Google Inc. nor the names of its +// contributors may be used to endorse or promote products derived from +// this software without specific prior written permission. +// +// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS +// "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT +// LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR +// A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT +// OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, +// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT +// LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, +// DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY +// THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT +// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE +// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + +#endregion + +using System; +using System.Collections.Generic; +using System.Linq; +using System.Threading.Tasks; + +namespace Grpc.Core.Utils +{ + /// + /// Extension methods that simplify work with gRPC streaming calls. + /// + public static class AsyncStreamExtensions + { + /// + /// Reads the entire stream and executes an async action for each element. + /// + public static async Task ForEach(this IAsyncStreamReader streamReader, Func asyncAction) + where T : class + { + while (true) + { + var elem = await streamReader.ReadNext(); + if (elem == null) + { + break; + } + await asyncAction(elem); + } + } + + /// + /// Reads the entire stream and creates a list containing all the elements read. + /// + public static async Task> ToList(this IAsyncStreamReader streamReader) + where T : class + { + var result = new List(); + while (true) + { + var elem = await streamReader.ReadNext(); + if (elem == null) + { + break; + } + result.Add(elem); + } + return result; + } + + /// + /// Writes all elements from given enumerable to the stream. + /// Closes the stream afterwards unless close = false. + /// + public static async Task WriteAll(this IClientStreamWriter streamWriter, IEnumerable elements, bool close = true) + where T : class + { + foreach (var element in elements) + { + await streamWriter.Write(element); + } + if (close) + { + await streamWriter.Close(); + } + } + + /// + /// Writes all elements from given enumerable to the stream. + /// + public static async Task WriteAll(this IServerStreamWriter streamWriter, IEnumerable elements) + where T : class + { + foreach (var element in elements) + { + await streamWriter.Write(element); + } + } + } +} diff --git a/src/csharp/Grpc.Core/Utils/RecordingObserver.cs b/src/csharp/Grpc.Core/Utils/RecordingObserver.cs deleted file mode 100644 index 7b43ab8ad5..0000000000 --- a/src/csharp/Grpc.Core/Utils/RecordingObserver.cs +++ /dev/null @@ -1,65 +0,0 @@ -#region Copyright notice and license - -// Copyright 2015, Google Inc. -// All rights reserved. -// -// Redistribution and use in source and binary forms, with or without -// modification, are permitted provided that the following conditions are -// met: -// -// * Redistributions of source code must retain the above copyright -// notice, this list of conditions and the following disclaimer. -// * Redistributions in binary form must reproduce the above -// copyright notice, this list of conditions and the following disclaimer -// in the documentation and/or other materials provided with the -// distribution. -// * Neither the name of Google Inc. nor the names of its -// contributors may be used to endorse or promote products derived from -// this software without specific prior written permission. -// -// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS -// "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT -// LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR -// A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT -// OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, -// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT -// LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, -// DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY -// THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT -// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE -// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. - -#endregion - -using System; -using System.Collections.Generic; -using System.Threading.Tasks; - -namespace Grpc.Core.Utils -{ - public class RecordingObserver : IObserver - { - TaskCompletionSource> tcs = new TaskCompletionSource>(); - List data = new List(); - - public void OnCompleted() - { - tcs.SetResult(data); - } - - public void OnError(Exception error) - { - tcs.SetException(error); - } - - public void OnNext(T value) - { - data.Add(value); - } - - public Task> ToList() - { - return tcs.Task; - } - } -} diff --git a/src/csharp/Grpc.Core/Utils/RecordingQueue.cs b/src/csharp/Grpc.Core/Utils/RecordingQueue.cs deleted file mode 100644 index 9749168af0..0000000000 --- a/src/csharp/Grpc.Core/Utils/RecordingQueue.cs +++ /dev/null @@ -1,83 +0,0 @@ -#region Copyright notice and license - -// Copyright 2015, Google Inc. -// All rights reserved. -// -// Redistribution and use in source and binary forms, with or without -// modification, are permitted provided that the following conditions are -// met: -// -// * Redistributions of source code must retain the above copyright -// notice, this list of conditions and the following disclaimer. -// * Redistributions in binary form must reproduce the above -// copyright notice, this list of conditions and the following disclaimer -// in the documentation and/or other materials provided with the -// distribution. -// * Neither the name of Google Inc. nor the names of its -// contributors may be used to endorse or promote products derived from -// this software without specific prior written permission. -// -// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS -// "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT -// LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR -// A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT -// OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, -// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT -// LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, -// DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY -// THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT -// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE -// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. - -#endregion - -using System; -using System.Collections.Concurrent; -using System.Collections.Generic; -using System.Threading.Tasks; - -namespace Grpc.Core.Utils -{ - // TODO: replace this by something that implements IAsyncEnumerator. - /// - /// Observer that allows us to await incoming messages one-by-one. - /// The implementation is not ideal and class will be probably replaced - /// by something more versatile in the future. - /// - public class RecordingQueue : IObserver - { - readonly BlockingCollection queue = new BlockingCollection(); - TaskCompletionSource tcs = new TaskCompletionSource(); - - public void OnCompleted() - { - tcs.SetResult(null); - } - - public void OnError(Exception error) - { - tcs.SetException(error); - } - - public void OnNext(T value) - { - queue.Add(value); - } - - public BlockingCollection Queue - { - get - { - return queue; - } - } - - public Task Finished - { - get - { - return tcs.Task; - } - } - } -} diff --git a/src/csharp/Grpc.Examples.Tests/MathClientServerTests.cs b/src/csharp/Grpc.Examples.Tests/MathClientServerTests.cs index fa5d6688a6..332795e0e5 100644 --- a/src/csharp/Grpc.Examples.Tests/MathClientServerTests.cs +++ b/src/csharp/Grpc.Examples.Tests/MathClientServerTests.cs @@ -63,7 +63,7 @@ namespace math.Tests server.Start(); channel = new Channel(host + ":" + port); - // TODO: get rid of the custom header here once we have dedicated tests + // TODO(jtattermusch): get rid of the custom header here once we have dedicated tests // for header support. var stubConfig = new StubConfiguration((headerBuilder) => { @@ -97,55 +97,67 @@ namespace math.Tests Assert.AreEqual(0, response.Remainder); } - // TODO: test division by zero + // TODO(jtattermusch): test division by zero [Test] public void DivAsync() { - DivReply response = client.DivAsync(new DivArgs.Builder { Dividend = 10, Divisor = 3 }.Build()).Result; - Assert.AreEqual(3, response.Quotient); - Assert.AreEqual(1, response.Remainder); + Task.Run(async () => + { + DivReply response = await client.DivAsync(new DivArgs.Builder { Dividend = 10, Divisor = 3 }.Build()); + Assert.AreEqual(3, response.Quotient); + Assert.AreEqual(1, response.Remainder); + }).Wait(); } [Test] public void Fib() { - var recorder = new RecordingObserver(); - client.Fib(new FibArgs.Builder { Limit = 6 }.Build(), recorder); + Task.Run(async () => + { + var call = client.Fib(new FibArgs.Builder { Limit = 6 }.Build()); - CollectionAssert.AreEqual(new List { 1, 1, 2, 3, 5, 8 }, - recorder.ToList().Result.ConvertAll((n) => n.Num_)); + var responses = await call.ResponseStream.ToList(); + CollectionAssert.AreEqual(new List { 1, 1, 2, 3, 5, 8 }, + responses.ConvertAll((n) => n.Num_)); + }).Wait(); } // TODO: test Fib with limit=0 and cancellation [Test] public void Sum() { - var clientStreamingResult = client.Sum(); - var numList = new List { 10, 20, 30 }.ConvertAll( - n => Num.CreateBuilder().SetNum_(n).Build()); - numList.Subscribe(clientStreamingResult.Inputs); - - Assert.AreEqual(60, clientStreamingResult.Task.Result.Num_); + Task.Run(async () => + { + var call = client.Sum(); + var numbers = new List { 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_); + }).Wait(); } [Test] public void DivMany() { - List divArgsList = new List + Task.Run(async () => { - new DivArgs.Builder { Dividend = 10, Divisor = 3 }.Build(), - new DivArgs.Builder { Dividend = 100, Divisor = 21 }.Build(), - new DivArgs.Builder { Dividend = 7, Divisor = 2 }.Build() - }; - - var recorder = new RecordingObserver(); - var requestObserver = client.DivMany(recorder); - divArgsList.Subscribe(requestObserver); - var result = recorder.ToList().Result; - - CollectionAssert.AreEqual(new long[] { 3, 4, 3 }, result.ConvertAll((divReply) => divReply.Quotient)); - CollectionAssert.AreEqual(new long[] { 1, 16, 1 }, result.ConvertAll((divReply) => divReply.Remainder)); + var divArgsList = new List + { + new DivArgs.Builder { Dividend = 10, Divisor = 3 }.Build(), + new DivArgs.Builder { Dividend = 100, Divisor = 21 }.Build(), + new DivArgs.Builder { Dividend = 7, Divisor = 2 }.Build() + }; + + 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)); + }).Wait(); } } } diff --git a/src/csharp/Grpc.Examples/MathExamples.cs b/src/csharp/Grpc.Examples/MathExamples.cs index 032372b2a1..dba5a7736c 100644 --- a/src/csharp/Grpc.Examples/MathExamples.cs +++ b/src/csharp/Grpc.Examples/MathExamples.cs @@ -61,9 +61,8 @@ namespace math public static async Task FibExample(MathGrpc.IMathServiceClient stub) { - var recorder = new RecordingObserver(); - stub.Fib(new FibArgs.Builder { Limit = 5 }.Build(), recorder); - List result = await recorder.ToList(); + var call = stub.Fib(new FibArgs.Builder { Limit = 5 }.Build()); + List result = await call.ResponseStream.ToList(); Console.WriteLine("Fib Result: " + string.Join("|", result)); } @@ -76,9 +75,9 @@ namespace math new Num.Builder { Num_ = 3 }.Build() }; - var clientStreamingResult = stub.Sum(); - numbers.Subscribe(clientStreamingResult.Inputs); - Console.WriteLine("Sum Result: " + await clientStreamingResult.Task); + var call = stub.Sum(); + await call.RequestStream.WriteAll(numbers); + Console.WriteLine("Sum Result: " + await call.Result); } public static async Task DivManyExample(MathGrpc.IMathServiceClient stub) @@ -89,12 +88,9 @@ namespace math new DivArgs.Builder { Dividend = 100, Divisor = 21 }.Build(), new DivArgs.Builder { Dividend = 7, Divisor = 2 }.Build() }; - - var recorder = new RecordingObserver(); - var inputs = stub.DivMany(recorder); - divArgsList.Subscribe(inputs); - var result = await recorder.ToList(); - Console.WriteLine("DivMany Result: " + string.Join("|", result)); + var call = stub.DivMany(); + await call.RequestStream.WriteAll(divArgsList); + Console.WriteLine("DivMany Result: " + string.Join("|", await call.ResponseStream.ToList())); } public static async Task DependendRequestsExample(MathGrpc.IMathServiceClient stub) @@ -106,9 +102,9 @@ namespace math new Num.Builder { Num_ = 3 }.Build() }; - var clientStreamingResult = stub.Sum(); - numbers.Subscribe(clientStreamingResult.Inputs); - Num sum = await clientStreamingResult.Task; + var sumCall = stub.Sum(); + await sumCall.RequestStream.WriteAll(numbers); + Num 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/MathGrpc.cs b/src/csharp/Grpc.Examples/MathGrpc.cs index 24e6a1de8e..60408b9018 100644 --- a/src/csharp/Grpc.Examples/MathGrpc.cs +++ b/src/csharp/Grpc.Examples/MathGrpc.cs @@ -82,11 +82,11 @@ namespace math Task DivAsync(DivArgs request, CancellationToken token = default(CancellationToken)); - void Fib(FibArgs request, IObserver responseObserver, CancellationToken token = default(CancellationToken)); + AsyncServerStreamingCall Fib(FibArgs request, CancellationToken token = default(CancellationToken)); - ClientStreamingAsyncResult Sum(CancellationToken token = default(CancellationToken)); + AsyncClientStreamingCall Sum(CancellationToken token = default(CancellationToken)); - IObserver DivMany(IObserver responseObserver, CancellationToken token = default(CancellationToken)); + AsyncDuplexStreamingCall DivMany(CancellationToken token = default(CancellationToken)); } public class MathServiceClientStub : AbstractStub, IMathServiceClient @@ -111,35 +111,35 @@ namespace math return Calls.AsyncUnaryCall(call, request, token); } - public void Fib(FibArgs request, IObserver responseObserver, CancellationToken token = default(CancellationToken)) + public AsyncServerStreamingCall Fib(FibArgs request, CancellationToken token = default(CancellationToken)) { var call = CreateCall(ServiceName, FibMethod); - Calls.AsyncServerStreamingCall(call, request, responseObserver, token); + return Calls.AsyncServerStreamingCall(call, request, token); } - public ClientStreamingAsyncResult Sum(CancellationToken token = default(CancellationToken)) + public AsyncClientStreamingCall Sum(CancellationToken token = default(CancellationToken)) { var call = CreateCall(ServiceName, SumMethod); return Calls.AsyncClientStreamingCall(call, token); } - public IObserver DivMany(IObserver responseObserver, CancellationToken token = default(CancellationToken)) + public AsyncDuplexStreamingCall DivMany(CancellationToken token = default(CancellationToken)) { var call = CreateCall(ServiceName, DivManyMethod); - return Calls.DuplexStreamingCall(call, responseObserver, token); + return Calls.AsyncDuplexStreamingCall(call, token); } } // server-side interface public interface IMathService { - void Div(DivArgs request, IObserver responseObserver); + Task Div(DivArgs request); - void Fib(FibArgs request, IObserver responseObserver); + Task Fib(FibArgs request, IServerStreamWriter responseStream); - IObserver Sum(IObserver responseObserver); + Task Sum(IAsyncStreamReader requestStream); - IObserver DivMany(IObserver responseObserver); + Task DivMany(IAsyncStreamReader requestStream, IServerStreamWriter responseStream); } public static ServerServiceDefinition BindService(IMathService serviceImpl) diff --git a/src/csharp/Grpc.Examples/MathServiceImpl.cs b/src/csharp/Grpc.Examples/MathServiceImpl.cs index 0b2357e0fa..83ec2a8c3d 100644 --- a/src/csharp/Grpc.Examples/MathServiceImpl.cs +++ b/src/csharp/Grpc.Examples/MathServiceImpl.cs @@ -36,6 +36,7 @@ using System.Collections.Generic; using System.Reactive.Linq; using System.Threading; using System.Threading.Tasks; +using Grpc.Core; using Grpc.Core.Utils; namespace math @@ -45,18 +46,16 @@ namespace math /// public class MathServiceImpl : MathGrpc.IMathService { - public void Div(DivArgs request, IObserver responseObserver) + public Task Div(DivArgs request) { - var response = DivInternal(request); - responseObserver.OnNext(response); - responseObserver.OnCompleted(); + return Task.FromResult(DivInternal(request)); } - public void Fib(FibArgs request, IObserver responseObserver) + public async Task Fib(FibArgs request, IServerStreamWriter responseStream) { if (request.Limit <= 0) { - // TODO: support cancellation.... + // TODO(jtattermusch): support cancellation throw new NotImplementedException("Not implemented yet"); } @@ -64,34 +63,27 @@ namespace math { foreach (var num in FibInternal(request.Limit)) { - responseObserver.OnNext(num); + await responseStream.Write(num); } - responseObserver.OnCompleted(); } } - public IObserver Sum(IObserver responseObserver) + public async Task Sum(IAsyncStreamReader requestStream) { - var recorder = new RecordingObserver(); - Task.Factory.StartNew(() => + long sum = 0; + await requestStream.ForEach(async num => { - List inputs = recorder.ToList().Result; - - long sum = 0; - foreach (Num num in inputs) - { - sum += num.Num_; - } - - responseObserver.OnNext(Num.CreateBuilder().SetNum_(sum).Build()); - responseObserver.OnCompleted(); + sum += num.Num_; }); - return recorder; + return Num.CreateBuilder().SetNum_(sum).Build(); } - public IObserver DivMany(IObserver responseObserver) + public async Task DivMany(IAsyncStreamReader requestStream, IServerStreamWriter responseStream) { - return new DivObserver(responseObserver); + await requestStream.ForEach(async divArgs => + { + await responseStream.Write(DivInternal(divArgs)); + }); } static DivReply DivInternal(DivArgs args) @@ -114,31 +106,6 @@ namespace math b = temp + b; yield return new Num.Builder { Num_ = a }.Build(); } - } - - private class DivObserver : IObserver - { - readonly IObserver responseObserver; - - public DivObserver(IObserver responseObserver) - { - this.responseObserver = responseObserver; - } - - public void OnCompleted() - { - responseObserver.OnCompleted(); - } - - public void OnError(Exception error) - { - throw new NotImplementedException(); - } - - public void OnNext(DivArgs value) - { - responseObserver.OnNext(DivInternal(value)); - } - } + } } } diff --git a/src/csharp/Grpc.IntegrationTesting/InteropClient.cs b/src/csharp/Grpc.IntegrationTesting/InteropClient.cs index 1fbae374b1..573ab30452 100644 --- a/src/csharp/Grpc.IntegrationTesting/InteropClient.cs +++ b/src/csharp/Grpc.IntegrationTesting/InteropClient.cs @@ -34,6 +34,7 @@ using System; using System.Collections.Generic; using System.Text.RegularExpressions; +using System.Threading.Tasks; using Google.ProtocolBuffers; using grpc.testing; @@ -199,113 +200,115 @@ namespace Grpc.IntegrationTesting public static void RunClientStreaming(TestServiceGrpc.ITestServiceClient client) { - Console.WriteLine("running client_streaming"); + Task.Run(async () => + { + Console.WriteLine("running client_streaming"); - var bodySizes = new List { 27182, 8, 1828, 45904 }; + var bodySizes = new List { 27182, 8, 1828, 45904 }.ConvertAll((size) => StreamingInputCallRequest.CreateBuilder().SetPayload(CreateZerosPayload(size)).Build()); - var context = client.StreamingInputCall(); - foreach (var size in bodySizes) - { - context.Inputs.OnNext( - StreamingInputCallRequest.CreateBuilder().SetPayload(CreateZerosPayload(size)).Build()); - } - context.Inputs.OnCompleted(); + var call = client.StreamingInputCall(); + await call.RequestStream.WriteAll(bodySizes); - var response = context.Task.Result; - Assert.AreEqual(74922, response.AggregatedPayloadSize); - Console.WriteLine("Passed!"); + var response = await call.Result; + Assert.AreEqual(74922, response.AggregatedPayloadSize); + Console.WriteLine("Passed!"); + }).Wait(); } public static void RunServerStreaming(TestServiceGrpc.ITestServiceClient client) { - Console.WriteLine("running server_streaming"); + Task.Run(async () => + { + Console.WriteLine("running server_streaming"); - var bodySizes = new List { 31415, 9, 2653, 58979 }; + var bodySizes = new List { 31415, 9, 2653, 58979 }; - var request = StreamingOutputCallRequest.CreateBuilder() + var request = StreamingOutputCallRequest.CreateBuilder() .SetResponseType(PayloadType.COMPRESSABLE) .AddRangeResponseParameters(bodySizes.ConvertAll( - (size) => ResponseParameters.CreateBuilder().SetSize(size).Build())) + (size) => ResponseParameters.CreateBuilder().SetSize(size).Build())) .Build(); - var recorder = new RecordingObserver(); - client.StreamingOutputCall(request, recorder); + var call = client.StreamingOutputCall(request); - var responseList = recorder.ToList().Result; - - foreach (var res in responseList) - { - Assert.AreEqual(PayloadType.COMPRESSABLE, res.Payload.Type); - } - CollectionAssert.AreEqual(bodySizes, responseList.ConvertAll((item) => item.Payload.Body.Length)); - Console.WriteLine("Passed!"); + 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)); + Console.WriteLine("Passed!"); + }).Wait(); } public static void RunPingPong(TestServiceGrpc.ITestServiceClient client) { - Console.WriteLine("running ping_pong"); + Task.Run(async () => + { + Console.WriteLine("running ping_pong"); - var recorder = new RecordingQueue(); - var inputs = client.FullDuplexCall(recorder); + var call = client.FullDuplexCall(); - StreamingOutputCallResponse response; + StreamingOutputCallResponse response; - inputs.OnNext(StreamingOutputCallRequest.CreateBuilder() + await call.RequestStream.Write(StreamingOutputCallRequest.CreateBuilder() .SetResponseType(PayloadType.COMPRESSABLE) .AddResponseParameters(ResponseParameters.CreateBuilder().SetSize(31415)) .SetPayload(CreateZerosPayload(27182)).Build()); - response = recorder.Queue.Take(); - Assert.AreEqual(PayloadType.COMPRESSABLE, response.Payload.Type); - Assert.AreEqual(31415, response.Payload.Body.Length); + response = await call.ResponseStream.ReadNext(); + Assert.AreEqual(PayloadType.COMPRESSABLE, response.Payload.Type); + Assert.AreEqual(31415, response.Payload.Body.Length); - inputs.OnNext(StreamingOutputCallRequest.CreateBuilder() + await call.RequestStream.Write(StreamingOutputCallRequest.CreateBuilder() .SetResponseType(PayloadType.COMPRESSABLE) .AddResponseParameters(ResponseParameters.CreateBuilder().SetSize(9)) .SetPayload(CreateZerosPayload(8)).Build()); - response = recorder.Queue.Take(); - Assert.AreEqual(PayloadType.COMPRESSABLE, response.Payload.Type); - Assert.AreEqual(9, response.Payload.Body.Length); + response = await call.ResponseStream.ReadNext(); + Assert.AreEqual(PayloadType.COMPRESSABLE, response.Payload.Type); + Assert.AreEqual(9, response.Payload.Body.Length); - inputs.OnNext(StreamingOutputCallRequest.CreateBuilder() + await call.RequestStream.Write(StreamingOutputCallRequest.CreateBuilder() .SetResponseType(PayloadType.COMPRESSABLE) .AddResponseParameters(ResponseParameters.CreateBuilder().SetSize(2653)) .SetPayload(CreateZerosPayload(1828)).Build()); - response = recorder.Queue.Take(); - Assert.AreEqual(PayloadType.COMPRESSABLE, response.Payload.Type); - Assert.AreEqual(2653, response.Payload.Body.Length); + response = await call.ResponseStream.ReadNext(); + Assert.AreEqual(PayloadType.COMPRESSABLE, response.Payload.Type); + Assert.AreEqual(2653, response.Payload.Body.Length); - inputs.OnNext(StreamingOutputCallRequest.CreateBuilder() + await call.RequestStream.Write(StreamingOutputCallRequest.CreateBuilder() .SetResponseType(PayloadType.COMPRESSABLE) .AddResponseParameters(ResponseParameters.CreateBuilder().SetSize(58979)) .SetPayload(CreateZerosPayload(45904)).Build()); - response = recorder.Queue.Take(); - Assert.AreEqual(PayloadType.COMPRESSABLE, response.Payload.Type); - Assert.AreEqual(58979, response.Payload.Body.Length); + response = await call.ResponseStream.ReadNext(); + Assert.AreEqual(PayloadType.COMPRESSABLE, response.Payload.Type); + Assert.AreEqual(58979, response.Payload.Body.Length); - inputs.OnCompleted(); + await call.RequestStream.Close(); - recorder.Finished.Wait(); - Assert.AreEqual(0, recorder.Queue.Count); + response = await call.ResponseStream.ReadNext(); + Assert.AreEqual(null, response); - Console.WriteLine("Passed!"); + Console.WriteLine("Passed!"); + }).Wait(); } public static void RunEmptyStream(TestServiceGrpc.ITestServiceClient client) { - Console.WriteLine("running empty_stream"); - - var recorder = new RecordingObserver(); - var inputs = client.FullDuplexCall(recorder); - inputs.OnCompleted(); + Task.Run(async () => + { + Console.WriteLine("running empty_stream"); + var call = client.FullDuplexCall(); + await call.Close(); - var responseList = recorder.ToList().Result; - Assert.AreEqual(0, responseList.Count); + var responseList = await call.ResponseStream.ToList(); + Assert.AreEqual(0, responseList.Count); - Console.WriteLine("Passed!"); + Console.WriteLine("Passed!"); + }).Wait(); } public static void RunServiceAccountCreds(TestServiceGrpc.ITestServiceClient client) diff --git a/src/csharp/Grpc.IntegrationTesting/InteropClientServerTest.cs b/src/csharp/Grpc.IntegrationTesting/InteropClientServerTest.cs index 1e76d3df21..e929b76b5e 100644 --- a/src/csharp/Grpc.IntegrationTesting/InteropClientServerTest.cs +++ b/src/csharp/Grpc.IntegrationTesting/InteropClientServerTest.cs @@ -87,7 +87,7 @@ namespace Grpc.IntegrationTesting [Test] public void LargeUnary() { - InteropClient.RunEmptyUnary(client); + InteropClient.RunLargeUnary(client); } [Test] diff --git a/src/csharp/Grpc.IntegrationTesting/TestServiceGrpc.cs b/src/csharp/Grpc.IntegrationTesting/TestServiceGrpc.cs index f63e0361a4..d1f8aa12c7 100644 --- a/src/csharp/Grpc.IntegrationTesting/TestServiceGrpc.cs +++ b/src/csharp/Grpc.IntegrationTesting/TestServiceGrpc.cs @@ -100,13 +100,13 @@ namespace grpc.testing Task UnaryCallAsync(SimpleRequest request, CancellationToken token = default(CancellationToken)); - void StreamingOutputCall(StreamingOutputCallRequest request, IObserver responseObserver, CancellationToken token = default(CancellationToken)); + AsyncServerStreamingCall StreamingOutputCall(StreamingOutputCallRequest request, CancellationToken token = default(CancellationToken)); - ClientStreamingAsyncResult StreamingInputCall(CancellationToken token = default(CancellationToken)); + AsyncClientStreamingCall StreamingInputCall(CancellationToken token = default(CancellationToken)); - IObserver FullDuplexCall(IObserver responseObserver, CancellationToken token = default(CancellationToken)); + AsyncDuplexStreamingCall FullDuplexCall(CancellationToken token = default(CancellationToken)); - IObserver HalfDuplexCall(IObserver responseObserver, CancellationToken token = default(CancellationToken)); + AsyncDuplexStreamingCall HalfDuplexCall(CancellationToken token = default(CancellationToken)); } public class TestServiceClientStub : AbstractStub, ITestServiceClient @@ -143,45 +143,45 @@ namespace grpc.testing return Calls.AsyncUnaryCall(call, request, token); } - public void StreamingOutputCall(StreamingOutputCallRequest request, IObserver responseObserver, CancellationToken token = default(CancellationToken)) + public AsyncServerStreamingCall StreamingOutputCall(StreamingOutputCallRequest request, CancellationToken token = default(CancellationToken)) { var call = CreateCall(ServiceName, StreamingOutputCallMethod); - Calls.AsyncServerStreamingCall(call, request, responseObserver, token); + return Calls.AsyncServerStreamingCall(call, request, token); } - public ClientStreamingAsyncResult StreamingInputCall(CancellationToken token = default(CancellationToken)) + public AsyncClientStreamingCall StreamingInputCall(CancellationToken token = default(CancellationToken)) { var call = CreateCall(ServiceName, StreamingInputCallMethod); return Calls.AsyncClientStreamingCall(call, token); } - public IObserver FullDuplexCall(IObserver responseObserver, CancellationToken token = default(CancellationToken)) + public AsyncDuplexStreamingCall FullDuplexCall(CancellationToken token = default(CancellationToken)) { var call = CreateCall(ServiceName, FullDuplexCallMethod); - return Calls.DuplexStreamingCall(call, responseObserver, token); + return Calls.AsyncDuplexStreamingCall(call, token); } - public IObserver HalfDuplexCall(IObserver responseObserver, CancellationToken token = default(CancellationToken)) + public AsyncDuplexStreamingCall HalfDuplexCall(CancellationToken token = default(CancellationToken)) { var call = CreateCall(ServiceName, HalfDuplexCallMethod); - return Calls.DuplexStreamingCall(call, responseObserver, token); + return Calls.AsyncDuplexStreamingCall(call, token); } } // server-side interface public interface ITestService { - void EmptyCall(Empty request, IObserver responseObserver); + Task EmptyCall(Empty request); - void UnaryCall(SimpleRequest request, IObserver responseObserver); + Task UnaryCall(SimpleRequest request); - void StreamingOutputCall(StreamingOutputCallRequest request, IObserver responseObserver); + Task StreamingOutputCall(StreamingOutputCallRequest request, IServerStreamWriter responseStream); - IObserver StreamingInputCall(IObserver responseObserver); + Task StreamingInputCall(IAsyncStreamReader requestStream); - IObserver FullDuplexCall(IObserver responseObserver); + Task FullDuplexCall(IAsyncStreamReader requestStream, IServerStreamWriter responseStream); - IObserver HalfDuplexCall(IObserver responseObserver); + Task HalfDuplexCall(IAsyncStreamReader requestStream, IServerStreamWriter responseStream); } public static ServerServiceDefinition BindService(ITestService serviceImpl) diff --git a/src/csharp/Grpc.IntegrationTesting/TestServiceImpl.cs b/src/csharp/Grpc.IntegrationTesting/TestServiceImpl.cs index 661b31b0ee..8b0cf3a2d0 100644 --- a/src/csharp/Grpc.IntegrationTesting/TestServiceImpl.cs +++ b/src/csharp/Grpc.IntegrationTesting/TestServiceImpl.cs @@ -36,6 +36,7 @@ using System.Collections.Generic; using System.Threading; using System.Threading.Tasks; using Google.ProtocolBuffers; +using Grpc.Core; using Grpc.Core.Utils; namespace grpc.testing @@ -45,88 +46,54 @@ namespace grpc.testing /// public class TestServiceImpl : TestServiceGrpc.ITestService { - public void EmptyCall(Empty request, IObserver responseObserver) + public Task EmptyCall(Empty request) { - responseObserver.OnNext(Empty.DefaultInstance); - responseObserver.OnCompleted(); + return Task.FromResult(Empty.DefaultInstance); } - public void UnaryCall(SimpleRequest request, IObserver responseObserver) + public Task UnaryCall(SimpleRequest request) { var response = SimpleResponse.CreateBuilder() .SetPayload(CreateZerosPayload(request.ResponseSize)).Build(); - // TODO: check we support ReponseType - responseObserver.OnNext(response); - responseObserver.OnCompleted(); + return Task.FromResult(response); } - public void StreamingOutputCall(StreamingOutputCallRequest request, IObserver responseObserver) + public async Task StreamingOutputCall(StreamingOutputCallRequest request, IServerStreamWriter responseStream) { foreach (var responseParam in request.ResponseParametersList) { var response = StreamingOutputCallResponse.CreateBuilder() .SetPayload(CreateZerosPayload(responseParam.Size)).Build(); - responseObserver.OnNext(response); + await responseStream.Write(response); } - responseObserver.OnCompleted(); } - public IObserver StreamingInputCall(IObserver responseObserver) + public async Task StreamingInputCall(IAsyncStreamReader requestStream) { - var recorder = new RecordingObserver(); - Task.Run(() => + int sum = 0; + await requestStream.ForEach(async request => { - int sum = 0; - foreach (var req in recorder.ToList().Result) - { - sum += req.Payload.Body.Length; - } - var response = StreamingInputCallResponse.CreateBuilder() - .SetAggregatedPayloadSize(sum).Build(); - responseObserver.OnNext(response); - responseObserver.OnCompleted(); + sum += request.Payload.Body.Length; }); - return recorder; + return StreamingInputCallResponse.CreateBuilder().SetAggregatedPayloadSize(sum).Build(); } - public IObserver FullDuplexCall(IObserver responseObserver) + public async Task FullDuplexCall(IAsyncStreamReader requestStream, IServerStreamWriter responseStream) { - return new FullDuplexObserver(responseObserver); - } - - public IObserver HalfDuplexCall(IObserver responseObserver) - { - throw new NotImplementedException(); - } - - private class FullDuplexObserver : IObserver - { - readonly IObserver responseObserver; - - public FullDuplexObserver(IObserver responseObserver) - { - this.responseObserver = responseObserver; - } - - public void OnCompleted() + await requestStream.ForEach(async request => { - responseObserver.OnCompleted(); - } - - public void OnError(Exception error) - { - throw new NotImplementedException(); - } - - public void OnNext(StreamingOutputCallRequest value) - { - foreach (var responseParam in value.ResponseParametersList) + foreach (var responseParam in request.ResponseParametersList) { var response = StreamingOutputCallResponse.CreateBuilder() .SetPayload(CreateZerosPayload(responseParam.Size)).Build(); - responseObserver.OnNext(response); + await responseStream.Write(response); } - } + }); + } + + public async Task HalfDuplexCall(IAsyncStreamReader requestStream, IServerStreamWriter responseStream) + { + throw new NotImplementedException(); } private static Payload CreateZerosPayload(int size) -- cgit v1.2.3 From e5c446004f2c1d3e2f2e5f0963f99332c6c94bc4 Mon Sep 17 00:00:00 2001 From: Jan Tattermusch Date: Fri, 1 May 2015 11:12:34 -0700 Subject: Added basic support for cancellation --- src/csharp/Grpc.Core.Tests/ClientServerTest.cs | 219 +++++++++++++++------ src/csharp/Grpc.Core/Calls.cs | 17 +- src/csharp/Grpc.Core/Internal/AsyncCallServer.cs | 6 + src/csharp/Grpc.Core/Internal/ServerCallHandler.cs | 83 ++++++-- .../Grpc.IntegrationTesting/InteropClient.cs | 65 ++++++ .../InteropClientServerTest.cs | 12 +- 6 files changed, 322 insertions(+), 80 deletions(-) diff --git a/src/csharp/Grpc.Core.Tests/ClientServerTest.cs b/src/csharp/Grpc.Core.Tests/ClientServerTest.cs index 9e510e82a6..c91efe53b4 100644 --- a/src/csharp/Grpc.Core.Tests/ClientServerTest.cs +++ b/src/csharp/Grpc.Core.Tests/ClientServerTest.cs @@ -47,23 +47,59 @@ namespace Grpc.Core.Tests const string Host = "localhost"; const string ServiceName = "/tests.Test"; - static readonly Method UnaryEchoStringMethod = new Method( + static readonly Method EchoMethod = new Method( MethodType.Unary, - "/tests.Test/UnaryEchoString", + "/tests.Test/Echo", + Marshallers.StringMarshaller, + Marshallers.StringMarshaller); + + static readonly Method ConcatAndEchoMethod = new Method( + MethodType.ClientStreaming, + "/tests.Test/ConcatAndEcho", + Marshallers.StringMarshaller, + Marshallers.StringMarshaller); + + static readonly Method NonexistentMethod = new Method( + MethodType.Unary, + "/tests.Test/NonexistentMethod", Marshallers.StringMarshaller, Marshallers.StringMarshaller); static readonly ServerServiceDefinition ServiceDefinition = ServerServiceDefinition.CreateBuilder(ServiceName) - .AddMethod(UnaryEchoStringMethod, HandleUnaryEchoString).Build(); + .AddMethod(EchoMethod, EchoHandler) + .AddMethod(ConcatAndEchoMethod, ConcatAndEchoHandler) + .Build(); + + Server server; + Channel channel; [TestFixtureSetUp] + public void InitClass() + { + GrpcEnvironment.Initialize(); + } + + [SetUp] public void Init() { GrpcEnvironment.Initialize(); + + server = new Server(); + server.AddServiceDefinition(ServiceDefinition); + int port = server.AddListeningPort(Host + ":0"); + server.Start(); + channel = new Channel(Host + ":" + port); } - [TestFixtureTearDown] + [TearDown] public void Cleanup() + { + channel.Dispose(); + server.ShutdownAsync().Wait(); + } + + [TestFixtureTearDown] + public void CleanupClass() { GrpcEnvironment.Shutdown(); } @@ -71,93 +107,160 @@ namespace Grpc.Core.Tests [Test] public void UnaryCall() { - var server = new Server(); - server.AddServiceDefinition(ServiceDefinition); - int port = server.AddListeningPort(Host + ":0"); - server.Start(); - - using (Channel channel = new Channel(Host + ":" + port)) - { - var call = new Call(ServiceName, UnaryEchoStringMethod, channel, Metadata.Empty); - Assert.AreEqual("ABC", Calls.BlockingUnaryCall(call, "ABC", default(CancellationToken))); - } - - server.ShutdownAsync().Wait(); + var call = new Call(ServiceName, EchoMethod, channel, Metadata.Empty); + Assert.AreEqual("ABC", Calls.BlockingUnaryCall(call, "ABC", CancellationToken.None)); } [Test] - public void CallOnDisposedChannel() + public void UnaryCall_ServerHandlerThrows() { - var server = new Server(); - server.AddServiceDefinition(ServiceDefinition); - int port = server.AddListeningPort(Host + ":0"); - server.Start(); - - Channel channel = new Channel(Host + ":" + port); - channel.Dispose(); - - var call = new Call(ServiceName, UnaryEchoStringMethod, channel, Metadata.Empty); + var call = new Call(ServiceName, EchoMethod, channel, Metadata.Empty); try { - Calls.BlockingUnaryCall(call, "ABC", default(CancellationToken)); - Assert.Fail(); + Calls.BlockingUnaryCall(call, "THROW", CancellationToken.None); + Assert.Fail(); } - catch (ObjectDisposedException e) + catch (RpcException e) { + Assert.AreEqual(StatusCode.Unknown, e.Status.StatusCode); } + } - server.ShutdownAsync().Wait(); + [Test] + public void AsyncUnaryCall() + { + var call = new Call(ServiceName, EchoMethod, channel, Metadata.Empty); + var result = Calls.AsyncUnaryCall(call, "ABC", CancellationToken.None).Result; + Assert.AreEqual("ABC", result); } [Test] - public void UnaryCallPerformance() + public void AsyncUnaryCall_ServerHandlerThrows() { - var server = new Server(); - server.AddServiceDefinition(ServiceDefinition); - int port = server.AddListeningPort(Host + ":0"); - server.Start(); + Task.Run(async () => + { + var call = new Call(ServiceName, EchoMethod, channel, Metadata.Empty); + try + { + await Calls.AsyncUnaryCall(call, "THROW", CancellationToken.None); + Assert.Fail(); + } + catch (RpcException e) + { + Assert.AreEqual(StatusCode.Unknown, e.Status.StatusCode); + } + }).Wait(); + } - using (Channel channel = new Channel(Host + ":" + port)) + [Test] + public void ClientStreamingCall() + { + Task.Run(async () => { - var call = new Call(ServiceName, UnaryEchoStringMethod, channel, Metadata.Empty); - BenchmarkUtil.RunBenchmark(100, 100, - () => { Calls.BlockingUnaryCall(call, "ABC", default(CancellationToken)); }); - } + var call = new Call(ServiceName, ConcatAndEchoMethod, channel, Metadata.Empty); + var callResult = Calls.AsyncClientStreamingCall(call, CancellationToken.None); - server.ShutdownAsync().Wait(); + await callResult.RequestStream.WriteAll(new string[] { "A", "B", "C" }); + Assert.AreEqual("ABC", await callResult.Result); + }).Wait(); } [Test] - public void UnknownMethodHandler() + public void ClientStreamingCall_ServerHandlerThrows() { - var server = new Server(); - server.AddServiceDefinition(ServerServiceDefinition.CreateBuilder(ServiceName).Build()); - int port = server.AddListeningPort(Host + ":0"); - server.Start(); + Task.Run(async () => + { + var call = new Call(ServiceName, ConcatAndEchoMethod, channel, Metadata.Empty); + var callResult = Calls.AsyncClientStreamingCall(call, CancellationToken.None); + // TODO(jtattermusch): if we send "A", "THROW", "C", server hangs. + await callResult.RequestStream.WriteAll(new string[] { "A", "B", "THROW" }); + + try + { + await callResult.Result; + } + catch(RpcException e) + { + Assert.AreEqual(StatusCode.Unknown, e.Status.StatusCode); + } + }).Wait(); + } - using (Channel channel = new Channel(Host + ":" + port)) + [Test] + public void ClientStreamingCall_CancelAfterBegin() + { + Task.Run(async () => { - var call = new Call(ServiceName, UnaryEchoStringMethod, channel, Metadata.Empty); + var call = new Call(ServiceName, ConcatAndEchoMethod, channel, Metadata.Empty); + + var cts = new CancellationTokenSource(); + var callResult = Calls.AsyncClientStreamingCall(call, cts.Token); + cts.Cancel(); + try { - Calls.BlockingUnaryCall(call, "ABC", default(CancellationToken)); - Assert.Fail(); + await callResult.Result; } - catch (RpcException e) + catch(RpcException e) { - Assert.AreEqual(StatusCode.Unimplemented, e.Status.StatusCode); + Assert.AreEqual(StatusCode.Cancelled, e.Status.StatusCode); } + }).Wait(); + } + + [Test] + public void UnaryCall_DisposedChannel() + { + channel.Dispose(); + + var call = new Call(ServiceName, EchoMethod, channel, Metadata.Empty); + Assert.Throws(typeof(ObjectDisposedException), () => Calls.BlockingUnaryCall(call, "ABC", CancellationToken.None)); + } + + [Test] + public void UnaryCallPerformance() + { + var call = new Call(ServiceName, EchoMethod, channel, Metadata.Empty); + BenchmarkUtil.RunBenchmark(100, 100, + () => { Calls.BlockingUnaryCall(call, "ABC", default(CancellationToken)); }); + } + + [Test] + public void UnknownMethodHandler() + { + var call = new Call(ServiceName, NonexistentMethod, channel, Metadata.Empty); + try + { + Calls.BlockingUnaryCall(call, "ABC", default(CancellationToken)); + Assert.Fail(); + } + catch (RpcException e) + { + Assert.AreEqual(StatusCode.Unimplemented, e.Status.StatusCode); } + } - server.ShutdownAsync().Wait(); + private static async Task EchoHandler(string request) + { + if (request == "THROW") + { + throw new Exception("This was thrown on purpose by a test"); + } + return request; } - /// - /// Handler for unaryEchoString method. - /// - private static Task HandleUnaryEchoString(string request) + private static async Task ConcatAndEchoHandler(IAsyncStreamReader requestStream) { - return Task.FromResult(request); + string result = ""; + await requestStream.ForEach(async (request) => + { + if (request == "THROW") + { + throw new Exception("This was thrown on purpose by a test"); + } + result += request; + }); + return result; } } } diff --git a/src/csharp/Grpc.Core/Calls.cs b/src/csharp/Grpc.Core/Calls.cs index 9365ccd9fb..c2397290fd 100644 --- a/src/csharp/Grpc.Core/Calls.cs +++ b/src/csharp/Grpc.Core/Calls.cs @@ -46,6 +46,8 @@ namespace Grpc.Core public static TResponse BlockingUnaryCall(Call call, TRequest req, CancellationToken token) { var asyncCall = new AsyncCall(call.RequestMarshaller.Serializer, call.ResponseMarshaller.Deserializer); + // TODO(jtattermusch): this gives a race that cancellation can be requested before the call even starts. + RegisterCancellationCallback(asyncCall, token); return asyncCall.UnaryCall(call.Channel, call.Name, req, call.Headers); } @@ -53,7 +55,9 @@ namespace Grpc.Core { var asyncCall = new AsyncCall(call.RequestMarshaller.Serializer, call.ResponseMarshaller.Deserializer); asyncCall.Initialize(call.Channel, GetCompletionQueue(), call.Name); - return await asyncCall.UnaryCallAsync(req, call.Headers); + var asyncResult = asyncCall.UnaryCallAsync(req, call.Headers); + RegisterCancellationCallback(asyncCall, token); + return await asyncResult; } public static AsyncServerStreamingCall AsyncServerStreamingCall(Call call, TRequest req, CancellationToken token) @@ -61,6 +65,7 @@ namespace Grpc.Core var asyncCall = new AsyncCall(call.RequestMarshaller.Serializer, call.ResponseMarshaller.Deserializer); asyncCall.Initialize(call.Channel, GetCompletionQueue(), call.Name); asyncCall.StartServerStreamingCall(req, call.Headers); + RegisterCancellationCallback(asyncCall, token); var responseStream = new ClientResponseStream(asyncCall); return new AsyncServerStreamingCall(responseStream); } @@ -70,6 +75,7 @@ namespace Grpc.Core var asyncCall = new AsyncCall(call.RequestMarshaller.Serializer, call.ResponseMarshaller.Deserializer); asyncCall.Initialize(call.Channel, GetCompletionQueue(), call.Name); var resultTask = asyncCall.ClientStreamingCallAsync(call.Headers); + RegisterCancellationCallback(asyncCall, token); var requestStream = new ClientRequestStream(asyncCall); return new AsyncClientStreamingCall(requestStream, resultTask); } @@ -79,11 +85,20 @@ namespace Grpc.Core var asyncCall = new AsyncCall(call.RequestMarshaller.Serializer, call.ResponseMarshaller.Deserializer); asyncCall.Initialize(call.Channel, GetCompletionQueue(), call.Name); asyncCall.StartDuplexStreamingCall(call.Headers); + RegisterCancellationCallback(asyncCall, token); var requestStream = new ClientRequestStream(asyncCall); var responseStream = new ClientResponseStream(asyncCall); return new AsyncDuplexStreamingCall(requestStream, responseStream); } + private static void RegisterCancellationCallback(AsyncCall asyncCall, CancellationToken token) + { + if (token.CanBeCanceled) + { + token.Register( () => asyncCall.Cancel() ); + } + } + /// /// Gets shared completion queue used for async calls. /// diff --git a/src/csharp/Grpc.Core/Internal/AsyncCallServer.cs b/src/csharp/Grpc.Core/Internal/AsyncCallServer.cs index 25dc15bbc0..449009336f 100644 --- a/src/csharp/Grpc.Core/Internal/AsyncCallServer.cs +++ b/src/csharp/Grpc.Core/Internal/AsyncCallServer.cs @@ -121,6 +121,12 @@ namespace Grpc.Core.Internal { finished = true; + if (readCompletionDelegate == null) + { + // allow disposal of native call + readingDone = true; + } + ReleaseResourcesIfPossible(); } // TODO: handle error ... diff --git a/src/csharp/Grpc.Core/Internal/ServerCallHandler.cs b/src/csharp/Grpc.Core/Internal/ServerCallHandler.cs index 2ae924b4a8..0416eada34 100644 --- a/src/csharp/Grpc.Core/Internal/ServerCallHandler.cs +++ b/src/csharp/Grpc.Core/Internal/ServerCallHandler.cs @@ -66,13 +66,21 @@ namespace Grpc.Core.Internal var requestStream = new ServerRequestStream(asyncCall); var responseStream = new ServerResponseStream(asyncCall); - var request = await requestStream.ReadNext(); - // TODO(jtattermusch): we need to read the full stream so that native callhandle gets deallocated. - Preconditions.CheckArgument(await requestStream.ReadNext() == null); - - var result = await handler(request); - await responseStream.Write(result); - await responseStream.WriteStatus(Status.DefaultSuccess); + Status status = Status.DefaultSuccess; + try + { + var request = await requestStream.ReadNext(); + // TODO(jtattermusch): we need to read the full stream so that native callhandle gets deallocated. + Preconditions.CheckArgument(await requestStream.ReadNext() == null); + var result = await handler(request); + await responseStream.Write(result); + } + catch (Exception e) + { + Console.WriteLine("Exception occured in handler: " + e); + status = HandlerUtils.StatusFromException(e); + } + await responseStream.WriteStatus(status); await finishedTask; } } @@ -99,12 +107,21 @@ namespace Grpc.Core.Internal var requestStream = new ServerRequestStream(asyncCall); var responseStream = new ServerResponseStream(asyncCall); - var request = await requestStream.ReadNext(); - // TODO(jtattermusch): we need to read the full stream so that native callhandle gets deallocated. - Preconditions.CheckArgument(await requestStream.ReadNext() == null); - - await handler(request, responseStream); - await responseStream.WriteStatus(Status.DefaultSuccess); + Status status = Status.DefaultSuccess; + try + { + var request = await requestStream.ReadNext(); + // TODO(jtattermusch): we need to read the full stream so that native callhandle gets deallocated. + Preconditions.CheckArgument(await requestStream.ReadNext() == null); + + await handler(request, responseStream); + } + catch (Exception e) + { + Console.WriteLine("Exception occured in handler: " + e); + status = HandlerUtils.StatusFromException(e); + } + await responseStream.WriteStatus(status); await finishedTask; } } @@ -131,9 +148,18 @@ namespace Grpc.Core.Internal var requestStream = new ServerRequestStream(asyncCall); var responseStream = new ServerResponseStream(asyncCall); - var result = await handler(requestStream); - await responseStream.Write(result); - await responseStream.WriteStatus(Status.DefaultSuccess); + Status status = Status.DefaultSuccess; + try + { + var result = await handler(requestStream); + await responseStream.Write(result); + } + catch (Exception e) + { + Console.WriteLine("Exception occured in handler: " + e); + status = HandlerUtils.StatusFromException(e); + } + await responseStream.WriteStatus(status); await finishedTask; } } @@ -160,8 +186,17 @@ namespace Grpc.Core.Internal var requestStream = new ServerRequestStream(asyncCall); var responseStream = new ServerResponseStream(asyncCall); - await handler(requestStream, responseStream); - await responseStream.WriteStatus(Status.DefaultSuccess); + Status status = Status.DefaultSuccess; + try + { + await handler(requestStream, responseStream); + } + catch (Exception e) + { + Console.WriteLine("Exception occured in handler: " + e); + status = HandlerUtils.StatusFromException(e); + } + await responseStream.WriteStatus(status); await finishedTask; } } @@ -173,15 +208,25 @@ namespace Grpc.Core.Internal // We don't care about the payload type here. var asyncCall = new AsyncCallServer( (payload) => payload, (payload) => payload); - + asyncCall.Initialize(call); var finishedTask = asyncCall.ServerSideCallAsync(); var requestStream = new ServerRequestStream(asyncCall); var responseStream = new ServerResponseStream(asyncCall); + await responseStream.WriteStatus(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; } } + + internal static class HandlerUtils + { + public static Status StatusFromException(Exception e) + { + // TODO(jtattermusch): what is the right status code here? + return new Status(StatusCode.Unknown, "Exception was thrown by handler."); + } + } } diff --git a/src/csharp/Grpc.IntegrationTesting/InteropClient.cs b/src/csharp/Grpc.IntegrationTesting/InteropClient.cs index 573ab30452..440702d06f 100644 --- a/src/csharp/Grpc.IntegrationTesting/InteropClient.cs +++ b/src/csharp/Grpc.IntegrationTesting/InteropClient.cs @@ -34,6 +34,7 @@ using System; using System.Collections.Generic; using System.Text.RegularExpressions; +using System.Threading; using System.Threading.Tasks; using Google.ProtocolBuffers; @@ -166,6 +167,12 @@ namespace Grpc.IntegrationTesting case "compute_engine_creds": RunComputeEngineCreds(client); break; + case "cancel_after_begin": + RunCancelAfterBegin(client); + break; + case "cancel_after_first_response": + RunCancelAfterFirstResponse(client); + break; case "benchmark_empty_unary": RunBenchmarkEmptyUnary(client); break; @@ -351,6 +358,64 @@ namespace Grpc.IntegrationTesting Console.WriteLine("Passed!"); } + public static void RunCancelAfterBegin(TestServiceGrpc.ITestServiceClient client) + { + Task.Run(async () => + { + Console.WriteLine("running cancel_after_begin"); + + var cts = new CancellationTokenSource(); + var call = client.StreamingInputCall(cts.Token); + cts.Cancel(); + + try + { + var response = await call.Result; + Assert.Fail(); + } + catch (RpcException e) + { + Assert.AreEqual(StatusCode.Cancelled, e.Status.StatusCode); + } + Console.WriteLine("Passed!"); + }).Wait(); + } + + public static void RunCancelAfterFirstResponse(TestServiceGrpc.ITestServiceClient client) + { + Task.Run(async () => + { + Console.WriteLine("running cancel_after_first_response"); + + var cts = new CancellationTokenSource(); + var call = client.FullDuplexCall(cts.Token); + + StreamingOutputCallResponse response; + + await call.RequestStream.Write(StreamingOutputCallRequest.CreateBuilder() + .SetResponseType(PayloadType.COMPRESSABLE) + .AddResponseParameters(ResponseParameters.CreateBuilder().SetSize(31415)) + .SetPayload(CreateZerosPayload(27182)).Build()); + + response = await call.ResponseStream.ReadNext(); + Assert.AreEqual(PayloadType.COMPRESSABLE, response.Payload.Type); + Assert.AreEqual(31415, response.Payload.Body.Length); + + cts.Cancel(); + + try + { + response = await call.ResponseStream.ReadNext(); + Assert.Fail(); + } + catch (RpcException e) + { + Assert.AreEqual(StatusCode.Cancelled, e.Status.StatusCode); + } + Console.WriteLine("Passed!"); + }).Wait(); + } + // This is not an official interop test, but it's useful. public static void RunBenchmarkEmptyUnary(TestServiceGrpc.ITestServiceClient client) { diff --git a/src/csharp/Grpc.IntegrationTesting/InteropClientServerTest.cs b/src/csharp/Grpc.IntegrationTesting/InteropClientServerTest.cs index e929b76b5e..45380227c2 100644 --- a/src/csharp/Grpc.IntegrationTesting/InteropClientServerTest.cs +++ b/src/csharp/Grpc.IntegrationTesting/InteropClientServerTest.cs @@ -114,8 +114,16 @@ namespace Grpc.IntegrationTesting InteropClient.RunEmptyStream(client); } - // TODO: add cancel_after_begin + [Test] + public void CancelAfterBegin() + { + InteropClient.RunCancelAfterBegin(client); + } - // TODO: add cancel_after_first_response + [Test] + public void CancelAfterFirstResponse() + { + InteropClient.RunCancelAfterFirstResponse(client); + } } } -- cgit v1.2.3 From 618647dc7402c65631573d67b1500114ffb74197 Mon Sep 17 00:00:00 2001 From: Jan Tattermusch Date: Fri, 1 May 2015 11:23:28 -0700 Subject: fixed some stylecop warnings --- src/csharp/Grpc.Core.Tests/ClientServerTest.cs | 4 ++-- src/csharp/Grpc.Core/Calls.cs | 2 +- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/src/csharp/Grpc.Core.Tests/ClientServerTest.cs b/src/csharp/Grpc.Core.Tests/ClientServerTest.cs index c91efe53b4..8c4b92fbad 100644 --- a/src/csharp/Grpc.Core.Tests/ClientServerTest.cs +++ b/src/csharp/Grpc.Core.Tests/ClientServerTest.cs @@ -179,7 +179,7 @@ namespace Grpc.Core.Tests { await callResult.Result; } - catch(RpcException e) + catch (RpcException e) { Assert.AreEqual(StatusCode.Unknown, e.Status.StatusCode); } @@ -201,7 +201,7 @@ namespace Grpc.Core.Tests { await callResult.Result; } - catch(RpcException e) + catch (RpcException e) { Assert.AreEqual(StatusCode.Cancelled, e.Status.StatusCode); } diff --git a/src/csharp/Grpc.Core/Calls.cs b/src/csharp/Grpc.Core/Calls.cs index c2397290fd..a8d2b9498e 100644 --- a/src/csharp/Grpc.Core/Calls.cs +++ b/src/csharp/Grpc.Core/Calls.cs @@ -95,7 +95,7 @@ namespace Grpc.Core { if (token.CanBeCanceled) { - token.Register( () => asyncCall.Cancel() ); + token.Register(() => asyncCall.Cancel()); } } -- cgit v1.2.3 From 1b54fcf31b32ae8c7f07ae733e781c184791a7c2 Mon Sep 17 00:00:00 2001 From: Jan Tattermusch Date: Fri, 1 May 2015 14:30:16 -0700 Subject: added stats with number of active native calls, useful for debugging --- src/csharp/Grpc.Core/Grpc.Core.csproj | 4 +- src/csharp/Grpc.Core/GrpcEnvironment.cs | 16 +++++++ src/csharp/Grpc.Core/Internal/AsyncCall.cs | 6 +++ src/csharp/Grpc.Core/Internal/AsyncCallBase.cs | 5 ++ src/csharp/Grpc.Core/Internal/AsyncCallServer.cs | 6 +++ src/csharp/Grpc.Core/Internal/AtomicCounter.cs | 61 ++++++++++++++++++++++++ src/csharp/Grpc.Core/Internal/DebugStats.cs | 45 +++++++++++++++++ 7 files changed, 142 insertions(+), 1 deletion(-) create mode 100644 src/csharp/Grpc.Core/Internal/AtomicCounter.cs create mode 100644 src/csharp/Grpc.Core/Internal/DebugStats.cs diff --git a/src/csharp/Grpc.Core/Grpc.Core.csproj b/src/csharp/Grpc.Core/Grpc.Core.csproj index fee742b220..9c91541d90 100644 --- a/src/csharp/Grpc.Core/Grpc.Core.csproj +++ b/src/csharp/Grpc.Core/Grpc.Core.csproj @@ -5,7 +5,7 @@ Debug AnyCPU - 10.0.0 + 8.0.30703 2.0 {CCC4440E-49F7-4790-B0AF-FEABB0837AE7} Library @@ -94,6 +94,8 @@ + + diff --git a/src/csharp/Grpc.Core/GrpcEnvironment.cs b/src/csharp/Grpc.Core/GrpcEnvironment.cs index 9c10a42e23..2e9e5a2ef6 100644 --- a/src/csharp/Grpc.Core/GrpcEnvironment.cs +++ b/src/csharp/Grpc.Core/GrpcEnvironment.cs @@ -86,6 +86,8 @@ namespace Grpc.Core { instance.Close(); instance = null; + + CheckDebugStats(); } } } @@ -132,5 +134,19 @@ namespace Grpc.Core // TODO: use proper logging here Console.WriteLine("GRPC shutdown."); } + + private static void CheckDebugStats() + { + var remainingClientCalls = DebugStats.ActiveClientCalls.Count; + if (remainingClientCalls != 0) + { + Console.WriteLine("Warning: Detected {0} client calls that weren't disposed properly.", remainingClientCalls); + } + var remainingServerCalls = DebugStats.ActiveServerCalls.Count; + if (remainingServerCalls != 0) + { + Console.WriteLine("Warning: Detected {0} server calls that weren't disposed properly.", remainingServerCalls); + } + } } } diff --git a/src/csharp/Grpc.Core/Internal/AsyncCall.cs b/src/csharp/Grpc.Core/Internal/AsyncCall.cs index fd94771ddd..3532f7347a 100644 --- a/src/csharp/Grpc.Core/Internal/AsyncCall.cs +++ b/src/csharp/Grpc.Core/Internal/AsyncCall.cs @@ -67,6 +67,7 @@ namespace Grpc.Core.Internal public void Initialize(Channel channel, CompletionQueueSafeHandle cq, string methodName) { var call = CallSafeHandle.Create(channel.Handle, cq, methodName, channel.Target, Timespec.InfFuture); + DebugStats.ActiveClientCalls.Increment(); InitializeInternal(call); } @@ -265,6 +266,11 @@ namespace Grpc.Core.Internal } } + protected override void OnReleaseResources() + { + DebugStats.ActiveClientCalls.Decrement(); + } + /// /// Handler for unary response completion. /// diff --git a/src/csharp/Grpc.Core/Internal/AsyncCallBase.cs b/src/csharp/Grpc.Core/Internal/AsyncCallBase.cs index 2bde4b3720..b911cdcc87 100644 --- a/src/csharp/Grpc.Core/Internal/AsyncCallBase.cs +++ b/src/csharp/Grpc.Core/Internal/AsyncCallBase.cs @@ -191,6 +191,7 @@ namespace Grpc.Core.Internal private void ReleaseResources() { + OnReleaseResources(); if (call != null) { call.Dispose(); @@ -199,6 +200,10 @@ namespace Grpc.Core.Internal disposed = true; } + protected virtual void OnReleaseResources() + { + } + protected void CheckSendingAllowed() { Preconditions.CheckState(started); diff --git a/src/csharp/Grpc.Core/Internal/AsyncCallServer.cs b/src/csharp/Grpc.Core/Internal/AsyncCallServer.cs index 449009336f..4775f2d07b 100644 --- a/src/csharp/Grpc.Core/Internal/AsyncCallServer.cs +++ b/src/csharp/Grpc.Core/Internal/AsyncCallServer.cs @@ -57,6 +57,7 @@ namespace Grpc.Core.Internal public void Initialize(CallSafeHandle call) { + DebugStats.ActiveServerCalls.Increment(); InitializeInternal(call); } @@ -112,6 +113,11 @@ namespace Grpc.Core.Internal } } + protected override void OnReleaseResources() + { + DebugStats.ActiveServerCalls.Decrement(); + } + /// /// Handles the server side close completion. /// diff --git a/src/csharp/Grpc.Core/Internal/AtomicCounter.cs b/src/csharp/Grpc.Core/Internal/AtomicCounter.cs new file mode 100644 index 0000000000..7ccda225dc --- /dev/null +++ b/src/csharp/Grpc.Core/Internal/AtomicCounter.cs @@ -0,0 +1,61 @@ +#region Copyright notice and license + +// Copyright 2015, Google Inc. +// All rights reserved. +// +// Redistribution and use in source and binary forms, with or without +// modification, are permitted provided that the following conditions are +// met: +// +// * Redistributions of source code must retain the above copyright +// notice, this list of conditions and the following disclaimer. +// * Redistributions in binary form must reproduce the above +// copyright notice, this list of conditions and the following disclaimer +// in the documentation and/or other materials provided with the +// distribution. +// * Neither the name of Google Inc. nor the names of its +// contributors may be used to endorse or promote products derived from +// this software without specific prior written permission. +// +// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS +// "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT +// LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR +// A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT +// OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, +// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT +// LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, +// DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY +// THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT +// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE +// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + +#endregion + +using System; +using System.Threading; + +namespace Grpc.Core.Internal +{ + internal class AtomicCounter + { + long counter = 0; + + public void Increment() + { + Interlocked.Increment(ref counter); + } + + public void Decrement() + { + Interlocked.Decrement(ref counter); + } + + public long Count + { + get + { + return counter; + } + } + } +} diff --git a/src/csharp/Grpc.Core/Internal/DebugStats.cs b/src/csharp/Grpc.Core/Internal/DebugStats.cs new file mode 100644 index 0000000000..476914f751 --- /dev/null +++ b/src/csharp/Grpc.Core/Internal/DebugStats.cs @@ -0,0 +1,45 @@ +#region Copyright notice and license + +// Copyright 2015, Google Inc. +// All rights reserved. +// +// Redistribution and use in source and binary forms, with or without +// modification, are permitted provided that the following conditions are +// met: +// +// * Redistributions of source code must retain the above copyright +// notice, this list of conditions and the following disclaimer. +// * Redistributions in binary form must reproduce the above +// copyright notice, this list of conditions and the following disclaimer +// in the documentation and/or other materials provided with the +// distribution. +// * Neither the name of Google Inc. nor the names of its +// contributors may be used to endorse or promote products derived from +// this software without specific prior written permission. +// +// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS +// "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT +// LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR +// A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT +// OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, +// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT +// LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, +// DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY +// THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT +// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE +// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + +#endregion + +using System; +using System.Threading; + +namespace Grpc.Core.Internal +{ + internal static class DebugStats + { + public static readonly AtomicCounter ActiveClientCalls = new AtomicCounter(); + + public static readonly AtomicCounter ActiveServerCalls = new AtomicCounter(); + } +} -- cgit v1.2.3 From 8c2dd9d864cb874f8fbe577faf8c3f72e6a077e4 Mon Sep 17 00:00:00 2001 From: Jan Tattermusch Date: Mon, 4 May 2015 09:20:43 -0700 Subject: Fixes for C# cancellation support --- src/csharp/Grpc.Core.Tests/ClientServerTest.cs | 28 +++--------- src/csharp/Grpc.Core/Internal/AsyncCallBase.cs | 17 ++++++-- src/csharp/Grpc.Core/Internal/AsyncCallServer.cs | 11 +++-- .../Internal/BatchContextSafeHandleNotOwned.cs | 8 ++++ src/csharp/Grpc.Core/Internal/ServerCallHandler.cs | 51 +++++++++++++++++++--- src/csharp/Grpc.Core/Status.cs | 5 +++ .../Grpc.IntegrationTesting/InteropClient.cs | 2 + src/csharp/ext/grpc_csharp_ext.c | 6 +++ 8 files changed, 93 insertions(+), 35 deletions(-) diff --git a/src/csharp/Grpc.Core.Tests/ClientServerTest.cs b/src/csharp/Grpc.Core.Tests/ClientServerTest.cs index 8c4b92fbad..26a1a683ba 100644 --- a/src/csharp/Grpc.Core.Tests/ClientServerTest.cs +++ b/src/csharp/Grpc.Core.Tests/ClientServerTest.cs @@ -165,27 +165,6 @@ namespace Grpc.Core.Tests }).Wait(); } - [Test] - public void ClientStreamingCall_ServerHandlerThrows() - { - Task.Run(async () => - { - var call = new Call(ServiceName, ConcatAndEchoMethod, channel, Metadata.Empty); - var callResult = Calls.AsyncClientStreamingCall(call, CancellationToken.None); - // TODO(jtattermusch): if we send "A", "THROW", "C", server hangs. - await callResult.RequestStream.WriteAll(new string[] { "A", "B", "THROW" }); - - try - { - await callResult.Result; - } - catch (RpcException e) - { - Assert.AreEqual(StatusCode.Unknown, e.Status.StatusCode); - } - }).Wait(); - } - [Test] public void ClientStreamingCall_CancelAfterBegin() { @@ -195,6 +174,9 @@ namespace Grpc.Core.Tests var cts = new CancellationTokenSource(); var callResult = Calls.AsyncClientStreamingCall(call, cts.Token); + + // TODO(jtattermusch): we need this to ensure call has been initiated once we cancel it. + await Task.Delay(1000); cts.Cancel(); try @@ -260,7 +242,9 @@ namespace Grpc.Core.Tests } result += request; }); - return result; + // simulate processing takes some time. + await Task.Delay(250); + return result; } } } diff --git a/src/csharp/Grpc.Core/Internal/AsyncCallBase.cs b/src/csharp/Grpc.Core/Internal/AsyncCallBase.cs index b911cdcc87..7cf0f6ff84 100644 --- a/src/csharp/Grpc.Core/Internal/AsyncCallBase.cs +++ b/src/csharp/Grpc.Core/Internal/AsyncCallBase.cs @@ -180,7 +180,8 @@ namespace Grpc.Core.Internal { if (!disposed && call != null) { - if (halfclosed && readingDone && finished) + bool noMoreSendCompletions = halfclosed || (cancelRequested && sendCompletionDelegate == null); + if (noMoreSendCompletions && readingDone && finished) { ReleaseResources(); return true; @@ -207,8 +208,9 @@ namespace Grpc.Core.Internal protected void CheckSendingAllowed() { Preconditions.CheckState(started); - Preconditions.CheckState(!disposed); Preconditions.CheckState(!errorOccured); + CheckNotCancelled(); + Preconditions.CheckState(!disposed); Preconditions.CheckState(!halfcloseRequested, "Already halfclosed."); Preconditions.CheckState(sendCompletionDelegate == null, "Only one write can be pending at a time"); @@ -221,7 +223,14 @@ namespace Grpc.Core.Internal Preconditions.CheckState(!errorOccured); Preconditions.CheckState(!readingDone, "Stream has already been closed."); - Preconditions.CheckState(readCompletionDelegate == null, "Only one write can be pending at a time"); + Preconditions.CheckState(readCompletionDelegate == null, "Only one read can be pending at a time"); + } + + protected void CheckNotCancelled() { + if (cancelRequested) + { + throw new OperationCanceledException("Remote call has been cancelled."); + } } protected byte[] UnsafeSerialize(TWrite msg) @@ -292,6 +301,8 @@ namespace Grpc.Core.Internal }); } + + /// /// Handles send completion. /// diff --git a/src/csharp/Grpc.Core/Internal/AsyncCallServer.cs b/src/csharp/Grpc.Core/Internal/AsyncCallServer.cs index 4775f2d07b..3c66c67dcc 100644 --- a/src/csharp/Grpc.Core/Internal/AsyncCallServer.cs +++ b/src/csharp/Grpc.Core/Internal/AsyncCallServer.cs @@ -123,18 +123,23 @@ namespace Grpc.Core.Internal /// private void HandleFinishedServerside(bool wasError, BatchContextSafeHandleNotOwned ctx) { + bool cancelled = ctx.GetReceivedCloseOnServerCancelled(); + lock (myLock) { finished = true; - if (readCompletionDelegate == null) + if (cancelled) { - // allow disposal of native call - readingDone = true; + // Once we cancel, we don't have to care that much + // about reads and writes. + Cancel(); } ReleaseResourcesIfPossible(); } + // TODO(jtattermusch): check if call was cancelled. + // TODO: handle error ... finishedServersideTcs.SetResult(null); diff --git a/src/csharp/Grpc.Core/Internal/BatchContextSafeHandleNotOwned.cs b/src/csharp/Grpc.Core/Internal/BatchContextSafeHandleNotOwned.cs index 3c54753756..b562abaa7a 100644 --- a/src/csharp/Grpc.Core/Internal/BatchContextSafeHandleNotOwned.cs +++ b/src/csharp/Grpc.Core/Internal/BatchContextSafeHandleNotOwned.cs @@ -61,6 +61,9 @@ namespace Grpc.Core.Internal [DllImport("grpc_csharp_ext.dll")] static extern IntPtr grpcsharp_batch_context_server_rpc_new_method(BatchContextSafeHandleNotOwned ctx); // returns const char* + [DllImport("grpc_csharp_ext.dll")] + static extern int grpcsharp_batch_context_recv_close_on_server_cancelled(BatchContextSafeHandleNotOwned ctx); + public BatchContextSafeHandleNotOwned(IntPtr handle) : base(false) { SetHandle(handle); @@ -94,5 +97,10 @@ namespace Grpc.Core.Internal { return Marshal.PtrToStringAnsi(grpcsharp_batch_context_server_rpc_new_method(this)); } + + public bool GetReceivedCloseOnServerCancelled() + { + return grpcsharp_batch_context_recv_close_on_server_cancelled(this) != 0; + } } } \ No newline at end of file diff --git a/src/csharp/Grpc.Core/Internal/ServerCallHandler.cs b/src/csharp/Grpc.Core/Internal/ServerCallHandler.cs index 0416eada34..01b2a11369 100644 --- a/src/csharp/Grpc.Core/Internal/ServerCallHandler.cs +++ b/src/csharp/Grpc.Core/Internal/ServerCallHandler.cs @@ -80,7 +80,14 @@ namespace Grpc.Core.Internal Console.WriteLine("Exception occured in handler: " + e); status = HandlerUtils.StatusFromException(e); } - await responseStream.WriteStatus(status); + try + { + await responseStream.WriteStatus(status); + } + catch (OperationCanceledException) + { + // Call has been already cancelled. + } await finishedTask; } } @@ -121,7 +128,15 @@ namespace Grpc.Core.Internal Console.WriteLine("Exception occured in handler: " + e); status = HandlerUtils.StatusFromException(e); } - await responseStream.WriteStatus(status); + + try + { + await responseStream.WriteStatus(status); + } + catch (OperationCanceledException) + { + // Call has been already cancelled. + } await finishedTask; } } @@ -151,15 +166,30 @@ namespace Grpc.Core.Internal Status status = Status.DefaultSuccess; try { - var result = await handler(requestStream); - await responseStream.Write(result); - } + var result = await handler(requestStream); + try + { + await responseStream.Write(result); + } + catch (OperationCanceledException) + { + status = Status.DefaultCancelled; + } + } catch (Exception e) { Console.WriteLine("Exception occured in handler: " + e); status = HandlerUtils.StatusFromException(e); } - await responseStream.WriteStatus(status); + + try + { + await responseStream.WriteStatus(status); + } + catch (OperationCanceledException) + { + // Call has been already cancelled. + } await finishedTask; } } @@ -196,7 +226,14 @@ namespace Grpc.Core.Internal Console.WriteLine("Exception occured in handler: " + e); status = HandlerUtils.StatusFromException(e); } - await responseStream.WriteStatus(status); + try + { + await responseStream.WriteStatus(status); + } + catch (OperationCanceledException) + { + // Call has been already cancelled. + } await finishedTask; } } diff --git a/src/csharp/Grpc.Core/Status.cs b/src/csharp/Grpc.Core/Status.cs index b588170694..754f6cb3ca 100644 --- a/src/csharp/Grpc.Core/Status.cs +++ b/src/csharp/Grpc.Core/Status.cs @@ -44,6 +44,11 @@ namespace Grpc.Core /// public static readonly Status DefaultSuccess = new Status(StatusCode.OK, ""); + /// + /// Default result of a cancelled RPC. StatusCode=Cancelled, empty details message. + /// + public static readonly Status DefaultCancelled = new Status(StatusCode.Cancelled, ""); + readonly StatusCode statusCode; readonly string detail; diff --git a/src/csharp/Grpc.IntegrationTesting/InteropClient.cs b/src/csharp/Grpc.IntegrationTesting/InteropClient.cs index 440702d06f..a433659a08 100644 --- a/src/csharp/Grpc.IntegrationTesting/InteropClient.cs +++ b/src/csharp/Grpc.IntegrationTesting/InteropClient.cs @@ -366,6 +366,8 @@ namespace Grpc.IntegrationTesting 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 diff --git a/src/csharp/ext/grpc_csharp_ext.c b/src/csharp/ext/grpc_csharp_ext.c index fb8b75798d..a8cc1b29a4 100644 --- a/src/csharp/ext/grpc_csharp_ext.c +++ b/src/csharp/ext/grpc_csharp_ext.c @@ -277,6 +277,12 @@ grpcsharp_batch_context_server_rpc_new_method( return ctx->server_rpc_new.call_details.method; } +GPR_EXPORT gpr_int32 GPR_CALLTYPE +grpcsharp_batch_context_recv_close_on_server_cancelled( + const grpcsharp_batch_context *ctx) { + return (gpr_int32) ctx->recv_close_on_server_cancelled; +} + /* Init & shutdown */ GPR_EXPORT void GPR_CALLTYPE grpcsharp_init(void) { grpc_init(); } -- cgit v1.2.3 From 32d95b9744567c9ead050d154181fadbcd0c8ea7 Mon Sep 17 00:00:00 2001 From: Jan Tattermusch Date: Mon, 4 May 2015 17:56:32 -0700 Subject: remove duplicate initialize --- src/csharp/Grpc.Core.Tests/ClientServerTest.cs | 6 ++---- 1 file changed, 2 insertions(+), 4 deletions(-) diff --git a/src/csharp/Grpc.Core.Tests/ClientServerTest.cs b/src/csharp/Grpc.Core.Tests/ClientServerTest.cs index 26a1a683ba..caa6220f2c 100644 --- a/src/csharp/Grpc.Core.Tests/ClientServerTest.cs +++ b/src/csharp/Grpc.Core.Tests/ClientServerTest.cs @@ -82,8 +82,6 @@ namespace Grpc.Core.Tests [SetUp] public void Init() { - GrpcEnvironment.Initialize(); - server = new Server(); server.AddServiceDefinition(ServiceDefinition); int port = server.AddListeningPort(Host + ":0"); @@ -137,7 +135,7 @@ namespace Grpc.Core.Tests [Test] public void AsyncUnaryCall_ServerHandlerThrows() { - Task.Run(async () => + Task.Run(async () => { var call = new Call(ServiceName, EchoMethod, channel, Metadata.Empty); try @@ -147,7 +145,7 @@ namespace Grpc.Core.Tests } catch (RpcException e) { - Assert.AreEqual(StatusCode.Unknown, e.Status.StatusCode); + Assert.AreEqual(StatusCode.Unknown, e.Status.StatusCode); } }).Wait(); } -- cgit v1.2.3 From a8447711be717a53fc52a3a14dd33fdb85fc80f2 Mon Sep 17 00:00:00 2001 From: Jan Tattermusch Date: Mon, 4 May 2015 17:56:52 -0700 Subject: stylecop fixes --- src/csharp/Grpc.Core/Internal/AsyncCallBase.cs | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/src/csharp/Grpc.Core/Internal/AsyncCallBase.cs b/src/csharp/Grpc.Core/Internal/AsyncCallBase.cs index 7cf0f6ff84..fc5bee40e2 100644 --- a/src/csharp/Grpc.Core/Internal/AsyncCallBase.cs +++ b/src/csharp/Grpc.Core/Internal/AsyncCallBase.cs @@ -226,7 +226,8 @@ namespace Grpc.Core.Internal Preconditions.CheckState(readCompletionDelegate == null, "Only one read can be pending at a time"); } - protected void CheckNotCancelled() { + protected void CheckNotCancelled() + { if (cancelRequested) { throw new OperationCanceledException("Remote call has been cancelled."); @@ -301,8 +302,6 @@ namespace Grpc.Core.Internal }); } - - /// /// Handles send completion. /// -- cgit v1.2.3