This is an archive of the discontinued Mercurial Phabricator instance.

Differential D3388

wireprotov2: add support for more response types
ClosedPublic

Authored by indygreg on Apr 14 2018, 8:21 PM.

Details

Reviewers
durin42
Group Reviewers
hg-reviewers
Commits
rHG564a3eec6e63: wireprotov2: add support for more response types
Summary

This adds types to represent error and generator responses from
server commands.

Diff Detail

Repository
rHG Mercurial
Lint
Lint Skipped
Unit
Unit Tests Skipped

Event Timeline

indygreg created this revision.Apr 14 2018, 8:21 PM
indygreg updated this revision to Diff 8304.Apr 15 2018, 1:37 PM
durin42 accepted this revision.Apr 16 2018, 6:53 PM
This revision is now accepted and ready to land.Apr 16 2018, 6:53 PM
Closed by commit rHG564a3eec6e63: wireprotov2: add support for more response types (authored by indygreg). · Explain WhyApr 16 2018, 7:11 PM
This revision was automatically updated to reflect the committed changes.

Revision Contents
Changeset List

Diff 8304

mercurial/wireprotoframing.py

yield stream.makeframe(requestid=requestid, yield stream.makeframe(requestid=requestid,
typeid=FRAME_TYPE_COMMAND_RESPONSE, typeid=FRAME_TYPE_COMMAND_RESPONSE,
flags=flags, flags=flags,
payload=chunk) payload=chunk)
if done: if done:
break break
def createbytesresponseframesfromgen(stream, requestid, gen,
maxframesize=DEFAULT_MAX_FRAME_SIZE):
overall = cbor.dumps({b'status': b'ok'}, canonical=True)
yield stream.makeframe(requestid=requestid,
typeid=FRAME_TYPE_COMMAND_RESPONSE,
flags=FLAG_COMMAND_RESPONSE_CONTINUATION,
payload=overall)
cb = util.chunkbuffer(gen)
flags = 0
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
flags ^= FLAG_COMMAND_RESPONSE_CONTINUATION
flags |= FLAG_COMMAND_RESPONSE_EOS
yield stream.makeframe(requestid=requestid,
typeid=FRAME_TYPE_COMMAND_RESPONSE,
flags=flags,
payload=b'')
def createcommanderrorresponse(stream, requestid, message, args=None):
m = {
b'status': b'error',
b'error': {
b'message': message,
}
}
if args:
m[b'error'][b'args'] = args
overall = cbor.dumps(m, canonical=True)
yield stream.makeframe(requestid=requestid,
typeid=FRAME_TYPE_COMMAND_RESPONSE,
flags=FLAG_COMMAND_RESPONSE_EOS,
payload=overall)
def createerrorframe(stream, requestid, msg, errtype): def createerrorframe(stream, requestid, msg, errtype):
# TODO properly handle frame size limits. # TODO properly handle frame size limits.
assert len(msg) <= DEFAULT_MAX_FRAME_SIZE assert len(msg) <= DEFAULT_MAX_FRAME_SIZE
payload = cbor.dumps({ payload = cbor.dumps({
b'type': errtype, b'type': errtype,
b'message': [{b'msg': msg}], b'message': [{b'msg': msg}],
}, canonical=True) }, canonical=True)
if self._deferoutput: if self._deferoutput:
self._bufferedframegens.append(result) self._bufferedframegens.append(result)
return 'noop', {} return 'noop', {}
else: else:
return 'sendframes', { return 'sendframes', {
'framegen': result, 'framegen': result,
} }
def oncommandresponsereadygen(self, stream, requestid, gen):
"""Signal that a bytes response is ready, with data as a generator."""
ensureserverstream(stream)
def sendframes():
for frame in createbytesresponseframesfromgen(stream, requestid,
gen):
yield frame
self._activecommands.remove(requestid)
return self._handlesendframes(sendframes())
def oninputeof(self): def oninputeof(self):
"""Signals that end of input has been received. """Signals that end of input has been received.
No more frames will be received. All pending activity should be No more frames will be received. All pending activity should be
completed. completed.
""" """
# TODO should we do anything about in-flight commands? # TODO should we do anything about in-flight commands?
if not self._deferoutput or not self._bufferedframegens: if not self._deferoutput or not self._bufferedframegens:
return 'noop', {} return 'noop', {}
# If we buffered all our responses, emit those. # If we buffered all our responses, emit those.
def makegen(): def makegen():
for gen in self._bufferedframegens: for gen in self._bufferedframegens:
for frame in gen: for frame in gen:
yield frame yield frame
return 'sendframes', { return 'sendframes', {
'framegen': makegen(), 'framegen': makegen(),
} }
def _handlesendframes(self, framegen):
if self._deferoutput:
self._bufferedframegens.append(framegen)
return 'noop', {}
else:
return 'sendframes', {
'framegen': framegen,
}
def onservererror(self, stream, requestid, msg): def onservererror(self, stream, requestid, msg):
ensureserverstream(stream) ensureserverstream(stream)
return 'sendframes', { def sendframes():
'framegen': createerrorframe(stream, requestid, msg, for frame in createerrorframe(stream, requestid, msg,
errtype='server'), 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())
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 = stream(streamid)
self._outgoingstreams[streamid] = s self._outgoingstreams[streamid] = s

mercurial/wireprototypes.py

def __init__(self, gen=None): def __init__(self, gen=None):
self.gen = gen self.gen = gen
class cborresponse(object): class cborresponse(object):
"""Encode the response value as CBOR.""" """Encode the response value as CBOR."""
def __init__(self, v): def __init__(self, v):
self.value = v self.value = v
class v2errorresponse(object):
"""Represents a command error for version 2 transports."""
def __init__(self, message, args=None):
self.message = message
self.args = args
class v2streamingresponse(object):
"""A response whose data is supplied by a generator.
The generator can either consist of data structures to CBOR
encode or a stream of already-encoded bytes.
"""
def __init__(self, gen, compressible=True):
self.gen = gen
self.compressible = compressible
# list of nodes encoding / decoding # list of nodes encoding / decoding
def decodelist(l, sep=' '): def decodelist(l, sep=' '):
if l: if l:
return [bin(v) for v in l.split(sep)] return [bin(v) for v in l.split(sep)]
return [] return []
def encodelist(l, sep=' '): def encodelist(l, sep=' '):
try: try:

mercurial/wireprotov2server.py

res.status = b'200 OK' res.status = b'200 OK'
res.headers[b'Content-Type'] = FRAMINGTYPE res.headers[b'Content-Type'] = FRAMINGTYPE
if isinstance(rsp, wireprototypes.cborresponse): if isinstance(rsp, wireprototypes.cborresponse):
encoded = cbor.dumps(rsp.value, canonical=True) encoded = cbor.dumps(rsp.value, canonical=True)
action, meta = reactor.oncommandresponseready(outstream, action, meta = reactor.oncommandresponseready(outstream,
command['requestid'], command['requestid'],
encoded) encoded)
elif isinstance(rsp, wireprototypes.v2streamingresponse):
action, meta = reactor.oncommandresponsereadygen(outstream,
command['requestid'],
rsp.gen)
elif isinstance(rsp, wireprototypes.v2errorresponse):
action, meta = reactor.oncommanderror(outstream,
command['requestid'],
rsp.message,
rsp.args)
else: else:
action, meta = reactor.onservererror( action, meta = reactor.onservererror(
_('unhandled response type from wire proto command')) _('unhandled response type from wire proto command'))
if action == 'sendframes': if action == 'sendframes':
res.setbodygen(meta['framegen']) res.setbodygen(meta['framegen'])
return True return True
elif action == 'noop': elif action == 'noop':