wireprotoframing.py
1388 lines
| 47.8 KiB
| text/x-python
|
PythonLexer
/ mercurial / wireprotoframing.py
|
r37069 | # wireprotoframing.py - unified framing protocol for wire protocol | ||
# | ||||
# Copyright 2018 Gregory Szorc <gregory.szorc@gmail.com> | ||||
# | ||||
# This software may be used and distributed according to the terms of the | ||||
# GNU General Public License version 2 or any later version. | ||||
# This file contains functionality to support the unified frame-based wire | ||||
# protocol. For details about the protocol, see | ||||
# `hg help internals.wireprotocol`. | ||||
from __future__ import absolute_import | ||||
|
r37561 | import collections | ||
|
r37069 | import struct | ||
|
r37070 | from .i18n import _ | ||
|
r37079 | from .thirdparty import ( | ||
attr, | ||||
) | ||||
|
r37069 | from . import ( | ||
|
r37492 | encoding, | ||
|
r37070 | error, | ||
|
r40061 | pycompat, | ||
|
r37069 | util, | ||
|
r40056 | wireprototypes, | ||
|
r37069 | ) | ||
|
r37102 | from .utils import ( | ||
|
r39477 | cborutil, | ||
|
r37102 | stringutil, | ||
) | ||||
|
r37069 | |||
|
r37304 | FRAME_HEADER_SIZE = 8 | ||
|
r37069 | DEFAULT_MAX_FRAME_SIZE = 32768 | ||
|
r37304 | STREAM_FLAG_BEGIN_STREAM = 0x01 | ||
STREAM_FLAG_END_STREAM = 0x02 | ||||
STREAM_FLAG_ENCODING_APPLIED = 0x04 | ||||
STREAM_FLAGS = { | ||||
b'stream-begin': STREAM_FLAG_BEGIN_STREAM, | ||||
b'stream-end': STREAM_FLAG_END_STREAM, | ||||
b'encoded': STREAM_FLAG_ENCODING_APPLIED, | ||||
} | ||||
|
r37308 | FRAME_TYPE_COMMAND_REQUEST = 0x01 | ||
|
r37741 | FRAME_TYPE_COMMAND_DATA = 0x02 | ||
|
r37742 | FRAME_TYPE_COMMAND_RESPONSE = 0x03 | ||
|
r37073 | FRAME_TYPE_ERROR_RESPONSE = 0x05 | ||
|
r37078 | FRAME_TYPE_TEXT_OUTPUT = 0x06 | ||
|
r37307 | FRAME_TYPE_PROGRESS = 0x07 | ||
|
r37304 | FRAME_TYPE_STREAM_SETTINGS = 0x08 | ||
|
r37069 | |||
FRAME_TYPES = { | ||||
|
r37308 | b'command-request': FRAME_TYPE_COMMAND_REQUEST, | ||
|
r37069 | b'command-data': FRAME_TYPE_COMMAND_DATA, | ||
|
r37742 | b'command-response': FRAME_TYPE_COMMAND_RESPONSE, | ||
|
r37073 | b'error-response': FRAME_TYPE_ERROR_RESPONSE, | ||
|
r37078 | b'text-output': FRAME_TYPE_TEXT_OUTPUT, | ||
|
r37307 | b'progress': FRAME_TYPE_PROGRESS, | ||
|
r37304 | b'stream-settings': FRAME_TYPE_STREAM_SETTINGS, | ||
|
r37069 | } | ||
|
r37308 | FLAG_COMMAND_REQUEST_NEW = 0x01 | ||
FLAG_COMMAND_REQUEST_CONTINUATION = 0x02 | ||||
FLAG_COMMAND_REQUEST_MORE_FRAMES = 0x04 | ||||
FLAG_COMMAND_REQUEST_EXPECT_DATA = 0x08 | ||||
|
r37069 | |||
|
r37308 | FLAGS_COMMAND_REQUEST = { | ||
b'new': FLAG_COMMAND_REQUEST_NEW, | ||||
b'continuation': FLAG_COMMAND_REQUEST_CONTINUATION, | ||||
b'more': FLAG_COMMAND_REQUEST_MORE_FRAMES, | ||||
b'have-data': FLAG_COMMAND_REQUEST_EXPECT_DATA, | ||||
|
r37069 | } | ||
FLAG_COMMAND_DATA_CONTINUATION = 0x01 | ||||
FLAG_COMMAND_DATA_EOS = 0x02 | ||||
FLAGS_COMMAND_DATA = { | ||||
b'continuation': FLAG_COMMAND_DATA_CONTINUATION, | ||||
b'eos': FLAG_COMMAND_DATA_EOS, | ||||
} | ||||
|
r37742 | FLAG_COMMAND_RESPONSE_CONTINUATION = 0x01 | ||
FLAG_COMMAND_RESPONSE_EOS = 0x02 | ||||
|
r37073 | |||
|
r37742 | FLAGS_COMMAND_RESPONSE = { | ||
b'continuation': FLAG_COMMAND_RESPONSE_CONTINUATION, | ||||
b'eos': FLAG_COMMAND_RESPONSE_EOS, | ||||
|
r37073 | } | ||
|
r37069 | # Maps frame types to their available flags. | ||
FRAME_TYPE_FLAGS = { | ||||
|
r37308 | FRAME_TYPE_COMMAND_REQUEST: FLAGS_COMMAND_REQUEST, | ||
|
r37069 | FRAME_TYPE_COMMAND_DATA: FLAGS_COMMAND_DATA, | ||
|
r37742 | FRAME_TYPE_COMMAND_RESPONSE: FLAGS_COMMAND_RESPONSE, | ||
|
r37744 | FRAME_TYPE_ERROR_RESPONSE: {}, | ||
|
r37078 | FRAME_TYPE_TEXT_OUTPUT: {}, | ||
|
r37307 | FRAME_TYPE_PROGRESS: {}, | ||
|
r37304 | FRAME_TYPE_STREAM_SETTINGS: {}, | ||
|
r37069 | } | ||
|
r37308 | ARGUMENT_RECORD_HEADER = struct.Struct(r'<HH') | ||
|
r37069 | |||
|
r37314 | def humanflags(mapping, value): | ||
"""Convert a numeric flags value to a human value, using a mapping table.""" | ||||
|
r37491 | namemap = {v: k for k, v in mapping.iteritems()} | ||
|
r37314 | flags = [] | ||
|
r37491 | val = 1 | ||
while value >= val: | ||||
|
r37314 | if value & val: | ||
|
r37491 | flags.append(namemap.get(val, '<unknown 0x%02x>' % val)) | ||
val <<= 1 | ||||
|
r37314 | |||
return b'|'.join(flags) | ||||
|
r37079 | @attr.s(slots=True) | ||
class frameheader(object): | ||||
"""Represents the data in a frame header.""" | ||||
length = attr.ib() | ||||
requestid = attr.ib() | ||||
|
r37304 | streamid = attr.ib() | ||
streamflags = attr.ib() | ||||
|
r37079 | typeid = attr.ib() | ||
flags = attr.ib() | ||||
|
r37314 | @attr.s(slots=True, repr=False) | ||
|
r37079 | class frame(object): | ||
"""Represents a parsed frame.""" | ||||
requestid = attr.ib() | ||||
|
r37304 | streamid = attr.ib() | ||
streamflags = attr.ib() | ||||
|
r37079 | typeid = attr.ib() | ||
flags = attr.ib() | ||||
payload = attr.ib() | ||||
|
r37492 | @encoding.strmethod | ||
|
r37314 | def __repr__(self): | ||
|
r37491 | typename = '<unknown 0x%02x>' % self.typeid | ||
|
r37314 | for name, value in FRAME_TYPES.iteritems(): | ||
if value == self.typeid: | ||||
typename = name | ||||
break | ||||
return ('frame(size=%d; request=%d; stream=%d; streamflags=%s; ' | ||||
'type=%s; flags=%s)' % ( | ||||
len(self.payload), self.requestid, self.streamid, | ||||
humanflags(STREAM_FLAGS, self.streamflags), typename, | ||||
|
r37490 | humanflags(FRAME_TYPE_FLAGS.get(self.typeid, {}), self.flags))) | ||
|
r37314 | |||
|
r37304 | def makeframe(requestid, streamid, streamflags, typeid, flags, payload): | ||
|
r37069 | """Assemble a frame into a byte array.""" | ||
# TODO assert size of payload. | ||||
frame = bytearray(FRAME_HEADER_SIZE + len(payload)) | ||||
|
r37075 | # 24 bits length | ||
# 16 bits request id | ||||
|
r37304 | # 8 bits stream id | ||
# 8 bits stream flags | ||||
|
r37075 | # 4 bits type | ||
# 4 bits flags | ||||
|
r37069 | l = struct.pack(r'<I', len(payload)) | ||
frame[0:3] = l[0:3] | ||||
|
r37304 | struct.pack_into(r'<HBB', frame, 3, requestid, streamid, streamflags) | ||
frame[7] = (typeid << 4) | flags | ||||
frame[8:] = payload | ||||
|
r37069 | |||
return frame | ||||
def makeframefromhumanstring(s): | ||||
|
r37075 | """Create a frame from a human readable string | ||
Strings have the form: | ||||
|
r37304 | <request-id> <stream-id> <stream-flags> <type> <flags> <payload> | ||
|
r37069 | |||
This can be used by user-facing applications and tests for creating | ||||
frames easily without having to type out a bunch of constants. | ||||
|
r37304 | Request ID and stream IDs are integers. | ||
|
r37075 | |||
|
r37304 | Stream flags, frame type, and flags can be specified by integer or | ||
named constant. | ||||
|
r37075 | |||
|
r37069 | Flags can be delimited by `|` to bitwise OR them together. | ||
|
r37306 | |||
If the payload begins with ``cbor:``, the following string will be | ||||
|
r37494 | evaluated as Python literal and the resulting object will be fed into | ||
|
r37306 | a CBOR encoder. Otherwise, the payload is interpreted as a Python | ||
byte string literal. | ||||
|
r37069 | """ | ||
|
r37304 | fields = s.split(b' ', 5) | ||
requestid, streamid, streamflags, frametype, frameflags, payload = fields | ||||
|
r37075 | |||
requestid = int(requestid) | ||||
|
r37304 | streamid = int(streamid) | ||
finalstreamflags = 0 | ||||
for flag in streamflags.split(b'|'): | ||||
if flag in STREAM_FLAGS: | ||||
finalstreamflags |= STREAM_FLAGS[flag] | ||||
else: | ||||
finalstreamflags |= int(flag) | ||||
|
r37069 | |||
if frametype in FRAME_TYPES: | ||||
frametype = FRAME_TYPES[frametype] | ||||
else: | ||||
frametype = int(frametype) | ||||
finalflags = 0 | ||||
validflags = FRAME_TYPE_FLAGS[frametype] | ||||
for flag in frameflags.split(b'|'): | ||||
if flag in validflags: | ||||
finalflags |= validflags[flag] | ||||
else: | ||||
finalflags |= int(flag) | ||||
|
r37306 | if payload.startswith(b'cbor:'): | ||
|
r39477 | payload = b''.join(cborutil.streamencode( | ||
stringutil.evalpythonliteral(payload[5:]))) | ||||
|
r37306 | |||
else: | ||||
payload = stringutil.unescapestr(payload) | ||||
|
r37069 | |||
|
r37304 | return makeframe(requestid=requestid, streamid=streamid, | ||
streamflags=finalstreamflags, typeid=frametype, | ||||
|
r37080 | flags=finalflags, payload=payload) | ||
|
r37069 | |||
|
r37070 | def parseheader(data): | ||
"""Parse a unified framing protocol frame header from a buffer. | ||||
The header is expected to be in the buffer at offset 0 and the | ||||
buffer is expected to be large enough to hold a full header. | ||||
""" | ||||
# 24 bits payload length (little endian) | ||||
|
r37304 | # 16 bits request ID | ||
# 8 bits stream ID | ||||
# 8 bits stream flags | ||||
|
r37070 | # 4 bits frame type | ||
# 4 bits frame flags | ||||
# ... payload | ||||
framelength = data[0] + 256 * data[1] + 16384 * data[2] | ||||
|
r37304 | requestid, streamid, streamflags = struct.unpack_from(r'<HBB', data, 3) | ||
typeflags = data[7] | ||||
|
r37070 | |||
frametype = (typeflags & 0xf0) >> 4 | ||||
frameflags = typeflags & 0x0f | ||||
|
r37304 | return frameheader(framelength, requestid, streamid, streamflags, | ||
frametype, frameflags) | ||||
|
r37070 | |||
def readframe(fh): | ||||
"""Read a unified framing protocol frame from a file object. | ||||
Returns a 3-tuple of (type, flags, payload) for the decoded frame or | ||||
None if no frame is available. May raise if a malformed frame is | ||||
seen. | ||||
""" | ||||
header = bytearray(FRAME_HEADER_SIZE) | ||||
readcount = fh.readinto(header) | ||||
if readcount == 0: | ||||
return None | ||||
if readcount != FRAME_HEADER_SIZE: | ||||
raise error.Abort(_('received incomplete frame: got %d bytes: %s') % | ||||
(readcount, header)) | ||||
|
r37079 | h = parseheader(header) | ||
|
r37070 | |||
|
r37079 | payload = fh.read(h.length) | ||
if len(payload) != h.length: | ||||
|
r37070 | raise error.Abort(_('frame length error: expected %d; got %d') % | ||
|
r37079 | (h.length, len(payload))) | ||
|
r37070 | |||
|
r37304 | return frame(h.requestid, h.streamid, h.streamflags, h.typeid, h.flags, | ||
payload) | ||||
|
r37070 | |||
|
r37308 | def createcommandframes(stream, requestid, cmd, args, datafh=None, | ||
|
r40060 | maxframesize=DEFAULT_MAX_FRAME_SIZE, | ||
redirect=None): | ||||
|
r37069 | """Create frames necessary to transmit a request to run a command. | ||
This is a generator of bytearrays. Each item represents a frame | ||||
ready to be sent over the wire to a peer. | ||||
""" | ||||
|
r37308 | data = {b'name': cmd} | ||
|
r37069 | if args: | ||
|
r37308 | data[b'args'] = args | ||
|
r37069 | |||
|
r40060 | if redirect: | ||
data[b'redirect'] = redirect | ||||
|
r39477 | data = b''.join(cborutil.streamencode(data)) | ||
|
r37069 | |||
|
r37308 | offset = 0 | ||
while True: | ||||
flags = 0 | ||||
|
r37069 | |||
|
r37308 | # Must set new or continuation flag. | ||
if not offset: | ||||
flags |= FLAG_COMMAND_REQUEST_NEW | ||||
else: | ||||
flags |= FLAG_COMMAND_REQUEST_CONTINUATION | ||||
|
r37069 | |||
|
r37308 | # Data frames is set on all frames. | ||
if datafh: | ||||
flags |= FLAG_COMMAND_REQUEST_EXPECT_DATA | ||||
|
r37069 | |||
|
r37308 | payload = data[offset:offset + maxframesize] | ||
offset += len(payload) | ||||
if len(payload) == maxframesize and offset < len(data): | ||||
flags |= FLAG_COMMAND_REQUEST_MORE_FRAMES | ||||
|
r37303 | yield stream.makeframe(requestid=requestid, | ||
|
r37308 | typeid=FRAME_TYPE_COMMAND_REQUEST, | ||
|
r37303 | flags=flags, | ||
payload=payload) | ||||
|
r37069 | |||
|
r37308 | if not (flags & FLAG_COMMAND_REQUEST_MORE_FRAMES): | ||
break | ||||
|
r37069 | if datafh: | ||
while True: | ||||
data = datafh.read(DEFAULT_MAX_FRAME_SIZE) | ||||
done = False | ||||
if len(data) == DEFAULT_MAX_FRAME_SIZE: | ||||
flags = FLAG_COMMAND_DATA_CONTINUATION | ||||
else: | ||||
flags = FLAG_COMMAND_DATA_EOS | ||||
assert datafh.read(1) == b'' | ||||
done = True | ||||
|
r37303 | yield stream.makeframe(requestid=requestid, | ||
typeid=FRAME_TYPE_COMMAND_DATA, | ||||
flags=flags, | ||||
payload=data) | ||||
|
r37069 | |||
if done: | ||||
break | ||||
|
r37070 | |||
|
r37742 | def createcommandresponseframesfrombytes(stream, requestid, data, | ||
maxframesize=DEFAULT_MAX_FRAME_SIZE): | ||||
|
r37073 | """Create a raw frame to send a bytes response from static bytes input. | ||
Returns a generator of bytearrays. | ||||
""" | ||||
|
r37743 | # Automatically send the overall CBOR response map. | ||
|
r39477 | overall = b''.join(cborutil.streamencode({b'status': b'ok'})) | ||
|
r37743 | if len(overall) > maxframesize: | ||
raise error.ProgrammingError('not yet implemented') | ||||
|
r37073 | |||
|
r37743 | # Simple case where we can fit the full response in a single frame. | ||
if len(overall) + len(data) <= maxframesize: | ||||
|
r37742 | flags = FLAG_COMMAND_RESPONSE_EOS | ||
|
r37303 | yield stream.makeframe(requestid=requestid, | ||
|
r37742 | typeid=FRAME_TYPE_COMMAND_RESPONSE, | ||
|
r37503 | flags=flags, | ||
|
r37743 | payload=overall + data) | ||
|
r37073 | return | ||
|
r37743 | # It's easier to send the overall CBOR map in its own frame than to track | ||
# offsets. | ||||
yield stream.makeframe(requestid=requestid, | ||||
typeid=FRAME_TYPE_COMMAND_RESPONSE, | ||||
flags=FLAG_COMMAND_RESPONSE_CONTINUATION, | ||||
payload=overall) | ||||
|
r37073 | offset = 0 | ||
while True: | ||||
chunk = data[offset:offset + maxframesize] | ||||
offset += len(chunk) | ||||
done = offset == len(data) | ||||
if done: | ||||
|
r37742 | flags = FLAG_COMMAND_RESPONSE_EOS | ||
|
r37073 | else: | ||
|
r37742 | flags = FLAG_COMMAND_RESPONSE_CONTINUATION | ||
|
r37073 | |||
|
r37303 | yield stream.makeframe(requestid=requestid, | ||
|
r37742 | typeid=FRAME_TYPE_COMMAND_RESPONSE, | ||
|
r37303 | flags=flags, | ||
payload=chunk) | ||||
|
r37073 | |||
if done: | ||||
break | ||||
|
r37746 | def createbytesresponseframesfromgen(stream, requestid, gen, | ||
maxframesize=DEFAULT_MAX_FRAME_SIZE): | ||||
|
r39595 | """Generator of frames from a generator of byte chunks. | ||
|
r37746 | |||
|
r39595 | This assumes that another frame will follow whatever this emits. i.e. | ||
this always emits the continuation flag and never emits the end-of-stream | ||||
flag. | ||||
""" | ||||
|
r37746 | cb = util.chunkbuffer(gen) | ||
|
r39595 | flags = FLAG_COMMAND_RESPONSE_CONTINUATION | ||
|
r37746 | |||
while True: | ||||
chunk = cb.read(maxframesize) | ||||
if not chunk: | ||||
break | ||||
yield stream.makeframe(requestid=requestid, | ||||
typeid=FRAME_TYPE_COMMAND_RESPONSE, | ||||
flags=flags, | ||||
payload=chunk) | ||||
flags |= FLAG_COMMAND_RESPONSE_CONTINUATION | ||||
|
r39595 | def createcommandresponseokframe(stream, requestid): | ||
overall = b''.join(cborutil.streamencode({b'status': b'ok'})) | ||||
return stream.makeframe(requestid=requestid, | ||||
typeid=FRAME_TYPE_COMMAND_RESPONSE, | ||||
flags=FLAG_COMMAND_RESPONSE_CONTINUATION, | ||||
payload=overall) | ||||
def createcommandresponseeosframe(stream, requestid): | ||||
"""Create an empty payload frame representing command end-of-stream.""" | ||||
return stream.makeframe(requestid=requestid, | ||||
typeid=FRAME_TYPE_COMMAND_RESPONSE, | ||||
flags=FLAG_COMMAND_RESPONSE_EOS, | ||||
payload=b'') | ||||
|
r37746 | |||
|
r40061 | def createalternatelocationresponseframe(stream, requestid, location): | ||
data = { | ||||
b'status': b'redirect', | ||||
b'location': { | ||||
b'url': location.url, | ||||
b'mediatype': location.mediatype, | ||||
} | ||||
} | ||||
for a in (r'size', r'fullhashes', r'fullhashseed', r'serverdercerts', | ||||
r'servercadercerts'): | ||||
value = getattr(location, a) | ||||
if value is not None: | ||||
data[b'location'][pycompat.bytestr(a)] = value | ||||
return stream.makeframe(requestid=requestid, | ||||
typeid=FRAME_TYPE_COMMAND_RESPONSE, | ||||
flags=FLAG_COMMAND_RESPONSE_CONTINUATION, | ||||
payload=b''.join(cborutil.streamencode(data))) | ||||
|
r37746 | def createcommanderrorresponse(stream, requestid, message, args=None): | ||
|
r39522 | # TODO should this be using a list of {'msg': ..., 'args': {}} so atom | ||
# formatting works consistently? | ||||
|
r37746 | m = { | ||
b'status': b'error', | ||||
b'error': { | ||||
b'message': message, | ||||
} | ||||
} | ||||
if args: | ||||
m[b'error'][b'args'] = args | ||||
|
r39477 | overall = b''.join(cborutil.streamencode(m)) | ||
|
r37746 | |||
yield stream.makeframe(requestid=requestid, | ||||
typeid=FRAME_TYPE_COMMAND_RESPONSE, | ||||
flags=FLAG_COMMAND_RESPONSE_EOS, | ||||
payload=overall) | ||||
|
r37744 | def createerrorframe(stream, requestid, msg, errtype): | ||
|
r37073 | # TODO properly handle frame size limits. | ||
assert len(msg) <= DEFAULT_MAX_FRAME_SIZE | ||||
|
r39477 | payload = b''.join(cborutil.streamencode({ | ||
|
r37744 | b'type': errtype, | ||
b'message': [{b'msg': msg}], | ||||
|
r39477 | })) | ||
|
r37073 | |||
|
r37303 | yield stream.makeframe(requestid=requestid, | ||
typeid=FRAME_TYPE_ERROR_RESPONSE, | ||||
|
r37744 | flags=0, | ||
payload=payload) | ||||
|
r37073 | |||
|
r37335 | def createtextoutputframe(stream, requestid, atoms, | ||
maxframesize=DEFAULT_MAX_FRAME_SIZE): | ||||
|
r37078 | """Create a text output frame to render text to people. | ||
``atoms`` is a 3-tuple of (formatting string, args, labels). | ||||
The formatting string contains ``%s`` tokens to be replaced by the | ||||
corresponding indexed entry in ``args``. ``labels`` is an iterable of | ||||
formatters to be applied at rendering time. In terms of the ``ui`` | ||||
class, each atom corresponds to a ``ui.write()``. | ||||
""" | ||||
|
r37335 | atomdicts = [] | ||
|
r37078 | |||
for (formatting, args, labels) in atoms: | ||||
# TODO look for localstr, other types here? | ||||
if not isinstance(formatting, bytes): | ||||
raise ValueError('must use bytes formatting strings') | ||||
for arg in args: | ||||
if not isinstance(arg, bytes): | ||||
raise ValueError('must use bytes for arguments') | ||||
for label in labels: | ||||
if not isinstance(label, bytes): | ||||
raise ValueError('must use bytes for labels') | ||||
|
r37335 | # Formatting string must be ASCII. | ||
formatting = formatting.decode(r'ascii', r'replace').encode(r'ascii') | ||||
|
r37078 | |||
# Arguments must be UTF-8. | ||||
args = [a.decode(r'utf-8', r'replace').encode(r'utf-8') for a in args] | ||||
# Labels must be ASCII. | ||||
labels = [l.decode(r'ascii', r'strict').encode(r'ascii') | ||||
for l in labels] | ||||
|
r37335 | atom = {b'msg': formatting} | ||
if args: | ||||
atom[b'args'] = args | ||||
if labels: | ||||
atom[b'labels'] = labels | ||||
|
r37078 | |||
|
r37335 | atomdicts.append(atom) | ||
|
r37078 | |||
|
r39477 | payload = b''.join(cborutil.streamencode(atomdicts)) | ||
|
r37078 | |||
|
r37335 | if len(payload) > maxframesize: | ||
|
r37078 | raise ValueError('cannot encode data in a single frame') | ||
|
r37303 | yield stream.makeframe(requestid=requestid, | ||
typeid=FRAME_TYPE_TEXT_OUTPUT, | ||||
flags=0, | ||||
|
r37335 | payload=payload) | ||
|
r37303 | |||
|
r39596 | class bufferingcommandresponseemitter(object): | ||
"""Helper object to emit command response frames intelligently. | ||||
Raw command response data is likely emitted in chunks much smaller | ||||
than what can fit in a single frame. This class exists to buffer | ||||
chunks until enough data is available to fit in a single frame. | ||||
TODO we'll need something like this when compression is supported. | ||||
So it might make sense to implement this functionality at the stream | ||||
level. | ||||
""" | ||||
def __init__(self, stream, requestid, maxframesize=DEFAULT_MAX_FRAME_SIZE): | ||||
self._stream = stream | ||||
self._requestid = requestid | ||||
self._maxsize = maxframesize | ||||
self._chunks = [] | ||||
self._chunkssize = 0 | ||||
def send(self, data): | ||||
"""Send new data for emission. | ||||
Is a generator of new frames that were derived from the new input. | ||||
If the special input ``None`` is received, flushes all buffered | ||||
data to frames. | ||||
""" | ||||
if data is None: | ||||
for frame in self._flush(): | ||||
yield frame | ||||
return | ||||
# There is a ton of potential to do more complicated things here. | ||||
# Our immediate goal is to coalesce small chunks into big frames, | ||||
# not achieve the fewest number of frames possible. So we go with | ||||
# a simple implementation: | ||||
# | ||||
# * If a chunk is too large for a frame, we flush and emit frames | ||||
# for the new chunk. | ||||
# * If a chunk can be buffered without total buffered size limits | ||||
# being exceeded, we do that. | ||||
# * If a chunk causes us to go over our buffering limit, we flush | ||||
# and then buffer the new chunk. | ||||
if len(data) > self._maxsize: | ||||
for frame in self._flush(): | ||||
yield frame | ||||
# Now emit frames for the big chunk. | ||||
offset = 0 | ||||
while True: | ||||
chunk = data[offset:offset + self._maxsize] | ||||
offset += len(chunk) | ||||
yield self._stream.makeframe( | ||||
self._requestid, | ||||
typeid=FRAME_TYPE_COMMAND_RESPONSE, | ||||
flags=FLAG_COMMAND_RESPONSE_CONTINUATION, | ||||
payload=chunk) | ||||
if offset == len(data): | ||||
return | ||||
# If we don't have enough to constitute a full frame, buffer and | ||||
# return. | ||||
if len(data) + self._chunkssize < self._maxsize: | ||||
self._chunks.append(data) | ||||
self._chunkssize += len(data) | ||||
return | ||||
# Else flush what we have and buffer the new chunk. We could do | ||||
# something more intelligent here, like break the chunk. Let's | ||||
# keep things simple for now. | ||||
for frame in self._flush(): | ||||
yield frame | ||||
self._chunks.append(data) | ||||
self._chunkssize = len(data) | ||||
def _flush(self): | ||||
payload = b''.join(self._chunks) | ||||
assert len(payload) <= self._maxsize | ||||
self._chunks[:] = [] | ||||
self._chunkssize = 0 | ||||
yield self._stream.makeframe( | ||||
self._requestid, | ||||
typeid=FRAME_TYPE_COMMAND_RESPONSE, | ||||
flags=FLAG_COMMAND_RESPONSE_CONTINUATION, | ||||
payload=payload) | ||||
|
r37303 | class stream(object): | ||
"""Represents a logical unidirectional series of frames.""" | ||||
|
r37304 | def __init__(self, streamid, active=False): | ||
self.streamid = streamid | ||||
|
r37673 | self._active = active | ||
|
r37304 | |||
|
r37303 | def makeframe(self, requestid, typeid, flags, payload): | ||
"""Create a frame to be sent out over this stream. | ||||
Only returns the frame instance. Does not actually send it. | ||||
""" | ||||
|
r37304 | streamflags = 0 | ||
if not self._active: | ||||
streamflags |= STREAM_FLAG_BEGIN_STREAM | ||||
self._active = True | ||||
return makeframe(requestid, self.streamid, streamflags, typeid, flags, | ||||
payload) | ||||
def ensureserverstream(stream): | ||||
if stream.streamid % 2: | ||||
raise error.ProgrammingError('server should only write to even ' | ||||
'numbered streams; %d is not even' % | ||||
stream.streamid) | ||||
|
r37078 | |||
|
r37070 | class serverreactor(object): | ||
"""Holds state of a server handling frame-based protocol requests. | ||||
This class is the "brain" of the unified frame-based protocol server | ||||
component. While the protocol is stateless from the perspective of | ||||
requests/commands, something needs to track which frames have been | ||||
received, what frames to expect, etc. This class is that thing. | ||||
Instances are modeled as a state machine of sorts. Instances are also | ||||
reactionary to external events. The point of this class is to encapsulate | ||||
the state of the connection and the exchange of frames, not to perform | ||||
work. Instead, callers tell this class when something occurs, like a | ||||
frame arriving. If that activity is worthy of a follow-up action (say | ||||
*run a command*), the return value of that handler will say so. | ||||
I/O and CPU intensive operations are purposefully delegated outside of | ||||
this class. | ||||
Consumers are expected to tell instances when events occur. They do so by | ||||
calling the various ``on*`` methods. These methods return a 2-tuple | ||||
describing any follow-up action(s) to take. The first element is the | ||||
name of an action to perform. The second is a data structure (usually | ||||
a dict) specific to that action that contains more information. e.g. | ||||
if the server wants to send frames back to the client, the data structure | ||||
will contain a reference to those frames. | ||||
Valid actions that consumers can be instructed to take are: | ||||
|
r37073 | sendframes | ||
Indicates that frames should be sent to the client. The ``framegen`` | ||||
key contains a generator of frames that should be sent. The server | ||||
assumes that all frames are sent to the client. | ||||
|
r37070 | error | ||
Indicates that an error occurred. Consumer should probably abort. | ||||
runcommand | ||||
Indicates that the consumer should run a wire protocol command. Details | ||||
of the command to run are given in the data structure. | ||||
wantframe | ||||
Indicates that nothing of interest happened and the server is waiting on | ||||
more frames from the client before anything interesting can be done. | ||||
|
r37074 | |||
noop | ||||
Indicates no additional action is required. | ||||
|
r37076 | |||
Known Issues | ||||
------------ | ||||
There are no limits to the number of partially received commands or their | ||||
size. A malicious client could stream command request data and exhaust the | ||||
server's memory. | ||||
Partially received commands are not acted upon when end of input is | ||||
reached. Should the server error if it receives a partial request? | ||||
Should the client send a message to abort a partially transmitted request | ||||
to facilitate graceful shutdown? | ||||
Active requests that haven't been responded to aren't tracked. This means | ||||
that if we receive a command and instruct its dispatch, another command | ||||
with its request ID can come in over the wire and there will be a race | ||||
between who responds to what. | ||||
|
r37070 | """ | ||
|
r37074 | def __init__(self, deferoutput=False): | ||
"""Construct a new server reactor. | ||||
``deferoutput`` can be used to indicate that no output frames should be | ||||
instructed to be sent until input has been exhausted. In this mode, | ||||
events that would normally generate output frames (such as a command | ||||
response being ready) will instead defer instructing the consumer to | ||||
send those frames. This is useful for half-duplex transports where the | ||||
sender cannot receive until all data has been transmitted. | ||||
""" | ||||
self._deferoutput = deferoutput | ||||
|
r37070 | self._state = 'idle' | ||
|
r37305 | self._nextoutgoingstreamid = 2 | ||
|
r37074 | self._bufferedframegens = [] | ||
|
r37304 | # stream id -> stream instance for all active streams from the client. | ||
self._incomingstreams = {} | ||||
|
r37305 | self._outgoingstreams = {} | ||
|
r37076 | # request id -> dict of commands that are actively being received. | ||
self._receivingcommands = {} | ||||
|
r37081 | # Request IDs that have been received and are actively being processed. | ||
# Once all output for a request has been sent, it is removed from this | ||||
# set. | ||||
self._activecommands = set() | ||||
|
r37070 | |||
|
r37079 | def onframerecv(self, frame): | ||
|
r37070 | """Process a frame that has been received off the wire. | ||
Returns a dict with an ``action`` key that details what action, | ||||
if any, the consumer should take next. | ||||
""" | ||||
|
r37304 | if not frame.streamid % 2: | ||
self._state = 'errored' | ||||
return self._makeerrorresult( | ||||
_('received frame with even numbered stream ID: %d') % | ||||
frame.streamid) | ||||
if frame.streamid not in self._incomingstreams: | ||||
if not frame.streamflags & STREAM_FLAG_BEGIN_STREAM: | ||||
self._state = 'errored' | ||||
return self._makeerrorresult( | ||||
_('received frame on unknown inactive stream without ' | ||||
'beginning of stream flag set')) | ||||
self._incomingstreams[frame.streamid] = stream(frame.streamid) | ||||
if frame.streamflags & STREAM_FLAG_ENCODING_APPLIED: | ||||
# TODO handle decoding frames | ||||
self._state = 'errored' | ||||
raise error.ProgrammingError('support for decoding stream payloads ' | ||||
'not yet implemented') | ||||
if frame.streamflags & STREAM_FLAG_END_STREAM: | ||||
del self._incomingstreams[frame.streamid] | ||||
|
r37070 | handlers = { | ||
'idle': self._onframeidle, | ||||
|
r37076 | 'command-receiving': self._onframecommandreceiving, | ||
|
r37070 | 'errored': self._onframeerrored, | ||
} | ||||
meth = handlers.get(self._state) | ||||
if not meth: | ||||
raise error.ProgrammingError('unhandled state: %s' % self._state) | ||||
|
r37079 | return meth(frame) | ||
|
r37070 | |||
|
r37742 | def oncommandresponseready(self, stream, requestid, data): | ||
|
r37073 | """Signal that a bytes response is ready to be sent to the client. | ||
The raw bytes response is passed as an argument. | ||||
""" | ||||
|
r37304 | ensureserverstream(stream) | ||
|
r37081 | def sendframes(): | ||
|
r37742 | for frame in createcommandresponseframesfrombytes(stream, requestid, | ||
data): | ||||
|
r37081 | yield frame | ||
self._activecommands.remove(requestid) | ||||
result = sendframes() | ||||
|
r37074 | |||
if self._deferoutput: | ||||
|
r37081 | self._bufferedframegens.append(result) | ||
|
r37074 | return 'noop', {} | ||
else: | ||||
return 'sendframes', { | ||||
|
r37081 | 'framegen': result, | ||
|
r37074 | } | ||
|
r39595 | def oncommandresponsereadyobjects(self, stream, requestid, objs): | ||
"""Signal that objects are ready to be sent to the client. | ||||
``objs`` is an iterable of objects (typically a generator) that will | ||||
be encoded via CBOR and added to frames, which will be sent to the | ||||
client. | ||||
""" | ||||
|
r37746 | ensureserverstream(stream) | ||
|
r39595 | # We need to take care over exception handling. Uncaught exceptions | ||
# when generating frames could lead to premature end of the frame | ||||
# stream and the possibility of the server or client process getting | ||||
# in a bad state. | ||||
# | ||||
# Keep in mind that if ``objs`` is a generator, advancing it could | ||||
# raise exceptions that originated in e.g. wire protocol command | ||||
# functions. That is why we differentiate between exceptions raised | ||||
# when iterating versus other exceptions that occur. | ||||
# | ||||
# In all cases, when the function finishes, the request is fully | ||||
# handled and no new frames for it should be seen. | ||||
|
r37746 | def sendframes(): | ||
|
r39595 | emitted = False | ||
|
r40061 | alternatelocationsent = False | ||
|
r39596 | emitter = bufferingcommandresponseemitter(stream, requestid) | ||
|
r39595 | while True: | ||
try: | ||||
o = next(objs) | ||||
except StopIteration: | ||||
|
r39596 | for frame in emitter.send(None): | ||
yield frame | ||||
|
r39595 | if emitted: | ||
yield createcommandresponseeosframe(stream, requestid) | ||||
break | ||||
except error.WireprotoCommandError as e: | ||||
for frame in createcommanderrorresponse( | ||||
stream, requestid, e.message, e.messageargs): | ||||
yield frame | ||||
break | ||||
except Exception as e: | ||||
|
r39870 | for frame in createerrorframe( | ||
stream, requestid, '%s' % stringutil.forcebytestr(e), | ||||
errtype='server'): | ||||
|
r39595 | yield frame | ||
break | ||||
try: | ||||
|
r40061 | # Alternate location responses can only be the first and | ||
# only object in the output stream. | ||||
if isinstance(o, wireprototypes.alternatelocationresponse): | ||||
if emitted: | ||||
raise error.ProgrammingError( | ||||
'alternatelocationresponse seen after initial ' | ||||
'output object') | ||||
yield createalternatelocationresponseframe( | ||||
stream, requestid, o) | ||||
alternatelocationsent = True | ||||
emitted = True | ||||
continue | ||||
if alternatelocationsent: | ||||
raise error.ProgrammingError( | ||||
'object follows alternatelocationresponse') | ||||
|
r39595 | if not emitted: | ||
yield createcommandresponseokframe(stream, requestid) | ||||
emitted = True | ||||
|
r40056 | # Objects emitted by command functions can be serializable | ||
# data structures or special types. | ||||
# TODO consider extracting the content normalization to a | ||||
# standalone function, as it may be useful for e.g. cachers. | ||||
# A pre-encoded object is sent directly to the emitter. | ||||
if isinstance(o, wireprototypes.encodedresponse): | ||||
for frame in emitter.send(o.data): | ||||
|
r39596 | yield frame | ||
|
r40056 | # A regular object is CBOR encoded. | ||
else: | ||||
for chunk in cborutil.streamencode(o): | ||||
for frame in emitter.send(chunk): | ||||
yield frame | ||||
|
r39595 | except Exception as e: | ||
for frame in createerrorframe(stream, requestid, | ||||
'%s' % e, | ||||
errtype='server'): | ||||
yield frame | ||||
break | ||||
|
r37746 | |||
self._activecommands.remove(requestid) | ||||
return self._handlesendframes(sendframes()) | ||||
|
r37074 | def oninputeof(self): | ||
"""Signals that end of input has been received. | ||||
No more frames will be received. All pending activity should be | ||||
completed. | ||||
""" | ||||
|
r37076 | # TODO should we do anything about in-flight commands? | ||
|
r37074 | if not self._deferoutput or not self._bufferedframegens: | ||
return 'noop', {} | ||||
# If we buffered all our responses, emit those. | ||||
def makegen(): | ||||
for gen in self._bufferedframegens: | ||||
for frame in gen: | ||||
yield frame | ||||
|
r37073 | return 'sendframes', { | ||
|
r37074 | 'framegen': makegen(), | ||
|
r37073 | } | ||
|
r37746 | def _handlesendframes(self, framegen): | ||
if self._deferoutput: | ||||
self._bufferedframegens.append(framegen) | ||||
return 'noop', {} | ||||
else: | ||||
return 'sendframes', { | ||||
'framegen': framegen, | ||||
} | ||||
|
r37744 | def onservererror(self, stream, requestid, msg): | ||
|
r37304 | ensureserverstream(stream) | ||
|
r37746 | def sendframes(): | ||
for frame in createerrorframe(stream, requestid, msg, | ||||
errtype='server'): | ||||
yield frame | ||||
self._activecommands.remove(requestid) | ||||
return self._handlesendframes(sendframes()) | ||||
def oncommanderror(self, stream, requestid, message, args=None): | ||||
"""Called when a command encountered an error before sending output.""" | ||||
ensureserverstream(stream) | ||||
def sendframes(): | ||||
for frame in createcommanderrorresponse(stream, requestid, message, | ||||
args): | ||||
yield frame | ||||
self._activecommands.remove(requestid) | ||||
return self._handlesendframes(sendframes()) | ||||
|
r37073 | |||
|
r37305 | def makeoutputstream(self): | ||
"""Create a stream to be used for sending data to the client.""" | ||||
streamid = self._nextoutgoingstreamid | ||||
self._nextoutgoingstreamid += 2 | ||||
s = stream(streamid) | ||||
self._outgoingstreams[streamid] = s | ||||
return s | ||||
|
r37070 | def _makeerrorresult(self, msg): | ||
return 'error', { | ||||
'message': msg, | ||||
} | ||||
|
r37076 | def _makeruncommandresult(self, requestid): | ||
entry = self._receivingcommands[requestid] | ||||
|
r37308 | |||
if not entry['requestdone']: | ||||
self._state = 'errored' | ||||
raise error.ProgrammingError('should not be called without ' | ||||
'requestdone set') | ||||
|
r37076 | del self._receivingcommands[requestid] | ||
if self._receivingcommands: | ||||
self._state = 'command-receiving' | ||||
else: | ||||
self._state = 'idle' | ||||
|
r37308 | # Decode the payloads as CBOR. | ||
entry['payload'].seek(0) | ||||
|
r39477 | request = cborutil.decodeall(entry['payload'].getvalue())[0] | ||
|
r37308 | |||
if b'name' not in request: | ||||
self._state = 'errored' | ||||
return self._makeerrorresult( | ||||
_('command request missing "name" field')) | ||||
if b'args' not in request: | ||||
request[b'args'] = {} | ||||
|
r37081 | assert requestid not in self._activecommands | ||
self._activecommands.add(requestid) | ||||
|
r37070 | return 'runcommand', { | ||
|
r37076 | 'requestid': requestid, | ||
|
r37308 | 'command': request[b'name'], | ||
'args': request[b'args'], | ||||
|
r40061 | 'redirect': request.get(b'redirect'), | ||
|
r37076 | 'data': entry['data'].getvalue() if entry['data'] else None, | ||
|
r37070 | } | ||
def _makewantframeresult(self): | ||||
return 'wantframe', { | ||||
'state': self._state, | ||||
} | ||||
|
r37308 | def _validatecommandrequestframe(self, frame): | ||
new = frame.flags & FLAG_COMMAND_REQUEST_NEW | ||||
continuation = frame.flags & FLAG_COMMAND_REQUEST_CONTINUATION | ||||
if new and continuation: | ||||
self._state = 'errored' | ||||
return self._makeerrorresult( | ||||
_('received command request frame with both new and ' | ||||
'continuation flags set')) | ||||
if not new and not continuation: | ||||
self._state = 'errored' | ||||
return self._makeerrorresult( | ||||
_('received command request frame with neither new nor ' | ||||
'continuation flags set')) | ||||
|
r37079 | def _onframeidle(self, frame): | ||
|
r37070 | # The only frame type that should be received in this state is a | ||
# command request. | ||||
|
r37308 | if frame.typeid != FRAME_TYPE_COMMAND_REQUEST: | ||
|
r37070 | self._state = 'errored' | ||
return self._makeerrorresult( | ||||
|
r37308 | _('expected command request frame; got %d') % frame.typeid) | ||
res = self._validatecommandrequestframe(frame) | ||||
if res: | ||||
return res | ||||
|
r37070 | |||
|
r37079 | if frame.requestid in self._receivingcommands: | ||
|
r37076 | self._state = 'errored' | ||
return self._makeerrorresult( | ||||
|
r37079 | _('request with ID %d already received') % frame.requestid) | ||
|
r37076 | |||
|
r37081 | if frame.requestid in self._activecommands: | ||
self._state = 'errored' | ||||
|
r37308 | return self._makeerrorresult( | ||
_('request with ID %d is already active') % frame.requestid) | ||||
new = frame.flags & FLAG_COMMAND_REQUEST_NEW | ||||
moreframes = frame.flags & FLAG_COMMAND_REQUEST_MORE_FRAMES | ||||
expectingdata = frame.flags & FLAG_COMMAND_REQUEST_EXPECT_DATA | ||||
|
r37081 | |||
|
r37308 | if not new: | ||
self._state = 'errored' | ||||
return self._makeerrorresult( | ||||
_('received command request frame without new flag set')) | ||||
payload = util.bytesio() | ||||
payload.write(frame.payload) | ||||
|
r37076 | |||
|
r37079 | self._receivingcommands[frame.requestid] = { | ||
|
r37308 | 'payload': payload, | ||
|
r37076 | 'data': None, | ||
|
r37308 | 'requestdone': not moreframes, | ||
'expectingdata': bool(expectingdata), | ||||
|
r37076 | } | ||
|
r37070 | |||
|
r37308 | # This is the final frame for this request. Dispatch it. | ||
if not moreframes and not expectingdata: | ||||
|
r37079 | return self._makeruncommandresult(frame.requestid) | ||
|
r37070 | |||
|
r37308 | assert moreframes or expectingdata | ||
self._state = 'command-receiving' | ||||
return self._makewantframeresult() | ||||
|
r37070 | |||
|
r37079 | def _onframecommandreceiving(self, frame): | ||
|
r37308 | if frame.typeid == FRAME_TYPE_COMMAND_REQUEST: | ||
# Process new command requests as such. | ||||
if frame.flags & FLAG_COMMAND_REQUEST_NEW: | ||||
return self._onframeidle(frame) | ||||
res = self._validatecommandrequestframe(frame) | ||||
if res: | ||||
return res | ||||
|
r37076 | |||
# All other frames should be related to a command that is currently | ||||
|
r37081 | # receiving but is not active. | ||
if frame.requestid in self._activecommands: | ||||
self._state = 'errored' | ||||
return self._makeerrorresult( | ||||
_('received frame for request that is still active: %d') % | ||||
frame.requestid) | ||||
|
r37079 | if frame.requestid not in self._receivingcommands: | ||
|
r37070 | self._state = 'errored' | ||
|
r37076 | return self._makeerrorresult( | ||
_('received frame for request that is not receiving: %d') % | ||||
|
r37079 | frame.requestid) | ||
|
r37076 | |||
|
r37079 | entry = self._receivingcommands[frame.requestid] | ||
|
r37076 | |||
|
r37308 | if frame.typeid == FRAME_TYPE_COMMAND_REQUEST: | ||
moreframes = frame.flags & FLAG_COMMAND_REQUEST_MORE_FRAMES | ||||
expectingdata = bool(frame.flags & FLAG_COMMAND_REQUEST_EXPECT_DATA) | ||||
if entry['requestdone']: | ||||
self._state = 'errored' | ||||
return self._makeerrorresult( | ||||
_('received command request frame when request frames ' | ||||
'were supposedly done')) | ||||
if expectingdata != entry['expectingdata']: | ||||
|
r37076 | self._state = 'errored' | ||
|
r37308 | return self._makeerrorresult( | ||
_('mismatch between expect data flag and previous frame')) | ||||
entry['payload'].write(frame.payload) | ||||
|
r37076 | |||
|
r37308 | if not moreframes: | ||
entry['requestdone'] = True | ||||
if not moreframes and not expectingdata: | ||||
return self._makeruncommandresult(frame.requestid) | ||||
return self._makewantframeresult() | ||||
|
r37076 | |||
|
r37079 | elif frame.typeid == FRAME_TYPE_COMMAND_DATA: | ||
|
r37076 | if not entry['expectingdata']: | ||
self._state = 'errored' | ||||
return self._makeerrorresult(_( | ||||
'received command data frame for request that is not ' | ||||
|
r37079 | 'expecting data: %d') % frame.requestid) | ||
|
r37076 | |||
if entry['data'] is None: | ||||
entry['data'] = util.bytesio() | ||||
|
r37079 | return self._handlecommanddataframe(frame, entry) | ||
|
r37308 | else: | ||
|
r37070 | self._state = 'errored' | ||
|
r37308 | return self._makeerrorresult(_( | ||
'received unexpected frame type: %d') % frame.typeid) | ||||
|
r37070 | |||
|
r37079 | def _handlecommanddataframe(self, frame, entry): | ||
assert frame.typeid == FRAME_TYPE_COMMAND_DATA | ||||
|
r37070 | |||
# TODO support streaming data instead of buffering it. | ||||
|
r37079 | entry['data'].write(frame.payload) | ||
|
r37070 | |||
|
r37079 | if frame.flags & FLAG_COMMAND_DATA_CONTINUATION: | ||
|
r37070 | return self._makewantframeresult() | ||
|
r37079 | elif frame.flags & FLAG_COMMAND_DATA_EOS: | ||
|
r37076 | entry['data'].seek(0) | ||
|
r37079 | return self._makeruncommandresult(frame.requestid) | ||
|
r37070 | else: | ||
self._state = 'errored' | ||||
return self._makeerrorresult(_('command data frame without ' | ||||
'flags')) | ||||
|
r37079 | def _onframeerrored(self, frame): | ||
|
r37070 | return self._makeerrorresult(_('server already errored')) | ||
|
r37561 | |||
class commandrequest(object): | ||||
"""Represents a request to run a command.""" | ||||
|
r40060 | def __init__(self, requestid, name, args, datafh=None, redirect=None): | ||
|
r37561 | self.requestid = requestid | ||
self.name = name | ||||
self.args = args | ||||
self.datafh = datafh | ||||
|
r40060 | self.redirect = redirect | ||
|
r37561 | self.state = 'pending' | ||
class clientreactor(object): | ||||
"""Holds state of a client issuing frame-based protocol requests. | ||||
This is like ``serverreactor`` but for client-side state. | ||||
Each instance is bound to the lifetime of a connection. For persistent | ||||
connection transports using e.g. TCP sockets and speaking the raw | ||||
framing protocol, there will be a single instance for the lifetime of | ||||
the TCP socket. For transports where there are multiple discrete | ||||
interactions (say tunneled within in HTTP request), there will be a | ||||
separate instance for each distinct interaction. | ||||
""" | ||||
def __init__(self, hasmultiplesend=False, buffersends=True): | ||||
"""Create a new instance. | ||||
``hasmultiplesend`` indicates whether multiple sends are supported | ||||
by the transport. When True, it is possible to send commands immediately | ||||
instead of buffering until the caller signals an intent to finish a | ||||
send operation. | ||||
``buffercommands`` indicates whether sends should be buffered until the | ||||
last request has been issued. | ||||
""" | ||||
self._hasmultiplesend = hasmultiplesend | ||||
self._buffersends = buffersends | ||||
self._canissuecommands = True | ||||
self._cansend = True | ||||
self._nextrequestid = 1 | ||||
# We only support a single outgoing stream for now. | ||||
self._outgoingstream = stream(1) | ||||
self._pendingrequests = collections.deque() | ||||
self._activerequests = {} | ||||
|
r37562 | self._incomingstreams = {} | ||
|
r37561 | |||
|
r40060 | def callcommand(self, name, args, datafh=None, redirect=None): | ||
|
r37561 | """Request that a command be executed. | ||
Receives the command name, a dict of arguments to pass to the command, | ||||
and an optional file object containing the raw data for the command. | ||||
Returns a 3-tuple of (request, action, action data). | ||||
""" | ||||
if not self._canissuecommands: | ||||
raise error.ProgrammingError('cannot issue new commands') | ||||
requestid = self._nextrequestid | ||||
self._nextrequestid += 2 | ||||
|
r40060 | request = commandrequest(requestid, name, args, datafh=datafh, | ||
redirect=redirect) | ||||
|
r37561 | |||
if self._buffersends: | ||||
self._pendingrequests.append(request) | ||||
return request, 'noop', {} | ||||
else: | ||||
if not self._cansend: | ||||
raise error.ProgrammingError('sends cannot be performed on ' | ||||
'this instance') | ||||
if not self._hasmultiplesend: | ||||
self._cansend = False | ||||
self._canissuecommands = False | ||||
return request, 'sendframes', { | ||||
'framegen': self._makecommandframes(request), | ||||
} | ||||
def flushcommands(self): | ||||
"""Request that all queued commands be sent. | ||||
If any commands are buffered, this will instruct the caller to send | ||||
them over the wire. If no commands are buffered it instructs the client | ||||
to no-op. | ||||
If instances aren't configured for multiple sends, no new command | ||||
requests are allowed after this is called. | ||||
""" | ||||
if not self._pendingrequests: | ||||
return 'noop', {} | ||||
if not self._cansend: | ||||
raise error.ProgrammingError('sends cannot be performed on this ' | ||||
'instance') | ||||
# If the instance only allows sending once, mark that we have fired | ||||
# our one shot. | ||||
if not self._hasmultiplesend: | ||||
self._canissuecommands = False | ||||
self._cansend = False | ||||
def makeframes(): | ||||
while self._pendingrequests: | ||||
request = self._pendingrequests.popleft() | ||||
for frame in self._makecommandframes(request): | ||||
yield frame | ||||
return 'sendframes', { | ||||
'framegen': makeframes(), | ||||
} | ||||
def _makecommandframes(self, request): | ||||
"""Emit frames to issue a command request. | ||||
As a side-effect, update request accounting to reflect its changed | ||||
state. | ||||
""" | ||||
self._activerequests[request.requestid] = request | ||||
request.state = 'sending' | ||||
res = createcommandframes(self._outgoingstream, | ||||
request.requestid, | ||||
request.name, | ||||
request.args, | ||||
|
r40060 | datafh=request.datafh, | ||
redirect=request.redirect) | ||||
|
r37561 | |||
for frame in res: | ||||
yield frame | ||||
request.state = 'sent' | ||||
|
r37562 | |||
def onframerecv(self, frame): | ||||
"""Process a frame that has been received off the wire. | ||||
Returns a 2-tuple of (action, meta) describing further action the | ||||
caller needs to take as a result of receiving this frame. | ||||
""" | ||||
if frame.streamid % 2: | ||||
return 'error', { | ||||
'message': ( | ||||
_('received frame with odd numbered stream ID: %d') % | ||||
frame.streamid), | ||||
} | ||||
if frame.streamid not in self._incomingstreams: | ||||
if not frame.streamflags & STREAM_FLAG_BEGIN_STREAM: | ||||
return 'error', { | ||||
'message': _('received frame on unknown stream ' | ||||
'without beginning of stream flag set'), | ||||
} | ||||
|
r37674 | self._incomingstreams[frame.streamid] = stream(frame.streamid) | ||
|
r37562 | if frame.streamflags & STREAM_FLAG_ENCODING_APPLIED: | ||
raise error.ProgrammingError('support for decoding stream ' | ||||
'payloads not yet implemneted') | ||||
if frame.streamflags & STREAM_FLAG_END_STREAM: | ||||
del self._incomingstreams[frame.streamid] | ||||
if frame.requestid not in self._activerequests: | ||||
return 'error', { | ||||
'message': (_('received frame for inactive request ID: %d') % | ||||
frame.requestid), | ||||
} | ||||
request = self._activerequests[frame.requestid] | ||||
request.state = 'receiving' | ||||
handlers = { | ||||
|
r37742 | FRAME_TYPE_COMMAND_RESPONSE: self._oncommandresponseframe, | ||
|
r37744 | FRAME_TYPE_ERROR_RESPONSE: self._onerrorresponseframe, | ||
|
r37562 | } | ||
meth = handlers.get(frame.typeid) | ||||
if not meth: | ||||
raise error.ProgrammingError('unhandled frame type: %d' % | ||||
frame.typeid) | ||||
return meth(request, frame) | ||||
|
r37742 | def _oncommandresponseframe(self, request, frame): | ||
if frame.flags & FLAG_COMMAND_RESPONSE_EOS: | ||||
|
r37562 | request.state = 'received' | ||
del self._activerequests[request.requestid] | ||||
return 'responsedata', { | ||||
'request': request, | ||||
|
r37742 | 'expectmore': frame.flags & FLAG_COMMAND_RESPONSE_CONTINUATION, | ||
'eos': frame.flags & FLAG_COMMAND_RESPONSE_EOS, | ||||
|
r37562 | 'data': frame.payload, | ||
} | ||||
|
r37744 | |||
def _onerrorresponseframe(self, request, frame): | ||||
request.state = 'errored' | ||||
del self._activerequests[request.requestid] | ||||
# The payload should be a CBOR map. | ||||
|
r39477 | m = cborutil.decodeall(frame.payload)[0] | ||
|
r37744 | |||
return 'error', { | ||||
'request': request, | ||||
'type': m['type'], | ||||
'message': m['message'], | ||||
} | ||||