wire passing of C# CallFlags to C core

pull/8828/head
Jan Tattermusch 8 years ago
parent 0faf80c348
commit 43416dac68
  1. 10
      src/csharp/Grpc.Core.Tests/Internal/FakeNativeCall.cs
  2. 10
      src/csharp/Grpc.Core/Internal/AsyncCall.cs
  3. 20
      src/csharp/Grpc.Core/Internal/CallSafeHandle.cs
  4. 17
      src/csharp/Grpc.Core/Internal/INativeCall.cs
  5. 10
      src/csharp/Grpc.Core/Internal/NativeMethods.cs
  6. 21
      src/csharp/ext/grpc_csharp_ext.c

@ -115,27 +115,27 @@ namespace Grpc.Core.Internal.Tests
return "PEER"; return "PEER";
} }
public void StartUnary(UnaryResponseClientHandler callback, byte[] payload, MetadataArraySafeHandle metadataArray, WriteFlags writeFlags) public void StartUnary(UnaryResponseClientHandler callback, byte[] payload, WriteFlags writeFlags, MetadataArraySafeHandle metadataArray, CallFlags callFlags)
{ {
UnaryResponseClientHandler = callback; UnaryResponseClientHandler = callback;
} }
public void StartUnary(BatchContextSafeHandle ctx, byte[] payload, MetadataArraySafeHandle metadataArray, WriteFlags writeFlags) public void StartUnary(BatchContextSafeHandle ctx, byte[] payload, WriteFlags writeFlags, MetadataArraySafeHandle metadataArray, CallFlags callFlags)
{ {
throw new NotImplementedException(); throw new NotImplementedException();
} }
public void StartClientStreaming(UnaryResponseClientHandler callback, MetadataArraySafeHandle metadataArray) public void StartClientStreaming(UnaryResponseClientHandler callback, MetadataArraySafeHandle metadataArray, CallFlags callFlags)
{ {
UnaryResponseClientHandler = callback; UnaryResponseClientHandler = callback;
} }
public void StartServerStreaming(ReceivedStatusOnClientHandler callback, byte[] payload, MetadataArraySafeHandle metadataArray, WriteFlags writeFlags) public void StartServerStreaming(ReceivedStatusOnClientHandler callback, byte[] payload, WriteFlags writeFlags, MetadataArraySafeHandle metadataArray, CallFlags callFlags)
{ {
ReceivedStatusOnClientHandler = callback; ReceivedStatusOnClientHandler = callback;
} }
public void StartDuplexStreaming(ReceivedStatusOnClientHandler callback, MetadataArraySafeHandle metadataArray) public void StartDuplexStreaming(ReceivedStatusOnClientHandler callback, MetadataArraySafeHandle metadataArray, CallFlags callFlags)
{ {
ReceivedStatusOnClientHandler = callback; ReceivedStatusOnClientHandler = callback;
} }

@ -106,7 +106,7 @@ namespace Grpc.Core.Internal
using (var metadataArray = MetadataArraySafeHandle.Create(details.Options.Headers)) using (var metadataArray = MetadataArraySafeHandle.Create(details.Options.Headers))
using (var ctx = BatchContextSafeHandle.Create()) using (var ctx = BatchContextSafeHandle.Create())
{ {
call.StartUnary(ctx, payload, metadataArray, GetWriteFlagsForCall()); call.StartUnary(ctx, payload, GetWriteFlagsForCall(), metadataArray, details.Options.Flags);
var ev = cq.Pluck(ctx.Handle); var ev = cq.Pluck(ctx.Handle);
@ -150,7 +150,7 @@ namespace Grpc.Core.Internal
unaryResponseTcs = new TaskCompletionSource<TResponse>(); unaryResponseTcs = new TaskCompletionSource<TResponse>();
using (var metadataArray = MetadataArraySafeHandle.Create(details.Options.Headers)) using (var metadataArray = MetadataArraySafeHandle.Create(details.Options.Headers))
{ {
call.StartUnary(HandleUnaryResponse, payload, metadataArray, GetWriteFlagsForCall()); call.StartUnary(HandleUnaryResponse, payload, GetWriteFlagsForCall(), metadataArray, details.Options.Flags);
} }
return unaryResponseTcs.Task; return unaryResponseTcs.Task;
} }
@ -174,7 +174,7 @@ namespace Grpc.Core.Internal
unaryResponseTcs = new TaskCompletionSource<TResponse>(); unaryResponseTcs = new TaskCompletionSource<TResponse>();
using (var metadataArray = MetadataArraySafeHandle.Create(details.Options.Headers)) using (var metadataArray = MetadataArraySafeHandle.Create(details.Options.Headers))
{ {
call.StartClientStreaming(HandleUnaryResponse, metadataArray); call.StartClientStreaming(HandleUnaryResponse, metadataArray, details.Options.Flags);
} }
return unaryResponseTcs.Task; return unaryResponseTcs.Task;
@ -200,7 +200,7 @@ namespace Grpc.Core.Internal
streamingResponseCallFinishedTcs = new TaskCompletionSource<object>(); streamingResponseCallFinishedTcs = new TaskCompletionSource<object>();
using (var metadataArray = MetadataArraySafeHandle.Create(details.Options.Headers)) using (var metadataArray = MetadataArraySafeHandle.Create(details.Options.Headers))
{ {
call.StartServerStreaming(HandleFinished, payload, metadataArray, GetWriteFlagsForCall()); call.StartServerStreaming(HandleFinished, payload, GetWriteFlagsForCall(), metadataArray, details.Options.Flags);
} }
call.StartReceiveInitialMetadata(HandleReceivedResponseHeaders); call.StartReceiveInitialMetadata(HandleReceivedResponseHeaders);
} }
@ -222,7 +222,7 @@ namespace Grpc.Core.Internal
streamingResponseCallFinishedTcs = new TaskCompletionSource<object>(); streamingResponseCallFinishedTcs = new TaskCompletionSource<object>();
using (var metadataArray = MetadataArraySafeHandle.Create(details.Options.Headers)) using (var metadataArray = MetadataArraySafeHandle.Create(details.Options.Headers))
{ {
call.StartDuplexStreaming(HandleFinished, metadataArray); call.StartDuplexStreaming(HandleFinished, metadataArray, details.Options.Flags);
} }
call.StartReceiveInitialMetadata(HandleReceivedResponseHeaders); call.StartReceiveInitialMetadata(HandleReceivedResponseHeaders);
} }

@ -63,50 +63,50 @@ namespace Grpc.Core.Internal
Native.grpcsharp_call_set_credentials(this, credentials).CheckOk(); Native.grpcsharp_call_set_credentials(this, credentials).CheckOk();
} }
public void StartUnary(UnaryResponseClientHandler callback, byte[] payload, MetadataArraySafeHandle metadataArray, WriteFlags writeFlags) public void StartUnary(UnaryResponseClientHandler callback, byte[] payload, WriteFlags writeFlags, MetadataArraySafeHandle metadataArray, CallFlags callFlags)
{ {
using (completionQueue.NewScope()) using (completionQueue.NewScope())
{ {
var ctx = BatchContextSafeHandle.Create(); var ctx = BatchContextSafeHandle.Create();
completionQueue.CompletionRegistry.RegisterBatchCompletion(ctx, (success, context) => callback(success, context.GetReceivedStatusOnClient(), context.GetReceivedMessage(), context.GetReceivedInitialMetadata())); completionQueue.CompletionRegistry.RegisterBatchCompletion(ctx, (success, context) => callback(success, context.GetReceivedStatusOnClient(), context.GetReceivedMessage(), context.GetReceivedInitialMetadata()));
Native.grpcsharp_call_start_unary(this, ctx, payload, new UIntPtr((ulong)payload.Length), metadataArray, writeFlags) Native.grpcsharp_call_start_unary(this, ctx, payload, new UIntPtr((ulong)payload.Length), writeFlags, metadataArray, callFlags)
.CheckOk(); .CheckOk();
} }
} }
public void StartUnary(BatchContextSafeHandle ctx, byte[] payload, MetadataArraySafeHandle metadataArray, WriteFlags writeFlags) public void StartUnary(BatchContextSafeHandle ctx, byte[] payload, WriteFlags writeFlags, MetadataArraySafeHandle metadataArray, CallFlags callFlags)
{ {
Native.grpcsharp_call_start_unary(this, ctx, payload, new UIntPtr((ulong)payload.Length), metadataArray, writeFlags) Native.grpcsharp_call_start_unary(this, ctx, payload, new UIntPtr((ulong)payload.Length), writeFlags, metadataArray, callFlags)
.CheckOk(); .CheckOk();
} }
public void StartClientStreaming(UnaryResponseClientHandler callback, MetadataArraySafeHandle metadataArray) public void StartClientStreaming(UnaryResponseClientHandler callback, MetadataArraySafeHandle metadataArray, CallFlags callFlags)
{ {
using (completionQueue.NewScope()) using (completionQueue.NewScope())
{ {
var ctx = BatchContextSafeHandle.Create(); var ctx = BatchContextSafeHandle.Create();
completionQueue.CompletionRegistry.RegisterBatchCompletion(ctx, (success, context) => callback(success, context.GetReceivedStatusOnClient(), context.GetReceivedMessage(), context.GetReceivedInitialMetadata())); completionQueue.CompletionRegistry.RegisterBatchCompletion(ctx, (success, context) => callback(success, context.GetReceivedStatusOnClient(), context.GetReceivedMessage(), context.GetReceivedInitialMetadata()));
Native.grpcsharp_call_start_client_streaming(this, ctx, metadataArray).CheckOk(); Native.grpcsharp_call_start_client_streaming(this, ctx, metadataArray, callFlags).CheckOk();
} }
} }
public void StartServerStreaming(ReceivedStatusOnClientHandler callback, byte[] payload, MetadataArraySafeHandle metadataArray, WriteFlags writeFlags) public void StartServerStreaming(ReceivedStatusOnClientHandler callback, byte[] payload, WriteFlags writeFlags, MetadataArraySafeHandle metadataArray, CallFlags callFlags)
{ {
using (completionQueue.NewScope()) using (completionQueue.NewScope())
{ {
var ctx = BatchContextSafeHandle.Create(); var ctx = BatchContextSafeHandle.Create();
completionQueue.CompletionRegistry.RegisterBatchCompletion(ctx, (success, context) => callback(success, context.GetReceivedStatusOnClient())); completionQueue.CompletionRegistry.RegisterBatchCompletion(ctx, (success, context) => callback(success, context.GetReceivedStatusOnClient()));
Native.grpcsharp_call_start_server_streaming(this, ctx, payload, new UIntPtr((ulong)payload.Length), metadataArray, writeFlags).CheckOk(); Native.grpcsharp_call_start_server_streaming(this, ctx, payload, new UIntPtr((ulong)payload.Length), writeFlags, metadataArray, callFlags).CheckOk();
} }
} }
public void StartDuplexStreaming(ReceivedStatusOnClientHandler callback, MetadataArraySafeHandle metadataArray) public void StartDuplexStreaming(ReceivedStatusOnClientHandler callback, MetadataArraySafeHandle metadataArray, CallFlags callFlags)
{ {
using (completionQueue.NewScope()) using (completionQueue.NewScope())
{ {
var ctx = BatchContextSafeHandle.Create(); var ctx = BatchContextSafeHandle.Create();
completionQueue.CompletionRegistry.RegisterBatchCompletion(ctx, (success, context) => callback(success, context.GetReceivedStatusOnClient())); completionQueue.CompletionRegistry.RegisterBatchCompletion(ctx, (success, context) => callback(success, context.GetReceivedStatusOnClient()));
Native.grpcsharp_call_start_duplex_streaming(this, ctx, metadataArray).CheckOk(); Native.grpcsharp_call_start_duplex_streaming(this, ctx, metadataArray, callFlags).CheckOk();
} }
} }

@ -31,6 +31,7 @@
#endregion #endregion
using System; using System;
using Grpc.Core;
namespace Grpc.Core.Internal namespace Grpc.Core.Internal
{ {
@ -54,19 +55,19 @@ namespace Grpc.Core.Internal
{ {
void Cancel(); void Cancel();
void CancelWithStatus(Grpc.Core.Status status); void CancelWithStatus(Status status);
string GetPeer(); string GetPeer();
void StartUnary(UnaryResponseClientHandler callback, byte[] payload, MetadataArraySafeHandle metadataArray, Grpc.Core.WriteFlags writeFlags); void StartUnary(UnaryResponseClientHandler callback, byte[] payload, WriteFlags writeFlags, MetadataArraySafeHandle metadataArray, CallFlags callFlags);
void StartUnary(BatchContextSafeHandle ctx, byte[] payload, MetadataArraySafeHandle metadataArray, Grpc.Core.WriteFlags writeFlags); void StartUnary(BatchContextSafeHandle ctx, byte[] payload, WriteFlags writeFlags, MetadataArraySafeHandle metadataArray, CallFlags callFlags);
void StartClientStreaming(UnaryResponseClientHandler callback, MetadataArraySafeHandle metadataArray); void StartClientStreaming(UnaryResponseClientHandler callback, MetadataArraySafeHandle metadataArray, CallFlags callFlags);
void StartServerStreaming(ReceivedStatusOnClientHandler callback, byte[] payload, MetadataArraySafeHandle metadataArray, Grpc.Core.WriteFlags writeFlags); void StartServerStreaming(ReceivedStatusOnClientHandler callback, byte[] payload, WriteFlags writeFlags, MetadataArraySafeHandle metadataArray, CallFlags callFlags);
void StartDuplexStreaming(ReceivedStatusOnClientHandler callback, MetadataArraySafeHandle metadataArray); void StartDuplexStreaming(ReceivedStatusOnClientHandler callback, MetadataArraySafeHandle metadataArray, CallFlags callFlags);
void StartReceiveMessage(ReceivedMessageHandler callback); void StartReceiveMessage(ReceivedMessageHandler callback);
@ -74,11 +75,11 @@ namespace Grpc.Core.Internal
void StartSendInitialMetadata(SendCompletionHandler callback, MetadataArraySafeHandle metadataArray); void StartSendInitialMetadata(SendCompletionHandler callback, MetadataArraySafeHandle metadataArray);
void StartSendMessage(SendCompletionHandler callback, byte[] payload, Grpc.Core.WriteFlags writeFlags, bool sendEmptyInitialMetadata); void StartSendMessage(SendCompletionHandler callback, byte[] payload, WriteFlags writeFlags, bool sendEmptyInitialMetadata);
void StartSendCloseFromClient(SendCompletionHandler callback); void StartSendCloseFromClient(SendCompletionHandler callback);
void StartSendStatusFromServer(SendCompletionHandler callback, Grpc.Core.Status status, MetadataArraySafeHandle metadataArray, bool sendEmptyInitialMetadata, byte[] optionalPayload, Grpc.Core.WriteFlags writeFlags); void StartSendStatusFromServer(SendCompletionHandler callback, Status status, MetadataArraySafeHandle metadataArray, bool sendEmptyInitialMetadata, byte[] optionalPayload, WriteFlags writeFlags);
void StartServerSide(ReceivedCloseOnServerHandler callback); void StartServerSide(ReceivedCloseOnServerHandler callback);
} }

@ -325,14 +325,14 @@ namespace Grpc.Core.Internal
public delegate CallError grpcsharp_call_cancel_delegate(CallSafeHandle call); public delegate CallError grpcsharp_call_cancel_delegate(CallSafeHandle call);
public delegate CallError grpcsharp_call_cancel_with_status_delegate(CallSafeHandle call, StatusCode status, string description); public delegate CallError grpcsharp_call_cancel_with_status_delegate(CallSafeHandle call, StatusCode status, string description);
public delegate CallError grpcsharp_call_start_unary_delegate(CallSafeHandle call, public delegate CallError grpcsharp_call_start_unary_delegate(CallSafeHandle call,
BatchContextSafeHandle ctx, byte[] sendBuffer, UIntPtr sendBufferLen, MetadataArraySafeHandle metadataArray, WriteFlags writeFlags); BatchContextSafeHandle ctx, byte[] sendBuffer, UIntPtr sendBufferLen, WriteFlags writeFlags, MetadataArraySafeHandle metadataArray, CallFlags metadataFlags);
public delegate CallError grpcsharp_call_start_client_streaming_delegate(CallSafeHandle call, public delegate CallError grpcsharp_call_start_client_streaming_delegate(CallSafeHandle call,
BatchContextSafeHandle ctx, MetadataArraySafeHandle metadataArray); BatchContextSafeHandle ctx, MetadataArraySafeHandle metadataArray, CallFlags metadataFlags);
public delegate CallError grpcsharp_call_start_server_streaming_delegate(CallSafeHandle call, public delegate CallError grpcsharp_call_start_server_streaming_delegate(CallSafeHandle call,
BatchContextSafeHandle ctx, byte[] sendBuffer, UIntPtr sendBufferLen, BatchContextSafeHandle ctx, byte[] sendBuffer, UIntPtr sendBufferLen, WriteFlags writeFlags,
MetadataArraySafeHandle metadataArray, WriteFlags writeFlags); MetadataArraySafeHandle metadataArray, CallFlags metadataFlags);
public delegate CallError grpcsharp_call_start_duplex_streaming_delegate(CallSafeHandle call, public delegate CallError grpcsharp_call_start_duplex_streaming_delegate(CallSafeHandle call,
BatchContextSafeHandle ctx, MetadataArraySafeHandle metadataArray); BatchContextSafeHandle ctx, MetadataArraySafeHandle metadataArray, CallFlags metadataFlags);
public delegate CallError grpcsharp_call_send_message_delegate(CallSafeHandle call, public delegate CallError grpcsharp_call_send_message_delegate(CallSafeHandle call,
BatchContextSafeHandle ctx, byte[] sendBuffer, UIntPtr sendBufferLen, WriteFlags writeFlags, bool sendEmptyInitialMetadata); BatchContextSafeHandle ctx, byte[] sendBuffer, UIntPtr sendBufferLen, WriteFlags writeFlags, bool sendEmptyInitialMetadata);
public delegate CallError grpcsharp_call_send_close_from_client_delegate(CallSafeHandle call, public delegate CallError grpcsharp_call_send_close_from_client_delegate(CallSafeHandle call,

@ -521,8 +521,8 @@ GPR_EXPORT void GPR_CALLTYPE grpcsharp_call_destroy(grpc_call *call) {
GPR_EXPORT grpc_call_error GPR_CALLTYPE GPR_EXPORT grpc_call_error GPR_CALLTYPE
grpcsharp_call_start_unary(grpc_call *call, grpcsharp_batch_context *ctx, grpcsharp_call_start_unary(grpc_call *call, grpcsharp_batch_context *ctx,
const char *send_buffer, size_t send_buffer_len, const char *send_buffer, size_t send_buffer_len, uint32_t write_flags,
grpc_metadata_array *initial_metadata, uint32_t write_flags) { grpc_metadata_array *initial_metadata, uint32_t initial_metadata_flags) {
/* TODO: don't use magic number */ /* TODO: don't use magic number */
grpc_op ops[6]; grpc_op ops[6];
memset(ops, 0, sizeof(ops)); memset(ops, 0, sizeof(ops));
@ -532,7 +532,7 @@ grpcsharp_call_start_unary(grpc_call *call, grpcsharp_batch_context *ctx,
ops[0].data.send_initial_metadata.count = ctx->send_initial_metadata.count; ops[0].data.send_initial_metadata.count = ctx->send_initial_metadata.count;
ops[0].data.send_initial_metadata.metadata = ops[0].data.send_initial_metadata.metadata =
ctx->send_initial_metadata.metadata; ctx->send_initial_metadata.metadata;
ops[0].flags = 0; ops[0].flags = initial_metadata_flags;
ops[0].reserved = NULL; ops[0].reserved = NULL;
ops[1].op = GRPC_OP_SEND_MESSAGE; ops[1].op = GRPC_OP_SEND_MESSAGE;
@ -575,7 +575,8 @@ grpcsharp_call_start_unary(grpc_call *call, grpcsharp_batch_context *ctx,
GPR_EXPORT grpc_call_error GPR_CALLTYPE GPR_EXPORT grpc_call_error GPR_CALLTYPE
grpcsharp_call_start_client_streaming(grpc_call *call, grpcsharp_call_start_client_streaming(grpc_call *call,
grpcsharp_batch_context *ctx, grpcsharp_batch_context *ctx,
grpc_metadata_array *initial_metadata) { grpc_metadata_array *initial_metadata,
uint32_t initial_metadata_flags) {
/* TODO: don't use magic number */ /* TODO: don't use magic number */
grpc_op ops[4]; grpc_op ops[4];
memset(ops, 0, sizeof(ops)); memset(ops, 0, sizeof(ops));
@ -585,7 +586,7 @@ grpcsharp_call_start_client_streaming(grpc_call *call,
ops[0].data.send_initial_metadata.count = ctx->send_initial_metadata.count; ops[0].data.send_initial_metadata.count = ctx->send_initial_metadata.count;
ops[0].data.send_initial_metadata.metadata = ops[0].data.send_initial_metadata.metadata =
ctx->send_initial_metadata.metadata; ctx->send_initial_metadata.metadata;
ops[0].flags = 0; ops[0].flags = initial_metadata_flags;
ops[0].reserved = NULL; ops[0].reserved = NULL;
ops[1].op = GRPC_OP_RECV_INITIAL_METADATA; ops[1].op = GRPC_OP_RECV_INITIAL_METADATA;
@ -617,7 +618,8 @@ grpcsharp_call_start_client_streaming(grpc_call *call,
GPR_EXPORT grpc_call_error GPR_CALLTYPE grpcsharp_call_start_server_streaming( GPR_EXPORT grpc_call_error GPR_CALLTYPE grpcsharp_call_start_server_streaming(
grpc_call *call, grpcsharp_batch_context *ctx, const char *send_buffer, grpc_call *call, grpcsharp_batch_context *ctx, const char *send_buffer,
size_t send_buffer_len, grpc_metadata_array *initial_metadata, uint32_t write_flags) { size_t send_buffer_len, uint32_t write_flags,
grpc_metadata_array *initial_metadata, uint32_t initial_metadata_flags) {
/* TODO: don't use magic number */ /* TODO: don't use magic number */
grpc_op ops[4]; grpc_op ops[4];
memset(ops, 0, sizeof(ops)); memset(ops, 0, sizeof(ops));
@ -627,7 +629,7 @@ GPR_EXPORT grpc_call_error GPR_CALLTYPE grpcsharp_call_start_server_streaming(
ops[0].data.send_initial_metadata.count = ctx->send_initial_metadata.count; ops[0].data.send_initial_metadata.count = ctx->send_initial_metadata.count;
ops[0].data.send_initial_metadata.metadata = ops[0].data.send_initial_metadata.metadata =
ctx->send_initial_metadata.metadata; ctx->send_initial_metadata.metadata;
ops[0].flags = 0; ops[0].flags = initial_metadata_flags;
ops[0].reserved = NULL; ops[0].reserved = NULL;
ops[1].op = GRPC_OP_SEND_MESSAGE; ops[1].op = GRPC_OP_SEND_MESSAGE;
@ -660,7 +662,8 @@ GPR_EXPORT grpc_call_error GPR_CALLTYPE grpcsharp_call_start_server_streaming(
GPR_EXPORT grpc_call_error GPR_CALLTYPE GPR_EXPORT grpc_call_error GPR_CALLTYPE
grpcsharp_call_start_duplex_streaming(grpc_call *call, grpcsharp_call_start_duplex_streaming(grpc_call *call,
grpcsharp_batch_context *ctx, grpcsharp_batch_context *ctx,
grpc_metadata_array *initial_metadata) { grpc_metadata_array *initial_metadata,
uint32_t initial_metadata_flags) {
/* TODO: don't use magic number */ /* TODO: don't use magic number */
grpc_op ops[2]; grpc_op ops[2];
memset(ops, 0, sizeof(ops)); memset(ops, 0, sizeof(ops));
@ -670,7 +673,7 @@ grpcsharp_call_start_duplex_streaming(grpc_call *call,
ops[0].data.send_initial_metadata.count = ctx->send_initial_metadata.count; ops[0].data.send_initial_metadata.count = ctx->send_initial_metadata.count;
ops[0].data.send_initial_metadata.metadata = ops[0].data.send_initial_metadata.metadata =
ctx->send_initial_metadata.metadata; ctx->send_initial_metadata.metadata;
ops[0].flags = 0; ops[0].flags = initial_metadata_flags;
ops[0].reserved = NULL; ops[0].reserved = NULL;
ops[1].op = GRPC_OP_RECV_STATUS_ON_CLIENT; ops[1].op = GRPC_OP_RECV_STATUS_ON_CLIENT;

Loading…
Cancel
Save