Python asyncio 模块，start_unix_server() 实例源码

def start(self):
        """Start accepting connections.

        Only use this method if you are not using the server as an asynchronous
        context manager.

        See:

        - :py:meth:`aiotk.TCPServer.wait_started`
        - :py:meth:`aiotk.TCPServer.close`
        """
        if self._server:
            raise Exception('Already running.')
        self._server = self._loop.create_task(asyncio.start_unix_server(
            self._client_connected, path=self._path,
        ))

    # NOTE: start is synchronous, so we can't await in there.

def startup(self, port=None, socket_path=None):
        if port:
            await asyncio.start_server(
                self._client_connected,
                port=port,
                loop=self.loop,
            )
            logger.info('server started on port %s' % (port,))
        elif socket_path:
            await asyncio.start_unix_server(
                self._client_connected,
                path=socket_path,
                loop=self.loop,
            )
            logger.info('server started on unix path %s' % (socket_path,))
        else:
            raise ValueError('one of port and socket_path must be provided.')

        for startup in self.startups:
            await startup(self.loop, self)

def run_unix_socket_server(on_connection, socket_path=None, loop=None):
    loop = loop or asyncio.get_event_loop()
    socket_path = socket_path or mktemp()

    class UnixSocketServerContextManager:
        async def __aenter__(self):
            server_task = await asyncio.start_unix_server(
                on_connection, path=socket_path, loop=loop)
            self.context = UnixSocketServer(socket_path, on_connection, server_task)
            return self.context

        async def __aexit__(self, *args):
            await self.context.stop()

        async def __await__(self):
            return await self.__aenter__()

    return UnixSocketServerContextManager()

def run_unix_socket_server(on_connection, socket_path=None, loop=None):
    loop = loop or asyncio.get_event_loop()
    socket_path = socket_path or mktemp()

    class UnixSocketServerContextManager:
        async def __aenter__(self):
            # Enjoy asyncio in it full glory: we must register connection
            # tasks to be able to cancel them when the server is closed.
            # Yes, there is like a rotten taste of C in my Python :'-(
            connections_made = set()

            def on_connection_register(reader, writer):
                coro = on_connection(reader, writer)
                fut = asyncio.ensure_future(coro)
                connections_made.add(fut)
                fut.add_done_callback(lambda res: connections_made.remove(fut))
                return fut

            server_task = await asyncio.start_unix_server(
                on_connection_register, path=socket_path, loop=loop)
            self.context = UnixSocketServer(
                socket_path, on_connection, server_task, connections_made)
            return self.context

        async def __aexit__(self, *args):
            await self.context.stop()

        async def __await__(self):
            return await self.__aenter__()

    return UnixSocketServerContextManager()

def start(path, chain=None, loop=None):
    '''
    :returns: initialized server
    :rtype: asyncio.Server
    '''
    if loop is None:
        loop = asyncio.get_event_loop()
    rpc = RPCServer(chain)
    server = loop.run_until_complete(asyncio.start_unix_server(
        connection_handler(rpc.execute),
        path,
        loop=loop,
        limit=MAXIMUM_REQUEST_BYTES,
    ))
    return server

def start_ipc_server(*args, **kwargs):
        try:
            os.unlink(kwargs['path'])
        except FileNotFoundError:
            pass

        return await start_unix_server(*args, **kwargs)

def uri_compile(uri):
    url = urllib.parse.urlparse(uri)
    rawprotos = url.scheme.split('+')
    err_str, protos = proto.get_protos(rawprotos)
    if err_str:
        raise argparse.ArgumentTypeError(err_str)
    if 'ssl' in rawprotos or 'secure' in rawprotos:
        import ssl
        if not hasattr(ssl, 'Purpose'):
            raise argparse.ArgumentTypeError('ssl support is available for Python 3.4 and above')
        sslserver = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
        sslclient = ssl.create_default_context(ssl.Purpose.SERVER_AUTH)
        if 'ssl' in rawprotos:
            sslclient.check_hostname = False
            sslclient.verify_mode = ssl.CERT_NONE
    else:
        sslserver = None
        sslclient = None
    cipher, _, loc = url.netloc.rpartition('@')
    if cipher:
        from pproxy.cipher import get_cipher
        err_str, cipher = get_cipher(cipher)
        if err_str:
            raise argparse.ArgumentTypeError(err_str)
    match = pattern_compile(url.query) if url.query else None
    if loc:
        host, _, port = loc.partition(':')
        port = int(port) if port else 8080
        connect = functools.partial(asyncio.open_connection, host=host, port=port, ssl=sslclient)
        server = functools.partial(asyncio.start_server, host=host, port=port, ssl=sslserver)
    else:
        connect = functools.partial(asyncio.open_unix_connection, path=url.path, ssl=sslclient, server_hostname='' if sslclient else None)
        server = functools.partial(asyncio.start_unix_server, path=url.path, ssl=sslserver)
    return types.SimpleNamespace(protos=protos, rproto=protos[0], cipher=cipher, auth=url.fragment.encode(), match=match, server=server, connect=connect, bind=loc or url.path, sslclient=sslclient, sslserver=sslserver)