Merge pull request #12 from d3rky/bugfix/LITE-17358-nack-created-messages

LITE-17358 nack with requeueing, add expiration for messages to 60s

Author: d3rky

dj_cqrs/transport/mixins.py

@@ -28,7 +28,7 @@ def log_consumed_denied(payload):
         :param dj_cqrs.dataclasses.TransportPayload payload: Transport payload from master model.
         """
         if payload.pk:
-            logger.info('CQRS is denied: pk = {} ({}).'.format(payload.pk, payload.cqrs_id))
+            logger.warning('CQRS is denied: pk = {} ({}).'.format(payload.pk, payload.cqrs_id))
 
     @staticmethod
     def log_produced(payload):

dj_cqrs/transport/rabbit_mq.py

@@ -99,7 +99,7 @@ def _consume_message(cls, ch, method, properties, body):
             ch.basic_ack(delivery_tag=method.delivery_tag)
             cls.log_consumed_accepted(payload)
         else:
-            ch.basic_nack(delivery_tag=method.delivery_tag, requeue=False)
+            ch.basic_nack(delivery_tag=method.delivery_tag)
             cls.log_consumed_denied(payload)
 
     @classmethod
@@ -111,7 +111,11 @@ def _produce_message(cls, channel, exchange, payload):
             routing_key=routing_key,
             body=ujson.dumps(payload.to_dict()),
             mandatory=True,
-            properties=BasicProperties(content_type='text/plain', delivery_mode=2)
+            properties=BasicProperties(
+                content_type='text/plain',
+                delivery_mode=2,  # make message persistent
+                expiration='60000',  # milliseconds
+            )
         )
 
     @staticmethod