| dlax | |
| yuja |
| hg-reviewers |
Constants make code easier to read than magic numbers.
I also threw in an explicit argument in a caller to further
increase code comprehension.
| Lint Skipped |
| Unit Tests Skipped |
| channel usable. But none of the part read from an abort are processed. In 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 | future, dropping the stream may become an option for channel we do not care to | ||||
| preserve. | preserve. | ||||
| """ | """ | ||||
| from __future__ import absolute_import, division | from __future__ import absolute_import, division | ||||
| import errno | import errno | ||||
| import os | |||||
| import re | import re | ||||
| import string | import string | ||||
| import struct | import struct | ||||
| import sys | import sys | ||||
| from .i18n import _ | from .i18n import _ | ||||
| from . import ( | from . import ( | ||||
| changegroup, | changegroup, | ||||
| def __enter__(self): | def __enter__(self): | ||||
| def func(): | def func(): | ||||
| itr = enumerate(self.unbundler.iterparts()) | itr = enumerate(self.unbundler.iterparts()) | ||||
| for count, p in itr: | for count, p in itr: | ||||
| self.count = count | self.count = count | ||||
| self.current = p | self.current = p | ||||
| yield p | yield p | ||||
| p.seek(0, 2) | p.seek(0, os.SEEK_END) | ||||
| self.current = None | self.current = None | ||||
| self.iterator = func() | self.iterator = func() | ||||
| return self.iterator | return self.iterator | ||||
| def __exit__(self, type, exc, tb): | def __exit__(self, type, exc, tb): | ||||
| if not self.iterator: | if not self.iterator: | ||||
| return | return | ||||
| # Only gracefully abort in a normal exception situation. User aborts | # Only gracefully abort in a normal exception situation. User aborts | ||||
| # like Ctrl+C throw a KeyboardInterrupt which is not a base Exception, | # like Ctrl+C throw a KeyboardInterrupt which is not a base Exception, | ||||
| # and should not gracefully cleanup. | # and should not gracefully cleanup. | ||||
| if isinstance(exc, Exception): | if isinstance(exc, Exception): | ||||
| # Any exceptions seeking to the end of the bundle at this point are | # Any exceptions seeking to the end of the bundle at this point are | ||||
| # almost certainly related to the underlying stream being bad. | # almost certainly related to the underlying stream being bad. | ||||
| # And, chances are that the exception we're handling is related to | # And, chances are that the exception we're handling is related to | ||||
| # getting in that bad state. So, we swallow the seeking error and | # getting in that bad state. So, we swallow the seeking error and | ||||
| # re-raise the original error. | # re-raise the original error. | ||||
| seekerror = False | seekerror = False | ||||
| try: | try: | ||||
| if self.current: | if self.current: | ||||
| # consume the part content to not corrupt the stream. | # consume the part content to not corrupt the stream. | ||||
| self.current.seek(0, 2) | self.current.seek(0, os.SEEK_END) | ||||
| for part in self.iterator: | for part in self.iterator: | ||||
| # consume the bundle content | # consume the bundle content | ||||
| part.seek(0, 2) | part.seek(0, os.SEEK_END) | ||||
| except Exception: | except Exception: | ||||
| seekerror = True | seekerror = True | ||||
| # Small hack to let caller code distinguish exceptions from bundle2 | # Small hack to let caller code distinguish exceptions from bundle2 | ||||
| # processing from processing the old format. This is mostly needed | # processing from processing the old format. This is mostly needed | ||||
| # to handle different return codes to unbundle according to the type | # to handle different return codes to unbundle according to the type | ||||
| # of bundle. We should probably clean up or drop this return code | # of bundle. We should probably clean up or drop this return code | ||||
| # craziness in a future version. | # craziness in a future version. | ||||
| indebug(self.ui, 'start extraction of bundle2 parts') | indebug(self.ui, 'start extraction of bundle2 parts') | ||||
| headerblock = self._readpartheader() | headerblock = self._readpartheader() | ||||
| while headerblock is not None: | while headerblock is not None: | ||||
| part = unbundlepart(self.ui, headerblock, self._fp) | part = unbundlepart(self.ui, headerblock, self._fp) | ||||
| yield part | yield part | ||||
| # Seek to the end of the part to force it's consumption so the next | # Seek to the end of the part to force it's consumption so the next | ||||
| # part can be read. But then seek back to the beginning so the | # part can be read. But then seek back to the beginning so the | ||||
| # code consuming this generator has a part that starts at 0. | # code consuming this generator has a part that starts at 0. | ||||
| part.seek(0, 2) | part.seek(0, os.SEEK_END) | ||||
| part.seek(0) | part.seek(0, os.SEEK_SET) | ||||
| headerblock = self._readpartheader() | headerblock = self._readpartheader() | ||||
| indebug(self.ui, 'end of bundle2 stream') | indebug(self.ui, 'end of bundle2 stream') | ||||
| def _readpartheader(self): | def _readpartheader(self): | ||||
| """reads a part header size and return the bytes blob | """reads a part header size and return the bytes blob | ||||
| returns None if empty""" | returns None if empty""" | ||||
| headersize = self._unpack(_fpartheadersize)[0] | headersize = self._unpack(_fpartheadersize)[0] | ||||
| hardabort = False | hardabort = False | ||||
| try: | try: | ||||
| _processpart(op, part) | _processpart(op, part) | ||||
| except (SystemExit, KeyboardInterrupt): | except (SystemExit, KeyboardInterrupt): | ||||
| hardabort = True | hardabort = True | ||||
| raise | raise | ||||
| finally: | finally: | ||||
| if not hardabort: | if not hardabort: | ||||
| part.seek(0, 2) | part.seek(0, os.SEEK_END) | ||||
| self.ui.debug('bundle2-input-stream-interrupt:' | self.ui.debug('bundle2-input-stream-interrupt:' | ||||
| ' closing out of band context\n') | ' closing out of band context\n') | ||||
| class interruptoperation(object): | class interruptoperation(object): | ||||
| """A limited operation to be use by part handler during interruption | """A limited operation to be use by part handler during interruption | ||||
| It only have access to an ui object. | It only have access to an ui object. | ||||
| """ | """ | ||||
| self.ui.debug('bundle2-input-part: total payload size %i\n' | self.ui.debug('bundle2-input-part: total payload size %i\n' | ||||
| % self._pos) | % self._pos) | ||||
| self.consumed = True | self.consumed = True | ||||
| return data | return data | ||||
| def tell(self): | def tell(self): | ||||
| return self._pos | return self._pos | ||||
| def seek(self, offset, whence=0): | def seek(self, offset, whence=os.SEEK_SET): | ||||
| if whence == 0: | if whence == os.SEEK_SET: | ||||
| newpos = offset | newpos = offset | ||||
| elif whence == 1: | elif whence == os.SEEK_CUR: | ||||
| newpos = self._pos + offset | newpos = self._pos + offset | ||||
| elif whence == 2: | elif whence == os.SEEK_END: | ||||
| if not self.consumed: | if not self.consumed: | ||||
| self.read() | self.read() | ||||
| newpos = self._chunkindex[-1][0] - offset | newpos = self._chunkindex[-1][0] - offset | ||||
| else: | else: | ||||
| raise ValueError('Unknown whence value: %r' % (whence,)) | raise ValueError('Unknown whence value: %r' % (whence,)) | ||||
| if newpos > self._chunkindex[-1][0] and not self.consumed: | if newpos > self._chunkindex[-1][0] and not self.consumed: | ||||
| self.read() | self.read() | ||||