'Attempt to connect and return ConnectionState'

def connect(self):

    if (self.state is ConnectionStates.DISCONNECTED):
        log.debug('%s:    creating    new    socket', self)
        if (self._init_afi == socket.AF_UNSPEC):
            if (self._gai is None):
                try:
                    self._gai = socket.getaddrinfo(self._init_host, self._init_port, socket.AF_UNSPEC, socket.SOCK_STREAM)
                except socket.gaierror as ex:
                    log.warning('DNS    lookup    failed    for    %s:%d,    exception    was    %s.    Is    your    advertised.listeners    (called    advertised.host.name    before    Kafka    9)    correct    and    resolvable?', self._init_host, self._init_port, ex)
                    self._gai = []
                self._gai_index = 0
            else:
                self._gai_index += 1
            while True:
                if (self._gai_index >= len(self._gai)):
                    error = 'Unable    to    connect    to    any    of    the    names    for    {0}:{1}'.format(self._init_host, self._init_port)
                    log.error(error)
                    self.close(Errors.ConnectionError(error))
                    return
                (afi, _, __, ___, sockaddr) = self._gai[self._gai_index]
                if (afi not in (socket.AF_INET, socket.AF_INET6)):
                    self._gai_index += 1
                    continue
                break
            (self.host, self.port) = sockaddr[:2]
            self._sock = socket.socket(afi, socket.SOCK_STREAM)
        else:
            self._sock = socket.socket(self._init_afi, socket.SOCK_STREAM)
        for option in self.config['socket_options']:
            log.debug('%s:    setting    socket    option    %s', self, option)
            self._sock.setsockopt(*option)
        self._sock.setblocking(False)
        if (self.config['security_protocol'] in ('SSL', 'SASL_SSL')):
            self._wrap_ssl()
        log.info('%s:    connecting    to    %s:%d', self, self.host, self.port)
        self.state = ConnectionStates.CONNECTING
        self.last_attempt = time.time()
        self.config['state_change_callback'](self)
    if (self.state is ConnectionStates.CONNECTING):
        request_timeout = (self.config['request_timeout_ms'] / 1000.0)
        ret = None
        try:
            ret = self._sock.connect_ex((self.host, self.port))
            if (self._gai is not None):
                self.afi = self._sock.family
                self._gai = None
        except socket.error as err:
            ret = err.errno
        if ((not ret) or (ret == errno.EISCONN)):
            log.debug('%s:    established    TCP    connection', self)
            if (self.config['security_protocol'] in ('SSL', 'SASL_SSL')):
                log.debug('%s:    initiating    SSL    handshake', self)
                self.state = ConnectionStates.HANDSHAKE
            elif (self.config['security_protocol'] == 'SASL_PLAINTEXT'):
                log.debug('%s:    initiating    SASL    authentication', self)
                self.state = ConnectionStates.AUTHENTICATING
            else:
                log.debug('%s:    Connection    complete.', self)
                self.state = ConnectionStates.CONNECTED
                self._reset_reconnect_backoff()
            self.config['state_change_callback'](self)
        elif (ret not in (errno.EINPROGRESS, errno.EALREADY, errno.EWOULDBLOCK, 10022)):
            log.error('Connect    attempt    to    %s    returned    error    %s.    Disconnecting.', self, ret)
            self.close(Errors.ConnectionError(ret))
        elif (time.time() > (request_timeout + self.last_attempt)):
            log.error('Connection    attempt    to    %s    timed    out', self)
            self.close(Errors.ConnectionError('timeout'))
        else:
            pass
    if (self.state is ConnectionStates.HANDSHAKE):
        if self._try_handshake():
            log.debug('%s:    completed    SSL    handshake.', self)
            if (self.config['security_protocol'] == 'SASL_SSL'):
                log.debug('%s:    initiating    SASL    authentication', self)
                self.state = ConnectionStates.AUTHENTICATING
            else:
                log.debug('%s:    Connection    complete.', self)
                self.state = ConnectionStates.CONNECTED
            self.config['state_change_callback'](self)
    if (self.state is ConnectionStates.AUTHENTICATING):
        assert (self.config['security_protocol'] in ('SASL_PLAINTEXT', 'SASL_SSL'))
        if self._try_authenticate():
            log.info('%s:    Authenticated    as    %s', self, self.config['sasl_plain_username'])
            log.debug('%s:    Connection    complete.', self)
            self.state = ConnectionStates.CONNECTED
            self._reset_reconnect_backoff()
            self.config['state_change_callback'](self)
    return self.state
