'Creates a new Connection, registers for pushed events, and refreshes
node/token and schema metadata.'

def _try_connect(self, host):

    log.debug('[control    connection]    Opening    new    connection    to    %s', host)
    while True:
        try:
            connection = self._cluster.connection_factory(host.address, is_control_connection=True)
            if self._is_shutdown:
                connection.close()
                raise DriverException('Reconnecting    during    shutdown')
            break
        except ProtocolVersionUnsupported as e:
            self._cluster.protocol_downgrade(host.address, e.startup_version)
    log.debug('[control    connection]    Established    new    connection    %r,    registering    watchers    and    refreshing    schema    and    topology', connection)
    self_weakref = weakref.ref(self, partial(_clear_watcher, weakref.proxy(connection)))
    try:
        connection.register_watchers({'TOPOLOGY_CHANGE': partial(_watch_callback, self_weakref, '_handle_topology_change'), 'STATUS_CHANGE': partial(_watch_callback, self_weakref, '_handle_status_change'), 'SCHEMA_CHANGE': partial(_watch_callback, self_weakref, '_handle_schema_change')}, register_timeout=self._timeout)
        sel_peers = (self._SELECT_PEERS if self._token_meta_enabled else self._SELECT_PEERS_NO_TOKENS)
        sel_local = (self._SELECT_LOCAL if self._token_meta_enabled else self._SELECT_LOCAL_NO_TOKENS)
        peers_query = QueryMessage(query=sel_peers, consistency_level=ConsistencyLevel.ONE)
        local_query = QueryMessage(query=sel_local, consistency_level=ConsistencyLevel.ONE)
        shared_results = connection.wait_for_responses(peers_query, local_query, timeout=self._timeout)
        self._refresh_node_list_and_token_map(connection, preloaded_results=shared_results)
        self._refresh_schema(connection, preloaded_results=shared_results, schema_agreement_wait=(-1))
    except Exception:
        connection.close()
        raise
    return connection
