Similar to `open_connection` but works with UNIX Domain Sockets.
(path=None, *,
limit=_DEFAULT_LIMIT, **kwds)
| 88 | # UNIX Domain Sockets are supported on this platform |
| 89 | |
| 90 | async def open_unix_connection(path=None, *, |
| 91 | limit=_DEFAULT_LIMIT, **kwds): |
| 92 | """Similar to `open_connection` but works with UNIX Domain Sockets.""" |
| 93 | loop = events.get_running_loop() |
| 94 | |
| 95 | reader = StreamReader(limit=limit, loop=loop) |
| 96 | protocol = StreamReaderProtocol(reader, loop=loop) |
| 97 | transport, _ = await loop.create_unix_connection( |
| 98 | lambda: protocol, path, **kwds) |
| 99 | writer = StreamWriter(transport, protocol, reader, loop) |
| 100 | return reader, writer |
| 101 | |
| 102 | async def start_unix_server(client_connected_cb, path=None, *, |
| 103 | limit=_DEFAULT_LIMIT, **kwds): |
nothing calls this directly
no test coverage detected
searching dependent graphs…