When using v3 of the native protocol, this is used instead of a connection pool per host (HostConnectionPool) due to the increased in-flight capacity of individual connections.
| 368 | |
| 369 | |
| 370 | class HostConnection(object): |
| 371 | """ |
| 372 | When using v3 of the native protocol, this is used instead of a connection |
| 373 | pool per host (HostConnectionPool) due to the increased in-flight capacity |
| 374 | of individual connections. |
| 375 | """ |
| 376 | |
| 377 | host = None |
| 378 | host_distance = None |
| 379 | is_shutdown = False |
| 380 | shutdown_on_error = False |
| 381 | |
| 382 | _session = None |
| 383 | _connection = None |
| 384 | _lock = None |
| 385 | _keyspace = None |
| 386 | |
| 387 | def __init__(self, host, host_distance, session): |
| 388 | self.host = host |
| 389 | self.host_distance = host_distance |
| 390 | self._session = weakref.proxy(session) |
| 391 | self._lock = Lock() |
| 392 | # this is used in conjunction with the connection streams. Not using the connection lock because the connection can be replaced in the lifetime of the pool. |
| 393 | self._stream_available_condition = Condition(self._lock) |
| 394 | self._is_replacing = False |
| 395 | # Contains connections which shouldn't be used anymore |
| 396 | # and are waiting until all requests time out or complete |
| 397 | # so that we can dispose of them. |
| 398 | self._trash = set() |
| 399 | |
| 400 | if host_distance == HostDistance.IGNORED: |
| 401 | log.debug("Not opening connection to ignored host %s", self.host) |
| 402 | return |
| 403 | elif host_distance == HostDistance.REMOTE and not session.cluster.connect_to_remote_hosts: |
| 404 | log.debug("Not opening connection to remote host %s", self.host) |
| 405 | return |
| 406 | |
| 407 | log.debug("Initializing connection for host %s", self.host) |
| 408 | self._connection = session.cluster.connection_factory(host.endpoint, on_orphaned_stream_released=self.on_orphaned_stream_released) |
| 409 | self._keyspace = session.keyspace |
| 410 | if self._keyspace: |
| 411 | self._connection.set_keyspace_blocking(self._keyspace) |
| 412 | log.debug("Finished initializing connection for host %s", self.host) |
| 413 | |
| 414 | def _get_connection(self): |
| 415 | if self.is_shutdown: |
| 416 | raise ConnectionException( |
| 417 | "Pool for %s is shutdown" % (self.host,), self.host) |
| 418 | |
| 419 | conn = self._connection |
| 420 | if not conn: |
| 421 | raise NoConnectionsAvailable() |
| 422 | return conn |
| 423 | |
| 424 | def borrow_connection(self, timeout): |
| 425 | conn = self._get_connection() |
| 426 | if conn.orphaned_threshold_reached: |
| 427 | with self._lock: |
no outgoing calls
no test coverage detected