streams.py 4.5 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167
  1. from __future__ import annotations
  2. from collections.abc import Generator
  3. class StreamReader:
  4. """
  5. Generator-based stream reader.
  6. This class doesn't support concurrent calls to :meth:`read_line`,
  7. :meth:`read_exact`, or :meth:`read_to_eof`. Make sure calls are
  8. serialized.
  9. """
  10. def __init__(self) -> None:
  11. self.buffer = bytearray()
  12. self.eof = False
  13. def read_line(
  14. self,
  15. m: int,
  16. too_long_exc_type: type[Exception] = RuntimeError,
  17. ) -> Generator[None, None, bytearray]:
  18. """
  19. Read a LF-terminated line from the stream.
  20. This is a generator-based coroutine.
  21. The return value includes the LF character.
  22. Args:
  23. m: Maximum number bytes to read; this is a security limit.
  24. too_long_exc_type: exception to raise if the line ends in more
  25. than ``m`` bytes; defaults to :exc:`RuntimeError`.
  26. Raises:
  27. EOFError: If the stream ends without a LF.
  28. RuntimeError: If the line ends in more than ``m`` bytes.
  29. """
  30. n = 0 # number of bytes to read
  31. p = 0 # number of bytes without a newline
  32. while True:
  33. n = self.buffer.find(b"\n", p) + 1
  34. if n > 0:
  35. break
  36. p = len(self.buffer)
  37. if p > m:
  38. raise too_long_exc_type(
  39. f"read {p} bytes, expected no more than {m} bytes"
  40. )
  41. if self.eof:
  42. raise EOFError(f"stream ends after {p} bytes, before end of line")
  43. yield
  44. if n > m:
  45. raise too_long_exc_type(f"read {n} bytes, expected no more than {m} bytes")
  46. r = self.buffer[:n]
  47. del self.buffer[:n]
  48. return r
  49. def read_exact(self, n: int) -> Generator[None, None, bytearray]:
  50. """
  51. Read a given number of bytes from the stream.
  52. This is a generator-based coroutine.
  53. Args:
  54. n: How many bytes to read.
  55. Raises:
  56. EOFError: If the stream ends in less than ``n`` bytes.
  57. """
  58. assert n >= 0
  59. while len(self.buffer) < n:
  60. if self.eof:
  61. p = len(self.buffer)
  62. raise EOFError(f"stream ends after {p} bytes, expected {n} bytes")
  63. yield
  64. r = self.buffer[:n]
  65. del self.buffer[:n]
  66. return r
  67. def read_to_eof(
  68. self,
  69. m: int,
  70. too_long_exc_type: type[Exception] = RuntimeError,
  71. ) -> Generator[None, None, bytearray]:
  72. """
  73. Read all bytes from the stream.
  74. This is a generator-based coroutine.
  75. Args:
  76. m: Maximum number bytes to read; this is a security limit.
  77. too_long_exc_type: exception to raise if the stream ends in more
  78. than ``m`` bytes; defaults to :exc:`RuntimeError`.
  79. Raises:
  80. RuntimeError: If the stream ends in more than ``m`` bytes.
  81. """
  82. while not self.eof:
  83. p = len(self.buffer)
  84. if p > m:
  85. raise too_long_exc_type(
  86. f"read {p} bytes, expected no more than {m} bytes"
  87. )
  88. yield
  89. r = self.buffer[:]
  90. del self.buffer[:]
  91. return r
  92. def at_eof(self) -> Generator[None, None, bool]:
  93. """
  94. Tell whether the stream has ended and all data was read.
  95. This is a generator-based coroutine.
  96. """
  97. while True:
  98. if self.buffer:
  99. return False
  100. if self.eof:
  101. return True
  102. # When all data was read but the stream hasn't ended, we can't
  103. # tell if until either feed_data() or feed_eof() is called.
  104. yield
  105. def feed_data(self, data: bytes | bytearray) -> None:
  106. """
  107. Write data to the stream.
  108. :meth:`feed_data` cannot be called after :meth:`feed_eof`.
  109. Args:
  110. data: Data to write.
  111. Raises:
  112. EOFError: If the stream has ended.
  113. """
  114. if self.eof:
  115. raise EOFError("stream ended")
  116. self.buffer += data
  117. def feed_eof(self) -> None:
  118. """
  119. End the stream.
  120. :meth:`feed_eof` cannot be called more than once.
  121. Raises:
  122. EOFError: If the stream has ended.
  123. """
  124. if self.eof:
  125. raise EOFError("stream ended")
  126. self.eof = True
  127. def discard(self) -> None:
  128. """
  129. Discard all buffered data, but don't end the stream.
  130. """
  131. del self.buffer[:]