bundle2: add a way to just forward the bundle2 stream to another user

There is use case for directly forward and bundle2 stream from the peer to a
file (eg: 'hg incoming --bundle'), However ssh peers have no way to know the
'getbundle' is over except for actually interpreting the bundle. So we need to
have the unbundle do the interpreting and forward job.

The function is marked as private to highlight that this is terrible and that we
are sorry.

mercurial/bundle2.py

@@ -641,6 +641,8 b' class unbundle20(unpackermixin):'
     This class is fed with a binary stream and yields parts through its
     `iterparts` methods."""
 
+    _magicstring = 'HG20'
+
     def __init__(self, ui, fp):
         """If header is specified, we do not read it out of the stream."""
         self.ui = ui
@@ -698,6 +700,46 b' class unbundle20(unpackermixin):'
         else:
             handler(self, name, value)
 
+    def _forwardchunks(self):
+        """utility to transfer a bundle2 as binary
+
+        This is made necessary by the fact the 'getbundle' command over 'ssh'
+        have no way to know then the reply end, relying on the bundle to be
+        interpreted to know its end. This is terrible and we are sorry, but we
+        needed to move forward to get general delta enabled.
+        """
+        yield self._magicstring
+        assert 'params' not in vars(self)
+        paramssize = self._unpack(_fstreamparamsize)[0]
+        if paramssize < 0:
+            raise error.BundleValueError('negative bundle param size: %i'
+                                         % paramssize)
+        yield _pack(_fstreamparamsize, paramssize)
+        if paramssize:
+            params = self._readexact(paramssize)
+            self._processallparams(params)
+            yield params
+            assert self._decompressor is util.decompressors[None]
+        # From there, payload might need to be decompressed
+        self._fp = self._decompressor(self._fp)
+        emptycount = 0
+        while emptycount < 2:
+            # so we can brainlessly loop
+            assert _fpartheadersize == _fpayloadsize
+            size = self._unpack(_fpartheadersize)[0]
+            yield _pack(_fpartheadersize, size)
+            if size:
+                emptycount = 0
+            else:
+                emptycount += 1
+                continue
+            if size == flaginterrupt:
+                continue
+            elif size < 0:
+                raise error.BundleValueError('negative chunk size: %i')
+            yield self._readexact(size)
+
+
     def iterparts(self):
         """yield all parts contained in the stream"""
         # make sure param have been loaded