diff --git a/src/csharp/Grpc.Core.Tests/Grpc.Core.Tests.csproj b/src/csharp/Grpc.Core.Tests/Grpc.Core.Tests.csproj
index 7070802d0f4447e38ab85eaa470123b55ca7170a..9364779df9edc967ca3b612c0b38ff92f1960b79 100644
--- a/src/csharp/Grpc.Core.Tests/Grpc.Core.Tests.csproj
+++ b/src/csharp/Grpc.Core.Tests/Grpc.Core.Tests.csproj
@@ -51,6 +51,7 @@
<Compile Include="Internal\ChannelArgsSafeHandleTest.cs" />
<Compile Include="ChannelOptionsTest.cs" />
<Compile Include="Internal\TimespecTest.cs" />
+ <Compile Include="TimeoutsTest.cs" />
</ItemGroup>
<Import Project="$(MSBuildBinPath)\Microsoft.CSharp.targets" />
<ItemGroup>
diff --git a/src/csharp/Grpc.Core.Tests/TimeoutsTest.cs b/src/csharp/Grpc.Core.Tests/TimeoutsTest.cs
new file mode 100644
index 0000000000000000000000000000000000000000..7c844c7953d08dd35894e115462424cf7f0e87f8
--- /dev/null
+++ b/src/csharp/Grpc.Core.Tests/TimeoutsTest.cs
@@ -0,0 +1,192 @@
+#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.Diagnostics;
+using System.Linq;
+using System.Threading;
+using System.Threading.Tasks;
+using Grpc.Core;
+using Grpc.Core.Internal;
+using Grpc.Core.Utils;
+using NUnit.Framework;
+
+namespace Grpc.Core.Tests
+{
+ /// <summary>
+ /// Tests for Deadline support.
+ /// </summary>
+ public class TimeoutsTest
+ {
+ const string Host = "localhost";
+ const string ServiceName = "/tests.Test";
+
+ static readonly Method<string, string> TestMethod = new Method<string, string>(
+ MethodType.Unary,
+ "/tests.Test/Test",
+ Marshallers.StringMarshaller,
+ Marshallers.StringMarshaller);
+
+ static readonly ServerServiceDefinition ServiceDefinition = ServerServiceDefinition.CreateBuilder(ServiceName)
+ .AddMethod(TestMethod, TestMethodHandler)
+ .Build();
+
+ // provides a way how to retrieve an out-of-band result value from server handler
+ static TaskCompletionSource<string> stringFromServerHandlerTcs;
+
+ Server server;
+ Channel channel;
+
+ [SetUp]
+ public void Init()
+ {
+ server = new Server();
+ server.AddServiceDefinition(ServiceDefinition);
+ int port = server.AddListeningPort(Host, Server.PickUnusedPort);
+ server.Start();
+ channel = new Channel(Host, port);
+
+ stringFromServerHandlerTcs = new TaskCompletionSource<string>();
+ }
+
+ [TearDown]
+ public void Cleanup()
+ {
+ channel.Dispose();
+ server.ShutdownAsync().Wait();
+ }
+
+ [TestFixtureTearDown]
+ public void CleanupClass()
+ {
+ GrpcEnvironment.Shutdown();
+ }
+
+ [Test]
+ public void InfiniteDeadline()
+ {
+ // no deadline specified, check server sees infinite deadline
+ var internalCall = new Call<string, string>(ServiceName, TestMethod, channel, Metadata.Empty);
+ Assert.AreEqual("DATETIME_MAXVALUE", Calls.BlockingUnaryCall(internalCall, "RETURN_DEADLINE", CancellationToken.None));
+
+ // DateTime.MaxValue deadline specified, check server sees infinite deadline
+ var internalCall2 = new Call<string, string>(ServiceName, TestMethod, channel, Metadata.Empty, DateTime.MaxValue);
+ Assert.AreEqual("DATETIME_MAXVALUE", Calls.BlockingUnaryCall(internalCall2, "RETURN_DEADLINE", CancellationToken.None));
+ }
+
+ [Test]
+ public void DeadlineTransferredToServer()
+ {
+ var remainingTimeClient = TimeSpan.FromDays(7);
+ var deadline = DateTime.UtcNow + remainingTimeClient;
+ Thread.Sleep(1000);
+ var internalCall = new Call<string, string>(ServiceName, TestMethod, channel, Metadata.Empty, deadline);
+
+ var serverDeadlineTicksString = Calls.BlockingUnaryCall(internalCall, "RETURN_DEADLINE", CancellationToken.None);
+ var serverDeadline = new DateTime(long.Parse(serverDeadlineTicksString), DateTimeKind.Utc);
+
+ // A fairly relaxed check that the deadline set by client and deadline seen by server
+ // are in agreement. C core takes care of the work with transferring deadline over the wire,
+ // so we don't need an exact check here.
+ Assert.IsTrue(Math.Abs((deadline - serverDeadline).TotalMilliseconds) < 5000);
+ }
+
+ [Test]
+ public void DeadlineExceededStatusOnTimeout()
+ {
+ // no deadline specified, check server sees infinite deadline
+ var deadline = DateTime.UtcNow.Add(TimeSpan.FromSeconds(1));
+ var internalCall = new Call<string, string>(ServiceName, TestMethod, channel, Metadata.Empty, deadline);
+
+ try
+ {
+ Calls.BlockingUnaryCall(internalCall, "TIMEOUT", CancellationToken.None);
+ Assert.Fail();
+ }
+ catch (RpcException e)
+ {
+ Assert.AreEqual(StatusCode.DeadlineExceeded, e.Status.StatusCode);
+ }
+ }
+
+ [Test]
+ public void ServerReceivesCancellationOnTimeout()
+ {
+ // no deadline specified, check server sees infinite deadline
+ var deadline = DateTime.UtcNow.Add(TimeSpan.FromSeconds(1));
+ var internalCall = new Call<string, string>(ServiceName, TestMethod, channel, Metadata.Empty, deadline);
+
+ try
+ {
+ Calls.BlockingUnaryCall(internalCall, "CHECK_CANCELLATION_RECEIVED", CancellationToken.None);
+ Assert.Fail();
+ }
+ catch (RpcException e)
+ {
+ Assert.AreEqual(StatusCode.DeadlineExceeded, e.Status.StatusCode);
+ }
+ Assert.AreEqual("CANCELLED", stringFromServerHandlerTcs.Task.Result);
+ }
+
+ private static async Task<string> TestMethodHandler(string request, ServerCallContext context)
+ {
+ if (request == "TIMEOUT")
+ {
+ await Task.Delay(60000);
+ return "";
+ }
+
+ if (request == "RETURN_DEADLINE")
+ {
+ if (context.Deadline == DateTime.MaxValue)
+ {
+ return "DATETIME_MAXVALUE";
+ }
+
+ return context.Deadline.Ticks.ToString();
+ }
+
+ if (request == "CHECK_CANCELLATION_RECEIVED")
+ {
+ // wait until cancellation token is fired.
+ var tcs = new TaskCompletionSource<object>();
+ context.CancellationToken.Register(() => { tcs.SetResult(null); });
+ await tcs.Task;
+ stringFromServerHandlerTcs.SetResult("CANCELLED");
+ return "";
+ }
+
+ return "";
+ }
+ }
+}
diff --git a/src/csharp/Grpc.Core/Call.cs b/src/csharp/Grpc.Core/Call.cs
index 37b452f020d390acd31a0deacc468be13688a562..94c5e26082264f720a367d1263ad715416bae536 100644
--- a/src/csharp/Grpc.Core/Call.cs
+++ b/src/csharp/Grpc.Core/Call.cs
@@ -47,14 +47,21 @@ namespace Grpc.Core
readonly Marshaller<TResponse> responseMarshaller;
readonly Channel channel;
readonly Metadata headers;
+ readonly DateTime deadline;
public Call(string serviceName, Method<TRequest, TResponse> method, Channel channel, Metadata headers)
+ : this(serviceName, method, channel, headers, DateTime.MaxValue)
+ {
+ }
+
+ public Call(string serviceName, Method<TRequest, TResponse> method, Channel channel, Metadata headers, DateTime deadline)
{
this.name = method.GetFullName(serviceName);
this.requestMarshaller = method.RequestMarshaller;
this.responseMarshaller = method.ResponseMarshaller;
this.channel = Preconditions.CheckNotNull(channel);
this.headers = Preconditions.CheckNotNull(headers);
+ this.deadline = deadline;
}
public Channel Channel
@@ -87,6 +94,14 @@ namespace Grpc.Core
}
}
+ public DateTime Deadline
+ {
+ get
+ {
+ return this.deadline;
+ }
+ }
+
public Marshaller<TRequest> RequestMarshaller
{
get
diff --git a/src/csharp/Grpc.Core/Calls.cs b/src/csharp/Grpc.Core/Calls.cs
index 359fe5374162d9292d747532eb6a049af333c22a..054fc2749170516f4c4ead69ea0439a4e18815c0 100644
--- a/src/csharp/Grpc.Core/Calls.cs
+++ b/src/csharp/Grpc.Core/Calls.cs
@@ -50,7 +50,7 @@ namespace Grpc.Core
var asyncCall = new AsyncCall<TRequest, TResponse>(call.RequestMarshaller.Serializer, call.ResponseMarshaller.Deserializer);
// TODO(jtattermusch): this gives a race that cancellation can be requested before the call even starts.
RegisterCancellationCallback(asyncCall, token);
- return asyncCall.UnaryCall(call.Channel, call.Name, req, call.Headers);
+ return asyncCall.UnaryCall(call.Channel, call.Name, req, call.Headers, call.Deadline);
}
public static AsyncUnaryCall<TResponse> AsyncUnaryCall<TRequest, TResponse>(Call<TRequest, TResponse> call, TRequest req, CancellationToken token)
@@ -58,8 +58,8 @@ namespace Grpc.Core
where TResponse : class
{
var asyncCall = new AsyncCall<TRequest, TResponse>(call.RequestMarshaller.Serializer, call.ResponseMarshaller.Deserializer);
- asyncCall.Initialize(call.Channel, call.Channel.CompletionQueue, call.Name);
- var asyncResult = asyncCall.UnaryCallAsync(req, call.Headers);
+ asyncCall.Initialize(call.Channel, call.Channel.CompletionQueue, call.Name, Timespec.FromDateTime(call.Deadline));
+ var asyncResult = asyncCall.UnaryCallAsync(req, call.Headers, call.Deadline);
RegisterCancellationCallback(asyncCall, token);
return new AsyncUnaryCall<TResponse>(asyncResult, asyncCall.GetStatus, asyncCall.GetTrailers, asyncCall.Cancel);
}
@@ -69,8 +69,8 @@ namespace Grpc.Core
where TResponse : class
{
var asyncCall = new AsyncCall<TRequest, TResponse>(call.RequestMarshaller.Serializer, call.ResponseMarshaller.Deserializer);
- asyncCall.Initialize(call.Channel, call.Channel.CompletionQueue, call.Name);
- asyncCall.StartServerStreamingCall(req, call.Headers);
+ asyncCall.Initialize(call.Channel, call.Channel.CompletionQueue, call.Name, Timespec.FromDateTime(call.Deadline));
+ asyncCall.StartServerStreamingCall(req, call.Headers, call.Deadline);
RegisterCancellationCallback(asyncCall, token);
var responseStream = new ClientResponseStream<TRequest, TResponse>(asyncCall);
return new AsyncServerStreamingCall<TResponse>(responseStream, asyncCall.GetStatus, asyncCall.GetTrailers, asyncCall.Cancel);
@@ -81,8 +81,8 @@ namespace Grpc.Core
where TResponse : class
{
var asyncCall = new AsyncCall<TRequest, TResponse>(call.RequestMarshaller.Serializer, call.ResponseMarshaller.Deserializer);
- asyncCall.Initialize(call.Channel, call.Channel.CompletionQueue, call.Name);
- var resultTask = asyncCall.ClientStreamingCallAsync(call.Headers);
+ asyncCall.Initialize(call.Channel, call.Channel.CompletionQueue, call.Name, Timespec.FromDateTime(call.Deadline));
+ var resultTask = asyncCall.ClientStreamingCallAsync(call.Headers, call.Deadline);
RegisterCancellationCallback(asyncCall, token);
var requestStream = new ClientRequestStream<TRequest, TResponse>(asyncCall);
return new AsyncClientStreamingCall<TRequest, TResponse>(requestStream, resultTask, asyncCall.GetStatus, asyncCall.GetTrailers, asyncCall.Cancel);
@@ -93,8 +93,8 @@ namespace Grpc.Core
where TResponse : class
{
var asyncCall = new AsyncCall<TRequest, TResponse>(call.RequestMarshaller.Serializer, call.ResponseMarshaller.Deserializer);
- asyncCall.Initialize(call.Channel, call.Channel.CompletionQueue, call.Name);
- asyncCall.StartDuplexStreamingCall(call.Headers);
+ asyncCall.Initialize(call.Channel, call.Channel.CompletionQueue, call.Name, Timespec.FromDateTime(call.Deadline));
+ asyncCall.StartDuplexStreamingCall(call.Headers, call.Deadline);
RegisterCancellationCallback(asyncCall, token);
var requestStream = new ClientRequestStream<TRequest, TResponse>(asyncCall);
var responseStream = new ClientResponseStream<TRequest, TResponse>(asyncCall);
diff --git a/src/csharp/Grpc.Core/ClientBase.cs b/src/csharp/Grpc.Core/ClientBase.cs
index a099f96aeab4c31fceccf7203bc90f4e12a6f093..fd3473128a975faebfc99fc59cdd75f5cb3a4dd6 100644
--- a/src/csharp/Grpc.Core/ClientBase.cs
+++ b/src/csharp/Grpc.Core/ClientBase.cs
@@ -76,7 +76,7 @@ namespace Grpc.Core
/// <summary>
/// Creates a new call to given method.
/// </summary>
- protected Call<TRequest, TResponse> CreateCall<TRequest, TResponse>(string serviceName, Method<TRequest, TResponse> method, Metadata metadata)
+ protected Call<TRequest, TResponse> CreateCall<TRequest, TResponse>(string serviceName, Method<TRequest, TResponse> method, Metadata metadata, DateTime? deadline)
where TRequest : class
where TResponse : class
{
@@ -87,8 +87,8 @@ namespace Grpc.Core
interceptor(metadata);
metadata.Freeze();
}
- metadata = metadata ?? Metadata.Empty;
- return new Call<TRequest, TResponse>(serviceName, method, channel, metadata);
+ return new Call<TRequest, TResponse>(serviceName, method, channel,
+ metadata ?? Metadata.Empty, deadline ?? DateTime.MaxValue);
}
}
}
diff --git a/src/csharp/Grpc.Core/Internal/AsyncCall.cs b/src/csharp/Grpc.Core/Internal/AsyncCall.cs
index f983dbb759cda8c9eed413af1812a1d3aad987e3..51022ac34fc1646fb2a607c542725c8e6c287164 100644
--- a/src/csharp/Grpc.Core/Internal/AsyncCall.cs
+++ b/src/csharp/Grpc.Core/Internal/AsyncCall.cs
@@ -61,10 +61,10 @@ namespace Grpc.Core.Internal
{
}
- public void Initialize(Channel channel, CompletionQueueSafeHandle cq, string methodName)
+ public void Initialize(Channel channel, CompletionQueueSafeHandle cq, string methodName, Timespec deadline)
{
this.channel = channel;
- var call = CallSafeHandle.Create(channel.Handle, channel.CompletionRegistry, cq, methodName, channel.Target, Timespec.InfFuture);
+ var call = channel.Handle.CreateCall(channel.CompletionRegistry, cq, methodName, channel.Target, deadline);
channel.Environment.DebugStats.ActiveClientCalls.Increment();
InitializeInternal(call);
}
@@ -76,7 +76,7 @@ namespace Grpc.Core.Internal
/// <summary>
/// Blocking unary request - unary response call.
/// </summary>
- public TResponse UnaryCall(Channel channel, string methodName, TRequest msg, Metadata headers)
+ public TResponse UnaryCall(Channel channel, string methodName, TRequest msg, Metadata headers, DateTime deadline)
{
using (CompletionQueueSafeHandle cq = CompletionQueueSafeHandle.Create())
{
@@ -86,7 +86,7 @@ namespace Grpc.Core.Internal
lock (myLock)
{
- Initialize(channel, cq, methodName);
+ Initialize(channel, cq, methodName, Timespec.FromDateTime(deadline));
started = true;
halfcloseRequested = true;
readingDone = true;
@@ -126,7 +126,7 @@ namespace Grpc.Core.Internal
/// <summary>
/// Starts a unary request - unary response call.
/// </summary>
- public Task<TResponse> UnaryCallAsync(TRequest msg, Metadata headers)
+ public Task<TResponse> UnaryCallAsync(TRequest msg, Metadata headers, DateTime deadline)
{
lock (myLock)
{
@@ -151,7 +151,7 @@ namespace Grpc.Core.Internal
/// Starts a streamed request - unary response call.
/// Use StartSendMessage and StartSendCloseFromClient to stream requests.
/// </summary>
- public Task<TResponse> ClientStreamingCallAsync(Metadata headers)
+ public Task<TResponse> ClientStreamingCallAsync(Metadata headers, DateTime deadline)
{
lock (myLock)
{
@@ -173,7 +173,7 @@ namespace Grpc.Core.Internal
/// <summary>
/// Starts a unary request - streamed response call.
/// </summary>
- public void StartServerStreamingCall(TRequest msg, Metadata headers)
+ public void StartServerStreamingCall(TRequest msg, Metadata headers, DateTime deadline)
{
lock (myLock)
{
@@ -196,7 +196,7 @@ namespace Grpc.Core.Internal
/// Starts a streaming request - streaming response call.
/// Use StartSendMessage and StartSendCloseFromClient to stream requests.
/// </summary>
- public void StartDuplexStreamingCall(Metadata headers)
+ public void StartDuplexStreamingCall(Metadata headers, DateTime deadline)
{
lock (myLock)
{
diff --git a/src/csharp/Grpc.Core/Internal/AsyncCallServer.cs b/src/csharp/Grpc.Core/Internal/AsyncCallServer.cs
index f809f4a84ca06b83ad19146b35b294bee127774c..702f1ce9e7eb2ed7e23d68302081f229c8e20abd 100644
--- a/src/csharp/Grpc.Core/Internal/AsyncCallServer.cs
+++ b/src/csharp/Grpc.Core/Internal/AsyncCallServer.cs
@@ -48,6 +48,7 @@ namespace Grpc.Core.Internal
internal class AsyncCallServer<TRequest, TResponse> : AsyncCallBase<TResponse, TRequest>
{
readonly TaskCompletionSource<object> finishedServersideTcs = new TaskCompletionSource<object>();
+ readonly CancellationTokenSource cancellationTokenSource = new CancellationTokenSource();
readonly GrpcEnvironment environment;
public AsyncCallServer(Func<TResponse, byte[]> serializer, Func<byte[], TRequest> deserializer, GrpcEnvironment environment) : base(serializer, deserializer)
@@ -118,6 +119,18 @@ namespace Grpc.Core.Internal
}
}
+ /// <summary>
+ /// Gets cancellation token that gets cancelled once close completion
+ /// is received and the cancelled flag is set.
+ /// </summary>
+ public CancellationToken CancellationToken
+ {
+ get
+ {
+ return cancellationTokenSource.Token;
+ }
+ }
+
protected override void OnReleaseResources()
{
environment.DebugStats.ActiveServerCalls.Decrement();
@@ -138,6 +151,8 @@ namespace Grpc.Core.Internal
{
// Once we cancel, we don't have to care that much
// about reads and writes.
+
+ // TODO(jtattermusch): is this still necessary?
Cancel();
}
@@ -145,6 +160,11 @@ namespace Grpc.Core.Internal
}
// TODO(jtattermusch): handle error
+ if (cancelled)
+ {
+ cancellationTokenSource.Cancel();
+ }
+
finishedServersideTcs.SetResult(null);
}
}
diff --git a/src/csharp/Grpc.Core/Internal/CallSafeHandle.cs b/src/csharp/Grpc.Core/Internal/CallSafeHandle.cs
index 19dbb83f2439f6a4c72a7550aa8ddea7b40a89a7..04e35a5efdff563eaf460abc2c076b51f6520f9f 100644
--- a/src/csharp/Grpc.Core/Internal/CallSafeHandle.cs
+++ b/src/csharp/Grpc.Core/Internal/CallSafeHandle.cs
@@ -45,9 +45,6 @@ namespace Grpc.Core.Internal
const uint GRPC_WRITE_BUFFER_HINT = 1;
CompletionRegistry completionRegistry;
- [DllImport("grpc_csharp_ext.dll")]
- static extern CallSafeHandle grpcsharp_channel_create_call(ChannelSafeHandle channel, CompletionQueueSafeHandle cq, string method, string host, Timespec deadline);
-
[DllImport("grpc_csharp_ext.dll")]
static extern GRPCCallError grpcsharp_call_cancel(CallSafeHandle call);
@@ -98,13 +95,6 @@ namespace Grpc.Core.Internal
{
}
- public static CallSafeHandle Create(ChannelSafeHandle channel, CompletionRegistry registry, CompletionQueueSafeHandle cq, string method, string host, Timespec deadline)
- {
- var result = grpcsharp_channel_create_call(channel, cq, method, host, deadline);
- result.SetCompletionRegistry(registry);
- return result;
- }
-
public void SetCompletionRegistry(CompletionRegistry completionRegistry)
{
this.completionRegistry = completionRegistry;
diff --git a/src/csharp/Grpc.Core/Internal/ChannelSafeHandle.cs b/src/csharp/Grpc.Core/Internal/ChannelSafeHandle.cs
index f046f4c6d0d4b0ee32b869ee2d09aea6d1c32aeb..53c506c872ed8ecaa22ad57143ad80d6bd1fcf31 100644
--- a/src/csharp/Grpc.Core/Internal/ChannelSafeHandle.cs
+++ b/src/csharp/Grpc.Core/Internal/ChannelSafeHandle.cs
@@ -46,6 +46,9 @@ namespace Grpc.Core.Internal
[DllImport("grpc_csharp_ext.dll")]
static extern ChannelSafeHandle grpcsharp_secure_channel_create(CredentialsSafeHandle credentials, string target, ChannelArgsSafeHandle channelArgs);
+ [DllImport("grpc_csharp_ext.dll")]
+ static extern CallSafeHandle grpcsharp_channel_create_call(ChannelSafeHandle channel, CompletionQueueSafeHandle cq, string method, string host, Timespec deadline);
+
[DllImport("grpc_csharp_ext.dll")]
static extern void grpcsharp_channel_destroy(IntPtr channel);
@@ -63,6 +66,13 @@ namespace Grpc.Core.Internal
return grpcsharp_secure_channel_create(credentials, target, channelArgs);
}
+ public CallSafeHandle CreateCall(CompletionRegistry registry, CompletionQueueSafeHandle cq, string method, string host, Timespec deadline)
+ {
+ var result = grpcsharp_channel_create_call(this, cq, method, host, deadline);
+ result.SetCompletionRegistry(registry);
+ return result;
+ }
+
protected override bool ReleaseHandle()
{
grpcsharp_channel_destroy(handle);
diff --git a/src/csharp/Grpc.Core/Internal/ServerCallHandler.cs b/src/csharp/Grpc.Core/Internal/ServerCallHandler.cs
index 3680b1e791b562249f9494d0e53f6babc23357f4..9bbcb172669f7b1a6766130fc6a90ab727754a0c 100644
--- a/src/csharp/Grpc.Core/Internal/ServerCallHandler.cs
+++ b/src/csharp/Grpc.Core/Internal/ServerCallHandler.cs
@@ -72,7 +72,7 @@ namespace Grpc.Core.Internal
var responseStream = new ServerResponseStream<TRequest, TResponse>(asyncCall);
Status status;
- var context = HandlerUtils.NewContext(newRpc);
+ var context = HandlerUtils.NewContext(newRpc, asyncCall.CancellationToken);
try
{
Preconditions.CheckArgument(await requestStream.MoveNext());
@@ -126,7 +126,7 @@ namespace Grpc.Core.Internal
var responseStream = new ServerResponseStream<TRequest, TResponse>(asyncCall);
Status status;
- var context = HandlerUtils.NewContext(newRpc);
+ var context = HandlerUtils.NewContext(newRpc, asyncCall.CancellationToken);
try
{
Preconditions.CheckArgument(await requestStream.MoveNext());
@@ -180,7 +180,7 @@ namespace Grpc.Core.Internal
var responseStream = new ServerResponseStream<TRequest, TResponse>(asyncCall);
Status status;
- var context = HandlerUtils.NewContext(newRpc);
+ var context = HandlerUtils.NewContext(newRpc, asyncCall.CancellationToken);
try
{
var result = await handler(requestStream, context);
@@ -238,7 +238,7 @@ namespace Grpc.Core.Internal
var responseStream = new ServerResponseStream<TRequest, TResponse>(asyncCall);
Status status;
- var context = HandlerUtils.NewContext(newRpc);
+ var context = HandlerUtils.NewContext(newRpc, asyncCall.CancellationToken);
try
{
await handler(requestStream, responseStream, context);
@@ -295,11 +295,11 @@ namespace Grpc.Core.Internal
return new Status(StatusCode.Unknown, "Exception was thrown by handler.");
}
- public static ServerCallContext NewContext(ServerRpcNew newRpc)
+ public static ServerCallContext NewContext(ServerRpcNew newRpc, CancellationToken cancellationToken)
{
return new ServerCallContext(
newRpc.Method, newRpc.Host, newRpc.Deadline.ToDateTime(),
- newRpc.RequestMetadata, CancellationToken.None);
+ newRpc.RequestMetadata, cancellationToken);
}
}
}