Chiebot-Mirror
/
grpc
8
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

allow sending trailers from server handler

Browse Source
pull/2597/head
Jan Tattermusch 10 years ago
parent 998eb9bcaf
commit a0bb06511e
5 changed files with 28 additions and 20 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 7
      src/csharp/Grpc.Core/Internal/AsyncCallServer.cs
  2. 6
      src/csharp/Grpc.Core/Internal/CallSafeHandle.cs
  3. 20
      src/csharp/Grpc.Core/Internal/ServerCallHandler.cs
  4. 4
      src/csharp/Grpc.Core/Internal/ServerResponseStream.cs
  5. 11
      src/csharp/ext/grpc_csharp_ext.c

7
src/csharp/Grpc.Core/Internal/AsyncCallServer.cs
Unescape View File

@ -101,14 +101,17 @@ namespace Grpc.Core.Internal
/// Only one pending send action is allowed at any given time.
/// completionDelegate is called when the operation finishes.
/// </summary>
public void StartSendStatusFromServer(Status status, AsyncCompletionDelegate<object> completionDelegate)
public void StartSendStatusFromServer(Status status, Metadata trailers, AsyncCompletionDelegate<object> completionDelegate)
{
lock (myLock)
{
Preconditions.CheckNotNull(completionDelegate, "Completion delegate cannot be null");
CheckSendingAllowed();
call.StartSendStatusFromServer(status, HandleHalfclosed);
using (var metadataArray = MetadataArraySafeHandle.Create(trailers))
{
call.StartSendStatusFromServer(status, HandleHalfclosed, metadataArray);
}
halfcloseRequested = true;
readingDone = true;
sendCompletionDelegate = completionDelegate;

6
src/csharp/Grpc.Core/Internal/CallSafeHandle.cs
Unescape View File

@ -81,7 +81,7 @@ namespace Grpc.Core.Internal
[DllImport("grpc_csharp_ext.dll")]
static extern GRPCCallError grpcsharp_call_send_status_from_server(CallSafeHandle call,
BatchContextSafeHandle ctx, StatusCode statusCode, string statusMessage);
BatchContextSafeHandle ctx, StatusCode statusCode, string statusMessage, MetadataArraySafeHandle metadataArray);
[DllImport("grpc_csharp_ext.dll")]
static extern GRPCCallError grpcsharp_call_recv_message(CallSafeHandle call,
@ -159,11 +159,11 @@ namespace Grpc.Core.Internal
grpcsharp_call_send_close_from_client(this, ctx).CheckOk();
}
public void StartSendStatusFromServer(Status status, BatchCompletionDelegate callback)
public void StartSendStatusFromServer(Status status, BatchCompletionDelegate callback, MetadataArraySafeHandle metadataArray)
{
var ctx = BatchContextSafeHandle.Create();
completionRegistry.RegisterBatchCompletion(ctx, callback);
grpcsharp_call_send_status_from_server(this, ctx, status.StatusCode, status.Detail).CheckOk();
grpcsharp_call_send_status_from_server(this, ctx, status.StatusCode, status.Detail, metadataArray).CheckOk();
}
public void StartReceiveMessage(BatchCompletionDelegate callback)

20
src/csharp/Grpc.Core/Internal/ServerCallHandler.cs
Unescape View File

@ -72,13 +72,13 @@ namespace Grpc.Core.Internal
var responseStream = new ServerResponseStream<TRequest, TResponse>(asyncCall);
Status status;
var context = HandlerUtils.NewContext(newRpc);
try
{
Preconditions.CheckArgument(await requestStream.MoveNext());
var request = requestStream.Current;
// TODO(jtattermusch): we need to read the full stream so that native callhandle gets deallocated.
Preconditions.CheckArgument(!await requestStream.MoveNext());
var context = HandlerUtils.NewContext(newRpc);
var result = await handler(context, request);
status = context.Status;
await responseStream.WriteAsync(result);
@ -90,7 +90,7 @@ namespace Grpc.Core.Internal
}
try
{
await responseStream.WriteStatusAsync(status);
await responseStream.WriteStatusAsync(status, context.ResponseTrailers);
}
catch (OperationCanceledException)
{
@ -126,14 +126,13 @@ namespace Grpc.Core.Internal
var responseStream = new ServerResponseStream<TRequest, TResponse>(asyncCall);
Status status;
var context = HandlerUtils.NewContext(newRpc);
try
{
Preconditions.CheckArgument(await requestStream.MoveNext());
var request = requestStream.Current;
// TODO(jtattermusch): we need to read the full stream so that native callhandle gets deallocated.
Preconditions.CheckArgument(!await requestStream.MoveNext());
var context = HandlerUtils.NewContext(newRpc);
await handler(context, request, responseStream);
status = context.Status;
}
@ -145,7 +144,7 @@ namespace Grpc.Core.Internal
try
{
await responseStream.WriteStatusAsync(status);
await responseStream.WriteStatusAsync(status, context.ResponseTrailers);
}
catch (OperationCanceledException)
{
@ -179,9 +178,10 @@ namespace Grpc.Core.Internal
var finishedTask = asyncCall.ServerSideCallAsync();
var requestStream = new ServerRequestStream<TRequest, TResponse>(asyncCall);
var responseStream = new ServerResponseStream<TRequest, TResponse>(asyncCall);
var context = HandlerUtils.NewContext(newRpc);
Status status;
var context = HandlerUtils.NewContext(newRpc);
try
{
var result = await handler(context, requestStream);
@ -203,7 +203,7 @@ namespace Grpc.Core.Internal
try
{
await responseStream.WriteStatusAsync(status);
await responseStream.WriteStatusAsync(status, context.ResponseTrailers);
}
catch (OperationCanceledException)
{
@ -237,9 +237,9 @@ namespace Grpc.Core.Internal
var finishedTask = asyncCall.ServerSideCallAsync();
var requestStream = new ServerRequestStream<TRequest, TResponse>(asyncCall);
var responseStream = new ServerResponseStream<TRequest, TResponse>(asyncCall);
var context = HandlerUtils.NewContext(newRpc);
Status status;
var context = HandlerUtils.NewContext(newRpc);
try
{
await handler(context, requestStream, responseStream);
@ -252,7 +252,7 @@ namespace Grpc.Core.Internal
}
try
{
await responseStream.WriteStatusAsync(status);
await responseStream.WriteStatusAsync(status, context.ResponseTrailers);
}
catch (OperationCanceledException)
{
@ -277,7 +277,7 @@ namespace Grpc.Core.Internal
var requestStream = new ServerRequestStream<byte[], byte[]>(asyncCall);
var responseStream = new ServerResponseStream<byte[], byte[]>(asyncCall);
await responseStream.WriteStatusAsync(new Status(StatusCode.Unimplemented, "No such method."));
await responseStream.WriteStatusAsync(new Status(StatusCode.Unimplemented, "No such method."), Metadata.Empty);
await finishedTask;
}
}

4
src/csharp/Grpc.Core/Internal/ServerResponseStream.cs
Unescape View File

@ -56,10 +56,10 @@ namespace Grpc.Core.Internal
return taskSource.Task;
}
public Task WriteStatusAsync(Status status)
public Task WriteStatusAsync(Status status, Metadata trailers)
{
var taskSource = new AsyncCompletionTaskSource<object>();
call.StartSendStatusFromServer(status, taskSource.CompletionDelegate);
call.StartSendStatusFromServer(status, trailers, taskSource.CompletionDelegate);
return taskSource.Task;
}
}

11
src/csharp/ext/grpc_csharp_ext.c
Unescape View File

@ -630,15 +630,20 @@ GPR_EXPORT grpc_call_error GPR_CALLTYPE
grpcsharp_call_send_status_from_server(grpc_call *call,
grpcsharp_batch_context *ctx,
grpc_status_code status_code,
const char *status_details) {
const char *status_details,
grpc_metadata_array *trailing_metadata) {
/* TODO: don't use magic number */
grpc_op ops[1];
ops[0].op = GRPC_OP_SEND_STATUS_FROM_SERVER;
ops[0].data.send_status_from_server.status = status_code;
ops[0].data.send_status_from_server.status_details =
gpr_strdup(status_details);
ops[0].data.send_status_from_server.trailing_metadata = NULL;
ops[0].data.send_status_from_server.trailing_metadata_count = 0;
grpcsharp_metadata_array_move(&(ctx->send_status_from_server.trailing_metadata),
trailing_metadata);
ops[0].data.send_status_from_server.trailing_metadata_count =
ctx->send_status_from_server.trailing_metadata.count;
ops[0].data.send_status_from_server.trailing_metadata =
ctx->send_status_from_server.trailing_metadata.metadata;
ops[0].flags = 0;
return grpc_call_start_batch(call, ops, sizeof(ops) / sizeof(ops[0]), ctx);

Write Preview
Loading…
Cancel
Save