def export()

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

• 33 lines of code
• 5 McCabe index (conditional complexity)

    def export(self):
        """
        Export the server.
        """
        if self.is_exported():
            return
        elif self.is_closed():
            raise RemotingError("The server is closed.")

        async def _inner_operation(_future: concurrent.futures.Future):
            try:
                running_loop = asyncio.get_running_loop()
                server = await running_loop.create_server(
                    lambda: self._url.attributes[common_constants.PROTOCOL_KEY](self._url),
                    self._url.host,
                    self._url.port,
                )

                # Serve the server forever
                async with server:
                    FutureHelper.set_result(_future, None)
                    await server.serve_forever()
            except Exception as e:
                FutureHelper.set_exception(_future, e)

        # Run the server logic in the event loop.
        future = concurrent.futures.Future()
        asyncio.run_coroutine_threadsafe(_inner_operation(future), self._event_loop.loop)

        try:
            exc = future.exception()
            if exc:
                raise RemotingError("Failed to export the server") from exc
            else:
                self._exported = True
            _LOGGER.info(
                "Exported the server. host: %s, port: %s",
                self._url.host,
                self._url.port,
            )
        finally:
            self._exporting = False