This is an archive of the discontinued Mercurial Phabricator instance.

Differential D1861

wireproto: drop support for reader interface from streamres (API)
ClosedPublic

Authored by indygreg on Jan 15 2018, 6:20 PM.

Details

Reviewers
None
Group Reviewers
hg-reviewers
Commits
rHG8cdb671dbd0b: wireproto: drop support for reader interface from streamres (API)
Summary

2add671bf55b and later commits overhauled support for compression
and output handling in the wire protocol.

Fast forward 14 months and all wire protocol commands except
the legacy "changegroup" and "changegroupsubset" commands feed
a generator to streamres.

I no longer think it is worth maintaining support for the old
"reader" API (which allows you to specify an object having a
read() method to obtain data).

This commit refactors the legacy wire protocol commands to feed
a generator to the streamres. We also drop support for the "reader"
argument and the code that was using it.

As part of the change, chunks over the SSH protocol have increased
in size for these commands. But these commands are really ancient,
so I doubt anyone will notice.

.. api::

wireproto.streamres.__init__ no longer accepts a "reader" argument.
Use the "gen" argument instead.

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.Jan 15 2018, 6:20 PM
Closed by commit rHG8cdb671dbd0b: wireproto: drop support for reader interface from streamres (API) (authored by indygreg). · Explain WhyJan 18 2018, 8:41 AM
This revision was automatically updated to reflect the committed changes.

Revision Contents
Changeset List

Diff 4894

mercurial/hgweb/protocol.py

for chunk in gen: for chunk in gen:
yield chunk yield chunk
rsp = wireproto.dispatch(repo, p, cmd) rsp = wireproto.dispatch(repo, p, cmd)
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, wireproto.streamres): elif isinstance(rsp, wireproto.streamres):
if rsp.reader:
gen = iter(lambda: rsp.reader.read(32768), '')
else:
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
# is here because we only compress streamres at the moment. # is here because we only compress streamres at the moment.
mediatype, engine, engineopts = p.responsetype(rsp.v1compressible) mediatype, engine, engineopts = p.responsetype(rsp.v1compressible)
if mediatype == HGTYPE and rsp.v1compressible: if mediatype == HGTYPE and rsp.v1compressible:
gen = engine.compressstream(gen, engineopts) gen = engine.compressstream(gen, engineopts)
elif mediatype == HGTYPE2: elif mediatype == HGTYPE2:

mercurial/sshserver.py

def sendresponse(self, v): def sendresponse(self, v):
self.fout.write("%d\n" % len(v)) self.fout.write("%d\n" % len(v))
self.fout.write(v) self.fout.write(v)
self.fout.flush() self.fout.flush()
def sendstream(self, source): def sendstream(self, source):
write = self.fout.write write = self.fout.write
for chunk in source.gen:
if source.reader:
gen = iter(lambda: source.reader.read(4096), '')
else:
gen = source.gen
for chunk in gen:
write(chunk) write(chunk)
self.fout.flush() self.fout.flush()
def sendpushresponse(self, rsp): def sendpushresponse(self, rsp):
self.sendresponse('') self.sendresponse('')
self.sendresponse(str(rsp.res)) self.sendresponse(str(rsp.res))
def sendpusherror(self, rsp): def sendpusherror(self, rsp):

mercurial/wireproto.py

# server side # server side
# wire protocol command can either return a string or one of these classes. # wire protocol command can either return a string or one of these classes.
class streamres(object): class streamres(object):
"""wireproto reply: binary stream """wireproto reply: binary stream
The call was successful and the result is a stream. The call was successful and the result is a stream.
Accepts either a generator or an object with a ``read(size)`` method. Accepts a generator containing chunks of data to be sent to the client.
``v1compressible`` indicates whether this data can be compressed to ``v1compressible`` indicates whether this data can be compressed to
"version 1" clients (technically: HTTP peers using "version 1" clients (technically: HTTP peers using
application/mercurial-0.1 media type). This flag should NOT be used on application/mercurial-0.1 media type). This flag should NOT be used on
new commands because new clients should support a more modern compression new commands because new clients should support a more modern compression
mechanism. mechanism.
""" """
def __init__(self, gen=None, reader=None, v1compressible=False): def __init__(self, gen=None, v1compressible=False):
self.gen = gen self.gen = gen
self.reader = reader
self.v1compressible = v1compressible self.v1compressible = v1compressible
class pushres(object): class pushres(object):
"""wireproto reply: success with simple integer return """wireproto reply: success with simple integer return
The call was successful and returned an integer contained in `self.res`. The call was successful and returned an integer contained in `self.res`.
""" """
def __init__(self, res): def __init__(self, res):
return ' '.join(_capabilities(repo, proto)) return ' '.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')
return streamres(reader=cg, v1compressible=True) gen = iter(lambda: cg.read(32768), '')
return streamres(gen=gen, v1compressible=True)
@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')
return streamres(reader=cg, v1compressible=True) gen = iter(lambda: cg.read(32768), '')
return streamres(gen=gen, v1compressible=True)
@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 repo.debugwireargs(one, two, **pycompat.strkwargs(opts))
@wireprotocommand('getbundle', '*') @wireprotocommand('getbundle', '*')