Revision 2ef10562 snf-common/synnefo/lib/amqp_puka.py
| b/snf-common/synnefo/lib/amqp_puka.py | ||
|---|---|---|
| 80 | 80 |
""" |
| 81 | 81 |
def __init__(self, hosts=settings.AMQP_HOSTS, max_retries=30, |
| 82 | 82 |
confirms=True, confirm_buffer=100): |
| 83 |
"""Format hosts as "amqp://username:pass@host:port" """ |
|
| 83 |
""" |
|
| 84 |
Format hosts as "amqp://username:pass@host:port" |
|
| 85 |
max_retries=0 defaults to unlimited retries |
|
| 86 |
|
|
| 87 |
""" |
|
| 84 | 88 |
|
| 85 | 89 |
self.hosts = hosts |
| 86 | 90 |
shuffle(self.hosts) |
| ... | ... | |
| 98 | 102 |
self.exchanges = [] |
| 99 | 103 |
|
| 100 | 104 |
def connect(self, retries=0): |
| 101 |
if retries > self.max_retries:
|
|
| 102 |
logger.error("Aborting after %s retries", retries - 1)
|
|
| 105 |
if self.max_retries and retries >= self.max_retries:
|
|
| 106 |
logger.error("Aborting after %d retries", retries)
|
|
| 103 | 107 |
raise AMQPConnectionError('Aborting after %d connection failures.'\
|
| 104 |
% (retries - 1))
|
|
| 108 |
% retries)
|
|
| 105 | 109 |
return |
| 106 | 110 |
|
| 107 | 111 |
# Pick up a host |
| ... | ... | |
| 117 | 121 |
promise = self.client.connect() |
| 118 | 122 |
self.client.wait(promise) |
| 119 | 123 |
except socket_error as e: |
| 120 |
logger.error('Cannot connect to host %s: %s', host, e)
|
|
| 121 |
if retries > 2 * len(self.hosts): |
|
| 124 |
if retries < len(self.hosts): |
|
| 125 |
logger.warning('Cannot connect to host %s: %s', host, e)
|
|
| 126 |
else: |
|
| 127 |
logger.error('Cannot connect to host %s: %s', host, e)
|
|
| 122 | 128 |
sleep(1) |
| 123 | 129 |
return self.connect(retries + 1) |
| 124 | 130 |
|
Also available in: Unified diff