Introduce spring-rabbitmq-client for RabbitMQ AMQP 1.0

Fixes: https://github.com/spring-projects/spring-amqp/issues/2991
Fixes: https://github.com/spring-projects/spring-amqp/issues/2992
Fixes: https://github.com/spring-projects/spring-amqp/issues/2993

* Add `QueueInformation.type` since RabbitMQ AMQP Client exposes such an info

build.gradle

@@ -59,6 +59,7 @@ ext {
 	micrometerVersion = '1.15.0-SNAPSHOT'
 	micrometerTracingVersion = '1.5.0-SNAPSHOT'
 	mockitoVersion = '5.15.2'
+	rabbitmqAmqpClientVersion = '0.4.0'
 	rabbitmqStreamVersion = '0.22.0'
 	rabbitmqVersion = '5.24.0'
 	reactorVersion = '2024.0.3'
@@ -472,6 +473,28 @@ project('spring-rabbit-stream') {
 	}
 }
 
+project('spring-rabbitmq-client') {
+	description = 'Spring RabbitMQ Client for AMQP 1.0'
+
+	dependencies {
+		api project(':spring-rabbit')
+		api "com.rabbitmq.client:amqp-client:$rabbitmqAmqpClientVersion"
+		api 'io.micrometer:micrometer-observation'
+
+		testApi project(':spring-rabbit-junit')
+
+		testRuntimeOnly 'com.fasterxml.jackson.core:jackson-databind'
+
+		testImplementation 'org.testcontainers:rabbitmq'
+		testImplementation 'org.testcontainers:junit-jupiter'
+		testImplementation 'org.apache.logging.log4j:log4j-slf4j-impl'
+		testImplementation 'io.micrometer:micrometer-observation-test'
+		testImplementation 'io.micrometer:micrometer-tracing-bridge-brave'
+		testImplementation 'io.micrometer:micrometer-tracing-test'
+		testImplementation 'io.micrometer:micrometer-tracing-integration-test'
+	}
+}
+
 project('spring-rabbit-junit') {
 	description = 'Spring Rabbit JUnit Support'
 

spring-amqp/src/main/java/org/springframework/amqp/core/QueueInformation.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2019-2024 the original author or authors.
+ * Copyright 2019-2025 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.
@@ -21,6 +21,8 @@ package org.springframework.amqp.core;
  *
  * @author Gary Russell
  * @author Ngoc Nhan
+ * @author Artem Bilan
+ *
  * @since 2.2
  *
  */
@@ -28,11 +30,13 @@ public class QueueInformation {
 
 	private final String name;
 
-	private final int messageCount;
+	private final long messageCount;
 
 	private final int consumerCount;
 
-	public QueueInformation(String name, int messageCount, int consumerCount) {
+	private String type = "classic";
+
+	public QueueInformation(String name, long messageCount, int consumerCount) {
 		this.name = name;
 		this.messageCount = messageCount;
 		this.consumerCount = consumerCount;
@@ -42,7 +46,7 @@ public class QueueInformation {
 		return this.name;
 	}
 
-	public int getMessageCount() {
+	public long getMessageCount() {
 		return this.messageCount;
 	}
 
@@ -50,11 +54,30 @@ public class QueueInformation {
 		return this.consumerCount;
 	}
 
+	/**
+	 * Return a queue type.
+	 * {@code classic} by default since AMQP 0.9.1 protocol does not return this info in {@code DeclareOk} reply.
+	 * @return a queue type
+	 * @since 4.0
+	 */
+	public String getType() {
+		return this.type;
+	}
+
+	/**
+	 * Set a queue type.
+	 * @param type the queue type: {@code quorum}, {@code classic} or {@code stream}
+	 * @since 4.0
+	 */
+	public void setType(String type) {
+		this.type = type;
+	}
+
 	@Override
 	public int hashCode() {
 		final int prime = 31;
 		int result = 1;
-		result = prime * result + ((this.name == null) ? 0 : this.name.hashCode());
+		result = prime * result + this.name.hashCode();
 		return result;
 	}
 
@@ -70,9 +93,6 @@ public class QueueInformation {
 			return false;
 		}
 		QueueInformation other = (QueueInformation) obj;
-		if (this.name == null) {
-			return other.name == null;
-		}
 		return this.name.equals(other.name);
 	}
 

spring-amqp/src/main/java/org/springframework/amqp/utils/JavaUtils.java

@@ -44,9 +44,9 @@ public final class JavaUtils {
 	}
 
 	/**
-	 * Invoke {@link Consumer#accept(Object)} with the value if the condition is true.
+	 * Invoke {@link Consumer#accept(Object)} with the value if it is not null and the condition is true.
 	 * @param condition the condition.
-	 * @param value the value.
+	 * @param value the value. Skipped if null.
 	 * @param consumer the consumer.
 	 * @param <T> the value type.
 	 * @return this.

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitAdmin.java

@@ -119,7 +119,7 @@ public class RabbitAdmin implements AmqpAdmin, ApplicationContextAware, Applicat
 	 */
 	public static final Object QUEUE_CONSUMER_COUNT = "QUEUE_CONSUMER_COUNT";
 
-	private static final String DELAYED_MESSAGE_EXCHANGE = "x-delayed-message";
+	public static final String DELAYED_MESSAGE_EXCHANGE = "x-delayed-message";
 
 	/** Logger available to subclasses. */
 	protected final Log logger = LogFactory.getLog(getClass()); // NOSONAR

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitAdminTests.java

@@ -163,7 +163,7 @@ public class RabbitAdminTests extends NeedsManagementTests {
 		}
 	}
 
-	private int messageCount(RabbitAdmin rabbitAdmin, String queueName) {
+	private long messageCount(RabbitAdmin rabbitAdmin, String queueName) {
 		QueueInformation info = rabbitAdmin.getQueueInfo(queueName);
 		assertThat(info).isNotNull();
 		return info.getMessageCount();

spring-rabbitmq-client/src/main/java/org/springframework/amqp/rabbitmq/client/AmqpConnectionFactoryBean.java

@@ -0,0 +1,286 @@
+/*
+ * Copyright 2025 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.amqp.rabbitmq.client;
+
+import java.time.Duration;
+import java.util.function.Consumer;
+
+import javax.net.ssl.SSLContext;
+
+import com.rabbitmq.client.amqp.AddressSelector;
+import com.rabbitmq.client.amqp.BackOffDelayPolicy;
+import com.rabbitmq.client.amqp.Connection;
+import com.rabbitmq.client.amqp.ConnectionBuilder;
+import com.rabbitmq.client.amqp.ConnectionSettings;
+import com.rabbitmq.client.amqp.CredentialsProvider;
+import com.rabbitmq.client.amqp.Environment;
+import com.rabbitmq.client.amqp.OAuth2Settings;
+import com.rabbitmq.client.amqp.Resource;
+import org.jspecify.annotations.Nullable;
+
+import org.springframework.beans.factory.config.AbstractFactoryBean;
+
+/**
+ * The {@link AbstractFactoryBean} for RabbitMQ AMQP 1.0 {@link Connection}.
+ * A Spring-friendly wrapper around {@link Environment#connectionBuilder()};
+ *
+ * @author Artem Bilan
+ *
+ * @since 4.0
+ */
+public class AmqpConnectionFactoryBean extends AbstractFactoryBean<Connection> {
+
+	private final ConnectionBuilder connectionBuilder;
+
+	public AmqpConnectionFactoryBean(Environment amqpEnvironment) {
+		this.connectionBuilder = amqpEnvironment.connectionBuilder();
+	}
+
+	public AmqpConnectionFactoryBean setHost(String host) {
+		this.connectionBuilder.host(host);
+		return this;
+	}
+
+	public AmqpConnectionFactoryBean setPort(int port) {
+		this.connectionBuilder.port(port);
+		return this;
+	}
+
+	public AmqpConnectionFactoryBean setUsername(String username) {
+		this.connectionBuilder.username(username);
+		return this;
+	}
+
+	public AmqpConnectionFactoryBean setPassword(String password) {
+		this.connectionBuilder.password(password);
+		return this;
+	}
+
+	public AmqpConnectionFactoryBean setVirtualHost(String virtualHost) {
+		this.connectionBuilder.virtualHost(virtualHost);
+		return this;
+	}
+
+	public AmqpConnectionFactoryBean setUri(String uri) {
+		this.connectionBuilder.uri(uri);
+		return this;
+	}
+
+	public AmqpConnectionFactoryBean setUris(String... uris) {
+		this.connectionBuilder.uris(uris);
+		return this;
+	}
+
+	public AmqpConnectionFactoryBean setIdleTimeout(Duration idleTimeout) {
+		this.connectionBuilder.idleTimeout(idleTimeout);
+		return this;
+	}
+
+	public AmqpConnectionFactoryBean setAddressSelector(AddressSelector addressSelector) {
+		this.connectionBuilder.addressSelector(addressSelector);
+		return this;
+	}
+
+	public AmqpConnectionFactoryBean setCredentialsProvider(CredentialsProvider credentialsProvider) {
+		this.connectionBuilder.credentialsProvider(credentialsProvider);
+		return this;
+	}
+
+	public AmqpConnectionFactoryBean setSaslMechanism(SaslMechanism saslMechanism) {
+		this.connectionBuilder.saslMechanism(saslMechanism.name());
+		return this;
+	}
+
+	public AmqpConnectionFactoryBean setTls(Consumer<Tls> tlsCustomizer) {
+		tlsCustomizer.accept(new Tls(this.connectionBuilder.tls()));
+		return this;
+	}
+
+	public AmqpConnectionFactoryBean setAffinity(Consumer<Affinity> affinityCustomizer) {
+		affinityCustomizer.accept(new Affinity(this.connectionBuilder.affinity()));
+		return this;
+	}
+
+	public AmqpConnectionFactoryBean setOAuth2(Consumer<OAuth2> oauth2Customizer) {
+		oauth2Customizer.accept(new OAuth2(this.connectionBuilder.oauth2()));
+		return this;
+	}
+
+	public AmqpConnectionFactoryBean setRecovery(Consumer<Recovery> recoveryCustomizer) {
+		recoveryCustomizer.accept(new Recovery(this.connectionBuilder.recovery()));
+		return this;
+	}
+
+	public AmqpConnectionFactoryBean setListeners(Resource.StateListener... listeners) {
+		this.connectionBuilder.listeners(listeners);
+		return this;
+	}
+
+	@Override
+	public @Nullable Class<?> getObjectType() {
+		return Connection.class;
+	}
+
+	@Override
+	protected Connection createInstance() {
+		return this.connectionBuilder.build();
+	}
+
+	@Override
+	protected void destroyInstance(@Nullable Connection instance) {
+		if (instance != null) {
+			instance.close();
+		}
+	}
+
+	public enum SaslMechanism {
+
+		PLAIN, ANONYMOUS, EXTERNAL
+
+	}
+
+	public static final class Tls {
+
+		private final ConnectionSettings.TlsSettings<? extends ConnectionBuilder> tls;
+
+		private Tls(ConnectionSettings.TlsSettings<? extends ConnectionBuilder> tls) {
+			this.tls = tls;
+		}
+
+		public Tls hostnameVerification() {
+			this.tls.hostnameVerification();
+			return this;
+		}
+
+		public Tls hostnameVerification(boolean hostnameVerification) {
+			this.tls.hostnameVerification(hostnameVerification);
+			return this;
+		}
+
+		public Tls sslContext(SSLContext sslContext) {
+			this.tls.sslContext(sslContext);
+			return this;
+		}
+
+		public Tls trustEverything() {
+			this.tls.trustEverything();
+			return this;
+		}
+
+	}
+
+	public static final class Affinity {
+
+		private final ConnectionSettings.Affinity<? extends ConnectionBuilder> affinity;
+
+		private Affinity(ConnectionSettings.Affinity<? extends ConnectionBuilder> affinity) {
+			this.affinity = affinity;
+		}
+
+		public Affinity queue(String queue) {
+			this.affinity.queue(queue);
+			return this;
+		}
+
+		public Affinity operation(ConnectionSettings.Affinity.Operation operation) {
+			this.affinity.operation(operation);
+			return this;
+		}
+
+		public Affinity reuse(boolean reuse) {
+			this.affinity.reuse(reuse);
+			return this;
+		}
+
+		public Affinity strategy(ConnectionSettings.AffinityStrategy strategy) {
+			this.affinity.strategy(strategy);
+			return this;
+		}
+
+	}
+
+	public static final class OAuth2 {
+
+		private final OAuth2Settings<? extends ConnectionBuilder> oAuth2Settings;
+
+		private OAuth2(OAuth2Settings<? extends ConnectionBuilder> oAuth2Settings) {
+			this.oAuth2Settings = oAuth2Settings;
+		}
+
+		public OAuth2 tokenEndpointUri(String uri) {
+			this.oAuth2Settings.tokenEndpointUri(uri);
+			return this;
+		}
+
+		public OAuth2 clientId(String clientId) {
+			this.oAuth2Settings.clientId(clientId);
+			return this;
+		}
+
+		public OAuth2 clientSecret(String clientSecret) {
+			this.oAuth2Settings.clientSecret(clientSecret);
+			return this;
+		}
+
+		public OAuth2 grantType(String grantType) {
+			this.oAuth2Settings.grantType(grantType);
+			return this;
+		}
+
+		public OAuth2 parameter(String name, String value) {
+			this.oAuth2Settings.parameter(name, value);
+			return this;
+		}
+
+		public OAuth2 shared(boolean shared) {
+			this.oAuth2Settings.shared(shared);
+			return this;
+		}
+
+		public OAuth2 sslContext(SSLContext sslContext) {
+			this.oAuth2Settings.tls().sslContext(sslContext);
+			return this;
+		}
+
+	}
+
+	public static final class Recovery {
+
+		private final ConnectionBuilder.RecoveryConfiguration recoveryConfiguration;
+
+		private Recovery(ConnectionBuilder.RecoveryConfiguration recoveryConfiguration) {
+			this.recoveryConfiguration = recoveryConfiguration;
+		}
+
+		public Recovery activated(boolean activated) {
+			this.recoveryConfiguration.activated(activated);
+			return this;
+		}
+
+		public Recovery backOffDelayPolicy(BackOffDelayPolicy backOffDelayPolicy) {
+			this.recoveryConfiguration.backOffDelayPolicy(backOffDelayPolicy);
+			return this;
+		}
+
+		public Recovery topology(boolean activated) {
+			this.recoveryConfiguration.topology(activated);
+			return this;
+		}
+
+	}
+
+}

spring-rabbitmq-client/src/main/java/org/springframework/amqp/rabbitmq/client/RabbitAmqpAdmin.java

@@ -0,0 +1,570 @@
+/*
+ * Copyright 2025 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.amqp.rabbitmq.client;
+
+import java.io.IOException;
+import java.util.Collection;
+import java.util.LinkedList;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Properties;
+import java.util.concurrent.atomic.AtomicReference;
+
+import com.rabbitmq.client.amqp.AmqpException;
+import com.rabbitmq.client.amqp.Connection;
+import com.rabbitmq.client.amqp.Management;
+import org.jspecify.annotations.Nullable;
+
+import org.springframework.amqp.core.AmqpAdmin;
+import org.springframework.amqp.core.Binding;
+import org.springframework.amqp.core.Declarable;
+import org.springframework.amqp.core.DeclarableCustomizer;
+import org.springframework.amqp.core.Declarables;
+import org.springframework.amqp.core.Exchange;
+import org.springframework.amqp.core.Queue;
+import org.springframework.amqp.core.QueueInformation;
+import org.springframework.amqp.rabbit.core.DeclarationExceptionEvent;
+import org.springframework.amqp.rabbit.core.RabbitAdmin;
+import org.springframework.beans.factory.BeanNameAware;
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.ApplicationContextAware;
+import org.springframework.context.ApplicationEventPublisher;
+import org.springframework.context.ApplicationEventPublisherAware;
+import org.springframework.context.SmartLifecycle;
+import org.springframework.core.log.LogAccessor;
+import org.springframework.core.task.SimpleAsyncTaskExecutor;
+import org.springframework.core.task.TaskExecutor;
+import org.springframework.jmx.export.annotation.ManagedOperation;
+import org.springframework.jmx.export.annotation.ManagedResource;
+import org.springframework.util.Assert;
+
+/**
+ * The {@link AmqpAdmin} implementation for RabbitMQ AMQP 1.0 client.
+ *
+ * @author Artem Bilan
+ *
+ * @since 4.0
+ */
+@ManagedResource(description = "Admin Tasks")
+public class RabbitAmqpAdmin
+		implements AmqpAdmin, ApplicationContextAware, ApplicationEventPublisherAware, BeanNameAware, SmartLifecycle {
+
+	private static final LogAccessor LOG = new LogAccessor(RabbitAmqpAdmin.class);
+
+	public static final String QUEUE_TYPE = "QUEUE_TYPE";
+
+	private final Connection amqpConnection;
+
+	private TaskExecutor taskExecutor = new SimpleAsyncTaskExecutor();
+
+	private boolean ignoreDeclarationExceptions;
+
+	private @Nullable ApplicationContext applicationContext;
+
+	private @Nullable ApplicationEventPublisher applicationEventPublisher;
+
+	@SuppressWarnings("NullAway.Init")
+	private String beanName;
+
+	private boolean explicitDeclarationsOnly;
+
+	private boolean autoStartup = true;
+
+	private volatile @Nullable DeclarationExceptionEvent lastDeclarationExceptionEvent;
+
+	private volatile boolean running = false;
+
+	public RabbitAmqpAdmin(Connection amqpConnection) {
+		this.amqpConnection = amqpConnection;
+	}
+
+	@Override
+	public void setApplicationContext(ApplicationContext applicationContext) {
+		this.applicationContext = applicationContext;
+	}
+
+	@Override
+	public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) {
+		this.applicationEventPublisher = applicationEventPublisher;
+	}
+
+	@Override
+	public void setBeanName(String name) {
+		this.beanName = name;
+	}
+
+	public void setIgnoreDeclarationExceptions(boolean ignoreDeclarationExceptions) {
+		this.ignoreDeclarationExceptions = ignoreDeclarationExceptions;
+	}
+
+	/**
+	 * Set a task executor to use for async operations. Currently only used
+	 * with {@link #purgeQueue(String, boolean)}.
+	 * @param taskExecutor the executor to use.
+	 */
+	public void setTaskExecutor(TaskExecutor taskExecutor) {
+		Assert.notNull(taskExecutor, "'taskExecutor' cannot be null");
+		this.taskExecutor = taskExecutor;
+	}
+
+	/**
+	 * Set to true to only declare {@link Declarable} beans that are explicitly configured
+	 * to be declared by this admin.
+	 * @param explicitDeclarationsOnly true to ignore beans with no admin declaration
+	 * configuration.
+	 */
+	public void setExplicitDeclarationsOnly(boolean explicitDeclarationsOnly) {
+		this.explicitDeclarationsOnly = explicitDeclarationsOnly;
+	}
+
+	/**
+	 * @return the last {@link DeclarationExceptionEvent} that was detected in this admin.
+	 */
+	public @Nullable DeclarationExceptionEvent getLastDeclarationExceptionEvent() {
+		return this.lastDeclarationExceptionEvent;
+	}
+
+	@Override
+	public boolean isAutoStartup() {
+		return this.autoStartup;
+	}
+
+	public void setAutoStartup(boolean autoStartup) {
+		this.autoStartup = autoStartup;
+	}
+
+	@Override
+	public int getPhase() {
+		return Integer.MIN_VALUE;
+	}
+
+	@Override
+	public void start() {
+		if (!this.running) {
+			initialize();
+			this.running = true;
+		}
+	}
+
+	@Override
+	public void stop() {
+		this.running = false;
+	}
+
+	@Override
+	public boolean isRunning() {
+		return this.running;
+	}
+
+	/**
+	 * Declares all the exchanges, queues and bindings in the enclosing application context, if any. It should be safe
+	 * (but unnecessary) to call this method more than once.
+	 */
+	@Override
+	public void initialize() {
+		redeclareBeanDeclarables();
+	}
+
+	/**
+	 * Process bean declarables.
+	 */
+	private void redeclareBeanDeclarables() {
+		if (this.applicationContext == null) {
+			LOG.debug("no ApplicationContext has been set, cannot auto-declare Exchanges, Queues, and Bindings");
+			return;
+		}
+
+		LOG.debug("Initializing declarations");
+		Collection<Exchange> contextExchanges = new LinkedList<>(
+				this.applicationContext.getBeansOfType(Exchange.class).values());
+		Collection<Queue> contextQueues = new LinkedList<>(
+				this.applicationContext.getBeansOfType(Queue.class).values());
+		Collection<Binding> contextBindings = new LinkedList<>(
+				this.applicationContext.getBeansOfType(Binding.class).values());
+		Collection<DeclarableCustomizer> customizers =
+				this.applicationContext.getBeansOfType(DeclarableCustomizer.class).values();
+
+		processDeclarables(contextExchanges, contextQueues, contextBindings,
+				this.applicationContext.getBeansOfType(Declarables.class, false, true).values());
+
+		final Collection<Exchange> exchanges = filterDeclarables(contextExchanges, customizers);
+		final Collection<Queue> queues = filterDeclarables(contextQueues, customizers);
+		final Collection<Binding> bindings = filterDeclarables(contextBindings, customizers);
+
+		for (Exchange exchange : exchanges) {
+			if ((!exchange.isDurable() || exchange.isAutoDelete())) {
+				LOG.info(() -> "Auto-declaring a non-durable or auto-delete Exchange ("
+						+ exchange.getName()
+						+ ") durable:" + exchange.isDurable() + ", auto-delete:" + exchange.isAutoDelete() + ". "
+						+ "It will be deleted by the broker if it shuts down, and can be redeclared by closing and "
+						+ "reopening the connection.");
+			}
+		}
+
+		for (Queue queue : queues) {
+			if ((!queue.isDurable() || queue.isAutoDelete() || queue.isExclusive())) {
+				LOG.info(() -> "Auto-declaring a non-durable, auto-delete, or exclusive Queue ("
+						+ queue.getName()
+						+ ") durable:" + queue.isDurable() + ", auto-delete:" + queue.isAutoDelete() + ", exclusive:"
+						+ queue.isExclusive() + ". "
+						+ "It will be redeclared if the broker stops and is restarted while the connection factory is "
+						+ "alive, but all messages will be lost.");
+			}
+		}
+
+		if (exchanges.isEmpty() && queues.isEmpty() && bindings.isEmpty()) {
+			LOG.debug("Nothing to declare");
+			return;
+		}
+
+		try (Management management = this.amqpConnection.management()) {
+			exchanges.forEach((exchange) -> doDeclareExchange(management, exchange));
+			queues.forEach((queue) -> doDeclareQueue(management, queue));
+			bindings.forEach((binding) -> doDeclareBinding(management, binding));
+		}
+
+		LOG.debug("Declarations finished");
+	}
+
+	/**
+	 * Remove any instances that should not be declared by this admin.
+	 * @param declarables the collection of {@link Declarable}s.
+	 * @param customizers a collection if {@link DeclarableCustomizer} beans.
+	 * @param <T> the declarable type.
+	 * @return a new collection containing {@link Declarable}s that should be declared by this
+	 * admin.
+	 */
+	@SuppressWarnings({"unchecked", "NullAway"}) // Dataflow analysis limitation
+	private <T extends Declarable> Collection<T> filterDeclarables(Collection<T> declarables,
+			Collection<DeclarableCustomizer> customizers) {
+
+		return declarables.stream()
+				.filter(dec -> dec.shouldDeclare() && declarableByMe(dec))
+				.map(dec -> {
+					if (customizers.isEmpty()) {
+						return dec;
+					}
+					AtomicReference<T> ref = new AtomicReference<>(dec);
+					customizers.forEach(cust -> ref.set((T) cust.apply(ref.get())));
+					return ref.get();
+				})
+				.toList();
+	}
+
+	private <T extends Declarable> boolean declarableByMe(T dec) {
+		return (dec.getDeclaringAdmins().isEmpty() && !this.explicitDeclarationsOnly) // NOSONAR boolean complexity
+				|| dec.getDeclaringAdmins().contains(this)
+				|| dec.getDeclaringAdmins().contains(this.beanName);
+	}
+
+	@Override
+	public void declareExchange(Exchange exchange) {
+		try (Management management = this.amqpConnection.management()) {
+			doDeclareExchange(management, exchange);
+		}
+	}
+
+	private void doDeclareExchange(Management management, Exchange exchange) {
+		Management.ExchangeSpecification exchangeSpecification =
+				management.exchange(exchange.getName())
+						.type(exchange.isDelayed() ? RabbitAdmin.DELAYED_MESSAGE_EXCHANGE : exchange.getType())
+//						.durable(exchange.isDurable())
+//						.internal(exchange.isInternal())
+						.autoDelete(exchange.isAutoDelete());
+		Map<String, Object> arguments = exchange.getArguments();
+		if (arguments != null) {
+			arguments.forEach(exchangeSpecification::argument);
+		}
+		if (exchange.isDelayed()) {
+			exchangeSpecification.argument("x-delayed-type", exchange.getType());
+		}
+		try {
+			exchangeSpecification.declare();
+		}
+		catch (AmqpException ex) {
+			logOrRethrowDeclarationException(exchange, "exchange", ex);
+		}
+	}
+
+	@Override
+	@ManagedOperation(description = "Delete an exchange from the broker")
+	public boolean deleteExchange(String exchangeName) {
+		if (isDeletingDefaultExchange(exchangeName)) {
+			return false;
+		}
+
+		try (Management management = this.amqpConnection.management()) {
+			management.exchangeDelete(exchangeName);
+		}
+		return true;
+	}
+
+	@Override
+	public @Nullable Queue declareQueue() {
+		try (Management management = this.amqpConnection.management()) {
+			return doDeclareQueue(management);
+		}
+	}
+
+	private @Nullable Queue doDeclareQueue(Management management) {
+		try {
+			Management.QueueInfo queueInfo =
+					management.queue()
+							.autoDelete(true)
+							.exclusive(true)
+							.classic()
+							//						.durable(false)
+							.queue()
+							.declare();
+			return new Queue(queueInfo.name(), false, true, true);
+		}
+		catch (AmqpException ex) {
+			logOrRethrowDeclarationException(null, "queue", ex);
+		}
+		return null;
+	}
+
+	@Override
+	public @Nullable String declareQueue(Queue queue) {
+		try (Management management = this.amqpConnection.management()) {
+			return doDeclareQueue(management, queue);
+		}
+	}
+
+	private @Nullable String doDeclareQueue(Management management, Queue queue) {
+		Management.QueueSpecification queueSpecification =
+				management.queue(queue.getName())
+						.autoDelete(queue.isAutoDelete())
+						.exclusive(queue.isExclusive())
+						.classic()
+//						.durable(queue.isDurable())
+						.queue();
+
+		queue.getArguments().forEach(queueSpecification::argument);
+
+		try {
+			String actualName = queueSpecification.declare().name();
+			queue.setActualName(actualName);
+			return actualName;
+		}
+		catch (AmqpException ex) {
+			logOrRethrowDeclarationException(queue, "queue", ex);
+		}
+		return null;
+	}
+
+	@Override
+	@ManagedOperation(description = "Delete a queue from the broker")
+	public boolean deleteQueue(String queueName) {
+		deleteQueue(queueName, false, false);
+		return true;
+	}
+
+	@Override
+	@ManagedOperation(description =
+			"Delete a queue from the broker if unused and empty (when corresponding arguments are true")
+	public void deleteQueue(String queueName, boolean unused, boolean empty) {
+		try (Management management = this.amqpConnection.management()) {
+			Management.QueueInfo queueInfo = management.queueInfo(queueName);
+			if ((!unused || queueInfo.consumerCount() == 0)
+					&& (!empty || queueInfo.messageCount() == 0)) {
+
+				management.queueDelete(queueName);
+			}
+		}
+	}
+
+	@Override
+	@ManagedOperation(description = "Purge a queue and optionally don't wait for the purge to occur")
+	public void purgeQueue(String queueName, boolean noWait) {
+		if (noWait) {
+			this.taskExecutor.execute(() -> purgeQueue(queueName));
+		}
+		else {
+			purgeQueue(queueName);
+		}
+	}
+
+	@Override
+	@ManagedOperation(description = "Purge a queue and return the number of messages purged")
+	public int purgeQueue(String queueName) {
+		try (Management management = this.amqpConnection.management()) {
+			management.queuePurge(queueName);
+		}
+		return 0;
+	}
+
+	@Override
+	public void declareBinding(Binding binding) {
+		try (Management management = this.amqpConnection.management()) {
+			doDeclareBinding(management, binding);
+		}
+	}
+
+	private void doDeclareBinding(Management management, Binding binding) {
+		try {
+			Management.BindingSpecification bindingSpecification =
+					management.binding()
+							.sourceExchange(binding.getExchange())
+							.key(binding.getRoutingKey())
+							.arguments(binding.getArguments());
+			if (binding.isDestinationQueue()) {
+				bindingSpecification.destinationQueue(binding.getDestination());
+			}
+			else {
+				bindingSpecification.destinationExchange(binding.getDestination());
+			}
+			bindingSpecification.bind();
+		}
+		catch (AmqpException ex) {
+			logOrRethrowDeclarationException(binding, "binding", ex);
+		}
+	}
+
+	@Override
+	public void removeBinding(Binding binding) {
+		if (binding.isDestinationQueue() && isRemovingImplicitQueueBinding(binding)) {
+			return;
+		}
+
+		try (Management management = this.amqpConnection.management()) {
+			Management.UnbindSpecification unbindSpecification =
+					management.unbind()
+							.sourceExchange(binding.getExchange())
+							.key(binding.getRoutingKey())
+							.arguments(binding.getArguments());
+
+			if (binding.isDestinationQueue()) {
+				unbindSpecification.destinationQueue(binding.getDestination());
+			}
+			else {
+				unbindSpecification.destinationExchange(binding.getDestination());
+			}
+			unbindSpecification.unbind();
+		}
+	}
+
+	/**
+	 * Returns 4 properties {@link RabbitAdmin#QUEUE_NAME}, {@link RabbitAdmin#QUEUE_MESSAGE_COUNT},
+	 * {@link RabbitAdmin#QUEUE_CONSUMER_COUNT}, {@link #QUEUE_TYPE}, or null if the queue doesn't exist.
+	 */
+	@Override
+	@ManagedOperation(description = "Get queue name, message count and consumer count")
+	public @Nullable Properties getQueueProperties(final String queueName) {
+		QueueInformation queueInfo = getQueueInfo(queueName);
+		if (queueInfo != null) {
+			Properties props = new Properties();
+			props.put(RabbitAdmin.QUEUE_NAME, queueInfo.getName());
+			props.put(RabbitAdmin.QUEUE_MESSAGE_COUNT, queueInfo.getMessageCount());
+			props.put(RabbitAdmin.QUEUE_CONSUMER_COUNT, queueInfo.getConsumerCount());
+			props.put(QUEUE_TYPE, queueInfo.getType());
+			return props;
+		}
+		else {
+			return null;
+		}
+	}
+
+	@Override
+	public @Nullable QueueInformation getQueueInfo(String queueName) {
+		try (Management management = this.amqpConnection.management()) {
+			Management.QueueInfo queueInfo = management.queueInfo(queueName);
+			QueueInformation queueInformation =
+					new QueueInformation(queueInfo.name(), queueInfo.messageCount(), queueInfo.consumerCount());
+			queueInformation.setType(queueInfo.type().name().toLowerCase());
+			return queueInformation;
+		}
+	}
+
+	private <T extends Throwable> void logOrRethrowDeclarationException(@Nullable Declarable element,
+			String elementType, T t) throws T {
+
+		publishDeclarationExceptionEvent(element, t);
+		if (this.ignoreDeclarationExceptions || (element != null && element.isIgnoreDeclarationExceptions())) {
+			if (LOG.isDebugEnabled()) {
+				LOG.debug(t, "Failed to declare " + elementType
+						+ ": " + (element == null ? "broker-generated" : element)
+						+ ", continuing...");
+			}
+			else if (LOG.isWarnEnabled()) {
+				Throwable cause = t;
+				if (t instanceof IOException && t.getCause() != null) {
+					cause = t.getCause();
+				}
+				LOG.warn("Failed to declare " + elementType
+						+ ": " + (element == null ? "broker-generated" : element)
+						+ ", continuing... " + cause);
+			}
+		}
+		else {
+			throw t;
+		}
+	}
+
+	private void publishDeclarationExceptionEvent(@Nullable Declarable element, Throwable ex) {
+		DeclarationExceptionEvent event = new DeclarationExceptionEvent(this, element, ex);
+		this.lastDeclarationExceptionEvent = event;
+		if (this.applicationEventPublisher != null) {
+			this.applicationEventPublisher.publishEvent(event);
+		}
+	}
+
+	private static boolean isDeletingDefaultExchange(String exchangeName) {
+		if (isDefaultExchange(exchangeName)) {
+			LOG.warn("Default exchange cannot be deleted.");
+			return true;
+		}
+		return false;
+	}
+
+	private static boolean isDefaultExchange(@Nullable String exchangeName) {
+		return exchangeName == null || RabbitAdmin.DEFAULT_EXCHANGE_NAME.equals(exchangeName);
+	}
+
+	private static boolean isRemovingImplicitQueueBinding(Binding binding) {
+		if (isImplicitQueueBinding(binding)) {
+			LOG.warn("Cannot remove implicit default exchange binding to queue.");
+			return true;
+		}
+		return false;
+	}
+
+	private static boolean isImplicitQueueBinding(Binding binding) {
+		return isDefaultExchange(binding.getExchange()) &&
+				Objects.equals(binding.getDestination(), binding.getRoutingKey());
+	}
+
+	private static void processDeclarables(Collection<Exchange> contextExchanges, Collection<Queue> contextQueues,
+			Collection<Binding> contextBindings, Collection<Declarables> declarables) {
+
+		declarables.forEach(d -> {
+			d.getDeclarables().forEach(declarable -> {
+				if (declarable instanceof Exchange exch) {
+					contextExchanges.add(exch);
+				}
+				else if (declarable instanceof Queue queue) {
+					contextQueues.add(queue);
+				}
+				else if (declarable instanceof Binding binding) {
+					contextBindings.add(binding);
+				}
+			});
+		});
+	}
+
+
+}

spring-rabbitmq-client/src/main/java/org/springframework/amqp/rabbitmq/client/RabbitAmqpTemplate.java

@@ -0,0 +1,494 @@
+/*
+ * Copyright 2025 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.amqp.rabbitmq.client;
+
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.util.Date;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+
+import com.rabbitmq.client.amqp.Connection;
+import com.rabbitmq.client.amqp.Consumer;
+import com.rabbitmq.client.amqp.Environment;
+import com.rabbitmq.client.amqp.Publisher;
+import com.rabbitmq.client.amqp.PublisherBuilder;
+import com.rabbitmq.client.amqp.Resource;
+import org.jspecify.annotations.Nullable;
+
+import org.springframework.amqp.AmqpException;
+import org.springframework.amqp.core.AmqpTemplate;
+import org.springframework.amqp.core.AsyncAmqpTemplate;
+import org.springframework.amqp.core.Message;
+import org.springframework.amqp.core.MessagePostProcessor;
+import org.springframework.amqp.core.MessageProperties;
+import org.springframework.amqp.core.ReceiveAndReplyCallback;
+import org.springframework.amqp.core.ReplyToAddressCallback;
+import org.springframework.amqp.rabbit.core.AmqpNackReceivedException;
+import org.springframework.amqp.support.converter.MessageConverter;
+import org.springframework.amqp.support.converter.SimpleMessageConverter;
+import org.springframework.amqp.support.converter.SmartMessageConverter;
+import org.springframework.amqp.utils.JavaUtils;
+import org.springframework.beans.factory.DisposableBean;
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.core.ParameterizedTypeReference;
+import org.springframework.util.Assert;
+
+/**
+ * The {@link AmqpTemplate} for RabbitMQ AMQP 1.0 protocol support.
+ * A Spring-friendly wrapper around {@link Environment#connectionBuilder()};
+ *
+ * @author Artem Bilan
+ *
+ * @since 4.0
+ */
+public class RabbitAmqpTemplate implements AsyncAmqpTemplate, InitializingBean, DisposableBean {
+
+	private final Connection connection;
+
+	private final PublisherBuilder publisherBuilder;
+
+	@SuppressWarnings("NullAway.Init")
+	private Publisher publisher;
+
+	private MessageConverter messageConverter = new SimpleMessageConverter();
+
+	private @Nullable String defaultExchange;
+
+	private @Nullable String defaultRoutingKey;
+
+	private @Nullable String defaultQueue;
+
+	private @Nullable String defaultReceiveQueue;
+
+	public RabbitAmqpTemplate(Connection amqpConnection) {
+		this.connection = amqpConnection;
+		this.publisherBuilder = amqpConnection.publisherBuilder();
+	}
+
+	public void setListeners(Resource.StateListener... listeners) {
+		this.publisherBuilder.listeners(listeners);
+	}
+
+	public void setPublishTimeout(Duration timeout) {
+		this.publisherBuilder.publishTimeout(timeout);
+	}
+
+	/**
+	 * Set a default exchange for publishing.
+	 * Cannot be real default AMQP exchange.
+	 * The {@link #setQueue(String)} is recommended instead.
+	 * Mutually exclusive with {@link #setQueue(String)}.
+	 * @param exchange the default exchange
+	 */
+	public void setExchange(String exchange) {
+		this.defaultExchange = exchange;
+	}
+
+	/**
+	 * Set a default routing key.
+	 * Mutually exclusive with {@link #setQueue(String)}.
+	 * @param key the default routing key.
+	 */
+	public void setKey(String key) {
+		this.defaultRoutingKey = key;
+	}
+
+	/**
+	 * Set default queue for publishing.
+	 * Mutually exclusive with {@link #setExchange(String)} and {@link #setKey(String)}.
+	 * @param queue the default queue.
+	 */
+	public void setQueue(String queue) {
+		this.defaultQueue = queue;
+	}
+
+	/**
+	 * Set a converter for {@link #convertAndSend(Object)} operations.
+	 * @param messageConverter the converter.
+	 */
+	public void setMessageConverter(MessageConverter messageConverter) {
+		this.messageConverter = messageConverter;
+	}
+
+	/**
+	 * The name of the default queue to receive messages from when none is specified explicitly.
+	 * @param queue the default queue name to use for receive operation.
+	 */
+	public void setDefaultReceiveQueue(String queue) {
+		this.defaultReceiveQueue = queue;
+	}
+
+	private String getRequiredQueue() throws IllegalStateException {
+		String name = this.defaultReceiveQueue;
+		Assert.state(name != null, "No 'queue' specified. Check configuration of this 'RabbitAmqpTemplate'.");
+		return name;
+	}
+
+	@Override
+	public void afterPropertiesSet() {
+		this.publisher = this.publisherBuilder.build();
+	}
+
+	@Override
+	public void destroy() {
+		this.publisher.close();
+	}
+
+	/**
+	 * Publish a message to the default exchange and routing key (if any) (or queue) configured on this template.
+	 * @param message to publish
+	 * @return the {@link CompletableFuture} as an async result of the message publication.
+	 */
+	public CompletableFuture<Boolean> send(Message message) {
+		return doSend(this.defaultExchange, this.defaultRoutingKey, this.defaultQueue, message);
+	}
+
+	/**
+	 * Publish the message to the provided queue.
+	 * @param queue to publish
+	 * @param message to publish
+	 * @return the {@link CompletableFuture} as an async result of the message publication.
+	 */
+	public CompletableFuture<Boolean> send(String queue, Message message) {
+		return doSend(null, null, queue, message);
+	}
+
+	public CompletableFuture<Boolean> send(String exchange, @Nullable String routingKey, Message message) {
+		return doSend(exchange, routingKey != null ? routingKey : this.defaultRoutingKey, null, message);
+	}
+
+	private CompletableFuture<Boolean> doSend(@Nullable String exchange, @Nullable String routingKey,
+			@Nullable String queue, Message message) {
+
+		MessageProperties messageProperties = message.getMessageProperties();
+
+		com.rabbitmq.client.amqp.Message amqpMessage =
+				this.publisher.message(message.getBody())
+						.contentEncoding(messageProperties.getContentEncoding())
+						.contentType(messageProperties.getContentType())
+						.messageId(messageProperties.getMessageId())
+						.correlationId(messageProperties.getCorrelationId())
+						.priority(messageProperties.getPriority().byteValue())
+						.replyTo(messageProperties.getReplyTo());
+
+		com.rabbitmq.client.amqp.Message.MessageAddressBuilder address = amqpMessage.toAddress();
+
+		Map<String, @Nullable Object> headers = messageProperties.getHeaders();
+		if (!headers.isEmpty()) {
+			headers.forEach((key, val) -> mapProp(key, val, amqpMessage));
+		}
+
+		JavaUtils.INSTANCE
+				.acceptIfNotNull(messageProperties.getUserId(),
+						(userId) -> amqpMessage.userId(userId.getBytes(StandardCharsets.UTF_8)))
+				.acceptIfNotNull(messageProperties.getTimestamp(),
+						(timestamp) -> amqpMessage.creationTime(timestamp.getTime()))
+				.acceptIfNotNull(messageProperties.getExpiration(),
+						(expiration) -> amqpMessage.absoluteExpiryTime(Long.parseLong(expiration)))
+				.acceptIfNotNull(exchange, address::exchange)
+				.acceptIfNotNull(routingKey, address::key)
+				.acceptIfNotNull(queue, address::queue);
+
+		CompletableFuture<Boolean> publishResult = new CompletableFuture<>();
+
+		this.publisher.publish(address.message(),
+				(context) -> {
+					switch (context.status()) {
+						case ACCEPTED -> publishResult.complete(true);
+						case REJECTED, RELEASED -> publishResult.completeExceptionally(
+								new AmqpNackReceivedException("The message was rejected", message));
+					}
+				});
+
+		return publishResult;
+	}
+
+	/**
+	 * Publish a message from converted body to the default exchange
+	 * and routing key (if any) (or queue) configured on this template.
+	 * @param message to publish
+	 * @return the {@link CompletableFuture} as an async result of the message publication.
+	 */
+	public CompletableFuture<Boolean> convertAndSend(Object message) {
+		return doConvertAndSend(this.defaultExchange, this.defaultRoutingKey, this.defaultQueue, message, null);
+	}
+
+	public CompletableFuture<Boolean> convertAndSend(String queue, Object message) {
+		return doConvertAndSend(null, null, queue, message, null);
+	}
+
+	public CompletableFuture<Boolean> convertAndSend(String exchange, @Nullable String routingKey, Object message) {
+		return doConvertAndSend(exchange, routingKey != null ? routingKey : this.defaultRoutingKey, null, message, null);
+	}
+
+	public CompletableFuture<Boolean> convertAndSend(Object message,
+			@Nullable MessagePostProcessor messagePostProcessor) {
+
+		return doConvertAndSend(null, null, null, message, messagePostProcessor);
+	}
+
+	public CompletableFuture<Boolean> convertAndSend(String queue, Object message,
+			@Nullable MessagePostProcessor messagePostProcessor) {
+
+		return doConvertAndSend(null, null, queue, message, messagePostProcessor);
+	}
+
+	public CompletableFuture<Boolean> convertAndSend(String exchange, @Nullable String routingKey, Object message,
+			@Nullable MessagePostProcessor messagePostProcessor) {
+
+		return doConvertAndSend(exchange, routingKey, null, message, messagePostProcessor);
+	}
+
+	private CompletableFuture<Boolean> doConvertAndSend(@Nullable String exchange, @Nullable String routingKey,
+			@Nullable String queue, Object data, @Nullable MessagePostProcessor messagePostProcessor) {
+
+		Message message =
+				data instanceof Message
+						? (Message) data
+						: this.messageConverter.toMessage(data, new MessageProperties());
+		if (messagePostProcessor != null) {
+			message = messagePostProcessor.postProcessMessage(message);
+		}
+		return doSend(exchange, routingKey, queue, message);
+	}
+
+	public CompletableFuture<Message> receive() {
+		return receive(getRequiredQueue());
+	}
+
+	@SuppressWarnings("try")
+	public CompletableFuture<Message> receive(String queueName) {
+		CompletableFuture<Message> messageFuture = new CompletableFuture<>();
+
+		Consumer consumer =
+				this.connection.consumerBuilder()
+						.queue(queueName)
+						.initialCredits(1)
+						.priority(10)
+						.messageHandler((context, message) -> {
+							context.accept();
+							messageFuture.complete(fromAmqpMessage(message));
+						})
+						.build();
+
+		return messageFuture
+				.orTimeout(1, TimeUnit.MINUTES)
+				.whenComplete((message, exception) -> consumer.close());
+	}
+
+	public CompletableFuture<Object> receiveAndConvert() {
+		return receiveAndConvert(getRequiredQueue());
+	}
+
+	public CompletableFuture<Object> receiveAndConvert(String queueName) {
+		return receive(queueName)
+				.thenApply(this.messageConverter::fromMessage);
+	}
+
+	/**
+	 * Receive a message from {@link #setDefaultReceiveQueue(String)} and convert its body
+	 * to the expected type.
+	 * The {@link #setMessageConverter(MessageConverter)} must be an implementation of {@link SmartMessageConverter}.
+	 * @param type the type to covert received result.
+	 * @return the CompletableFuture with a result.
+	 */
+	public <T> CompletableFuture<T> receiveAndConvert(ParameterizedTypeReference<T> type) {
+		return receiveAndConvert(getRequiredQueue(), type);
+	}
+
+	/**
+	 * Receive a message from {@link #setDefaultReceiveQueue(String)} and convert its body
+	 * to the expected type.
+	 * The {@link #setMessageConverter(MessageConverter)} must be an implementation of {@link SmartMessageConverter}.
+	 * @param queueName the queue to consume message from.
+	 * @param type the type to covert received result.
+	 * @return the CompletableFuture with a result.
+	 */
+	@SuppressWarnings("unchecked")
+	public <T> CompletableFuture<T> receiveAndConvert(String queueName, ParameterizedTypeReference<T> type) {
+		SmartMessageConverter smartMessageConverter = getRequiredSmartMessageConverter();
+		return receive(queueName)
+				.thenApply((message) -> (T) smartMessageConverter.fromMessage(message, type));
+	}
+
+	private SmartMessageConverter getRequiredSmartMessageConverter() throws IllegalStateException {
+		Assert.state(this.messageConverter instanceof SmartMessageConverter,
+				"template's message converter must be a SmartMessageConverter");
+		return (SmartMessageConverter) this.messageConverter;
+	}
+
+	public <R, S> boolean receiveAndReply(ReceiveAndReplyCallback<R, S> callback) throws AmqpException {
+		throw new UnsupportedOperationException();
+	}
+
+	public <R, S> boolean receiveAndReply(String queueName, ReceiveAndReplyCallback<R, S> callback) throws AmqpException {
+		throw new UnsupportedOperationException();
+	}
+
+	public <R, S> boolean receiveAndReply(ReceiveAndReplyCallback<R, S> callback, String replyExchange, String replyRoutingKey) throws AmqpException {
+		throw new UnsupportedOperationException();
+	}
+
+	public <R, S> boolean receiveAndReply(String queueName, ReceiveAndReplyCallback<R, S> callback, String replyExchange, String replyRoutingKey) throws AmqpException {
+		throw new UnsupportedOperationException();
+	}
+
+	public <R, S> boolean receiveAndReply(ReceiveAndReplyCallback<R, S> callback, ReplyToAddressCallback<S> replyToAddressCallback) throws AmqpException {
+		throw new UnsupportedOperationException();
+	}
+
+	public <R, S> boolean receiveAndReply(String queueName, ReceiveAndReplyCallback<R, S> callback, ReplyToAddressCallback<S> replyToAddressCallback) throws AmqpException {
+		throw new UnsupportedOperationException();
+	}
+
+	@Override
+	public CompletableFuture<Message> sendAndReceive(Message message) {
+		throw new UnsupportedOperationException();
+	}
+
+	@Override
+	public CompletableFuture<Message> sendAndReceive(String routingKey, Message message) {
+		throw new UnsupportedOperationException();
+	}
+
+	@Override
+	public CompletableFuture<Message> sendAndReceive(String exchange, String routingKey, Message message) {
+		throw new UnsupportedOperationException();
+	}
+
+	@Override
+	public <C> CompletableFuture<C> convertSendAndReceive(Object object) {
+		throw new UnsupportedOperationException();
+	}
+
+	@Override
+	public <C> CompletableFuture<C> convertSendAndReceive(String routingKey, Object object) {
+		throw new UnsupportedOperationException();
+	}
+
+	@Override
+	public <C> CompletableFuture<C> convertSendAndReceive(String exchange, String routingKey, Object object) {
+		throw new UnsupportedOperationException();
+	}
+
+	@Override
+	public <C> CompletableFuture<C> convertSendAndReceive(Object object, MessagePostProcessor messagePostProcessor) {
+		throw new UnsupportedOperationException();
+	}
+
+	@Override
+	public <C> CompletableFuture<C> convertSendAndReceive(String routingKey, Object object, MessagePostProcessor messagePostProcessor) {
+		throw new UnsupportedOperationException();
+	}
+
+	@Override
+	public <C> CompletableFuture<C> convertSendAndReceive(String exchange, String routingKey, Object object, @Nullable MessagePostProcessor messagePostProcessor) {
+		throw new UnsupportedOperationException();
+	}
+
+	@Override
+	public <C> CompletableFuture<C> convertSendAndReceiveAsType(Object object, ParameterizedTypeReference<C> responseType) {
+		throw new UnsupportedOperationException();
+	}
+
+	@Override
+	public <C> CompletableFuture<C> convertSendAndReceiveAsType(String routingKey, Object object, ParameterizedTypeReference<C> responseType) {
+		throw new UnsupportedOperationException();
+	}
+
+	@Override
+	public <C> CompletableFuture<C> convertSendAndReceiveAsType(String exchange, String routingKey, Object object, ParameterizedTypeReference<C> responseType) {
+		throw new UnsupportedOperationException();
+	}
+
+	@Override
+	public <C> CompletableFuture<C> convertSendAndReceiveAsType(Object object, MessagePostProcessor messagePostProcessor, ParameterizedTypeReference<C> responseType) {
+		throw new UnsupportedOperationException();
+	}
+
+	@Override
+	public <C> CompletableFuture<C> convertSendAndReceiveAsType(String routingKey, Object object, @Nullable MessagePostProcessor messagePostProcessor, @Nullable ParameterizedTypeReference<C> responseType) {
+		throw new UnsupportedOperationException();
+	}
+
+	@Override
+	public <C> CompletableFuture<C> convertSendAndReceiveAsType(String exchange, String routingKey, Object object, @Nullable MessagePostProcessor messagePostProcessor, @Nullable ParameterizedTypeReference<C> responseType) {
+		throw new UnsupportedOperationException();
+	}
+
+	private static void mapProp(String key, @Nullable Object val, com.rabbitmq.client.amqp.Message amqpMessage) {
+		if (val == null) {
+			return;
+		}
+		if (val instanceof String string) {
+			amqpMessage.property(key, string);
+		}
+		else if (val instanceof Long longValue) {
+			amqpMessage.property(key, longValue);
+		}
+		else if (val instanceof Integer intValue) {
+			amqpMessage.property(key, intValue);
+		}
+		else if (val instanceof Short shortValue) {
+			amqpMessage.property(key, shortValue);
+		}
+		else if (val instanceof Byte byteValue) {
+			amqpMessage.property(key, byteValue);
+		}
+		else if (val instanceof Double doubleValue) {
+			amqpMessage.property(key, doubleValue);
+		}
+		else if (val instanceof Float floatValue) {
+			amqpMessage.property(key, floatValue);
+		}
+		else if (val instanceof Character character) {
+			amqpMessage.property(key, character);
+		}
+		else if (val instanceof UUID uuid) {
+			amqpMessage.property(key, uuid);
+		}
+		else if (val instanceof byte[] bytes) {
+			amqpMessage.property(key, bytes);
+		}
+		else if (val instanceof Boolean booleanValue) {
+			amqpMessage.property(key, booleanValue);
+		}
+	}
+
+	private static Message fromAmqpMessage(com.rabbitmq.client.amqp.Message amqpMessage) {
+		MessageProperties messageProperties = new MessageProperties();
+
+		JavaUtils.INSTANCE
+				.acceptIfNotNull(amqpMessage.messageIdAsString(), messageProperties::setMessageId)
+				.acceptIfNotNull(amqpMessage.userId(),
+						(usr) -> messageProperties.setUserId(new String(usr, StandardCharsets.UTF_8)))
+				.acceptIfNotNull(amqpMessage.correlationIdAsString(), messageProperties::setCorrelationId)
+				.acceptIfNotNull(amqpMessage.contentType(), messageProperties::setContentType)
+				.acceptIfNotNull(amqpMessage.contentEncoding(), messageProperties::setContentEncoding)
+				.acceptIfNotNull(amqpMessage.absoluteExpiryTime(),
+						(exp) -> messageProperties.setExpiration(Long.toString(exp)))
+				.acceptIfNotNull(amqpMessage.creationTime(), (time) -> messageProperties.setTimestamp(new Date(time)));
+
+		amqpMessage.forEachProperty(messageProperties::setHeader);
+
+		return new Message(amqpMessage.body(), messageProperties);
+	}
+
+}

spring-rabbitmq-client/src/main/java/org/springframework/amqp/rabbitmq/client/package-info.java

@@ -0,0 +1,5 @@
+/**
+ * Provides Spring support for RabbitMQ AMQP 1.0 Client.
+ */
+@org.jspecify.annotations.NullMarked
+package org.springframework.amqp.rabbitmq.client;

spring-rabbitmq-client/src/test/java/org/springframework/amqp/rabbitmq/client/RabbitAmqpAdminTests.java

@@ -0,0 +1,145 @@
+/*
+ * Copyright 2025 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.amqp.rabbitmq.client;
+
+import java.time.Duration;
+import java.util.concurrent.CompletableFuture;
+
+import org.junit.jupiter.api.Test;
+
+import org.springframework.amqp.core.Binding;
+import org.springframework.amqp.core.BindingBuilder;
+import org.springframework.amqp.core.Declarables;
+import org.springframework.amqp.core.DirectExchange;
+import org.springframework.amqp.core.Exchange;
+import org.springframework.amqp.core.Queue;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.test.context.ContextConfiguration;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * @author Artem Bilan
+ *
+ * @since 4.0
+ */
+@ContextConfiguration
+public class RabbitAmqpAdminTests extends RabbitAmqpTestBase {
+
+	@Autowired
+	RabbitAmqpTemplate template;
+
+	@Autowired
+	RabbitAmqpAdmin admin;
+
+	@Autowired
+	@Qualifier("ds")
+	Declarables declarables;
+
+	@Test
+	void verifyBeanDeclarations() {
+		CompletableFuture<Void> publishFutures =
+				CompletableFuture.allOf(
+						template.convertAndSend("e1", "k1", "test1"),
+						template.convertAndSend("e2", "k2", "test2"),
+						template.convertAndSend("e2", "k2", "test3"),
+						template.convertAndSend("e3", "k3", "test4"),
+						template.convertAndSend("e4", "k4", "test5"));
+		assertThat(publishFutures).succeedsWithin(Duration.ofSeconds(10));
+
+		assertThat(template.receiveAndConvert("q1")).succeedsWithin(Duration.ofSeconds(10)).isEqualTo("test1");
+		assertThat(template.receiveAndConvert("q2")).succeedsWithin(Duration.ofSeconds(10)).isEqualTo("test2");
+		assertThat(template.receiveAndConvert("q2")).succeedsWithin(Duration.ofSeconds(10)).isEqualTo("test3");
+		assertThat(template.receiveAndConvert("q3")).succeedsWithin(Duration.ofSeconds(10)).isEqualTo("test4");
+		assertThat(template.receiveAndConvert("q4")).succeedsWithin(Duration.ofSeconds(10)).isEqualTo("test5");
+
+		admin.deleteQueue("q1");
+		admin.deleteQueue("q2");
+		admin.deleteQueue("q3");
+		admin.deleteQueue("q4");
+		admin.deleteExchange("e1");
+		admin.deleteExchange("e2");
+		admin.deleteExchange("e3");
+		admin.deleteExchange("e4");
+
+		assertThat(declarables.getDeclarablesByType(Queue.class))
+				.hasSize(1)
+				.extracting(Queue::getName)
+				.contains("q4");
+		assertThat(declarables.getDeclarablesByType(Exchange.class))
+				.hasSize(1)
+				.extracting(Exchange::getName)
+				.contains("e4");
+		assertThat(declarables.getDeclarablesByType(Binding.class))
+				.hasSize(1)
+				.extracting(Binding::getDestination)
+				.contains("q4");
+	}
+
+	@Configuration
+	public static class Config {
+
+		@Bean
+		DirectExchange e1() {
+			return new DirectExchange("e1", false, false);
+		}
+
+		@Bean
+		Queue q1() {
+			return new Queue("q1", false, false, false);
+		}
+
+		@Bean
+		Binding b1() {
+			return BindingBuilder.bind(q1()).to(e1()).with("k1");
+		}
+
+		@Bean
+		Declarables es() {
+			return new Declarables(
+					new DirectExchange("e2", false, false),
+					new DirectExchange("e3", false, false));
+		}
+
+		@Bean
+		Declarables qs() {
+			return new Declarables(
+					new Queue("q2", false, false, false),
+					new Queue("q3", false, false, false));
+		}
+
+		@Bean
+		Declarables bs() {
+			return new Declarables(
+					new Binding("q2", Binding.DestinationType.QUEUE, "e2", "k2", null),
+					new Binding("q3", Binding.DestinationType.QUEUE, "e3", "k3", null));
+		}
+
+		@Bean
+		Declarables ds() {
+			return new Declarables(
+					new DirectExchange("e4", false, false),
+					new Queue("q4", false, false, false),
+					new Binding("q4", Binding.DestinationType.QUEUE, "e4", "k4", null));
+		}
+
+	}
+
+}

spring-rabbitmq-client/src/test/java/org/springframework/amqp/rabbitmq/client/RabbitAmqpTemplateTests.java

@@ -0,0 +1,107 @@
+/*
+ * Copyright 2025 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.amqp.rabbitmq.client;
+
+import java.time.Duration;
+
+import com.rabbitmq.client.amqp.Connection;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import org.springframework.amqp.core.Binding;
+import org.springframework.amqp.core.BindingBuilder;
+import org.springframework.amqp.core.DirectExchange;
+import org.springframework.amqp.core.Queue;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.test.context.ContextConfiguration;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * @author Artem Bilan
+ *
+ * @since 4.0
+ */
+@ContextConfiguration
+public class RabbitAmqpTemplateTests extends RabbitAmqpTestBase {
+
+	@Autowired
+	Connection connection;
+
+	RabbitAmqpTemplate rabbitAmqpTemplate;
+
+	@BeforeEach
+	void setUp() {
+		this.rabbitAmqpTemplate = new RabbitAmqpTemplate(this.connection);
+		this.rabbitAmqpTemplate.afterPropertiesSet();
+	}
+
+	@AfterEach
+	void tearDown() {
+		this.rabbitAmqpTemplate.destroy();
+	}
+
+	@Test
+	void defaultExchangeAndRoutingKey() {
+		this.rabbitAmqpTemplate.setExchange("e1");
+		this.rabbitAmqpTemplate.setKey("k1");
+
+		assertThat(this.rabbitAmqpTemplate.convertAndSend("test1"))
+				.succeedsWithin(Duration.ofSeconds(10));
+
+		assertThat(this.rabbitAmqpTemplate.receiveAndConvert("q1"))
+				.succeedsWithin(Duration.ofSeconds(10))
+				.isEqualTo("test1");
+	}
+
+	@Test
+	void defaultQueues() {
+		this.rabbitAmqpTemplate.setQueue("q1");
+		this.rabbitAmqpTemplate.setDefaultReceiveQueue("q1");
+
+		assertThat(this.rabbitAmqpTemplate.convertAndSend("test2"))
+				.succeedsWithin(Duration.ofSeconds(10));
+
+		assertThat(this.rabbitAmqpTemplate.receiveAndConvert())
+				.succeedsWithin(Duration.ofSeconds(10))
+				.isEqualTo("test2");
+	}
+
+	@Configuration
+	static class Config {
+
+		@Bean
+		DirectExchange e1() {
+			return new DirectExchange("e1", false, false);
+		}
+
+		@Bean
+		Queue q1() {
+			return new Queue("q1", false, false, false);
+		}
+
+		@Bean
+		Binding b1() {
+			return BindingBuilder.bind(q1()).to(e1()).with("k1");
+		}
+
+	}
+
+}

spring-rabbitmq-client/src/test/java/org/springframework/amqp/rabbitmq/client/RabbitAmqpTestBase.java

@@ -0,0 +1,69 @@
+/*
+ * Copyright 2025 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.amqp.rabbitmq.client;
+
+import com.rabbitmq.client.amqp.Connection;
+import com.rabbitmq.client.amqp.Environment;
+import com.rabbitmq.client.amqp.impl.AmqpEnvironmentBuilder;
+
+import org.springframework.amqp.rabbit.junit.AbstractTestContainerTests;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
+
+/**
+ * The {@link AbstractTestContainerTests} extension
+ *
+ * @author Artem Bilan
+ *
+ * @since 4.0
+ */
+@SpringJUnitConfig
+@DirtiesContext
+abstract class RabbitAmqpTestBase extends AbstractTestContainerTests {
+
+	@Configuration
+	public static class AmqpCommonConfig {
+
+		@Bean
+		Environment environment() {
+			return new AmqpEnvironmentBuilder()
+					.connectionSettings()
+					.port(amqpPort())
+					.environmentBuilder()
+					.build();
+		}
+
+		@Bean
+		AmqpConnectionFactoryBean connection(Environment environment) {
+			return new AmqpConnectionFactoryBean(environment);
+		}
+
+		@Bean
+		RabbitAmqpAdmin admin(Connection connection) {
+			return new RabbitAmqpAdmin(connection);
+		}
+
+		@Bean
+		RabbitAmqpTemplate rabbitTemplate(Connection connection) {
+			return new RabbitAmqpTemplate(connection);
+		}
+
+	}
+
+}

spring-rabbitmq-client/src/test/resources/log4j2-test.xml

@@ -0,0 +1,14 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<Configuration>
+	<Appenders>
+		<Console name="STDOUT" target="SYSTEM_OUT">
+			<PatternLayout pattern="%d %5p %c [%t] : %m%n" />
+		</Console>
+	</Appenders>
+	<Loggers>
+		<Logger name="org.springframework.amqp.rabbit" level="info"/>
+		<Root level="warn">
+			<AppenderRef ref="STDOUT" />
+		</Root>
+	</Loggers>
+</Configuration>