httppeer: use compression engine API for decompressing responses
authorGregory Szorc <gregory.szorc@gmail.com>
Sun, 20 Nov 2016 13:55:53 -0800
changeset 30465 40a1871eea5e
parent 30464 e16e234b9ca3
child 30466 2add671bf55b
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.
mercurial/httppeer.py
--- a/mercurial/httppeer.py	Sat Nov 19 18:31:40 2016 -0800
+++ b/mercurial/httppeer.py	Sun Nov 20 13:55:53 2016 -0800
@@ -12,7 +12,6 @@
 import os
 import socket
 import tempfile
-import zlib
 
 from .i18n import _
 from .node import nullid
@@ -30,16 +29,26 @@
 urlerr = util.urlerr
 urlreq = util.urlreq
 
-def zgenerator(f):
-    zd = zlib.decompressobj()
+# FUTURE: consider refactoring this API to use generators. This will
+# require a compression engine API to emit generators.
+def decompressresponse(response, engine):
     try:
-        for chunk in util.filechunkiter(f):
-            while chunk:
-                yield zd.decompress(chunk, 2**18)
-                chunk = zd.unconsumed_tail
+        reader = engine.decompressorreader(response)
     except httplib.HTTPException:
         raise IOError(None, _('connection ended unexpectedly'))
-    yield zd.flush()
+
+    # 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):
@@ -202,7 +211,7 @@
                                       (safeurl, version))
 
         if _compressible:
-            return util.chunkbuffer(zgenerator(resp))
+            return decompressresponse(resp, util.compengines['zlib'])
 
         return resp