Merge pull request #19 from elarahq/retry_on_close

added retry

Author: azitabh

.gitignore

@@ -59,3 +59,4 @@ target/
 # Other
 .DS_Store
 ._.DS_Store
+*.swp

README.rst

@@ -6,6 +6,10 @@ A rabbitmq publisher and  receiver that implements a safe and reliable rabbitmq
 
 Changelog
 =========
+2.0.2
+-----
+
+* Added retry on connection lost.
 
 2.0.1
 -----

rmq/rmqproducer/rabbitmq_producer.py

@@ -3,6 +3,7 @@
 import signal
 import logging
 from connection import RMQConnectionPool
+from random import randint
 
 
 class Publisher(object):
@@ -19,6 +20,7 @@ class Publisher(object):
 
     """
 
+
     def __init__(self, amqp_url, exchange, **kwargs):
         """Create a new instance of the Publisher class, passing in the
         parameters used to connect to RabbitMQ. It established connection and
@@ -100,6 +102,7 @@ def connect(self):
         if connection is None:
             connection = pika.SelectConnection(pika.URLParameters(self._url),
                                                self.on_connection_open,
+                                               self.on_connection_error,
                                                stop_ioloop_on_close=False)
             self._connection = connection
         else:
@@ -122,6 +125,12 @@ def on_connection_open(self, unused_connection):
         self._LOGGER.info('Connection opened')
         self.add_on_connection_close_callback()
         self.open_channel()
+    
+    
+    def on_connection_error(self, connection, error):
+        self._LOGGER.warning("Publisher: Connection lost retrying in {time}".format(time=5))
+        connection.add_timeout(5, self.reconnect)
+
 
     def add_on_connection_close_callback(self):
         """This method adds an on close callback that will be invoked by pika
@@ -149,7 +158,7 @@ def on_connection_closed(self, connection, reply_code, reply_text):
         else:
             self._LOGGER.warning('Connection closed, reopening in %d seconds: (%s) %s',
                                  self.reconnect_time, reply_code, reply_text)
-            self._connection.add_timeout(self.reconnect_time, self.reconnect)
+            self._connection.add_timeout(5, self.reconnect)
 
     def reconnect(self):
         """Will be invoked by the IOLoop timer if the connection is

rmq/rmqreceiver/rabbitmq_receiver.py

@@ -2,6 +2,7 @@
 import pika
 import signal
 import logging
+from random import randint
 
 
 class Receiver(object):
@@ -94,6 +95,7 @@ def connect(self):
         self._LOGGER.info('Connecting to %s', self._url)
         return pika.SelectConnection(pika.URLParameters(self._url),
                                      self.on_connection_open,
+                                     self.on_connection_error,
                                      stop_ioloop_on_close=False)
 
     def on_connection_open(self, unused_connection):
@@ -119,6 +121,13 @@ def add_on_connection_close_callback(self):
         self._LOGGER.info('Adding connection close callback')
         self._connection.add_on_close_callback(self.on_connection_closed)
 
+   
+
+
+    def on_connection_error(self, connection, error):
+        self._LOGGER.warning("Connection failed.Retrying in next 5 seconds")
+        connection.add_timeout(5, self.reconnect)
+
     def on_connection_closed(self, connection, reply_code, reply_text):
         """Invoked by pika when the connection to RabbitMQ is closed unexpectedly.
 

setup.py

@@ -6,7 +6,7 @@ def readme():
         return f.read()
 
 setup(name='rmq-pub-sub',
-      version='2.0.1',
+      version='2.0.2',
       description='Rabbitmq Receiver and Publisher',
       url='https://github.com/loconsolutions/python-rabbitmq-pubsub',
       author='Rahul Kumar and Bipul Karnani',

test.py

@@ -0,0 +1,10 @@
+from rmq import Receiver, Publisher
+
+
+def callback(self, ch, method,properties, body):
+    print("message received")
+
+
+publisher = Publisher('amqp://guest:guest@10.1.6.46:5672/%2F/?socket_timeout=10', 'analytics_new')
+receiver = Receiver(callback, 'amqp://guest:guest@10.1.6.46:5672/%2F/?socket_timeout=10', 'analytics_new', binding_keys=["#"], queue='dsl_profiling_service')
+receiver.run()