diff options
Diffstat (limited to 'src/csharp/Grpc.Core')
-rw-r--r-- | src/csharp/Grpc.Core/Internal/AsyncCall.cs | 37 | ||||
-rw-r--r-- | src/csharp/Grpc.Core/Internal/AsyncCallBase.cs | 20 | ||||
-rw-r--r-- | src/csharp/Grpc.Core/Internal/AsyncCallServer.cs | 22 | ||||
-rw-r--r-- | src/csharp/Grpc.Core/Internal/CallSafeHandle.cs | 8 | ||||
-rw-r--r-- | src/csharp/Grpc.Core/Internal/INativeCall.cs | 27 |
5 files changed, 85 insertions, 29 deletions
diff --git a/src/csharp/Grpc.Core/Internal/AsyncCall.cs b/src/csharp/Grpc.Core/Internal/AsyncCall.cs index 09fb722c81..aa2161267a 100644 --- a/src/csharp/Grpc.Core/Internal/AsyncCall.cs +++ b/src/csharp/Grpc.Core/Internal/AsyncCall.cs @@ -27,7 +27,7 @@ namespace Grpc.Core.Internal /// <summary> /// Manages client side native call lifecycle. /// </summary> - internal class AsyncCall<TRequest, TResponse> : AsyncCallBase<TRequest, TResponse> + internal class AsyncCall<TRequest, TResponse> : AsyncCallBase<TRequest, TResponse>, IUnaryResponseClientCallback, IReceivedStatusOnClientCallback, IReceivedResponseHeadersCallback { static readonly ILogger Logger = GrpcEnvironment.Logger.ForType<AsyncCall<TRequest, TResponse>>(); @@ -138,7 +138,7 @@ namespace Grpc.Core.Internal unaryResponseTcs = new TaskCompletionSource<TResponse>(); using (var metadataArray = MetadataArraySafeHandle.Create(details.Options.Headers)) { - call.StartUnary(HandleUnaryResponse, payload, GetWriteFlagsForCall(), metadataArray, details.Options.Flags); + call.StartUnary(UnaryResponseClientCallback, payload, GetWriteFlagsForCall(), metadataArray, details.Options.Flags); } return unaryResponseTcs.Task; } @@ -162,7 +162,7 @@ namespace Grpc.Core.Internal unaryResponseTcs = new TaskCompletionSource<TResponse>(); using (var metadataArray = MetadataArraySafeHandle.Create(details.Options.Headers)) { - call.StartClientStreaming(HandleUnaryResponse, metadataArray, details.Options.Flags); + call.StartClientStreaming(UnaryResponseClientCallback, metadataArray, details.Options.Flags); } return unaryResponseTcs.Task; @@ -188,9 +188,9 @@ namespace Grpc.Core.Internal streamingResponseCallFinishedTcs = new TaskCompletionSource<object>(); using (var metadataArray = MetadataArraySafeHandle.Create(details.Options.Headers)) { - call.StartServerStreaming(HandleFinished, payload, GetWriteFlagsForCall(), metadataArray, details.Options.Flags); + call.StartServerStreaming(ReceivedStatusOnClientCallback, payload, GetWriteFlagsForCall(), metadataArray, details.Options.Flags); } - call.StartReceiveInitialMetadata(HandleReceivedResponseHeaders); + call.StartReceiveInitialMetadata(ReceivedResponseHeadersCallback); } } @@ -210,9 +210,9 @@ namespace Grpc.Core.Internal streamingResponseCallFinishedTcs = new TaskCompletionSource<object>(); using (var metadataArray = MetadataArraySafeHandle.Create(details.Options.Headers)) { - call.StartDuplexStreaming(HandleFinished, metadataArray, details.Options.Flags); + call.StartDuplexStreaming(ReceivedStatusOnClientCallback, metadataArray, details.Options.Flags); } - call.StartReceiveInitialMetadata(HandleReceivedResponseHeaders); + call.StartReceiveInitialMetadata(ReceivedResponseHeadersCallback); } } @@ -256,7 +256,7 @@ namespace Grpc.Core.Internal halfcloseRequested = true; return TaskUtils.CompletedTask; } - call.StartSendCloseFromClient(HandleSendFinished); + call.StartSendCloseFromClient(SendCompletionCallback); halfcloseRequested = true; streamingWriteTcs = new TaskCompletionSource<object>(); @@ -516,5 +516,26 @@ namespace Grpc.Core.Internal streamingResponseCallFinishedTcs.SetResult(null); } + + IUnaryResponseClientCallback UnaryResponseClientCallback => this; + + void IUnaryResponseClientCallback.OnUnaryResponseClient(bool success, ClientSideStatus receivedStatus, byte[] receivedMessage, Metadata responseHeaders) + { + HandleUnaryResponse(success, receivedStatus, receivedMessage, responseHeaders); + } + + IReceivedStatusOnClientCallback ReceivedStatusOnClientCallback => this; + + void IReceivedStatusOnClientCallback.OnReceivedStatusOnClient(bool success, ClientSideStatus receivedStatus) + { + HandleFinished(success, receivedStatus); + } + + IReceivedResponseHeadersCallback ReceivedResponseHeadersCallback => this; + + void IReceivedResponseHeadersCallback.OnReceivedResponseHeaders(bool success, Metadata responseHeaders) + { + HandleReceivedResponseHeaders(success, responseHeaders); + } } } diff --git a/src/csharp/Grpc.Core/Internal/AsyncCallBase.cs b/src/csharp/Grpc.Core/Internal/AsyncCallBase.cs index f379c85e00..3273c26b88 100644 --- a/src/csharp/Grpc.Core/Internal/AsyncCallBase.cs +++ b/src/csharp/Grpc.Core/Internal/AsyncCallBase.cs @@ -35,7 +35,7 @@ namespace Grpc.Core.Internal /// Base for handling both client side and server side calls. /// Manages native call lifecycle and provides convenience methods. /// </summary> - internal abstract class AsyncCallBase<TWrite, TRead> + internal abstract class AsyncCallBase<TWrite, TRead> : IReceivedMessageCallback, ISendCompletionCallback { static readonly ILogger Logger = GrpcEnvironment.Logger.ForType<AsyncCallBase<TWrite, TRead>>(); protected static readonly Status DeserializeResponseFailureStatus = new Status(StatusCode.Internal, "Failed to deserialize response message."); @@ -126,7 +126,7 @@ namespace Grpc.Core.Internal return earlyResult; } - call.StartSendMessage(HandleSendFinished, payload, writeFlags, !initialMetadataSent); + call.StartSendMessage(SendCompletionCallback, payload, writeFlags, !initialMetadataSent); initialMetadataSent = true; streamingWritesCounter++; @@ -154,7 +154,7 @@ namespace Grpc.Core.Internal GrpcPreconditions.CheckState(streamingReadTcs == null, "Only one read can be pending at a time"); GrpcPreconditions.CheckState(!disposed); - call.StartReceiveMessage(HandleReadFinished); + call.StartReceiveMessage(ReceivedMessageCallback); streamingReadTcs = new TaskCompletionSource<TRead>(); return streamingReadTcs.Task; } @@ -342,5 +342,19 @@ namespace Grpc.Core.Internal } origTcs.SetResult(msg); } + + protected ISendCompletionCallback SendCompletionCallback => this; + + void ISendCompletionCallback.OnSendCompletion(bool success) + { + HandleSendFinished(success); + } + + IReceivedMessageCallback ReceivedMessageCallback => this; + + void IReceivedMessageCallback.OnReceivedMessage(bool success, byte[] receivedMessage) + { + HandleReadFinished(success, receivedMessage); + } } } diff --git a/src/csharp/Grpc.Core/Internal/AsyncCallServer.cs b/src/csharp/Grpc.Core/Internal/AsyncCallServer.cs index 271a6ffadf..4f44514974 100644 --- a/src/csharp/Grpc.Core/Internal/AsyncCallServer.cs +++ b/src/csharp/Grpc.Core/Internal/AsyncCallServer.cs @@ -31,7 +31,7 @@ namespace Grpc.Core.Internal /// <summary> /// Manages server side native call lifecycle. /// </summary> - internal class AsyncCallServer<TRequest, TResponse> : AsyncCallBase<TResponse, TRequest> + internal class AsyncCallServer<TRequest, TResponse> : AsyncCallBase<TResponse, TRequest>, IReceivedCloseOnServerCallback, ISendStatusFromServerCompletionCallback { readonly TaskCompletionSource<object> finishedServersideTcs = new TaskCompletionSource<object>(); readonly CancellationTokenSource cancellationTokenSource = new CancellationTokenSource(); @@ -70,7 +70,7 @@ namespace Grpc.Core.Internal started = true; - call.StartServerSide(HandleFinishedServerside); + call.StartServerSide(ReceiveCloseOnServerCallback); return finishedServersideTcs.Task; } } @@ -114,7 +114,7 @@ namespace Grpc.Core.Internal using (var metadataArray = MetadataArraySafeHandle.Create(headers)) { - call.StartSendInitialMetadata(HandleSendFinished, metadataArray); + call.StartSendInitialMetadata(SendCompletionCallback, metadataArray); } this.initialMetadataSent = true; @@ -140,7 +140,7 @@ namespace Grpc.Core.Internal using (var metadataArray = MetadataArraySafeHandle.Create(trailers)) { - call.StartSendStatusFromServer(HandleSendStatusFromServerFinished, status, metadataArray, !initialMetadataSent, + call.StartSendStatusFromServer(SendStatusFromServerCompletionCallback, status, metadataArray, !initialMetadataSent, payload, writeFlags); } halfcloseRequested = true; @@ -227,5 +227,19 @@ namespace Grpc.Core.Internal finishedServersideTcs.SetResult(null); } + + IReceivedCloseOnServerCallback ReceiveCloseOnServerCallback => this; + + void IReceivedCloseOnServerCallback.OnReceivedCloseOnServerHandler(bool success, bool cancelled) + { + HandleFinishedServerside(success, cancelled); + } + + ISendStatusFromServerCompletionCallback SendStatusFromServerCompletionCallback => this; + + void ISendStatusFromServerCompletionCallback.OnSendStatusFromServerCompletion(bool success) + { + HandleSendStatusFromServerFinished(success); + } } } diff --git a/src/csharp/Grpc.Core/Internal/CallSafeHandle.cs b/src/csharp/Grpc.Core/Internal/CallSafeHandle.cs index 589f6cbe16..7316bc2787 100644 --- a/src/csharp/Grpc.Core/Internal/CallSafeHandle.cs +++ b/src/csharp/Grpc.Core/Internal/CallSafeHandle.cs @@ -44,6 +44,8 @@ namespace Grpc.Core.Internal (success, context, state) => ((IReceivedResponseHeadersCallback)state).OnReceivedResponseHeaders(success, context.GetReceivedInitialMetadata()); static readonly BatchCompletionDelegate CompletionHandler_ISendCompletionCallback = (success, context, state) => ((ISendCompletionCallback)state).OnSendCompletion(success); + static readonly BatchCompletionDelegate CompletionHandler_ISendStatusFromServerCompletionCallback = + (success, context, state) => ((ISendStatusFromServerCompletionCallback)state).OnSendStatusFromServerCompletion(success); static readonly BatchCompletionDelegate CompletionHandler_IReceivedCloseOnServerCallback = (success, context, state) => ((IReceivedCloseOnServerCallback)state).OnReceivedCloseOnServerHandler(success, context.GetReceivedCloseOnServerCancelled()); @@ -81,7 +83,7 @@ namespace Grpc.Core.Internal .CheckOk(); } - public void StartClientStreaming(UnaryResponseClientHandler callback, MetadataArraySafeHandle metadataArray, CallFlags callFlags) + public void StartClientStreaming(IUnaryResponseClientCallback callback, MetadataArraySafeHandle metadataArray, CallFlags callFlags) { using (completionQueue.NewScope()) { @@ -131,14 +133,14 @@ namespace Grpc.Core.Internal } } - public void StartSendStatusFromServer(ISendCompletionCallback callback, Status status, MetadataArraySafeHandle metadataArray, bool sendEmptyInitialMetadata, + public void StartSendStatusFromServer(ISendStatusFromServerCompletionCallback callback, Status status, MetadataArraySafeHandle metadataArray, bool sendEmptyInitialMetadata, byte[] optionalPayload, WriteFlags writeFlags) { using (completionQueue.NewScope()) { var ctx = BatchContextSafeHandle.Create(); var optionalPayloadLength = optionalPayload != null ? new UIntPtr((ulong)optionalPayload.Length) : UIntPtr.Zero; - completionQueue.CompletionRegistry.RegisterBatchCompletion(ctx, CompletionHandler_ISendCompletionCallback, callback); + completionQueue.CompletionRegistry.RegisterBatchCompletion(ctx, CompletionHandler_ISendStatusFromServerCompletionCallback, callback); var statusDetailBytes = MarshalUtils.GetBytesUTF8(status.Detail); Native.grpcsharp_call_send_status_from_server(this, ctx, status.StatusCode, statusDetailBytes, new UIntPtr((ulong)statusDetailBytes.Length), metadataArray, sendEmptyInitialMetadata ? 1 : 0, optionalPayload, optionalPayloadLength, writeFlags).CheckOk(); diff --git a/src/csharp/Grpc.Core/Internal/INativeCall.cs b/src/csharp/Grpc.Core/Internal/INativeCall.cs index cedf82a037..3474ad6208 100644 --- a/src/csharp/Grpc.Core/Internal/INativeCall.cs +++ b/src/csharp/Grpc.Core/Internal/INativeCall.cs @@ -59,6 +59,11 @@ namespace Grpc.Core.Internal void OnSendCompletion(bool success); } + internal interface ISendStatusFromServerCompletionCallback + { + void OnSendStatusFromServerCompletion(bool success); + } + internal interface IReceivedCloseOnServerCallback { void OnReceivedCloseOnServerHandler(bool success, bool cancelled); @@ -75,28 +80,28 @@ namespace Grpc.Core.Internal string GetPeer(); - void StartUnary(UnaryResponseClientHandler callback, byte[] payload, WriteFlags writeFlags, MetadataArraySafeHandle metadataArray, CallFlags callFlags); + void StartUnary(IUnaryResponseClientCallback callback, byte[] payload, WriteFlags writeFlags, MetadataArraySafeHandle metadataArray, CallFlags callFlags); void StartUnary(BatchContextSafeHandle ctx, byte[] payload, WriteFlags writeFlags, MetadataArraySafeHandle metadataArray, CallFlags callFlags); - void StartClientStreaming(UnaryResponseClientHandler callback, MetadataArraySafeHandle metadataArray, CallFlags callFlags); + void StartClientStreaming(IUnaryResponseClientCallback callback, MetadataArraySafeHandle metadataArray, CallFlags callFlags); - void StartServerStreaming(ReceivedStatusOnClientHandler callback, byte[] payload, WriteFlags writeFlags, MetadataArraySafeHandle metadataArray, CallFlags callFlags); + void StartServerStreaming(IReceivedStatusOnClientCallback callback, byte[] payload, WriteFlags writeFlags, MetadataArraySafeHandle metadataArray, CallFlags callFlags); - void StartDuplexStreaming(ReceivedStatusOnClientHandler callback, MetadataArraySafeHandle metadataArray, CallFlags callFlags); + void StartDuplexStreaming(IReceivedStatusOnClientCallback callback, MetadataArraySafeHandle metadataArray, CallFlags callFlags); - void StartReceiveMessage(ReceivedMessageHandler callback); + void StartReceiveMessage(IReceivedMessageCallback callback); - void StartReceiveInitialMetadata(ReceivedResponseHeadersHandler callback); + void StartReceiveInitialMetadata(IReceivedResponseHeadersCallback callback); - void StartSendInitialMetadata(SendCompletionHandler callback, MetadataArraySafeHandle metadataArray); + void StartSendInitialMetadata(ISendCompletionCallback callback, MetadataArraySafeHandle metadataArray); - void StartSendMessage(SendCompletionHandler callback, byte[] payload, WriteFlags writeFlags, bool sendEmptyInitialMetadata); + void StartSendMessage(ISendCompletionCallback callback, byte[] payload, WriteFlags writeFlags, bool sendEmptyInitialMetadata); - void StartSendCloseFromClient(SendCompletionHandler callback); + void StartSendCloseFromClient(ISendCompletionCallback callback); - void StartSendStatusFromServer(SendCompletionHandler callback, Status status, MetadataArraySafeHandle metadataArray, bool sendEmptyInitialMetadata, byte[] optionalPayload, WriteFlags writeFlags); + void StartSendStatusFromServer(ISendStatusFromServerCompletionCallback callback, Status status, MetadataArraySafeHandle metadataArray, bool sendEmptyInitialMetadata, byte[] optionalPayload, WriteFlags writeFlags); - void StartServerSide(ReceivedCloseOnServerHandler callback); + void StartServerSide(IReceivedCloseOnServerCallback callback); } } |