Merge pull request #30 from cloudblue/lite-18256-requeue-fix

LITE-18058 Fix for requeuing: local replica routing key is used instead of global one

Author: maxipavlovic

dj_cqrs/controller/consumer.py

@@ -1,5 +1,6 @@
-#  Copyright © 2020 Ingram Micro Inc. All rights reserved.
+#  Copyright © 2021 Ingram Micro Inc. All rights reserved.
 
+import copy
 import logging
 from contextlib import ExitStack
 
@@ -16,6 +17,7 @@ def consume(payload):
 
     :param dj_cqrs.dataclasses.TransportPayload payload: Consumed payload from master service.
     """
+    payload = copy.deepcopy(payload)
     return route_signal_to_replica_model(
         payload.signal_type, payload.cqrs_id, payload.instance_data,
         previous_data=payload.previous_data,

dj_cqrs/transport/rabbit_mq.py

@@ -196,6 +196,8 @@ def _add_to_dead_letter_queue(cls, channel, payload):
     @classmethod
     def _requeue_message(cls, channel, delivery_tag, payload):
         payload.retries += 1
+        payload.is_requeue = True
+
         cls.produce(payload)
         cls._nack(channel, delivery_tag)
         cls.log_requeued(payload)
@@ -230,6 +232,9 @@ def _get_produced_message_routing_key(cls, payload):
         elif getattr(payload, 'is_dead_letter', False):
             dead_letter_queue_name = cls._get_consumer_settings()[-1]
             routing_key = 'cqrs.{}.{}'.format(dead_letter_queue_name, routing_key)
+        elif getattr(payload, 'is_requeue', False):
+            queue = cls._get_consumer_settings()[0]
+            routing_key = 'cqrs.{}.{}'.format(queue, routing_key)
 
         return routing_key
 
@@ -249,7 +254,7 @@ def _get_consumer_rmq_objects(
         for cqrs_id, replica_model in ReplicaRegistry.models.items():
             channel.queue_bind(exchange=exchange, queue=queue_name, routing_key=cqrs_id)
 
-            # Every service must have specific SYNC routes
+            # Every service must have specific SYNC or requeue routes
             channel.queue_bind(
                 exchange=exchange,
                 queue=queue_name,

integration_tests/tests/conftest.py

@@ -37,7 +37,7 @@ def clean_rabbit_transport_connection():
 @pytest.fixture
 def replica_channel(settings):
     if current_transport is not RabbitMQTransport:
-        pytest.skip("Dead letter queue is implemented only for RabbitMQTransport.")
+        pytest.skip("Replica channel is implemented only for RabbitMQTransport.")
 
     connection = BlockingConnection(
         parameters=URLParameters(settings.CQRS['url']),

tests/test_controller.py

@@ -32,6 +32,31 @@ def test_consumer(mocker):
     factory_mock.assert_called_once_with('a', 'b', {}, previous_data={'e': 'f'})
 
 
+def test_changed_payload_data_during_consume(mocker):
+    def change_data(*args, **kwargs):
+        instance_data = args[2]
+        instance_data['instance_key'] = 'changed instance'
+        kwargs['previous_data']['previous_key'] = 'changed previous'
+
+    factory_mock = mocker.patch(
+        'dj_cqrs.controller.consumer.route_signal_to_replica_model',
+        side_effect=change_data,
+    )
+
+    payload = TransportPayload(
+        SignalType.SAVE,
+        cqrs_id='b',
+        instance_data={'instance_key': 'initial instance'},
+        instance_pk='c',
+        previous_data={'previous_key': 'initial previous'},
+    )
+    consume(payload)
+
+    assert factory_mock.call_count == 1
+    assert payload.instance_data == {'instance_key': 'initial instance'}
+    assert payload.previous_data == {'previous_key': 'initial previous'}
+
+
 @pytest.mark.django_db(transaction=True)
 def test_route_signal_to_replica_model_with_db(django_assert_num_queries):
     with django_assert_num_queries(1):

tests/test_transport/test_rabbit_mq.py

@@ -464,6 +464,16 @@ def test_get_produced_message_routing_key_dead_letter(settings):
     assert routing_key == 'cqrs.dead_letter_replica.CQRS_ID'
 
 
+def test_get_produced_message_routing_key_requeue(settings):
+    settings.CQRS['queue'] = 'replica'
+    payload = TransportPayload(SignalType.SAVE, 'CQRS_ID', {}, None)
+    payload.is_requeue = True
+
+    routing_key = PublicRabbitMQTransport.get_produced_message_routing_key(payload)
+
+    assert routing_key == 'cqrs.replica.CQRS_ID'
+
+
 def test_process_delay_messages(mocker, caplog):
     channel = mocker.MagicMock()
     produce = mocker.patch('dj_cqrs.transport.rabbit_mq.RabbitMQTransport.produce')
@@ -483,6 +493,7 @@ def test_process_delay_messages(mocker, caplog):
     produce_payload = produce.call_args[0][0]
     assert produce_payload is payload
     assert produce_payload.retries == 1
+    assert getattr(produce_payload, 'is_requeue', False)
 
     assert 'CQRS is requeued: pk = 1 (CQRS_ID)' in caplog.text