| 1205 | self._io_buffer._segment_consumed = False |
| 1206 | |
| 1207 | def process_io_buffer(self): |
| 1208 | while True: |
| 1209 | if self._is_checksumming_enabled and self._io_buffer.readable_io_bytes(): |
| 1210 | self._process_segment_buffer() |
| 1211 | self._io_buffer.reset_io_buffer() |
| 1212 | |
| 1213 | if self._is_checksumming_enabled and not self._io_buffer.has_consumed_segment: |
| 1214 | # We couldn't read an entire segment from the io buffer, so return |
| 1215 | # control to allow more bytes to be read off the wire |
| 1216 | return |
| 1217 | |
| 1218 | if not self._current_frame: |
| 1219 | pos = self._read_frame_header() |
| 1220 | else: |
| 1221 | pos = self._io_buffer.readable_cql_frame_bytes() |
| 1222 | |
| 1223 | if not self._current_frame or pos < self._current_frame.end_pos: |
| 1224 | if self._is_checksumming_enabled and self._io_buffer.readable_io_bytes(): |
| 1225 | # We have a multi-segments message, and we need to read more |
| 1226 | # data to complete the current cql frame |
| 1227 | continue |
| 1228 | |
| 1229 | # we don't have a complete header yet, or we |
| 1230 | # already saw a header, but we don't have a |
| 1231 | # complete message yet |
| 1232 | return |
| 1233 | else: |
| 1234 | frame = self._current_frame |
| 1235 | self._io_buffer.cql_frame_buffer.seek(frame.body_offset) |
| 1236 | msg = self._io_buffer.cql_frame_buffer.read(frame.end_pos - frame.body_offset) |
| 1237 | self.process_msg(frame, msg) |
| 1238 | self._io_buffer.reset_cql_frame_buffer() |
| 1239 | self._current_frame = None |
| 1240 | |
| 1241 | @defunct_on_error |
| 1242 | def process_msg(self, header, body): |