Wrapper around non-seekable streams. Note that the implementation is tied to the decoder, not checking for dangerous arguments for the sake of performance. The read bytes are kept in an internal cache until setting _markedPosition which may reset the cache.
| 11 | from pyasn1.type import univ |
| 12 | |
| 13 | class CachingStreamWrapper(io.IOBase): |
| 14 | """Wrapper around non-seekable streams. |
| 15 | |
| 16 | Note that the implementation is tied to the decoder, |
| 17 | not checking for dangerous arguments for the sake |
| 18 | of performance. |
| 19 | |
| 20 | The read bytes are kept in an internal cache until |
| 21 | setting _markedPosition which may reset the cache. |
| 22 | """ |
| 23 | def __init__(self, raw): |
| 24 | self._raw = raw |
| 25 | self._cache = io.BytesIO() |
| 26 | self._markedPosition = 0 |
| 27 | |
| 28 | def peek(self, n): |
| 29 | result = self.read(n) |
| 30 | self._cache.seek(-len(result), os.SEEK_CUR) |
| 31 | return result |
| 32 | |
| 33 | def seekable(self): |
| 34 | return True |
| 35 | |
| 36 | def seek(self, n=-1, whence=os.SEEK_SET): |
| 37 | # Note that this not safe for seeking forward. |
| 38 | return self._cache.seek(n, whence) |
| 39 | |
| 40 | def read(self, n=-1): |
| 41 | read_from_cache = self._cache.read(n) |
| 42 | if n != -1: |
| 43 | n -= len(read_from_cache) |
| 44 | if not n: # 0 bytes left to read |
| 45 | return read_from_cache |
| 46 | |
| 47 | read_from_raw = self._raw.read(n) |
| 48 | |
| 49 | self._cache.write(read_from_raw) |
| 50 | |
| 51 | return read_from_cache + read_from_raw |
| 52 | |
| 53 | @property |
| 54 | def markedPosition(self): |
| 55 | """Position where the currently processed element starts. |
| 56 | |
| 57 | This is used for back-tracking in SingleItemDecoder.__call__ |
| 58 | and (indefLen)ValueDecoder and should not be used for other purposes. |
| 59 | The client is not supposed to ever seek before this position. |
| 60 | """ |
| 61 | return self._markedPosition |
| 62 | |
| 63 | @markedPosition.setter |
| 64 | def markedPosition(self, value): |
| 65 | # By setting the value, we ensure we won't seek back before it. |
| 66 | # `value` should be the same as the current position |
| 67 | # We don't check for this for performance reasons. |
| 68 | self._markedPosition = value |
| 69 | |
| 70 | # Whenever we set _marked_position, we know for sure |