conftest.py 文件源码

def publish(config):
    conn = Connection(config[AMQP_URI_CONFIG_KEY])

    def publish(payload, routing_key, exchange=None, **kwargs):
        """Publish an AMQP message."""
        with kombu_connections[conn].acquire(block=True) as connection:
            if exchange is not None:
                exchange.maybe_bind(connection)
            with producers[conn].acquire(block=True) as producer:
                producer.publish(
                    payload,
                    exchange=exchange,
                    serializer='json',
                    routing_key=routing_key,
                    **kwargs)

    return publish