How to send a message to the queue created by spring cloud stream with RabbitMQ broker running in testcontainers?

Question

I would like to create an integration test for my pack and label method(pipe composed), which is based on Spring Cloud Function. I know the configured input binding is waiting for message from the order-accepted.dispatcher-service queue of order-accepted exchange, and Spring Cloud Stream will automatically sends message to order-dispatched exchange. So I only need to send message to the order-accepted.dispatcher-service queue, and check message received from order-dispatched exchange. But I don't know routing key , How am I able to send message to the order-accepted.dispatcher-service queue? Below are the configuration file and test class code that I have so far:

1. application.yml:

server:
  port: 9003

spring:
  application:
    name: dispatcher-service
  cloud:
    function:
      definition: pack|label
    stream:
      bindings:
        packlabel-in-0:
          destination: order-accepted
          group: ${spring.application.name}
        packlabel-out-0:
          destination: order-dispatched

  rabbitmq:
    host: localhost
    port: 5672
    username: user
    password: password
    connection-timeout: 5s

2. Integration test class(How should I implement this integration test code?):

import org.junit.jupiter.api.Test;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.core.ParameterizedTypeReference;
import org.springframework.test.context.DynamicPropertyRegistry;
import org.springframework.test.context.DynamicPropertySource;
import org.testcontainers.containers.RabbitMQContainer;
import org.testcontainers.junit.jupiter.Container;
import org.testcontainers.junit.jupiter.Testcontainers;
import org.testcontainers.utility.DockerImageName;

import java.util.concurrent.TimeUnit;

import static org.assertj.core.api.Assertions.assertThat;
import static org.awaitility.Awaitility.await;


@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
@Testcontainers
class DispatcherServiceApplicationTests {

    @Container
    static RabbitMQContainer rabbitMQ = new RabbitMQContainer(DockerImageName.parse("rabbitmq:3.10-management"));


    @DynamicPropertySource
    static void rabbitMQProperties(DynamicPropertyRegistry registry){
        registry.add("spring.rabbitmq.host", rabbitMQ::getHost);
        registry.add("spring.rabbitmq.port", rabbitMQ::getAmqpPort);
        registry.add("spring.rabbitmq.username", rabbitMQ::getAdminUsername);
        registry.add("spring.rabbitmq.password", rabbitMQ::getAdminPassword);
    }

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Autowired
    private RabbitAdmin rabbitAdmin;

    @Test
    void contextLoads() {
    }

    @Test
    void packAndLabel(){
        long orderId = 121;

        Queue inputQueue = this.rabbitAdmin.declareQueue();
        assert inputQueue != null;
        Binding inputBinding = new Binding(inputQueue.getName(), Binding.DestinationType.QUEUE, "order-accepted", "#", null);

        Queue outputQueue = this.rabbitAdmin.declareQueue();
        assert outputQueue != null;
      
        Binding outputBinding = new Binding(outputQueue.getName(), Binding.DestinationType.QUEUE, "order-dispatched", "#", null);

        this.rabbitAdmin.declareBinding(inputBinding);
        this.rabbitAdmin.declareBinding(outputBinding);

        // I think this only send message to order-accepted exchange rather than order-accepted.dispatcher-service queue.
        rabbitTemplate.convertAndSend("order-accepted", "#", new OrderAcceptedMessage(orderId));


        await().atMost(5, TimeUnit.SECONDS).untilAsserted(() -> {
            OrderDispatchedMessage message = rabbitTemplate.receiveAndConvert(outputQueue.getName(),
                10000, new ParameterizedTypeReference<OrderDispatchedMessage>(){});

            assert message != null;
            assertThat(message.orderId()).isEqualTo(orderId);
            System.out.println("------------------------------------: " + message.orderId());
        });
    }
}

Answer 1

We are using testcontainers ourselves so you can look at some of the tests we have in Rabbit. Here is one of them - https://github.com/spring-cloud/spring-cloud-stream/blob/8f5e7d75f23f9463ecc4f1e8372685b1c01d97d4/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/integration/RabbitBinderModuleTests.java#L99