Skip to content

Commit

Permalink
Made queues non-durable. Always retry after disconnect. (#120)
Browse files Browse the repository at this point in the history
  • Loading branch information
xarg authored and miguelgrinberg committed Jul 26, 2017
1 parent 97f4faa commit 358a8e1
Showing 1 changed file with 21 additions and 6 deletions.
27 changes: 21 additions & 6 deletions socketio/kombu_manager.py
Original file line number Diff line number Diff line change
Expand Up @@ -71,18 +71,33 @@ def _exchange(self):
def _queue(self):
queue_name = 'flask-socketio.' + str(uuid.uuid4())
return kombu.Queue(queue_name, self._exchange(),
durable=False,
queue_arguments={'x-expires': 300000})

def _producer(self):
return self._connection().Producer(exchange=self._exchange())

def __error_callback(self, exception, interval):
self.server.logger.exception('Sleeping {}s'.format(interval))

def _publish(self, data):
self.producer.publish(pickle.dumps(data))
connection = self._connection()
publish = connection.ensure(self.producer, self.producer.publish,
errback=self.__error_callback)
publish(pickle.dumps(data))

def _listen(self):
reader_queue = self._queue()
with self._connection().SimpleQueue(reader_queue) as queue:
while True:
message = queue.get(block=True)
message.ack()
yield message.payload

while True:
connection = self._connection().ensure_connection(
errback=self.__error_callback)
try:
with connection.SimpleQueue(reader_queue) as queue:
while True:
message = queue.get(block=True)
message.ack()
yield message.payload
except connection.connection_errors:
self.server.logger.exception("Connection error "
"while reading from queue")

0 comments on commit 358a8e1

Please sign in to comment.