Method before_task_publish_handler

t/integration/test_canvas.py:861–886

We want to republish t1 to ensure that the chain is executed twice

(sender=None, body=None, exchange=None, routing_key=None, headers=None,
                                        properties=None,
                                        declare=None, retry_policy=None, **kwargs)

        # Republish t1 to cause the chain to be executed twice
        @before_task_publish.connect
        def before_task_publish_handler(sender=None, body=None, exchange=None, routing_key=None, headers=None,
                                        properties=None,
                                        declare=None, retry_policy=None, **kwargs):
            """ We want to republish t1 to ensure that the chain is executed twice """

            metadata = {
                'body': body,
                'exchange': exchange,
                'routing_key': routing_key,
                'properties': properties,
                'headers': headers,
            }

            with celery_session_app.producer_pool.acquire(block=True) as producer:
                # Publish t1 to the message broker, just before it's going to be published which causes duplication
                return producer.publish(
                    metadata['body'],
                    exchange=metadata['exchange'],
                    routing_key=metadata['routing_key'],
                    retry=None,
                    retry_policy=retry_policy,
                    serializer='json',
                    delivery_mode=None,
                    headers=headers,
                    **kwargs
                )

        # Clean redis key
        redis_connection = get_redis_connection()