comparison mercurial/httppeer.py @ 30465:40a1871eea5e

httppeer: use compression engine API for decompressing responses

In preparation for supporting multiple compression formats on the
wire protocol, we need all users of the wire protocol to use
compression engine APIs.

This commit ports the HTTP wire protocol client to use the
compression engine API.

The code for handling the HTTPException is a bit hacky. Essentially,
HTTPException could be thrown by any read() from the socket. However,
as part of porting the API, we no longer have a generator wrapping
the socket and we don't have a single place where we can trap the
exception. We solve this by introducing a proxy class that intercepts
read() and converts the exception appropriately.

In the future, we could introduce a new compression engine API that
supports emitting a generator of decompressed chunks. This would
eliminate the need for the proxy class. As I said when I introduced
the decompressorreader() API, I'm not fond of it and would support
transitioning to something better. This can be done as a follow-up,
preferably once all the code is using the compression engine API and
we have a better idea of the API needs of all the consumers.

30464:e16e234b9ca3

import errno
import os
import socket
import tempfile
import zlib

from .i18n import _
from .node import nullid
from . import (
    bundle2,

httplib = util.httplib
urlerr = util.urlerr
urlreq = util.urlreq

def zgenerator(f):
    zd = zlib.decompressobj()
    try:
        for chunk in util.filechunkiter(f):
            while chunk:
                yield zd.decompress(chunk, 2**18)
                chunk = zd.unconsumed_tail
    except httplib.HTTPException:
        raise IOError(None, _('connection ended unexpectedly'))
    yield zd.flush()

class httppeer(wireproto.wirepeer):
    def __init__(self, ui, path):
        self.path = path
        self.caps = None

            if version_info > (0, 1):
                raise error.RepoError(_("'%s' uses newer protocol %s") %
                                      (safeurl, version))

        if _compressible:
            return util.chunkbuffer(zgenerator(resp))

        return resp

    def _call(self, cmd, **args):
        fp = self._callstream(cmd, **args)

30465:40a1871eea5e

import errno
import os
import socket
import tempfile

from .i18n import _
from .node import nullid
from . import (
    bundle2,

httplib = util.httplib
urlerr = util.urlerr
urlreq = util.urlreq

# FUTURE: consider refactoring this API to use generators. This will
# require a compression engine API to emit generators.
def decompressresponse(response, engine):
    try:
        reader = engine.decompressorreader(response)
    except httplib.HTTPException:
        raise IOError(None, _('connection ended unexpectedly'))

    # We need to wrap reader.read() so HTTPException on subsequent
    # reads is also converted.
    origread = reader.read
    class readerproxy(reader.__class__):
        def read(self, *args, **kwargs):
            try:
                return origread(*args, **kwargs)
            except httplib.HTTPException:
                raise IOError(None, _('connection ended unexpectedly'))

    reader.__class__ = readerproxy
    return reader

class httppeer(wireproto.wirepeer):
    def __init__(self, ui, path):
        self.path = path
        self.caps = None

            if version_info > (0, 1):
                raise error.RepoError(_("'%s' uses newer protocol %s") %
                                      (safeurl, version))

        if _compressible:
            return decompressresponse(resp, util.compengines['zlib'])

        return resp

    def _call(self, cmd, **args):
        fp = self._callstream(cmd, **args)