Merge pull request #4 from maxipavlovic/bugfix/LITE-15603

LITE-15603 Any exception in CQRS processing is handled in RabbitMQ consumer

Author: d3rky

dj_cqrs/controller/consumer.py

@@ -33,15 +33,15 @@ def route_signal_to_replica_model(signal_type, cqrs_id, instance_data, previous_
 
     if model_cls:
         if signal_type == SignalType.DELETE:
-            with transaction.atomic():
+            with transaction.atomic(savepoint=False):
                 return model_cls.cqrs_delete(instance_data)
 
         elif signal_type == SignalType.SAVE:
-            with transaction.atomic():
+            with transaction.atomic(savepoint=False):
                 return model_cls.cqrs_save(instance_data, previous_data=previous_data)
 
         elif signal_type == SignalType.SYNC:
-            with transaction.atomic():
+            with transaction.atomic(savepoint=False):
                 return model_cls.cqrs_save(
                     instance_data,
                     previous_data=previous_data,

dj_cqrs/transport/rabbit_mq.py

@@ -88,7 +88,12 @@ def _consume_message(cls, ch, method, properties, body):
         )
 
         cls.log_consumed(payload)
-        instance = consumer.consume(payload)
+
+        instance = None
+        try:
+            instance = consumer.consume(payload)
+        except Exception:
+            logger.error('CQRS service exception', exc_info=True)
 
         if instance:
             ch.basic_ack(delivery_tag=method.delivery_tag)