aboutsummaryrefslogtreecommitdiffhomepage
path: root/src/csharp/Grpc.Core/Internal/ServerCallHandler.cs
diff options
context:
space:
mode:
Diffstat (limited to 'src/csharp/Grpc.Core/Internal/ServerCallHandler.cs')
-rw-r--r--src/csharp/Grpc.Core/Internal/ServerCallHandler.cs79
1 files changed, 49 insertions, 30 deletions
diff --git a/src/csharp/Grpc.Core/Internal/ServerCallHandler.cs b/src/csharp/Grpc.Core/Internal/ServerCallHandler.cs
index 594e46b159..3680b1e791 100644
--- a/src/csharp/Grpc.Core/Internal/ServerCallHandler.cs
+++ b/src/csharp/Grpc.Core/Internal/ServerCallHandler.cs
@@ -34,6 +34,7 @@
using System;
using System.Collections.Generic;
using System.Linq;
+using System.Threading;
using System.Threading.Tasks;
using Grpc.Core.Internal;
using Grpc.Core.Utils;
@@ -42,7 +43,7 @@ namespace Grpc.Core.Internal
{
internal interface IServerCallHandler
{
- Task HandleCall(string methodName, CallSafeHandle call, GrpcEnvironment environment);
+ Task HandleCall(ServerRpcNew newRpc, GrpcEnvironment environment);
}
internal class UnaryServerCallHandler<TRequest, TResponse> : IServerCallHandler
@@ -58,27 +59,28 @@ namespace Grpc.Core.Internal
this.handler = handler;
}
- public async Task HandleCall(string methodName, CallSafeHandle call, GrpcEnvironment environment)
+ public async Task HandleCall(ServerRpcNew newRpc, GrpcEnvironment environment)
{
var asyncCall = new AsyncCallServer<TRequest, TResponse>(
method.ResponseMarshaller.Serializer,
method.RequestMarshaller.Deserializer,
environment);
- asyncCall.Initialize(call);
+ asyncCall.Initialize(newRpc.Call);
var finishedTask = asyncCall.ServerSideCallAsync();
var requestStream = new ServerRequestStream<TRequest, TResponse>(asyncCall);
var responseStream = new ServerResponseStream<TRequest, TResponse>(asyncCall);
- Status status = Status.DefaultSuccess;
+ 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 = new ServerCallContext(); // TODO(jtattermusch): initialize the context
- var result = await handler(context, request);
+ var result = await handler(request, context);
+ status = context.Status;
await responseStream.WriteAsync(result);
}
catch (Exception e)
@@ -88,7 +90,7 @@ namespace Grpc.Core.Internal
}
try
{
- await responseStream.WriteStatusAsync(status);
+ await responseStream.WriteStatusAsync(status, context.ResponseTrailers);
}
catch (OperationCanceledException)
{
@@ -111,28 +113,28 @@ namespace Grpc.Core.Internal
this.handler = handler;
}
- public async Task HandleCall(string methodName, CallSafeHandle call, GrpcEnvironment environment)
+ public async Task HandleCall(ServerRpcNew newRpc, GrpcEnvironment environment)
{
var asyncCall = new AsyncCallServer<TRequest, TResponse>(
method.ResponseMarshaller.Serializer,
method.RequestMarshaller.Deserializer,
environment);
- asyncCall.Initialize(call);
+ asyncCall.Initialize(newRpc.Call);
var finishedTask = asyncCall.ServerSideCallAsync();
var requestStream = new ServerRequestStream<TRequest, TResponse>(asyncCall);
var responseStream = new ServerResponseStream<TRequest, TResponse>(asyncCall);
- Status status = Status.DefaultSuccess;
+ 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 = new ServerCallContext(); // TODO(jtattermusch): initialize the context
- await handler(context, request, responseStream);
+ await handler(request, responseStream, context);
+ status = context.Status;
}
catch (Exception e)
{
@@ -142,7 +144,7 @@ namespace Grpc.Core.Internal
try
{
- await responseStream.WriteStatusAsync(status);
+ await responseStream.WriteStatusAsync(status, context.ResponseTrailers);
}
catch (OperationCanceledException)
{
@@ -165,23 +167,24 @@ namespace Grpc.Core.Internal
this.handler = handler;
}
- public async Task HandleCall(string methodName, CallSafeHandle call, GrpcEnvironment environment)
+ public async Task HandleCall(ServerRpcNew newRpc, GrpcEnvironment environment)
{
var asyncCall = new AsyncCallServer<TRequest, TResponse>(
method.ResponseMarshaller.Serializer,
method.RequestMarshaller.Deserializer,
environment);
- asyncCall.Initialize(call);
+ asyncCall.Initialize(newRpc.Call);
var finishedTask = asyncCall.ServerSideCallAsync();
var requestStream = new ServerRequestStream<TRequest, TResponse>(asyncCall);
var responseStream = new ServerResponseStream<TRequest, TResponse>(asyncCall);
- var context = new ServerCallContext(); // TODO(jtattermusch): initialize the context
- Status status = Status.DefaultSuccess;
+ Status status;
+ var context = HandlerUtils.NewContext(newRpc);
try
{
- var result = await handler(context, requestStream);
+ var result = await handler(requestStream, context);
+ status = context.Status;
try
{
await responseStream.WriteAsync(result);
@@ -199,7 +202,7 @@ namespace Grpc.Core.Internal
try
{
- await responseStream.WriteStatusAsync(status);
+ await responseStream.WriteStatusAsync(status, context.ResponseTrailers);
}
catch (OperationCanceledException)
{
@@ -222,23 +225,24 @@ namespace Grpc.Core.Internal
this.handler = handler;
}
- public async Task HandleCall(string methodName, CallSafeHandle call, GrpcEnvironment environment)
+ public async Task HandleCall(ServerRpcNew newRpc, GrpcEnvironment environment)
{
var asyncCall = new AsyncCallServer<TRequest, TResponse>(
method.ResponseMarshaller.Serializer,
method.RequestMarshaller.Deserializer,
environment);
- asyncCall.Initialize(call);
+ asyncCall.Initialize(newRpc.Call);
var finishedTask = asyncCall.ServerSideCallAsync();
var requestStream = new ServerRequestStream<TRequest, TResponse>(asyncCall);
var responseStream = new ServerResponseStream<TRequest, TResponse>(asyncCall);
- var context = new ServerCallContext(); // TODO(jtattermusch): initialize the context
- Status status = Status.DefaultSuccess;
+ Status status;
+ var context = HandlerUtils.NewContext(newRpc);
try
{
- await handler(context, requestStream, responseStream);
+ await handler(requestStream, responseStream, context);
+ status = context.Status;
}
catch (Exception e)
{
@@ -247,7 +251,7 @@ namespace Grpc.Core.Internal
}
try
{
- await responseStream.WriteStatusAsync(status);
+ await responseStream.WriteStatusAsync(status, context.ResponseTrailers);
}
catch (OperationCanceledException)
{
@@ -259,18 +263,19 @@ namespace Grpc.Core.Internal
internal class NoSuchMethodCallHandler : IServerCallHandler
{
- public async Task HandleCall(string methodName, CallSafeHandle call, GrpcEnvironment environment)
+ public static readonly NoSuchMethodCallHandler Instance = new NoSuchMethodCallHandler();
+
+ public async Task HandleCall(ServerRpcNew newRpc, GrpcEnvironment environment)
{
// We don't care about the payload type here.
var asyncCall = new AsyncCallServer<byte[], byte[]>(
(payload) => payload, (payload) => payload, environment);
- asyncCall.Initialize(call);
+ asyncCall.Initialize(newRpc.Call);
var finishedTask = asyncCall.ServerSideCallAsync();
- 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;
}
}
@@ -279,8 +284,22 @@ namespace Grpc.Core.Internal
{
public static Status StatusFromException(Exception e)
{
+ var rpcException = e as RpcException;
+ if (rpcException != null)
+ {
+ // use the status thrown by handler.
+ return rpcException.Status;
+ }
+
// TODO(jtattermusch): what is the right status code here?
return new Status(StatusCode.Unknown, "Exception was thrown by handler.");
}
+
+ public static ServerCallContext NewContext(ServerRpcNew newRpc)
+ {
+ return new ServerCallContext(
+ newRpc.Method, newRpc.Host, newRpc.Deadline.ToDateTime(),
+ newRpc.RequestMetadata, CancellationToken.None);
+ }
}
}
generated by cgit v1.2.3 (git 2.39.1) at 2024-10-02 15:14:23 +0000