diff options
Diffstat (limited to 'src/csharp/Grpc.Core/Internal')
13 files changed, 252 insertions, 73 deletions
diff --git a/src/csharp/Grpc.Core/Internal/AsyncCall.cs b/src/csharp/Grpc.Core/Internal/AsyncCall.cs index f549c52876..49d0a111ef 100644 --- a/src/csharp/Grpc.Core/Internal/AsyncCall.cs +++ b/src/csharp/Grpc.Core/Internal/AsyncCall.cs @@ -265,7 +265,7 @@ namespace Grpc.Core.Internal // the halfclose has already been done implicitly, so just return // completed task here. halfcloseRequested = true; - return Task.FromResult<object>(null); + return TaskUtils.CompletedTask; } call.StartSendCloseFromClient(HandleSendFinished); @@ -341,6 +341,11 @@ namespace Grpc.Core.Internal get { return true; } } + protected override Exception GetRpcExceptionClientOnly() + { + return new RpcException(finishedStatus.Value.Status); + } + protected override Task CheckSendAllowedOrEarlyResult() { var earlyResult = CheckSendPreconditionsClientSide(); @@ -452,6 +457,7 @@ namespace Grpc.Core.Internal using (Profilers.ForCurrentThread().NewScope("AsyncCall.HandleUnaryResponse")) { + TaskCompletionSource<object> delayedStreamingWriteTcs = null; TResponse msg = default(TResponse); var deserializeException = TryDeserialize(receivedMessage, out msg); @@ -465,13 +471,23 @@ namespace Grpc.Core.Internal } finishedStatus = receivedStatus; + if (isStreamingWriteCompletionDelayed) + { + delayedStreamingWriteTcs = streamingWriteTcs; + streamingWriteTcs = null; + } + ReleaseResourcesIfPossible(); } responseHeadersTcs.SetResult(responseHeaders); - var status = receivedStatus.Status; + if (delayedStreamingWriteTcs != null) + { + delayedStreamingWriteTcs.SetException(GetRpcExceptionClientOnly()); + } + var status = receivedStatus.Status; if (status.StatusCode != StatusCode.OK) { unaryResponseTcs.SetException(new RpcException(status)); @@ -490,16 +506,27 @@ namespace Grpc.Core.Internal // NOTE: because this event is a result of batch containing GRPC_OP_RECV_STATUS_ON_CLIENT, // success will be always set to true. + TaskCompletionSource<object> delayedStreamingWriteTcs = null; + lock (myLock) { finished = true; finishedStatus = receivedStatus; + if (isStreamingWriteCompletionDelayed) + { + delayedStreamingWriteTcs = streamingWriteTcs; + streamingWriteTcs = null; + } ReleaseResourcesIfPossible(); } - var status = receivedStatus.Status; + if (delayedStreamingWriteTcs != null) + { + delayedStreamingWriteTcs.SetException(GetRpcExceptionClientOnly()); + } + var status = receivedStatus.Status; if (status.StatusCode != StatusCode.OK) { streamingCallFinishedTcs.SetException(new RpcException(status)); diff --git a/src/csharp/Grpc.Core/Internal/AsyncCallBase.cs b/src/csharp/Grpc.Core/Internal/AsyncCallBase.cs index eb9c3ea62d..9f9d260e7e 100644 --- a/src/csharp/Grpc.Core/Internal/AsyncCallBase.cs +++ b/src/csharp/Grpc.Core/Internal/AsyncCallBase.cs @@ -69,6 +69,7 @@ namespace Grpc.Core.Internal protected TaskCompletionSource<TRead> streamingReadTcs; // Completion of a pending streaming read if not null. protected TaskCompletionSource<object> streamingWriteTcs; // Completion of a pending streaming write or send close from client if not null. protected TaskCompletionSource<object> sendStatusFromServerTcs; + protected bool isStreamingWriteCompletionDelayed; // Only used for the client side. protected bool readingDone; // True if last read (i.e. read with null payload) was already received. protected bool halfcloseRequested; // True if send close have been initiated. @@ -200,6 +201,12 @@ namespace Grpc.Core.Internal get; } + /// <summary> + /// Returns an exception to throw for a failed send operation. + /// It is only allowed to call this method for a call that has already finished. + /// </summary> + protected abstract Exception GetRpcExceptionClientOnly(); + private void ReleaseResources() { if (call != null) @@ -252,18 +259,43 @@ namespace Grpc.Core.Internal /// </summary> protected void HandleSendFinished(bool success) { + bool delayCompletion = false; TaskCompletionSource<object> origTcs = null; lock (myLock) { - origTcs = streamingWriteTcs; - streamingWriteTcs = null; + if (!success && !finished && IsClient) { + // We should be setting this only once per call, following writes will be short circuited + // because they cannot start until the entire call finishes. + GrpcPreconditions.CheckState(!isStreamingWriteCompletionDelayed); + + // leave streamingWriteTcs set, it will be completed once call finished. + isStreamingWriteCompletionDelayed = true; + delayCompletion = true; + } + else + { + origTcs = streamingWriteTcs; + streamingWriteTcs = null; + } ReleaseResourcesIfPossible(); } if (!success) { - origTcs.SetException(new InvalidOperationException("Send failed")); + if (!delayCompletion) + { + if (IsClient) + { + GrpcPreconditions.CheckState(finished); // implied by !success && !delayCompletion && IsClient + origTcs.SetException(GetRpcExceptionClientOnly()); + } + else + { + origTcs.SetException (new IOException("Error sending from server.")); + } + } + // if delayCompletion == true, postpone SetException until call finishes. } else { @@ -283,7 +315,7 @@ namespace Grpc.Core.Internal if (!success) { - sendStatusFromServerTcs.SetException(new InvalidOperationException("Error sending status from server.")); + sendStatusFromServerTcs.SetException(new IOException("Error sending status from server.")); } else { diff --git a/src/csharp/Grpc.Core/Internal/AsyncCallServer.cs b/src/csharp/Grpc.Core/Internal/AsyncCallServer.cs index 56c23ba3ef..50fdfa9006 100644 --- a/src/csharp/Grpc.Core/Internal/AsyncCallServer.cs +++ b/src/csharp/Grpc.Core/Internal/AsyncCallServer.cs @@ -33,6 +33,7 @@ using System; using System.Diagnostics; +using System.IO; using System.Runtime.CompilerServices; using System.Runtime.InteropServices; using System.Threading; @@ -193,6 +194,11 @@ namespace Grpc.Core.Internal get { return false; } } + protected override Exception GetRpcExceptionClientOnly() + { + throw new InvalidOperationException("Call be only called for client calls"); + } + protected override void OnAfterReleaseResources() { server.RemoveCallReference(this); diff --git a/src/csharp/Grpc.Core/Internal/BatchContextSafeHandle.cs b/src/csharp/Grpc.Core/Internal/BatchContextSafeHandle.cs index c28a6f64d3..26449ee539 100644 --- a/src/csharp/Grpc.Core/Internal/BatchContextSafeHandle.cs +++ b/src/csharp/Grpc.Core/Internal/BatchContextSafeHandle.cs @@ -93,21 +93,6 @@ namespace Grpc.Core.Internal return data; } - // Gets data of server_rpc_new completion. - public ServerRpcNew GetServerRpcNew(Server server) - { - var call = Native.grpcsharp_batch_context_server_rpc_new_call(this); - - var method = Marshal.PtrToStringAnsi(Native.grpcsharp_batch_context_server_rpc_new_method(this)); - var host = Marshal.PtrToStringAnsi(Native.grpcsharp_batch_context_server_rpc_new_host(this)); - var deadline = Native.grpcsharp_batch_context_server_rpc_new_deadline(this); - - IntPtr metadataArrayPtr = Native.grpcsharp_batch_context_server_rpc_new_request_metadata(this); - var metadata = MetadataArraySafeHandle.ReadMetadataFromPtrUnsafe(metadataArrayPtr); - - return new ServerRpcNew(server, call, method, host, deadline, metadata); - } - // Gets data of receive_close_on_server completion. public bool GetReceivedCloseOnServerCancelled() { diff --git a/src/csharp/Grpc.Core/Internal/CompletionRegistry.cs b/src/csharp/Grpc.Core/Internal/CompletionRegistry.cs index 628844f242..7e2f0e9c6c 100644 --- a/src/csharp/Grpc.Core/Internal/CompletionRegistry.cs +++ b/src/csharp/Grpc.Core/Internal/CompletionRegistry.cs @@ -44,6 +44,8 @@ namespace Grpc.Core.Internal internal delegate void BatchCompletionDelegate(bool success, BatchContextSafeHandle ctx); + internal delegate void RequestCallCompletionDelegate(bool success, RequestCallContextSafeHandle ctx); + internal class CompletionRegistry { static readonly ILogger Logger = GrpcEnvironment.Logger.ForType<CompletionRegistry>(); @@ -68,6 +70,12 @@ namespace Grpc.Core.Internal Register(ctx.Handle, opCallback); } + public void RegisterRequestCallCompletion(RequestCallContextSafeHandle ctx, RequestCallCompletionDelegate callback) + { + OpCompletionDelegate opCallback = ((success) => HandleRequestCallCompletion(success, ctx, callback)); + Register(ctx.Handle, opCallback); + } + public OpCompletionDelegate Extract(IntPtr key) { OpCompletionDelegate value; @@ -84,7 +92,26 @@ namespace Grpc.Core.Internal } catch (Exception e) { - Logger.Error(e, "Exception occured while invoking completion delegate."); + Logger.Error(e, "Exception occured while invoking batch completion delegate."); + } + finally + { + if (ctx != null) + { + ctx.Dispose(); + } + } + } + + private static void HandleRequestCallCompletion(bool success, RequestCallContextSafeHandle ctx, RequestCallCompletionDelegate callback) + { + try + { + callback(success, ctx); + } + catch (Exception e) + { + Logger.Error(e, "Exception occured while invoking request call completion delegate."); } finally { diff --git a/src/csharp/Grpc.Core/Internal/InterceptingCallInvoker.cs b/src/csharp/Grpc.Core/Internal/InterceptingCallInvoker.cs index ef48dc7121..0c63e2092a 100644 --- a/src/csharp/Grpc.Core/Internal/InterceptingCallInvoker.cs +++ b/src/csharp/Grpc.Core/Internal/InterceptingCallInvoker.cs @@ -48,7 +48,7 @@ namespace Grpc.Core.Internal readonly Func<CallOptions, CallOptions> callOptionsInterceptor; /// <summary> - /// Initializes a new instance of the <see cref="Grpc.Core.InterceptingCallInvoker"/> class. + /// Initializes a new instance of the <see cref="Grpc.Core.Internal.InterceptingCallInvoker"/> class. /// </summary> public InterceptingCallInvoker(CallInvoker callInvoker, Func<string, string> hostInterceptor = null, diff --git a/src/csharp/Grpc.Core/Internal/NativeMetadataCredentialsPlugin.cs b/src/csharp/Grpc.Core/Internal/NativeMetadataCredentialsPlugin.cs index 26af6311d5..b3714481eb 100644 --- a/src/csharp/Grpc.Core/Internal/NativeMetadataCredentialsPlugin.cs +++ b/src/csharp/Grpc.Core/Internal/NativeMetadataCredentialsPlugin.cs @@ -78,7 +78,10 @@ namespace Grpc.Core.Internal { var context = new AuthInterceptorContext(Marshal.PtrToStringAnsi(serviceUrlPtr), Marshal.PtrToStringAnsi(methodNamePtr)); - StartGetMetadata(context, callbackPtr, userDataPtr); + // Don't await, we are in a native callback and need to return. + #pragma warning disable 4014 + GetMetadataAsync(context, callbackPtr, userDataPtr); + #pragma warning restore 4014 } catch (Exception e) { @@ -87,7 +90,7 @@ namespace Grpc.Core.Internal } } - private async Task StartGetMetadata(AuthInterceptorContext context, IntPtr callbackPtr, IntPtr userDataPtr) + private async Task GetMetadataAsync(AuthInterceptorContext context, IntPtr callbackPtr, IntPtr userDataPtr) { try { diff --git a/src/csharp/Grpc.Core/Internal/NativeMethods.cs b/src/csharp/Grpc.Core/Internal/NativeMethods.cs index f457c9dbf1..40ba7e30cb 100644 --- a/src/csharp/Grpc.Core/Internal/NativeMethods.cs +++ b/src/csharp/Grpc.Core/Internal/NativeMethods.cs @@ -64,14 +64,17 @@ namespace Grpc.Core.Internal public readonly Delegates.grpcsharp_batch_context_recv_status_on_client_status_delegate grpcsharp_batch_context_recv_status_on_client_status; public readonly Delegates.grpcsharp_batch_context_recv_status_on_client_details_delegate grpcsharp_batch_context_recv_status_on_client_details; public readonly Delegates.grpcsharp_batch_context_recv_status_on_client_trailing_metadata_delegate grpcsharp_batch_context_recv_status_on_client_trailing_metadata; - public readonly Delegates.grpcsharp_batch_context_server_rpc_new_call_delegate grpcsharp_batch_context_server_rpc_new_call; - public readonly Delegates.grpcsharp_batch_context_server_rpc_new_method_delegate grpcsharp_batch_context_server_rpc_new_method; - public readonly Delegates.grpcsharp_batch_context_server_rpc_new_host_delegate grpcsharp_batch_context_server_rpc_new_host; - public readonly Delegates.grpcsharp_batch_context_server_rpc_new_deadline_delegate grpcsharp_batch_context_server_rpc_new_deadline; - public readonly Delegates.grpcsharp_batch_context_server_rpc_new_request_metadata_delegate grpcsharp_batch_context_server_rpc_new_request_metadata; public readonly Delegates.grpcsharp_batch_context_recv_close_on_server_cancelled_delegate grpcsharp_batch_context_recv_close_on_server_cancelled; public readonly Delegates.grpcsharp_batch_context_destroy_delegate grpcsharp_batch_context_destroy; + public readonly Delegates.grpcsharp_request_call_context_create_delegate grpcsharp_request_call_context_create; + public readonly Delegates.grpcsharp_request_call_context_call_delegate grpcsharp_request_call_context_call; + public readonly Delegates.grpcsharp_request_call_context_method_delegate grpcsharp_request_call_context_method; + public readonly Delegates.grpcsharp_request_call_context_host_delegate grpcsharp_request_call_context_host; + public readonly Delegates.grpcsharp_request_call_context_deadline_delegate grpcsharp_request_call_context_deadline; + public readonly Delegates.grpcsharp_request_call_context_request_metadata_delegate grpcsharp_request_call_context_request_metadata; + public readonly Delegates.grpcsharp_request_call_context_destroy_delegate grpcsharp_request_call_context_destroy; + public readonly Delegates.grpcsharp_composite_call_credentials_create_delegate grpcsharp_composite_call_credentials_create; public readonly Delegates.grpcsharp_call_credentials_release_delegate grpcsharp_call_credentials_release; @@ -170,14 +173,17 @@ namespace Grpc.Core.Internal this.grpcsharp_batch_context_recv_status_on_client_status = GetMethodDelegate<Delegates.grpcsharp_batch_context_recv_status_on_client_status_delegate>(library); this.grpcsharp_batch_context_recv_status_on_client_details = GetMethodDelegate<Delegates.grpcsharp_batch_context_recv_status_on_client_details_delegate>(library); this.grpcsharp_batch_context_recv_status_on_client_trailing_metadata = GetMethodDelegate<Delegates.grpcsharp_batch_context_recv_status_on_client_trailing_metadata_delegate>(library); - this.grpcsharp_batch_context_server_rpc_new_call = GetMethodDelegate<Delegates.grpcsharp_batch_context_server_rpc_new_call_delegate>(library); - this.grpcsharp_batch_context_server_rpc_new_method = GetMethodDelegate<Delegates.grpcsharp_batch_context_server_rpc_new_method_delegate>(library); - this.grpcsharp_batch_context_server_rpc_new_host = GetMethodDelegate<Delegates.grpcsharp_batch_context_server_rpc_new_host_delegate>(library); - this.grpcsharp_batch_context_server_rpc_new_deadline = GetMethodDelegate<Delegates.grpcsharp_batch_context_server_rpc_new_deadline_delegate>(library); - this.grpcsharp_batch_context_server_rpc_new_request_metadata = GetMethodDelegate<Delegates.grpcsharp_batch_context_server_rpc_new_request_metadata_delegate>(library); this.grpcsharp_batch_context_recv_close_on_server_cancelled = GetMethodDelegate<Delegates.grpcsharp_batch_context_recv_close_on_server_cancelled_delegate>(library); this.grpcsharp_batch_context_destroy = GetMethodDelegate<Delegates.grpcsharp_batch_context_destroy_delegate>(library); + this.grpcsharp_request_call_context_create = GetMethodDelegate<Delegates.grpcsharp_request_call_context_create_delegate>(library); + this.grpcsharp_request_call_context_call = GetMethodDelegate<Delegates.grpcsharp_request_call_context_call_delegate>(library); + this.grpcsharp_request_call_context_method = GetMethodDelegate<Delegates.grpcsharp_request_call_context_method_delegate>(library); + this.grpcsharp_request_call_context_host = GetMethodDelegate<Delegates.grpcsharp_request_call_context_host_delegate>(library); + this.grpcsharp_request_call_context_deadline = GetMethodDelegate<Delegates.grpcsharp_request_call_context_deadline_delegate>(library); + this.grpcsharp_request_call_context_request_metadata = GetMethodDelegate<Delegates.grpcsharp_request_call_context_request_metadata_delegate>(library); + this.grpcsharp_request_call_context_destroy = GetMethodDelegate<Delegates.grpcsharp_request_call_context_destroy_delegate>(library); + this.grpcsharp_composite_call_credentials_create = GetMethodDelegate<Delegates.grpcsharp_composite_call_credentials_create_delegate>(library); this.grpcsharp_call_credentials_release = GetMethodDelegate<Delegates.grpcsharp_call_credentials_release_delegate>(library); @@ -302,14 +308,17 @@ namespace Grpc.Core.Internal public delegate StatusCode grpcsharp_batch_context_recv_status_on_client_status_delegate(BatchContextSafeHandle ctx); public delegate IntPtr grpcsharp_batch_context_recv_status_on_client_details_delegate(BatchContextSafeHandle ctx); // returns const char* public delegate IntPtr grpcsharp_batch_context_recv_status_on_client_trailing_metadata_delegate(BatchContextSafeHandle ctx); - public delegate CallSafeHandle grpcsharp_batch_context_server_rpc_new_call_delegate(BatchContextSafeHandle ctx); - public delegate IntPtr grpcsharp_batch_context_server_rpc_new_method_delegate(BatchContextSafeHandle ctx); // returns const char* - public delegate IntPtr grpcsharp_batch_context_server_rpc_new_host_delegate(BatchContextSafeHandle ctx); // returns const char* - public delegate Timespec grpcsharp_batch_context_server_rpc_new_deadline_delegate(BatchContextSafeHandle ctx); - public delegate IntPtr grpcsharp_batch_context_server_rpc_new_request_metadata_delegate(BatchContextSafeHandle ctx); public delegate int grpcsharp_batch_context_recv_close_on_server_cancelled_delegate(BatchContextSafeHandle ctx); public delegate void grpcsharp_batch_context_destroy_delegate(IntPtr ctx); + public delegate RequestCallContextSafeHandle grpcsharp_request_call_context_create_delegate(); + public delegate CallSafeHandle grpcsharp_request_call_context_call_delegate(RequestCallContextSafeHandle ctx); + public delegate IntPtr grpcsharp_request_call_context_method_delegate(RequestCallContextSafeHandle ctx); // returns const char* + public delegate IntPtr grpcsharp_request_call_context_host_delegate(RequestCallContextSafeHandle ctx); // returns const char* + public delegate Timespec grpcsharp_request_call_context_deadline_delegate(RequestCallContextSafeHandle ctx); + public delegate IntPtr grpcsharp_request_call_context_request_metadata_delegate(RequestCallContextSafeHandle ctx); + public delegate void grpcsharp_request_call_context_destroy_delegate(IntPtr ctx); + public delegate CallCredentialsSafeHandle grpcsharp_composite_call_credentials_create_delegate(CallCredentialsSafeHandle creds1, CallCredentialsSafeHandle creds2); public delegate void grpcsharp_call_credentials_release_delegate(IntPtr credentials); @@ -393,7 +402,7 @@ namespace Grpc.Core.Internal public delegate int grpcsharp_server_add_insecure_http2_port_delegate(ServerSafeHandle server, string addr); public delegate int grpcsharp_server_add_secure_http2_port_delegate(ServerSafeHandle server, string addr, ServerCredentialsSafeHandle creds); public delegate void grpcsharp_server_start_delegate(ServerSafeHandle server); - public delegate CallError grpcsharp_server_request_call_delegate(ServerSafeHandle server, CompletionQueueSafeHandle cq, BatchContextSafeHandle ctx); + public delegate CallError grpcsharp_server_request_call_delegate(ServerSafeHandle server, CompletionQueueSafeHandle cq, RequestCallContextSafeHandle ctx); public delegate void grpcsharp_server_cancel_all_calls_delegate(ServerSafeHandle server); public delegate void grpcsharp_server_shutdown_and_notify_callback_delegate(ServerSafeHandle server, CompletionQueueSafeHandle cq, BatchContextSafeHandle ctx); public delegate void grpcsharp_server_destroy_delegate(IntPtr server); diff --git a/src/csharp/Grpc.Core/Internal/RequestCallContextSafeHandle.cs b/src/csharp/Grpc.Core/Internal/RequestCallContextSafeHandle.cs new file mode 100644 index 0000000000..ea7819d7b1 --- /dev/null +++ b/src/csharp/Grpc.Core/Internal/RequestCallContextSafeHandle.cs @@ -0,0 +1,85 @@ +#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 Grpc.Core; + +namespace Grpc.Core.Internal +{ + /// <summary> + /// grpcsharp_request_call_context + /// </summary> + internal class RequestCallContextSafeHandle : SafeHandleZeroIsInvalid + { + static readonly NativeMethods Native = NativeMethods.Get(); + + private RequestCallContextSafeHandle() + { + } + + public static RequestCallContextSafeHandle Create() + { + return Native.grpcsharp_request_call_context_create(); + } + + public IntPtr Handle + { + get + { + return handle; + } + } + + // Gets data of server_rpc_new completion. + public ServerRpcNew GetServerRpcNew(Server server) + { + var call = Native.grpcsharp_request_call_context_call(this); + + var method = Marshal.PtrToStringAnsi(Native.grpcsharp_request_call_context_method(this)); + var host = Marshal.PtrToStringAnsi(Native.grpcsharp_request_call_context_host(this)); + var deadline = Native.grpcsharp_request_call_context_deadline(this); + + IntPtr metadataArrayPtr = Native.grpcsharp_request_call_context_request_metadata(this); + var metadata = MetadataArraySafeHandle.ReadMetadataFromPtrUnsafe(metadataArrayPtr); + + return new ServerRpcNew(server, call, method, host, deadline, metadata); + } + + protected override bool ReleaseHandle() + { + Native.grpcsharp_request_call_context_destroy(handle); + return true; + } + } +} diff --git a/src/csharp/Grpc.Core/Internal/SafeHandleZeroIsInvalid.cs b/src/csharp/Grpc.Core/Internal/SafeHandleZeroIsInvalid.cs index 230faacff6..a637a54f58 100644 --- a/src/csharp/Grpc.Core/Internal/SafeHandleZeroIsInvalid.cs +++ b/src/csharp/Grpc.Core/Internal/SafeHandleZeroIsInvalid.cs @@ -45,10 +45,6 @@ namespace Grpc.Core.Internal { } - public SafeHandleZeroIsInvalid(bool ownsHandle) : base(IntPtr.Zero, ownsHandle) - { - } - public override bool IsInvalid { get @@ -56,11 +52,5 @@ namespace Grpc.Core.Internal return handle == IntPtr.Zero; } } - - protected override bool ReleaseHandle() - { - // handle is not owned. - return true; - } } } diff --git a/src/csharp/Grpc.Core/Internal/ServerCallHandler.cs b/src/csharp/Grpc.Core/Internal/ServerCallHandler.cs index 6a2f520163..600811ddd6 100644 --- a/src/csharp/Grpc.Core/Internal/ServerCallHandler.cs +++ b/src/csharp/Grpc.Core/Internal/ServerCallHandler.cs @@ -76,7 +76,7 @@ namespace Grpc.Core.Internal Status status; Tuple<TResponse,WriteFlags> responseTuple = null; - var context = HandlerUtils.NewContext(newRpc, asyncCall.Peer, responseStream, asyncCall.CancellationToken); + var context = HandlerUtils.NewContext(newRpc, responseStream, asyncCall.CancellationToken); try { GrpcPreconditions.CheckArgument(await requestStream.MoveNext().ConfigureAwait(false)); @@ -134,7 +134,7 @@ namespace Grpc.Core.Internal var responseStream = new ServerResponseStream<TRequest, TResponse>(asyncCall); Status status; - var context = HandlerUtils.NewContext(newRpc, asyncCall.Peer, responseStream, asyncCall.CancellationToken); + var context = HandlerUtils.NewContext(newRpc, responseStream, asyncCall.CancellationToken); try { GrpcPreconditions.CheckArgument(await requestStream.MoveNext().ConfigureAwait(false)); @@ -193,7 +193,7 @@ namespace Grpc.Core.Internal Status status; Tuple<TResponse,WriteFlags> responseTuple = null; - var context = HandlerUtils.NewContext(newRpc, asyncCall.Peer, responseStream, asyncCall.CancellationToken); + var context = HandlerUtils.NewContext(newRpc, responseStream, asyncCall.CancellationToken); try { var response = await handler(requestStream, context).ConfigureAwait(false); @@ -250,7 +250,7 @@ namespace Grpc.Core.Internal var responseStream = new ServerResponseStream<TRequest, TResponse>(asyncCall); Status status; - var context = HandlerUtils.NewContext(newRpc, asyncCall.Peer, responseStream, asyncCall.CancellationToken); + var context = HandlerUtils.NewContext(newRpc, responseStream, asyncCall.CancellationToken); try { await handler(requestStream, responseStream, context).ConfigureAwait(false); @@ -277,20 +277,31 @@ namespace Grpc.Core.Internal } } - internal class NoSuchMethodCallHandler : IServerCallHandler + internal class UnimplementedMethodCallHandler : IServerCallHandler { - public static readonly NoSuchMethodCallHandler Instance = new NoSuchMethodCallHandler(); + public static readonly UnimplementedMethodCallHandler Instance = new UnimplementedMethodCallHandler(); - public async Task HandleCall(ServerRpcNew newRpc, CompletionQueueSafeHandle cq) + DuplexStreamingServerCallHandler<byte[], byte[]> callHandlerImpl; + + public UnimplementedMethodCallHandler() { - // We don't care about the payload type here. - var asyncCall = new AsyncCallServer<byte[], byte[]>( - (payload) => payload, (payload) => payload, newRpc.Server); - - asyncCall.Initialize(newRpc.Call, cq); - var finishedTask = asyncCall.ServerSideCallAsync(); - await asyncCall.SendStatusFromServerAsync(new Status(StatusCode.Unimplemented, ""), Metadata.Empty, null).ConfigureAwait(false); - await finishedTask.ConfigureAwait(false); + var marshaller = new Marshaller<byte[]>((payload) => payload, (payload) => payload); + var method = new Method<byte[], byte[]>(MethodType.DuplexStreaming, "", "", marshaller, marshaller); + this.callHandlerImpl = new DuplexStreamingServerCallHandler<byte[], byte[]>(method, new DuplexStreamingServerMethod<byte[], byte[]>(UnimplementedMethod)); + } + + /// <summary> + /// Handler used for unimplemented method. + /// </summary> + private Task UnimplementedMethod(IAsyncStreamReader<byte[]> requestStream, IServerStreamWriter<byte[]> responseStream, ServerCallContext ctx) + { + ctx.Status = new Status(StatusCode.Unimplemented, ""); + return TaskUtils.CompletedTask; + } + + public Task HandleCall(ServerRpcNew newRpc, CompletionQueueSafeHandle cq) + { + return callHandlerImpl.HandleCall(newRpc, cq); } } @@ -313,13 +324,13 @@ namespace Grpc.Core.Internal return writeOptions != null ? writeOptions.Flags : default(WriteFlags); } - public static ServerCallContext NewContext<TRequest, TResponse>(ServerRpcNew newRpc, string peer, ServerResponseStream<TRequest, TResponse> serverResponseStream, CancellationToken cancellationToken) + public static ServerCallContext NewContext<TRequest, TResponse>(ServerRpcNew newRpc, ServerResponseStream<TRequest, TResponse> serverResponseStream, CancellationToken cancellationToken) where TRequest : class where TResponse : class { DateTime realtimeDeadline = newRpc.Deadline.ToClockType(ClockType.Realtime).ToDateTime(); - return new ServerCallContext(newRpc.Call, newRpc.Method, newRpc.Host, peer, realtimeDeadline, + return new ServerCallContext(newRpc.Call, newRpc.Method, newRpc.Host, realtimeDeadline, newRpc.RequestMetadata, cancellationToken, serverResponseStream.WriteResponseHeadersAsync, serverResponseStream); } } diff --git a/src/csharp/Grpc.Core/Internal/ServerSafeHandle.cs b/src/csharp/Grpc.Core/Internal/ServerSafeHandle.cs index 014a8db78f..7d7b838616 100644 --- a/src/csharp/Grpc.Core/Internal/ServerSafeHandle.cs +++ b/src/csharp/Grpc.Core/Internal/ServerSafeHandle.cs @@ -85,12 +85,12 @@ namespace Grpc.Core.Internal } } - public void RequestCall(BatchCompletionDelegate callback, CompletionQueueSafeHandle completionQueue) + public void RequestCall(RequestCallCompletionDelegate callback, CompletionQueueSafeHandle completionQueue) { using (completionQueue.NewScope()) { - var ctx = BatchContextSafeHandle.Create(); - completionQueue.CompletionRegistry.RegisterBatchCompletion(ctx, callback); + var ctx = RequestCallContextSafeHandle.Create(); + completionQueue.CompletionRegistry.RegisterRequestCallCompletion(ctx, callback); Native.grpcsharp_server_request_call(this, completionQueue, ctx).CheckOk(); } } diff --git a/src/csharp/Grpc.Core/Internal/UnmanagedLibrary.cs b/src/csharp/Grpc.Core/Internal/UnmanagedLibrary.cs index 31e1402849..1553bdd687 100644 --- a/src/csharp/Grpc.Core/Internal/UnmanagedLibrary.cs +++ b/src/csharp/Grpc.Core/Internal/UnmanagedLibrary.cs @@ -134,7 +134,11 @@ namespace Grpc.Core.Internal { throw new MissingMethodException(string.Format("The native method \"{0}\" does not exist", methodName)); } - return Marshal.GetDelegateForFunctionPointer(ptr, typeof(T)) as T; +#if NETSTANDARD1_5 + return Marshal.GetDelegateForFunctionPointer<T>(ptr); // non-generic version is obsolete +#else + return Marshal.GetDelegateForFunctionPointer(ptr, typeof(T)) as T; // generic version not available in .NET45 +#endif } /// <summary> |