From c0b3721d61fed71a3528cecfa6797cdc26722d3a Mon Sep 17 00:00:00 2001 From: Jan Tattermusch Date: Fri, 13 Mar 2015 08:35:41 -0700 Subject: [PATCH] added support for metadata --- .../Grpc.Core.Tests/ClientServerTest.cs | 14 +- .../Grpc.Core.Tests/Grpc.Core.Tests.csproj | 4 + .../Internal/MetadataArraySafeHandleTest.cs | 62 +++++++ src/csharp/Grpc.Core/Call.cs | 55 +++--- src/csharp/Grpc.Core/Calls.cs | 30 ++-- src/csharp/Grpc.Core/Grpc.Core.csproj | 7 + src/csharp/Grpc.Core/Internal/AsyncCall.cs | 37 ++-- .../Grpc.Core/Internal/CallSafeHandle.cs | 33 ++-- .../Internal/MetadataArraySafeHandle.cs | 72 ++++++++ src/csharp/Grpc.Core/Marshaller.cs | 5 +- src/csharp/Grpc.Core/Metadata.cs | 126 ++++++++++++++ .../Grpc.Core/ServerServiceDefinition.cs | 15 +- src/csharp/Grpc.Core/Stub/AbstractStub.cs | 73 ++++++++ .../Grpc.Core/Stub/StubConfiguration.cs | 64 +++++++ .../MathClientServerTests.cs | 9 +- src/csharp/Grpc.Examples/MathGrpc.cs | 36 ++-- .../Grpc.IntegrationTesting.csproj | 3 +- .../TestServiceGrpc.cs | 41 +++-- src/csharp/ext/grpc_csharp_ext.c | 162 ++++++++++++++---- 19 files changed, 694 insertions(+), 154 deletions(-) create mode 100644 src/csharp/Grpc.Core.Tests/Internal/MetadataArraySafeHandleTest.cs create mode 100644 src/csharp/Grpc.Core/Internal/MetadataArraySafeHandle.cs create mode 100644 src/csharp/Grpc.Core/Metadata.cs create mode 100644 src/csharp/Grpc.Core/Stub/AbstractStub.cs create mode 100644 src/csharp/Grpc.Core/Stub/StubConfiguration.cs diff --git a/src/csharp/Grpc.Core.Tests/ClientServerTest.cs b/src/csharp/Grpc.Core.Tests/ClientServerTest.cs index e73159b3507..807f5a6ded4 100644 --- a/src/csharp/Grpc.Core.Tests/ClientServerTest.cs +++ b/src/csharp/Grpc.Core.Tests/ClientServerTest.cs @@ -46,6 +46,8 @@ namespace Grpc.Core.Tests { string host = "localhost"; + string serviceName = "/tests.Test"; + Method unaryEchoStringMethod = new Method( MethodType.Unary, "/tests.Test/UnaryEchoString", @@ -69,7 +71,7 @@ namespace Grpc.Core.Tests { Server server = new Server(); server.AddServiceDefinition( - ServerServiceDefinition.CreateBuilder("someService") + ServerServiceDefinition.CreateBuilder(serviceName) .AddMethod(unaryEchoStringMethod, HandleUnaryEchoString).Build()); int port = server.AddPort(host + ":0"); @@ -77,7 +79,7 @@ namespace Grpc.Core.Tests using (Channel channel = new Channel(host + ":" + port)) { - var call = new Call(unaryEchoStringMethod, channel); + var call = new Call(serviceName, unaryEchoStringMethod, channel, Metadata.Empty); Assert.AreEqual("ABC", Calls.BlockingUnaryCall(call, "ABC", default(CancellationToken))); @@ -92,7 +94,7 @@ namespace Grpc.Core.Tests { Server server = new Server(); server.AddServiceDefinition( - ServerServiceDefinition.CreateBuilder("someService") + ServerServiceDefinition.CreateBuilder(serviceName) .AddMethod(unaryEchoStringMethod, HandleUnaryEchoString).Build()); int port = server.AddPort(host + ":0"); @@ -100,7 +102,7 @@ namespace Grpc.Core.Tests using (Channel channel = new Channel(host + ":" + port)) { - var call = new Call(unaryEchoStringMethod, channel); + var call = new Call(serviceName, unaryEchoStringMethod, channel, Metadata.Empty); BenchmarkUtil.RunBenchmark(100, 1000, () => { Calls.BlockingUnaryCall(call, "ABC", default(CancellationToken)); }); } @@ -113,14 +115,14 @@ namespace Grpc.Core.Tests { Server server = new Server(); server.AddServiceDefinition( - ServerServiceDefinition.CreateBuilder("someService").Build()); + ServerServiceDefinition.CreateBuilder(serviceName).Build()); int port = server.AddPort(host + ":0"); server.Start(); using (Channel channel = new Channel(host + ":" + port)) { - var call = new Call(unaryEchoStringMethod, channel); + var call = new Call(serviceName, unaryEchoStringMethod, channel, Metadata.Empty); try { diff --git a/src/csharp/Grpc.Core.Tests/Grpc.Core.Tests.csproj b/src/csharp/Grpc.Core.Tests/Grpc.Core.Tests.csproj index a365320f052..eac8d16fb15 100644 --- a/src/csharp/Grpc.Core.Tests/Grpc.Core.Tests.csproj +++ b/src/csharp/Grpc.Core.Tests/Grpc.Core.Tests.csproj @@ -42,6 +42,7 @@ + @@ -56,4 +57,7 @@ + + + \ No newline at end of file diff --git a/src/csharp/Grpc.Core.Tests/Internal/MetadataArraySafeHandleTest.cs b/src/csharp/Grpc.Core.Tests/Internal/MetadataArraySafeHandleTest.cs new file mode 100644 index 00000000000..2f6013483d9 --- /dev/null +++ b/src/csharp/Grpc.Core.Tests/Internal/MetadataArraySafeHandleTest.cs @@ -0,0 +1,62 @@ +#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; +using Grpc.Core.Internal; +using Grpc.Core.Utils; +using NUnit.Framework; + +namespace Grpc.Core.Internal.Tests +{ + public class MetadataArraySafeHandleTest + { + [Test] + public void CreateEmptyAndDestroy() + { + var metadata = Metadata.CreateBuilder().Build(); + var nativeMetadata = MetadataArraySafeHandle.Create(metadata); + nativeMetadata.Dispose(); + } + + [Test] + public void CreateAndDestroy() + { + var metadata = Metadata.CreateBuilder() + .Add(new Metadata.MetadataEntry("host", "somehost")) + .Add(new Metadata.MetadataEntry("header2", "header value")).Build(); + var nativeMetadata = MetadataArraySafeHandle.Create(metadata); + nativeMetadata.Dispose(); + } + } +} diff --git a/src/csharp/Grpc.Core/Call.cs b/src/csharp/Grpc.Core/Call.cs index d84d5940c28..fe5f40f5e90 100644 --- a/src/csharp/Grpc.Core/Call.cs +++ b/src/csharp/Grpc.Core/Call.cs @@ -33,65 +33,70 @@ using System; using Grpc.Core.Internal; +using Grpc.Core.Utils; namespace Grpc.Core { public class Call { - readonly string methodName; - readonly Func requestSerializer; - readonly Func responseDeserializer; + readonly string name; + readonly Marshaller requestMarshaller; + readonly Marshaller responseMarshaller; readonly Channel channel; + readonly Metadata headers; - public Call(string methodName, - Func requestSerializer, - Func responseDeserializer, - TimeSpan timeout, - Channel channel) + public Call(string serviceName, Method method, Channel channel, Metadata headers) { - this.methodName = methodName; - this.requestSerializer = requestSerializer; - this.responseDeserializer = responseDeserializer; - this.channel = channel; + this.name = Preconditions.CheckNotNull(serviceName) + "/" + method.Name; + this.requestMarshaller = method.RequestMarshaller; + this.responseMarshaller = method.ResponseMarshaller; + this.channel = Preconditions.CheckNotNull(channel); + this.headers = Preconditions.CheckNotNull(headers); } - public Call(Method method, Channel channel) + public Channel Channel { - this.methodName = method.Name; - this.requestSerializer = method.RequestMarshaller.Serializer; - this.responseDeserializer = method.ResponseMarshaller.Deserializer; - this.channel = channel; + get + { + return this.channel; + } } - public Channel Channel + /// + /// Full methods name including the service name. + /// + public string Name { get { - return this.channel; + return name; } } - public string MethodName + /// + /// Headers to send at the beginning of the call. + /// + public Metadata Headers { get { - return this.methodName; + return headers; } } - public Func RequestSerializer + public Marshaller RequestMarshaller { get { - return this.requestSerializer; + return requestMarshaller; } } - public Func ResponseDeserializer + public Marshaller ResponseMarshaller { get { - return this.responseDeserializer; + return responseMarshaller; } } } diff --git a/src/csharp/Grpc.Core/Calls.cs b/src/csharp/Grpc.Core/Calls.cs index cc1d67afcae..280387b323b 100644 --- a/src/csharp/Grpc.Core/Calls.cs +++ b/src/csharp/Grpc.Core/Calls.cs @@ -45,30 +45,29 @@ namespace Grpc.Core { public static TResponse BlockingUnaryCall(Call call, TRequest req, CancellationToken token) { - var asyncCall = new AsyncCall(call.RequestSerializer, call.ResponseDeserializer); - return asyncCall.UnaryCall(call.Channel, call.MethodName, req); + var asyncCall = new AsyncCall(call.RequestMarshaller.Serializer, call.ResponseMarshaller.Deserializer); + return asyncCall.UnaryCall(call.Channel, call.Name, req, call.Headers); } public static async Task AsyncUnaryCall(Call call, TRequest req, CancellationToken token) { - var asyncCall = new AsyncCall(call.RequestSerializer, call.ResponseDeserializer); - asyncCall.Initialize(call.Channel, GetCompletionQueue(), call.MethodName); - return await asyncCall.UnaryCallAsync(req); + var asyncCall = new AsyncCall(call.RequestMarshaller.Serializer, call.ResponseMarshaller.Deserializer); + asyncCall.Initialize(call.Channel, GetCompletionQueue(), call.Name); + return await asyncCall.UnaryCallAsync(req, call.Headers); } public static void AsyncServerStreamingCall(Call call, TRequest req, IObserver outputs, CancellationToken token) { - var asyncCall = new AsyncCall(call.RequestSerializer, call.ResponseDeserializer); - - asyncCall.Initialize(call.Channel, GetCompletionQueue(), call.MethodName); - asyncCall.StartServerStreamingCall(req, outputs); + var asyncCall = new AsyncCall(call.RequestMarshaller.Serializer, call.ResponseMarshaller.Deserializer); + asyncCall.Initialize(call.Channel, GetCompletionQueue(), call.Name); + asyncCall.StartServerStreamingCall(req, outputs, call.Headers); } public static ClientStreamingAsyncResult AsyncClientStreamingCall(Call call, CancellationToken token) { - var asyncCall = new AsyncCall(call.RequestSerializer, call.ResponseDeserializer); - asyncCall.Initialize(call.Channel, GetCompletionQueue(), call.MethodName); - var task = asyncCall.ClientStreamingCallAsync(); + 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); } @@ -80,10 +79,9 @@ namespace Grpc.Core public static IObserver DuplexStreamingCall(Call call, IObserver outputs, CancellationToken token) { - var asyncCall = new AsyncCall(call.RequestSerializer, call.ResponseDeserializer); - asyncCall.Initialize(call.Channel, GetCompletionQueue(), call.MethodName); - - asyncCall.StartDuplexStreamingCall(outputs); + 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); } diff --git a/src/csharp/Grpc.Core/Grpc.Core.csproj b/src/csharp/Grpc.Core/Grpc.Core.csproj index 29f1a0604a2..78ba32b2776 100644 --- a/src/csharp/Grpc.Core/Grpc.Core.csproj +++ b/src/csharp/Grpc.Core/Grpc.Core.csproj @@ -79,6 +79,10 @@ + + + + @@ -96,4 +100,7 @@ + + + \ No newline at end of file diff --git a/src/csharp/Grpc.Core/Internal/AsyncCall.cs b/src/csharp/Grpc.Core/Internal/AsyncCall.cs index 04fc28d988a..bc72cb78de3 100644 --- a/src/csharp/Grpc.Core/Internal/AsyncCall.cs +++ b/src/csharp/Grpc.Core/Internal/AsyncCall.cs @@ -77,7 +77,7 @@ namespace Grpc.Core.Internal /// /// Blocking unary request - unary response call. /// - public TResponse UnaryCall(Channel channel, string methodName, TRequest msg) + public TResponse UnaryCall(Channel channel, string methodName, TRequest msg, Metadata headers) { using (CompletionQueueSafeHandle cq = CompletionQueueSafeHandle.Create()) { @@ -92,7 +92,11 @@ namespace Grpc.Core.Internal halfcloseRequested = true; readingDone = true; } - call.BlockingUnary(cq, payload, unaryResponseHandler); + + using (var metadataArray = MetadataArraySafeHandle.Create(headers)) + { + call.BlockingUnary(cq, payload, unaryResponseHandler, metadataArray); + } try { @@ -109,7 +113,7 @@ namespace Grpc.Core.Internal /// /// Starts a unary request - unary response call. /// - public Task UnaryCallAsync(TRequest msg) + public Task UnaryCallAsync(TRequest msg, Metadata headers) { lock (myLock) { @@ -122,8 +126,10 @@ namespace Grpc.Core.Internal byte[] payload = UnsafeSerialize(msg); unaryResponseTcs = new TaskCompletionSource(); - call.StartUnary(payload, unaryResponseHandler); - + using (var metadataArray = MetadataArraySafeHandle.Create(headers)) + { + call.StartUnary(payload, unaryResponseHandler, metadataArray); + } return unaryResponseTcs.Task; } } @@ -132,7 +138,7 @@ namespace Grpc.Core.Internal /// Starts a streamed request - unary response call. /// Use StartSendMessage and StartSendCloseFromClient to stream requests. /// - public Task ClientStreamingCallAsync() + public Task ClientStreamingCallAsync(Metadata headers) { lock (myLock) { @@ -142,7 +148,10 @@ namespace Grpc.Core.Internal readingDone = true; unaryResponseTcs = new TaskCompletionSource(); - call.StartClientStreaming(unaryResponseHandler); + using (var metadataArray = MetadataArraySafeHandle.Create(headers)) + { + call.StartClientStreaming(unaryResponseHandler, metadataArray); + } return unaryResponseTcs.Task; } @@ -151,7 +160,7 @@ namespace Grpc.Core.Internal /// /// Starts a unary request - streamed response call. /// - public void StartServerStreamingCall(TRequest msg, IObserver readObserver) + public void StartServerStreamingCall(TRequest msg, IObserver readObserver, Metadata headers) { lock (myLock) { @@ -165,7 +174,10 @@ namespace Grpc.Core.Internal byte[] payload = UnsafeSerialize(msg); - call.StartServerStreaming(payload, finishedHandler); + using (var metadataArray = MetadataArraySafeHandle.Create(headers)) + { + call.StartServerStreaming(payload, finishedHandler, metadataArray); + } StartReceiveMessage(); } @@ -175,7 +187,7 @@ namespace Grpc.Core.Internal /// Starts a streaming request - streaming response call. /// Use StartSendMessage and StartSendCloseFromClient to stream requests. /// - public void StartDuplexStreamingCall(IObserver readObserver) + public void StartDuplexStreamingCall(IObserver readObserver, Metadata headers) { lock (myLock) { @@ -185,7 +197,10 @@ namespace Grpc.Core.Internal this.readObserver = readObserver; - call.StartDuplexStreaming(finishedHandler); + using (var metadataArray = MetadataArraySafeHandle.Create(headers)) + { + call.StartDuplexStreaming(finishedHandler, metadataArray); + } StartReceiveMessage(); } diff --git a/src/csharp/Grpc.Core/Internal/CallSafeHandle.cs b/src/csharp/Grpc.Core/Internal/CallSafeHandle.cs index a8cef4a68b1..14add60c728 100644 --- a/src/csharp/Grpc.Core/Internal/CallSafeHandle.cs +++ b/src/csharp/Grpc.Core/Internal/CallSafeHandle.cs @@ -57,25 +57,28 @@ namespace Grpc.Core.Internal [DllImport("grpc_csharp_ext.dll")] static extern GRPCCallError grpcsharp_call_start_unary(CallSafeHandle call, [MarshalAs(UnmanagedType.FunctionPtr)] CompletionCallbackDelegate callback, - byte[] send_buffer, UIntPtr send_buffer_len); + byte[] send_buffer, UIntPtr send_buffer_len, MetadataArraySafeHandle metadataArray); [DllImport("grpc_csharp_ext.dll")] static extern void grpcsharp_call_blocking_unary(CallSafeHandle call, CompletionQueueSafeHandle dedicatedCq, [MarshalAs(UnmanagedType.FunctionPtr)] CompletionCallbackDelegate callback, - byte[] send_buffer, UIntPtr send_buffer_len); + byte[] send_buffer, UIntPtr send_buffer_len, MetadataArraySafeHandle metadataArray); [DllImport("grpc_csharp_ext.dll")] static extern GRPCCallError grpcsharp_call_start_client_streaming(CallSafeHandle call, - [MarshalAs(UnmanagedType.FunctionPtr)] CompletionCallbackDelegate callback); + [MarshalAs(UnmanagedType.FunctionPtr)] CompletionCallbackDelegate callback, + MetadataArraySafeHandle metadataArray); [DllImport("grpc_csharp_ext.dll")] static extern GRPCCallError grpcsharp_call_start_server_streaming(CallSafeHandle call, [MarshalAs(UnmanagedType.FunctionPtr)] CompletionCallbackDelegate callback, - byte[] send_buffer, UIntPtr send_buffer_len); + byte[] send_buffer, UIntPtr send_buffer_len, + MetadataArraySafeHandle metadataArray); [DllImport("grpc_csharp_ext.dll")] static extern GRPCCallError grpcsharp_call_start_duplex_streaming(CallSafeHandle call, - [MarshalAs(UnmanagedType.FunctionPtr)] CompletionCallbackDelegate callback); + [MarshalAs(UnmanagedType.FunctionPtr)] CompletionCallbackDelegate callback, + MetadataArraySafeHandle metadataArray); [DllImport("grpc_csharp_ext.dll")] static extern GRPCCallError grpcsharp_call_send_message(CallSafeHandle call, @@ -109,29 +112,29 @@ namespace Grpc.Core.Internal return grpcsharp_channel_create_call(channel, cq, method, host, deadline); } - public void StartUnary(byte[] payload, CompletionCallbackDelegate callback) + public void StartUnary(byte[] payload, CompletionCallbackDelegate callback, MetadataArraySafeHandle metadataArray) { - AssertCallOk(grpcsharp_call_start_unary(this, callback, payload, new UIntPtr((ulong)payload.Length))); + AssertCallOk(grpcsharp_call_start_unary(this, callback, payload, new UIntPtr((ulong)payload.Length), metadataArray)); } - public void BlockingUnary(CompletionQueueSafeHandle dedicatedCq, byte[] payload, CompletionCallbackDelegate callback) + public void BlockingUnary(CompletionQueueSafeHandle dedicatedCq, byte[] payload, CompletionCallbackDelegate callback, MetadataArraySafeHandle metadataArray) { - grpcsharp_call_blocking_unary(this, dedicatedCq, callback, payload, new UIntPtr((ulong)payload.Length)); + grpcsharp_call_blocking_unary(this, dedicatedCq, callback, payload, new UIntPtr((ulong)payload.Length), metadataArray); } - public void StartClientStreaming(CompletionCallbackDelegate callback) + public void StartClientStreaming(CompletionCallbackDelegate callback, MetadataArraySafeHandle metadataArray) { - AssertCallOk(grpcsharp_call_start_client_streaming(this, callback)); + AssertCallOk(grpcsharp_call_start_client_streaming(this, callback, metadataArray)); } - public void StartServerStreaming(byte[] payload, CompletionCallbackDelegate callback) + public void StartServerStreaming(byte[] payload, CompletionCallbackDelegate callback, MetadataArraySafeHandle metadataArray) { - AssertCallOk(grpcsharp_call_start_server_streaming(this, callback, payload, new UIntPtr((ulong)payload.Length))); + AssertCallOk(grpcsharp_call_start_server_streaming(this, callback, payload, new UIntPtr((ulong)payload.Length), metadataArray)); } - public void StartDuplexStreaming(CompletionCallbackDelegate callback) + public void StartDuplexStreaming(CompletionCallbackDelegate callback, MetadataArraySafeHandle metadataArray) { - AssertCallOk(grpcsharp_call_start_duplex_streaming(this, callback)); + AssertCallOk(grpcsharp_call_start_duplex_streaming(this, callback, metadataArray)); } public void StartSendMessage(byte[] payload, CompletionCallbackDelegate callback) diff --git a/src/csharp/Grpc.Core/Internal/MetadataArraySafeHandle.cs b/src/csharp/Grpc.Core/Internal/MetadataArraySafeHandle.cs new file mode 100644 index 00000000000..c9c4d954c91 --- /dev/null +++ b/src/csharp/Grpc.Core/Internal/MetadataArraySafeHandle.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.InteropServices; +using System.Threading.Tasks; + +namespace Grpc.Core.Internal +{ + /// + /// grpc_metadata_array from + /// + internal class MetadataArraySafeHandle : SafeHandleZeroIsInvalid + { + [DllImport("grpc_csharp_ext.dll")] + static extern MetadataArraySafeHandle grpcsharp_metadata_array_create(UIntPtr capacity); + + [DllImport("grpc_csharp_ext.dll", CharSet = CharSet.Ansi)] + static extern void grpcsharp_metadata_array_add(MetadataArraySafeHandle array, string key, byte[] value, UIntPtr valueLength); + + [DllImport("grpc_csharp_ext.dll")] + static extern void grpcsharp_metadata_array_destroy_full(IntPtr array); + + private MetadataArraySafeHandle() + { + } + + public static MetadataArraySafeHandle Create(Metadata metadata) + { + var entries = metadata.Entries; + var metadataArray = grpcsharp_metadata_array_create(new UIntPtr((ulong)entries.Count)); + for (int i = 0; i < entries.Count; i++) + { + grpcsharp_metadata_array_add(metadataArray, entries[i].Key, entries[i].ValueBytes, new UIntPtr((ulong)entries[i].ValueBytes.Length)); + } + return metadataArray; + } + + protected override bool ReleaseHandle() + { + grpcsharp_metadata_array_destroy_full(handle); + return true; + } + } +} diff --git a/src/csharp/Grpc.Core/Marshaller.cs b/src/csharp/Grpc.Core/Marshaller.cs index e73e7b762ef..8b1a9290741 100644 --- a/src/csharp/Grpc.Core/Marshaller.cs +++ b/src/csharp/Grpc.Core/Marshaller.cs @@ -32,6 +32,7 @@ #endregion using System; +using Grpc.Core.Utils; namespace Grpc.Core { @@ -45,8 +46,8 @@ namespace Grpc.Core public Marshaller(Func serializer, Func deserializer) { - this.serializer = serializer; - this.deserializer = deserializer; + this.serializer = Preconditions.CheckNotNull(serializer); + this.deserializer = Preconditions.CheckNotNull(deserializer); } public Func Serializer diff --git a/src/csharp/Grpc.Core/Metadata.cs b/src/csharp/Grpc.Core/Metadata.cs new file mode 100644 index 00000000000..eccec26a616 --- /dev/null +++ b/src/csharp/Grpc.Core/Metadata.cs @@ -0,0 +1,126 @@ +#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.Collections.Immutable; +using System.Runtime.InteropServices; +using System.Text; + +namespace Grpc.Core +{ + /// + /// gRPC call metadata. + /// + public class Metadata + { + public static readonly Metadata Empty = new Metadata(ImmutableList.Empty); + + readonly ImmutableList entries; + + public Metadata(ImmutableList entries) + { + this.entries = entries; + } + + public ImmutableList Entries + { + get + { + return this.entries; + } + } + + public static Builder CreateBuilder() + { + return new Builder(); + } + + public struct MetadataEntry + { + readonly string key; + readonly byte[] valueBytes; + + public MetadataEntry(string key, byte[] valueBytes) + { + this.key = key; + this.valueBytes = valueBytes; + } + + public MetadataEntry(string key, string value) + { + this.key = key; + this.valueBytes = Encoding.ASCII.GetBytes(value); + } + + public string Key + { + get + { + return this.key; + } + } + + // TODO: using ByteString would guarantee immutability. + public byte[] ValueBytes + { + get + { + return this.valueBytes; + } + } + } + + public class Builder + { + readonly List entries = new List(); + + public List Entries + { + get + { + return entries; + } + } + + public Builder Add(MetadataEntry entry) + { + entries.Add(entry); + return this; + } + + public Metadata Build() + { + return new Metadata(entries.ToImmutableList()); + } + } + } +} diff --git a/src/csharp/Grpc.Core/ServerServiceDefinition.cs b/src/csharp/Grpc.Core/ServerServiceDefinition.cs index 004415477ca..f08c7d88f3f 100644 --- a/src/csharp/Grpc.Core/ServerServiceDefinition.cs +++ b/src/csharp/Grpc.Core/ServerServiceDefinition.cs @@ -43,12 +43,10 @@ namespace Grpc.Core /// public class ServerServiceDefinition { - readonly string serviceName; readonly ImmutableDictionary callHandlers; - private ServerServiceDefinition(string serviceName, ImmutableDictionary callHandlers) + private ServerServiceDefinition(ImmutableDictionary callHandlers) { - this.serviceName = serviceName; this.callHandlers = callHandlers; } @@ -79,7 +77,7 @@ namespace Grpc.Core Method method, UnaryRequestServerMethod handler) { - callHandlers.Add(method.Name, ServerCalls.UnaryRequestCall(method, handler)); + callHandlers.Add(GetFullMethodName(serviceName, method.Name), ServerCalls.UnaryRequestCall(method, handler)); return this; } @@ -87,13 +85,18 @@ namespace Grpc.Core Method method, StreamingRequestServerMethod handler) { - callHandlers.Add(method.Name, ServerCalls.StreamingRequestCall(method, handler)); + callHandlers.Add(GetFullMethodName(serviceName, method.Name), ServerCalls.StreamingRequestCall(method, handler)); return this; } public ServerServiceDefinition Build() { - return new ServerServiceDefinition(serviceName, callHandlers.ToImmutableDictionary()); + return new ServerServiceDefinition(callHandlers.ToImmutableDictionary()); + } + + private string GetFullMethodName(string serviceName, string methodName) + { + return serviceName + "/" + methodName; } } } diff --git a/src/csharp/Grpc.Core/Stub/AbstractStub.cs b/src/csharp/Grpc.Core/Stub/AbstractStub.cs new file mode 100644 index 00000000000..cf5ab958c5b --- /dev/null +++ b/src/csharp/Grpc.Core/Stub/AbstractStub.cs @@ -0,0 +1,73 @@ +#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: support adding timeout to methods. + /// + /// Base for client-side stubs. + /// + public abstract class AbstractStub + where TConfig : StubConfiguration + { + readonly Channel channel; + readonly TConfig config; + + public AbstractStub(Channel channel, TConfig config) + { + this.channel = channel; + this.config = config; + } + + public Channel Channel + { + get + { + return this.channel; + } + } + + /// + /// Creates a new call to given method. + /// + protected Call CreateCall(string serviceName, Method method) + { + var headerBuilder = Metadata.CreateBuilder(); + config.HeaderInterceptor(headerBuilder); + return new Call(serviceName, method, channel, headerBuilder.Build()); + } + } +} diff --git a/src/csharp/Grpc.Core/Stub/StubConfiguration.cs b/src/csharp/Grpc.Core/Stub/StubConfiguration.cs new file mode 100644 index 00000000000..5bcb5b40d2d --- /dev/null +++ b/src/csharp/Grpc.Core/Stub/StubConfiguration.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 Grpc.Core.Internal; +using Grpc.Core.Utils; + +namespace Grpc.Core +{ + public delegate void HeaderInterceptorDelegate(Metadata.Builder headerBuilder); + + public class StubConfiguration + { + /// + /// The default stub configuration. + /// + public static readonly StubConfiguration Default = new StubConfiguration((headerBuilder) => { }); + + readonly HeaderInterceptorDelegate headerInterceptor; + + public StubConfiguration(HeaderInterceptorDelegate headerInterceptor) + { + this.headerInterceptor = Preconditions.CheckNotNull(headerInterceptor); + } + + public HeaderInterceptorDelegate HeaderInterceptor + { + get + { + return headerInterceptor; + } + } + } +} diff --git a/src/csharp/Grpc.Examples.Tests/MathClientServerTests.cs b/src/csharp/Grpc.Examples.Tests/MathClientServerTests.cs index c86da65af47..32a523f2135 100644 --- a/src/csharp/Grpc.Examples.Tests/MathClientServerTests.cs +++ b/src/csharp/Grpc.Examples.Tests/MathClientServerTests.cs @@ -61,7 +61,14 @@ namespace math.Tests int port = server.AddPort(host + ":0"); server.Start(); channel = new Channel(host + ":" + port); - client = MathGrpc.NewStub(channel); + + // TODO: get rid of the custom header here once we have dedicated tests + // for header support. + var stubConfig = new StubConfiguration((headerBuilder) => + { + headerBuilder.Add(new Metadata.MetadataEntry("customHeader", "abcdef")); + }); + client = MathGrpc.NewStub(channel, stubConfig); } [TestFixtureTearDown] diff --git a/src/csharp/Grpc.Examples/MathGrpc.cs b/src/csharp/Grpc.Examples/MathGrpc.cs index 33a9ca92876..24e6a1de8e3 100644 --- a/src/csharp/Grpc.Examples/MathGrpc.cs +++ b/src/csharp/Grpc.Examples/MathGrpc.cs @@ -45,6 +45,8 @@ namespace math /// public class MathGrpc { + static readonly string ServiceName = "/math.Math"; + static readonly Marshaller DivArgsMarshaller = Marshallers.Create((arg) => arg.ToByteArray(), DivArgs.ParseFrom); static readonly Marshaller DivReplyMarshaller = Marshallers.Create((arg) => arg.ToByteArray(), DivReply.ParseFrom); static readonly Marshaller NumMarshaller = Marshallers.Create((arg) => arg.ToByteArray(), Num.ParseFrom); @@ -52,25 +54,25 @@ namespace math static readonly Method DivMethod = new Method( MethodType.Unary, - "/math.Math/Div", + "Div", DivArgsMarshaller, DivReplyMarshaller); static readonly Method FibMethod = new Method( MethodType.ServerStreaming, - "/math.Math/Fib", + "Fib", FibArgsMarshaller, NumMarshaller); static readonly Method SumMethod = new Method( MethodType.ClientStreaming, - "/math.Math/Sum", + "Sum", NumMarshaller, NumMarshaller); static readonly Method DivManyMethod = new Method( MethodType.DuplexStreaming, - "/math.Math/DivMany", + "DivMany", DivArgsMarshaller, DivReplyMarshaller); @@ -87,42 +89,43 @@ namespace math IObserver DivMany(IObserver responseObserver, CancellationToken token = default(CancellationToken)); } - public class MathServiceClientStub : IMathServiceClient + public class MathServiceClientStub : AbstractStub, IMathServiceClient { - readonly Channel channel; + public MathServiceClientStub(Channel channel) : this(channel, StubConfiguration.Default) + { + } - public MathServiceClientStub(Channel channel) + public MathServiceClientStub(Channel channel, StubConfiguration config) : base(channel, config) { - this.channel = channel; } public DivReply Div(DivArgs request, CancellationToken token = default(CancellationToken)) { - var call = new Grpc.Core.Call(DivMethod, channel); + var call = CreateCall(ServiceName, DivMethod); return Calls.BlockingUnaryCall(call, request, token); } public Task DivAsync(DivArgs request, CancellationToken token = default(CancellationToken)) { - var call = new Grpc.Core.Call(DivMethod, channel); + var call = CreateCall(ServiceName, DivMethod); return Calls.AsyncUnaryCall(call, request, token); } public void Fib(FibArgs request, IObserver responseObserver, CancellationToken token = default(CancellationToken)) { - var call = new Grpc.Core.Call(FibMethod, channel); + var call = CreateCall(ServiceName, FibMethod); Calls.AsyncServerStreamingCall(call, request, responseObserver, token); } public ClientStreamingAsyncResult Sum(CancellationToken token = default(CancellationToken)) { - var call = new Grpc.Core.Call(SumMethod, channel); + var call = CreateCall(ServiceName, SumMethod); return Calls.AsyncClientStreamingCall(call, token); } public IObserver DivMany(IObserver responseObserver, CancellationToken token = default(CancellationToken)) { - var call = new Grpc.Core.Call(DivManyMethod, channel); + var call = CreateCall(ServiceName, DivManyMethod); return Calls.DuplexStreamingCall(call, responseObserver, token); } } @@ -141,7 +144,7 @@ namespace math public static ServerServiceDefinition BindService(IMathService serviceImpl) { - return ServerServiceDefinition.CreateBuilder("/math.Math/") + return ServerServiceDefinition.CreateBuilder(ServiceName) .AddMethod(DivMethod, serviceImpl.Div) .AddMethod(FibMethod, serviceImpl.Fib) .AddMethod(SumMethod, serviceImpl.Sum) @@ -152,5 +155,10 @@ namespace math { return new MathServiceClientStub(channel); } + + public static IMathServiceClient NewStub(Channel channel, StubConfiguration config) + { + return new MathServiceClientStub(channel, config); + } } } diff --git a/src/csharp/Grpc.IntegrationTesting/Grpc.IntegrationTesting.csproj b/src/csharp/Grpc.IntegrationTesting/Grpc.IntegrationTesting.csproj index cfb258711ad..c3e5f03074a 100644 --- a/src/csharp/Grpc.IntegrationTesting/Grpc.IntegrationTesting.csproj +++ b/src/csharp/Grpc.IntegrationTesting/Grpc.IntegrationTesting.csproj @@ -39,8 +39,7 @@ ..\packages\Google.ProtocolBuffers.2.4.1.521\lib\net40\Google.ProtocolBuffers.dll - - False + ..\packages\System.Collections.Immutable.1.1.34-rc\lib\portable-net45+win8+wp8+wpa81\System.Collections.Immutable.dll diff --git a/src/csharp/Grpc.IntegrationTesting/TestServiceGrpc.cs b/src/csharp/Grpc.IntegrationTesting/TestServiceGrpc.cs index 9b0251c3ca9..f63e0361a45 100644 --- a/src/csharp/Grpc.IntegrationTesting/TestServiceGrpc.cs +++ b/src/csharp/Grpc.IntegrationTesting/TestServiceGrpc.cs @@ -44,6 +44,8 @@ namespace grpc.testing /// public class TestServiceGrpc { + static readonly string ServiceName = "/grpc.testing.TestService"; + static readonly Marshaller EmptyMarshaller = Marshallers.Create((arg) => arg.ToByteArray(), Empty.ParseFrom); static readonly Marshaller SimpleRequestMarshaller = Marshallers.Create((arg) => arg.ToByteArray(), SimpleRequest.ParseFrom); static readonly Marshaller SimpleResponseMarshaller = Marshallers.Create((arg) => arg.ToByteArray(), SimpleResponse.ParseFrom); @@ -54,37 +56,37 @@ namespace grpc.testing static readonly Method EmptyCallMethod = new Method( MethodType.Unary, - "/grpc.testing.TestService/EmptyCall", + "EmptyCall", EmptyMarshaller, EmptyMarshaller); static readonly Method UnaryCallMethod = new Method( MethodType.Unary, - "/grpc.testing.TestService/UnaryCall", + "UnaryCall", SimpleRequestMarshaller, SimpleResponseMarshaller); static readonly Method StreamingOutputCallMethod = new Method( MethodType.ServerStreaming, - "/grpc.testing.TestService/StreamingOutputCall", + "StreamingOutputCall", StreamingOutputCallRequestMarshaller, StreamingOutputCallResponseMarshaller); static readonly Method StreamingInputCallMethod = new Method( MethodType.ClientStreaming, - "/grpc.testing.TestService/StreamingInputCall", + "StreamingInputCall", StreamingInputCallRequestMarshaller, StreamingInputCallResponseMarshaller); static readonly Method FullDuplexCallMethod = new Method( MethodType.DuplexStreaming, - "/grpc.testing.TestService/FullDuplexCall", + "FullDuplexCall", StreamingOutputCallRequestMarshaller, StreamingOutputCallResponseMarshaller); static readonly Method HalfDuplexCallMethod = new Method( MethodType.DuplexStreaming, - "/grpc.testing.TestService/HalfDuplexCall", + "HalfDuplexCall", StreamingOutputCallRequestMarshaller, StreamingOutputCallResponseMarshaller); @@ -107,60 +109,61 @@ namespace grpc.testing IObserver HalfDuplexCall(IObserver responseObserver, CancellationToken token = default(CancellationToken)); } - public class TestServiceClientStub : ITestServiceClient + public class TestServiceClientStub : AbstractStub, ITestServiceClient { - readonly Channel channel; + public TestServiceClientStub(Channel channel) : base(channel, StubConfiguration.Default) + { + } - public TestServiceClientStub(Channel channel) + public TestServiceClientStub(Channel channel, StubConfiguration config) : base(channel, config) { - this.channel = channel; } public Empty EmptyCall(Empty request, CancellationToken token = default(CancellationToken)) { - var call = new Grpc.Core.Call(EmptyCallMethod, channel); + var call = CreateCall(ServiceName, EmptyCallMethod); return Calls.BlockingUnaryCall(call, request, token); } public Task EmptyCallAsync(Empty request, CancellationToken token = default(CancellationToken)) { - var call = new Grpc.Core.Call(EmptyCallMethod, channel); + var call = CreateCall(ServiceName, EmptyCallMethod); return Calls.AsyncUnaryCall(call, request, token); } public SimpleResponse UnaryCall(SimpleRequest request, CancellationToken token = default(CancellationToken)) { - var call = new Grpc.Core.Call(UnaryCallMethod, channel); + var call = CreateCall(ServiceName, UnaryCallMethod); return Calls.BlockingUnaryCall(call, request, token); } public Task UnaryCallAsync(SimpleRequest request, CancellationToken token = default(CancellationToken)) { - var call = new Grpc.Core.Call(UnaryCallMethod, channel); + var call = CreateCall(ServiceName, UnaryCallMethod); return Calls.AsyncUnaryCall(call, request, token); } public void StreamingOutputCall(StreamingOutputCallRequest request, IObserver responseObserver, CancellationToken token = default(CancellationToken)) { - var call = new Grpc.Core.Call(StreamingOutputCallMethod, channel); + var call = CreateCall(ServiceName, StreamingOutputCallMethod); Calls.AsyncServerStreamingCall(call, request, responseObserver, token); } public ClientStreamingAsyncResult StreamingInputCall(CancellationToken token = default(CancellationToken)) { - var call = new Grpc.Core.Call(StreamingInputCallMethod, channel); + var call = CreateCall(ServiceName, StreamingInputCallMethod); return Calls.AsyncClientStreamingCall(call, token); } public IObserver FullDuplexCall(IObserver responseObserver, CancellationToken token = default(CancellationToken)) { - var call = new Grpc.Core.Call(FullDuplexCallMethod, channel); + var call = CreateCall(ServiceName, FullDuplexCallMethod); return Calls.DuplexStreamingCall(call, responseObserver, token); } public IObserver HalfDuplexCall(IObserver responseObserver, CancellationToken token = default(CancellationToken)) { - var call = new Grpc.Core.Call(HalfDuplexCallMethod, channel); + var call = CreateCall(ServiceName, HalfDuplexCallMethod); return Calls.DuplexStreamingCall(call, responseObserver, token); } } @@ -183,7 +186,7 @@ namespace grpc.testing public static ServerServiceDefinition BindService(ITestService serviceImpl) { - return ServerServiceDefinition.CreateBuilder("/grpc.testing.TestService/") + return ServerServiceDefinition.CreateBuilder(ServiceName) .AddMethod(EmptyCallMethod, serviceImpl.EmptyCall) .AddMethod(UnaryCallMethod, serviceImpl.UnaryCall) .AddMethod(StreamingOutputCallMethod, serviceImpl.StreamingOutputCall) diff --git a/src/csharp/ext/grpc_csharp_ext.c b/src/csharp/ext/grpc_csharp_ext.c index 51abb632f7a..9a1c908d11b 100644 --- a/src/csharp/ext/grpc_csharp_ext.c +++ b/src/csharp/ext/grpc_csharp_ext.c @@ -102,34 +102,114 @@ grpcsharp_batch_context *grpcsharp_batch_context_create() { return ctx; } -/** - * Destroys metadata array including keys and values. +/* + * Destroys array->metadata. + * The array pointer itself is not freed. + */ +void grpcsharp_metadata_array_destroy_metadata_only( + grpc_metadata_array *array) { + gpr_free(array->metadata); +} + +/* + * Destroys keys, values and array->metadata. + * The array pointer itself is not freed. + */ +void grpcsharp_metadata_array_destroy_metadata_including_entries( + grpc_metadata_array *array) { + size_t i; + if (array->metadata) { + for (i = 0; i < array->count; i++) { + gpr_free((void *)array->metadata[i].key); + gpr_free((void *)array->metadata[i].value); + } + } + gpr_free(array->metadata); +} + +/* + * Fully destroys the metadata array. + */ +GPR_EXPORT void GPR_CALLTYPE +grpcsharp_metadata_array_destroy_full(grpc_metadata_array *array) { + if (!array) { + return; + } + grpcsharp_metadata_array_destroy_metadata_including_entries(array); + gpr_free(array); +} + +/* + * Creates an empty metadata array with given capacity. + * Array can later be destroyed by grpc_metadata_array_destroy_full. */ -void grpcsharp_metadata_array_destroy_recursive(grpc_metadata_array *array) { - if (!array->metadata) { +GPR_EXPORT grpc_metadata_array *GPR_CALLTYPE +grpcsharp_metadata_array_create(size_t capacity) { + grpc_metadata_array *array = + (grpc_metadata_array *)gpr_malloc(sizeof(grpc_metadata_array)); + grpc_metadata_array_init(array); + array->capacity = capacity; + array->count = 0; + if (capacity > 0) { + array->metadata = + (grpc_metadata *)gpr_malloc(sizeof(grpc_metadata) * capacity); + memset(array->metadata, 0, sizeof(grpc_metadata) * capacity); + } else { + array->metadata = NULL; + } + return array; +} + +GPR_EXPORT void GPR_CALLTYPE +grpcsharp_metadata_array_add(grpc_metadata_array *array, const char *key, + const char *value, size_t value_length) { + size_t i = array->count; + GPR_ASSERT(array->count < array->capacity); + array->metadata[i].key = gpr_strdup(key); + array->metadata[i].value = (char *)gpr_malloc(value_length); + memcpy((void *)array->metadata[i].value, value, value_length); + array->metadata[i].value_length = value_length; + array->count++; +} + +/* Move contents of metadata array */ +void grpcsharp_metadata_array_move(grpc_metadata_array *dest, + grpc_metadata_array *src) { + if (!src) { + dest->capacity = 0; + dest->count = 0; + dest->metadata = NULL; return; } - /* TODO: destroy also keys and values */ - grpc_metadata_array_destroy(array); + + dest->capacity = src->capacity; + dest->count = src->count; + dest->metadata = src->metadata; + + src->capacity = 0; + src->count = 0; + src->metadata = NULL; } void grpcsharp_batch_context_destroy(grpcsharp_batch_context *ctx) { if (!ctx) { return; } - grpcsharp_metadata_array_destroy_recursive(&(ctx->send_initial_metadata)); + grpcsharp_metadata_array_destroy_metadata_including_entries( + &(ctx->send_initial_metadata)); grpc_byte_buffer_destroy(ctx->send_message); - grpcsharp_metadata_array_destroy_recursive( + grpcsharp_metadata_array_destroy_metadata_including_entries( &(ctx->send_status_from_server.trailing_metadata)); gpr_free(ctx->send_status_from_server.status_details); - grpc_metadata_array_destroy(&(ctx->recv_initial_metadata)); + grpcsharp_metadata_array_destroy_metadata_only(&(ctx->recv_initial_metadata)); grpc_byte_buffer_destroy(ctx->recv_message); - grpc_metadata_array_destroy(&(ctx->recv_status_on_client.trailing_metadata)); + grpcsharp_metadata_array_destroy_metadata_only( + &(ctx->recv_status_on_client.trailing_metadata)); gpr_free((void *)ctx->recv_status_on_client.status_details); /* NOTE: ctx->server_rpc_new.call is not destroyed because callback handler is @@ -137,7 +217,8 @@ void grpcsharp_batch_context_destroy(grpcsharp_batch_context *ctx) { to take its ownership. */ grpc_call_details_destroy(&(ctx->server_rpc_new.call_details)); - grpc_metadata_array_destroy(&(ctx->server_rpc_new.request_metadata)); + grpcsharp_metadata_array_destroy_metadata_only( + &(ctx->server_rpc_new.request_metadata)); gpr_free(ctx); } @@ -346,17 +427,19 @@ grpcsharp_call_start_write_from_copied_buffer(grpc_call *call, GPR_EXPORT grpc_call_error GPR_CALLTYPE grpcsharp_call_start_unary(grpc_call *call, callback_funcptr callback, - const char *send_buffer, size_t send_buffer_len) { + const char *send_buffer, size_t send_buffer_len, + grpc_metadata_array *initial_metadata) { /* TODO: don't use magic number */ grpc_op ops[6]; grpcsharp_batch_context *ctx = grpcsharp_batch_context_create(); ctx->callback = callback; - /* TODO: implement sending the metadata... */ ops[0].op = GRPC_OP_SEND_INITIAL_METADATA; - /* ctx->send_initial_metadata is already zeroed out. */ - ops[0].data.send_initial_metadata.count = 0; - ops[0].data.send_initial_metadata.metadata = NULL; + grpcsharp_metadata_array_move(&(ctx->send_initial_metadata), + initial_metadata); + ops[0].data.send_initial_metadata.count = ctx->send_initial_metadata.count; + ops[0].data.send_initial_metadata.metadata = + ctx->send_initial_metadata.metadata; ops[1].op = GRPC_OP_SEND_MESSAGE; ctx->send_message = string_to_byte_buffer(send_buffer, send_buffer_len); @@ -389,9 +472,11 @@ GPR_EXPORT void GPR_CALLTYPE grpcsharp_call_blocking_unary(grpc_call *call, grpc_completion_queue *dedicated_cq, callback_funcptr callback, - const char *send_buffer, size_t send_buffer_len) { + const char *send_buffer, size_t send_buffer_len, + grpc_metadata_array *initial_metadata) { GPR_ASSERT(grpcsharp_call_start_unary(call, callback, send_buffer, - send_buffer_len) == GRPC_CALL_OK); + send_buffer_len, + initial_metadata) == GRPC_CALL_OK); /* TODO: we would like to use pluck, but we don't know the tag */ GPR_ASSERT(grpcsharp_completion_queue_next_with_callback(dedicated_cq) == @@ -403,17 +488,19 @@ grpcsharp_call_blocking_unary(grpc_call *call, GPR_EXPORT grpc_call_error GPR_CALLTYPE grpcsharp_call_start_client_streaming(grpc_call *call, - callback_funcptr callback) { + callback_funcptr callback, + grpc_metadata_array *initial_metadata) { /* TODO: don't use magic number */ grpc_op ops[4]; grpcsharp_batch_context *ctx = grpcsharp_batch_context_create(); ctx->callback = callback; - /* TODO: implement sending the metadata... */ ops[0].op = GRPC_OP_SEND_INITIAL_METADATA; - /* ctx->send_initial_metadata is already zeroed out. */ - ops[0].data.send_initial_metadata.count = 0; - ops[0].data.send_initial_metadata.metadata = NULL; + grpcsharp_metadata_array_move(&(ctx->send_initial_metadata), + initial_metadata); + ops[0].data.send_initial_metadata.count = ctx->send_initial_metadata.count; + ops[0].data.send_initial_metadata.metadata = + ctx->send_initial_metadata.metadata; ops[1].op = GRPC_OP_RECV_INITIAL_METADATA; ops[1].data.recv_initial_metadata = &(ctx->recv_initial_metadata); @@ -435,21 +522,20 @@ grpcsharp_call_start_client_streaming(grpc_call *call, return grpc_call_start_batch(call, ops, sizeof(ops) / sizeof(ops[0]), ctx); } -GPR_EXPORT grpc_call_error GPR_CALLTYPE -grpcsharp_call_start_server_streaming(grpc_call *call, - callback_funcptr callback, - const char *send_buffer, - size_t send_buffer_len) { +GPR_EXPORT grpc_call_error GPR_CALLTYPE grpcsharp_call_start_server_streaming( + grpc_call *call, callback_funcptr callback, const char *send_buffer, + size_t send_buffer_len, grpc_metadata_array *initial_metadata) { /* TODO: don't use magic number */ grpc_op ops[5]; grpcsharp_batch_context *ctx = grpcsharp_batch_context_create(); ctx->callback = callback; - /* TODO: implement sending the metadata... */ ops[0].op = GRPC_OP_SEND_INITIAL_METADATA; - /* ctx->send_initial_metadata is already zeroed out. */ - ops[0].data.send_initial_metadata.count = 0; - ops[0].data.send_initial_metadata.metadata = NULL; + grpcsharp_metadata_array_move(&(ctx->send_initial_metadata), + initial_metadata); + ops[0].data.send_initial_metadata.count = ctx->send_initial_metadata.count; + ops[0].data.send_initial_metadata.metadata = + ctx->send_initial_metadata.metadata; ops[1].op = GRPC_OP_SEND_MESSAGE; ctx->send_message = string_to_byte_buffer(send_buffer, send_buffer_len); @@ -476,17 +562,19 @@ grpcsharp_call_start_server_streaming(grpc_call *call, GPR_EXPORT grpc_call_error GPR_CALLTYPE grpcsharp_call_start_duplex_streaming(grpc_call *call, - callback_funcptr callback) { + callback_funcptr callback, + grpc_metadata_array *initial_metadata) { /* TODO: don't use magic number */ grpc_op ops[3]; grpcsharp_batch_context *ctx = grpcsharp_batch_context_create(); ctx->callback = callback; - /* TODO: implement sending the metadata... */ ops[0].op = GRPC_OP_SEND_INITIAL_METADATA; - /* ctx->send_initial_metadata is already zeroed out. */ - ops[0].data.send_initial_metadata.count = 0; - ops[0].data.send_initial_metadata.metadata = NULL; + grpcsharp_metadata_array_move(&(ctx->send_initial_metadata), + initial_metadata); + ops[0].data.send_initial_metadata.count = ctx->send_initial_metadata.count; + ops[0].data.send_initial_metadata.metadata = + ctx->send_initial_metadata.metadata; ops[1].op = GRPC_OP_RECV_INITIAL_METADATA; ops[1].data.recv_initial_metadata = &(ctx->recv_initial_metadata);