Git-Mirrors/brozzler
1
0
Fork 0
mirror of https://github.com/internetarchive/brozzler.git synced 2025-02-24 08:39:59 -05:00
Code Issues Actions Packages Projects Releases Wiki Activity

ARI-3814 try to recover from rabbitmq communication problems

Browse Source
This commit is contained in:
Noah Levitt 2014-04-07 11:45:12 -07:00
parent 4e72cbae58
commit aacb886b62
1 changed files with 23 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

36
umbra/umbra.py
View File

@ -143,19 +143,29 @@ class Umbra:
self.amqp_thread.join() self.amqp_thread.join()
def consume_amqp(self): def consume_amqp(self):
self.umbra_exchange = Exchange(name='umbra', type='direct', durable=True) while not self.amqp_stop.is_set():
url_queue = Queue('urls', routing_key='url', exchange=self.umbra_exchange) try:
self.logger.info("connecting to amqp exchange={} at {}".format(self.umbra_exchange.name, self.amqp_url)) self.umbra_exchange = Exchange(name='umbra', type='direct', durable=True)
with Connection(self.amqp_url) as conn: url_queue = Queue('urls', routing_key='url', exchange=self.umbra_exchange)
self.producer = conn.Producer(serializer='json') self.logger.info("connecting to amqp exchange={} at {}".format(self.umbra_exchange.name, self.amqp_url))
self.producer_lock = threading.Lock() with Connection(self.amqp_url) as conn:
with conn.Consumer(url_queue, callbacks=[self.fetch_url]) as consumer: if self.producer_lock is None:
import socket self.producer_lock = threading.Lock()
while not self.amqp_stop.is_set(): with self.producer_lock:
try: self.producer = conn.Producer(serializer='json')
conn.drain_events(timeout=0.5) with conn.Consumer(url_queue, callbacks=[self.fetch_url]) as consumer:
except socket.timeout: import socket
pass while not self.amqp_stop.is_set():
import random
if random.randrange(0, 30) == 0:
raise BaseException("test exception!")
try:
conn.drain_events(timeout=0.5)
except socket.timeout:
pass
except BaseException as e:
self.logger.error("amqp exception {}".format(e))
self.logger.error("attempting to reopen amqp connection")
def fetch_url(self, body, message): def fetch_url(self, body, message):
client_id = body['clientId'] client_id = body['clientId']