Загрузить файлы в «venv/Lib/site-packages/anyio/streams»

venv/Lib/site-packages/anyio/streams/buffered.py

@@ -0,0 +1,198 @@
+from __future__ import annotations
+
+__all__ = (
+    "BufferedByteReceiveStream",
+    "BufferedByteStream",
+    "BufferedConnectable",
+)
+
+import sys
+from collections.abc import Callable, Iterable, Mapping
+from dataclasses import dataclass, field
+from typing import Any, SupportsIndex
+
+from .. import ClosedResourceError, DelimiterNotFound, EndOfStream, IncompleteRead
+from ..abc import (
+    AnyByteReceiveStream,
+    AnyByteStream,
+    AnyByteStreamConnectable,
+    ByteReceiveStream,
+    ByteStream,
+    ByteStreamConnectable,
+)
+
+if sys.version_info >= (3, 12):
+    from typing import override
+else:
+    from typing_extensions import override
+
+
+@dataclass(eq=False)
+class BufferedByteReceiveStream(ByteReceiveStream):
+    """
+    Wraps any bytes-based receive stream and uses a buffer to provide sophisticated
+    receiving capabilities in the form of a byte stream.
+    """
+
+    receive_stream: AnyByteReceiveStream
+    _buffer: bytearray = field(init=False, default_factory=bytearray)
+    _closed: bool = field(init=False, default=False)
+
+    async def aclose(self) -> None:
+        await self.receive_stream.aclose()
+        self._closed = True
+
+    @property
+    def buffer(self) -> bytes:
+        """The bytes currently in the buffer."""
+        return bytes(self._buffer)
+
+    @property
+    def extra_attributes(self) -> Mapping[Any, Callable[[], Any]]:
+        return self.receive_stream.extra_attributes
+
+    def feed_data(self, data: Iterable[SupportsIndex], /) -> None:
+        """
+        Append data directly into the buffer.
+
+        Any data in the buffer will be consumed by receive operations before receiving
+        anything from the wrapped stream.
+
+        :param data: the data to append to the buffer (can be bytes or anything else
+            that supports ``__index__()``)
+
+        """
+        self._buffer.extend(data)
+
+    async def receive(self, max_bytes: int = 65536) -> bytes:
+        if self._closed:
+            raise ClosedResourceError
+
+        if self._buffer:
+            chunk = bytes(self._buffer[:max_bytes])
+            del self._buffer[:max_bytes]
+            return chunk
+        elif isinstance(self.receive_stream, ByteReceiveStream):
+            return await self.receive_stream.receive(max_bytes)
+        else:
+            # With a bytes-oriented object stream, we need to handle any surplus bytes
+            # we get from the receive() call
+            chunk = await self.receive_stream.receive()
+            if len(chunk) > max_bytes:
+                # Save the surplus bytes in the buffer
+                self._buffer.extend(chunk[max_bytes:])
+                return chunk[:max_bytes]
+            else:
+                return chunk
+
+    async def receive_exactly(self, nbytes: int) -> bytes:
+        """
+        Read exactly the given amount of bytes from the stream.
+
+        :param nbytes: the number of bytes to read
+        :return: the bytes read
+        :raises ~anyio.IncompleteRead: if the stream was closed before the requested
+            amount of bytes could be read from the stream
+
+        """
+        while True:
+            remaining = nbytes - len(self._buffer)
+            if remaining <= 0:
+                retval = self._buffer[:nbytes]
+                del self._buffer[:nbytes]
+                return bytes(retval)
+
+            try:
+                if isinstance(self.receive_stream, ByteReceiveStream):
+                    chunk = await self.receive_stream.receive(remaining)
+                else:
+                    chunk = await self.receive_stream.receive()
+            except EndOfStream as exc:
+                raise IncompleteRead from exc
+
+            self._buffer.extend(chunk)
+
+    async def receive_until(self, delimiter: bytes, max_bytes: int) -> bytes:
+        """
+        Read from the stream until the delimiter is found or max_bytes have been read.
+
+        :param delimiter: the marker to look for in the stream
+        :param max_bytes: maximum number of bytes that will be read before raising
+            :exc:`~anyio.DelimiterNotFound`
+        :return: the bytes read (not including the delimiter)
+        :raises ~anyio.IncompleteRead: if the stream was closed before the delimiter
+            was found
+        :raises ~anyio.DelimiterNotFound: if the delimiter is not found within the
+            bytes read up to the maximum allowed
+
+        """
+        delimiter_size = len(delimiter)
+        offset = 0
+        while True:
+            # Check if the delimiter can be found in the current buffer
+            index = self._buffer.find(delimiter, offset)
+            if index >= 0:
+                found = self._buffer[:index]
+                del self._buffer[: index + len(delimiter) :]
+                return bytes(found)
+
+            # Check if the buffer is already at or over the limit
+            if len(self._buffer) >= max_bytes:
+                raise DelimiterNotFound(max_bytes)
+
+            # Read more data into the buffer from the socket
+            try:
+                data = await self.receive_stream.receive()
+            except EndOfStream as exc:
+                raise IncompleteRead from exc
+
+            # Move the offset forward and add the new data to the buffer
+            offset = max(len(self._buffer) - delimiter_size + 1, 0)
+            self._buffer.extend(data)
+
+
+class BufferedByteStream(BufferedByteReceiveStream, ByteStream):
+    """
+    A full-duplex variant of :class:`BufferedByteReceiveStream`. All writes are passed
+    through to the wrapped stream as-is.
+    """
+
+    def __init__(self, stream: AnyByteStream):
+        """
+        :param stream: the stream to be wrapped
+
+        """
+        super().__init__(stream)
+        self._stream = stream
+
+    @override
+    async def send_eof(self) -> None:
+        await self._stream.send_eof()
+
+    @override
+    async def send(self, item: bytes) -> None:
+        await self._stream.send(item)
+
+
+class BufferedConnectable(ByteStreamConnectable):
+    """
+    Wraps a byte stream connectable to produce :class:`BufferedByteStream` connections.
+
+    Use this when you want the streams returned by :meth:`connect` to have the buffered
+    receive API (e.g. :meth:`~BufferedByteReceiveStream.receive_exactly` and
+    :meth:`~BufferedByteReceiveStream.receive_until`).
+
+    :param connectable: the byte stream connectable to wrap
+    """
+
+    def __init__(self, connectable: AnyByteStreamConnectable):
+        """
+        :param connectable: the connectable to wrap
+
+        """
+        self.connectable = connectable
+
+    @override
+    async def connect(self) -> BufferedByteStream:
+        stream = await self.connectable.connect()
+        return BufferedByteStream(stream)

venv/Lib/site-packages/anyio/streams/file.py

@@ -0,0 +1,154 @@
+from __future__ import annotations
+
+__all__ = (
+    "FileReadStream",
+    "FileStreamAttribute",
+    "FileWriteStream",
+)
+
+from collections.abc import Callable, Mapping
+from io import SEEK_SET, UnsupportedOperation
+from os import PathLike
+from pathlib import Path
+from typing import IO, Any
+
+from .. import (
+    BrokenResourceError,
+    ClosedResourceError,
+    EndOfStream,
+    TypedAttributeSet,
+    to_thread,
+    typed_attribute,
+)
+from ..abc import ByteReceiveStream, ByteSendStream
+
+
+class FileStreamAttribute(TypedAttributeSet):
+    #: the open file descriptor
+    file: IO[bytes] = typed_attribute()
+    #: the path of the file on the file system, if available (file must be a real file)
+    path: Path = typed_attribute()
+    #: the file number, if available (file must be a real file or a TTY)
+    fileno: int = typed_attribute()
+
+
+class _BaseFileStream:
+    def __init__(self, file: IO[bytes]):
+        self._file = file
+
+    async def aclose(self) -> None:
+        await to_thread.run_sync(self._file.close)
+
+    @property
+    def extra_attributes(self) -> Mapping[Any, Callable[[], Any]]:
+        attributes: dict[Any, Callable[[], Any]] = {
+            FileStreamAttribute.file: lambda: self._file,
+        }
+
+        if hasattr(self._file, "name"):
+            attributes[FileStreamAttribute.path] = lambda: Path(self._file.name)
+
+        try:
+            self._file.fileno()
+        except UnsupportedOperation:
+            pass
+        else:
+            attributes[FileStreamAttribute.fileno] = lambda: self._file.fileno()
+
+        return attributes
+
+
+class FileReadStream(_BaseFileStream, ByteReceiveStream):
+    """
+    A byte stream that reads from a file in the file system.
+
+    :param file: a file that has been opened for reading in binary mode
+
+    .. versionadded:: 3.0
+    """
+
+    @classmethod
+    async def from_path(cls, path: str | PathLike[str]) -> FileReadStream:
+        """
+        Create a file read stream by opening the given file.
+
+        :param path: path of the file to read from
+
+        """
+        file = await to_thread.run_sync(Path(path).open, "rb")
+        return cls(file)
+
+    async def receive(self, max_bytes: int = 65536) -> bytes:
+        try:
+            data = await to_thread.run_sync(self._file.read, max_bytes)
+        except ValueError:
+            raise ClosedResourceError from None
+        except OSError as exc:
+            raise BrokenResourceError from exc
+
+        if data:
+            return data
+        else:
+            raise EndOfStream
+
+    async def seek(self, position: int, whence: int = SEEK_SET) -> int:
+        """
+        Seek the file to the given position.
+
+        .. seealso:: :meth:`io.IOBase.seek`
+
+        .. note:: Not all file descriptors are seekable.
+
+        :param position: position to seek the file to
+        :param whence: controls how ``position`` is interpreted
+        :return: the new absolute position
+        :raises OSError: if the file is not seekable
+
+        """
+        return await to_thread.run_sync(self._file.seek, position, whence)
+
+    async def tell(self) -> int:
+        """
+        Return the current stream position.
+
+        .. note:: Not all file descriptors are seekable.
+
+        :return: the current absolute position
+        :raises OSError: if the file is not seekable
+
+        """
+        return await to_thread.run_sync(self._file.tell)
+
+
+class FileWriteStream(_BaseFileStream, ByteSendStream):
+    """
+    A byte stream that writes to a file in the file system.
+
+    :param file: a file that has been opened for writing in binary mode
+
+    .. versionadded:: 3.0
+    """
+
+    @classmethod
+    async def from_path(
+        cls, path: str | PathLike[str], append: bool = False
+    ) -> FileWriteStream:
+        """
+        Create a file write stream by opening the given file for writing.
+
+        :param path: path of the file to write to
+        :param append: if ``True``, open the file for appending; if ``False``, any
+            existing file at the given path will be truncated
+
+        """
+        mode = "ab" if append else "wb"
+        file = await to_thread.run_sync(Path(path).open, mode)
+        return cls(file)
+
+    async def send(self, item: bytes) -> None:
+        try:
+            await to_thread.run_sync(self._file.write, item)
+        except ValueError:
+            raise ClosedResourceError from None
+        except OSError as exc:
+            raise BrokenResourceError from exc

venv/Lib/site-packages/anyio/streams/memory.py

@@ -0,0 +1,325 @@
+from __future__ import annotations
+
+__all__ = (
+    "MemoryObjectReceiveStream",
+    "MemoryObjectSendStream",
+    "MemoryObjectStreamStatistics",
+)
+
+import warnings
+from collections import OrderedDict, deque
+from dataclasses import dataclass, field
+from types import TracebackType
+from typing import Generic, NamedTuple, TypeVar
+
+from .. import (
+    BrokenResourceError,
+    ClosedResourceError,
+    EndOfStream,
+    WouldBlock,
+)
+from .._core._testing import TaskInfo, get_current_task
+from ..abc import Event, ObjectReceiveStream, ObjectSendStream
+from ..lowlevel import checkpoint
+
+T_Item = TypeVar("T_Item")
+T_co = TypeVar("T_co", covariant=True)
+T_contra = TypeVar("T_contra", contravariant=True)
+
+
+class MemoryObjectStreamStatistics(NamedTuple):
+    current_buffer_used: int  #: number of items stored in the buffer
+    #: maximum number of items that can be stored on this stream (or :data:`math.inf`)
+    max_buffer_size: float
+    open_send_streams: int  #: number of unclosed clones of the send stream
+    open_receive_streams: int  #: number of unclosed clones of the receive stream
+    #: number of tasks blocked on :meth:`MemoryObjectSendStream.send`
+    tasks_waiting_send: int
+    #: number of tasks blocked on :meth:`MemoryObjectReceiveStream.receive`
+    tasks_waiting_receive: int
+
+
+@dataclass(eq=False)
+class _MemoryObjectItemReceiver(Generic[T_Item]):
+    task_info: TaskInfo = field(init=False, default_factory=get_current_task)
+    item: T_Item = field(init=False)
+
+    def __repr__(self) -> str:
+        # When item is not defined, we get following error with default __repr__:
+        # AttributeError: 'MemoryObjectItemReceiver' object has no attribute 'item'
+        item = getattr(self, "item", None)
+        return f"{self.__class__.__name__}(task_info={self.task_info}, item={item!r})"
+
+
+@dataclass(eq=False)
+class _MemoryObjectStreamState(Generic[T_Item]):
+    max_buffer_size: float = field()
+    buffer: deque[T_Item] = field(init=False, default_factory=deque)
+    open_send_channels: int = field(init=False, default=0)
+    open_receive_channels: int = field(init=False, default=0)
+    waiting_receivers: OrderedDict[Event, _MemoryObjectItemReceiver[T_Item]] = field(
+        init=False, default_factory=OrderedDict
+    )
+    waiting_senders: OrderedDict[Event, T_Item] = field(
+        init=False, default_factory=OrderedDict
+    )
+
+    def statistics(self) -> MemoryObjectStreamStatistics:
+        return MemoryObjectStreamStatistics(
+            len(self.buffer),
+            self.max_buffer_size,
+            self.open_send_channels,
+            self.open_receive_channels,
+            len(self.waiting_senders),
+            len(self.waiting_receivers),
+        )
+
+
+@dataclass(eq=False)
+class MemoryObjectReceiveStream(Generic[T_co], ObjectReceiveStream[T_co]):
+    _state: _MemoryObjectStreamState[T_co]
+    _closed: bool = field(init=False, default=False)
+
+    def __post_init__(self) -> None:
+        self._state.open_receive_channels += 1
+
+    def receive_nowait(self) -> T_co:
+        """
+        Receive the next item if it can be done without waiting.
+
+        :return: the received item
+        :raises ~anyio.ClosedResourceError: if this send stream has been closed
+        :raises ~anyio.EndOfStream: if the buffer is empty and this stream has been
+            closed from the sending end
+        :raises ~anyio.WouldBlock: if there are no items in the buffer and no tasks
+            waiting to send
+
+        """
+        if self._closed:
+            raise ClosedResourceError
+
+        if self._state.waiting_senders:
+            # Get the item from the next sender
+            send_event, item = self._state.waiting_senders.popitem(last=False)
+            self._state.buffer.append(item)
+            send_event.set()
+
+        if self._state.buffer:
+            return self._state.buffer.popleft()
+        elif not self._state.open_send_channels:
+            raise EndOfStream
+
+        raise WouldBlock
+
+    async def receive(self) -> T_co:
+        await checkpoint()
+        try:
+            return self.receive_nowait()
+        except WouldBlock:
+            # Add ourselves in the queue
+            receive_event = Event()
+            receiver = _MemoryObjectItemReceiver[T_co]()
+            self._state.waiting_receivers[receive_event] = receiver
+
+            try:
+                await receive_event.wait()
+            finally:
+                self._state.waiting_receivers.pop(receive_event, None)
+
+            try:
+                return receiver.item
+            except AttributeError:
+                raise EndOfStream from None
+
+    def clone(self) -> MemoryObjectReceiveStream[T_co]:
+        """
+        Create a clone of this receive stream.
+
+        Each clone can be closed separately. Only when all clones have been closed will
+        the receiving end of the memory stream be considered closed by the sending ends.
+
+        :return: the cloned stream
+
+        """
+        if self._closed:
+            raise ClosedResourceError
+
+        return MemoryObjectReceiveStream(_state=self._state)
+
+    def close(self) -> None:
+        """
+        Close the stream.
+
+        This works the exact same way as :meth:`aclose`, but is provided as a special
+        case for the benefit of synchronous callbacks.
+
+        """
+        if not self._closed:
+            self._closed = True
+            self._state.open_receive_channels -= 1
+            if self._state.open_receive_channels == 0:
+                send_events = list(self._state.waiting_senders.keys())
+                for event in send_events:
+                    event.set()
+
+    async def aclose(self) -> None:
+        self.close()
+
+    def statistics(self) -> MemoryObjectStreamStatistics:
+        """
+        Return statistics about the current state of this stream.
+
+        .. versionadded:: 3.0
+        """
+        return self._state.statistics()
+
+    def __enter__(self) -> MemoryObjectReceiveStream[T_co]:
+        return self
+
+    def __exit__(
+        self,
+        exc_type: type[BaseException] | None,
+        exc_val: BaseException | None,
+        exc_tb: TracebackType | None,
+    ) -> None:
+        self.close()
+
+    def __del__(self) -> None:
+        if not self._closed:
+            warnings.warn(
+                f"Unclosed <{self.__class__.__name__} at {id(self):x}>",
+                ResourceWarning,
+                stacklevel=1,
+                source=self,
+            )
+
+
+@dataclass(eq=False)
+class MemoryObjectSendStream(Generic[T_contra], ObjectSendStream[T_contra]):
+    _state: _MemoryObjectStreamState[T_contra]
+    _closed: bool = field(init=False, default=False)
+
+    def __post_init__(self) -> None:
+        self._state.open_send_channels += 1
+
+    def send_nowait(self, item: T_contra) -> None:
+        """
+        Send an item immediately if it can be done without waiting.
+
+        :param item: the item to send
+        :raises ~anyio.ClosedResourceError: if this send stream has been closed
+        :raises ~anyio.BrokenResourceError: if the stream has been closed from the
+            receiving end
+        :raises ~anyio.WouldBlock: if the buffer is full and there are no tasks waiting
+            to receive
+
+        """
+        if self._closed:
+            raise ClosedResourceError
+        if not self._state.open_receive_channels:
+            raise BrokenResourceError
+
+        while self._state.waiting_receivers:
+            receive_event, receiver = self._state.waiting_receivers.popitem(last=False)
+            if not receiver.task_info.has_pending_cancellation():
+                receiver.item = item
+                receive_event.set()
+                return
+
+        if len(self._state.buffer) < self._state.max_buffer_size:
+            self._state.buffer.append(item)
+        else:
+            raise WouldBlock
+
+    async def send(self, item: T_contra) -> None:
+        """
+        Send an item to the stream.
+
+        If the buffer is full, this method blocks until there is again room in the
+        buffer or the item can be sent directly to a receiver.
+
+        :param item: the item to send
+        :raises ~anyio.ClosedResourceError: if this send stream has been closed
+        :raises ~anyio.BrokenResourceError: if the stream has been closed from the
+            receiving end
+
+        """
+        await checkpoint()
+        try:
+            self.send_nowait(item)
+        except WouldBlock:
+            # Wait until there's someone on the receiving end
+            send_event = Event()
+            self._state.waiting_senders[send_event] = item
+            try:
+                await send_event.wait()
+            except BaseException:
+                self._state.waiting_senders.pop(send_event, None)
+                raise
+
+            if send_event in self._state.waiting_senders:
+                del self._state.waiting_senders[send_event]
+                raise BrokenResourceError from None
+
+    def clone(self) -> MemoryObjectSendStream[T_contra]:
+        """
+        Create a clone of this send stream.
+
+        Each clone can be closed separately. Only when all clones have been closed will
+        the sending end of the memory stream be considered closed by the receiving ends.
+
+        :return: the cloned stream
+
+        """
+        if self._closed:
+            raise ClosedResourceError
+
+        return MemoryObjectSendStream(_state=self._state)
+
+    def close(self) -> None:
+        """
+        Close the stream.
+
+        This works the exact same way as :meth:`aclose`, but is provided as a special
+        case for the benefit of synchronous callbacks.
+
+        """
+        if not self._closed:
+            self._closed = True
+            self._state.open_send_channels -= 1
+            if self._state.open_send_channels == 0:
+                receive_events = list(self._state.waiting_receivers.keys())
+                self._state.waiting_receivers.clear()
+                for event in receive_events:
+                    event.set()
+
+    async def aclose(self) -> None:
+        self.close()
+
+    def statistics(self) -> MemoryObjectStreamStatistics:
+        """
+        Return statistics about the current state of this stream.
+
+        .. versionadded:: 3.0
+        """
+        return self._state.statistics()
+
+    def __enter__(self) -> MemoryObjectSendStream[T_contra]:
+        return self
+
+    def __exit__(
+        self,
+        exc_type: type[BaseException] | None,
+        exc_val: BaseException | None,
+        exc_tb: TracebackType | None,
+    ) -> None:
+        self.close()
+
+    def __del__(self) -> None:
+        if not self._closed:
+            warnings.warn(
+                f"Unclosed <{self.__class__.__name__} at {id(self):x}>",
+                ResourceWarning,
+                stacklevel=1,
+                source=self,
+            )

venv/Lib/site-packages/anyio/streams/stapled.py

@@ -0,0 +1,147 @@
+from __future__ import annotations
+
+__all__ = (
+    "MultiListener",
+    "StapledByteStream",
+    "StapledObjectStream",
+)
+
+from collections.abc import Callable, Mapping, Sequence
+from dataclasses import dataclass
+from typing import Any, Generic, TypeVar
+
+from ..abc import (
+    ByteReceiveStream,
+    ByteSendStream,
+    ByteStream,
+    Listener,
+    ObjectReceiveStream,
+    ObjectSendStream,
+    ObjectStream,
+    TaskGroup,
+)
+
+T_Item = TypeVar("T_Item")
+T_Stream = TypeVar("T_Stream")
+
+
+@dataclass(eq=False)
+class StapledByteStream(ByteStream):
+    """
+    Combines two byte streams into a single, bidirectional byte stream.
+
+    Extra attributes will be provided from both streams, with the receive stream
+    providing the values in case of a conflict.
+
+    :param ByteSendStream send_stream: the sending byte stream
+    :param ByteReceiveStream receive_stream: the receiving byte stream
+    """
+
+    send_stream: ByteSendStream
+    receive_stream: ByteReceiveStream
+
+    async def receive(self, max_bytes: int = 65536) -> bytes:
+        return await self.receive_stream.receive(max_bytes)
+
+    async def send(self, item: bytes) -> None:
+        await self.send_stream.send(item)
+
+    async def send_eof(self) -> None:
+        await self.send_stream.aclose()
+
+    async def aclose(self) -> None:
+        await self.send_stream.aclose()
+        await self.receive_stream.aclose()
+
+    @property
+    def extra_attributes(self) -> Mapping[Any, Callable[[], Any]]:
+        return {
+            **self.send_stream.extra_attributes,
+            **self.receive_stream.extra_attributes,
+        }
+
+
+@dataclass(eq=False)
+class StapledObjectStream(Generic[T_Item], ObjectStream[T_Item]):
+    """
+    Combines two object streams into a single, bidirectional object stream.
+
+    Extra attributes will be provided from both streams, with the receive stream
+    providing the values in case of a conflict.
+
+    :param ObjectSendStream send_stream: the sending object stream
+    :param ObjectReceiveStream receive_stream: the receiving object stream
+    """
+
+    send_stream: ObjectSendStream[T_Item]
+    receive_stream: ObjectReceiveStream[T_Item]
+
+    async def receive(self) -> T_Item:
+        return await self.receive_stream.receive()
+
+    async def send(self, item: T_Item) -> None:
+        await self.send_stream.send(item)
+
+    async def send_eof(self) -> None:
+        await self.send_stream.aclose()
+
+    async def aclose(self) -> None:
+        await self.send_stream.aclose()
+        await self.receive_stream.aclose()
+
+    @property
+    def extra_attributes(self) -> Mapping[Any, Callable[[], Any]]:
+        return {
+            **self.send_stream.extra_attributes,
+            **self.receive_stream.extra_attributes,
+        }
+
+
+@dataclass(eq=False)
+class MultiListener(Generic[T_Stream], Listener[T_Stream]):
+    """
+    Combines multiple listeners into one, serving connections from all of them at once.
+
+    Any MultiListeners in the given collection of listeners will have their listeners
+    moved into this one.
+
+    Extra attributes are provided from each listener, with each successive listener
+    overriding any conflicting attributes from the previous one.
+
+    :param listeners: listeners to serve
+    :type listeners: Sequence[Listener[T_Stream]]
+    """
+
+    listeners: Sequence[Listener[T_Stream]]
+
+    def __post_init__(self) -> None:
+        listeners: list[Listener[T_Stream]] = []
+        for listener in self.listeners:
+            if isinstance(listener, MultiListener):
+                listeners.extend(listener.listeners)
+                del listener.listeners[:]  # type: ignore[attr-defined]
+            else:
+                listeners.append(listener)
+
+        self.listeners = listeners
+
+    async def serve(
+        self, handler: Callable[[T_Stream], Any], task_group: TaskGroup | None = None
+    ) -> None:
+        from .. import create_task_group
+
+        async with create_task_group() as tg:
+            for listener in self.listeners:
+                tg.start_soon(listener.serve, handler, task_group)
+
+    async def aclose(self) -> None:
+        for listener in self.listeners:
+            await listener.aclose()
+
+    @property
+    def extra_attributes(self) -> Mapping[Any, Callable[[], Any]]:
+        attributes: dict = {}
+        for listener in self.listeners:
+            attributes.update(listener.extra_attributes)
+
+        return attributes