1 27
import os
2 27
from contextlib import contextmanager
3

4 27
import trio
5 27
from trio.socket import socket, SOCK_STREAM
6

7 27
try:
8 27
    from trio.socket import AF_UNIX
9

10 16
    has_unix = True
11 11
except ImportError:
12 11
    has_unix = False
13

14

15 27
@contextmanager
16 13
def close_on_error(obj):
17 16
    try:
18 16
        yield obj
19 16
    except:
20 16
        obj.close()
21 16
        raise
22

23

24 27
async def open_unix_socket(filename):
25
    """Opens a connection to the specified
26
    `Unix domain socket <https://en.wikipedia.org/wiki/Unix_domain_socket>`__.
27

28
    You must have read/write permission on the specified file to connect.
29

30
    Args:
31
      filename (str or bytes): The filename to open the connection to.
32

33
    Returns:
34
      SocketStream: a :class:`~trio.abc.Stream` connected to the given file.
35

36
    Raises:
37
      OSError: If the socket file could not be connected to.
38
      RuntimeError: If AF_UNIX sockets are not supported.
39
    """
40 27
    if not has_unix:
41 0
        raise RuntimeError("Unix sockets are not supported on this platform")
42

43
    # much more simplified logic vs tcp sockets - one socket type and only one
44
    # possible location to connect to
45 16
    sock = socket(AF_UNIX, SOCK_STREAM)
46 16
    with close_on_error(sock):
47 16
        await sock.connect(os.fspath(filename))
48

49 16
    return trio.SocketStream(sock)

Read our documentation on viewing source code .

Loading