bundle2: read the whole bundle from stream on abort

When the bundle processing abort on unknown mandatory parts, we now makes sure
all the bundle content is read. This avoid leaving the communication channel in
an unrecoverable state.

mercurial/bundle2.py

@@ -130,8 +130,11 The matching of a part to its handler is
 part type is used to know if a part is mandatory or advisory. If the Part type
 contains any uppercase char it is considered mandatory. When no handler is
 known for a Mandatory part, the process is aborted and an exception is raised.
-If the part is advisory and no handler is known, the part is ignored.
-
+If the part is advisory and no handler is known, the part is ignored. When the
+process is aborted, the full bundle is still read from the stream to keep the
+channel usable. But none of the part read from an abort are processed. In the
+future, dropping the stream may become an option for channel we do not care to
+preserve.
 """
 
 import util
@@ -205,7 +208,9 def processbundle(repo, stream):
     # - replace this is a init function soon.
     # - exception catching
     unbundler.params
-    for part in unbundler:
+    iterparts = iter(unbundler)
+    try:
+        for part in iterparts:
         parttype = part.type
         # part key are matched lower case
         key = parttype.lower()
@@ -216,12 +221,16 def processbundle(repo, stream):
             if key != parttype: # mandatory parts
                 # todo:
                 # - use a more precise exception
-                # - consume the bundle2 stream anyway.
                 raise
             ui.debug('ignoring unknown advisory part %r\n' % key)
-            # todo: consume the part (once we use streamed parts)
+                # todo:
+                # - consume the part once we use streaming
             continue
         handler(repo, part)
+    except Exception:
+        for part in iterparts:
+            pass # consume the bundle content
+        raise
 
 class bundle20(object):
     """represent an outgoing bundle2 container

tests/test-bundle2.t

@@ -72,9 +72,13 Create an extension to test bundle2 API
   > def cmdunbundle2(ui, repo):
   >     """process a bundle2 stream from stdin on the current repo"""
   >     try:
+  >         try:
   >         bundle2.processbundle(repo, sys.stdin)
   >     except KeyError, exc:
   >         raise util.Abort('missing support for %s' % exc)
+  >     finally:
+  >         remains = sys.stdin.read()
+  >         ui.write('%i unread bytes\n' % len(remains))
   > 
   > @command('statbundle2', [], '')
   > def cmdstatbundle2(ui, repo):
@@ -385,6 +389,7 Process the bundle
   ignoring unknown advisory part 'test:math'
   part header size: 0
   end of bundle2 stream
+  0 unread bytes
 
 
   $ hg bundle2 --parts --unknown ../unknown.hg2
@@ -394,5 +399,6 Process the bundle
       Patali Dirapata, Cromda Cromda Ripalo, Pata Pata, Ko Ko Ko
       Bokoro Dipoulito, Rondi Rondi Pepino, Pata Pata, Ko Ko Ko
       Emana Karassoli, Loucra Loucra Ponponto, Pata Pata, Ko Ko Ko.
+  0 unread bytes
   abort: missing support for 'test:unknown'
   [255]