Repository: agronholm/anyio

Issue:
Using "recvmsg" with anyio
Using `wait_socket_readable` fails when using asyncio.

Seems that asyncio decides to add a socket reader as soon as the socket is opened – which causes `BaseSelectorEventLoop._ensure_fd_no_transport` to go splat when I try to check for readability.

This test program succeeds with trio, but not with asyncio:
```python3
#!/usr/bin/python3

# boring stuff
import anyio
PORT=59493
async def main():
    async def handler(sock):
        await sock.send(b'123')
        await anyio.sleep(9999)
    async def server(evt):
        listener = await anyio.create_tcp_listener(local_port=PORT)
        await evt.set()
        await listener.serve(handler)

    async with anyio.create_task_group() as tg:
        evt = anyio.create_event()
        await tg.spawn(server, evt)
        await evt.wait()
# boring stuff ends

        async with await anyio.connect_tcp('localhost', PORT) as client:
            sock = client.extra(anyio.abc.SocketAttribute.raw_socket)
            await anyio.wait_socket_readable(sock.fileno())
            data, aux, *_ = sock.recvmsg(4096,1024)
            assert data == b'123', data

            await tg.cancel_scope.cancel()
anyio.run(main, backend="asyncio")
```
I'm doing this because depending on the options it's called with, the library I'm anyio-izing may need to pass file descriptors along.

Discussion:
agronholm: That's understandable. Thanks for taking the time to peek at it at least!

Patch:
diff --git a/docs/networking.rst b/docs/networking.rst
index 83ff4ba..023af9d 100644
--- a/docs/networking.rst
+++ b/docs/networking.rst
@@ -101,6 +101,53 @@ And the listener::
 
     run(main)
 
+Sending and receiving file descriptors
+++++++++++++++++++++++++++++++++++++++
+
+UNIX sockets can be used to pass open file descriptors (sockets and files) to another process.
+The receiving end can then use either :func:`os.fdopen` or :func:`socket.socket` to get a usable
+file or socket object, respectively.
+
+The following is an example where a client connects to a UNIX socket server and receives the
+descriptor of a file opened on the server, reads the contents of the file and then prints them on
+standard output.
+
+Client::
+
+    import os
+
+    from anyio import connect_unix, run
+
+
+    async def main():
+        async with await connect_unix('/tmp/mysock') as client:
+            _, fds = await client.receive_fds(0, 1)
+            with os.fdopen(fds[0]) as file:
+                print(file.read())
+
+    run(main)
+
+Server::
+
+    from pathlib import Path
+
+    from anyio import create_unix_listener, run
+
+
+    async def handle(client):
+        async with client:
+            with path.open('r') as file:
+                await client.send_fds(b'this message is ignored', [file])
+
+
+    async def main():
+        listener = await create_unix_listener('/tmp/mysock')
+        await listener.serve(handle)
+
+    path = Path('/tmp/examplefile')
+    path.write_text('Test file')
+    run(main)
+
 Working with UDP sockets
 ------------------------
 
@@ -139,4 +186,3 @@ you send data to the peer::
             await udp.send(b'Hi there!\n')
 
     run(main)
-
diff --git a/docs/versionhistory.rst b/docs/versionhistory.rst
index 606eff7..bff403b 100644
--- a/docs/versionhistory.rst
+++ b/docs/versionhistory.rst
@@ -48,6 +48,8 @@ This library adheres to `Semantic Versioning 2.0 <http://semver.org/>`_.
   non-generic
 - Dropped Curio as a backend (see the :doc:`FAQ <faq>` as for why)
 - Added the ``run_sync_from_thread()`` function
+- Added ``UNIXSocketStream`` as a ``SocketStream`` subclass, capable of sending and receiving
+  file descriptors
 - Added the ``FileReadStream`` and ``FileWriteStream`` classes
 - Added the ``TaskGroup.start()`` method and a corresponding ``BlockingPortal.start_task()`` method
 - Added the ``name`` argument to ``BlockingPortal.spawn_task()``
diff --git a/src/anyio/_backends/_asyncio.py b/src/anyio/_backends/_asyncio.py
index 8b85c43..b730f02 100644
--- a/src/anyio/_backends/_asyncio.py
+++ b/src/anyio/_backends/_asyncio.py
@@ -1,3 +1,4 @@
+import array
 import asyncio
 import concurrent.futures
 import math
@@ -9,12 +10,13 @@ from contextlib import contextmanager
 from dataclasses import dataclass
 from functools import partial, wraps
 from inspect import isgenerator
+from io import IOBase
 from socket import AddressFamily, SocketKind, SocketType
 from threading import Thread
 from types import TracebackType
 from typing import (
-    Any, Awaitable, Callable, Coroutine, Deque, Dict, Generator, List, Optional, Sequence, Set,
-    Tuple, Type, TypeVar, Union, cast)
+    Any, Awaitable, Callable, Collection, Coroutine, Deque, Dict, Generator, List, Optional,
+    Sequence, Set, Tuple, Type, TypeVar, Union, cast)
 from weakref import WeakKeyDictionary
 
 from .. import CapacityLimiterStatistics, EventStatistics, TaskInfo, abc
@@ -863,7 +865,162 @@ class SocketStream(abc.SocketStream):
             self._transport.abort()
 
 
-class SocketListener(abc.SocketListener):
+class UNIXSocketStream(abc.SocketStream):
+    _receive_future: Optional[asyncio.Future] = None
+    _send_future: Optional[asyncio.Future] = None
+    _closing = False
+
+    def __init__(self, raw_socket: socket.SocketType):
+        self.__raw_socket = raw_socket
+        self._loop = get_running_loop()
+        self._receive_guard = ResourceGuard('reading from')
+        self._send_guard = ResourceGuard('writing to')
+
+    @property
+    def _raw_socket(self) -> SocketType:
+        return self.__raw_socket
+
+    def _wait_until_readable(self, loop: asyncio.AbstractEventLoop) -> asyncio.Future:
+        def callback(f):
+            del self._receive_future
+            loop.remove_reader(self.__raw_socket)
+
+        f = self._receive_future = asyncio.Future()
+        self._loop.add_reader(self.__raw_socket, f.set_result, None)
+        f.add_done_callback(callback)
+        return f
+
+    def _wait_until_writable(self, loop: asyncio.AbstractEventLoop) -> asyncio.Future:
+        def callback(f):
+            del self._send_future
+            loop.remove_writer(self.__raw_socket)
+
+        f = self._send_future = asyncio.Future()
+        self._loop.add_writer(self.__raw_socket, f.set_result, None)
+        f.add_done_callback(callback)
+        return f
+
+    async def send_eof(self) -> None:
+        with self._send_guard:
+            self._raw_socket.shutdown(socket.SHUT_WR)
+
+    async def receive(self, max_bytes: int = 65536) -> bytes:
+        loop = get_running_loop()
+        await checkpoint()
+        with self._receive_guard:
+            while True:
+                try:
+                    data = self.__raw_socket.recv(max_bytes)
+                except BlockingIOError:
+                    await self._wait_until_readable(loop)
+                except OSError as exc:
+                    if self._closing:
+                        raise ClosedResourceError from None
+                    else:
+                        raise BrokenResourceError from exc
+                else:
+                    if not data:
+                        raise EndOfStream
+
+                    return data
+
+    async def send(self, item: bytes) -> None:
+        loop = get_running_loop()
+        await checkpoint()
+        with self._send_guard:
+            view = memoryview(item)
+            while view:
+                try:
+                    bytes_sent = self.__raw_socket.send(item)
+                except BlockingIOError:
+                    await self._wait_until_writable(loop)
+                except OSError as exc:
+                    if self._closing:
+                        raise ClosedResourceError from None
+                    else:
+                        raise BrokenResourceError from exc
+                else:
+                    view = view[bytes_sent:]
+
+    async def receive_fds(self, msglen: int, maxfds: int) -> Tuple[bytes, List[int]]:
+        if not isinstance(msglen, int) or msglen < 0:
+            raise ValueError('msglen must be a non-negative integer')
+        if not isinstance(maxfds, int) or maxfds < 1:
+            raise ValueError('maxfds must be a positive integer')
+
+        loop = get_running_loop()
+        fds = array.array("i")
+        await checkpoint()
+        with self._receive_guard:
+            while True:
+                try:
+                    message, ancdata, flags, addr = self.__raw_socket.recvmsg(
+                        msglen, socket.CMSG_LEN(maxfds * fds.itemsize))
+                except BlockingIOError:
+                    await self._wait_until_readable(loop)
+                except OSError as exc:
+                    if self._closing:
+                        raise ClosedResourceError from None
+                    else:
+                        raise BrokenResourceError from exc
+                else:
+                    if not message and not ancdata:
+                        raise EndOfStream
+
+                    break
+
+        for cmsg_level, cmsg_type, cmsg_data in ancdata:
+            if cmsg_level != socket.SOL_SOCKET or cmsg_type != socket.SCM_RIGHTS:
+                raise RuntimeError(f'Received unexpected ancillary data; message = {message}, '
+                                   f'cmsg_level = {cmsg_level}, cmsg_type = {cmsg_type}')
+
+            fds.frombytes(cmsg_data[:len(cmsg_data) - (len(cmsg_data) % fds.itemsize)])
+
+        return message, list(fds)
+
+    async def send_fds(self, message: bytes, fds: Collection[Union[int, IOBase]]) -> None:
+        if not message:
+            raise ValueError('message must not be empty')
+        if not fds:
+            raise ValueError('fds must not be empty')
+
+        loop = get_running_loop()
+        filenos: List[int] = []
+        for fd in fds:
+            if isinstance(fd, int):
+                filenos.append(fd)
+            elif isinstance(fd, IOBase):
+                filenos.append(fd.fileno())
+
+        fdarray = array.array("i", filenos)
+        await checkpoint()
+        with self._send_guard:
+            while True:
+                try:
+                    self.__raw_socket.sendmsg([message],
+                                              [(socket.SOL_SOCKET, socket.SCM_RIGHTS, fdarray)])
+                    break
+                except BlockingIOError:
+                    await self._wait_until_writable(loop)
+                except OSError as exc:
+                    if self._closing:
+                        raise ClosedResourceError from None
+                    else:
+                        raise BrokenResourceError from exc
+
+    async def aclose(self) -> None:
+        if not self._closing:
+            self._closing = True
+            if self.__raw_socket.fileno() != -1:
+                self.__raw_socket.close()
+
+            if self._receive_future:
+                self._receive_future.set_result(None)
+            if self._send_future:
+                self._send_future.set_result(None)
+
+
+class TCPSocketListener(abc.SocketListener):
     def __init__(self, raw_socket: socket.SocketType):
         self.__raw_socket = raw_socket
         self._loop = cast(asyncio.BaseEventLoop, get_running_loop())
@@ -888,9 +1045,7 @@ class SocketListener(abc.SocketListener):
 
                 raise
 
-        if client_sock.family in (socket.AF_INET, socket.AF_INET6):
-            client_sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
-
+        client_sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
         transport, protocol = await self._loop.connect_accepted_socket(StreamProtocol, client_sock)
         return SocketStream(cast(asyncio.Transport, transport), cast(StreamProtocol, protocol))
 
@@ -904,6 +1059,40 @@ class SocketListener(abc.SocketListener):
         self._raw_socket.close()
 
 
+class UNIXSocketListener(abc.SocketListener):
+    def __init__(self, raw_socket: socket.SocketType):
+        self.__raw_socket = raw_socket
+        self._loop = get_running_loop()
+        self._accept_guard = ResourceGuard('accepting connections from')
+        self._closed = False
+
+    async def accept(self) -> abc.SocketStream:
+        await checkpoint()
+        with self._accept_guard:
+            while True:
+                try:
+                    client_sock, _ = self.__raw_socket.accept()
+                    return UNIXSocketStream(client_sock)
+                except BlockingIOError:
+                    f: asyncio.Future = asyncio.Future()
+                    self._loop.add_reader(self.__raw_socket, f.set_result, None)
+                    f.add_done_callback(lambda _: self._loop.remove_reader(self.__raw_socket))
+                    await f
+                except OSError as exc:
+                    if self._closed:
+                        raise ClosedResourceError from None
+                    else:
+                        raise BrokenResourceError from exc
+
+    async def aclose(self) -> None:
+        self._closed = True
+        self.__raw_socket.close()
+
+    @property
+    def _raw_socket(self) -> SocketType:
+        return self.__raw_socket
+
+
 class UDPSocket(abc.UDPSocket):
     def __init__(self, transport: asyncio.DatagramTransport, protocol: DatagramProtocol):
         self._transport = transport
@@ -1009,13 +1198,21 @@ async def connect_tcp(host: str, port: int,
     return SocketStream(transport, protocol)
 
 
-async def connect_unix(path: str) -> SocketStream:
-    transport, protocol = cast(
-        Tuple[asyncio.Transport, StreamProtocol],
-        await get_running_loop().create_unix_connection(StreamProtocol, path)
-    )
-    transport.pause_reading()
-    return SocketStream(transport, protocol)
+async def connect_unix(path: str) -> UNIXSocketStream:
+    await checkpoint()
+    loop = get_running_loop()
+    raw_socket = socket.socket(socket.AF_UNIX)
+    raw_socket.setblocking(False)
+    while True:
+        try:
+            raw_socket.connect(path)
+        except BlockingIOError:
+            f: asyncio.Future = asyncio.Future()
+            loop.add_writer(raw_socket, f.set_result, None)
+            f.add_done_callback(lambda _: loop.remove_writer(raw_socket))
+            await f
+        else:
+            return UNIXSocketStream(raw_socket)
 
 
 async def create_udp_socket(
diff --git a/src/anyio/_backends/_trio.py b/src/anyio/_backends/_trio.py
index 9dbe201..0b0a946 100644
--- a/src/anyio/_backends/_trio.py
+++ b/src/anyio/_backends/_trio.py
@@ -1,11 +1,13 @@
+import array
 import socket
 from concurrent.futures import Future
 from dataclasses import dataclass
 from functools import partial
+from io import IOBase
 from types import TracebackType
 from typing import (
-    Any, Awaitable, Callable, Coroutine, Dict, Generic, List, NoReturn, Optional, Tuple, Type,
-    TypeVar, Union)
+    Any, Awaitable, Callable, Collection, Coroutine, Dict, Generic, List, NoReturn, Optional,
+    Tuple, Type, TypeVar, Union)
 
 import trio.from_thread
 from outcome import Error, Value
@@ -288,7 +290,63 @@ class SocketStream(_TrioSocketMixin, abc.SocketStream):
         self._trio_socket.shutdown(socket.SHUT_WR)
 
 
-class SocketListener(_TrioSocketMixin, abc.SocketListener):
+class UNIXSocketStream(SocketStream, abc.UNIXSocketStream):
+    async def receive_fds(self, msglen: int, maxfds: int) -> Tuple[bytes, List[int]]:
+        if not isinstance(msglen, int) or msglen < 0:
+            raise ValueError('msglen must be a non-negative integer')
+        if not isinstance(maxfds, int) or maxfds < 1:
+            raise ValueError('maxfds must be a positive integer')
+
+        fds = array.array("i")
+        await checkpoint()
+        with self._receive_guard:
+            while True:
+                try:
+                    message, ancdata, flags, addr = await self._trio_socket.recvmsg(
+                        msglen, socket.CMSG_LEN(maxfds * fds.itemsize))
+                except BaseException as exc:
+                    self._convert_socket_error(exc)
+                else:
+                    if not message and not ancdata:
+                        raise EndOfStream
+
+                    break
+
+        for cmsg_level, cmsg_type, cmsg_data in ancdata:
+            if cmsg_level != socket.SOL_SOCKET or cmsg_type != socket.SCM_RIGHTS:
+                raise RuntimeError(f'Received unexpected ancillary data; message = {message}, '
+                                   f'cmsg_level = {cmsg_level}, cmsg_type = {cmsg_type}')
+
+            fds.frombytes(cmsg_data[:len(cmsg_data) - (len(cmsg_data) % fds.itemsize)])
+
+        return message, list(fds)
+
+    async def send_fds(self, message: bytes, fds: Collection[Union[int, IOBase]]) -> None:
+        if not message:
+            raise ValueError('message must not be empty')
+        if not fds:
+            raise ValueError('fds must not be empty')
+
+        filenos: List[int] = []
+        for fd in fds:
+            if isinstance(fd, int):
+                filenos.append(fd)
+            elif isinstance(fd, IOBase):
+                filenos.append(fd.fileno())
+
+        fdarray = array.array("i", filenos)
+        await checkpoint()
+        with self._send_guard:
+            while True:
+                try:
+                    await self._trio_socket.sendmsg(
+                        [message], [(socket.SOL_SOCKET, socket.SCM_RIGHTS, fdarray)])
+                    break
+                except BaseException as exc:
+                    self._convert_socket_error(exc)
+
+
+class TCPSocketListener(_TrioSocketMixin, abc.SocketListener):
     def __init__(self, raw_socket: socket.SocketType):
         super().__init__(trio.socket.from_stdlib_socket(raw_socket))
         self._accept_guard = ResourceGuard('accepting connections from')
@@ -300,12 +358,25 @@ class SocketListener(_TrioSocketMixin, abc.SocketListener):
             except BaseException as exc:
                 self._convert_socket_error(exc)
 
-        if trio_socket.family in (socket.AF_INET, socket.AF_INET6):
-            trio_socket.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
-
+        trio_socket.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
         return SocketStream(trio_socket)
 
 
+class UNIXSocketListener(_TrioSocketMixin, abc.SocketListener):
+    def __init__(self, raw_socket: socket.SocketType):
+        super().__init__(trio.socket.from_stdlib_socket(raw_socket))
+        self._accept_guard = ResourceGuard('accepting connections from')
+
+    async def accept(self) -> UNIXSocketStream:
+        with self._accept_guard:
+            try:
+                trio_socket, _addr = await self._trio_socket.accept()
+            except BaseException as exc:
+                self._convert_socket_error(exc)
+
+        return UNIXSocketStream(trio_socket)
+
+
 class UDPSocket(_TrioSocketMixin[IPSockAddrType], abc.UDPSocket):
     def __init__(self, trio_socket):
         super().__init__(trio_socket)
@@ -366,7 +437,7 @@ async def connect_tcp(host: str, port: int,
     return SocketStream(trio_socket)
 
 
-async def connect_unix(path: str) -> SocketStream:
+async def connect_unix(path: str) -> UNIXSocketStream:
     trio_socket = trio.socket.socket(socket.AF_UNIX)
     try:
         await trio_socket.connect(path)
@@ -374,7 +445,7 @@ async def connect_unix(path: str) -> SocketStream:
         trio_socket.close()
         raise
 
-    return SocketStream(trio_socket)
+    return UNIXSocketStream(trio_socket)
 
 
 async def create_udp_socket(
diff --git a/src/anyio/_core/_sockets.py b/src/anyio/_core/_sockets.py
index e50ddb2..814ebe7 100644
--- a/src/anyio/_core/_sockets.py
+++ b/src/anyio/_core/_sockets.py
@@ -9,7 +9,7 @@ from typing import Awaitable, List, Optional, Tuple, Union, cast, overload
 
 from ..abc import (
     ConnectedUDPSocket, Event, IPAddressType, IPSockAddrType, SocketListener, SocketStream,
-    UDPSocket)
+    UDPSocket, UNIXSocketStream)
 from ..streams.stapled import MultiListener
 from ..streams.tls import TLSStream
 from ._eventloop import get_asynclib
@@ -194,7 +194,7 @@ async def connect_tcp(
     return connected_stream
 
 
-async def connect_unix(path: Union[str, PathLike]) -> SocketStream:
+async def connect_unix(path: Union[str, PathLike]) -> UNIXSocketStream:
     """
     Connect to the given UNIX socket.
 
@@ -257,7 +257,7 @@ async def create_tcp_listener(
 
             raw_socket.bind(sockaddr)
             raw_socket.listen(backlog)
-            listener = asynclib.SocketListener(raw_socket)
+            listener = asynclib.TCPSocketListener(raw_socket)
             listeners.append(listener)
     except BaseException:
         for listener in listeners:
@@ -293,7 +293,7 @@ async def create_unix_listener(
             await run_sync_in_worker_thread(chmod, path, mode, cancellable=True)
 
         raw_socket.listen(backlog)
-        return get_asynclib().SocketListener(raw_socket)
+        return get_asynclib().UNIXSocketListener(raw_socket)
     except BaseException:
         raw_socket.close()
         raise
diff --git a/src/anyio/abc/__init__.py b/src/anyio/abc/__init__.py
index d80852a..0834ccf 100644
--- a/src/anyio/abc/__init__.py
+++ b/src/anyio/abc/__init__.py
@@ -1,8 +1,8 @@
 __all__ = ('AsyncResource', 'IPAddressType', 'IPSockAddrType', 'SocketAttribute', 'SocketStream',
-           'SocketListener', 'UDPSocket', 'UDPPacketType', 'ConnectedUDPSocket',
-           'UnreliableObjectReceiveStream', 'UnreliableObjectSendStream', 'UnreliableObjectStream',
-           'ObjectReceiveStream', 'ObjectSendStream', 'ObjectStream', 'ByteReceiveStream',
-           'ByteSendStream', 'ByteStream', 'AnyUnreliableByteReceiveStream',
+           'SocketListener', 'UDPSocket', 'UNIXSocketStream', 'UDPPacketType',
+           'ConnectedUDPSocket', 'UnreliableObjectReceiveStream', 'UnreliableObjectSendStream',
+           'UnreliableObjectStream', 'ObjectReceiveStream', 'ObjectSendStream', 'ObjectStream',
+           'ByteReceiveStream', 'ByteSendStream', 'ByteStream', 'AnyUnreliableByteReceiveStream',
            'AnyUnreliableByteSendStream', 'AnyUnreliableByteStream', 'AnyByteReceiveStream',
            'AnyByteSendStream', 'AnyByteStream', 'Listener', 'Process', 'Event',
            'Condition', 'Lock', 'Semaphore', 'CapacityLimiter', 'CancelScope', 'TaskGroup',
@@ -11,7 +11,7 @@ __all__ = ('AsyncResource', 'IPAddressType', 'IPSockAddrType', 'SocketAttribute'
 from ._resources import AsyncResource
 from ._sockets import (
     ConnectedUDPSocket, IPAddressType, IPSockAddrType, SocketAttribute, SocketListener,
-    SocketStream, UDPPacketType, UDPSocket)
+    SocketStream, UDPPacketType, UDPSocket, UNIXSocketStream)
 from ._streams import (
     AnyByteReceiveStream, AnyByteSendStream, AnyByteStream, AnyUnreliableByteReceiveStream,
     AnyUnreliableByteSendStream, AnyUnreliableByteStream, ByteReceiveStream, ByteSendStream,
diff --git a/src/anyio/abc/_sockets.py b/src/anyio/abc/_sockets.py
index 1562a8d..b1195fa 100644
--- a/src/anyio/abc/_sockets.py
+++ b/src/anyio/abc/_sockets.py
@@ -1,7 +1,9 @@
 from abc import abstractmethod
+from io import IOBase
 from ipaddress import IPv4Address, IPv6Address
 from socket import AddressFamily, SocketType
-from typing import Any, AsyncContextManager, Callable, Optional, Tuple, TypeVar, Union
+from typing import (
+    Any, AsyncContextManager, Callable, Collection, List, Optional, Tuple, TypeVar, Union)
 
 from .._core._typedattr import TypedAttributeProvider, TypedAttributeSet, typed_attribute
 from ._streams import ByteStream, Listener, T_Stream, UnreliableObjectStream
@@ -78,6 +80,28 @@ class SocketStream(ByteStream, _SocketProvider):
     """
 
 
+class UNIXSocketStream(SocketStream):
+    @abstractmethod
+    async def send_fds(self, message: bytes, fds: Collection[Union[int, IOBase]]) -> None:
+        """
+        Send file descriptors along with a message to the peer.
+
+        :param message: a non-empty bytestring
+        :param fds: a collection of files (either numeric file descriptors or open file or socket
+            objects)
+        """
+
+    @abstractmethod
+    async def receive_fds(self, msglen: int, maxfds: int) -> Tuple[bytes, List[int]]:
+        """
+        Receive file descriptors along with a message from the peer.
+
+        :param msglen: length of the message to expect from the peer
+        :param maxfds: maximum number of file descriptors to expect from the peer
+        :return: a tuple of (message, file descriptors)
+        """
+
+
 class SocketListener(Listener[SocketStream], _SocketProvider):
     """
     Listens to incoming socket connections.


Test patch:
diff --git a/tests/test_sockets.py b/tests/test_sockets.py
index 60c4072..746f9c7 100644
--- a/tests/test_sockets.py
+++ b/tests/test_sockets.py
@@ -1,3 +1,5 @@
+import array
+import os
 import platform
 import socket
 import sys
@@ -501,6 +503,72 @@ class TestUNIXStream:
         thread.join()
         assert response == buffer
 
+    async def test_receive_fds(self, server_sock, socket_path, tmp_path):
+        def serve():
+            path1 = tmp_path / 'file1'
+            path2 = tmp_path / 'file2'
+            path1.write_text('Hello, ')
+            path2.write_text('World!')
+            with path1.open() as file1, path2.open() as file2:
+                fdarray = array.array('i', [file1.fileno(), file2.fileno()])
+                client, _ = server_sock.accept()
+                with client:
+                    client.sendmsg([b'test'], [(socket.SOL_SOCKET, socket.SCM_RIGHTS, fdarray)])
+
+        async with await connect_unix(socket_path) as stream:
+            thread = Thread(target=serve, daemon=True)
+            thread.start()
+            message, fds = await stream.receive_fds(10, 2)
+            thread.join()
+
+        text = ''
+        for fd in fds:
+            with os.fdopen(fd) as file:
+                text += file.read()
+
+        assert message == b'test'
+        assert text == 'Hello, World!'
+
+    async def test_receive_fds_bad_args(self, server_sock, socket_path):
+        async with await connect_unix(socket_path) as stream:
+            for msglen in (-1, 'foo'):
+                with pytest.raises(ValueError, match='msglen must be a non-negative integer'):
+                    await stream.receive_fds(msglen, 0)
+
+            for maxfds in (0, 'foo'):
+                with pytest.raises(ValueError, match='maxfds must be a positive integer'):
+                    await stream.receive_fds(0, maxfds)
+
+    async def test_send_fds(self, server_sock, socket_path, tmp_path):
+        def serve():
+            fds = array.array('i')
+            client, _ = server_sock.accept()
+            msg, ancdata, *_ = client.recvmsg(10, socket.CMSG_LEN(2 * fds.itemsize))
+            client.close()
+            assert msg == b'test'
+            for cmsg_level, cmsg_type, cmsg_data in ancdata:
+                assert cmsg_level == socket.SOL_SOCKET
+                assert cmsg_type == socket.SCM_RIGHTS
+                fds.frombytes(cmsg_data[:len(cmsg_data) - (len(cmsg_data) % fds.itemsize)])
+
+            text = ''
+            for fd in fds:
+                with os.fdopen(fd) as file:
+                    text += file.read()
+
+            assert text == 'Hello, World!'
+
+        path1 = tmp_path / 'file1'
+        path2 = tmp_path / 'file2'
+        path1.write_text('Hello, ')
+        path2.write_text('World!')
+        with path1.open() as file1, path2.open() as file2, fail_after(2):
+            async with await connect_unix(socket_path) as stream:
+                thread = Thread(target=serve, daemon=True)
+                thread.start()
+                await stream.send_fds(b'test', [file1, file2])
+                thread.join()
+
     async def test_send_eof(self, server_sock, socket_path):
         def serve():
             client, _ = server_sock.accept()
@@ -543,6 +611,14 @@ class TestUNIXStream:
         thread.join()
         assert chunks == [b'bl', b'ah']
 
+    async def test_send_fds_bad_args(self, server_sock, socket_path):
+        async with await connect_unix(socket_path) as stream:
+            with pytest.raises(ValueError, match='message must not be empty'):
+                await stream.send_fds(b'', [0])
+
+            with pytest.raises(ValueError, match='fds must not be empty'):
+                await stream.send_fds(b'test', [])
+
     async def test_concurrent_send(self, server_sock, socket_path):
         async def send_data():
             while True: