"""Parser is a generator function (NOT coroutine).
Parser receives data with generator's send() method and sends data to
destination DataQueue. Parser receives ParserBuffer and DataQueue objects
as a parameters of the parser call, all subsequent send() calls should
send bytes objects. Parser sends parsed `term` to destination buffer with
DataQueue.feed_data() method. DataQueue object should implement two methods.
feed_data() - parser uses this method to send parsed protocol data.
feed_eof() - parser uses this method for indication of end of parsing stream.
To indicate end of incoming data stream EofStream exception should be sent
into parser. Parser could throw exceptions.
There are three stages:
* Data flow chain:
1. Application creates StreamParser object for storing incoming data.
2. StreamParser creates ParserBuffer as internal data buffer.
3. Application create parser and set it into stream buffer:
parser = HttpRequestParser()
data_queue = stream.set_parser(parser)
3. At this stage StreamParser creates DataQueue object and passes it
and internal buffer into parser as an arguments.
def set_parser(self, parser):
output = DataQueue()
self.p = parser(output, self._input)
return output
4. Application waits data on output.read()
while True:
msg = yield from output.read()
...
* Data flow:
1. asyncio's transport reads data from socket and sends data to protocol
with data_received() call.
2. Protocol sends data to StreamParser with feed_data() call.
3. StreamParser sends data into parser with generator's send() method.
4. Parser processes incoming data and sends parsed data
to DataQueue with feed_data()
5. Application received parsed data from DataQueue.read()
* Eof:
1. StreamParser receives eof with feed_eof() call.
2. StreamParser throws EofStream exception into parser.
3. Then it unsets parser.
_SocketSocketTransport ->
-> "protocol" -> StreamParser -> "parser" -> DataQueue <- "application"
"""
import asyncio
import asyncio.streams
import inspect
import socket
from . import errors
from .streams import FlowControlDataQueue, EofStream
__all__ = ('EofStream', 'StreamParser', 'StreamProtocol',
'ParserBuffer', 'LinesParser', 'ChunksParser')
DEFAULT_LIMIT = 2 ** 16
if hasattr(socket, 'TCP_CORK'): # pragma: no cover
CORK = socket.TCP_CORK
elif hasattr(socket, 'TCP_NOPUSH'): # pragma: no cover
CORK = socket.TCP_NOPUSH
else: # pragma: no cover
CORK = None
[docs]class StreamParser:
"""StreamParser manages incoming bytes stream and protocol parsers.
StreamParser uses ParserBuffer as internal buffer.
set_parser() sets current parser, it creates DataQueue object
and sends ParserBuffer and DataQueue into parser generator.
unset_parser() sends EofStream into parser and then removes it.
"""
def __init__(self, *, loop=None, buf=None,
limit=DEFAULT_LIMIT, eof_exc_class=RuntimeError, **kwargs):
self._loop = loop
self._eof = False
self._exception = None
self._parser = None
self._output = None
self._limit = limit
self._eof_exc_class = eof_exc_class
self._buffer = buf if buf is not None else ParserBuffer()
self.paused = False
self.transport = None
@property
def output(self):
return self._output
[docs] def set_transport(self, transport):
assert transport is None or self.transport is None, \
'Transport already set'
self.transport = transport
[docs] def at_eof(self):
return self._eof
[docs] def exception(self):
return self._exception
[docs] def set_exception(self, exc):
if isinstance(exc, ConnectionError):
exc, old_exc = self._eof_exc_class(), exc
exc.__cause__ = old_exc
exc.__context__ = old_exc
self._exception = exc
if self._output is not None:
self._output.set_exception(exc)
self._output = None
self._parser = None
[docs] def feed_data(self, data):
"""send data to current parser or store in buffer."""
if data is None:
return
if self._parser:
try:
self._parser.send(data)
except StopIteration:
self._output.feed_eof()
self._output = None
self._parser = None
except Exception as exc:
self._output.set_exception(exc)
self._output = None
self._parser = None
else:
self._buffer.feed_data(data)
[docs] def feed_eof(self):
"""send eof to all parsers, recursively."""
if self._parser:
try:
if self._buffer:
self._parser.send(b'')
self._parser.throw(EofStream())
except StopIteration:
self._output.feed_eof()
except EofStream:
self._output.set_exception(self._eof_exc_class())
except Exception as exc:
self._output.set_exception(exc)
self._parser = None
self._output = None
self._eof = True
[docs] def set_parser(self, parser, output=None):
"""set parser to stream. return parser's DataQueue."""
if self._parser:
self.unset_parser()
if output is None:
output = FlowControlDataQueue(
self, limit=self._limit, loop=self._loop)
if self._exception:
output.set_exception(self._exception)
return output
# init parser
p = parser(output, self._buffer)
assert inspect.isgenerator(p), 'Generator is required'
try:
# initialize parser with data and parser buffers
next(p)
except StopIteration:
pass
except Exception as exc:
output.set_exception(exc)
else:
# parser still require more data
self._parser = p
self._output = output
if self._eof:
self.unset_parser()
return output
[docs] def unset_parser(self):
"""unset parser, send eof to the parser and then remove it."""
if self._parser is None:
return
# TODO: write test
if hasattr(self._loop, 'is_closed'):
if self._loop.is_closed():
# TODO: log something
return
try:
self._parser.throw(EofStream())
except StopIteration:
self._output.feed_eof()
except EofStream:
self._output.set_exception(self._eof_exc_class())
except Exception as exc:
self._output.set_exception(exc)
finally:
self._output = None
self._parser = None
class StreamWriter(asyncio.streams.StreamWriter):
def __init__(self, transport, protocol, reader, loop):
self._transport = transport
self._protocol = protocol
self._reader = reader
self._loop = loop
self._tcp_nodelay = False
self._tcp_cork = False
self._socket = transport.get_extra_info('socket')
@property
def tcp_nodelay(self):
return self._tcp_nodelay
def set_tcp_nodelay(self, value):
value = bool(value)
if self._tcp_nodelay == value:
return
self._tcp_nodelay = value
if self._socket is None:
return
if self._socket.family not in (socket.AF_INET, socket.AF_INET6):
return
if self._tcp_cork:
self._tcp_cork = False
if CORK is not None: # pragma: no branch
self._socket.setsockopt(socket.IPPROTO_TCP, CORK, False)
self._socket.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, value)
@property
def tcp_cork(self):
return self._tcp_cork
def set_tcp_cork(self, value):
value = bool(value)
if self._tcp_cork == value:
return
self._tcp_cork = value
if self._socket is None:
return
if self._socket.family not in (socket.AF_INET, socket.AF_INET6):
return
if self._tcp_nodelay:
self._socket.setsockopt(socket.IPPROTO_TCP,
socket.TCP_NODELAY,
False)
self._tcp_nodelay = False
if CORK is not None: # pragma: no branch
self._socket.setsockopt(socket.IPPROTO_TCP, CORK, value)
[docs]class StreamProtocol(asyncio.streams.FlowControlMixin, asyncio.Protocol):
"""Helper class to adapt between Protocol and StreamReader."""
def __init__(self, *, loop=None, disconnect_error=RuntimeError, **kwargs):
super().__init__(loop=loop)
self.transport = None
self.writer = None
self.reader = StreamParser(
loop=loop, eof_exc_class=disconnect_error, **kwargs)
[docs] def is_connected(self):
return self.transport is not None
[docs] def connection_made(self, transport):
self.transport = transport
self.reader.set_transport(transport)
self.writer = StreamWriter(transport, self, self.reader, self._loop)
[docs] def connection_lost(self, exc):
self.transport = self.writer = None
self.reader.set_transport(None)
if exc is None:
self.reader.feed_eof()
else:
self.reader.set_exception(exc)
super().connection_lost(exc)
[docs] def data_received(self, data):
self.reader.feed_data(data)
[docs] def eof_received(self):
self.reader.feed_eof()
class _ParserBufferHelper:
__slots__ = ('exception', 'data')
def __init__(self, exception, data):
self.exception = exception
self.data = data
[docs]class ParserBuffer:
"""ParserBuffer is NOT a bytearray extension anymore.
ParserBuffer provides helper methods for parsers.
"""
__slots__ = ('_helper', '_writer', '_data')
def __init__(self, *args):
self._data = bytearray(*args)
self._helper = _ParserBufferHelper(None, self._data)
self._writer = self._feed_data(self._helper)
next(self._writer)
[docs] def exception(self):
return self._helper.exception
[docs] def set_exception(self, exc):
self._helper.exception = exc
@staticmethod
def _feed_data(helper):
while True:
chunk = yield
if chunk:
helper.data.extend(chunk)
if helper.exception:
raise helper.exception
[docs] def feed_data(self, data):
if not self._helper.exception:
self._writer.send(data)
[docs] def read(self, size):
"""read() reads specified amount of bytes."""
while True:
if self._helper.exception:
raise self._helper.exception
if len(self._data) >= size:
data = self._data[:size]
del self._data[:size]
return data
self._writer.send((yield))
[docs] def readsome(self, size=None):
"""reads size of less amount of bytes."""
while True:
if self._helper.exception:
raise self._helper.exception
length = len(self._data)
if length > 0:
if size is None or length < size:
size = length
data = self._data[:size]
del self._data[:size]
return data
self._writer.send((yield))
[docs] def readuntil(self, stop, limit=None):
assert isinstance(stop, bytes) and stop, \
'bytes is required: {!r}'.format(stop)
stop_len = len(stop)
while True:
if self._helper.exception:
raise self._helper.exception
pos = self._data.find(stop)
if pos >= 0:
end = pos + stop_len
size = end
if limit is not None and size > limit:
raise errors.LineLimitExceededParserError(
'Line is too long.', limit)
data = self._data[:size]
del self._data[:size]
return data
else:
if limit is not None and len(self._data) > limit:
raise errors.LineLimitExceededParserError(
'Line is too long.', limit)
self._writer.send((yield))
[docs] def wait(self, size):
"""wait() waits for specified amount of bytes
then returns data without changing internal buffer."""
while True:
if self._helper.exception:
raise self._helper.exception
if len(self._data) >= size:
return self._data[:size]
self._writer.send((yield))
[docs] def waituntil(self, stop, limit=None):
"""waituntil() reads until `stop` bytes sequence."""
assert isinstance(stop, bytes) and stop, \
'bytes is required: {!r}'.format(stop)
stop_len = len(stop)
while True:
if self._helper.exception:
raise self._helper.exception
pos = self._data.find(stop)
if pos >= 0:
size = pos + stop_len
if limit is not None and size > limit:
raise errors.LineLimitExceededParserError(
'Line is too long. %s' % bytes(self._data), limit)
return self._data[:size]
else:
if limit is not None and len(self._data) > limit:
raise errors.LineLimitExceededParserError(
'Line is too long. %s' % bytes(self._data), limit)
self._writer.send((yield))
[docs] def skip(self, size):
"""skip() skips specified amount of bytes."""
while len(self._data) < size:
if self._helper.exception:
raise self._helper.exception
self._writer.send((yield))
del self._data[:size]
[docs] def skipuntil(self, stop):
"""skipuntil() reads until `stop` bytes sequence."""
assert isinstance(stop, bytes) and stop, \
'bytes is required: {!r}'.format(stop)
stop_len = len(stop)
while True:
if self._helper.exception:
raise self._helper.exception
stop_line = self._data.find(stop)
if stop_line >= 0:
size = stop_line + stop_len
del self._data[:size]
return
self._writer.send((yield))
[docs] def extend(self, data):
self._data.extend(data)
def __len__(self):
return len(self._data)
def __bytes__(self):
return bytes(self._data)
[docs]class LinesParser:
"""Lines parser.
Lines parser splits a bytes stream into a chunks of data, each chunk ends
with \\n symbol."""
def __init__(self, limit=DEFAULT_LIMIT):
self._limit = limit
def __call__(self, out, buf):
try:
while True:
chunk = yield from buf.readuntil(b'\n', self._limit)
out.feed_data(chunk, len(chunk))
except EofStream:
pass
[docs]class ChunksParser:
"""Chunks parser.
Chunks parser splits a bytes stream into a specified
size chunks of data."""
def __init__(self, size=8192):
self._size = size
def __call__(self, out, buf):
try:
while True:
chunk = yield from buf.read(self._size)
out.feed_data(chunk, len(chunk))
except EofStream:
pass
