| 4670 | self.send_request() |
| 4671 | |
| 4672 | def _set_result(self, host, connection, pool, response): |
| 4673 | try: |
| 4674 | self.coordinator_host = host |
| 4675 | if pool: |
| 4676 | pool.return_connection(connection) |
| 4677 | |
| 4678 | trace_id = getattr(response, 'trace_id', None) |
| 4679 | if trace_id: |
| 4680 | if not self._query_traces: |
| 4681 | self._query_traces = [] |
| 4682 | self._query_traces.append(QueryTrace(trace_id, self.session)) |
| 4683 | |
| 4684 | self._warnings = getattr(response, 'warnings', None) |
| 4685 | self._custom_payload = getattr(response, 'custom_payload', None) |
| 4686 | |
| 4687 | if isinstance(response, ResultMessage): |
| 4688 | if response.kind == RESULT_KIND_SET_KEYSPACE: |
| 4689 | session = getattr(self, 'session', None) |
| 4690 | # since we're running on the event loop thread, we need to |
| 4691 | # use a non-blocking method for setting the keyspace on |
| 4692 | # all connections in this session, otherwise the event |
| 4693 | # loop thread will deadlock waiting for keyspaces to be |
| 4694 | # set. This uses a callback chain which ends with |
| 4695 | # self._set_keyspace_completed() being called in the |
| 4696 | # event loop thread. |
| 4697 | if session: |
| 4698 | session._set_keyspace_for_all_pools( |
| 4699 | response.new_keyspace, self._set_keyspace_completed) |
| 4700 | elif response.kind == RESULT_KIND_SCHEMA_CHANGE: |
| 4701 | # refresh the schema before responding, but do it in another |
| 4702 | # thread instead of the event loop thread |
| 4703 | self.is_schema_agreed = False |
| 4704 | self.session.submit( |
| 4705 | refresh_schema_and_set_result, |
| 4706 | self.session.cluster.control_connection, |
| 4707 | self, connection, **response.schema_change_event) |
| 4708 | elif response.kind == RESULT_KIND_ROWS: |
| 4709 | self._paging_state = response.paging_state |
| 4710 | self._col_names = response.column_names |
| 4711 | self._col_types = response.column_types |
| 4712 | if getattr(self.message, 'continuous_paging_options', None): |
| 4713 | self._handle_continuous_paging_first_response(connection, response) |
| 4714 | else: |
| 4715 | self._set_final_result(self.row_factory(response.column_names, response.parsed_rows)) |
| 4716 | elif response.kind == RESULT_KIND_VOID: |
| 4717 | self._set_final_result(None) |
| 4718 | else: |
| 4719 | self._set_final_result(response) |
| 4720 | elif isinstance(response, ErrorMessage): |
| 4721 | retry_policy = self._retry_policy |
| 4722 | |
| 4723 | if isinstance(response, ReadTimeoutErrorMessage): |
| 4724 | if self._metrics is not None: |
| 4725 | self._metrics.on_read_timeout() |
| 4726 | retry = retry_policy.on_read_timeout( |
| 4727 | self.query, retry_num=self._query_retries, **response.info) |
| 4728 | elif isinstance(response, WriteTimeoutErrorMessage): |
| 4729 | if self._metrics is not None: |