# Copyright 2013-2016 DataStax, Inc.

#

# Licensed 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.

"""

This module houses the main classes you will interact with,

:class:`.Cluster` and :class:`.Session`.

"""

from __future__ import absolute_import

import atexit

from collections import defaultdict, Mapping

from concurrent.futures import ThreadPoolExecutor

import logging

from random import random

import socket

import sys

import time

from threading import Lock, RLock, Thread, Event

import warnings

import six

from six.moves import range

from six.moves import queue as Queue

import weakref

from weakref import WeakValueDictionary

try:

from weakref import WeakSet

except ImportError:

from cassandra.util import WeakSet # NOQA

from functools import partial, wraps

from itertools import groupby

from cassandra import (ConsistencyLevel, AuthenticationFailed,

OperationTimedOut, UnsupportedOperation,

SchemaTargetType)

from cassandra.connection import (ConnectionException, ConnectionShutdown,

ConnectionHeartbeat, ProtocolVersionUnsupported)

from cassandra.cqltypes import UserType

from cassandra.encoder import Encoder

from cassandra.protocol import (QueryMessage, ResultMessage,

ErrorMessage, ReadTimeoutErrorMessage,

WriteTimeoutErrorMessage,

UnavailableErrorMessage,

OverloadedErrorMessage,

PrepareMessage, ExecuteMessage,

PreparedQueryNotFound,

IsBootstrappingErrorMessage,

BatchMessage, RESULT_KIND_PREPARED,

RESULT_KIND_SET_KEYSPACE, RESULT_KIND_ROWS,

RESULT_KIND_SCHEMA_CHANGE, MIN_SUPPORTED_VERSION,

ProtocolHandler)

from cassandra.metadata import Metadata, protect_name, murmur3

from cassandra.policies import (TokenAwarePolicy, DCAwareRoundRobinPolicy, SimpleConvictionPolicy,

ExponentialReconnectionPolicy, HostDistance,

RetryPolicy)

from cassandra.pool import (Host, _ReconnectionHandler, _HostReconnectionHandler,

HostConnectionPool, HostConnection,

NoConnectionsAvailable)

from cassandra.query import (SimpleStatement, PreparedStatement, BoundStatement,

BatchStatement, bind_params, QueryTrace, Statement,

named_tuple_factory, dict_factory, FETCH_SIZE_UNSET)

def _is_eventlet_monkey_patched():

if 'eventlet.patcher' not in sys.modules:

return False

import eventlet.patcher

return eventlet.patcher.is_monkey_patched('socket')

def _is_gevent_monkey_patched():

if 'gevent.monkey' not in sys.modules:

return False

import gevent.socket

return socket.socket is gevent.socket.socket

# default to gevent when we are monkey patched with gevent, eventlet when

# monkey patched with eventlet, otherwise if libev is available, use that as

# the default because it's fastest. Otherwise, use asyncore.

if _is_gevent_monkey_patched():

from cassandra.io.geventreactor import GeventConnection as DefaultConnection

elif _is_eventlet_monkey_patched():

from cassandra.io.eventletreactor import EventletConnection as DefaultConnection

else:

try:

from cassandra.io.libevreactor import LibevConnection as DefaultConnection # NOQA

except ImportError:

from cassandra.io.asyncorereactor import AsyncoreConnection as DefaultConnection # NOQA

# Forces load of utf8 encoding module to avoid deadlock that occurs

# if code that is being imported tries to import the module in a seperate

# thread.

# See http://bugs.python.org/issue10923

"".encode('utf8')

log = logging.getLogger(__name__)

DEFAULT_MIN_REQUESTS = 5

DEFAULT_MAX_REQUESTS = 100

DEFAULT_MIN_CONNECTIONS_PER_LOCAL_HOST = 2

DEFAULT_MAX_CONNECTIONS_PER_LOCAL_HOST = 8

DEFAULT_MIN_CONNECTIONS_PER_REMOTE_HOST = 1

DEFAULT_MAX_CONNECTIONS_PER_REMOTE_HOST = 2

_NOT_SET = object()

class NoHostAvailable(Exception):

"""

Raised when an operation is attempted but all connections are

busy, defunct, closed, or resulted in errors when used.

"""

errors = None

"""

A map of the form ``{ip: exception}`` which details the particular

Exception that was caught for each host the operation was attempted

against.

"""

def __init__(self, message, errors):

Exception.__init__(self, message, errors)

self.errors = errors

def _future_completed(future):

""" Helper for run_in_executor() """

exc = future.exception()

if exc:

log.debug("Failed to run task on executor", exc_info=exc)

def run_in_executor(f):

"""

A decorator to run the given method in the ThreadPoolExecutor.

"""

@wraps(f)

def new_f(self, *args, **kwargs):

if self.is_shutdown:

return

try:

future = self.executor.submit(f, self, *args, **kwargs)

future.add_done_callback(_future_completed)

except Exception:

log.exception("Failed to submit task to executor")

return new_f

def _shutdown_cluster(cluster):

if cluster and not cluster.is_shutdown:

cluster.shutdown()

# murmur3 implementation required for TokenAware is only available for CPython

import platform

if platform.python_implementation() == 'CPython':

def default_lbp_factory():

if murmur3 is not None:

return TokenAwarePolicy(DCAwareRoundRobinPolicy())

return DCAwareRoundRobinPolicy()

else:

def default_lbp_factory():

return DCAwareRoundRobinPolicy()

class Cluster(object):

"""

The main class to use when interacting with a Cassandra cluster.

Typically, one instance of this class will be created for each

separate Cassandra cluster that your application interacts with.

Example usage::

>>> from cassandra.cluster import Cluster

>>> cluster = Cluster(['192.168.1.1', '192.168.1.2'])

>>> session = cluster.connect()

>>> session.execute("CREATE KEYSPACE ...")

>>> ...

>>> cluster.shutdown()

"""

contact_points = ['127.0.0.1']

"""

The list of contact points to try connecting for cluster discovery.

Defaults to loopback interface.

Note: When using :class:`.DCAwareLoadBalancingPolicy` with no explicit

local_dc set (as is the default), the DC is chosen from an arbitrary

host in contact_points. In this case, contact_points should contain

only nodes from a single, local DC.

"""

port = 9042

"""

The server-side port to open connections to. Defaults to 9042.

"""

cql_version = None

"""

If a specific version of CQL should be used, this may be set to that

string version. Otherwise, the highest CQL version supported by the

server will be automatically used.

"""

protocol_version = 4

"""

The maximum version of the native protocol to use.

The driver will automatically downgrade version based on a negotiation with

the server, but it is most efficient to set this to the maximum supported

by your version of Cassandra. Setting this will also prevent conflicting

versions negotiated if your cluster is upgraded.

Version 2 of the native protocol adds support for lightweight transactions,

batch operations, and automatic query paging. The v2 protocol is

supported by Cassandra 2.0+.

Version 3 of the native protocol adds support for protocol-level

client-side timestamps (see :attr:`.Session.use_client_timestamp`),

serial consistency levels for :class:`~.BatchStatement`, and an

improved connection pool.

Version 4 of the native protocol adds a number of new types, server warnings,

new failure messages, and custom payloads. Details in the

`project docs <https://github.com/apache/cassandra/blob/trunk/doc/native_protocol_v4.spec>`_

The following table describes the native protocol versions that

are supported by each version of Cassandra:

+-------------------+-------------------+

| Cassandra Version | Protocol Versions |

+===================+===================+

| 1.2 | 1 |

+-------------------+-------------------+

| 2.0 | 1, 2 |

+-------------------+-------------------+

| 2.1 | 1, 2, 3 |

+-------------------+-------------------+

| 2.2 | 1, 2, 3, 4 |

+-------------------+-------------------+

"""

compression = True

"""

Controls compression for communications between the driver and Cassandra.

If left as the default of :const:`True`, either lz4 or snappy compression

may be used, depending on what is supported by both the driver

and Cassandra. If both are fully supported, lz4 will be preferred.

You may also set this to 'snappy' or 'lz4' to request that specific

compression type.

Setting this to :const:`False` disables compression.

"""

_auth_provider = None

_auth_provider_callable = None

@property

def auth_provider(self):

"""

When :attr:`~.Cluster.protocol_version` is 2 or higher, this should

be an instance of a subclass of :class:`~cassandra.auth.AuthProvider`,

such as :class:`~.PlainTextAuthProvider`.

When :attr:`~.Cluster.protocol_version` is 1, this should be

a function that accepts one argument, the IP address of a node,

and returns a dict of credentials for that node.

When not using authentication, this should be left as :const:`None`.

"""

return self._auth_provider

@auth_provider.setter # noqa

def auth_provider(self, value):

if not value:

self._auth_provider = value

return

try:

self._auth_provider_callable = value.new_authenticator

except AttributeError:

if self.protocol_version > 1:

raise TypeError("auth_provider must implement the cassandra.auth.AuthProvider "

"interface when protocol_version >= 2")

elif not callable(value):

raise TypeError("auth_provider must be callable when protocol_version == 1")

self._auth_provider_callable = value

self._auth_provider = value

load_balancing_policy = None

"""

An instance of :class:`.policies.LoadBalancingPolicy` or

one of its subclasses.

.. versionchanged:: 2.6.0

Defaults to :class:`~.TokenAwarePolicy` (:class:`~.DCAwareRoundRobinPolicy`).

when using CPython (where the murmur3 extension is available). :class:`~.DCAwareRoundRobinPolicy`

otherwise. Default local DC will be chosen from contact points.

**Please see** :class:`~.DCAwareRoundRobinPolicy` **for a discussion on default behavior with respect to

DC locality and remote nodes.**

"""

reconnection_policy = ExponentialReconnectionPolicy(1.0, 600.0)

"""

An instance of :class:`.policies.ReconnectionPolicy`. Defaults to an instance

of :class:`.ExponentialReconnectionPolicy` with a base delay of one second and

a max delay of ten minutes.

"""

default_retry_policy = RetryPolicy()

"""

A default :class:`.policies.RetryPolicy` instance to use for all

:class:`.Statement` objects which do not have a :attr:`~.Statement.retry_policy`

explicitly set.

"""

conviction_policy_factory = SimpleConvictionPolicy

"""

A factory function which creates instances of

:class:`.policies.ConvictionPolicy`. Defaults to

:class:`.policies.SimpleConvictionPolicy`.

"""

connect_to_remote_hosts = True

"""

If left as :const:`True`, hosts that are considered :attr:`~.HostDistance.REMOTE`

by the :attr:`~.Cluster.load_balancing_policy` will have a connection

opened to them. Otherwise, they will not have a connection opened to them.

Note that the default load balancing policy ignores remote hosts by default.

.. versionadded:: 2.1.0

"""

metrics_enabled = False

"""

Whether or not metric collection is enabled. If enabled, :attr:`.metrics`

will be an instance of :class:`~cassandra.metrics.Metrics`.

"""

metrics = None

"""

An instance of :class:`cassandra.metrics.Metrics` if :attr:`.metrics_enabled` is

:const:`True`, else :const:`None`.

"""

ssl_options = None

"""

A optional dict which will be used as kwargs for ``ssl.wrap_socket()``

when new sockets are created. This should be used when client encryption

is enabled in Cassandra.

By default, a ``ca_certs`` value should be supplied (the value should be

a string pointing to the location of the CA certs file), and you probably

want to specify ``ssl_version`` as ``ssl.PROTOCOL_TLSv1`` to match

Cassandra's default protocol.

"""

sockopts = None

"""

An optional list of tuples which will be used as arguments to

``socket.setsockopt()`` for all created sockets.

Note: some drivers find setting TCPNODELAY beneficial in the context of

their execution model. It was not found generally beneficial for this driver.

To try with your own workload, set ``sockopts = [(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)]``

"""

max_schema_agreement_wait = 10

"""

The maximum duration (in seconds) that the driver will wait for schema

agreement across the cluster. Defaults to ten seconds.

If set <= 0, the driver will bypass schema agreement waits altogether.

"""

metadata = None

"""

An instance of :class:`cassandra.metadata.Metadata`.

"""

connection_class = DefaultConnection

"""

This determines what event loop system will be used for managing

I/O with Cassandra. These are the current options:

* :class:`cassandra.io.asyncorereactor.AsyncoreConnection`

* :class:`cassandra.io.libevreactor.LibevConnection`

* :class:`cassandra.io.eventletreactor.EventletConnection` (requires monkey-patching - see doc for details)

* :class:`cassandra.io.geventreactor.GeventConnection` (requires monkey-patching - see doc for details)

* :class:`cassandra.io.twistedreactor.TwistedConnection`

By default, ``AsyncoreConnection`` will be used, which uses

the ``asyncore`` module in the Python standard library.

If ``libev`` is installed, ``LibevConnection`` will be used instead.

If ``gevent`` or ``eventlet`` monkey-patching is detected, the corresponding

connection class will be used automatically.

"""

control_connection_timeout = 2.0

"""

A timeout, in seconds, for queries made by the control connection, such

as querying the current schema and information about nodes in the cluster.

If set to :const:`None`, there will be no timeout for these queries.

"""

idle_heartbeat_interval = 30

"""

Interval, in seconds, on which to heartbeat idle connections. This helps

keep connections open through network devices that expire idle connections.

It also helps discover bad connections early in low-traffic scenarios.

Setting to zero disables heartbeats.

"""

schema_event_refresh_window = 2

"""

Window, in seconds, within which a schema component will be refreshed after

receiving a schema_change event.

The driver delays a random amount of time in the range [0.0, window)

before executing the refresh. This serves two purposes:

1.) Spread the refresh for deployments with large fanout from C* to client tier,

preventing a 'thundering herd' problem with many clients refreshing simultaneously.

2.) Remove redundant refreshes. Redundant events arriving within the delay period

are discarded, and only one refresh is executed.

Setting this to zero will execute refreshes immediately.

Setting this negative will disable schema refreshes in response to push events

(refreshes will still occur in response to schema change responses to DDL statements

executed by Sessions of this Cluster).

"""

topology_event_refresh_window = 10

"""

Window, in seconds, within which the node and token list will be refreshed after

receiving a topology_change event.

Setting this to zero will execute refreshes immediately.

Setting this negative will disable node refreshes in response to push events

(refreshes will still occur in response to new nodes observed on "UP" events).

See :attr:`.schema_event_refresh_window` for discussion of rationale

"""

connect_timeout = 5

"""

Timeout, in seconds, for creating new connections.

This timeout covers the entire connection negotiation, including TCP

establishment, options passing, and authentication.

"""

sessions = None

control_connection = None

scheduler = None

executor = None

is_shutdown = False

_is_setup = False

_prepared_statements = None

_prepared_statement_lock = None

_idle_heartbeat = None

_user_types = None

"""

A map of {keyspace: {type_name: UserType}}

"""

_listeners = None

_listener_lock = None

def __init__(self,

contact_points=["127.0.0.1"],

port=9042,

compression=True,

auth_provider=None,

load_balancing_policy=None,

reconnection_policy=None,

default_retry_policy=None,

conviction_policy_factory=None,

metrics_enabled=False,

connection_class=None,

ssl_options=None,

sockopts=None,

cql_version=None,

protocol_version=4,

executor_threads=2,

max_schema_agreement_wait=10,

control_connection_timeout=2.0,

idle_heartbeat_interval=30,

schema_event_refresh_window=2,

topology_event_refresh_window=10,

connect_timeout=5):

"""

Any of the mutable Cluster attributes may be set as keyword arguments

to the constructor.

"""

if contact_points is not None:

if isinstance(contact_points, six.string_types):

raise TypeError("contact_points should not be a string, it should be a sequence (e.g. list) of strings")

self.contact_points = contact_points

self.port = port

self.compression = compression

self.protocol_version = protocol_version

self.auth_provider = auth_provider

if load_balancing_policy is not None:

if isinstance(load_balancing_policy, type):

raise TypeError("load_balancing_policy should not be a class, it should be an instance of that class")

self.load_balancing_policy = load_balancing_policy

else:

self.load_balancing_policy = default_lbp_factory()

if reconnection_policy is not None:

if isinstance(reconnection_policy, type):

raise TypeError("reconnection_policy should not be a class, it should be an instance of that class")

self.reconnection_policy = reconnection_policy

if default_retry_policy is not None:

if isinstance(default_retry_policy, type):

raise TypeError("default_retry_policy should not be a class, it should be an instance of that class")

self.default_retry_policy = default_retry_policy

if conviction_policy_factory is not None:

if not callable(conviction_policy_factory):

raise ValueError("conviction_policy_factory must be callable")

self.conviction_policy_factory = conviction_policy_factory

if connection_class is not None:

self.connection_class = connection_class

self.metrics_enabled = metrics_enabled

self.ssl_options = ssl_options

self.sockopts = sockopts

self.cql_version = cql_version

self.max_schema_agreement_wait = max_schema_agreement_wait

self.control_connection_timeout = control_connection_timeout

self.idle_heartbeat_interval = idle_heartbeat_interval

self.schema_event_refresh_window = schema_event_refresh_window

self.topology_event_refresh_window = topology_event_refresh_window

self.connect_timeout = connect_timeout

self._listeners = set()

self._listener_lock = Lock()

# let Session objects be GC'ed (and shutdown) when the user no longer

# holds a reference.

self.sessions = WeakSet()

self.metadata = Metadata()

self.control_connection = None

self._prepared_statements = WeakValueDictionary()

self._prepared_statement_lock = Lock()

self._user_types = defaultdict(dict)

self._min_requests_per_connection = {

HostDistance.LOCAL: DEFAULT_MIN_REQUESTS,

HostDistance.REMOTE: DEFAULT_MIN_REQUESTS

}

self._max_requests_per_connection = {

HostDistance.LOCAL: DEFAULT_MAX_REQUESTS,

HostDistance.REMOTE: DEFAULT_MAX_REQUESTS

}

self._core_connections_per_host = {

HostDistance.LOCAL: DEFAULT_MIN_CONNECTIONS_PER_LOCAL_HOST,

HostDistance.REMOTE: DEFAULT_MIN_CONNECTIONS_PER_REMOTE_HOST

}

self._max_connections_per_host = {

HostDistance.LOCAL: DEFAULT_MAX_CONNECTIONS_PER_LOCAL_HOST,

HostDistance.REMOTE: DEFAULT_MAX_CONNECTIONS_PER_REMOTE_HOST

}

self.executor = ThreadPoolExecutor(max_workers=executor_threads)

self.scheduler = _Scheduler(self.executor)

self._lock = RLock()

if self.metrics_enabled:

from cassandra.metrics import Metrics

self.metrics = Metrics(weakref.proxy(self))

self.control_connection = ControlConnection(

self, self.control_connection_timeout,

self.schema_event_refresh_window, self.topology_event_refresh_window)

def register_user_type(self, keyspace, user_type, klass):

"""

Registers a class to use to represent a particular user-defined type.

Query parameters for this user-defined type will be assumed to be

instances of `klass`. Result sets for this user-defined type will

be instances of `klass`. If no class is registered for a user-defined

type, a namedtuple will be used for result sets, and non-prepared

statements may not encode parameters for this type correctly.

`keyspace` is the name of the keyspace that the UDT is defined in.

`user_type` is the string name of the UDT to register the mapping

for.

`klass` should be a class with attributes whose names match the

fields of the user-defined type. The constructor must accepts kwargs

for each of the fields in the UDT.

This method should only be called after the type has been created

within Cassandra.

Example::

cluster = Cluster(protocol_version=3)

session = cluster.connect()

session.set_keyspace('mykeyspace')

session.execute("CREATE TYPE address (street text, zipcode int)")

session.execute("CREATE TABLE users (id int PRIMARY KEY, location address)")

# create a class to map to the "address" UDT

class Address(object):

def __init__(self, street, zipcode):

self.street = street

self.zipcode = zipcode

cluster.register_user_type('mykeyspace', 'address', Address)

# insert a row using an instance of Address

session.execute("INSERT INTO users (id, location) VALUES (%s, %s)",

(0, Address("123 Main St.", 78723)))

# results will include Address instances

results = session.execute("SELECT * FROM users")

row = results[0]

print row.id, row.location.street, row.location.zipcode

"""

if self.protocol_version < 3:

log.warning("User Type serialization is only supported in native protocol version 3+ (%d in use). "

"CQL encoding for simple statements will still work, but named tuples will "

"be returned when reading type %s.%s.", self.protocol_version, keyspace, user_type)

self._user_types[keyspace][user_type] = klass

for session in self.sessions:

session.user_type_registered(keyspace, user_type, klass)

UserType.evict_udt_class(keyspace, user_type)

def get_min_requests_per_connection(self, host_distance):

return self._min_requests_per_connection[host_distance]

def set_min_requests_per_connection(self, host_distance, min_requests):

"""

Sets a threshold for concurrent requests per connection, below which

connections will be considered for disposal (down to core connections;

see :meth:`~Cluster.set_core_connections_per_host`).

Pertains to connection pool management in protocol versions {1,2}.

"""

if self.protocol_version >= 3:

raise UnsupportedOperation(

"Cluster.set_min_requests_per_connection() only has an effect "

"when using protocol_version 1 or 2.")

self._min_requests_per_connection[host_distance] = min_requests

def get_max_requests_per_connection(self, host_distance):

return self._max_requests_per_connection[host_distance]

def set_max_requests_per_connection(self, host_distance, max_requests):

"""

Sets a threshold for concurrent requests per connection, above which new

connections will be created to a host (up to max connections;

see :meth:`~Cluster.set_max_connections_per_host`).

Pertains to connection pool management in protocol versions {1,2}.

"""

if self.protocol_version >= 3:

raise UnsupportedOperation(

"Cluster.set_max_requests_per_connection() only has an effect "

"when using protocol_version 1 or 2.")

self._max_requests_per_connection[host_distance] = max_requests

def get_core_connections_per_host(self, host_distance):

"""

Gets the minimum number of connections per Session that will be opened

for each host with :class:`~.HostDistance` equal to `host_distance`.

The default is 2 for :attr:`~HostDistance.LOCAL` and 1 for

:attr:`~HostDistance.REMOTE`.

This property is ignored if :attr:`~.Cluster.protocol_version` is

3 or higher.

"""

return self._core_connections_per_host[host_distance]

def set_core_connections_per_host(self, host_distance, core_connections):

"""

Sets the minimum number of connections per Session that will be opened

for each host with :class:`~.HostDistance` equal to `host_distance`.

The default is 2 for :attr:`~HostDistance.LOCAL` and 1 for

:attr:`~HostDistance.REMOTE`.

Protocol version 1 and 2 are limited in the number of concurrent

requests they can send per connection. The driver implements connection

pooling to support higher levels of concurrency.

If :attr:`~.Cluster.protocol_version` is set to 3 or higher, this

is not supported (there is always one connection per host, unless

the host is remote and :attr:`connect_to_remote_hosts` is :const:`False`)

and using this will result in an :exc:`~.UnsupporteOperation`.

"""

if self.protocol_version >= 3:

raise UnsupportedOperation(

"Cluster.set_core_connections_per_host() only has an effect "

"when using protocol_version 1 or 2.")

old = self._core_connections_per_host[host_distance]

self._core_connections_per_host[host_distance] = core_connections

if old < core_connections:

self._ensure_core_connections()

def get_max_connections_per_host(self, host_distance):

"""

Gets the maximum number of connections per Session that will be opened

for each host with :class:`~.HostDistance` equal to `host_distance`.

The default is 8 for :attr:`~HostDistance.LOCAL` and 2 for

:attr:`~HostDistance.REMOTE`.

This property is ignored if :attr:`~.Cluster.protocol_version` is

3 or higher.

"""

return self._max_connections_per_host[host_distance]

def set_max_connections_per_host(self, host_distance, max_connections):

"""

Sets the maximum number of connections per Session that will be opened

for each host with :class:`~.HostDistance` equal to `host_distance`.

The default is 2 for :attr:`~HostDistance.LOCAL` and 1 for

:attr:`~HostDistance.REMOTE`.

If :attr:`~.Cluster.protocol_version` is set to 3 or higher, this

is not supported (there is always one connection per host, unless

the host is remote and :attr:`connect_to_remote_hosts` is :const:`False`)

and using this will result in an :exc:`~.UnsupporteOperation`.

"""

if self.protocol_version >= 3:

raise UnsupportedOperation(

"Cluster.set_max_connections_per_host() only has an effect "

"when using protocol_version 1 or 2.")

self._max_connections_per_host[host_distance] = max_connections

def connection_factory(self, address, *args, **kwargs):

"""

Called to create a new connection with proper configuration.

Intended for internal use only.

"""

kwargs = self._make_connection_kwargs(address, kwargs)

return self.connection_class.factory(address, self.connect_timeout, *args, **kwargs)

def _make_connection_factory(self, host, *args, **kwargs):

kwargs = self._make_connection_kwargs(host.address, kwargs)

return partial(self.connection_class.factory, host.address, self.connect_timeout, *args, **kwargs)

def _make_connection_kwargs(self, address, kwargs_dict):

if self._auth_provider_callable:

kwargs_dict.setdefault('authenticator', self._auth_provider_callable(address))

kwargs_dict.setdefault('port', self.port)

kwargs_dict.setdefault('compression', self.compression)

kwargs_dict.setdefault('sockopts', self.sockopts)

kwargs_dict.setdefault('ssl_options', self.ssl_options)

kwargs_dict.setdefault('cql_version', self.cql_version)

kwargs_dict.setdefault('protocol_version', self.protocol_version)

kwargs_dict.setdefault('user_type_map', self._user_types)

return kwargs_dict

def protocol_downgrade(self, host_addr, previous_version):

new_version = previous_version - 1

if new_version < self.protocol_version:

if new_version >= MIN_SUPPORTED_VERSION:

log.warning("Downgrading core protocol version from %d to %d for %s", self.protocol_version, new_version, host_addr)

self.protocol_version = new_version

else:

raise Exception("Cannot downgrade protocol version (%d) below minimum supported version: %d" % (new_version, MIN_SUPPORTED_VERSION))

def connect(self, keyspace=None):

"""

Creates and returns a new :class:`~.Session` object. If `keyspace`

is specified, that keyspace will be the default keyspace for

operations on the ``Session``.

"""

with self._lock:

if self.is_shutdown:

raise Exception("Cluster is already shut down")

if not self._is_setup:

log.debug("Connecting to cluster, contact points: %s; protocol version: %s",

self.contact_points, self.protocol_version)

self.connection_class.initialize_reactor()

atexit.register(partial(_shutdown_cluster, self))

for address in self.contact_points:

host, new = self.add_host(address, signal=False)

if new:

host.set_up()

for listener in self.listeners:

listener.on_add(host)

self.load_balancing_policy.populate(

weakref.proxy(self), self.metadata.all_hosts())

try:

self.control_connection.connect()

log.debug("Control connection created")

except Exception:

log.exception("Control connection failed to connect, "

"shutting down Cluster:")

self.shutdown()

raise

self.load_balancing_policy.check_supported()

if self.idle_heartbeat_interval:

self._idle_heartbeat = ConnectionHeartbeat(self.idle_heartbeat_interval, self.get_connection_holders)

self._is_setup = True

session = self._new_session()

if keyspace:

session.set_keyspace(keyspace)

return session

def get_connection_holders(self):

holders = []

for s in self.sessions:

holders.extend(s.get_pools())

holders.append(self.control_connection)

return holders

def shutdown(self):

"""

Closes all sessions and connection associated with this Cluster.

To ensure all connections are properly closed, **you should always

call shutdown() on a Cluster instance when you are done with it**.

Once shutdown, a Cluster should not be used for any purpose.

"""

with self._lock:

if self.is_shutdown:

return

else:

self.is_shutdown = True

if self._idle_heartbeat:

self._idle_heartbeat.stop()

self.scheduler.shutdown()

self.control_connection.shutdown()

for session in self.sessions:

session.shutdown()

self.executor.shutdown()

def _new_session(self):

session = Session(self, self.metadata.all_hosts())

self._session_register_user_types(session)

self.sessions.add(session)

return session

def _session_register_user_types(self, session):

for keyspace, type_map in six.iteritems(self._user_types):

for udt_name, klass in six.iteritems(type_map):

session.user_type_registered(keyspace, udt_name, klass)

def _cleanup_failed_on_up_handling(self, host):

self.load_balancing_policy.on_down(host)

self.control_connection.on_down(host)

for session in self.sessions:

session.remove_pool(host)

self._start_reconnector(host, is_host_addition=False)

def _on_up_future_completed(self, host, futures, results, lock, finished_future):

with lock:

futures.discard(finished_future)

try:

results.append(finished_future.result())

except Exception as exc:

results.append(exc)

if futures:

return

try:

# all futures have completed at this point

for exc in [f for f in results if isinstance(f, Exception)]:

log.error("Unexpected failure while marking node %s up:", host, exc_info=exc)

self._cleanup_failed_on_up_handling(host)

return

if not all(results):

log.debug("Connection pool could not be created, not marking node %s up", host)

self._cleanup_failed_on_up_handling(host)

return

log.info("Connection pools established for node %s", host)

# mark the host as up and notify all listeners

host.set_up()

for listener in self.listeners:

listener.on_up(host)

finally:

with host.lock:

host._currently_handling_node_up = False

# see if there are any pools to add or remove now that the host is marked up

for session in self.sessions:

session.update_created_pools()

def on_up(self, host):

"""

Intended for internal use only.

"""

if self.is_shutdown:

return

log.debug("Waiting to acquire lock for handling up status of node %s", host)

with host.lock:

if host._currently_handling_node_up:

log.debug("Another thread is already handling up status of node %s", host)

return

if host.is_up:

log.debug("Host %s was already marked up", host)

return

host._currently_handling_node_up = True

log.debug("Starting to handle up status of node %s", host)

have_future = False

futures = set()

try:

log.info("Host %s may be up; will prepare queries and open connection pool", host)

reconnector = host.get_and_set_reconnection_handler(None)

if reconnector:

log.debug("Now that host %s is up, cancelling the reconnection handler", host)

reconnector.cancel()

self._prepare_all_queries(host)

log.debug("Done preparing all queries for host %s, ", host)

for session in self.sessions:

session.remove_pool(host)

log.debug("Signalling to load balancing policy that host %s is up", host)

self.load_balancing_policy.on_up(host)

log.debug("Signalling to control connection that host %s is up", host)

self.control_connection.on_up(host)

log.debug("Attempting to open new connection pools for host %s", host)

futures_lock = Lock()

futures_results = []

callback = partial(self._on_up_future_completed, host, futures, futures_results, futures_lock)

for session in self.sessions:

future = session.add_or_renew_pool(host, is_host_addition=False)

if future is not None: