Spring/spring-amqp
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-1491: Support Optional/null Payloads

Browse Source 
Resolves https://github.com/spring-projects/spring-amqp/issues/1491

**cherry-pick to 2.4.x**

* Improve connection factory bean in test.
# Conflicts:
#	src/reference/asciidoc/whats-new.adoc
This commit is contained in:
Artem Bilan
2022-08-09 15:47:55 -04:00
parent 67bfec93f1
commit 3d3dfa5d70
6 changed files with 271 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
spring-amqp/src/main/java/org/springframework/amqp/support/converter/AbstractJackson2MessageConverter.java
View File

@@ -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;
}

72
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/annotation/RabbitListenerAnnotationBeanPostProcessor.java
View File

@@ -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<HandlerMethodArgumentResolver> customArgumentsResolver =
new ArrayList<>(RabbitListenerAnnotationBeanPostProcessor.this.registrar.getCustomMethodArgumentResolvers());
List<HandlerMethodArgumentResolver> 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<ObjectError> 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.

9
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/MessagingMessageListenerAdapter.java
View File

@@ -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;
}

142
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/OptionalPayloadTests.java Normal file
View File

@@ -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<String> 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<String> optional) {
this.deOptionaled.add(optional.orElse("baz"));
this.latch.countDown();
}
}
}

32
src/reference/asciidoc/amqp.adoc
View File

@@ -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<Thing> 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

3
src/reference/asciidoc/whats-new.adoc
View File

@@ -14,6 +14,9 @@ See <<async-annotation-driven-enable-signature>> for more information.
Async reply types now include `CompleteableFuture`
See <<async-returns>> for more information.
`MessageConverter` s can now return `Optional.empty()` for a null value; this is currently implemented by the `Jackson2JsonMessageConverter`.
See <<Jackson2JsonMessageConverter-from-message>> for more information.
==== `RabbitAdmin` Changes
A new property `recoverManualDeclarations` allows recovery of manually declared queues/exchanges/bindings.