From f486a30e636f68f1ea08f0b562f7787ef21e3452 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Wed, 26 Feb 2025 13:13:15 -0500 Subject: [PATCH] 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 | 23 + .../amqp/core/QueueInformation.java | 36 +- .../springframework/amqp/utils/JavaUtils.java | 4 +- .../amqp/rabbit/core/RabbitAdmin.java | 2 +- .../amqp/rabbit/core/RabbitAdminTests.java | 2 +- .../client/AmqpConnectionFactoryBean.java | 286 +++++++++ .../amqp/rabbitmq/client/RabbitAmqpAdmin.java | 570 ++++++++++++++++++ .../rabbitmq/client/RabbitAmqpTemplate.java | 494 +++++++++++++++ .../amqp/rabbitmq/client/package-info.java | 5 + .../rabbitmq/client/RabbitAmqpAdminTests.java | 145 +++++ .../client/RabbitAmqpTemplateTests.java | 107 ++++ .../rabbitmq/client/RabbitAmqpTestBase.java | 69 +++ .../src/test/resources/log4j2-test.xml | 14 + 13 files changed, 1745 insertions(+), 12 deletions(-) create mode 100644 spring-rabbitmq-client/src/main/java/org/springframework/amqp/rabbitmq/client/AmqpConnectionFactoryBean.java create mode 100644 spring-rabbitmq-client/src/main/java/org/springframework/amqp/rabbitmq/client/RabbitAmqpAdmin.java create mode 100644 spring-rabbitmq-client/src/main/java/org/springframework/amqp/rabbitmq/client/RabbitAmqpTemplate.java create mode 100644 spring-rabbitmq-client/src/main/java/org/springframework/amqp/rabbitmq/client/package-info.java create mode 100644 spring-rabbitmq-client/src/test/java/org/springframework/amqp/rabbitmq/client/RabbitAmqpAdminTests.java create mode 100644 spring-rabbitmq-client/src/test/java/org/springframework/amqp/rabbitmq/client/RabbitAmqpTemplateTests.java create mode 100644 spring-rabbitmq-client/src/test/java/org/springframework/amqp/rabbitmq/client/RabbitAmqpTestBase.java create mode 100644 spring-rabbitmq-client/src/test/resources/log4j2-test.xml diff --git a/build.gradle b/build.gradle index 35e67de8..f226c681 100644 --- a/build.gradle +++ b/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' diff --git a/spring-amqp/src/main/java/org/springframework/amqp/core/QueueInformation.java b/spring-amqp/src/main/java/org/springframework/amqp/core/QueueInformation.java index f2d8df24..23237749 100644 --- a/spring-amqp/src/main/java/org/springframework/amqp/core/QueueInformation.java +++ b/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); } diff --git a/spring-amqp/src/main/java/org/springframework/amqp/utils/JavaUtils.java b/spring-amqp/src/main/java/org/springframework/amqp/utils/JavaUtils.java index edd108e5..ee151353 100644 --- a/spring-amqp/src/main/java/org/springframework/amqp/utils/JavaUtils.java +++ b/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 the value type. * @return this. diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitAdmin.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitAdmin.java index d04a61f2..18313990 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitAdmin.java +++ b/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 diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitAdminTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitAdminTests.java index 07465a9d..2cacb164 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitAdminTests.java +++ b/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(); diff --git a/spring-rabbitmq-client/src/main/java/org/springframework/amqp/rabbitmq/client/AmqpConnectionFactoryBean.java b/spring-rabbitmq-client/src/main/java/org/springframework/amqp/rabbitmq/client/AmqpConnectionFactoryBean.java new file mode 100644 index 00000000..b0769f5e --- /dev/null +++ b/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 { + + 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 tlsCustomizer) { + tlsCustomizer.accept(new Tls(this.connectionBuilder.tls())); + return this; + } + + public AmqpConnectionFactoryBean setAffinity(Consumer affinityCustomizer) { + affinityCustomizer.accept(new Affinity(this.connectionBuilder.affinity())); + return this; + } + + public AmqpConnectionFactoryBean setOAuth2(Consumer oauth2Customizer) { + oauth2Customizer.accept(new OAuth2(this.connectionBuilder.oauth2())); + return this; + } + + public AmqpConnectionFactoryBean setRecovery(Consumer 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 tls; + + private Tls(ConnectionSettings.TlsSettings 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 affinity; + + private Affinity(ConnectionSettings.Affinity 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 oAuth2Settings; + + private OAuth2(OAuth2Settings 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; + } + + } + +} diff --git a/spring-rabbitmq-client/src/main/java/org/springframework/amqp/rabbitmq/client/RabbitAmqpAdmin.java b/spring-rabbitmq-client/src/main/java/org/springframework/amqp/rabbitmq/client/RabbitAmqpAdmin.java new file mode 100644 index 00000000..226a3413 --- /dev/null +++ b/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 contextExchanges = new LinkedList<>( + this.applicationContext.getBeansOfType(Exchange.class).values()); + Collection contextQueues = new LinkedList<>( + this.applicationContext.getBeansOfType(Queue.class).values()); + Collection contextBindings = new LinkedList<>( + this.applicationContext.getBeansOfType(Binding.class).values()); + Collection customizers = + this.applicationContext.getBeansOfType(DeclarableCustomizer.class).values(); + + processDeclarables(contextExchanges, contextQueues, contextBindings, + this.applicationContext.getBeansOfType(Declarables.class, false, true).values()); + + final Collection exchanges = filterDeclarables(contextExchanges, customizers); + final Collection queues = filterDeclarables(contextQueues, customizers); + final Collection 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 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 Collection filterDeclarables(Collection declarables, + Collection customizers) { + + return declarables.stream() + .filter(dec -> dec.shouldDeclare() && declarableByMe(dec)) + .map(dec -> { + if (customizers.isEmpty()) { + return dec; + } + AtomicReference ref = new AtomicReference<>(dec); + customizers.forEach(cust -> ref.set((T) cust.apply(ref.get()))); + return ref.get(); + }) + .toList(); + } + + private 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 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 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 contextExchanges, Collection contextQueues, + Collection contextBindings, Collection 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); + } + }); + }); + } + + +} diff --git a/spring-rabbitmq-client/src/main/java/org/springframework/amqp/rabbitmq/client/RabbitAmqpTemplate.java b/spring-rabbitmq-client/src/main/java/org/springframework/amqp/rabbitmq/client/RabbitAmqpTemplate.java new file mode 100644 index 00000000..d6b89aef --- /dev/null +++ b/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 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 send(String queue, Message message) { + return doSend(null, null, queue, message); + } + + public CompletableFuture send(String exchange, @Nullable String routingKey, Message message) { + return doSend(exchange, routingKey != null ? routingKey : this.defaultRoutingKey, null, message); + } + + private CompletableFuture 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 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 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 convertAndSend(Object message) { + return doConvertAndSend(this.defaultExchange, this.defaultRoutingKey, this.defaultQueue, message, null); + } + + public CompletableFuture convertAndSend(String queue, Object message) { + return doConvertAndSend(null, null, queue, message, null); + } + + public CompletableFuture convertAndSend(String exchange, @Nullable String routingKey, Object message) { + return doConvertAndSend(exchange, routingKey != null ? routingKey : this.defaultRoutingKey, null, message, null); + } + + public CompletableFuture convertAndSend(Object message, + @Nullable MessagePostProcessor messagePostProcessor) { + + return doConvertAndSend(null, null, null, message, messagePostProcessor); + } + + public CompletableFuture convertAndSend(String queue, Object message, + @Nullable MessagePostProcessor messagePostProcessor) { + + return doConvertAndSend(null, null, queue, message, messagePostProcessor); + } + + public CompletableFuture convertAndSend(String exchange, @Nullable String routingKey, Object message, + @Nullable MessagePostProcessor messagePostProcessor) { + + return doConvertAndSend(exchange, routingKey, null, message, messagePostProcessor); + } + + private CompletableFuture 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 receive() { + return receive(getRequiredQueue()); + } + + @SuppressWarnings("try") + public CompletableFuture receive(String queueName) { + CompletableFuture 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 receiveAndConvert() { + return receiveAndConvert(getRequiredQueue()); + } + + public CompletableFuture 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 CompletableFuture receiveAndConvert(ParameterizedTypeReference 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 CompletableFuture receiveAndConvert(String queueName, ParameterizedTypeReference 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 boolean receiveAndReply(ReceiveAndReplyCallback callback) throws AmqpException { + throw new UnsupportedOperationException(); + } + + public boolean receiveAndReply(String queueName, ReceiveAndReplyCallback callback) throws AmqpException { + throw new UnsupportedOperationException(); + } + + public boolean receiveAndReply(ReceiveAndReplyCallback callback, String replyExchange, String replyRoutingKey) throws AmqpException { + throw new UnsupportedOperationException(); + } + + public boolean receiveAndReply(String queueName, ReceiveAndReplyCallback callback, String replyExchange, String replyRoutingKey) throws AmqpException { + throw new UnsupportedOperationException(); + } + + public boolean receiveAndReply(ReceiveAndReplyCallback callback, ReplyToAddressCallback replyToAddressCallback) throws AmqpException { + throw new UnsupportedOperationException(); + } + + public boolean receiveAndReply(String queueName, ReceiveAndReplyCallback callback, ReplyToAddressCallback replyToAddressCallback) throws AmqpException { + throw new UnsupportedOperationException(); + } + + @Override + public CompletableFuture sendAndReceive(Message message) { + throw new UnsupportedOperationException(); + } + + @Override + public CompletableFuture sendAndReceive(String routingKey, Message message) { + throw new UnsupportedOperationException(); + } + + @Override + public CompletableFuture sendAndReceive(String exchange, String routingKey, Message message) { + throw new UnsupportedOperationException(); + } + + @Override + public CompletableFuture convertSendAndReceive(Object object) { + throw new UnsupportedOperationException(); + } + + @Override + public CompletableFuture convertSendAndReceive(String routingKey, Object object) { + throw new UnsupportedOperationException(); + } + + @Override + public CompletableFuture convertSendAndReceive(String exchange, String routingKey, Object object) { + throw new UnsupportedOperationException(); + } + + @Override + public CompletableFuture convertSendAndReceive(Object object, MessagePostProcessor messagePostProcessor) { + throw new UnsupportedOperationException(); + } + + @Override + public CompletableFuture convertSendAndReceive(String routingKey, Object object, MessagePostProcessor messagePostProcessor) { + throw new UnsupportedOperationException(); + } + + @Override + public CompletableFuture convertSendAndReceive(String exchange, String routingKey, Object object, @Nullable MessagePostProcessor messagePostProcessor) { + throw new UnsupportedOperationException(); + } + + @Override + public CompletableFuture convertSendAndReceiveAsType(Object object, ParameterizedTypeReference responseType) { + throw new UnsupportedOperationException(); + } + + @Override + public CompletableFuture convertSendAndReceiveAsType(String routingKey, Object object, ParameterizedTypeReference responseType) { + throw new UnsupportedOperationException(); + } + + @Override + public CompletableFuture convertSendAndReceiveAsType(String exchange, String routingKey, Object object, ParameterizedTypeReference responseType) { + throw new UnsupportedOperationException(); + } + + @Override + public CompletableFuture convertSendAndReceiveAsType(Object object, MessagePostProcessor messagePostProcessor, ParameterizedTypeReference responseType) { + throw new UnsupportedOperationException(); + } + + @Override + public CompletableFuture convertSendAndReceiveAsType(String routingKey, Object object, @Nullable MessagePostProcessor messagePostProcessor, @Nullable ParameterizedTypeReference responseType) { + throw new UnsupportedOperationException(); + } + + @Override + public CompletableFuture convertSendAndReceiveAsType(String exchange, String routingKey, Object object, @Nullable MessagePostProcessor messagePostProcessor, @Nullable ParameterizedTypeReference 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); + } + +} diff --git a/spring-rabbitmq-client/src/main/java/org/springframework/amqp/rabbitmq/client/package-info.java b/spring-rabbitmq-client/src/main/java/org/springframework/amqp/rabbitmq/client/package-info.java new file mode 100644 index 00000000..ae2d6944 --- /dev/null +++ b/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; \ No newline at end of file diff --git a/spring-rabbitmq-client/src/test/java/org/springframework/amqp/rabbitmq/client/RabbitAmqpAdminTests.java b/spring-rabbitmq-client/src/test/java/org/springframework/amqp/rabbitmq/client/RabbitAmqpAdminTests.java new file mode 100644 index 00000000..6dedc04b --- /dev/null +++ b/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 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)); + } + + } + +} diff --git a/spring-rabbitmq-client/src/test/java/org/springframework/amqp/rabbitmq/client/RabbitAmqpTemplateTests.java b/spring-rabbitmq-client/src/test/java/org/springframework/amqp/rabbitmq/client/RabbitAmqpTemplateTests.java new file mode 100644 index 00000000..fbe714cd --- /dev/null +++ b/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"); + } + + } + +} diff --git a/spring-rabbitmq-client/src/test/java/org/springframework/amqp/rabbitmq/client/RabbitAmqpTestBase.java b/spring-rabbitmq-client/src/test/java/org/springframework/amqp/rabbitmq/client/RabbitAmqpTestBase.java new file mode 100644 index 00000000..bd4eca2c --- /dev/null +++ b/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); + } + + } + +} diff --git a/spring-rabbitmq-client/src/test/resources/log4j2-test.xml b/spring-rabbitmq-client/src/test/resources/log4j2-test.xml new file mode 100644 index 00000000..a4931c98 --- /dev/null +++ b/spring-rabbitmq-client/src/test/resources/log4j2-test.xml @@ -0,0 +1,14 @@ + + + + + + + + + + + + + +