Revision e6f5bb10 logic/dispatcher.py
| b/logic/dispatcher.py | ||
|---|---|---|
| 95 | 95 |
except socket.error: |
| 96 | 96 |
self.logger.error("Server went away, reconnecting...")
|
| 97 | 97 |
self._init() |
| 98 |
except Exception, e: |
|
| 99 |
self.logger.exception("Caught unexpected exception")
|
|
| 98 | 100 |
|
| 99 | 101 |
[self.chan.basic_cancel(clienttag) for clienttag in self.clienttags] |
| 100 | 102 |
self.chan.connection.close() |
| ... | ... | |
| 137 | 139 |
callback = getattr(callbacks, binding[3]) |
| 138 | 140 |
except AttributeError: |
| 139 | 141 |
self.logger.error("Cannot find callback %s" % binding[3])
|
| 140 |
continue
|
|
| 142 |
raise SystemExit(1)
|
|
| 141 | 143 |
|
| 142 | 144 |
self.chan.queue_bind(queue=binding[0], exchange=binding[1], |
| 143 | 145 |
routing_key=binding[2]) |
| ... | ... | |
| 170 | 172 |
DB_HANDLER_KEY_OP ='ganeti.%s.event.op' % prefix |
| 171 | 173 |
# notifications of type "ganeti-net-status" |
| 172 | 174 |
DB_HANDLER_KEY_NET ='ganeti.%s.event.net' % prefix |
| 173 |
# Build process monitoring event
|
|
| 174 |
BUILD_MONITOR_HANDLER = 'ganeti.%s.event.progress' %prefix |
|
| 175 |
# notifications of type "ganeti-create-progress"
|
|
| 176 |
BUILD_MONITOR_HANDLER = 'ganeti.%s.event.progress' % prefix
|
|
| 175 | 177 |
|
| 176 | 178 |
BINDINGS = [ |
| 177 | 179 |
# Queue # Exchange # RouteKey # Handler |
| ... | ... | |
| 268 | 270 |
|
| 269 | 271 |
|
| 270 | 272 |
def purge_exchanges(): |
| 271 |
""" |
|
| 272 |
Delete declared exchanges from RabbitMQ, after removing all queues first |
|
| 273 |
""" |
|
| 273 |
"""Delete declared exchanges from RabbitMQ, after removing all queues""" |
|
| 274 | 274 |
global QUEUES, BINDINGS |
| 275 | 275 |
purge_queues() |
| 276 | 276 |
|
| 277 | 277 |
conn = get_connection() |
| 278 | 278 |
chan = conn.channel() |
| 279 | 279 |
|
| 280 |
print "Exchnages to be deleted: ", settings.EXCHANGES
|
|
| 280 |
print "Exchanges to be deleted: ", settings.EXCHANGES
|
|
| 281 | 281 |
|
| 282 | 282 |
if not get_user_confirmation(): |
| 283 | 283 |
return |
| ... | ... | |
| 292 | 292 |
|
| 293 | 293 |
|
| 294 | 294 |
def drain_queue(queue): |
| 295 |
""" |
|
| 296 |
Strip a (declared) queue from all outstanding messages |
|
| 297 |
""" |
|
| 295 |
"""Strip a (declared) queue from all outstanding messages""" |
|
| 298 | 296 |
global QUEUES, BINDINGS |
| 299 | 297 |
if not queue: |
| 300 | 298 |
return |
| ... | ... | |
| 340 | 338 |
|
| 341 | 339 |
|
| 342 | 340 |
def get_connection(): |
| 343 |
conn = amqp.Connection( host=settings.RABBIT_HOST,
|
|
| 344 |
userid=settings.RABBIT_USERNAME, |
|
| 345 |
password=settings.RABBIT_PASSWORD, |
|
| 346 |
virtual_host=settings.RABBIT_VHOST) |
|
| 341 |
conn = amqp.Connection(host=settings.RABBIT_HOST, |
|
| 342 |
userid=settings.RABBIT_USERNAME,
|
|
| 343 |
password=settings.RABBIT_PASSWORD,
|
|
| 344 |
virtual_host=settings.RABBIT_VHOST)
|
|
| 347 | 345 |
return conn |
| 348 | 346 |
|
| 349 | 347 |
|
Also available in: Unified diff