Creates a new Connection, registers for pushed events, and refreshes node/token and schema metadata.
(self, host)
| 3645 | raise NoHostAvailable("Unable to connect to any servers", errors) |
| 3646 | |
| 3647 | def _try_connect(self, host): |
| 3648 | """ |
| 3649 | Creates a new Connection, registers for pushed events, and refreshes |
| 3650 | node/token and schema metadata. |
| 3651 | """ |
| 3652 | log.debug("[control connection] Opening new connection to %s", host) |
| 3653 | |
| 3654 | while True: |
| 3655 | try: |
| 3656 | connection = self._cluster.connection_factory(host.endpoint, is_control_connection=True) |
| 3657 | if self._is_shutdown: |
| 3658 | connection.close() |
| 3659 | raise DriverException("Reconnecting during shutdown") |
| 3660 | break |
| 3661 | except ProtocolVersionUnsupported as e: |
| 3662 | self._cluster.protocol_downgrade(host.endpoint, e.startup_version) |
| 3663 | except ProtocolException as e: |
| 3664 | # protocol v5 is out of beta in C* >=4.0-beta5 and is now the default driver |
| 3665 | # protocol version. If the protocol version was not explicitly specified, |
| 3666 | # and that the server raises a beta protocol error, we should downgrade. |
| 3667 | if not self._cluster._protocol_version_explicit and e.is_beta_protocol_error: |
| 3668 | self._cluster.protocol_downgrade(host.endpoint, self._cluster.protocol_version) |
| 3669 | else: |
| 3670 | raise |
| 3671 | |
| 3672 | log.debug("[control connection] Established new connection %r, " |
| 3673 | "registering watchers and refreshing schema and topology", |
| 3674 | connection) |
| 3675 | |
| 3676 | # use weak references in both directions |
| 3677 | # _clear_watcher will be called when this ControlConnection is about to be finalized |
| 3678 | # _watch_callback will get the actual callback from the Connection and relay it to |
| 3679 | # this object (after a dereferencing a weakref) |
| 3680 | self_weakref = weakref.ref(self, partial(_clear_watcher, weakref.proxy(connection))) |
| 3681 | try: |
| 3682 | connection.register_watchers({ |
| 3683 | "TOPOLOGY_CHANGE": partial(_watch_callback, self_weakref, '_handle_topology_change'), |
| 3684 | "STATUS_CHANGE": partial(_watch_callback, self_weakref, '_handle_status_change'), |
| 3685 | "SCHEMA_CHANGE": partial(_watch_callback, self_weakref, '_handle_schema_change') |
| 3686 | }, register_timeout=self._timeout) |
| 3687 | |
| 3688 | sel_peers = self._get_peers_query(self.PeersQueryType.PEERS, connection) |
| 3689 | sel_local = self._SELECT_LOCAL if self._token_meta_enabled else self._SELECT_LOCAL_NO_TOKENS |
| 3690 | peers_query = QueryMessage(query=sel_peers, consistency_level=ConsistencyLevel.ONE) |
| 3691 | local_query = QueryMessage(query=sel_local, consistency_level=ConsistencyLevel.ONE) |
| 3692 | (peers_success, peers_result), (local_success, local_result) = connection.wait_for_responses( |
| 3693 | peers_query, local_query, timeout=self._timeout, fail_on_error=False) |
| 3694 | |
| 3695 | if not local_success: |
| 3696 | raise local_result |
| 3697 | |
| 3698 | if not peers_success: |
| 3699 | # error with the peers v2 query, fallback to peers v1 |
| 3700 | self._uses_peers_v2 = False |
| 3701 | sel_peers = self._get_peers_query(self.PeersQueryType.PEERS, connection) |
| 3702 | peers_query = QueryMessage(query=sel_peers, consistency_level=ConsistencyLevel.ONE) |
| 3703 | peers_result = connection.wait_for_response( |
| 3704 | peers_query, timeout=self._timeout) |
no test coverage detected