Add limited constant retry strategy.

async_grpc/client.h

@@ -21,14 +21,12 @@
 #include "async_grpc/retry.h"
 #include "async_grpc/rpc_handler_interface.h"
 #include "async_grpc/rpc_service_method_traits.h"
-
+#include "glog/logging.h"
 #include "grpc++/grpc++.h"
 #include "grpc++/impl/codegen/client_unary_call.h"
 #include "grpc++/impl/codegen/proto_utils.h"
 #include "grpc++/impl/codegen/sync_stream.h"
 
-#include "glog/logging.h"
-
 namespace async_grpc {
 
 // Wraps a method invocation for all rpc types, unary, client streaming,

async_grpc/completion_queue_pool.cc

@@ -14,10 +14,11 @@
  * limitations under the License.
  */
 
+#include "async_grpc/completion_queue_pool.h"
+
 #include <cstdlib>
 
 #include "async_grpc/async_client.h"
-#include "async_grpc/completion_queue_pool.h"
 #include "common/make_unique.h"
 #include "glog/logging.h"
 
@@ -89,8 +90,7 @@ void CompletionQueuePool::Shutdown() {
 }
 
 CompletionQueuePool::CompletionQueuePool()
-    : number_completion_queues_(kDefaultNumberCompletionQueues) {
-}
+    : number_completion_queues_(kDefaultNumberCompletionQueues) {}
 
 CompletionQueuePool::~CompletionQueuePool() {
   LOG(INFO) << "~CompletionQueuePool";

async_grpc/completion_queue_thread.h

@@ -18,6 +18,7 @@
 #define CPP_GRPC_COMMON_COMPLETION_QUEUE_THREAD_H_
 
 #include <grpc++/grpc++.h>
+
 #include <memory>
 #include <thread>
 

async_grpc/opencensus_span.cc

@@ -37,7 +37,8 @@ void OpencensusSpan::End() { span_.End(); }
 
 OpencensusSpan::OpencensusSpan(const std::string& name,
                                const OpencensusSpan* parent)
-    : span_(opencensus::trace::Span::StartSpan(name, parent ? &parent->span_: nullptr)) {}
+    : span_(opencensus::trace::Span::StartSpan(
+          name, parent ? &parent->span_ : nullptr)) {}
 
 }  // namespace async_grpc
 

async_grpc/retry.cc

@@ -14,19 +14,20 @@
  * limitations under the License.
  */
 
+#include "async_grpc/retry.h"
+
 #include <chrono>
 #include <cmath>
 #include <thread>
 
-#include "async_grpc/retry.h"
 #include "glog/logging.h"
 
 namespace async_grpc {
 
 RetryStrategy CreateRetryStrategy(RetryIndicator retry_indicator,
                                   RetryDelayCalculator retry_delay_calculator) {
   return [retry_indicator, retry_delay_calculator](
-      int failed_attempts, const ::grpc::Status &status) {
+             int failed_attempts, const ::grpc::Status &status) {
     if (!retry_indicator(failed_attempts, status)) {
       return optional<Duration>();
     }
@@ -40,6 +41,15 @@ RetryIndicator CreateLimitedRetryIndicator(int max_attempts) {
   };
 }
 
+RetryIndicator CreateLimitedRetryIndicator(
+    int max_attempts, const std::set<::grpc::StatusCode> &unrecoverable_codes) {
+  return [max_attempts, unrecoverable_codes](int failed_attempts,
+                                             const ::grpc::Status &status) {
+    return (failed_attempts < max_attempts) &&
+           !unrecoverable_codes.count(status.error_code());
+  };
+}
+
 RetryIndicator CreateUnlimitedRetryIndicator() {
   return [](int failed_attempts, const ::grpc::Status &status) { return true; };
 }
@@ -48,7 +58,7 @@ RetryIndicator CreateUnlimitedRetryIndicator(
     const std::set<::grpc::StatusCode> &unrecoverable_codes) {
   return
       [unrecoverable_codes](int failed_attempts, const ::grpc::Status &status) {
-        return unrecoverable_codes.count(status.error_code()) <= 0;
+        return !unrecoverable_codes.count(status.error_code());
       };
 }
 
@@ -80,6 +90,20 @@ RetryStrategy CreateUnlimitedConstantDelayStrategy(Duration delay) {
                              CreateConstantDelayCalculator(delay));
 }
 
+RetryStrategy CreateLimitedConstantDelayStrategy(Duration delay,
+                                                 int max_attempts) {
+  return CreateRetryStrategy(CreateLimitedRetryIndicator(max_attempts),
+                             CreateConstantDelayCalculator(delay));
+}
+
+RetryStrategy CreateLimitedConstantDelayStrategy(
+    Duration delay, int max_attempts,
+    const std::set<::grpc::StatusCode> &unrecoverable_codes) {
+  return CreateRetryStrategy(
+      CreateLimitedRetryIndicator(max_attempts, unrecoverable_codes),
+      CreateConstantDelayCalculator(delay));
+}
+
 RetryStrategy CreateUnlimitedConstantDelayStrategy(
     Duration delay, const std::set<::grpc::StatusCode> &unrecoverable_codes) {
   return CreateRetryStrategy(CreateUnlimitedRetryIndicator(unrecoverable_codes),

async_grpc/retry.h

@@ -38,6 +38,8 @@ RetryStrategy CreateRetryStrategy(RetryIndicator retry_indicator,
                                   RetryDelayCalculator retry_delay_calculator);
 
 RetryIndicator CreateLimitedRetryIndicator(int max_attempts);
+RetryIndicator CreateLimitedRetryIndicator(
+    int max_attempts, const std::set<::grpc::StatusCode> &unrecoverable_codes);
 RetryIndicator CreateUnlimitedRetryIndicator();
 RetryIndicator CreateUnlimitedRetryIndicator(
     const std::set<::grpc::StatusCode> &unrecoverable_codes);
@@ -47,6 +49,11 @@ RetryDelayCalculator CreateConstantDelayCalculator(Duration delay);
 RetryStrategy CreateLimitedBackoffStrategy(Duration min_delay,
                                            float backoff_factor,
                                            int max_attempts);
+RetryStrategy CreateLimitedConstantDelayStrategy(Duration delay,
+                                                 int max_attempts);
+RetryStrategy CreateLimitedConstantDelayStrategy(
+    Duration delay, int max_attempts,
+    const std::set<::grpc::StatusCode> &unrecoverable_codes);
 RetryStrategy CreateUnlimitedConstantDelayStrategy(Duration delay);
 RetryStrategy CreateUnlimitedConstantDelayStrategy(
     Duration delay, const std::set<::grpc::StatusCode> &unrecoverable_codes);

async_grpc/rpc.cc

@@ -15,9 +15,9 @@
  */
 
 #include "async_grpc/rpc.h"
-#include "async_grpc/service.h"
 
 #include "async_grpc/common/make_unique.h"
+#include "async_grpc/service.h"
 #include "glog/logging.h"
 
 namespace async_grpc {

async_grpc/server.cc

@@ -46,12 +46,14 @@ void Server::Builder::SetServerAddress(const std::string& server_address) {
 }
 
 void Server::Builder::SetMaxReceiveMessageSize(int max_receive_message_size) {
-  CHECK_GT(max_receive_message_size, 0) << "max_receive_message_size must be larger than 0.";
+  CHECK_GT(max_receive_message_size, 0)
+      << "max_receive_message_size must be larger than 0.";
   options_.max_receive_message_size = max_receive_message_size;
 }
 
 void Server::Builder::SetMaxSendMessageSize(int max_send_message_size) {
-  CHECK_GT(max_send_message_size, 0) << "max_send_message_size must be larger than 0.";
+  CHECK_GT(max_send_message_size, 0)
+      << "max_send_message_size must be larger than 0.";
   options_.max_send_message_size = max_send_message_size;
 }
 
@@ -63,19 +65,19 @@ void Server::Builder::EnableTracing() {
 #endif
 }
 
-void Server::Builder::DisableTracing() {
-  options_.enable_tracing = false;
-}
+void Server::Builder::DisableTracing() { options_.enable_tracing = false; }
 
-void Server::Builder::SetTracingSamplerProbability(double tracing_sampler_probability) {
+void Server::Builder::SetTracingSamplerProbability(
+    double tracing_sampler_probability) {
   options_.tracing_sampler_probability = tracing_sampler_probability;
 }
 
 void Server::Builder::SetTracingTaskName(const std::string& tracing_task_name) {
   options_.tracing_task_name = tracing_task_name;
 }
 
-void Server::Builder::SetTracingGcpProjectId(const std::string& tracing_gcp_project_id) {
+void Server::Builder::SetTracingGcpProjectId(
+    const std::string& tracing_gcp_project_id) {
   options_.tracing_gcp_project_id = tracing_gcp_project_id;
 }
 
@@ -178,7 +180,6 @@ void Server::Start() {
   }
 #endif
 
-
   // Start the gRPC server process.
   server_ = server_builder_.BuildAndStart();
 

async_grpc/server.h

@@ -30,13 +30,12 @@
 #include "async_grpc/rpc_handler.h"
 #include "async_grpc/rpc_service_method_traits.h"
 #include "async_grpc/service.h"
-
 #include "grpc++/grpc++.h"
 
 namespace async_grpc {
 namespace {
 
-constexpr int kDefaultMaxMessageSize = 10 * 1024 * 1024;  // 10 MB
+constexpr int kDefaultMaxMessageSize = 10 * 1024 * 1024;    // 10 MB
 constexpr double kDefaultTracingSamplerProbability = 0.01;  // 1 Percent
 
 }  // namespace

async_grpc/service.cc

@@ -14,10 +14,9 @@
  * limitations under the License.
  */
 
-#include "async_grpc/server.h"
-
 #include <cstdlib>
 
+#include "async_grpc/server.h"
 #include "glog/logging.h"
 #include "grpc++/impl/codegen/proto_utils.h"