diff --git a/build.gradle b/build.gradle index f226c681..3f06a8d3 100644 --- a/build.gradle +++ b/build.gradle @@ -479,7 +479,6 @@ project('spring-rabbitmq-client') { dependencies { api project(':spring-rabbit') api "com.rabbitmq.client:amqp-client:$rabbitmqAmqpClientVersion" - api 'io.micrometer:micrometer-observation' testApi project(':spring-rabbit-junit') @@ -488,10 +487,6 @@ project('spring-rabbitmq-client') { 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' } } diff --git a/spring-amqp/src/main/java/org/springframework/amqp/core/AmqpAcknowledgment.java b/spring-amqp/src/main/java/org/springframework/amqp/core/AmqpAcknowledgment.java new file mode 100644 index 00000000..8b9b94d0 --- /dev/null +++ b/spring-amqp/src/main/java/org/springframework/amqp/core/AmqpAcknowledgment.java @@ -0,0 +1,58 @@ +/* + * 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.core; + +/** + * An abstraction over acknowledgments. + * + * @author Artem Bilan + * + * @since 4.0 + */ +@FunctionalInterface +public interface AmqpAcknowledgment { + + /** + * Acknowledge the message. + * @param status the status. + */ + void acknowledge(Status status); + + default void acknowledge() { + acknowledge(Status.ACCEPT); + } + + enum Status { + + /** + * Mark the message as accepted. + */ + ACCEPT, + + /** + * Mark the message as rejected. + */ + REJECT, + + /** + * Reject the message and requeue so that it will be redelivered. + */ + REQUEUE + + } + +} diff --git a/spring-amqp/src/main/java/org/springframework/amqp/core/MessageProperties.java b/spring-amqp/src/main/java/org/springframework/amqp/core/MessageProperties.java index 758e03af..08c55e57 100644 --- a/spring-amqp/src/main/java/org/springframework/amqp/core/MessageProperties.java +++ b/spring-amqp/src/main/java/org/springframework/amqp/core/MessageProperties.java @@ -157,6 +157,8 @@ public class MessageProperties implements Serializable { private transient @Nullable Object targetBean; + private transient @Nullable AmqpAcknowledgment amqpAcknowledgment; + public void setHeader(String key, Object value) { this.headers.put(key, value); } @@ -641,6 +643,25 @@ public class MessageProperties implements Serializable { } } + /** + * Return the {@link AmqpAcknowledgment} for consumer if any. + * @return the {@link AmqpAcknowledgment} for consumer if any. + * @since 4.0 + */ + public @Nullable AmqpAcknowledgment getAmqpAcknowledgment() { + return this.amqpAcknowledgment; + } + + /** + * Set an {@link AmqpAcknowledgment} for manual acks in the target message processor. + * This is only in-application a consumer side logic. + * @param amqpAcknowledgment the {@link AmqpAcknowledgment} to use in the application. + * @since 4.0 + */ + public void setAmqpAcknowledgment(AmqpAcknowledgment amqpAcknowledgment) { + this.amqpAcknowledgment = amqpAcknowledgment; + } + @Override // NOSONAR complexity public int hashCode() { final int prime = 31; diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractMessageListenerContainer.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractMessageListenerContainer.java index 6a24f0fb..130428dd 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractMessageListenerContainer.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractMessageListenerContainer.java @@ -752,7 +752,7 @@ public abstract class AbstractMessageListenerContainer extends ObservableListene * to be sent to the dead letter exchange. Setting to false causes all rejections to not * be requeued. When true, the default can be overridden by the listener throwing an * {@link AmqpRejectAndDontRequeueException}. Default true. - * @param defaultRequeueRejected true to reject by default. + * @param defaultRequeueRejected true to requeue by default. */ public void setDefaultRequeueRejected(boolean defaultRequeueRejected) { this.defaultRequeueRejected = defaultRequeueRejected; diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/support/ContainerUtils.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/support/ContainerUtils.java index 06d83ebc..f5709479 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/support/ContainerUtils.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/support/ContainerUtils.java @@ -1,5 +1,5 @@ /* - * Copyright 2018-2022 the original author or authors. + * Copyright 2018-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. @@ -17,8 +17,10 @@ package org.springframework.amqp.rabbit.listener.support; import org.apache.commons.logging.Log; +import org.jspecify.annotations.Nullable; import org.springframework.amqp.AmqpRejectAndDontRequeueException; +import org.springframework.amqp.ImmediateAcknowledgeAmqpException; import org.springframework.amqp.ImmediateRequeueAmqpException; import org.springframework.amqp.rabbit.listener.exception.MessageRejectedWhileStoppingException; @@ -59,7 +61,11 @@ public final class ContainerUtils { shouldRequeue = true; break; } - t = t.getCause(); + Throwable cause = t.getCause(); + if (cause == t) { + break; + } + t = cause; } if (logger.isDebugEnabled()) { logger.debug("Rejecting messages (requeue=" + shouldRequeue + ")"); @@ -75,8 +81,41 @@ public final class ContainerUtils { * @since 2.2 */ public static boolean isRejectManual(Throwable ex) { - return ex instanceof AmqpRejectAndDontRequeueException aradrex - && aradrex.isRejectManual(); + AmqpRejectAndDontRequeueException amqpRejectAndDontRequeueException = + findInCause(ex, AmqpRejectAndDontRequeueException.class); + return amqpRejectAndDontRequeueException != null && amqpRejectAndDontRequeueException.isRejectManual(); + } + + /** + * Return true for {@link ImmediateAcknowledgeAmqpException}. + * @param ex the exception to traverse. + * @return true if an {@link ImmediateAcknowledgeAmqpException} is present in the cause chain. + * @since 4.0 + */ + public static boolean isImmediateAcknowledge(Throwable ex) { + return findInCause(ex, ImmediateAcknowledgeAmqpException.class) != null; + } + + /** + * Return true for {@link AmqpRejectAndDontRequeueException}. + * @param ex the exception to traverse. + * @return true if an {@link AmqpRejectAndDontRequeueException} is present in the cause chain. + * @since 4.0 + */ + public static boolean isAmqpReject(Throwable ex) { + return findInCause(ex, AmqpRejectAndDontRequeueException.class) != null; + } + + @SuppressWarnings("unchecked") + private static @Nullable T findInCause(Throwable throwable, Class exceptionToFind) { + if (exceptionToFind.isAssignableFrom(throwable.getClass())) { + return (T) throwable; + } + Throwable cause = throwable.getCause(); + if (cause == null || cause == throwable) { + return null; + } + return findInCause(cause, exceptionToFind); } } 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 index 2ba589bd..c2a6eba6 100644 --- 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 @@ -16,11 +16,7 @@ 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; @@ -182,38 +178,20 @@ public class RabbitAmqpTemplate implements AsyncAmqpTemplate, InitializingBean, 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 amqpMessage = this.publisher.message(); 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); + amqpMessage = address.message(); + + RabbitAmqpUtils.toAmqpMessage(message, amqpMessage); + CompletableFuture publishResult = new CompletableFuture<>(); - this.publisher.publish(address.message(), + this.publisher.publish(amqpMessage, (context) -> { switch (context.status()) { case ACCEPTED -> publishResult.complete(true); @@ -299,7 +277,7 @@ public class RabbitAmqpTemplate implements AsyncAmqpTemplate, InitializingBean, .priority(10) .messageHandler((context, message) -> { context.accept(); - messageFuture.complete(fromAmqpMessage(message)); + messageFuture.complete(RabbitAmqpUtils.fromAmqpMessage(message, null)); }) .build(); @@ -452,62 +430,4 @@ public class RabbitAmqpTemplate implements AsyncAmqpTemplate, InitializingBean, 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/RabbitAmqpUtils.java b/spring-rabbitmq-client/src/main/java/org/springframework/amqp/rabbitmq/client/RabbitAmqpUtils.java new file mode 100644 index 00000000..340b82d9 --- /dev/null +++ b/spring-rabbitmq-client/src/main/java/org/springframework/amqp/rabbitmq/client/RabbitAmqpUtils.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.nio.charset.StandardCharsets; +import java.util.Date; +import java.util.Map; +import java.util.UUID; + +import com.rabbitmq.client.amqp.Consumer; +import org.jspecify.annotations.Nullable; + +import org.springframework.amqp.core.Message; +import org.springframework.amqp.core.MessageProperties; +import org.springframework.amqp.utils.JavaUtils; + +/** + * The utilities for RabbitMQ AMQP 1.0 protocol API. + */ +public final class RabbitAmqpUtils { + + /** + * Convert {@link com.rabbitmq.client.amqp.Message} into {@link Message}. + * @param amqpMessage the {@link com.rabbitmq.client.amqp.Message} convert from. + * @param context the {@link Consumer.Context} for manual message settlement. + * @return the {@link Message} mapped from a {@link com.rabbitmq.client.amqp.Message}. + */ + public static Message fromAmqpMessage(com.rabbitmq.client.amqp.Message amqpMessage, + Consumer.@Nullable Context context) { + + 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); + + if (context != null) { + messageProperties.setAmqpAcknowledgment((status) -> { + switch (status) { + case ACCEPT -> context.accept(); + case REJECT -> context.discard(); + case REQUEUE -> context.requeue(); + } + }); + } + + return new Message(amqpMessage.body(), messageProperties); + } + + /** + * Convert {@link com.rabbitmq.client.amqp.Message} into {@link Message}. + * @param amqpMessage the {@link com.rabbitmq.client.amqp.Message} convert from. + */ + public static void toAmqpMessage(Message message, com.rabbitmq.client.amqp.Message amqpMessage) { + MessageProperties messageProperties = message.getMessageProperties(); + + amqpMessage + .body(message.getBody()) + .contentEncoding(messageProperties.getContentEncoding()) + .contentType(messageProperties.getContentType()) + .messageId(messageProperties.getMessageId()) + .correlationId(messageProperties.getCorrelationId()) + .priority(messageProperties.getPriority().byteValue()) + .replyTo(messageProperties.getReplyTo()); + + 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))); + } + + 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 RabbitAmqpUtils() { + } + +} diff --git a/spring-rabbitmq-client/src/main/java/org/springframework/amqp/rabbitmq/client/config/RabbitAmqpListenerContainerFactory.java b/spring-rabbitmq-client/src/main/java/org/springframework/amqp/rabbitmq/client/config/RabbitAmqpListenerContainerFactory.java new file mode 100644 index 00000000..926580f5 --- /dev/null +++ b/spring-rabbitmq-client/src/main/java/org/springframework/amqp/rabbitmq/client/config/RabbitAmqpListenerContainerFactory.java @@ -0,0 +1,97 @@ +/* + * Copyright 2021-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.config; + +import com.rabbitmq.client.amqp.Connection; +import org.aopalliance.aop.Advice; +import org.jspecify.annotations.Nullable; + +import org.springframework.amqp.rabbit.config.BaseRabbitListenerContainerFactory; +import org.springframework.amqp.rabbit.config.ContainerCustomizer; +import org.springframework.amqp.rabbit.listener.MethodRabbitListenerEndpoint; +import org.springframework.amqp.rabbit.listener.RabbitListenerEndpoint; +import org.springframework.amqp.rabbitmq.client.listener.RabbitAmqpListenerContainer; +import org.springframework.amqp.rabbitmq.client.listener.RabbitAmqpMessageListenerAdapter; +import org.springframework.amqp.utils.JavaUtils; + +/** + * Factory for {@link RabbitAmqpListenerContainer}. + * To use it as default one, has to be configured with a + * {@link org.springframework.amqp.rabbit.annotation.RabbitListenerAnnotationBeanPostProcessor#DEFAULT_RABBIT_LISTENER_CONTAINER_FACTORY_BEAN_NAME}. + * + * @author Artem Bilan + * + * @since 4.0 + * + */ +public class RabbitAmqpListenerContainerFactory + extends BaseRabbitListenerContainerFactory { + + private final Connection connection; + + private @Nullable ContainerCustomizer containerCustomizer; + + /** + * Construct an instance using the provided amqpConnection. + * @param amqpConnection the connection. + */ + public RabbitAmqpListenerContainerFactory(Connection amqpConnection) { + this.connection = amqpConnection; + } + + /** + * Set a {@link ContainerCustomizer} that is invoked after a container is created and + * configured to enable further customization of the container. + * @param containerCustomizer the customizer. + */ + public void setContainerCustomizer(ContainerCustomizer containerCustomizer) { + this.containerCustomizer = containerCustomizer; + } + + @Override + public RabbitAmqpListenerContainer createListenerContainer(@Nullable RabbitListenerEndpoint endpoint) { + if (endpoint instanceof MethodRabbitListenerEndpoint methodRabbitListenerEndpoint) { + methodRabbitListenerEndpoint.setAdapterProvider( + (batch, bean, method, returnExceptions, errorHandler, batchingStrategy) -> + new RabbitAmqpMessageListenerAdapter(bean, method, returnExceptions, errorHandler)); + } + RabbitAmqpListenerContainer container = createContainerInstance(); + Advice[] adviceChain = getAdviceChain(); + JavaUtils.INSTANCE + .acceptIfNotNull(adviceChain, container::setAdviceChain) + .acceptIfNotNull(getDefaultRequeueRejected(), container::setDefaultRequeue); + + applyCommonOverrides(endpoint, container); + + if (endpoint != null) { + JavaUtils.INSTANCE + .acceptIfNotNull(endpoint.getAckMode(), + (ackMode) -> container.setAutoSettle(!ackMode.isManual())) + .acceptIfNotNull(endpoint.getConcurrency(), + (concurrency) -> container.setConsumersPerQueue(Integer.parseInt(concurrency))); + } + if (this.containerCustomizer != null) { + this.containerCustomizer.configure(container); + } + return container; + } + + protected RabbitAmqpListenerContainer createContainerInstance() { + return new RabbitAmqpListenerContainer(this.connection); + } + +} diff --git a/spring-rabbitmq-client/src/main/java/org/springframework/amqp/rabbitmq/client/config/package-info.java b/spring-rabbitmq-client/src/main/java/org/springframework/amqp/rabbitmq/client/config/package-info.java new file mode 100644 index 00000000..3b874dd2 --- /dev/null +++ b/spring-rabbitmq-client/src/main/java/org/springframework/amqp/rabbitmq/client/config/package-info.java @@ -0,0 +1,5 @@ +/** + * Provides classes for Spring application context support. + */ +@org.jspecify.annotations.NullMarked +package org.springframework.amqp.rabbitmq.client.config; \ No newline at end of file diff --git a/spring-rabbitmq-client/src/main/java/org/springframework/amqp/rabbitmq/client/listener/RabbitAmqpListenerContainer.java b/spring-rabbitmq-client/src/main/java/org/springframework/amqp/rabbitmq/client/listener/RabbitAmqpListenerContainer.java new file mode 100644 index 00000000..479af40f --- /dev/null +++ b/spring-rabbitmq-client/src/main/java/org/springframework/amqp/rabbitmq/client/listener/RabbitAmqpListenerContainer.java @@ -0,0 +1,392 @@ +/* + * 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.listener; + +import java.time.Duration; +import java.util.Arrays; +import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.locks.Lock; +import java.util.concurrent.locks.ReentrantLock; + +import com.rabbitmq.client.amqp.Connection; +import com.rabbitmq.client.amqp.Consumer; +import com.rabbitmq.client.amqp.Resource; +import org.aopalliance.aop.Advice; +import org.apache.commons.logging.LogFactory; +import org.jspecify.annotations.Nullable; + +import org.springframework.amqp.AmqpRejectAndDontRequeueException; +import org.springframework.amqp.core.AcknowledgeMode; +import org.springframework.amqp.core.AmqpAcknowledgment; +import org.springframework.amqp.core.Message; +import org.springframework.amqp.core.MessageListener; +import org.springframework.amqp.rabbit.listener.ConditionalRejectingErrorHandler; +import org.springframework.amqp.rabbit.listener.MessageListenerContainer; +import org.springframework.amqp.rabbit.listener.support.ContainerUtils; +import org.springframework.amqp.rabbitmq.client.RabbitAmqpUtils; +import org.springframework.aop.framework.ProxyFactory; +import org.springframework.aop.support.DefaultPointcutAdvisor; +import org.springframework.core.log.LogAccessor; +import org.springframework.util.Assert; +import org.springframework.util.ErrorHandler; +import org.springframework.util.LinkedMultiValueMap; +import org.springframework.util.MultiValueMap; +import org.springframework.util.ObjectUtils; + +/** + * A listener container for RabbitMQ AMQP 1.0 Consumer. + * + * @author Artem Bilan + * + * @since 4.0 + * + */ +public class RabbitAmqpListenerContainer implements MessageListenerContainer { + + private static final LogAccessor LOG = new LogAccessor(LogFactory.getLog(RabbitAmqpListenerContainer.class)); + + private final Lock lock = new ReentrantLock(); + + private final Connection connection; + + private final MultiValueMap queueToConsumers = new LinkedMultiValueMap<>(); + + private String @Nullable [] queues; + + private Advice @Nullable [] adviceChain; + + private int initialCredits = 100; + + private int priority; + + private Resource.StateListener @Nullable [] stateListeners; + + private boolean autoSettle = true; + + private boolean defaultRequeue = true; + + private int consumersPerQueue = 1; + + private @Nullable MessageListener messageListener; + + private ErrorHandler errorHandler = new ConditionalRejectingErrorHandler(); + + private boolean autoStartup = true; + + private @Nullable String listenerId; + + private Duration gracefulShutdownPeriod = Duration.ofSeconds(30); + + /** + * Construct an instance using the provided connection. + * @param connection to use. + */ + public RabbitAmqpListenerContainer(Connection connection) { + this.connection = connection; + } + + @Override + public void setQueueNames(String... queueNames) { + this.queues = Arrays.copyOf(queueNames, queueNames.length); + } + + public void setInitialCredits(int initialCredits) { + this.initialCredits = initialCredits; + } + + public void setPriority(int priority) { + this.priority = priority; + } + + public void setStateListeners(Resource.StateListener... stateListeners) { + this.stateListeners = Arrays.copyOf(stateListeners, stateListeners.length); + } + + @Override + public void setAutoStartup(boolean autoStart) { + this.autoStartup = autoStart; + } + + @Override + public boolean isAutoStartup() { + return this.autoStartup; + } + + /** + * Set an advice chain to apply to the listener. + * @param advices the advice chain. + * @since 2.4.5 + */ + public void setAdviceChain(Advice... advices) { + Assert.notNull(advices, "'advices' cannot be null"); + Assert.noNullElements(advices, "'advices' cannot have null elements"); + this.adviceChain = Arrays.copyOf(advices, advices.length); + } + + /** + * Set to {@code false} to propagate a + * {@link org.springframework.amqp.core.MessageProperties#setAmqpAcknowledgment(AmqpAcknowledgment)} + * for target {@link MessageListener} manual settlement. + * In case of {@link RabbitAmqpMessageListener}, the native {@link Consumer.Context} + * should be used for manual settlement. + * @param autoSettle to call {@link Consumer.Context#accept()} automatically. + */ + public void setAutoSettle(boolean autoSettle) { + this.autoSettle = autoSettle; + } + + /** + * Set the default behavior when a message processing has failed. + * When true, messages will be requeued, when false, they will be discarded. + * When true, the default can be overridden by the listener throwing an + * {@link AmqpRejectAndDontRequeueException}. Default true. + * @param defaultRequeue true to requeue by default. + */ + public void setDefaultRequeue(boolean defaultRequeue) { + this.defaultRequeue = defaultRequeue; + } + + public void setGracefulShutdownPeriod(Duration gracefulShutdownPeriod) { + this.gracefulShutdownPeriod = gracefulShutdownPeriod; + } + + /** + * Each queue runs in its own consumer; set this property to create multiple + * consumers for each queue. + * Can be treated as {@code concurrency}, but per queue. + * @param consumersPerQueue the consumers per queue. + */ + public void setConsumersPerQueue(int consumersPerQueue) { + this.consumersPerQueue = consumersPerQueue; + } + + public void setErrorHandler(ErrorHandler errorHandler) { + this.errorHandler = errorHandler; + } + + @Override + public void setListenerId(String id) { + this.listenerId = id; + } + + @Override + public void setupMessageListener(MessageListener messageListener) { + this.messageListener = messageListener; + if (!ObjectUtils.isEmpty(this.adviceChain)) { + ProxyFactory factory = new ProxyFactory(messageListener); + for (Advice advice : this.adviceChain) { + factory.addAdvisor(new DefaultPointcutAdvisor(advice)); + } + factory.setInterfaces(messageListener.getClass().getInterfaces()); + this.messageListener = (MessageListener) factory.getProxy(getClass().getClassLoader()); + } + } + + @Override + public @Nullable Object getMessageListener() { + return this.messageListener; + } + + @Override + public void afterPropertiesSet() { + Assert.state(this.queues != null, "At least one queue has to be provided for consuming."); + Assert.state(this.messageListener != null, "The 'messageListener' must be provided."); + + this.messageListener.containerAckMode(this.autoSettle ? AcknowledgeMode.AUTO : AcknowledgeMode.MANUAL); + } + + @Override + public boolean isRunning() { + this.lock.lock(); + try { + return !this.queueToConsumers.isEmpty(); + } + finally { + this.lock.unlock(); + } + } + + @Override + @SuppressWarnings("NullAway") // Dataflow analysis limitation + public void start() { + this.lock.lock(); + try { + if (this.queueToConsumers.isEmpty()) { + for (String queue : this.queues) { + for (int i = 0; i < this.consumersPerQueue; i++) { + Consumer consumer = + this.connection.consumerBuilder() + .queue(queue) + .priority(this.priority) + .initialCredits(this.initialCredits) + .listeners(this.stateListeners) + .messageHandler(this::invokeListener) + .build(); + this.queueToConsumers.add(queue, consumer); + } + } + } + } + finally { + this.lock.unlock(); + } + } + + private void invokeListener(Consumer.Context context, com.rabbitmq.client.amqp.Message amqpMessage) { + try { + doInvokeListener(context, amqpMessage); + if (this.autoSettle) { + context.accept(); + } + } + catch (Exception ex) { + if (!handleSpecialErrors(ex, context)) { + try { + this.errorHandler.handleError(ex); + // If error handler does not re-throw an exception, treat it as a successful processing result. + context.accept(); + } + catch (Exception rethrow) { + if (!handleSpecialErrors(rethrow, context)) { + if (this.defaultRequeue) { + context.requeue(); + } + else { + context.discard(); + } + LOG.error(rethrow, () -> + "The 'errorHandler' has thrown an exception. The '" + amqpMessage + "' is " + + (this.defaultRequeue ? "re-queued." : "discarded.")); + } + } + } + } + } + + @SuppressWarnings("NullAway") // Dataflow analysis limitation + private void doInvokeListener(Consumer.Context context, com.rabbitmq.client.amqp.Message amqpMessage) { + Consumer.@Nullable Context contextToUse = this.autoSettle ? null : context; + if (this.messageListener instanceof RabbitAmqpMessageListener amqpMessageListener) { + amqpMessageListener.onAmqpMessage(amqpMessage, contextToUse); + } + else { + Message message = RabbitAmqpUtils.fromAmqpMessage(amqpMessage, contextToUse); + this.messageListener.onMessage(message); + } + } + + private boolean handleSpecialErrors(Exception ex, Consumer.Context context) { + if (ContainerUtils.shouldRequeue(this.defaultRequeue, ex, LOG.getLog())) { + context.requeue(); + return true; + } + if (ContainerUtils.isAmqpReject(ex)) { + context.discard(); + return true; + } + if (ContainerUtils.isImmediateAcknowledge(ex)) { + context.accept(); + return true; + } + return false; + } + + @Override + public void stop() { + stop(() -> { + }); + } + + @Override + @SuppressWarnings("unchecked") + public void stop(Runnable callback) { + this.lock.lock(); + try { + CompletableFuture[] completableFutures = + this.queueToConsumers.values().stream() + .flatMap(List::stream) + .peek(Consumer::pause) + .map((consumer) -> + CompletableFuture.supplyAsync(() -> { + try (consumer) { + while (consumer.unsettledMessageCount() > 0) { + Thread.sleep(100); + } + } + catch (InterruptedException ex) { + Thread.currentThread().interrupt(); + throw new RuntimeException(ex); + } + return null; + })) + .toArray(CompletableFuture[]::new); + + CompletableFuture.allOf(completableFutures) + .orTimeout(this.gracefulShutdownPeriod.toMillis(), TimeUnit.MILLISECONDS) + .whenComplete((unused, throwable) -> { + this.queueToConsumers.clear(); + callback.run(); + }); + } + finally { + this.lock.unlock(); + } + } + + /** + * Pause all the consumer for all queues. + */ + public void pause() { + this.queueToConsumers.values() + .stream() + .flatMap(List::stream) + .forEach(Consumer::pause); + } + + /** + * Resume all the consumer for all queues. + */ + public void resume() { + this.queueToConsumers.values() + .stream() + .flatMap(List::stream) + .forEach(Consumer::unpause); + } + + /** + * Pause all the consumer for specific queue. + */ + public void pause(String queueName) { + List consumers = this.queueToConsumers.get(queueName); + if (consumers != null) { + consumers.forEach(Consumer::pause); + } + } + + /** + * Resume all the consumer for specific queue. + */ + public void resume(String queueName) { + List consumers = this.queueToConsumers.get(queueName); + if (consumers != null) { + consumers.forEach(Consumer::unpause); + } + } + +} diff --git a/spring-rabbitmq-client/src/main/java/org/springframework/amqp/rabbitmq/client/listener/RabbitAmqpMessageListener.java b/spring-rabbitmq-client/src/main/java/org/springframework/amqp/rabbitmq/client/listener/RabbitAmqpMessageListener.java new file mode 100644 index 00000000..70cc52cb --- /dev/null +++ b/spring-rabbitmq-client/src/main/java/org/springframework/amqp/rabbitmq/client/listener/RabbitAmqpMessageListener.java @@ -0,0 +1,42 @@ +/* + * Copyright 2021-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.listener; + +import com.rabbitmq.client.amqp.Consumer; +import com.rabbitmq.client.amqp.Message; +import org.jspecify.annotations.Nullable; + +import org.springframework.amqp.core.MessageListener; + +/** + * A message listener that receives native AMQP 1.0 messages from RabbitMQ. + * + * @author Artem Bilan + * + * @since 4.0 + */ +public interface RabbitAmqpMessageListener extends MessageListener { + + /** + * Process an AMQP message. + * @param message the message to process. + * @param context the consumer context to settle message. + * Null if container is configured for {@code autoSettle}. + */ + void onAmqpMessage(Message message, Consumer.@Nullable Context context); + +} diff --git a/spring-rabbitmq-client/src/main/java/org/springframework/amqp/rabbitmq/client/listener/RabbitAmqpMessageListenerAdapter.java b/spring-rabbitmq-client/src/main/java/org/springframework/amqp/rabbitmq/client/listener/RabbitAmqpMessageListenerAdapter.java new file mode 100644 index 00000000..98ba024e --- /dev/null +++ b/spring-rabbitmq-client/src/main/java/org/springframework/amqp/rabbitmq/client/listener/RabbitAmqpMessageListenerAdapter.java @@ -0,0 +1,72 @@ +/* + * 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.listener; + +import java.lang.reflect.Method; + +import com.rabbitmq.client.amqp.Consumer; +import org.jspecify.annotations.Nullable; + +import org.springframework.amqp.rabbit.listener.adapter.InvocationResult; +import org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter; +import org.springframework.amqp.rabbit.listener.api.RabbitListenerErrorHandler; +import org.springframework.amqp.rabbit.support.ListenerExecutionFailedException; +import org.springframework.amqp.rabbitmq.client.RabbitAmqpUtils; + +/** + * A {@link MessagingMessageListenerAdapter} extension for the {@link RabbitAmqpMessageListener}. + * Provides these arguments for the {@link #getHandlerAdapter()} invocation: + *
    + *
  • {@link com.rabbitmq.client.amqp.Message} - the native AMQP 1.0 message without any conversions
    • + *
  • {@link org.springframework.amqp.core.Message} - Spring AMQP message abstraction as conversion result from the native AMQP 1.0 message
    • + *
  • {@link org.springframework.messaging.Message} - Spring Messaging abstraction as conversion result from the Spring AMQP message
    • + *
  • {@link Consumer.Context} - RabbitMQ AMQP client consumer settlement API.
    • + *
  • {@link org.springframework.amqp.core.AmqpAcknowledgment} - Spring AMQP acknowledgment abstraction: delegates to the {@link Consumer.Context}
    • + *
+ * + * @author Artem Bilan + * + * @since 4.0 + */ +public class RabbitAmqpMessageListenerAdapter extends MessagingMessageListenerAdapter + implements RabbitAmqpMessageListener { + + public RabbitAmqpMessageListenerAdapter(@Nullable Object bean, @Nullable Method method, boolean returnExceptions, + @Nullable RabbitListenerErrorHandler errorHandler) { + + super(bean, method, returnExceptions, errorHandler); + } + + @Override + public void onAmqpMessage(com.rabbitmq.client.amqp.Message amqpMessage, Consumer.@Nullable Context context) { + try { + org.springframework.amqp.core.Message springMessage = RabbitAmqpUtils.fromAmqpMessage(amqpMessage, context); + org.springframework.messaging.Message messagingMessage = toMessagingMessage(springMessage); + InvocationResult result = getHandlerAdapter() + .invoke(messagingMessage, + springMessage, springMessage.getMessageProperties().getAmqpAcknowledgment(), + amqpMessage, context); + if (result.getReturnValue() != null) { + logger.warn("Replies are not currently supported with RabbitMQ AMQP 1.0 listeners"); + } + } + catch (Exception ex) { + throw new ListenerExecutionFailedException("Failed to invoke listener", ex); + } + } + +} diff --git a/spring-rabbitmq-client/src/main/java/org/springframework/amqp/rabbitmq/client/listener/package-info.java b/spring-rabbitmq-client/src/main/java/org/springframework/amqp/rabbitmq/client/listener/package-info.java new file mode 100644 index 00000000..23d84893 --- /dev/null +++ b/spring-rabbitmq-client/src/main/java/org/springframework/amqp/rabbitmq/client/listener/package-info.java @@ -0,0 +1,5 @@ +/** + * Provides Spring support for RabbitMQ AMQP 1.0 Consumer. + */ +@org.jspecify.annotations.NullMarked +package org.springframework.amqp.rabbitmq.client.listener; \ No newline at end of file