Add raw interface for use with proxies

Author: mpenick

src/connection.cpp

@@ -254,6 +254,11 @@ void Connection::on_read(const char* buf, size_t size) {
                 host_->address_string().c_str());
 
       if (response->stream() < 0) {
+          if (!response->decode_response_body(false)) {
+            LOG_ERROR("Error decoding event response body");
+            defunct();
+            continue;
+          }
         if (response->opcode() == CQL_OPCODE_EVENT) {
           listener_->on_event(response->response_body());
         } else {
@@ -266,6 +271,12 @@ void Connection::on_read(const char* buf, size_t size) {
         RequestCallback::Ptr callback;
 
         if (stream_manager_.get(response->stream(), callback)) {
+          if (!response->decode_response_body(callback->request()->is_raw())) {
+            LOG_ERROR("Error decoding response body");
+            defunct();
+            continue;
+          }
+
           switch (callback->state()) {
             case RequestCallback::REQUEST_STATE_READING:
               pending_reads_.remove(callback.get());

src/future.cpp

@@ -57,14 +57,28 @@ const CassResult* cass_future_get_result(CassFuture* future) {
   }
 
   Response::Ptr response(static_cast<ResponseFuture*>(future->from())->response());
-  if (!response || response->opcode() == CQL_OPCODE_ERROR) {
+  if (!response || response->opcode() == CQL_OPCODE_ERROR || response->is_raw()) {
     return NULL;
   }
 
   response->inc_ref();
   return CassResult::to(static_cast<ResultResponse*>(response.get()));
 }
 
+const CassRawResult* cass_future_get_raw_result(CassFuture* future) {
+  if (future->type() != Future::FUTURE_TYPE_RESPONSE) {
+    return NULL;
+  }
+
+  Response::Ptr response(static_cast<ResponseFuture*>(future->from())->response());
+  if (!response || response->opcode() == CQL_OPCODE_ERROR || !response->is_raw()) {
+    return NULL;
+  }
+
+  response->inc_ref();
+  return CassRawResult::to(static_cast<RawResponse*>(response.get()));
+}
+
 const CassPrepared* cass_future_get_prepared(CassFuture* future) {
   if (future->type() != Future::FUTURE_TYPE_RESPONSE) {
     return NULL;

src/request.hpp

@@ -187,6 +187,8 @@ class Request : public RefCounted<Request> {
   void set_host(const Address& host) { host_.reset(new Address(host)); }
   const Address* host() const { return host_.get(); }
 
+  virtual bool is_raw() const { return false; }
+
   virtual int encode(ProtocolVersion version, RequestCallback* callback, BufferVec* bufs) const = 0;
 
 private:
@@ -212,6 +214,24 @@ class RoutableRequest : public Request {
   virtual bool get_routing_key(String* routing_key) const = 0;
 };
 
+class RawRequest : public Request {
+public:
+  RawRequest(uint8_t opcode, const char* buf, size_t buf_size)
+    : Request(opcode)
+    , buf_(buf, buf_size) { }
+
+  virtual bool is_raw() const { return true; }
+
+  virtual int encode(ProtocolVersion version, RequestCallback* callback, BufferVec* bufs) const {
+    bufs->push_back(buf_);
+    return static_cast<int>(buf_.size());
+  }
+
+private:
+  Buffer buf_;
+};
+
+
 }}} // namespace datastax::internal::core
 
 EXTERNAL_TYPE(datastax::internal::core::CustomPayload, CassCustomPayload)

src/request_handler.cpp

@@ -462,6 +462,11 @@ void RequestExecution::on_set(ResponseMessage* response) {
   current_host_->decrement_inflight_requests();
   Connection* connection = connection_;
 
+  if (response->response_body()->is_raw()) {
+    set_response(response->response_body());
+    return;
+  }
+
   switch (response->opcode()) {
     case CQL_OPCODE_RESULT:
       on_result_response(connection, response);

src/response.cpp

@@ -28,6 +28,18 @@
 
 using namespace datastax::internal::core;
 
+extern "C" {
+
+void cass_raw_result_free(const CassRawResult* result) { result->dec_ref(); }
+
+cass_uint8_t cass_raw_result_opcode(const CassRawResult* result) { return result->opcode(); }
+
+const char* cass_raw_result_frame(const CassRawResult* result) { return result->data(); }
+
+size_t cass_raw_result_frame_length(const CassRawResult* result) { return result->length(); }
+
+}
+
 /**
  * A dummy invalid protocol error response that's used to handle responses
  * encoded with deprecated protocol versions.
@@ -58,9 +70,7 @@ bool Response::decode_custom_payload(Decoder& decoder) {
 bool Response::decode_warnings(Decoder& decoder) { return decoder.decode_warnings(warnings_); }
 
 bool ResponseMessage::allocate_body(int8_t opcode) {
-  response_body_.reset();
   switch (opcode) {
-
     case CQL_OPCODE_ERROR:
       response_body_.reset(new ErrorResponse());
       return true;
@@ -144,13 +154,11 @@ ssize_t ResponseMessage::decode(const char* input, size_t size) {
       // If a deprecated version of the protocol is encountered then we fake
       // an invalid protocol error.
       if (version_ < CASS_PROTOCOL_VERSION_V3) {
-        response_body_.reset(new InvalidProtocolErrorResponse());
-      } else if (!allocate_body(opcode_) || !response_body_) {
-        return -1;
+        invalid_protocol_error_ = true;
+      } else {
+        buffer_ = RefBuffer::Ptr(RefBuffer::create(length_));
+        body_buffer_pos_ = buffer_->data();
       }
-
-      response_body_->set_buffer(length_);
-      body_buffer_pos_ = response_body_->data();
     } else {
       // We haven't received all the data for the header. We consume the
       // entire buffer.
@@ -171,25 +179,9 @@ ssize_t ResponseMessage::decode(const char* input, size_t size) {
     memcpy(body_buffer_pos_, input_pos, needed);
     body_buffer_pos_ += needed;
     input_pos += needed;
-    assert(body_buffer_pos_ == response_body_->data() + length_);
-    Decoder decoder(response_body_->data(), length_, ProtocolVersion(version_));
+    assert(body_buffer_pos_ == buffer_->data() + length_);
 
-    if (flags_ & CASS_FLAG_TRACING) {
-      if (!response_body_->decode_trace_id(decoder)) return -1;
-    }
-
-    if (flags_ & CASS_FLAG_WARNING) {
-      if (!response_body_->decode_warnings(decoder)) return -1;
-    }
-
-    if (flags_ & CASS_FLAG_CUSTOM_PAYLOAD) {
-      if (!response_body_->decode_custom_payload(decoder)) return -1;
-    }
-
-    if (!response_body_->decode(decoder)) {
-      is_body_error_ = true;
-      return -1;
-    }
+    response_decoder_ =  Decoder(buffer_->data(), length_, ProtocolVersion(version_));
 
     is_body_ready_ = true;
   } else {
@@ -202,3 +194,40 @@ ssize_t ResponseMessage::decode(const char* input, size_t size) {
 
   return input_pos - input;
 }
+
+bool ResponseMessage::decode_response_body(bool is_raw) {
+  if (invalid_protocol_error_) {
+    response_body_.reset(new InvalidProtocolErrorResponse());
+    return true;
+  }
+
+  if (is_raw) {
+    response_body_.reset(new RawResponse(opcode_, length_));
+    response_body_->set_buffer(buffer_);
+    return true;
+  }
+
+  if (!allocate_body(opcode_)) {
+    return false;
+  }
+
+  response_body_->set_buffer(buffer_);
+
+  if (flags_ & CASS_FLAG_TRACING) {
+    if (!response_body_->decode_trace_id(response_decoder_)) return false;
+  }
+
+  if (flags_ & CASS_FLAG_WARNING) {
+    if (!response_body_->decode_warnings(response_decoder_)) return false;
+  }
+
+  if (flags_ & CASS_FLAG_CUSTOM_PAYLOAD) {
+    if (!response_body_->decode_custom_payload(response_decoder_)) return false;
+  }
+
+  if (!response_body_->decode(response_decoder_)) {
+    return false;
+  }
+
+  return true;
+}

src/response.hpp

@@ -46,7 +46,7 @@ class Response : public RefCounted<Response> {
 
   const RefBuffer::Ptr& buffer() const { return buffer_; }
 
-  void set_buffer(size_t size) { buffer_ = RefBuffer::Ptr(RefBuffer::create(size)); }
+  void set_buffer(const RefBuffer::Ptr& buffer) { buffer_ = buffer; }
 
   bool has_tracing_id() const;
 
@@ -62,6 +62,8 @@ class Response : public RefCounted<Response> {
 
   bool decode_warnings(Decoder& decoder);
 
+  virtual bool is_raw() const { return false; }
+
   virtual bool decode(Decoder& decoder) = 0;
 
 private:
@@ -75,6 +77,24 @@ class Response : public RefCounted<Response> {
   DISALLOW_COPY_AND_ASSIGN(Response);
 };
 
+class RawResponse : public Response {
+public:
+  RawResponse(uint8_t opcode, size_t length)
+    : Response(opcode)
+    , length_(length) { }
+
+  size_t length() const { return length_; }
+
+  virtual bool is_raw() const { return true; }
+
+  virtual bool decode(Decoder& decoder) {
+    return true; //  Ignore decoding the body
+  }
+
+private:
+  size_t length_;
+};
+
 class ResponseMessage : public Allocated {
 public:
   ResponseMessage()
@@ -88,8 +108,8 @@ class ResponseMessage : public Allocated {
       , is_header_received_(false)
       , header_buffer_pos_(header_buffer_)
       , is_body_ready_(false)
-      , is_body_error_(false)
-      , body_buffer_pos_(NULL) {}
+      , body_buffer_pos_(NULL)
+      , invalid_protocol_error_(false) { }
 
   uint8_t flags() const { return flags_; }
 
@@ -103,6 +123,8 @@ class ResponseMessage : public Allocated {
 
   ssize_t decode(const char* input, size_t size);
 
+  bool decode_response_body(bool is_raw);
+
 private:
   bool allocate_body(int8_t opcode);
 
@@ -120,14 +142,22 @@ class ResponseMessage : public Allocated {
   char* header_buffer_pos_;
 
   bool is_body_ready_;
-  bool is_body_error_;
-  Response::Ptr response_body_;
   char* body_buffer_pos_;
+  RefBuffer::Ptr buffer_;
+  bool invalid_protocol_error_;
+  Response::Ptr response_body_;
+
+  Decoder response_decoder_;
 
 private:
   DISALLOW_COPY_AND_ASSIGN(ResponseMessage);
 };
 
 }}} // namespace datastax::internal::core
 
+
+typedef struct CassRawResult_ CassRawResult;
+
+EXTERNAL_TYPE(datastax::internal::core::RawResponse, CassRawResult)
+
 #endif

src/session.cpp

@@ -99,6 +99,12 @@ CassFuture* cass_session_execute_batch(CassSession* session, const CassBatch* ba
   return CassFuture::to(future.get());
 }
 
+CassFuture* cass_session_execute_raw(CassSession* session, cass_uint8_t opcode, cass_uint8_t flags, const char* frame, size_t frame_size) {
+  Future::Ptr future(session->execute(Request::ConstPtr(new RawRequest(opcode, frame, frame_size))));
+  future->inc_ref();
+  return CassFuture::to(future.get());
+}
+
 const CassSchemaMeta* cass_session_get_schema_meta(const CassSession* session) {
   return CassSchemaMeta::to(new Metadata::SchemaSnapshot(session->cluster()->schema_snapshot()));
 }