Use this in order to avoid deadlocking the event loop thread. When the operation completes, `callback` will be called with two arguments: this connection and an Exception if an error occurred, otherwise :const:`None`. This method will always increment :attr:
(self, keyspace, callback)
| 1531 | raise conn_exc |
| 1532 | |
| 1533 | def set_keyspace_async(self, keyspace, callback): |
| 1534 | """ |
| 1535 | Use this in order to avoid deadlocking the event loop thread. |
| 1536 | When the operation completes, `callback` will be called with |
| 1537 | two arguments: this connection and an Exception if an error |
| 1538 | occurred, otherwise :const:`None`. |
| 1539 | |
| 1540 | This method will always increment :attr:`.in_flight` attribute, even if |
| 1541 | it doesn't need to make a request, just to maintain an |
| 1542 | ":attr:`.in_flight` is incremented" invariant. |
| 1543 | """ |
| 1544 | # Here we increment in_flight unconditionally, whether we need to issue |
| 1545 | # a request or not. This is bad, but allows callers -- specifically |
| 1546 | # _set_keyspace_for_all_conns -- to assume that we increment |
| 1547 | # self.in_flight during this call. This allows the passed callback to |
| 1548 | # safely call HostConnection{Pool,}.return_connection on this |
| 1549 | # Connection. |
| 1550 | # |
| 1551 | # We use a busy wait on the lock here because: |
| 1552 | # - we'll only spin if the connection is at max capacity, which is very |
| 1553 | # unlikely for a set_keyspace call |
| 1554 | # - it allows us to avoid signaling a condition every time a request completes |
| 1555 | while True: |
| 1556 | with self.lock: |
| 1557 | if self.in_flight < self.max_request_id: |
| 1558 | self.in_flight += 1 |
| 1559 | break |
| 1560 | time.sleep(0.001) |
| 1561 | |
| 1562 | if not keyspace or keyspace == self.keyspace: |
| 1563 | callback(self, None) |
| 1564 | return |
| 1565 | |
| 1566 | query = QueryMessage(query='USE "%s"' % (keyspace,), |
| 1567 | consistency_level=ConsistencyLevel.ONE) |
| 1568 | |
| 1569 | def process_result(result): |
| 1570 | if isinstance(result, ResultMessage): |
| 1571 | self.keyspace = keyspace |
| 1572 | callback(self, None) |
| 1573 | elif isinstance(result, InvalidRequestException): |
| 1574 | callback(self, result.to_exception()) |
| 1575 | else: |
| 1576 | callback(self, self.defunct(ConnectionException( |
| 1577 | "Problem while setting keyspace: %r" % (result,), self.endpoint))) |
| 1578 | |
| 1579 | # We've incremented self.in_flight above, so we "have permission" to |
| 1580 | # acquire a new request id |
| 1581 | request_id = self.get_request_id() |
| 1582 | |
| 1583 | self.send_msg(query, request_id, process_result) |
| 1584 | |
| 1585 | @property |
| 1586 | def is_idle(self): |
no test coverage detected