add not acknowledge to worker

Author: vitorleal

.gitignore

@@ -8,3 +8,4 @@ build
 dist
 docs/_build
 .pypirc
+message_queue.egg-info

examples/publisher.py

@@ -4,21 +4,20 @@
 
 if __name__ == '__main__':
     # Instantiate the AMQP adapter with the host configuration
-    adapter = message_queue.AMQPAdapter(host='192.168.99.100')
+    adapter = message_queue.AMQPAdapter(host='127.0.0.1')
 
     # Configurate queue
     adapter.configurate_queue(queue='python.publish.test')
 
     # Instantiate publisher
     publisher = message_queue.Publisher(adapter)
 
-    # Create a new message
-    message = message_queue.Message({
-        'id': 12345,
-        'message': 'test publish'
-    })
-
     # Publish message
     for i in xrange(10000):
+        message = message_queue.Message({
+            'id': i,
+            'message': 'test publish'
+        })
+
         publisher.publish(message)
 

examples/subscriber.py

@@ -4,11 +4,20 @@
 
 
 def my_worker(channel, method, properties, body):
-    print json.loads(body)
+    try:
+        data = json.loads(body)
+        print data
+
+    except:
+        print 'Error parsing body: %r' % body
+        # return False to not acknowledge the message
+        return False
+
 
 if __name__ == '__main__':
     # Instantiate the AMQP adapter with the host configuration
-    adapter = message_queue.AMQPAdapter(host='192.168.99.100')
+    adapter = message_queue.AMQPAdapter(host='127.0.0.1')
+
     # Configurate queue
     adapter.configurate_queue(queue='python.publish.test')
 

message_queue/__init__.py

@@ -1,4 +1,4 @@
-__version__ = '0.1.1'
+__version__ = '0.1.6'
 
 from message_queue.logger import *
 

message_queue/adapters/amqp_adapter.py

@@ -46,7 +46,7 @@ def configurate_queue(self, **kwargs):
         if not self.queue:
             self.queue = kwargs.get('queue', '')
             self.basic_ack = kwargs.get('basic_ack', True)
-            self.prefetch_count = kwargs.get('prefetch_count', 0)
+            self.prefetch_count = kwargs.get('prefetch_count', 1)
 
             self.channel.queue_declare(
                 queue       = self.queue,
@@ -64,7 +64,7 @@ def configurate_queue(self, **kwargs):
                          self.queue, self.basic_ack, self.prefetch_count)
 
     def connect(self):
-        """Connect usgin BlockingConnection.
+        """Connect to AMQP server usgin BlockingConnection.
 
         """
         try:
@@ -108,7 +108,7 @@ def format_message(self, message):
         _message = {}
         _message['body'] = message['body']
         _message['routing_key'] = self.queue
-        _message['exchange']   = exchange
+        _message['exchange'] = exchange
         _message['properties'] = pika.BasicProperties(
             content_type='application/json',
             delivery_mode=delivery_mode,
@@ -124,8 +124,8 @@ def consume(self, worker):
         :param function worker: Method that consume the message
 
         """
-        self._consume_worker = worker
-        self.channel.basic_consume(self.consume_callback, self.queue)
+        callback = self.consume_callback(worker)
+        self.channel.basic_consume(callback, self.queue)
 
         try:
             self.channel.start_consuming()
@@ -134,24 +134,40 @@ def consume(self, worker):
             self.channel.stop_consuming()
             self.close()
 
-    def consume_callback(self, channel, method, properties, body):
-        """Message consume callback
+    def consume_callback(self, worker):
+        """Decorate worker to exectue on consume callback.
 
-        :param pika.channel.Channel channel: The channel object
-        :param pika.Spec.Basic.Deliver method: basic_deliver method
-        :param pika.Spec.BasicProperties properties: properties
-        :param str|unicode body: The message body
+        :param function worker: Worker to execture in the consume callback
 
         """
-        self._consume_worker(channel, method, properties, body)
-        self.consume_acknowledge(channel, method.delivery_tag)
+        def callback(channel, method, properties, body):
+            """Message consume callback.
 
-    def consume_acknowledge(self, channel, tag):
-        """Message acknowledge
+            :param pika.channel.Channel channel: The channel object
+            :param pika.Spec.Basic.Deliver method: basic_deliver method
+            :param pika.Spec.BasicProperties properties: properties
+            :param str|unicode body: The message body
+
+            """
+            # Execute the worker
+            acknowledge = worker(channel, method, properties, body)
+
+            # Acknowledge the message or not
+            self._consume_acknowledge(channel, method.delivery_tag, acknowledge)
+
+        return callback
+
+    def _consume_acknowledge(self, channel, tag, acknowledge=True):
+        """Message acknowledge.
 
         :param pika.channel.Channel channel: Channel to acknowledge the message
         :param int tag: Message tag to acknowledge
+        :param bool acknowledge: If should acknowledge the message or not
 
         """
+        if acknowledge is False:
+            channel.basic_nack(delivery_tag=tag)
+            return
+
         channel.basic_ack(delivery_tag=tag)
 

setup.py

@@ -2,15 +2,15 @@
 
 setuptools.setup(
     name='message-queue',
-    version='0.1.1',
+    version='0.1.6',
     description='Message Queue',
     long_description='Message Queue python library to publish and subscribe to queues with diferent types of adapters.',
     url='https://github.com/ingresse/message-queue-python',
     author='Ingresse',
     author_email='vitor.leal@ingresse.com',
     license='BSD',
     packages=setuptools.find_packages(),
-    install_requires=['pika'],
-    download_url = 'https://github.com/ingresse/message-queue-python/tarball/0.1.1',
+    install_requires=['pika>=0.10.0'],
+    download_url = 'https://github.com/ingresse/message-queue-python/tarball/0.1.6',
 )