Added retry policy logic and a couple implementations

Author: Michael Penick

include/cassandra.h

@@ -377,6 +377,7 @@ typedef struct CassMetrics_ {
 } CassMetrics;
 
 typedef enum CassConsistency_ {
+  CASS_CONSISTENCY_UNKNOWN      = 0xFFFF,
   CASS_CONSISTENCY_ANY          = 0x0000,
   CASS_CONSISTENCY_ONE          = 0x0001,
   CASS_CONSISTENCY_TWO          = 0x0002,

src/auth_requests.cpp

@@ -18,7 +18,7 @@
 
 namespace cass {
 
-int CredentialsRequest::encode(int version, BufferVec* bufs, EncodingCache* cache) const {
+int CredentialsRequest::encode(int version, Handler* handler, BufferVec* bufs) const {
   if (version != 1) {
     return -1;
   }
@@ -39,7 +39,7 @@ int CredentialsRequest::encode(int version, BufferVec* bufs, EncodingCache* cach
   return length;
 }
 
-int AuthResponseRequest::encode(int version, BufferVec* bufs, EncodingCache* cache) const {
+int AuthResponseRequest::encode(int version, Handler* handler, BufferVec* bufs) const {
   if (version < 2) {
     return -1;
   }

src/auth_requests.hpp

@@ -31,7 +31,7 @@ class CredentialsRequest : public Request {
     , credentials_(credentials) {}
 
 private:
-  int encode(int version, BufferVec* bufs, EncodingCache* cache) const;
+  int encode(int version, Handler* handler, BufferVec* bufs) const;
 
 private:
   V1Authenticator::Credentials credentials_;
@@ -47,7 +47,7 @@ class AuthResponseRequest : public Request {
   ScopedPtr<Authenticator>& auth() { return auth_; }
 
 private:
-  int encode(int version, BufferVec* bufs, EncodingCache* cache) const;
+  int encode(int version, Handler* handler, BufferVec* bufs) const;
 
 private:
   std::string token_;

src/batch_request.cpp

@@ -49,7 +49,7 @@ CassError cass_batch_add_statement(CassBatch* batch, CassStatement* statement) {
 
 namespace cass {
 
-int BatchRequest::encode(int version, BufferVec* bufs, EncodingCache* cache) const {
+int BatchRequest::encode(int version, Handler* handler, BufferVec* bufs) const {
   int length = 0;
   uint8_t flags = 0;
 
@@ -76,7 +76,7 @@ int BatchRequest::encode(int version, BufferVec* bufs, EncodingCache* cache) con
     if (statement->has_names_for_values()) {
       return ENCODE_ERROR_BATCH_WITH_NAMED_VALUES;
     }
-    int32_t result = (*i)->encode_batch(version, bufs, cache);
+    int32_t result = (*i)->encode_batch(version, bufs, handler->encoding_cache());
     if (result < 0) {
       return result;
     }

src/batch_request.hpp

@@ -55,7 +55,7 @@ class BatchRequest : public RoutableRequest {
   virtual bool get_routing_key(std::string* routing_key, EncodingCache* cache) const;
 
 private:
-  int encode(int version, BufferVec* bufs, EncodingCache* cache) const;
+  int encode(int version, Handler* handler, BufferVec* bufs) const;
 
 private:
   typedef std::map<std::string, ExecuteRequest*> PreparedMap;

src/config.hpp

@@ -21,6 +21,7 @@
 #include "cassandra.h"
 #include "dc_aware_policy.hpp"
 #include "latency_aware_policy.hpp"
+#include "retry_policy.hpp"
 #include "ssl.hpp"
 #include "token_aware_policy.hpp"
 
@@ -63,7 +64,8 @@ class Config {
       , latency_aware_routing_(false)
       , tcp_nodelay_enable_(false)
       , tcp_keepalive_enable_(false)
-      , tcp_keepalive_delay_secs_(0) {}
+      , tcp_keepalive_delay_secs_(0)
+      , retry_policy_(new DefaultRetryPolicy()) { }
 
   unsigned thread_count_io() const { return thread_count_io_; }
 
@@ -270,6 +272,15 @@ class Config {
     tcp_keepalive_delay_secs_ = delay_secs;
   }
 
+  RetryPolicy* retry_policy() const {
+    return retry_policy_.get();
+  }
+
+  void set_retry_policy(RetryPolicy* retry_policy) {
+    if (retry_policy == NULL) return;
+    retry_policy_.reset(retry_policy);
+  }
+
 private:
   int port_;
   int protocol_version_;
@@ -302,6 +313,7 @@ class Config {
   bool tcp_nodelay_enable_;
   bool tcp_keepalive_enable_;
   unsigned tcp_keepalive_delay_secs_;
+  SharedRefPtr<RetryPolicy> retry_policy_;
 };
 
 } // namespace cass

src/connection.cpp

@@ -76,14 +76,14 @@ void Connection::StartupHandler::on_set(ResponseMessage* response) {
       ErrorResponse* error
           = static_cast<ErrorResponse*>(response->response_body().get());
       if (error->code() == CQL_ERROR_PROTOCOL_ERROR &&
-          error->message().find("Invalid or unsupported protocol version") != std::string::npos) {
+          error->message().to_string().find("Invalid or unsupported protocol version") != std::string::npos) {
         connection_->is_invalid_protocol_ = true;
         LOG_WARN("Host %s received invalid protocol response %s",
                  connection_->addr_string_.c_str(), error->error_message().c_str());
         connection_->defunct();
         return;
       } else if (error->code() == CQL_ERROR_BAD_CREDENTIALS) {
-        connection_->auth_error_ = error->message();
+        connection_->auth_error_ = error->message().to_string();
       }
       connection_->notify_error("Received error response " + error->error_message());
       break;

src/error_response.cpp

@@ -26,18 +26,51 @@ namespace cass {
 
 std::string ErrorResponse::error_message() const {
   std::ostringstream ss;
-  ss << "'" << message() << "'"
+  ss << "'" << message().to_string() << "'"
      << " (0x" << std::hex << std::uppercase << std::setw(8) << std::setfill('0')
      << CASS_ERROR(CASS_ERROR_SOURCE_SERVER, code()) << ")";
   return ss.str();
 }
 
 bool ErrorResponse::decode(int version, char* buffer, size_t size) {
+  uint8_t data_present;
+  StringRef write_type;
+
   char* pos = decode_int32(buffer, code_);
-  pos = decode_string(pos, &message_, message_size_);
+  pos = decode_string_ref(pos, &message_);
+
   switch (code_) {
+    case CQL_ERROR_UNAVAILABLE:
+      pos = decode_uint16(pos, cl_);
+      pos = decode_int32(pos, received_);
+      pos = decode_int32(pos, alive_);
+      break;
+    case CQL_ERROR_READ_TIMEOUT:
+      pos = decode_uint16(pos, cl_);
+      pos = decode_int32(pos, received_);
+      pos = decode_int32(pos, required_);
+      decode_byte(pos, data_present);
+      data_present_ = data_present > 0;
+      break;
+    case CQL_ERROR_WRITE_TIMEOUT:
+      pos = decode_uint16(pos, cl_);
+      pos = decode_int32(pos, received_);
+      pos = decode_int32(pos, required_);
+      decode_string_ref(pos, &write_type);
+      if (write_type == "SIMPLE") {
+        write_type_ = RetryPolicy::SIMPLE;
+      } else if(write_type == "BATCH") {
+        write_type_ = RetryPolicy::BATCH;
+      } else if(write_type == "UNLOGGED_BATCH") {
+        write_type_ = RetryPolicy::UNLOGGED_BATCH;
+      } else if(write_type == "COUNTER") {
+        write_type_ = RetryPolicy::COUNTER;
+      } else if(write_type == "BATCH_LOG") {
+        write_type_ = RetryPolicy::BATCH_LOG;
+      }
+      break;
     case CQL_ERROR_UNPREPARED:
-      decode_string(pos, &prepared_id_, prepared_id_size_);
+      decode_string_ref(pos, &prepared_id_);
       break;
   }
   return true;

src/error_response.hpp

@@ -20,6 +20,8 @@
 #include "response.hpp"
 #include "constants.hpp"
 #include "scoped_ptr.hpp"
+#include "string_ref.hpp"
+#include "retry_policy.hpp"
 
 #include <uv.h>
 
@@ -32,40 +34,38 @@ class ErrorResponse : public Response {
 public:
   ErrorResponse()
       : Response(CQL_OPCODE_ERROR)
-      , code_(0xFFFFFFFF)
-      , message_(NULL)
-      , message_size_(0)
-      , prepared_id_(NULL)
-      , prepared_id_size_(0) {}
-
-  ErrorResponse(int32_t code, const char* input, size_t input_size)
-      : Response(CQL_OPCODE_ERROR)
-      , guard(new char[input_size])
-      , code_(code)
-      , message_(guard.get())
-      , message_size_(input_size) {
-    memcpy(message_, input, input_size);
-  }
+      , code_(0xFFFFFFFF) { }
 
   int32_t code() const { return code_; }
+  StringRef message() const { return message_; }
 
-  std::string prepared_id() const {
-    return std::string(prepared_id_, prepared_id_size_);
-  }
+  StringRef prepared_id() const { return prepared_id_; }
 
-  std::string message() const { return std::string(message_, message_size_); }
+  CassConsistency consistency() const {
+    return static_cast<CassConsistency>(cl_);
+  }
+  int32_t received() const { return received_; }
+  int32_t required() const { return required_; }
+  int32_t alive() const { return alive_; }
+  bool data_present() const { return data_present_; }
+  RetryPolicy::WriteType write_type() const { return write_type_; }
 
   std::string error_message() const;
 
   bool decode(int version, char* buffer, size_t size);
 
 private:
-  ScopedPtr<char> guard;
   int32_t code_;
-  char* message_;
-  size_t message_size_;
-  char* prepared_id_;
-  size_t prepared_id_size_;
+  StringRef message_;
+
+  StringRef prepared_id_;
+
+  uint16_t cl_;
+  int32_t received_;
+  int32_t required_;
+  int32_t alive_;
+  bool data_present_;
+  RetryPolicy::WriteType write_type_;
 };
 
 bool check_error_or_invalid_response(const std::string& prefix, uint8_t expected_opcode,

src/execute_request.cpp

@@ -16,6 +16,8 @@
 
 #include "execute_request.hpp"
 
+#include "handler.hpp"
+
 namespace cass {
 
 int32_t ExecuteRequest::encode_batch(int version, BufferVec* bufs, EncodingCache* cache) const {
@@ -40,15 +42,15 @@ int32_t ExecuteRequest::encode_batch(int version, BufferVec* bufs, EncodingCache
   return length;
 }
 
-int ExecuteRequest::encode(int version, BufferVec* bufs, EncodingCache* cache) const {
+int ExecuteRequest::encode(int version, Handler* handler, BufferVec* bufs) const {
   if (version == 1) {
-    return internal_encode_v1(bufs, cache);
+    return internal_encode_v1(handler, bufs);
   } else {
-    return internal_encode(version, bufs, cache);
+    return internal_encode(version, handler, bufs);
   }
 }
 
-int ExecuteRequest::internal_encode_v1(BufferVec* bufs, EncodingCache* cache) const {
+int ExecuteRequest::internal_encode_v1(Handler* handler, BufferVec* bufs) const {
   size_t length = 0;
   const int version = 1;
 
@@ -68,23 +70,23 @@ int ExecuteRequest::internal_encode_v1(BufferVec* bufs, EncodingCache* cache) co
                                  prepared_id.size());
     buf.encode_uint16(pos, elements_count());
     // <value_1>...<value_n>
-    length += copy_buffers(version, bufs, cache);
+    length += copy_buffers(version, bufs, handler->encoding_cache());
   }
 
   {
     // <consistency> [short]
     size_t buf_size = sizeof(uint16_t);
 
     Buffer buf(buf_size);
-    buf.encode_uint16(0, consistency());
+    buf.encode_uint16(0, handler->consistency());
     bufs->push_back(buf);
     length += buf_size;
   }
 
   return length;
 }
 
-int ExecuteRequest::internal_encode(int version, BufferVec* bufs, EncodingCache* cache) const {
+int ExecuteRequest::internal_encode(int version, Handler* handler, BufferVec* bufs) const {
   int length = 0;
   uint8_t flags = this->flags();
 
@@ -123,12 +125,12 @@ int ExecuteRequest::internal_encode(int version, BufferVec* bufs, EncodingCache*
     size_t pos = buf.encode_string(0,
                                  prepared_id.data(),
                                  prepared_id.size());
-    pos = buf.encode_uint16(pos, consistency());
+    pos = buf.encode_uint16(pos, handler->consistency());
     pos = buf.encode_byte(pos, flags);
 
     if (elements_count() > 0) {
       buf.encode_uint16(pos, elements_count());
-      length += copy_buffers(version, bufs, cache);
+      length += copy_buffers(version, bufs, handler->encoding_cache());
     }
   }