diff --git a/spring-integration-core/src/main/java/org/springframework/integration/core/MessageProducer.java b/spring-integration-core/src/main/java/org/springframework/integration/core/MessageProducer.java index f39d6eff55..543fe7b9f6 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/core/MessageProducer.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/core/MessageProducer.java @@ -16,6 +16,7 @@ package org.springframework.integration.core; +import org.springframework.lang.Nullable; import org.springframework.messaging.MessageChannel; /** @@ -49,6 +50,7 @@ public interface MessageProducer { * @return the channel. * @since 4.3 */ + @Nullable MessageChannel getOutputChannel(); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageSourcePollingTemplate.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageSourcePollingTemplate.java index ca4dcfb0ce..f1adeda947 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageSourcePollingTemplate.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageSourcePollingTemplate.java @@ -1,5 +1,5 @@ /* - * Copyright 2018 the original author or authors. + * Copyright 2018-2019 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. @@ -20,9 +20,9 @@ import org.springframework.integration.StaticMessageHeaderAccessor; import org.springframework.integration.acks.AckUtils; import org.springframework.integration.acks.AcknowledgmentCallback; import org.springframework.integration.core.MessageSource; +import org.springframework.integration.support.utils.IntegrationUtils; import org.springframework.messaging.Message; import org.springframework.messaging.MessageHandler; -import org.springframework.messaging.MessageHandlingException; import org.springframework.util.Assert; /** @@ -56,7 +56,9 @@ public class MessageSourcePollingTemplate implements PollingOperations { } catch (Exception e) { AckUtils.autoNack(ackCallback); - throw new MessageHandlingException(message, e); + throw IntegrationUtils.wrapInHandlingExceptionIfNecessary(message, + () -> "error occurred during handling message in 'MessageSourcePollingTemplate' [" + + this + "]", e); } return true; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageHandler.java index 5cc1bb294a..5296b06d3d 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageHandler.java @@ -33,6 +33,7 @@ import org.springframework.integration.support.management.MessageHandlerMetrics; import org.springframework.integration.support.management.MetricsContext; import org.springframework.integration.support.management.Statistics; import org.springframework.integration.support.management.TrackableComponent; +import org.springframework.integration.support.management.metrics.MeterFacade; import org.springframework.integration.support.management.metrics.MetricsCaptor; import org.springframework.integration.support.management.metrics.SampleFacade; import org.springframework.integration.support.management.metrics.TimerFacade; @@ -338,7 +339,7 @@ public abstract class AbstractMessageHandler extends IntegrationObjectSupport @Override public void destroy() throws Exception { - this.timers.forEach(t -> t.remove()); + this.timers.forEach(MeterFacade::remove); } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProcessor.java index 6b50c61923..f8a669157e 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProcessor.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2018 the original author or authors. + * Copyright 2002-2019 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.integration.handler; import org.springframework.integration.util.AbstractExpressionEvaluator; +import org.springframework.lang.Nullable; import org.springframework.messaging.Message; /** @@ -28,6 +29,7 @@ import org.springframework.messaging.Message; */ public abstract class AbstractMessageProcessor extends AbstractExpressionEvaluator implements MessageProcessor { + @Nullable public abstract T processMessage(Message message); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProducingHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProducingHandler.java index acd795f80e..cff14db384 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProducingHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProducingHandler.java @@ -68,8 +68,10 @@ public abstract class AbstractMessageProducingHandler extends AbstractMessageHan private boolean async; + @Nullable private String outputChannelName; + @Nullable private MessageChannel outputChannel; private String[] notPropagatedHeaders; @@ -205,6 +207,7 @@ public abstract class AbstractMessageProducingHandler extends AbstractMessageHan } @Override + @Nullable public MessageChannel getOutputChannel() { if (this.outputChannelName != null) { this.outputChannel = getChannelResolver().resolveDestination(this.outputChannelName); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractReplyProducingMessageHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractReplyProducingMessageHandler.java index b5ff1d3187..23610517c5 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractReplyProducingMessageHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractReplyProducingMessageHandler.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2018 the original author or authors. + * Copyright 2002-2019 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. @@ -24,6 +24,7 @@ import org.aopalliance.aop.Advice; import org.springframework.aop.framework.ProxyFactory; import org.springframework.beans.factory.BeanClassLoaderAware; import org.springframework.integration.handler.advice.HandleMessageAdvice; +import org.springframework.lang.Nullable; import org.springframework.messaging.Message; import org.springframework.util.Assert; import org.springframework.util.ClassUtils; @@ -137,6 +138,7 @@ public abstract class AbstractReplyProducingMessageHandler extends AbstractMessa } } + @Nullable protected Object doInvokeAdvisedRequestHandler(Message message) { return this.advisedRequestHandler.handleRequestMessage(message); } @@ -149,6 +151,7 @@ public abstract class AbstractReplyProducingMessageHandler extends AbstractMessa * @param requestMessage The request message. * @return The result of handling the message, or {@code null}. */ + @Nullable protected abstract Object handleRequestMessage(Message requestMessage); @@ -166,6 +169,7 @@ public abstract class AbstractReplyProducingMessageHandler extends AbstractMessa */ public interface RequestHandler { + @Nullable Object handleRequestMessage(Message requestMessage); /** @@ -189,6 +193,7 @@ public abstract class AbstractReplyProducingMessageHandler extends AbstractMessa } @Override + @Nullable public Object handleRequestMessage(Message requestMessage) { return AbstractReplyProducingMessageHandler.this.handleRequestMessage(requestMessage); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractReplyProducingPostProcessingMessageHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractReplyProducingPostProcessingMessageHandler.java index 4e3c2ecf1b..7dc47949fa 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractReplyProducingPostProcessingMessageHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractReplyProducingPostProcessingMessageHandler.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-2019 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,10 +16,13 @@ package org.springframework.integration.handler; +import org.springframework.lang.Nullable; import org.springframework.messaging.Message; /** * @author Gary Russell + * @author Artem Bilan + * * @since 3.0 * */ @@ -42,6 +45,7 @@ public abstract class AbstractReplyProducingPostProcessingMessageHandler } @Override + @Nullable protected final Object handleRequestMessage(Message requestMessage) { Object result = this.doHandleRequestMessage(requestMessage); if (this.postProcessWithinAdvice || !this.hasAdviceChain()) { @@ -51,6 +55,7 @@ public abstract class AbstractReplyProducingPostProcessingMessageHandler } @Override + @Nullable protected final Object doInvokeAdvisedRequestHandler(Message message) { Object result = super.doInvokeAdvisedRequestHandler(message); if (!this.postProcessWithinAdvice) { @@ -59,6 +64,7 @@ public abstract class AbstractReplyProducingPostProcessingMessageHandler return result; } + @Nullable protected abstract Object doHandleRequestMessage(Message requestMessage); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/BeanNameMessageProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/BeanNameMessageProcessor.java index 77404ad292..ffec9212f2 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/BeanNameMessageProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/BeanNameMessageProcessor.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2017 the original author or authors. + * Copyright 2016-2019 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. @@ -19,8 +19,8 @@ package org.springframework.integration.handler; import org.springframework.beans.BeansException; import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.BeanFactoryAware; +import org.springframework.lang.Nullable; import org.springframework.messaging.Message; -import org.springframework.util.Assert; /** * An "artificial" {@link MessageProcessor} for lazy-load of target bean by its name. @@ -29,6 +29,7 @@ import org.springframework.util.Assert; * @param the expected {@link #processMessage} result type. * * @author Artem Bilan + * * @since 5.0 */ public class BeanNameMessageProcessor implements MessageProcessor, BeanFactoryAware { @@ -48,11 +49,11 @@ public class BeanNameMessageProcessor implements MessageProcessor, BeanFac @Override public void setBeanFactory(BeanFactory beanFactory) throws BeansException { - Assert.notNull(beanFactory, "'beanFactory' must not be null"); this.beanFactory = beanFactory; } @Override + @Nullable public T processMessage(Message message) { if (this.delegate == null) { Object target = this.beanFactory.getBean(this.beanName); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/DiscardingMessageHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/DiscardingMessageHandler.java index f0bf0ccba0..c0afebcf82 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/DiscardingMessageHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/DiscardingMessageHandler.java @@ -1,5 +1,5 @@ /* - * Copyright 2016 the original author or authors. + * Copyright 2016-2019 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,6 +16,7 @@ package org.springframework.integration.handler; +import org.springframework.lang.Nullable; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandler; @@ -32,6 +33,7 @@ public interface DiscardingMessageHandler extends MessageHandler { * Return the discard channel. * @return the channel. */ + @Nullable MessageChannel getDiscardChannel(); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/ExpressionCommandMessageProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/ExpressionCommandMessageProcessor.java index e1f5b48f06..8d4847f855 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/ExpressionCommandMessageProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/ExpressionCommandMessageProcessor.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-2019 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. @@ -31,7 +31,9 @@ import org.springframework.expression.MethodExecutor; import org.springframework.expression.MethodFilter; import org.springframework.expression.MethodResolver; import org.springframework.expression.spel.support.ReflectiveMethodResolver; +import org.springframework.lang.Nullable; import org.springframework.messaging.Message; +import org.springframework.util.CollectionUtils; /** * A MessageProcessor implementation that expects an Expression or expressionString @@ -40,6 +42,8 @@ import org.springframework.messaging.Message; * @author Dave Syer * @author Mark Fisher * @author Gary Russell + * @author Artem Bilan + * * @since 2.0 */ public class ExpressionCommandMessageProcessor extends AbstractMessageProcessor { @@ -47,26 +51,31 @@ public class ExpressionCommandMessageProcessor extends AbstractMessageProcessor< public ExpressionCommandMessageProcessor() { } - public ExpressionCommandMessageProcessor(MethodFilter methodFilter) { + public ExpressionCommandMessageProcessor(@Nullable MethodFilter methodFilter) { this(methodFilter, null); } - public ExpressionCommandMessageProcessor(MethodFilter methodFilter, BeanFactory beanFactory) { + public ExpressionCommandMessageProcessor(@Nullable MethodFilter methodFilter, @Nullable BeanFactory beanFactory) { if (beanFactory != null) { - this.setBeanFactory(beanFactory); + setBeanFactory(beanFactory); } if (methodFilter != null) { MethodResolver methodResolver = new ExpressionCommandMethodResolver(methodFilter); - this.getEvaluationContext(false).setMethodResolvers(Collections.singletonList(methodResolver)); + getEvaluationContext(false).setMethodResolvers(Collections.singletonList(methodResolver)); } } + @Override + public final void setBeanFactory(BeanFactory beanFactory) { + super.setBeanFactory(beanFactory); + } /** * Evaluates the Message payload expression as a command. * @throws IllegalArgumentException if the payload is not an Exception or String */ @Override + @Nullable public Object processMessage(Message message) { Object expression = message.getPayload(); if (expression instanceof Expression) { @@ -91,18 +100,17 @@ public class ExpressionCommandMessageProcessor extends AbstractMessageProcessor< @Override public MethodExecutor resolve(EvaluationContext context, - Object targetObject, String name, List argumentTypes) throws AccessException { - this.validateMethod(targetObject, name, (argumentTypes != null ? argumentTypes.size() : 0)); + Object targetObject, String name, List argumentTypes) + throws AccessException { + + validateMethod(targetObject, name, !CollectionUtils.isEmpty(argumentTypes) ? argumentTypes.size() : 0); return super.resolve(context, targetObject, name, argumentTypes); } private void validateMethod(Object targetObject, String name, int argumentCount) { - if (this.methodFilter == null) { - return; - } Class type = (targetObject instanceof Class ? (Class) targetObject : targetObject.getClass()); Method[] methods = type.getMethods(); - List candidates = new ArrayList(); + List candidates = new ArrayList<>(); for (Method method : methods) { if (method.getName().equals(name) && method.getParameterTypes().length == argumentCount) { candidates.add(method); @@ -111,10 +119,12 @@ public class ExpressionCommandMessageProcessor extends AbstractMessageProcessor< List supportedMethods = this.methodFilter.filter(candidates); if (supportedMethods.size() == 0) { String methodDescription = (candidates.size() > 0) ? candidates.get(0).toString() : name; - throw new EvaluationException("The method '" + methodDescription + "' is not supported by this command processor. " + + throw new EvaluationException("The method '" + methodDescription + + "' is not supported by this command processor. " + "If using the Control Bus, consider adding @ManagedOperation or @ManagedAttribute."); } } + } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/ExpressionEvaluatingMessageProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/ExpressionEvaluatingMessageProcessor.java index 04955867e9..ec5c12f670 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/ExpressionEvaluatingMessageProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/ExpressionEvaluatingMessageProcessor.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-2019 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. @@ -18,6 +18,7 @@ package org.springframework.integration.handler; import org.springframework.expression.Expression; import org.springframework.expression.ParseException; +import org.springframework.lang.Nullable; import org.springframework.messaging.Message; import org.springframework.util.Assert; @@ -28,6 +29,7 @@ import org.springframework.util.Assert; * @author Mark Fisher * @author Artem Bilan * @author Gary Russell + * * @since 2.0 */ public class ExpressionEvaluatingMessageProcessor extends AbstractMessageProcessor { @@ -51,7 +53,7 @@ public class ExpressionEvaluatingMessageProcessor extends AbstractMessageProc * @param expression The expression. * @param expectedType The expected type. */ - public ExpressionEvaluatingMessageProcessor(Expression expression, Class expectedType) { + public ExpressionEvaluatingMessageProcessor(Expression expression, @Nullable Class expectedType) { Assert.notNull(expression, "The expression must not be null"); try { this.expression = expression; @@ -84,7 +86,7 @@ public class ExpressionEvaluatingMessageProcessor extends AbstractMessageProc * @param expectedType the expected result type. * @since 5.0 */ - public ExpressionEvaluatingMessageProcessor(String expression, Class expectedType) { + public ExpressionEvaluatingMessageProcessor(String expression, @Nullable Class expectedType) { try { this.expression = EXPRESSION_PARSER.parseExpression(expression); this.expectedType = expectedType; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/LambdaMessageProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/LambdaMessageProcessor.java index 6d1c69dcd5..1e1583f57d 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/LambdaMessageProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/LambdaMessageProcessor.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2018 the original author or authors. + * Copyright 2016-2019 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. @@ -29,6 +29,7 @@ import org.springframework.beans.BeansException; import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.BeanFactoryAware; import org.springframework.integration.context.IntegrationContextUtils; +import org.springframework.integration.support.utils.IntegrationUtils; import org.springframework.messaging.Message; import org.springframework.messaging.MessageHandlingException; import org.springframework.messaging.converter.MessageConverter; @@ -47,7 +48,7 @@ import org.springframework.util.ReflectionUtils; */ public class LambdaMessageProcessor implements MessageProcessor, BeanFactoryAware { - private static final Log logger = LogFactory.getLog(LambdaMessageProcessor.class); // NOSONAR lower case static + private static final Log LOGGER = LogFactory.getLog(LambdaMessageProcessor.class); private final Object target; @@ -103,15 +104,17 @@ public class LambdaMessageProcessor implements MessageProcessor, BeanFac } catch (InvocationTargetException e) { if (e.getTargetException() instanceof ClassCastException) { - logger.error("Could not invoke the method due to a class cast exception, if using a lambda in the DSL, " - + "consider using an overloaded EIP method that takes a Class argument to explicitly " - + "specify the type. An example of when this often occurs is if the lambda is configured to " - + "receive a Message argument.", e.getCause()); + LOGGER.error("Could not invoke the method due to a class cast exception, " + + "if using a lambda in the DSL, consider using an overloaded EIP method " + + "that takes a Class argument to explicitly specify the type. " + + "An example of when this often occurs is if the lambda is configured to " + + "receive a Message argument.", e.getCause()); } throw new MessageHandlingException(message, e.getCause()); } catch (Exception e) { - throw new MessageHandlingException(message, e); + throw IntegrationUtils.wrapInHandlingExceptionIfNecessary(message, + () -> "error occurred during processing message in 'LambdaMessageProcessor' [" + this + "]", e); } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/LoggingHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/LoggingHandler.java index ecf8b434ce..5286f7648f 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/LoggingHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/LoggingHandler.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2018 the original author or authors. + * Copyright 2002-2019 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. @@ -19,6 +19,7 @@ package org.springframework.integration.handler; import java.io.PrintWriter; import java.io.StringWriter; import java.io.Writer; +import java.util.function.Function; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -27,6 +28,8 @@ import org.springframework.expression.EvaluationContext; import org.springframework.expression.Expression; import org.springframework.integration.dispatcher.AggregateMessageDeliveryException; import org.springframework.integration.expression.ExpressionUtils; +import org.springframework.integration.expression.FunctionExpression; +import org.springframework.lang.Nullable; import org.springframework.messaging.Message; import org.springframework.util.Assert; import org.springframework.util.StringUtils; @@ -41,6 +44,7 @@ import org.springframework.util.StringUtils; * @author Gary Russell * @author Artem Bilan * @author Andriy Kryvtsun + * * @since 1.0.1 */ public class LoggingHandler extends AbstractMessageHandler { @@ -51,17 +55,17 @@ public class LoggingHandler extends AbstractMessageHandler { } - private volatile Level level; + private Level level; - private volatile boolean expressionSet; + private Expression expression = new FunctionExpression>(Message::getPayload); - private volatile Expression expression = EXPRESSION_PARSER.parseExpression("payload"); + private boolean expressionSet; - private volatile EvaluationContext evaluationContext = ExpressionUtils.createStandardEvaluationContext(); + private EvaluationContext evaluationContext = ExpressionUtils.createStandardEvaluationContext(); - private volatile boolean shouldLogFullMessageSet; + private boolean shouldLogFullMessageSet; - private volatile Log messageLogger = this.logger; + private Log messageLogger = this.logger; /** * Create a LoggingHandler with the given log level (case-insensitive). @@ -153,9 +157,10 @@ public class LoggingHandler extends AbstractMessageHandler { public void setShouldLogFullMessage(boolean shouldLogFullMessage) { Assert.isTrue(!(this.expressionSet), "Cannot set both 'expression' AND 'shouldLogFullMessage' properties"); this.shouldLogFullMessageSet = true; - this.expression = shouldLogFullMessage - ? EXPRESSION_PARSER.parseExpression("#root") - : EXPRESSION_PARSER.parseExpression("payload"); + this.expression = + shouldLogFullMessage + ? new FunctionExpression>(Function.identity()) + : new FunctionExpression>(Message::getPayload); } @Override @@ -170,7 +175,7 @@ public class LoggingHandler extends AbstractMessageHandler { } @Override - protected void handleMessageInternal(Message message) throws Exception { + protected void handleMessageInternal(Message message) { switch (this.level) { case FATAL: if (this.messageLogger.isFatalEnabled()) { @@ -207,6 +212,7 @@ public class LoggingHandler extends AbstractMessageHandler { } } + @Nullable private Object createLogMessage(Message message) { Object logMessage = this.expression.getValue(this.evaluationContext, message); return logMessage instanceof Throwable diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/MessageProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/MessageProcessor.java index a4121e99b6..abd2c8c730 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/MessageProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/MessageProcessor.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-2019 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,6 +16,7 @@ package org.springframework.integration.handler; +import org.springframework.lang.Nullable; import org.springframework.messaging.Message; /** @@ -37,6 +38,8 @@ import org.springframework.messaging.Message; * message-handling components. As such, it is subject to change. * * @author Mark Fisher + * @author Artem Bilan + * * @since 2.0 */ @FunctionalInterface @@ -48,6 +51,7 @@ public interface MessageProcessor { * @param message The message to process. * @return The result. */ + @Nullable T processMessage(Message message); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/MethodInvokingMessageHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/MethodInvokingMessageHandler.java index 18a06bf4a9..62fa00ecb1 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/MethodInvokingMessageHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/MethodInvokingMessageHandler.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2018 the original author or authors. + * Copyright 2002-2019 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. @@ -34,18 +34,18 @@ import org.springframework.util.Assert; */ public class MethodInvokingMessageHandler extends AbstractMessageHandler implements Lifecycle { - private volatile MethodInvokingMessageProcessor processor; + private final MethodInvokingMessageProcessor processor; - private volatile String componentType; + private String componentType; public MethodInvokingMessageHandler(Object object, Method method) { Assert.isTrue(method.getReturnType().equals(void.class), "MethodInvokingMessageHandler requires a void-returning method"); - this.processor = new MethodInvokingMessageProcessor(object, method); + this.processor = new MethodInvokingMessageProcessor<>(object, method); } public MethodInvokingMessageHandler(Object object, String methodName) { - this.processor = new MethodInvokingMessageProcessor(object, methodName); + this.processor = new MethodInvokingMessageProcessor<>(object, methodName); } @Override @@ -79,7 +79,7 @@ public class MethodInvokingMessageHandler extends AbstractMessageHandler impleme } @Override - protected void handleMessageInternal(Message message) throws Exception { + protected void handleMessageInternal(Message message) { Object result = this.processor.processMessage(message); if (result != null) { throw new MessagingException(message, "the MethodInvokingMessageHandler method must " diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/MethodInvokingMessageProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/MethodInvokingMessageProcessor.java index 160ac050ca..b0d3f29664 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/MethodInvokingMessageProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/MethodInvokingMessageProcessor.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2018 the original author or authors. + * Copyright 2002-2019 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. @@ -23,9 +23,9 @@ import org.springframework.beans.factory.BeanFactory; import org.springframework.context.Lifecycle; import org.springframework.core.convert.ConversionService; import org.springframework.integration.handler.support.MessagingMethodInvokerHelper; -import org.springframework.lang.NonNull; +import org.springframework.integration.support.utils.IntegrationUtils; +import org.springframework.lang.Nullable; import org.springframework.messaging.Message; -import org.springframework.messaging.MessageHandlingException; /** * A MessageProcessor implementation that invokes a method on a target Object. @@ -69,7 +69,7 @@ public class MethodInvokingMessageProcessor extends AbstractMessageProcessor< } @Override - public void setBeanFactory(@NonNull BeanFactory beanFactory) { + public void setBeanFactory(@Nullable BeanFactory beanFactory) { super.setBeanFactory(beanFactory); this.delegate.setBeanFactory(beanFactory); } @@ -101,12 +101,15 @@ public class MethodInvokingMessageProcessor extends AbstractMessageProcessor< } @Override + @Nullable public T processMessage(Message message) { try { return this.delegate.process(message); } catch (Exception e) { - throw new MessageHandlingException(message, e); + throw IntegrationUtils.wrapInHandlingExceptionIfNecessary(message, + () -> "error occurred during processing message in 'MethodInvokingMessageProcessor' [" + this + "]", + e); } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/ServiceActivatingHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/ServiceActivatingHandler.java index d1b267d470..9cff81863a 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/ServiceActivatingHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/ServiceActivatingHandler.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2017 the original author or authors. + * Copyright 2002-2019 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. @@ -22,6 +22,7 @@ import org.springframework.beans.factory.BeanFactoryAware; import org.springframework.context.Lifecycle; import org.springframework.core.convert.ConversionService; import org.springframework.integration.annotation.ServiceActivator; +import org.springframework.lang.Nullable; import org.springframework.messaging.Message; /** @@ -35,15 +36,15 @@ public class ServiceActivatingHandler extends AbstractReplyProducingMessageHandl public ServiceActivatingHandler(final Object object) { - this(new MethodInvokingMessageProcessor(object, ServiceActivator.class)); + this(new MethodInvokingMessageProcessor<>(object, ServiceActivator.class)); } public ServiceActivatingHandler(Object object, Method method) { - this(new MethodInvokingMessageProcessor(object, method)); + this(new MethodInvokingMessageProcessor<>(object, method)); } public ServiceActivatingHandler(Object object, String methodName) { - this(new MethodInvokingMessageProcessor(object, methodName)); + this(new MethodInvokingMessageProcessor<>(object, methodName)); } public ServiceActivatingHandler(MessageProcessor processor) { @@ -89,6 +90,7 @@ public class ServiceActivatingHandler extends AbstractReplyProducingMessageHandl } @Override + @Nullable protected Object handleRequestMessage(Message message) { return this.processor.processMessage(message); } @@ -96,7 +98,7 @@ public class ServiceActivatingHandler extends AbstractReplyProducingMessageHandl @Override public String toString() { return "ServiceActivator for [" + this.processor + "]" - + (this.getComponentName() == null ? "" : " (" + this.getComponentName() + ")"); + + (getComponentName() == null ? "" : " (" + getComponentName() + ")"); } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/support/MessagingMethodInvokerHelper.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/support/MessagingMethodInvokerHelper.java index 80d74ad683..c0ee267f30 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/support/MessagingMethodInvokerHelper.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/support/MessagingMethodInvokerHelper.java @@ -84,7 +84,7 @@ import org.springframework.integration.util.ClassUtils; import org.springframework.integration.util.FixedMethodFilter; import org.springframework.integration.util.MessagingAnnotationUtils; import org.springframework.integration.util.UniqueMethodFilter; -import org.springframework.lang.NonNull; +import org.springframework.lang.Nullable; import org.springframework.messaging.Message; import org.springframework.messaging.MessageHandlingException; import org.springframework.messaging.MessageHeaders; @@ -285,7 +285,7 @@ public class MessagingMethodInvokerHelper extends AbstractExpressionEvaluator } @Override - public void setBeanFactory(@NonNull BeanFactory beanFactory) { + public void setBeanFactory(BeanFactory beanFactory) { super.setBeanFactory(beanFactory); ((DefaultMessageHandlerMethodFactory) this.messageHandlerMethodFactory).setBeanFactory(beanFactory); @@ -308,11 +308,13 @@ public class MessagingMethodInvokerHelper extends AbstractExpressionEvaluator } } + @Nullable public T process(Message message) throws Exception { ParametersWrapper parameters = new ParametersWrapper(message); return processInternal(parameters); } + @Nullable public T process(Collection> messages, Map headers) throws Exception { ParametersWrapper parameters = new ParametersWrapper(messages, headers); return processInternal(parameters); @@ -468,6 +470,7 @@ public class MessagingMethodInvokerHelper extends AbstractExpressionEvaluator } @SuppressWarnings("unchecked") + @Nullable private T processInternal(ParametersWrapper parameters) throws Exception { if (!this.initialized) { initialize(); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/utils/IntegrationUtils.java b/spring-integration-core/src/main/java/org/springframework/integration/support/utils/IntegrationUtils.java index e4d07ba251..825affd66a 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/support/utils/IntegrationUtils.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/utils/IntegrationUtils.java @@ -1,5 +1,5 @@ /* - * Copyright 2014-2018 the original author or authors. + * Copyright 2014-2019 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. @@ -37,6 +37,8 @@ import org.springframework.util.Assert; * * @author Gary Russell * @author Marius Bogoevici + * @author Artem Bilan + * * @since 4.0 * */ @@ -113,7 +115,6 @@ public final class IntegrationUtils { /** * Utility method for null-safe conversion from String to byte[] - * * @param value the String to be converted * @param encoding the encoding * @return the byte[] corresponding to the given String and encoding, null if provided String argument was null @@ -130,7 +131,6 @@ public final class IntegrationUtils { /** * Utility method for null-safe conversion from byte[] to String - * * @param bytes the byte[] to be converted * @param encoding the encoding * @return the String corresponding to the given byte[] and encoding, null if provided byte[] argument was null @@ -151,19 +151,19 @@ public final class IntegrationUtils { * in a new {@link MessageDeliveryException} with the message. * @param message the message. * @param text a Supplier for the new exception's message text. - * @param e the exception. + * @param ex the exception. * @return the wrapper, if necessary, or the original exception. * @since 5.0.4 */ public static RuntimeException wrapInDeliveryExceptionIfNecessary(Message message, Supplier text, - Exception e) { + Throwable ex) { - RuntimeException runtimeException = (e instanceof RuntimeException) - ? (RuntimeException) e - : new MessageDeliveryException(message, text.get(), e); - if (!(e instanceof MessagingException) || - ((MessagingException) e).getFailedMessage() == null) { - runtimeException = new MessageDeliveryException(message, text.get(), e); + RuntimeException runtimeException = (ex instanceof RuntimeException) + ? (RuntimeException) ex + : new MessageDeliveryException(message, text.get(), ex); + if (!(ex instanceof MessagingException) || + ((MessagingException) ex).getFailedMessage() == null) { + runtimeException = new MessageDeliveryException(message, text.get(), ex); } return runtimeException; } @@ -174,19 +174,19 @@ public final class IntegrationUtils { * in a new {@link MessageHandlingException} with the message. * @param message the message. * @param text a Supplier for the new exception's message text. - * @param e the exception. + * @param ex the exception. * @return the wrapper, if necessary, or the original exception. * @since 5.0.4 */ public static RuntimeException wrapInHandlingExceptionIfNecessary(Message message, Supplier text, - Exception e) { + Throwable ex) { - RuntimeException runtimeException = (e instanceof RuntimeException) - ? (RuntimeException) e - : new MessageHandlingException(message, text.get(), e); - if (!(e instanceof MessagingException) || - ((MessagingException) e).getFailedMessage() == null) { - runtimeException = new MessageHandlingException(message, text.get(), e); + RuntimeException runtimeException = (ex instanceof RuntimeException) + ? (RuntimeException) ex + : new MessageHandlingException(message, text.get(), ex); + if (!(ex instanceof MessagingException) || + ((MessagingException) ex).getFailedMessage() == null) { + runtimeException = new MessageHandlingException(message, text.get(), ex); } return runtimeException; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/util/AbstractExpressionEvaluator.java b/spring-integration-core/src/main/java/org/springframework/integration/util/AbstractExpressionEvaluator.java index 6147a244ea..e7118bc60e 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/util/AbstractExpressionEvaluator.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/util/AbstractExpressionEvaluator.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2018 the original author or authors. + * Copyright 2002-2019 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,7 +35,6 @@ import org.springframework.integration.support.MessageBuilderFactory; import org.springframework.integration.support.utils.IntegrationUtils; import org.springframework.lang.Nullable; import org.springframework.messaging.Message; -import org.springframework.messaging.MessageHandlingException; /** * @author Mark Fisher @@ -64,13 +63,11 @@ public abstract class AbstractExpressionEvaluator implements BeanFactoryAware, I * Specify a BeanFactory in order to enable resolution via @beanName in the expression. */ @Override - public void setBeanFactory(final @Nullable BeanFactory beanFactory) { - if (beanFactory != null) { - this.beanFactory = beanFactory; - this.typeConverter.setBeanFactory(beanFactory); - if (this.evaluationContext != null && this.evaluationContext.getBeanResolver() == null) { - this.evaluationContext.setBeanResolver(new BeanFactoryResolver(beanFactory)); - } + public void setBeanFactory(BeanFactory beanFactory) { + this.beanFactory = beanFactory; + this.typeConverter.setBeanFactory(beanFactory); + if (this.evaluationContext != null && this.evaluationContext.getBeanResolver() == null) { + this.evaluationContext.setBeanResolver(new BeanFactoryResolver(beanFactory)); } } @@ -89,7 +86,7 @@ public abstract class AbstractExpressionEvaluator implements BeanFactoryAware, I } @Override - public final void afterPropertiesSet() throws Exception { + public final void afterPropertiesSet() { getEvaluationContext(); if (this.beanFactory != null) { this.messageBuilderFactory = IntegrationUtils.getMessageBuilderFactory(this.beanFactory); @@ -99,7 +96,7 @@ public abstract class AbstractExpressionEvaluator implements BeanFactoryAware, I } protected StandardEvaluationContext getEvaluationContext() { - return this.getEvaluationContext(true); + return getEvaluationContext(true); } /** @@ -126,50 +123,52 @@ public abstract class AbstractExpressionEvaluator implements BeanFactoryAware, I return this.evaluationContext; } - protected T evaluateExpression(Expression expression, Message message, Class expectedType) { + @Nullable + protected T evaluateExpression(Expression expression, Message message, @Nullable Class expectedType) { try { return evaluateExpression(expression, (Object) message, expectedType); } - catch (EvaluationException e) { - Throwable cause = e.getCause(); - if (this.logger.isDebugEnabled()) { - this.logger.debug("SpEL Expression evaluation failed with EvaluationException.", e); + catch (Exception ex) { + this.logger.debug("SpEL Expression evaluation failed with Exception.", ex); + Throwable cause = null; + if (ex instanceof EvaluationException) { + cause = ex.getCause(); } - throw new MessageHandlingException(message, "Expression evaluation failed: " - + expression.getExpressionString(), cause == null ? e : cause); - } - catch (Exception e) { - if (this.logger.isDebugEnabled()) { - this.logger.debug("SpEL Expression evaluation failed with Exception." + e); - } - throw new MessageHandlingException(message, "Expression evaluation failed: " - + expression.getExpressionString(), e); + throw IntegrationUtils.wrapInHandlingExceptionIfNecessary(message, + () -> "Expression evaluation failed: " + expression.getExpressionString(), + cause == null ? ex : cause); } } + @Nullable protected Object evaluateExpression(String expression, Object input) { - return this.evaluateExpression(expression, input, null); + return evaluateExpression(expression, input, null); } - protected T evaluateExpression(String expression, Object input, Class expectedType) { + @Nullable + protected T evaluateExpression(String expression, Object input, @Nullable Class expectedType) { return EXPRESSION_PARSER.parseExpression(expression) - .getValue(this.getEvaluationContext(), input, expectedType); + .getValue(getEvaluationContext(), input, expectedType); } + @Nullable protected Object evaluateExpression(Expression expression, Object input) { - return this.evaluateExpression(expression, input, null); + return evaluateExpression(expression, input, null); } - protected T evaluateExpression(Expression expression, Class expectedType) { - return expression.getValue(this.getEvaluationContext(), expectedType); + @Nullable + protected T evaluateExpression(Expression expression, @Nullable Class expectedType) { + return expression.getValue(getEvaluationContext(), expectedType); } + @Nullable protected Object evaluateExpression(Expression expression) { - return expression.getValue(this.getEvaluationContext()); + return expression.getValue(getEvaluationContext()); } - protected T evaluateExpression(Expression expression, Object input, Class expectedType) { - return expression.getValue(this.getEvaluationContext(), input, expectedType); + @Nullable + protected T evaluateExpression(Expression expression, Object input, @Nullable Class expectedType) { + return expression.getValue(getEvaluationContext(), input, expectedType); } protected void onInit() { diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/LoggingChannelAdapterParserTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/LoggingChannelAdapterParserTests.java index 2cbc9ed8ff..a4757319c0 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/LoggingChannelAdapterParserTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/LoggingChannelAdapterParserTests.java @@ -17,22 +17,21 @@ package org.springframework.integration.config.xml; import static org.assertj.core.api.Assertions.assertThat; -import static org.assertj.core.api.Assertions.fail; +import static org.assertj.core.api.Assertions.assertThatExceptionOfType; import org.junit.Test; import org.junit.runner.RunWith; -import org.springframework.beans.BeansException; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.beans.factory.parsing.BeanDefinitionParsingException; import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.core.Ordered; import org.springframework.integration.endpoint.EventDrivenConsumer; +import org.springframework.integration.expression.FunctionExpression; import org.springframework.integration.handler.LoggingHandler; import org.springframework.integration.test.util.TestUtils; -import org.springframework.test.context.ContextConfiguration; -import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; +import org.springframework.test.context.junit4.SpringRunner; /** * @author Mark Fisher @@ -41,14 +40,15 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; * * @since 2.1 */ -@ContextConfiguration -@RunWith(SpringJUnit4ClassRunner.class) +@RunWith(SpringRunner.class) public class LoggingChannelAdapterParserTests { - @Autowired @Qualifier("logger.adapter") + @Autowired + @Qualifier("logger.adapter") private EventDrivenConsumer loggerConsumer; - @Autowired @Qualifier("loggerWithExpression.adapter") + @Autowired + @Qualifier("loggerWithExpression.adapter") private EventDrivenConsumer loggerWithExpression; @@ -58,33 +58,30 @@ public class LoggingChannelAdapterParserTests { assertThat(TestUtils.getPropertyValue(loggingHandler, "messageLogger.logger.name")) .isEqualTo("org.springframework.integration.test.logger"); assertThat(TestUtils.getPropertyValue(loggingHandler, "order")).isEqualTo(1); - assertThat(TestUtils.getPropertyValue(loggingHandler, "level").toString()).isEqualTo("WARN"); - assertThat(TestUtils.getPropertyValue(loggingHandler, "expression.expression")).isEqualTo("#root"); + assertThat(TestUtils.getPropertyValue(loggingHandler, "level")).isEqualTo(LoggingHandler.Level.WARN); + assertThat(TestUtils.getPropertyValue(loggingHandler, "expression")).isInstanceOf(FunctionExpression.class); } @Test public void verifyExpressionAndOtherDefaultConfig() { - LoggingHandler loggingHandler = TestUtils.getPropertyValue(loggerWithExpression, "handler", LoggingHandler.class); + LoggingHandler loggingHandler = + TestUtils.getPropertyValue(loggerWithExpression, "handler", LoggingHandler.class); assertThat(TestUtils.getPropertyValue(loggingHandler, "messageLogger.logger.name")) .isEqualTo("org.springframework.integration.handler.LoggingHandler"); assertThat(TestUtils.getPropertyValue(loggingHandler, "order")).isEqualTo(Ordered.LOWEST_PRECEDENCE); - assertThat(TestUtils.getPropertyValue(loggingHandler, "level").toString()).isEqualTo("INFO"); + assertThat(TestUtils.getPropertyValue(loggingHandler, "level")).isEqualTo(LoggingHandler.Level.INFO); assertThat(TestUtils.getPropertyValue(loggingHandler, "expression.expression")).isEqualTo("payload.foo"); assertThat(TestUtils.getPropertyValue(loggingHandler, "evaluationContext.beanResolver")).isNotNull(); } @Test public void failConfigLogFullMessageAndExpression() { - try { - new ClassPathXmlApplicationContext("LoggingChannelAdapterParserTests-fail-context.xml", this.getClass()) - .close(); - fail("BeanDefinitionParsingException expected"); - } - catch (BeansException e) { - assertThat(e instanceof BeanDefinitionParsingException).isTrue(); - assertThat(e.getMessage() - .contains("The 'expression' and 'log-full-message' attributes are mutually exclusive.")).isTrue(); - } + assertThatExceptionOfType(BeanDefinitionParsingException.class) + .isThrownBy(() -> + new ClassPathXmlApplicationContext( + "LoggingChannelAdapterParserTests-fail-context.xml", + getClass())) + .withMessageContaining("The 'expression' and 'log-full-message' attributes are mutually exclusive."); } } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/handler/MethodInvokingMessageProcessorTests.java b/spring-integration-core/src/test/java/org/springframework/integration/handler/MethodInvokingMessageProcessorTests.java index 1a933bfba8..46c3c784aa 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/handler/MethodInvokingMessageProcessorTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/handler/MethodInvokingMessageProcessorTests.java @@ -17,7 +17,7 @@ package org.springframework.integration.handler; import static org.assertj.core.api.Assertions.assertThat; -import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.assertj.core.api.Assertions.assertThatExceptionOfType; import static org.assertj.core.api.Assertions.fail; import static org.mockito.AdditionalAnswers.returnsFirstArg; import static org.mockito.ArgumentMatchers.anyString; @@ -76,6 +76,7 @@ import org.springframework.messaging.MessageHandlingException; import org.springframework.messaging.MessageHeaders; import org.springframework.messaging.MessagingException; import org.springframework.messaging.PollableChannel; +import org.springframework.messaging.converter.MessageConversionException; import org.springframework.messaging.handler.annotation.Header; import org.springframework.messaging.handler.annotation.Payload; import org.springframework.messaging.handler.annotation.support.DefaultMessageHandlerMethodFactory; @@ -394,13 +395,14 @@ public class MethodInvokingMessageProcessorTests { assertThat(result).isEqualTo(456); } - @Test(expected = MessageHandlingException.class) + @Test public void conversionFailureWithAnnotatedMethod() throws Exception { AnnotatedTestService service = new AnnotatedTestService(); Method method = service.getClass().getMethod("integerMethod", Integer.class); MethodInvokingMessageProcessor processor = new MethodInvokingMessageProcessor(service, method); processor.setBeanFactory(mock(BeanFactory.class)); - processor.processMessage(new GenericMessage<>("foo")); + assertThatExceptionOfType(MessageConversionException.class) + .isThrownBy(() -> processor.processMessage(new GenericMessage<>("foo"))); } @Test @@ -420,8 +422,9 @@ public class MethodInvokingMessageProcessorTests { Method method = service.getClass().getMethod("integerMethod", Integer.class); MethodInvokingMessageProcessor processor = new MethodInvokingMessageProcessor(service, method); processor.setBeanFactory(mock(BeanFactory.class)); - assertThatThrownBy(() -> processor.processMessage(new GenericMessage<>("foo"))) - .isInstanceOf(MessageHandlingException.class); + assertThatExceptionOfType(MessageConversionException.class) + .isThrownBy(() -> processor.processMessage(new GenericMessage<>("foo"))) + .withMessageContaining("Failed to convert message payload 'foo' to 'java.lang.Integer'"); } @Test @@ -430,8 +433,9 @@ public class MethodInvokingMessageProcessorTests { Method method = service.getClass().getMethod("error", String.class); MethodInvokingMessageProcessor processor = new MethodInvokingMessageProcessor(service, method); processor.setBeanFactory(mock(BeanFactory.class)); - assertThatThrownBy(() -> processor.processMessage(new GenericMessage<>("foo"))) - .hasCauseInstanceOf(UnsupportedOperationException.class); + assertThatExceptionOfType(MessageHandlingException.class) + .isThrownBy(() -> processor.processMessage(new GenericMessage<>("foo"))) + .withCauseInstanceOf(UnsupportedOperationException.class); } @Test @@ -440,8 +444,9 @@ public class MethodInvokingMessageProcessorTests { Method method = service.getClass().getMethod("checked", String.class); MethodInvokingMessageProcessor processor = new MethodInvokingMessageProcessor(service, method); processor.setBeanFactory(mock(BeanFactory.class)); - assertThatThrownBy(() -> processor.processMessage(new GenericMessage<>("foo"))) - .hasCauseInstanceOf(CheckedException.class); + assertThatExceptionOfType(MessageHandlingException.class) + .isThrownBy(() -> processor.processMessage(new GenericMessage<>("foo"))) + .withCauseInstanceOf(CheckedException.class); } @Test @@ -451,8 +456,9 @@ public class MethodInvokingMessageProcessorTests { MethodInvokingMessageProcessor processor = new MethodInvokingMessageProcessor(service, method); processor.setUseSpelInvoker(true); processor.setBeanFactory(mock(BeanFactory.class)); - assertThatThrownBy(() -> processor.processMessage(new GenericMessage<>("foo"))) - .hasCauseInstanceOf(SpelEvaluationException.class); + assertThatExceptionOfType(MessageHandlingException.class) + .isThrownBy(() -> processor.processMessage(new GenericMessage<>("foo"))) + .withCauseInstanceOf(SpelEvaluationException.class); } @Test diff --git a/spring-integration-core/src/test/java/org/springframework/integration/handler/ServiceActivatorDefaultFrameworkMethodTests.java b/spring-integration-core/src/test/java/org/springframework/integration/handler/ServiceActivatorDefaultFrameworkMethodTests.java index ad725283d2..6b5eadb32a 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/handler/ServiceActivatorDefaultFrameworkMethodTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/handler/ServiceActivatorDefaultFrameworkMethodTests.java @@ -17,6 +17,7 @@ package org.springframework.integration.handler; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatExceptionOfType; import static org.assertj.core.api.Assertions.fail; import java.util.concurrent.atomic.AtomicReference; @@ -43,8 +44,7 @@ import org.springframework.messaging.MessageHandlingException; import org.springframework.messaging.MessagingException; import org.springframework.messaging.PollableChannel; import org.springframework.messaging.support.ErrorMessage; -import org.springframework.test.context.ContextConfiguration; -import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; +import org.springframework.test.context.junit4.SpringRunner; import org.springframework.util.concurrent.ListenableFuture; import org.springframework.util.concurrent.SettableListenableFuture; @@ -57,8 +57,7 @@ import org.springframework.util.concurrent.SettableListenableFuture; * * @since 2.0.1 */ -@ContextConfiguration -@RunWith(SpringJUnit4ClassRunner.class) +@RunWith(SpringRunner.class) public class ServiceActivatorDefaultFrameworkMethodTests { @Autowired @@ -109,19 +108,14 @@ public class ServiceActivatorDefaultFrameworkMethodTests { assertThat(reply.getHeaders().get("history").toString()) .isEqualTo("gatewayTestInputChannel,gatewayTestService,gateway,requestChannel,replyChannel"); - message = MessageBuilder.withPayload("foo").setReplyChannel(replyChannel).build(); - try { - this.gatewayTestInputChannel.send(message); - fail("Exception expected"); - } - catch (Exception e) { - assertThat(e).isInstanceOf(MessageHandlingException.class); - assertThat(e.getCause()).isInstanceOf(MessageTransformationException.class); - assertThat(e.getCause().getCause()).isInstanceOf(MessageHandlingException.class); - assertThat(e.getCause().getCause().getCause()).isInstanceOf(IllegalStateException.class); - assertThat(e.getMessage()).contains("Expression evaluation failed"); - assertThat(e.getMessage()).contains("Wrong payload"); - } + Message message2 = MessageBuilder.withPayload("foo").setReplyChannel(replyChannel).build(); + + assertThatExceptionOfType(MessageTransformationException.class) + .isThrownBy(() -> this.gatewayTestInputChannel.send(message2)) + .withCauseInstanceOf(MessageHandlingException.class) + .withRootCauseInstanceOf(IllegalStateException.class) + .withMessageContaining("Expression evaluation failed") + .withMessageContaining("Wrong payload"); } @Test diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/FileWritingMessageHandler.java b/spring-integration-file/src/main/java/org/springframework/integration/file/FileWritingMessageHandler.java index 8f9471a8c2..5684522871 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/FileWritingMessageHandler.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/FileWritingMessageHandler.java @@ -57,6 +57,7 @@ import org.springframework.integration.handler.MessageTriggerAction; import org.springframework.integration.support.locks.DefaultLockRegistry; import org.springframework.integration.support.locks.LockRegistry; import org.springframework.integration.support.locks.PassThruLockRegistry; +import org.springframework.integration.support.utils.IntegrationUtils; import org.springframework.integration.util.WhileLockedProcessor; import org.springframework.messaging.Message; import org.springframework.messaging.MessageHandlingException; @@ -526,7 +527,8 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand timestamp); } catch (Exception e) { - throw new MessageHandlingException(requestMessage, "failed to write Message payload to file", e); + throw IntegrationUtils.wrapInHandlingExceptionIfNecessary(requestMessage, + () -> "failed to write Message payload to file", e); } } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandler.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandler.java index a2e898e186..2707d79ae7 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandler.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandler.java @@ -33,6 +33,7 @@ import org.springframework.integration.ip.tcp.connection.ConnectionFactory; import org.springframework.integration.ip.tcp.connection.TcpConnection; import org.springframework.integration.ip.tcp.connection.TcpConnectionFailedCorrelationEvent; import org.springframework.integration.ip.tcp.connection.TcpSender; +import org.springframework.integration.support.utils.IntegrationUtils; import org.springframework.messaging.Message; import org.springframework.messaging.MessageHandlingException; import org.springframework.util.Assert; @@ -119,7 +120,8 @@ public class TcpSendingMessageHandler extends AbstractMessageHandler implements catch (Exception ex) { logger.error("Error sending message", ex); connection.close(); - throw wrapToMessageHandlingExceptionIfNecessary(message, "Error sending message", ex); + throw IntegrationUtils.wrapInHandlingExceptionIfNecessary(message, + () -> "Error sending message", ex); } finally { if (this.isSingleUse) { // close after replying @@ -136,17 +138,6 @@ public class TcpSendingMessageHandler extends AbstractMessageHandler implements } } - private MessageHandlingException wrapToMessageHandlingExceptionIfNecessary(Message message, String description, - Throwable cause) { - - if (cause instanceof MessageHandlingException) { - throw (MessageHandlingException) cause; - } - else { - throw new MessageHandlingException(message, description, cause); - } - } - private void handleMessageAsClient(Message message) { // we own the connection TcpConnection connection = null; @@ -190,12 +181,16 @@ public class TcpSendingMessageHandler extends AbstractMessageHandler implements connection.send(message); } catch (Exception ex) { - String connectionId = null; + final String connectionId; if (connection != null) { connectionId = connection.getConnectionId(); } - throw wrapToMessageHandlingExceptionIfNecessary(message, - "Failed to handle message using " + connectionId, ex); + else { + connectionId = null; + } + + throw IntegrationUtils.wrapInHandlingExceptionIfNecessary(message, + () -> "Failed to handle message using " + connectionId, ex); } return connection; } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastSendingMessageHandler.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastSendingMessageHandler.java index bbe4b2ac19..f7a3badb79 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastSendingMessageHandler.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastSendingMessageHandler.java @@ -38,10 +38,9 @@ import org.springframework.expression.EvaluationContext; import org.springframework.expression.Expression; import org.springframework.integration.context.IntegrationContextUtils; import org.springframework.integration.ip.AbstractInternetProtocolSendingMessageHandler; +import org.springframework.integration.support.utils.IntegrationUtils; import org.springframework.lang.Nullable; import org.springframework.messaging.Message; -import org.springframework.messaging.MessageDeliveryException; -import org.springframework.messaging.MessageHandlingException; import org.springframework.messaging.MessagingException; import org.springframework.scheduling.concurrent.CustomizableThreadFactory; import org.springframework.util.Assert; @@ -253,8 +252,7 @@ public class UnicastSendingMessageHandler extends } @Override - public void handleMessageInternal(Message message) throws MessageHandlingException, - MessageDeliveryException { + public void handleMessageInternal(Message message) { if (this.acknowledge) { Assert.state(this.isRunning(), "When 'acknowledge' is enabled, adapter must be running"); startAckThread(); @@ -284,12 +282,12 @@ public class UnicastSendingMessageHandler extends } } } - catch (MessagingException e) { - throw e; - } - catch (Exception e) { - closeSocketIfNeeded(); - throw new MessageHandlingException(message, "failed to send UDP packet", e); + catch (Exception ex) { + if (!(ex instanceof MessagingException)) { + closeSocketIfNeeded(); + } + throw IntegrationUtils.wrapInHandlingExceptionIfNecessary(message, + () -> "Failed to send UDP packet", ex); } finally { if (countdownLatch != null) { diff --git a/spring-integration-jpa/src/main/java/org/springframework/integration/jpa/support/parametersource/ExpressionEvaluatingParameterSourceUtils.java b/spring-integration-jpa/src/main/java/org/springframework/integration/jpa/support/parametersource/ExpressionEvaluatingParameterSourceUtils.java index 3ad8164031..5b2c4be906 100644 --- a/spring-integration-jpa/src/main/java/org/springframework/integration/jpa/support/parametersource/ExpressionEvaluatingParameterSourceUtils.java +++ b/spring-integration-jpa/src/main/java/org/springframework/integration/jpa/support/parametersource/ExpressionEvaluatingParameterSourceUtils.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-2019 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. @@ -24,38 +24,38 @@ import org.springframework.expression.Expression; import org.springframework.expression.spel.support.StandardEvaluationContext; import org.springframework.integration.jpa.support.JpaParameter; import org.springframework.integration.util.AbstractExpressionEvaluator; +import org.springframework.lang.Nullable; import org.springframework.util.Assert; /** * * @author Gunnar Hillert * @author Gary Russell + * @author Artem Bilan + * * @since 2.2 * */ final class ExpressionEvaluatingParameterSourceUtils { private ExpressionEvaluatingParameterSourceUtils() { - throw new AssertionError(); } /** * Utility method that converts a Collection of {@link JpaParameter} to * a Map containing only static parameters. - * * @param jpaParameters Must not be null. * @return Map containing only the static parameters. Will never be null. */ public static Map convertStaticParameters(Collection jpaParameters) { - Assert.notNull(jpaParameters, "The Collection of jpaParameters must not be null."); for (JpaParameter parameter : jpaParameters) { Assert.notNull(parameter, "'jpaParameters' must not contain null values."); } - final Map staticParameters = new HashMap(); + final Map staticParameters = new HashMap<>(); for (JpaParameter parameter : jpaParameters) { if (parameter.getValue() != null) { @@ -78,6 +78,7 @@ final class ExpressionEvaluatingParameterSourceUtils { } @Override + @Nullable public Object evaluateExpression(Expression expression, Object input) { return super.evaluateExpression(expression, input); } diff --git a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/support/DefaultPahoMessageConverter.java b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/support/DefaultPahoMessageConverter.java index c60ae76ed3..d8b6261255 100644 --- a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/support/DefaultPahoMessageConverter.java +++ b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/support/DefaultPahoMessageConverter.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2018 the original author or authors. + * Copyright 2002-2019 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. @@ -29,7 +29,6 @@ import org.springframework.integration.support.DefaultMessageBuilderFactory; import org.springframework.integration.support.MessageBuilderFactory; import org.springframework.integration.support.utils.IntegrationUtils; import org.springframework.messaging.Message; -import org.springframework.messaging.MessageHandlingException; import org.springframework.messaging.MessageHeaders; import org.springframework.messaging.converter.MessageConversionException; import org.springframework.util.Assert; @@ -174,7 +173,6 @@ public class DefaultPahoMessageConverter implements MqttMessageConverter, BeanFa /** * True if the converter should not convert the message payload to a String. * Ignored if a {@link BytesMessageMapper} is provided. - * * @param payloadAsBytes The payloadAsBytes to set. * @see #setBytesMessageMapper(BytesMessageMapper) */ @@ -274,7 +272,8 @@ public class DefaultPahoMessageConverter implements MqttMessageConverter, BeanFa return this.bytesMessageMapper.fromMessage(message); } catch (Exception e) { - throw new MessageHandlingException(message, "Failed to map outbound message", e); + throw IntegrationUtils.wrapInHandlingExceptionIfNecessary(message, + () -> "Failed to map outbound message", e); } } else { diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/outbound/RedisStoreWritingMessageHandler.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/outbound/RedisStoreWritingMessageHandler.java index fc3aba27bd..f97b14df06 100644 --- a/spring-integration-redis/src/main/java/org/springframework/integration/redis/outbound/RedisStoreWritingMessageHandler.java +++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/outbound/RedisStoreWritingMessageHandler.java @@ -1,5 +1,5 @@ /* - * Copyright 2007-2018 the original author or authors. + * Copyright 2007-2019 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. @@ -44,8 +44,8 @@ import org.springframework.integration.expression.ExpressionUtils; import org.springframework.integration.expression.FunctionExpression; import org.springframework.integration.handler.AbstractMessageHandler; import org.springframework.integration.redis.support.RedisHeaders; +import org.springframework.integration.support.utils.IntegrationUtils; import org.springframework.messaging.Message; -import org.springframework.messaging.MessageHandlingException; import org.springframework.util.Assert; import org.springframework.util.NumberUtils; @@ -289,11 +289,11 @@ public class RedisStoreWritingMessageHandler extends AbstractMessageHandler { */ @SuppressWarnings("unchecked") @Override - protected void handleMessageInternal(Message message) throws Exception { + protected void handleMessageInternal(Message message) { String key = this.keyExpression.getValue(this.evaluationContext, message, String.class); Assert.hasText(key, () -> "Failed to determine a key for the Redis store based on the message: " + message); - RedisStore store = this.createStoreView(key); + RedisStore store = createStoreView(key); Assert.state(this.initialized, "handler not initialized - afterPropertiesSet() must be called before the first use"); @@ -314,8 +314,9 @@ public class RedisStoreWritingMessageHandler extends AbstractMessageHandler { writeToProperties((RedisProperties) store, message); } } - catch (Exception e) { - throw new MessageHandlingException(message, "Failed to store Message data in Redis collection", e); + catch (Exception ex) { + throw IntegrationUtils.wrapInHandlingExceptionIfNecessary(message, + () -> "Failed to store Message data in Redis collection", ex); } } @@ -488,7 +489,8 @@ public class RedisStoreWritingMessageHandler extends AbstractMessageHandler { return 1d; } else { - Assert.isInstanceOf(Number.class, scoreHeader, "Header " + RedisHeaders.ZSET_SCORE + " must be a Number"); + Assert.isInstanceOf(Number.class, scoreHeader, + () -> "Header " + RedisHeaders.ZSET_SCORE + " must be a Number"); Number score = (Number) scoreHeader; return Double.valueOf(score.toString()); } diff --git a/spring-integration-rmi/src/main/java/org/springframework/integration/rmi/RmiOutboundGateway.java b/spring-integration-rmi/src/main/java/org/springframework/integration/rmi/RmiOutboundGateway.java index 2fb697621a..f4fa588556 100644 --- a/spring-integration-rmi/src/main/java/org/springframework/integration/rmi/RmiOutboundGateway.java +++ b/spring-integration-rmi/src/main/java/org/springframework/integration/rmi/RmiOutboundGateway.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2017 the original author or authors. + * Copyright 2002-2019 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 RmiOutboundGateway extends AbstractReplyProducingMessageHandler { if (!(requestMessage.getPayload() instanceof Serializable)) { throw new MessageHandlingException(requestMessage, this.getComponentName() + " expects a Serializable payload type " + - "but encountered [" + requestMessage.getPayload().getClass().getName() + "]"); + "but encountered [" + requestMessage.getPayload().getClass().getName() + "]"); } try { return this.proxy.exchange(requestMessage); @@ -86,8 +86,8 @@ public class RmiOutboundGateway extends AbstractReplyProducingMessageHandler { throw new MessageHandlingException(requestMessage, e); } catch (RemoteAccessException e) { - throw new MessageHandlingException(requestMessage, "Remote failure in RmiOutboundGateway: " + - this.getComponentName(), e); + throw new MessageHandlingException(requestMessage, + "Remote failure in RmiOutboundGateway: " + getComponentName(), e); } } diff --git a/spring-integration-rmi/src/test/java/org/springframework/integration/rmi/BackToBackTests.java b/spring-integration-rmi/src/test/java/org/springframework/integration/rmi/BackToBackTests.java index 365143fb20..86ea8abddb 100644 --- a/spring-integration-rmi/src/test/java/org/springframework/integration/rmi/BackToBackTests.java +++ b/spring-integration-rmi/src/test/java/org/springframework/integration/rmi/BackToBackTests.java @@ -17,7 +17,7 @@ package org.springframework.integration.rmi; import static org.assertj.core.api.Assertions.assertThat; -import static org.assertj.core.api.Assertions.fail; +import static org.assertj.core.api.Assertions.assertThatExceptionOfType; import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.verify; @@ -26,7 +26,10 @@ import org.junit.runner.RunWith; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.support.AbstractApplicationContext; +import org.springframework.integration.MessageDispatchingException; import org.springframework.messaging.Message; +import org.springframework.messaging.MessageDeliveryException; +import org.springframework.messaging.MessageHandlingException; import org.springframework.messaging.PollableChannel; import org.springframework.messaging.SubscribableChannel; import org.springframework.messaging.support.GenericMessage; @@ -38,6 +41,7 @@ import org.springframework.transaction.TransactionDefinition; /** * @author Gary Russell * @author Artem Bilan + * * @since 3.0 * */ @@ -75,7 +79,7 @@ public class BackToBackTests { @Test public void testBad() { - bad.send(new GenericMessage("foo")); + bad.send(new GenericMessage<>("foo")); Message reply = this.reply.receive(0); assertThat(reply).isNotNull(); assertThat(reply.getPayload()).isEqualTo("error:foo"); @@ -84,13 +88,11 @@ public class BackToBackTests { @Test public void testUgly() { context.setId("context"); - try { - ugly.send(new GenericMessage("foo")); - fail("Expected exception"); - } - catch (Exception e) { - assertThat(e.getCause().getMessage()).contains("Dispatcher has no subscribers for channel 'context.baz'."); - } + assertThatExceptionOfType(MessageHandlingException.class) + .isThrownBy(() -> ugly.send(new GenericMessage<>("foo"))) + .withCauseInstanceOf(MessageDeliveryException.class) + .withRootCauseInstanceOf(MessageDispatchingException.class) + .withMessageContaining("Dispatcher has no subscribers for channel 'context.baz'."); } } diff --git a/spring-integration-scripting/src/main/java/org/springframework/integration/scripting/AbstractScriptExecutingMessageProcessor.java b/spring-integration-scripting/src/main/java/org/springframework/integration/scripting/AbstractScriptExecutingMessageProcessor.java index cda2cac732..7114cb45ae 100644 --- a/spring-integration-scripting/src/main/java/org/springframework/integration/scripting/AbstractScriptExecutingMessageProcessor.java +++ b/spring-integration-scripting/src/main/java/org/springframework/integration/scripting/AbstractScriptExecutingMessageProcessor.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-2019 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. @@ -23,8 +23,8 @@ import org.springframework.beans.factory.BeanClassLoaderAware; import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.BeanFactoryAware; import org.springframework.integration.handler.MessageProcessor; +import org.springframework.integration.support.utils.IntegrationUtils; import org.springframework.messaging.Message; -import org.springframework.messaging.MessageHandlingException; import org.springframework.scripting.ScriptSource; import org.springframework.util.Assert; @@ -33,9 +33,12 @@ import org.springframework.util.Assert; * * @author Mark Fisher * @author Stefan Reuter + * @author Artem Bilan + * * @since 2.0 */ -public abstract class AbstractScriptExecutingMessageProcessor implements MessageProcessor, BeanClassLoaderAware, BeanFactoryAware { +public abstract class AbstractScriptExecutingMessageProcessor + implements MessageProcessor, BeanClassLoaderAware, BeanFactoryAware { private final ScriptVariableGenerator scriptVariableGenerator; @@ -44,7 +47,7 @@ public abstract class AbstractScriptExecutingMessageProcessor implements Mess protected volatile BeanFactory beanFactory; protected AbstractScriptExecutingMessageProcessor() { - this.scriptVariableGenerator = new DefaultScriptVariableGenerator(); + this(new DefaultScriptVariableGenerator()); } protected AbstractScriptExecutingMessageProcessor(ScriptVariableGenerator scriptVariableGenerator) { @@ -61,10 +64,10 @@ public abstract class AbstractScriptExecutingMessageProcessor implements Mess try { ScriptSource source = this.getScriptSource(message); Map variables = this.scriptVariableGenerator.generateScriptVariables(message); - return this.executeScript(source, variables); + return executeScript(source, variables); } catch (Exception e) { - throw new MessageHandlingException(message, "failed to execute script", e); + throw IntegrationUtils.wrapInHandlingExceptionIfNecessary(message, () -> "Failed to execute script.", e); } } @@ -81,7 +84,6 @@ public abstract class AbstractScriptExecutingMessageProcessor implements Mess /** * Subclasses must implement this method to create a script source, * optionally using the message to locate or create the script. - * * @param message the message being processed * @return a ScriptSource to use to create a script */ @@ -90,8 +92,7 @@ public abstract class AbstractScriptExecutingMessageProcessor implements Mess /** * Subclasses must implement this method. In doing so, the execution context * for the script should be populated with the provided script variables. - * - * @param scriptSource The script source. + *78546 @param scriptSource The script source. * @param variables The variables. * @return The result of the execution. * @throws Exception Any Exception. diff --git a/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/inbound/StompInboundChannelAdapter.java b/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/inbound/StompInboundChannelAdapter.java index 590a064239..1d3919b363 100644 --- a/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/inbound/StompInboundChannelAdapter.java +++ b/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/inbound/StompInboundChannelAdapter.java @@ -17,6 +17,7 @@ package org.springframework.integration.stomp.inbound; import java.lang.reflect.Type; +import java.util.Arrays; import java.util.HashMap; import java.util.LinkedHashSet; import java.util.Map; @@ -33,13 +34,13 @@ import org.springframework.integration.stomp.StompSessionManager; import org.springframework.integration.stomp.event.StompReceiptEvent; import org.springframework.integration.stomp.support.StompHeaderMapper; import org.springframework.integration.support.management.IntegrationManagedResource; +import org.springframework.integration.support.utils.IntegrationUtils; import org.springframework.jmx.export.annotation.ManagedAttribute; import org.springframework.jmx.export.annotation.ManagedOperation; import org.springframework.jmx.export.annotation.ManagedResource; import org.springframework.lang.Nullable; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; -import org.springframework.messaging.MessageHandlingException; import org.springframework.messaging.simp.stomp.StompCommand; import org.springframework.messaging.simp.stomp.StompFrameHandler; import org.springframework.messaging.simp.stomp.StompHeaderAccessor; @@ -133,14 +134,14 @@ public class StompInboundChannelAdapter extends MessageProducerSupport implement Assert.notNull(destination, "'destination' cannot be null"); this.destinationLock.lock(); try { - for (String d : destination) { - if (this.destinations.add(d)) { - if (this.logger.isDebugEnabled()) { - logger.debug("Subscribe to destination '" + d + "'."); - } - subscribeDestination(d); - } - } + Arrays.stream(destination) + .filter(this.destinations::add) + .forEach(d -> { + if (this.logger.isDebugEnabled()) { + logger.debug("Subscribe to destination '" + d + "'."); + } + subscribeDestination(d); + }); } finally { this.destinationLock.unlock(); @@ -156,22 +157,22 @@ public class StompInboundChannelAdapter extends MessageProducerSupport implement Assert.notNull(destination, "'destination' cannot be null"); this.destinationLock.lock(); try { - for (String d : destination) { - if (this.destinations.remove(d)) { - if (this.logger.isDebugEnabled()) { - logger.debug("Removed '" + d + "' from subscriptions."); - } - StompSession.Subscription subscription = this.subscriptions.get(d); - if (subscription != null) { - subscription.unsubscribe(); - } - else { + Arrays.stream(destination) + .filter(this.destinations::remove) + .forEach(d -> { if (this.logger.isDebugEnabled()) { - logger.debug("No subscription for destination '" + d + "'."); + logger.debug("Removed '" + d + "' from subscriptions."); } - } - } - } + StompSession.Subscription subscription = this.subscriptions.get(d); + if (subscription != null) { + subscription.unsubscribe(); + } + else { + if (this.logger.isDebugEnabled()) { + logger.debug("No subscription for destination '" + d + "'."); + } + } + }); } finally { this.destinationLock.unlock(); @@ -227,12 +228,11 @@ public class StompInboundChannelAdapter extends MessageProducerSupport implement message = (Message) body; } else { - Map headersToCopy = - StompInboundChannelAdapter.this.headerMapper.toHeaders(headers); message = getMessageBuilderFactory() .withPayload(body) - .copyHeaders(headersToCopy) + .copyHeaders( + StompInboundChannelAdapter.this.headerMapper.toHeaders(headers)) .build(); } sendMessage(message); @@ -274,36 +274,32 @@ public class StompInboundChannelAdapter extends MessageProducerSupport implement @Override public void afterConnected(StompSession session, StompHeaders connectedHeaders) { StompInboundChannelAdapter.this.stompSession = session; - for (String destination : StompInboundChannelAdapter.this.destinations) { - subscribeDestination(destination); - } + StompInboundChannelAdapter.this.destinations.forEach(StompInboundChannelAdapter.this::subscribeDestination); } @Override - public void handleException(StompSession session, @Nullable StompCommand command, StompHeaders headers, - byte[] payload, Throwable exception) { + public void handleException(StompSession session, StompCommand command, StompHeaders headers, byte[] payload, + Throwable exception) { + + String exceptionMessage = "STOMP Frame handling error."; MessageChannel errorChannel = getErrorChannel(); + if (errorChannel != null) { - Message failedMessage; - // TODO 5.2 Copy all the STOMP headers for error message without any mapping - Map headersToCopy = StompInboundChannelAdapter.this.headerMapper.toHeaders(headers); - if (command != null) { - StompHeaderAccessor headerAccessor = StompHeaderAccessor.create(command); - headerAccessor.copyHeaders(headersToCopy); - failedMessage = MessageBuilder.createMessage(payload, headerAccessor.getMessageHeaders()); - } - else { - failedMessage = - MessageBuilder.withPayload(payload) - .copyHeaders(headersToCopy) - .build(); - } + StompHeaderAccessor headerAccessor = StompHeaderAccessor.create(command); + headerAccessor.copyHeaders(StompInboundChannelAdapter.this.headerMapper.toHeaders(headers)); + Message failedMessage = + MessageBuilder.createMessage(payload, headerAccessor.getMessageHeaders()); + + Exception ex = + IntegrationUtils.wrapInHandlingExceptionIfNecessary(failedMessage, + () -> exceptionMessage, exception); + getMessagingTemplate() - .send(errorChannel, new ErrorMessage(new MessageHandlingException(failedMessage, exception))); + .send(errorChannel, new ErrorMessage(ex)); } else { - logger.error("STOMP Frame handling error.", exception); + logger.error(exceptionMessage, exception); } } diff --git a/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/inbound/WebSocketInboundChannelAdapter.java b/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/inbound/WebSocketInboundChannelAdapter.java index e10f38517c..d258f0d7cb 100644 --- a/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/inbound/WebSocketInboundChannelAdapter.java +++ b/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/inbound/WebSocketInboundChannelAdapter.java @@ -1,5 +1,5 @@ /* - * Copyright 2014-2018 the original author or authors. + * Copyright 2014-2019 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. @@ -29,6 +29,7 @@ import org.springframework.context.Lifecycle; import org.springframework.integration.channel.FixedSubscriberChannel; import org.springframework.integration.endpoint.MessageProducerSupport; import org.springframework.integration.support.json.JacksonPresent; +import org.springframework.integration.support.utils.IntegrationUtils; import org.springframework.integration.websocket.IntegrationWebSocketContainer; import org.springframework.integration.websocket.ServerWebSocketContainer; import org.springframework.integration.websocket.WebSocketListener; @@ -37,7 +38,6 @@ import org.springframework.integration.websocket.support.PassThruSubProtocolHand import org.springframework.integration.websocket.support.SubProtocolHandlerRegistry; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; -import org.springframework.messaging.MessageHandlingException; import org.springframework.messaging.converter.ByteArrayMessageConverter; import org.springframework.messaging.converter.CompositeMessageConverter; import org.springframework.messaging.converter.DefaultContentTypeResolver; @@ -74,8 +74,6 @@ public class WebSocketInboundChannelAdapter extends MessageProducerSupport private final List defaultConverters = new ArrayList<>(3); - private ApplicationEventPublisher eventPublisher; - { this.defaultConverters.add(new StringMessageConverter()); this.defaultConverters.add(new ByteArrayMessageConverter()); @@ -98,18 +96,20 @@ public class WebSocketInboundChannelAdapter extends MessageProducerSupport private final MessageChannel subProtocolHandlerChannel; - private final AtomicReference> payloadType = new AtomicReference>(String.class); + private final AtomicReference> payloadType = new AtomicReference<>(String.class); - private volatile List messageConverters; + private ApplicationEventPublisher eventPublisher; - private volatile boolean mergeWithDefaultConverters = false; + private List messageConverters; - private volatile boolean active; + private boolean mergeWithDefaultConverters = false; - private volatile boolean useBroker; + private boolean useBroker; private AbstractBrokerMessageHandler brokerHandler; + private volatile boolean active; + public WebSocketInboundChannelAdapter(IntegrationWebSocketContainer webSocketContainer) { this(webSocketContainer, new SubProtocolHandlerRegistry(new PassThruSubProtocolHandler())); } @@ -121,14 +121,16 @@ public class WebSocketInboundChannelAdapter extends MessageProducerSupport this.webSocketContainer = webSocketContainer; this.server = this.webSocketContainer instanceof ServerWebSocketContainer; this.subProtocolHandlerRegistry = protocolHandlerRegistry; - this.subProtocolHandlerChannel = new FixedSubscriberChannel(message -> { - try { - handleMessageAndSend(message); - } - catch (Exception e) { - throw new MessageHandlingException(message, e); - } - }); + this.subProtocolHandlerChannel = + new FixedSubscriberChannel(message -> { + try { + handleMessageAndSend(message); + } + catch (Exception e) { + throw IntegrationUtils.wrapInHandlingExceptionIfNecessary(message, + () -> "Failed to handle and process message.", e); + } + }); } /** @@ -138,7 +140,7 @@ public class WebSocketInboundChannelAdapter extends MessageProducerSupport */ public void setMessageConverters(List messageConverters) { Assert.noNullElements(messageConverters.toArray(), "'messageConverters' must not contain null entries"); - this.messageConverters = new ArrayList(messageConverters); + this.messageConverters = new ArrayList<>(messageConverters); } @@ -347,7 +349,8 @@ public class WebSocketInboundChannelAdapter extends MessageProducerSupport this.subProtocolHandlerRegistry.findProtocolHandler(session).handleMessageToClient(session, ackMessage); } catch (Exception e) { - throw new MessageHandlingException(message, "Error sending connect ack message", e); + throw IntegrationUtils.wrapInHandlingExceptionIfNecessary(message, + () -> "Error sending connect ack message", e); } } diff --git a/spring-integration-xml/src/main/java/org/springframework/integration/xml/splitter/XPathMessageSplitter.java b/spring-integration-xml/src/main/java/org/springframework/integration/xml/splitter/XPathMessageSplitter.java index 0fed3cadcd..2524caf81b 100644 --- a/spring-integration-xml/src/main/java/org/springframework/integration/xml/splitter/XPathMessageSplitter.java +++ b/spring-integration-xml/src/main/java/org/springframework/integration/xml/splitter/XPathMessageSplitter.java @@ -41,11 +41,11 @@ import org.w3c.dom.Node; import org.w3c.dom.NodeList; import org.springframework.integration.splitter.AbstractMessageSplitter; +import org.springframework.integration.support.utils.IntegrationUtils; import org.springframework.integration.util.FunctionIterator; import org.springframework.integration.xml.DefaultXmlPayloadConverter; import org.springframework.integration.xml.XmlPayloadConverter; import org.springframework.messaging.Message; -import org.springframework.messaging.MessageHandlingException; import org.springframework.messaging.converter.MessageConversionException; import org.springframework.util.Assert; import org.springframework.xml.DocumentBuilderFactoryUtils; @@ -221,11 +221,12 @@ public class XPathMessageSplitter extends AbstractMessageSplitter { } return result; } - catch (ParserConfigurationException e) { - throw new MessageConversionException(message, "failed to create DocumentBuilder", e); + catch (ParserConfigurationException ex) { + throw new MessageConversionException(message, "failed to create DocumentBuilder", ex); } - catch (Exception e) { - throw new MessageHandlingException(message, "failed to split Message payload", e); + catch (Exception ex) { + throw IntegrationUtils.wrapInHandlingExceptionIfNecessary(message, + () -> "Failed to split Message payload", ex); } }