| /* |
| Licensed to the Apache Software Foundation (ASF) under one |
| or more contributor license agreements. See the NOTICE file |
| distributed with this work for additional information |
| regarding copyright ownership. The ASF licenses this file |
| to you 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 "request_handler.hpp" |
| |
| #include "batch_request.hpp" |
| #include "connection.hpp" |
| #include "connection_pool_manager.hpp" |
| #include "constants.hpp" |
| #include "error_response.hpp" |
| #include "execute_request.hpp" |
| #include "metrics.hpp" |
| #include "prepare_request.hpp" |
| #include "protocol.hpp" |
| #include "response.hpp" |
| #include "result_response.hpp" |
| #include "row.hpp" |
| #include "session.hpp" |
| |
| #include <uv.h> |
| |
| using namespace datastax; |
| using namespace datastax::internal; |
| using namespace datastax::internal::core; |
| |
| static String to_hex(const String& byte_id) { |
| static const char half_byte_to_hex[] = { '0', '1', '2', '3', '4', '5', '6', '7', |
| '8', '9', 'a', 'b', 'c', 'd', 'e', 'f' }; |
| OStringStream ss; |
| |
| const char* data = byte_id.data(); |
| for (size_t i = 0; i < byte_id.length(); ++i) { |
| uint8_t byte = static_cast<uint8_t>(data[i]); |
| ss << half_byte_to_hex[(byte >> 4) & 0x0F]; |
| ss << half_byte_to_hex[byte & 0x0F]; |
| } |
| return ss.str(); |
| } |
| |
| class SingleHostQueryPlan : public QueryPlan { |
| public: |
| SingleHostQueryPlan(const Address& address) |
| : host_(new Host(address)) {} |
| |
| virtual Host::Ptr compute_next() { |
| Host::Ptr temp = host_; |
| host_.reset(); // Only return the host once |
| return temp; |
| } |
| |
| private: |
| Host::Ptr host_; |
| }; |
| |
| class PrepareCallback : public SimpleRequestCallback { |
| public: |
| PrepareCallback(const String& query, const String& id, RequestExecution* request_execution); |
| |
| private: |
| class PrepareRequest : public core::PrepareRequest { |
| public: |
| PrepareRequest(const String& query, const String& keyspace, uint64_t request_timeout_ms) |
| : core::PrepareRequest(query) { |
| set_keyspace(keyspace); |
| set_request_timeout_ms(request_timeout_ms); |
| } |
| }; |
| |
| private: |
| virtual void on_internal_set(ResponseMessage* response); |
| virtual void on_internal_error(CassError code, const String& message); |
| virtual void on_internal_timeout(); |
| |
| private: |
| RequestExecution::Ptr request_execution_; |
| String id_; |
| }; |
| |
| PrepareCallback::PrepareCallback(const String& query, const String& id, |
| RequestExecution* request_execution) |
| : SimpleRequestCallback( |
| Request::ConstPtr(new PrepareRequest(query, request_execution->request()->keyspace(), |
| request_execution->request_timeout_ms()))) |
| , request_execution_(request_execution) |
| , id_(id) {} |
| |
| void PrepareCallback::on_internal_set(ResponseMessage* response) { |
| switch (response->opcode()) { |
| case CQL_OPCODE_RESULT: { |
| ResultResponse* result = static_cast<ResultResponse*>(response->response_body().get()); |
| if (result->kind() == CASS_RESULT_KIND_PREPARED) { |
| String result_id = result->prepared_id().to_string(); |
| if (id_ != result_id) { |
| request_execution_->notify_prepared_id_mismatch(id_, result_id); |
| } else { |
| request_execution_->notify_result_metadata_changed(request(), result); |
| request_execution_->on_retry_current_host(); |
| } |
| } else { |
| request_execution_->on_retry_next_host(); |
| } |
| } break; |
| case CQL_OPCODE_ERROR: |
| request_execution_->on_retry_next_host(); |
| break; |
| default: |
| break; |
| } |
| } |
| |
| void PrepareCallback::on_internal_error(CassError code, const String& message) { |
| request_execution_->on_retry_next_host(); |
| } |
| |
| void PrepareCallback::on_internal_timeout() { request_execution_->on_retry_next_host(); } |
| |
| class NopRequestListener : public RequestListener { |
| public: |
| virtual void on_prepared_metadata_changed(const String& id, |
| const PreparedMetadata::Entry::Ptr& entry) {} |
| |
| virtual void on_keyspace_changed(const String& keyspace, KeyspaceChangedResponse response) {} |
| |
| virtual bool on_wait_for_tracing_data(const RequestHandler::Ptr& request_handler, |
| const Host::Ptr& current_host, |
| const Response::Ptr& response) { |
| return false; |
| } |
| |
| virtual bool on_wait_for_schema_agreement(const RequestHandler::Ptr& request_handler, |
| const Host::Ptr& current_host, |
| const Response::Ptr& response) { |
| return false; |
| } |
| |
| virtual bool on_prepare_all(const RequestHandler::Ptr& request_handler, |
| const Host::Ptr& current_host, const Response::Ptr& response) { |
| return false; |
| } |
| |
| virtual void on_done() {} |
| }; |
| |
| static NopRequestListener nop_request_listener__; |
| |
| RequestHandler::RequestHandler(const Request::ConstPtr& request, const ResponseFuture::Ptr& future, |
| Metrics* metrics) |
| : wrapper_(request) |
| , future_(future) |
| , is_done_(false) |
| , running_executions_(0) |
| , start_time_ns_(uv_hrtime()) |
| , listener_(&nop_request_listener__) |
| , manager_(NULL) |
| , metrics_(metrics) {} |
| |
| RequestHandler::~RequestHandler() { |
| if (Logger::log_level() >= CASS_LOG_TRACE) { |
| OStringStream ss; |
| for (RequestTryVec::const_iterator it = request_tries_.begin(), end = request_tries_.end(); |
| it != end; ++it) { |
| if (it != request_tries_.begin()) ss << ", "; |
| ss << "(" << it->address << ", "; |
| if (it->error != CASS_OK) { |
| ss << cass_error_desc(it->error); |
| } else { |
| ss << it->latency; |
| } |
| ss << ")"; |
| } |
| LOG_TRACE("Speculative execution attempts: [%s]", ss.str().c_str()); |
| } |
| } |
| |
| void RequestHandler::set_prepared_metadata(const PreparedMetadata::Entry::Ptr& entry) { |
| wrapper_.set_prepared_metadata(entry); |
| } |
| |
| void RequestHandler::init(const ExecutionProfile& profile, ConnectionPoolManager* manager, |
| const TokenMap* token_map, TimestampGenerator* timestamp_generator, |
| RequestListener* listener) { |
| manager_ = manager; |
| listener_ = listener ? listener : &nop_request_listener__; |
| wrapper_.init(profile, timestamp_generator); |
| |
| // Attempt to use the statement's keyspace first then if not set then use the session's keyspace |
| const String& keyspace(!request()->keyspace().empty() ? request()->keyspace() |
| : manager_->keyspace()); |
| |
| // If a specific host is set then bypass the load balancing policy and use a |
| // specialized single host query plan. |
| if (request()->host()) { |
| query_plan_.reset(new SingleHostQueryPlan(*request()->host())); |
| } else { |
| query_plan_.reset(profile.load_balancing_policy()->new_query_plan(keyspace, this, token_map)); |
| } |
| |
| execution_plan_.reset( |
| profile.speculative_execution_policy()->new_plan(keyspace, wrapper_.request().get())); |
| } |
| |
| void RequestHandler::execute() { |
| RequestExecution::Ptr request_execution(new RequestExecution(this)); |
| running_executions_++; |
| internal_retry(request_execution.get()); |
| } |
| |
| void RequestHandler::retry(RequestExecution* request_execution, Protected) { |
| internal_retry(request_execution); |
| } |
| |
| void RequestHandler::start_request(uv_loop_t* loop, Protected) { |
| if (!timer_.is_running()) { |
| uint64_t request_timeout_ms = wrapper_.request_timeout_ms(); |
| if (request_timeout_ms > 0) { // 0 means no timeout |
| timer_.start(loop, request_timeout_ms, bind_callback(&RequestHandler::on_timeout, this)); |
| } |
| } |
| } |
| |
| Host::Ptr RequestHandler::next_host(Protected) { return query_plan_->compute_next(); } |
| |
| int64_t RequestHandler::next_execution(const Host::Ptr& current_host, Protected) { |
| return execution_plan_->next_execution(current_host); |
| } |
| |
| void RequestHandler::add_attempted_address(const Address& address, Protected) { |
| future_->add_attempted_address(address); |
| } |
| |
| void RequestHandler::notify_result_metadata_changed(const String& prepared_id, const String& query, |
| const String& keyspace, |
| const String& result_metadata_id, |
| const ResultResponse::ConstPtr& result_response, |
| Protected) { |
| PreparedMetadata::Entry::Ptr entry( |
| new PreparedMetadata::Entry(query, keyspace, result_metadata_id, result_response)); |
| listener_->on_prepared_metadata_changed(prepared_id, entry); |
| } |
| |
| void RequestHandler::notify_keyspace_changed(const String& keyspace, const Host::Ptr& current_host, |
| const Response::Ptr& response) { |
| listener_->on_keyspace_changed( |
| keyspace, KeyspaceChangedResponse(RequestHandler::Ptr(this), current_host, response)); |
| } |
| |
| bool RequestHandler::wait_for_tracing_data(const Host::Ptr& current_host, |
| const Response::Ptr& response) { |
| return listener_->on_wait_for_tracing_data(Ptr(this), current_host, response); |
| } |
| |
| bool RequestHandler::wait_for_schema_agreement(const Host::Ptr& current_host, |
| const Response::Ptr& response) { |
| return listener_->on_wait_for_schema_agreement(Ptr(this), current_host, response); |
| } |
| |
| bool RequestHandler::prepare_all(const Host::Ptr& current_host, const Response::Ptr& response) { |
| return listener_->on_prepare_all(Ptr(this), current_host, response); |
| } |
| |
| void RequestHandler::set_response(const Host::Ptr& host, const Response::Ptr& response) { |
| stop_request(); |
| running_executions_--; |
| |
| if (future_->set_response(host->address(), response)) { |
| if (metrics_) { |
| metrics_->record_request(uv_hrtime() - start_time_ns_); |
| } |
| } else { |
| // This request is a speculative execution for whom we already processed |
| // a response (another speculative execution). So consider this one an |
| // aborted speculative execution. |
| if (metrics_) { |
| metrics_->record_speculative_request(uv_hrtime() - start_time_ns_); |
| } |
| } |
| |
| if (Logger::log_level() >= CASS_LOG_TRACE) { |
| request_tries_.push_back(RequestTry(host->address(), uv_hrtime() - start_time_ns_)); |
| } |
| } |
| |
| void RequestHandler::set_error(CassError code, const String& message) { |
| stop_request(); |
| bool skip = (code == CASS_ERROR_LIB_NO_HOSTS_AVAILABLE && --running_executions_ > 0); |
| if (!skip) { |
| future_->set_error(code, message); |
| } |
| } |
| |
| void RequestHandler::set_error(const Host::Ptr& host, CassError code, const String& message) { |
| stop_request(); |
| bool skip = (code == CASS_ERROR_LIB_NO_HOSTS_AVAILABLE && --running_executions_ > 0); |
| if (!skip) { |
| if (host) { |
| future_->set_error_with_address(host->address(), code, message); |
| } else { |
| set_error(code, message); |
| } |
| } |
| if (Logger::log_level() >= CASS_LOG_TRACE) { |
| request_tries_.push_back(RequestTry(host->address(), code)); |
| } |
| } |
| |
| void RequestHandler::set_error_with_error_response(const Host::Ptr& host, |
| const Response::Ptr& error, CassError code, |
| const String& message) { |
| stop_request(); |
| running_executions_--; |
| future_->set_error_with_response(host->address(), error, code, message); |
| if (Logger::log_level() >= CASS_LOG_TRACE) { |
| request_tries_.push_back(RequestTry(host->address(), code)); |
| } |
| } |
| |
| void RequestHandler::stop_timer() { timer_.stop(); } |
| |
| void RequestHandler::on_timeout(Timer* timer) { |
| if (metrics_) { |
| metrics_->request_timeouts.inc(); |
| } |
| set_error(CASS_ERROR_LIB_REQUEST_TIMED_OUT, "Request timed out"); |
| LOG_DEBUG("Request timed out"); |
| } |
| |
| void RequestHandler::stop_request() { |
| if (!is_done_) { |
| listener_->on_done(); |
| is_done_ = true; |
| } |
| timer_.stop(); |
| } |
| |
| void RequestHandler::internal_retry(RequestExecution* request_execution) { |
| if (is_done_) { |
| LOG_DEBUG("Canceling speculative execution (%p) for request (%p) on host %s", |
| static_cast<void*>(request_execution), static_cast<void*>(this), |
| request_execution->current_host() |
| ? request_execution->current_host()->address_string().c_str() |
| : "<no current host>"); |
| return; |
| } |
| |
| bool is_done = false; |
| while (!is_done && request_execution->current_host()) { |
| PooledConnection::Ptr connection = |
| manager_->find_least_busy(request_execution->current_host()->address()); |
| if (connection) { |
| int32_t result = connection->write(request_execution); |
| |
| if (result > 0) { |
| is_done = true; |
| } else { |
| switch (result) { |
| case SocketRequest::SOCKET_REQUEST_ERROR_CLOSED: |
| // This should never happen, but retry with next host if it does. |
| request_execution->next_host(); |
| break; |
| |
| case SocketRequest::SOCKET_REQUEST_ERROR_NO_HANDLER: |
| set_error(CASS_ERROR_LIB_WRITE_ERROR, |
| "Socket is not properly configured with a handler"); |
| is_done = true; |
| break; |
| |
| case Request::REQUEST_ERROR_NO_AVAILABLE_STREAM_IDS: |
| // Retry with next host |
| request_execution->next_host(); |
| break; |
| |
| case Request::REQUEST_ERROR_BATCH_WITH_NAMED_VALUES: |
| case Request::REQUEST_ERROR_PARAMETER_UNSET: |
| case Request::REQUEST_ERROR_UNSUPPORTED_PROTOCOL: |
| case Request::REQUEST_ERROR_NO_DATA_WRITTEN: |
| // Already handled with a specific error. |
| is_done = true; |
| break; |
| |
| default: |
| set_error(CASS_ERROR_LIB_WRITE_ERROR, "Unspecified write error occurred"); |
| is_done = true; |
| break; |
| } |
| } |
| } else { |
| // No connection available on the current host, move to the next host. |
| request_execution->next_host(); |
| } |
| } |
| |
| if (!request_execution->current_host()) { |
| set_error(CASS_ERROR_LIB_NO_HOSTS_AVAILABLE, "All hosts in current policy attempted " |
| "and were either unavailable or failed"); |
| } |
| } |
| |
| RequestExecution::RequestExecution(RequestHandler* request_handler) |
| : RequestCallback(request_handler->wrapper()) |
| , request_handler_(request_handler) |
| , current_host_(request_handler->next_host(RequestHandler::Protected())) |
| , num_retries_(0) |
| , start_time_ns_(uv_hrtime()) {} |
| |
| void RequestExecution::on_execute_next(Timer* timer) { request_handler_->execute(); } |
| |
| void RequestExecution::on_retry_current_host() { retry_current_host(); } |
| |
| void RequestExecution::on_retry_next_host() { |
| if (current_host_) current_host_->decrement_inflight_requests(); |
| retry_next_host(); |
| } |
| |
| void RequestExecution::retry_current_host() { |
| // Reset the request so it can be executed again |
| set_state(REQUEST_STATE_NEW); |
| |
| request_handler_->retry(this, RequestHandler::Protected()); |
| } |
| |
| void RequestExecution::retry_next_host() { |
| next_host(); |
| retry_current_host(); |
| } |
| |
| void RequestExecution::on_write(Connection* connection) { |
| assert(current_host_ && "Tried to start on a non-existent host"); |
| current_host_->increment_inflight_requests(); |
| connection_ = connection; |
| if (request()->record_attempted_addresses()) { |
| request_handler_->add_attempted_address(current_host_->address(), RequestHandler::Protected()); |
| } |
| request_handler_->start_request(connection->loop(), RequestHandler::Protected()); |
| if (request()->is_idempotent()) { |
| int64_t timeout = request_handler_->next_execution(current_host_, RequestHandler::Protected()); |
| if (timeout == 0) { |
| request_handler_->execute(); |
| } else if (timeout > 0) { |
| schedule_timer_.start(connection->loop(), timeout, |
| bind_callback(&RequestExecution::on_execute_next, this)); |
| } |
| } |
| } |
| |
| void RequestExecution::on_set(ResponseMessage* response) { |
| assert(connection_ != NULL); |
| assert(current_host_ && "Tried to set on a non-existent host"); |
| |
| current_host_->decrement_inflight_requests(); |
| Connection* connection = connection_; |
| |
| switch (response->opcode()) { |
| case CQL_OPCODE_RESULT: |
| on_result_response(connection, response); |
| break; |
| case CQL_OPCODE_ERROR: |
| on_error_response(connection, response); |
| break; |
| default: |
| connection->defunct(); |
| set_error(CASS_ERROR_LIB_UNEXPECTED_RESPONSE, "Unexpected response"); |
| break; |
| } |
| } |
| |
| void RequestExecution::on_error(CassError code, const String& message) { |
| if (current_host_) current_host_->decrement_inflight_requests(); |
| set_error(code, message); |
| } |
| |
| void RequestExecution::notify_result_metadata_changed(const Request* request, |
| ResultResponse* result_response) { |
| // Attempt to use the per-query keyspace first (v5+/DSEv2+ only) then |
| // the keyspace in the result metadata. |
| String keyspace; |
| if (result_response->protocol_version().supports_set_keyspace() && !request->keyspace().empty()) { |
| keyspace = request->keyspace(); |
| } else { |
| keyspace = result_response->quoted_keyspace(); |
| } |
| |
| if (request->opcode() == CQL_OPCODE_EXECUTE && result_response->kind() == CASS_RESULT_KIND_ROWS) { |
| const ExecuteRequest* execute = static_cast<const ExecuteRequest*>(request); |
| request_handler_->notify_result_metadata_changed( |
| execute->prepared()->id(), execute->prepared()->query(), keyspace, |
| result_response->new_metadata_id().to_string(), ResultResponse::ConstPtr(result_response), |
| RequestHandler::Protected()); |
| } else if (request->opcode() == CQL_OPCODE_PREPARE && |
| result_response->kind() == CASS_RESULT_KIND_PREPARED) { |
| const PrepareRequest* prepare = static_cast<const PrepareRequest*>(request); |
| request_handler_->notify_result_metadata_changed( |
| result_response->prepared_id().to_string(), prepare->query(), keyspace, |
| result_response->result_metadata_id().to_string(), |
| ResultResponse::ConstPtr(result_response), RequestHandler::Protected()); |
| } else { |
| assert(false && "Invalid response type for a result metadata change"); |
| } |
| } |
| |
| void RequestExecution::notify_prepared_id_mismatch(const String& expected_id, |
| const String& received_id) { |
| OStringStream ss; |
| ss << "ID mismatch while trying to prepare query (expected ID " << to_hex(expected_id) |
| << ", received ID " << to_hex(received_id) |
| << "). This prepared statement won't work anymore. This usually happens when you run a " |
| "'USE...' query after the statement was prepared."; |
| String message = ss.str(); |
| request_handler_->set_error(CASS_ERROR_LIB_UNEXPECTED_RESPONSE, message); |
| } |
| |
| void RequestExecution::on_result_response(Connection* connection, ResponseMessage* response) { |
| ResultResponse* result = static_cast<ResultResponse*>(response->response_body().get()); |
| |
| switch (result->kind()) { |
| case CASS_RESULT_KIND_ROWS: |
| current_host_->update_latency(uv_hrtime() - start_time_ns_); |
| |
| // Execute statements with no metadata get their metadata from |
| // result_metadata() returned when the statement was prepared. |
| if (request()->opcode() == CQL_OPCODE_EXECUTE) { |
| if (result->no_metadata()) { |
| if (!skip_metadata()) { |
| // Caused by a race condition in C* 2.1.0 |
| on_error(CASS_ERROR_LIB_UNEXPECTED_RESPONSE, |
| "Expected metadata but no metadata in response (see CASSANDRA-8054)"); |
| return; |
| } |
| result->set_metadata(prepared_metadata_entry()->result()->result_metadata()); |
| } else if (result->metadata_changed()) { |
| notify_result_metadata_changed(request(), result); |
| } |
| } |
| |
| if (!response->response_body()->has_tracing_id() || |
| !request_handler_->wait_for_tracing_data(current_host(), response->response_body())) { |
| set_response(response->response_body()); |
| } |
| break; |
| |
| case CASS_RESULT_KIND_SCHEMA_CHANGE: { |
| if (!request_handler_->wait_for_schema_agreement(current_host(), response->response_body())) { |
| set_response(response->response_body()); |
| } |
| break; |
| } |
| |
| case CASS_RESULT_KIND_SET_KEYSPACE: |
| // The response is set after the keyspace is propagated to all threads. |
| request_handler_->notify_keyspace_changed(result->quoted_keyspace(), current_host_, |
| response->response_body()); |
| break; |
| |
| case CASS_RESULT_KIND_PREPARED: |
| notify_result_metadata_changed(request(), result); |
| if (!request_handler_->prepare_all(current_host(), response->response_body())) { |
| set_response(response->response_body()); |
| } |
| break; |
| |
| default: |
| set_response(response->response_body()); |
| break; |
| } |
| } |
| |
| void RequestExecution::on_error_response(Connection* connection, ResponseMessage* response) { |
| ErrorResponse* error = static_cast<ErrorResponse*>(response->response_body().get()); |
| |
| RetryPolicy::RetryDecision decision = RetryPolicy::RetryDecision::return_error(); |
| |
| switch (error->code()) { |
| case CQL_ERROR_READ_TIMEOUT: |
| if (retry_policy()) { |
| decision = retry_policy()->on_read_timeout(request(), error->consistency(), |
| error->received(), error->required(), |
| error->data_present() > 0, num_retries_); |
| } |
| break; |
| |
| case CQL_ERROR_WRITE_TIMEOUT: |
| if (retry_policy() && request()->is_idempotent()) { |
| decision = |
| retry_policy()->on_write_timeout(request(), error->consistency(), error->received(), |
| error->required(), error->write_type(), num_retries_); |
| } |
| break; |
| |
| case CQL_ERROR_UNAVAILABLE: |
| if (retry_policy()) { |
| decision = retry_policy()->on_unavailable( |
| request(), error->consistency(), error->required(), error->received(), num_retries_); |
| } |
| break; |
| |
| case CQL_ERROR_OVERLOADED: |
| LOG_WARN("Host %s is overloaded.", connection->address_string().c_str()); |
| if (retry_policy() && request()->is_idempotent()) { |
| decision = retry_policy()->on_request_error(request(), consistency(), error, num_retries_); |
| } |
| break; |
| |
| case CQL_ERROR_SERVER_ERROR: |
| LOG_WARN("Received server error '%s' from host %s. Defuncting the connection...", |
| error->message().to_string().c_str(), connection->address_string().c_str()); |
| connection->defunct(); |
| if (retry_policy() && request()->is_idempotent()) { |
| decision = retry_policy()->on_request_error(request(), consistency(), error, num_retries_); |
| } |
| break; |
| |
| case CQL_ERROR_IS_BOOTSTRAPPING: |
| LOG_ERROR("Query sent to bootstrapping host %s. Retrying on the next host...", |
| connection->address_string().c_str()); |
| retry_next_host(); |
| return; // Done |
| |
| case CQL_ERROR_UNPREPARED: |
| on_error_unprepared(connection, error); |
| return; // Done |
| |
| default: |
| // Return the error response |
| break; |
| } |
| |
| // Process retry decision |
| switch (decision.type()) { |
| case RetryPolicy::RetryDecision::RETURN_ERROR: |
| set_error_with_error_response( |
| response->response_body(), |
| static_cast<CassError>(CASS_ERROR(CASS_ERROR_SOURCE_SERVER, error->code())), |
| error->message().to_string()); |
| break; |
| |
| case RetryPolicy::RetryDecision::RETRY: |
| set_retry_consistency(decision.retry_consistency()); |
| if (decision.retry_current_host()) { |
| retry_current_host(); |
| } else { |
| retry_next_host(); |
| } |
| num_retries_++; |
| break; |
| |
| case RetryPolicy::RetryDecision::IGNORE: |
| set_response(Response::Ptr(new ResultResponse())); |
| break; |
| } |
| } |
| |
| void RequestExecution::on_error_unprepared(Connection* connection, ErrorResponse* error) { |
| LOG_DEBUG("Unprepared error response returned for request: %s", |
| error->message().to_string().c_str()); |
| |
| String query; |
| String id = error->prepared_id().to_string(); |
| if (request()->opcode() == CQL_OPCODE_EXECUTE) { |
| const ExecuteRequest* execute = static_cast<const ExecuteRequest*>(request()); |
| query = execute->prepared()->query(); |
| } else if (request()->opcode() == CQL_OPCODE_BATCH) { |
| const BatchRequest* batch = static_cast<const BatchRequest*>(request()); |
| if (!batch->find_prepared_query(id, &query)) { |
| set_error(CASS_ERROR_LIB_UNEXPECTED_RESPONSE, |
| "Unable to find prepared statement in batch statement"); |
| return; |
| } |
| } else { |
| connection->defunct(); |
| set_error(CASS_ERROR_LIB_UNEXPECTED_RESPONSE, "Received unprepared error for invalid " |
| "request type or invalid prepared id"); |
| return; |
| } |
| |
| RequestCallback::Ptr callback(new PrepareCallback(query, id, this)); |
| if (connection->write_and_flush(callback) < 0) { |
| // Try to prepare on the same host but on a different connection |
| retry_current_host(); |
| } |
| } |
| |
| void RequestExecution::set_response(const Response::Ptr& response) { |
| request_handler_->set_response(current_host_, response); |
| } |
| |
| void RequestExecution::set_error(CassError code, const String& message) { |
| request_handler_->set_error(current_host_, code, message); |
| } |
| |
| void RequestExecution::set_error_with_error_response(const Response::Ptr& error, CassError code, |
| const String& message) { |
| request_handler_->set_error_with_error_response(current_host_, error, code, message); |
| } |
