compression: use sysstr to specify attribute to fetch for priority

These are attributes so they should be `str`.

mercurial/utils/compression.py

@@ -1,810 +1,810 b''
 # compression.py - Mercurial utility functions for compression
 #
 # This software may be used and distributed according to the terms of the
 # GNU General Public License version 2 or any later version.
 
 
 import bz2
 import collections
 import zlib
 
 from ..pycompat import getattr
 from .. import (
     error,
     i18n,
     pycompat,
 )
 from . import stringutil
 
 safehasattr = pycompat.safehasattr
 
 
 _ = i18n._
 
 # compression code
 
 SERVERROLE = b'server'
 CLIENTROLE = b'client'
 
 compewireprotosupport = collections.namedtuple(
     'compenginewireprotosupport',
     ('name', 'serverpriority', 'clientpriority'),
 )
 
 
 class propertycache:
     def __init__(self, func):
         self.func = func
         self.name = func.__name__
 
     def __get__(self, obj, type=None):
         result = self.func(obj)
         self.cachevalue(obj, result)
         return result
 
     def cachevalue(self, obj, value):
         # __dict__ assignment required to bypass __setattr__ (eg: repoview)
         obj.__dict__[self.name] = value
 
 
 class compressormanager:
     """Holds registrations of various compression engines.
 
     This class essentially abstracts the differences between compression
     engines to allow new compression formats to be added easily, possibly from
     extensions.
 
     Compressors are registered against the global instance by calling its
     ``register()`` method.
     """
 
     def __init__(self):
         self._engines = {}
         # Bundle spec human name to engine name.
         self._bundlenames = {}
         # Internal bundle identifier to engine name.
         self._bundletypes = {}
         # Revlog header to engine name.
         self._revlogheaders = {}
         # Wire proto identifier to engine name.
         self._wiretypes = {}
 
     def __getitem__(self, key):
         return self._engines[key]
 
     def __contains__(self, key):
         return key in self._engines
 
     def __iter__(self):
         return iter(self._engines.keys())
 
     def register(self, engine):
         """Register a compression engine with the manager.
 
         The argument must be a ``compressionengine`` instance.
         """
         if not isinstance(engine, compressionengine):
             raise ValueError(_(b'argument must be a compressionengine'))
 
         name = engine.name()
 
         if name in self._engines:
             raise error.Abort(
                 _(b'compression engine %s already registered') % name
             )
 
         bundleinfo = engine.bundletype()
         if bundleinfo:
             bundlename, bundletype = bundleinfo
 
             if bundlename in self._bundlenames:
                 raise error.Abort(
                     _(b'bundle name %s already registered') % bundlename
                 )
             if bundletype in self._bundletypes:
                 raise error.Abort(
                     _(b'bundle type %s already registered by %s')
                     % (bundletype, self._bundletypes[bundletype])
                 )
 
             # No external facing name declared.
             if bundlename:
                 self._bundlenames[bundlename] = name
 
             self._bundletypes[bundletype] = name
 
         wiresupport = engine.wireprotosupport()
         if wiresupport:
             wiretype = wiresupport.name
             if wiretype in self._wiretypes:
                 raise error.Abort(
                     _(
                         b'wire protocol compression %s already '
                         b'registered by %s'
                     )
                     % (wiretype, self._wiretypes[wiretype])
                 )
 
             self._wiretypes[wiretype] = name
 
         revlogheader = engine.revlogheader()
         if revlogheader and revlogheader in self._revlogheaders:
             raise error.Abort(
                 _(b'revlog header %s already registered by %s')
                 % (revlogheader, self._revlogheaders[revlogheader])
             )
 
         if revlogheader:
             self._revlogheaders[revlogheader] = name
 
         self._engines[name] = engine
 
     @property
     def supportedbundlenames(self):
         return set(self._bundlenames.keys())
 
     @property
     def supportedbundletypes(self):
         return set(self._bundletypes.keys())
 
     def forbundlename(self, bundlename):
         """Obtain a compression engine registered to a bundle name.
 
         Will raise KeyError if the bundle type isn't registered.
 
         Will abort if the engine is known but not available.
         """
         engine = self._engines[self._bundlenames[bundlename]]
         if not engine.available():
             raise error.Abort(
                 _(b'compression engine %s could not be loaded') % engine.name()
             )
         return engine
 
     def forbundletype(self, bundletype):
         """Obtain a compression engine registered to a bundle type.
 
         Will raise KeyError if the bundle type isn't registered.
 
         Will abort if the engine is known but not available.
         """
         engine = self._engines[self._bundletypes[bundletype]]
         if not engine.available():
             raise error.Abort(
                 _(b'compression engine %s could not be loaded') % engine.name()
             )
         return engine
 
     def supportedwireengines(self, role, onlyavailable=True):
         """Obtain compression engines that support the wire protocol.
 
         Returns a list of engines in prioritized order, most desired first.
 
         If ``onlyavailable`` is set, filter out engines that can't be
         loaded.
         """
         assert role in (SERVERROLE, CLIENTROLE)
 
-        attr = b'serverpriority' if role == SERVERROLE else b'clientpriority'
+        attr = 'serverpriority' if role == SERVERROLE else 'clientpriority'
 
         engines = [self._engines[e] for e in self._wiretypes.values()]
         if onlyavailable:
             engines = [e for e in engines if e.available()]
 
         def getkey(e):
             # Sort first by priority, highest first. In case of tie, sort
             # alphabetically. This is arbitrary, but ensures output is
             # stable.
             w = e.wireprotosupport()
             return -1 * getattr(w, attr), w.name
 
         return list(sorted(engines, key=getkey))
 
     def forwiretype(self, wiretype):
         engine = self._engines[self._wiretypes[wiretype]]
         if not engine.available():
             raise error.Abort(
                 _(b'compression engine %s could not be loaded') % engine.name()
             )
         return engine
 
     def forrevlogheader(self, header):
         """Obtain a compression engine registered to a revlog header.
 
         Will raise KeyError if the revlog header value isn't registered.
         """
         return self._engines[self._revlogheaders[header]]
 
 
 compengines = compressormanager()
 
 
 class compressionengine:
     """Base class for compression engines.
 
     Compression engines must implement the interface defined by this class.
     """
 
     def name(self):
         """Returns the name of the compression engine.
 
         This is the key the engine is registered under.
 
         This method must be implemented.
         """
         raise NotImplementedError()
 
     def available(self):
         """Whether the compression engine is available.
 
         The intent of this method is to allow optional compression engines
         that may not be available in all installations (such as engines relying
         on C extensions that may not be present).
         """
         return True
 
     def bundletype(self):
         """Describes bundle identifiers for this engine.
 
         If this compression engine isn't supported for bundles, returns None.
 
         If this engine can be used for bundles, returns a 2-tuple of strings of
         the user-facing "bundle spec" compression name and an internal
         identifier used to denote the compression format within bundles. To
         exclude the name from external usage, set the first element to ``None``.
 
         If bundle compression is supported, the class must also implement
         ``compressstream`` and `decompressorreader``.
 
         The docstring of this method is used in the help system to tell users
         about this engine.
         """
         return None
 
     def wireprotosupport(self):
         """Declare support for this compression format on the wire protocol.
 
         If this compression engine isn't supported for compressing wire
         protocol payloads, returns None.
 
         Otherwise, returns ``compenginewireprotosupport`` with the following
         fields:
 
         * String format identifier
         * Integer priority for the server
         * Integer priority for the client
 
         The integer priorities are used to order the advertisement of format
         support by server and client. The highest integer is advertised
         first. Integers with non-positive values aren't advertised.
 
         The priority values are somewhat arbitrary and only used for default
         ordering. The relative order can be changed via config options.
 
         If wire protocol compression is supported, the class must also implement
         ``compressstream`` and ``decompressorreader``.
         """
         return None
 
     def revlogheader(self):
         """Header added to revlog chunks that identifies this engine.
 
         If this engine can be used to compress revlogs, this method should
         return the bytes used to identify chunks compressed with this engine.
         Else, the method should return ``None`` to indicate it does not
         participate in revlog compression.
         """
         return None
 
     def compressstream(self, it, opts=None):
         """Compress an iterator of chunks.
 
         The method receives an iterator (ideally a generator) of chunks of
         bytes to be compressed. It returns an iterator (ideally a generator)
         of bytes of chunks representing the compressed output.
 
         Optionally accepts an argument defining how to perform compression.
         Each engine treats this argument differently.
         """
         raise NotImplementedError()
 
     def decompressorreader(self, fh):
         """Perform decompression on a file object.
 
         Argument is an object with a ``read(size)`` method that returns
         compressed data. Return value is an object with a ``read(size)`` that
         returns uncompressed data.
         """
         raise NotImplementedError()
 
     def revlogcompressor(self, opts=None):
         """Obtain an object that can be used to compress revlog entries.
 
         The object has a ``compress(data)`` method that compresses binary
         data. This method returns compressed binary data or ``None`` if
         the data could not be compressed (too small, not compressible, etc).
         The returned data should have a header uniquely identifying this
         compression format so decompression can be routed to this engine.
         This header should be identified by the ``revlogheader()`` return
         value.
 
         The object has a ``decompress(data)`` method that decompresses
         data. The method will only be called if ``data`` begins with
         ``revlogheader()``. The method should return the raw, uncompressed
         data or raise a ``StorageError``.
 
         The object is reusable but is not thread safe.
         """
         raise NotImplementedError()
 
 
 class _CompressedStreamReader:
     def __init__(self, fh):
         if safehasattr(fh, 'unbufferedread'):
             self._reader = fh.unbufferedread
         else:
             self._reader = fh.read
         self._pending = []
         self._pos = 0
         self._eof = False
 
     def _decompress(self, chunk):
         raise NotImplementedError()
 
     def read(self, l):
         buf = []
         while True:
             while self._pending:
                 if len(self._pending[0]) > l + self._pos:
                     newbuf = self._pending[0]
                     buf.append(newbuf[self._pos : self._pos + l])
                     self._pos += l
                     return b''.join(buf)
 
                 newbuf = self._pending.pop(0)
                 if self._pos:
                     buf.append(newbuf[self._pos :])
                     l -= len(newbuf) - self._pos
                 else:
                     buf.append(newbuf)
                     l -= len(newbuf)
                 self._pos = 0
 
             if self._eof:
                 return b''.join(buf)
             chunk = self._reader(65536)
             self._decompress(chunk)
             if not chunk and not self._pending and not self._eof:
                 # No progress and no new data, bail out
                 return b''.join(buf)
 
 
 class _GzipCompressedStreamReader(_CompressedStreamReader):
     def __init__(self, fh):
         super(_GzipCompressedStreamReader, self).__init__(fh)
         self._decompobj = zlib.decompressobj()
 
     def _decompress(self, chunk):
         newbuf = self._decompobj.decompress(chunk)
         if newbuf:
             self._pending.append(newbuf)
         d = self._decompobj.copy()
         try:
             d.decompress(b'x')
             d.flush()
             if d.unused_data == b'x':
                 self._eof = True
         except zlib.error:
             pass
 
 
 class _BZ2CompressedStreamReader(_CompressedStreamReader):
     def __init__(self, fh):
         super(_BZ2CompressedStreamReader, self).__init__(fh)
         self._decompobj = bz2.BZ2Decompressor()
 
     def _decompress(self, chunk):
         newbuf = self._decompobj.decompress(chunk)
         if newbuf:
             self._pending.append(newbuf)
         try:
             while True:
                 newbuf = self._decompobj.decompress(b'')
                 if newbuf:
                     self._pending.append(newbuf)
                 else:
                     break
         except EOFError:
             self._eof = True
 
 
 class _TruncatedBZ2CompressedStreamReader(_BZ2CompressedStreamReader):
     def __init__(self, fh):
         super(_TruncatedBZ2CompressedStreamReader, self).__init__(fh)
         newbuf = self._decompobj.decompress(b'BZ')
         if newbuf:
             self._pending.append(newbuf)
 
 
 class _ZstdCompressedStreamReader(_CompressedStreamReader):
     def __init__(self, fh, zstd):
         super(_ZstdCompressedStreamReader, self).__init__(fh)
         self._zstd = zstd
         self._decompobj = zstd.ZstdDecompressor().decompressobj()
 
     def _decompress(self, chunk):
         newbuf = self._decompobj.decompress(chunk)
         if newbuf:
             self._pending.append(newbuf)
         try:
             while True:
                 newbuf = self._decompobj.decompress(b'')
                 if newbuf:
                     self._pending.append(newbuf)
                 else:
                     break
         except self._zstd.ZstdError:
             self._eof = True
 
 
 class _zlibengine(compressionengine):
     def name(self):
         return b'zlib'
 
     def bundletype(self):
         """zlib compression using the DEFLATE algorithm.
 
         All Mercurial clients should support this format. The compression
         algorithm strikes a reasonable balance between compression ratio
         and size.
         """
         return b'gzip', b'GZ'
 
     def wireprotosupport(self):
         return compewireprotosupport(b'zlib', 20, 20)
 
     def revlogheader(self):
         return b'x'
 
     def compressstream(self, it, opts=None):
         opts = opts or {}
 
         z = zlib.compressobj(opts.get(b'level', -1))
         for chunk in it:
             data = z.compress(chunk)
             # Not all calls to compress emit data. It is cheaper to inspect
             # here than to feed empty chunks through generator.
             if data:
                 yield data
 
         yield z.flush()
 
     def decompressorreader(self, fh):
         return _GzipCompressedStreamReader(fh)
 
     class zlibrevlogcompressor:
         def __init__(self, level=None):
             self._level = level
 
         def compress(self, data):
             insize = len(data)
             # Caller handles empty input case.
             assert insize > 0
 
             if insize < 44:
                 return None
 
             elif insize <= 1000000:
                 if self._level is None:
                     compressed = zlib.compress(data)
                 else:
                     compressed = zlib.compress(data, self._level)
                 if len(compressed) < insize:
                     return compressed
                 return None
 
             # zlib makes an internal copy of the input buffer, doubling
             # memory usage for large inputs. So do streaming compression
             # on large inputs.
             else:
                 if self._level is None:
                     z = zlib.compressobj()
                 else:
                     z = zlib.compressobj(level=self._level)
                 parts = []
                 pos = 0
                 while pos < insize:
                     pos2 = pos + 2 ** 20
                     parts.append(z.compress(data[pos:pos2]))
                     pos = pos2
                 parts.append(z.flush())
 
                 if sum(map(len, parts)) < insize:
                     return b''.join(parts)
                 return None
 
         def decompress(self, data):
             try:
                 return zlib.decompress(data)
             except zlib.error as e:
                 raise error.StorageError(
                     _(b'revlog decompress error: %s')
                     % stringutil.forcebytestr(e)
                 )
 
     def revlogcompressor(self, opts=None):
         level = None
         if opts is not None:
             level = opts.get(b'zlib.level')
         return self.zlibrevlogcompressor(level)
 
 
 compengines.register(_zlibengine())
 
 
 class _bz2engine(compressionengine):
     def name(self):
         return b'bz2'
 
     def bundletype(self):
         """An algorithm that produces smaller bundles than ``gzip``.
 
         All Mercurial clients should support this format.
 
         This engine will likely produce smaller bundles than ``gzip`` but
         will be significantly slower, both during compression and
         decompression.
 
         If available, the ``zstd`` engine can yield similar or better
         compression at much higher speeds.
         """
         return b'bzip2', b'BZ'
 
     # We declare a protocol name but don't advertise by default because
     # it is slow.
     def wireprotosupport(self):
         return compewireprotosupport(b'bzip2', 0, 0)
 
     def compressstream(self, it, opts=None):
         opts = opts or {}
         z = bz2.BZ2Compressor(opts.get(b'level', 9))
         for chunk in it:
             data = z.compress(chunk)
             if data:
                 yield data
 
         yield z.flush()
 
     def decompressorreader(self, fh):
         return _BZ2CompressedStreamReader(fh)
 
 
 compengines.register(_bz2engine())
 
 
 class _truncatedbz2engine(compressionengine):
     def name(self):
         return b'bz2truncated'
 
     def bundletype(self):
         return None, b'_truncatedBZ'
 
     # We don't implement compressstream because it is hackily handled elsewhere.
 
     def decompressorreader(self, fh):
         return _TruncatedBZ2CompressedStreamReader(fh)
 
 
 compengines.register(_truncatedbz2engine())
 
 
 class _noopengine(compressionengine):
     def name(self):
         return b'none'
 
     def bundletype(self):
         """No compression is performed.
 
         Use this compression engine to explicitly disable compression.
         """
         return b'none', b'UN'
 
     # Clients always support uncompressed payloads. Servers don't because
     # unless you are on a fast network, uncompressed payloads can easily
     # saturate your network pipe.
     def wireprotosupport(self):
         return compewireprotosupport(b'none', 0, 10)
 
     # revlog special cases the uncompressed case, but implementing
     # revlogheader allows forcing uncompressed storage.
     def revlogheader(self):
         return b'\0'
 
     def compressstream(self, it, opts=None):
         return it
 
     def decompressorreader(self, fh):
         return fh
 
     class nooprevlogcompressor:
         def compress(self, data):
             return None
 
     def revlogcompressor(self, opts=None):
         return self.nooprevlogcompressor()
 
 
 compengines.register(_noopengine())
 
 
 class _zstdengine(compressionengine):
     def name(self):
         return b'zstd'
 
     @propertycache
     def _module(self):
         # Not all installs have the zstd module available. So defer importing
         # until first access.
         try:
             from .. import zstd  # pytype: disable=import-error
 
             # Force delayed import.
             zstd.__version__
             return zstd
         except ImportError:
             return None
 
     def available(self):
         return bool(self._module)
 
     def bundletype(self):
         """A modern compression algorithm that is fast and highly flexible.
 
         Only supported by Mercurial 4.1 and newer clients.
 
         With the default settings, zstd compression is both faster and yields
         better compression than ``gzip``. It also frequently yields better
         compression than ``bzip2`` while operating at much higher speeds.
 
         If this engine is available and backwards compatibility is not a
         concern, it is likely the best available engine.
         """
         return b'zstd', b'ZS'
 
     def wireprotosupport(self):
         return compewireprotosupport(b'zstd', 50, 50)
 
     def revlogheader(self):
         return b'\x28'
 
     def compressstream(self, it, opts=None):
         opts = opts or {}
         # zstd level 3 is almost always significantly faster than zlib
         # while providing no worse compression. It strikes a good balance
         # between speed and compression.
         level = opts.get(b'level', 3)
         # default to single-threaded compression
         threads = opts.get(b'threads', 0)
 
         zstd = self._module
         z = zstd.ZstdCompressor(level=level, threads=threads).compressobj()
         for chunk in it:
             data = z.compress(chunk)
             if data:
                 yield data
 
         yield z.flush()
 
     def decompressorreader(self, fh):
         return _ZstdCompressedStreamReader(fh, self._module)
 
     class zstdrevlogcompressor:
         def __init__(self, zstd, level=3):
             # TODO consider omitting frame magic to save 4 bytes.
             # This writes content sizes into the frame header. That is
             # extra storage. But it allows a correct size memory allocation
             # to hold the result.
             self._cctx = zstd.ZstdCompressor(level=level)
             self._dctx = zstd.ZstdDecompressor()
             self._compinsize = zstd.COMPRESSION_RECOMMENDED_INPUT_SIZE
             self._decompinsize = zstd.DECOMPRESSION_RECOMMENDED_INPUT_SIZE
 
         def compress(self, data):
             insize = len(data)
             # Caller handles empty input case.
             assert insize > 0
 
             if insize < 50:
                 return None
 
             elif insize <= 1000000:
                 compressed = self._cctx.compress(data)
                 if len(compressed) < insize:
                     return compressed
                 return None
             else:
                 z = self._cctx.compressobj()
                 chunks = []
                 pos = 0
                 while pos < insize:
                     pos2 = pos + self._compinsize
                     chunk = z.compress(data[pos:pos2])
                     if chunk:
                         chunks.append(chunk)
                     pos = pos2
                 chunks.append(z.flush())
 
                 if sum(map(len, chunks)) < insize:
                     return b''.join(chunks)
                 return None
 
         def decompress(self, data):
             insize = len(data)
 
             try:
                 # This was measured to be faster than other streaming
                 # decompressors.
                 dobj = self._dctx.decompressobj()
                 chunks = []
                 pos = 0
                 while pos < insize:
                     pos2 = pos + self._decompinsize
                     chunk = dobj.decompress(data[pos:pos2])
                     if chunk:
                         chunks.append(chunk)
                     pos = pos2
                 # Frame should be exhausted, so no finish() API.
 
                 return b''.join(chunks)
             except Exception as e:
                 raise error.StorageError(
                     _(b'revlog decompress error: %s')
                     % stringutil.forcebytestr(e)
                 )
 
     def revlogcompressor(self, opts=None):
         opts = opts or {}
         level = opts.get(b'zstd.level')
         if level is None:
             level = opts.get(b'level')
         if level is None:
             level = 3
         return self.zstdrevlogcompressor(self._module, level=level)
 
 
 compengines.register(_zstdengine())
 
 
 def bundlecompressiontopics():
     """Obtains a list of available bundle compressions for use in help."""
     # help.makeitemsdocs() expects a dict of names to items with a .__doc__.
     items = {}
 
     # We need to format the docstring. So use a dummy object/type to hold it
     # rather than mutating the original.
     class docobject:
         pass
 
     for name in compengines:
         engine = compengines[name]
 
         if not engine.available():
             continue
 
         bt = engine.bundletype()
         if not bt or not bt[0]:
             continue
 
         doc = b'``%s``\n    %s' % (bt[0], pycompat.getdoc(engine.bundletype))
 
         value = docobject()
         value.__doc__ = pycompat.sysstr(doc)
         value._origdoc = engine.bundletype.__doc__
         value._origfunc = engine.bundletype
 
         items[bt[0]] = value
 
     return items
 
 
 i18nfunctions = bundlecompressiontopics().values()

mercurial/wireprototypes.py

@@ -1,449 +1,447 b''
 # Copyright 2018 Gregory Szorc <[e-mail]>
 #
 # This software may be used and distributed according to the terms of the
 # GNU General Public License version 2 or any later version.
 
 
 from .node import (
     bin,
     hex,
 )
 from .i18n import _
 from .pycompat import getattr
 from .thirdparty import attr
 from . import (
     error,
     util,
 )
 from .interfaces import util as interfaceutil
 from .utils import compression
 
 # Names of the SSH protocol implementations.
 SSHV1 = b'ssh-v1'
 
 NARROWCAP = b'exp-narrow-1'
 ELLIPSESCAP1 = b'exp-ellipses-1'
 ELLIPSESCAP = b'exp-ellipses-2'
 SUPPORTED_ELLIPSESCAP = (ELLIPSESCAP1, ELLIPSESCAP)
 
 # All available wire protocol transports.
 TRANSPORTS = {
     SSHV1: {
         b'transport': b'ssh',
         b'version': 1,
     },
     b'http-v1': {
         b'transport': b'http',
         b'version': 1,
     },
 }
 
 
 class bytesresponse:
     """A wire protocol response consisting of raw bytes."""
 
     def __init__(self, data):
         self.data = data
 
 
 class ooberror:
     """wireproto reply: failure of a batch of operation
 
     Something failed during a batch call. The error message is stored in
     `self.message`.
     """
 
     def __init__(self, message):
         self.message = message
 
 
 class pushres:
     """wireproto reply: success with simple integer return
 
     The call was successful and returned an integer contained in `self.res`.
     """
 
     def __init__(self, res, output):
         self.res = res
         self.output = output
 
 
 class pusherr:
     """wireproto reply: failure
 
     The call failed. The `self.res` attribute contains the error message.
     """
 
     def __init__(self, res, output):
         self.res = res
         self.output = output
 
 
 class streamres:
     """wireproto reply: binary stream
 
     The call was successful and the result is a stream.
 
     Accepts a generator containing chunks of data to be sent to the client.
 
     ``prefer_uncompressed`` indicates that the data is expected to be
     uncompressable and that the stream should therefore use the ``none``
     engine.
     """
 
     def __init__(self, gen=None, prefer_uncompressed=False):
         self.gen = gen
         self.prefer_uncompressed = prefer_uncompressed
 
 
 class streamreslegacy:
     """wireproto reply: uncompressed binary stream
 
     The call was successful and the result is a stream.
 
     Accepts a generator containing chunks of data to be sent to the client.
 
     Like ``streamres``, but sends an uncompressed data for "version 1" clients
     using the application/mercurial-0.1 media type.
     """
 
     def __init__(self, gen=None):
         self.gen = gen
 
 
 # list of nodes encoding / decoding
 def decodelist(l, sep=b' '):
     if l:
         return [bin(v) for v in l.split(sep)]
     return []
 
 
 def encodelist(l, sep=b' '):
     try:
         return sep.join(map(hex, l))
     except TypeError:
         raise
 
 
 # batched call argument encoding
 
 
 def escapebatcharg(plain):
     return (
         plain.replace(b':', b':c')
         .replace(b',', b':o')
         .replace(b';', b':s')
         .replace(b'=', b':e')
     )
 
 
 def unescapebatcharg(escaped):
     return (
         escaped.replace(b':e', b'=')
         .replace(b':s', b';')
         .replace(b':o', b',')
         .replace(b':c', b':')
     )
 
 
 # mapping of options accepted by getbundle and their types
 #
 # Meant to be extended by extensions. It is the extension's responsibility to
 # ensure such options are properly processed in exchange.getbundle.
 #
 # supported types are:
 #
 # :nodes: list of binary nodes, transmitted as space-separated hex nodes
 # :csv:   list of values, transmitted as comma-separated values
 # :scsv:  set of values, transmitted as comma-separated values
 # :plain: string with no transformation needed.
 GETBUNDLE_ARGUMENTS = {
     b'heads': b'nodes',
     b'bookmarks': b'boolean',
     b'common': b'nodes',
     b'obsmarkers': b'boolean',
     b'phases': b'boolean',
     b'bundlecaps': b'scsv',
     b'listkeys': b'csv',
     b'cg': b'boolean',
     b'cbattempted': b'boolean',
     b'stream': b'boolean',
     b'includepats': b'csv',
     b'excludepats': b'csv',
 }
 
 
 class baseprotocolhandler(interfaceutil.Interface):
     """Abstract base class for wire protocol handlers.
 
     A wire protocol handler serves as an interface between protocol command
     handlers and the wire protocol transport layer. Protocol handlers provide
     methods to read command arguments, redirect stdio for the duration of
     the request, handle response types, etc.
     """
 
     name = interfaceutil.Attribute(
         """The name of the protocol implementation.
 
         Used for uniquely identifying the transport type.
         """
     )
 
     def getargs(args):
         """return the value for arguments in <args>
 
         For version 1 transports, returns a list of values in the same
         order they appear in ``args``. For version 2 transports, returns
         a dict mapping argument name to value.
         """
 
     def getprotocaps():
         """Returns the list of protocol-level capabilities of client
 
         Returns a list of capabilities as declared by the client for
         the current request (or connection for stateful protocol handlers)."""
 
     def getpayload():
         """Provide a generator for the raw payload.
 
         The caller is responsible for ensuring that the full payload is
         processed.
         """
 
     def mayberedirectstdio():
         """Context manager to possibly redirect stdio.
 
         The context manager yields a file-object like object that receives
         stdout and stderr output when the context manager is active. Or it
         yields ``None`` if no I/O redirection occurs.
 
         The intent of this context manager is to capture stdio output
         so it may be sent in the response. Some transports support streaming
         stdio to the client in real time. For these transports, stdio output
         won't be captured.
         """
 
     def client():
         """Returns a string representation of this client (as bytes)."""
 
     def addcapabilities(repo, caps):
         """Adds advertised capabilities specific to this protocol.
 
         Receives the list of capabilities collected so far.
 
         Returns a list of capabilities. The passed in argument can be returned.
         """
 
     def checkperm(perm):
         """Validate that the client has permissions to perform a request.
 
         The argument is the permission required to proceed. If the client
         doesn't have that permission, the exception should raise or abort
         in a protocol specific manner.
         """
 
 
 class commandentry:
     """Represents a declared wire protocol command."""
 
     def __init__(
         self,
         func,
         args=b'',
         transports=None,
         permission=b'push',
         cachekeyfn=None,
         extracapabilitiesfn=None,
     ):
         self.func = func
         self.args = args
         self.transports = transports or set()
         self.permission = permission
         self.cachekeyfn = cachekeyfn
         self.extracapabilitiesfn = extracapabilitiesfn
 
     def _merge(self, func, args):
         """Merge this instance with an incoming 2-tuple.
 
         This is called when a caller using the old 2-tuple API attempts
         to replace an instance. The incoming values are merged with
         data not captured by the 2-tuple and a new instance containing
         the union of the two objects is returned.
         """
         return commandentry(
             func,
             args=args,
             transports=set(self.transports),
             permission=self.permission,
         )
 
     # Old code treats instances as 2-tuples. So expose that interface.
     def __iter__(self):
         yield self.func
         yield self.args
 
     def __getitem__(self, i):
         if i == 0:
             return self.func
         elif i == 1:
             return self.args
         else:
             raise IndexError(b'can only access elements 0 and 1')
 
 
 class commanddict(dict):
     """Container for registered wire protocol commands.
 
     It behaves like a dict. But __setitem__ is overwritten to allow silent
     coercion of values from 2-tuples for API compatibility.
     """
 
     def __setitem__(self, k, v):
         if isinstance(v, commandentry):
             pass
         # Cast 2-tuples to commandentry instances.
         elif isinstance(v, tuple):
             if len(v) != 2:
                 raise ValueError(b'command tuples must have exactly 2 elements')
 
             # It is common for extensions to wrap wire protocol commands via
             # e.g. ``wireproto.commands[x] = (newfn, args)``. Because callers
             # doing this aren't aware of the new API that uses objects to store
             # command entries, we automatically merge old state with new.
             if k in self:
                 v = self[k]._merge(v[0], v[1])
             else:
                 # Use default values from @wireprotocommand.
                 v = commandentry(
                     v[0],
                     args=v[1],
                     transports=set(TRANSPORTS),
                     permission=b'push',
                 )
         else:
             raise ValueError(
                 b'command entries must be commandentry instances '
                 b'or 2-tuples'
             )
 
         return super(commanddict, self).__setitem__(k, v)
 
     def commandavailable(self, command, proto):
         """Determine if a command is available for the requested protocol."""
         assert proto.name in TRANSPORTS
 
         entry = self.get(command)
 
         if not entry:
             return False
 
         if proto.name not in entry.transports:
             return False
 
         return True
 
 
 def supportedcompengines(ui, role):
     """Obtain the list of supported compression engines for a request."""
     assert role in (compression.CLIENTROLE, compression.SERVERROLE)
 
     compengines = compression.compengines.supportedwireengines(role)
 
     # Allow config to override default list and ordering.
     if role == compression.SERVERROLE:
         configengines = ui.configlist(b'server', b'compressionengines')
         config = b'server.compressionengines'
     else:
         # This is currently implemented mainly to facilitate testing. In most
         # cases, the server should be in charge of choosing a compression engine
         # because a server has the most to lose from a sub-optimal choice. (e.g.
         # CPU DoS due to an expensive engine or a network DoS due to poor
         # compression ratio).
         configengines = ui.configlist(
             b'experimental', b'clientcompressionengines'
         )
         config = b'experimental.clientcompressionengines'
 
     # No explicit config. Filter out the ones that aren't supposed to be
     # advertised and return default ordering.
     if not configengines:
-        attr = (
-            b'serverpriority' if role == util.SERVERROLE else b'clientpriority'
-        )
+        attr = 'serverpriority' if role == util.SERVERROLE else 'clientpriority'
         return [
             e for e in compengines if getattr(e.wireprotosupport(), attr) > 0
         ]
 
     # If compression engines are listed in the config, assume there is a good
     # reason for it (like server operators wanting to achieve specific
     # performance characteristics). So fail fast if the config references
     # unusable compression engines.
     validnames = {e.name() for e in compengines}
     invalidnames = {e for e in configengines if e not in validnames}
     if invalidnames:
         raise error.Abort(
             _(b'invalid compression engine defined in %s: %s')
             % (config, b', '.join(sorted(invalidnames)))
         )
 
     compengines = [e for e in compengines if e.name() in configengines]
     compengines = sorted(
         compengines, key=lambda e: configengines.index(e.name())
     )
 
     if not compengines:
         raise error.Abort(
             _(
                 b'%s config option does not specify any known '
                 b'compression engines'
             )
             % config,
             hint=_(b'usable compression engines: %s')
             % b', '.sorted(validnames),  # pytype: disable=attribute-error
         )
 
     return compengines
 
 
 @attr.s
 class encodedresponse:
     """Represents response data that is already content encoded.
 
     Wire protocol version 2 only.
 
     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
     and the content from this object is used instead.
     """
 
     data = attr.ib()
 
 
 @attr.s
 class alternatelocationresponse:
     """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)
 
 
 @attr.s
 class indefinitebytestringresponse:
     """Represents an object to be encoded to an indefinite length bytestring.
 
     Instances are initialized from an iterable of chunks, with each chunk being
     a bytes instance.
     """
 
     chunks = attr.ib()