This is an archive of the discontinued Mercurial Phabricator instance.

Differential D4920

wireprotov2: establish dedicated classes for input and output streams
ClosedPublic

Authored by indygreg on Oct 8 2018, 8:29 PM.

Details

Reviewers
None
Group Reviewers
hg-reviewers
Commits
rHG5d44c4d1d516: wireprotov2: establish dedicated classes for input and output streams
Summary

Streams are unidirectional. As part of implementing encoding/decoding
support, it became clear that it didn't make sense for a generic
"stream" class to hold functionality related to both encoding and
decoding. So we create new classes to represent the flavor of
stream.

Diff Detail

Repository
rHG Mercurial
Lint
Automatic diff as part of commit; lint not applicable.
Unit
Automatic diff as part of commit; unit tests not applicable.

Event Timeline

indygreg created this revision.Oct 8 2018, 8:29 PM
Closed by commit rHG5d44c4d1d516: wireprotov2: establish dedicated classes for input and output streams (authored by indygreg). · Explain WhyOct 10 2018, 10:57 AM
This revision was automatically updated to reflect the committed changes.

Revision Contents
Changeset List

PathPackages
Mmercurial/wireprotoframing.py (15 lines)

Diff 11784

mercurial/wireprotoframing.py

streamflags = 0 streamflags = 0
if not self._active: if not self._active:
streamflags |= STREAM_FLAG_BEGIN_STREAM streamflags |= STREAM_FLAG_BEGIN_STREAM
self._active = True self._active = True
return makeframe(requestid, self.streamid, streamflags, typeid, flags, return makeframe(requestid, self.streamid, streamflags, typeid, flags,
payload) payload)
class inputstream(stream):
"""Represents a stream used for receiving data."""
def setdecoder(self, name, extraobjs): def setdecoder(self, name, extraobjs):
"""Set the decoder for this stream. """Set the decoder for this stream.
Receives the stream profile name and any additional CBOR objects Receives the stream profile name and any additional CBOR objects
decoded from the stream encoding settings frame payloads. decoded from the stream encoding settings frame payloads.
""" """
class outputstream(stream):
"""Represents a stream used for sending data."""
def ensureserverstream(stream): def ensureserverstream(stream):
if stream.streamid % 2: if stream.streamid % 2:
raise error.ProgrammingError('server should only write to even ' raise error.ProgrammingError('server should only write to even '
'numbered streams; %d is not even' % 'numbered streams; %d is not even' %
stream.streamid) stream.streamid)
DEFAULT_PROTOCOL_SETTINGS = { DEFAULT_PROTOCOL_SETTINGS = {
'contentencodings': [b'identity'], 'contentencodings': [b'identity'],
if frame.streamid not in self._incomingstreams: if frame.streamid not in self._incomingstreams:
if not frame.streamflags & STREAM_FLAG_BEGIN_STREAM: if not frame.streamflags & STREAM_FLAG_BEGIN_STREAM:
self._state = 'errored' self._state = 'errored'
return self._makeerrorresult( return self._makeerrorresult(
_('received frame on unknown inactive stream without ' _('received frame on unknown inactive stream without '
'beginning of stream flag set')) 'beginning of stream flag set'))
self._incomingstreams[frame.streamid] = stream(frame.streamid) self._incomingstreams[frame.streamid] = inputstream(frame.streamid)
if frame.streamflags & STREAM_FLAG_ENCODING_APPLIED: if frame.streamflags & STREAM_FLAG_ENCODING_APPLIED:
# TODO handle decoding frames # TODO handle decoding frames
self._state = 'errored' self._state = 'errored'
raise error.ProgrammingError('support for decoding stream payloads ' raise error.ProgrammingError('support for decoding stream payloads '
'not yet implemented') 'not yet implemented')
if frame.streamflags & STREAM_FLAG_END_STREAM: if frame.streamflags & STREAM_FLAG_END_STREAM:
return self._handlesendframes(sendframes()) return self._handlesendframes(sendframes())
def makeoutputstream(self): def makeoutputstream(self):
"""Create a stream to be used for sending data to the client.""" """Create a stream to be used for sending data to the client."""
streamid = self._nextoutgoingstreamid streamid = self._nextoutgoingstreamid
self._nextoutgoingstreamid += 2 self._nextoutgoingstreamid += 2
s = stream(streamid) s = outputstream(streamid)
self._outgoingstreams[streamid] = s self._outgoingstreams[streamid] = s
return s return s
def _makeerrorresult(self, msg): def _makeerrorresult(self, msg):
return 'error', { return 'error', {
'message': msg, 'message': msg,
} }
self._hasmultiplesend = hasmultiplesend self._hasmultiplesend = hasmultiplesend
self._buffersends = buffersends self._buffersends = buffersends
self._canissuecommands = True self._canissuecommands = True
self._cansend = True self._cansend = True
self._nextrequestid = 1 self._nextrequestid = 1
# We only support a single outgoing stream for now. # We only support a single outgoing stream for now.
self._outgoingstream = stream(1) self._outgoingstream = outputstream(1)
self._pendingrequests = collections.deque() self._pendingrequests = collections.deque()
self._activerequests = {} self._activerequests = {}
self._incomingstreams = {} self._incomingstreams = {}
self._streamsettingsdecoders = {} self._streamsettingsdecoders = {}
def callcommand(self, name, args, datafh=None, redirect=None): def callcommand(self, name, args, datafh=None, redirect=None):
"""Request that a command be executed. """Request that a command be executed.
if frame.streamid not in self._incomingstreams: if frame.streamid not in self._incomingstreams:
if not frame.streamflags & STREAM_FLAG_BEGIN_STREAM: if not frame.streamflags & STREAM_FLAG_BEGIN_STREAM:
return 'error', { return 'error', {
'message': _('received frame on unknown stream ' 'message': _('received frame on unknown stream '
'without beginning of stream flag set'), 'without beginning of stream flag set'),
} }
self._incomingstreams[frame.streamid] = stream(frame.streamid) self._incomingstreams[frame.streamid] = inputstream(
frame.streamid)
if frame.streamflags & STREAM_FLAG_ENCODING_APPLIED: if frame.streamflags & STREAM_FLAG_ENCODING_APPLIED:
raise error.ProgrammingError('support for decoding stream ' raise error.ProgrammingError('support for decoding stream '
'payloads not yet implemneted') 'payloads not yet implemneted')
if frame.streamflags & STREAM_FLAG_END_STREAM: if frame.streamflags & STREAM_FLAG_END_STREAM:
del self._incomingstreams[frame.streamid] del self._incomingstreams[frame.streamid]