while (!pending->is_empty()) {

Handler* handler = pending->front();

pending->remove(handler);

if (handler->state() == Handler::REQUEST_STATE_WRITING ||

handler->state() == Handler::REQUEST_STATE_READING) {

handler->on_timeout();

handler->stop_timer();

}

handler->dec_ref();

}

const Config& config,

Metrics* metrics,

const Address& address,

const std::string& keyspace,

int protocol_version,

Listener* listener)

: state_(CONNECTION_STATE_NEW)

, error_code_(CONNECTION_OK)

, ssl_error_code_(CASS_OK)

, pending_writes_size_(0)

, loop_(loop)

, config_(config)

, metrics_(metrics)

, address_(address)

, addr_string_(address.to_string())

, keyspace_(keyspace)

, protocol_version_(protocol_version)

, listener_(listener)

, response_(new ResponseMessage())

, stream_manager_(protocol_version)

, ssl_session_(NULL)

, idle_start_time_ms_(0)

, heartbeat_outstanding_(false) {

socket_.data = this;

uv_tcp_init(loop_, &socket_);

if (uv_tcp_nodelay(&socket_,

config.tcp_nodelay_enable() ? 1 : 0) != 0) {

LOG_WARN("Unable to set tcp nodelay");

}

if (uv_tcp_keepalive(&socket_,

config.tcp_keepalive_enable() ? 1 : 0,

config.tcp_keepalive_delay_secs()) != 0) {

LOG_WARN("Unable to set tcp keepalive");

}

SslContext* ssl_context = config_.ssl_context();

if (ssl_context != NULL) {

ssl_session_.reset(ssl_context->create_session(address_));

}

if (state_ == CONNECTION_STATE_NEW) {

set_state(CONNECTION_STATE_CONNECTING);

connect_timer_.start(loop_, config_.connect_timeout_ms(), this,

on_connect_timeout);

Connector::connect(&socket_, address_, this, on_connect);

}

int stream = stream_manager_.acquire(handler);

if (stream < 0) {

return false;

}

handler->inc_ref(); // Connection reference

handler->set_connection(this);

handler->set_stream(stream);

if (pending_writes_.is_empty() || pending_writes_.back()->is_flushed()) {

if (ssl_session_) {

pending_writes_.add_to_back(new PendingWriteSsl(this));

} else {

pending_writes_.add_to_back(new PendingWrite(this));

}

}

PendingWriteBase* pending_write = pending_writes_.back();

int32_t request_size = pending_write->write(handler);

if (request_size < 0) {

stream_manager_.release(stream);

switch (request_size) {

case Request::ENCODE_ERROR_BATCH_WITH_NAMED_VALUES:

case Request::ENCODE_ERROR_PARAMETER_UNSET:

// Already handled

break;

default:

handler->on_error(CASS_ERROR_LIB_MESSAGE_ENCODE,

"Operation unsupported by this protocol version");

break;

}

handler->dec_ref();

return true; // Don't retry

}

pending_writes_size_ += request_size;

if (pending_writes_size_ > config_.write_bytes_high_water_mark()) {

LOG_WARN("Exceeded write bytes water mark (current: %u water mark: %u) on connection to host %s",

static_cast<unsigned int>(pending_writes_size_),

config_.write_bytes_high_water_mark(),

addr_string_.c_str());

metrics_->exceeded_write_bytes_water_mark.inc();

set_state(CONNECTION_STATE_OVERWHELMED);

}

LOG_TRACE("Sending message type %s with stream %d",

opcode_to_string(handler->request()->opcode()).c_str(), stream);

handler->set_state(Handler::REQUEST_STATE_WRITING);

handler->start_timer(loop_,

config_.request_timeout_ms(),

handler,

Connection::on_timeout);

if (flush_immediately) {

pending_write->flush();

}

if (reset_idle_time) {

idle_start_time_ms_ = 0;

restart_heartbeat_timer();

}

return true;

if (pending_writes_.is_empty()) return;

pending_writes_.back()->flush();

PendingSchemaAgreement* pending_schema_agreement = new PendingSchemaAgreement(handler);

pending_schema_agreements_.add_to_back(pending_schema_agreement);

pending_schema_agreement->timer.start(loop_,

wait,

pending_schema_agreement,

Connection::on_pending_schema_agreement);

assert(close_state == CONNECTION_STATE_CLOSE ||

close_state == CONNECTION_STATE_CLOSE_DEFUNCT);

if (state_ != CONNECTION_STATE_CLOSE &&

state_ != CONNECTION_STATE_CLOSE_DEFUNCT) {

uv_handle_t* handle = copy_cast<uv_tcp_t*, uv_handle_t*>(&socket_);

if (!uv_is_closing(handle)) {

heartbeat_timer_.stop();

connect_timer_.stop();

if (state_ == CONNECTION_STATE_CONNECTED ||

state_ == CONNECTION_STATE_READY) {

uv_read_stop(copy_cast<uv_tcp_t*, uv_stream_t*>(&socket_));

}

set_state(close_state);

uv_close(handle, on_close);

}

}

// Only update if the state changed

if (new_state == state_) return;

ConnectionState old_state = state_;

switch (state_) {

case CONNECTION_STATE_NEW:

assert(new_state == CONNECTION_STATE_CONNECTING &&

"Invalid connection state after new");

state_ = new_state;

break;

case CONNECTION_STATE_CONNECTING:

assert((new_state == CONNECTION_STATE_CONNECTED ||

new_state == CONNECTION_STATE_CLOSE ||

new_state == CONNECTION_STATE_CLOSE_DEFUNCT) &&

"Invalid connection state after connecting");

state_ = new_state;

break;

case CONNECTION_STATE_CONNECTED:

assert((new_state == CONNECTION_STATE_REGISTERING_EVENTS ||

new_state == CONNECTION_STATE_READY ||

new_state == CONNECTION_STATE_CLOSE ||

new_state == CONNECTION_STATE_CLOSE_DEFUNCT) &&

"Invalid connection state after connected");

state_ = new_state;

break;

case CONNECTION_STATE_REGISTERING_EVENTS:

assert((new_state == CONNECTION_STATE_READY ||

new_state == CONNECTION_STATE_CLOSE ||

new_state == CONNECTION_STATE_CLOSE_DEFUNCT) &&

"Invalid connection state after registering for events");

state_ = new_state;

break;

case CONNECTION_STATE_READY:

assert((new_state == CONNECTION_STATE_OVERWHELMED ||

new_state == CONNECTION_STATE_CLOSE ||

new_state == CONNECTION_STATE_CLOSE_DEFUNCT) &&

"Invalid connection state after ready");

state_ = new_state;

break;

case CONNECTION_STATE_OVERWHELMED:

assert((new_state == CONNECTION_STATE_READY ||

new_state == CONNECTION_STATE_CLOSE ||

new_state == CONNECTION_STATE_CLOSE_DEFUNCT) &&

"Invalid connection state after being overwhelmed");

state_ = new_state;

break;

case CONNECTION_STATE_CLOSE:

assert(false && "No state change after close");

break;

case CONNECTION_STATE_CLOSE_DEFUNCT:

assert(false && "No state change after close defunct");

break;

}

// Only change the availability if the state changes to/from being ready

if (new_state == CONNECTION_STATE_READY || old_state == CONNECTION_STATE_READY) {

listener_->on_availability_change(this);

}

char* buffer = input;

size_t remaining = size;

while (remaining != 0) {

ssize_t consumed = response_->decode(buffer, remaining);

if (consumed <= 0) {

notify_error("Error consuming message");

remaining = 0;

continue;

}

if (response_->is_body_ready()) {

ScopedPtr<ResponseMessage> response(response_.release());

response_.reset(new ResponseMessage());

LOG_TRACE("Consumed message type %s with stream %d, input %u, remaining %u on host %s",

opcode_to_string(response->opcode()).c_str(),

static_cast<int>(response->stream()),

static_cast<unsigned int>(size),

static_cast<unsigned int>(remaining),

addr_string_.c_str());

if (response->stream() < 0) {

if (response->opcode() == CQL_OPCODE_EVENT) {

listener_->on_event(static_cast<EventResponse*>(response->response_body().get()));

} else {

notify_error("Invalid response opcode for event stream: " +

opcode_to_string(response->opcode()));

}

} else {

Handler* handler = NULL;

if (stream_manager_.get_pending_and_release(response->stream(), handler)) {

switch (handler->state()) {

case Handler::REQUEST_STATE_READING:

maybe_set_keyspace(response.get());

pending_reads_.remove(handler);

handler->stop_timer();

handler->set_state(Handler::REQUEST_STATE_DONE);

handler->on_set(response.get());

handler->dec_ref();

break;

case Handler::REQUEST_STATE_WRITING:

// There are cases when the read callback will happen

// before the write callback. If this happens we have

// to allow the write callback to cleanup.

maybe_set_keyspace(response.get());

handler->set_state(Handler::REQUEST_STATE_READ_BEFORE_WRITE);

handler->on_set(response.get());

break;

case Handler::REQUEST_STATE_TIMEOUT:

pending_reads_.remove(handler);

handler->set_state(Handler::REQUEST_STATE_DONE);

handler->dec_ref();

break;

case Handler::REQUEST_STATE_TIMEOUT_WRITE_OUTSTANDING:

// We must wait for the write callback before we can do the cleanup

handler->set_state(Handler::REQUEST_STATE_READ_BEFORE_WRITE);

break;

default:

assert(false && "Invalid request state after receiving response");

break;

}

} else {

notify_error("Invalid stream ID");

}

}

}

remaining -= consumed;

buffer += consumed;

}

if (response->opcode() == CQL_OPCODE_RESULT) {

ResultResponse* result =

static_cast<ResultResponse*>(response->response_body().get());

if (result->kind() == CASS_RESULT_KIND_SET_KEYSPACE) {

keyspace_ = result->keyspace().to_string();

}

}

Connection* connection = static_cast<Connection*>(connector->data());

if (!connection->connect_timer_.is_running()) {

return; // Timed out

}

if (connector->status() == 0) {

LOG_DEBUG("Connected to host %s", connection->addr_string_.c_str());

if (connection->ssl_session_) {

uv_read_start(copy_cast<uv_tcp_t*, uv_stream_t*>(&connection->socket_),

Connection::alloc_buffer_ssl, Connection::on_read_ssl);

} else {

uv_read_start(copy_cast<uv_tcp_t*, uv_stream_t*>(&connection->socket_),

Connection::alloc_buffer, Connection::on_read);

}

connection->set_state(CONNECTION_STATE_CONNECTED);

if (connection->ssl_session_) {

connection->ssl_handshake();

} else {

connection->on_connected();

}

} else {

LOG_ERROR("Connect error '%s' on host %s",

UV_ERRSTR(connector->status(),

connection->loop_),

connection->addr_string_.c_str() );

connection->notify_error("Unable to connect");

}

Connection* connection = static_cast<Connection*>(timer->data());

connection->notify_error("Connection timeout", CONNECTION_ERROR_TIMEOUT);

connection->metrics_->connection_timeouts.inc();

Connection* connection = static_cast<Connection*>(handle->data);

LOG_DEBUG("Connection to host %s closed",

connection->addr_string_.c_str());

cleanup_pending_handlers(&connection->pending_reads_);

while (!connection->pending_writes_.is_empty()) {

PendingWriteBase* pending_write

= connection->pending_writes_.front();

connection->pending_writes_.remove(pending_write);

delete pending_write;

}

while (!connection->pending_schema_agreements_.is_empty()) {

PendingSchemaAgreement* pending_schema_aggreement

= connection->pending_schema_agreements_.front();

connection->pending_schema_agreements_.remove(pending_schema_aggreement);

pending_schema_aggreement->stop_timer();

pending_schema_aggreement->handler->on_closing();

delete pending_schema_aggreement;

}

connection->listener_->on_close(connection);

delete connection;

if (suggested_size <= BUFFER_REUSE_SIZE) {

if (!buffer_reuse_list_.empty()) {

uv_buf_t ret = buffer_reuse_list_.top();

buffer_reuse_list_.pop();

return ret;

}

return uv_buf_init(new char[BUFFER_REUSE_SIZE], BUFFER_REUSE_SIZE);

}

return uv_buf_init(new char[suggested_size], suggested_size);

if (buf.len == BUFFER_REUSE_SIZE && buffer_reuse_list_.size() < MAX_BUFFER_REUSE_NO) {

buffer_reuse_list_.push(buf);

return;

}

delete[] buf.base;

Connection* connection = static_cast<Connection*>(handle->data);

char* base = connection->ssl_session_->incoming().peek_writable(&suggested_size);

return uv_buf_init(base, suggested_size);

Connection* connection = static_cast<Connection*>(handle->data);

buf->base = connection->ssl_session_->incoming().peek_writable(&suggested_size);

buf->len = suggested_size;

Handler* handler = static_cast<Handler*>(timer->data());

Connection* connection = handler->connection();

LOG_INFO("Request timed out to host %s", connection->addr_string_.c_str());

// TODO (mpenick): We need to handle the case where we have too many

// timeout requests and we run out of stream ids. The java-driver

// uses a threshold to defunct the connection.

handler->set_state(Handler::REQUEST_STATE_TIMEOUT);

handler->on_timeout();

connection->metrics_->request_timeouts.inc();

if (protocol_version_ == 1) {

send_credentials();

} else {

send_initial_auth_response();

}

const std::string& token) {

AuthResponseRequest* auth_response

= new AuthResponseRequest(request->auth()->evaluate_challenge(token),

request->auth());

write(new StartupHandler(this, auth_response));

const std::string& token) {

request->auth()->on_authenticate_success(token);

on_ready();

if (state_ == CONNECTION_STATE_CONNECTED && listener_->event_types() != 0) {

set_state(CONNECTION_STATE_REGISTERING_EVENTS);

write(new StartupHandler(this, new RegisterRequest(listener_->event_types())));

return;

}

if (keyspace_.empty()) {

notify_ready();

} else {

write(new StartupHandler(this, new QueryRequest("use \"" + keyspace_ + "\"")));

}

SupportedResponse* supported =

static_cast<SupportedResponse*>(response->response_body().get());

// TODO(mstump) do something with the supported info

(void)supported;

write(new StartupHandler(this, new StartupRequest()));

PendingSchemaAgreement* pending_schema_agreement

= static_cast<PendingSchemaAgreement*>(timer->data());

Connection* connection = pending_schema_agreement->handler->connection();

connection->pending_schema_agreements_.remove(pending_schema_agreement);

pending_schema_agreement->handler->execute();

delete pending_schema_agreement;

connect_timer_.stop();

restart_heartbeat_timer();

set_state(CONNECTION_STATE_READY);

listener_->on_ready(this);

error_message_ = message;

error_code_ = code;

switch (code) {

case CONNECTION_OK:

assert(false && "Notified error without an error");

return;

case CONNECTION_ERROR_INVALID_PROTOCOL:

LOG_WARN("Host %s received invalid protocol response %s",

addr_string_.c_str(), message.c_str());

break;

case CONNECTION_ERROR_SSL:

ssl_error_code_ = ssl_session_->error_code();

log_error(message);

break;

default:

log_error(message);

break;

}

defunct();

if (state_ == CONNECTION_STATE_READY) {

LOG_ERROR("Host %s had the following error: %s",

addr_string_.c_str(), error.c_str());

} else {

LOG_ERROR("Host %s had the following error on startup: %s",

addr_string_.c_str(), error.c_str());

}

defunct();

if (!ssl_session_->is_handshake_done()) {

ssl_session_->do_handshake();

if (ssl_session_->has_error()) {

notify_error("Error during SSL handshake: " + ssl_session_->error_message(), CONNECTION_ERROR_SSL);

return;

}

}

char buf[SslHandshakeWriter::MAX_BUFFER_SIZE];

size_t size = ssl_session_->outgoing().read(buf, sizeof(buf));

if (size > 0) {

if (!SslHandshakeWriter::write(this, buf, size)) {

notify_error("Error writing data during SSL handshake");

return;

}

}

if (ssl_session_->is_handshake_done()) {

ssl_session_->verify();

if (ssl_session_->has_error()) {

notify_error("Error verifying peer certificate: " + ssl_session_->error_message(), CONNECTION_ERROR_SSL);

return;

}

on_connected();

}

ScopedPtr<V1Authenticator> v1_auth(config_.auth_provider()->new_authenticator_v1(address_));

if (v1_auth) {

V1Authenticator::Credentials credentials;

v1_auth->get_credentials(&credentials);

write(new StartupHandler(this, new CredentialsRequest(credentials)));

} else {

send_initial_auth_response();

}

SharedRefPtr<Authenticator> auth(

config_.auth_provider()->new_authenticator(address_));

if (!auth) {

notify_error("Authentication required but no auth provider set", CONNECTION_ERROR_AUTH);

} else {

AuthResponseRequest* auth_response

= new AuthResponseRequest(auth->initial_response(), auth);

write(new StartupHandler(this, auth_response));

}

if (config_.connection_heartbeat_interval_secs() > 0) {

heartbeat_timer_.start(loop_,

1000 * config_.connection_heartbeat_interval_secs(),

this, on_heartbeat);

}

Connection* connection = static_cast<Connection*>(timer->data());

if (connection->idle_start_time_ms_ == 0) {

connection->idle_start_time_ms_ = get_time_since_epoch_ms();

} else if ((get_time_since_epoch_ms() - connection->idle_start_time_ms_) / 1000 >

connection->config().connection_idle_timeout_secs()){

connection->notify_error("Failed to send a heartbeat within connection idle interval. "

"Terminating connection...",

CONNECTION_ERROR_TIMEOUT);

return;

}

if (!connection->heartbeat_outstanding_) {

if (!connection->internal_write(new HeartbeatHandler(connection), true, false)) {

// Recycling only this connection with a timeout error. This is unlikely and

// it means the connection ran out of stream IDs as a result of requests

// that never returned and as a result timed out.

connection->notify_error("No streams IDs available for heartbeat request. "

"Terminating connection...",

CONNECTION_ERROR_TIMEOUT);

return;

}

connection->heartbeat_outstanding_ = true;

}

connection->restart_heartbeat_timer();