diff --git a/spring-amqp/src/main/java/org/springframework/amqp/support/converter/AbstractJackson2MessageConverter.java b/spring-amqp/src/main/java/org/springframework/amqp/support/converter/AbstractJackson2MessageConverter.java index 5c102712..14f07da9 100644 --- a/spring-amqp/src/main/java/org/springframework/amqp/support/converter/AbstractJackson2MessageConverter.java +++ b/spring-amqp/src/main/java/org/springframework/amqp/support/converter/AbstractJackson2MessageConverter.java @@ -21,6 +21,7 @@ import java.lang.reflect.Modifier; import java.lang.reflect.Type; import java.nio.charset.Charset; import java.nio.charset.StandardCharsets; +import java.util.Optional; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -96,6 +97,8 @@ public abstract class AbstractJackson2MessageConverter extends AbstractMessageCo private boolean alwaysConvertToInferredType; + private boolean nullAsOptionalEmpty; + /** * Construct with the provided {@link ObjectMapper} instance. * @param objectMapper the {@link ObjectMapper} to use. @@ -148,6 +151,15 @@ public abstract class AbstractJackson2MessageConverter extends AbstractMessageCo this.supportedCTCharset = this.supportedContentType.getParameter("charset"); } + /** + * When true, if jackson decodes the body as {@code null} convert to {@link Optional#empty()} + * instead of returning the original body. Default false. + * @param nullAsOptionalEmpty true to return empty. + * @since 2.4.7 + */ + public void setNullAsOptionalEmpty(boolean nullAsOptionalEmpty) { + this.nullAsOptionalEmpty = nullAsOptionalEmpty; + } @Nullable public ClassMapper getClassMapper() { @@ -316,7 +328,12 @@ public abstract class AbstractJackson2MessageConverter extends AbstractMessageCo } } if (content == null) { - content = message.getBody(); + if (this.nullAsOptionalEmpty) { + content = Optional.empty(); + } + else { + content = message.getBody(); + } } return content; } diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/annotation/RabbitListenerAnnotationBeanPostProcessor.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/annotation/RabbitListenerAnnotationBeanPostProcessor.java index 74983a73..aebf241f 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/annotation/RabbitListenerAnnotationBeanPostProcessor.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/annotation/RabbitListenerAnnotationBeanPostProcessor.java @@ -18,6 +18,7 @@ package org.springframework.amqp.rabbit.annotation; import java.lang.reflect.AnnotatedElement; import java.lang.reflect.Method; +import java.lang.reflect.Type; import java.nio.charset.Charset; import java.nio.charset.StandardCharsets; import java.util.ArrayList; @@ -28,6 +29,7 @@ import java.util.HashMap; import java.util.HashSet; import java.util.List; import java.util.Map; +import java.util.Optional; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; @@ -73,6 +75,7 @@ import org.springframework.beans.factory.config.ConfigurableBeanFactory; import org.springframework.beans.factory.config.ConfigurableListableBeanFactory; import org.springframework.context.EnvironmentAware; import org.springframework.context.expression.StandardBeanExpressionResolver; +import org.springframework.core.MethodParameter; import org.springframework.core.Ordered; import org.springframework.core.annotation.AnnotationUtils; import org.springframework.core.annotation.MergedAnnotations; @@ -82,9 +85,14 @@ import org.springframework.core.convert.converter.Converter; import org.springframework.core.convert.support.DefaultConversionService; import org.springframework.core.env.Environment; import org.springframework.core.task.TaskExecutor; +import org.springframework.format.support.DefaultFormattingConversionService; import org.springframework.lang.Nullable; +import org.springframework.messaging.Message; +import org.springframework.messaging.converter.GenericMessageConverter; import org.springframework.messaging.handler.annotation.support.DefaultMessageHandlerMethodFactory; import org.springframework.messaging.handler.annotation.support.MessageHandlerMethodFactory; +import org.springframework.messaging.handler.annotation.support.MethodArgumentNotValidException; +import org.springframework.messaging.handler.annotation.support.PayloadMethodArgumentResolver; import org.springframework.messaging.handler.invocation.HandlerMethodArgumentResolver; import org.springframework.messaging.handler.invocation.InvocableHandlerMethod; import org.springframework.util.Assert; @@ -92,6 +100,7 @@ import org.springframework.util.ClassUtils; import org.springframework.util.CollectionUtils; import org.springframework.util.ReflectionUtils; import org.springframework.util.StringUtils; +import org.springframework.validation.ObjectError; import org.springframework.validation.Validator; /** @@ -977,6 +986,9 @@ public class RabbitListenerAnnotationBeanPostProcessor */ private class RabbitHandlerMethodFactoryAdapter implements MessageHandlerMethodFactory { + private final DefaultFormattingConversionService defaultFormattingConversionService = + new DefaultFormattingConversionService(); + private MessageHandlerMethodFactory factory; RabbitHandlerMethodFactoryAdapter() { @@ -1005,20 +1017,70 @@ public class RabbitListenerAnnotationBeanPostProcessor defaultFactory.setValidator(validator); } defaultFactory.setBeanFactory(RabbitListenerAnnotationBeanPostProcessor.this.beanFactory); - DefaultConversionService conversionService = new DefaultConversionService(); - conversionService.addConverter( + this.defaultFormattingConversionService.addConverter( new BytesToStringConverter(RabbitListenerAnnotationBeanPostProcessor.this.charset)); - defaultFactory.setConversionService(conversionService); + defaultFactory.setConversionService(this.defaultFormattingConversionService); - List customArgumentsResolver = - new ArrayList<>(RabbitListenerAnnotationBeanPostProcessor.this.registrar.getCustomMethodArgumentResolvers()); + List customArgumentsResolver = new ArrayList<>( + RabbitListenerAnnotationBeanPostProcessor.this.registrar.getCustomMethodArgumentResolvers()); defaultFactory.setCustomArgumentResolvers(customArgumentsResolver); + GenericMessageConverter messageConverter = new GenericMessageConverter( + this.defaultFormattingConversionService); + defaultFactory.setMessageConverter(messageConverter); + // Has to be at the end - look at PayloadMethodArgumentResolver documentation + customArgumentsResolver.add(new OptionalEmptyAwarePayloadArgumentResolver(messageConverter, validator)); defaultFactory.afterPropertiesSet(); return defaultFactory; } } + private static class OptionalEmptyAwarePayloadArgumentResolver extends PayloadMethodArgumentResolver { + + OptionalEmptyAwarePayloadArgumentResolver( + org.springframework.messaging.converter.MessageConverter messageConverter, + @Nullable Validator validator) { + + super(messageConverter, validator); + } + + @Override + public Object resolveArgument(MethodParameter parameter, Message message) throws Exception { // NOSONAR + Object resolved = null; + try { + resolved = super.resolveArgument(parameter, message); + } + catch (MethodArgumentNotValidException ex) { + if (message.getPayload().equals(Optional.empty())) { + Type type = parameter.getGenericParameterType(); + List allErrors = ex.getBindingResult().getAllErrors(); + if (allErrors.size() == 1 + && allErrors.get(0).getDefaultMessage().equals("Payload value must not be empty")) { + return Optional.empty(); + } + } + throw ex; + } + /* + * Replace Optional.empty() list elements with null. + */ + if (resolved instanceof List) { + List list = ((List) resolved); + for (int i = 0; i < list.size(); i++) { + if (list.get(i).equals(Optional.empty())) { + list.set(i, null); + } + } + } + return resolved; + } + + @Override + protected boolean isEmptyPayload(Object payload) { + return payload == null || payload.equals(Optional.empty()); + } + + } /** * The metadata holder of the class with {@link RabbitListener} * and {@link RabbitHandler} annotations. diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/MessagingMessageListenerAdapter.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/MessagingMessageListenerAdapter.java index baba04c4..97a2d57d 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/MessagingMessageListenerAdapter.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/MessagingMessageListenerAdapter.java @@ -21,6 +21,7 @@ import java.lang.reflect.ParameterizedType; import java.lang.reflect.Type; import java.lang.reflect.WildcardType; import java.util.List; +import java.util.Optional; import org.springframework.amqp.core.MessageProperties; import org.springframework.amqp.rabbit.listener.api.RabbitListenerErrorHandler; @@ -404,7 +405,15 @@ public class MessagingMessageListenerAdapter extends AbstractAdaptableMessageLis } } } + return checkOptional(genericParameterType); + } + protected Type checkOptional(Type genericParameterType) { + if (genericParameterType instanceof ParameterizedType + && ((ParameterizedType) genericParameterType).getRawType().equals(Optional.class)) { + + return ((ParameterizedType) genericParameterType).getActualTypeArguments()[0]; + } return genericParameterType; } diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/OptionalPayloadTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/OptionalPayloadTests.java new file mode 100644 index 00000000..1d5a4764 --- /dev/null +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/OptionalPayloadTests.java @@ -0,0 +1,142 @@ +/* + * Copyright 2022 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.rabbit.annotation; + +import static org.assertj.core.api.Assertions.assertThat; + +import java.util.ArrayList; +import java.util.List; +import java.util.Optional; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +import org.junit.jupiter.api.Test; + +import org.springframework.amqp.AmqpException; +import org.springframework.amqp.core.MessageBuilder; +import org.springframework.amqp.core.MessagePropertiesBuilder; +import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory; +import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; +import org.springframework.amqp.rabbit.connection.ConnectionFactory; +import org.springframework.amqp.rabbit.core.RabbitTemplate; +import org.springframework.amqp.rabbit.junit.RabbitAvailable; +import org.springframework.amqp.rabbit.junit.RabbitAvailableCondition; +import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.messaging.handler.annotation.Payload; +import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.ObjectMapper; + +/** + * @author Gary Russell + * @since 2.8 + * + */ +@SpringJUnitConfig +@RabbitAvailable(queues = { "op.1", "op.2" }) +public class OptionalPayloadTests { + + @Test + void optionals(@Autowired RabbitTemplate template, @Autowired Listener listener) + throws JsonProcessingException, AmqpException, InterruptedException { + + ObjectMapper objectMapper = new ObjectMapper(); + template.send("op.1", MessageBuilder.withBody(objectMapper.writeValueAsBytes("foo")) + .andProperties(MessagePropertiesBuilder.newInstance() + .setContentType("application/json") + .build()) + .build()); + template.send("op.1", MessageBuilder.withBody(objectMapper.writeValueAsBytes(null)) + .andProperties(MessagePropertiesBuilder.newInstance() + .setContentType("application/json") + .build()) + .build()); + template.send("op.2", MessageBuilder.withBody(objectMapper.writeValueAsBytes("bar")) + .andProperties(MessagePropertiesBuilder.newInstance() + .setContentType("application/json") + .build()) + .build()); + template.send("op.2", MessageBuilder.withBody(objectMapper.writeValueAsBytes(null)) + .andProperties(MessagePropertiesBuilder.newInstance() + .setContentType("application/json") + .build()) + .build()); + assertThat(listener.latch.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(listener.deOptionaled).containsExactlyInAnyOrder("foo", null, "bar", "baz"); + } + + @Configuration + @EnableRabbit + public static class Config { + + @Bean + RabbitTemplate template() { + return new RabbitTemplate(rabbitConnectionFactory()); + } + + @Bean + SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory() { + SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory(); + factory.setConnectionFactory(rabbitConnectionFactory()); + factory.setMessageConverter(converter()); + return factory; + } + + @Bean + ConnectionFactory rabbitConnectionFactory() { + return new CachingConnectionFactory(RabbitAvailableCondition.getBrokerRunning().getConnectionFactory()); + } + + @Bean + Jackson2JsonMessageConverter converter() { + Jackson2JsonMessageConverter converter = new Jackson2JsonMessageConverter(); + converter.setNullAsOptionalEmpty(true); + return converter; + } + + @Bean + Listener listener() { + return new Listener(); + } + + } + + static class Listener { + + final CountDownLatch latch = new CountDownLatch(4); + + List deOptionaled = new ArrayList<>(); + + @RabbitListener(queues = "op.1") + void listen(@Payload(required = false) String payload) { + this.deOptionaled.add(payload); + this.latch.countDown(); + } + + @RabbitListener(queues = "op.2") + void listen(Optional optional) { + this.deOptionaled.add(optional.orElse("baz")); + this.latch.countDown(); + } + + } + +} diff --git a/src/reference/asciidoc/amqp.adoc b/src/reference/asciidoc/amqp.adoc index 6846bd4a..50328bf2 100644 --- a/src/reference/asciidoc/amqp.adoc +++ b/src/reference/asciidoc/amqp.adoc @@ -4078,6 +4078,38 @@ converter to determine the type. IMPORTANT: Starting with version 1.6.11, `Jackson2JsonMessageConverter` and, therefore, `DefaultJackson2JavaTypeMapper` (`DefaultClassMapper`) provide the `trustedPackages` option to overcome https://pivotal.io/security/cve-2017-4995[Serialization Gadgets] vulnerability. By default and for backward compatibility, the `Jackson2JsonMessageConverter` trusts all packages -- that is, it uses `*` for the option. +Starting with version 2.4.7, the converter can be configured to return `Optional.empty()` if Jackson returns `null` after deserializing the message body. +This facilitates `@RabbitListener` s to receive null payloads, in two ways: + +==== +[source, java] +---- +@RabbitListener(queues = "op.1") +void listen(@Payload(required = false) Thing payload) { + handleOptional(payload); // payload might be null +} + +@RabbitListener(queues = "op.2") +void listen(Optional optional) { + handleOptional(optional.orElse(this.emptyThing)); +} +---- +==== + +To enable this feature, set `setNullAsOptionalEmpty` to `true`; when `false` (default), the converter falls back to the raw message body (`byte[]`). + +==== +[source, java] +---- +@Bean +Jackson2JsonMessageConverter converter() { + Jackson2JsonMessageConverter converter = new Jackson2JsonMessageConverter(); + converter.setNullAsOptionalEmpty(true); + return converter; +} +---- +==== + [[jackson-abstract]] ====== Deserializing Abstract Classes diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index d067dce3..4b8bb210 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -14,6 +14,9 @@ See <> for more information. Async reply types now include `CompleteableFuture` See <> for more information. +`MessageConverter` s can now return `Optional.empty()` for a null value; this is currently implemented by the `Jackson2JsonMessageConverter`. +See <> for more information. + ==== `RabbitAdmin` Changes A new property `recoverManualDeclarations` allows recovery of manually declared queues/exchanges/bindings.