| 1651 | |
| 1652 | |
| 1653 | class HeartbeatFuture(object): |
| 1654 | def __init__(self, connection, owner): |
| 1655 | self._exception = None |
| 1656 | self._event = Event() |
| 1657 | self.connection = connection |
| 1658 | self.owner = owner |
| 1659 | log.debug("Sending options message heartbeat on idle connection (%s) %s", |
| 1660 | id(connection), connection.endpoint) |
| 1661 | with connection.lock: |
| 1662 | if connection.in_flight < connection.max_request_id: |
| 1663 | connection.in_flight += 1 |
| 1664 | connection.send_msg(OptionsMessage(), connection.get_request_id(), self._options_callback) |
| 1665 | else: |
| 1666 | self._exception = Exception("Failed to send heartbeat because connection 'in_flight' exceeds threshold") |
| 1667 | self._event.set() |
| 1668 | |
| 1669 | def wait(self, timeout): |
| 1670 | self._event.wait(timeout) |
| 1671 | if self._event.is_set(): |
| 1672 | if self._exception: |
| 1673 | raise self._exception |
| 1674 | else: |
| 1675 | raise OperationTimedOut("Connection heartbeat timeout after %s seconds" % (timeout,), self.connection.endpoint) |
| 1676 | |
| 1677 | def _options_callback(self, response): |
| 1678 | if isinstance(response, SupportedMessage): |
| 1679 | log.debug("Received options response on connection (%s) from %s", |
| 1680 | id(self.connection), self.connection.endpoint) |
| 1681 | else: |
| 1682 | if isinstance(response, ConnectionException): |
| 1683 | self._exception = response |
| 1684 | else: |
| 1685 | self._exception = ConnectionException("Received unexpected response to OptionsMessage: %s" |
| 1686 | % (response,)) |
| 1687 | self._event.set() |
| 1688 | |
| 1689 | |
| 1690 | class ConnectionHeartbeat(Thread): |