Merge pull request grpc#2756 from jtattermusch/csharp_channel_state

C# Improving Channel API
ksophocleous · Aug 4, 2015 · 8956e60 · 8956e60
2 parents 847dfff + 9fb1d20
commit 8956e60
Show file tree

Hide file tree

Showing 14 changed files with 321 additions and 41 deletions.
diff --git a/src/csharp/Grpc.Auth/OAuth2Interceptors.cs b/src/csharp/Grpc.Auth/OAuth2Interceptors.cs
@@ -119,7 +119,5 @@ public static Metadata.Entry CreateBearerTokenHeader(string accessToken)
                 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
@@ -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
@@ -276,6 +276,30 @@ public void PeerInfoPresent()
             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
@@ -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
@@ -70,10 +70,8 @@ public void NUnitVersionTest1()
         [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
@@ -134,7 +134,8 @@ public void DeadlineInThePast()
             }
             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 @@ public void DeadlineExceededStatusOnTimeout()
             }
             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 @@ public void ServerReceivesCancellationOnTimeout()
             }
             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
@@ -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 @@ 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;
-        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,35 +80,96 @@ public Channel(string host, Credentials credentials, IEnumerable<ChannelOption>
                     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) :
             this(string.Format("{0}:{1}", host, port), credentials, options)
         {
         }
 
-        public void Dispose()
+        /// <summary>
+        /// Gets current connectivity state of this channel.
+        /// </summary>
+        public ChannelState State
         {
-            Dispose(true);
-            GC.SuppressFinalize(this);
+            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;
         }
 
-        internal string Target
+        /// <summary> Address of the remote endpoint in URI format.</summary>
+        public string Target
         {
             get
             {
-                return target;
+                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 ChannelSafeHandle Handle
         {
             get
@@ -159,26 +224,5 @@ private static string GetUserAgentString()
             // 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
@@ -135,6 +135,9 @@ public static class ChannelOptions
         /// <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";
-Original file line number
+Diff line change
@@ Expand Up @@
                     return new Metadata.Entry(AuthorizationHeader, Schema + " " + accessToken);
                 }
             }
         }
     }