This is an archive of the discontinued Mercurial Phabricator instance.

Differential D4777

wireprotov2: server support for sending content redirects
ClosedPublic

Authored by indygreg on Sep 26 2018, 9:09 PM.

Details

Reviewers
None
Group Reviewers
hg-reviewers
Commits
rHGb099e6032f38: wireprotov2: server support for sending content redirects
Summary

A "content redirect" can be sent in place of inline response content.

In terms of code, we model a content redirect as a special type of
response object holding the attributes describing that redirect.

Sending a content redirect thus becomes as simple as the object
emission layer sending an instance of that type. A cacher using
externally-addressable content storage could replace the outgoing
object stream with an object advertising its location.

The bulk of the code in this commit is teaching the output layer
which handles the object stream to recognize alternate location
objects. The rules are that if an alternate location object is
present, it must be the first and only object in the object stream.
Otherwise the server emits an error.

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.

Revision Contents
Changeset List

Diff 11628

mercurial/wireprotoframing.py

from .i18n import _ from .i18n import _
from .thirdparty import ( from .thirdparty import (
attr, attr,
) )
from . import ( from . import (
encoding, encoding,
error, error,
pycompat,
util, util,
wireprototypes, wireprototypes,
) )
from .utils import ( from .utils import (
cborutil, cborutil,
stringutil, stringutil,
) )
def createcommandresponseeosframe(stream, requestid): def createcommandresponseeosframe(stream, requestid):
"""Create an empty payload frame representing command end-of-stream.""" """Create an empty payload frame representing command end-of-stream."""
return stream.makeframe(requestid=requestid, return stream.makeframe(requestid=requestid,
typeid=FRAME_TYPE_COMMAND_RESPONSE, typeid=FRAME_TYPE_COMMAND_RESPONSE,
flags=FLAG_COMMAND_RESPONSE_EOS, flags=FLAG_COMMAND_RESPONSE_EOS,
payload=b'') payload=b'')
def createalternatelocationresponseframe(stream, requestid, location):
data = {
b'status': b'redirect',
b'location': {
b'url': location.url,
b'mediatype': location.mediatype,
}
}
for a in (r'size', r'fullhashes', r'fullhashseed', r'serverdercerts',
r'servercadercerts'):
value = getattr(location, a)
if value is not None:
data[b'location'][pycompat.bytestr(a)] = value
return stream.makeframe(requestid=requestid,
typeid=FRAME_TYPE_COMMAND_RESPONSE,
flags=FLAG_COMMAND_RESPONSE_CONTINUATION,
payload=b''.join(cborutil.streamencode(data)))
def createcommanderrorresponse(stream, requestid, message, args=None): def createcommanderrorresponse(stream, requestid, message, args=None):
# TODO should this be using a list of {'msg': ..., 'args': {}} so atom # TODO should this be using a list of {'msg': ..., 'args': {}} so atom
# formatting works consistently? # formatting works consistently?
m = { m = {
b'status': b'error', b'status': b'error',
b'error': { b'error': {
b'message': message, b'message': message,
} }
# functions. That is why we differentiate between exceptions raised # functions. That is why we differentiate between exceptions raised
# when iterating versus other exceptions that occur. # when iterating versus other exceptions that occur.
# #
# In all cases, when the function finishes, the request is fully # In all cases, when the function finishes, the request is fully
# handled and no new frames for it should be seen. # handled and no new frames for it should be seen.
def sendframes(): def sendframes():
emitted = False emitted = False
alternatelocationsent = False
emitter = bufferingcommandresponseemitter(stream, requestid) emitter = bufferingcommandresponseemitter(stream, requestid)
while True: while True:
try: try:
o = next(objs) o = next(objs)
except StopIteration: except StopIteration:
for frame in emitter.send(None): for frame in emitter.send(None):
yield frame yield frame
stream, requestid, '%s' % stringutil.forcebytestr(e), stream, requestid, '%s' % stringutil.forcebytestr(e),
errtype='server'): errtype='server'):
yield frame yield frame
break break
try: try:
# Alternate location responses can only be the first and
# only object in the output stream.
if isinstance(o, wireprototypes.alternatelocationresponse):
if emitted:
raise error.ProgrammingError(
'alternatelocationresponse seen after initial '
'output object')
yield createalternatelocationresponseframe(
stream, requestid, o)
alternatelocationsent = True
emitted = True
continue
if alternatelocationsent:
raise error.ProgrammingError(
'object follows alternatelocationresponse')
if not emitted: if not emitted:
yield createcommandresponseokframe(stream, requestid) yield createcommandresponseokframe(stream, requestid)
emitted = True emitted = True
# Objects emitted by command functions can be serializable # Objects emitted by command functions can be serializable
# data structures or special types. # data structures or special types.
# TODO consider extracting the content normalization to a # TODO consider extracting the content normalization to a
# standalone function, as it may be useful for e.g. cachers. # standalone function, as it may be useful for e.g. cachers.
assert requestid not in self._activecommands assert requestid not in self._activecommands
self._activecommands.add(requestid) self._activecommands.add(requestid)
return 'runcommand', { return 'runcommand', {
'requestid': requestid, 'requestid': requestid,
'command': request[b'name'], 'command': request[b'name'],
'args': request[b'args'], 'args': request[b'args'],
'redirect': request.get(b'redirect'),
'data': entry['data'].getvalue() if entry['data'] else None, 'data': entry['data'].getvalue() if entry['data'] else None,
} }
def _makewantframeresult(self): def _makewantframeresult(self):
return 'wantframe', { return 'wantframe', {
'state': self._state, 'state': self._state,
} }

mercurial/wireprototypes.py

Wire protocol version 2 only. Wire protocol version 2 only.
Commands typically emit Python objects that are encoded and sent over the Commands typically emit Python objects that are encoded and sent over the
wire. If commands emit an object of this type, the encoding step is bypassed wire. If commands emit an object of this type, the encoding step is bypassed
and the content from this object is used instead. and the content from this object is used instead.
""" """
data = attr.ib() data = attr.ib()
@attr.s
class alternatelocationresponse(object):
"""Represents a response available at an alternate location.
Instances are sent in place of actual response objects when the server
is sending a "content redirect" response.
Only compatible with wire protocol version 2.
"""
url = attr.ib()
mediatype = attr.ib()
size = attr.ib(default=None)
fullhashes = attr.ib(default=None)
fullhashseed = attr.ib(default=None)
serverdercerts = attr.ib(default=None)
servercadercerts = attr.ib(default=None)

mercurial/wireprotov2server.py

res.headers[b'Content-Type'] = b'text/plain' res.headers[b'Content-Type'] = b'text/plain'
res.setbodybytes(_('command in frame must match command in URL')) res.setbodybytes(_('command in frame must match command in URL'))
return True return True
res.status = b'200 OK' res.status = b'200 OK'
res.headers[b'Content-Type'] = FRAMINGTYPE res.headers[b'Content-Type'] = FRAMINGTYPE
try: try:
objs = dispatch(repo, proto, command['command']) objs = dispatch(repo, proto, command['command'], command['redirect'])
action, meta = reactor.oncommandresponsereadyobjects( action, meta = reactor.oncommandresponsereadyobjects(
outstream, command['requestid'], objs) outstream, command['requestid'], objs)
except error.WireprotoCommandError as e: except error.WireprotoCommandError as e:
action, meta = reactor.oncommanderror( action, meta = reactor.oncommanderror(
outstream, command['requestid'], e.message, e.messageargs) outstream, command['requestid'], e.message, e.messageargs)
return False return False
else: else:
raise error.ProgrammingError('unhandled event from reactor: %s' % raise error.ProgrammingError('unhandled event from reactor: %s' %
action) action)
def getdispatchrepo(repo, proto, command): def getdispatchrepo(repo, proto, command):
return repo.filtered('served') return repo.filtered('served')
def dispatch(repo, proto, command): def dispatch(repo, proto, command, redirect):
"""Run a wire protocol command. """Run a wire protocol command.
Returns an iterable of objects that will be sent to the client. Returns an iterable of objects that will be sent to the client.
""" """
repo = getdispatchrepo(repo, proto, command) repo = getdispatchrepo(repo, proto, command)
entry = COMMANDS[command] entry = COMMANDS[command]
func = entry.func func = entry.func
spec = entry.args spec = entry.args
args = proto.getargs(spec) args = proto.getargs(spec)
# There is some duplicate boilerplate code here for calling the command and # There is some duplicate boilerplate code here for calling the command and
# emitting objects. It is either that or a lot of indented code that looks # emitting objects. It is either that or a lot of indented code that looks
# like a pyramid (since there are a lot of code paths that result in not # like a pyramid (since there are a lot of code paths that result in not
# using the cacher). # using the cacher).
callcommand = lambda: func(repo, proto, **pycompat.strkwargs(args)) callcommand = lambda: func(repo, proto, **pycompat.strkwargs(args))
# Request is not cacheable. Don't bother instantiating a cacher. # Request is not cacheable. Don't bother instantiating a cacher.
if not entry.cachekeyfn: if not entry.cachekeyfn:
for o in callcommand(): for o in callcommand():
yield o yield o
return return
if redirect:
redirecttargets = redirect[b'targets']
redirecthashes = redirect[b'hashes']
else:
redirecttargets = []
redirecthashes = []
cacher = makeresponsecacher(repo, proto, command, args, cacher = makeresponsecacher(repo, proto, command, args,
cborutil.streamencode) cborutil.streamencode,
redirecttargets=redirecttargets,
redirecthashes=redirecthashes)
# But we have no cacher. Do default handling. # But we have no cacher. Do default handling.
if not cacher: if not cacher:
for o in callcommand(): for o in callcommand():
yield o yield o
return return
with cacher: with cacher:
hasher = hashlib.sha1() hasher = hashlib.sha1()
for chunk in cborutil.streamencode(state): for chunk in cborutil.streamencode(state):
hasher.update(chunk) hasher.update(chunk)
return pycompat.sysbytes(hasher.hexdigest()) return pycompat.sysbytes(hasher.hexdigest())
return cachekeyfn return cachekeyfn
def makeresponsecacher(repo, proto, command, args, objencoderfn): def makeresponsecacher(repo, proto, command, args, objencoderfn,
redirecttargets, redirecthashes):
"""Construct a cacher for a cacheable command. """Construct a cacher for a cacheable command.
Returns an ``iwireprotocolcommandcacher`` instance. Returns an ``iwireprotocolcommandcacher`` instance.
Extensions can monkeypatch this function to provide custom caching Extensions can monkeypatch this function to provide custom caching
backends. backends.
""" """
return None return None

tests/test-http-api-httpv2.t

s> host: $LOCALIP:$HGPORT\r\n (glob) s> host: $LOCALIP:$HGPORT\r\n (glob)
s> \r\n s> \r\n
s> \'\x00\x00\x01\x00\x01\x01\x11\xa2Dargs\xa2Dbar1CvalCfooDval1DnameHcommand1 s> \'\x00\x00\x01\x00\x01\x01\x11\xa2Dargs\xa2Dbar1CvalCfooDval1DnameHcommand1
s> makefile('rb', None) s> makefile('rb', None)
s> HTTP/1.1 200 OK\r\n s> HTTP/1.1 200 OK\r\n
s> Server: testing stub value\r\n s> Server: testing stub value\r\n
s> Date: $HTTP_DATE$\r\n s> Date: $HTTP_DATE$\r\n
s> Content-Type: text/plain\r\n s> Content-Type: text/plain\r\n
s> Content-Length: 205\r\n s> Content-Length: 223\r\n
s> \r\n s> \r\n
s> received: 1 1 1 \xa2Dargs\xa2Dbar1CvalCfooDval1DnameHcommand1\n s> received: 1 1 1 \xa2Dargs\xa2Dbar1CvalCfooDval1DnameHcommand1\n
s> ["runcommand", {"args": {"bar1": "val", "foo": "val1"}, "command": "command1", "data": null, "requestid": 1}]\n s> ["runcommand", {"args": {"bar1": "val", "foo": "val1"}, "command": "command1", "data": null, "redirect": null, "requestid": 1}]\n
s> received: <no frame>\n s> received: <no frame>\n
s> {"action": "noop"} s> {"action": "noop"}
Multiple requests to regular command URL are not allowed Multiple requests to regular command URL are not allowed
$ sendhttpraw << EOF $ sendhttpraw << EOF
> httprequest POST api/$HTTPV2/ro/customreadonly > httprequest POST api/$HTTPV2/ro/customreadonly
> accept: $MEDIATYPE > accept: $MEDIATYPE

tests/test-wireproto-content-redirects.t

$ . $TESTDIR/wireprotohelpers.sh $ . $TESTDIR/wireprotohelpers.sh
$ cat >> $HGRCPATH << EOF
> [extensions]
> blackbox =
> [blackbox]
> track = simplecache
> EOF
$ hg init server $ hg init server
$ enablehttpv2 server $ enablehttpv2 server
$ cd server $ cd server
$ cat >> .hg/hgrc << EOF $ cat >> .hg/hgrc << EOF
> [extensions] > [extensions]
> simplecache = $TESTDIR/wireprotosimplecache.py > simplecache = $TESTDIR/wireprotosimplecache.py
> [simplecache]
> cacheapi = true
> EOF > EOF
$ echo a0 > a $ echo a0 > a
$ echo b0 > b $ echo b0 > b
$ hg -q commit -A -m 'commit 0' $ hg -q commit -A -m 'commit 0'
$ echo a1 > a $ echo a1 > a
$ hg commit -m 'commit 1' $ hg commit -m 'commit 1'
b'https://example.com/' b'https://example.com/'
] ]
} }
] ]
} }
} }
] ]
Set up the server to issue content redirects to its built-in API server.
$ cat > redirects.py << EOF
> [
> {
> b'name': b'local',
> b'protocol': b'http',
> b'uris': [b'http://example.com/'],
> },
> ]
> EOF
Request to eventual cache URL should return 404 (validating the cache server works)
$ sendhttpraw << EOF
> httprequest GET api/simplecache/missingkey
> user-agent: test
> EOF
using raw connection to peer
s> GET /api/simplecache/missingkey HTTP/1.1\r\n
s> Accept-Encoding: identity\r\n
s> user-agent: test\r\n
s> host: $LOCALIP:$HGPORT\r\n (glob)
s> \r\n
s> makefile('rb', None)
s> HTTP/1.1 404 Not Found\r\n
s> Server: testing stub value\r\n
s> Date: $HTTP_DATE$\r\n
s> Content-Type: text/plain\r\n
s> Content-Length: 22\r\n
s> \r\n
s> key not found in cache
Send a cacheable request
$ sendhttpv2peer << EOF
> command manifestdata
> nodes eval:[b'\x99\x2f\x47\x79\x02\x9a\x3d\xf8\xd0\x66\x6d\x00\xbb\x92\x4f\x69\x63\x4e\x26\x41']
> tree eval:b''
> fields eval:[b'parents']
> EOF
creating http peer for wire protocol version 2
sending manifestdata command
s> POST /api/exp-http-v2-0002/ro/manifestdata HTTP/1.1\r\n
s> Accept-Encoding: identity\r\n
s> accept: application/mercurial-exp-framing-0005\r\n
s> content-type: application/mercurial-exp-framing-0005\r\n
s> content-length: 128\r\n
s> host: $LOCALIP:$HGPORT\r\n (glob)
s> user-agent: Mercurial debugwireproto\r\n
s> \r\n
s> x\x00\x00\x01\x00\x01\x01\x11\xa3Dargs\xa3Ffields\x81GparentsEnodes\x81T\x99/Gy\x02\x9a=\xf8\xd0fm\x00\xbb\x92OicN&ADtree@DnameLmanifestdataHredirect\xa2Fhashes\x82Fsha256Dsha1Gtargets\x81Elocal
s> makefile('rb', None)
s> HTTP/1.1 200 OK\r\n
s> Server: testing stub value\r\n
s> Date: $HTTP_DATE$\r\n
s> Content-Type: application/mercurial-exp-framing-0005\r\n
s> Transfer-Encoding: chunked\r\n
s> \r\n
s> 13\r\n
s> \x0b\x00\x00\x01\x00\x02\x011
s> \xa1FstatusBok
s> \r\n
received frame(size=11; request=1; stream=2; streamflags=stream-begin; type=command-response; flags=continuation)
s> 63\r\n
s> [\x00\x00\x01\x00\x02\x001
s> \xa1Jtotalitems\x01\xa2DnodeT\x99/Gy\x02\x9a=\xf8\xd0fm\x00\xbb\x92OicN&AGparents\x82T\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00T\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00
s> \r\n
received frame(size=91; request=1; stream=2; streamflags=; type=command-response; flags=continuation)
s> 8\r\n
s> \x00\x00\x00\x01\x00\x02\x002
s> \r\n
s> 0\r\n
s> \r\n
received frame(size=0; request=1; stream=2; streamflags=; type=command-response; flags=eos)
response: gen[
{
b'totalitems': 1
},
{
b'node': b'\x99/Gy\x02\x9a=\xf8\xd0fm\x00\xbb\x92OicN&A',
b'parents': [
b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00',
b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00'
]
}
]
Cached entry should be available on server
$ sendhttpraw << EOF
> httprequest GET api/simplecache/c045a581599d58608efd3d93d8129841f2af04a0
> user-agent: test
> EOF
using raw connection to peer
s> GET /api/simplecache/c045a581599d58608efd3d93d8129841f2af04a0 HTTP/1.1\r\n
s> Accept-Encoding: identity\r\n
s> user-agent: test\r\n
s> host: $LOCALIP:$HGPORT\r\n (glob)
s> \r\n
s> makefile('rb', None)
s> HTTP/1.1 200 OK\r\n
s> Server: testing stub value\r\n
s> Date: $HTTP_DATE$\r\n
s> Content-Type: application/mercurial-cbor\r\n
s> Content-Length: 91\r\n
s> \r\n
s> \xa1Jtotalitems\x01\xa2DnodeT\x99/Gy\x02\x9a=\xf8\xd0fm\x00\xbb\x92OicN&AGparents\x82T\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00T\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00
cbor> [
{
b'totalitems': 1
},
{
b'node': b'\x99/Gy\x02\x9a=\xf8\xd0fm\x00\xbb\x92OicN&A',
b'parents': [
b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00',
b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00'
]
}
]
2nd request should result in content redirect response
$ sendhttpv2peer << EOF
> command manifestdata
> nodes eval:[b'\x99\x2f\x47\x79\x02\x9a\x3d\xf8\xd0\x66\x6d\x00\xbb\x92\x4f\x69\x63\x4e\x26\x41']
> tree eval:b''
> fields eval:[b'parents']
> EOF
creating http peer for wire protocol version 2
sending manifestdata command
s> POST /api/exp-http-v2-0002/ro/manifestdata HTTP/1.1\r\n
s> Accept-Encoding: identity\r\n
s> accept: application/mercurial-exp-framing-0005\r\n
s> content-type: application/mercurial-exp-framing-0005\r\n
s> content-length: 128\r\n
s> host: $LOCALIP:$HGPORT\r\n (glob)
s> user-agent: Mercurial debugwireproto\r\n
s> \r\n
s> x\x00\x00\x01\x00\x01\x01\x11\xa3Dargs\xa3Ffields\x81GparentsEnodes\x81T\x99/Gy\x02\x9a=\xf8\xd0fm\x00\xbb\x92OicN&ADtree@DnameLmanifestdataHredirect\xa2Fhashes\x82Fsha256Dsha1Gtargets\x81Elocal
s> makefile('rb', None)
s> HTTP/1.1 200 OK\r\n
s> Server: testing stub value\r\n
s> Date: $HTTP_DATE$\r\n
s> Content-Type: application/mercurial-exp-framing-0005\r\n
s> Transfer-Encoding: chunked\r\n
s> \r\n
s> *\r\n (glob)
s> \x*\x00\x00\x01\x00\x02\x011 (glob)
s> \xa2Hlocation\xa2ImediatypeX\x1aapplication/mercurial-cborCurl*http://*:$HGPORT/api/simplecache/c045a581599d58608efd3d93d8129841f2af04a0FstatusHredirect (glob)
s> \r\n
received frame(size=*; request=1; stream=2; streamflags=stream-begin; type=command-response; flags=continuation) (glob)
s> 8\r\n
s> \x00\x00\x00\x01\x00\x02\x001
s> \r\n
s> 8\r\n
received frame(size=0; request=1; stream=2; streamflags=; type=command-response; flags=continuation)
s> \x00\x00\x00\x01\x00\x02\x002
s> \r\n
s> 0\r\n
s> \r\n
received frame(size=0; request=1; stream=2; streamflags=; type=command-response; flags=eos)
abort: redirect responses not yet supported
[255]
$ cat error.log $ cat error.log
$ killdaemons.py $ killdaemons.py
$ cat .hg/blackbox.log
*> cacher constructed for manifestdata (glob)
*> cache miss for c045a581599d58608efd3d93d8129841f2af04a0 (glob)
*> storing cache entry for c045a581599d58608efd3d93d8129841f2af04a0 (glob)
*> cacher constructed for manifestdata (glob)
*> cache hit for c045a581599d58608efd3d93d8129841f2af04a0 (glob)
*> sending content redirect for c045a581599d58608efd3d93d8129841f2af04a0 to http://*:$HGPORT/api/simplecache/c045a581599d58608efd3d93d8129841f2af04a0 (glob)

tests/test-wireproto-serverreactor.py

stream = framing.stream(1) stream = framing.stream(1)
results = list(sendcommandframes(reactor, stream, 1, b'mycommand', {})) results = list(sendcommandframes(reactor, stream, 1, b'mycommand', {}))
self.assertEqual(len(results), 1) self.assertEqual(len(results), 1)
self.assertaction(results[0], b'runcommand') self.assertaction(results[0], b'runcommand')
self.assertEqual(results[0][1], { self.assertEqual(results[0][1], {
b'requestid': 1, b'requestid': 1,
b'command': b'mycommand', b'command': b'mycommand',
b'args': {}, b'args': {},
b'redirect': None,
b'data': None, b'data': None,
}) })
result = reactor.oninputeof() result = reactor.oninputeof()
self.assertaction(result, b'noop') self.assertaction(result, b'noop')
def test1argument(self): def test1argument(self):
reactor = makereactor() reactor = makereactor()
stream = framing.stream(1) stream = framing.stream(1)
results = list(sendcommandframes(reactor, stream, 41, b'mycommand', results = list(sendcommandframes(reactor, stream, 41, b'mycommand',
{b'foo': b'bar'})) {b'foo': b'bar'}))
self.assertEqual(len(results), 1) self.assertEqual(len(results), 1)
self.assertaction(results[0], b'runcommand') self.assertaction(results[0], b'runcommand')
self.assertEqual(results[0][1], { self.assertEqual(results[0][1], {
b'requestid': 41, b'requestid': 41,
b'command': b'mycommand', b'command': b'mycommand',
b'args': {b'foo': b'bar'}, b'args': {b'foo': b'bar'},
b'redirect': None,
b'data': None, b'data': None,
}) })
def testmultiarguments(self): def testmultiarguments(self):
reactor = makereactor() reactor = makereactor()
stream = framing.stream(1) stream = framing.stream(1)
results = list(sendcommandframes(reactor, stream, 1, b'mycommand', results = list(sendcommandframes(reactor, stream, 1, b'mycommand',
{b'foo': b'bar', b'biz': b'baz'})) {b'foo': b'bar', b'biz': b'baz'}))
self.assertEqual(len(results), 1) self.assertEqual(len(results), 1)
self.assertaction(results[0], b'runcommand') self.assertaction(results[0], b'runcommand')
self.assertEqual(results[0][1], { self.assertEqual(results[0][1], {
b'requestid': 1, b'requestid': 1,
b'command': b'mycommand', b'command': b'mycommand',
b'args': {b'foo': b'bar', b'biz': b'baz'}, b'args': {b'foo': b'bar', b'biz': b'baz'},
b'redirect': None,
b'data': None, b'data': None,
}) })
def testsimplecommanddata(self): def testsimplecommanddata(self):
reactor = makereactor() reactor = makereactor()
stream = framing.stream(1) stream = framing.stream(1)
results = list(sendcommandframes(reactor, stream, 1, b'mycommand', {}, results = list(sendcommandframes(reactor, stream, 1, b'mycommand', {},
util.bytesio(b'data!'))) util.bytesio(b'data!')))
self.assertEqual(len(results), 2) self.assertEqual(len(results), 2)
self.assertaction(results[0], b'wantframe') self.assertaction(results[0], b'wantframe')
self.assertaction(results[1], b'runcommand') self.assertaction(results[1], b'runcommand')
self.assertEqual(results[1][1], { self.assertEqual(results[1][1], {
b'requestid': 1, b'requestid': 1,
b'command': b'mycommand', b'command': b'mycommand',
b'args': {}, b'args': {},
b'redirect': None,
b'data': b'data!', b'data': b'data!',
}) })
def testmultipledataframes(self): def testmultipledataframes(self):
frames = [ frames = [
ffs(b'1 1 stream-begin command-request new|have-data ' ffs(b'1 1 stream-begin command-request new|have-data '
b"cbor:{b'name': b'mycommand'}"), b"cbor:{b'name': b'mycommand'}"),
ffs(b'1 1 0 command-data continuation data1'), ffs(b'1 1 0 command-data continuation data1'),
ffs(b'1 1 0 command-data continuation data2'), ffs(b'1 1 0 command-data continuation data2'),
ffs(b'1 1 0 command-data eos data3'), ffs(b'1 1 0 command-data eos data3'),
] ]
reactor = makereactor() reactor = makereactor()
results = list(sendframes(reactor, frames)) results = list(sendframes(reactor, frames))
self.assertEqual(len(results), 4) self.assertEqual(len(results), 4)
for i in range(3): for i in range(3):
self.assertaction(results[i], b'wantframe') self.assertaction(results[i], b'wantframe')
self.assertaction(results[3], b'runcommand') self.assertaction(results[3], b'runcommand')
self.assertEqual(results[3][1], { self.assertEqual(results[3][1], {
b'requestid': 1, b'requestid': 1,
b'command': b'mycommand', b'command': b'mycommand',
b'args': {}, b'args': {},
b'redirect': None,
b'data': b'data1data2data3', b'data': b'data1data2data3',
}) })
def testargumentanddata(self): def testargumentanddata(self):
frames = [ frames = [
ffs(b'1 1 stream-begin command-request new|have-data ' ffs(b'1 1 stream-begin command-request new|have-data '
b"cbor:{b'name': b'command', b'args': {b'key': b'val'," b"cbor:{b'name': b'command', b'args': {b'key': b'val',"
b"b'foo': b'bar'}}"), b"b'foo': b'bar'}}"),
ffs(b'1 1 0 command-data continuation value1'), ffs(b'1 1 0 command-data continuation value1'),
ffs(b'1 1 0 command-data eos value2'), ffs(b'1 1 0 command-data eos value2'),
] ]
reactor = makereactor() reactor = makereactor()
results = list(sendframes(reactor, frames)) results = list(sendframes(reactor, frames))
self.assertaction(results[-1], b'runcommand') self.assertaction(results[-1], b'runcommand')
self.assertEqual(results[-1][1], { self.assertEqual(results[-1][1], {
b'requestid': 1, b'requestid': 1,
b'command': b'command', b'command': b'command',
b'args': { b'args': {
b'key': b'val', b'key': b'val',
b'foo': b'bar', b'foo': b'bar',
}, },
b'redirect': None,
b'data': b'value1value2', b'data': b'value1value2',
}) })
def testnewandcontinuation(self): def testnewandcontinuation(self):
result = self._sendsingleframe(makereactor(), result = self._sendsingleframe(makereactor(),
ffs(b'1 1 stream-begin command-request new|continuation ')) ffs(b'1 1 stream-begin command-request new|continuation '))
self.assertaction(result, b'error') self.assertaction(result, b'error')
self.assertEqual(result[1], { self.assertEqual(result[1], {
list(result[1][b'framegen']) list(result[1][b'framegen'])
for i in range(3): for i in range(3):
self.assertaction(results[i], b'runcommand') self.assertaction(results[i], b'runcommand')
self.assertEqual(results[i][1], { self.assertEqual(results[i][1], {
b'requestid': 1, b'requestid': 1,
b'command': b'command', b'command': b'command',
b'args': {}, b'args': {},
b'redirect': None,
b'data': None, b'data': None,
}) })
def testconflictingrequestid(self): def testconflictingrequestid(self):
"""Request ID for new command matching in-flight command is illegal.""" """Request ID for new command matching in-flight command is illegal."""
results = list(sendframes(makereactor(), [ results = list(sendframes(makereactor(), [
ffs(b'1 1 stream-begin command-request new|more ' ffs(b'1 1 stream-begin command-request new|more '
b"cbor:{b'name': b'command'}"), b"cbor:{b'name': b'command'}"),
b'runcommand', b'runcommand',
b'runcommand', b'runcommand',
]) ])
self.assertEqual(results[4][1], { self.assertEqual(results[4][1], {
b'requestid': 3, b'requestid': 3,
b'command': b'command3', b'command': b'command3',
b'args': {b'biz': b'baz', b'key': b'val'}, b'args': {b'biz': b'baz', b'key': b'val'},
b'redirect': None,
b'data': None, b'data': None,
}) })
self.assertEqual(results[5][1], { self.assertEqual(results[5][1], {
b'requestid': 1, b'requestid': 1,
b'command': b'command1', b'command': b'command1',
b'args': {b'foo': b'bar', b'key1': b'val'}, b'args': {b'foo': b'bar', b'key1': b'val'},
b'redirect': None,
b'data': None, b'data': None,
}) })
def testmissingcommanddataframe(self): def testmissingcommanddataframe(self):
# The reactor doesn't currently handle partially received commands. # The reactor doesn't currently handle partially received commands.
# So this test is failing to do anything with request 1. # So this test is failing to do anything with request 1.
frames = [ frames = [
ffs(b'1 1 stream-begin command-request new|have-data ' ffs(b'1 1 stream-begin command-request new|have-data '

tests/wireprotosimplecache.py

# wireprotosimplecache.py - Extension providing in-memory wire protocol cache # wireprotosimplecache.py - Extension providing in-memory wire protocol cache
# #
# 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
from mercurial import ( from mercurial import (
extensions, extensions,
registrar, registrar,
repository, repository,
util, util,
wireprotoserver,
wireprototypes, wireprototypes,
wireprotov2server, wireprotov2server,
) )
from mercurial.utils import ( from mercurial.utils import (
interfaceutil, interfaceutil,
stringutil, stringutil,
) )
CACHE = None CACHE = None
configtable = {} configtable = {}
configitem = registrar.configitem(configtable) configitem = registrar.configitem(configtable)
configitem('simplecache', 'cacheapi',
default=False)
configitem('simplecache', 'cacheobjects', configitem('simplecache', 'cacheobjects',
default=False) default=False)
configitem('simplecache', 'redirectsfile', configitem('simplecache', 'redirectsfile',
default=None) default=None)
# API handler that makes cached keys available.
def handlecacherequest(rctx, req, res, checkperm, urlparts):
if rctx.repo.ui.configbool('simplecache', 'cacheobjects'):
res.status = b'500 Internal Server Error'
res.setbodybytes(b'cacheobjects not supported for api server')
return
if not urlparts:
res.status = b'200 OK'
res.headers[b'Content-Type'] = b'text/plain'
res.setbodybytes(b'simple cache server')
return
key = b'/'.join(urlparts)
if key not in CACHE:
res.status = b'404 Not Found'
res.headers[b'Content-Type'] = b'text/plain'
res.setbodybytes(b'key not found in cache')
return
res.status = b'200 OK'
res.headers[b'Content-Type'] = b'application/mercurial-cbor'
res.setbodybytes(CACHE[key])
def cachedescriptor(req, repo):
return {}
wireprotoserver.API_HANDLERS[b'simplecache'] = {
'config': (b'simplecache', b'cacheapi'),
'handler': handlecacherequest,
'apidescriptor': cachedescriptor,
}
@interfaceutil.implementer(repository.iwireprotocolcommandcacher) @interfaceutil.implementer(repository.iwireprotocolcommandcacher)
class memorycacher(object): class memorycacher(object):
def __init__(self, ui, command, encodefn): def __init__(self, ui, command, encodefn, redirecttargets, redirecthashes,
req):
self.ui = ui self.ui = ui
self.encodefn = encodefn self.encodefn = encodefn
self.redirecttargets = redirecttargets
self.redirecthashes = redirecthashes
self.req = req
self.key = None self.key = None
self.cacheobjects = ui.configbool('simplecache', 'cacheobjects') self.cacheobjects = ui.configbool('simplecache', 'cacheobjects')
self.cacheapi = ui.configbool('simplecache', 'cacheapi')
self.buffered = [] self.buffered = []
ui.log('simplecache', 'cacher constructed for %s\n', command) ui.log('simplecache', 'cacher constructed for %s\n', command)
def __enter__(self): def __enter__(self):
return self return self
def __exit__(self, exctype, excvalue, exctb): def __exit__(self, exctype, excvalue, exctb):
def lookup(self): def lookup(self):
if self.key not in CACHE: if self.key not in CACHE:
self.ui.log('simplecache', 'cache miss for %s\n', self.key) self.ui.log('simplecache', 'cache miss for %s\n', self.key)
return None return None
entry = CACHE[self.key] entry = CACHE[self.key]
self.ui.log('simplecache', 'cache hit for %s\n', self.key) self.ui.log('simplecache', 'cache hit for %s\n', self.key)
redirectable = True
if not self.cacheapi:
redirectable = False
elif not self.redirecttargets:
redirectable = False
else:
clienttargets = set(self.redirecttargets)
ourtargets = set(t[b'name'] for t in loadredirecttargets(self.ui))
# We only ever redirect to a single target (for now). So we don't
# need to store which target matched.
if not clienttargets & ourtargets:
redirectable = False
if redirectable:
paths = self.req.dispatchparts[:-3]
paths.append(b'simplecache')
paths.append(self.key)
url = b'%s/%s' % (self.req.advertisedbaseurl, b'/'.join(paths))
#url = b'http://example.com/%s' % self.key
self.ui.log('simplecache', 'sending content redirect for %s to '
'%s\n', self.key, url)
response = wireprototypes.alternatelocationresponse(
url=url,
mediatype=b'application/mercurial-cbor')
return {'objs': [response]}
if self.cacheobjects: if self.cacheobjects:
return { return {
'objs': entry, 'objs': entry,
} }
else: else:
return { return {
'objs': [wireprototypes.encodedresponse(entry)], 'objs': [wireprototypes.encodedresponse(entry)],
} }
self.ui.log('simplecache', 'storing cache entry for %s\n', self.key) self.ui.log('simplecache', 'storing cache entry for %s\n', self.key)
if self.cacheobjects: if self.cacheobjects:
CACHE[self.key] = self.buffered CACHE[self.key] = self.buffered
else: else:
CACHE[self.key] = b''.join(self.buffered) CACHE[self.key] = b''.join(self.buffered)
return [] return []
def makeresponsecacher(orig, repo, proto, command, args, objencoderfn): def makeresponsecacher(orig, repo, proto, command, args, objencoderfn,
return memorycacher(repo.ui, command, objencoderfn) redirecttargets, redirecthashes):
return memorycacher(repo.ui, command, objencoderfn, redirecttargets,
redirecthashes, proto._req)
def loadredirecttargets(ui): def loadredirecttargets(ui):
path = ui.config('simplecache', 'redirectsfile') path = ui.config('simplecache', 'redirectsfile')
if not path: if not path:
return [] return []
with open(path, 'rb') as fh: with open(path, 'rb') as fh:
s = fh.read() s = fh.read()