| 1240 | |
| 1241 | @defunct_on_error |
| 1242 | def process_msg(self, header, body): |
| 1243 | self.msg_received = True |
| 1244 | stream_id = header.stream |
| 1245 | if stream_id < 0: |
| 1246 | callback = None |
| 1247 | decoder = ProtocolHandler.decode_message |
| 1248 | result_metadata = None |
| 1249 | else: |
| 1250 | if stream_id in self._continuous_paging_sessions: |
| 1251 | paging_session = self._continuous_paging_sessions[stream_id] |
| 1252 | callback = paging_session.on_message |
| 1253 | decoder = paging_session.decoder |
| 1254 | result_metadata = None |
| 1255 | else: |
| 1256 | need_notify_of_release = False |
| 1257 | with self.lock: |
| 1258 | if stream_id in self.orphaned_request_ids: |
| 1259 | self.in_flight -= 1 |
| 1260 | self.orphaned_request_ids.remove(stream_id) |
| 1261 | need_notify_of_release = True |
| 1262 | if need_notify_of_release and self._on_orphaned_stream_released: |
| 1263 | self._on_orphaned_stream_released() |
| 1264 | |
| 1265 | try: |
| 1266 | callback, decoder, result_metadata = self._requests.pop(stream_id) |
| 1267 | # This can only happen if the stream_id was |
| 1268 | # removed due to an OperationTimedOut |
| 1269 | except KeyError: |
| 1270 | with self.lock: |
| 1271 | self.request_ids.append(stream_id) |
| 1272 | return |
| 1273 | |
| 1274 | try: |
| 1275 | response = decoder(header.version, self.user_type_map, stream_id, |
| 1276 | header.flags, header.opcode, body, self.decompressor, result_metadata) |
| 1277 | except Exception as exc: |
| 1278 | log.exception("Error decoding response from Cassandra. " |
| 1279 | "%s; buffer: %r", header, self._iobuf.getvalue()) |
| 1280 | if callback is not None: |
| 1281 | callback(exc) |
| 1282 | self.defunct(exc) |
| 1283 | return |
| 1284 | |
| 1285 | try: |
| 1286 | if stream_id >= 0: |
| 1287 | if isinstance(response, ProtocolException): |
| 1288 | if 'unsupported protocol version' in response.message: |
| 1289 | self.is_unsupported_proto_version = True |
| 1290 | else: |
| 1291 | log.error("Closing connection %s due to protocol error: %s", self, response.summary_msg()) |
| 1292 | self.defunct(response) |
| 1293 | if callback is not None: |
| 1294 | callback(response) |
| 1295 | else: |
| 1296 | self.handle_pushed(response) |
| 1297 | except Exception: |
| 1298 | log.exception("Callback handler errored, ignoring:") |
| 1299 | |