diff options
author | Jan Tattermusch <jtattermusch@google.com> | 2015-07-22 12:51:06 -0700 |
---|---|---|
committer | Jan Tattermusch <jtattermusch@google.com> | 2015-08-04 11:30:00 -0700 |
commit | d8bbdeae42ce6ae6077e4e5b4b4f4c673acecf57 (patch) | |
tree | 75ca549460decd14b348376a8776c52ecece6d8b | |
parent | 0175e18133bf8cd74363ef08b8f3485038e3a36d (diff) |
Added channel state API
-rw-r--r-- | src/csharp/Grpc.Auth/OAuth2Interceptors.cs | 2 | ||||
-rw-r--r-- | src/csharp/Grpc.Core.Tests/ChannelTest.cs | 82 | ||||
-rw-r--r-- | src/csharp/Grpc.Core.Tests/ClientServerTest.cs | 24 | ||||
-rw-r--r-- | src/csharp/Grpc.Core.Tests/Grpc.Core.Tests.csproj | 1 | ||||
-rw-r--r-- | src/csharp/Grpc.Core.Tests/NUnitVersionTest.cs | 4 | ||||
-rw-r--r-- | src/csharp/Grpc.Core/Channel.cs | 74 | ||||
-rw-r--r-- | src/csharp/Grpc.Core/ChannelState.cs | 69 | ||||
-rw-r--r-- | src/csharp/Grpc.Core/Grpc.Core.csproj | 1 | ||||
-rw-r--r-- | src/csharp/Grpc.Core/Internal/ChannelSafeHandle.cs | 20 | ||||
-rw-r--r-- | src/csharp/ext/grpc_csharp_ext.c | 12 |
10 files changed, 280 insertions, 9 deletions
diff --git a/src/csharp/Grpc.Auth/OAuth2Interceptors.cs b/src/csharp/Grpc.Auth/OAuth2Interceptors.cs index c785ca5a16..cc9d2c175f 100644 --- a/src/csharp/Grpc.Auth/OAuth2Interceptors.cs +++ b/src/csharp/Grpc.Auth/OAuth2Interceptors.cs @@ -119,7 +119,5 @@ namespace Grpc.Auth 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 0000000000..bfe001b292 --- /dev/null +++ b/src/csharp/Grpc.Core.Tests/ChannelTest.cs @@ -0,0 +1,82 @@ +#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 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 540fe756c0..35924868ca 100644 --- a/src/csharp/Grpc.Core.Tests/ClientServerTest.cs +++ b/src/csharp/Grpc.Core.Tests/ClientServerTest.cs @@ -276,6 +276,30 @@ namespace Grpc.Core.Tests 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 242a60d098..f2bf459dc5 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 600df1a18d..3fa6ad09c0 100644 --- a/src/csharp/Grpc.Core.Tests/NUnitVersionTest.cs +++ b/src/csharp/Grpc.Core.Tests/NUnitVersionTest.cs @@ -70,10 +70,8 @@ namespace Grpc.Core.Tests [Test] public async Task NUnitVersionTest2() { - testRunCount ++; + testRunCount++; await Task.Delay(10); } - - } } diff --git a/src/csharp/Grpc.Core/Channel.cs b/src/csharp/Grpc.Core/Channel.cs index 18e6f2fda5..8a71afa01a 100644 --- a/src/csharp/Grpc.Core/Channel.cs +++ b/src/csharp/Grpc.Core/Channel.cs @@ -37,6 +37,8 @@ using System.Threading; using System.Threading.Tasks; using Grpc.Core.Internal; +using Grpc.Core.Logging; +using Grpc.Core.Utils; namespace Grpc.Core { @@ -45,6 +47,8 @@ namespace Grpc.Core /// </summary> public class Channel : IDisposable { + static readonly ILogger Logger = GrpcEnvironment.Logger.ForType<Channel>(); + readonly GrpcEnvironment environment; readonly ChannelSafeHandle handle; readonly List<ChannelOption> options; @@ -53,13 +57,14 @@ namespace Grpc.Core /// <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>(); @@ -82,8 +87,8 @@ namespace Grpc.Core /// <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,6 +96,67 @@ namespace Grpc.Core { } + /// <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> + /// 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); diff --git a/src/csharp/Grpc.Core/ChannelState.cs b/src/csharp/Grpc.Core/ChannelState.cs new file mode 100644 index 0000000000..d293b98f75 --- /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 940a6b8ac0..641b54baba 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/ChannelSafeHandle.cs b/src/csharp/Grpc.Core/Internal/ChannelSafeHandle.cs index 20815efbd3..c017560b56 100644 --- a/src/csharp/Grpc.Core/Internal/ChannelSafeHandle.cs +++ b/src/csharp/Grpc.Core/Internal/ChannelSafeHandle.cs @@ -50,6 +50,13 @@ namespace Grpc.Core.Internal 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 void grpcsharp_channel_destroy(IntPtr channel); private ChannelSafeHandle() @@ -73,6 +80,19 @@ namespace Grpc.Core.Internal 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); + } + 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 49a0471042..0e3a83d244 100644 --- a/src/csharp/ext/grpc_csharp_ext.c +++ b/src/csharp/ext/grpc_csharp_ext.c @@ -382,6 +382,18 @@ grpcsharp_channel_create_call(grpc_channel *channel, grpc_completion_queue *cq, 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, NULL, + deadline, cq, ctx); +} + /* Channel args */ GPR_EXPORT grpc_channel_args *GPR_CALLTYPE |