_streams.py 7.3 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233
  1. from __future__ import annotations
  2. from abc import ABCMeta, abstractmethod
  3. from collections.abc import Callable
  4. from typing import Any, Generic, TypeAlias, TypeVar
  5. from .._core._exceptions import EndOfStream
  6. from .._core._typedattr import TypedAttributeProvider
  7. from ._resources import AsyncResource
  8. from ._tasks import TaskGroup
  9. T_Item = TypeVar("T_Item")
  10. T_co = TypeVar("T_co", covariant=True)
  11. T_contra = TypeVar("T_contra", contravariant=True)
  12. class UnreliableObjectReceiveStream(
  13. Generic[T_co], AsyncResource, TypedAttributeProvider
  14. ):
  15. """
  16. An interface for receiving objects.
  17. This interface makes no guarantees that the received messages arrive in the order in
  18. which they were sent, or that no messages are missed.
  19. Asynchronously iterating over objects of this type will yield objects matching the
  20. given type parameter.
  21. """
  22. def __aiter__(self) -> UnreliableObjectReceiveStream[T_co]:
  23. return self
  24. async def __anext__(self) -> T_co:
  25. try:
  26. return await self.receive()
  27. except EndOfStream:
  28. raise StopAsyncIteration from None
  29. @abstractmethod
  30. async def receive(self) -> T_co:
  31. """
  32. Receive the next item.
  33. :raises ~anyio.ClosedResourceError: if the receive stream has been explicitly
  34. closed
  35. :raises ~anyio.EndOfStream: if this stream has been closed from the other end
  36. :raises ~anyio.BrokenResourceError: if this stream has been rendered unusable
  37. due to external causes
  38. """
  39. class UnreliableObjectSendStream(
  40. Generic[T_contra], AsyncResource, TypedAttributeProvider
  41. ):
  42. """
  43. An interface for sending objects.
  44. This interface makes no guarantees that the messages sent will reach the
  45. recipient(s) in the same order in which they were sent, or at all.
  46. """
  47. @abstractmethod
  48. async def send(self, item: T_contra) -> None:
  49. """
  50. Send an item to the peer(s).
  51. :param item: the item to send
  52. :raises ~anyio.ClosedResourceError: if the send stream has been explicitly
  53. closed
  54. :raises ~anyio.BrokenResourceError: if this stream has been rendered unusable
  55. due to external causes
  56. """
  57. class UnreliableObjectStream(
  58. UnreliableObjectReceiveStream[T_Item], UnreliableObjectSendStream[T_Item]
  59. ):
  60. """
  61. A bidirectional message stream which does not guarantee the order or reliability of
  62. message delivery.
  63. """
  64. class ObjectReceiveStream(UnreliableObjectReceiveStream[T_co]):
  65. """
  66. A receive message stream which guarantees that messages are received in the same
  67. order in which they were sent, and that no messages are missed.
  68. """
  69. class ObjectSendStream(UnreliableObjectSendStream[T_contra]):
  70. """
  71. A send message stream which guarantees that messages are delivered in the same order
  72. in which they were sent, without missing any messages in the middle.
  73. """
  74. class ObjectStream(
  75. ObjectReceiveStream[T_Item],
  76. ObjectSendStream[T_Item],
  77. UnreliableObjectStream[T_Item],
  78. ):
  79. """
  80. A bidirectional message stream which guarantees the order and reliability of message
  81. delivery.
  82. """
  83. @abstractmethod
  84. async def send_eof(self) -> None:
  85. """
  86. Send an end-of-file indication to the peer.
  87. You should not try to send any further data to this stream after calling this
  88. method. This method is idempotent (does nothing on successive calls).
  89. """
  90. class ByteReceiveStream(AsyncResource, TypedAttributeProvider):
  91. """
  92. An interface for receiving bytes from a single peer.
  93. Iterating this byte stream will yield a byte string of arbitrary length, but no more
  94. than 65536 bytes.
  95. """
  96. def __aiter__(self) -> ByteReceiveStream:
  97. return self
  98. async def __anext__(self) -> bytes:
  99. try:
  100. return await self.receive()
  101. except EndOfStream:
  102. raise StopAsyncIteration from None
  103. @abstractmethod
  104. async def receive(self, max_bytes: int = 65536) -> bytes:
  105. """
  106. Receive at most ``max_bytes`` bytes from the peer.
  107. .. note:: Implementers of this interface should not return an empty
  108. :class:`bytes` object, and users should ignore them.
  109. :param max_bytes: maximum number of bytes to receive
  110. :return: the received bytes
  111. :raises ~anyio.EndOfStream: if this stream has been closed from the other end
  112. """
  113. class ByteSendStream(AsyncResource, TypedAttributeProvider):
  114. """An interface for sending bytes to a single peer."""
  115. @abstractmethod
  116. async def send(self, item: bytes) -> None:
  117. """
  118. Send the given bytes to the peer.
  119. :param item: the bytes to send
  120. """
  121. class ByteStream(ByteReceiveStream, ByteSendStream):
  122. """A bidirectional byte stream."""
  123. @abstractmethod
  124. async def send_eof(self) -> None:
  125. """
  126. Send an end-of-file indication to the peer.
  127. You should not try to send any further data to this stream after calling this
  128. method. This method is idempotent (does nothing on successive calls).
  129. """
  130. #: Type alias for all unreliable bytes-oriented receive streams.
  131. AnyUnreliableByteReceiveStream: TypeAlias = (
  132. UnreliableObjectReceiveStream[bytes] | ByteReceiveStream
  133. )
  134. #: Type alias for all unreliable bytes-oriented send streams.
  135. AnyUnreliableByteSendStream: TypeAlias = (
  136. UnreliableObjectSendStream[bytes] | ByteSendStream
  137. )
  138. #: Type alias for all unreliable bytes-oriented streams.
  139. AnyUnreliableByteStream: TypeAlias = UnreliableObjectStream[bytes] | ByteStream
  140. #: Type alias for all bytes-oriented receive streams.
  141. AnyByteReceiveStream: TypeAlias = ObjectReceiveStream[bytes] | ByteReceiveStream
  142. #: Type alias for all bytes-oriented send streams.
  143. AnyByteSendStream: TypeAlias = ObjectSendStream[bytes] | ByteSendStream
  144. #: Type alias for all bytes-oriented streams.
  145. AnyByteStream: TypeAlias = ObjectStream[bytes] | ByteStream
  146. class Listener(Generic[T_co], AsyncResource, TypedAttributeProvider):
  147. """An interface for objects that let you accept incoming connections."""
  148. @abstractmethod
  149. async def serve(
  150. self, handler: Callable[[T_co], Any], task_group: TaskGroup | None = None
  151. ) -> None:
  152. """
  153. Accept incoming connections as they come in and start tasks to handle them.
  154. :param handler: a callable that will be used to handle each accepted connection
  155. :param task_group: the task group that will be used to start tasks for handling
  156. each accepted connection (if omitted, an ad-hoc task group will be created)
  157. """
  158. class ObjectStreamConnectable(Generic[T_co], metaclass=ABCMeta):
  159. @abstractmethod
  160. async def connect(self) -> ObjectStream[T_co]:
  161. """
  162. Connect to the remote endpoint.
  163. :return: an object stream connected to the remote end
  164. :raises ConnectionFailed: if the connection fails
  165. """
  166. class ByteStreamConnectable(metaclass=ABCMeta):
  167. @abstractmethod
  168. async def connect(self) -> ByteStream:
  169. """
  170. Connect to the remote endpoint.
  171. :return: a bytestream connected to the remote end
  172. :raises ConnectionFailed: if the connection fails
  173. """
  174. #: Type alias for all connectables returning bytestreams or bytes-oriented object streams
  175. AnyByteStreamConnectable: TypeAlias = (
  176. ObjectStreamConnectable[bytes] | ByteStreamConnectable
  177. )