Spring AMQP

https://docs.spring.io/spring-amqp/reference/
Admin
Spring AMQP applies core Spring concepts to the development of AMQP-based messaging solutions,...
Tokens:123,553
Snippets:1,221
Trust Score:10
Update:1 month ago
Show doc for...
Tokens:
Raw
### Basic SimpleMessageListenerContainer Java setup

Source: https://docs.spring.io/spring-amqp/reference/amqp/receiving-messages/async-consumer.html

Creates a SimpleMessageListenerContainer programmatically, setting the connection factory, queue name, and a MessageListenerAdapter. This is the most basic example for consuming messages from a queue.

```java
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
container.setConnectionFactory(rabbitConnectionFactory);
container.setQueueNames("some.queue");
container.setMessageListener(new MessageListenerAdapter(somePojo));
```

--------------------------------

### Create a CachingConnectionFactory connection

Source: https://docs.spring.io/spring-amqp/reference/amqp/connections.html

Creates a CachingConnectionFactory with a hostname, sets the username and password, and creates a connection. The example shows the basic setup for obtaining a connection from the factory.

```java
CachingConnectionFactory connectionFactory = new CachingConnectionFactory("somehost");
connectionFactory.setUsername("guest");
connectionFactory.setPassword("guest");

Connection connection = connectionFactory.createConnection();
```

--------------------------------

### Import statements for Spring AMQP examples

Source: https://docs.spring.io/spring-amqp/reference/introduction/quick-tour.html

Required imports for the imperative Java example that follows.

```java
import org.springframework.amqp.core.AmqpAdmin;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
```

--------------------------------

### Get RabbitMQ queue info with WebClient

Source: https://docs.spring.io/spring-amqp/reference/amqp/management-rest-api.html

Example of using WebClient to retrieve queue information from the RabbitMQ management REST API. It creates a client with basic authentication, builds a URI for the queue, and blocks for up to 10 seconds for the response.

```java
public Map<String, Object> queueInfo(String queueName) throws URISyntaxException {
		WebClient client = createClient("admin", "admin");
		URI uri = queueUri(queueName);
		return client.get()
				.uri(uri)
				.accept(MediaType.APPLICATION_JSON)
				.retrieve()
				.bodyToMono(new ParameterizedTypeReference<Map<String, Object>>() {
				})
				.block(Duration.ofSeconds(10));
	}

	private URI queueUri(String queue) throws URISyntaxException {
		URI uri = new URI("http://localhost:15672/api/")
				.resolve("/api/queues/" + UriUtils.encodePathSegment("/", StandardCharsets.UTF_8) + "/" + queue);
		return uri;
	}

	private WebClient createClient(String adminUser, String adminPassword) {
		return WebClient.builder()
				.filter(ExchangeFilterFunctions.basicAuthentication(adminUser, adminPassword))
				.build();
	}
```

--------------------------------

### Listener method with Channel and Message parameters

Source: https://docs.spring.io/spring-amqp/reference/amqp/receiving-messages/async-consumer.html

Example listener method that receives the extracted message, Channel, and Message, as returned by buildListenerArguments.

```java
public void handleMessage(Object object, Channel channel, Message message) throws IOException {
    ...
}
```

--------------------------------

### XML configuration example for Spring AMQP

Source: https://docs.spring.io/spring-amqp/reference/introduction/quick-tour.html

Externalizes resource configuration to XML. The Java code loads the context and uses AmqpTemplate to send and receive. The XML defines connection factory, template, admin, and queue.

```java
ApplicationContext context =
    new GenericXmlApplicationContext("classpath:/rabbit-context.xml");
AmqpTemplate template = context.getBean(AmqpTemplate.class);
template.convertAndSend("myqueue", "foo");
String foo = (String) template.receiveAndConvert("myqueue");
```

```xml
<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xmlns:rabbit="http://www.springframework.org/schema/rabbit"
       xsi:schemaLocation="http://www.springframework.org/schema/rabbit
           https://www.springframework.org/schema/rabbit/spring-rabbit.xsd
           http://www.springframework.org/schema/beans
           https://www.springframework.org/schema/beans/spring-beans.xsd">

    <rabbit:connection-factory id="connectionFactory"/>

    <rabbit:template id="amqpTemplate" connection-factory="connectionFactory"/>

    <rabbit:admin connection-factory="connectionFactory"/>

    <rabbit:queue name="myqueue"/>

</beans>
```

--------------------------------

### Customize NodeLocator with setNodeLocator

Source: https://docs.spring.io/spring-amqp/reference/amqp/connections.html

Shows how to implement a custom NodeLocator by overriding createClient, restCall, and optionally close methods. The example uses MyClient and includes placeholder logic.

```java
lqcf.setNodeLocator(new NodeLocator<MyClient>() {

    @Override
    public MyClient createClient(String userName, String password) {
        ...
    }

    @Override
    public Map<String, Object> restCall(MyClient client, String baseUri, String vhost, String queue) throws URISyntaxException {
        ...
    }

});
```

--------------------------------

### Logback AmqpAppender configuration

Source: https://docs.spring.io/spring-amqp/reference/logging.html

Configure a Logback appender to publish log messages to RabbitMQ using the `AmqpAppender` class. The example sets common properties such as `addresses`, `routingKeyPattern`, and `deliveryMode`.

```xml
<appender name="AMQP" class="org.springframework.amqp.rabbit.logback.AmqpAppender">
    <layout>
        <pattern><![CDATA[ %d %p %t [%c] - <%m>%n ]]></pattern>
    </layout>
    <addresses>foo:5672,bar:5672</addresses>
    <abbreviation>36</abbreviation>
    <includeCallerData>false</includeCallerData>
    <applicationId>myApplication</applicationId>
    <routingKeyPattern>%property{applicationId}.%c.%p</routingKeyPattern>
    <generateId>true</generateId>
    <charset>UTF-8</charset>
    <durable>false</durable>
    <deliveryMode>NON_PERSISTENT</deliveryMode>
    <declareExchange>true</declareExchange>
    <addMdcAsHeaders>false</addMdcAsHeaders>
</appender>
```

--------------------------------

### Define RabbitAmqpAdmin bean

Source: https://docs.spring.io/spring-amqp/reference/rabbitmq-amqp-client.html

Add this bean to manage RabbitMQ topology (exchanges, queues, bindings) over AMQP 1.0. It scans for bean definitions in its start() callback.

```java
@Bean
RabbitAmqpAdmin admin(AmqpConnectionFactory connectionFactory) {
    return new RabbitAmqpAdmin(connectionFactory);
}

```

--------------------------------

### Logging ack and nack confirms

Source: https://docs.spring.io/spring-amqp/reference/amqp/template.html

Example of using invoke with lambda callbacks to log ack and nack confirmations. The callbacks receive the delivery tag and multiple flag.

```java
Collection<?> messages = getMessagesToSend();
Boolean result = this.template.invoke(t -> {
    messages.forEach(m -> t.convertAndSend(ROUTE, m));
    t.waitForConfirmsOrDie(10_000);
    return true;
}, (tag, multiple) -> {
        log.info("Ack: " + tag + ":" + multiple);
}, (tag, multiple) -> {
        log.info("Nack: " + tag + ":" + multiple);
}));

```

--------------------------------

### Defer queue declaration with auto-startup=false admin

Source: https://docs.spring.io/spring-amqp/reference/amqp/receiving-messages/async-consumer.html

Declares a queue, exchange, and listener container with auto-startup disabled so elements are not declared during context initialization. The container uses the admin to declare elements when started.

```xml
<rabbit:queue id="otherAnon" declared-by="containerAdmin" />

<rabbit:direct-exchange name="otherExchange" auto-delete="true" declared-by="containerAdmin">
    <rabbit:bindings>
        <rabbit:binding queue="otherAnon" key="otherAnon" />
    </rabbit:bindings>
</rabbit:direct-exchange>

<rabbit:listener-container id="container2" auto-startup="false">
    <rabbit:listener id="listener2" ref="foo" queues="otherAnon" admin="containerAdmin" />
</rabbit:listener-container>

<rabbit:admin id="containerAdmin" connection-factory="rabbitConnectionFactory"
    auto-startup="false" />
```

--------------------------------

### Manual acknowledgment using Channel header

Source: https://docs.spring.io/spring-amqp/reference/amqp/receiving-messages/async-annotation-driven/error-handling.html

Shows how to access the Channel from the message headers to reject a failed message when using AcknowledgeMode.MANUAL. The example uses basicReject with the delivery tag and requeue flag.

```java
public Object handleError(Message amqpMessage, org.springframework.messaging.Message<?> message,
          ListenerExecutionFailedException exception) {
              ...
              message.getHeaders().get(AmqpHeaders.CHANNEL, Channel.class)
                  .basicReject(message.getHeaders().get(AmqpHeaders.DELIVERY_TAG, Long.class),
                               true);
          }
```

--------------------------------

### Configure RabbitTemplate with routing connection factory and selector expression

Source: https://docs.spring.io/spring-amqp/reference/amqp/connections.html

Configures a `RabbitTemplate` with a `SimpleRoutingConnectionFactory` that routes based on the `x-use-publisher-confirms` header. Uses a caching connection factory for confirms and a pooled channel connection factory otherwise. Requires consistent confirms/returns settings unless disabled.

```java
@Bean
public RabbitTemplate rabbitTemplate() {
    final com.rabbitmq.client.ConnectionFactory cf = new com.rabbitmq.client.ConnectionFactory();
    cf.setHost("localhost");
    cf.setPort(5672);

    CachingConnectionFactory cachingConnectionFactory = new CachingConnectionFactory(cf);
    cachingConnectionFactory.setPublisherConfirmType(CachingConnectionFactory.ConfirmType.CORRELATED);

    PooledChannelConnectionFactory pooledChannelConnectionFactory = new PooledChannelConnectionFactory(cf);

    final Map<Object, ConnectionFactory> connectionFactoryMap = new HashMap<>(2);
    connectionFactoryMap.put("true", cachingConnectionFactory);
    connectionFactoryMap.put("false", pooledChannelConnectionFactory);

    final AbstractRoutingConnectionFactory routingConnectionFactory = new SimpleRoutingConnectionFactory();
    routingConnectionFactory.setConsistentConfirmsReturns(false);
    routingConnectionFactory.setDefaultTargetConnectionFactory(pooledChannelConnectionFactory);
    routingConnectionFactory.setTargetConnectionFactories(connectionFactoryMap);

    final RabbitTemplate rabbitTemplate = new RabbitTemplate(routingConnectionFactory);

    final Expression sendExpression = new SpelExpressionParser().parseExpression(
            "messageProperties.headers['x-use-publisher-confirms'] ?: false");
    rabbitTemplate.setSendConnectionFactorySelectorExpression(sendExpression);
}
```

--------------------------------

### Spring Boot Application with ThreadChannelConnectionFactory

Source: https://docs.spring.io/spring-amqp/reference/amqp/template.html

Complete example of a Spring Boot application that uses ThreadChannelConnectionFactory to maintain message ordering when publishing from multiple threads. The Application class defines the main method, a TaskExecutor bean, a ThreadChannelConnectionFactory bean, a RabbitListener, a Queue bean, and an ApplicationRunner that triggers the service. The Service class uses the RabbitTemplate to publish messages and switches the thread context using prepareSwitchContext and switchContext methods. Note that after calling prepareSwitchContext, the current thread's channel is transferred, and the thread-bound channel must be closed when no longer needed.

```java
@SpringBootApplication
public class Application {

	private static final Logger log = LoggerFactory.getLogger(Application.class);

	public static void main(String[] args) {
		SpringApplication.run(Application.class, args);
	}

	@Bean
	TaskExecutor exec() {
		ThreadPoolTaskExecutor exec = new ThreadPoolTaskExecutor();
		exec.setCorePoolSize(10);
		return exec;
	}

	@Bean
	ThreadChannelConnectionFactory tccf() {
		ConnectionFactory rabbitConnectionFactory = new ConnectionFactory();
		rabbitConnectionFactory.setHost("localhost");
		return new ThreadChannelConnectionFactory(rabbitConnectionFactory);
	}

	@RabbitListener(queues = "queue")
	void listen(String in) {
		log.info(in);
	}

	@Bean
	Queue queue() {
		return new Queue("queue");
	}


	@Bean
	public ApplicationRunner runner(Service service, TaskExecutor exec) {
		return args -> {
			exec.execute(() -> service.mainService("test"));
		};
	}

}

@Component
class Service {

	private static final Logger LOG = LoggerFactory.getLogger(Service.class);

	private final RabbitTemplate template;

	private final TaskExecutor exec;

	private final ThreadChannelConnectionFactory connFactory;

	Service(RabbitTemplate template, TaskExecutor exec,
			ThreadChannelConnectionFactory tccf) {

		this.template = template;
		this.exec = exec;
		this.connFactory = tccf;
	}

	void mainService(String toSend) {
		LOG.info("Publishing from main service");
		this.template.convertAndSend("queue", toSend);
		Object context = this.connFactory.prepareSwitchContext();
		this.exec.execute(() -> secondaryService(toSend.toUpperCase(), context));
	}

	void secondaryService(String toSend, Object threadContext) {
		LOG.info("Publishing from secondary service");
		this.connFactory.switchContext(threadContext);
		this.template.convertAndSend("queue", toSend);
		this.connFactory.closeThreadChannel();
	}

}
```

--------------------------------

### Using @RabbitListenerTest to spy on RabbitListener methods

Source: https://docs.spring.io/spring-amqp/reference/testing.html

Annotate a @Configuration class with @RabbitListenerTest to enable RabbitListenerTestHarness. The example shows a listener with two @RabbitListener methods (foo and bar) and a test that uses the harness to get spies, stub answers, and verify invocations. Note that final methods cannot be spied or advised, and only listeners with an id attribute can be spied or advised.

```java
@Configuration
@RabbitListenerTest
public class Config {

    @Bean
    public Listener listener() {
        return new Listener();
    }

    ...

}

public class Listener {

    @RabbitListener(id="foo", queues="#{queue1.name}")
    public String foo(String foo) {
        return foo.toUpperCase();
    }

    @RabbitListener(id="bar", queues="#{queue2.name}")
    public void foo(@Payload String foo, @Header("amqp_receivedRoutingKey") String rk) {
        ...
    }

}

@SpringJUnitConfig
public class MyTests {

    @Autowired
    private RabbitListenerTestHarness harness; __**(1)**

    @Test
    public void testTwoWay() throws Exception {
        assertEquals("FOO", this.rabbitTemplate.convertSendAndReceive(this.queue1.getName(), "foo"));

        Listener listener = this.harness.getSpy("foo"); __**(2)**
        assertNotNull(listener);
        verify(listener).foo("foo");
    }

    @Test
    public void testOneWay() throws Exception {
        Listener listener = this.harness.getSpy("bar");
        assertNotNull(listener);

        LatchCountDownAndCallRealMethodAnswer answer = this.harness.getLatchAnswerFor("bar", 2); __**(3)**
        doAnswer(answer).when(listener).foo(anyString(), anyString()); __**(4)**

        this.rabbitTemplate.convertAndSend(this.queue2.getName(), "bar");
        this.rabbitTemplate.convertAndSend(this.queue2.getName(), "baz");

        assertTrue(answer.await(10));
        verify(listener).foo("bar", this.queue2.getName());
        verify(listener).foo("baz", this.queue2.getName());
    }

}
Copied!

```

--------------------------------

### Multi-broker configuration with SimpleRoutingConnectionFactory and RabbitListener containers

Source: https://docs.spring.io/spring-amqp/reference/amqp/multi-rabbit.html

Configures three CachingConnectionFactory beans (cf1, cf2, cf3) for different hosts, a SimpleRoutingConnectionFactory that routes between them, and RabbitAdmin beans for each. Also defines RabbitListenerContainerFactory beans (factory1, factory2, factory3) and a RabbitTemplate using the routing factory. The Listeners component uses @RabbitListener with queuesToDeclare and containerFactory attributes to associate each listener with a specific broker.

```java
@SpringBootApplication(exclude = RabbitAutoConfiguration.class)
public class Application {

    public static void main(String[] args) {
        SpringApplication.run(Application.class, args);
    }

    @Bean
    CachingConnectionFactory cf1() {
        return new CachingConnectionFactory("localhost");
    }

    @Bean
    CachingConnectionFactory cf2() {
        return new CachingConnectionFactory("otherHost");
    }

    @Bean
    CachingConnectionFactory cf3() {
        return new CachingConnectionFactory("thirdHost");
    }

    @Bean
    SimpleRoutingConnectionFactory rcf(CachingConnectionFactory cf1,
            CachingConnectionFactory cf2, CachingConnectionFactory cf3) {

        SimpleRoutingConnectionFactory rcf = new SimpleRoutingConnectionFactory();
        rcf.setDefaultTargetConnectionFactory(cf1);
        rcf.setTargetConnectionFactories(Map.of("one", cf1, "two", cf2, "three", cf3));
        return rcf;
    }

    @Bean("factory1-admin")
    RabbitAdmin admin1(CachingConnectionFactory cf1) {
        return new RabbitAdmin(cf1);
    }

    @Bean("factory2-admin")
    RabbitAdmin admin2(CachingConnectionFactory cf2) {
        return new RabbitAdmin(cf2);
    }

    @Bean("factory3-admin")
    RabbitAdmin admin3(CachingConnectionFactory cf3) {
        return new RabbitAdmin(cf3);
    }

    @Bean
    public RabbitListenerEndpointRegistry rabbitListenerEndpointRegistry() {
        return new RabbitListenerEndpointRegistry();
    }

    @Bean
    public RabbitListenerAnnotationBeanPostProcessor postProcessor(RabbitListenerEndpointRegistry registry) {
        MultiRabbitListenerAnnotationBeanPostProcessor postProcessor
                = new MultiRabbitListenerAnnotationBeanPostProcessor();
        postProcessor.setEndpointRegistry(registry);
        postProcessor.setContainerFactoryBeanName("defaultContainerFactory");
        return postProcessor;
    }

    @Bean
    public SimpleRabbitListenerContainerFactory factory1(CachingConnectionFactory cf1) {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(cf1);
        return factory;
    }

    @Bean
    public SimpleRabbitListenerContainerFactory factory2(CachingConnectionFactory cf2) {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(cf2);
        return factory;
    }

    @Bean
    public SimpleRabbitListenerContainerFactory factory3(CachingConnectionFactory cf3) {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(cf3);
        return factory;
    }

    @Bean
    RabbitTemplate template(SimpleRoutingConnectionFactory rcf) {
        return new RabbitTemplate(rcf);
    }

    @Bean
    ConnectionFactoryContextWrapper wrapper(SimpleRoutingConnectionFactory rcf) {
        return new ConnectionFactoryContextWrapper(rcf);
    }

}

@Component
class Listeners {

    @RabbitListener(queuesToDeclare = @Queue("q1"), containerFactory = "factory1")
    public void listen1(String in) {

    }

    @RabbitListener(queuesToDeclare = @Queue("q2"), containerFactory = "factory2")
    public void listen2(String in) {

    }

    @RabbitListener(queuesToDeclare = @Queue("q3"), containerFactory = "factory3")
    public void listen3(String in) {

    }

}

```

--------------------------------

### AsyncRabbitTemplate convertSendAndReceive with get() and whenComplete()

Source: https://docs.spring.io/spring-amqp/reference/amqp/request-reply.html

Shows two approaches for using AsyncRabbitTemplate: synchronously retrieving the result with future.get() and registering an asynchronous callback with whenComplete(). The get() method can throw ExecutionException, and whenComplete() receives a result and exception. The template must be autowired.

```java
@Autowired
private AsyncRabbitTemplate template;

...

public void doSomeWorkAndGetResultLater() {

    ...

    CompletableFuture<String> future = this.template.convertSendAndReceive("foo");

    // do some more work

    String reply = null;
    try {
        reply = future.get(10, TimeUnit.SECONDS);
    }
    catch (ExecutionException e) {
        ...
    }

    ...

}

public void doSomeWorkAndGetResultAsync() {

    ...

    RabbitConverterFuture<String> future = this.template.convertSendAndReceive("foo");
    future.whenComplete((result, ex) -> {
        if (ex == null) {
            // success
        }
        else {
            // failure
        }
    });

    ...

}
Copied!
```

--------------------------------

### Configuration and listener with @RabbitListenerTest and capture advice

Source: https://docs.spring.io/spring-amqp/reference/testing.html

Defines a configuration class with @RabbitListenerTest(spy = false, capture = true) and a listener service with two @RabbitListener methods. The listener throws a RuntimeException when the payload is 'ex' after a failed flag. This setup enables capturing invocation data for testing.

```java
@Configuration
@ComponentScan
@RabbitListenerTest(spy = false, capture = true)
public class Config {

}

@Service
public class Listener {

    private boolean failed;

    @RabbitListener(id="foo", queues="#{queue1.name}")
    public String foo(String foo) {
        return foo.toUpperCase();
    }

    @RabbitListener(id="bar", queues="#{queue2.name}")
    public void foo(@Payload String foo, @Header("amqp_receivedRoutingKey") String rk) {
        if (!failed && foo.equals("ex")) {
            failed = true;
            throw new RuntimeException(foo);
        }
        failed = false;
    }

}
```

--------------------------------

### Add spring-rabbit-stream Gradle dependency

Source: https://docs.spring.io/spring-amqp/reference/stream.html

Declares the spring-rabbit-stream dependency in a Gradle build. Version 4.1.1 is specified in the example.

```groovy
implementation 'org.springframework.amqp:spring-rabbit-stream:4.1.1'
```

--------------------------------

### Add spring-rabbit-stream Maven dependency

Source: https://docs.spring.io/spring-amqp/reference/stream.html

Declares the spring-rabbit-stream dependency in a Maven pom.xml. Version 4.1.1 is specified in the example.

```xml
<dependency>
  <groupId>org.springframework.amqp</groupId>
  <artifactId>spring-rabbit-stream</artifactId>
  <version>4.1.1</version>
</dependency>
```

--------------------------------

### Configure exchange, queue, binding and test default operations

Source: https://docs.spring.io/spring-amqp/reference/rabbitmq-amqp-client.html

Shows how to declare a DirectExchange, a durable Queue with dead-letter exchange, and a Binding, then uses RabbitAmqpTemplate defaults to send and receive a message. The test asserts successful send and receive within 10 seconds.

```java
@Bean
DirectExchange e1() {
    return new DirectExchange("e1");
}

@Bean
Queue q1() {
    return QueueBuilder.durable("q1").deadLetterExchange("dlx1").build();
}

@Bean
Binding b1() {
    return BindingBuilder.bind(q1()).to(e1()).with("k1");
}

...

@Test
void defaultExchangeAndRoutingKey() {
    this.rabbitAmqpTemplate.setExchange("e1");
    this.rabbitAmqpTemplate.setRoutingKey("k1");
	this.rabbitAmqpTemplate.setReceiveQueue("q1");

    assertThat(this.rabbitAmqpTemplate.convertAndSend("test1"))
            .succeedsWithin(Duration.ofSeconds(10));

    assertThat(this.rabbitAmqpTemplate.receiveAndConvert())
            .succeedsWithin(Duration.ofSeconds(10))
            .isEqualTo("test1");
}

```

--------------------------------

### Gradle dependency for spring-rabbit

Source: https://docs.spring.io/spring-amqp/reference/introduction/quick-tour.html

Declare the spring-rabbit dependency in your Gradle build to get the Spring AMQP libraries and their dependencies.

```groovy
implementation 'org.springframework.amqp:spring-rabbit:4.1.1'
```

--------------------------------

### Maven dependency for spring-rabbit

Source: https://docs.spring.io/spring-amqp/reference/introduction/quick-tour.html

Declare the spring-rabbit dependency in your Maven build to get the Spring AMQP libraries and their dependencies.

```xml
<dependency>
  <groupId>org.springframework.amqp</groupId>
  <artifactId>spring-rabbit</artifactId>
  <version>4.1.1</version>
</dependency>
```

--------------------------------

### Consumer main method

Source: https://docs.spring.io/spring-amqp/reference/sample-apps.html

Main method for the Consumer class. It creates an ApplicationContext, retrieves the AmqpTemplate bean, and calls receiveAndConvert() to receive a message, printing the result. Expect to see 'Received: Hello World' in the console when run after the Producer.

```java
public static void main(String[] args) {
    ApplicationContext context =
        new AnnotationConfigApplicationContext(RabbitConfiguration.class);
    AmqpTemplate amqpTemplate = context.getBean(AmqpTemplate.class);
    System.out.println("Received: " + amqpTemplate.receiveAndConvert());
}
```

--------------------------------

### Custom BlockingQueue for Logback AmqpAppender

Source: https://docs.spring.io/spring-amqp/reference/logging.html

Provides a custom BlockingQueue implementation for the Logback AmqpAppender by overriding createEventQueue(). The example returns an ArrayBlockingQueue.

```java
public class MyEnhancedAppender extends AmqpAppender {

    @Override
    protected BlockingQueue<Event> createEventQueue() {
        return new ArrayBlockingQueue();
    }

}
```

--------------------------------

### Configure MessageListenerAdapter with lambda

Source: https://docs.spring.io/spring-amqp/reference/amqp/receiving-messages/async-consumer.html

Shows how to use a lambda with ReplyingMessageListener to configure the adapter, with placeholder for message processing logic.

```java
new MessageListenerAdapter((ReplyingMessageListener<String, String>) data -> {
    ...
    return result;
}));
```

--------------------------------

### Configure inbound-channel-adapter to receive AMQP messages

Source: https://docs.spring.io/spring-amqp/reference/integration-reference.html

Use this XML snippet to configure an inbound channel adapter that receives messages from a queue. It requires a channel, queue-names, and a connection-factory. The example uses 'fromAMQP' channel, 'some.queue' queue, and 'rabbitConnectionFactory'.

```xml
<amqp:inbound-channel-adapter channel="fromAMQP"
                              queue-names="some.queue"
                              connection-factory="rabbitConnectionFactory"/>
```

--------------------------------

### Create RabbitAmqpTemplate bean

Source: https://docs.spring.io/spring-amqp/reference/rabbitmq-amqp-client.html

Defines a @Bean method that returns a RabbitAmqpTemplate, requiring an AmqpConnectionFactory. This is the typical minimal setup for using the template.

```java
@Bean
RabbitAmqpTemplate rabbitTemplate(AmqpConnectionFactory connectionFactory) {
    return new RabbitAmqpTemplate(connectionFactory);
}

```

--------------------------------

### RPC with RabbitAmqpTemplate using receiveAndReply

Source: https://docs.spring.io/spring-amqp/reference/rabbitmq-amqp-client.html

Demonstrates an RPC client-server pattern. The client sends a request and waits for a reply; the server uses receiveAndReply to process the request and return a response. Asserts both results succeed within 10 seconds.

```java
@Test
void verifyRpc() {
    String testRequest = "rpc-request";
    String testReply = "rpc-reply";

    CompletableFuture<Object> rpcClientResult = this.template.convertSendAndReceive("e1", "k1", testRequest);

    AtomicReference<String> receivedRequest = new AtomicReference<>();
    CompletableFuture<Boolean> rpcServerResult =
            this.rabbitAmqpTemplate.<String, String>receiveAndReply("q1",
                     payload -> {
                         receivedRequest.set(payload);
                         return testReply;
                     });

    assertThat(rpcServerResult).succeedsWithin(Duration.ofSeconds(10)).isEqualTo(true);
    assertThat(rpcClientResult).succeedsWithin(Duration.ofSeconds(10)).isEqualTo(testReply);
    assertThat(receivedRequest.get()).isEqualTo(testRequest);
}

```

--------------------------------

### Send with template exchange property

Source: https://docs.spring.io/spring-amqp/reference/amqp/sending-messages.html

Sets the exchange property on the template and uses the two-argument send overload. Functionally equivalent to the previous example.

```java
amqpTemplate.setExchange("marketData.topic");
amqpTemplate.send("quotes.nasdaq.FOO", new Message("12.34".getBytes(), someProperties));
```

--------------------------------

### Configure SimpleRoutingConnectionFactory in XML

Source: https://docs.spring.io/spring-amqp/reference/amqp/connections.html

Configures a SimpleRoutingConnectionFactory with target connection factories mapped by virtual host. The rabbit template references this factory.

```xml
<bean id="connectionFactory"
      class="org.springframework.amqp.rabbit.connection.SimpleRoutingConnectionFactory">
    <property name="targetConnectionFactories">
        <map>
            <entry key="#{connectionFactory1.virtualHost}" ref="connectionFactory1"/>
            <entry key="#{connectionFactory2.virtualHost}" ref="connectionFactory2"/>
        </map>
    </property>
</bean>

<rabbit:template id="template" connection-factory="connectionFactory" />

```

--------------------------------

### SimpleMessageListenerContainer bean definition

Source: https://docs.spring.io/spring-amqp/reference/sample-apps.html

Configures a SimpleMessageListenerContainer with a connection factory, queue name, and a MessageListenerAdapter wrapping the HelloWorldHandler. The container starts automatically by default.

```java
@Bean
public SimpleMessageListenerContainer listenerContainer() {
    SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
    container.setConnectionFactory(connectionFactory());
    container.setQueueName(this.helloWorldQueueName);
    container.setMessageListener(new MessageListenerAdapter(new HelloWorldHandler()));
    return container;
}
```

--------------------------------

### TestRabbitTemplate configuration with @Configuration and @EnableRabbit

Source: https://docs.spring.io/spring-amqp/reference/testing.html

Configuration class defining beans for TestRabbitTemplate, a mocked ConnectionFactory, and a SimpleMessageListenerContainer. It also declares @RabbitListener methods for queues 'foo', 'bar', and 'baz'. The mocked connection factory returns mocked channels to avoid a real broker.

```java
@Configuration
@EnableRabbit
public static class Config {

    public String fooIn = "";

    public String barIn = "";

    public String smlc1In = "smlc1:";

    @Bean
    public TestRabbitTemplate template() throws IOException {
        return new TestRabbitTemplate(connectionFactory());
    }

    @Bean
    public ConnectionFactory connectionFactory() throws IOException {
        ConnectionFactory factory = mock(ConnectionFactory.class);
        Connection connection = mock(Connection.class);
        Channel channel = mock(Channel.class);
        willReturn(connection).given(factory).createConnection();
        willReturn(channel).given(connection).createChannel(anyBoolean());
        given(channel.isOpen()).willReturn(true);
        return factory;
    }

    @Bean
    public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory() throws IOException {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory());
        return factory;
    }

    @RabbitListener(queues = "foo")
    public void foo(String in) {
        this.fooIn += "foo:" + in;
    }

    @RabbitListener(queues = "bar")
    public void bar(String in) {
        this.barIn += "bar:" + in;
    }

    @RabbitListener(queues = "baz")
    public String baz(String in) {
        return "baz:" + in;
    }

    @Bean
    public SimpleMessageListenerContainer smlc1() throws IOException {
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory());
        container.setQueueNames("foo", "bar");
        container.setMessageListener(new MessageListenerAdapter(new Object() {

            public void handleMessage(String in) {
                smlc1In += in;
            }

        }));
        return container;
    }

}
Copied!

```

--------------------------------

### Configure outbound-channel-adapter to send AMQP messages

Source: https://docs.spring.io/spring-amqp/reference/integration-reference.html

Use this XML snippet to configure an outbound channel adapter that sends messages to an exchange. It requires a channel, exchange-name, and amqp-template. Optionally, a routing-key can be provided. The example uses 'toAMQP' channel, 'some.exchange' exchange, 'foo' routing key, and 'rabbitTemplate'.

```xml
<amqp:outbound-channel-adapter channel="toAMQP"
                               exchange-name="some.exchange"
                               routing-key="foo"
                               amqp-template="rabbitTemplate"/>
```

--------------------------------

### Runtime SpEL expression in @SendTo

Source: https://docs.spring.io/spring-amqp/reference/amqp/receiving-messages/async-annotation-driven/reply.html

Starting with version 1.6, @SendTo can be a SpEL expression evaluated at runtime against the request and reply. The expression must evaluate to a String.

```java
@RabbitListener(queues = "test.sendTo.spel")
@SendTo("!{'some.reply.queue.with.' + result.queueName}")
public Bar capitalizeWithSendToSpel(Foo foo) {
    return processTheFooAndReturnABar(foo);
}
```
Tokens:
Raw