diff --git a/src/csharp/Grpc.Core.Tests/ClientServerTest.cs b/src/csharp/Grpc.Core.Tests/ClientServerTest.cs index caa6220f2ccfecafc1140f1df21042535f4bb172..b69b933aba605f42851ee5916e1cf546992318b8 100644 --- a/src/csharp/Grpc.Core.Tests/ClientServerTest.cs +++ b/src/csharp/Grpc.Core.Tests/ClientServerTest.cs @@ -84,7 +84,7 @@ namespace Grpc.Core.Tests { server = new Server(); server.AddServiceDefinition(ServiceDefinition); - int port = server.AddListeningPort(Host + ":0"); + int port = server.AddListeningPort(Host, Server.PickUnusedPort); server.Start(); channel = new Channel(Host + ":" + port); } @@ -220,7 +220,7 @@ namespace Grpc.Core.Tests } } - private static async Task<string> EchoHandler(string request) + private static async Task<string> EchoHandler(ServerCallContext context, string request) { if (request == "THROW") { @@ -229,7 +229,7 @@ namespace Grpc.Core.Tests return request; } - private static async Task<string> ConcatAndEchoHandler(IAsyncStreamReader<string> requestStream) + private static async Task<string> ConcatAndEchoHandler(ServerCallContext context, IAsyncStreamReader<string> requestStream) { string result = ""; await requestStream.ForEach(async (request) => diff --git a/src/csharp/Grpc.Core.Tests/ServerTest.cs b/src/csharp/Grpc.Core.Tests/ServerTest.cs index 2a1855da675bdd26c810af0e13e3c1c8bfd01c64..02c773c9ccc625cd62f564426ddc12b03cb7fad6 100644 --- a/src/csharp/Grpc.Core.Tests/ServerTest.cs +++ b/src/csharp/Grpc.Core.Tests/ServerTest.cs @@ -47,7 +47,7 @@ namespace Grpc.Core.Tests GrpcEnvironment.Initialize(); Server server = new Server(); - server.AddListeningPort("localhost:0"); + server.AddListeningPort("localhost", Server.PickUnusedPort); server.Start(); server.ShutdownAsync().Wait(); diff --git a/src/csharp/Grpc.Core/AsyncClientStreamingCall.cs b/src/csharp/Grpc.Core/AsyncClientStreamingCall.cs index e81ce01ebbc969ab599a239ff56f8a155947b605..b95776f66d46bc143e54602d1f8452303be85b94 100644 --- a/src/csharp/Grpc.Core/AsyncClientStreamingCall.cs +++ b/src/csharp/Grpc.Core/AsyncClientStreamingCall.cs @@ -40,7 +40,9 @@ namespace Grpc.Core /// <summary> /// Return type for client streaming calls. /// </summary> - public struct AsyncClientStreamingCall<TRequest, TResponse> + public sealed class AsyncClientStreamingCall<TRequest, TResponse> + where TRequest : class + where TResponse : class { readonly IClientStreamWriter<TRequest> requestStream; readonly Task<TResponse> result; diff --git a/src/csharp/Grpc.Core/AsyncDuplexStreamingCall.cs b/src/csharp/Grpc.Core/AsyncDuplexStreamingCall.cs index 1cb30f47795aed61baaee934445183df0f607b02..ee0543741673b59a53b369be3f84da9f7e99c817 100644 --- a/src/csharp/Grpc.Core/AsyncDuplexStreamingCall.cs +++ b/src/csharp/Grpc.Core/AsyncDuplexStreamingCall.cs @@ -40,7 +40,9 @@ namespace Grpc.Core /// <summary> /// Return type for bidirectional streaming calls. /// </summary> - public struct AsyncDuplexStreamingCall<TRequest, TResponse> + public sealed class AsyncDuplexStreamingCall<TRequest, TResponse> + where TRequest : class + where TResponse : class { readonly IClientStreamWriter<TRequest> requestStream; readonly IAsyncStreamReader<TResponse> responseStream; diff --git a/src/csharp/Grpc.Core/AsyncServerStreamingCall.cs b/src/csharp/Grpc.Core/AsyncServerStreamingCall.cs index d614916fb7c1fe3c082bcabbe77b153aa77f0c10..73b96149850ff4e0e3203893e95ae56b71e29f73 100644 --- a/src/csharp/Grpc.Core/AsyncServerStreamingCall.cs +++ b/src/csharp/Grpc.Core/AsyncServerStreamingCall.cs @@ -40,7 +40,8 @@ namespace Grpc.Core /// <summary> /// Return type for server streaming calls. /// </summary> - public struct AsyncServerStreamingCall<TResponse> + public sealed class AsyncServerStreamingCall<TResponse> + where TResponse : class { readonly IAsyncStreamReader<TResponse> responseStream; diff --git a/src/csharp/Grpc.Core/Call.cs b/src/csharp/Grpc.Core/Call.cs index 070dfb569d6477da2fff25106f6606ba9ed87cd5..771cc083dae6d6b2fa1c5e1d6d3ad6c47a22076e 100644 --- a/src/csharp/Grpc.Core/Call.cs +++ b/src/csharp/Grpc.Core/Call.cs @@ -41,6 +41,8 @@ namespace Grpc.Core /// Abstraction of a call to be invoked on a client. /// </summary> public class Call<TRequest, TResponse> + where TRequest : class + where TResponse : class { readonly string name; readonly Marshaller<TRequest> requestMarshaller; diff --git a/src/csharp/Grpc.Core/Calls.cs b/src/csharp/Grpc.Core/Calls.cs index a8d2b9498e999af80d8242a9d42ba3f3b5cc9097..ba42a2d4f879918b76fda1c47392f14eeb79f8f8 100644 --- a/src/csharp/Grpc.Core/Calls.cs +++ b/src/csharp/Grpc.Core/Calls.cs @@ -44,6 +44,8 @@ namespace Grpc.Core public static class Calls { public static TResponse BlockingUnaryCall<TRequest, TResponse>(Call<TRequest, TResponse> call, TRequest req, CancellationToken token) + where TRequest : class + where TResponse : class { var asyncCall = new AsyncCall<TRequest, TResponse>(call.RequestMarshaller.Serializer, call.ResponseMarshaller.Deserializer); // TODO(jtattermusch): this gives a race that cancellation can be requested before the call even starts. @@ -52,6 +54,8 @@ namespace Grpc.Core } public static async Task<TResponse> AsyncUnaryCall<TRequest, TResponse>(Call<TRequest, TResponse> call, TRequest req, CancellationToken token) + where TRequest : class + where TResponse : class { var asyncCall = new AsyncCall<TRequest, TResponse>(call.RequestMarshaller.Serializer, call.ResponseMarshaller.Deserializer); asyncCall.Initialize(call.Channel, GetCompletionQueue(), call.Name); @@ -61,6 +65,8 @@ namespace Grpc.Core } public static AsyncServerStreamingCall<TResponse> AsyncServerStreamingCall<TRequest, TResponse>(Call<TRequest, TResponse> call, TRequest req, CancellationToken token) + where TRequest : class + where TResponse : class { var asyncCall = new AsyncCall<TRequest, TResponse>(call.RequestMarshaller.Serializer, call.ResponseMarshaller.Deserializer); asyncCall.Initialize(call.Channel, GetCompletionQueue(), call.Name); @@ -71,6 +77,8 @@ namespace Grpc.Core } public static AsyncClientStreamingCall<TRequest, TResponse> AsyncClientStreamingCall<TRequest, TResponse>(Call<TRequest, TResponse> call, CancellationToken token) + where TRequest : class + where TResponse : class { var asyncCall = new AsyncCall<TRequest, TResponse>(call.RequestMarshaller.Serializer, call.ResponseMarshaller.Deserializer); asyncCall.Initialize(call.Channel, GetCompletionQueue(), call.Name); @@ -81,6 +89,8 @@ namespace Grpc.Core } public static AsyncDuplexStreamingCall<TRequest, TResponse> AsyncDuplexStreamingCall<TRequest, TResponse>(Call<TRequest, TResponse> call, CancellationToken token) + where TRequest : class + where TResponse : class { var asyncCall = new AsyncCall<TRequest, TResponse>(call.RequestMarshaller.Serializer, call.ResponseMarshaller.Deserializer); asyncCall.Initialize(call.Channel, GetCompletionQueue(), call.Name); diff --git a/src/csharp/Grpc.Core/Grpc.Core.csproj b/src/csharp/Grpc.Core/Grpc.Core.csproj index 9c91541d904b3504bb143ced2566a1bb62f7697d..f5f2cf5f220e8f7dc44c38b3209832d456f2cb43 100644 --- a/src/csharp/Grpc.Core/Grpc.Core.csproj +++ b/src/csharp/Grpc.Core/Grpc.Core.csproj @@ -96,6 +96,7 @@ <Compile Include="Internal\ServerResponseStream.cs" /> <Compile Include="Internal\AtomicCounter.cs" /> <Compile Include="Internal\DebugStats.cs" /> + <Compile Include="ServerCallContext.cs" /> </ItemGroup> <ItemGroup> <None Include="packages.config" /> diff --git a/src/csharp/Grpc.Core/IAsyncStreamReader.cs b/src/csharp/Grpc.Core/IAsyncStreamReader.cs index 61cf57f7e0d3c4ad3f4f8a4892e88fc547c005a1..699741cd054420c48915615af10aaf629de060e9 100644 --- a/src/csharp/Grpc.Core/IAsyncStreamReader.cs +++ b/src/csharp/Grpc.Core/IAsyncStreamReader.cs @@ -44,9 +44,10 @@ namespace Grpc.Core /// </summary> /// <typeparam name="T"></typeparam> public interface IAsyncStreamReader<T> + where T : class { /// <summary> - /// Reads a single message. Returns default(T) if the last message was already read. + /// Reads a single message. Returns null if the last message was already read. /// A following read can only be started when the previous one finishes. /// </summary> Task<T> ReadNext(); diff --git a/src/csharp/Grpc.Core/IAsyncStreamWriter.cs b/src/csharp/Grpc.Core/IAsyncStreamWriter.cs index 724bae8f3130eb7a97115b271b558cbdc38f2932..4bd8bfb8df2ed829bf7743ff01359f95318e702a 100644 --- a/src/csharp/Grpc.Core/IAsyncStreamWriter.cs +++ b/src/csharp/Grpc.Core/IAsyncStreamWriter.cs @@ -44,6 +44,7 @@ namespace Grpc.Core /// </summary> /// <typeparam name="T"></typeparam> public interface IAsyncStreamWriter<T> + where T : class { /// <summary> /// Writes a single message. Only one write can be pending at a time. diff --git a/src/csharp/Grpc.Core/IClientStreamWriter.cs b/src/csharp/Grpc.Core/IClientStreamWriter.cs index 6da42e9ccc5c030a6eed980d4d962ad14f25d729..0847a928e6c185e2a6734eff5f8664d5ed69007d 100644 --- a/src/csharp/Grpc.Core/IClientStreamWriter.cs +++ b/src/csharp/Grpc.Core/IClientStreamWriter.cs @@ -44,6 +44,7 @@ namespace Grpc.Core /// </summary> /// <typeparam name="T"></typeparam> public interface IClientStreamWriter<T> : IAsyncStreamWriter<T> + where T : class { /// <summary> /// Closes the stream. Can only be called once there is no pending write. No writes should follow calling this. diff --git a/src/csharp/Grpc.Core/IServerStreamWriter.cs b/src/csharp/Grpc.Core/IServerStreamWriter.cs index e76397d8a0d9200964df735cf9ff0292f87ac56c..199a585a3fecf7b1336c8ac8b41183d456aea259 100644 --- a/src/csharp/Grpc.Core/IServerStreamWriter.cs +++ b/src/csharp/Grpc.Core/IServerStreamWriter.cs @@ -43,6 +43,7 @@ namespace Grpc.Core /// A writable stream of messages that is used in server-side handlers. /// </summary> public interface IServerStreamWriter<T> : IAsyncStreamWriter<T> + where T : class { } } diff --git a/src/csharp/Grpc.Core/Internal/AsyncCallServer.cs b/src/csharp/Grpc.Core/Internal/AsyncCallServer.cs index 3c66c67dccea634d3a2587ad9b433c8c172e3958..171d0c799d1e576487f1dad666d9315d6727f827 100644 --- a/src/csharp/Grpc.Core/Internal/AsyncCallServer.cs +++ b/src/csharp/Grpc.Core/Internal/AsyncCallServer.cs @@ -138,9 +138,7 @@ namespace Grpc.Core.Internal ReleaseResourcesIfPossible(); } - // TODO(jtattermusch): check if call was cancelled. - - // TODO: handle error ... + // TODO(jtattermusch): handle error finishedServersideTcs.SetResult(null); } diff --git a/src/csharp/Grpc.Core/Internal/ClientRequestStream.cs b/src/csharp/Grpc.Core/Internal/ClientRequestStream.cs index 6854922a6f7165f7ece8ab180594d80a8aca3546..16970587328540e59d8df70c6f624fdffee4cd0c 100644 --- a/src/csharp/Grpc.Core/Internal/ClientRequestStream.cs +++ b/src/csharp/Grpc.Core/Internal/ClientRequestStream.cs @@ -38,6 +38,8 @@ namespace Grpc.Core.Internal /// Writes requests asynchronously to an underlying AsyncCall object. /// </summary> internal class ClientRequestStream<TRequest, TResponse> : IClientStreamWriter<TRequest> + where TRequest : class + where TResponse : class { readonly AsyncCall<TRequest, TResponse> call; diff --git a/src/csharp/Grpc.Core/Internal/ClientResponseStream.cs b/src/csharp/Grpc.Core/Internal/ClientResponseStream.cs index 7fa511faa82b8099110c5ec4d0018a4d87551ec3..b2378cade66d14a9413cc667868b068449876190 100644 --- a/src/csharp/Grpc.Core/Internal/ClientResponseStream.cs +++ b/src/csharp/Grpc.Core/Internal/ClientResponseStream.cs @@ -38,6 +38,8 @@ using System.Threading.Tasks; namespace Grpc.Core.Internal { internal class ClientResponseStream<TRequest, TResponse> : IAsyncStreamReader<TResponse> + where TRequest : class + where TResponse : class { readonly AsyncCall<TRequest, TResponse> call; diff --git a/src/csharp/Grpc.Core/Internal/ServerCallHandler.cs b/src/csharp/Grpc.Core/Internal/ServerCallHandler.cs index 01b2a113699a7edffbd26d36f39612fee4bf96e9..95d8e97869276aefd56d3243fb07819fd424cf07 100644 --- a/src/csharp/Grpc.Core/Internal/ServerCallHandler.cs +++ b/src/csharp/Grpc.Core/Internal/ServerCallHandler.cs @@ -45,6 +45,8 @@ namespace Grpc.Core.Internal } internal class UnaryServerCallHandler<TRequest, TResponse> : IServerCallHandler + where TRequest : class + where TResponse : class { readonly Method<TRequest, TResponse> method; readonly UnaryServerMethod<TRequest, TResponse> handler; @@ -72,7 +74,8 @@ namespace Grpc.Core.Internal 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); + var context = new ServerCallContext(); // TODO(jtattermusch): initialize the context + var result = await handler(context, request); await responseStream.Write(result); } catch (Exception e) @@ -93,6 +96,8 @@ namespace Grpc.Core.Internal } internal class ServerStreamingServerCallHandler<TRequest, TResponse> : IServerCallHandler + where TRequest : class + where TResponse : class { readonly Method<TRequest, TResponse> method; readonly ServerStreamingServerMethod<TRequest, TResponse> handler; @@ -121,7 +126,8 @@ namespace Grpc.Core.Internal // 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); + var context = new ServerCallContext(); // TODO(jtattermusch): initialize the context + await handler(context, request, responseStream); } catch (Exception e) { @@ -142,6 +148,8 @@ namespace Grpc.Core.Internal } internal class ClientStreamingServerCallHandler<TRequest, TResponse> : IServerCallHandler + where TRequest : class + where TResponse : class { readonly Method<TRequest, TResponse> method; readonly ClientStreamingServerMethod<TRequest, TResponse> handler; @@ -162,11 +170,12 @@ namespace Grpc.Core.Internal var finishedTask = asyncCall.ServerSideCallAsync(); var requestStream = new ServerRequestStream<TRequest, TResponse>(asyncCall); var responseStream = new ServerResponseStream<TRequest, TResponse>(asyncCall); + var context = new ServerCallContext(); // TODO(jtattermusch): initialize the context Status status = Status.DefaultSuccess; try { - var result = await handler(requestStream); + var result = await handler(context, requestStream); try { await responseStream.Write(result); @@ -195,6 +204,8 @@ namespace Grpc.Core.Internal } internal class DuplexStreamingServerCallHandler<TRequest, TResponse> : IServerCallHandler + where TRequest : class + where TResponse : class { readonly Method<TRequest, TResponse> method; readonly DuplexStreamingServerMethod<TRequest, TResponse> handler; @@ -215,11 +226,12 @@ namespace Grpc.Core.Internal var finishedTask = asyncCall.ServerSideCallAsync(); var requestStream = new ServerRequestStream<TRequest, TResponse>(asyncCall); var responseStream = new ServerResponseStream<TRequest, TResponse>(asyncCall); + var context = new ServerCallContext(); // TODO(jtattermusch): initialize the context Status status = Status.DefaultSuccess; try { - await handler(requestStream, responseStream); + await handler(context, requestStream, responseStream); } catch (Exception e) { diff --git a/src/csharp/Grpc.Core/Internal/ServerCalls.cs b/src/csharp/Grpc.Core/Internal/ServerCalls.cs index 5c6b335c7f1c9ee5d01a39d8d3c3b844a4d9c2dc..81279678b95699c7f0f8aa48ad517d83b3faa0e8 100644 --- a/src/csharp/Grpc.Core/Internal/ServerCalls.cs +++ b/src/csharp/Grpc.Core/Internal/ServerCalls.cs @@ -41,21 +41,29 @@ namespace Grpc.Core.Internal internal static class ServerCalls { public static IServerCallHandler UnaryCall<TRequest, TResponse>(Method<TRequest, TResponse> method, UnaryServerMethod<TRequest, TResponse> handler) + where TRequest : class + where TResponse : class { return new UnaryServerCallHandler<TRequest, TResponse>(method, handler); } public static IServerCallHandler ClientStreamingCall<TRequest, TResponse>(Method<TRequest, TResponse> method, ClientStreamingServerMethod<TRequest, TResponse> handler) + where TRequest : class + where TResponse : class { return new ClientStreamingServerCallHandler<TRequest, TResponse>(method, handler); } public static IServerCallHandler ServerStreamingCall<TRequest, TResponse>(Method<TRequest, TResponse> method, ServerStreamingServerMethod<TRequest, TResponse> handler) + where TRequest : class + where TResponse : class { return new ServerStreamingServerCallHandler<TRequest, TResponse>(method, handler); } public static IServerCallHandler DuplexStreamingCall<TRequest, TResponse>(Method<TRequest, TResponse> method, DuplexStreamingServerMethod<TRequest, TResponse> handler) + where TRequest : class + where TResponse : class { return new DuplexStreamingServerCallHandler<TRequest, TResponse>(method, handler); } diff --git a/src/csharp/Grpc.Core/Internal/ServerRequestStream.cs b/src/csharp/Grpc.Core/Internal/ServerRequestStream.cs index aa311059c32b1d3b0358c71d508262fc20aec427..d9ee0c815b2570af7ffb4a4d9834697506616b70 100644 --- a/src/csharp/Grpc.Core/Internal/ServerRequestStream.cs +++ b/src/csharp/Grpc.Core/Internal/ServerRequestStream.cs @@ -38,6 +38,8 @@ using System.Threading.Tasks; namespace Grpc.Core.Internal { internal class ServerRequestStream<TRequest, TResponse> : IAsyncStreamReader<TRequest> + where TRequest : class + where TResponse : class { readonly AsyncCallServer<TRequest, TResponse> call; diff --git a/src/csharp/Grpc.Core/Internal/ServerResponseStream.cs b/src/csharp/Grpc.Core/Internal/ServerResponseStream.cs index 686017c048dcf1ad3532f33df886d8eb7693c7fb..da688d504f2e8d492596224ec7eb43f8a34a2f35 100644 --- a/src/csharp/Grpc.Core/Internal/ServerResponseStream.cs +++ b/src/csharp/Grpc.Core/Internal/ServerResponseStream.cs @@ -39,6 +39,8 @@ namespace Grpc.Core.Internal /// Writes responses asynchronously to an underlying AsyncCallServer object. /// </summary> internal class ServerResponseStream<TRequest, TResponse> : IServerStreamWriter<TResponse> + where TRequest : class + where TResponse : class { readonly AsyncCallServer<TRequest, TResponse> call; diff --git a/src/csharp/Grpc.Core/Server.cs b/src/csharp/Grpc.Core/Server.cs index a3000cee462131ce90cc619e715bba07df6007bf..0df46bb25b2b9ea44ba71ca8f1604c8b0e45a3e2 100644 --- a/src/csharp/Grpc.Core/Server.cs +++ b/src/csharp/Grpc.Core/Server.cs @@ -47,6 +47,11 @@ namespace Grpc.Core /// </summary> public class Server { + /// <summary> + /// Pass this value as port to have the server choose an unused listening port for you. + /// </summary> + public const int PickUnusedPort = 0; + // TODO(jtattermusch) : make sure the delegate doesn't get garbage collected while // native callbacks are in the completion queue. readonly ServerShutdownCallbackDelegate serverShutdownHandler; @@ -89,29 +94,25 @@ namespace Grpc.Core /// Add a non-secure port on which server should listen. /// Only call this before Start(). /// </summary> - public int AddListeningPort(string addr) + /// <returns>The port on which server will be listening.</returns> + /// <param name="host">the host</param> + /// <param name="port">the port. If zero, an unused port is chosen automatically.</param> + public int AddListeningPort(string host, int port) { - lock (myLock) - { - Preconditions.CheckState(!startRequested); - return handle.AddListeningPort(addr); - } + return AddListeningPortInternal(host, port, null); } /// <summary> - /// Add a secure port on which server should listen. + /// Add a non-secure port on which server should listen. /// Only call this before Start(). /// </summary> - public int AddListeningPort(string addr, ServerCredentials credentials) + /// <returns>The port on which server will be listening.</returns> + /// <param name="host">the host</param> + /// <param name="port">the port. If zero, , an unused port is chosen automatically.</param> + public int AddListeningPort(string host, int port, ServerCredentials credentials) { - lock (myLock) - { - Preconditions.CheckState(!startRequested); - using (var nativeCredentials = credentials.ToNativeCredentials()) - { - return handle.AddListeningPort(addr, nativeCredentials); - } - } + Preconditions.CheckNotNull(credentials); + return AddListeningPortInternal(host, port, credentials); } /// <summary> @@ -164,6 +165,26 @@ namespace Grpc.Core handle.Dispose(); } + private int AddListeningPortInternal(string host, int port, ServerCredentials credentials) + { + lock (myLock) + { + Preconditions.CheckState(!startRequested); + var address = string.Format("{0}:{1}", host, port); + if (credentials != null) + { + using (var nativeCredentials = credentials.ToNativeCredentials()) + { + return handle.AddListeningPort(address, nativeCredentials); + } + } + else + { + return handle.AddListeningPort(address); + } + } + } + /// <summary> /// Allows one new RPC call to be received by server. /// </summary> diff --git a/src/csharp/Grpc.Core/ServerCallContext.cs b/src/csharp/Grpc.Core/ServerCallContext.cs new file mode 100644 index 0000000000000000000000000000000000000000..e873b3e88a74d0036737f7830e0726ac29a4b60c --- /dev/null +++ b/src/csharp/Grpc.Core/ServerCallContext.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.Runtime.CompilerServices; +using System.Threading.Tasks; + +namespace Grpc.Core +{ + /// <summary> + /// Context for a server-side call. + /// </summary> + public sealed class ServerCallContext + { + + // TODO(jtattermusch): add cancellationToken + + // TODO(jtattermusch): add deadline info + + // TODO(jtattermusch): expose initial metadata sent by client for reading + + // TODO(jtattermusch): expose method to send initial metadata back to client + + // TODO(jtattermusch): allow setting status and trailing metadata to send after handler completes. + } +} diff --git a/src/csharp/Grpc.Core/ServerMethods.cs b/src/csharp/Grpc.Core/ServerMethods.cs index 6646bb5a89b4f82daee7ce4c1f869a17f9753b85..377b78eb302e5b1171fb87580855a1c87a75368a 100644 --- a/src/csharp/Grpc.Core/ServerMethods.cs +++ b/src/csharp/Grpc.Core/ServerMethods.cs @@ -42,20 +42,28 @@ namespace Grpc.Core /// <summary> /// Server-side handler for unary call. /// </summary> - public delegate Task<TResponse> UnaryServerMethod<TRequest, TResponse>(TRequest request); + public delegate Task<TResponse> UnaryServerMethod<TRequest, TResponse>(ServerCallContext context, TRequest request) + where TRequest : class + where TResponse : class; /// <summary> /// Server-side handler for client streaming call. /// </summary> - public delegate Task<TResponse> ClientStreamingServerMethod<TRequest, TResponse>(IAsyncStreamReader<TRequest> requestStream); + public delegate Task<TResponse> ClientStreamingServerMethod<TRequest, TResponse>(ServerCallContext context, IAsyncStreamReader<TRequest> requestStream) + where TRequest : class + where TResponse : class; /// <summary> /// Server-side handler for server streaming call. /// </summary> - public delegate Task ServerStreamingServerMethod<TRequest, TResponse>(TRequest request, IServerStreamWriter<TResponse> responseStream); + public delegate Task ServerStreamingServerMethod<TRequest, TResponse>(ServerCallContext context, TRequest request, IServerStreamWriter<TResponse> responseStream) + where TRequest : class + where TResponse : class; /// <summary> /// Server-side handler for bidi streaming call. /// </summary> - public delegate Task DuplexStreamingServerMethod<TRequest, TResponse>(IAsyncStreamReader<TRequest> requestStream, IServerStreamWriter<TResponse> responseStream); + public delegate Task DuplexStreamingServerMethod<TRequest, TResponse>(ServerCallContext context, IAsyncStreamReader<TRequest> requestStream, IServerStreamWriter<TResponse> responseStream) + where TRequest : class + where TResponse : class; } diff --git a/src/csharp/Grpc.Core/ServerServiceDefinition.cs b/src/csharp/Grpc.Core/ServerServiceDefinition.cs index 01b1dc8f7bdceaf82169b4a1c13e0dbd77ef9a3f..81846beb2f7c2bc62f2a605535a42b5e0f499890 100644 --- a/src/csharp/Grpc.Core/ServerServiceDefinition.cs +++ b/src/csharp/Grpc.Core/ServerServiceDefinition.cs @@ -76,6 +76,8 @@ namespace Grpc.Core public Builder AddMethod<TRequest, TResponse>( Method<TRequest, TResponse> method, UnaryServerMethod<TRequest, TResponse> handler) + where TRequest : class + where TResponse : class { callHandlers.Add(GetFullMethodName(serviceName, method.Name), ServerCalls.UnaryCall(method, handler)); return this; @@ -84,6 +86,8 @@ namespace Grpc.Core public Builder AddMethod<TRequest, TResponse>( Method<TRequest, TResponse> method, ClientStreamingServerMethod<TRequest, TResponse> handler) + where TRequest : class + where TResponse : class { callHandlers.Add(GetFullMethodName(serviceName, method.Name), ServerCalls.ClientStreamingCall(method, handler)); return this; @@ -92,6 +96,8 @@ namespace Grpc.Core public Builder AddMethod<TRequest, TResponse>( Method<TRequest, TResponse> method, ServerStreamingServerMethod<TRequest, TResponse> handler) + where TRequest : class + where TResponse : class { callHandlers.Add(GetFullMethodName(serviceName, method.Name), ServerCalls.ServerStreamingCall(method, handler)); return this; @@ -100,6 +106,8 @@ namespace Grpc.Core public Builder AddMethod<TRequest, TResponse>( Method<TRequest, TResponse> method, DuplexStreamingServerMethod<TRequest, TResponse> handler) + where TRequest : class + where TResponse : class { callHandlers.Add(GetFullMethodName(serviceName, method.Name), ServerCalls.DuplexStreamingCall(method, handler)); return this; diff --git a/src/csharp/Grpc.Core/Stub/AbstractStub.cs b/src/csharp/Grpc.Core/Stub/AbstractStub.cs index cf5ab958c5bda0af14c873a804fc1c26e02ec103..4a8b2543579f8e64b79c89b95d264ce30944b301 100644 --- a/src/csharp/Grpc.Core/Stub/AbstractStub.cs +++ b/src/csharp/Grpc.Core/Stub/AbstractStub.cs @@ -64,6 +64,8 @@ namespace Grpc.Core /// Creates a new call to given method. /// </summary> protected Call<TRequest, TResponse> CreateCall<TRequest, TResponse>(string serviceName, Method<TRequest, TResponse> method) + where TRequest : class + where TResponse : class { var headerBuilder = Metadata.CreateBuilder(); config.HeaderInterceptor(headerBuilder); diff --git a/src/csharp/Grpc.Examples.MathServer/MathServer.cs b/src/csharp/Grpc.Examples.MathServer/MathServer.cs index abc7ef05e4b28d57a072eff4e101665728eea6c3..cfde9b42c76eb736550686978c3381d66e54d46d 100644 --- a/src/csharp/Grpc.Examples.MathServer/MathServer.cs +++ b/src/csharp/Grpc.Examples.MathServer/MathServer.cs @@ -46,7 +46,7 @@ namespace math Server server = new Server(); server.AddServiceDefinition(MathGrpc.BindService(new MathServiceImpl())); - int port = server.AddListeningPort(host + ":23456"); + int port = server.AddListeningPort(host, 23456); server.Start(); Console.WriteLine("MathServer listening on port " + port); diff --git a/src/csharp/Grpc.Examples.Tests/MathClientServerTests.cs b/src/csharp/Grpc.Examples.Tests/MathClientServerTests.cs index 332795e0e54c1516ee421fc033db554d88ded72b..4ada95edd6da174fae3a1abe6ae4c5429b77d266 100644 --- a/src/csharp/Grpc.Examples.Tests/MathClientServerTests.cs +++ b/src/csharp/Grpc.Examples.Tests/MathClientServerTests.cs @@ -59,7 +59,7 @@ namespace math.Tests server = new Server(); server.AddServiceDefinition(MathGrpc.BindService(new MathServiceImpl())); - int port = server.AddListeningPort(host + ":0"); + int port = server.AddListeningPort(host, Server.PickUnusedPort); server.Start(); channel = new Channel(host + ":" + port); diff --git a/src/csharp/Grpc.Examples/MathGrpc.cs b/src/csharp/Grpc.Examples/MathGrpc.cs index 60408b901848e94be15f2d14021594c2bccf54d6..03f5c31cb7acc77e7c846f93e08daa3c9cad1847 100644 --- a/src/csharp/Grpc.Examples/MathGrpc.cs +++ b/src/csharp/Grpc.Examples/MathGrpc.cs @@ -133,13 +133,13 @@ namespace math // server-side interface public interface IMathService { - Task<DivReply> Div(DivArgs request); + Task<DivReply> Div(ServerCallContext context, DivArgs request); - Task Fib(FibArgs request, IServerStreamWriter<Num> responseStream); + Task Fib(ServerCallContext context, FibArgs request, IServerStreamWriter<Num> responseStream); - Task<Num> Sum(IAsyncStreamReader<Num> requestStream); + Task<Num> Sum(ServerCallContext context, IAsyncStreamReader<Num> requestStream); - Task DivMany(IAsyncStreamReader<DivArgs> requestStream, IServerStreamWriter<DivReply> responseStream); + Task DivMany(ServerCallContext context, IAsyncStreamReader<DivArgs> requestStream, IServerStreamWriter<DivReply> responseStream); } public static ServerServiceDefinition BindService(IMathService serviceImpl) diff --git a/src/csharp/Grpc.Examples/MathServiceImpl.cs b/src/csharp/Grpc.Examples/MathServiceImpl.cs index 83ec2a8c3df20fcda82be12209088e0812adadc6..800dee8735412a037d76c831a9e953c042feff62 100644 --- a/src/csharp/Grpc.Examples/MathServiceImpl.cs +++ b/src/csharp/Grpc.Examples/MathServiceImpl.cs @@ -46,12 +46,12 @@ namespace math /// </summary> public class MathServiceImpl : MathGrpc.IMathService { - public Task<DivReply> Div(DivArgs request) + public Task<DivReply> Div(ServerCallContext context, DivArgs request) { return Task.FromResult(DivInternal(request)); } - public async Task Fib(FibArgs request, IServerStreamWriter<Num> responseStream) + public async Task Fib(ServerCallContext context, FibArgs request, IServerStreamWriter<Num> responseStream) { if (request.Limit <= 0) { @@ -68,7 +68,7 @@ namespace math } } - public async Task<Num> Sum(IAsyncStreamReader<Num> requestStream) + public async Task<Num> Sum(ServerCallContext context, IAsyncStreamReader<Num> requestStream) { long sum = 0; await requestStream.ForEach(async num => @@ -78,7 +78,7 @@ namespace math return Num.CreateBuilder().SetNum_(sum).Build(); } - public async Task DivMany(IAsyncStreamReader<DivArgs> requestStream, IServerStreamWriter<DivReply> responseStream) + public async Task DivMany(ServerCallContext context, IAsyncStreamReader<DivArgs> requestStream, IServerStreamWriter<DivReply> responseStream) { await requestStream.ForEach(async divArgs => { diff --git a/src/csharp/Grpc.IntegrationTesting/InteropClientServerTest.cs b/src/csharp/Grpc.IntegrationTesting/InteropClientServerTest.cs index 45380227c25e13da56799bb26e1f75a2911a60e1..9e49ce0d174d800bb829cac7e6338cc1321771cf 100644 --- a/src/csharp/Grpc.IntegrationTesting/InteropClientServerTest.cs +++ b/src/csharp/Grpc.IntegrationTesting/InteropClientServerTest.cs @@ -59,7 +59,7 @@ namespace Grpc.IntegrationTesting server = new Server(); server.AddServiceDefinition(TestServiceGrpc.BindService(new TestServiceImpl())); - int port = server.AddListeningPort(host + ":0", TestCredentials.CreateTestServerCredentials()); + int port = server.AddListeningPort(host, Server.PickUnusedPort, TestCredentials.CreateTestServerCredentials()); server.Start(); var channelArgs = ChannelArgs.CreateBuilder() diff --git a/src/csharp/Grpc.IntegrationTesting/InteropServer.cs b/src/csharp/Grpc.IntegrationTesting/InteropServer.cs index ad5200774f5840ec6dd2869434ac29ae99aae929..ca54aed04188c4b69ccf57a5c8bd5786ed483f11 100644 --- a/src/csharp/Grpc.IntegrationTesting/InteropServer.cs +++ b/src/csharp/Grpc.IntegrationTesting/InteropServer.cs @@ -93,16 +93,17 @@ namespace Grpc.IntegrationTesting var server = new Server(); server.AddServiceDefinition(TestServiceGrpc.BindService(new TestServiceImpl())); - string addr = "0.0.0.0:" + options.port; + string host = "0.0.0.0"; + int port = options.port.Value; if (options.useTls) { - server.AddListeningPort(addr, TestCredentials.CreateTestServerCredentials()); + server.AddListeningPort(host, port, TestCredentials.CreateTestServerCredentials()); } else { - server.AddListeningPort(addr); + server.AddListeningPort(host, options.port.Value); } - Console.WriteLine("Running server on " + addr); + Console.WriteLine("Running server on " + string.Format("{0}:{1}", host, port)); server.Start(); server.ShutdownTask.Wait(); diff --git a/src/csharp/Grpc.IntegrationTesting/TestServiceGrpc.cs b/src/csharp/Grpc.IntegrationTesting/TestServiceGrpc.cs index d1f8aa12c78187f806a317ee455b5967f0f30f1e..9f14dad6c0c4e7170282743587bb27e723e5c9f1 100644 --- a/src/csharp/Grpc.IntegrationTesting/TestServiceGrpc.cs +++ b/src/csharp/Grpc.IntegrationTesting/TestServiceGrpc.cs @@ -171,17 +171,17 @@ namespace grpc.testing // server-side interface public interface ITestService { - Task<Empty> EmptyCall(Empty request); + Task<Empty> EmptyCall(ServerCallContext context, Empty request); - Task<SimpleResponse> UnaryCall(SimpleRequest request); + Task<SimpleResponse> UnaryCall(ServerCallContext context, SimpleRequest request); - Task StreamingOutputCall(StreamingOutputCallRequest request, IServerStreamWriter<StreamingOutputCallResponse> responseStream); + Task StreamingOutputCall(ServerCallContext context, StreamingOutputCallRequest request, IServerStreamWriter<StreamingOutputCallResponse> responseStream); - Task<StreamingInputCallResponse> StreamingInputCall(IAsyncStreamReader<StreamingInputCallRequest> requestStream); + Task<StreamingInputCallResponse> StreamingInputCall(ServerCallContext context, IAsyncStreamReader<StreamingInputCallRequest> requestStream); - Task FullDuplexCall(IAsyncStreamReader<StreamingOutputCallRequest> requestStream, IServerStreamWriter<StreamingOutputCallResponse> responseStream); + Task FullDuplexCall(ServerCallContext context, IAsyncStreamReader<StreamingOutputCallRequest> requestStream, IServerStreamWriter<StreamingOutputCallResponse> responseStream); - Task HalfDuplexCall(IAsyncStreamReader<StreamingOutputCallRequest> requestStream, IServerStreamWriter<StreamingOutputCallResponse> responseStream); + Task HalfDuplexCall(ServerCallContext context, IAsyncStreamReader<StreamingOutputCallRequest> requestStream, IServerStreamWriter<StreamingOutputCallResponse> responseStream); } public static ServerServiceDefinition BindService(ITestService serviceImpl) diff --git a/src/csharp/Grpc.IntegrationTesting/TestServiceImpl.cs b/src/csharp/Grpc.IntegrationTesting/TestServiceImpl.cs index 8b0cf3a2d0596c8319f8db6dfd77478701999071..40f32b5a88f0ad989d6416bdd0b8abd2fd4e8e07 100644 --- a/src/csharp/Grpc.IntegrationTesting/TestServiceImpl.cs +++ b/src/csharp/Grpc.IntegrationTesting/TestServiceImpl.cs @@ -46,19 +46,19 @@ namespace grpc.testing /// </summary> public class TestServiceImpl : TestServiceGrpc.ITestService { - public Task<Empty> EmptyCall(Empty request) + public Task<Empty> EmptyCall(ServerCallContext context, Empty request) { return Task.FromResult(Empty.DefaultInstance); } - public Task<SimpleResponse> UnaryCall(SimpleRequest request) + public Task<SimpleResponse> UnaryCall(ServerCallContext context, SimpleRequest request) { var response = SimpleResponse.CreateBuilder() .SetPayload(CreateZerosPayload(request.ResponseSize)).Build(); return Task.FromResult(response); } - public async Task StreamingOutputCall(StreamingOutputCallRequest request, IServerStreamWriter<StreamingOutputCallResponse> responseStream) + public async Task StreamingOutputCall(ServerCallContext context, StreamingOutputCallRequest request, IServerStreamWriter<StreamingOutputCallResponse> responseStream) { foreach (var responseParam in request.ResponseParametersList) { @@ -68,7 +68,7 @@ namespace grpc.testing } } - public async Task<StreamingInputCallResponse> StreamingInputCall(IAsyncStreamReader<StreamingInputCallRequest> requestStream) + public async Task<StreamingInputCallResponse> StreamingInputCall(ServerCallContext context, IAsyncStreamReader<StreamingInputCallRequest> requestStream) { int sum = 0; await requestStream.ForEach(async request => @@ -78,7 +78,7 @@ namespace grpc.testing return StreamingInputCallResponse.CreateBuilder().SetAggregatedPayloadSize(sum).Build(); } - public async Task FullDuplexCall(IAsyncStreamReader<StreamingOutputCallRequest> requestStream, IServerStreamWriter<StreamingOutputCallResponse> responseStream) + public async Task FullDuplexCall(ServerCallContext context, IAsyncStreamReader<StreamingOutputCallRequest> requestStream, IServerStreamWriter<StreamingOutputCallResponse> responseStream) { await requestStream.ForEach(async request => { @@ -91,7 +91,7 @@ namespace grpc.testing }); } - public async Task HalfDuplexCall(IAsyncStreamReader<StreamingOutputCallRequest> requestStream, IServerStreamWriter<StreamingOutputCallResponse> responseStream) + public async Task HalfDuplexCall(ServerCallContext context, IAsyncStreamReader<StreamingOutputCallRequest> requestStream, IServerStreamWriter<StreamingOutputCallResponse> responseStream) { throw new NotImplementedException(); }