diff --git a/spring-amqp/src/main/java/org/springframework/amqp/support/converter/AbstractMessageConverter.java b/spring-amqp/src/main/java/org/springframework/amqp/support/converter/AbstractMessageConverter.java index dcde2077..3a648a9e 100644 --- a/spring-amqp/src/main/java/org/springframework/amqp/support/converter/AbstractMessageConverter.java +++ b/spring-amqp/src/main/java/org/springframework/amqp/support/converter/AbstractMessageConverter.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-2018 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. @@ -16,14 +16,18 @@ package org.springframework.amqp.support.converter; +import java.lang.reflect.Type; import java.util.UUID; import org.springframework.amqp.core.Message; import org.springframework.amqp.core.MessageProperties; +import org.springframework.lang.Nullable; /** * Convenient base class for {@link MessageConverter} implementations. + * * @author Dave Syer + * @author Gary Russell * */ public abstract class AbstractMessageConverter implements MessageConverter { @@ -50,10 +54,18 @@ public abstract class AbstractMessageConverter implements MessageConverter { @Override public final Message toMessage(Object object, MessageProperties messageProperties) throws MessageConversionException { + + return toMessage(object, messageProperties, null); + } + + @Override + public final Message toMessage(Object object, MessageProperties messageProperties, @Nullable Type genericType) + throws MessageConversionException { + if (messageProperties == null) { messageProperties = new MessageProperties(); } - Message message = createMessage(object, messageProperties); + Message message = createMessage(object, messageProperties, genericType); messageProperties = message.getMessageProperties(); if (this.createMessageIds && messageProperties.getMessageId() == null) { messageProperties.setMessageId(UUID.randomUUID().toString()); @@ -64,14 +76,23 @@ public abstract class AbstractMessageConverter implements MessageConverter { /** * Crate a message from the payload object and message properties provided. The message id will be added to the * properties if necessary later. - * * @param object the payload * @param messageProperties the message properties (headers) + * @param genericType the type to convert from - used to populate type headers. * @return a message + * @since 2.1 + */ + protected Message createMessage(Object object, MessageProperties messageProperties, @Nullable Type genericType) { + return createMessage(object, messageProperties); + } + + /** + * Crate a message from the payload object and message properties provided. The message id will be added to the + * properties if necessary later. + * @param object the payload. + * @param messageProperties the message properties (headers). + * @return a message. */ protected abstract Message createMessage(Object object, MessageProperties messageProperties); - @Override - public abstract Object fromMessage(Message message) throws MessageConversionException; - } diff --git a/spring-amqp/src/main/java/org/springframework/amqp/support/converter/DefaultJackson2JavaTypeMapper.java b/spring-amqp/src/main/java/org/springframework/amqp/support/converter/DefaultJackson2JavaTypeMapper.java index ba8df42a..5cb0ca88 100644 --- a/spring-amqp/src/main/java/org/springframework/amqp/support/converter/DefaultJackson2JavaTypeMapper.java +++ b/spring-amqp/src/main/java/org/springframework/amqp/support/converter/DefaultJackson2JavaTypeMapper.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2017 the original author or authors. + * Copyright 2002-2018 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. @@ -30,6 +30,7 @@ import com.fasterxml.jackson.databind.type.TypeFactory; /** * Jackson 2 type mapper. + * * @author Mark Pollack * @author Sam Nelson * @author Andreas Asplund @@ -104,6 +105,11 @@ public class DefaultJackson2JavaTypeMapper extends AbstractJavaTypeMapper } } + @Override + public void addTrustedPackages(String... packages) { + setTrustedPackages(packages); + } + @Override public JavaType toJavaType(MessageProperties properties) { boolean hasInferredTypeHeader = hasInferredTypeHeader(properties); diff --git a/spring-amqp/src/main/java/org/springframework/amqp/support/converter/Jackson2JavaTypeMapper.java b/spring-amqp/src/main/java/org/springframework/amqp/support/converter/Jackson2JavaTypeMapper.java index c18b1d30..2011c652 100644 --- a/spring-amqp/src/main/java/org/springframework/amqp/support/converter/Jackson2JavaTypeMapper.java +++ b/spring-amqp/src/main/java/org/springframework/amqp/support/converter/Jackson2JavaTypeMapper.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2017 the original author or authors. + * Copyright 2002-2018 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. @@ -35,15 +35,40 @@ public interface Jackson2JavaTypeMapper extends ClassMapper { /** * The precedence for type conversion - inferred from the method parameter or message * headers. Only applies if both exist. + * @since 1.6 */ enum TypePrecedence { INFERRED, TYPE_ID } + /** + * Set the message properties according to the type. + * @param javaType the type. + * @param properties the properties. + */ void fromJavaType(JavaType javaType, MessageProperties properties); + /** + * Determine the type from the message properties. + * @param properties the properties. + * @return the type. + */ JavaType toJavaType(MessageProperties properties); + /** + * Get the type precedence. + * @return the precedence. + * @since 1.6 + */ TypePrecedence getTypePrecedence(); + /** + * Add trusted packages. + * @param packages the packages. + * @since 2.1 + */ + default void addTrustedPackages(String... packages) { + // no op + } + } diff --git a/spring-amqp/src/main/java/org/springframework/amqp/support/converter/Jackson2JsonMessageConverter.java b/spring-amqp/src/main/java/org/springframework/amqp/support/converter/Jackson2JsonMessageConverter.java index e5624dc5..fbce94dd 100644 --- a/spring-amqp/src/main/java/org/springframework/amqp/support/converter/Jackson2JsonMessageConverter.java +++ b/spring-amqp/src/main/java/org/springframework/amqp/support/converter/Jackson2JsonMessageConverter.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2017 the original author or authors. + * Copyright 2002-2018 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,6 +17,7 @@ package org.springframework.amqp.support.converter; import java.io.IOException; +import java.lang.reflect.Type; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -224,6 +225,14 @@ public class Jackson2JsonMessageConverter extends AbstractJsonMessageConverter i @Override protected Message createMessage(Object objectToConvert, MessageProperties messageProperties) throws MessageConversionException { + + return createMessage(objectToConvert, messageProperties, null); + } + + @Override + protected Message createMessage(Object objectToConvert, MessageProperties messageProperties, Type genericType) + throws MessageConversionException { + byte[] bytes; try { String jsonString = this.jsonObjectMapper @@ -231,22 +240,19 @@ public class Jackson2JsonMessageConverter extends AbstractJsonMessageConverter i bytes = jsonString.getBytes(getDefaultCharset()); } catch (IOException e) { - throw new MessageConversionException( - "Failed to convert Message content", e); + throw new MessageConversionException("Failed to convert Message content", e); } messageProperties.setContentType(MessageProperties.CONTENT_TYPE_JSON); messageProperties.setContentEncoding(getDefaultCharset()); messageProperties.setContentLength(bytes.length); if (getClassMapper() == null) { - getJavaTypeMapper().fromJavaType(this.jsonObjectMapper.constructType(objectToConvert.getClass()), - messageProperties); - + getJavaTypeMapper().fromJavaType(this.jsonObjectMapper.constructType( + genericType == null ? objectToConvert.getClass() : genericType), messageProperties); } else { getClassMapper().fromClass(objectToConvert.getClass(), messageProperties); - } return new Message(bytes, messageProperties); diff --git a/spring-amqp/src/main/java/org/springframework/amqp/support/converter/MessageConverter.java b/spring-amqp/src/main/java/org/springframework/amqp/support/converter/MessageConverter.java index 1a6a335f..d5b07ec4 100644 --- a/spring-amqp/src/main/java/org/springframework/amqp/support/converter/MessageConverter.java +++ b/spring-amqp/src/main/java/org/springframework/amqp/support/converter/MessageConverter.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2017 the original author or authors. + * Copyright 2002-2018 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. @@ -16,13 +16,18 @@ package org.springframework.amqp.support.converter; +import java.lang.reflect.Type; + import org.springframework.amqp.core.Message; import org.springframework.amqp.core.MessageProperties; +import org.springframework.lang.Nullable; /** * Message converter interface. + * * @author Mark Fisher * @author Mark Pollack + * @author Gary Russell */ public interface MessageConverter { @@ -35,6 +40,22 @@ public interface MessageConverter { */ Message toMessage(Object object, MessageProperties messageProperties) throws MessageConversionException; + /** + * Convert a Java object to a Message. + * The default implementation calls {@link #toMessage(Object, MessageProperties)}. + * @param object the object to convert + * @param messageProperties The message properties. + * @param genericType the type to use to populate type headers. + * @return the Message + * @throws MessageConversionException in case of conversion failure + * @since 2.1 + */ + default Message toMessage(Object object, MessageProperties messageProperties, @Nullable Type genericType) + throws MessageConversionException { + + return toMessage(object, messageProperties); + } + /** * Convert from a Message to a Java object. * @param message the message to convert diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/MethodRabbitListenerEndpoint.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/MethodRabbitListenerEndpoint.java index 8d3798c8..25812793 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/MethodRabbitListenerEndpoint.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/MethodRabbitListenerEndpoint.java @@ -115,7 +115,7 @@ public class MethodRabbitListenerEndpoint extends AbstractRabbitListenerEndpoint Assert.state(this.messageHandlerMethodFactory != null, "Could not create message listener - MessageHandlerMethodFactory not set"); MessagingMessageListenerAdapter messageListener = createMessageListenerInstance(); - messageListener.setHandlerMethod(configureListenerAdapter(messageListener)); + messageListener.setHandlerAdapter(configureListenerAdapter(messageListener)); String replyToAddress = getDefaultReplyToAddress(); if (replyToAddress != null) { messageListener.setResponseAddress(replyToAddress); diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/AbstractAdaptableMessageListener.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/AbstractAdaptableMessageListener.java index 080a8d0c..057856fd 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/AbstractAdaptableMessageListener.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/AbstractAdaptableMessageListener.java @@ -16,6 +16,7 @@ package org.springframework.amqp.rabbit.listener.adapter; +import java.lang.reflect.Type; import java.util.Arrays; import org.apache.commons.logging.Log; @@ -179,8 +180,7 @@ public abstract class AbstractAdaptableMessageListener implements MessageListene * Set a post processor to process the reply immediately before * {@code Channel#basicPublish()}. Often used to compress the data. * @param replyPostProcessor the reply post processor. - * @deprecated in favor of - * {@link #setBeforeSendReplyPostProcessors(MessagePostProcessor...)}. + * @deprecated in favor of {@link #setBeforeSendReplyPostProcessors(MessagePostProcessor...)}. */ @Deprecated public void setReplyPostProcessor(MessagePostProcessor replyPostProcessor) { @@ -271,13 +271,12 @@ public abstract class AbstractAdaptableMessageListener implements MessageListene * @param resultArg the result object to handle (never null) * @param request the original request message * @param channel the Rabbit channel to operate on (may be null) - * @throws Exception if thrown by Rabbit API methods * @see #buildMessage * @see #postProcessResponse - * @see #getReplyToAddress(Message, Object, Object) + * @see #getReplyToAddress(Message, Object, InvocationResult) * @see #sendResponse */ - protected void handleResult(Object resultArg, Message request, Channel channel) throws Exception { + protected void handleResult(InvocationResult resultArg, Message request, Channel channel) { handleResult(resultArg, request, channel, null); } @@ -289,21 +288,19 @@ public abstract class AbstractAdaptableMessageListener implements MessageListene * @param channel the Rabbit channel to operate on (may be null) * @param source the source data for the method invocation - e.g. * {@code o.s.messaging.Message}; may be null - * @throws Exception if thrown by Rabbit API methods * @see #buildMessage * @see #postProcessResponse - * @see #getReplyToAddress(Message, Object, Object) + * @see #getReplyToAddress(Message, Object, InvocationResult) * @see #sendResponse */ - protected void handleResult(Object resultArg, Message request, Channel channel, Object source) throws Exception { + protected void handleResult(InvocationResult resultArg, Message request, Channel channel, Object source) { if (channel != null) { if (this.logger.isDebugEnabled()) { this.logger.debug("Listener method returned result [" + resultArg + "] - generating response message for it"); } try { - Object result = resultArg instanceof ResultHolder ? ((ResultHolder) resultArg).result : resultArg; - Message response = buildMessage(channel, result); + Message response = buildMessage(channel, resultArg.getReturnValue(), resultArg.getReturnType()); postProcessResponse(request, response); Address replyTo = getReplyToAddress(request, source, resultArg); sendResponse(channel, replyTo, response); @@ -324,16 +321,17 @@ public abstract class AbstractAdaptableMessageListener implements MessageListene /** * Build a Rabbit message to be sent as response based on the given result object. - * @param channel the Rabbit Channel to operate on - * @param result the content of the message, as returned from the listener method - * @return the Rabbit Message (never null) - * @throws Exception if thrown by Rabbit API methods + * @param channel the Rabbit Channel to operate on. + * @param result the content of the message, as returned from the listener method. + * @param genericType the generic type to populate type headers. + * @return the Rabbit Message (never null). + * @throws Exception if thrown by Rabbit API methods. * @see #setMessageConverter */ - protected Message buildMessage(Channel channel, Object result) throws Exception { + protected Message buildMessage(Channel channel, Object result, Type genericType) throws Exception { MessageConverter converter = getMessageConverter(); if (converter != null && !(result instanceof Message)) { - return converter.toMessage(result, new MessageProperties()); + return converter.toMessage(result, new MessageProperties(), genericType); } else { if (!(result instanceof Message)) { @@ -351,9 +349,8 @@ public abstract class AbstractAdaptableMessageListener implements MessageListene * otherwise to the request message id. * @param request the original incoming Rabbit message * @param response the outgoing Rabbit message about to be sent - * @throws Exception if thrown by Rabbit API methods */ - protected void postProcessResponse(Message request, Message response) throws Exception { + protected void postProcessResponse(Message request, Message response) { String correlation = request.getMessageProperties().getCorrelationId(); if (correlation == null) { @@ -376,24 +373,23 @@ public abstract class AbstractAdaptableMessageListener implements MessageListene * @param source the source data (e.g. {@code o.s.messaging.Message}). * @param result the result. * @return the reply-to Address (never null) - * @throws Exception if thrown by Rabbit API methods * @throws org.springframework.amqp.AmqpException if no {@link Address} can be determined * @see #setResponseAddress(String) * @see #setResponseRoutingKey(String) * @see org.springframework.amqp.core.Message#getMessageProperties() * @see org.springframework.amqp.core.MessageProperties#getReplyTo() */ - protected Address getReplyToAddress(Message request, Object source, Object result) throws Exception { + protected Address getReplyToAddress(Message request, Object source, InvocationResult result) { Address replyTo = request.getMessageProperties().getReplyToAddress(); if (replyTo == null) { if (this.responseAddress == null && this.responseExchange != null) { this.responseAddress = new Address(this.responseExchange, this.responseRoutingKey); } - if (result instanceof ResultHolder) { - replyTo = evaluateReplyTo(request, source, result, ((ResultHolder) result).sendTo); + if (result.getSendTo() != null) { + replyTo = evaluateReplyTo(request, source, result.getReturnValue(), result.getSendTo()); } else if (this.responseExpression != null) { - replyTo = evaluateReplyTo(request, source, result, this.responseExpression); + replyTo = evaluateReplyTo(request, source, result.getReturnValue(), this.responseExpression); } else if (this.responseAddress == null) { throw new AmqpException( @@ -427,10 +423,9 @@ public abstract class AbstractAdaptableMessageListener implements MessageListene * @param channel the Rabbit channel to operate on * @param replyTo the Rabbit ReplyTo string to use when sending. Currently interpreted to be the routing key. * @param messageIn the Rabbit message to send - * @throws Exception if thrown by Rabbit API methods * @see #postProcessResponse(Message, Message) */ - protected void sendResponse(Channel channel, Address replyTo, Message messageIn) throws Exception { + protected void sendResponse(Channel channel, Address replyTo, Message messageIn) { Message message = messageIn; if (this.beforeSendReplyPostProcessors != null) { for (MessagePostProcessor postProcessor : this.beforeSendReplyPostProcessors) { @@ -458,30 +453,8 @@ public abstract class AbstractAdaptableMessageListener implements MessageListene * * @param channel The channel. * @param response the outgoing Rabbit message about to be sent - * @throws Exception if thrown by Rabbit API methods */ - protected void postProcessChannel(Channel channel, Message response) throws Exception { - } - - /** - * Result holder. - */ - public static final class ResultHolder { - - private final Object result; - - private final Expression sendTo; - - public ResultHolder(Object result, Expression sendTo) { - this.result = result; - this.sendTo = sendTo; - } - - @Override - public String toString() { - return this.result.toString(); - } - + protected void postProcessChannel(Channel channel, Message response) { } /** @@ -495,7 +468,7 @@ public abstract class AbstractAdaptableMessageListener implements MessageListene private final Object result; - public ReplyExpressionRoot(Message request, Object source, Object result) { + protected ReplyExpressionRoot(Message request, Object source, Object result) { this.request = request; this.source = source; this.result = result; diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/DelegatingInvocableHandler.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/DelegatingInvocableHandler.java index 9e2d3506..63437fe6 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/DelegatingInvocableHandler.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/DelegatingInvocableHandler.java @@ -51,6 +51,7 @@ import org.springframework.util.Assert; * Matches must be unambiguous. * * @author Gary Russell + * * @since 1.5 * */ @@ -62,13 +63,11 @@ public class DelegatingInvocableHandler { private final List handlers; - private final ConcurrentMap, InvocableHandlerMethod> cachedHandlers = - new ConcurrentHashMap, InvocableHandlerMethod>(); + private final ConcurrentMap, InvocableHandlerMethod> cachedHandlers = new ConcurrentHashMap<>(); private final InvocableHandlerMethod defaultHandler; - private final Map handlerSendTo = - new HashMap(); + private final Map handlerSendTo = new HashMap<>(); private final Object bean; @@ -85,6 +84,7 @@ public class DelegatingInvocableHandler { */ public DelegatingInvocableHandler(List handlers, Object bean, BeanExpressionResolver beanExpressionResolver, BeanExpressionContext beanExpressionContext) { + this(handlers, null, bean, beanExpressionResolver, beanExpressionContext); } @@ -100,7 +100,8 @@ public class DelegatingInvocableHandler { public DelegatingInvocableHandler(List handlers, @Nullable InvocableHandlerMethod defaultHandler, Object bean, BeanExpressionResolver beanExpressionResolver, BeanExpressionContext beanExpressionContext) { - this.handlers = new ArrayList(handlers); + + this.handlers = new ArrayList<>(handlers); this.defaultHandler = defaultHandler; this.bean = bean; this.resolver = beanExpressionResolver; @@ -122,17 +123,17 @@ public class DelegatingInvocableHandler { * @throws Exception raised if no suitable argument resolver can be found, * or the method raised an exception. */ - public Object invoke(Message message, Object... providedArgs) throws Exception { + public InvocationResult invoke(Message message, Object... providedArgs) throws Exception { Class payloadClass = message.getPayload().getClass(); InvocableHandlerMethod handler = getHandlerForPayload(payloadClass); Object result = handler.invoke(message, providedArgs); if (message.getHeaders().get(AmqpHeaders.REPLY_TO) == null) { Expression replyTo = this.handlerSendTo.get(handler); if (replyTo != null) { - result = new AbstractAdaptableMessageListener.ResultHolder(result, replyTo); + return new InvocationResult(result, replyTo, handler.getMethod().getGenericReturnType()); } } - return result; + return new InvocationResult(result, null, handler.getMethod().getGenericReturnType()); } /** diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/HandlerAdapter.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/HandlerAdapter.java index 6dc6aee6..a98c6902 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/HandlerAdapter.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/HandlerAdapter.java @@ -46,9 +46,10 @@ public class HandlerAdapter { this.delegatingHandler = delegatingHandler; } - public Object invoke(Message message, Object... providedArgs) throws Exception { + public InvocationResult invoke(Message message, Object... providedArgs) throws Exception { if (this.invokerHandlerMethod != null) { - return this.invokerHandlerMethod.invoke(message, providedArgs); + return new InvocationResult(this.invokerHandlerMethod.invoke(message, providedArgs), + null, this.invokerHandlerMethod.getMethod().getGenericReturnType()); } else if (this.delegatingHandler.hasDefaultHandler()) { // Needed to avoid returning raw Message which matches Object diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/InvocationResult.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/InvocationResult.java new file mode 100644 index 00000000..c5947c4c --- /dev/null +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/InvocationResult.java @@ -0,0 +1,64 @@ +/* + * Copyright 2018 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 + * + * http://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.listener.adapter; + +import java.lang.reflect.Type; + +import org.springframework.expression.Expression; + +/** + * The result of a listener method invocation. + * + * @author Gary Russell + * + * @since 2.1 + */ +public final class InvocationResult { + + private final Object returnValue; + + private final Expression sendTo; + + private final Type returnType; + + public InvocationResult(Object result, Expression sendTo, Type returnType) { + this.returnValue = result; + this.sendTo = sendTo; + this.returnType = returnType; + } + + public Object getReturnValue() { + return this.returnValue; + } + + public Expression getSendTo() { + return this.sendTo; + } + + + public Type getReturnType() { + return this.returnType; + } + + @Override + public String toString() { + return "InvocationResult [returnValue=" + this.returnValue + + (this.sendTo != null ? ", sendTo=" + this.sendTo : "") + + ", returnType=" + this.returnType + "]"; + } + +} diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/MessageListenerAdapter.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/MessageListenerAdapter.java index d1217509..32bd96d5 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/MessageListenerAdapter.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/MessageListenerAdapter.java @@ -297,7 +297,7 @@ public class MessageListenerAdapter extends AbstractAdaptableMessageListener { Object[] listenerArguments = buildListenerArguments(convertedMessage); Object result = invokeListenerMethod(methodName, listenerArguments, message); if (result != null) { - handleResult(result, message, channel); + handleResult(new InvocationResult(result, null, null), message, channel); } else { logger.trace("No result object given - no result to handle"); 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 f9cc55a0..531377fa 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 @@ -58,7 +58,7 @@ import com.rabbitmq.client.Channel; */ public class MessagingMessageListenerAdapter extends AbstractAdaptableMessageListener { - private HandlerAdapter handlerMethod; + private HandlerAdapter handlerAdapter; private final MessagingMessageConverterAdapter messagingMessageConverter; @@ -84,10 +84,10 @@ public class MessagingMessageListenerAdapter extends AbstractAdaptableMessageLis /** * Set the {@link HandlerAdapter} to use to invoke the method * processing an incoming {@link org.springframework.amqp.core.Message}. - * @param handlerMethod {@link HandlerAdapter} instance. + * @param handlerAdapter {@link HandlerAdapter} instance. */ - public void setHandlerMethod(HandlerAdapter handlerMethod) { - this.handlerMethod = handlerMethod; + public void setHandlerAdapter(HandlerAdapter handlerAdapter) { + this.handlerAdapter = handlerAdapter; } /** @@ -116,9 +116,10 @@ public class MessagingMessageListenerAdapter extends AbstractAdaptableMessageLis if (logger.isDebugEnabled()) { logger.debug("Processing [" + message + "]"); } + InvocationResult result = null; try { - Object result = invokeHandler(amqpMessage, channel, message); - if (result != null) { + result = invokeHandler(amqpMessage, channel, message); + if (result.getReturnValue() != null) { handleResult(result, amqpMessage, channel, message); } else { @@ -128,9 +129,9 @@ public class MessagingMessageListenerAdapter extends AbstractAdaptableMessageLis catch (ListenerExecutionFailedException e) { if (this.errorHandler != null) { try { - Object result = this.errorHandler.handleError(amqpMessage, message, e); - if (result != null) { - handleResult(result, amqpMessage, channel, message); + Object errorResult = this.errorHandler.handleError(amqpMessage, message, e); + if (errorResult != null) { + handleResult(new InvocationResult(errorResult, null, null), amqpMessage, channel, message); } else { logger.trace("Error handler returned no result"); @@ -152,10 +153,11 @@ public class MessagingMessageListenerAdapter extends AbstractAdaptableMessageLis throw exceptionToThrow; } try { - handleResult(new RemoteInvocationResult(throwableToReturn), amqpMessage, channel, message); + handleResult(new InvocationResult(new RemoteInvocationResult(throwableToReturn), null, null), + amqpMessage, channel, message); } catch (ReplyFailureException rfe) { - if (void.class.equals(this.handlerMethod.getReturnType(message.getPayload()))) { + if (void.class.equals(this.handlerAdapter.getReturnType(message.getPayload()))) { throw exceptionToThrow; } else { @@ -176,10 +178,10 @@ public class MessagingMessageListenerAdapter extends AbstractAdaptableMessageLis * @param message the messaging message. * @return the result of invoking the handler. */ - private Object invokeHandler(org.springframework.amqp.core.Message amqpMessage, Channel channel, + private InvocationResult invokeHandler(org.springframework.amqp.core.Message amqpMessage, Channel channel, Message message) { try { - return this.handlerMethod.invoke(message, amqpMessage, channel); + return this.handlerAdapter.invoke(message, amqpMessage, channel); } catch (MessagingException ex) { throw new ListenerExecutionFailedException(createMessagingErrorMessage("Listener method could not " + @@ -187,34 +189,36 @@ public class MessagingMessageListenerAdapter extends AbstractAdaptableMessageLis } catch (Exception ex) { throw new ListenerExecutionFailedException("Listener method '" + - this.handlerMethod.getMethodAsString(message.getPayload()) + "' threw exception", ex, amqpMessage); + this.handlerAdapter.getMethodAsString(message.getPayload()) + "' threw exception", ex, amqpMessage); } } private String createMessagingErrorMessage(String description, Object payload) { return description + "\n" + "Endpoint handler details:\n" - + "Method [" + this.handlerMethod.getMethodAsString(payload) + "]\n" - + "Bean [" + this.handlerMethod.getBean() + "]"; + + "Method [" + this.handlerAdapter.getMethodAsString(payload) + "]\n" + + "Bean [" + this.handlerAdapter.getBean() + "]"; } /** * Build a Rabbit message to be sent as response based on the given result object. * @param channel the Rabbit Channel to operate on * @param result the content of the message, as returned from the listener method + * @param genericType the generic type of the result. * @return the Rabbit Message (never null) * @throws Exception if thrown by Rabbit API methods * @see #setMessageConverter */ @Override - protected org.springframework.amqp.core.Message buildMessage(Channel channel, Object result) throws Exception { + protected org.springframework.amqp.core.Message buildMessage(Channel channel, Object result, Type genericType) + throws Exception { MessageConverter converter = getMessageConverter(); if (converter != null && !(result instanceof org.springframework.amqp.core.Message)) { if (result instanceof org.springframework.messaging.Message) { return this.messagingMessageConverter.toMessage(result, new MessageProperties()); } else { - return converter.toMessage(result, new MessageProperties()); + return converter.toMessage(result, new MessageProperties(), genericType); } } else { diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/EnableRabbitIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/EnableRabbitIntegrationTests.java index 4b99d553..8e6d35fc 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/EnableRabbitIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/EnableRabbitIntegrationTests.java @@ -41,6 +41,7 @@ import java.lang.annotation.Target; import java.util.ArrayList; import java.util.Arrays; import java.util.Collection; +import java.util.Collections; import java.util.Date; import java.util.HashMap; import java.util.List; @@ -160,7 +161,7 @@ public class EnableRabbitIntegrationTests { "test.converted.args2", "test.converted.message", "test.notconverted.message", "test.notconverted.channel", "test.notconverted.messagechannel", "test.notconverted.messagingmessage", "test.converted.foomessage", "test.notconverted.messagingmessagenotgeneric", "test.simple.direct", - "test.simple.direct2", + "test.simple.direct2", "test.generic.list", "test.generic.map", "amqp656dlq", "test.simple.declare", "test.return.exceptions", "test.pojo.errors", "test.pojo.errors2"); @Autowired @@ -669,6 +670,17 @@ public class EnableRabbitIntegrationTests { assertSame(value, typeCache.get(Foo1.class)); } + @Test + public void testGenericReturnTypes() { + Object returned = this.jsonRabbitTemplate.convertSendAndReceive("", "test.generic.list", new JsonObject("baz")); + assertThat(returned, instanceOf(List.class)); + assertThat(((List) returned).get(0), instanceOf(JsonObject.class)); + + returned = this.jsonRabbitTemplate.convertSendAndReceive("", "test.generic.map", new JsonObject("baz")); + assertThat(returned, instanceOf(Map.class)); + assertThat(((Map) returned).get("key"), instanceOf(JsonObject.class)); + } + interface TxService { @Transactional @@ -955,6 +967,43 @@ public class EnableRabbitIntegrationTests { throw new Exception("return this"); } + @RabbitListener(queues = "test.generic.list", containerFactory = "simpleJsonListenerContainerFactory") + public List genericList(JsonObject in) { + return Collections.singletonList(in); + } + + @RabbitListener(queues = "test.generic.map", containerFactory = "simpleJsonListenerContainerFactory") + public Map genericMap(JsonObject in) { + return Collections.singletonMap("key", in); + } + + } + + public static class JsonObject { + + private String bar; + + public JsonObject() { + super(); + } + + public JsonObject(String bar) { + this.bar = bar; + } + + public String getBar() { + return this.bar; + } + + public void setBar(String bar) { + this.bar = bar; + } + + @Override + public String toString() { + return "JsonObject [bar=" + this.bar + "]"; + } + } public static class Foo1 { @@ -1147,7 +1196,9 @@ public class EnableRabbitIntegrationTests { factory.setConnectionFactory(rabbitConnectionFactory()); factory.setErrorHandler(errorHandler()); factory.setConsumerTagStrategy(consumerTagStrategy()); - factory.setMessageConverter(new Jackson2JsonMessageConverter()); + Jackson2JsonMessageConverter messageConverter = new Jackson2JsonMessageConverter(); + messageConverter.getJavaTypeMapper().addTrustedPackages("*"); + factory.setMessageConverter(messageConverter); factory.setReceiveTimeout(10L); return factory; } diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/adapter/MessagingMessageListenerAdapterTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/adapter/MessagingMessageListenerAdapterTests.java index 1b422326..76004d87 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/adapter/MessagingMessageListenerAdapterTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/adapter/MessagingMessageListenerAdapterTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2014-2017 the original author or authors. + * Copyright 2014-2018 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. @@ -77,7 +77,7 @@ public class MessagingMessageListenerAdapterTests { Channel session = mock(Channel.class); MessagingMessageListenerAdapter listener = getSimpleInstance("echo", Message.class); - org.springframework.amqp.core.Message replyMessage = listener.buildMessage(session, result); + org.springframework.amqp.core.Message replyMessage = listener.buildMessage(session, result, null); assertNotNull("reply should never be null", replyMessage); assertEquals("Response", new String(replyMessage.getBody())); @@ -229,7 +229,7 @@ public class MessagingMessageListenerAdapterTests { protected MessagingMessageListenerAdapter createInstance(Method m, boolean returnExceptions) { MessagingMessageListenerAdapter adapter = new MessagingMessageListenerAdapter(null, m, returnExceptions, null); - adapter.setHandlerMethod(new HandlerAdapter(factory.createInvocableHandlerMethod(sample, m))); + adapter.setHandlerAdapter(new HandlerAdapter(factory.createInvocableHandlerMethod(sample, m))); return adapter; } @@ -246,7 +246,7 @@ public class MessagingMessageListenerAdapterTests { methods.add(this.factory.createInvocableHandlerMethod(sample, m1)); methods.add(this.factory.createInvocableHandlerMethod(sample, m2)); DelegatingInvocableHandler handler = new DelegatingInvocableHandler(methods, this.sample, null, null); - adapter.setHandlerMethod(new HandlerAdapter(handler)); + adapter.setHandlerAdapter(new HandlerAdapter(handler)); return adapter; }