def connect()

in src/dubbo/remoting/aio/aio_transporter.py [0:0]

• 21 lines of code
• 6 McCabe index (conditional complexity)

    def connect(self) -> None:
        """
        Connect to the server.
        """
        with self._global_lock:
            if self.is_connected():
                return
            elif self.is_closed():
                raise RemotingError("The client is closed.")

            # Run the connection logic in the event loop.
            if self._event_loop.stopped:
                raise RemotingError("The event loop is stopped.")
            elif not self._event_loop.started:
                self._event_loop.start()

            future = concurrent.futures.Future()
            asyncio.run_coroutine_threadsafe(self._do_connect(future), self._event_loop.loop)

            try:
                self._protocol = future.result(timeout=3)
                _LOGGER.info(
                    "Connected to the server. host: %s, port: %s",
                    self._url.host,
                    self._url.port,
                )
            except Exception:
                raise RemotingError(f"Failed to connect to the server. host: {self._url.host}, port: {self._url.port}")