Add AMQP 1.0 generic auto-configuration

This commit adds auto-configuration for the generic AMQP 1.0 client
in Spring AMQP 4.1, using Qpid ProtonJ. Support for connection factory,
client and listener endpoints are available.

The "spring.amqp" namespace exposes settings to connect to an AMQP 1.0
compliant broker and customize client and listeners settings. Docker
compose and testcontainers support using RabbitMQ, ActiveMQ, and Artemis
have been added too.

Closes gh-49678

buildSrc/src/main/java/org/springframework/boot/build/context/properties/DocumentConfigurationProperties.java

@@ -171,6 +171,7 @@ public abstract class DocumentConfigurationProperties extends DefaultTask {
 
 	private void integrationPrefixes(Config prefix) {
 		prefix.accept("spring.activemq");
+		prefix.accept("spring.amqp");
 		prefix.accept("spring.artemis");
 		prefix.accept("spring.batch");
 		prefix.accept("spring.integration");

documentation/spring-boot-docs/build.gradle

@@ -94,6 +94,7 @@ dependencies {
 	implementation(project(path: ":loader:spring-boot-loader-tools"))
 	implementation(project(path: ":module:spring-boot-actuator"))
 	implementation(project(path: ":module:spring-boot-actuator-autoconfigure"))
+	implementation(project(path: ":module:spring-boot-amqp"))
 	implementation(project(path: ":module:spring-boot-cache"))
 	implementation(project(path: ":module:spring-boot-cache-test"))
 	implementation(project(path: ":module:spring-boot-data-cassandra"))

documentation/spring-boot-docs/src/docs/antora/modules/ROOT/pages/redirect.adoc

@@ -1716,13 +1716,13 @@
 * xref:reference:messaging/amqp.adoc#messaging.amqp.rabbitmq[#features.messaging.amqp.rabbit]
 * xref:reference:messaging/amqp.adoc#messaging.amqp.rabbitmq[#messaging.amqp.rabbit]
 * xref:reference:messaging/amqp.adoc#messaging.amqp.rabbitmq[#messaging.amqp.rabbitmq]
-* xref:reference:messaging/amqp.adoc#messaging.amqp.receiving[#boot-features-using-amqp-receiving]
+* xref:reference:messaging/amqp.adoc#messaging.amqp.rabbitmq.receiving[#boot-features-using-amqp-receiving]
-* xref:reference:messaging/amqp.adoc#messaging.amqp.receiving[#features.messaging.amqp.receiving]
+* xref:reference:messaging/amqp.adoc#messaging.amqp.rabbitmq.receiving[#features.messaging.amqp.receiving]
-* xref:reference:messaging/amqp.adoc#messaging.amqp.receiving[#messaging.amqp.receiving]
+* xref:reference:messaging/amqp.adoc#messaging.amqp.rabbitmq.receiving[#messaging.amqp.receiving]
-* xref:reference:messaging/amqp.adoc#messaging.amqp.sending-stream[#messaging.amqp.sending-stream]
+* xref:reference:messaging/amqp.adoc#messaging.amqp.rabbitmq.sending-stream[#messaging.amqp.sending-stream]
-* xref:reference:messaging/amqp.adoc#messaging.amqp.sending[#boot-features-using-amqp-sending]
+* xref:reference:messaging/amqp.adoc#messaging.amqp.rabbitmq.sending[#boot-features-using-amqp-sending]
-* xref:reference:messaging/amqp.adoc#messaging.amqp.sending[#features.messaging.amqp.sending]
+* xref:reference:messaging/amqp.adoc#messaging.amqp.rabbitmq.sending[#features.messaging.amqp.sending]
-* xref:reference:messaging/amqp.adoc#messaging.amqp.sending[#messaging.amqp.sending]
+* xref:reference:messaging/amqp.adoc#messaging.amqp.rabbitmq.sending[#messaging.amqp.sending]
 * xref:reference:messaging/amqp.adoc#messaging.amqp[#boot-features-amqp]
 * xref:reference:messaging/amqp.adoc#messaging.amqp[#features.messaging.amqp]
 * xref:reference:messaging/amqp.adoc#messaging.amqp[#messaging.amqp]

documentation/spring-boot-docs/src/docs/antora/modules/reference/pages/features/dev-services.adoc

@@ -86,6 +86,9 @@ The following service connections are currently supported:
 | javadoc:org.springframework.boot.activemq.autoconfigure.ActiveMQConnectionDetails[]
 | Containers named "symptoma/activemq", "apache/activemq-classic" or "apache/activemq"
 
+| javadoc:org.springframework.boot.amqp.autoconfigure.AmqpConnectionDetails[]
+| Containers named "rabbitmq" or "apache/activemq-classic" or "apache/activemq-artemis"
+
 | javadoc:org.springframework.boot.artemis.autoconfigure.ArtemisConnectionDetails[]
 | Containers named "apache/activemq-artemis" or "apache/artemis"
 

documentation/spring-boot-docs/src/docs/antora/modules/reference/pages/messaging/amqp.adoc

@@ -3,7 +3,78 @@
 
 The Advanced Message Queuing Protocol (AMQP) is a platform-neutral, wire-level protocol for message-oriented middleware.
 The Spring AMQP project applies core Spring concepts to the development of AMQP-based messaging solutions.
-Spring Boot offers several conveniences for working with AMQP through RabbitMQ, including the `spring-boot-starter-rabbitmq` starter.
+Spring Boot offers several conveniences for working with AMQP: generic AMQP 1.0 support is provided by the `spring-boot-starter-amqp` starter while specific RabbitMQ support is available via the `spring-boot-starter-rabbitmq` starter.
+
+
+
+[[messaging.amqp.generic]]
+== Generic AMQP 1.0 Support
+
+AMQP 1.0 is supported by several brokers and messaging services beyond RabbitMQ, including ActiveMQ, Azure Service Bus, and others.
+Spring AMQP provides {url-spring-amqp-docs}/amqp10-client.html[generic support for AMQP 1.0] via `org.springframework.amqp:spring-amqp-client` that is based on the https://github.com/apache/qpid-protonj2/blob/main/protonj2-client/README.md[Qpid ProtonJ2 Client Library].
+
+AMQP configuration is controlled by external configuration properties in `+spring.amqp.*+`.
+For example, you might declare the following in your configuration:
+
+[configprops,yaml]
+----
+spring:
+  amqp:
+    host: "localhost"
+    port: 5672
+    username: "admin"
+    password: "secret"
+----
+
+To configure javadoc:org.apache.qpid.protonj2.client.ConnectionOptions[connection options] of the auto-configured javadoc:org.springframework.amqp.client.AmqpConnectionFactory[], define a javadoc:org.springframework.boot.amqp.autoconfigure.ConnectionOptionsCustomizer[] bean.
+
+[[messaging.amqp.generic.sending]]
+=== Sending a Message
+
+Spring's javadoc:org.springframework.amqp.client.AmqpClient[] is auto-configured, and you can autowire it directly into your own beans, as shown in the following example:
+
+include-code::MyBean[]
+
+If a javadoc:org.springframework.amqp.support.converter.MessageConverter[] bean is defined, it is associated automatically with the auto-configured javadoc:org.springframework.amqp.client.AmqpClient[].
+If no such converter is defined and Jackson is available, javadoc:org.springframework.amqp.support.converter.JacksonJsonMessageConverter[] is used.
+
+Client-specific settings can be configured as follows:
+
+[configprops,yaml]
+----
+spring:
+  amqp:
+    client:
+      default-to-address: "/queues/default_queue"
+      completion-timeout: "500ms"
+----
+
+To further configure the auto-configured javadoc:org.springframework.amqp.client.AmqpClient[], define a javadoc:org.springframework.boot.amqp.autoconfigure.AmqpClientCustomizer[] bean.
+
+
+
+[[messaging.amqp.generic.receiving]]
+=== Receiving a Message
+
+When the generic AMQP 1.0 infrastructure is present, any bean can be annotated with javadoc:org.springframework.amqp.client.annotation.AmqpListener[format=annotation] to create a listener endpoint.
+If no javadoc:org.springframework.amqp.client.config.MethodAmqpMessageListenerContainerFactory[] has been defined, a default one is automatically configured.
+If a javadoc:org.springframework.amqp.support.converter.MessageConverter[] or a javadoc:org.springframework.amqp.client.listener.AmqpListenerErrorHandler[] bean is defined, it is automatically associated with the default factory.
+
+The following sample component creates a listener endpoint on the `/queues/someQueue` address:
+
+include-code::MyBean[]
+
+TIP: See javadoc:org.springframework.amqp.client.config.EnableAmqp[format=annotation] for more details.
+
+If you need to create more javadoc:org.springframework.amqp.client.config.MethodAmqpMessageListenerContainerFactory[] instances or if you want to override the default, Spring Boot provides a javadoc:org.springframework.boot.amqp.autoconfigure.AmqpMessageListenerContainerFactoryConfigurer[] that you can use to initialize a javadoc:org.springframework.amqp.client.config.MethodAmqpMessageListenerContainerFactory[] with the same settings as the factory used by the auto-configuration.
+
+For instance, the following configuration class exposes another factory that uses a specific javadoc:org.springframework.amqp.support.converter.MessageConverter[]:
+
+include-code::custom/MyAmqpConfiguration[]
+
+Then you can use the factory in any javadoc:org.springframework.amqp.client.annotation.AmqpListener[format=annotation]-annotated method, as follows:
+
+include-code::custom/MyBean[]
 
 
 
@@ -49,8 +120,8 @@ TIP: See https://spring.io/blog/2010/06/14/understanding-amqp-the-protocol-used-
 
 
 
-[[messaging.amqp.sending]]
+[[messaging.amqp.rabbitmq.sending]]
-== Sending a Message
+=== Sending a Message
 
 Spring's javadoc:org.springframework.amqp.core.AmqpTemplate[] and javadoc:org.springframework.amqp.core.AmqpAdmin[] are auto-configured, and you can autowire them directly into your own beans, as shown in the following example:
 
@@ -82,8 +153,8 @@ If there's a bean of type javadoc:org.springframework.amqp.rabbit.support.microm
 
 
 
-[[messaging.amqp.sending-stream]]
+[[messaging.amqp.rabbitmq.sending-stream]]
-== Sending a Message To A Stream
+=== Sending a Message To A Stream
 
 To send a message to a particular stream, specify the name of the stream, as shown in the following example:
 
@@ -101,14 +172,14 @@ If you need to create more javadoc:org.springframework.rabbit.stream.producer.Ra
 
 
 
-[[messaging.amqp.sending-stream.ssl]]
+[[messaging.amqp.rabbitmq.sending-stream.ssl]]
-=== SSL
+==== SSL
 To use SSL with RabbitMQ Streams, set configprop:spring.rabbitmq.stream.ssl.enabled[] to `true` or set configprop:spring.rabbitmq.stream.ssl.bundle[] to configure the xref:features/ssl.adoc#features.ssl.bundles[SSL bundle] to use.
 
 
 
-[[messaging.amqp.receiving]]
+[[messaging.amqp.rabbitmq.receiving]]
-== Receiving a Message
+=== Receiving a Message
 
 When the Rabbit infrastructure is present, any bean can be annotated with javadoc:org.springframework.amqp.rabbit.annotation.RabbitListener[format=annotation] to create a listener endpoint.
 If no javadoc:org.springframework.amqp.rabbit.listener.RabbitListenerContainerFactory[] has been defined, a default javadoc:org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory[] is automatically configured and you can switch to a direct container using the configprop:spring.rabbitmq.listener.type[] property.

documentation/spring-boot-docs/src/docs/antora/modules/reference/pages/testing/testcontainers.adoc

@@ -128,6 +128,9 @@ The following service connection factories are provided in the `spring-boot-test
 | javadoc:org.springframework.boot.activemq.autoconfigure.ActiveMQConnectionDetails[]
 | Containers named "symptoma/activemq" or javadoc:org.testcontainers.activemq.ActiveMQContainer[]
 
+| javadoc:org.springframework.boot.amqp.autoconfigure.AmqpConnectionDetails[]
+| Containers of types javadoc:{url-testcontainers-rabbitmq-javadoc}/org.testcontainers.rabbitmq.RabbitMQContainer[] or avadoc:org.testcontainers.activemq.ActiveMQContainer[] or javadoc:org.testcontainers.activemq.ArtemisContainer[]
+
 | javadoc:org.springframework.boot.artemis.autoconfigure.ArtemisConnectionDetails[]
 | Containers of type javadoc:org.testcontainers.activemq.ArtemisContainer[]
 

documentation/spring-boot-docs/src/main/java/org/springframework/boot/docs/messaging/amqp/generic/receiving/MyBean.java

@@ -0,0 +1,30 @@
+/*
+ * Copyright 2012-present the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.boot.docs.messaging.amqp.generic.receiving;
+
+import org.springframework.amqp.client.annotation.AmqpListener;
+import org.springframework.stereotype.Component;
+
+@Component
+public class MyBean {
+
+	@AmqpListener(addresses = "/queues/someQueue")
+	public void processMessage(String content) {
+		// ...
+	}
+
+}

documentation/spring-boot-docs/src/main/java/org/springframework/boot/docs/messaging/amqp/generic/receiving/custom/MyAmqpConfiguration.java

@@ -0,0 +1,38 @@
+/*
+ * Copyright 2012-present the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.boot.docs.messaging.amqp.generic.receiving.custom;
+
+import org.springframework.amqp.client.AmqpConnectionFactory;
+import org.springframework.amqp.client.config.MethodAmqpMessageListenerContainerFactory;
+import org.springframework.boot.amqp.autoconfigure.AmqpMessageListenerContainerFactoryConfigurer;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+@Configuration(proxyBeanMethods = false)
+public class MyAmqpConfiguration {
+
+	@Bean
+	public MethodAmqpMessageListenerContainerFactory myFactory(AmqpMessageListenerContainerFactoryConfigurer configurer,
+			AmqpConnectionFactory connectionFactory) {
+		MethodAmqpMessageListenerContainerFactory factory = new MethodAmqpMessageListenerContainerFactory(
+				connectionFactory);
+		configurer.configure(factory);
+		factory.setMessageConverter(new MyMessageConverter());
+		return factory;
+	}
+
+}

documentation/spring-boot-docs/src/main/java/org/springframework/boot/docs/messaging/amqp/generic/receiving/custom/MyBean.java

@@ -0,0 +1,30 @@
+/*
+ * Copyright 2012-present the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.boot.docs.messaging.amqp.generic.receiving.custom;
+
+import org.springframework.amqp.client.annotation.AmqpListener;
+import org.springframework.stereotype.Component;
+
+@Component
+public class MyBean {
+
+	@AmqpListener(addresses = "/queues/someQueue", containerFactory = "myFactory")
+	public void processMessage(String content) {
+		// ...
+	}
+
+}

documentation/spring-boot-docs/src/main/java/org/springframework/boot/docs/messaging/amqp/generic/receiving/custom/MyMessageConverter.java

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package org.springframework.boot.docs.messaging.amqp.receiving.custom;
+package org.springframework.boot.docs.messaging.amqp.generic.receiving.custom;
 
 import org.springframework.amqp.core.Message;
 import org.springframework.amqp.core.MessageProperties;

documentation/spring-boot-docs/src/main/java/org/springframework/boot/docs/messaging/amqp/generic/sending/MyBean.java

@@ -0,0 +1,37 @@
+/*
+ * Copyright 2012-present the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.boot.docs.messaging.amqp.generic.sending;
+
+import org.springframework.amqp.client.AmqpClient;
+import org.springframework.stereotype.Component;
+
+@Component
+public class MyBean {
+
+	private final AmqpClient amqpClient;
+
+	public MyBean(AmqpClient amqpClient) {
+		this.amqpClient = amqpClient;
+	}
+
+	// @fold:on // ...
+	public void sendMessage(String msg) {
+		this.amqpClient.to("/queues/test").body(msg).send();
+	}
+	// @fold:off
+
+}

documentation/spring-boot-docs/src/main/java/org/springframework/boot/docs/messaging/amqp/rabbitmq/receiving/MyBean.java

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package org.springframework.boot.docs.messaging.amqp.receiving;
+package org.springframework.boot.docs.messaging.amqp.rabbitmq.receiving;
 
 import org.springframework.amqp.rabbit.annotation.RabbitListener;
 import org.springframework.stereotype.Component;

documentation/spring-boot-docs/src/main/java/org/springframework/boot/docs/messaging/amqp/rabbitmq/receiving/custom/MyBean.java

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package org.springframework.boot.docs.messaging.amqp.receiving.custom;
+package org.springframework.boot.docs.messaging.amqp.rabbitmq.receiving.custom;
 
 import org.springframework.amqp.rabbit.annotation.RabbitListener;
 import org.springframework.stereotype.Component;

documentation/spring-boot-docs/src/main/java/org/springframework/boot/docs/messaging/amqp/rabbitmq/receiving/custom/MyMessageConverter.java

@@ -0,0 +1,36 @@
+/*
+ * Copyright 2012-present the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.boot.docs.messaging.amqp.rabbitmq.receiving.custom;
+
+import org.springframework.amqp.core.Message;
+import org.springframework.amqp.core.MessageProperties;
+import org.springframework.amqp.support.converter.MessageConversionException;
+import org.springframework.amqp.support.converter.MessageConverter;
+
+class MyMessageConverter implements MessageConverter {
+
+	@Override
+	public Message toMessage(Object object, MessageProperties messageProperties) throws MessageConversionException {
+		return null;
+	}
+
+	@Override
+	public Object fromMessage(Message message) throws MessageConversionException {
+		return null;
+	}
+
+}

documentation/spring-boot-docs/src/main/java/org/springframework/boot/docs/messaging/amqp/rabbitmq/receiving/custom/MyRabbitConfiguration.java

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package org.springframework.boot.docs.messaging.amqp.receiving.custom;
+package org.springframework.boot.docs.messaging.amqp.rabbitmq.receiving.custom;
 
 import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
 import org.springframework.amqp.rabbit.connection.ConnectionFactory;

documentation/spring-boot-docs/src/main/java/org/springframework/boot/docs/messaging/amqp/rabbitmq/sending/MyBean.java

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package org.springframework.boot.docs.messaging.amqp.sending;
+package org.springframework.boot.docs.messaging.amqp.rabbitmq.sending;
 
 import org.springframework.amqp.core.AmqpAdmin;
 import org.springframework.amqp.core.AmqpTemplate;

documentation/spring-boot-docs/src/main/kotlin/org/springframework/boot/docs/messaging/amqp/generic/receiving/MyBean.kt

@@ -0,0 +1,31 @@
+/*
+ * Copyright 2012-present the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.boot.docs.messaging.amqp.generic.receiving
+
+import org.springframework.amqp.client.annotation.AmqpListener
+import org.springframework.stereotype.Component
+
+@Suppress("UNUSED_PARAMETER")
+@Component
+class MyBean {
+
+	@AmqpListener(addresses = ["/queues/someQueue"])
+	fun processMessage(content: String?) {
+		// ...
+	}
+
+}

documentation/spring-boot-docs/src/main/kotlin/org/springframework/boot/docs/messaging/amqp/generic/receiving/custom/MyAmqpConfiguration.kt

@@ -0,0 +1,39 @@
+/*
+ * Copyright 2012-present the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.boot.docs.messaging.amqp.generic.receiving.custom
+
+import org.springframework.amqp.client.AmqpConnectionFactory
+import org.springframework.amqp.client.config.MethodAmqpMessageListenerContainerFactory
+import org.springframework.boot.amqp.autoconfigure.AmqpMessageListenerContainerFactoryConfigurer
+import org.springframework.context.annotation.Bean
+import org.springframework.context.annotation.Configuration
+
+@Configuration(proxyBeanMethods = false)
+class MyAmqpConfiguration {
+
+	@Bean
+	fun myFactory(
+		configurer: AmqpMessageListenerContainerFactoryConfigurer,
+		connectionFactory: AmqpConnectionFactory
+	): MethodAmqpMessageListenerContainerFactory {
+		val factory = MethodAmqpMessageListenerContainerFactory(connectionFactory)
+		configurer.configure(factory)
+		factory.setMessageConverter(MyMessageConverter())
+		return factory
+	}
+
+}

documentation/spring-boot-docs/src/main/kotlin/org/springframework/boot/docs/messaging/amqp/generic/receiving/custom/MyBean.kt

@@ -0,0 +1,31 @@
+/*
+ * Copyright 2012-present the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.boot.docs.messaging.amqp.generic.receiving.custom
+
+import org.springframework.amqp.client.annotation.AmqpListener
+import org.springframework.stereotype.Component
+
+@Suppress("UNUSED_PARAMETER")
+@Component
+class MyBean {
+
+	@AmqpListener(addresses = ["/queues/someQueue"], containerFactory = "myFactory")
+	fun processMessage(content: String?) {
+		// ...
+	}
+
+}

documentation/spring-boot-docs/src/main/kotlin/org/springframework/boot/docs/messaging/amqp/generic/receiving/custom/MyMessageConverter.kt

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package org.springframework.boot.docs.messaging.amqp.receiving.custom
+package org.springframework.boot.docs.messaging.amqp.generic.receiving.custom
 
 import org.springframework.amqp.core.Message
 import org.springframework.amqp.core.MessageProperties
@@ -31,4 +31,3 @@ internal class MyMessageConverter : MessageConverter {
 	}
 
 }
-

documentation/spring-boot-docs/src/main/kotlin/org/springframework/boot/docs/messaging/amqp/generic/sending/MyBean.kt

@@ -0,0 +1,32 @@
+/*
+ * Copyright 2012-present the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.boot.docs.messaging.amqp.generic.sending
+
+import org.springframework.amqp.client.AmqpClient
+import org.springframework.stereotype.Component
+
+@Component
+class MyBean(private val amqpClient: AmqpClient) {
+
+	// @fold:on // ...
+	fun someOtherMethod(msg: String) {
+		amqpClient.to("/queues/test").body(msg).send()
+	}
+	// @fold:off
+
+}
+

documentation/spring-boot-docs/src/main/kotlin/org/springframework/boot/docs/messaging/amqp/rabbitmq/receiving/MyBean.kt

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package org.springframework.boot.docs.messaging.amqp.receiving
+package org.springframework.boot.docs.messaging.amqp.rabbitmq.receiving
 
 import org.springframework.amqp.rabbit.annotation.RabbitListener
 import org.springframework.stereotype.Component

documentation/spring-boot-docs/src/main/kotlin/org/springframework/boot/docs/messaging/amqp/rabbitmq/receiving/custom/MyBean.kt

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package org.springframework.boot.docs.messaging.amqp.receiving.custom
+package org.springframework.boot.docs.messaging.amqp.rabbitmq.receiving.custom
 
 import org.springframework.amqp.rabbit.annotation.RabbitListener
 import org.springframework.stereotype.Component

documentation/spring-boot-docs/src/main/kotlin/org/springframework/boot/docs/messaging/amqp/rabbitmq/receiving/custom/MyMessageConverter.kt

@@ -0,0 +1,34 @@
+/*
+ * Copyright 2012-present the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.boot.docs.messaging.amqp.rabbitmq.receiving.custom
+
+import org.springframework.amqp.core.Message
+import org.springframework.amqp.core.MessageProperties
+import org.springframework.amqp.support.converter.MessageConverter
+
+internal class MyMessageConverter : MessageConverter {
+
+	override fun toMessage(`object`: Any, messageProperties: MessageProperties): Message {
+		return Message(byteArrayOf())
+	}
+
+	override fun fromMessage(message: Message): Any {
+		return Any()
+	}
+
+}
+

documentation/spring-boot-docs/src/main/kotlin/org/springframework/boot/docs/messaging/amqp/rabbitmq/receiving/custom/MyRabbitConfiguration.kt

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package org.springframework.boot.docs.messaging.amqp.receiving.custom
+package org.springframework.boot.docs.messaging.amqp.rabbitmq.receiving.custom
 
 import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory
 import org.springframework.amqp.rabbit.connection.CachingConnectionFactory

documentation/spring-boot-docs/src/main/kotlin/org/springframework/boot/docs/messaging/amqp/rabbitmq/sending/MyBean.kt

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package org.springframework.boot.docs.messaging.amqp.sending
+package org.springframework.boot.docs.messaging.amqp.rabbitmq.sending
 
 import org.springframework.amqp.core.AmqpAdmin
 import org.springframework.amqp.core.AmqpTemplate

module/spring-boot-amqp/build.gradle

@@ -0,0 +1,61 @@
+/*
+ * Copyright 2012-present the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the License);
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+plugins {
+	id "java-library"
+	id "org.springframework.boot.docker-test"
+	id "org.springframework.boot.auto-configuration"
+	id "org.springframework.boot.configuration-properties"
+	id "org.springframework.boot.deployed"
+	id "org.springframework.boot.optional-dependencies"
+}
+
+description = "Spring Boot support for AMQP 1.0"
+
+dependencies {
+	api(project(":core:spring-boot"))
+	api("org.apache.qpid:protonj2-client")
+	api("org.springframework:spring-messaging")
+	api("org.springframework.amqp:spring-amqp-client")
+
+	implementation(project(":module:spring-boot-transaction"))
+
+	optional(project(":core:spring-boot-autoconfigure"))
+	optional(project(":core:spring-boot-docker-compose"))
+	optional(project(":module:spring-boot-jackson"))
+	optional(project(":core:spring-boot-testcontainers"))
+	optional("org.testcontainers:testcontainers-activemq")
+	optional("org.testcontainers:testcontainers-rabbitmq")
+
+	dockerTestImplementation(project(":core:spring-boot-test"))
+	dockerTestImplementation(project(":test-support:spring-boot-docker-test-support"))
+	dockerTestImplementation(testFixtures(project(":core:spring-boot-docker-compose")))
+	dockerTestImplementation("ch.qos.logback:logback-classic")
+	dockerTestImplementation("org.testcontainers:testcontainers-junit-jupiter")
+
+	testImplementation(project(":core:spring-boot-test"))
+	testImplementation(project(":test-support:spring-boot-test-support"))
+
+	testRuntimeOnly("ch.qos.logback:logback-classic")
+}
+
+tasks.named("compileTestJava") {
+	options.nullability.checking = "tests"
+}
+
+tasks.named("compileDockerTestJava") {
+	options.nullability.checking = "tests"
+}

module/spring-boot-amqp/src/dockerTest/java/org/springframework/boot/amqp/autoconfigure/AmqpAutoConfigurationIntegrationTests.java

@@ -0,0 +1,130 @@
+/*
+ * Copyright 2012-present the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.boot.amqp.autoconfigure;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+import org.awaitility.Awaitility;
+import org.junit.jupiter.api.Test;
+import org.testcontainers.junit.jupiter.Container;
+import org.testcontainers.junit.jupiter.Testcontainers;
+
+import org.springframework.amqp.client.AmqpClient;
+import org.springframework.amqp.client.annotation.AmqpListener;
+import org.springframework.boot.autoconfigure.AutoConfigurations;
+import org.springframework.boot.jackson.autoconfigure.JacksonAutoConfiguration;
+import org.springframework.boot.test.context.runner.ApplicationContextRunner;
+import org.springframework.boot.testsupport.container.RabbitMqManagementContainer;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Integration tests for {@link AmqpAutoConfiguration}.
+ *
+ * @author Stephane Nicoll
+ */
+@Testcontainers(disabledWithoutDocker = true)
+class AmqpAutoConfigurationIntegrationTests {
+
+	@Container
+	static final RabbitMqManagementContainer container = new RabbitMqManagementContainer();
+
+	private final ApplicationContextRunner contextRunner = new ApplicationContextRunner()
+		.withConfiguration(AutoConfigurations.of(AmqpAutoConfiguration.class))
+		.withPropertyValues("spring.amqp.host=" + container.getHost(), "spring.amqp.port=" + container.getAmqpPort());
+
+	@Test
+	void sendAndReceiveUsingAmqpClient() {
+		String queue = container.createRandomQueue();
+		this.contextRunner.run((context) -> {
+			AmqpClient amqpClient = context.getBean(AmqpClient.class);
+			assertThat(amqpClient.to(queue).body("Hello World").send().get(1, TimeUnit.SECONDS)).isTrue();
+			assertThat(amqpClient.from(queue).receiveAndConvert().get(1, TimeUnit.SECONDS)).isEqualTo("Hello World");
+		});
+	}
+
+	@Test
+	void sendAndReceiveUsingJson() {
+		String queue = container.createRandomQueue();
+		this.contextRunner.withConfiguration(AutoConfigurations.of(JacksonAutoConfiguration.class)).run((context) -> {
+			AmqpClient amqpClient = context.getBean(AmqpClient.class);
+			assertThat(amqpClient.to(queue).body(new TestMessage("hello", 42)).send().get(1, TimeUnit.SECONDS))
+				.isTrue();
+			assertThat(amqpClient.from(queue).receiveAndConvert().get(1, TimeUnit.SECONDS))
+				.isEqualTo(new TestMessage("hello", 42));
+		});
+	}
+
+	@Test
+	void sendAndReceiveUsingListener() {
+		String queue = container.createRandomQueue();
+		this.contextRunner.withPropertyValues("test.queue=" + queue)
+			.withUserConfiguration(TestListener.class)
+			.run((context) -> {
+				AmqpClient amqpClient = context.getBean(AmqpClient.class);
+				TestListener listener = context.getBean(TestListener.class);
+				amqpClient.to(queue).body("Hello World").send();
+				Awaitility.waitAtMost(Duration.ofMinutes(1))
+					.untilAsserted(() -> assertThat(listener.messages).containsOnly("Hello World"));
+			});
+	}
+
+	@Test
+	void sendAndReceiveUsingListenerAndJson() {
+		String queue = container.createRandomQueue();
+		this.contextRunner.withPropertyValues("test.queue=" + queue)
+			.withConfiguration(AutoConfigurations.of(JacksonAutoConfiguration.class))
+			.withUserConfiguration(TestMessageListener.class)
+			.run((context) -> {
+				AmqpClient amqpClient = context.getBean(AmqpClient.class);
+				TestMessageListener listener = context.getBean(TestMessageListener.class);
+				amqpClient.to(queue).body(new TestMessage("hello", 42)).send();
+				Awaitility.waitAtMost(Duration.ofMinutes(1))
+					.untilAsserted(() -> assertThat(listener.messages).containsOnly(new TestMessage("hello", 42)));
+			});
+	}
+
+	record TestMessage(String value, int counter) {
+
+	}
+
+	static class TestListener {
+
+		private final List<String> messages = new ArrayList<>();
+
+		@AmqpListener(addresses = "${test.queue}")
+		void processMessage(String message) {
+			this.messages.add(message);
+		}
+
+	}
+
+	static class TestMessageListener {
+
+		private final List<TestMessage> messages = new ArrayList<>();
+
+		@AmqpListener(addresses = "${test.queue}")
+		void processMessage(TestMessage message) {
+			this.messages.add(message);
+		}
+
+	}
+
+}

module/spring-boot-amqp/src/dockerTest/java/org/springframework/boot/amqp/docker/compose/ActiveMqAmqpDockerComposeConnectionDetailsFactoryIntegrationTests.java

@@ -0,0 +1,42 @@
+/*
+ * Copyright 2012-present the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.boot.amqp.docker.compose;
+
+import org.springframework.boot.amqp.autoconfigure.AmqpConnectionDetails;
+import org.springframework.boot.amqp.autoconfigure.AmqpConnectionDetails.Address;
+import org.springframework.boot.docker.compose.service.connection.test.DockerComposeTest;
+import org.springframework.boot.testsupport.container.TestImage;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Integration tests for {@link ActiveMqAmqpDockerComposeConnectionDetailsFactory}.
+ *
+ * @author Stephane Nicoll
+ */
+class ActiveMqAmqpDockerComposeConnectionDetailsFactoryIntegrationTests {
+
+	@DockerComposeTest(composeFile = "activemq-compose.yaml", image = TestImage.ACTIVE_MQ_CLASSIC)
+	void runCreatesConnectionDetails(AmqpConnectionDetails connectionDetails) {
+		assertThat(connectionDetails.getUsername()).isEqualTo("root");
+		assertThat(connectionDetails.getPassword()).isEqualTo("secret");
+		Address address = connectionDetails.getAddress();
+		assertThat(address.host()).isNotNull();
+		assertThat(address.port()).isGreaterThan(0);
+	}
+
+}

module/spring-boot-amqp/src/dockerTest/java/org/springframework/boot/amqp/docker/compose/ArtemisAmqpDockerComposeConnectionDetailsFactoryIntegrationTests.java

@@ -0,0 +1,42 @@
+/*
+ * Copyright 2012-present the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.boot.amqp.docker.compose;
+
+import org.springframework.boot.amqp.autoconfigure.AmqpConnectionDetails;
+import org.springframework.boot.amqp.autoconfigure.AmqpConnectionDetails.Address;
+import org.springframework.boot.docker.compose.service.connection.test.DockerComposeTest;
+import org.springframework.boot.testsupport.container.TestImage;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Integration tests for {@link ArtemisAmqpDockerComposeConnectionDetailsFactory}.
+ *
+ * @author Stephane Nicoll
+ */
+class ArtemisAmqpDockerComposeConnectionDetailsFactoryIntegrationTests {
+
+	@DockerComposeTest(composeFile = "artemis-compose.yaml", image = TestImage.ARTEMIS)
+	void runCreatesConnectionDetails(AmqpConnectionDetails connectionDetails) {
+		assertThat(connectionDetails.getUsername()).isEqualTo("root");
+		assertThat(connectionDetails.getPassword()).isEqualTo("secret");
+		Address address = connectionDetails.getAddress();
+		assertThat(address.host()).isNotNull();
+		assertThat(address.port()).isGreaterThan(0);
+	}
+
+}

module/spring-boot-amqp/src/dockerTest/java/org/springframework/boot/amqp/docker/compose/RabbitMqAmqpDockerComposeConnectionDetailsFactoryIntegrationTests.java

@@ -0,0 +1,42 @@
+/*
+ * Copyright 2012-present the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.boot.amqp.docker.compose;
+
+import org.springframework.boot.amqp.autoconfigure.AmqpConnectionDetails;
+import org.springframework.boot.amqp.autoconfigure.AmqpConnectionDetails.Address;
+import org.springframework.boot.docker.compose.service.connection.test.DockerComposeTest;
+import org.springframework.boot.testsupport.container.TestImage;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Integration tests for {@link RabbitMqAmqpDockerComposeConnectionDetailsFactory}.
+ *
+ * @author Stephane Nicoll
+ */
+class RabbitMqAmqpDockerComposeConnectionDetailsFactoryIntegrationTests {
+
+	@DockerComposeTest(composeFile = "rabbitmq-compose.yaml", image = TestImage.RABBITMQ)
+	void runCreatesConnectionDetails(AmqpConnectionDetails connectionDetails) {
+		assertThat(connectionDetails.getUsername()).isEqualTo("myuser");
+		assertThat(connectionDetails.getPassword()).isEqualTo("secret");
+		Address address = connectionDetails.getAddress();
+		assertThat(address.host()).isNotNull();
+		assertThat(address.port()).isGreaterThan(0);
+	}
+
+}

module/spring-boot-amqp/src/dockerTest/java/org/springframework/boot/amqp/testcontainers/ActiveMqAmqpContainerConnectionDetailsFactoryIntegrationTests.java

@@ -0,0 +1,73 @@
+/*
+ * Copyright 2012-present the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.boot.amqp.testcontainers;
+
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.jupiter.api.Test;
+import org.testcontainers.junit.jupiter.Container;
+import org.testcontainers.junit.jupiter.Testcontainers;
+
+import org.springframework.amqp.client.AmqpClient;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.amqp.autoconfigure.AmqpAutoConfiguration;
+import org.springframework.boot.amqp.autoconfigure.AmqpConnectionDetails;
+import org.springframework.boot.autoconfigure.ImportAutoConfiguration;
+import org.springframework.boot.testcontainers.service.connection.ServiceConnection;
+import org.springframework.boot.testsupport.container.ActiveMQClassicContainer;
+import org.springframework.boot.testsupport.container.TestImage;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Tests for {@link ActiveMqAmqpContainerConnectionDetailsFactory}.
+ *
+ * @author Stephane Nicoll
+ */
+@SpringJUnitConfig
+@Testcontainers(disabledWithoutDocker = true)
+class ActiveMqAmqpContainerConnectionDetailsFactoryIntegrationTests {
+
+	@Container
+	@ServiceConnection
+	static final ActiveMQClassicContainer container = TestImage.container(ActiveMQClassicContainer.class);
+
+	@Autowired(required = false)
+	private AmqpConnectionDetails connectionDetails;
+
+	@Autowired
+	private AmqpClient amqpClient;
+
+	@Test
+	void connectionCanBeMadeToActiveMqContainer() throws Exception {
+		assertThat(this.connectionDetails).isNotNull();
+		String address = UUID.randomUUID().toString();
+		this.amqpClient.to(address).body("test message").send();
+		Object message = this.amqpClient.from(address).receiveAndConvert().get(4, TimeUnit.MINUTES);
+		assertThat(message).isEqualTo("test message");
+	}
+
+	@Configuration(proxyBeanMethods = false)
+	@ImportAutoConfiguration(AmqpAutoConfiguration.class)
+	static class TestConfiguration {
+
+	}
+
+}

module/spring-boot-amqp/src/dockerTest/java/org/springframework/boot/amqp/testcontainers/ArtemisAmqpContainerConnectionDetailsFactoryIntegrationTests.java

@@ -0,0 +1,94 @@
+/*
+ * Copyright 2012-present the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.boot.amqp.testcontainers;
+
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.jupiter.api.Test;
+import org.testcontainers.containers.Container.ExecResult;
+import org.testcontainers.junit.jupiter.Container;
+import org.testcontainers.junit.jupiter.Testcontainers;
+
+import org.springframework.amqp.client.AmqpClient;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.amqp.autoconfigure.AmqpAutoConfiguration;
+import org.springframework.boot.amqp.autoconfigure.AmqpConnectionDetails;
+import org.springframework.boot.autoconfigure.ImportAutoConfiguration;
+import org.springframework.boot.testcontainers.service.connection.ServiceConnection;
+import org.springframework.boot.testsupport.container.ArtemisLegacyContainer;
+import org.springframework.boot.testsupport.container.TestImage;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Tests for {@link ArtemisAmqpContainerConnectionDetailsFactory}.
+ *
+ * @author Stephane Nicoll
+ */
+@SpringJUnitConfig
+@Testcontainers(disabledWithoutDocker = true)
+class ArtemisAmqpContainerConnectionDetailsFactoryIntegrationTests {
+
+	@Container
+	@ServiceConnection
+	static final ArtemisLegacyContainer container = TestImage.container(ArtemisLegacyContainer.class);
+
+	@Autowired(required = false)
+	private AmqpConnectionDetails connectionDetails;
+
+	@Autowired
+	private AmqpClient amqpClient;
+
+	@Test
+	void connectionCanBeMadeToArtemisContainer() throws Exception {
+		assertThat(this.connectionDetails).isNotNull();
+		String address = createRandomQueue();
+		this.amqpClient.to(address).body("test message").send();
+		Object message = this.amqpClient.from(address).receiveAndConvert().get(4, TimeUnit.MINUTES);
+		assertThat(message).isEqualTo("test message");
+	}
+
+	private String createRandomQueue() {
+		String name = UUID.randomUUID().toString();
+		createQueue(name);
+		return name;
+	}
+
+	private void createQueue(String name) {
+		try {
+			ExecResult execResult = container.execInContainer("/var/lib/artemis-instance/bin/artemis", "queue",
+					"create", "--name=" + name, "--auto-create-address", "--anycast", "--silent",
+					"--user=" + container.getUser(), "--password=" + container.getPassword());
+			if (execResult.getExitCode() != 0) {
+				throw new IllegalStateException("Failed to create queue: " + execResult);
+			}
+		}
+		catch (Exception ex) {
+			throw new IllegalStateException("Failed to create queue " + name, ex);
+		}
+	}
+
+	@Configuration(proxyBeanMethods = false)
+	@ImportAutoConfiguration(AmqpAutoConfiguration.class)
+	static class TestConfiguration {
+
+	}
+
+}

module/spring-boot-amqp/src/dockerTest/java/org/springframework/boot/amqp/testcontainers/RabbitMqAmqpContainerConnectionDetailsFactoryIntegrationTests.java

@@ -0,0 +1,71 @@
+/*
+ * Copyright 2012-present the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.boot.amqp.testcontainers;
+
+import java.util.concurrent.TimeUnit;
+
+import org.junit.jupiter.api.Test;
+import org.testcontainers.junit.jupiter.Container;
+import org.testcontainers.junit.jupiter.Testcontainers;
+
+import org.springframework.amqp.client.AmqpClient;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.amqp.autoconfigure.AmqpAutoConfiguration;
+import org.springframework.boot.amqp.autoconfigure.AmqpConnectionDetails;
+import org.springframework.boot.autoconfigure.ImportAutoConfiguration;
+import org.springframework.boot.testcontainers.service.connection.ServiceConnection;
+import org.springframework.boot.testsupport.container.RabbitMqManagementContainer;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Tests for {@link RabbitMqAmqpContainerConnectionDetailsFactory}.
+ *
+ * @author Stephane Nicoll
+ */
+@SpringJUnitConfig
+@Testcontainers(disabledWithoutDocker = true)
+class RabbitMqAmqpContainerConnectionDetailsFactoryIntegrationTests {
+
+	@Container
+	@ServiceConnection
+	static final RabbitMqManagementContainer container = new RabbitMqManagementContainer();
+
+	@Autowired(required = false)
+	private AmqpConnectionDetails connectionDetails;
+
+	@Autowired
+	private AmqpClient amqpClient;
+
+	@Test
+	void connectionCanBeMadeToRabbitMqContainer() throws Exception {
+		assertThat(this.connectionDetails).isNotNull();
+		String address = container.createRandomQueue();
+		this.amqpClient.to(address).body("test message").send();
+		Object message = this.amqpClient.from(address).receiveAndConvert().get(4, TimeUnit.MINUTES);
+		assertThat(message).isEqualTo("test message");
+	}
+
+	@Configuration(proxyBeanMethods = false)
+	@ImportAutoConfiguration(AmqpAutoConfiguration.class)
+	static class TestConfiguration {
+
+	}
+
+}

module/spring-boot-amqp/src/dockerTest/resources/logback-test.xml

@@ -0,0 +1,4 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<configuration>
+	<include resource="org/springframework/boot/logging/logback/base.xml"/>
+</configuration>

module/spring-boot-amqp/src/dockerTest/resources/org/springframework/boot/amqp/docker/compose/activemq-compose.yaml

@@ -0,0 +1,9 @@
+services:
+  activemq:
+    image: '{imageName}'
+    ports:
+      - '61616'
+      - '5672'
+    environment:
+      ACTIVEMQ_CONNECTION_USER: 'root'
+      ACTIVEMQ_CONNECTION_PASSWORD: 'secret'

module/spring-boot-amqp/src/dockerTest/resources/org/springframework/boot/amqp/docker/compose/artemis-compose.yaml

@@ -0,0 +1,9 @@
+services:
+  artemis:
+    image: '{imageName}'
+    ports:
+      - '61616'
+      - '5672'
+    environment:
+      ARTEMIS_USER: 'root'
+      ARTEMIS_PASSWORD: 'secret'

module/spring-boot-amqp/src/dockerTest/resources/org/springframework/boot/amqp/docker/compose/rabbitmq-compose.yaml

@@ -0,0 +1,8 @@
+services:
+  rabbitmq:
+    image: '{imageName}'
+    environment:
+      - 'RABBITMQ_DEFAULT_USER=myuser'
+      - 'RABBITMQ_DEFAULT_PASS=secret'
+    ports:
+      - '5672'

module/spring-boot-amqp/src/dockerTest/resources/spring.properties

@@ -0,0 +1 @@
+spring.test.context.cache.maxSize=1

module/spring-boot-amqp/src/main/java/org/springframework/boot/amqp/autoconfigure/AmqpAnnotationDrivenConfiguration.java

@@ -0,0 +1,100 @@
+/*
+ * Copyright 2012-present the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.boot.amqp.autoconfigure;
+
+import org.springframework.amqp.client.AmqpConnectionFactory;
+import org.springframework.amqp.client.annotation.AmqpListener;
+import org.springframework.amqp.client.config.AmqpDefaultConfiguration;
+import org.springframework.amqp.client.config.EnableAmqp;
+import org.springframework.amqp.client.config.MethodAmqpMessageListenerContainerFactory;
+import org.springframework.amqp.client.listener.AmqpListenerErrorHandler;
+import org.springframework.amqp.support.converter.MessageConverter;
+import org.springframework.beans.factory.ObjectProvider;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnThreading;
+import org.springframework.boot.thread.Threading;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.core.task.VirtualThreadTaskExecutor;
+
+/**
+ * Configuration for Spring AMQP annotation driven endpoints using {@link AmqpListener}.
+ *
+ * @author Stephane Nicoll
+ */
+@Configuration(proxyBeanMethods = false)
+@ConditionalOnClass(EnableAmqp.class)
+class AmqpAnnotationDrivenConfiguration {
+
+	private static final String AMQP_LISTENER_CONTAINER_FACTORY_CONFIGURER_BEAN_NAME = "amqpMessageListenerContainerFactoryConfigurer";
+
+	private final AmqpProperties properties;
+
+	private final ObjectProvider<MessageConverter> messageConverter;
+
+	private final ObjectProvider<AmqpListenerErrorHandler> listenerErrorHandler;
+
+	AmqpAnnotationDrivenConfiguration(AmqpProperties properties, ObjectProvider<MessageConverter> messageConverter,
+			ObjectProvider<AmqpListenerErrorHandler> listenerErrorHandler) {
+		this.properties = properties;
+		this.messageConverter = messageConverter;
+		this.listenerErrorHandler = listenerErrorHandler;
+	}
+
+	@Bean(name = AMQP_LISTENER_CONTAINER_FACTORY_CONFIGURER_BEAN_NAME)
+	@ConditionalOnMissingBean
+	@ConditionalOnThreading(Threading.PLATFORM)
+	AmqpMessageListenerContainerFactoryConfigurer amqpMessageListenerContainerFactoryConfigurer() {
+		return messageListenerConfigurer();
+	}
+
+	@Bean(name = AMQP_LISTENER_CONTAINER_FACTORY_CONFIGURER_BEAN_NAME)
+	@ConditionalOnMissingBean
+	@ConditionalOnThreading(Threading.VIRTUAL)
+	AmqpMessageListenerContainerFactoryConfigurer amqpMessageListenerContainerFactoryConfigurerVirtualThreads() {
+		AmqpMessageListenerContainerFactoryConfigurer configurer = messageListenerConfigurer();
+		configurer.setTaskExecutor(new VirtualThreadTaskExecutor("amqp-simple-"));
+		return configurer;
+	}
+
+	@Bean(name = AmqpDefaultConfiguration.DEFAULT_AMQP_LISTENER_CONTAINER_FACTORY_BEAN_NAME)
+	@ConditionalOnMissingBean(name = AmqpDefaultConfiguration.DEFAULT_AMQP_LISTENER_CONTAINER_FACTORY_BEAN_NAME)
+	MethodAmqpMessageListenerContainerFactory amqpMessageListenerContainerFactory(
+			AmqpMessageListenerContainerFactoryConfigurer configurer, AmqpConnectionFactory connectionFactory) {
+		MethodAmqpMessageListenerContainerFactory factory = new MethodAmqpMessageListenerContainerFactory(
+				connectionFactory);
+		configurer.configure(factory);
+		return factory;
+	}
+
+	private AmqpMessageListenerContainerFactoryConfigurer messageListenerConfigurer() {
+		AmqpMessageListenerContainerFactoryConfigurer configurer = new AmqpMessageListenerContainerFactoryConfigurer(
+				this.properties);
+		configurer.setMessageConverter(this.messageConverter.getIfUnique());
+		configurer.setListenerErrorHandler(this.listenerErrorHandler.getIfUnique());
+		return configurer;
+	}
+
+	@Configuration(proxyBeanMethods = false)
+	@EnableAmqp
+	@ConditionalOnMissingBean(name = AmqpDefaultConfiguration.AMQP_LISTENER_ANNOTATION_PROCESSOR_BEAN_NAME)
+	static class EnableAmqpConfiguration {
+
+	}
+
+}

module/spring-boot-amqp/src/main/java/org/springframework/boot/amqp/autoconfigure/AmqpAutoConfiguration.java

@@ -0,0 +1,134 @@
+/*
+ * Copyright 2012-present the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.boot.amqp.autoconfigure;
+
+import java.util.stream.Stream;
+
+import org.apache.qpid.protonj2.client.Client;
+import org.apache.qpid.protonj2.client.ClientOptions;
+import org.apache.qpid.protonj2.client.ConnectionOptions;
+import tools.jackson.databind.json.JsonMapper;
+
+import org.springframework.amqp.client.AmqpClient;
+import org.springframework.amqp.client.AmqpConnectionFactory;
+import org.springframework.amqp.client.SingleAmqpConnectionFactory;
+import org.springframework.amqp.support.converter.JacksonJsonMessageConverter;
+import org.springframework.amqp.support.converter.MessageConverter;
+import org.springframework.beans.factory.ObjectProvider;
+import org.springframework.boot.amqp.autoconfigure.AmqpConnectionDetails.Address;
+import org.springframework.boot.autoconfigure.AutoConfiguration;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnSingleCandidate;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.boot.context.properties.PropertyMapper;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Import;
+
+/**
+ * Auto-configuration for generic AMQP 1.0 support.
+ *
+ * @author Stephane Nicoll
+ * @since 4.2.0
+ */
+@AutoConfiguration(afterName = "org.springframework.boot.jackson.autoconfigure.JacksonAutoConfiguration")
+@ConditionalOnClass({ Client.class, AmqpConnectionFactory.class })
+@EnableConfigurationProperties(AmqpProperties.class)
+@Import(AmqpAnnotationDrivenConfiguration.class)
+public final class AmqpAutoConfiguration {
+
+	@Configuration(proxyBeanMethods = false)
+	static class AmqpConnectionFactoryConfiguration {
+
+		@Bean
+		@ConditionalOnMissingBean
+		Client protonClient() {
+			return Client.create(new ClientOptions());
+		}
+
+		@Bean
+		@ConditionalOnMissingBean
+		AmqpConnectionDetails amqpConnectionDetails(AmqpProperties properties) {
+			return new PropertiesAmqpConnectionDetails(properties);
+		}
+
+		@Bean
+		@ConditionalOnMissingBean
+		AmqpConnectionFactory amqpConnectionFactory(AmqpConnectionDetails connectionDetails, Client protonClient,
+				ObjectProvider<ConnectionOptionsCustomizer> connectionOptionsCustomizers) {
+			ConnectionOptions connectionOptions = createConnectionOptions(connectionDetails,
+					connectionOptionsCustomizers.orderedStream());
+			return createAmqpConnectionFactory(connectionDetails, protonClient).setConnectionOptions(connectionOptions);
+		}
+
+		private SingleAmqpConnectionFactory createAmqpConnectionFactory(AmqpConnectionDetails connectionDetails,
+				Client protonClient) {
+			SingleAmqpConnectionFactory connectionFactory = new SingleAmqpConnectionFactory(protonClient);
+			PropertyMapper map = PropertyMapper.get();
+			Address address = connectionDetails.getAddress();
+			map.from(address::host).to(connectionFactory::setHost);
+			map.from(address::port).to(connectionFactory::setPort);
+			return connectionFactory;
+		}
+
+		private ConnectionOptions createConnectionOptions(AmqpConnectionDetails connectionDetails,
+				Stream<ConnectionOptionsCustomizer> customizers) {
+			ConnectionOptions connectionOptions = new ConnectionOptions();
+			PropertyMapper map = PropertyMapper.get();
+			map.from(connectionDetails::getUsername).to(connectionOptions::user);
+			map.from(connectionDetails::getPassword).to(connectionOptions::password);
+			customizers.forEach((customizer) -> customizer.customize(connectionOptions));
+			return connectionOptions;
+		}
+
+	}
+
+	@Configuration(proxyBeanMethods = false)
+	@ConditionalOnBean(JsonMapper.class)
+	@ConditionalOnSingleCandidate(JsonMapper.class)
+	static class JacksonMessageConverterConfiguration {
+
+		@Bean
+		@ConditionalOnMissingBean(MessageConverter.class)
+		JacksonJsonMessageConverter jacksonJsonMessageConverter(JsonMapper jsonMapper) {
+			return new JacksonJsonMessageConverter(jsonMapper);
+		}
+
+	}
+
+	@Configuration(proxyBeanMethods = false)
+	static class AmqpClientConfiguration {
+
+		@Bean
+		@ConditionalOnMissingBean
+		AmqpClient amqpClient(AmqpConnectionFactory amqpConnectionFactory, AmqpProperties properties,
+				ObjectProvider<MessageConverter> messageConverter, ObjectProvider<AmqpClientCustomizer> customizers) {
+			AmqpClient.Builder builder = AmqpClient.builder(amqpConnectionFactory);
+			AmqpProperties.Client client = properties.getClient();
+			PropertyMapper map = PropertyMapper.get();
+			map.from(client.getDefaultToAddress()).to(builder::defaultToAddress);
+			map.from(client.getCompletionTimeout()).to(builder::completionTimeout);
+			messageConverter.ifAvailable(builder::messageConverter);
+			customizers.orderedStream().forEach((customizer) -> customizer.customize(builder));
+			return builder.build();
+		}
+
+	}
+
+}

module/spring-boot-amqp/src/main/java/org/springframework/boot/amqp/autoconfigure/AmqpClientCustomizer.java

@@ -0,0 +1,37 @@
+/*
+ * Copyright 2012-present the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.boot.amqp.autoconfigure;
+
+import org.springframework.amqp.client.AmqpClient;
+
+/**
+ * Callback interface that can be used to customize the auto-configured
+ * {@link AmqpClient.Builder}.
+ *
+ * @author Stephane Nicoll
+ * @since 4.2.0
+ */
+@FunctionalInterface
+public interface AmqpClientCustomizer {
+
+	/**
+	 * Callback to customize a {@link AmqpClient.Builder} instance.
+	 * @param amqpClientBuilder the client builder to customize
+	 */
+	void customize(AmqpClient.Builder amqpClientBuilder);
+
+}

module/spring-boot-amqp/src/main/java/org/springframework/boot/amqp/autoconfigure/AmqpConnectionDetails.java

@@ -0,0 +1,62 @@
+/*
+ * Copyright 2012-present the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.boot.amqp.autoconfigure;
+
+import org.jspecify.annotations.Nullable;
+
+import org.springframework.boot.autoconfigure.service.connection.ConnectionDetails;
+
+/**
+ * Details required to establish a connection to an AMQP broker.
+ *
+ * @author Stephane Nicoll
+ * @since 4.2.0
+ */
+public interface AmqpConnectionDetails extends ConnectionDetails {
+
+	/**
+	 * Return the {@link Address} of the broker.
+	 * @return the address
+	 */
+	Address getAddress();
+
+	/**
+	 * Login user to authenticate to the broker.
+	 * @return the login user to authenticate to the broker or {@code null}
+	 */
+	default @Nullable String getUsername() {
+		return null;
+	}
+
+	/**
+	 * Password used to authenticate to the broker.
+	 * @return the password to authenticate to the broker or {@code null}
+	 */
+	default @Nullable String getPassword() {
+		return null;
+	}
+
+	/**
+	 * An AMQP broker address.
+	 *
+	 * @param host the host
+	 * @param port the port
+	 */
+	record Address(String host, int port) {
+	}
+
+}

module/spring-boot-amqp/src/main/java/org/springframework/boot/amqp/autoconfigure/AmqpMessageListenerContainerFactoryConfigurer.java

@@ -0,0 +1,86 @@
+/*
+ * Copyright 2012-present the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.boot.amqp.autoconfigure;
+
+import java.util.concurrent.Executor;
+
+import org.jspecify.annotations.Nullable;
+
+import org.springframework.amqp.client.config.AmqpMessageListenerContainerFactory;
+import org.springframework.amqp.client.config.MethodAmqpMessageListenerContainerFactory;
+import org.springframework.amqp.client.listener.AmqpListenerErrorHandler;
+import org.springframework.amqp.support.converter.MessageConverter;
+import org.springframework.boot.amqp.autoconfigure.AmqpProperties.Listener;
+import org.springframework.boot.context.properties.PropertyMapper;
+
+/**
+ * Configure {@link AmqpMessageListenerContainerFactory} with sensible defaults tuned
+ * using configuration properties. Handle additional settings exposed by
+ * {@link MethodAmqpMessageListenerContainerFactory} as well.
+ * <p>
+ * Can be injected into application code and used to define a custom
+ * {@code AmqpMessageListenerContainerFactory} whose configuration is based upon that
+ * produced by auto-configuration.
+ *
+ * @author Stephane Nicoll
+ * @since 4.2.0
+ */
+public class AmqpMessageListenerContainerFactoryConfigurer {
+
+	private final AmqpProperties amqpProperties;
+
+	private @Nullable Executor taskExecutor;
+
+	private @Nullable MessageConverter messageConverter;
+
+	private @Nullable AmqpListenerErrorHandler listenerErrorHandler;
+
+	public AmqpMessageListenerContainerFactoryConfigurer(AmqpProperties amqpProperties) {
+		this.amqpProperties = amqpProperties;
+	}
+
+	protected void setTaskExecutor(@Nullable Executor taskExecutor) {
+		this.taskExecutor = taskExecutor;
+	}
+
+	protected void setMessageConverter(@Nullable MessageConverter messageConverter) {
+		this.messageConverter = messageConverter;
+	}
+
+	protected void setListenerErrorHandler(@Nullable AmqpListenerErrorHandler listenerErrorHandler) {
+		this.listenerErrorHandler = listenerErrorHandler;
+	}
+
+	public void configure(AmqpMessageListenerContainerFactory factory) {
+		PropertyMapper map = PropertyMapper.get();
+		Listener listener = this.amqpProperties.getListener();
+
+		map.from(listener::isAutoStartup).to(factory::setAutoStartup);
+		map.from(listener::getConcurrency).to(factory::setConcurrency);
+		map.from(listener::getInitialCredits).to(factory::setInitialCredits);
+		map.from(listener::getReceiveTimeout).to(factory::setReceiveTimeout);
+		map.from(listener::getGracefulShutdownPeriod).to(factory::setGracefulShutdownPeriod);
+		map.from(this.taskExecutor).to(factory::setTaskExecutor);
+
+		if (factory instanceof MethodAmqpMessageListenerContainerFactory methodFactory) {
+			map.from(listener::isDefaultRequeueRejected).to(methodFactory::setDefaultRequeueRejected);
+			map.from(this.messageConverter).to(methodFactory::setMessageConverter);
+			map.from(this.listenerErrorHandler).to(methodFactory::setListenerErrorHandler);
+		}
+	}
+
+}

module/spring-boot-amqp/src/main/java/org/springframework/boot/amqp/autoconfigure/AmqpProperties.java

@@ -0,0 +1,213 @@
+/*
+ * Copyright 2012-present the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.boot.amqp.autoconfigure;
+
+import java.time.Duration;
+
+import org.jspecify.annotations.Nullable;
+
+import org.springframework.boot.context.properties.ConfigurationProperties;
+
+/**
+ * Configuration properties for AMQP 1.0 connectivity.
+ *
+ * @author Stephane Nicoll
+ * @since 4.2.0
+ */
+@ConfigurationProperties("spring.amqp")
+public class AmqpProperties {
+
+	/**
+	 * AMQP broker host.
+	 */
+	private String host = "localhost";
+
+	/**
+	 * AMQP broker port.
+	 */
+	private int port = 5672;
+
+	/**
+	 * Login user to authenticate to the broker.
+	 */
+	private @Nullable String username;
+
+	/**
+	 * Password used to authenticate to the broker.
+	 */
+	private @Nullable String password;
+
+	private final Client client = new Client();
+
+	private final Listener listener = new Listener();
+
+	public String getHost() {
+		return this.host;
+	}
+
+	public void setHost(String host) {
+		this.host = host;
+	}
+
+	public int getPort() {
+		return this.port;
+	}
+
+	public void setPort(int port) {
+		this.port = port;
+	}
+
+	public @Nullable String getUsername() {
+		return this.username;
+	}
+
+	public void setUsername(@Nullable String username) {
+		this.username = username;
+	}
+
+	public @Nullable String getPassword() {
+		return this.password;
+	}
+
+	public void setPassword(@Nullable String password) {
+		this.password = password;
+	}
+
+	public Client getClient() {
+		return this.client;
+	}
+
+	public Listener getListener() {
+		return this.listener;
+	}
+
+	/**
+	 * Client-level settings.
+	 */
+	public static class Client {
+
+		/**
+		 * Default destination address for send operations when none is specified.
+		 */
+		private @Nullable String defaultToAddress;
+
+		/**
+		 * Maximum time to wait for request operations to complete.
+		 */
+		private Duration completionTimeout = Duration.ofSeconds(60);
+
+		public @Nullable String getDefaultToAddress() {
+			return this.defaultToAddress;
+		}
+
+		public void setDefaultToAddress(@Nullable String defaultToAddress) {
+			this.defaultToAddress = defaultToAddress;
+		}
+
+		public Duration getCompletionTimeout() {
+			return this.completionTimeout;
+		}
+
+		public void setCompletionTimeout(Duration completionTimeout) {
+			this.completionTimeout = completionTimeout;
+		}
+
+	}
+
+	public static class Listener {
+
+		/**
+		 * Whether to start the container automatically on startup.
+		 */
+		private boolean autoStartup = true;
+
+		/**
+		 * Number of consumers.
+		 */
+		private int concurrency = 1;
+
+		/**
+		 * Initial number of credits to grant to the AMQP receiver.
+		 */
+		private int initialCredits = 100;
+
+		/**
+		 * Timeout for receive operations.
+		 */
+		private Duration receiveTimeout = Duration.ofSeconds(1);
+
+		/**
+		 * How long to wait for all consumers to shut down successfully.
+		 */
+		private Duration gracefulShutdownPeriod = Duration.ofSeconds(30);
+
+		/**
+		 * Whether to requeue messages in case of error.
+		 */
+		private boolean defaultRequeueRejected = true;
+
+		public boolean isAutoStartup() {
+			return this.autoStartup;
+		}
+
+		public void setAutoStartup(boolean autoStartup) {
+			this.autoStartup = autoStartup;
+		}
+
+		public int getConcurrency() {
+			return this.concurrency;
+		}
+
+		public void setConcurrency(int concurrency) {
+			this.concurrency = concurrency;
+		}
+
+		public int getInitialCredits() {
+			return this.initialCredits;
+		}
+
+		public void setInitialCredits(int initialCredits) {
+			this.initialCredits = initialCredits;
+		}
+
+		public Duration getReceiveTimeout() {
+			return this.receiveTimeout;
+		}
+
+		public void setReceiveTimeout(Duration receiveTimeout) {
+			this.receiveTimeout = receiveTimeout;
+		}
+
+		public Duration getGracefulShutdownPeriod() {
+			return this.gracefulShutdownPeriod;
+		}
+
+		public void setGracefulShutdownPeriod(Duration gracefulShutdownPeriod) {
+			this.gracefulShutdownPeriod = gracefulShutdownPeriod;
+		}
+
+		public boolean isDefaultRequeueRejected() {
+			return this.defaultRequeueRejected;
+		}
+
+		public void setDefaultRequeueRejected(boolean defaultRequeueRejected) {
+			this.defaultRequeueRejected = defaultRequeueRejected;
+		}
+
+	}
+
+}

module/spring-boot-amqp/src/main/java/org/springframework/boot/amqp/autoconfigure/ConnectionOptionsCustomizer.java

@@ -0,0 +1,39 @@
+/*
+ * Copyright 2012-present the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.boot.amqp.autoconfigure;
+
+import org.apache.qpid.protonj2.client.ConnectionOptions;
+
+import org.springframework.amqp.client.AmqpConnectionFactory;
+
+/**
+ * Callback interface for customizing {@link ConnectionOptions} on the auto-configured
+ * {@link AmqpConnectionFactory}.
+ *
+ * @author Stephane Nicoll
+ * @since 4.2.0
+ */
+@FunctionalInterface
+public interface ConnectionOptionsCustomizer {
+
+	/**
+	 * Customize the {@link ConnectionOptions}.
+	 * @param connectionOptions the connection options to customize
+	 */
+	void customize(ConnectionOptions connectionOptions);
+
+}

module/spring-boot-amqp/src/main/java/org/springframework/boot/amqp/autoconfigure/PropertiesAmqpConnectionDetails.java

@@ -0,0 +1,49 @@
+/*
+ * Copyright 2012-present the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.boot.amqp.autoconfigure;
+
+import org.jspecify.annotations.Nullable;
+
+/**
+ * Adapts {@link AmqpProperties} to {@link AmqpConnectionDetails}.
+ *
+ * @author Stephane Nicoll
+ */
+class PropertiesAmqpConnectionDetails implements AmqpConnectionDetails {
+
+	private final AmqpProperties properties;
+
+	PropertiesAmqpConnectionDetails(AmqpProperties properties) {
+		this.properties = properties;
+	}
+
+	@Override
+	public Address getAddress() {
+		return new Address(this.properties.getHost(), this.properties.getPort());
+	}
+
+	@Override
+	public @Nullable String getUsername() {
+		return this.properties.getUsername();
+	}
+
+	@Override
+	public @Nullable String getPassword() {
+		return this.properties.getPassword();
+	}
+
+}

module/spring-boot-amqp/src/main/java/org/springframework/boot/amqp/autoconfigure/package-info.java

@@ -0,0 +1,23 @@
+/*
+ * Copyright 2012-present the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+/**
+ * Auto-configuration for generic AMQP 1.0 support.
+ */
+@NullMarked
+package org.springframework.boot.amqp.autoconfigure;
+
+import org.jspecify.annotations.NullMarked;

module/spring-boot-amqp/src/main/java/org/springframework/boot/amqp/docker/compose/AbstractAmqpDockerComposeConnectionDetailsFactory.java

@@ -0,0 +1,89 @@
+/*
+ * Copyright 2012-present the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.boot.amqp.docker.compose;
+
+import org.jspecify.annotations.Nullable;
+
+import org.springframework.boot.amqp.autoconfigure.AmqpConnectionDetails;
+import org.springframework.boot.docker.compose.core.RunningService;
+import org.springframework.boot.docker.compose.service.connection.DockerComposeConnectionDetailsFactory;
+import org.springframework.boot.docker.compose.service.connection.DockerComposeConnectionSource;
+
+/**
+ * Base {@link DockerComposeConnectionDetailsFactory} to create
+ * {@link AmqpConnectionDetails} from an AMQP 1.0-compliant service.
+ *
+ * @author Stephane Nicoll
+ */
+abstract class AbstractAmqpDockerComposeConnectionDetailsFactory
+		extends DockerComposeConnectionDetailsFactory<AmqpConnectionDetails> {
+
+	protected AbstractAmqpDockerComposeConnectionDetailsFactory(String connectionName) {
+		super(connectionName);
+	}
+
+	protected AbstractAmqpDockerComposeConnectionDetailsFactory(String[] connectionNames,
+			String... requiredClassNames) {
+		super(connectionNames, requiredClassNames);
+	}
+
+	@Override
+	protected final @Nullable AmqpConnectionDetails getDockerComposeConnectionDetails(
+			DockerComposeConnectionSource source) {
+		RunningService runningService = source.getRunningService();
+		AmqpEnvironment environment = getAmqpEnvironment(runningService);
+		return new AmqpDockerComposeConnectionDetails(runningService, environment);
+	}
+
+	protected abstract AmqpEnvironment getAmqpEnvironment(RunningService service);
+
+	static class AmqpDockerComposeConnectionDetails extends DockerComposeConnectionDetails
+			implements AmqpConnectionDetails {
+
+		private static final int AMQP_PORT = 5672;
+
+		private final Address address;
+
+		private final @Nullable String username;
+
+		private final @Nullable String password;
+
+		AmqpDockerComposeConnectionDetails(RunningService runningService, AmqpEnvironment environment) {
+			super(runningService);
+			this.address = new Address(runningService.host(), runningService.ports().get(AMQP_PORT));
+			this.username = environment.getUsername();
+			this.password = environment.getPassword();
+		}
+
+		@Override
+		public Address getAddress() {
+			return this.address;
+		}
+
+		@Override
+		public @Nullable String getUsername() {
+			return this.username;
+		}
+
+		@Override
+		public @Nullable String getPassword() {
+			return this.password;
+		}
+
+	}
+
+}

module/spring-boot-amqp/src/main/java/org/springframework/boot/amqp/docker/compose/ActiveMqAmqpDockerComposeConnectionDetailsFactory.java

@@ -0,0 +1,46 @@
+/*
+ * Copyright 2012-present the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.boot.amqp.docker.compose;
+
+import java.util.Map;
+
+import org.jspecify.annotations.Nullable;
+
+import org.springframework.boot.amqp.autoconfigure.AmqpConnectionDetails;
+import org.springframework.boot.docker.compose.core.RunningService;
+import org.springframework.boot.docker.compose.service.connection.DockerComposeConnectionDetailsFactory;
+
+/**
+ * {@link DockerComposeConnectionDetailsFactory} to create {@link AmqpConnectionDetails}
+ * for an {@code activemq} service.
+ *
+ * @author Stephane Nicoll
+ * @author Eddú Meléndez
+ */
+class ActiveMqAmqpDockerComposeConnectionDetailsFactory extends AbstractAmqpDockerComposeConnectionDetailsFactory {
+
+	ActiveMqAmqpDockerComposeConnectionDetailsFactory() {
+		super(new String[] { "apache/activemq-classic", "apache/activemq" });
+	}
+
+	@Override
+	protected AmqpEnvironment getAmqpEnvironment(RunningService service) {
+		Map<String, @Nullable String> env = service.env();
+		return new AmqpEnvironment(env.get("ACTIVEMQ_CONNECTION_USER"), env.get("ACTIVEMQ_CONNECTION_PASSWORD"));
+	}
+
+}

module/spring-boot-amqp/src/main/java/org/springframework/boot/amqp/docker/compose/AmqpEnvironment.java

@@ -0,0 +1,45 @@
+/*
+ * Copyright 2012-present the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.boot.amqp.docker.compose;
+
+import org.jspecify.annotations.Nullable;
+
+/**
+ * Generic AMQP environment details.
+ *
+ * @author Stephane Nicoll
+ */
+class AmqpEnvironment {
+
+	private final @Nullable String username;
+
+	private final @Nullable String password;
+
+	AmqpEnvironment(@Nullable String username, @Nullable String password) {
+		this.username = username;
+		this.password = password;
+	}
+
+	@Nullable String getUsername() {
+		return this.username;
+	}
+
+	@Nullable String getPassword() {
+		return this.password;
+	}
+
+}

module/spring-boot-amqp/src/main/java/org/springframework/boot/amqp/docker/compose/ArtemisAmqpDockerComposeConnectionDetailsFactory.java

@@ -0,0 +1,45 @@
+/*
+ * Copyright 2012-present the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.boot.amqp.docker.compose;
+
+import java.util.Map;
+
+import org.jspecify.annotations.Nullable;
+
+import org.springframework.boot.amqp.autoconfigure.AmqpConnectionDetails;
+import org.springframework.boot.docker.compose.core.RunningService;
+import org.springframework.boot.docker.compose.service.connection.DockerComposeConnectionDetailsFactory;
+
+/**
+ * {@link DockerComposeConnectionDetailsFactory} to create {@link AmqpConnectionDetails}
+ * for a {@code artemis} service.
+ *
+ * @author Stephane Nicoll
+ */
+class ArtemisAmqpDockerComposeConnectionDetailsFactory extends AbstractAmqpDockerComposeConnectionDetailsFactory {
+
+	ArtemisAmqpDockerComposeConnectionDetailsFactory() {
+		super(new String[] { "apache/activemq-artemis", "apache/artemis" });
+	}
+
+	@Override
+	protected AmqpEnvironment getAmqpEnvironment(RunningService service) {
+		Map<String, @Nullable String> env = service.env();
+		return new AmqpEnvironment(env.get("ARTEMIS_USER"), env.get("ARTEMIS_PASSWORD"));
+	}
+
+}

module/spring-boot-amqp/src/main/java/org/springframework/boot/amqp/docker/compose/RabbitMqAmqpDockerComposeConnectionDetailsFactory.java

@@ -0,0 +1,47 @@
+/*
+ * Copyright 2012-present the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.boot.amqp.docker.compose;
+
+import java.util.Map;
+
+import org.jspecify.annotations.Nullable;
+
+import org.springframework.boot.amqp.autoconfigure.AmqpConnectionDetails;
+import org.springframework.boot.docker.compose.core.RunningService;
+import org.springframework.boot.docker.compose.service.connection.DockerComposeConnectionDetailsFactory;
+
+/**
+ * {@link DockerComposeConnectionDetailsFactory} to create {@link AmqpConnectionDetails}
+ * for a {@code rabbitmq} service.
+ *
+ * @author Stephane Nicoll
+ */
+class RabbitMqAmqpDockerComposeConnectionDetailsFactory extends AbstractAmqpDockerComposeConnectionDetailsFactory {
+
+	RabbitMqAmqpDockerComposeConnectionDetailsFactory() {
+		super("rabbitmq");
+	}
+
+	@Override
+	protected AmqpEnvironment getAmqpEnvironment(RunningService service) {
+		Map<String, @Nullable String> env = service.env();
+		String username = env.getOrDefault("RABBITMQ_DEFAULT_USER", env.getOrDefault("RABBITMQ_USERNAME", "guest"));
+		String password = env.getOrDefault("RABBITMQ_DEFAULT_PASS", env.getOrDefault("RABBITMQ_PASSWORD", "guest"));
+		return new AmqpEnvironment(username, password);
+	}
+
+}

module/spring-boot-amqp/src/main/java/org/springframework/boot/amqp/docker/compose/package-info.java

@@ -0,0 +1,23 @@
+/*
+ * Copyright 2012-present the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+/**
+ * Support for Docker Compose generic AMQP service connections.
+ */
+@NullMarked
+package org.springframework.boot.amqp.docker.compose;
+
+import org.jspecify.annotations.NullMarked;

module/spring-boot-amqp/src/main/java/org/springframework/boot/amqp/testcontainers/AbstractAmqpContainerConnectionDetailsFactory.java

@@ -0,0 +1,53 @@
+/*
+ * Copyright 2012-present the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.boot.amqp.testcontainers;
+
+import org.testcontainers.containers.Container;
+
+import org.springframework.boot.amqp.autoconfigure.AmqpConnectionDetails;
+import org.springframework.boot.testcontainers.service.connection.ContainerConnectionDetailsFactory;
+import org.springframework.boot.testcontainers.service.connection.ContainerConnectionSource;
+import org.springframework.boot.testcontainers.service.connection.ServiceConnection;
+
+/**
+ * Base {@link ContainerConnectionDetailsFactory} to create {@link AmqpConnectionDetails}
+ * from a {@link ServiceConnection @ServiceConnection}-annotated container that is
+ * compliant with AMQP 1.0.
+ *
+ * @param <C> the container type
+ * @author Stephane Nicoll
+ */
+abstract class AbstractAmqpContainerConnectionDetailsFactory<C extends Container<?>>
+		extends ContainerConnectionDetailsFactory<C, AmqpConnectionDetails> {
+
+	protected static class AmqpContainerConnectionDetails<C extends Container<?>> extends ContainerConnectionDetails<C>
+			implements AmqpConnectionDetails {
+
+		private static final int AMQP_PORT = 5672;
+
+		protected AmqpContainerConnectionDetails(ContainerConnectionSource<C> source) {
+			super(source);
+		}
+
+		@Override
+		public Address getAddress() {
+			return new Address(getContainer().getHost(), getContainer().getMappedPort(AMQP_PORT));
+		}
+
+	}
+
+}

module/spring-boot-amqp/src/main/java/org/springframework/boot/amqp/testcontainers/ActiveMqAmqpContainerConnectionDetailsFactory.java

@@ -0,0 +1,60 @@
+/*
+ * Copyright 2012-present the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.boot.amqp.testcontainers;
+
+import org.jspecify.annotations.Nullable;
+import org.testcontainers.activemq.ActiveMQContainer;
+
+import org.springframework.boot.amqp.autoconfigure.AmqpConnectionDetails;
+import org.springframework.boot.testcontainers.service.connection.ContainerConnectionDetailsFactory;
+import org.springframework.boot.testcontainers.service.connection.ContainerConnectionSource;
+import org.springframework.boot.testcontainers.service.connection.ServiceConnection;
+
+/**
+ * {@link ContainerConnectionDetailsFactory} to create {@link AmqpConnectionDetails} from
+ * a {@link ServiceConnection @ServiceConnection}-annotated {@link ActiveMQContainer}.
+ *
+ * @author Stephane Nicoll
+ */
+class ActiveMqAmqpContainerConnectionDetailsFactory
+		extends AbstractAmqpContainerConnectionDetailsFactory<ActiveMQContainer> {
+
+	@Override
+	protected @Nullable AmqpConnectionDetails getContainerConnectionDetails(
+			ContainerConnectionSource<ActiveMQContainer> source) {
+		return new ActiveMqAmqpContainerConnectionDetails(source);
+	}
+
+	static class ActiveMqAmqpContainerConnectionDetails extends AmqpContainerConnectionDetails<ActiveMQContainer> {
+
+		ActiveMqAmqpContainerConnectionDetails(ContainerConnectionSource<ActiveMQContainer> source) {
+			super(source);
+		}
+
+		@Override
+		public @Nullable String getUsername() {
+			return getContainer().getUser();
+		}
+
+		@Override
+		public @Nullable String getPassword() {
+			return getContainer().getPassword();
+		}
+
+	}
+
+}

module/spring-boot-amqp/src/main/java/org/springframework/boot/amqp/testcontainers/ArtemisAmqpContainerConnectionDetailsFactory.java

@@ -0,0 +1,60 @@
+/*
+ * Copyright 2012-present the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.boot.amqp.testcontainers;
+
+import org.jspecify.annotations.Nullable;
+import org.testcontainers.activemq.ArtemisContainer;
+
+import org.springframework.boot.amqp.autoconfigure.AmqpConnectionDetails;
+import org.springframework.boot.testcontainers.service.connection.ContainerConnectionDetailsFactory;
+import org.springframework.boot.testcontainers.service.connection.ContainerConnectionSource;
+import org.springframework.boot.testcontainers.service.connection.ServiceConnection;
+
+/**
+ * {@link ContainerConnectionDetailsFactory} to create {@link AmqpConnectionDetails} from
+ * a {@link ServiceConnection @ServiceConnection}-annotated {@link ArtemisContainer}.
+ *
+ * @author Stephane Nicoll
+ */
+class ArtemisAmqpContainerConnectionDetailsFactory
+		extends AbstractAmqpContainerConnectionDetailsFactory<ArtemisContainer> {
+
+	@Override
+	protected @Nullable AmqpConnectionDetails getContainerConnectionDetails(
+			ContainerConnectionSource<ArtemisContainer> source) {
+		return new ArtemisAmqpContainerConnectionDetails(source);
+	}
+
+	static class ArtemisAmqpContainerConnectionDetails extends AmqpContainerConnectionDetails<ArtemisContainer> {
+
+		ArtemisAmqpContainerConnectionDetails(ContainerConnectionSource<ArtemisContainer> source) {
+			super(source);
+		}
+
+		@Override
+		public @Nullable String getUsername() {
+			return getContainer().getUser();
+		}
+
+		@Override
+		public @Nullable String getPassword() {
+			return getContainer().getPassword();
+		}
+
+	}
+
+}

module/spring-boot-amqp/src/main/java/org/springframework/boot/amqp/testcontainers/RabbitMqAmqpContainerConnectionDetailsFactory.java

@@ -0,0 +1,71 @@
+/*
+ * Copyright 2012-present the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.boot.amqp.testcontainers;
+
+import java.net.URI;
+
+import org.testcontainers.rabbitmq.RabbitMQContainer;
+
+import org.springframework.boot.amqp.autoconfigure.AmqpConnectionDetails;
+import org.springframework.boot.testcontainers.service.connection.ContainerConnectionDetailsFactory;
+import org.springframework.boot.testcontainers.service.connection.ContainerConnectionSource;
+import org.springframework.boot.testcontainers.service.connection.ServiceConnection;
+
+/**
+ * {@link ContainerConnectionDetailsFactory} to create {@link AmqpConnectionDetails} from
+ * a {@link ServiceConnection @ServiceConnection}-annotated {@link RabbitMQContainer}.
+ *
+ * @author Stephane Nicoll
+ */
+class RabbitMqAmqpContainerConnectionDetailsFactory
+		extends AbstractAmqpContainerConnectionDetailsFactory<RabbitMQContainer> {
+
+	@Override
+	protected AmqpConnectionDetails getContainerConnectionDetails(ContainerConnectionSource<RabbitMQContainer> source) {
+		return new RabbitMqAmqpContainerConnectionDetails(source);
+	}
+
+	/**
+	 * {@link AmqpConnectionDetails} for RabbitMQ backed by a
+	 * {@link ContainerConnectionSource}.
+	 */
+	static final class RabbitMqAmqpContainerConnectionDetails
+			extends AmqpContainerConnectionDetails<RabbitMQContainer> {
+
+		private RabbitMqAmqpContainerConnectionDetails(ContainerConnectionSource<RabbitMQContainer> source) {
+			super(source);
+		}
+
+		@Override
+		public Address getAddress() {
+			URI uri = URI.create(getContainer().getAmqpUrl());
+			return new Address(uri.getHost(), uri.getPort());
+		}
+
+		@Override
+		public String getUsername() {
+			return getContainer().getAdminUsername();
+		}
+
+		@Override
+		public String getPassword() {
+			return getContainer().getAdminPassword();
+		}
+
+	}
+
+}

module/spring-boot-amqp/src/main/java/org/springframework/boot/amqp/testcontainers/package-info.java

@@ -0,0 +1,23 @@
+/*
+ * Copyright 2012-present the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+/**
+ * Support for testcontainers generic AMQP service connections.
+ */
+@NullMarked
+package org.springframework.boot.amqp.testcontainers;
+
+import org.jspecify.annotations.NullMarked;

module/spring-boot-amqp/src/main/resources/META-INF/spring.factories

@@ -0,0 +1,8 @@
+# Connection Details Factories
+org.springframework.boot.autoconfigure.service.connection.ConnectionDetailsFactory=\
+org.springframework.boot.amqp.docker.compose.ActiveMqAmqpDockerComposeConnectionDetailsFactory,\
+org.springframework.boot.amqp.docker.compose.ArtemisAmqpDockerComposeConnectionDetailsFactory,\
+org.springframework.boot.amqp.docker.compose.RabbitMqAmqpDockerComposeConnectionDetailsFactory,\
+org.springframework.boot.amqp.testcontainers.ActiveMqAmqpContainerConnectionDetailsFactory,\
+org.springframework.boot.amqp.testcontainers.ArtemisAmqpContainerConnectionDetailsFactory,\
+org.springframework.boot.amqp.testcontainers.RabbitMqAmqpContainerConnectionDetailsFactory

module/spring-boot-amqp/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports

@@ -0,0 +1 @@
+org.springframework.boot.amqp.autoconfigure.AmqpAutoConfiguration

module/spring-boot-amqp/src/test/java/org/springframework/boot/amqp/autoconfigure/AmqpAutoConfigurationTests.java

@@ -0,0 +1,381 @@
+/*
+ * Copyright 2012-present the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.boot.amqp.autoconfigure;
+
+import java.time.Duration;
+import java.util.Objects;
+import java.util.function.Consumer;
+
+import org.apache.qpid.protonj2.client.Client;
+import org.apache.qpid.protonj2.client.ConnectionOptions;
+import org.assertj.core.api.InstanceOfAssertFactories;
+import org.junit.jupiter.api.Test;
+import org.mockito.InOrder;
+import tools.jackson.databind.json.JsonMapper;
+
+import org.springframework.amqp.client.AmqpClient;
+import org.springframework.amqp.client.AmqpConnectionFactory;
+import org.springframework.amqp.client.annotation.AmqpListener;
+import org.springframework.amqp.client.config.AmqpDefaultConfiguration;
+import org.springframework.amqp.client.config.AmqpMessageListenerContainerFactory;
+import org.springframework.amqp.client.config.MethodAmqpListenerEndpoint;
+import org.springframework.amqp.client.config.MethodAmqpMessageListenerContainerFactory;
+import org.springframework.amqp.client.listener.AmqpMessageListenerContainer;
+import org.springframework.amqp.support.converter.JacksonJsonMessageConverter;
+import org.springframework.amqp.support.converter.MessageConverter;
+import org.springframework.boot.amqp.autoconfigure.AmqpConnectionDetails.Address;
+import org.springframework.boot.amqp.autoconfigure.AmqpProperties.Listener;
+import org.springframework.boot.autoconfigure.AutoConfigurations;
+import org.springframework.boot.jackson.autoconfigure.JacksonAutoConfiguration;
+import org.springframework.boot.test.context.runner.ApplicationContextRunner;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.core.annotation.Order;
+import org.springframework.util.ReflectionUtils;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.BDDMockito.given;
+import static org.mockito.BDDMockito.then;
+import static org.mockito.Mockito.inOrder;
+import static org.mockito.Mockito.mock;
+
+/**
+ * Tests for {@link AmqpAutoConfiguration}.
+ *
+ * @author Stephane Nicoll
+ */
+class AmqpAutoConfigurationTests {
+
+	private final ApplicationContextRunner contextRunner = new ApplicationContextRunner()
+		.withConfiguration(AutoConfigurations.of(AmqpAutoConfiguration.class));
+
+	@Test
+	void autoConfigurationConfiguresConnectionFactoryWithDefaultSettings() {
+		this.contextRunner.run((context) -> {
+			AmqpConnectionFactory connectionFactory = context.getBean(AmqpConnectionFactory.class);
+			assertThat(connectionFactory).hasFieldOrPropertyWithValue("host", "localhost");
+			assertThat(connectionFactory).hasFieldOrPropertyWithValue("port", 5672);
+			assertThat(connectionFactory)
+				.extracting("connectionOptions", InstanceOfAssertFactories.type(ConnectionOptions.class))
+				.satisfies((connectOptions) -> {
+					assertThat(connectOptions).hasFieldOrPropertyWithValue("user", null);
+					assertThat(connectOptions).hasFieldOrPropertyWithValue("password", null);
+				});
+		});
+	}
+
+	@Test
+	void autoConfigurationConfiguresConnectionFactoryWithOverrides() {
+		this.contextRunner
+			.withPropertyValues("spring.amqp.host=amqp.example.com", "spring.amqp.port=1234",
+					"spring.amqp.username=user", "spring.amqp.password=secret")
+			.run((context) -> {
+				AmqpConnectionFactory connectionFactory = context.getBean(AmqpConnectionFactory.class);
+				assertThat(connectionFactory).hasFieldOrPropertyWithValue("host", "amqp.example.com");
+				assertThat(connectionFactory).hasFieldOrPropertyWithValue("port", 1234);
+				assertThat(connectionFactory)
+					.extracting("connectionOptions", InstanceOfAssertFactories.type(ConnectionOptions.class))
+					.satisfies((connectOptions) -> {
+						assertThat(connectOptions).hasFieldOrPropertyWithValue("user", "user");
+						assertThat(connectOptions).hasFieldOrPropertyWithValue("password", "secret");
+					});
+			});
+	}
+
+	@Test
+	void autoConfigurationUsesAmqpConnectionDetailsForConnectionFactory() {
+		AmqpConnectionDetails connectionDetails = mock();
+		given(connectionDetails.getAddress()).willReturn(new Address("details.example.com", 9999));
+		given(connectionDetails.getUsername()).willReturn("from-details");
+		given(connectionDetails.getPassword()).willReturn("details-secret");
+		this.contextRunner.withBean(AmqpConnectionDetails.class, () -> connectionDetails)
+			.withPropertyValues("spring.amqp.host=ignored.example.com", "spring.amqp.port=1111",
+					"spring.amqp.username=ignored", "spring.amqp.password=ignored")
+			.run((context) -> {
+				AmqpConnectionFactory connectionFactory = context.getBean(AmqpConnectionFactory.class);
+				assertThat(connectionFactory).hasFieldOrPropertyWithValue("host", "details.example.com");
+				assertThat(connectionFactory).hasFieldOrPropertyWithValue("port", 9999);
+				assertThat(connectionFactory)
+					.extracting("connectionOptions", InstanceOfAssertFactories.type(ConnectionOptions.class))
+					.satisfies((connectOptions) -> {
+						assertThat(connectOptions).hasFieldOrPropertyWithValue("user", "from-details");
+						assertThat(connectOptions).hasFieldOrPropertyWithValue("password", "details-secret");
+					});
+			});
+	}
+
+	@Test
+	void autoConfigurationAppliesConnectionOptionsCustomizer() {
+		this.contextRunner
+			.withBean(ConnectionOptionsCustomizer.class, () -> (connectOptions) -> connectOptions.user("custom-user"))
+			.withPropertyValues("spring.amqp.username=user")
+			.run((context) -> {
+				AmqpConnectionFactory connectionFactory = context.getBean(AmqpConnectionFactory.class);
+				assertThat(connectionFactory)
+					.extracting("connectionOptions", InstanceOfAssertFactories.type(ConnectionOptions.class))
+					.satisfies((connectOptions) -> assertThat(connectOptions).hasFieldOrPropertyWithValue("user",
+							"custom-user"));
+			});
+	}
+
+	@Test
+	void autoConfigurationAppliesConnectionOptionsCustomizersInOrder() {
+		this.contextRunner.withUserConfiguration(ConnectionOptionsCustomizersConfiguration.class).run((context) -> {
+			ConnectionOptionsCustomizer first = context.getBean("first", ConnectionOptionsCustomizer.class);
+			ConnectionOptionsCustomizer second = context.getBean("second", ConnectionOptionsCustomizer.class);
+			InOrder ordered = inOrder(first, second);
+			ordered.verify(first).customize(any());
+			ordered.verify(second).customize(any());
+		});
+	}
+
+	@Test
+	void autoConfigurationBacksOffWhenUserProvidesProtonClient() {
+		Client protonClient = mock();
+		this.contextRunner.withBean(Client.class, () -> protonClient).run((context) -> {
+			assertThat(context).hasSingleBean(Client.class);
+			assertThat(context.getBean(Client.class)).isSameAs(protonClient);
+		});
+	}
+
+	@Test
+	void autoConfigurationBacksOffWhenUserProvidesAmqpConnectionDetails() {
+		AmqpConnectionDetails connectionDetails = mock();
+		given(connectionDetails.getAddress()).willReturn(new Address("details.example.com", 9999));
+		this.contextRunner.withBean(AmqpConnectionDetails.class, () -> connectionDetails).run((context) -> {
+			assertThat(context).hasSingleBean(AmqpConnectionDetails.class);
+			assertThat(context.getBean(AmqpConnectionDetails.class)).isSameAs(connectionDetails);
+		});
+	}
+
+	@Test
+	void autoConfigurationBacksOffWhenUserProvidesAmqpConnectionFactory() {
+		AmqpConnectionFactory amqpConnectionFactory = mock();
+		this.contextRunner.withBean(AmqpConnectionFactory.class, () -> amqpConnectionFactory).run((context) -> {
+			assertThat(context).hasSingleBean(AmqpConnectionFactory.class);
+			assertThat(context.getBean(AmqpConnectionFactory.class)).isSameAs(amqpConnectionFactory);
+		});
+	}
+
+	@Test
+	void autoConfigurationConfiguresAmqpClientWithOverrides() {
+		this.contextRunner
+			.withPropertyValues("spring.amqp.client.default-to-address=/queues/default_queue",
+					"spring.amqp.client.completion-timeout=20s")
+			.run((context) -> {
+				AmqpClient amqpClient = context.getBean(AmqpClient.class);
+				assertThat(amqpClient).hasFieldOrPropertyWithValue("defaultToAddress", "/queues/default_queue");
+				assertThat(amqpClient).hasFieldOrPropertyWithValue("completionTimeout", Duration.ofSeconds(20));
+			});
+	}
+
+	@Test
+	void autoConfigurationConfiguresAmqpClientWithJacksonMessageConverter() {
+		this.contextRunner.withConfiguration(AutoConfigurations.of(JacksonAutoConfiguration.class)).run((context) -> {
+			assertThat(context).hasSingleBean(JsonMapper.class);
+			AmqpClient amqpClient = context.getBean(AmqpClient.class);
+			assertThat(amqpClient)
+				.extracting("messageConverter", InstanceOfAssertFactories.type(MessageConverter.class))
+				.satisfies((messageConverter) -> assertThat(messageConverter)
+					.isInstanceOf(JacksonJsonMessageConverter.class)
+					.hasFieldOrPropertyWithValue("objectMapper", context.getBean(JsonMapper.class)));
+		});
+	}
+
+	@Test
+	void autoConfigurationConfiguresAmqpClientWithMessageConverter() {
+		MessageConverter messageConverter = mock(MessageConverter.class);
+		this.contextRunner.withBean(MessageConverter.class, () -> messageConverter).run((context) -> {
+			AmqpClient amqpClient = context.getBean(AmqpClient.class);
+			assertThat(amqpClient).hasFieldOrPropertyWithValue("messageConverter", messageConverter);
+		});
+	}
+
+	@Test
+	void autoConfigurationBacksOffWhenUserProvidesAmqpClient() {
+		AmqpClient amqpClient = mock();
+		this.contextRunner.withBean(AmqpClient.class, () -> amqpClient).run((context) -> {
+			assertThat(context).hasSingleBean(AmqpClient.class);
+			assertThat(context.getBean(AmqpClient.class)).isSameAs(amqpClient);
+		});
+	}
+
+	@Test
+	void autoConfigurationUsesUserMessageConverterWhenDefinedAlongsideJackson() {
+		MessageConverter messageConverter = mock(MessageConverter.class);
+		this.contextRunner.withConfiguration(AutoConfigurations.of(JacksonAutoConfiguration.class))
+			.withBean(MessageConverter.class, () -> messageConverter)
+			.run((context) -> {
+				assertThat(context).doesNotHaveBean(JacksonJsonMessageConverter.class);
+				AmqpClient amqpClient = context.getBean(AmqpClient.class);
+				assertThat(amqpClient).hasFieldOrPropertyWithValue("messageConverter", messageConverter);
+			});
+	}
+
+	@Test
+	void autoConfigurationAppliesAmqpClientCustomizer() {
+		this.contextRunner
+			.withBean(AmqpClientCustomizer.class,
+					() -> (client) -> client.defaultToAddress("/queues/custom_default_queue"))
+			.withPropertyValues("spring.amqp.client.default-to-address=/queues/default_queue")
+			.run((context) -> {
+				AmqpClient amqpClient = context.getBean(AmqpClient.class);
+				assertThat(amqpClient).hasFieldOrPropertyWithValue("defaultToAddress", "/queues/custom_default_queue");
+			});
+	}
+
+	@Test
+	void autoConfigurationAppliesAmqpClientCustomizersInOrder() {
+		this.contextRunner.withUserConfiguration(AmqpClientCustomizersConfiguration.class).run((context) -> {
+			AmqpClientCustomizer first = context.getBean("first", AmqpClientCustomizer.class);
+			AmqpClientCustomizer second = context.getBean("second", AmqpClientCustomizer.class);
+			InOrder ordered = inOrder(first, second);
+			ordered.verify(first).customize(any());
+			ordered.verify(second).customize(any());
+		});
+	}
+
+	@Test
+	void autoConfigureCreatesDefaultContainerFactory() {
+		this.contextRunner.run((context) -> assertThat(context).hasSingleBean(AmqpMessageListenerContainerFactory.class)
+			.hasSingleBean(MethodAmqpMessageListenerContainerFactory.class)
+			.hasBean(AmqpDefaultConfiguration.DEFAULT_AMQP_LISTENER_CONTAINER_FACTORY_BEAN_NAME));
+	}
+
+	@Test
+	void autoConfigureConfiguresContainerFactoryWithDefaultSettings() {
+		this.contextRunner.run((context) -> {
+			AmqpMessageListenerContainerFactory factory = context.getBean(AmqpMessageListenerContainerFactory.class);
+			TestAmqpListenerEndpoint endpoint = new TestAmqpListenerEndpoint();
+			AmqpMessageListenerContainer container = factory.createContainer(endpoint);
+			Listener listener = new AmqpProperties().getListener();
+			assertThat(container).satisfies(listenerContainerConfiguration(listener.isAutoStartup(),
+					listener.getConcurrency(), listener.getInitialCredits(), listener.getReceiveTimeout(),
+					listener.getGracefulShutdownPeriod()));
+			assertThat(endpoint.isDefaultRequeueRejected()).isEqualTo(listener.isDefaultRequeueRejected());
+		});
+	}
+
+	@Test
+	void autoConfigureConfiguresContainerFactoryWithOverrides() {
+		this.contextRunner
+			.withPropertyValues("spring.amqp.listener.auto-startup=false", "spring.amqp.listener.concurrency=3",
+					"spring.amqp.listener.initial-credits=500", "spring.amqp.listener.receive-timeout=25s",
+					"spring.amqp.listener.graceful-shutdown-period=35s",
+					"spring.amqp.listener.default-requeue-rejected=false")
+			.run((context) -> {
+				AmqpMessageListenerContainerFactory factory = context
+					.getBean(AmqpMessageListenerContainerFactory.class);
+				TestAmqpListenerEndpoint endpoint = new TestAmqpListenerEndpoint();
+				AmqpMessageListenerContainer container = factory.createContainer(endpoint);
+				assertThat(container).satisfies(
+						listenerContainerConfiguration(false, 3, 500, Duration.ofSeconds(25), Duration.ofSeconds(35)));
+				assertThat(endpoint.isDefaultRequeueRejected()).isEqualTo(false);
+			});
+	}
+
+	@Test
+	void autoConfigurationCreatesContainerUsingAmqpMessageListenerContainerFactory() {
+		AmqpMessageListenerContainerFactory factory = mock(AmqpMessageListenerContainerFactory.class);
+		this.contextRunner
+			.withBean(AmqpDefaultConfiguration.DEFAULT_AMQP_LISTENER_CONTAINER_FACTORY_BEAN_NAME,
+					AmqpMessageListenerContainerFactory.class, () -> factory)
+			.withUserConfiguration(TestListener.class)
+			.run((context) -> then(factory).should().createContainer(any()));
+	}
+
+	private Consumer<AmqpMessageListenerContainer> listenerContainerConfiguration(boolean autoStartup, int concurrency,
+			int initialCredit, Duration receiveTimeout, Duration gracefulShutdownPeriod) {
+		return (container) -> assertThat(container).hasFieldOrPropertyWithValue("autoStartup", autoStartup)
+			.hasFieldOrPropertyWithValue("consumersPerQueue", concurrency)
+			.hasFieldOrPropertyWithValue("initialCredits", initialCredit)
+			.hasFieldOrPropertyWithValue("receiveTimeout", receiveTimeout)
+			.hasFieldOrPropertyWithValue("gracefulShutdownPeriod", gracefulShutdownPeriod);
+	}
+
+	@Test
+	void autoConfigurationBacksOffWhenUserProvidesContainerFactory() {
+		AmqpMessageListenerContainerFactory containerFactory = mock(AmqpMessageListenerContainerFactory.class);
+		this.contextRunner
+			.withBean(AmqpDefaultConfiguration.DEFAULT_AMQP_LISTENER_CONTAINER_FACTORY_BEAN_NAME,
+					AmqpMessageListenerContainerFactory.class, () -> containerFactory)
+			.run((context) -> {
+				assertThat(context).hasSingleBean(AmqpMessageListenerContainerFactory.class)
+					.doesNotHaveBean(MethodAmqpMessageListenerContainerFactory.class);
+				assertThat(context.getBean(AmqpMessageListenerContainerFactory.class)).isEqualTo(containerFactory);
+			});
+	}
+
+	@Configuration(proxyBeanMethods = false)
+	static class ConnectionOptionsCustomizersConfiguration {
+
+		@Bean
+		@Order(2)
+		ConnectionOptionsCustomizer second() {
+			return mock();
+		}
+
+		@Bean
+		@Order(1)
+		ConnectionOptionsCustomizer first() {
+			return mock();
+		}
+
+	}
+
+	@Configuration(proxyBeanMethods = false)
+	static class AmqpClientCustomizersConfiguration {
+
+		@Bean
+		@Order(2)
+		AmqpClientCustomizer second() {
+			return mock();
+		}
+
+		@Bean
+		@Order(1)
+		AmqpClientCustomizer first() {
+			return mock();
+		}
+
+	}
+
+	static class TestListener {
+
+		@AmqpListener
+		void listen(String in) {
+
+		}
+
+	}
+
+	static class TestAmqpListenerEndpoint extends MethodAmqpListenerEndpoint {
+
+		TestAmqpListenerEndpoint() {
+			super(TestListener.class,
+					Objects.requireNonNull(ReflectionUtils.findMethod(TestListener.class, "listen", String.class)));
+		}
+
+		@Override
+		protected boolean isDefaultRequeueRejected() {
+			return super.isDefaultRequeueRejected();
+		}
+
+	}
+
+}

module/spring-boot-amqp/src/test/java/org/springframework/boot/amqp/autoconfigure/AmqpMessageListenerContainerFactoryConfigurerTests.java

@@ -0,0 +1,101 @@
+/*
+ * Copyright 2012-present the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.boot.amqp.autoconfigure;
+
+import java.time.Duration;
+import java.util.concurrent.Executor;
+
+import org.junit.jupiter.api.Test;
+
+import org.springframework.amqp.client.config.AmqpMessageListenerContainerFactory;
+import org.springframework.amqp.client.config.MethodAmqpMessageListenerContainerFactory;
+import org.springframework.amqp.client.listener.AmqpListenerErrorHandler;
+import org.springframework.amqp.support.converter.MessageConverter;
+import org.springframework.boot.amqp.autoconfigure.AmqpProperties.Listener;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.BDDMockito.then;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+
+/**
+ * Tests for {@link AmqpMessageListenerContainerFactoryConfigurer}.
+ *
+ * @author Stephane Nicoll
+ */
+class AmqpMessageListenerContainerFactoryConfigurerTests {
+
+	@Test
+	void configureAppliesListenerPropertiesToAmqpMessageListenerContainerFactory() {
+		AmqpProperties properties = new AmqpProperties();
+		Listener listener = properties.getListener();
+		listener.setAutoStartup(false);
+		listener.setConcurrency(3);
+		listener.setInitialCredits(50);
+		listener.setReceiveTimeout(Duration.ofMillis(500));
+		listener.setGracefulShutdownPeriod(Duration.ofSeconds(60));
+		Executor taskExecutor = mock(Executor.class);
+		AmqpMessageListenerContainerFactoryConfigurer configurer = new AmqpMessageListenerContainerFactoryConfigurer(
+				properties);
+		configurer.setTaskExecutor(taskExecutor);
+
+		AmqpMessageListenerContainerFactory factory = mock(AmqpMessageListenerContainerFactory.class);
+		configurer.configure(factory);
+		then(factory).should().setAutoStartup(false);
+		then(factory).should().setConcurrency(3);
+		then(factory).should().setInitialCredits(50);
+		then(factory).should().setReceiveTimeout(Duration.ofMillis(500));
+		then(factory).should().setGracefulShutdownPeriod(Duration.ofSeconds(60));
+		then(factory).should().setTaskExecutor(taskExecutor);
+	}
+
+	@Test
+	void configureAppliesMethodAmqpMessageListenerContainerFactoryProperties() {
+		AmqpProperties properties = new AmqpProperties();
+		properties.getListener().setDefaultRequeueRejected(false);
+
+		AmqpMessageListenerContainerFactoryConfigurer configurer = new AmqpMessageListenerContainerFactoryConfigurer(
+				properties);
+		MessageConverter messageConverter = mock(MessageConverter.class);
+		AmqpListenerErrorHandler errorHandler = mock(AmqpListenerErrorHandler.class);
+		configurer.setMessageConverter(messageConverter);
+		configurer.setListenerErrorHandler(errorHandler);
+
+		MethodAmqpMessageListenerContainerFactory factory = mock(MethodAmqpMessageListenerContainerFactory.class);
+		configurer.configure(factory);
+
+		then(factory).should().setDefaultRequeueRejected(false);
+		then(factory).should().setMessageConverter(messageConverter);
+		then(factory).should().setListenerErrorHandler(errorHandler);
+	}
+
+	@Test
+	void configureSkipsOptionalDependenciesWhenNotSet() {
+		AmqpProperties properties = new AmqpProperties();
+		AmqpMessageListenerContainerFactoryConfigurer configurer = new AmqpMessageListenerContainerFactoryConfigurer(
+				properties);
+
+		MethodAmqpMessageListenerContainerFactory factory = mock(MethodAmqpMessageListenerContainerFactory.class);
+		configurer.configure(factory);
+
+		then(factory).should(never()).setTaskExecutor(any());
+		then(factory).should(never()).setMessageConverter(any());
+		then(factory).should(never()).setListenerErrorHandler(any());
+		then(factory).should().setDefaultRequeueRejected(properties.getListener().isDefaultRequeueRejected());
+	}
+
+}

module/spring-boot-amqp/src/test/java/org/springframework/boot/amqp/docker/compose/ActiveMqAmqpDockerComposeConnectionDetailsFactoryTests.java

@@ -0,0 +1,67 @@
+/*
+ * Copyright 2012-present the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.boot.amqp.docker.compose;
+
+import java.util.Collections;
+import java.util.Map;
+
+import org.junit.jupiter.api.Test;
+
+import org.springframework.boot.docker.compose.core.RunningService;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.BDDMockito.given;
+import static org.mockito.Mockito.mock;
+
+/**
+ * Tests for {@link ActiveMqAmqpDockerComposeConnectionDetailsFactory}.
+ *
+ * @author Stephane Nicoll
+ */
+class ActiveMqAmqpDockerComposeConnectionDetailsFactoryTests {
+
+	@Test
+	void getUserWhenHasNoActiveMqUser() {
+		AmqpEnvironment environment = getAmqpEnvironment(Collections.emptyMap());
+		assertThat(environment.getUsername()).isNull();
+	}
+
+	@Test
+	void getUserWhenHasActiveMqUser() {
+		AmqpEnvironment environment = getAmqpEnvironment(Map.of("ACTIVEMQ_CONNECTION_USER", "me"));
+		assertThat(environment.getUsername()).isEqualTo("me");
+	}
+
+	@Test
+	void getPasswordWhenHasNoActiveMqPassword() {
+		AmqpEnvironment environment = getAmqpEnvironment(Collections.emptyMap());
+		assertThat(environment.getPassword()).isNull();
+	}
+
+	@Test
+	void getPasswordWhenHasActiveMqPassword() {
+		AmqpEnvironment environment = getAmqpEnvironment(Map.of("ACTIVEMQ_CONNECTION_PASSWORD", "secret"));
+		assertThat(environment.getPassword()).isEqualTo("secret");
+	}
+
+	private AmqpEnvironment getAmqpEnvironment(Map<String, String> env) {
+		RunningService runningService = mock(RunningService.class);
+		given(runningService.env()).willReturn(env);
+		return new ActiveMqAmqpDockerComposeConnectionDetailsFactory().getAmqpEnvironment(runningService);
+	}
+
+}

module/spring-boot-amqp/src/test/java/org/springframework/boot/amqp/docker/compose/ArtemisAmqpDockerComposeConnectionDetailsFactoryTests.java

@@ -0,0 +1,67 @@
+/*
+ * Copyright 2012-present the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.boot.amqp.docker.compose;
+
+import java.util.Collections;
+import java.util.Map;
+
+import org.junit.jupiter.api.Test;
+
+import org.springframework.boot.docker.compose.core.RunningService;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.BDDMockito.given;
+import static org.mockito.Mockito.mock;
+
+/**
+ * Tests for {@link ArtemisAmqpDockerComposeConnectionDetailsFactory}.
+ *
+ * @author Stephane Nicoll
+ */
+class ArtemisAmqpDockerComposeConnectionDetailsFactoryTests {
+
+	@Test
+	void getUserWhenHasNoActiveMqUser() {
+		AmqpEnvironment environment = getAmqpEnvironment(Collections.emptyMap());
+		assertThat(environment.getUsername()).isNull();
+	}
+
+	@Test
+	void getUserWhenHasActiveMqUser() {
+		AmqpEnvironment environment = getAmqpEnvironment(Map.of("ARTEMIS_USER", "me"));
+		assertThat(environment.getUsername()).isEqualTo("me");
+	}
+
+	@Test
+	void getPasswordWhenHasNoActiveMqPassword() {
+		AmqpEnvironment environment = getAmqpEnvironment(Collections.emptyMap());
+		assertThat(environment.getPassword()).isNull();
+	}
+
+	@Test
+	void getPasswordWhenHasActiveMqPassword() {
+		AmqpEnvironment environment = getAmqpEnvironment(Map.of("ARTEMIS_PASSWORD", "secret"));
+		assertThat(environment.getPassword()).isEqualTo("secret");
+	}
+
+	private AmqpEnvironment getAmqpEnvironment(Map<String, String> env) {
+		RunningService runningService = mock(RunningService.class);
+		given(runningService.env()).willReturn(env);
+		return new ArtemisAmqpDockerComposeConnectionDetailsFactory().getAmqpEnvironment(runningService);
+	}
+
+}

module/spring-boot-amqp/src/test/java/org/springframework/boot/amqp/docker/compose/RabbitMqAmqpDockerComposeConnectionDetailsFactoryTests.java

@@ -0,0 +1,79 @@
+/*
+ * Copyright 2012-present the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.boot.amqp.docker.compose;
+
+import java.util.Collections;
+import java.util.Map;
+
+import org.junit.jupiter.api.Test;
+
+import org.springframework.boot.docker.compose.core.RunningService;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.BDDMockito.given;
+import static org.mockito.Mockito.mock;
+
+/**
+ * Tests for {@link RabbitMqAmqpDockerComposeConnectionDetailsFactory}.
+ *
+ * @author Stephane Nicoll
+ */
+class RabbitMqAmqpDockerComposeConnectionDetailsFactoryTests {
+
+	@Test
+	void getUsernameWhenNoRabbitMqDefaultUser() {
+		AmqpEnvironment environment = getAmqpEnvironment(Collections.emptyMap());
+		assertThat(environment.getUsername()).isEqualTo("guest");
+	}
+
+	@Test
+	void getUsernameWhenHasRabbitMqDefaultUser() {
+		AmqpEnvironment environment = getAmqpEnvironment(Map.of("RABBITMQ_DEFAULT_USER", "me"));
+		assertThat(environment.getUsername()).isEqualTo("me");
+	}
+
+	@Test
+	void getUsernameWhenHasRabbitMqUsername() {
+		AmqpEnvironment environment = getAmqpEnvironment(Map.of("RABBITMQ_USERNAME", "me"));
+		assertThat(environment.getUsername()).isEqualTo("me");
+	}
+
+	@Test
+	void getPasswordWhenNoRabbitMqDefaultPass() {
+		AmqpEnvironment environment = getAmqpEnvironment(Collections.emptyMap());
+		assertThat(environment.getPassword()).isEqualTo("guest");
+	}
+
+	@Test
+	void getPasswordWhenHasRabbitMqDefaultPass() {
+		AmqpEnvironment environment = getAmqpEnvironment(Map.of("RABBITMQ_DEFAULT_PASS", "secret"));
+		assertThat(environment.getPassword()).isEqualTo("secret");
+	}
+
+	@Test
+	void getPasswordWhenHasRabbitMqPassword() {
+		AmqpEnvironment environment = getAmqpEnvironment(Map.of("RABBITMQ_PASSWORD", "secret"));
+		assertThat(environment.getPassword()).isEqualTo("secret");
+	}
+
+	private AmqpEnvironment getAmqpEnvironment(Map<String, String> env) {
+		RunningService runningService = mock(RunningService.class);
+		given(runningService.env()).willReturn(env);
+		return new RabbitMqAmqpDockerComposeConnectionDetailsFactory().getAmqpEnvironment(runningService);
+	}
+
+}

module/spring-boot-autoconfigure-classic-modules/build.gradle

@@ -28,6 +28,9 @@ dependencies {
 	api(project(":module:spring-boot-activemq")) {
 		transitive = false
 	}
+	api(project(":module:spring-boot-amqp")) {
+		transitive = false
+	}
 	api(project(":module:spring-boot-artemis")) {
 		transitive = false
 	}

platform/spring-boot-dependencies/build.gradle

@@ -2123,6 +2123,7 @@ bom {
 				"spring-boot-activemq",
 				"spring-boot-actuator",
 				"spring-boot-actuator-autoconfigure",
+				"spring-boot-amqp",
 				"spring-boot-artemis",
 				"spring-boot-autoconfigure",
 				"spring-boot-autoconfigure-classic",
@@ -2244,6 +2245,8 @@ bom {
 				"spring-boot-starter-activemq-test",
 				"spring-boot-starter-actuator",
 				"spring-boot-starter-actuator-test",
+				"spring-boot-starter-amqp",
+				"spring-boot-starter-amqp-test",
 				"spring-boot-starter-artemis",
 				"spring-boot-starter-artemis-test",
 				"spring-boot-starter-aspectj",
@@ -2362,8 +2365,8 @@ bom {
 				"spring-boot-starter-quartz-test",
 				"spring-boot-starter-r2dbc",
 				"spring-boot-starter-r2dbc-test",
-				"spring-boot-starter-rabbitmqmq",
+				"spring-boot-starter-rabbitmq",
-				"spring-boot-starter-rabbitmqmq-test",
+				"spring-boot-starter-rabbitmq-test",
 				"spring-boot-starter-reactor-netty",
 				"spring-boot-starter-restclient",
 				"spring-boot-starter-restclient-test",

settings.gradle

@@ -80,6 +80,7 @@ include "core:spring-boot-testcontainers"
 include "module:spring-boot-activemq"
 include "module:spring-boot-actuator"
 include "module:spring-boot-actuator-autoconfigure"
+include "module:spring-boot-amqp"
 include "module:spring-boot-artemis"
 include "module:spring-boot-autoconfigure-classic"
 include "module:spring-boot-autoconfigure-classic-modules"
@@ -211,6 +212,8 @@ include "starter:spring-boot-starter-activemq"
 include "starter:spring-boot-starter-activemq-test"
 include "starter:spring-boot-starter-actuator"
 include "starter:spring-boot-starter-actuator-test"
+include "starter:spring-boot-starter-amqp"
+include "starter:spring-boot-starter-amqp-test"
 include "starter:spring-boot-starter-artemis"
 include "starter:spring-boot-starter-artemis-test"
 include "starter:spring-boot-starter-aspectj"
@@ -401,6 +404,7 @@ include ":smoke-test:spring-boot-smoke-test-actuator-extension"
 include ":smoke-test:spring-boot-smoke-test-actuator-log4j2"
 include ":smoke-test:spring-boot-smoke-test-actuator-noweb"
 include ":smoke-test:spring-boot-smoke-test-actuator-ui"
+include ":smoke-test:spring-boot-smoke-test-amqp"
 include ":smoke-test:spring-boot-smoke-test-ant"
 include ":smoke-test:spring-boot-smoke-test-artemis"
 include ":smoke-test:spring-boot-smoke-test-aspectj"

smoke-test/spring-boot-smoke-test-amqp/build.gradle

@@ -0,0 +1,41 @@
+/*
+ * Copyright 2012-present the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the License);
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+plugins {
+	id "java"
+	id "org.springframework.boot.docker-test"
+}
+
+description = "Spring Boot AMQP smoke test"
+
+dependencies {
+	implementation(project(":starter:spring-boot-starter-amqp"))
+
+	dockerTestImplementation(project(":core:spring-boot-testcontainers"))
+	dockerTestImplementation(project(":starter:spring-boot-starter-amqp-test"))
+	dockerTestImplementation(project(":test-support:spring-boot-docker-test-support"))
+	dockerTestImplementation("org.awaitility:awaitility")
+	dockerTestImplementation("org.testcontainers:testcontainers-junit-jupiter")
+	dockerTestImplementation("org.testcontainers:testcontainers-rabbitmq")
+}
+
+tasks.named("compileTestJava") {
+	options.nullability.checking = "tests"
+}
+
+tasks.named("compileDockerTestJava") {
+	options.nullability.checking = "tests"
+}

smoke-test/spring-boot-smoke-test-amqp/src/dockerTest/java/smoketest/amqp/SampleAmqpApplicationTests.java

@@ -0,0 +1,57 @@
+/*
+ * Copyright 2012-present the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package smoketest.amqp;
+
+import java.time.Duration;
+
+import org.awaitility.Awaitility;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.testcontainers.junit.jupiter.Container;
+import org.testcontainers.junit.jupiter.Testcontainers;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.boot.test.system.CapturedOutput;
+import org.springframework.boot.test.system.OutputCaptureExtension;
+import org.springframework.boot.testcontainers.service.connection.ServiceConnection;
+import org.springframework.boot.testsupport.container.RabbitMqManagementContainer;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+@SpringBootTest
+@Testcontainers(disabledWithoutDocker = true)
+@ExtendWith(OutputCaptureExtension.class)
+class SampleAmqpApplicationTests {
+
+	@Container
+	@ServiceConnection
+	static final RabbitMqManagementContainer container = new RabbitMqManagementContainer();
+
+	@BeforeAll
+	static void setupBroker() {
+		container.createQueue("foo");
+	}
+
+	@Test
+	void sendSimpleMessage(@Autowired Sender sender, CapturedOutput output) {
+		sender.send("Test message");
+		Awaitility.waitAtMost(Duration.ofMinutes(1)).untilAsserted(() -> assertThat(output).contains("Test message"));
+	}
+
+}

smoke-test/spring-boot-smoke-test-amqp/src/main/java/smoketest/amqp/Listener.java

@@ -0,0 +1,35 @@
+/*
+ * Copyright 2012-present the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package smoketest.amqp;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import org.springframework.amqp.client.annotation.AmqpListener;
+import org.springframework.stereotype.Component;
+
+@Component
+class Listener {
+
+	private static final Log logger = LogFactory.getLog(Listener.class);
+
+	@AmqpListener(addresses = "/queues/foo")
+	void process(String foo) {
+		logger.info(foo);
+	}
+
+}

smoke-test/spring-boot-smoke-test-amqp/src/main/java/smoketest/amqp/SampleAmqpApplication.java

@@ -0,0 +1,36 @@
+/*
+ * Copyright 2012-present the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package smoketest.amqp;
+
+import org.springframework.boot.ApplicationRunner;
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.context.annotation.Bean;
+
+@SpringBootApplication
+public class SampleAmqpApplication {
+
+	@Bean
+	public ApplicationRunner runner(Sender sender) {
+		return (args) -> sender.send("Hello");
+	}
+
+	public static void main(String[] args) {
+		SpringApplication.run(SampleAmqpApplication.class, args);
+	}
+
+}

smoke-test/spring-boot-smoke-test-amqp/src/main/java/smoketest/amqp/Sender.java

@@ -0,0 +1,35 @@
+/*
+ * Copyright 2012-present the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package smoketest.amqp;
+
+import org.springframework.amqp.client.AmqpClient;
+import org.springframework.stereotype.Component;
+
+@Component
+class Sender {
+
+	private final AmqpClient amqpClient;
+
+	Sender(AmqpClient amqpClient) {
+		this.amqpClient = amqpClient;
+	}
+
+	void send(String message) {
+		this.amqpClient.to("/queues/foo").body(message).send();
+	}
+
+}

smoke-test/spring-boot-smoke-test-amqp/src/main/java/smoketest/amqp/package-info.java

@@ -0,0 +1,20 @@
+/*
+ * Copyright 2012-present the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+@NullMarked
+package smoketest.amqp;
+
+import org.jspecify.annotations.NullMarked;

starter/spring-boot-starter-amqp-test/build.gradle

@@ -0,0 +1,26 @@
+/*
+ * Copyright 2012-present the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the License);
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+plugins {
+	id "org.springframework.boot.starter"
+}
+
+description = "Starter for testing generic AMQP 1.0 support"
+
+dependencies {
+	api(project(":starter:spring-boot-starter-amqp"))
+	api(project(":starter:spring-boot-starter-test"))
+}

starter/spring-boot-starter-amqp/build.gradle

@@ -0,0 +1,27 @@
+/*
+ * Copyright 2012-present the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the License);
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+plugins {
+	id "org.springframework.boot.starter"
+}
+
+description = "Starter for generic AMQP 1.0 support"
+
+dependencies {
+	api(project(":starter:spring-boot-starter"))
+
+	api(project(":module:spring-boot-amqp"))
+}

test-support/spring-boot-docker-test-support/src/main/java/org/springframework/boot/testsupport/container/RabbitMqManagementContainer.java

@@ -0,0 +1,57 @@
+/*
+ * Copyright 2012-present the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.boot.testsupport.container;
+
+import java.util.UUID;
+
+import org.testcontainers.rabbitmq.RabbitMQContainer;
+
+/**
+ * A {@link RabbitMQContainer} with management plugin enabled.
+ *
+ * @author Stephane Nicoll
+ */
+public class RabbitMqManagementContainer extends RabbitMQContainer {
+
+	public RabbitMqManagementContainer() {
+		super(TestImage.RABBITMQ.toString());
+	}
+
+	/**
+	 * Create a queue with the given name.
+	 * @param name the name of the queue
+	 * @return the AMQP target of the queue
+	 */
+	public String createQueue(String name) {
+		try {
+			execInContainer("rabbitmqadmin", "queues", "declare", "--name", name);
+			return "/queues/" + name;
+		}
+		catch (Exception ex) {
+			throw new RuntimeException("Could not create queue", ex);
+		}
+	}
+
+	/**
+	 * Create a queue with a random, unique, name.
+	 * @return the AMQP target of the queue
+	 */
+	public String createRandomQueue() {
+		return createQueue(UUID.randomUUID().toString());
+	}
+
+}

test-support/spring-boot-docker-test-support/src/main/java/org/springframework/boot/testsupport/container/TestImage.java

@@ -260,9 +260,10 @@ public enum TestImage {
 				.withStartupTimeout(Duration.ofMinutes(3))),
 
 	/**
-	 * A container image suitable for testing RabbitMQ.
+	 * A container image suitable for testing RabbitMQ with support for both AMQP 0.9 and
+	 * 1.0. Queues can be declared using {@code rabbitmqadmin}.
 	 */
-	RABBITMQ("rabbitmq", "3.11-alpine", () -> RabbitMQContainer.class,
+	RABBITMQ("rabbitmq", "4.2-management", () -> RabbitMQContainer.class,
 			(container) -> ((RabbitMQContainer) container).withStartupTimeout(Duration.ofMinutes(4))),
 
 	/**