alternative to io.BytesIO making debug easier and added a few conveniance methods
| 34 | |
| 35 | |
| 36 | class Buffer(object): |
| 37 | |
| 38 | """ |
| 39 | alternative to io.BytesIO making debug easier |
| 40 | and added a few conveniance methods |
| 41 | """ |
| 42 | |
| 43 | def __init__(self, data, start_pos=0, size=-1): |
| 44 | # self.logger = logging.getLogger(__name__) |
| 45 | self._data = data |
| 46 | self._cur_pos = start_pos |
| 47 | if size == -1: |
| 48 | size = len(data) - start_pos |
| 49 | self._size = size |
| 50 | |
| 51 | def __str__(self): |
| 52 | return "Buffer(size:{0}, data:{1})".format( |
| 53 | self._size, |
| 54 | self._data[self._cur_pos:self._cur_pos + self._size]) |
| 55 | __repr__ = __str__ |
| 56 | |
| 57 | def __len__(self): |
| 58 | return self._size |
| 59 | |
| 60 | def read(self, size): |
| 61 | """ |
| 62 | read and pop number of bytes for buffer |
| 63 | """ |
| 64 | if size > self._size: |
| 65 | raise NotEnoughData("Not enough data left in buffer, request for {0}, we have {1}".format(size, self)) |
| 66 | # self.logger.debug("Request for %s bytes, from %s", size, self) |
| 67 | self._size -= size |
| 68 | pos = self._cur_pos |
| 69 | self._cur_pos += size |
| 70 | data = self._data[pos:self._cur_pos] |
| 71 | # self.logger.debug("Returning: %s ", data) |
| 72 | return data |
| 73 | |
| 74 | def copy(self, size=-1): |
| 75 | """ |
| 76 | return a shadow copy, optionnaly only copy 'size' bytes |
| 77 | """ |
| 78 | if size == -1 or size > self._size: |
| 79 | size = self._size |
| 80 | return Buffer(self._data, self._cur_pos, size) |
| 81 | |
| 82 | def skip(self, size): |
| 83 | """ |
| 84 | skip size bytes in buffer |
| 85 | """ |
| 86 | if size > self._size: |
| 87 | raise NotEnoughData("Not enough data left in buffer, request for {0}, we have {1}".format(size, self)) |
| 88 | self._size -= size |
| 89 | self._cur_pos += size |
| 90 | |
| 91 | |
| 92 | class SocketWrapper(object): |
no outgoing calls
no test coverage detected