This is an archive of the discontinued Mercurial Phabricator instance.

Differential D2089

wireproto: introduce type for raw byte responses (API)
ClosedPublic

Authored by indygreg on Feb 7 2018, 7:27 PM.

Details

Reviewers
durin42
Group Reviewers
hg-reviewers
Commits
rHG2f7290555c96: wireproto: introduce type for raw byte responses (API)
Summary

Right now we simply return a str/bytes instance for simple
responses. I want all wire protocol response types to be strongly
typed. So let's invent and use a type for raw bytes responses.

.. api::

Wire protocol command handlers now return a
wireprototypes.bytesresponse instead of a raw bytes instance.
Protocol handlers will continue handling bytes instances. However,
any extensions wrapping wire protocol commands will need to handle
the new type.

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.Feb 7 2018, 7:27 PM
indygreg updated this revision to Diff 5325.Feb 7 2018, 7:31 PM
indygreg edited the summary of this revision. (Show Details)Feb 7 2018, 11:32 PM
indygreg updated this revision to Diff 5341.
durin42 accepted this revision.Feb 12 2018, 4:25 PM
This revision is now accepted and ready to land.Feb 12 2018, 4:25 PM
Closed by commit rHG2f7290555c96: wireproto: introduce type for raw byte responses (API) (authored by indygreg). · Explain WhyFeb 12 2018, 4:34 PM
This revision was automatically updated to reflect the committed changes.

Revision Contents
Changeset List

Diff 5534

hgext/largefiles/proto.py

# Copyright 2011 Fog Creek Software # Copyright 2011 Fog Creek Software
# #
# This software may be used and distributed according to the terms of the # This software may be used and distributed according to the terms of the
# GNU General Public License version 2 or any later version. # GNU General Public License version 2 or any later version.
from __future__ import absolute_import from __future__ import absolute_import
import os import os
import re import re
from mercurial.i18n import _ from mercurial.i18n import _
from mercurial import ( from mercurial import (
error, error,
httppeer, httppeer,
util, util,
wireproto, wireproto,
wireprototypes,
) )
from . import ( from . import (
lfutil, lfutil,
) )
urlerr = util.urlerr urlerr = util.urlerr
urlreq = util.urlreq urlreq = util.urlreq
'''Server command for checking if a largefile is present - returns '2\n' if '''Server command for checking if a largefile is present - returns '2\n' if
the largefile is missing, '0\n' if it seems to be in good condition. the largefile is missing, '0\n' if it seems to be in good condition.
The value 1 is reserved for mismatched checksum, but that is too expensive The value 1 is reserved for mismatched checksum, but that is too expensive
to be verified on every stat and must be caught be running 'hg verify' to be verified on every stat and must be caught be running 'hg verify'
server side.''' server side.'''
filename = lfutil.findfile(repo, sha) filename = lfutil.findfile(repo, sha)
if not filename: if not filename:
return '2\n' return wireprototypes.bytesresponse('2\n')
return '0\n' return wireprototypes.bytesresponse('0\n')
def wirereposetup(ui, repo): def wirereposetup(ui, repo):
class lfileswirerepository(repo.__class__): class lfileswirerepository(repo.__class__):
def putlfile(self, sha, fd): def putlfile(self, sha, fd):
# unfortunately, httprepository._callpush tries to convert its # unfortunately, httprepository._callpush tries to convert its
# input file-like into a bundle before sending it, so we can't use # input file-like into a bundle before sending it, so we can't use
# it ... # it ...
if issubclass(self.__class__, httppeer.httppeer): if issubclass(self.__class__, httppeer.httppeer):

mercurial/wireproto.py

streamclone, streamclone,
util, util,
wireprototypes, wireprototypes,
) )
urlerr = util.urlerr urlerr = util.urlerr
urlreq = util.urlreq urlreq = util.urlreq
bytesresponse = wireprototypes.bytesresponse
ooberror = wireprototypes.ooberror ooberror = wireprototypes.ooberror
pushres = wireprototypes.pushres pushres = wireprototypes.pushres
pusherr = wireprototypes.pusherr pusherr = wireprototypes.pusherr
streamres = wireprototypes.streamres streamres = wireprototypes.streamres
streamres_legacy = wireprototypes.streamreslegacy streamres_legacy = wireprototypes.streamreslegacy
bundle2requiredmain = _('incompatible Mercurial client; bundle2 required') bundle2requiredmain = _('incompatible Mercurial client; bundle2 required')
bundle2requiredhint = _('see https://www.mercurial-scm.org/wiki/' bundle2requiredhint = _('see https://www.mercurial-scm.org/wiki/'
data['*'] = star data['*'] = star
else: else:
data[k] = vals[k] data[k] = vals[k]
result = func(repo, proto, *[data[k] for k in keys]) result = func(repo, proto, *[data[k] for k in keys])
else: else:
result = func(repo, proto) result = func(repo, proto)
if isinstance(result, ooberror): if isinstance(result, ooberror):
return result return result
# For now, all batchable commands must return bytesresponse or
# raw bytes (for backwards compatibility).
assert isinstance(result, (bytesresponse, bytes))
if isinstance(result, bytesresponse):
result = result.data
res.append(escapearg(result)) res.append(escapearg(result))
return ';'.join(res)
return bytesresponse(';'.join(res))
@wireprotocommand('between', 'pairs') @wireprotocommand('between', 'pairs')
def between(repo, proto, pairs): def between(repo, proto, pairs):
pairs = [decodelist(p, '-') for p in pairs.split(" ")] pairs = [decodelist(p, '-') for p in pairs.split(" ")]
r = [] r = []
for b in repo.between(pairs): for b in repo.between(pairs):
r.append(encodelist(b) + "\n") r.append(encodelist(b) + "\n")
return "".join(r)
return bytesresponse(''.join(r))
@wireprotocommand('branchmap') @wireprotocommand('branchmap')
def branchmap(repo, proto): def branchmap(repo, proto):
branchmap = repo.branchmap() branchmap = repo.branchmap()
heads = [] heads = []
for branch, nodes in branchmap.iteritems(): for branch, nodes in branchmap.iteritems():
branchname = urlreq.quote(encoding.fromlocal(branch)) branchname = urlreq.quote(encoding.fromlocal(branch))
branchnodes = encodelist(nodes) branchnodes = encodelist(nodes)
heads.append('%s %s' % (branchname, branchnodes)) heads.append('%s %s' % (branchname, branchnodes))
return '\n'.join(heads)
return bytesresponse('\n'.join(heads))
@wireprotocommand('branches', 'nodes') @wireprotocommand('branches', 'nodes')
def branches(repo, proto, nodes): def branches(repo, proto, nodes):
nodes = decodelist(nodes) nodes = decodelist(nodes)
r = [] r = []
for b in repo.branches(nodes): for b in repo.branches(nodes):
r.append(encodelist(b) + "\n") r.append(encodelist(b) + "\n")
return "".join(r)
return bytesresponse(''.join(r))
@wireprotocommand('clonebundles', '') @wireprotocommand('clonebundles', '')
def clonebundles(repo, proto): def clonebundles(repo, proto):
"""Server command for returning info for available bundles to seed clones. """Server command for returning info for available bundles to seed clones.
Clients will parse this response and determine what bundle to fetch. Clients will parse this response and determine what bundle to fetch.
Extensions may wrap this command to filter or dynamically emit data Extensions may wrap this command to filter or dynamically emit data
depending on the request. e.g. you could advertise URLs for the closest depending on the request. e.g. you could advertise URLs for the closest
data center given the client's IP address. data center given the client's IP address.
""" """
return repo.vfs.tryread('clonebundles.manifest') return bytesresponse(repo.vfs.tryread('clonebundles.manifest'))
wireprotocaps = ['lookup', 'changegroupsubset', 'branchmap', 'pushkey', wireprotocaps = ['lookup', 'changegroupsubset', 'branchmap', 'pushkey',
'known', 'getbundle', 'unbundlehash', 'batch'] 'known', 'getbundle', 'unbundlehash', 'batch']
def _capabilities(repo, proto): def _capabilities(repo, proto):
"""return a list of capabilities for a repo """return a list of capabilities for a repo
This function exists to allow extensions to easily wrap capabilities This function exists to allow extensions to easily wrap capabilities
caps.append('compression=%s' % comptypes) caps.append('compression=%s' % comptypes)
return caps return caps
# If you are writing an extension and consider wrapping this function. Wrap # If you are writing an extension and consider wrapping this function. Wrap
# `_capabilities` instead. # `_capabilities` instead.
@wireprotocommand('capabilities') @wireprotocommand('capabilities')
def capabilities(repo, proto): def capabilities(repo, proto):
return ' '.join(_capabilities(repo, proto)) return bytesresponse(' '.join(_capabilities(repo, proto)))
@wireprotocommand('changegroup', 'roots') @wireprotocommand('changegroup', 'roots')
def changegroup(repo, proto, roots): def changegroup(repo, proto, roots):
nodes = decodelist(roots) nodes = decodelist(roots)
outgoing = discovery.outgoing(repo, missingroots=nodes, outgoing = discovery.outgoing(repo, missingroots=nodes,
missingheads=repo.heads()) missingheads=repo.heads())
cg = changegroupmod.makechangegroup(repo, outgoing, '01', 'serve') cg = changegroupmod.makechangegroup(repo, outgoing, '01', 'serve')
gen = iter(lambda: cg.read(32768), '') gen = iter(lambda: cg.read(32768), '')
return streamres(gen=gen) return streamres(gen=gen)
@wireprotocommand('changegroupsubset', 'bases heads') @wireprotocommand('changegroupsubset', 'bases heads')
def changegroupsubset(repo, proto, bases, heads): def changegroupsubset(repo, proto, bases, heads):
bases = decodelist(bases) bases = decodelist(bases)
heads = decodelist(heads) heads = decodelist(heads)
outgoing = discovery.outgoing(repo, missingroots=bases, outgoing = discovery.outgoing(repo, missingroots=bases,
missingheads=heads) missingheads=heads)
cg = changegroupmod.makechangegroup(repo, outgoing, '01', 'serve') cg = changegroupmod.makechangegroup(repo, outgoing, '01', 'serve')
gen = iter(lambda: cg.read(32768), '') gen = iter(lambda: cg.read(32768), '')
return streamres(gen=gen) return streamres(gen=gen)
@wireprotocommand('debugwireargs', 'one two *') @wireprotocommand('debugwireargs', 'one two *')
def debugwireargs(repo, proto, one, two, others): def debugwireargs(repo, proto, one, two, others):
# only accept optional args from the known set # only accept optional args from the known set
opts = options('debugwireargs', ['three', 'four'], others) opts = options('debugwireargs', ['three', 'four'], others)
return repo.debugwireargs(one, two, **pycompat.strkwargs(opts)) return bytesresponse(repo.debugwireargs(one, two,
**pycompat.strkwargs(opts)))
@wireprotocommand('getbundle', '*') @wireprotocommand('getbundle', '*')
def getbundle(repo, proto, others): def getbundle(repo, proto, others):
opts = options('getbundle', gboptsmap.keys(), others) opts = options('getbundle', gboptsmap.keys(), others)
for k, v in opts.iteritems(): for k, v in opts.iteritems():
keytype = gboptsmap[k] keytype = gboptsmap[k]
if keytype == 'nodes': if keytype == 'nodes':
opts[k] = decodelist(v) opts[k] = decodelist(v)
chunks = bundler.getchunks() chunks = bundler.getchunks()
prefercompressed = False prefercompressed = False
return streamres(gen=chunks, prefer_uncompressed=not prefercompressed) return streamres(gen=chunks, prefer_uncompressed=not prefercompressed)
@wireprotocommand('heads') @wireprotocommand('heads')
def heads(repo, proto): def heads(repo, proto):
h = repo.heads() h = repo.heads()
return encodelist(h) + "\n" return bytesresponse(encodelist(h) + '\n')
@wireprotocommand('hello') @wireprotocommand('hello')
def hello(repo, proto): def hello(repo, proto):
'''the hello command returns a set of lines describing various '''the hello command returns a set of lines describing various
interesting things about the server, in an RFC822-like format. interesting things about the server, in an RFC822-like format.
Currently the only one defined is "capabilities", which Currently the only one defined is "capabilities", which
consists of a line in the form: consists of a line in the form:
capabilities: space separated list of tokens capabilities: space separated list of tokens
''' '''
return "capabilities: %s\n" % (capabilities(repo, proto)) caps = capabilities(repo, proto).data
return bytesresponse('capabilities: %s\n' % caps)
@wireprotocommand('listkeys', 'namespace') @wireprotocommand('listkeys', 'namespace')
def listkeys(repo, proto, namespace): def listkeys(repo, proto, namespace):
d = repo.listkeys(encoding.tolocal(namespace)).items() d = repo.listkeys(encoding.tolocal(namespace)).items()
return pushkeymod.encodekeys(d) return bytesresponse(pushkeymod.encodekeys(d))
@wireprotocommand('lookup', 'key') @wireprotocommand('lookup', 'key')
def lookup(repo, proto, key): def lookup(repo, proto, key):
try: try:
k = encoding.tolocal(key) k = encoding.tolocal(key)
c = repo[k] c = repo[k]
r = c.hex() r = c.hex()
success = 1 success = 1
except Exception as inst: except Exception as inst:
r = str(inst) r = str(inst)
success = 0 success = 0
return "%d %s\n" % (success, r) return bytesresponse('%d %s\n' % (success, r))
@wireprotocommand('known', 'nodes *') @wireprotocommand('known', 'nodes *')
def known(repo, proto, nodes, others): def known(repo, proto, nodes, others):
return ''.join(b and "1" or "0" for b in repo.known(decodelist(nodes))) v = ''.join(b and '1' or '0' for b in repo.known(decodelist(nodes)))
return bytesresponse(v)
@wireprotocommand('pushkey', 'namespace key old new') @wireprotocommand('pushkey', 'namespace key old new')
def pushkey(repo, proto, namespace, key, old, new): def pushkey(repo, proto, namespace, key, old, new):
# compatibility with pre-1.8 clients which were accidentally # compatibility with pre-1.8 clients which were accidentally
# sending raw binary nodes rather than utf-8-encoded hex # sending raw binary nodes rather than utf-8-encoded hex
if len(new) == 20 and util.escapestr(new) != new: if len(new) == 20 and util.escapestr(new) != new:
# looks like it could be a binary node # looks like it could be a binary node
try: try:
new.decode('utf-8') new.decode('utf-8')
new = encoding.tolocal(new) # but cleanly decodes as UTF-8 new = encoding.tolocal(new) # but cleanly decodes as UTF-8
except UnicodeDecodeError: except UnicodeDecodeError:
pass # binary, leave unmodified pass # binary, leave unmodified
else: else:
new = encoding.tolocal(new) # normal path new = encoding.tolocal(new) # normal path
with proto.mayberedirectstdio() as output: with proto.mayberedirectstdio() as output:
r = repo.pushkey(encoding.tolocal(namespace), encoding.tolocal(key), r = repo.pushkey(encoding.tolocal(namespace), encoding.tolocal(key),
encoding.tolocal(old), new) or False encoding.tolocal(old), new) or False
output = output.getvalue() if output else '' output = output.getvalue() if output else ''
return '%s\n%s' % (int(r), output) return bytesresponse('%s\n%s' % (int(r), output))
@wireprotocommand('stream_out') @wireprotocommand('stream_out')
def stream(repo, proto): def stream(repo, proto):
'''If the server supports streaming clone, it advertises the "stream" '''If the server supports streaming clone, it advertises the "stream"
capability with a value representing the version and flags of the repo capability with a value representing the version and flags of the repo
it is serving. Client checks to see if it understands the format. it is serving. Client checks to see if it understands the format.
''' '''
return streamres_legacy(streamclone.generatev1wireproto(repo)) return streamres_legacy(streamclone.generatev1wireproto(repo))

mercurial/wireprotoserver.py

req.respond(HTTP_OK, HGERRTYPE, req.respond(HTTP_OK, HGERRTYPE,
body=_('requested wire protocol command is not available ' body=_('requested wire protocol command is not available '
'over HTTP')) 'over HTTP'))
return [] return []
if isinstance(rsp, bytes): if isinstance(rsp, bytes):
req.respond(HTTP_OK, HGTYPE, body=rsp) req.respond(HTTP_OK, HGTYPE, body=rsp)
return [] return []
elif isinstance(rsp, wireprototypes.bytesresponse):
req.respond(HTTP_OK, HGTYPE, body=rsp.data)
return []
elif isinstance(rsp, wireprototypes.streamreslegacy): elif isinstance(rsp, wireprototypes.streamreslegacy):
gen = rsp.gen gen = rsp.gen
req.respond(HTTP_OK, HGTYPE) req.respond(HTTP_OK, HGTYPE)
return gen return gen
elif isinstance(rsp, wireprototypes.streamres): elif isinstance(rsp, wireprototypes.streamres):
gen = rsp.gen gen = rsp.gen
# This code for compression should not be streamres specific. It # This code for compression should not be streamres specific. It
def serve_one(self): def serve_one(self):
cmd = self._fin.readline()[:-1] cmd = self._fin.readline()[:-1]
if cmd and wireproto.commands.commandavailable(cmd, self._proto): if cmd and wireproto.commands.commandavailable(cmd, self._proto):
rsp = wireproto.dispatch(self._repo, self._proto, cmd) rsp = wireproto.dispatch(self._repo, self._proto, cmd)
if isinstance(rsp, bytes): if isinstance(rsp, bytes):
_sshv1respondbytes(self._fout, rsp) _sshv1respondbytes(self._fout, rsp)
elif isinstance(rsp, wireprototypes.bytesresponse):
_sshv1respondbytes(self._fout, rsp.data)
elif isinstance(rsp, wireprototypes.streamres): elif isinstance(rsp, wireprototypes.streamres):
_sshv1respondstream(self._fout, rsp) _sshv1respondstream(self._fout, rsp)
elif isinstance(rsp, wireprototypes.streamreslegacy): elif isinstance(rsp, wireprototypes.streamreslegacy):
_sshv1respondstream(self._fout, rsp) _sshv1respondstream(self._fout, rsp)
elif isinstance(rsp, wireprototypes.pushres): elif isinstance(rsp, wireprototypes.pushres):
_sshv1respondbytes(self._fout, b'') _sshv1respondbytes(self._fout, b'')
_sshv1respondbytes(self._fout, bytes(rsp.res)) _sshv1respondbytes(self._fout, bytes(rsp.res))
elif isinstance(rsp, wireprototypes.pusherr): elif isinstance(rsp, wireprototypes.pusherr):

mercurial/wireprototypes.py

# Copyright 2018 Gregory Szorc <gregory.szorc@gmail.com> # Copyright 2018 Gregory Szorc <gregory.szorc@gmail.com>
# #
# This software may be used and distributed according to the terms of the # This software may be used and distributed according to the terms of the
# GNU General Public License version 2 or any later version. # GNU General Public License version 2 or any later version.
from __future__ import absolute_import from __future__ import absolute_import
class bytesresponse(object):
"""A wire protocol response consisting of raw bytes."""
def __init__(self, data):
self.data = data
class ooberror(object): class ooberror(object):
"""wireproto reply: failure of a batch of operation """wireproto reply: failure of a batch of operation
Something failed during a batch call. The error message is stored in Something failed during a batch call. The error message is stored in
`self.message`. `self.message`.
""" """
def __init__(self, message): def __init__(self, message):
self.message = message self.message = message

tests/sshprotoext.py

def serve_forever(self): def serve_forever(self):
l = self._fin.readline() l = self._fin.readline()
assert l == b'hello\n' assert l == b'hello\n'
# Respond to unknown commands with an empty reply. # Respond to unknown commands with an empty reply.
wireprotoserver._sshv1respondbytes(self._fout, b'') wireprotoserver._sshv1respondbytes(self._fout, b'')
l = self._fin.readline() l = self._fin.readline()
assert l == b'between\n' assert l == b'between\n'
rsp = wireproto.dispatch(self._repo, self._proto, b'between') rsp = wireproto.dispatch(self._repo, self._proto, b'between')
wireprotoserver._sshv1respondbytes(self._fout, rsp) wireprotoserver._sshv1respondbytes(self._fout, rsp.data)
super(prehelloserver, self).serve_forever() super(prehelloserver, self).serve_forever()
class upgradev2server(wireprotoserver.sshserver): class upgradev2server(wireprotoserver.sshserver):
"""Tests behavior for clients that issue upgrade to version 2.""" """Tests behavior for clients that issue upgrade to version 2."""
def serve_forever(self): def serve_forever(self):
name = wireprotoserver.SSHV2 name = wireprotoserver.SSHV2
l = self._fin.readline() l = self._fin.readline()
assert l.startswith(b'upgrade ') assert l.startswith(b'upgrade ')
token, caps = l[:-1].split(b' ')[1:] token, caps = l[:-1].split(b' ')[1:]
assert caps == b'proto=%s' % name assert caps == b'proto=%s' % name
# Filter hello and between requests. # Filter hello and between requests.
l = self._fin.readline() l = self._fin.readline()
assert l == b'hello\n' assert l == b'hello\n'
l = self._fin.readline() l = self._fin.readline()
assert l == b'between\n' assert l == b'between\n'
l = self._fin.readline() l = self._fin.readline()
assert l == 'pairs 81\n' assert l == 'pairs 81\n'
self._fin.read(81) self._fin.read(81)
# Send the upgrade response. # Send the upgrade response.
self._fout.write(b'upgraded %s %s\n' % (token, name)) self._fout.write(b'upgraded %s %s\n' % (token, name))
servercaps = wireproto.capabilities(self._repo, self._proto) servercaps = wireproto.capabilities(self._repo, self._proto)
rsp = b'capabilities: %s' % servercaps rsp = b'capabilities: %s' % servercaps.data
self._fout.write(b'%d\n' % len(rsp)) self._fout.write(b'%d\n' % len(rsp))
self._fout.write(rsp) self._fout.write(rsp)
self._fout.write(b'\n') self._fout.write(b'\n')
self._fout.flush() self._fout.flush()
super(upgradev2server, self).serve_forever() super(upgradev2server, self).serve_forever()
def performhandshake(orig, ui, stdin, stdout, stderr): def performhandshake(orig, ui, stdin, stdout, stderr):

tests/test-wireproto.py

from __future__ import absolute_import, print_function from __future__ import absolute_import, print_function
from mercurial import ( from mercurial import (
error,
util, util,
wireproto, wireproto,
wireprototypes,
) )
stringio = util.stringio stringio = util.stringio
class proto(object): class proto(object):
def __init__(self, args): def __init__(self, args):
self.args = args self.args = args
def getargs(self, spec): def getargs(self, spec):
args = self.args args = self.args
def close(self): def close(self):
pass pass
def capabilities(self): def capabilities(self):
return ['batch'] return ['batch']
def _call(self, cmd, **args): def _call(self, cmd, **args):
return wireproto.dispatch(self.serverrepo, proto(args), cmd) res = wireproto.dispatch(self.serverrepo, proto(args), cmd)
if isinstance(res, wireprototypes.bytesresponse):
return res.data
elif isinstance(res, bytes):
return res
else:
raise error.Abort('dummy client does not support response type')
def _callstream(self, cmd, **args): def _callstream(self, cmd, **args):
return stringio(self._call(cmd, **args)) return stringio(self._call(cmd, **args))
@wireproto.batchable @wireproto.batchable
def greet(self, name): def greet(self, name):
f = wireproto.future() f = wireproto.future()
yield {'name': mangle(name)}, f yield {'name': mangle(name)}, f