virtual void on_pool_up(const Address& address) {}

virtual void on_pool_down(const Address& address) {}

virtual void on_pool_critical_error(const Address& address, Connector::ConnectionError code,

const String& message) {}

virtual void on_close(ConnectionPoolManager* manager) {}

ProtocolVersion protocol_version,

const String& keyspace,

ConnectionPoolManagerListener* listener,

Metrics* metrics,

const ConnectionPoolSettings& settings,

const ShardPortCalculator* shard_port_calculator)

: loop_(loop)

, protocol_version_(protocol_version)

, settings_(settings)

, listener_(listener ? listener : &nop_connection_pool_manager_listener__)

, close_state_(CLOSE_STATE_OPEN)

, keyspace_(keyspace)

, metrics_(metrics)

, shard_port_calculator_(shard_port_calculator)

#ifdef CASS_INTERNAL_DIAGNOSTICS

, flush_bytes_("flushed")

#endif

{

inc_ref(); // Reference for the lifetime of the connection pools

set_pointer_keys(to_flush_);

for (ConnectionPool::Map::const_iterator it = pools.begin(), end = pools.end(); it != end; ++it) {

it->second->set_listener(this);

add_pool(it->second);

}

ConnectionPool::Map::const_iterator it = pools_.find(address);

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

return PooledConnection::Ptr();

}

return it->second->find_least_busy(token);

for (DenseHashSet<ConnectionPool*>::const_iterator it = to_flush_.begin(), end = to_flush_.end();

it != end; ++it) {

(*it)->flush();

}

to_flush_.clear();

AddressVec result;

result.reserve(pools_.size());

for (ConnectionPool::Map::const_iterator it = pools_.begin(), end = pools_.end(); it != end;

++it) {

result.push_back(it->first);

}

return result;

ConnectionPool::Map::iterator it = pools_.find(host->address());

if (it != pools_.end()) return;

for (ConnectionPoolConnector::Vec::iterator it = pending_pools_.begin(),

end = pending_pools_.end();

it != end; ++it) {

if ((*it)->address() == host->address()) return;

}

ConnectionPoolConnector::Ptr connector(new ConnectionPoolConnector(

host, protocol_version_, bind_callback(&ConnectionPoolManager::on_connect, this)));

pending_pools_.push_back(connector);

connector->with_listener(this)

->with_keyspace(keyspace_)

->with_metrics(metrics_)

->with_shard_port_calculator(shard_port_calculator_)

->with_settings(settings_)

->connect_with_scout(loop_);

ConnectionPool::Map::iterator it = pools_.find(address);

if (it == pools_.end()) return;

// The connection pool will remove itself from the manager when all of its

// connections are closed.

it->second->close();

if (close_state_ == CLOSE_STATE_OPEN) {

close_state_ = CLOSE_STATE_CLOSING;

// Make copies of pool/connector data structures to prevent iterator

// invalidation.

ConnectionPool::Map pools(pools_);

for (ConnectionPool::Map::iterator it = pools.begin(), end = pools.end(); it != end; ++it) {

it->second->close();

}

ConnectionPoolConnector::Vec pending_pools(pending_pools_);

for (ConnectionPoolConnector::Vec::iterator it = pending_pools.begin(),

end = pending_pools.end();

it != end; ++it) {

(*it)->cancel();

}

close_state_ = CLOSE_STATE_WAITING_FOR_POOLS;

maybe_closed();

}

ConnectionPool::Map::iterator it = pools_.find(address);

if (it == pools_.end()) return;

it->second->attempt_immediate_connect();

keyspace_ = keyspace;

for (ConnectionPool::Map::iterator it = pools_.begin(), end = pools_.end(); it != end; ++it) {

it->second->set_keyspace(keyspace);

}

Connector::ConnectionError code,

const String& message) {

listener_->on_pool_critical_error(address, code, message);

if (to_flush_.empty()) {

listener_->on_requires_flush();

}

to_flush_.insert(pool);

pools_.erase(pool->address());

to_flush_.erase(pool);

maybe_closed();

// Close the manager once all current and pending pools are terminated.

if (close_state_ == CLOSE_STATE_WAITING_FOR_POOLS && pools_.empty() && pending_pools_.empty()) {

close_state_ = CLOSE_STATE_CLOSED;

listener_->on_close(this);

dec_ref();

}

pending_pools_.erase(std::remove(pending_pools_.begin(), pending_pools_.end(), pool_connector),

pending_pools_.end());

if (close_state_ != CLOSE_STATE_OPEN) {

maybe_closed();

return;

}

if (pool_connector->is_ok()) {

add_pool(pool_connector->release_pool());

} else {

listener_->on_pool_critical_error(pool_connector->address(), pool_connector->error_code(),

pool_connector->error_message());

}