wireprotov1peer.py
670 lines
| 22.3 KiB
| text/x-python
|
PythonLexer
/ mercurial / wireprotov1peer.py
|
r37632 | # wireprotov1peer.py - Client-side functionality for wire protocol version 1. | ||
# | ||||
|
r47575 | # Copyright 2005-2010 Olivia Mackall <olivia@selenic.com> | ||
|
r37632 | # | ||
# This software may be used and distributed according to the terms of the | ||||
# GNU General Public License version 2 or any later version. | ||||
from __future__ import absolute_import | ||||
|
r37648 | import sys | ||
|
r37649 | import weakref | ||
|
r37632 | |||
from .i18n import _ | ||||
|
r43346 | from .node import bin | ||
|
r43359 | from .pycompat import ( | ||
getattr, | ||||
setattr, | ||||
) | ||||
|
r37632 | from . import ( | ||
bundle2, | ||||
changegroup as changegroupmod, | ||||
encoding, | ||||
error, | ||||
pushkey as pushkeymod, | ||||
pycompat, | ||||
util, | ||||
wireprototypes, | ||||
) | ||||
|
r43078 | from .interfaces import ( | ||
repository, | ||||
|
r43079 | util as interfaceutil, | ||
|
r37828 | ) | ||
|
r44517 | from .utils import hashutil | ||
|
r37632 | |||
urlreq = util.urlreq | ||||
|
r43346 | |||
|
r37633 | def batchable(f): | ||
|
r46554 | """annotation for batchable methods | ||
|
r37633 | |||
Such methods must implement a coroutine as follows: | ||||
@batchable | ||||
def sample(self, one, two=None): | ||||
# Build list of encoded arguments suitable for your wire protocol: | ||||
|
r47209 | encoded_args = [('one', encode(one),), ('two', encode(two),)] | ||
|
r37633 | # Create future for injection of encoded result: | ||
|
r47209 | encoded_res_future = future() | ||
|
r37633 | # Return encoded arguments and future: | ||
|
r47209 | yield encoded_args, encoded_res_future | ||
|
r37633 | # Assuming the future to be filled with the result from the batched | ||
# request now. Decode it: | ||||
|
r47209 | yield decode(encoded_res_future.value) | ||
|
r37633 | |||
The decorator returns a function which wraps this coroutine as a plain | ||||
method, but adds the original method as an attribute called "batchable", | ||||
which is used by remotebatch to split the call into separate encoding and | ||||
decoding phases. | ||||
|
r46554 | """ | ||
|
r43346 | |||
|
r37633 | def plain(*args, **opts): | ||
batchable = f(*args, **opts) | ||||
|
r47209 | encoded_args_or_res, encoded_res_future = next(batchable) | ||
if not encoded_res_future: | ||||
return encoded_args_or_res # a local result in this case | ||||
|
r37633 | self = args[0] | ||
cmd = pycompat.bytesurl(f.__name__) # ensure cmd is ascii bytestr | ||||
|
r47209 | encoded_res_future.set(self._submitone(cmd, encoded_args_or_res)) | ||
|
r37633 | return next(batchable) | ||
|
r43346 | |||
|
r37633 | setattr(plain, 'batchable', f) | ||
|
r39624 | setattr(plain, '__name__', f.__name__) | ||
|
r37633 | return plain | ||
|
r43346 | |||
|
r37633 | class future(object): | ||
'''placeholder for a value to be set later''' | ||||
|
r43346 | |||
|
r37633 | def set(self, value): | ||
|
r43347 | if util.safehasattr(self, b'value'): | ||
raise error.RepoError(b"future is already set") | ||||
|
r37633 | self.value = value | ||
|
r43346 | |||
|
r37632 | def encodebatchcmds(req): | ||
"""Return a ``cmds`` argument value for the ``batch`` command.""" | ||||
escapearg = wireprototypes.escapebatcharg | ||||
cmds = [] | ||||
for op, argsdict in req: | ||||
# Old servers didn't properly unescape argument names. So prevent | ||||
# the sending of argument names that may not be decoded properly by | ||||
# servers. | ||||
assert all(escapearg(k) == k for k in argsdict) | ||||
|
r43347 | args = b','.join( | ||
b'%s=%s' % (escapearg(k), escapearg(v)) | ||||
|
r43376 | for k, v in pycompat.iteritems(argsdict) | ||
|
r43346 | ) | ||
|
r43347 | cmds.append(b'%s %s' % (op, args)) | ||
|
r37632 | |||
|
r43347 | return b';'.join(cmds) | ||
|
r37632 | |||
|
r43346 | |||
|
r37649 | class unsentfuture(pycompat.futures.Future): | ||
"""A Future variation to represent an unsent command. | ||||
Because we buffer commands and don't submit them immediately, calling | ||||
``result()`` on an unsent future could deadlock. Futures for buffered | ||||
commands are represented by this type, which wraps ``result()`` to | ||||
call ``sendcommands()``. | ||||
""" | ||||
def result(self, timeout=None): | ||||
if self.done(): | ||||
return pycompat.futures.Future.result(self, timeout) | ||||
self._peerexecutor.sendcommands() | ||||
# This looks like it will infinitely recurse. However, | ||||
# sendcommands() should modify __class__. This call serves as a check | ||||
# on that. | ||||
return self.result(timeout) | ||||
|
r43346 | |||
|
r37828 | @interfaceutil.implementer(repository.ipeercommandexecutor) | ||
|
r37648 | class peerexecutor(object): | ||
def __init__(self, peer): | ||||
self._peer = peer | ||||
self._sent = False | ||||
self._closed = False | ||||
self._calls = [] | ||||
|
r37649 | self._futures = weakref.WeakSet() | ||
self._responseexecutor = None | ||||
self._responsef = None | ||||
|
r37648 | |||
def __enter__(self): | ||||
return self | ||||
def __exit__(self, exctype, excvalee, exctb): | ||||
self.close() | ||||
def callcommand(self, command, args): | ||||
if self._sent: | ||||
|
r43346 | raise error.ProgrammingError( | ||
|
r43387 | b'callcommand() cannot be used after commands are sent' | ||
|
r43346 | ) | ||
|
r37648 | |||
if self._closed: | ||||
|
r43346 | raise error.ProgrammingError( | ||
|
r43387 | b'callcommand() cannot be used after close()' | ||
|
r43346 | ) | ||
|
r37648 | |||
# Commands are dispatched through methods on the peer. | ||||
fn = getattr(self._peer, pycompat.sysstr(command), None) | ||||
if not fn: | ||||
raise error.ProgrammingError( | ||||
|
r43347 | b'cannot call command %s: method of same name not available ' | ||
b'on peer' % command | ||||
|
r43346 | ) | ||
|
r37648 | |||
# Commands are either batchable or they aren't. If a command | ||||
# isn't batchable, we send it immediately because the executor | ||||
# can no longer accept new commands after a non-batchable command. | ||||
|
r37649 | # If a command is batchable, we queue it for later. But we have | ||
# to account for the case of a non-batchable command arriving after | ||||
# a batchable one and refuse to service it. | ||||
def addcall(): | ||||
f = pycompat.futures.Future() | ||||
self._futures.add(f) | ||||
self._calls.append((command, args, fn, f)) | ||||
return f | ||||
|
r37648 | |||
if getattr(fn, 'batchable', False): | ||||
|
r37649 | f = addcall() | ||
# But since we don't issue it immediately, we wrap its result() | ||||
# to trigger sending so we avoid deadlocks. | ||||
f.__class__ = unsentfuture | ||||
f._peerexecutor = self | ||||
|
r37648 | else: | ||
if self._calls: | ||||
raise error.ProgrammingError( | ||||
|
r43347 | b'%s is not batchable and cannot be called on a command ' | ||
b'executor along with other commands' % command | ||||
|
r43346 | ) | ||
|
r37648 | |||
|
r37649 | f = addcall() | ||
# Non-batchable commands can never coexist with another command | ||||
# in this executor. So send the command immediately. | ||||
self.sendcommands() | ||||
|
r37648 | return f | ||
def sendcommands(self): | ||||
if self._sent: | ||||
return | ||||
if not self._calls: | ||||
return | ||||
self._sent = True | ||||
|
r37649 | # Unhack any future types so caller seens a clean type and to break | ||
# cycle between us and futures. | ||||
for f in self._futures: | ||||
if isinstance(f, unsentfuture): | ||||
f.__class__ = pycompat.futures.Future | ||||
f._peerexecutor = None | ||||
|
r37648 | calls = self._calls | ||
# Mainly to destroy references to futures. | ||||
self._calls = None | ||||
|
r37649 | # Simple case of a single command. We call it synchronously. | ||
|
r37648 | if len(calls) == 1: | ||
command, args, fn, f = calls[0] | ||||
# Future was cancelled. Ignore it. | ||||
if not f.set_running_or_notify_cancel(): | ||||
return | ||||
try: | ||||
result = fn(**pycompat.strkwargs(args)) | ||||
except Exception: | ||||
|
r37687 | pycompat.future_set_exception_info(f, sys.exc_info()[1:]) | ||
|
r37648 | else: | ||
f.set_result(result) | ||||
return | ||||
|
r37649 | # Batch commands are a bit harder. First, we have to deal with the | ||
# @batchable coroutine. That's a bit annoying. Furthermore, we also | ||||
# need to preserve streaming. i.e. it should be possible for the | ||||
# futures to resolve as data is coming in off the wire without having | ||||
# to wait for the final byte of the final response. We do this by | ||||
# spinning up a thread to read the responses. | ||||
requests = [] | ||||
states = [] | ||||
for command, args, fn, f in calls: | ||||
# Future was cancelled. Ignore it. | ||||
if not f.set_running_or_notify_cancel(): | ||||
continue | ||||
try: | ||||
|
r43346 | batchable = fn.batchable( | ||
fn.__self__, **pycompat.strkwargs(args) | ||||
) | ||||
|
r37649 | except Exception: | ||
|
r37687 | pycompat.future_set_exception_info(f, sys.exc_info()[1:]) | ||
|
r37649 | return | ||
# Encoded arguments and future holding remote result. | ||||
try: | ||||
|
r47209 | encoded_args_or_res, fremote = next(batchable) | ||
|
r37649 | except Exception: | ||
|
r37687 | pycompat.future_set_exception_info(f, sys.exc_info()[1:]) | ||
|
r37649 | return | ||
|
r41085 | if not fremote: | ||
|
r47209 | f.set_result(encoded_args_or_res) | ||
|
r41085 | else: | ||
|
r47209 | requests.append((command, encoded_args_or_res)) | ||
|
r41085 | states.append((command, f, batchable, fremote)) | ||
|
r37649 | |||
if not requests: | ||||
return | ||||
# This will emit responses in order they were executed. | ||||
wireresults = self._peer._submitbatch(requests) | ||||
# The use of a thread pool executor here is a bit weird for something | ||||
# that only spins up a single thread. However, thread management is | ||||
# hard and it is easy to encounter race conditions, deadlocks, etc. | ||||
# concurrent.futures already solves these problems and its thread pool | ||||
# executor has minimal overhead. So we use it. | ||||
self._responseexecutor = pycompat.futures.ThreadPoolExecutor(1) | ||||
|
r43346 | self._responsef = self._responseexecutor.submit( | ||
self._readbatchresponse, states, wireresults | ||||
) | ||||
|
r37648 | |||
def close(self): | ||||
self.sendcommands() | ||||
|
r37649 | if self._closed: | ||
return | ||||
|
r37648 | self._closed = True | ||
|
r37649 | if not self._responsef: | ||
return | ||||
# We need to wait on our in-flight response and then shut down the | ||||
# executor once we have a result. | ||||
try: | ||||
self._responsef.result() | ||||
finally: | ||||
self._responseexecutor.shutdown(wait=True) | ||||
self._responsef = None | ||||
self._responseexecutor = None | ||||
# If any of our futures are still in progress, mark them as | ||||
# errored. Otherwise a result() could wait indefinitely. | ||||
for f in self._futures: | ||||
if not f.done(): | ||||
|
r43346 | f.set_exception( | ||
error.ResponseError( | ||||
|
r47430 | _(b'unfulfilled batch command response'), None | ||
|
r43346 | ) | ||
) | ||||
|
r37649 | |||
self._futures = None | ||||
def _readbatchresponse(self, states, wireresults): | ||||
# Executes in a thread to read data off the wire. | ||||
for command, f, batchable, fremote in states: | ||||
# Grab raw result off the wire and teach the internal future | ||||
# about it. | ||||
|
r47430 | try: | ||
remoteresult = next(wireresults) | ||||
except StopIteration: | ||||
# This can happen in particular because next(batchable) | ||||
# in the previous iteration can call peer._abort, which | ||||
# may close the peer. | ||||
f.set_exception( | ||||
error.ResponseError( | ||||
_(b'unfulfilled batch command response'), None | ||||
) | ||||
) | ||||
else: | ||||
fremote.set(remoteresult) | ||||
|
r37649 | |||
|
r47430 | # And ask the coroutine to decode that value. | ||
try: | ||||
result = next(batchable) | ||||
except Exception: | ||||
pycompat.future_set_exception_info(f, sys.exc_info()[1:]) | ||||
else: | ||||
f.set_result(result) | ||||
|
r37649 | |||
|
r43346 | |||
@interfaceutil.implementer( | ||||
repository.ipeercommands, repository.ipeerlegacycommands | ||||
) | ||||
|
r37653 | class wirepeer(repository.peer): | ||
|
r37632 | """Client-side interface for communicating with a peer repository. | ||
Methods commonly call wire protocol commands of the same name. | ||||
See also httppeer.py and sshpeer.py for protocol-specific | ||||
implementations of this interface. | ||||
""" | ||||
|
r43346 | |||
|
r37648 | def commandexecutor(self): | ||
return peerexecutor(self) | ||||
|
r37632 | # Begin of ipeercommands interface. | ||
|
r37667 | def clonebundles(self): | ||
|
r43347 | self.requirecap(b'clonebundles', _(b'clone bundles')) | ||
return self._call(b'clonebundles') | ||||
|
r37667 | |||
|
r37632 | @batchable | ||
def lookup(self, key): | ||||
|
r43347 | self.requirecap(b'lookup', _(b'look up remote revision')) | ||
|
r37632 | f = future() | ||
|
r43347 | yield {b'key': encoding.fromlocal(key)}, f | ||
|
r37632 | d = f.value | ||
|
r43347 | success, data = d[:-1].split(b" ", 1) | ||
|
r37632 | if int(success): | ||
yield bin(data) | ||||
else: | ||||
self._abort(error.RepoError(data)) | ||||
@batchable | ||||
def heads(self): | ||||
f = future() | ||||
yield {}, f | ||||
d = f.value | ||||
try: | ||||
yield wireprototypes.decodelist(d[:-1]) | ||||
except ValueError: | ||||
|
r43347 | self._abort(error.ResponseError(_(b"unexpected response:"), d)) | ||
|
r37632 | |||
@batchable | ||||
def known(self, nodes): | ||||
f = future() | ||||
|
r43347 | yield {b'nodes': wireprototypes.encodelist(nodes)}, f | ||
|
r37632 | d = f.value | ||
try: | ||||
|
r38350 | yield [bool(int(b)) for b in pycompat.iterbytestr(d)] | ||
|
r37632 | except ValueError: | ||
|
r43347 | self._abort(error.ResponseError(_(b"unexpected response:"), d)) | ||
|
r37632 | |||
@batchable | ||||
def branchmap(self): | ||||
f = future() | ||||
yield {}, f | ||||
d = f.value | ||||
try: | ||||
branchmap = {} | ||||
for branchpart in d.splitlines(): | ||||
|
r43347 | branchname, branchheads = branchpart.split(b' ', 1) | ||
|
r37632 | branchname = encoding.tolocal(urlreq.unquote(branchname)) | ||
branchheads = wireprototypes.decodelist(branchheads) | ||||
branchmap[branchname] = branchheads | ||||
yield branchmap | ||||
except TypeError: | ||||
|
r43347 | self._abort(error.ResponseError(_(b"unexpected response:"), d)) | ||
|
r37632 | |||
@batchable | ||||
def listkeys(self, namespace): | ||||
|
r43347 | if not self.capable(b'pushkey'): | ||
|
r37632 | yield {}, None | ||
f = future() | ||||
|
r43347 | self.ui.debug(b'preparing listkeys for "%s"\n' % namespace) | ||
yield {b'namespace': encoding.fromlocal(namespace)}, f | ||||
|
r37632 | d = f.value | ||
|
r43346 | self.ui.debug( | ||
|
r43347 | b'received listkey for "%s": %i bytes\n' % (namespace, len(d)) | ||
|
r43346 | ) | ||
|
r37632 | yield pushkeymod.decodekeys(d) | ||
@batchable | ||||
def pushkey(self, namespace, key, old, new): | ||||
|
r43347 | if not self.capable(b'pushkey'): | ||
|
r37632 | yield False, None | ||
f = future() | ||||
|
r43347 | self.ui.debug(b'preparing pushkey for "%s:%s"\n' % (namespace, key)) | ||
|
r43346 | yield { | ||
|
r43347 | b'namespace': encoding.fromlocal(namespace), | ||
b'key': encoding.fromlocal(key), | ||||
b'old': encoding.fromlocal(old), | ||||
b'new': encoding.fromlocal(new), | ||||
|
r43346 | }, f | ||
|
r37632 | d = f.value | ||
|
r43347 | d, output = d.split(b'\n', 1) | ||
|
r37632 | try: | ||
d = bool(int(d)) | ||||
except ValueError: | ||||
raise error.ResponseError( | ||||
|
r43347 | _(b'push failed (unexpected response):'), d | ||
|
r43346 | ) | ||
|
r37632 | for l in output.splitlines(True): | ||
|
r43347 | self.ui.status(_(b'remote: '), l) | ||
|
r37632 | yield d | ||
def stream_out(self): | ||||
|
r43347 | return self._callstream(b'stream_out') | ||
|
r37632 | |||
def getbundle(self, source, **kwargs): | ||||
kwargs = pycompat.byteskwargs(kwargs) | ||||
|
r43347 | self.requirecap(b'getbundle', _(b'look up remote changes')) | ||
|
r37632 | opts = {} | ||
|
r43347 | bundlecaps = kwargs.get(b'bundlecaps') or set() | ||
|
r43376 | for key, value in pycompat.iteritems(kwargs): | ||
|
r37632 | if value is None: | ||
continue | ||||
keytype = wireprototypes.GETBUNDLE_ARGUMENTS.get(key) | ||||
if keytype is None: | ||||
raise error.ProgrammingError( | ||||
|
r43347 | b'Unexpectedly None keytype for key %s' % key | ||
|
r43346 | ) | ||
|
r43347 | elif keytype == b'nodes': | ||
|
r37632 | value = wireprototypes.encodelist(value) | ||
|
r43347 | elif keytype == b'csv': | ||
value = b','.join(value) | ||||
elif keytype == b'scsv': | ||||
value = b','.join(sorted(value)) | ||||
elif keytype == b'boolean': | ||||
value = b'%i' % bool(value) | ||||
elif keytype != b'plain': | ||||
raise KeyError(b'unknown getbundle option type %s' % keytype) | ||||
|
r37632 | opts[key] = value | ||
|
r43347 | f = self._callcompressable(b"getbundle", **pycompat.strkwargs(opts)) | ||
if any((cap.startswith(b'HG2') for cap in bundlecaps)): | ||||
|
r37632 | return bundle2.getunbundler(self.ui, f) | ||
else: | ||||
|
r43347 | return changegroupmod.cg1unpacker(f, b'UN') | ||
|
r37632 | |||
|
r37664 | def unbundle(self, bundle, heads, url): | ||
|
r46554 | """Send cg (a readable file-like object representing the | ||
|
r37632 | changegroup to push, typically a chunkbuffer object) to the | ||
remote server as a bundle. | ||||
When pushing a bundle10 stream, return an integer indicating the | ||||
result of the push (see changegroup.apply()). | ||||
When pushing a bundle20 stream, return a bundle20 stream. | ||||
`url` is the url the client thinks it's pushing to, which is | ||||
visible to hooks. | ||||
|
r46554 | """ | ||
|
r37632 | |||
|
r43347 | if heads != [b'force'] and self.capable(b'unbundlehash'): | ||
|
r37632 | heads = wireprototypes.encodelist( | ||
|
r44517 | [b'hashed', hashutil.sha1(b''.join(sorted(heads))).digest()] | ||
|
r43346 | ) | ||
|
r37632 | else: | ||
heads = wireprototypes.encodelist(heads) | ||||
|
r43347 | if util.safehasattr(bundle, b'deltaheader'): | ||
|
r37632 | # this a bundle10, do the old style call sequence | ||
|
r43347 | ret, output = self._callpush(b"unbundle", bundle, heads=heads) | ||
if ret == b"": | ||||
raise error.ResponseError(_(b'push failed:'), output) | ||||
|
r37632 | try: | ||
ret = int(ret) | ||||
except ValueError: | ||||
raise error.ResponseError( | ||||
|
r43347 | _(b'push failed (unexpected response):'), ret | ||
|
r43346 | ) | ||
|
r37632 | |||
for l in output.splitlines(True): | ||||
|
r43347 | self.ui.status(_(b'remote: '), l) | ||
|
r37632 | else: | ||
# bundle2 push. Send a stream, fetch a stream. | ||||
|
r43347 | stream = self._calltwowaystream(b'unbundle', bundle, heads=heads) | ||
|
r37632 | ret = bundle2.getunbundler(self.ui, stream) | ||
return ret | ||||
# End of ipeercommands interface. | ||||
# Begin of ipeerlegacycommands interface. | ||||
def branches(self, nodes): | ||||
n = wireprototypes.encodelist(nodes) | ||||
|
r43347 | d = self._call(b"branches", nodes=n) | ||
|
r37632 | try: | ||
br = [tuple(wireprototypes.decodelist(b)) for b in d.splitlines()] | ||||
return br | ||||
except ValueError: | ||||
|
r43347 | self._abort(error.ResponseError(_(b"unexpected response:"), d)) | ||
|
r37632 | |||
def between(self, pairs): | ||||
|
r43346 | batch = 8 # avoid giant requests | ||
|
r37632 | r = [] | ||
|
r38806 | for i in pycompat.xrange(0, len(pairs), batch): | ||
|
r43347 | n = b" ".join( | ||
|
r43346 | [ | ||
|
r43347 | wireprototypes.encodelist(p, b'-') | ||
|
r43346 | for p in pairs[i : i + batch] | ||
] | ||||
) | ||||
|
r43347 | d = self._call(b"between", pairs=n) | ||
|
r37632 | try: | ||
|
r43346 | r.extend( | ||
l and wireprototypes.decodelist(l) or [] | ||||
for l in d.splitlines() | ||||
) | ||||
|
r37632 | except ValueError: | ||
|
r43347 | self._abort(error.ResponseError(_(b"unexpected response:"), d)) | ||
|
r37632 | return r | ||
|
r37653 | def changegroup(self, nodes, source): | ||
|
r37632 | n = wireprototypes.encodelist(nodes) | ||
|
r43347 | f = self._callcompressable(b"changegroup", roots=n) | ||
return changegroupmod.cg1unpacker(f, b'UN') | ||||
|
r37632 | |||
|
r37653 | def changegroupsubset(self, bases, heads, source): | ||
|
r43347 | self.requirecap(b'changegroupsubset', _(b'look up remote changes')) | ||
|
r37632 | bases = wireprototypes.encodelist(bases) | ||
heads = wireprototypes.encodelist(heads) | ||||
|
r43346 | f = self._callcompressable( | ||
|
r43347 | b"changegroupsubset", bases=bases, heads=heads | ||
|
r43346 | ) | ||
|
r43347 | return changegroupmod.cg1unpacker(f, b'UN') | ||
|
r37632 | |||
# End of ipeerlegacycommands interface. | ||||
def _submitbatch(self, req): | ||||
"""run batch request <req> on the server | ||||
Returns an iterator of the raw responses from the server. | ||||
""" | ||||
ui = self.ui | ||||
|
r43347 | if ui.debugflag and ui.configbool(b'devel', b'debug.peer-request'): | ||
ui.debug(b'devel-peer-request: batched-content\n') | ||||
|
r37632 | for op, args in req: | ||
|
r43347 | msg = b'devel-peer-request: - %s (%d arguments)\n' | ||
|
r37632 | ui.debug(msg % (op, len(args))) | ||
unescapearg = wireprototypes.unescapebatcharg | ||||
|
r43347 | rsp = self._callstream(b"batch", cmds=encodebatchcmds(req)) | ||
|
r37632 | chunk = rsp.read(1024) | ||
work = [chunk] | ||||
while chunk: | ||||
|
r43347 | while b';' not in chunk and chunk: | ||
|
r37632 | chunk = rsp.read(1024) | ||
work.append(chunk) | ||||
|
r43347 | merged = b''.join(work) | ||
while b';' in merged: | ||||
one, merged = merged.split(b';', 1) | ||||
|
r37632 | yield unescapearg(one) | ||
chunk = rsp.read(1024) | ||||
work = [merged, chunk] | ||||
|
r43347 | yield unescapearg(b''.join(work)) | ||
|
r37632 | |||
def _submitone(self, op, args): | ||||
return self._call(op, **pycompat.strkwargs(args)) | ||||
def debugwireargs(self, one, two, three=None, four=None, five=None): | ||||
# don't pass optional arguments left at their default value | ||||
opts = {} | ||||
if three is not None: | ||||
|
r43906 | opts['three'] = three | ||
|
r37632 | if four is not None: | ||
|
r43906 | opts['four'] = four | ||
|
r43347 | return self._call(b'debugwireargs', one=one, two=two, **opts) | ||
|
r37632 | |||
def _call(self, cmd, **args): | ||||
"""execute <cmd> on the server | ||||
The command is expected to return a simple string. | ||||
returns the server reply as a string.""" | ||||
raise NotImplementedError() | ||||
def _callstream(self, cmd, **args): | ||||
"""execute <cmd> on the server | ||||
The command is expected to return a stream. Note that if the | ||||
command doesn't return a stream, _callstream behaves | ||||
differently for ssh and http peers. | ||||
returns the server reply as a file like object. | ||||
""" | ||||
raise NotImplementedError() | ||||
def _callcompressable(self, cmd, **args): | ||||
"""execute <cmd> on the server | ||||
The command is expected to return a stream. | ||||
The stream may have been compressed in some implementations. This | ||||
function takes care of the decompression. This is the only difference | ||||
with _callstream. | ||||
returns the server reply as a file like object. | ||||
""" | ||||
raise NotImplementedError() | ||||
def _callpush(self, cmd, fp, **args): | ||||
"""execute a <cmd> on server | ||||
The command is expected to be related to a push. Push has a special | ||||
return method. | ||||
returns the server reply as a (ret, output) tuple. ret is either | ||||
empty (error) or a stringified int. | ||||
""" | ||||
raise NotImplementedError() | ||||
def _calltwowaystream(self, cmd, fp, **args): | ||||
"""execute <cmd> on server | ||||
The command will send a stream to the server and get a stream in reply. | ||||
""" | ||||
raise NotImplementedError() | ||||
def _abort(self, exception): | ||||
|
r46554 | """clearly abort the wire protocol connection and raise the exception""" | ||
|
r37632 | raise NotImplementedError() | ||