aboutsummaryrefslogtreecommitdiffhomepage
path: root/src/csharp
diff options
context:
space:
mode:
Diffstat (limited to 'src/csharp')
-rw-r--r--src/csharp/.gitignore1
-rw-r--r--src/csharp/Grpc.Core.Tests/ChannelTest.cs33
-rw-r--r--src/csharp/Grpc.Core.Tests/ClientServerTest.cs15
-rw-r--r--src/csharp/Grpc.Core.Tests/CompressionTest.cs8
-rw-r--r--src/csharp/Grpc.Core.Tests/ContextPropagationTest.cs8
-rw-r--r--src/csharp/Grpc.Core.Tests/Grpc.Core.Tests.csproj3
-rw-r--r--src/csharp/Grpc.Core.Tests/GrpcEnvironmentTest.cs33
-rw-r--r--src/csharp/Grpc.Core.Tests/Internal/AsyncCallTest.cs222
-rw-r--r--src/csharp/Grpc.Core.Tests/MetadataTest.cs120
-rw-r--r--src/csharp/Grpc.Core.Tests/ResponseHeadersTest.cs79
-rw-r--r--src/csharp/Grpc.Core.Tests/ServerTest.cs5
-rw-r--r--src/csharp/Grpc.Core.Tests/ShutdownTest.cs77
-rw-r--r--src/csharp/Grpc.Core.Tests/TimeoutsTest.cs8
-rw-r--r--src/csharp/Grpc.Core/AsyncClientStreamingCall.cs15
-rw-r--r--src/csharp/Grpc.Core/AsyncDuplexStreamingCall.cs16
-rw-r--r--src/csharp/Grpc.Core/AsyncServerStreamingCall.cs16
-rw-r--r--src/csharp/Grpc.Core/AsyncUnaryCall.cs15
-rw-r--r--src/csharp/Grpc.Core/Calls.cs8
-rw-r--r--src/csharp/Grpc.Core/Channel.cs51
-rw-r--r--src/csharp/Grpc.Core/ClientBase.cs3
-rw-r--r--src/csharp/Grpc.Core/ContextPropagationToken.cs1
-rw-r--r--src/csharp/Grpc.Core/Grpc.Core.csproj1
-rw-r--r--src/csharp/Grpc.Core/GrpcEnvironment.cs44
-rw-r--r--src/csharp/Grpc.Core/Internal/AsyncCall.cs149
-rw-r--r--src/csharp/Grpc.Core/Internal/AsyncCallBase.cs64
-rw-r--r--src/csharp/Grpc.Core/Internal/AsyncCallServer.cs23
-rw-r--r--src/csharp/Grpc.Core/Internal/BatchContextSafeHandle.cs16
-rw-r--r--src/csharp/Grpc.Core/Internal/CallSafeHandle.cs53
-rw-r--r--src/csharp/Grpc.Core/Internal/ChannelSafeHandle.cs7
-rw-r--r--src/csharp/Grpc.Core/Internal/ClientResponseStream.cs8
-rw-r--r--src/csharp/Grpc.Core/Internal/DebugStats.cs14
-rw-r--r--src/csharp/Grpc.Core/Internal/GrpcThreadPool.cs3
-rw-r--r--src/csharp/Grpc.Core/Internal/INativeCall.cs85
-rw-r--r--src/csharp/Grpc.Core/Internal/MetadataArraySafeHandle.cs5
-rw-r--r--src/csharp/Grpc.Core/Internal/ServerCallHandler.cs10
-rw-r--r--src/csharp/Grpc.Core/Internal/ServerSafeHandle.cs4
-rw-r--r--src/csharp/Grpc.Core/Logging/ConsoleLogger.cs14
-rw-r--r--src/csharp/Grpc.Core/Metadata.cs112
-rw-r--r--src/csharp/Grpc.Core/Server.cs57
-rw-r--r--src/csharp/Grpc.Examples.MathClient/MathClient.cs20
-rw-r--r--src/csharp/Grpc.Examples.MathServer/MathServer.cs1
-rw-r--r--src/csharp/Grpc.Examples.Tests/MathClientServerTests.cs3
-rw-r--r--src/csharp/Grpc.HealthCheck.Tests/HealthClientServerTest.cs3
-rw-r--r--src/csharp/Grpc.IntegrationTesting/InteropClient.cs40
-rw-r--r--src/csharp/Grpc.IntegrationTesting/InteropClientServerTest.cs9
-rw-r--r--src/csharp/Grpc.IntegrationTesting/InteropServer.cs2
-rw-r--r--src/csharp/Grpc.IntegrationTesting/SslCredentialsTest.cs3
-rw-r--r--src/csharp/ext/grpc_csharp_ext.c55
48 files changed, 1182 insertions, 360 deletions
diff --git a/src/csharp/.gitignore b/src/csharp/.gitignore
index ae48956567..48365e32a5 100644
--- a/src/csharp/.gitignore
+++ b/src/csharp/.gitignore
@@ -5,4 +5,5 @@ test-results
packages
Grpc.v12.suo
TestResult.xml
+/TestResults
*.nupkg
diff --git a/src/csharp/Grpc.Core.Tests/ChannelTest.cs b/src/csharp/Grpc.Core.Tests/ChannelTest.cs
index 2787572924..dfbd92879e 100644
--- a/src/csharp/Grpc.Core.Tests/ChannelTest.cs
+++ b/src/csharp/Grpc.Core.Tests/ChannelTest.cs
@@ -41,12 +41,6 @@ namespace Grpc.Core.Tests
{
public class ChannelTest
{
- [TestFixtureTearDown]
- public void CleanupClass()
- {
- GrpcEnvironment.Shutdown();
- }
-
[Test]
public void Constructor_RejectsInvalidParams()
{
@@ -56,36 +50,33 @@ namespace Grpc.Core.Tests
[Test]
public void State_IdleAfterCreation()
{
- using (var channel = new Channel("localhost", Credentials.Insecure))
- {
- Assert.AreEqual(ChannelState.Idle, channel.State);
- }
+ var channel = new Channel("localhost", Credentials.Insecure);
+ Assert.AreEqual(ChannelState.Idle, channel.State);
+ channel.ShutdownAsync().Wait();
}
[Test]
public void WaitForStateChangedAsync_InvalidArgument()
{
- using (var channel = new Channel("localhost", Credentials.Insecure))
- {
- Assert.Throws(typeof(ArgumentException), () => channel.WaitForStateChangedAsync(ChannelState.FatalFailure));
- }
+ var channel = new Channel("localhost", Credentials.Insecure);
+ Assert.Throws(typeof(ArgumentException), () => channel.WaitForStateChangedAsync(ChannelState.FatalFailure));
+ channel.ShutdownAsync().Wait();
}
[Test]
public void ResolvedTarget()
{
- using (var channel = new Channel("127.0.0.1", Credentials.Insecure))
- {
- Assert.IsTrue(channel.ResolvedTarget.Contains("127.0.0.1"));
- }
+ var channel = new Channel("127.0.0.1", Credentials.Insecure);
+ Assert.IsTrue(channel.ResolvedTarget.Contains("127.0.0.1"));
+ channel.ShutdownAsync().Wait();
}
[Test]
- public void Dispose_IsIdempotent()
+ public void Shutdown_AllowedOnlyOnce()
{
var channel = new Channel("localhost", Credentials.Insecure);
- channel.Dispose();
- channel.Dispose();
+ channel.ShutdownAsync().Wait();
+ Assert.Throws(typeof(InvalidOperationException), () => channel.ShutdownAsync().GetAwaiter().GetResult());
}
}
}
diff --git a/src/csharp/Grpc.Core.Tests/ClientServerTest.cs b/src/csharp/Grpc.Core.Tests/ClientServerTest.cs
index e49fdb5268..68279a2007 100644
--- a/src/csharp/Grpc.Core.Tests/ClientServerTest.cs
+++ b/src/csharp/Grpc.Core.Tests/ClientServerTest.cs
@@ -63,16 +63,10 @@ namespace Grpc.Core.Tests
[TearDown]
public void Cleanup()
{
- channel.Dispose();
+ channel.ShutdownAsync().Wait();
server.ShutdownAsync().Wait();
}
- [TestFixtureTearDown]
- public void CleanupClass()
- {
- GrpcEnvironment.Shutdown();
- }
-
[Test]
public async Task UnaryCall()
{
@@ -208,13 +202,6 @@ namespace Grpc.Core.Tests
}
[Test]
- public void UnaryCall_DisposedChannel()
- {
- channel.Dispose();
- Assert.Throws(typeof(ObjectDisposedException), () => Calls.BlockingUnaryCall(helper.CreateUnaryCall(), "ABC"));
- }
-
- [Test]
public void UnaryCallPerformance()
{
helper.UnaryHandler = new UnaryServerMethod<string, string>(async (request, context) =>
diff --git a/src/csharp/Grpc.Core.Tests/CompressionTest.cs b/src/csharp/Grpc.Core.Tests/CompressionTest.cs
index 9547683f60..378c81851c 100644
--- a/src/csharp/Grpc.Core.Tests/CompressionTest.cs
+++ b/src/csharp/Grpc.Core.Tests/CompressionTest.cs
@@ -62,16 +62,10 @@ namespace Grpc.Core.Tests
[TearDown]
public void Cleanup()
{
- channel.Dispose();
+ channel.ShutdownAsync().Wait();
server.ShutdownAsync().Wait();
}
- [TestFixtureTearDown]
- public void CleanupClass()
- {
- GrpcEnvironment.Shutdown();
- }
-
[Test]
public void WriteOptions_Unary()
{
diff --git a/src/csharp/Grpc.Core.Tests/ContextPropagationTest.cs b/src/csharp/Grpc.Core.Tests/ContextPropagationTest.cs
index db5f953b0e..2db3f286f7 100644
--- a/src/csharp/Grpc.Core.Tests/ContextPropagationTest.cs
+++ b/src/csharp/Grpc.Core.Tests/ContextPropagationTest.cs
@@ -62,16 +62,10 @@ namespace Grpc.Core.Tests
[TearDown]
public void Cleanup()
{
- channel.Dispose();
+ channel.ShutdownAsync().Wait();
server.ShutdownAsync().Wait();
}
- [TestFixtureTearDown]
- public void CleanupClass()
- {
- GrpcEnvironment.Shutdown();
- }
-
[Test]
public async Task PropagateCancellation()
{
diff --git a/src/csharp/Grpc.Core.Tests/Grpc.Core.Tests.csproj b/src/csharp/Grpc.Core.Tests/Grpc.Core.Tests.csproj
index d6a8f52570..b571fe9025 100644
--- a/src/csharp/Grpc.Core.Tests/Grpc.Core.Tests.csproj
+++ b/src/csharp/Grpc.Core.Tests/Grpc.Core.Tests.csproj
@@ -64,6 +64,8 @@
<Link>Version.cs</Link>
</Compile>
<Compile Include="ClientBaseTest.cs" />
+ <Compile Include="ShutdownTest.cs" />
+ <Compile Include="Internal\AsyncCallTest.cs" />
<Compile Include="Properties\AssemblyInfo.cs" />
<Compile Include="ClientServerTest.cs" />
<Compile Include="ServerTest.cs" />
@@ -82,6 +84,7 @@
<Compile Include="ResponseHeadersTest.cs" />
<Compile Include="CompressionTest.cs" />
<Compile Include="ContextPropagationTest.cs" />
+ <Compile Include="MetadataTest.cs" />
</ItemGroup>
<Import Project="$(MSBuildBinPath)\Microsoft.CSharp.targets" />
<ItemGroup>
diff --git a/src/csharp/Grpc.Core.Tests/GrpcEnvironmentTest.cs b/src/csharp/Grpc.Core.Tests/GrpcEnvironmentTest.cs
index 4ed93c7eca..78295cf6d4 100644
--- a/src/csharp/Grpc.Core.Tests/GrpcEnvironmentTest.cs
+++ b/src/csharp/Grpc.Core.Tests/GrpcEnvironmentTest.cs
@@ -43,31 +43,40 @@ namespace Grpc.Core.Tests
[Test]
public void InitializeAndShutdownGrpcEnvironment()
{
- var env = GrpcEnvironment.GetInstance();
+ var env = GrpcEnvironment.AddRef();
Assert.IsNotNull(env.CompletionQueue);
- GrpcEnvironment.Shutdown();
+ GrpcEnvironment.Release();
}
[Test]
public void SubsequentInvocations()
{
- var env1 = GrpcEnvironment.GetInstance();
- var env2 = GrpcEnvironment.GetInstance();
- Assert.IsTrue(object.ReferenceEquals(env1, env2));
- GrpcEnvironment.Shutdown();
- GrpcEnvironment.Shutdown();
+ var env1 = GrpcEnvironment.AddRef();
+ var env2 = GrpcEnvironment.AddRef();
+ Assert.AreSame(env1, env2);
+ GrpcEnvironment.Release();
+ GrpcEnvironment.Release();
}
[Test]
public void InitializeAfterShutdown()
{
- var env1 = GrpcEnvironment.GetInstance();
- GrpcEnvironment.Shutdown();
+ Assert.AreEqual(0, GrpcEnvironment.GetRefCount());
- var env2 = GrpcEnvironment.GetInstance();
- GrpcEnvironment.Shutdown();
+ var env1 = GrpcEnvironment.AddRef();
+ GrpcEnvironment.Release();
- Assert.IsFalse(object.ReferenceEquals(env1, env2));
+ var env2 = GrpcEnvironment.AddRef();
+ GrpcEnvironment.Release();
+
+ Assert.AreNotSame(env1, env2);
+ }
+
+ [Test]
+ public void ReleaseWithoutAddRef()
+ {
+ Assert.AreEqual(0, GrpcEnvironment.GetRefCount());
+ Assert.Throws(typeof(InvalidOperationException), () => GrpcEnvironment.Release());
}
[Test]
diff --git a/src/csharp/Grpc.Core.Tests/Internal/AsyncCallTest.cs b/src/csharp/Grpc.Core.Tests/Internal/AsyncCallTest.cs
new file mode 100644
index 0000000000..685c5f7d6c
--- /dev/null
+++ b/src/csharp/Grpc.Core.Tests/Internal/AsyncCallTest.cs
@@ -0,0 +1,222 @@
+#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.InteropServices;
+using System.Threading.Tasks;
+
+using Grpc.Core.Internal;
+using NUnit.Framework;
+
+namespace Grpc.Core.Internal.Tests
+{
+ public class AsyncCallTest
+ {
+ Channel channel;
+ FakeNativeCall fakeCall;
+ AsyncCall<string, string> asyncCall;
+
+ [SetUp]
+ public void Init()
+ {
+ channel = new Channel("localhost", Credentials.Insecure);
+
+ fakeCall = new FakeNativeCall();
+
+ var callDetails = new CallInvocationDetails<string, string>(channel, "someMethod", null, Marshallers.StringMarshaller, Marshallers.StringMarshaller, new CallOptions());
+ asyncCall = new AsyncCall<string, string>(callDetails, fakeCall);
+ }
+
+ [TearDown]
+ public void Cleanup()
+ {
+ channel.ShutdownAsync().Wait();
+ }
+
+ [Test]
+ public void AsyncUnary_CompletionSuccess()
+ {
+ var resultTask = asyncCall.UnaryCallAsync("abc");
+ fakeCall.UnaryResponseClientHandler(true, new ClientSideStatus(Status.DefaultSuccess, new Metadata()), new byte[] { 1, 2, 3 }, new Metadata());
+ Assert.IsTrue(resultTask.IsCompleted);
+ Assert.IsTrue(fakeCall.IsDisposed);
+ Assert.AreEqual(Status.DefaultSuccess, asyncCall.GetStatus());
+ }
+
+ [Test]
+ public void AsyncUnary_CompletionFailure()
+ {
+ var resultTask = asyncCall.UnaryCallAsync("abc");
+ fakeCall.UnaryResponseClientHandler(false, new ClientSideStatus(new Status(StatusCode.Internal, ""), null), new byte[] { 1, 2, 3 }, new Metadata());
+
+ Assert.IsTrue(resultTask.IsCompleted);
+ Assert.IsTrue(fakeCall.IsDisposed);
+
+ Assert.AreEqual(StatusCode.Internal, asyncCall.GetStatus().StatusCode);
+ Assert.IsNull(asyncCall.GetTrailers());
+ var ex = Assert.Throws<RpcException>(() => resultTask.GetAwaiter().GetResult());
+ Assert.AreEqual(StatusCode.Internal, ex.Status.StatusCode);
+ }
+
+ internal class FakeNativeCall : INativeCall
+ {
+ public UnaryResponseClientHandler UnaryResponseClientHandler
+ {
+ get;
+ set;
+ }
+
+ public ReceivedStatusOnClientHandler ReceivedStatusOnClientHandler
+ {
+ get;
+ set;
+ }
+
+ public ReceivedMessageHandler ReceivedMessageHandler
+ {
+ get;
+ set;
+ }
+
+ public ReceivedResponseHeadersHandler ReceivedResponseHeadersHandler
+ {
+ get;
+ set;
+ }
+
+ public SendCompletionHandler SendCompletionHandler
+ {
+ get;
+ set;
+ }
+
+ public ReceivedCloseOnServerHandler ReceivedCloseOnServerHandler
+ {
+ get;
+ set;
+ }
+
+ public bool IsCancelled
+ {
+ get;
+ set;
+ }
+
+ public bool IsDisposed
+ {
+ get;
+ set;
+ }
+
+ public void Cancel()
+ {
+ IsCancelled = true;
+ }
+
+ public void CancelWithStatus(Status status)
+ {
+ IsCancelled = true;
+ }
+
+ public string GetPeer()
+ {
+ return "PEER";
+ }
+
+ public void StartUnary(UnaryResponseClientHandler callback, byte[] payload, MetadataArraySafeHandle metadataArray, WriteFlags writeFlags)
+ {
+ UnaryResponseClientHandler = callback;
+ }
+
+ public void StartUnary(BatchContextSafeHandle ctx, byte[] payload, MetadataArraySafeHandle metadataArray, WriteFlags writeFlags)
+ {
+ throw new NotImplementedException();
+ }
+
+ public void StartClientStreaming(UnaryResponseClientHandler callback, MetadataArraySafeHandle metadataArray)
+ {
+ UnaryResponseClientHandler = callback;
+ }
+
+ public void StartServerStreaming(ReceivedStatusOnClientHandler callback, byte[] payload, MetadataArraySafeHandle metadataArray, WriteFlags writeFlags)
+ {
+ ReceivedStatusOnClientHandler = callback;
+ }
+
+ public void StartDuplexStreaming(ReceivedStatusOnClientHandler callback, MetadataArraySafeHandle metadataArray)
+ {
+ ReceivedStatusOnClientHandler = callback;
+ }
+
+ public void StartReceiveMessage(ReceivedMessageHandler callback)
+ {
+ ReceivedMessageHandler = callback;
+ }
+
+ public void StartReceiveInitialMetadata(ReceivedResponseHeadersHandler callback)
+ {
+ ReceivedResponseHeadersHandler = callback;
+ }
+
+ public void StartSendInitialMetadata(SendCompletionHandler callback, MetadataArraySafeHandle metadataArray)
+ {
+ SendCompletionHandler = callback;
+ }
+
+ public void StartSendMessage(SendCompletionHandler callback, byte[] payload, WriteFlags writeFlags, bool sendEmptyInitialMetadata)
+ {
+ SendCompletionHandler = callback;
+ }
+
+ public void StartSendCloseFromClient(SendCompletionHandler callback)
+ {
+ SendCompletionHandler = callback;
+ }
+
+ public void StartSendStatusFromServer(SendCompletionHandler callback, Status status, MetadataArraySafeHandle metadataArray, bool sendEmptyInitialMetadata)
+ {
+ SendCompletionHandler = callback;
+ }
+
+ public void StartServerSide(ReceivedCloseOnServerHandler callback)
+ {
+ ReceivedCloseOnServerHandler = callback;
+ }
+
+ public void Dispose()
+ {
+ IsDisposed = true;
+ }
+ }
+ }
+} \ No newline at end of file
diff --git a/src/csharp/Grpc.Core.Tests/MetadataTest.cs b/src/csharp/Grpc.Core.Tests/MetadataTest.cs
new file mode 100644
index 0000000000..c00f945d6a
--- /dev/null
+++ b/src/csharp/Grpc.Core.Tests/MetadataTest.cs
@@ -0,0 +1,120 @@
+#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.Runtime.InteropServices;
+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
+{
+ public class MetadataTest
+ {
+ [Test]
+ public void AsciiEntry()
+ {
+ var entry = new Metadata.Entry("ABC", "XYZ");
+ Assert.IsFalse(entry.IsBinary);
+ Assert.AreEqual("abc", entry.Key); // key is in lowercase.
+ Assert.AreEqual("XYZ", entry.Value);
+ CollectionAssert.AreEqual(new[] { (byte)'X', (byte)'Y', (byte)'Z' }, entry.ValueBytes);
+
+ Assert.Throws(typeof(ArgumentException), () => new Metadata.Entry("abc-bin", "xyz"));
+
+ Assert.AreEqual("[Entry: key=abc, value=XYZ]", entry.ToString());
+ }
+
+ [Test]
+ public void BinaryEntry()
+ {
+ var bytes = new byte[] { 1, 2, 3 };
+ var entry = new Metadata.Entry("ABC-BIN", bytes);
+ Assert.IsTrue(entry.IsBinary);
+ Assert.AreEqual("abc-bin", entry.Key); // key is in lowercase.
+ Assert.Throws(typeof(InvalidOperationException), () => { var v = entry.Value; });
+ CollectionAssert.AreEqual(bytes, entry.ValueBytes);
+
+ Assert.Throws(typeof(ArgumentException), () => new Metadata.Entry("abc", bytes));
+
+ Assert.AreEqual("[Entry: key=abc-bin, valueBytes=System.Byte[]]", entry.ToString());
+ }
+
+ [Test]
+ public void Entry_ConstructionPreconditions()
+ {
+ Assert.Throws(typeof(ArgumentNullException), () => new Metadata.Entry(null, "xyz"));
+ Assert.Throws(typeof(ArgumentNullException), () => new Metadata.Entry("abc", (string)null));
+ Assert.Throws(typeof(ArgumentNullException), () => new Metadata.Entry("abc-bin", (byte[])null));
+ }
+
+ [Test]
+ public void Entry_Immutable()
+ {
+ var origBytes = new byte[] { 1, 2, 3 };
+ var bytes = new byte[] { 1, 2, 3 };
+ var entry = new Metadata.Entry("ABC-BIN", bytes);
+ bytes[0] = 255; // changing the array passed to constructor should have any effect.
+ CollectionAssert.AreEqual(origBytes, entry.ValueBytes);
+
+ entry.ValueBytes[0] = 255;
+ CollectionAssert.AreEqual(origBytes, entry.ValueBytes);
+ }
+
+ [Test]
+ public void Entry_CreateUnsafe_Ascii()
+ {
+ var bytes = new byte[] { (byte)'X', (byte)'y' };
+ var entry = Metadata.Entry.CreateUnsafe("abc", bytes);
+ Assert.IsFalse(entry.IsBinary);
+ Assert.AreEqual("abc", entry.Key);
+ Assert.AreEqual("Xy", entry.Value);
+ CollectionAssert.AreEqual(bytes, entry.ValueBytes);
+ }
+
+ [Test]
+ public void Entry_CreateUnsafe_Binary()
+ {
+ var bytes = new byte[] { 1, 2, 3 };
+ var entry = Metadata.Entry.CreateUnsafe("abc-bin", bytes);
+ Assert.IsTrue(entry.IsBinary);
+ Assert.AreEqual("abc-bin", entry.Key);
+ Assert.Throws(typeof(InvalidOperationException), () => { var v = entry.Value; });
+ CollectionAssert.AreEqual(bytes, entry.ValueBytes);
+ }
+ }
+}
diff --git a/src/csharp/Grpc.Core.Tests/ResponseHeadersTest.cs b/src/csharp/Grpc.Core.Tests/ResponseHeadersTest.cs
index 981b8ea3c8..a1648f3671 100644
--- a/src/csharp/Grpc.Core.Tests/ResponseHeadersTest.cs
+++ b/src/csharp/Grpc.Core.Tests/ResponseHeadersTest.cs
@@ -32,13 +32,16 @@
#endregion
using System;
+using System.Collections.Generic;
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
@@ -69,14 +72,82 @@ namespace Grpc.Core.Tests
[TearDown]
public void Cleanup()
{
- channel.Dispose();
+ channel.ShutdownAsync().Wait();
server.ShutdownAsync().Wait();
}
- [TestFixtureTearDown]
- public void CleanupClass()
+ [Test]
+ public async Task ResponseHeadersAsync_UnaryCall()
+ {
+ helper.UnaryHandler = new UnaryServerMethod<string, string>(async (request, context) =>
+ {
+ await context.WriteResponseHeadersAsync(headers);
+ return "PASS";
+ });
+
+ var call = Calls.AsyncUnaryCall(helper.CreateUnaryCall(), "");
+ var responseHeaders = await call.ResponseHeadersAsync;
+
+ Assert.AreEqual(headers.Count, responseHeaders.Count);
+ Assert.AreEqual("ascii-header", responseHeaders[0].Key);
+ Assert.AreEqual("abcdefg", responseHeaders[0].Value);
+
+ Assert.AreEqual("PASS", await call.ResponseAsync);
+ }
+
+ [Test]
+ public async Task ResponseHeadersAsync_ClientStreamingCall()
+ {
+ helper.ClientStreamingHandler = new ClientStreamingServerMethod<string, string>(async (requestStream, context) =>
+ {
+ await context.WriteResponseHeadersAsync(headers);
+ return "PASS";
+ });
+
+ var call = Calls.AsyncClientStreamingCall(helper.CreateClientStreamingCall());
+ await call.RequestStream.CompleteAsync();
+ var responseHeaders = await call.ResponseHeadersAsync;
+
+ Assert.AreEqual("ascii-header", responseHeaders[0].Key);
+ Assert.AreEqual("PASS", await call.ResponseAsync);
+ }
+
+ [Test]
+ public async Task ResponseHeadersAsync_ServerStreamingCall()
+ {
+ helper.ServerStreamingHandler = new ServerStreamingServerMethod<string, string>(async (request, responseStream, context) =>
+ {
+ await context.WriteResponseHeadersAsync(headers);
+ await responseStream.WriteAsync("PASS");
+ });
+
+ var call = Calls.AsyncServerStreamingCall(helper.CreateServerStreamingCall(), "");
+ var responseHeaders = await call.ResponseHeadersAsync;
+
+ Assert.AreEqual("ascii-header", responseHeaders[0].Key);
+ CollectionAssert.AreEqual(new[] { "PASS" }, await call.ResponseStream.ToListAsync());
+ }
+
+ [Test]
+ public async Task ResponseHeadersAsync_DuplexStreamingCall()
{
- GrpcEnvironment.Shutdown();
+ helper.DuplexStreamingHandler = new DuplexStreamingServerMethod<string, string>(async (requestStream, responseStream, context) =>
+ {
+ await context.WriteResponseHeadersAsync(headers);
+ while (await requestStream.MoveNext())
+ {
+ await responseStream.WriteAsync(requestStream.Current);
+ }
+ });
+
+ var call = Calls.AsyncDuplexStreamingCall(helper.CreateDuplexStreamingCall());
+ var responseHeaders = await call.ResponseHeadersAsync;
+
+ var messages = new[] { "PASS" };
+ await call.RequestStream.WriteAllAsync(messages);
+
+ Assert.AreEqual("ascii-header", responseHeaders[0].Key);
+ CollectionAssert.AreEqual(messages, await call.ResponseStream.ToListAsync());
}
[Test]
diff --git a/src/csharp/Grpc.Core.Tests/ServerTest.cs b/src/csharp/Grpc.Core.Tests/ServerTest.cs
index 485006ebac..e7193c843b 100644
--- a/src/csharp/Grpc.Core.Tests/ServerTest.cs
+++ b/src/csharp/Grpc.Core.Tests/ServerTest.cs
@@ -51,7 +51,6 @@ namespace Grpc.Core.Tests
};
server.Start();
server.ShutdownAsync().Wait();
- GrpcEnvironment.Shutdown();
}
[Test]
@@ -67,8 +66,7 @@ namespace Grpc.Core.Tests
Assert.Greater(boundPort.BoundPort, 0);
server.Start();
- server.ShutdownAsync();
- GrpcEnvironment.Shutdown();
+ server.ShutdownAsync().Wait();
}
[Test]
@@ -83,7 +81,6 @@ namespace Grpc.Core.Tests
Assert.Throws(typeof(InvalidOperationException), () => server.Services.Add(ServerServiceDefinition.CreateBuilder("serviceName").Build()));
server.ShutdownAsync().Wait();
- GrpcEnvironment.Shutdown();
}
}
}
diff --git a/src/csharp/Grpc.Core.Tests/ShutdownTest.cs b/src/csharp/Grpc.Core.Tests/ShutdownTest.cs
new file mode 100644
index 0000000000..a2be7ddd5e
--- /dev/null
+++ b/src/csharp/Grpc.Core.Tests/ShutdownTest.cs
@@ -0,0 +1,77 @@
+#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
+{
+ public class ShutdownTest
+ {
+ const string Host = "127.0.0.1";
+
+ MockServiceHelper helper;
+ Server server;
+ Channel channel;
+
+ [SetUp]
+ public void Init()
+ {
+ helper = new MockServiceHelper(Host);
+ server = helper.GetServer();
+ server.Start();
+ channel = helper.GetChannel();
+ }
+
+ [Test]
+ public async Task AbandonedCall()
+ {
+ helper.DuplexStreamingHandler = new DuplexStreamingServerMethod<string, string>(async (requestStream, responseStream, context) =>
+ {
+ await requestStream.ToListAsync();
+ });
+
+ var call = Calls.AsyncDuplexStreamingCall(helper.CreateDuplexStreamingCall(new CallOptions(deadline: DateTime.UtcNow.AddMilliseconds(1))));
+
+ channel.ShutdownAsync().Wait();
+ server.ShutdownAsync().Wait();
+ }
+ }
+}
diff --git a/src/csharp/Grpc.Core.Tests/TimeoutsTest.cs b/src/csharp/Grpc.Core.Tests/TimeoutsTest.cs
index d875d601b9..41f661f62d 100644
--- a/src/csharp/Grpc.Core.Tests/TimeoutsTest.cs
+++ b/src/csharp/Grpc.Core.Tests/TimeoutsTest.cs
@@ -65,16 +65,10 @@ namespace Grpc.Core.Tests
[TearDown]
public void Cleanup()
{
- channel.Dispose();
+ channel.ShutdownAsync().Wait();
server.ShutdownAsync().Wait();
}
- [TestFixtureTearDown]
- public void CleanupClass()
- {
- GrpcEnvironment.Shutdown();
- }
-
[Test]
public void InfiniteDeadline()
{
diff --git a/src/csharp/Grpc.Core/AsyncClientStreamingCall.cs b/src/csharp/Grpc.Core/AsyncClientStreamingCall.cs
index fb9b562c77..dbaa3085c5 100644
--- a/src/csharp/Grpc.Core/AsyncClientStreamingCall.cs
+++ b/src/csharp/Grpc.Core/AsyncClientStreamingCall.cs
@@ -44,14 +44,16 @@ namespace Grpc.Core
{
readonly IClientStreamWriter<TRequest> requestStream;
readonly Task<TResponse> responseAsync;
+ readonly Task<Metadata> responseHeadersAsync;
readonly Func<Status> getStatusFunc;
readonly Func<Metadata> getTrailersFunc;
readonly Action disposeAction;
- public AsyncClientStreamingCall(IClientStreamWriter<TRequest> requestStream, Task<TResponse> responseAsync, Func<Status> getStatusFunc, Func<Metadata> getTrailersFunc, Action disposeAction)
+ public AsyncClientStreamingCall(IClientStreamWriter<TRequest> requestStream, Task<TResponse> responseAsync, Task<Metadata> responseHeadersAsync, Func<Status> getStatusFunc, Func<Metadata> getTrailersFunc, Action disposeAction)
{
this.requestStream = requestStream;
this.responseAsync = responseAsync;
+ this.responseHeadersAsync = responseHeadersAsync;
this.getStatusFunc = getStatusFunc;
this.getTrailersFunc = getTrailersFunc;
this.disposeAction = disposeAction;
@@ -69,6 +71,17 @@ namespace Grpc.Core
}
/// <summary>
+ /// Asynchronous access to response headers.
+ /// </summary>
+ public Task<Metadata> ResponseHeadersAsync
+ {
+ get
+ {
+ return this.responseHeadersAsync;
+ }
+ }
+
+ /// <summary>
/// Async stream to send streaming requests.
/// </summary>
public IClientStreamWriter<TRequest> RequestStream
diff --git a/src/csharp/Grpc.Core/AsyncDuplexStreamingCall.cs b/src/csharp/Grpc.Core/AsyncDuplexStreamingCall.cs
index 183c84216a..ee7ba29695 100644
--- a/src/csharp/Grpc.Core/AsyncDuplexStreamingCall.cs
+++ b/src/csharp/Grpc.Core/AsyncDuplexStreamingCall.cs
@@ -32,6 +32,7 @@
#endregion
using System;
+using System.Threading.Tasks;
namespace Grpc.Core
{
@@ -42,14 +43,16 @@ namespace Grpc.Core
{
readonly IClientStreamWriter<TRequest> requestStream;
readonly IAsyncStreamReader<TResponse> responseStream;
+ readonly Task<Metadata> responseHeadersAsync;
readonly Func<Status> getStatusFunc;
readonly Func<Metadata> getTrailersFunc;
readonly Action disposeAction;
- public AsyncDuplexStreamingCall(IClientStreamWriter<TRequest> requestStream, IAsyncStreamReader<TResponse> responseStream, Func<Status> getStatusFunc, Func<Metadata> getTrailersFunc, Action disposeAction)
+ public AsyncDuplexStreamingCall(IClientStreamWriter<TRequest> requestStream, IAsyncStreamReader<TResponse> responseStream, Task<Metadata> responseHeadersAsync, Func<Status> getStatusFunc, Func<Metadata> getTrailersFunc, Action disposeAction)
{
this.requestStream = requestStream;
this.responseStream = responseStream;
+ this.responseHeadersAsync = responseHeadersAsync;
this.getStatusFunc = getStatusFunc;
this.getTrailersFunc = getTrailersFunc;
this.disposeAction = disposeAction;
@@ -78,6 +81,17 @@ namespace Grpc.Core
}
/// <summary>
+ /// Asynchronous access to response headers.
+ /// </summary>
+ public Task<Metadata> ResponseHeadersAsync
+ {
+ get
+ {
+ return this.responseHeadersAsync;
+ }
+ }
+
+ /// <summary>
/// Gets the call status if the call has already finished.
/// Throws InvalidOperationException otherwise.
/// </summary>
diff --git a/src/csharp/Grpc.Core/AsyncServerStreamingCall.cs b/src/csharp/Grpc.Core/AsyncServerStreamingCall.cs
index ab2049f269..2853a79ce6 100644
--- a/src/csharp/Grpc.Core/AsyncServerStreamingCall.cs
+++ b/src/csharp/Grpc.Core/AsyncServerStreamingCall.cs
@@ -32,6 +32,7 @@
#endregion
using System;
+using System.Threading.Tasks;
namespace Grpc.Core
{
@@ -41,13 +42,15 @@ namespace Grpc.Core
public sealed class AsyncServerStreamingCall<TResponse> : IDisposable
{
readonly IAsyncStreamReader<TResponse> responseStream;
+ readonly Task<Metadata> responseHeadersAsync;
readonly Func<Status> getStatusFunc;
readonly Func<Metadata> getTrailersFunc;
readonly Action disposeAction;
- public AsyncServerStreamingCall(IAsyncStreamReader<TResponse> responseStream, Func<Status> getStatusFunc, Func<Metadata> getTrailersFunc, Action disposeAction)
+ public AsyncServerStreamingCall(IAsyncStreamReader<TResponse> responseStream, Task<Metadata> responseHeadersAsync, Func<Status> getStatusFunc, Func<Metadata> getTrailersFunc, Action disposeAction)
{
this.responseStream = responseStream;
+ this.responseHeadersAsync = responseHeadersAsync;
this.getStatusFunc = getStatusFunc;
this.getTrailersFunc = getTrailersFunc;
this.disposeAction = disposeAction;
@@ -65,6 +68,17 @@ namespace Grpc.Core
}
/// <summary>
+ /// Asynchronous access to response headers.
+ /// </summary>
+ public Task<Metadata> ResponseHeadersAsync
+ {
+ get
+ {
+ return this.responseHeadersAsync;
+ }
+ }
+
+ /// <summary>
/// Gets the call status if the call has already finished.
/// Throws InvalidOperationException otherwise.
/// </summary>
diff --git a/src/csharp/Grpc.Core/AsyncUnaryCall.cs b/src/csharp/Grpc.Core/AsyncUnaryCall.cs
index 224e343916..154a17a33e 100644
--- a/src/csharp/Grpc.Core/AsyncUnaryCall.cs
+++ b/src/csharp/Grpc.Core/AsyncUnaryCall.cs
@@ -43,13 +43,15 @@ namespace Grpc.Core
public sealed class AsyncUnaryCall<TResponse> : IDisposable
{
readonly Task<TResponse> responseAsync;
+ readonly Task<Metadata> responseHeadersAsync;
readonly Func<Status> getStatusFunc;
readonly Func<Metadata> getTrailersFunc;
readonly Action disposeAction;
- public AsyncUnaryCall(Task<TResponse> responseAsync, Func<Status> getStatusFunc, Func<Metadata> getTrailersFunc, Action disposeAction)
+ public AsyncUnaryCall(Task<TResponse> responseAsync, Task<Metadata> responseHeadersAsync, Func<Status> getStatusFunc, Func<Metadata> getTrailersFunc, Action disposeAction)
{
this.responseAsync = responseAsync;
+ this.responseHeadersAsync = responseHeadersAsync;
this.getStatusFunc = getStatusFunc;
this.getTrailersFunc = getTrailersFunc;
this.disposeAction = disposeAction;
@@ -67,6 +69,17 @@ namespace Grpc.Core
}
/// <summary>
+ /// Asynchronous access to response headers.
+ /// </summary>
+ public Task<Metadata> ResponseHeadersAsync
+ {
+ get
+ {
+ return this.responseHeadersAsync;
+ }
+ }
+
+ /// <summary>
/// Allows awaiting this object directly.
/// </summary>
public TaskAwaiter<TResponse> GetAwaiter()
diff --git a/src/csharp/Grpc.Core/Calls.cs b/src/csharp/Grpc.Core/Calls.cs
index 7067456638..e57ac89db3 100644
--- a/src/csharp/Grpc.Core/Calls.cs
+++ b/src/csharp/Grpc.Core/Calls.cs
@@ -74,7 +74,7 @@ namespace Grpc.Core
{
var asyncCall = new AsyncCall<TRequest, TResponse>(call);
var asyncResult = asyncCall.UnaryCallAsync(req);
- return new AsyncUnaryCall<TResponse>(asyncResult, asyncCall.GetStatus, asyncCall.GetTrailers, asyncCall.Cancel);
+ return new AsyncUnaryCall<TResponse>(asyncResult, asyncCall.ResponseHeadersAsync, asyncCall.GetStatus, asyncCall.GetTrailers, asyncCall.Cancel);
}
/// <summary>
@@ -93,7 +93,7 @@ namespace Grpc.Core
var asyncCall = new AsyncCall<TRequest, TResponse>(call);
asyncCall.StartServerStreamingCall(req);
var responseStream = new ClientResponseStream<TRequest, TResponse>(asyncCall);
- return new AsyncServerStreamingCall<TResponse>(responseStream, asyncCall.GetStatus, asyncCall.GetTrailers, asyncCall.Cancel);
+ return new AsyncServerStreamingCall<TResponse>(responseStream, asyncCall.ResponseHeadersAsync, asyncCall.GetStatus, asyncCall.GetTrailers, asyncCall.Cancel);
}
/// <summary>
@@ -110,7 +110,7 @@ namespace Grpc.Core
var asyncCall = new AsyncCall<TRequest, TResponse>(call);
var resultTask = asyncCall.ClientStreamingCallAsync();
var requestStream = new ClientRequestStream<TRequest, TResponse>(asyncCall);
- return new AsyncClientStreamingCall<TRequest, TResponse>(requestStream, resultTask, asyncCall.GetStatus, asyncCall.GetTrailers, asyncCall.Cancel);
+ return new AsyncClientStreamingCall<TRequest, TResponse>(requestStream, resultTask, asyncCall.ResponseHeadersAsync, asyncCall.GetStatus, asyncCall.GetTrailers, asyncCall.Cancel);
}
/// <summary>
@@ -130,7 +130,7 @@ namespace Grpc.Core
asyncCall.StartDuplexStreamingCall();
var requestStream = new ClientRequestStream<TRequest, TResponse>(asyncCall);
var responseStream = new ClientResponseStream<TRequest, TResponse>(asyncCall);
- return new AsyncDuplexStreamingCall<TRequest, TResponse>(requestStream, responseStream, asyncCall.GetStatus, asyncCall.GetTrailers, asyncCall.Cancel);
+ return new AsyncDuplexStreamingCall<TRequest, TResponse>(requestStream, responseStream, asyncCall.ResponseHeadersAsync, asyncCall.GetStatus, asyncCall.GetTrailers, asyncCall.Cancel);
}
}
}
diff --git a/src/csharp/Grpc.Core/Channel.cs b/src/csharp/Grpc.Core/Channel.cs
index 64c6adf2bf..c11b320a64 100644
--- a/src/csharp/Grpc.Core/Channel.cs
+++ b/src/csharp/Grpc.Core/Channel.cs
@@ -45,15 +45,19 @@ namespace Grpc.Core
/// <summary>
/// gRPC Channel
/// </summary>
- public class Channel : IDisposable
+ public class Channel
{
static readonly ILogger Logger = GrpcEnvironment.Logger.ForType<Channel>();
+ readonly object myLock = new object();
+ readonly AtomicCounter activeCallCounter = new AtomicCounter();
+
readonly string target;
readonly GrpcEnvironment environment;
readonly ChannelSafeHandle handle;
readonly List<ChannelOption> options;
- bool disposed;
+
+ bool shutdownRequested;
/// <summary>
/// Creates a channel that connects to a specific host.
@@ -65,7 +69,7 @@ namespace Grpc.Core
public Channel(string target, Credentials credentials, IEnumerable<ChannelOption> options = null)
{
this.target = Preconditions.CheckNotNull(target, "target");
- this.environment = GrpcEnvironment.GetInstance();
+ this.environment = GrpcEnvironment.AddRef();
this.options = options != null ? new List<ChannelOption>(options) : new List<ChannelOption>();
EnsureUserAgentChannelOption(this.options);
@@ -172,12 +176,26 @@ namespace Grpc.Core
}
/// <summary>
- /// Destroys the underlying channel.
+ /// Waits until there are no more active calls for this channel and then cleans up
+ /// resources used by this channel.
/// </summary>
- public void Dispose()
+ public async Task ShutdownAsync()
{
- Dispose(true);
- GC.SuppressFinalize(this);
+ lock (myLock)
+ {
+ Preconditions.CheckState(!shutdownRequested);
+ shutdownRequested = true;
+ }
+
+ var activeCallCount = activeCallCounter.Count;
+ if (activeCallCount > 0)
+ {
+ Logger.Warning("Channel shutdown was called but there are still {0} active calls for that channel.", activeCallCount);
+ }
+
+ handle.Dispose();
+
+ await Task.Run(() => GrpcEnvironment.Release());
}
internal ChannelSafeHandle Handle
@@ -196,13 +214,20 @@ namespace Grpc.Core
}
}
- protected virtual void Dispose(bool disposing)
+ internal void AddCallReference(object call)
{
- if (disposing && handle != null && !disposed)
- {
- disposed = true;
- handle.Dispose();
- }
+ activeCallCounter.Increment();
+
+ bool success = false;
+ handle.DangerousAddRef(ref success);
+ Preconditions.CheckState(success);
+ }
+
+ internal void RemoveCallReference(object call)
+ {
+ handle.DangerousRelease();
+
+ activeCallCounter.Decrement();
}
private static void EnsureUserAgentChannelOption(List<ChannelOption> options)
diff --git a/src/csharp/Grpc.Core/ClientBase.cs b/src/csharp/Grpc.Core/ClientBase.cs
index 7bc100ca60..903449439b 100644
--- a/src/csharp/Grpc.Core/ClientBase.cs
+++ b/src/csharp/Grpc.Core/ClientBase.cs
@@ -119,7 +119,8 @@ namespace Grpc.Core
internal static string GetAuthUriBase(string target)
{
var match = ChannelTargetPattern.Match(target);
- if (!match.Success) {
+ if (!match.Success)
+ {
return null;
}
return "https://" + match.Groups[2].Value + "/";
diff --git a/src/csharp/Grpc.Core/ContextPropagationToken.cs b/src/csharp/Grpc.Core/ContextPropagationToken.cs
index 2e4bfc9e47..a5bf1b5a70 100644
--- a/src/csharp/Grpc.Core/ContextPropagationToken.cs
+++ b/src/csharp/Grpc.Core/ContextPropagationToken.cs
@@ -132,7 +132,6 @@ namespace Grpc.Core
bool propagateDeadline;
bool propagateCancellation;
-
/// <summary>
/// Creates new context propagation options.
/// </summary>
diff --git a/src/csharp/Grpc.Core/Grpc.Core.csproj b/src/csharp/Grpc.Core/Grpc.Core.csproj
index 055aff1444..ad2af17bc7 100644
--- a/src/csharp/Grpc.Core/Grpc.Core.csproj
+++ b/src/csharp/Grpc.Core/Grpc.Core.csproj
@@ -49,6 +49,7 @@
<Compile Include="AsyncDuplexStreamingCall.cs" />
<Compile Include="AsyncServerStreamingCall.cs" />
<Compile Include="IClientStreamWriter.cs" />
+ <Compile Include="Internal\INativeCall.cs" />
<Compile Include="IServerStreamWriter.cs" />
<Compile Include="IAsyncStreamWriter.cs" />
<Compile Include="IAsyncStreamReader.cs" />
diff --git a/src/csharp/Grpc.Core/GrpcEnvironment.cs b/src/csharp/Grpc.Core/GrpcEnvironment.cs
index 30d8c80235..e7c04185c2 100644
--- a/src/csharp/Grpc.Core/GrpcEnvironment.cs
+++ b/src/csharp/Grpc.Core/GrpcEnvironment.cs
@@ -58,6 +58,7 @@ namespace Grpc.Core
static object staticLock = new object();
static GrpcEnvironment instance;
+ static int refCount;
static ILogger logger = new ConsoleLogger();
@@ -67,13 +68,14 @@ namespace Grpc.Core
bool isClosed;
/// <summary>
- /// Returns an instance of initialized gRPC environment.
- /// Subsequent invocations return the same instance unless Shutdown has been called first.
+ /// Returns a reference-counted instance of initialized gRPC environment.
+ /// Subsequent invocations return the same instance unless reference count has dropped to zero previously.
/// </summary>
- internal static GrpcEnvironment GetInstance()
+ internal static GrpcEnvironment AddRef()
{
lock (staticLock)
{
+ refCount++;
if (instance == null)
{
instance = new GrpcEnvironment();
@@ -83,14 +85,16 @@ namespace Grpc.Core
}
/// <summary>
- /// Shuts down the gRPC environment if it was initialized before.
- /// Blocks until the environment has been fully shutdown.
+ /// Decrements the reference count for currently active environment and shuts down the gRPC environment if reference count drops to zero.
+ /// (and blocks until the environment has been fully shutdown).
/// </summary>
- public static void Shutdown()
+ internal static void Release()
{
lock (staticLock)
{
- if (instance != null)
+ Preconditions.CheckState(refCount > 0);
+ refCount--;
+ if (refCount == 0)
{
instance.Close();
instance = null;
@@ -98,6 +102,14 @@ namespace Grpc.Core
}
}
+ internal static int GetRefCount()
+ {
+ lock (staticLock)
+ {
+ return refCount;
+ }
+ }
+
/// <summary>
/// Gets application-wide logger used by gRPC.
/// </summary>
@@ -125,12 +137,10 @@ namespace Grpc.Core
private GrpcEnvironment()
{
NativeLogRedirector.Redirect();
- grpcsharp_init();
+ GrpcNativeInit();
completionRegistry = new CompletionRegistry(this);
threadPool = new GrpcThreadPool(this, THREAD_POOL_SIZE);
threadPool.Start();
- // TODO: use proper logging here
- Logger.Info("gRPC initialized.");
}
/// <summary>
@@ -175,6 +185,16 @@ namespace Grpc.Core
return Marshal.PtrToStringAnsi(ptr);
}
+ internal static void GrpcNativeInit()
+ {
+ grpcsharp_init();
+ }
+
+ internal static void GrpcNativeShutdown()
+ {
+ grpcsharp_shutdown();
+ }
+
/// <summary>
/// Shuts down this environment.
/// </summary>
@@ -185,12 +205,10 @@ namespace Grpc.Core
throw new InvalidOperationException("Close has already been called");
}
threadPool.Stop();
- grpcsharp_shutdown();
+ GrpcNativeShutdown();
isClosed = true;
debugStats.CheckOK();
-
- Logger.Info("gRPC shutdown.");
}
}
}
diff --git a/src/csharp/Grpc.Core/Internal/AsyncCall.cs b/src/csharp/Grpc.Core/Internal/AsyncCall.cs
index 2c3e3d75ea..be5d611a53 100644
--- a/src/csharp/Grpc.Core/Internal/AsyncCall.cs
+++ b/src/csharp/Grpc.Core/Internal/AsyncCall.cs
@@ -51,22 +51,35 @@ namespace Grpc.Core.Internal
static readonly ILogger Logger = GrpcEnvironment.Logger.ForType<AsyncCall<TRequest, TResponse>>();
readonly CallInvocationDetails<TRequest, TResponse> details;
+ readonly INativeCall injectedNativeCall; // for testing
// Completion of a pending unary response if not null.
TaskCompletionSource<TResponse> unaryResponseTcs;
+ // Indicates that steaming call has finished.
+ TaskCompletionSource<object> streamingCallFinishedTcs = new TaskCompletionSource<object>();
+
+ // Response headers set here once received.
+ TaskCompletionSource<Metadata> responseHeadersTcs = new TaskCompletionSource<Metadata>();
+
// Set after status is received. Used for both unary and streaming response calls.
ClientSideStatus? finishedStatus;
- bool readObserverCompleted; // True if readObserver has already been completed.
-
public AsyncCall(CallInvocationDetails<TRequest, TResponse> callDetails)
- : base(callDetails.RequestMarshaller.Serializer, callDetails.ResponseMarshaller.Deserializer)
+ : base(callDetails.RequestMarshaller.Serializer, callDetails.ResponseMarshaller.Deserializer, callDetails.Channel.Environment)
{
this.details = callDetails.WithOptions(callDetails.Options.Normalize());
this.initialMetadataSent = true; // we always send metadata at the very beginning of the call.
}
+ /// <summary>
+ /// This constructor should only be used for testing.
+ /// </summary>
+ public AsyncCall(CallInvocationDetails<TRequest, TResponse> callDetails, INativeCall injectedNativeCall) : this(callDetails)
+ {
+ this.injectedNativeCall = injectedNativeCall;
+ }
+
// TODO: this method is not Async, so it shouldn't be in AsyncCall class, but
// it is reusing fair amount of code in this class, so we are leaving it here.
/// <summary>
@@ -100,7 +113,7 @@ namespace Grpc.Core.Internal
bool success = (ev.success != 0);
try
{
- HandleUnaryResponse(success, ctx);
+ HandleUnaryResponse(success, ctx.GetReceivedStatusOnClient(), ctx.GetReceivedMessage(), ctx.GetReceivedInitialMetadata());
}
catch (Exception e)
{
@@ -125,7 +138,7 @@ namespace Grpc.Core.Internal
Preconditions.CheckState(!started);
started = true;
- Initialize(details.Channel.Environment.CompletionQueue);
+ Initialize(environment.CompletionQueue);
halfcloseRequested = true;
readingDone = true;
@@ -152,7 +165,7 @@ namespace Grpc.Core.Internal
Preconditions.CheckState(!started);
started = true;
- Initialize(details.Channel.Environment.CompletionQueue);
+ Initialize(environment.CompletionQueue);
readingDone = true;
@@ -176,10 +189,9 @@ namespace Grpc.Core.Internal
Preconditions.CheckState(!started);
started = true;
- Initialize(details.Channel.Environment.CompletionQueue);
+ Initialize(environment.CompletionQueue);
halfcloseRequested = true;
- halfclosed = true; // halfclose not confirmed yet, but it will be once finishedHandler is called.
byte[] payload = UnsafeSerialize(msg);
@@ -187,6 +199,7 @@ namespace Grpc.Core.Internal
{
call.StartServerStreaming(HandleFinished, payload, metadataArray, GetWriteFlagsForCall());
}
+ call.StartReceiveInitialMetadata(HandleReceivedResponseHeaders);
}
}
@@ -201,12 +214,13 @@ namespace Grpc.Core.Internal
Preconditions.CheckState(!started);
started = true;
- Initialize(details.Channel.Environment.CompletionQueue);
+ Initialize(environment.CompletionQueue);
using (var metadataArray = MetadataArraySafeHandle.Create(details.Options.Headers))
{
call.StartDuplexStreaming(HandleFinished, metadataArray);
}
+ call.StartReceiveInitialMetadata(HandleReceivedResponseHeaders);
}
}
@@ -248,6 +262,28 @@ namespace Grpc.Core.Internal
}
/// <summary>
+ /// Get the task that completes once if streaming call finishes with ok status and throws RpcException with given status otherwise.
+ /// </summary>
+ public Task StreamingCallFinishedTask
+ {
+ get
+ {
+ return streamingCallFinishedTcs.Task;
+ }
+ }
+
+ /// <summary>
+ /// Get the task that completes once response headers are received.
+ /// </summary>
+ public Task<Metadata> ResponseHeadersAsync
+ {
+ get
+ {
+ return responseHeadersTcs.Task;
+ }
+ }
+
+ /// <summary>
/// Gets the resulting status if the call has already finished.
/// Throws InvalidOperationException otherwise.
/// </summary>
@@ -281,51 +317,31 @@ namespace Grpc.Core.Internal
}
}
- /// <summary>
- /// On client-side, we only fire readCompletionDelegate once all messages have been read
- /// and status has been received.
- /// </summary>
- protected override void ProcessLastRead(AsyncCompletionDelegate<TResponse> completionDelegate)
+ protected override void OnAfterReleaseResources()
{
- if (completionDelegate != null && readingDone && finishedStatus.HasValue)
- {
- bool shouldComplete;
- lock (myLock)
- {
- shouldComplete = !readObserverCompleted;
- readObserverCompleted = true;
- }
-
- if (shouldComplete)
- {
- var status = finishedStatus.Value.Status;
- if (status.StatusCode != StatusCode.OK)
- {
- FireCompletion(completionDelegate, default(TResponse), new RpcException(status));
- }
- else
- {
- FireCompletion(completionDelegate, default(TResponse), null);
- }
- }
- }
+ details.Channel.RemoveCallReference(this);
}
- protected override void OnReleaseResources()
+ private void Initialize(CompletionQueueSafeHandle cq)
{
- details.Channel.Environment.DebugStats.ActiveClientCalls.Decrement();
+ var call = CreateNativeCall(cq);
+ details.Channel.AddCallReference(this);
+ InitializeInternal(call);
+ RegisterCancellationCallback();
}
- private void Initialize(CompletionQueueSafeHandle cq)
+ private INativeCall CreateNativeCall(CompletionQueueSafeHandle cq)
{
+ if (injectedNativeCall != null)
+ {
+ return injectedNativeCall; // allows injecting a mock INativeCall in tests.
+ }
+
var parentCall = details.Options.PropagationToken != null ? details.Options.PropagationToken.ParentCall : CallSafeHandle.NullInstance;
- var call = details.Channel.Handle.CreateCall(details.Channel.Environment.CompletionRegistry,
+ return details.Channel.Handle.CreateCall(environment.CompletionRegistry,
parentCall, ContextPropagationToken.DefaultMask, cq,
details.Method, details.Host, Timespec.FromDateTime(details.Options.Deadline.Value));
- details.Channel.Environment.DebugStats.ActiveClientCalls.Increment();
- InitializeInternal(call);
- RegisterCancellationCallback();
}
// Make sure that once cancellationToken for this call is cancelled, Cancel() will be called.
@@ -348,31 +364,31 @@ namespace Grpc.Core.Internal
}
/// <summary>
- /// Handler for unary response completion.
+ /// Handles receive status completion for calls with streaming response.
/// </summary>
- private void HandleUnaryResponse(bool success, BatchContextSafeHandle ctx)
+ private void HandleReceivedResponseHeaders(bool success, Metadata responseHeaders)
{
- var fullStatus = ctx.GetReceivedStatusOnClient();
+ responseHeadersTcs.SetResult(responseHeaders);
+ }
+ /// <summary>
+ /// Handler for unary response completion.
+ /// </summary>
+ private void HandleUnaryResponse(bool success, ClientSideStatus receivedStatus, byte[] receivedMessage, Metadata responseHeaders)
+ {
lock (myLock)
{
finished = true;
- finishedStatus = fullStatus;
-
- halfclosed = true;
+ finishedStatus = receivedStatus;
ReleaseResourcesIfPossible();
}
- if (!success)
- {
- unaryResponseTcs.SetException(new RpcException(new Status(StatusCode.Internal, "Internal error occured.")));
- return;
- }
+ responseHeadersTcs.SetResult(responseHeaders);
- var status = fullStatus.Status;
+ var status = receivedStatus.Status;
- if (status.StatusCode != StatusCode.OK)
+ if (!success || status.StatusCode != StatusCode.OK)
{
unaryResponseTcs.SetException(new RpcException(status));
return;
@@ -380,7 +396,7 @@ namespace Grpc.Core.Internal
// TODO: handle deserialization error
TResponse msg;
- TryDeserialize(ctx.GetReceivedMessage(), out msg);
+ TryDeserialize(receivedMessage, out msg);
unaryResponseTcs.SetResult(msg);
}
@@ -388,22 +404,25 @@ namespace Grpc.Core.Internal
/// <summary>
/// Handles receive status completion for calls with streaming response.
/// </summary>
- private void HandleFinished(bool success, BatchContextSafeHandle ctx)
+ private void HandleFinished(bool success, ClientSideStatus receivedStatus)
{
- var fullStatus = ctx.GetReceivedStatusOnClient();
-
- AsyncCompletionDelegate<TResponse> origReadCompletionDelegate = null;
lock (myLock)
{
finished = true;
- finishedStatus = fullStatus;
-
- origReadCompletionDelegate = readCompletionDelegate;
+ finishedStatus = receivedStatus;
ReleaseResourcesIfPossible();
}
- ProcessLastRead(origReadCompletionDelegate);
+ var status = receivedStatus.Status;
+
+ if (!success || status.StatusCode != StatusCode.OK)
+ {
+ streamingCallFinishedTcs.SetException(new RpcException(status));
+ return;
+ }
+
+ streamingCallFinishedTcs.SetResult(null);
}
}
} \ No newline at end of file
diff --git a/src/csharp/Grpc.Core/Internal/AsyncCallBase.cs b/src/csharp/Grpc.Core/Internal/AsyncCallBase.cs
index 6ca4bbdafc..4d20394644 100644
--- a/src/csharp/Grpc.Core/Internal/AsyncCallBase.cs
+++ b/src/csharp/Grpc.Core/Internal/AsyncCallBase.cs
@@ -54,30 +54,30 @@ namespace Grpc.Core.Internal
readonly Func<TWrite, byte[]> serializer;
readonly Func<byte[], TRead> deserializer;
+ protected readonly GrpcEnvironment environment;
protected readonly object myLock = new object();
- protected CallSafeHandle call;
+ protected INativeCall call;
protected bool disposed;
protected bool started;
- protected bool errorOccured;
protected bool cancelRequested;
protected AsyncCompletionDelegate<object> sendCompletionDelegate; // Completion of a pending send or sendclose if not null.
protected AsyncCompletionDelegate<TRead> readCompletionDelegate; // Completion of a pending send or sendclose if not null.
- protected bool readingDone;
- protected bool halfcloseRequested;
- protected bool halfclosed;
+ protected bool readingDone; // True if last read (i.e. read with null payload) was already received.
+ protected bool halfcloseRequested; // True if send close have been initiated.
protected bool finished; // True if close has been received from the peer.
protected bool initialMetadataSent;
- protected long streamingWritesCounter;
+ protected long streamingWritesCounter; // Number of streaming send operations started so far.
- public AsyncCallBase(Func<TWrite, byte[]> serializer, Func<byte[], TRead> deserializer)
+ public AsyncCallBase(Func<TWrite, byte[]> serializer, Func<byte[], TRead> deserializer, GrpcEnvironment environment)
{
this.serializer = Preconditions.CheckNotNull(serializer);
this.deserializer = Preconditions.CheckNotNull(deserializer);
+ this.environment = Preconditions.CheckNotNull(environment);
}
/// <summary>
@@ -114,7 +114,7 @@ namespace Grpc.Core.Internal
}
}
- protected void InitializeInternal(CallSafeHandle call)
+ protected void InitializeInternal(INativeCall call)
{
lock (myLock)
{
@@ -159,16 +159,6 @@ namespace Grpc.Core.Internal
}
}
- // TODO(jtattermusch): find more fitting name for this method.
- /// <summary>
- /// Default behavior just completes the read observer, but more sofisticated behavior might be required
- /// by subclasses.
- /// </summary>
- protected virtual void ProcessLastRead(AsyncCompletionDelegate<TRead> completionDelegate)
- {
- FireCompletion(completionDelegate, default(TRead), null);
- }
-
/// <summary>
/// If there are no more pending actions and no new actions can be started, releases
/// the underlying native resources.
@@ -177,7 +167,7 @@ namespace Grpc.Core.Internal
{
if (!disposed && call != null)
{
- bool noMoreSendCompletions = halfclosed || (cancelRequested && sendCompletionDelegate == null);
+ bool noMoreSendCompletions = sendCompletionDelegate == null && (halfcloseRequested || cancelRequested || finished);
if (noMoreSendCompletions && readingDone && finished)
{
ReleaseResources();
@@ -189,34 +179,33 @@ namespace Grpc.Core.Internal
private void ReleaseResources()
{
- OnReleaseResources();
if (call != null)
{
call.Dispose();
}
disposed = true;
+ OnAfterReleaseResources();
}
- protected virtual void OnReleaseResources()
+ protected virtual void OnAfterReleaseResources()
{
}
protected void CheckSendingAllowed()
{
Preconditions.CheckState(started);
- Preconditions.CheckState(!errorOccured);
CheckNotCancelled();
Preconditions.CheckState(!disposed);
Preconditions.CheckState(!halfcloseRequested, "Already halfclosed.");
+ Preconditions.CheckState(!finished, "Already finished.");
Preconditions.CheckState(sendCompletionDelegate == null, "Only one write can be pending at a time");
}
- protected void CheckReadingAllowed()
+ protected virtual void CheckReadingAllowed()
{
Preconditions.CheckState(started);
Preconditions.CheckState(!disposed);
- Preconditions.CheckState(!errorOccured);
Preconditions.CheckState(!readingDone, "Stream has already been closed.");
Preconditions.CheckState(readCompletionDelegate == null, "Only one read can be pending at a time");
@@ -280,7 +269,7 @@ namespace Grpc.Core.Internal
/// <summary>
/// Handles send completion.
/// </summary>
- protected void HandleSendFinished(bool success, BatchContextSafeHandle ctx)
+ protected void HandleSendFinished(bool success)
{
AsyncCompletionDelegate<object> origCompletionDelegate = null;
lock (myLock)
@@ -304,12 +293,11 @@ namespace Grpc.Core.Internal
/// <summary>
/// Handles halfclose completion.
/// </summary>
- protected void HandleHalfclosed(bool success, BatchContextSafeHandle ctx)
+ protected void HandleHalfclosed(bool success)
{
AsyncCompletionDelegate<object> origCompletionDelegate = null;
lock (myLock)
{
- halfclosed = true;
origCompletionDelegate = sendCompletionDelegate;
sendCompletionDelegate = null;
@@ -329,23 +317,17 @@ namespace Grpc.Core.Internal
/// <summary>
/// Handles streaming read completion.
/// </summary>
- protected void HandleReadFinished(bool success, BatchContextSafeHandle ctx)
+ protected void HandleReadFinished(bool success, byte[] receivedMessage)
{
- var payload = ctx.GetReceivedMessage();
-
AsyncCompletionDelegate<TRead> origCompletionDelegate = null;
lock (myLock)
{
origCompletionDelegate = readCompletionDelegate;
- if (payload != null)
- {
- readCompletionDelegate = null;
- }
- else
+ readCompletionDelegate = null;
+
+ if (receivedMessage == null)
{
- // This was the last read. Keeping the readCompletionDelegate
- // to be either fired by this handler or by client-side finished
- // handler.
+ // This was the last read.
readingDone = true;
}
@@ -354,17 +336,17 @@ namespace Grpc.Core.Internal
// TODO: handle the case when error occured...
- if (payload != null)
+ if (receivedMessage != null)
{
// TODO: handle deserialization error
TRead msg;
- TryDeserialize(payload, out msg);
+ TryDeserialize(receivedMessage, out msg);
FireCompletion(origCompletionDelegate, msg, null);
}
else
{
- ProcessLastRead(origCompletionDelegate);
+ FireCompletion(origCompletionDelegate, default(TRead), null);
}
}
}
diff --git a/src/csharp/Grpc.Core/Internal/AsyncCallServer.cs b/src/csharp/Grpc.Core/Internal/AsyncCallServer.cs
index 3710a65d6b..5c47251030 100644
--- a/src/csharp/Grpc.Core/Internal/AsyncCallServer.cs
+++ b/src/csharp/Grpc.Core/Internal/AsyncCallServer.cs
@@ -49,17 +49,18 @@ namespace Grpc.Core.Internal
{
readonly TaskCompletionSource<object> finishedServersideTcs = new TaskCompletionSource<object>();
readonly CancellationTokenSource cancellationTokenSource = new CancellationTokenSource();
- readonly GrpcEnvironment environment;
+ readonly Server server;
- public AsyncCallServer(Func<TResponse, byte[]> serializer, Func<byte[], TRequest> deserializer, GrpcEnvironment environment) : base(serializer, deserializer)
+ public AsyncCallServer(Func<TResponse, byte[]> serializer, Func<byte[], TRequest> deserializer, GrpcEnvironment environment, Server server) : base(serializer, deserializer, environment)
{
- this.environment = Preconditions.CheckNotNull(environment);
+ this.server = Preconditions.CheckNotNull(server);
}
public void Initialize(CallSafeHandle call)
{
call.SetCompletionRegistry(environment.CompletionRegistry);
- environment.DebugStats.ActiveServerCalls.Increment();
+
+ server.AddCallReference(this);
InitializeInternal(call);
}
@@ -168,18 +169,22 @@ namespace Grpc.Core.Internal
}
}
- protected override void OnReleaseResources()
+ protected override void CheckReadingAllowed()
{
- environment.DebugStats.ActiveServerCalls.Decrement();
+ base.CheckReadingAllowed();
+ Preconditions.CheckArgument(!cancelRequested);
+ }
+
+ protected override void OnAfterReleaseResources()
+ {
+ server.RemoveCallReference(this);
}
/// <summary>
/// Handles the server side close completion.
/// </summary>
- private void HandleFinishedServerside(bool success, BatchContextSafeHandle ctx)
+ private void HandleFinishedServerside(bool success, bool cancelled)
{
- bool cancelled = ctx.GetReceivedCloseOnServerCancelled();
-
lock (myLock)
{
finished = true;
diff --git a/src/csharp/Grpc.Core/Internal/BatchContextSafeHandle.cs b/src/csharp/Grpc.Core/Internal/BatchContextSafeHandle.cs
index 6a2add54db..3a96414bea 100644
--- a/src/csharp/Grpc.Core/Internal/BatchContextSafeHandle.cs
+++ b/src/csharp/Grpc.Core/Internal/BatchContextSafeHandle.cs
@@ -134,7 +134,7 @@ namespace Grpc.Core.Internal
}
// Gets data of server_rpc_new completion.
- public ServerRpcNew GetServerRpcNew()
+ public ServerRpcNew GetServerRpcNew(Server server)
{
var call = grpcsharp_batch_context_server_rpc_new_call(this);
@@ -145,7 +145,7 @@ namespace Grpc.Core.Internal
IntPtr metadataArrayPtr = grpcsharp_batch_context_server_rpc_new_request_metadata(this);
var metadata = MetadataArraySafeHandle.ReadMetadataFromPtrUnsafe(metadataArrayPtr);
- return new ServerRpcNew(call, method, host, deadline, metadata);
+ return new ServerRpcNew(server, call, method, host, deadline, metadata);
}
// Gets data of receive_close_on_server completion.
@@ -198,14 +198,16 @@ namespace Grpc.Core.Internal
/// </summary>
internal struct ServerRpcNew
{
+ readonly Server server;
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)
+ public ServerRpcNew(Server server, CallSafeHandle call, string method, string host, Timespec deadline, Metadata requestMetadata)
{
+ this.server = server;
this.call = call;
this.method = method;
this.host = host;
@@ -213,6 +215,14 @@ namespace Grpc.Core.Internal
this.requestMetadata = requestMetadata;
}
+ public Server Server
+ {
+ get
+ {
+ return this.server;
+ }
+ }
+
public CallSafeHandle Call
{
get
diff --git a/src/csharp/Grpc.Core/Internal/CallSafeHandle.cs b/src/csharp/Grpc.Core/Internal/CallSafeHandle.cs
index 3cb01e29bd..0f187529e8 100644
--- a/src/csharp/Grpc.Core/Internal/CallSafeHandle.cs
+++ b/src/csharp/Grpc.Core/Internal/CallSafeHandle.cs
@@ -40,7 +40,7 @@ namespace Grpc.Core.Internal
/// <summary>
/// grpc_call from <grpc/grpc.h>
/// </summary>
- internal class CallSafeHandle : SafeHandleZeroIsInvalid
+ internal class CallSafeHandle : SafeHandleZeroIsInvalid, INativeCall
{
public static readonly CallSafeHandle NullInstance = new CallSafeHandle();
@@ -87,6 +87,10 @@ namespace Grpc.Core.Internal
BatchContextSafeHandle ctx);
[DllImport("grpc_csharp_ext.dll")]
+ static extern GRPCCallError grpcsharp_call_recv_initial_metadata(CallSafeHandle call,
+ BatchContextSafeHandle ctx);
+
+ [DllImport("grpc_csharp_ext.dll")]
static extern GRPCCallError grpcsharp_call_start_serverside(CallSafeHandle call,
BatchContextSafeHandle ctx);
@@ -109,10 +113,10 @@ namespace Grpc.Core.Internal
this.completionRegistry = completionRegistry;
}
- public void StartUnary(BatchCompletionDelegate callback, byte[] payload, MetadataArraySafeHandle metadataArray, WriteFlags writeFlags)
+ public void StartUnary(UnaryResponseClientHandler callback, byte[] payload, MetadataArraySafeHandle metadataArray, WriteFlags writeFlags)
{
var ctx = BatchContextSafeHandle.Create();
- completionRegistry.RegisterBatchCompletion(ctx, callback);
+ completionRegistry.RegisterBatchCompletion(ctx, (success, context) => callback(success, context.GetReceivedStatusOnClient(), context.GetReceivedMessage(), context.GetReceivedInitialMetadata()));
grpcsharp_call_start_unary(this, ctx, payload, new UIntPtr((ulong)payload.Length), metadataArray, writeFlags)
.CheckOk();
}
@@ -123,66 +127,73 @@ namespace Grpc.Core.Internal
.CheckOk();
}
- public void StartClientStreaming(BatchCompletionDelegate callback, MetadataArraySafeHandle metadataArray)
+ public void StartClientStreaming(UnaryResponseClientHandler callback, MetadataArraySafeHandle metadataArray)
{
var ctx = BatchContextSafeHandle.Create();
- completionRegistry.RegisterBatchCompletion(ctx, callback);
+ completionRegistry.RegisterBatchCompletion(ctx, (success, context) => callback(success, context.GetReceivedStatusOnClient(), context.GetReceivedMessage(), context.GetReceivedInitialMetadata()));
grpcsharp_call_start_client_streaming(this, ctx, metadataArray).CheckOk();
}
- public void StartServerStreaming(BatchCompletionDelegate callback, byte[] payload, MetadataArraySafeHandle metadataArray, WriteFlags writeFlags)
+ public void StartServerStreaming(ReceivedStatusOnClientHandler callback, byte[] payload, MetadataArraySafeHandle metadataArray, WriteFlags writeFlags)
{
var ctx = BatchContextSafeHandle.Create();
- completionRegistry.RegisterBatchCompletion(ctx, callback);
+ completionRegistry.RegisterBatchCompletion(ctx, (success, context) => callback(success, context.GetReceivedStatusOnClient()));
grpcsharp_call_start_server_streaming(this, ctx, payload, new UIntPtr((ulong)payload.Length), metadataArray, writeFlags).CheckOk();
}
- public void StartDuplexStreaming(BatchCompletionDelegate callback, MetadataArraySafeHandle metadataArray)
+ public void StartDuplexStreaming(ReceivedStatusOnClientHandler callback, MetadataArraySafeHandle metadataArray)
{
var ctx = BatchContextSafeHandle.Create();
- completionRegistry.RegisterBatchCompletion(ctx, callback);
+ completionRegistry.RegisterBatchCompletion(ctx, (success, context) => callback(success, context.GetReceivedStatusOnClient()));
grpcsharp_call_start_duplex_streaming(this, ctx, metadataArray).CheckOk();
}
- public void StartSendMessage(BatchCompletionDelegate callback, byte[] payload, WriteFlags writeFlags, bool sendEmptyInitialMetadata)
+ public void StartSendMessage(SendCompletionHandler callback, byte[] payload, WriteFlags writeFlags, bool sendEmptyInitialMetadata)
{
var ctx = BatchContextSafeHandle.Create();
- completionRegistry.RegisterBatchCompletion(ctx, callback);
+ completionRegistry.RegisterBatchCompletion(ctx, (success, context) => callback(success));
grpcsharp_call_send_message(this, ctx, payload, new UIntPtr((ulong)payload.Length), writeFlags, sendEmptyInitialMetadata).CheckOk();
}
- public void StartSendCloseFromClient(BatchCompletionDelegate callback)
+ public void StartSendCloseFromClient(SendCompletionHandler callback)
{
var ctx = BatchContextSafeHandle.Create();
- completionRegistry.RegisterBatchCompletion(ctx, callback);
+ completionRegistry.RegisterBatchCompletion(ctx, (success, context) => callback(success));
grpcsharp_call_send_close_from_client(this, ctx).CheckOk();
}
- public void StartSendStatusFromServer(BatchCompletionDelegate callback, Status status, MetadataArraySafeHandle metadataArray, bool sendEmptyInitialMetadata)
+ public void StartSendStatusFromServer(SendCompletionHandler callback, Status status, MetadataArraySafeHandle metadataArray, bool sendEmptyInitialMetadata)
{
var ctx = BatchContextSafeHandle.Create();
- completionRegistry.RegisterBatchCompletion(ctx, callback);
+ completionRegistry.RegisterBatchCompletion(ctx, (success, context) => callback(success));
grpcsharp_call_send_status_from_server(this, ctx, status.StatusCode, status.Detail, metadataArray, sendEmptyInitialMetadata).CheckOk();
}
- public void StartReceiveMessage(BatchCompletionDelegate callback)
+ public void StartReceiveMessage(ReceivedMessageHandler callback)
{
var ctx = BatchContextSafeHandle.Create();
- completionRegistry.RegisterBatchCompletion(ctx, callback);
+ completionRegistry.RegisterBatchCompletion(ctx, (success, context) => callback(success, context.GetReceivedMessage()));
grpcsharp_call_recv_message(this, ctx).CheckOk();
}
- public void StartServerSide(BatchCompletionDelegate callback)
+ public void StartReceiveInitialMetadata(ReceivedResponseHeadersHandler callback)
+ {
+ var ctx = BatchContextSafeHandle.Create();
+ completionRegistry.RegisterBatchCompletion(ctx, (success, context) => callback(success, context.GetReceivedInitialMetadata()));
+ grpcsharp_call_recv_initial_metadata(this, ctx).CheckOk();
+ }
+
+ public void StartServerSide(ReceivedCloseOnServerHandler callback)
{
var ctx = BatchContextSafeHandle.Create();
- completionRegistry.RegisterBatchCompletion(ctx, callback);
+ completionRegistry.RegisterBatchCompletion(ctx, (success, context) => callback(success, context.GetReceivedCloseOnServerCancelled()));
grpcsharp_call_start_serverside(this, ctx).CheckOk();
}
- public void StartSendInitialMetadata(BatchCompletionDelegate callback, MetadataArraySafeHandle metadataArray)
+ public void StartSendInitialMetadata(SendCompletionHandler callback, MetadataArraySafeHandle metadataArray)
{
var ctx = BatchContextSafeHandle.Create();
- completionRegistry.RegisterBatchCompletion(ctx, callback);
+ completionRegistry.RegisterBatchCompletion(ctx, (success, context) => callback(success));
grpcsharp_call_send_initial_metadata(this, ctx, metadataArray).CheckOk();
}
diff --git a/src/csharp/Grpc.Core/Internal/ChannelSafeHandle.cs b/src/csharp/Grpc.Core/Internal/ChannelSafeHandle.cs
index 7f03bf4ea5..8cef566c14 100644
--- a/src/csharp/Grpc.Core/Internal/ChannelSafeHandle.cs
+++ b/src/csharp/Grpc.Core/Internal/ChannelSafeHandle.cs
@@ -68,11 +68,17 @@ namespace Grpc.Core.Internal
public static ChannelSafeHandle CreateInsecure(string target, ChannelArgsSafeHandle channelArgs)
{
+ // Increment reference count for the native gRPC environment to make sure we don't do grpc_shutdown() before destroying the server handle.
+ // Doing so would make object finalizer crash if we end up abandoning the handle.
+ GrpcEnvironment.GrpcNativeInit();
return grpcsharp_insecure_channel_create(target, channelArgs);
}
public static ChannelSafeHandle CreateSecure(CredentialsSafeHandle credentials, string target, ChannelArgsSafeHandle channelArgs)
{
+ // Increment reference count for the native gRPC environment to make sure we don't do grpc_shutdown() before destroying the server handle.
+ // Doing so would make object finalizer crash if we end up abandoning the handle.
+ GrpcEnvironment.GrpcNativeInit();
return grpcsharp_secure_channel_create(credentials, target, channelArgs);
}
@@ -107,6 +113,7 @@ namespace Grpc.Core.Internal
protected override bool ReleaseHandle()
{
grpcsharp_channel_destroy(handle);
+ GrpcEnvironment.GrpcNativeShutdown();
return true;
}
}
diff --git a/src/csharp/Grpc.Core/Internal/ClientResponseStream.cs b/src/csharp/Grpc.Core/Internal/ClientResponseStream.cs
index 6c44521038..b4a7335c7c 100644
--- a/src/csharp/Grpc.Core/Internal/ClientResponseStream.cs
+++ b/src/csharp/Grpc.Core/Internal/ClientResponseStream.cs
@@ -72,7 +72,13 @@ namespace Grpc.Core.Internal
call.StartReadMessage(taskSource.CompletionDelegate);
var result = await taskSource.Task;
this.current = result;
- return result != null;
+
+ if (result == null)
+ {
+ await call.StreamingCallFinishedTask;
+ return false;
+ }
+ return true;
}
public void Dispose()
diff --git a/src/csharp/Grpc.Core/Internal/DebugStats.cs b/src/csharp/Grpc.Core/Internal/DebugStats.cs
index 8793450ff3..1bea1adf9e 100644
--- a/src/csharp/Grpc.Core/Internal/DebugStats.cs
+++ b/src/csharp/Grpc.Core/Internal/DebugStats.cs
@@ -38,10 +38,6 @@ namespace Grpc.Core.Internal
{
internal class DebugStats
{
- public readonly AtomicCounter ActiveClientCalls = new AtomicCounter();
-
- public readonly AtomicCounter ActiveServerCalls = new AtomicCounter();
-
public readonly AtomicCounter PendingBatchCompletions = new AtomicCounter();
/// <summary>
@@ -49,16 +45,6 @@ namespace Grpc.Core.Internal
/// </summary>
public void CheckOK()
{
- var remainingClientCalls = ActiveClientCalls.Count;
- if (remainingClientCalls != 0)
- {
- DebugWarning(string.Format("Detected {0} client calls that weren't disposed properly.", remainingClientCalls));
- }
- var remainingServerCalls = ActiveServerCalls.Count;
- if (remainingServerCalls != 0)
- {
- DebugWarning(string.Format("Detected {0} server calls that weren't disposed properly.", remainingServerCalls));
- }
var pendingBatchCompletions = PendingBatchCompletions.Count;
if (pendingBatchCompletions != 0)
{
diff --git a/src/csharp/Grpc.Core/Internal/GrpcThreadPool.cs b/src/csharp/Grpc.Core/Internal/GrpcThreadPool.cs
index cb4c7c821e..4b7124ee74 100644
--- a/src/csharp/Grpc.Core/Internal/GrpcThreadPool.cs
+++ b/src/csharp/Grpc.Core/Internal/GrpcThreadPool.cs
@@ -83,8 +83,6 @@ namespace Grpc.Core.Internal
lock (myLock)
{
cq.Shutdown();
-
- Logger.Info("Waiting for GRPC threads to finish.");
foreach (var thread in threads)
{
thread.Join();
@@ -136,7 +134,6 @@ namespace Grpc.Core.Internal
}
}
while (ev.type != GRPCCompletionType.Shutdown);
- Logger.Info("Completion queue has shutdown successfully, thread {0} exiting.", Thread.CurrentThread.Name);
}
}
}
diff --git a/src/csharp/Grpc.Core/Internal/INativeCall.cs b/src/csharp/Grpc.Core/Internal/INativeCall.cs
new file mode 100644
index 0000000000..cbef599139
--- /dev/null
+++ b/src/csharp/Grpc.Core/Internal/INativeCall.cs
@@ -0,0 +1,85 @@
+#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;
+
+namespace Grpc.Core.Internal
+{
+ internal delegate void UnaryResponseClientHandler(bool success, ClientSideStatus receivedStatus, byte[] receivedMessage, Metadata responseHeaders);
+
+ // Received status for streaming response calls.
+ internal delegate void ReceivedStatusOnClientHandler(bool success, ClientSideStatus receivedStatus);
+
+ internal delegate void ReceivedMessageHandler(bool success, byte[] receivedMessage);
+
+ internal delegate void ReceivedResponseHeadersHandler(bool success, Metadata responseHeaders);
+
+ internal delegate void SendCompletionHandler(bool success);
+
+ internal delegate void ReceivedCloseOnServerHandler(bool success, bool cancelled);
+
+ /// <summary>
+ /// Abstraction of a native call object.
+ /// </summary>
+ internal interface INativeCall : IDisposable
+ {
+ void Cancel();
+
+ void CancelWithStatus(Grpc.Core.Status status);
+
+ string GetPeer();
+
+ void StartUnary(UnaryResponseClientHandler callback, byte[] payload, MetadataArraySafeHandle metadataArray, Grpc.Core.WriteFlags writeFlags);
+
+ void StartUnary(BatchContextSafeHandle ctx, byte[] payload, MetadataArraySafeHandle metadataArray, Grpc.Core.WriteFlags writeFlags);
+
+ void StartClientStreaming(UnaryResponseClientHandler callback, MetadataArraySafeHandle metadataArray);
+
+ void StartServerStreaming(ReceivedStatusOnClientHandler callback, byte[] payload, MetadataArraySafeHandle metadataArray, Grpc.Core.WriteFlags writeFlags);
+
+ void StartDuplexStreaming(ReceivedStatusOnClientHandler callback, MetadataArraySafeHandle metadataArray);
+
+ void StartReceiveMessage(ReceivedMessageHandler callback);
+
+ void StartReceiveInitialMetadata(ReceivedResponseHeadersHandler callback);
+
+ void StartSendInitialMetadata(SendCompletionHandler callback, MetadataArraySafeHandle metadataArray);
+
+ void StartSendMessage(SendCompletionHandler callback, byte[] payload, Grpc.Core.WriteFlags writeFlags, bool sendEmptyInitialMetadata);
+
+ void StartSendCloseFromClient(SendCompletionHandler callback);
+
+ void StartSendStatusFromServer(SendCompletionHandler callback, Grpc.Core.Status status, MetadataArraySafeHandle metadataArray, bool sendEmptyInitialMetadata);
+
+ void StartServerSide(ReceivedCloseOnServerHandler callback);
+ }
+}
diff --git a/src/csharp/Grpc.Core/Internal/MetadataArraySafeHandle.cs b/src/csharp/Grpc.Core/Internal/MetadataArraySafeHandle.cs
index 427c16fac6..83994f6762 100644
--- a/src/csharp/Grpc.Core/Internal/MetadataArraySafeHandle.cs
+++ b/src/csharp/Grpc.Core/Internal/MetadataArraySafeHandle.cs
@@ -70,7 +70,8 @@ namespace Grpc.Core.Internal
var metadataArray = grpcsharp_metadata_array_create(new UIntPtr((ulong)metadata.Count));
for (int i = 0; i < metadata.Count; i++)
{
- grpcsharp_metadata_array_add(metadataArray, metadata[i].Key, metadata[i].ValueBytes, new UIntPtr((ulong)metadata[i].ValueBytes.Length));
+ var valueBytes = metadata[i].GetSerializedValueUnsafe();
+ grpcsharp_metadata_array_add(metadataArray, metadata[i].Key, valueBytes, new UIntPtr((ulong)valueBytes.Length));
}
return metadataArray;
}
@@ -94,7 +95,7 @@ namespace Grpc.Core.Internal
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));
+ metadata.Add(Metadata.Entry.CreateUnsafe(key, bytes));
}
return metadata;
}
diff --git a/src/csharp/Grpc.Core/Internal/ServerCallHandler.cs b/src/csharp/Grpc.Core/Internal/ServerCallHandler.cs
index 688f9f6fec..59f4c5727c 100644
--- a/src/csharp/Grpc.Core/Internal/ServerCallHandler.cs
+++ b/src/csharp/Grpc.Core/Internal/ServerCallHandler.cs
@@ -67,7 +67,7 @@ namespace Grpc.Core.Internal
var asyncCall = new AsyncCallServer<TRequest, TResponse>(
method.ResponseMarshaller.Serializer,
method.RequestMarshaller.Deserializer,
- environment);
+ environment, newRpc.Server);
asyncCall.Initialize(newRpc.Call);
var finishedTask = asyncCall.ServerSideCallAsync();
@@ -123,7 +123,7 @@ namespace Grpc.Core.Internal
var asyncCall = new AsyncCallServer<TRequest, TResponse>(
method.ResponseMarshaller.Serializer,
method.RequestMarshaller.Deserializer,
- environment);
+ environment, newRpc.Server);
asyncCall.Initialize(newRpc.Call);
var finishedTask = asyncCall.ServerSideCallAsync();
@@ -179,7 +179,7 @@ namespace Grpc.Core.Internal
var asyncCall = new AsyncCallServer<TRequest, TResponse>(
method.ResponseMarshaller.Serializer,
method.RequestMarshaller.Deserializer,
- environment);
+ environment, newRpc.Server);
asyncCall.Initialize(newRpc.Call);
var finishedTask = asyncCall.ServerSideCallAsync();
@@ -239,7 +239,7 @@ namespace Grpc.Core.Internal
var asyncCall = new AsyncCallServer<TRequest, TResponse>(
method.ResponseMarshaller.Serializer,
method.RequestMarshaller.Deserializer,
- environment);
+ environment, newRpc.Server);
asyncCall.Initialize(newRpc.Call);
var finishedTask = asyncCall.ServerSideCallAsync();
@@ -278,7 +278,7 @@ namespace Grpc.Core.Internal
{
// We don't care about the payload type here.
var asyncCall = new AsyncCallServer<byte[], byte[]>(
- (payload) => payload, (payload) => payload, environment);
+ (payload) => payload, (payload) => payload, environment, newRpc.Server);
asyncCall.Initialize(newRpc.Call);
var finishedTask = asyncCall.ServerSideCallAsync();
diff --git a/src/csharp/Grpc.Core/Internal/ServerSafeHandle.cs b/src/csharp/Grpc.Core/Internal/ServerSafeHandle.cs
index f9b44b1acf..5ee7ac14e8 100644
--- a/src/csharp/Grpc.Core/Internal/ServerSafeHandle.cs
+++ b/src/csharp/Grpc.Core/Internal/ServerSafeHandle.cs
@@ -74,6 +74,9 @@ namespace Grpc.Core.Internal
public static ServerSafeHandle NewServer(CompletionQueueSafeHandle cq, ChannelArgsSafeHandle args)
{
+ // Increment reference count for the native gRPC environment to make sure we don't do grpc_shutdown() before destroying the server handle.
+ // Doing so would make object finalizer crash if we end up abandoning the handle.
+ GrpcEnvironment.GrpcNativeInit();
return grpcsharp_server_create(cq, args);
}
@@ -109,6 +112,7 @@ namespace Grpc.Core.Internal
protected override bool ReleaseHandle()
{
grpcsharp_server_destroy(handle);
+ GrpcEnvironment.GrpcNativeShutdown();
return true;
}
diff --git a/src/csharp/Grpc.Core/Logging/ConsoleLogger.cs b/src/csharp/Grpc.Core/Logging/ConsoleLogger.cs
index 382481d871..35561d25d8 100644
--- a/src/csharp/Grpc.Core/Logging/ConsoleLogger.cs
+++ b/src/csharp/Grpc.Core/Logging/ConsoleLogger.cs
@@ -51,7 +51,19 @@ namespace Grpc.Core.Logging
private ConsoleLogger(Type forType)
{
this.forType = forType;
- this.forTypeString = forType != null ? forType.FullName + " " : "";
+ if (forType != null)
+ {
+ var namespaceStr = forType.Namespace ?? "";
+ if (namespaceStr.Length > 0)
+ {
+ namespaceStr += ".";
+ }
+ this.forTypeString = namespaceStr + forType.Name + " ";
+ }
+ else
+ {
+ this.forTypeString = "";
+ }
}
/// <summary>
diff --git a/src/csharp/Grpc.Core/Metadata.cs b/src/csharp/Grpc.Core/Metadata.cs
index 9db2abf46e..a589b50caa 100644
--- a/src/csharp/Grpc.Core/Metadata.cs
+++ b/src/csharp/Grpc.Core/Metadata.cs
@@ -46,6 +46,11 @@ namespace Grpc.Core
public sealed class Metadata : IList<Metadata.Entry>
{
/// <summary>
+ /// All binary headers should have this suffix.
+ /// </summary>
+ public const string BinaryHeaderSuffix = "-bin";
+
+ /// <summary>
/// An read-only instance of metadata containing no entries.
/// </summary>
public static readonly Metadata Empty = new Metadata().Freeze();
@@ -181,23 +186,49 @@ namespace Grpc.Core
private static readonly Encoding Encoding = Encoding.ASCII;
readonly string key;
- string value;
- byte[] valueBytes;
+ readonly string value;
+ readonly byte[] valueBytes;
+
+ private Entry(string key, string value, byte[] valueBytes)
+ {
+ this.key = key;
+ this.value = value;
+ this.valueBytes = valueBytes;
+ }
+ /// <summary>
+ /// Initializes a new instance of the <see cref="Grpc.Core.Metadata+Entry"/> struct with a binary value.
+ /// </summary>
+ /// <param name="key">Metadata key, needs to have suffix indicating a binary valued metadata entry.</param>
+ /// <param name="valueBytes">Value bytes.</param>
public Entry(string key, byte[] valueBytes)
{
- this.key = Preconditions.CheckNotNull(key, "key");
+ this.key = NormalizeKey(key);
+ Preconditions.CheckArgument(this.key.EndsWith(BinaryHeaderSuffix),
+ "Key for binary valued metadata entry needs to have suffix indicating binary value.");
this.value = null;
- this.valueBytes = Preconditions.CheckNotNull(valueBytes, "valueBytes");
+ Preconditions.CheckNotNull(valueBytes, "valueBytes");
+ this.valueBytes = new byte[valueBytes.Length];
+ Buffer.BlockCopy(valueBytes, 0, this.valueBytes, 0, valueBytes.Length); // defensive copy to guarantee immutability
}
+ /// <summary>
+ /// Initializes a new instance of the <see cref="Grpc.Core.Metadata+Entry"/> struct holding an ASCII value.
+ /// </summary>
+ /// <param name="key">Metadata key, must not use suffix indicating a binary valued metadata entry.</param>
+ /// <param name="value">Value string. Only ASCII characters are allowed.</param>
public Entry(string key, string value)
{
- this.key = Preconditions.CheckNotNull(key, "key");
+ this.key = NormalizeKey(key);
+ Preconditions.CheckArgument(!this.key.EndsWith(BinaryHeaderSuffix),
+ "Key for ASCII valued metadata entry cannot have suffix indicating binary value.");
this.value = Preconditions.CheckNotNull(value, "value");
this.valueBytes = null;
}
+ /// <summary>
+ /// Gets the metadata entry key.
+ /// </summary>
public string Key
{
get
@@ -206,33 +237,86 @@ namespace Grpc.Core
}
}
+ /// <summary>
+ /// Gets the binary value of this metadata entry.
+ /// </summary>
public byte[] ValueBytes
{
get
{
if (valueBytes == null)
{
- valueBytes = Encoding.GetBytes(value);
+ return Encoding.GetBytes(value);
}
- return valueBytes;
+
+ // defensive copy to guarantee immutability
+ var bytes = new byte[valueBytes.Length];
+ Buffer.BlockCopy(valueBytes, 0, bytes, 0, valueBytes.Length);
+ return bytes;
}
}
+ /// <summary>
+ /// Gets the string value of this metadata entry.
+ /// </summary>
public string Value
{
get
{
- if (value == null)
- {
- value = Encoding.GetString(valueBytes);
- }
- return value;
+ Preconditions.CheckState(!IsBinary, "Cannot access string value of a binary metadata entry");
+ return value ?? Encoding.GetString(valueBytes);
}
}
-
+
+ /// <summary>
+ /// Returns <c>true</c> if this entry is a binary-value entry.
+ /// </summary>
+ public bool IsBinary
+ {
+ get
+ {
+ return value == null;
+ }
+ }
+
+ /// <summary>
+ /// Returns a <see cref="System.String"/> that represents the current <see cref="Grpc.Core.Metadata+Entry"/>.
+ /// </summary>
public override string ToString()
{
- return string.Format("[Entry: key={0}, value={1}]", Key, Value);
+ if (IsBinary)
+ {
+ return string.Format("[Entry: key={0}, valueBytes={1}]", key, valueBytes);
+ }
+
+ return string.Format("[Entry: key={0}, value={1}]", key, value);
+ }
+
+ /// <summary>
+ /// Gets the serialized value for this entry. For binary metadata entries, this leaks
+ /// the internal <c>valueBytes</c> byte array and caller must not change contents of it.
+ /// </summary>
+ internal byte[] GetSerializedValueUnsafe()
+ {
+ return valueBytes ?? Encoding.GetBytes(value);
+ }
+
+ /// <summary>
+ /// Creates a binary value or ascii value metadata entry from data received from the native layer.
+ /// We trust C core to give us well-formed data, so we don't perform any checks or defensive copying.
+ /// </summary>
+ internal static Entry CreateUnsafe(string key, byte[] valueBytes)
+ {
+ if (key.EndsWith(BinaryHeaderSuffix))
+ {
+ return new Entry(key, null, valueBytes);
+ }
+ return new Entry(key, Encoding.GetString(valueBytes), null);
+ }
+
+ private static string NormalizeKey(string key)
+ {
+ return Preconditions.CheckNotNull(key, "key").ToLower();
}
}
}
diff --git a/src/csharp/Grpc.Core/Server.cs b/src/csharp/Grpc.Core/Server.cs
index c76f126026..28f1686e20 100644
--- a/src/csharp/Grpc.Core/Server.cs
+++ b/src/csharp/Grpc.Core/Server.cs
@@ -50,6 +50,8 @@ namespace Grpc.Core
{
static readonly ILogger Logger = GrpcEnvironment.Logger.ForType<Server>();
+ readonly AtomicCounter activeCallCounter = new AtomicCounter();
+
readonly ServiceDefinitionCollection serviceDefinitions;
readonly ServerPortCollection ports;
readonly GrpcEnvironment environment;
@@ -73,7 +75,7 @@ namespace Grpc.Core
{
this.serviceDefinitions = new ServiceDefinitionCollection(this);
this.ports = new ServerPortCollection(this);
- this.environment = GrpcEnvironment.GetInstance();
+ this.environment = GrpcEnvironment.AddRef();
this.options = options != null ? new List<ChannelOption>(options) : new List<ChannelOption>();
using (var channelArgs = ChannelOptions.CreateChannelArgs(this.options))
{
@@ -106,6 +108,17 @@ namespace Grpc.Core
}
/// <summary>
+ /// To allow awaiting termination of the server.
+ /// </summary>
+ public Task ShutdownTask
+ {
+ get
+ {
+ return shutdownTcs.Task;
+ }
+ }
+
+ /// <summary>
/// Starts the server.
/// </summary>
public void Start()
@@ -136,18 +149,9 @@ namespace Grpc.Core
handle.ShutdownAndNotify(HandleServerShutdown, environment);
await shutdownTcs.Task;
- handle.Dispose();
- }
+ DisposeHandle();
- /// <summary>
- /// To allow awaiting termination of the server.
- /// </summary>
- public Task ShutdownTask
- {
- get
- {
- return shutdownTcs.Task;
- }
+ await Task.Run(() => GrpcEnvironment.Release());
}
/// <summary>
@@ -166,7 +170,22 @@ namespace Grpc.Core
handle.ShutdownAndNotify(HandleServerShutdown, environment);
handle.CancelAllCalls();
await shutdownTcs.Task;
- handle.Dispose();
+ DisposeHandle();
+ }
+
+ internal void AddCallReference(object call)
+ {
+ activeCallCounter.Increment();
+
+ bool success = false;
+ handle.DangerousAddRef(ref success);
+ Preconditions.CheckState(success);
+ }
+
+ internal void RemoveCallReference(object call)
+ {
+ handle.DangerousRelease();
+ activeCallCounter.Decrement();
}
/// <summary>
@@ -227,6 +246,16 @@ namespace Grpc.Core
}
}
+ private void DisposeHandle()
+ {
+ var activeCallCount = activeCallCounter.Count;
+ if (activeCallCount > 0)
+ {
+ Logger.Warning("Server shutdown has finished but there are still {0} active calls for that server.", activeCallCount);
+ }
+ handle.Dispose();
+ }
+
/// <summary>
/// Selects corresponding handler for given call and handles the call.
/// </summary>
@@ -254,7 +283,7 @@ namespace Grpc.Core
{
if (success)
{
- ServerRpcNew newRpc = ctx.GetServerRpcNew();
+ ServerRpcNew newRpc = ctx.GetServerRpcNew(this);
// after server shutdown, the callback returns with null call
if (!newRpc.Call.IsInvalid)
diff --git a/src/csharp/Grpc.Examples.MathClient/MathClient.cs b/src/csharp/Grpc.Examples.MathClient/MathClient.cs
index f9839d99f1..abd95cb905 100644
--- a/src/csharp/Grpc.Examples.MathClient/MathClient.cs
+++ b/src/csharp/Grpc.Examples.MathClient/MathClient.cs
@@ -39,23 +39,21 @@ namespace math
{
public static void Main(string[] args)
{
- using (Channel channel = new Channel("127.0.0.1", 23456, Credentials.Insecure))
- {
- Math.IMathClient client = new Math.MathClient(channel);
- MathExamples.DivExample(client);
+ var channel = new Channel("127.0.0.1", 23456, Credentials.Insecure);
+ Math.IMathClient client = new Math.MathClient(channel);
+ MathExamples.DivExample(client);
- MathExamples.DivAsyncExample(client).Wait();
+ MathExamples.DivAsyncExample(client).Wait();
- MathExamples.FibExample(client).Wait();
+ MathExamples.FibExample(client).Wait();
- MathExamples.SumExample(client).Wait();
+ MathExamples.SumExample(client).Wait();
- MathExamples.DivManyExample(client).Wait();
+ MathExamples.DivManyExample(client).Wait();
- MathExamples.DependendRequestsExample(client).Wait();
- }
+ MathExamples.DependendRequestsExample(client).Wait();
- GrpcEnvironment.Shutdown();
+ channel.ShutdownAsync().Wait();
}
}
}
diff --git a/src/csharp/Grpc.Examples.MathServer/MathServer.cs b/src/csharp/Grpc.Examples.MathServer/MathServer.cs
index 5f7e717b0c..26bef646ec 100644
--- a/src/csharp/Grpc.Examples.MathServer/MathServer.cs
+++ b/src/csharp/Grpc.Examples.MathServer/MathServer.cs
@@ -56,7 +56,6 @@ namespace math
Console.ReadKey();
server.ShutdownAsync().Wait();
- GrpcEnvironment.Shutdown();
}
}
}
diff --git a/src/csharp/Grpc.Examples.Tests/MathClientServerTests.cs b/src/csharp/Grpc.Examples.Tests/MathClientServerTests.cs
index fdef950f09..36c1c947bd 100644
--- a/src/csharp/Grpc.Examples.Tests/MathClientServerTests.cs
+++ b/src/csharp/Grpc.Examples.Tests/MathClientServerTests.cs
@@ -68,9 +68,8 @@ namespace math.Tests
[TestFixtureTearDown]
public void Cleanup()
{
- channel.Dispose();
+ channel.ShutdownAsync().Wait();
server.ShutdownAsync().Wait();
- GrpcEnvironment.Shutdown();
}
[Test]
diff --git a/src/csharp/Grpc.HealthCheck.Tests/HealthClientServerTest.cs b/src/csharp/Grpc.HealthCheck.Tests/HealthClientServerTest.cs
index 024377e216..80c35fb197 100644
--- a/src/csharp/Grpc.HealthCheck.Tests/HealthClientServerTest.cs
+++ b/src/csharp/Grpc.HealthCheck.Tests/HealthClientServerTest.cs
@@ -71,10 +71,9 @@ namespace Grpc.HealthCheck.Tests
[TestFixtureTearDown]
public void Cleanup()
{
- channel.Dispose();
+ channel.ShutdownAsync().Wait();
server.ShutdownAsync().Wait();
- GrpcEnvironment.Shutdown();
}
[Test]
diff --git a/src/csharp/Grpc.IntegrationTesting/InteropClient.cs b/src/csharp/Grpc.IntegrationTesting/InteropClient.cs
index f4b0a1028f..24c22273fb 100644
--- a/src/csharp/Grpc.IntegrationTesting/InteropClient.cs
+++ b/src/csharp/Grpc.IntegrationTesting/InteropClient.cs
@@ -37,13 +37,15 @@ using System.Text.RegularExpressions;
using System.Threading;
using System.Threading.Tasks;
+using Google.Apis.Auth.OAuth2;
using Google.ProtocolBuffers;
+
using grpc.testing;
using Grpc.Auth;
using Grpc.Core;
using Grpc.Core.Utils;
+
using NUnit.Framework;
-using Google.Apis.Auth.OAuth2;
namespace Grpc.IntegrationTesting
{
@@ -118,12 +120,10 @@ namespace Grpc.IntegrationTesting
};
}
- using (Channel channel = new Channel(options.serverHost, options.serverPort.Value, credentials, channelOptions))
- {
- TestService.TestServiceClient client = new TestService.TestServiceClient(channel);
- await RunTestCaseAsync(options.testCase, client);
- }
- GrpcEnvironment.Shutdown();
+ var channel = new Channel(options.serverHost, options.serverPort.Value, credentials, channelOptions);
+ TestService.TestServiceClient client = new TestService.TestServiceClient(channel);
+ await RunTestCaseAsync(options.testCase, client);
+ channel.ShutdownAsync().Wait();
}
private async Task RunTestCaseAsync(string testCase, TestService.TestServiceClient client)
@@ -169,6 +169,9 @@ namespace Grpc.IntegrationTesting
case "cancel_after_first_response":
await RunCancelAfterFirstResponseAsync(client);
break;
+ case "timeout_on_sleeping_server":
+ await RunTimeoutOnSleepingServerAsync(client);
+ break;
case "benchmark_empty_unary":
RunBenchmarkEmptyUnary(client);
break;
@@ -458,6 +461,29 @@ namespace Grpc.IntegrationTesting
Console.WriteLine("Passed!");
}
+ public static async Task RunTimeoutOnSleepingServerAsync(TestService.ITestServiceClient client)
+ {
+ Console.WriteLine("running timeout_on_sleeping_server");
+
+ var deadline = DateTime.UtcNow.AddMilliseconds(1);
+ using (var call = client.FullDuplexCall(deadline: deadline))
+ {
+ try
+ {
+ await call.RequestStream.WriteAsync(StreamingOutputCallRequest.CreateBuilder()
+ .SetPayload(CreateZerosPayload(27182)).Build());
+ }
+ catch (InvalidOperationException)
+ {
+ // Deadline was reached before write has started. Eat the exception and continue.
+ }
+
+ var ex = Assert.Throws<RpcException>(async () => await call.ResponseStream.MoveNext());
+ Assert.AreEqual(StatusCode.DeadlineExceeded, ex.Status.StatusCode);
+ }
+ Console.WriteLine("Passed!");
+ }
+
// This is not an official interop test, but it's useful.
public static void RunBenchmarkEmptyUnary(TestService.ITestServiceClient client)
{
diff --git a/src/csharp/Grpc.IntegrationTesting/InteropClientServerTest.cs b/src/csharp/Grpc.IntegrationTesting/InteropClientServerTest.cs
index 6fa721bc1c..f3158aeb45 100644
--- a/src/csharp/Grpc.IntegrationTesting/InteropClientServerTest.cs
+++ b/src/csharp/Grpc.IntegrationTesting/InteropClientServerTest.cs
@@ -75,9 +75,8 @@ namespace Grpc.IntegrationTesting
[TestFixtureTearDown]
public void Cleanup()
{
- channel.Dispose();
+ channel.ShutdownAsync().Wait();
server.ShutdownAsync().Wait();
- GrpcEnvironment.Shutdown();
}
[Test]
@@ -127,5 +126,11 @@ namespace Grpc.IntegrationTesting
{
await InteropClient.RunCancelAfterFirstResponseAsync(client);
}
+
+ [Test]
+ public async Task TimeoutOnSleepingServerAsync()
+ {
+ await InteropClient.RunTimeoutOnSleepingServerAsync(client);
+ }
}
}
diff --git a/src/csharp/Grpc.IntegrationTesting/InteropServer.cs b/src/csharp/Grpc.IntegrationTesting/InteropServer.cs
index 504fd11857..0cc8b2cde1 100644
--- a/src/csharp/Grpc.IntegrationTesting/InteropServer.cs
+++ b/src/csharp/Grpc.IntegrationTesting/InteropServer.cs
@@ -107,8 +107,6 @@ namespace Grpc.IntegrationTesting
server.Start();
server.ShutdownTask.Wait();
-
- GrpcEnvironment.Shutdown();
}
private static ServerOptions ParseArguments(string[] args)
diff --git a/src/csharp/Grpc.IntegrationTesting/SslCredentialsTest.cs b/src/csharp/Grpc.IntegrationTesting/SslCredentialsTest.cs
index 1c398eb84e..842795374f 100644
--- a/src/csharp/Grpc.IntegrationTesting/SslCredentialsTest.cs
+++ b/src/csharp/Grpc.IntegrationTesting/SslCredentialsTest.cs
@@ -85,9 +85,8 @@ namespace Grpc.IntegrationTesting
[TestFixtureTearDown]
public void Cleanup()
{
- channel.Dispose();
+ channel.ShutdownAsync().Wait();
server.ShutdownAsync().Wait();
- GrpcEnvironment.Shutdown();
}
[Test]
diff --git a/src/csharp/ext/grpc_csharp_ext.c b/src/csharp/ext/grpc_csharp_ext.c
index fc9470f93f..489e219c49 100644
--- a/src/csharp/ext/grpc_csharp_ext.c
+++ b/src/csharp/ext/grpc_csharp_ext.c
@@ -595,7 +595,7 @@ GPR_EXPORT grpc_call_error GPR_CALLTYPE grpcsharp_call_start_server_streaming(
grpc_call *call, grpcsharp_batch_context *ctx, const char *send_buffer,
size_t send_buffer_len, grpc_metadata_array *initial_metadata, gpr_uint32 write_flags) {
/* TODO: don't use magic number */
- grpc_op ops[5];
+ grpc_op ops[4];
ops[0].op = GRPC_OP_SEND_INITIAL_METADATA;
grpcsharp_metadata_array_move(&(ctx->send_initial_metadata),
initial_metadata);
@@ -615,23 +615,18 @@ GPR_EXPORT grpc_call_error GPR_CALLTYPE grpcsharp_call_start_server_streaming(
ops[2].flags = 0;
ops[2].reserved = NULL;
- ops[3].op = GRPC_OP_RECV_INITIAL_METADATA;
- ops[3].data.recv_initial_metadata = &(ctx->recv_initial_metadata);
- ops[3].flags = 0;
- ops[3].reserved = NULL;
-
- ops[4].op = GRPC_OP_RECV_STATUS_ON_CLIENT;
- ops[4].data.recv_status_on_client.trailing_metadata =
+ ops[3].op = GRPC_OP_RECV_STATUS_ON_CLIENT;
+ ops[3].data.recv_status_on_client.trailing_metadata =
&(ctx->recv_status_on_client.trailing_metadata);
- ops[4].data.recv_status_on_client.status =
+ ops[3].data.recv_status_on_client.status =
&(ctx->recv_status_on_client.status);
/* not using preallocation for status_details */
- ops[4].data.recv_status_on_client.status_details =
+ ops[3].data.recv_status_on_client.status_details =
&(ctx->recv_status_on_client.status_details);
- ops[4].data.recv_status_on_client.status_details_capacity =
+ ops[3].data.recv_status_on_client.status_details_capacity =
&(ctx->recv_status_on_client.status_details_capacity);
- ops[4].flags = 0;
- ops[4].reserved = NULL;
+ ops[3].flags = 0;
+ ops[3].reserved = NULL;
return grpc_call_start_batch(call, ops, sizeof(ops) / sizeof(ops[0]), ctx,
NULL);
@@ -642,7 +637,7 @@ grpcsharp_call_start_duplex_streaming(grpc_call *call,
grpcsharp_batch_context *ctx,
grpc_metadata_array *initial_metadata) {
/* TODO: don't use magic number */
- grpc_op ops[3];
+ grpc_op ops[2];
ops[0].op = GRPC_OP_SEND_INITIAL_METADATA;
grpcsharp_metadata_array_move(&(ctx->send_initial_metadata),
initial_metadata);
@@ -652,28 +647,36 @@ grpcsharp_call_start_duplex_streaming(grpc_call *call,
ops[0].flags = 0;
ops[0].reserved = NULL;
- ops[1].op = GRPC_OP_RECV_INITIAL_METADATA;
- ops[1].data.recv_initial_metadata = &(ctx->recv_initial_metadata);
- ops[1].flags = 0;
- ops[1].reserved = NULL;
-
- ops[2].op = GRPC_OP_RECV_STATUS_ON_CLIENT;
- ops[2].data.recv_status_on_client.trailing_metadata =
+ ops[1].op = GRPC_OP_RECV_STATUS_ON_CLIENT;
+ ops[1].data.recv_status_on_client.trailing_metadata =
&(ctx->recv_status_on_client.trailing_metadata);
- ops[2].data.recv_status_on_client.status =
+ ops[1].data.recv_status_on_client.status =
&(ctx->recv_status_on_client.status);
/* not using preallocation for status_details */
- ops[2].data.recv_status_on_client.status_details =
+ ops[1].data.recv_status_on_client.status_details =
&(ctx->recv_status_on_client.status_details);
- ops[2].data.recv_status_on_client.status_details_capacity =
+ ops[1].data.recv_status_on_client.status_details_capacity =
&(ctx->recv_status_on_client.status_details_capacity);
- ops[2].flags = 0;
- ops[2].reserved = NULL;
+ ops[1].flags = 0;
+ ops[1].reserved = NULL;
return grpc_call_start_batch(call, ops, sizeof(ops) / sizeof(ops[0]), ctx,
NULL);
}
+GPR_EXPORT grpc_call_error GPR_CALLTYPE grpcsharp_call_recv_initial_metadata(
+ grpc_call *call, grpcsharp_batch_context *ctx) {
+ /* TODO: don't use magic number */
+ grpc_op ops[1];
+ ops[0].op = GRPC_OP_RECV_INITIAL_METADATA;
+ ops[0].data.recv_initial_metadata = &(ctx->recv_initial_metadata);
+ ops[0].flags = 0;
+ ops[0].reserved = NULL;
+
+ return grpc_call_start_batch(call, ops, sizeof(ops) / sizeof(ops[0]), ctx,
+ NULL);
+}
+
GPR_EXPORT grpc_call_error GPR_CALLTYPE
grpcsharp_call_send_message(grpc_call *call, grpcsharp_batch_context *ctx,
const char *send_buffer, size_t send_buffer_len,