Unary client supports timeout

CMakeLists.txt

@@ -62,6 +62,7 @@ set(ALL_LIBRARY_SRCS
     async_grpc/token_file_credentials.cc)
 
 set(ALL_TESTS
+    async_grpc/client_test.cc
     async_grpc/server_test.cc
     async_grpc/type_traits_test.cc)
 

async_grpc/BUILD.bazel

@@ -71,6 +71,7 @@ cc_library(
 
 [cc_test(
     name = src.replace("/", "_").replace(".cc", ""),
+    size = "small",
     srcs = [src],
     deps = [
         ":async_grpc",

async_grpc/client.h

@@ -17,6 +17,7 @@
 #ifndef CPP_GRPC_CLIENT_H
 #define CPP_GRPC_CLIENT_H
 
+#include "async_grpc/common/optional.h"
 #include "async_grpc/retry.h"
 #include "async_grpc/rpc_handler_interface.h"
 #include "async_grpc/rpc_service_method_traits.h"
@@ -47,30 +48,37 @@ class Client<RpcServiceMethodConcept, ::grpc::internal::RpcMethod::NORMAL_RPC> {
   using ResponseType = typename RpcServiceMethod::ResponseType;
 
  public:
-  Client(std::shared_ptr<::grpc::Channel> channel, RetryStrategy retry_strategy)
+  Client(std::shared_ptr<::grpc::Channel> channel)
       : channel_(channel),
         client_context_(common::make_unique<::grpc::ClientContext>()),
         rpc_method_name_(RpcServiceMethod::MethodName()),
         rpc_method_(rpc_method_name_.c_str(), RpcServiceMethod::StreamType,
-                    channel_),
-        retry_strategy_(retry_strategy) {}
+                    channel_) {}
 
-  Client(std::shared_ptr<::grpc::Channel> channel)
+  // 'timeout' is used for every 'Write' separately, but multiple retries count
+  // towards a single timeout.
+  Client(std::shared_ptr<::grpc::Channel> channel, common::Duration timeout,
+         RetryStrategy retry_strategy = nullptr)
       : channel_(channel),
         client_context_(common::make_unique<::grpc::ClientContext>()),
         rpc_method_name_(RpcServiceMethod::MethodName()),
         rpc_method_(rpc_method_name_.c_str(), RpcServiceMethod::StreamType,
-                    channel_) {}
+                    channel_),
+        timeout_(timeout),
+        retry_strategy_(retry_strategy) {}
 
   bool Write(const RequestType& request, ::grpc::Status* status = nullptr) {
     ::grpc::Status internal_status;
+    if (timeout_.has_value()) {
+      deadline_ = std::chrono::system_clock::now() + timeout_.value();
+    }
+    Reset();
     bool result = RetryWithStrategy(retry_strategy_,
                                     [this, &request, &internal_status] {
                                       WriteImpl(request, &internal_status);
                                       return internal_status;
                                     },
                                     [this] { Reset(); });
-
     if (status != nullptr) {
       *status = internal_status;
     }
@@ -82,6 +90,9 @@ class Client<RpcServiceMethodConcept, ::grpc::internal::RpcMethod::NORMAL_RPC> {
  private:
   void Reset() {
     client_context_ = common::make_unique<::grpc::ClientContext>();
+    if (deadline_.has_value()) {
+      client_context_->set_deadline(deadline_.value());
+    }
   }
 
   bool WriteImpl(const RequestType& request, ::grpc::Status* status) {
@@ -101,6 +112,8 @@ class Client<RpcServiceMethodConcept, ::grpc::internal::RpcMethod::NORMAL_RPC> {
   std::unique_ptr<::grpc::ClientContext> client_context_;
   const std::string rpc_method_name_;
   const ::grpc::internal::RpcMethod rpc_method_;
+  common::optional<common::Duration> timeout_;
+  common::optional<std::chrono::system_clock::time_point> deadline_;
 
   ResponseType response_;
   RetryStrategy retry_strategy_;
@@ -143,10 +156,6 @@ class Client<RpcServiceMethodConcept,
   const ResponseType& response() { return response_; }
 
  private:
-  void Reset() {
-    client_context_ = common::make_unique<::grpc::ClientContext>();
-  }
-
   bool WriteImpl(const RequestType& request, ::grpc::Status* status) {
     InstantiateClientWriterIfNeeded();
     return client_writer_->Write(request);
@@ -204,10 +213,6 @@ class Client<RpcServiceMethodConcept,
   }
 
  private:
-  void Reset() {
-    client_context_ = common::make_unique<::grpc::ClientContext>();
-  }
-
   bool WriteImpl(const RequestType& request, ::grpc::Status* status) {
     InstantiateClientReader(request);
     return true;
@@ -267,10 +272,6 @@ class Client<RpcServiceMethodConcept,
   }
 
  private:
-  void Reset() {
-    client_context_ = common::make_unique<::grpc::ClientContext>();
-  }
-
   bool WriteImpl(const RequestType& request, ::grpc::Status* status) {
     InstantiateClientReaderWriterIfNeeded();
     return client_reader_writer_->Write(request);

async_grpc/client_test.cc

@@ -0,0 +1,61 @@
+/*
+ * Copyright 2018 The Cartographer Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include "async_grpc/client.h"
+
+#include "async_grpc/proto/math_service.pb.h"
+#include "async_grpc/retry.h"
+#include "glog/logging.h"
+#include "grpc++/grpc++.h"
+#include "gtest/gtest.h"
+
+namespace async_grpc {
+namespace {
+
+struct GetEchoMethod {
+  static constexpr const char* MethodName() {
+    return "/async_grpc.proto.Math/GetEcho";
+  }
+  using IncomingType = proto::GetEchoRequest;
+  using OutgoingType = proto::GetEchoResponse;
+};
+
+const char* kWrongAddress = "wrong-domain-does-not-exist:50051";
+
+TEST(ClientTest, TimesOut) {
+  auto client_channel = ::grpc::CreateChannel(
+      kWrongAddress, ::grpc::InsecureChannelCredentials());
+  Client<GetEchoMethod> client(client_channel, common::FromSeconds(0.1));
+  proto::GetEchoRequest request;
+  grpc::Status status;
+  EXPECT_FALSE(client.Write(request, &status));
+  EXPECT_EQ(status.error_code(), grpc::StatusCode::DEADLINE_EXCEEDED);
+}
+
+TEST(ClientTest, TimesOutWithRetries) {
+  auto client_channel = ::grpc::CreateChannel(
+      kWrongAddress, ::grpc::InsecureChannelCredentials());
+  Client<GetEchoMethod> client(
+      client_channel, common::FromSeconds(0.5),
+      CreateLimitedBackoffStrategy(common::FromSeconds(0.1), 1, 3));
+  proto::GetEchoRequest request;
+  grpc::Status status;
+  EXPECT_FALSE(client.Write(request, &status));
+  EXPECT_EQ(status.error_code(), grpc::StatusCode::DEADLINE_EXCEEDED);
+}
+
+}  // namespace
+}  // namespace async_grpc

async_grpc/server_test.cc

@@ -285,8 +285,9 @@ TEST_F(ServerTest, RetryWithUnrecoverableError) {
   server_->Start();
 
   Client<GetSquareMethod> client(
-      client_channel_, CreateUnlimitedConstantDelayStrategy(
-                           common::FromSeconds(1), {::grpc::INTERNAL}));
+      client_channel_, common::FromSeconds(5),
+      CreateUnlimitedConstantDelayStrategy(common::FromSeconds(1),
+                                           {::grpc::INTERNAL}));
   proto::GetSquareRequest request;
   request.set_input(-11);
   EXPECT_FALSE(client.Write(request));