ClusterRunClose(const Cluster::Ptr& cluster)

: cluster_(cluster) {}

void run(EventLoop* event_loop) { cluster_->internal_close(); }

Cluster::Ptr cluster_;

ClusterNotifyUp(const Cluster::Ptr& cluster, const Address& address)

: cluster_(cluster)

, address_(address) {}

void run(EventLoop* event_loop) { cluster_->internal_notify_host_up(address_); }

Cluster::Ptr cluster_;

Address address_;

ClusterNotifyDown(const Cluster::Ptr& cluster, const Address& address)

: cluster_(cluster)

, address_(address) {}

void run(EventLoop* event_loop) { cluster_->internal_notify_host_down(address_); }

Cluster::Ptr cluster_;

Address address_;

ClusterStartEvents(const Cluster::Ptr& cluster)

: cluster_(cluster) {}

void run(EventLoop* event_loop) { cluster_->internal_start_events(); }

Cluster::Ptr cluster_;

ClusterStartClientMonitor(const Cluster::Ptr& cluster, const String& client_id,

const String& session_id, const Config& config)

: cluster_(cluster)

, client_id_(client_id)

, session_id_(session_id)

, config_(config) {}

void run(EventLoop* event_loop) {

cluster_->internal_start_monitor_reporting(client_id_, session_id_, config_);

}

Cluster::Ptr cluster_;

String client_id_;

String session_id_;

Config config_;

virtual void on_connect(Cluster* cluster) {}

virtual void on_host_up(const Host::Ptr& host) {}

virtual void on_host_down(const Host::Ptr& host) {}

virtual void on_host_added(const Host::Ptr& host) {}

virtual void on_host_removed(const Host::Ptr& host) {}

virtual void on_token_map_updated(const TokenMap::Ptr& token_map) {}

virtual void on_close(Cluster* cluster) {}

switch (event.type) {

case HOST_UP:

listener->on_host_up(event.host);

break;

case HOST_DOWN:

listener->on_host_down(event.host);

break;

case HOST_ADD:

listener->on_host_added(event.host);

break;

case HOST_REMOVE:

listener->on_host_removed(event.host);

break;

case HOST_MAYBE_UP:

listener->on_host_maybe_up(event.host);

break;

case HOST_READY:

listener->on_host_ready(event.host);

break;

case TOKEN_MAP_UPDATE:

listener->on_token_map_updated(event.token_map);

break;

}

for (ClusterEvent::Vec::const_iterator it = events.begin(), end = events.end(); it != end; ++it) {

process_event(*it, listener);

}

ScopedMutex l(&mutex_);

const_iterator it = find(address);

if (it == end()) return Host::Ptr();

return it->second;

: load_balancing_policy(new RoundRobinPolicy())

, port(CASS_DEFAULT_PORT)

, reconnection_policy(new ExponentialReconnectionPolicy())

, prepare_on_up_or_add_host(CASS_DEFAULT_PREPARE_ON_UP_OR_ADD_HOST)

, max_prepares_per_flush(CASS_DEFAULT_MAX_PREPARES_PER_FLUSH)

, disable_events_on_startup(false)

, cluster_metadata_resolver_factory(new DefaultClusterMetadataResolverFactory()) {

load_balancing_policies.push_back(load_balancing_policy);

: control_connection_settings(config)

, load_balancing_policy(config.load_balancing_policy())

, load_balancing_policies(config.load_balancing_policies())

, port(config.port())

, reconnection_policy(config.reconnection_policy())

, prepare_on_up_or_add_host(config.prepare_on_up_or_add_host())

, max_prepares_per_flush(CASS_DEFAULT_MAX_PREPARES_PER_FLUSH)

, disable_events_on_startup(false)

EventLoop* event_loop, const Host::Ptr& connected_host, const HostMap& hosts,

const ControlConnectionSchema& schema,

const LoadBalancingPolicy::Ptr& load_balancing_policy,

const LoadBalancingPolicy::Vec& load_balancing_policies, const String& local_dc,

const StringMultimap& supported_options, const ClusterSettings& settings)

: connection_(connection)

, listener_(listener ? listener : &nop_cluster_listener__)

, event_loop_(event_loop)

, load_balancing_policy_(load_balancing_policy)

, load_balancing_policies_(load_balancing_policies)

, settings_(settings)

, is_closing_(false)

, connected_host_(connected_host)

, hosts_(hosts)

, local_dc_(local_dc)

, supported_options_(supported_options)

, is_recording_events_(settings.disable_events_on_startup) {

inc_ref();

connection_->set_listener(this);

query_plan_.reset(load_balancing_policy_->new_query_plan("", NULL, NULL));

update_schema(schema);

update_token_map(hosts, connected_host_->partitioner(), schema);

listener_->on_reconnect(this);

HostMap available;

for (HostMap::const_iterator it = hosts_.begin(), end = hosts_.end(); it != end; ++it) {

if (!is_host_ignored(it->second)) {

available[it->first] = it->second;

}

}

return available;

// Update the hosts and properly notify the listener

HostMap existing(hosts_);

for (HostMap::const_iterator it = hosts.begin(), end = hosts.end(); it != end; ++it) {

HostMap::iterator find_it = existing.find(it->first);

if (find_it != existing.end()) {

existing.erase(find_it); // Already exists mark as visited

} else {

notify_host_add(it->second); // A new host has been added

}

}

// Any hosts that existed before, but aren't in the new hosts

// need to be marked as removed.

for (HostMap::const_iterator it = existing.begin(), end = existing.end(); it != end; ++it) {

notify_host_remove(it->first);

}

metadata_.clear_and_update_back(connection_->server_version());

if (schema.keyspaces) {

metadata_.update_keyspaces(schema.keyspaces.get(), false);

}

if (schema.tables) {

metadata_.update_tables(schema.tables.get());

}

if (schema.views) {

metadata_.update_views(schema.views.get());

}

if (schema.columns) {

metadata_.update_columns(schema.columns.get());

}

if (schema.indexes) {

metadata_.update_indexes(schema.indexes.get());

}

if (schema.user_types) {

metadata_.update_user_types(schema.user_types.get());

}

if (schema.functions) {

metadata_.update_functions(schema.functions.get());

}

if (schema.aggregates) {

metadata_.update_aggregates(schema.aggregates.get());

}

if (schema.virtual_keyspaces) {

metadata_.update_keyspaces(schema.virtual_keyspaces.get(), true);

}

if (schema.virtual_tables) {

metadata_.update_tables(schema.virtual_tables.get());

}

if (schema.virtual_columns) {

metadata_.update_columns(schema.virtual_columns.get());

}

metadata_.swap_to_back_and_update_front();

const ControlConnectionSchema& schema) {

if (settings_.control_connection_settings.use_token_aware_routing && schema.keyspaces) {

// Create a new token map and populate it

token_map_ = TokenMap::from_partitioner(partitioner);

if (!token_map_) {

return; // Partition is not supported

}

token_map_->add_keyspaces(connection_->server_version(), schema.keyspaces.get());

for (HostMap::const_iterator it = hosts.begin(), end = hosts.end(); it != end; ++it) {

token_map_->add_host(it->second);

}

token_map_->build();

}

if (!reconnection_schedule_) {

reconnection_schedule_.reset(settings_.reconnection_policy->new_reconnection_schedule());

}

uint64_t delay_ms = reconnection_schedule_->next_delay_ms();

if (delay_ms > 0) {

timer_.start(connection_->loop(), delay_ms,

bind_callback(&Cluster::on_schedule_reconnect, this));

} else {

handle_schedule_reconnect();

}

const Host::Ptr& host = query_plan_->compute_next();

if (host) {

reconnector_.reset(new ControlConnector(host, connection_->protocol_version(),

bind_callback(&Cluster::on_reconnect, this)));

reconnector_->with_settings(settings_.control_connection_settings)

->connect(connection_->loop());

} else {

// No more hosts, refresh the query plan and schedule a re-connection

LOG_TRACE("Control connection query plan has no more hosts. "

"Reset query plan and schedule reconnect");

query_plan_.reset(load_balancing_policy_->new_query_plan("", NULL, NULL));

schedule_reconnect();

}

reconnector_.reset();

if (is_closing_) {

handle_close();

return;

}

if (connector->is_ok()) {

connection_ = connector->release_connection();

connection_->set_listener(this);

// Incrementally update the hosts (notifying the listener)

update_hosts(connector->hosts());

// Get the newly connected host

connected_host_ = hosts_[connection_->address()];

assert(connected_host_ && "Connected host not found in hosts map");

update_schema(connector->schema());

update_token_map(connector->hosts(), connected_host_->partitioner(), connector->schema());

// Notify the listener that we've built a new token map

if (token_map_) {

notify_or_record(ClusterEvent(token_map_));

}

LOG_INFO("Control connection connected to %s", connected_host_->address_string().c_str());

listener_->on_reconnect(this);

reconnection_schedule_.reset();

} else if (!connector->is_canceled()) {

LOG_ERROR(

"Unable to reestablish a control connection to host %s because of the following error: %s",

connector->address().to_string().c_str(), connector->error_message().c_str());

schedule_reconnect();

}

is_closing_ = true;

monitor_reporting_timer_.stop();

if (timer_.is_running()) {

timer_.stop();

handle_close();

} else if (reconnector_) {

reconnector_->cancel();

} else if (connection_) {

connection_->close();

}

for (LoadBalancingPolicy::Vec::const_iterator it = load_balancing_policies_.begin(),

end = load_balancing_policies_.end();

it != end; ++it) {

(*it)->close_handles();

}

connection_.reset();

listener_->on_close(this);

dec_ref();

LockedHostMap::const_iterator it = hosts_.find(address);

if (it == hosts_.end()) {

LOG_WARN("Attempting to mark host %s that we don't have as UP", address.to_string().c_str());

return;

}

Host::Ptr host(it->second);

if (load_balancing_policy_->is_host_up(address)) {

// Already marked up so don't repeat duplicate notifications.

if (!is_host_ignored(host)) {

notify_or_record(ClusterEvent(ClusterEvent::HOST_READY, host));

}

return;

}

for (LoadBalancingPolicy::Vec::const_iterator it = load_balancing_policies_.begin(),

end = load_balancing_policies_.end();

it != end; ++it) {

(*it)->on_host_up(host);

}

if (is_host_ignored(host)) {

return; // Ignore host

}

if (!prepare_host(host, bind_callback(&Cluster::on_prepare_host_up, this))) {

notify_host_up_after_prepare(host);

}

LockedHostMap::const_iterator it = hosts_.find(address);

if (it == hosts_.end()) {

// Using DEBUG level here because this can happen normally as the result of

// a remove event.

LOG_DEBUG("Attempting to mark host %s that we don't have as DOWN", address.to_string().c_str());

return;

}

Host::Ptr host(it->second);

if (!load_balancing_policy_->is_host_up(address)) {

// Already marked down so don't repeat duplicate notifications.

return;

}

for (LoadBalancingPolicy::Vec::const_iterator it = load_balancing_policies_.begin(),

end = load_balancing_policies_.end();

it != end; ++it) {

(*it)->on_host_down(address);

}

notify_or_record(ClusterEvent(ClusterEvent::HOST_DOWN, host));

// Ignore if closing or already processed events

if (!is_closing_ && is_recording_events_) {

is_recording_events_ = false;

ClusterEvent::process_events(recorded_events_, listener_);

recorded_events_.clear();

}

const Config& config) {

monitor_reporting_.reset(create_monitor_reporting(client_id, session_id, config));

if (!is_closing_ && monitor_reporting_->interval_ms(connection_->dse_server_version()) > 0) {

monitor_reporting_->send_startup_message(connection_->connection(), config, available_hosts(),

load_balancing_policies_);

monitor_reporting_timer_.start(

event_loop_->loop(), monitor_reporting_->interval_ms(connection_->dse_server_version()),

bind_callback(&Cluster::on_monitor_reporting, this));

}

if (!is_closing_) {

monitor_reporting_->send_status_message(connection_->connection(), available_hosts());

monitor_reporting_timer_.start(

event_loop_->loop(), monitor_reporting_->interval_ms(connection_->dse_server_version()),

bind_callback(&Cluster::on_monitor_reporting, this));

}

LockedHostMap::const_iterator host_it = hosts_.find(host->address());

if (host_it != hosts_.end()) {

LOG_WARN("Attempting to add host %s that we already have", host->address_string().c_str());

// If an entry already exists then notify that the node has been removed

// then re-add it.

for (LoadBalancingPolicy::Vec::const_iterator it = load_balancing_policies_.begin(),

end = load_balancing_policies_.end();

it != end; ++it) {

(*it)->on_host_removed(host_it->second);

}

notify_or_record(ClusterEvent(ClusterEvent::HOST_REMOVE, host));

}

hosts_[host->address()] = host;

for (LoadBalancingPolicy::Vec::const_iterator it = load_balancing_policies_.begin(),

end = load_balancing_policies_.end();

it != end; ++it) {

(*it)->on_host_added(host);

}

if (is_host_ignored(host)) {

return; // Ignore host

}

if (!prepare_host(host, bind_callback(&Cluster::on_prepare_host_add, this))) {

notify_host_add_after_prepare(host);

}

if (token_map_) {

token_map_ = token_map_->copy();

token_map_->update_host_and_build(host);

notify_or_record(ClusterEvent(token_map_));

}

notify_or_record(ClusterEvent(ClusterEvent::HOST_ADD, host));

LockedHostMap::const_iterator it = hosts_.find(address);

if (it == hosts_.end()) {

LOG_WARN("Attempting removing host %s that we don't have", address.to_string().c_str());

return;

}

Host::Ptr host(it->second);

if (token_map_) {

token_map_ = token_map_->copy();

token_map_->remove_host_and_build(host);

notify_or_record(ClusterEvent(token_map_));

}

// If not marked down yet then explicitly trigger the event.

if (load_balancing_policy_->is_host_up(address)) {

notify_or_record(ClusterEvent(ClusterEvent::HOST_DOWN, host));

}

hosts_.erase(address);

for (LoadBalancingPolicy::Vec::const_iterator it = load_balancing_policies_.begin(),

end = load_balancing_policies_.end();

it != end; ++it) {

(*it)->on_host_removed(host);

}

notify_or_record(ClusterEvent(ClusterEvent::HOST_REMOVE, host));

if (is_recording_events_) {

recorded_events_.push_back(event);

} else {

ClusterEvent::process_event(event, listener_);

}

if (connection_ && settings_.prepare_on_up_or_add_host) {

PrepareHostHandler::Ptr prepare_host_handler(

new PrepareHostHandler(host, prepared_metadata_.copy(), callback,

connection_->protocol_version(), settings_.max_prepares_per_flush));

prepare_host_handler->prepare(connection_->loop(),

settings_.control_connection_settings.connection_settings);

return true;

}

return false;

const String& keyspace_name, const String& target_name) {

switch (type) {

case KEYSPACE:

// Virtual keyspaces are not updated (always false)

metadata_.update_keyspaces(result.get(), false);

if (token_map_) {

token_map_ = token_map_->copy();

token_map_->update_keyspaces_and_build(connection_->server_version(), result.get());

notify_or_record(ClusterEvent(token_map_));

}

break;

case TABLE:

metadata_.update_tables(result.get());

break;

case VIEW:

metadata_.update_views(result.get());

break;

case COLUMN:

metadata_.update_columns(result.get());

break;

case INDEX:

metadata_.update_indexes(result.get());

break;

case USER_TYPE:

metadata_.update_user_types(result.get());

break;

case FUNCTION:

metadata_.update_functions(result.get());

break;

case AGGREGATE:

metadata_.update_aggregates(result.get());

break;

}

const String& target_name) {

switch (type) {

case KEYSPACE:

metadata_.drop_keyspace(keyspace_name);

if (token_map_) {

token_map_ = token_map_->copy();

token_map_->drop_keyspace(keyspace_name);

notify_or_record(ClusterEvent(token_map_));

}

break;

case TABLE:

metadata_.drop_table_or_view(keyspace_name, target_name);

break;

case VIEW:

metadata_.drop_table_or_view(keyspace_name, target_name);

break;

case USER_TYPE:

metadata_.drop_user_type(keyspace_name, target_name);

break;

case FUNCTION:

metadata_.drop_function(keyspace_name, target_name);

break;

case AGGREGATE:

metadata_.drop_aggregate(keyspace_name, target_name);

break;

default:

break;

}

LockedHostMap::const_iterator it = hosts_.find(address);

if (it == hosts_.end()) {

LOG_WARN("Received UP event for an unknown host %s", address.to_string().c_str());

return;

}

notify_or_record(ClusterEvent(ClusterEvent::HOST_MAYBE_UP, it->second));

if (!is_closing_) {

LOG_WARN("Lost control connection to host %s", connection_->address_string().c_str());

schedule_reconnect();

} else {

handle_close();

}