(self, connection)
| 504 | self._stream_available_condition.notify() |
| 505 | |
| 506 | def _replace(self, connection): |
| 507 | with self._lock: |
| 508 | if self.is_shutdown: |
| 509 | return |
| 510 | |
| 511 | log.debug("Replacing connection (%s) to %s", id(connection), self.host) |
| 512 | try: |
| 513 | conn = self._session.cluster.connection_factory(self.host.endpoint, on_orphaned_stream_released=self.on_orphaned_stream_released) |
| 514 | if self._keyspace: |
| 515 | conn.set_keyspace_blocking(self._keyspace) |
| 516 | self._connection = conn |
| 517 | except Exception: |
| 518 | log.warning("Failed reconnecting %s. Retrying." % (self.host.endpoint,)) |
| 519 | self._session.submit(self._replace, connection) |
| 520 | else: |
| 521 | with connection.lock: |
| 522 | with self._lock: |
| 523 | if connection.orphaned_threshold_reached: |
| 524 | if connection.in_flight == len(connection.orphaned_request_ids): |
| 525 | connection.close() |
| 526 | else: |
| 527 | self._trash.add(connection) |
| 528 | self._is_replacing = False |
| 529 | self._stream_available_condition.notify() |
| 530 | |
| 531 | def shutdown(self): |
| 532 | with self._lock: |
nothing calls this directly
no test coverage detected