Merge pull request #2756 from jtattermusch/csharp_channel_state
C# Improving Channel API
diff --git a/src/csharp/Grpc.Auth/OAuth2Interceptors.cs b/src/csharp/Grpc.Auth/OAuth2Interceptors.cs
index c785ca5..cc9d2c1 100644
--- a/src/csharp/Grpc.Auth/OAuth2Interceptors.cs
+++ b/src/csharp/Grpc.Auth/OAuth2Interceptors.cs
@@ -119,7 +119,5 @@
return new Metadata.Entry(AuthorizationHeader, Schema + " " + accessToken);
}
}
-
-
}
}
diff --git a/src/csharp/Grpc.Core.Tests/ChannelTest.cs b/src/csharp/Grpc.Core.Tests/ChannelTest.cs
new file mode 100644
index 0000000..60b4517
--- /dev/null
+++ b/src/csharp/Grpc.Core.Tests/ChannelTest.cs
@@ -0,0 +1,91 @@
+#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 Grpc.Core;
+using Grpc.Core.Internal;
+using Grpc.Core.Utils;
+using NUnit.Framework;
+
+namespace Grpc.Core.Tests
+{
+ public class ChannelTest
+ {
+ [TestFixtureTearDown]
+ public void CleanupClass()
+ {
+ GrpcEnvironment.Shutdown();
+ }
+
+ [Test]
+ public void Constructor_RejectsInvalidParams()
+ {
+ Assert.Throws(typeof(NullReferenceException), () => new Channel(null, Credentials.Insecure));
+ }
+
+ [Test]
+ public void State_IdleAfterCreation()
+ {
+ using (var channel = new Channel("localhost", Credentials.Insecure))
+ {
+ Assert.AreEqual(ChannelState.Idle, channel.State);
+ }
+ }
+
+ [Test]
+ public void WaitForStateChangedAsync_InvalidArgument()
+ {
+ using (var channel = new Channel("localhost", Credentials.Insecure))
+ {
+ Assert.Throws(typeof(ArgumentException), () => channel.WaitForStateChangedAsync(ChannelState.FatalFailure));
+ }
+ }
+
+ [Test]
+ public void Target()
+ {
+ using (var channel = new Channel("127.0.0.1", Credentials.Insecure))
+ {
+ Assert.IsTrue(channel.Target.Contains("127.0.0.1"));
+ }
+ }
+
+ [Test]
+ public void Dispose_IsIdempotent()
+ {
+ var channel = new Channel("localhost", Credentials.Insecure);
+ channel.Dispose();
+ channel.Dispose();
+ }
+ }
+}
diff --git a/src/csharp/Grpc.Core.Tests/ClientServerTest.cs b/src/csharp/Grpc.Core.Tests/ClientServerTest.cs
index 540fe75..3592486 100644
--- a/src/csharp/Grpc.Core.Tests/ClientServerTest.cs
+++ b/src/csharp/Grpc.Core.Tests/ClientServerTest.cs
@@ -276,6 +276,30 @@
Assert.IsTrue(peer.Contains(Host));
}
+ [Test]
+ public async Task Channel_WaitForStateChangedAsync()
+ {
+ Assert.Throws(typeof(TaskCanceledException),
+ async () => await channel.WaitForStateChangedAsync(channel.State, DateTime.UtcNow.AddMilliseconds(10)));
+
+ var stateChangedTask = channel.WaitForStateChangedAsync(channel.State);
+
+ var internalCall = new Call<string, string>(ServiceName, EchoMethod, channel, Metadata.Empty);
+ await Calls.AsyncUnaryCall(internalCall, "abc", CancellationToken.None);
+
+ await stateChangedTask;
+ Assert.AreEqual(ChannelState.Ready, channel.State);
+ }
+
+ [Test]
+ public async Task Channel_ConnectAsync()
+ {
+ await channel.ConnectAsync();
+ Assert.AreEqual(ChannelState.Ready, channel.State);
+ await channel.ConnectAsync(DateTime.UtcNow.AddMilliseconds(1000));
+ Assert.AreEqual(ChannelState.Ready, channel.State);
+ }
+
private static async Task<string> EchoHandler(string request, ServerCallContext context)
{
foreach (Metadata.Entry metadataEntry in context.RequestHeaders)
diff --git a/src/csharp/Grpc.Core.Tests/Grpc.Core.Tests.csproj b/src/csharp/Grpc.Core.Tests/Grpc.Core.Tests.csproj
index 242a60d..f2bf459 100644
--- a/src/csharp/Grpc.Core.Tests/Grpc.Core.Tests.csproj
+++ b/src/csharp/Grpc.Core.Tests/Grpc.Core.Tests.csproj
@@ -76,6 +76,7 @@
<Compile Include="Internal\TimespecTest.cs" />
<Compile Include="TimeoutsTest.cs" />
<Compile Include="NUnitVersionTest.cs" />
+ <Compile Include="ChannelTest.cs" />
</ItemGroup>
<Import Project="$(MSBuildBinPath)\Microsoft.CSharp.targets" />
<ItemGroup>
diff --git a/src/csharp/Grpc.Core.Tests/NUnitVersionTest.cs b/src/csharp/Grpc.Core.Tests/NUnitVersionTest.cs
index 600df1a..3fa6ad0 100644
--- a/src/csharp/Grpc.Core.Tests/NUnitVersionTest.cs
+++ b/src/csharp/Grpc.Core.Tests/NUnitVersionTest.cs
@@ -70,10 +70,8 @@
[Test]
public async Task NUnitVersionTest2()
{
- testRunCount ++;
+ testRunCount++;
await Task.Delay(10);
}
-
-
}
}
diff --git a/src/csharp/Grpc.Core.Tests/TimeoutsTest.cs b/src/csharp/Grpc.Core.Tests/TimeoutsTest.cs
index 010ffd8..a09273b 100644
--- a/src/csharp/Grpc.Core.Tests/TimeoutsTest.cs
+++ b/src/csharp/Grpc.Core.Tests/TimeoutsTest.cs
@@ -134,7 +134,8 @@
}
catch (RpcException e)
{
- Assert.AreEqual(StatusCode.DeadlineExceeded, e.Status.StatusCode);
+ // We can't guarantee the status code always DeadlineExceeded. See issue #2685.
+ Assert.Contains(e.Status.StatusCode, new[] { StatusCode.DeadlineExceeded, StatusCode.Internal });
}
}
@@ -151,7 +152,8 @@
}
catch (RpcException e)
{
- Assert.AreEqual(StatusCode.DeadlineExceeded, e.Status.StatusCode);
+ // We can't guarantee the status code always DeadlineExceeded. See issue #2685.
+ Assert.Contains(e.Status.StatusCode, new[] { StatusCode.DeadlineExceeded, StatusCode.Internal });
}
}
@@ -168,7 +170,8 @@
}
catch (RpcException e)
{
- Assert.AreEqual(StatusCode.DeadlineExceeded, e.Status.StatusCode);
+ // We can't guarantee the status code is always DeadlineExceeded. See issue #2685.
+ Assert.Contains(e.Status.StatusCode, new[] { StatusCode.DeadlineExceeded, StatusCode.Internal });
}
Assert.AreEqual("CANCELLED", stringFromServerHandlerTcs.Task.Result);
}
diff --git a/src/csharp/Grpc.Core/Channel.cs b/src/csharp/Grpc.Core/Channel.cs
index 18e6f2f..0b69610 100644
--- a/src/csharp/Grpc.Core/Channel.cs
+++ b/src/csharp/Grpc.Core/Channel.cs
@@ -37,6 +37,8 @@
using System.Threading.Tasks;
using Grpc.Core.Internal;
+using Grpc.Core.Logging;
+using Grpc.Core.Utils;
namespace Grpc.Core
{
@@ -45,21 +47,23 @@
/// </summary>
public class Channel : IDisposable
{
+ static readonly ILogger Logger = GrpcEnvironment.Logger.ForType<Channel>();
+
readonly GrpcEnvironment environment;
readonly ChannelSafeHandle handle;
readonly List<ChannelOption> options;
- readonly string target;
bool disposed;
/// <summary>
/// Creates a channel that connects to a specific host.
- /// Port will default to 80 for an unsecure channel and to 443 a secure channel.
+ /// Port will default to 80 for an unsecure channel and to 443 for a secure channel.
/// </summary>
- /// <param name="host">The DNS name of IP address of the host.</param>
+ /// <param name="host">The name or IP address of the host.</param>
/// <param name="credentials">Credentials to secure the channel.</param>
/// <param name="options">Channel options.</param>
public Channel(string host, Credentials credentials, IEnumerable<ChannelOption> options = null)
{
+ Preconditions.CheckNotNull(host);
this.environment = GrpcEnvironment.GetInstance();
this.options = options != null ? new List<ChannelOption>(options) : new List<ChannelOption>();
@@ -76,14 +80,13 @@
this.handle = ChannelSafeHandle.CreateInsecure(host, nativeChannelArgs);
}
}
- this.target = GetOverridenTarget(host, this.options);
}
/// <summary>
/// Creates a channel that connects to a specific host and port.
/// </summary>
- /// <param name="host">DNS name or IP address</param>
- /// <param name="port">the port</param>
+ /// <param name="host">The name or IP address of the host.</param>
+ /// <param name="port">The port.</param>
/// <param name="credentials">Credentials to secure the channel.</param>
/// <param name="options">Channel options.</param>
public Channel(string host, int port, Credentials credentials, IEnumerable<ChannelOption> options = null) :
@@ -91,20 +94,82 @@
{
}
+ /// <summary>
+ /// Gets current connectivity state of this channel.
+ /// </summary>
+ public ChannelState State
+ {
+ get
+ {
+ return handle.CheckConnectivityState(false);
+ }
+ }
+
+ /// <summary>
+ /// Returned tasks completes once channel state has become different from
+ /// given lastObservedState.
+ /// If deadline is reached or and error occurs, returned task is cancelled.
+ /// </summary>
+ public Task WaitForStateChangedAsync(ChannelState lastObservedState, DateTime? deadline = null)
+ {
+ Preconditions.CheckArgument(lastObservedState != ChannelState.FatalFailure,
+ "FatalFailure is a terminal state. No further state changes can occur.");
+ var tcs = new TaskCompletionSource<object>();
+ var deadlineTimespec = deadline.HasValue ? Timespec.FromDateTime(deadline.Value) : Timespec.InfFuture;
+ var handler = new BatchCompletionDelegate((success, ctx) =>
+ {
+ if (success)
+ {
+ tcs.SetResult(null);
+ }
+ else
+ {
+ tcs.SetCanceled();
+ }
+ });
+ handle.WatchConnectivityState(lastObservedState, deadlineTimespec, environment.CompletionQueue, environment.CompletionRegistry, handler);
+ return tcs.Task;
+ }
+
+ /// <summary> Address of the remote endpoint in URI format.</summary>
+ public string Target
+ {
+ get
+ {
+ return handle.GetTarget();
+ }
+ }
+
+ /// <summary>
+ /// Allows explicitly requesting channel to connect without starting an RPC.
+ /// Returned task completes once state Ready was seen. If the deadline is reached,
+ /// or channel enters the FatalFailure state, the task is cancelled.
+ /// There is no need to call this explicitly unless your use case requires that.
+ /// Starting an RPC on a new channel will request connection implicitly.
+ /// </summary>
+ public async Task ConnectAsync(DateTime? deadline = null)
+ {
+ var currentState = handle.CheckConnectivityState(true);
+ while (currentState != ChannelState.Ready)
+ {
+ if (currentState == ChannelState.FatalFailure)
+ {
+ throw new OperationCanceledException("Channel has reached FatalFailure state.");
+ }
+ await WaitForStateChangedAsync(currentState, deadline);
+ currentState = handle.CheckConnectivityState(false);
+ }
+ }
+
+ /// <summary>
+ /// Destroys the underlying channel.
+ /// </summary>
public void Dispose()
{
Dispose(true);
GC.SuppressFinalize(this);
}
- internal string Target
- {
- get
- {
- return target;
- }
- }
-
internal ChannelSafeHandle Handle
{
get
@@ -159,26 +224,5 @@
// TODO(jtattermusch): it would be useful to also provide .NET/mono version.
return string.Format("grpc-csharp/{0}", VersionInfo.CurrentVersion);
}
-
- /// <summary>
- /// Look for SslTargetNameOverride option and return its value instead of originalTarget
- /// if found.
- /// </summary>
- private static string GetOverridenTarget(string originalTarget, IEnumerable<ChannelOption> options)
- {
- if (options == null)
- {
- return originalTarget;
- }
- foreach (var option in options)
- {
- if (option.Type == ChannelOption.OptionType.String
- && option.Name == ChannelOptions.SslTargetNameOverride)
- {
- return option.StringValue;
- }
- }
- return originalTarget;
- }
}
}
diff --git a/src/csharp/Grpc.Core/ChannelOptions.cs b/src/csharp/Grpc.Core/ChannelOptions.cs
index 9fe03d2..f70408d 100644
--- a/src/csharp/Grpc.Core/ChannelOptions.cs
+++ b/src/csharp/Grpc.Core/ChannelOptions.cs
@@ -135,6 +135,9 @@
/// <summary>Initial sequence number for http2 transports</summary>
public const string Http2InitialSequenceNumber = "grpc.http2.initial_sequence_number";
+ /// <summary>Default authority for calls.</summary>
+ public const string DefaultAuthority = "grpc.default_authority";
+
/// <summary>Primary user agent: goes at the start of the user-agent metadata</summary>
public const string PrimaryUserAgentString = "grpc.primary_user_agent";
diff --git a/src/csharp/Grpc.Core/ChannelState.cs b/src/csharp/Grpc.Core/ChannelState.cs
new file mode 100644
index 0000000..d293b98
--- /dev/null
+++ b/src/csharp/Grpc.Core/ChannelState.cs
@@ -0,0 +1,69 @@
+#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
+{
+ /// <summary>
+ /// Connectivity state of a channel.
+ /// Based on grpc_connectivity_state from grpc/grpc.h
+ /// </summary>
+ public enum ChannelState
+ {
+ /// <summary>
+ /// Channel is idle
+ /// </summary>
+ Idle,
+
+ /// <summary>
+ /// Channel is connecting
+ /// </summary>
+ Connecting,
+
+ /// <summary>
+ /// Channel is ready for work
+ /// </summary>
+ Ready,
+
+ /// <summary>
+ /// Channel has seen a failure but expects to recover
+ /// </summary>
+ TransientFailure,
+
+ /// <summary>
+ /// Channel has seen a failure that it cannot recover from
+ /// </summary>
+ FatalFailure
+ }
+}
diff --git a/src/csharp/Grpc.Core/Grpc.Core.csproj b/src/csharp/Grpc.Core/Grpc.Core.csproj
index 940a6b8..641b54b 100644
--- a/src/csharp/Grpc.Core/Grpc.Core.csproj
+++ b/src/csharp/Grpc.Core/Grpc.Core.csproj
@@ -115,6 +115,7 @@
<Compile Include="Logging\ILogger.cs" />
<Compile Include="Logging\ConsoleLogger.cs" />
<Compile Include="Internal\NativeLogRedirector.cs" />
+ <Compile Include="ChannelState.cs" />
</ItemGroup>
<ItemGroup>
<None Include="Grpc.Core.nuspec" />
diff --git a/src/csharp/Grpc.Core/Internal/AsyncCall.cs b/src/csharp/Grpc.Core/Internal/AsyncCall.cs
index bfcb936..48f4664 100644
--- a/src/csharp/Grpc.Core/Internal/AsyncCall.cs
+++ b/src/csharp/Grpc.Core/Internal/AsyncCall.cs
@@ -67,7 +67,7 @@
public void Initialize(Channel channel, CompletionQueueSafeHandle cq, string methodName, Timespec deadline)
{
this.channel = channel;
- var call = channel.Handle.CreateCall(channel.CompletionRegistry, cq, methodName, channel.Target, deadline);
+ var call = channel.Handle.CreateCall(channel.CompletionRegistry, cq, methodName, null, deadline);
channel.Environment.DebugStats.ActiveClientCalls.Increment();
InitializeInternal(call);
}
diff --git a/src/csharp/Grpc.Core/Internal/ChannelSafeHandle.cs b/src/csharp/Grpc.Core/Internal/ChannelSafeHandle.cs
index 20815ef..7324ebd 100644
--- a/src/csharp/Grpc.Core/Internal/ChannelSafeHandle.cs
+++ b/src/csharp/Grpc.Core/Internal/ChannelSafeHandle.cs
@@ -50,6 +50,16 @@
static extern CallSafeHandle grpcsharp_channel_create_call(ChannelSafeHandle channel, CompletionQueueSafeHandle cq, string method, string host, Timespec deadline);
[DllImport("grpc_csharp_ext.dll")]
+ static extern ChannelState grpcsharp_channel_check_connectivity_state(ChannelSafeHandle channel, int tryToConnect);
+
+ [DllImport("grpc_csharp_ext.dll")]
+ static extern void grpcsharp_channel_watch_connectivity_state(ChannelSafeHandle channel, ChannelState lastObservedState,
+ Timespec deadline, CompletionQueueSafeHandle cq, BatchContextSafeHandle ctx);
+
+ [DllImport("grpc_csharp_ext.dll")]
+ static extern CStringSafeHandle grpcsharp_channel_get_target(ChannelSafeHandle call);
+
+ [DllImport("grpc_csharp_ext.dll")]
static extern void grpcsharp_channel_destroy(IntPtr channel);
private ChannelSafeHandle()
@@ -73,6 +83,27 @@
return result;
}
+ public ChannelState CheckConnectivityState(bool tryToConnect)
+ {
+ return grpcsharp_channel_check_connectivity_state(this, tryToConnect ? 1 : 0);
+ }
+
+ public void WatchConnectivityState(ChannelState lastObservedState, Timespec deadline, CompletionQueueSafeHandle cq,
+ CompletionRegistry completionRegistry, BatchCompletionDelegate callback)
+ {
+ var ctx = BatchContextSafeHandle.Create();
+ completionRegistry.RegisterBatchCompletion(ctx, callback);
+ grpcsharp_channel_watch_connectivity_state(this, lastObservedState, deadline, cq, ctx);
+ }
+
+ public string GetTarget()
+ {
+ using (var cstring = grpcsharp_channel_get_target(this))
+ {
+ return cstring.GetValue();
+ }
+ }
+
protected override bool ReleaseHandle()
{
grpcsharp_channel_destroy(handle);
diff --git a/src/csharp/ext/grpc_csharp_ext.c b/src/csharp/ext/grpc_csharp_ext.c
index 49a0471..28c2791 100644
--- a/src/csharp/ext/grpc_csharp_ext.c
+++ b/src/csharp/ext/grpc_csharp_ext.c
@@ -382,6 +382,22 @@
return grpc_channel_create_call(channel, cq, method, host, deadline);
}
+GPR_EXPORT grpc_connectivity_state GPR_CALLTYPE
+grpcsharp_channel_check_connectivity_state(grpc_channel *channel, gpr_int32 try_to_connect) {
+ return grpc_channel_check_connectivity_state(channel, try_to_connect);
+}
+
+GPR_EXPORT void GPR_CALLTYPE grpcsharp_channel_watch_connectivity_state(
+ grpc_channel *channel, grpc_connectivity_state last_observed_state,
+ gpr_timespec deadline, grpc_completion_queue *cq, grpcsharp_batch_context *ctx) {
+ grpc_channel_watch_connectivity_state(channel, last_observed_state,
+ deadline, cq, ctx);
+}
+
+GPR_EXPORT char *GPR_CALLTYPE grpcsharp_channel_get_target(grpc_channel *channel) {
+ return grpc_channel_get_target(channel);
+}
+
/* Channel args */
GPR_EXPORT grpc_channel_args *GPR_CALLTYPE
diff --git a/tools/run_tests/run_tests.py b/tools/run_tests/run_tests.py
index fa74949..b93f584 100755
--- a/tools/run_tests/run_tests.py
+++ b/tools/run_tests/run_tests.py
@@ -254,6 +254,7 @@
def test_specs(self, config, travis):
assemblies = ['Grpc.Core.Tests',
'Grpc.Examples.Tests',
+ 'Grpc.HealthCheck.Tests',
'Grpc.IntegrationTesting']
if self.platform == 'windows':
cmd = 'tools\\run_tests\\run_csharp.bat'