Merge github.com:grpc/grpc into error

10 files changed, 191 insertions, 22 deletions

diff --git a/src/cpp/server/server.cc b/src/cpp/server/server.cc
index fafe31e84c..f955a31494 100644
--- a/src/cpp/server/server.cc
+++ b/src/cpp/server/server.cc
@@ -33,6 +33,7 @@
 
 #include <grpc++/server.h>
 
+#include <sstream>
 #include <utility>
 
 #include <grpc++/completion_queue.h>
@@ -41,6 +42,7 @@
 #include <grpc++/impl/grpc_library.h>
 #include <grpc++/impl/method_handler_impl.h>
 #include <grpc++/impl/rpc_service_method.h>
+#include <grpc++/impl/server_initializer.h>
 #include <grpc++/impl/service_type.h>
 #include <grpc++/security/server_credentials.h>
 #include <grpc++/server_context.h>
@@ -284,7 +286,8 @@ Server::Server(ThreadPoolInterface* thread_pool, bool thread_pool_owned,
       has_generic_service_(false),
       server_(nullptr),
       thread_pool_(thread_pool),
-      thread_pool_owned_(thread_pool_owned) {
+      thread_pool_owned_(thread_pool_owned),
+      server_initializer_(new ServerInitializer(this)) {
   g_gli_initializer.summon();
   gpr_once_init(&g_once_init_callbacks, InitGlobalCallbacks);
   global_callbacks_ = g_callbacks;
@@ -341,6 +344,7 @@ bool Server::RegisterService(const grpc::string* host, Service* service) {
                "Can only register an asynchronous service against one server.");
     service->server_ = this;
   }
+  const char* method_name = nullptr;
   for (auto it = service->methods_.begin(); it != service->methods_.end();
        ++it) {
     if (it->get() == nullptr) {  // Handled by generic service if any.
@@ -360,6 +364,17 @@ bool Server::RegisterService(const grpc::string* host, Service* service) {
     } else {
       sync_methods_->emplace_back(method, tag);
     }
+    method_name = method->name();
+  }
+
+  // Parse service name.
+  if (method_name != nullptr) {
+    std::stringstream ss(method_name);
+    grpc::string service_name;
+    if (std::getline(ss, service_name, '/') &&
+        std::getline(ss, service_name, '/')) {
+      services_.push_back(service_name);
+    }
   }
   return true;
 }
@@ -598,4 +613,6 @@ void Server::RunRpc() {
   }
 }
 
+ServerInitializer* Server::initializer() { return server_initializer_.get(); }
+
 }  // namespace grpc
diff --git a/src/cpp/server/server_builder.cc b/src/cpp/server/server_builder.cc
index 68cc38258c..9658a56745 100644
--- a/src/cpp/server/server_builder.cc
+++ b/src/cpp/server/server_builder.cc
@@ -41,9 +41,23 @@
 
 namespace grpc {
 
+static std::vector<std::unique_ptr<ServerBuilderPlugin> (*)()>*
+    g_plugin_factory_list;
+static gpr_once once_init_plugin_list = GPR_ONCE_INIT;
+
+static void do_plugin_list_init(void) {
+  g_plugin_factory_list =
+      new std::vector<std::unique_ptr<ServerBuilderPlugin> (*)()>();
+}
+
 ServerBuilder::ServerBuilder()
     : max_message_size_(-1), generic_service_(nullptr) {
   grpc_compression_options_init(&compression_options_);
+  gpr_once_init(&once_init_plugin_list, do_plugin_list_init);
+  for (auto factory : (*g_plugin_factory_list)) {
+    std::unique_ptr<ServerBuilderPlugin> plugin = factory();
+    plugins_[plugin->name()] = std::move(plugin);
+  }
 }
 
 std::unique_ptr<ServerCompletionQueue> ServerBuilder::AddCompletionQueue() {
@@ -96,6 +110,15 @@ std::unique_ptr<Server> ServerBuilder::BuildAndStart() {
   ChannelArguments args;
   for (auto option = options_.begin(); option != options_.end(); ++option) {
     (*option)->UpdateArguments(&args);
+    (*option)->UpdatePlugins(&plugins_);
+  }
+  if (thread_pool == nullptr) {
+    for (auto plugin = plugins_.begin(); plugin != plugins_.end(); plugin++) {
+      if ((*plugin).second->has_sync_methods()) {
+        thread_pool.reset(CreateDefaultThreadPool());
+        break;
+      }
+    }
   }
   if (max_message_size_ > 0) {
     args.SetInt(GRPC_ARG_MAX_MESSAGE_LENGTH, max_message_size_);
@@ -104,6 +127,7 @@ std::unique_ptr<Server> ServerBuilder::BuildAndStart() {
               compression_options_.enabled_algorithms_bitset);
   std::unique_ptr<Server> server(
       new Server(thread_pool.release(), true, max_message_size_, &args));
+  ServerInitializer* initializer = server->initializer();
   for (auto cq = cqs_.begin(); cq != cqs_.end(); ++cq) {
     grpc_server_register_completion_queue(server->server_, (*cq)->cq(),
                                           nullptr);
@@ -114,6 +138,9 @@ std::unique_ptr<Server> ServerBuilder::BuildAndStart() {
       return nullptr;
     }
   }
+  for (auto plugin = plugins_.begin(); plugin != plugins_.end(); plugin++) {
+    (*plugin).second->InitServer(initializer);
+  }
   if (generic_service_) {
     server->RegisterAsyncGenericService(generic_service_);
   } else {
@@ -137,7 +164,16 @@ std::unique_ptr<Server> ServerBuilder::BuildAndStart() {
   if (!server->Start(cqs_data, cqs_.size())) {
     return nullptr;
   }
+  for (auto plugin = plugins_.begin(); plugin != plugins_.end(); plugin++) {
+    (*plugin).second->Finish(initializer);
+  }
   return server;
 }
 
+void ServerBuilder::InternalAddPluginFactory(
+    std::unique_ptr<ServerBuilderPlugin> (*CreatePlugin)()) {
+  gpr_once_init(&once_init_plugin_list, do_plugin_list_init);
+  (*g_plugin_factory_list).push_back(CreatePlugin);
+}
+
 }  // namespace grpc
diff --git a/src/csharp/Grpc.Core.Tests/ChannelTest.cs b/src/csharp/Grpc.Core.Tests/ChannelTest.cs
index 6330f50fae..850d70ce92 100644
--- a/src/csharp/Grpc.Core.Tests/ChannelTest.cs
+++ b/src/csharp/Grpc.Core.Tests/ChannelTest.cs
@@ -32,6 +32,7 @@
 #endregion
 
 using System;
+using System.Threading.Tasks;
 using Grpc.Core;
 using Grpc.Core.Internal;
 using Grpc.Core.Utils;
@@ -89,5 +90,43 @@ namespace Grpc.Core.Tests
             channel.ShutdownAsync().Wait();
             Assert.ThrowsAsync(typeof(InvalidOperationException), async () => await channel.ShutdownAsync());
         }
+
+        [Test]
+        public async Task ShutdownTokenCancelledAfterShutdown()
+        {
+            var channel = new Channel("localhost", ChannelCredentials.Insecure);
+            Assert.IsFalse(channel.ShutdownToken.IsCancellationRequested);
+            var shutdownTask = channel.ShutdownAsync();
+            Assert.IsTrue(channel.ShutdownToken.IsCancellationRequested);
+            await shutdownTask;
+        }
+
+        [Test]
+        public async Task StateIsFatalFailureAfterShutdown()
+        {
+            var channel = new Channel("localhost", ChannelCredentials.Insecure);
+            await channel.ShutdownAsync();
+            Assert.AreEqual(ChannelState.FatalFailure, channel.State);
+        }
+
+        [Test]
+        public async Task ShutdownFinishesWaitForStateChangedAsync()
+        {
+            var channel = new Channel("localhost", ChannelCredentials.Insecure);
+            var stateChangedTask = channel.WaitForStateChangedAsync(ChannelState.Idle);
+            var shutdownTask = channel.ShutdownAsync();
+            await stateChangedTask;
+            await shutdownTask;
+        }
+
+        [Test]
+        public async Task OperationsThrowAfterShutdown()
+        {
+            var channel = new Channel("localhost", ChannelCredentials.Insecure);
+            await channel.ShutdownAsync();
+            Assert.ThrowsAsync(typeof(ObjectDisposedException), async () => await channel.WaitForStateChangedAsync(ChannelState.Idle));
+            Assert.Throws(typeof(ObjectDisposedException), () => { var x = channel.ResolvedTarget; });
+            Assert.ThrowsAsync(typeof(TaskCanceledException), async () => await channel.ConnectAsync());
+        }
     }
 }
diff --git a/src/csharp/Grpc.Core/Channel.cs b/src/csharp/Grpc.Core/Channel.cs
index 89981b1849..93a6e6a3d9 100644
--- a/src/csharp/Grpc.Core/Channel.cs
+++ b/src/csharp/Grpc.Core/Channel.cs
@@ -32,6 +32,7 @@
 using System;
 using System.Collections.Generic;
 using System.Linq;
+using System.Threading;
 using System.Threading.Tasks;
 
 using Grpc.Core.Internal;
@@ -51,6 +52,7 @@ namespace Grpc.Core
 
         readonly object myLock = new object();
         readonly AtomicCounter activeCallCounter = new AtomicCounter();
+        readonly CancellationTokenSource shutdownTokenSource = new CancellationTokenSource();
 
         readonly string target;
         readonly GrpcEnvironment environment;
@@ -101,12 +103,13 @@ namespace Grpc.Core
 
         /// <summary>
         /// Gets current connectivity state of this channel.
+        /// After channel is has been shutdown, <c>ChannelState.FatalFailure</c> will be returned.
         /// </summary>
         public ChannelState State
         {
             get
             {
-                return handle.CheckConnectivityState(false);        
+                return GetConnectivityState(false);
             }
         }
 
@@ -155,6 +158,17 @@ namespace Grpc.Core
         }
 
         /// <summary>
+        /// Returns a token that gets cancelled once <c>ShutdownAsync</c> is invoked.
+        /// </summary>
+        public CancellationToken ShutdownToken
+        {
+            get
+            {
+                return this.shutdownTokenSource.Token;
+            }
+        }
+
+        /// <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.
@@ -164,7 +178,7 @@ namespace Grpc.Core
         /// <param name="deadline">The deadline. <c>null</c> indicates no deadline.</param>
         public async Task ConnectAsync(DateTime? deadline = null)
         {
-            var currentState = handle.CheckConnectivityState(true);
+            var currentState = GetConnectivityState(true);
             while (currentState != ChannelState.Ready)
             {
                 if (currentState == ChannelState.FatalFailure)
@@ -172,7 +186,7 @@ namespace Grpc.Core
                     throw new OperationCanceledException("Channel has reached FatalFailure state.");
                 }
                 await WaitForStateChangedAsync(currentState, deadline).ConfigureAwait(false);
-                currentState = handle.CheckConnectivityState(false);
+                currentState = GetConnectivityState(false);
             }
         }
 
@@ -188,6 +202,8 @@ namespace Grpc.Core
                 shutdownRequested = true;
             }
 
+            shutdownTokenSource.Cancel();
+
             var activeCallCount = activeCallCounter.Count;
             if (activeCallCount > 0)
             {
@@ -231,6 +247,18 @@ namespace Grpc.Core
             activeCallCounter.Decrement();
         }
 
+        private ChannelState GetConnectivityState(bool tryToConnect)
+        {
+            try
+            {
+                return handle.CheckConnectivityState(tryToConnect);
+            }
+            catch (ObjectDisposedException)
+            {
+                return ChannelState.FatalFailure;
+            }
+        }
+
         private static void EnsureUserAgentChannelOption(Dictionary<string, ChannelOption> options)
         {
             var key = ChannelOptions.PrimaryUserAgentString;
diff --git a/src/csharp/Grpc.Examples.Tests/MathClientServerTests.cs b/src/csharp/Grpc.Examples.Tests/MathClientServerTests.cs
index 875202b950..ee11105efe 100644
--- a/src/csharp/Grpc.Examples.Tests/MathClientServerTests.cs
+++ b/src/csharp/Grpc.Examples.Tests/MathClientServerTests.cs
@@ -92,7 +92,7 @@ namespace Math.Tests
         public void DivByZero()
         {
             var ex = Assert.Throws<RpcException>(() => client.Div(new DivArgs { Dividend = 0, Divisor = 0 }));
-            Assert.AreEqual(StatusCode.Unknown, ex.Status.StatusCode);
+            Assert.AreEqual(StatusCode.InvalidArgument, ex.Status.StatusCode);
         }
 
         [Test]
diff --git a/src/csharp/Grpc.Examples/MathExamples.cs b/src/csharp/Grpc.Examples/MathExamples.cs
index 6075420974..d260830b94 100644
--- a/src/csharp/Grpc.Examples/MathExamples.cs
+++ b/src/csharp/Grpc.Examples/MathExamples.cs
@@ -32,6 +32,7 @@
 using System;
 using System.Collections.Generic;
 using System.Threading.Tasks;
+using Grpc.Core;
 using Grpc.Core.Utils;
 
 namespace Math
@@ -109,5 +110,42 @@ namespace Math
             DivReply result = await client.DivAsync(new DivArgs { Dividend = sum.Num_, Divisor = numbers.Count });
             Console.WriteLine("Avg Result: " + result);
         }
+
+        /// <summary>
+        /// Shows how to handle a call ending with non-OK status.
+        /// </summary>
+        public static async Task HandleErrorExample(Math.MathClient client)
+        {
+            try
+            {
+                 DivReply result = await client.DivAsync(new DivArgs { Dividend = 5, Divisor = 0 });
+            }
+            catch (RpcException ex)
+            {
+                Console.WriteLine(string.Format("RPC ended with status {0}", ex.Status));
+            }
+        }
+
+        /// <summary>
+        /// Shows how to send request headers and how to access response headers
+        /// and response trailers.
+        /// </summary>
+        public static async Task MetadataExample(Math.MathClient client)
+        {
+            var requestHeaders = new Metadata
+            {
+                { "custom-header", "custom-value" }
+            };
+
+            var call = client.DivAsync(new DivArgs { Dividend = 5, Divisor = 0 }, requestHeaders);
+
+            // Get response headers
+            Metadata responseHeaders = await call.ResponseHeadersAsync;
+
+            var result = await call;
+
+            // Get response trailers after the call has finished.
+            Metadata responseTrailers = call.GetTrailers();
+        }
     }
 }
diff --git a/src/csharp/Grpc.Examples/MathServiceImpl.cs b/src/csharp/Grpc.Examples/MathServiceImpl.cs
index 79c56e57a8..a28020f62f 100644
--- a/src/csharp/Grpc.Examples/MathServiceImpl.cs
+++ b/src/csharp/Grpc.Examples/MathServiceImpl.cs
@@ -52,23 +52,15 @@ namespace Math
 
         public override async Task Fib(FibArgs request, IServerStreamWriter<Num> responseStream, ServerCallContext context)
         {
-            if (request.Limit <= 0)
-            {
-                // keep streaming the sequence until cancelled.
-                IEnumerator<Num> fibEnumerator = FibInternal(long.MaxValue).GetEnumerator();
-                while (!context.CancellationToken.IsCancellationRequested && fibEnumerator.MoveNext())
-                {
-                    await responseStream.WriteAsync(fibEnumerator.Current);
-                    await Task.Delay(100);
-                }
-            }
+            var limit = request.Limit > 0 ? request.Limit : long.MaxValue;
+            var fibEnumerator = FibInternal(limit).GetEnumerator();
 
-            if (request.Limit > 0)
+            // Keep streaming the sequence until the call is cancelled.
+            // Use CancellationToken from ServerCallContext to detect the cancellation.
+            while (!context.CancellationToken.IsCancellationRequested && fibEnumerator.MoveNext())
             {
-                foreach (var num in FibInternal(request.Limit))
-                {
-                    await responseStream.WriteAsync(num);
-                }
+                await responseStream.WriteAsync(fibEnumerator.Current);
+                await Task.Delay(100);
             }
         }
 
@@ -89,6 +81,13 @@ namespace Math
 
         static DivReply DivInternal(DivArgs args)
         {
+            if (args.Divisor == 0)
+            {
+                // One can finish the RPC with non-ok status by throwing RpcException instance.
+                // Alternatively, resulting status can be set using ServerCallContext.Status
+                throw new RpcException(new Status(StatusCode.InvalidArgument, "Division by zero"));
+            }
+
             long quotient = args.Dividend / args.Divisor;
             long remainder = args.Dividend % args.Divisor;
             return new DivReply { Quotient = quotient, Remainder = remainder };
diff --git a/src/csharp/Grpc.IntegrationTesting/StressTestClient.cs b/src/csharp/Grpc.IntegrationTesting/StressTestClient.cs
index 8db691cb04..4d6ca7ece5 100644
--- a/src/csharp/Grpc.IntegrationTesting/StressTestClient.cs
+++ b/src/csharp/Grpc.IntegrationTesting/StressTestClient.cs
@@ -311,7 +311,7 @@ namespace Grpc.IntegrationTesting
                 var snapshot = histogram.GetSnapshot(true);
                 var elapsedSnapshot = wallClockStopwatch.GetElapsedSnapshot(true);
 
-                return (long) (snapshot.Count / elapsedSnapshot.Seconds);
+                return (long) (snapshot.Count / elapsedSnapshot.TotalSeconds);
             }
         }
     }
diff --git a/src/proto/grpc/testing/echo_messages.proto b/src/proto/grpc/testing/echo_messages.proto
index 1be1966f10..b405acf043 100644
--- a/src/proto/grpc/testing/echo_messages.proto
+++ b/src/proto/grpc/testing/echo_messages.proto
@@ -32,6 +32,12 @@ syntax = "proto3";
 
 package grpc.testing;
 
+// Message to be echoed back serialized in trailer.
+message DebugInfo {
+  repeated string stack_entries = 1;
+  string detail = 2;
+}
+
 message RequestParams {
   bool echo_deadline = 1;
   int32 client_cancel_after_us = 2;
@@ -43,6 +49,7 @@ message RequestParams {
   string expected_client_identity = 8; // will force check_auth_context.
   bool skip_cancelled_check = 9;
   string expected_transport_security_type = 10;
+  DebugInfo debug_info = 11;
 }
 
 message EchoRequest {
diff --git a/src/ruby/lib/grpc/generic/client_stub.rb b/src/ruby/lib/grpc/generic/client_stub.rb
index 68e167a69f..12946fe819 100644
--- a/src/ruby/lib/grpc/generic/client_stub.rb
+++ b/src/ruby/lib/grpc/generic/client_stub.rb
@@ -49,7 +49,12 @@ module GRPC
         fail(TypeError, '!Channel') unless alt_chan.is_a?(Core::Channel)
         return alt_chan
       end
-      kw['grpc.primary_user_agent'] = "grpc-ruby/#{VERSION}"
+      if kw['grpc.primary_user_agent'].nil?
+        kw['grpc.primary_user_agent'] = ''
+      else
+        kw['grpc.primary_user_agent'] += ' '
+      end
+      kw['grpc.primary_user_agent'] += "grpc-ruby/#{VERSION}"
       unless creds.is_a?(Core::ChannelCredentials) || creds.is_a?(Symbol)
         fail(TypeError, '!ChannelCredentials or Symbol')
       end