Merge pull request #2597 from jtattermusch/csharp_metadata

Improving C# metadata API
pull/2608/head
Stanley Cheung 9 years ago
commit 88b8a036ce
  1. 4
      src/compiler/csharp_generator.cc
  2. 117
      src/csharp/Grpc.Core.Tests/ClientServerTest.cs
  3. 21
      src/csharp/Grpc.Core.Tests/Internal/MetadataArraySafeHandleTest.cs
  4. 13
      src/csharp/Grpc.Core.Tests/TimespecTest.cs
  5. 18
      src/csharp/Grpc.Core/AsyncClientStreamingCall.cs
  6. 25
      src/csharp/Grpc.Core/AsyncDuplexStreamingCall.cs
  7. 24
      src/csharp/Grpc.Core/AsyncServerStreamingCall.cs
  8. 106
      src/csharp/Grpc.Core/AsyncUnaryCall.cs
  9. 10
      src/csharp/Grpc.Core/Calls.cs
  10. 8
      src/csharp/Grpc.Core/Grpc.Core.csproj
  11. 43
      src/csharp/Grpc.Core/Internal/AsyncCall.cs
  12. 7
      src/csharp/Grpc.Core/Internal/AsyncCallServer.cs
  13. 146
      src/csharp/Grpc.Core/Internal/BatchContextSafeHandle.cs
  14. 6
      src/csharp/Grpc.Core/Internal/CallSafeHandle.cs
  15. 46
      src/csharp/Grpc.Core/Internal/MetadataArraySafeHandle.cs
  16. 80
      src/csharp/Grpc.Core/Internal/ServerCallHandler.cs
  17. 4
      src/csharp/Grpc.Core/Internal/ServerResponseStream.cs
  18. 14
      src/csharp/Grpc.Core/Internal/Timespec.cs
  19. 6
      src/csharp/Grpc.Core/Metadata.cs
  20. 24
      src/csharp/Grpc.Core/Server.cs
  21. 91
      src/csharp/Grpc.Core/ServerCallContext.cs
  22. 8
      src/csharp/Grpc.Core/ServerMethods.cs
  23. 2
      src/csharp/Grpc.Examples.Tests/MathClientServerTests.cs
  24. 7
      src/csharp/Grpc.Examples/MathExamples.cs
  25. 12
      src/csharp/Grpc.Examples/MathGrpc.cs
  26. 8
      src/csharp/Grpc.Examples/MathServiceImpl.cs
  27. 4
      src/csharp/Grpc.HealthCheck.Tests/HealthClientServerTest.cs
  28. 2
      src/csharp/Grpc.HealthCheck.Tests/HealthServiceImplTest.cs
  29. 6
      src/csharp/Grpc.HealthCheck/HealthGrpc.cs
  30. 2
      src/csharp/Grpc.HealthCheck/HealthServiceImpl.cs
  31. 4
      src/csharp/Grpc.IntegrationTesting/InteropClient.cs
  32. 20
      src/csharp/Grpc.IntegrationTesting/TestGrpc.cs
  33. 12
      src/csharp/Grpc.IntegrationTesting/TestServiceImpl.cs
  34. 64
      src/csharp/ext/grpc_csharp_ext.c

@ -149,7 +149,7 @@ std::string GetMethodRequestParamMaybe(const MethodDescriptor *method) {
std::string GetMethodReturnTypeClient(const MethodDescriptor *method) {
switch (GetMethodType(method)) {
case METHODTYPE_NO_STREAMING:
return "Task<" + GetClassName(method->output_type()) + ">";
return "AsyncUnaryCall<" + GetClassName(method->output_type()) + ">";
case METHODTYPE_CLIENT_STREAMING:
return "AsyncClientStreamingCall<" + GetClassName(method->input_type())
+ ", " + GetClassName(method->output_type()) + ">";
@ -298,7 +298,7 @@ void GenerateServerInterface(Printer* out, const ServiceDescriptor *service) {
out->Indent();
for (int i = 0; i < service->method_count(); i++) {
const MethodDescriptor *method = service->method(i);
out->Print("$returntype$ $methodname$(ServerCallContext context, $request$$response_stream_maybe$);\n",
out->Print("$returntype$ $methodname$($request$$response_stream_maybe$, ServerCallContext context);\n",
"methodname", method->name(), "returntype",
GetMethodReturnTypeServer(method), "request",
GetMethodRequestParamServer(method), "response_stream_maybe",

@ -99,17 +99,17 @@ namespace Grpc.Core.Tests
[Test]
public void UnaryCall()
{
var call = new Call<string, string>(ServiceName, EchoMethod, channel, Metadata.Empty);
Assert.AreEqual("ABC", Calls.BlockingUnaryCall(call, "ABC", CancellationToken.None));
var internalCall = new Call<string, string>(ServiceName, EchoMethod, channel, Metadata.Empty);
Assert.AreEqual("ABC", Calls.BlockingUnaryCall(internalCall, "ABC", CancellationToken.None));
}
[Test]
public void UnaryCall_ServerHandlerThrows()
{
var call = new Call<string, string>(ServiceName, EchoMethod, channel, Metadata.Empty);
var internalCall = new Call<string, string>(ServiceName, EchoMethod, channel, Metadata.Empty);
try
{
Calls.BlockingUnaryCall(call, "THROW", CancellationToken.None);
Calls.BlockingUnaryCall(internalCall, "THROW", CancellationToken.None);
Assert.Fail();
}
catch (RpcException e)
@ -118,11 +118,41 @@ namespace Grpc.Core.Tests
}
}
[Test]
public void UnaryCall_ServerHandlerThrowsRpcException()
{
var internalCall = new Call<string, string>(ServiceName, EchoMethod, channel, Metadata.Empty);
try
{
Calls.BlockingUnaryCall(internalCall, "THROW_UNAUTHENTICATED", CancellationToken.None);
Assert.Fail();
}
catch (RpcException e)
{
Assert.AreEqual(StatusCode.Unauthenticated, e.Status.StatusCode);
}
}
[Test]
public void UnaryCall_ServerHandlerSetsStatus()
{
var internalCall = new Call<string, string>(ServiceName, EchoMethod, channel, Metadata.Empty);
try
{
Calls.BlockingUnaryCall(internalCall, "SET_UNAUTHENTICATED", CancellationToken.None);
Assert.Fail();
}
catch (RpcException e)
{
Assert.AreEqual(StatusCode.Unauthenticated, e.Status.StatusCode);
}
}
[Test]
public void AsyncUnaryCall()
{
var call = new Call<string, string>(ServiceName, EchoMethod, channel, Metadata.Empty);
var result = Calls.AsyncUnaryCall(call, "ABC", CancellationToken.None).Result;
var internalCall = new Call<string, string>(ServiceName, EchoMethod, channel, Metadata.Empty);
var result = Calls.AsyncUnaryCall(internalCall, "ABC", CancellationToken.None).ResponseAsync.Result;
Assert.AreEqual("ABC", result);
}
@ -131,10 +161,10 @@ namespace Grpc.Core.Tests
{
Task.Run(async () =>
{
var call = new Call<string, string>(ServiceName, EchoMethod, channel, Metadata.Empty);
var internalCall = new Call<string, string>(ServiceName, EchoMethod, channel, Metadata.Empty);
try
{
await Calls.AsyncUnaryCall(call, "THROW", CancellationToken.None);
await Calls.AsyncUnaryCall(internalCall, "THROW", CancellationToken.None);
Assert.Fail();
}
catch (RpcException e)
@ -149,11 +179,11 @@ namespace Grpc.Core.Tests
{
Task.Run(async () =>
{
var call = new Call<string, string>(ServiceName, ConcatAndEchoMethod, channel, Metadata.Empty);
var callResult = Calls.AsyncClientStreamingCall(call, CancellationToken.None);
var internalCall = new Call<string, string>(ServiceName, ConcatAndEchoMethod, channel, Metadata.Empty);
var call = Calls.AsyncClientStreamingCall(internalCall, CancellationToken.None);
await callResult.RequestStream.WriteAll(new string[] { "A", "B", "C" });
Assert.AreEqual("ABC", await callResult.Result);
await call.RequestStream.WriteAll(new string[] { "A", "B", "C" });
Assert.AreEqual("ABC", await call.ResponseAsync);
}).Wait();
}
@ -162,10 +192,10 @@ namespace Grpc.Core.Tests
{
Task.Run(async () =>
{
var call = new Call<string, string>(ServiceName, ConcatAndEchoMethod, channel, Metadata.Empty);
var internalCall = new Call<string, string>(ServiceName, ConcatAndEchoMethod, channel, Metadata.Empty);
var cts = new CancellationTokenSource();
var callResult = Calls.AsyncClientStreamingCall(call, cts.Token);
var call = Calls.AsyncClientStreamingCall(internalCall, cts.Token);
// TODO(jtattermusch): we need this to ensure call has been initiated once we cancel it.
await Task.Delay(1000);
@ -173,7 +203,7 @@ namespace Grpc.Core.Tests
try
{
await callResult.Result;
await call.ResponseAsync;
}
catch (RpcException e)
{
@ -182,30 +212,54 @@ namespace Grpc.Core.Tests
}).Wait();
}
[Test]
public void AsyncUnaryCall_EchoMetadata()
{
var headers = new Metadata
{
new Metadata.Entry("asciiHeader", "abcdefg"),
new Metadata.Entry("binaryHeader-bin", new byte[] { 1, 2, 3, 0, 0xff } ),
};
var internalCall = new Call<string, string>(ServiceName, EchoMethod, channel, headers);
var call = Calls.AsyncUnaryCall(internalCall, "ABC", CancellationToken.None);
Assert.AreEqual("ABC", call.ResponseAsync.Result);
Assert.AreEqual(StatusCode.OK, call.GetStatus().StatusCode);
var trailers = call.GetTrailers();
Assert.AreEqual(2, trailers.Count);
Assert.AreEqual(headers[0].Key, trailers[0].Key);
Assert.AreEqual(headers[0].Value, trailers[0].Value);
Assert.AreEqual(headers[1].Key, trailers[1].Key);
CollectionAssert.AreEqual(headers[1].ValueBytes, trailers[1].ValueBytes);
}
[Test]
public void UnaryCall_DisposedChannel()
{
channel.Dispose();
var call = new Call<string, string>(ServiceName, EchoMethod, channel, Metadata.Empty);
Assert.Throws(typeof(ObjectDisposedException), () => Calls.BlockingUnaryCall(call, "ABC", CancellationToken.None));
var internalCall = new Call<string, string>(ServiceName, EchoMethod, channel, Metadata.Empty);
Assert.Throws(typeof(ObjectDisposedException), () => Calls.BlockingUnaryCall(internalCall, "ABC", CancellationToken.None));
}
[Test]
public void UnaryCallPerformance()
{
var call = new Call<string, string>(ServiceName, EchoMethod, channel, Metadata.Empty);
var internalCall = new Call<string, string>(ServiceName, EchoMethod, channel, Metadata.Empty);
BenchmarkUtil.RunBenchmark(100, 100,
() => { Calls.BlockingUnaryCall(call, "ABC", default(CancellationToken)); });
() => { Calls.BlockingUnaryCall(internalCall, "ABC", default(CancellationToken)); });
}
[Test]
public void UnknownMethodHandler()
{
var call = new Call<string, string>(ServiceName, NonexistentMethod, channel, Metadata.Empty);
var internalCall = new Call<string, string>(ServiceName, NonexistentMethod, channel, Metadata.Empty);
try
{
Calls.BlockingUnaryCall(call, "ABC", default(CancellationToken));
Calls.BlockingUnaryCall(internalCall, "ABC", default(CancellationToken));
Assert.Fail();
}
catch (RpcException e)
@ -214,16 +268,33 @@ namespace Grpc.Core.Tests
}
}
private static async Task<string> EchoHandler(ServerCallContext context, string request)
private static async Task<string> EchoHandler(string request, ServerCallContext context)
{
foreach (Metadata.Entry metadataEntry in context.RequestHeaders)
{
Console.WriteLine("Echoing header " + metadataEntry.Key + " as trailer");
context.ResponseTrailers.Add(metadataEntry);
}
if (request == "THROW")
{
throw new Exception("This was thrown on purpose by a test");
}
if (request == "THROW_UNAUTHENTICATED")
{
throw new RpcException(new Status(StatusCode.Unauthenticated, ""));
}
if (request == "SET_UNAUTHENTICATED")
{
context.Status = new Status(StatusCode.Unauthenticated, "");
}
return request;
}
private static async Task<string> ConcatAndEchoHandler(ServerCallContext context, IAsyncStreamReader<string> requestStream)
private static async Task<string> ConcatAndEchoHandler(IAsyncStreamReader<string> requestStream, ServerCallContext context)
{
string result = "";
await requestStream.ForEach(async (request) =>

@ -59,5 +59,26 @@ namespace Grpc.Core.Internal.Tests
var nativeMetadata = MetadataArraySafeHandle.Create(metadata);
nativeMetadata.Dispose();
}
[Test]
public void ReadMetadataFromPtrUnsafe()
{
var metadata = new Metadata
{
new Metadata.Entry("host", "somehost"),
new Metadata.Entry("header2", "header value"),
};
var nativeMetadata = MetadataArraySafeHandle.Create(metadata);
var copy = MetadataArraySafeHandle.ReadMetadataFromPtrUnsafe(nativeMetadata.Handle);
Assert.AreEqual(2, copy.Count);
Assert.AreEqual("host", copy[0].Key);
Assert.AreEqual("somehost", copy[0].Value);
Assert.AreEqual("header2", copy[1].Key);
Assert.AreEqual("header value", copy[1].Value);
nativeMetadata.Dispose();
}
}
}

@ -58,6 +58,19 @@ namespace Grpc.Core.Internal.Tests
Assert.AreEqual(Timespec.NativeSize, Marshal.SizeOf(typeof(Timespec)));
}
[Test]
public void ToDateTime()
{
Assert.AreEqual(new DateTime(1970, 1, 1, 0, 0, 0, DateTimeKind.Utc),
new Timespec(IntPtr.Zero, 0).ToDateTime());
Assert.AreEqual(new DateTime(1970, 1, 1, 0, 0, 10, DateTimeKind.Utc).AddTicks(50),
new Timespec(new IntPtr(10), 5000).ToDateTime());
Assert.AreEqual(new DateTime(2015, 7, 21, 4, 21, 48, DateTimeKind.Utc),
new Timespec(new IntPtr(1437452508), 0).ToDateTime());
}
[Test]
public void Add()
{

@ -43,24 +43,28 @@ namespace Grpc.Core
public sealed class AsyncClientStreamingCall<TRequest, TResponse> : IDisposable
{
readonly IClientStreamWriter<TRequest> requestStream;
readonly Task<TResponse> result;
readonly Task<TResponse> responseAsync;
readonly Func<Status> getStatusFunc;
readonly Func<Metadata> getTrailersFunc;
readonly Action disposeAction;
public AsyncClientStreamingCall(IClientStreamWriter<TRequest> requestStream, Task<TResponse> result, Action disposeAction)
public AsyncClientStreamingCall(IClientStreamWriter<TRequest> requestStream, Task<TResponse> responseAsync, Func<Status> getStatusFunc, Func<Metadata> getTrailersFunc, Action disposeAction)
{
this.requestStream = requestStream;
this.result = result;
this.responseAsync = responseAsync;
this.getStatusFunc = getStatusFunc;
this.getTrailersFunc = getTrailersFunc;
this.disposeAction = disposeAction;
}
/// <summary>
/// Asynchronous call result.
/// </summary>
public Task<TResponse> Result
public Task<TResponse> ResponseAsync
{
get
{
return this.result;
return this.responseAsync;
}
}
@ -81,11 +85,11 @@ namespace Grpc.Core
/// <returns></returns>
public TaskAwaiter<TResponse> GetAwaiter()
{
return result.GetAwaiter();
return responseAsync.GetAwaiter();
}
/// <summary>
/// Provides means to provide after the call.
/// Provides means to cleanup after the call.
/// If the call has already finished normally (request stream has been completed and call result has been received), doesn't do anything.
/// Otherwise, requests cancellation of the call which should terminate all pending async operations associated with the call.
/// As a result, all resources being used by the call should be released eventually.

@ -44,14 +44,19 @@ namespace Grpc.Core
{
readonly IClientStreamWriter<TRequest> requestStream;
readonly IAsyncStreamReader<TResponse> responseStream;
readonly Func<Status> getStatusFunc;
readonly Func<Metadata> getTrailersFunc;
readonly Action disposeAction;
public AsyncDuplexStreamingCall(IClientStreamWriter<TRequest> requestStream, IAsyncStreamReader<TResponse> responseStream, Action disposeAction)
public AsyncDuplexStreamingCall(IClientStreamWriter<TRequest> requestStream, IAsyncStreamReader<TResponse> responseStream, Func<Status> getStatusFunc, Func<Metadata> getTrailersFunc, Action disposeAction)
{
this.requestStream = requestStream;
this.responseStream = responseStream;
this.getStatusFunc = getStatusFunc;
this.getTrailersFunc = getTrailersFunc;
this.disposeAction = disposeAction;
}
/// <summary>
/// Async stream to read streaming responses.
@ -75,6 +80,24 @@ namespace Grpc.Core
}
}
/// <summary>
/// Gets the call status if the call has already finished.
/// Throws InvalidOperationException otherwise.
/// </summary>
public Status GetStatus()
{
return getStatusFunc();
}
/// <summary>
/// Gets the call trailing metadata if the call has already finished.
/// Throws InvalidOperationException otherwise.
/// </summary>
public Metadata GetTrailers()
{
return getTrailersFunc();
}
/// <summary>
/// Provides means to cleanup after the call.
/// If the call has already finished normally (request stream has been completed and response stream has been fully read), doesn't do anything.

@ -43,11 +43,15 @@ namespace Grpc.Core
public sealed class AsyncServerStreamingCall<TResponse> : IDisposable
{
readonly IAsyncStreamReader<TResponse> responseStream;
readonly Func<Status> getStatusFunc;
readonly Func<Metadata> getTrailersFunc;
readonly Action disposeAction;
public AsyncServerStreamingCall(IAsyncStreamReader<TResponse> responseStream, Action disposeAction)
public AsyncServerStreamingCall(IAsyncStreamReader<TResponse> responseStream, Func<Status> getStatusFunc, Func<Metadata> getTrailersFunc, Action disposeAction)
{
this.responseStream = responseStream;
this.getStatusFunc = getStatusFunc;
this.getTrailersFunc = getTrailersFunc;
this.disposeAction = disposeAction;
}
@ -62,6 +66,24 @@ namespace Grpc.Core
}
}
/// <summary>
/// Gets the call status if the call has already finished.
/// Throws InvalidOperationException otherwise.
/// </summary>
public Status GetStatus()
{
return getStatusFunc();
}
/// <summary>
/// Gets the call trailing metadata if the call has already finished.
/// Throws InvalidOperationException otherwise.
/// </summary>
public Metadata GetTrailers()
{
return getTrailersFunc();
}
/// <summary>
/// Provides means to cleanup after the call.
/// If the call has already finished normally (response stream has been fully read), doesn't do anything.

@ -0,0 +1,106 @@
#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.CompilerServices;
using System.Threading.Tasks;
namespace Grpc.Core
{
/// <summary>
/// Return type for single request - single response call.
/// </summary>
public sealed class AsyncUnaryCall<TResponse> : IDisposable
{
readonly Task<TResponse> responseAsync;
readonly Func<Status> getStatusFunc;
readonly Func<Metadata> getTrailersFunc;
readonly Action disposeAction;
public AsyncUnaryCall(Task<TResponse> responseAsync, Func<Status> getStatusFunc, Func<Metadata> getTrailersFunc, Action disposeAction)
{
this.responseAsync = responseAsync;
this.getStatusFunc = getStatusFunc;
this.getTrailersFunc = getTrailersFunc;
this.disposeAction = disposeAction;
}
/// <summary>
/// Asynchronous call result.
/// </summary>
public Task<TResponse> ResponseAsync
{
get
{
return this.responseAsync;
}
}
/// <summary>
/// Allows awaiting this object directly.
/// </summary>
public TaskAwaiter<TResponse> GetAwaiter()
{
return responseAsync.GetAwaiter();
}
/// <summary>
/// Gets the call status if the call has already finished.
/// Throws InvalidOperationException otherwise.
/// </summary>
public Status GetStatus()
{
return getStatusFunc();
}
/// <summary>
/// Gets the call trailing metadata if the call has already finished.
/// Throws InvalidOperationException otherwise.
/// </summary>
public Metadata GetTrailers()
{
return getTrailersFunc();
}
/// <summary>
/// Provides means to cleanup after the call.
/// If the call has already finished normally (request stream has been completed and call result has been received), doesn't do anything.
/// Otherwise, requests cancellation of the call which should terminate all pending async operations associated with the call.
/// As a result, all resources being used by the call should be released eventually.
/// </summary>
public void Dispose()
{
disposeAction.Invoke();
}
}
}

@ -53,7 +53,7 @@ namespace Grpc.Core
return asyncCall.UnaryCall(call.Channel, call.Name, req, call.Headers);
}
public static async Task<TResponse> AsyncUnaryCall<TRequest, TResponse>(Call<TRequest, TResponse> call, TRequest req, CancellationToken token)
public static AsyncUnaryCall<TResponse> AsyncUnaryCall<TRequest, TResponse>(Call<TRequest, TResponse> call, TRequest req, CancellationToken token)
where TRequest : class
where TResponse : class
{
@ -61,7 +61,7 @@ namespace Grpc.Core
asyncCall.Initialize(call.Channel, call.Channel.CompletionQueue, call.Name);
var asyncResult = asyncCall.UnaryCallAsync(req, call.Headers);
RegisterCancellationCallback(asyncCall, token);
return await asyncResult;
return new AsyncUnaryCall<TResponse>(asyncResult, asyncCall.GetStatus, asyncCall.GetTrailers, asyncCall.Cancel);
}
public static AsyncServerStreamingCall<TResponse> AsyncServerStreamingCall<TRequest, TResponse>(Call<TRequest, TResponse> call, TRequest req, CancellationToken token)
@ -73,7 +73,7 @@ namespace Grpc.Core
asyncCall.StartServerStreamingCall(req, call.Headers);
RegisterCancellationCallback(asyncCall, token);
var responseStream = new ClientResponseStream<TRequest, TResponse>(asyncCall);
return new AsyncServerStreamingCall<TResponse>(responseStream, asyncCall.Cancel);
return new AsyncServerStreamingCall<TResponse>(responseStream, asyncCall.GetStatus, asyncCall.GetTrailers, asyncCall.Cancel);
}
public static AsyncClientStreamingCall<TRequest, TResponse> AsyncClientStreamingCall<TRequest, TResponse>(Call<TRequest, TResponse> call, CancellationToken token)
@ -85,7 +85,7 @@ namespace Grpc.Core
var resultTask = asyncCall.ClientStreamingCallAsync(call.Headers);
RegisterCancellationCallback(asyncCall, token);
var requestStream = new ClientRequestStream<TRequest, TResponse>(asyncCall);
return new AsyncClientStreamingCall<TRequest, TResponse>(requestStream, resultTask, asyncCall.Cancel);
return new AsyncClientStreamingCall<TRequest, TResponse>(requestStream, resultTask, asyncCall.GetStatus, asyncCall.GetTrailers, asyncCall.Cancel);
}
public static AsyncDuplexStreamingCall<TRequest, TResponse> AsyncDuplexStreamingCall<TRequest, TResponse>(Call<TRequest, TResponse> call, CancellationToken token)
@ -98,7 +98,7 @@ namespace Grpc.Core
RegisterCancellationCallback(asyncCall, token);
var requestStream = new ClientRequestStream<TRequest, TResponse>(asyncCall);
var responseStream = new ClientResponseStream<TRequest, TResponse>(asyncCall);
return new AsyncDuplexStreamingCall<TRequest, TResponse>(requestStream, responseStream, asyncCall.Cancel);
return new AsyncDuplexStreamingCall<TRequest, TResponse>(requestStream, responseStream, asyncCall.GetStatus, asyncCall.GetTrailers, asyncCall.Cancel);
}
private static void RegisterCancellationCallback<TRequest, TResponse>(AsyncCall<TRequest, TResponse> asyncCall, CancellationToken token)

@ -33,13 +33,12 @@
</PropertyGroup>
<ItemGroup>
<Reference Include="System" />
<Reference Include="System.Collections.Immutable, Version=1.1.36.0, Culture=neutral, PublicKeyToken=b03f5f7f11d50a3a, processorArchitecture=MSIL">
<SpecificVersion>False</SpecificVersion>
<HintPath>..\packages\System.Collections.Immutable.1.1.36\lib\portable-net45+win8+wp8+wpa81\System.Collections.Immutable.dll</HintPath>
</Reference>
<Reference Include="System.Interactive.Async">
<HintPath>..\packages\Ix-Async.1.2.3\lib\net45\System.Interactive.Async.dll</HintPath>
</Reference>
<Reference Include="System.Collections.Immutable">
<HintPath>..\packages\System.Collections.Immutable.1.1.36\lib\portable-net45+win8+wp8+wpa81\System.Collections.Immutable.dll</HintPath>
</Reference>
</ItemGroup>
<ItemGroup>
<Compile Include="AsyncDuplexStreamingCall.cs" />
@ -102,6 +101,7 @@
<Compile Include="Internal\CompletionRegistry.cs" />
<Compile Include="Internal\BatchContextSafeHandle.cs" />
<Compile Include="ChannelOptions.cs" />
<Compile Include="AsyncUnaryCall.cs" />
</ItemGroup>
<ItemGroup>
<None Include="Grpc.Core.nuspec" />

@ -52,8 +52,8 @@ namespace Grpc.Core.Internal
// Completion of a pending unary response if not null.
TaskCompletionSource<TResponse> unaryResponseTcs;
// Set after status is received. Only used for streaming response calls.
Status? finishedStatus;
// Set after status is received. Used for both unary and streaming response calls.
ClientSideStatus? finishedStatus;
bool readObserverCompleted; // True if readObserver has already been completed.
@ -248,6 +248,32 @@ namespace Grpc.Core.Internal
}
}
/// <summary>
/// Gets the resulting status if the call has already finished.
/// Throws InvalidOperationException otherwise.
/// </summary>
public Status GetStatus()
{
lock (myLock)
{
Preconditions.CheckState(finishedStatus.HasValue, "Status can only be accessed once the call has finished.");
return finishedStatus.Value.Status;
}
}
/// <summary>
/// Gets the trailing metadata if the call has already finished.
/// Throws InvalidOperationException otherwise.
/// </summary>
public Metadata GetTrailers()
{
lock (myLock)
{
Preconditions.CheckState(finishedStatus.HasValue, "Trailers can only be accessed once the call has finished.");
return finishedStatus.Value.Trailers;
}
}
/// <summary>
/// On client-side, we only fire readCompletionDelegate once all messages have been read
/// and status has been received.
@ -265,7 +291,7 @@ namespace Grpc.Core.Internal
if (shouldComplete)
{
var status = finishedStatus.Value;
var status = finishedStatus.Value.Status;
if (status.StatusCode != StatusCode.OK)
{
FireCompletion(completionDelegate, default(TResponse), new RpcException(status));
@ -288,9 +314,13 @@ namespace Grpc.Core.Internal
/// </summary>
private void HandleUnaryResponse(bool success, BatchContextSafeHandle ctx)
{
var fullStatus = ctx.GetReceivedStatusOnClient();
lock (myLock)
{
finished = true;
finishedStatus = fullStatus;
halfclosed = true;
ReleaseResourcesIfPossible();
@ -302,7 +332,8 @@ namespace Grpc.Core.Internal
return;
}
var status = ctx.GetReceivedStatus();
var status = fullStatus.Status;
if (status.StatusCode != StatusCode.OK)
{
unaryResponseTcs.SetException(new RpcException(status));
@ -321,13 +352,13 @@ namespace Grpc.Core.Internal
/// </summary>
private void HandleFinished(bool success, BatchContextSafeHandle ctx)
{
var status = ctx.GetReceivedStatus();
var fullStatus = ctx.GetReceivedStatusOnClient();
AsyncCompletionDelegate<TResponse> origReadCompletionDelegate = null;
lock (myLock)
{
finished = true;
finishedStatus = status;
finishedStatus = fullStatus;
origReadCompletionDelegate = readCompletionDelegate;

@ -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;

@ -38,7 +38,6 @@ using Grpc.Core;
namespace Grpc.Core.Internal
{
/// <summary>
/// Not owned version of
/// grpcsharp_batch_context
/// </summary>
internal class BatchContextSafeHandle : SafeHandleZeroIsInvalid
@ -46,6 +45,9 @@ namespace Grpc.Core.Internal
[DllImport("grpc_csharp_ext.dll")]
static extern BatchContextSafeHandle grpcsharp_batch_context_create();
[DllImport("grpc_csharp_ext.dll")]
static extern IntPtr grpcsharp_batch_context_recv_initial_metadata(BatchContextSafeHandle ctx);
[DllImport("grpc_csharp_ext.dll")]
static extern IntPtr grpcsharp_batch_context_recv_message_length(BatchContextSafeHandle ctx);
@ -58,12 +60,24 @@ namespace Grpc.Core.Internal
[DllImport("grpc_csharp_ext.dll")]
static extern IntPtr grpcsharp_batch_context_recv_status_on_client_details(BatchContextSafeHandle ctx); // returns const char*
[DllImport("grpc_csharp_ext.dll")]
static extern IntPtr grpcsharp_batch_context_recv_status_on_client_trailing_metadata(BatchContextSafeHandle ctx);
[DllImport("grpc_csharp_ext.dll")]
static extern CallSafeHandle grpcsharp_batch_context_server_rpc_new_call(BatchContextSafeHandle ctx);
[DllImport("grpc_csharp_ext.dll")]
static extern IntPtr grpcsharp_batch_context_server_rpc_new_method(BatchContextSafeHandle ctx); // returns const char*
[DllImport("grpc_csharp_ext.dll")]
static extern IntPtr grpcsharp_batch_context_server_rpc_new_host(BatchContextSafeHandle ctx); // returns const char*
[DllImport("grpc_csharp_ext.dll")]
static extern Timespec grpcsharp_batch_context_server_rpc_new_deadline(BatchContextSafeHandle ctx);
[DllImport("grpc_csharp_ext.dll")]
static extern IntPtr grpcsharp_batch_context_server_rpc_new_request_metadata(BatchContextSafeHandle ctx);
[DllImport("grpc_csharp_ext.dll")]
static extern int grpcsharp_batch_context_recv_close_on_server_cancelled(BatchContextSafeHandle ctx);
@ -87,13 +101,26 @@ namespace Grpc.Core.Internal
}
}
public Status GetReceivedStatus()
// Gets data of recv_initial_metadata completion.
public Metadata GetReceivedInitialMetadata()
{
IntPtr metadataArrayPtr = grpcsharp_batch_context_recv_initial_metadata(this);
return MetadataArraySafeHandle.ReadMetadataFromPtrUnsafe(metadataArrayPtr);
}
// Gets data of recv_status_on_client completion.
public ClientSideStatus GetReceivedStatusOnClient()
{
// TODO: can the native method return string directly?
string details = Marshal.PtrToStringAnsi(grpcsharp_batch_context_recv_status_on_client_details(this));
return new Status(grpcsharp_batch_context_recv_status_on_client_status(this), details);
var status = new Status(grpcsharp_batch_context_recv_status_on_client_status(this), details);
IntPtr metadataArrayPtr = grpcsharp_batch_context_recv_status_on_client_trailing_metadata(this);
var metadata = MetadataArraySafeHandle.ReadMetadataFromPtrUnsafe(metadataArrayPtr);
return new ClientSideStatus(status, metadata);
}
// Gets data of recv_message completion.
public byte[] GetReceivedMessage()
{
IntPtr len = grpcsharp_batch_context_recv_message_length(this);
@ -106,16 +133,22 @@ namespace Grpc.Core.Internal
return data;
}
public CallSafeHandle GetServerRpcNewCall()
// Gets data of server_rpc_new completion.
public ServerRpcNew GetServerRpcNew()
{
return grpcsharp_batch_context_server_rpc_new_call(this);
}
var call = grpcsharp_batch_context_server_rpc_new_call(this);
public string GetServerRpcNewMethod()
{
return Marshal.PtrToStringAnsi(grpcsharp_batch_context_server_rpc_new_method(this));
var method = Marshal.PtrToStringAnsi(grpcsharp_batch_context_server_rpc_new_method(this));
var host = Marshal.PtrToStringAnsi(grpcsharp_batch_context_server_rpc_new_host(this));
var deadline = grpcsharp_batch_context_server_rpc_new_deadline(this);
IntPtr metadataArrayPtr = grpcsharp_batch_context_server_rpc_new_request_metadata(this);
var metadata = MetadataArraySafeHandle.ReadMetadataFromPtrUnsafe(metadataArrayPtr);
return new ServerRpcNew(call, method, host, deadline, metadata);
}
// Gets data of receive_close_on_server completion.
public bool GetReceivedCloseOnServerCancelled()
{
return grpcsharp_batch_context_recv_close_on_server_cancelled(this) != 0;
@ -127,4 +160,97 @@ namespace Grpc.Core.Internal
return true;
}
}
/// <summary>
/// Status + metadata received on client side when call finishes.
/// (when receive_status_on_client operation finishes).
/// </summary>
internal struct ClientSideStatus
{
readonly Status status;
readonly Metadata trailers;
public ClientSideStatus(Status status, Metadata trailers)
{
this.status = status;
this.trailers = trailers;
}
public Status Status
{
get
{
return this.status;
}
}
public Metadata Trailers
{
get
{
return this.trailers;
}
}
}
/// <summary>
/// Details of a newly received RPC.
/// </summary>
internal struct ServerRpcNew
{
readonly CallSafeHandle call;
readonly string method;
readonly string host;
readonly Timespec deadline;
readonly Metadata requestMetadata;
public ServerRpcNew(CallSafeHandle call, string method, string host, Timespec deadline, Metadata requestMetadata)
{
this.call = call;
this.method = method;
this.host = host;
this.deadline = deadline;
this.requestMetadata = requestMetadata;
}
public CallSafeHandle Call
{
get
{
return this.call;
}
}
public string Method
{
get
{
return this.method;
}
}
public string Host
{
get
{
return this.host;
}
}
public Timespec Deadline
{
get
{
return this.deadline;
}
}
public Metadata RequestMetadata
{
get
{
return this.requestMetadata;
}
}
}
}

@ -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)

@ -45,13 +45,25 @@ namespace Grpc.Core.Internal
[DllImport("grpc_csharp_ext.dll", CharSet = CharSet.Ansi)]
static extern void grpcsharp_metadata_array_add(MetadataArraySafeHandle array, string key, byte[] value, UIntPtr valueLength);
[DllImport("grpc_csharp_ext.dll")]
static extern UIntPtr grpcsharp_metadata_array_count(IntPtr metadataArray);
[DllImport("grpc_csharp_ext.dll")]
static extern IntPtr grpcsharp_metadata_array_get_key(IntPtr metadataArray, UIntPtr index);
[DllImport("grpc_csharp_ext.dll")]
static extern IntPtr grpcsharp_metadata_array_get_value(IntPtr metadataArray, UIntPtr index);
[DllImport("grpc_csharp_ext.dll")]
static extern UIntPtr grpcsharp_metadata_array_get_value_length(IntPtr metadataArray, UIntPtr index);
[DllImport("grpc_csharp_ext.dll")]
static extern void grpcsharp_metadata_array_destroy_full(IntPtr array);
private MetadataArraySafeHandle()
{
}
public static MetadataArraySafeHandle Create(Metadata metadata)
{
// TODO(jtattermusch): we might wanna check that the metadata is readonly
@ -63,6 +75,38 @@ namespace Grpc.Core.Internal
return metadataArray;
}
/// <summary>
/// Reads metadata from pointer to grpc_metadata_array
/// </summary>
public static Metadata ReadMetadataFromPtrUnsafe(IntPtr metadataArray)
{
if (metadataArray == IntPtr.Zero)
{
return null;
}
ulong count = grpcsharp_metadata_array_count(metadataArray).ToUInt64();
var metadata = new Metadata();
for (ulong i = 0; i < count; i ++)
{
var index = new UIntPtr(i);
string key = Marshal.PtrToStringAnsi(grpcsharp_metadata_array_get_key(metadataArray, index));
var bytes = new byte[grpcsharp_metadata_array_get_value_length(metadataArray, index).ToUInt64()];
Marshal.Copy(grpcsharp_metadata_array_get_value(metadataArray, index), bytes, 0, bytes.Length);
metadata.Add(new Metadata.Entry(key, bytes));
}
return metadata;
}
internal IntPtr Handle
{
get
{
return handle;
}
}
protected override bool ReleaseHandle()
{
grpcsharp_metadata_array_destroy_full(handle);

@ -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,25 @@ 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 +203,7 @@ namespace Grpc.Core.Internal
try
{
await responseStream.WriteStatusAsync(status);
await responseStream.WriteStatusAsync(status, context.ResponseTrailers);
}
catch (OperationCanceledException)
{
@ -222,23 +226,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 +252,7 @@ namespace Grpc.Core.Internal
}
try
{
await responseStream.WriteStatusAsync(status);
await responseStream.WriteStatusAsync(status, context.ResponseTrailers);
}
catch (OperationCanceledException)
{
@ -259,18 +264,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 +285,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);
}
}
}

@ -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;
}
}

@ -43,6 +43,8 @@ namespace Grpc.Core.Internal
const int NanosPerSecond = 1000 * 1000 * 1000;
const int NanosPerTick = 100;
static readonly DateTime UnixEpoch = new DateTime(1970, 1, 1, 0, 0, 0, 0, DateTimeKind.Utc);
[DllImport("grpc_csharp_ext.dll")]
static extern Timespec gprsharp_now();
@ -52,6 +54,13 @@ namespace Grpc.Core.Internal
[DllImport("grpc_csharp_ext.dll")]
static extern int gprsharp_sizeof_timespec();
public Timespec(IntPtr tv_sec, int tv_nsec)
{
this.tv_sec = tv_sec;
this.tv_nsec = tv_nsec;
this.clock_type = GPRClockType.Realtime;
}
// NOTE: on linux 64bit sizeof(gpr_timespec) = 16, on windows 32bit sizeof(gpr_timespec) = 8
// so IntPtr seems to have the right size to work on both.
public System.IntPtr tv_sec;
@ -76,6 +85,11 @@ namespace Grpc.Core.Internal
return gprsharp_now();
}
}
public DateTime ToDateTime()
{
return UnixEpoch.AddTicks(tv_sec.ToInt64() * (NanosPerSecond / NanosPerTick) + tv_nsec / NanosPerTick);
}
internal static int NativeSize
{

@ -220,6 +220,12 @@ namespace Grpc.Core
return value;
}
}
public override string ToString()
{
return string.Format("[Entry: key={0}, value={1}]", Key, Value);
}
}
}
}

@ -218,16 +218,16 @@ namespace Grpc.Core
/// <summary>
/// Selects corresponding handler for given call and handles the call.
/// </summary>
private async Task InvokeCallHandler(CallSafeHandle call, string method)
private async Task HandleCallAsync(ServerRpcNew newRpc)
{
try
{
IServerCallHandler callHandler;
if (!callHandlers.TryGetValue(method, out callHandler))
if (!callHandlers.TryGetValue(newRpc.Method, out callHandler))
{
callHandler = new NoSuchMethodCallHandler();
callHandler = NoSuchMethodCallHandler.Instance;
}
await callHandler.HandleCall(method, call, environment);
await callHandler.HandleCall(newRpc, environment);
}
catch (Exception e)
{
@ -240,15 +240,15 @@ namespace Grpc.Core
/// </summary>
private void HandleNewServerRpc(bool success, BatchContextSafeHandle ctx)
{
// TODO: handle error
CallSafeHandle call = ctx.GetServerRpcNewCall();
string method = ctx.GetServerRpcNewMethod();
// after server shutdown, the callback returns with null call
if (!call.IsInvalid)
if (success)
{
Task.Run(async () => await InvokeCallHandler(call, method));
ServerRpcNew newRpc = ctx.GetServerRpcNew();
// after server shutdown, the callback returns with null call
if (!newRpc.Call.IsInvalid)
{
Task.Run(async () => await HandleCallAsync(newRpc));
}
}
AllowOneRpc();

@ -33,6 +33,7 @@
using System;
using System.Runtime.CompilerServices;
using System.Threading;
using System.Threading.Tasks;
namespace Grpc.Core
@ -42,14 +43,94 @@ namespace Grpc.Core
/// </summary>
public sealed class ServerCallContext
{
// TODO(jtattermusch): add cancellationToken
// TODO(jtattermusch): expose method to send initial metadata back to client
// TODO(jtattermusch): add deadline info
// TODO(jtattermusch): allow setting status and trailing metadata to send after handler completes.
// TODO(jtattermusch): expose initial metadata sent by client for reading
private readonly string method;
private readonly string host;
private readonly DateTime deadline;
private readonly Metadata requestHeaders;
private readonly CancellationToken cancellationToken;
// TODO(jtattermusch): expose method to send initial metadata back to client
private Status status = Status.DefaultSuccess;
private readonly Metadata responseTrailers = new Metadata();
// TODO(jtattermusch): allow setting status and trailing metadata to send after handler completes.
public ServerCallContext(string method, string host, DateTime deadline, Metadata requestHeaders, CancellationToken cancellationToken)
{
this.method = method;
this.host = host;
this.deadline = deadline;
this.requestHeaders = requestHeaders;
this.cancellationToken = cancellationToken;
}
/// <summary> Name of method called in this RPC. </summary>
public string Method
{
get
{
return this.method;
}
}
/// <summary> Name of host called in this RPC. </summary>
public string Host
{
get
{
return this.host;
}
}
/// <summary> Deadline for this RPC. </summary>
public DateTime Deadline
{
get
{
return this.deadline;
}
}
/// <summary> Initial metadata sent by client. </summary>
public Metadata RequestHeaders
{
get
{
return this.requestHeaders;
}
}
// TODO(jtattermusch): support signalling cancellation.
/// <summary> Cancellation token signals when call is cancelled. </summary>
public CancellationToken CancellationToken
{
get
{
return this.cancellationToken;
}
}
/// <summary> Trailers to send back to client after RPC finishes.</summary>
public Metadata ResponseTrailers
{
get
{
return this.responseTrailers;
}
}
/// <summary> Status to send back to client after RPC finishes.</summary>
public Status Status
{
get
{
return this.status;
}
set
{
status = value;
}
}
}
}

@ -42,28 +42,28 @@ namespace Grpc.Core
/// <summary>
/// Server-side handler for unary call.
/// </summary>
public delegate Task<TResponse> UnaryServerMethod<TRequest, TResponse>(ServerCallContext context, TRequest request)
public delegate Task<TResponse> UnaryServerMethod<TRequest, TResponse>(TRequest request, ServerCallContext context)
where TRequest : class
where TResponse : class;
/// <summary>
/// Server-side handler for client streaming call.
/// </summary>
public delegate Task<TResponse> ClientStreamingServerMethod<TRequest, TResponse>(ServerCallContext context, IAsyncStreamReader<TRequest> requestStream)
public delegate Task<TResponse> ClientStreamingServerMethod<TRequest, TResponse>(IAsyncStreamReader<TRequest> requestStream, ServerCallContext context)
where TRequest : class
where TResponse : class;
/// <summary>
/// Server-side handler for server streaming call.
/// </summary>
public delegate Task ServerStreamingServerMethod<TRequest, TResponse>(ServerCallContext context, TRequest request, IServerStreamWriter<TResponse> responseStream)
public delegate Task ServerStreamingServerMethod<TRequest, TResponse>(TRequest request, IServerStreamWriter<TResponse> responseStream, ServerCallContext context)
where TRequest : class
where TResponse : class;
/// <summary>
/// Server-side handler for bidi streaming call.
/// </summary>
public delegate Task DuplexStreamingServerMethod<TRequest, TResponse>(ServerCallContext context, IAsyncStreamReader<TRequest> requestStream, IServerStreamWriter<TResponse> responseStream)
public delegate Task DuplexStreamingServerMethod<TRequest, TResponse>(IAsyncStreamReader<TRequest> requestStream, IServerStreamWriter<TResponse> responseStream, ServerCallContext context)
where TRequest : class
where TResponse : class;
}

@ -144,7 +144,7 @@ namespace math.Tests
n => Num.CreateBuilder().SetNum_(n).Build());
await call.RequestStream.WriteAll(numbers);
var result = await call.Result;
var result = await call.ResponseAsync;
Assert.AreEqual(60, result.Num_);
}
}).Wait();

@ -46,8 +46,7 @@ namespace math
public static async Task DivAsyncExample(Math.IMathClient client)
{
Task<DivReply> resultTask = client.DivAsync(new DivArgs.Builder { Dividend = 4, Divisor = 5 }.Build());
DivReply result = await resultTask;
DivReply result = await client.DivAsync(new DivArgs.Builder { Dividend = 4, Divisor = 5 }.Build());
Console.WriteLine("DivAsync Result: " + result);
}
@ -72,7 +71,7 @@ namespace math
using (var call = client.Sum())
{
await call.RequestStream.WriteAll(numbers);
Console.WriteLine("Sum Result: " + await call.Result);
Console.WriteLine("Sum Result: " + await call.ResponseAsync);
}
}
@ -104,7 +103,7 @@ namespace math
using (var sumCall = client.Sum())
{
await sumCall.RequestStream.WriteAll(numbers);
sum = await sumCall.Result;
sum = await sumCall.ResponseAsync;
}
DivReply result = await client.DivAsync(new DivArgs.Builder { Dividend = sum.Num_, Divisor = numbers.Count }.Build());

@ -45,7 +45,7 @@ namespace math {
public interface IMathClient
{
global::math.DivReply Div(global::math.DivArgs request, Metadata headers = null, CancellationToken cancellationToken = default(CancellationToken));
Task<global::math.DivReply> DivAsync(global::math.DivArgs request, Metadata headers = null, CancellationToken cancellationToken = default(CancellationToken));
AsyncUnaryCall<global::math.DivReply> DivAsync(global::math.DivArgs request, Metadata headers = null, CancellationToken cancellationToken = default(CancellationToken));
AsyncDuplexStreamingCall<global::math.DivArgs, global::math.DivReply> DivMany(Metadata headers = null, CancellationToken cancellationToken = default(CancellationToken));
AsyncServerStreamingCall<global::math.Num> Fib(global::math.FibArgs request, Metadata headers = null, CancellationToken cancellationToken = default(CancellationToken));
AsyncClientStreamingCall<global::math.Num, global::math.Num> Sum(Metadata headers = null, CancellationToken cancellationToken = default(CancellationToken));
@ -54,10 +54,10 @@ namespace math {
// server-side interface
public interface IMath
{
Task<global::math.DivReply> Div(ServerCallContext context, global::math.DivArgs request);
Task DivMany(ServerCallContext context, IAsyncStreamReader<global::math.DivArgs> requestStream, IServerStreamWriter<global::math.DivReply> responseStream);
Task Fib(ServerCallContext context, global::math.FibArgs request, IServerStreamWriter<global::math.Num> responseStream);
Task<global::math.Num> Sum(ServerCallContext context, IAsyncStreamReader<global::math.Num> requestStream);
Task<global::math.DivReply> Div(global::math.DivArgs request, ServerCallContext context);
Task DivMany(IAsyncStreamReader<global::math.DivArgs> requestStream, IServerStreamWriter<global::math.DivReply> responseStream, ServerCallContext context);
Task Fib(global::math.FibArgs request, IServerStreamWriter<global::math.Num> responseStream, ServerCallContext context);
Task<global::math.Num> Sum(IAsyncStreamReader<global::math.Num> requestStream, ServerCallContext context);
}
// client stub
@ -71,7 +71,7 @@ namespace math {
var call = CreateCall(__ServiceName, __Method_Div, headers);
return Calls.BlockingUnaryCall(call, request, cancellationToken);
}
public Task<global::math.DivReply> DivAsync(global::math.DivArgs request, Metadata headers = null, CancellationToken cancellationToken = default(CancellationToken))
public AsyncUnaryCall<global::math.DivReply> DivAsync(global::math.DivArgs request, Metadata headers = null, CancellationToken cancellationToken = default(CancellationToken))
{
var call = CreateCall(__ServiceName, __Method_Div, headers);
return Calls.AsyncUnaryCall(call, request, cancellationToken);

@ -45,12 +45,12 @@ namespace math
/// </summary>
public class MathServiceImpl : Math.IMath
{
public Task<DivReply> Div(ServerCallContext context, DivArgs request)
public Task<DivReply> Div(DivArgs request, ServerCallContext context)
{
return Task.FromResult(DivInternal(request));
}
public async Task Fib(ServerCallContext context, FibArgs request, IServerStreamWriter<Num> responseStream)
public async Task Fib(FibArgs request, IServerStreamWriter<Num> responseStream, ServerCallContext context)
{
if (request.Limit <= 0)
{
@ -67,7 +67,7 @@ namespace math
}
}
public async Task<Num> Sum(ServerCallContext context, IAsyncStreamReader<Num> requestStream)
public async Task<Num> Sum(IAsyncStreamReader<Num> requestStream, ServerCallContext context)
{
long sum = 0;
await requestStream.ForEach(async num =>
@ -77,7 +77,7 @@ namespace math
return Num.CreateBuilder().SetNum_(sum).Build();
}
public async Task DivMany(ServerCallContext context, IAsyncStreamReader<DivArgs> requestStream, IServerStreamWriter<DivReply> responseStream)
public async Task DivMany(IAsyncStreamReader<DivArgs> requestStream, IServerStreamWriter<DivReply> responseStream, ServerCallContext context)
{
await requestStream.ForEach(async divArgs =>
{

@ -87,9 +87,7 @@ namespace Grpc.HealthCheck.Tests
[Test]
public void ServiceDoesntExist()
{
// TODO(jtattermusch): currently, this returns wrong status code, because we don't enable sending arbitrary status code from
// server handlers yet.
Assert.Throws(typeof(RpcException), () => client.Check(HealthCheckRequest.CreateBuilder().SetHost("").SetService("nonexistent.service").Build()));
Assert.Throws(Is.TypeOf(typeof(RpcException)).And.Property("Status").Property("StatusCode").EqualTo(StatusCode.NotFound), () => client.Check(HealthCheckRequest.CreateBuilder().SetHost("").SetService("nonexistent.service").Build()));
}
// TODO(jtattermusch): add test with timeout once timeouts are supported

@ -101,7 +101,7 @@ namespace Grpc.HealthCheck.Tests
private static HealthCheckResponse.Types.ServingStatus GetStatusHelper(HealthServiceImpl impl, string host, string service)
{
return impl.Check(null, HealthCheckRequest.CreateBuilder().SetHost(host).SetService(service).Build()).Result.Status;
return impl.Check(HealthCheckRequest.CreateBuilder().SetHost(host).SetService(service).Build(), null).Result.Status;
}
}
}

@ -25,13 +25,13 @@ namespace Grpc.Health.V1Alpha {
public interface IHealthClient
{
global::Grpc.Health.V1Alpha.HealthCheckResponse Check(global::Grpc.Health.V1Alpha.HealthCheckRequest request, Metadata headers = null, CancellationToken cancellationToken = default(CancellationToken));
Task<global::Grpc.Health.V1Alpha.HealthCheckResponse> CheckAsync(global::Grpc.Health.V1Alpha.HealthCheckRequest request, Metadata headers = null, CancellationToken cancellationToken = default(CancellationToken));
AsyncUnaryCall<global::Grpc.Health.V1Alpha.HealthCheckResponse> CheckAsync(global::Grpc.Health.V1Alpha.HealthCheckRequest request, Metadata headers = null, CancellationToken cancellationToken = default(CancellationToken));
}
// server-side interface
public interface IHealth
{
Task<global::Grpc.Health.V1Alpha.HealthCheckResponse> Check(ServerCallContext context, global::Grpc.Health.V1Alpha.HealthCheckRequest request);
Task<global::Grpc.Health.V1Alpha.HealthCheckResponse> Check(global::Grpc.Health.V1Alpha.HealthCheckRequest request, ServerCallContext context);
}
// client stub
@ -45,7 +45,7 @@ namespace Grpc.Health.V1Alpha {
var call = CreateCall(__ServiceName, __Method_Check, headers);
return Calls.BlockingUnaryCall(call, request, cancellationToken);
}
public Task<global::Grpc.Health.V1Alpha.HealthCheckResponse> CheckAsync(global::Grpc.Health.V1Alpha.HealthCheckRequest request, Metadata headers = null, CancellationToken cancellationToken = default(CancellationToken))
public AsyncUnaryCall<global::Grpc.Health.V1Alpha.HealthCheckResponse> CheckAsync(global::Grpc.Health.V1Alpha.HealthCheckRequest request, Metadata headers = null, CancellationToken cancellationToken = default(CancellationToken))
{
var call = CreateCall(__ServiceName, __Method_Check, headers);
return Calls.AsyncUnaryCall(call, request, cancellationToken);

@ -95,7 +95,7 @@ namespace Grpc.HealthCheck
}
}
public Task<HealthCheckResponse> Check(ServerCallContext context, HealthCheckRequest request)
public Task<HealthCheckResponse> Check(HealthCheckRequest request, ServerCallContext context)
{
lock (myLock)
{

@ -219,7 +219,7 @@ namespace Grpc.IntegrationTesting
{
await call.RequestStream.WriteAll(bodySizes);
var response = await call.Result;
var response = await call.ResponseAsync;
Assert.AreEqual(74922, response.AggregatedPayloadSize);
}
Console.WriteLine("Passed!");
@ -421,7 +421,7 @@ namespace Grpc.IntegrationTesting
try
{
var response = await call.Result;
var response = await call.ResponseAsync;
Assert.Fail();
}
catch (RpcException e)

@ -60,9 +60,9 @@ namespace grpc.testing {
public interface ITestServiceClient
{
global::grpc.testing.Empty EmptyCall(global::grpc.testing.Empty request, Metadata headers = null, CancellationToken cancellationToken = default(CancellationToken));
Task<global::grpc.testing.Empty> EmptyCallAsync(global::grpc.testing.Empty request, Metadata headers = null, CancellationToken cancellationToken = default(CancellationToken));
AsyncUnaryCall<global::grpc.testing.Empty> EmptyCallAsync(global::grpc.testing.Empty request, Metadata headers = null, CancellationToken cancellationToken = default(CancellationToken));
global::grpc.testing.SimpleResponse UnaryCall(global::grpc.testing.SimpleRequest request, Metadata headers = null, CancellationToken cancellationToken = default(CancellationToken));
Task<global::grpc.testing.SimpleResponse> UnaryCallAsync(global::grpc.testing.SimpleRequest request, Metadata headers = null, CancellationToken cancellationToken = default(CancellationToken));
AsyncUnaryCall<global::grpc.testing.SimpleResponse> UnaryCallAsync(global::grpc.testing.SimpleRequest request, Metadata headers = null, CancellationToken cancellationToken = default(CancellationToken));
AsyncServerStreamingCall<global::grpc.testing.StreamingOutputCallResponse> StreamingOutputCall(global::grpc.testing.StreamingOutputCallRequest request, Metadata headers = null, CancellationToken cancellationToken = default(CancellationToken));
AsyncClientStreamingCall<global::grpc.testing.StreamingInputCallRequest, global::grpc.testing.StreamingInputCallResponse> StreamingInputCall(Metadata headers = null, CancellationToken cancellationToken = default(CancellationToken));
AsyncDuplexStreamingCall<global::grpc.testing.StreamingOutputCallRequest, global::grpc.testing.StreamingOutputCallResponse> FullDuplexCall(Metadata headers = null, CancellationToken cancellationToken = default(CancellationToken));
@ -72,12 +72,12 @@ namespace grpc.testing {
// server-side interface
public interface ITestService
{
Task<global::grpc.testing.Empty> EmptyCall(ServerCallContext context, global::grpc.testing.Empty request);
Task<global::grpc.testing.SimpleResponse> UnaryCall(ServerCallContext context, global::grpc.testing.SimpleRequest request);
Task StreamingOutputCall(ServerCallContext context, global::grpc.testing.StreamingOutputCallRequest request, IServerStreamWriter<global::grpc.testing.StreamingOutputCallResponse> responseStream);
Task<global::grpc.testing.StreamingInputCallResponse> StreamingInputCall(ServerCallContext context, IAsyncStreamReader<global::grpc.testing.StreamingInputCallRequest> requestStream);
Task FullDuplexCall(ServerCallContext context, IAsyncStreamReader<global::grpc.testing.StreamingOutputCallRequest> requestStream, IServerStreamWriter<global::grpc.testing.StreamingOutputCallResponse> responseStream);
Task HalfDuplexCall(ServerCallContext context, IAsyncStreamReader<global::grpc.testing.StreamingOutputCallRequest> requestStream, IServerStreamWriter<global::grpc.testing.StreamingOutputCallResponse> responseStream);
Task<global::grpc.testing.Empty> EmptyCall(global::grpc.testing.Empty request, ServerCallContext context);
Task<global::grpc.testing.SimpleResponse> UnaryCall(global::grpc.testing.SimpleRequest request, ServerCallContext context);
Task StreamingOutputCall(global::grpc.testing.StreamingOutputCallRequest request, IServerStreamWriter<global::grpc.testing.StreamingOutputCallResponse> responseStream, ServerCallContext context);
Task<global::grpc.testing.StreamingInputCallResponse> StreamingInputCall(IAsyncStreamReader<global::grpc.testing.StreamingInputCallRequest> requestStream, ServerCallContext context);
Task FullDuplexCall(IAsyncStreamReader<global::grpc.testing.StreamingOutputCallRequest> requestStream, IServerStreamWriter<global::grpc.testing.StreamingOutputCallResponse> responseStream, ServerCallContext context);
Task HalfDuplexCall(IAsyncStreamReader<global::grpc.testing.StreamingOutputCallRequest> requestStream, IServerStreamWriter<global::grpc.testing.StreamingOutputCallResponse> responseStream, ServerCallContext context);
}
// client stub
@ -91,7 +91,7 @@ namespace grpc.testing {
var call = CreateCall(__ServiceName, __Method_EmptyCall, headers);
return Calls.BlockingUnaryCall(call, request, cancellationToken);
}
public Task<global::grpc.testing.Empty> EmptyCallAsync(global::grpc.testing.Empty request, Metadata headers = null, CancellationToken cancellationToken = default(CancellationToken))
public AsyncUnaryCall<global::grpc.testing.Empty> EmptyCallAsync(global::grpc.testing.Empty request, Metadata headers = null, CancellationToken cancellationToken = default(CancellationToken))
{
var call = CreateCall(__ServiceName, __Method_EmptyCall, headers);
return Calls.AsyncUnaryCall(call, request, cancellationToken);
@ -101,7 +101,7 @@ namespace grpc.testing {
var call = CreateCall(__ServiceName, __Method_UnaryCall, headers);
return Calls.BlockingUnaryCall(call, request, cancellationToken);
}
public Task<global::grpc.testing.SimpleResponse> UnaryCallAsync(global::grpc.testing.SimpleRequest request, Metadata headers = null, CancellationToken cancellationToken = default(CancellationToken))
public AsyncUnaryCall<global::grpc.testing.SimpleResponse> UnaryCallAsync(global::grpc.testing.SimpleRequest request, Metadata headers = null, CancellationToken cancellationToken = default(CancellationToken))
{
var call = CreateCall(__ServiceName, __Method_UnaryCall, headers);
return Calls.AsyncUnaryCall(call, request, cancellationToken);

@ -46,19 +46,19 @@ namespace grpc.testing
/// </summary>
public class TestServiceImpl : TestService.ITestService
{
public Task<Empty> EmptyCall(ServerCallContext context, Empty request)
public Task<Empty> EmptyCall(Empty request, ServerCallContext context)
{
return Task.FromResult(Empty.DefaultInstance);
}
public Task<SimpleResponse> UnaryCall(ServerCallContext context, SimpleRequest request)
public Task<SimpleResponse> UnaryCall(SimpleRequest request, ServerCallContext context)
{
var response = SimpleResponse.CreateBuilder()
.SetPayload(CreateZerosPayload(request.ResponseSize)).Build();
return Task.FromResult(response);
}
public async Task StreamingOutputCall(ServerCallContext context, StreamingOutputCallRequest request, IServerStreamWriter<StreamingOutputCallResponse> responseStream)
public async Task StreamingOutputCall(StreamingOutputCallRequest request, IServerStreamWriter<StreamingOutputCallResponse> responseStream, ServerCallContext context)
{
foreach (var responseParam in request.ResponseParametersList)
{
@ -68,7 +68,7 @@ namespace grpc.testing
}
}
public async Task<StreamingInputCallResponse> StreamingInputCall(ServerCallContext context, IAsyncStreamReader<StreamingInputCallRequest> requestStream)
public async Task<StreamingInputCallResponse> StreamingInputCall(IAsyncStreamReader<StreamingInputCallRequest> requestStream, ServerCallContext context)
{
int sum = 0;
await requestStream.ForEach(async request =>
@ -78,7 +78,7 @@ namespace grpc.testing
return StreamingInputCallResponse.CreateBuilder().SetAggregatedPayloadSize(sum).Build();
}
public async Task FullDuplexCall(ServerCallContext context, IAsyncStreamReader<StreamingOutputCallRequest> requestStream, IServerStreamWriter<StreamingOutputCallResponse> responseStream)
public async Task FullDuplexCall(IAsyncStreamReader<StreamingOutputCallRequest> requestStream, IServerStreamWriter<StreamingOutputCallResponse> responseStream, ServerCallContext context)
{
await requestStream.ForEach(async request =>
{
@ -91,7 +91,7 @@ namespace grpc.testing
});
}
public async Task HalfDuplexCall(ServerCallContext context, IAsyncStreamReader<StreamingOutputCallRequest> requestStream, IServerStreamWriter<StreamingOutputCallResponse> responseStream)
public async Task HalfDuplexCall(IAsyncStreamReader<StreamingOutputCallRequest> requestStream, IServerStreamWriter<StreamingOutputCallResponse> responseStream, ServerCallContext context)
{
throw new NotImplementedException();
}

@ -167,6 +167,29 @@ grpcsharp_metadata_array_add(grpc_metadata_array *array, const char *key,
array->count++;
}
GPR_EXPORT gpr_intptr GPR_CALLTYPE
grpcsharp_metadata_array_count(grpc_metadata_array *array) {
return (gpr_intptr) array->count;
}
GPR_EXPORT const char *GPR_CALLTYPE
grpcsharp_metadata_array_get_key(grpc_metadata_array *array, size_t index) {
GPR_ASSERT(index < array->count);
return array->metadata[index].key;
}
GPR_EXPORT const char *GPR_CALLTYPE
grpcsharp_metadata_array_get_value(grpc_metadata_array *array, size_t index) {
GPR_ASSERT(index < array->count);
return array->metadata[index].value;
}
GPR_EXPORT gpr_intptr GPR_CALLTYPE
grpcsharp_metadata_array_get_value_length(grpc_metadata_array *array, size_t index) {
GPR_ASSERT(index < array->count);
return (gpr_intptr) array->metadata[index].value_length;
}
/* Move contents of metadata array */
void grpcsharp_metadata_array_move(grpc_metadata_array *dest,
grpc_metadata_array *src) {
@ -218,6 +241,12 @@ GPR_EXPORT void GPR_CALLTYPE grpcsharp_batch_context_destroy(grpcsharp_batch_con
gpr_free(ctx);
}
GPR_EXPORT const grpc_metadata_array *GPR_CALLTYPE
grpcsharp_batch_context_recv_initial_metadata(
const grpcsharp_batch_context *ctx) {
return &(ctx->recv_initial_metadata);
}
GPR_EXPORT gpr_intptr GPR_CALLTYPE grpcsharp_batch_context_recv_message_length(
const grpcsharp_batch_context *ctx) {
if (!ctx->recv_message) {
@ -260,6 +289,12 @@ grpcsharp_batch_context_recv_status_on_client_details(
return ctx->recv_status_on_client.status_details;
}
GPR_EXPORT const grpc_metadata_array *GPR_CALLTYPE
grpcsharp_batch_context_recv_status_on_client_trailing_metadata(
const grpcsharp_batch_context *ctx) {
return &(ctx->recv_status_on_client.trailing_metadata);
}
GPR_EXPORT grpc_call *GPR_CALLTYPE grpcsharp_batch_context_server_rpc_new_call(
const grpcsharp_batch_context *ctx) {
return ctx->server_rpc_new.call;
@ -271,6 +306,24 @@ grpcsharp_batch_context_server_rpc_new_method(
return ctx->server_rpc_new.call_details.method;
}
GPR_EXPORT const char *GPR_CALLTYPE
grpcsharp_batch_context_server_rpc_new_host(
const grpcsharp_batch_context *ctx) {
return ctx->server_rpc_new.call_details.host;
}
GPR_EXPORT gpr_timespec GPR_CALLTYPE
grpcsharp_batch_context_server_rpc_new_deadline(
const grpcsharp_batch_context *ctx) {
return ctx->server_rpc_new.call_details.deadline;
}
GPR_EXPORT const grpc_metadata_array *GPR_CALLTYPE
grpcsharp_batch_context_server_rpc_new_request_metadata(
const grpcsharp_batch_context *ctx) {
return &(ctx->server_rpc_new.request_metadata);
}
GPR_EXPORT gpr_int32 GPR_CALLTYPE
grpcsharp_batch_context_recv_close_on_server_cancelled(
const grpcsharp_batch_context *ctx) {
@ -589,15 +642,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);

Loading…
Cancel
Save