x/python-ganttclient
Code Issues Proposed changes

Reverted some changes

Browse Source
This commit is contained in:
Josh Kearney
2010-11-19 16:42:37 -06:00
parent 3bb2b2b4be
commit bc4a038e3b
1 changed files with 22 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

32
nova/rpc.py
View File

@@ -83,6 +83,8 @@ class Consumer(messaging.Consumer):
Contains methods for connecting the fetch method to async loops
"""
def __init__(self, *args, **kwargs):
self.failed_connection = False
while True:
try:
super(Consumer, self).__init__(*args, **kwargs)
@@ -91,27 +93,33 @@ class Consumer(messaging.Consumer):
logging.warning("AMQP server on %s:%d is unreachable. " \
"Trying again in 30 seconds." % (
FLAGS.rabbit_host,
FLAGS.rabbit_port
))
FLAGS.rabbit_port))
time.sleep(30)
continue
def fetch(self, no_ack=None, auto_ack=None, enable_callbacks=False):
"""Wraps the parent fetch with some logic for failed connections"""
# TODO(vish): the logic for failed connections and logging should be
# refactored into some sort of connection manager object
try:
super(Consumer, self).fetch(no_ack, auto_ack, enable_callbacks)
except: # Catching all because carrot sucks
try:
if self.failed_connection:
# NOTE(vish): conn is defined in the parent class, we can
# recreate it as long as we create the backend too
# pylint: disable-msg=W0201
self.connection = Connection.recreate()
self.backend = self.connection.create_backend()
self.declare()
except: # Catching all because carrot sucks
logging.warning("AMQP server on %s:%d is unreachable. " \
"Trying again in 30 seconds." % (
FLAGS.rabbit_host,
FLAGS.rabbit_port
))
time.sleep(30)
super(Consumer, self).fetch(no_ack, auto_ack, enable_callbacks)
if self.failed_connection:
logging.error("Reconnected to queue")
self.failed_connection = False
# NOTE(vish): This is catching all errors because we really don't
# exceptions to be logged 10 times a second if some
# persistent failure occurs.
except Exception: # pylint: disable-msg=W0703
if not self.failed_connection:
logging.exception("Failed to fetch message from queue")
self.failed_connection = True
def attach_to_eventlet(self):
"""Only needed for unit tests!"""