This is an archive of the discontinued Mercurial Phabricator instance.

Differential D4918

wireprotov2: handle stream encoding settings frames
ClosedPublic

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

Details

Reviewers
None
Group Reviewers
hg-reviewers
Commits
rHG57782791b7e9: wireprotov2: handle stream encoding settings frames
Summary

Like what we just did for the server reactor, we teach the client
reactor to handle stream encoding settings frames. The code is
very similar.

We define a method on the stream class to handle processing the data
within the decoded frames. However, it doesn't yet do anything useful.

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 rHG57782791b7e9: wireprotov2: handle stream encoding settings frames (authored by indygreg). · Explain WhyOct 10 2018, 10:57 AM
This revision was automatically updated to reflect the committed changes.

Revision Contents
Changeset List

Diff 11782

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)
def setdecoder(self, name, extraobjs):
"""Set the decoder for this stream.
Receives the stream profile name and any additional CBOR objects
decoded from the stream encoding settings frame payloads.
"""
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'],
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 = stream(1)
self._pendingrequests = collections.deque() self._pendingrequests = collections.deque()
self._activerequests = {} self._activerequests = {}
self._incomingstreams = {} self._incomingstreams = {}
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.
Receives the command name, a dict of arguments to pass to the command, 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. and an optional file object containing the raw data for the command.
Returns a 3-tuple of (request, action, action data). Returns a 3-tuple of (request, action, action data).
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]
if frame.typeid == FRAME_TYPE_STREAM_SETTINGS:
return self._onstreamsettingsframe(frame)
if frame.requestid not in self._activerequests: if frame.requestid not in self._activerequests:
return 'error', { return 'error', {
'message': (_('received frame for inactive request ID: %d') % 'message': (_('received frame for inactive request ID: %d') %
frame.requestid), frame.requestid),
} }
request = self._activerequests[frame.requestid] request = self._activerequests[frame.requestid]
request.state = 'receiving' request.state = 'receiving'
handlers = { handlers = {
FRAME_TYPE_COMMAND_RESPONSE: self._oncommandresponseframe, FRAME_TYPE_COMMAND_RESPONSE: self._oncommandresponseframe,
FRAME_TYPE_ERROR_RESPONSE: self._onerrorresponseframe, FRAME_TYPE_ERROR_RESPONSE: self._onerrorresponseframe,
} }
meth = handlers.get(frame.typeid) meth = handlers.get(frame.typeid)
if not meth: if not meth:
raise error.ProgrammingError('unhandled frame type: %d' % raise error.ProgrammingError('unhandled frame type: %d' %
frame.typeid) frame.typeid)
return meth(request, frame) return meth(request, frame)
def _onstreamsettingsframe(self, frame):
assert frame.typeid == FRAME_TYPE_STREAM_SETTINGS
more = frame.flags & FLAG_STREAM_ENCODING_SETTINGS_CONTINUATION
eos = frame.flags & FLAG_STREAM_ENCODING_SETTINGS_EOS
if more and eos:
return 'error', {
'message': (_('stream encoding settings frame cannot have both '
'continuation and end of stream flags set')),
}
if not more and not eos:
return 'error', {
'message': _('stream encoding settings frame must have '
'continuation or end of stream flag set'),
}
if frame.streamid not in self._streamsettingsdecoders:
decoder = cborutil.bufferingdecoder()
self._streamsettingsdecoders[frame.streamid] = decoder
decoder = self._streamsettingsdecoders[frame.streamid]
try:
decoder.decode(frame.payload)
except Exception as e:
return 'error', {
'message': (_('error decoding CBOR from stream encoding '
'settings frame: %s') %
stringutil.forcebytestr(e)),
}
if more:
return 'noop', {}
assert eos
decoded = decoder.getavailable()
del self._streamsettingsdecoders[frame.streamid]
if not decoded:
return 'error', {
'message': _('stream encoding settings frame did not contain '
'CBOR data'),
}
try:
self._incomingstreams[frame.streamid].setdecoder(decoded[0],
decoded[1:])
except Exception as e:
return 'error', {
'message': (_('error setting stream decoder: %s') %
stringutil.forcebytestr(e)),
}
return 'noop', {}
def _oncommandresponseframe(self, request, frame): def _oncommandresponseframe(self, request, frame):
if frame.flags & FLAG_COMMAND_RESPONSE_EOS: if frame.flags & FLAG_COMMAND_RESPONSE_EOS:
request.state = 'received' request.state = 'received'
del self._activerequests[request.requestid] del self._activerequests[request.requestid]
return 'responsedata', { return 'responsedata', {
'request': request, 'request': request,
'expectmore': frame.flags & FLAG_COMMAND_RESPONSE_CONTINUATION, 'expectmore': frame.flags & FLAG_COMMAND_RESPONSE_CONTINUATION,

tests/test-wireproto-clientreactor.py

from __future__ import absolute_import from __future__ import absolute_import
import unittest import unittest
from mercurial import ( from mercurial import (
error, error,
wireprotoframing as framing, wireprotoframing as framing,
) )
from mercurial.utils import (
cborutil,
)
ffs = framing.makeframefromhumanstring ffs = framing.makeframefromhumanstring
def sendframe(reactor, frame): def sendframe(reactor, frame):
"""Send a frame bytearray to a reactor.""" """Send a frame bytearray to a reactor."""
header = framing.parseheader(frame) header = framing.parseheader(frame)
payload = frame[framing.FRAME_HEADER_SIZE:] payload = frame[framing.FRAME_HEADER_SIZE:]
assert len(payload) == header.length assert len(payload) == header.length
self.assertEqual(len(frames), 1) self.assertEqual(len(frames), 1)
self.assertEqual(frames[0], self.assertEqual(frames[0],
ffs(b'1 1 stream-begin command-request new ' ffs(b'1 1 stream-begin command-request new '
b"cbor:{b'name': b'foo', " b"cbor:{b'name': b'foo', "
b"b'redirect': {b'targets': [b'a', b'b'], " b"b'redirect': {b'targets': [b'a', b'b'], "
b"b'hashes': [b'sha256']}}")) b"b'hashes': [b'sha256']}}"))
class StreamSettingsTests(unittest.TestCase):
def testnoflags(self):
reactor = framing.clientreactor(buffersends=False)
request, action, meta = reactor.callcommand(b'foo', {})
for f in meta[b'framegen']:
pass
action, meta = sendframe(reactor,
ffs(b'1 2 stream-begin stream-settings 0 '))
self.assertEqual(action, b'error')
self.assertEqual(meta, {
b'message': b'stream encoding settings frame must have '
b'continuation or end of stream flag set',
})
def testconflictflags(self):
reactor = framing.clientreactor(buffersends=False)
request, action, meta = reactor.callcommand(b'foo', {})
for f in meta[b'framegen']:
pass
action, meta = sendframe(reactor,
ffs(b'1 2 stream-begin stream-settings continuation|eos '))
self.assertEqual(action, b'error')
self.assertEqual(meta, {
b'message': b'stream encoding settings frame cannot have both '
b'continuation and end of stream flags set',
})
def testemptypayload(self):
reactor = framing.clientreactor(buffersends=False)
request, action, meta = reactor.callcommand(b'foo', {})
for f in meta[b'framegen']:
pass
action, meta = sendframe(reactor,
ffs(b'1 2 stream-begin stream-settings eos '))
self.assertEqual(action, b'error')
self.assertEqual(meta, {
b'message': b'stream encoding settings frame did not contain '
b'CBOR data'
})
def testbadcbor(self):
reactor = framing.clientreactor(buffersends=False)
request, action, meta = reactor.callcommand(b'foo', {})
for f in meta[b'framegen']:
pass
action, meta = sendframe(reactor,
ffs(b'1 2 stream-begin stream-settings eos badvalue'))
self.assertEqual(action, b'error')
def testsingleobject(self):
reactor = framing.clientreactor(buffersends=False)
request, action, meta = reactor.callcommand(b'foo', {})
for f in meta[b'framegen']:
pass
action, meta = sendframe(reactor,
ffs(b'1 2 stream-begin stream-settings eos cbor:b"identity"'))
self.assertEqual(action, b'noop')
self.assertEqual(meta, {})
def testmultipleobjects(self):
reactor = framing.clientreactor(buffersends=False)
request, action, meta = reactor.callcommand(b'foo', {})
for f in meta[b'framegen']:
pass
data = b''.join([
b''.join(cborutil.streamencode(b'identity')),
b''.join(cborutil.streamencode({b'foo', b'bar'})),
])
action, meta = sendframe(reactor,
ffs(b'1 2 stream-begin stream-settings eos %s' % data))
self.assertEqual(action, b'noop')
self.assertEqual(meta, {})
def testmultipleframes(self):
reactor = framing.clientreactor(buffersends=False)
request, action, meta = reactor.callcommand(b'foo', {})
for f in meta[b'framegen']:
pass
data = b''.join(cborutil.streamencode(b'identity'))
action, meta = sendframe(reactor,
ffs(b'1 2 stream-begin stream-settings continuation %s' %
data[0:3]))
self.assertEqual(action, b'noop')
self.assertEqual(meta, {})
action, meta = sendframe(reactor,
ffs(b'1 2 0 stream-settings eos %s' % data[3:]))
self.assertEqual(action, b'noop')
self.assertEqual(meta, {})
if __name__ == '__main__': if __name__ == '__main__':
import silenttestrunner import silenttestrunner
silenttestrunner.main(__name__) silenttestrunner.main(__name__)