Skip to content

Instantly share code, notes, and snippets.

@Bogdanp
Created October 17, 2018 05:39
Show Gist options
  • Save Bogdanp/077f14e0a16b7e1710d7721fc6eea1d4 to your computer and use it in GitHub Desktop.
Save Bogdanp/077f14e0a16b7e1710d7721fc6eea1d4 to your computer and use it in GitHub Desktop.
import typing
def iter_lines(sock: socket.socket, bufsize: int = 16_384) -> typing.Generator[bytes, None, bytes]:
"""Given a socket, read all the individual CRLF-separated lines
and yield each one until an empty one is found. Returns the
remainder after the empty line.
"""
buff = b""
while True:
data = sock.recv(bufsize)
if not data:
return b""
buff += data
while True:
try:
i = buff.index(b"\r\n")
line, buff = buff[:i], buff[i + 2:]
if not line:
return buff
yield line
except IndexError:
break
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment