(self, data: bytes, seq: Optional[int] = None)
| 104 | self.incomplete = [] # type: List[Tuple[int, int]] |
| 105 | |
| 106 | def append(self, data: bytes, seq: Optional[int] = None) -> None: |
| 107 | if not data: |
| 108 | return |
| 109 | data_len = len(data) |
| 110 | if seq is None: |
| 111 | seq = self.content_len |
| 112 | seq = seq - 1 - self.noff |
| 113 | if seq < 0: |
| 114 | # Data is located before the start of the current buffer |
| 115 | # (e.g. the first fragment was missing) |
| 116 | self.content = bytearray(b"\x00" * (-seq)) + self.content |
| 117 | self.content_len += (-seq) |
| 118 | self.noff += seq |
| 119 | seq = 0 |
| 120 | if seq + data_len > self.content_len: |
| 121 | # Data is located after the end of the current buffer |
| 122 | self.content += b"\x00" * (seq - self.content_len + data_len) |
| 123 | # As data was missing, mark it. |
| 124 | # self.incomplete.append((self.content_len, seq)) |
| 125 | self.content_len = seq + data_len |
| 126 | assert len(self.content) == self.content_len |
| 127 | # XXX removes empty space marker. |
| 128 | # for ifrag in self.incomplete: |
| 129 | # if [???]: |
| 130 | # self.incomplete.remove([???]) |
| 131 | memoryview(self.content)[seq:seq + data_len] = data |
| 132 | |
| 133 | def shiftleft(self, i: int) -> None: |
| 134 | self.content = self.content[i:] |
no outgoing calls