From d2e974a6de2cc1869dbce668593c939fece28c27 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Wed, 27 Feb 2019 15:17:51 -0500 Subject: [PATCH] Fix new Sonar smells (#2768) * Fix new Sonar smells * Fix some old Sonar smells as well * Fix Micrometer leaks in the `PollableChannel` when we register meters, but don't remove them. * * Fix NPE around `MetricsCaptor` in channels * * Fix new smells according test report * * Further Sonar smell fixes * * More smell fixes for `MessagingMethodInvokerHelper` * Remove `throws Exception` from `AbstractMessageHandler.destroy()` * * Fix complexity in the `MessagingMethodInvokerHelper.processInvokeExceptionAndFallbackToExpressionIfAny()` --- .../amqp/channel/AbstractAmqpChannel.java | 4 +- .../AbstractSubscribableAmqpChannel.java | 4 +- .../amqp/channel/PollableAmqpChannel.java | 138 ++++---- .../AbstractCorrelatingMessageHandler.java | 122 +++---- .../aggregator/BarrierMessageHandler.java | 18 +- .../ExpressionEvaluatingReleaseStrategy.java | 4 +- .../MethodInvokingMessageListProcessor.java | 26 +- .../channel/AbstractExecutorChannel.java | 21 +- .../channel/AbstractMessageChannel.java | 108 ++++--- .../channel/AbstractPollableChannel.java | 75 ++--- .../endpoint/MessageProducerSupport.java | 61 ++-- .../graph/IntegrationGraphServer.java | 56 ++-- .../handler/AbstractMessageHandler.java | 2 +- .../AbstractMessageProducingHandler.java | 96 +++--- .../MethodInvokingMessageProcessor.java | 21 +- .../support/MessagingMethodInvokerHelper.java | 275 +++++++++------- .../AbstractIntegrationMessageBuilder.java | 4 +- .../support/utils/IntegrationUtils.java | 4 +- .../util/AbstractExpressionEvaluator.java | 4 +- .../config/AggregatorParserTests.java | 3 +- .../MethodInvokingMessageProcessorTests.java | 33 +- ...ourceInboundChannelAdapterParserTests.java | 12 +- .../config/FileListFilterFactoryBean.java | 4 +- .../FileReadingMessageSourceFactoryBean.java | 41 ++- .../AbstractRemoteFileOutboundGateway.java | 298 +++++++++--------- .../ftp/gateway/FtpOutboundGateway.java | 4 +- .../ip/udp/UnicastSendingMessageHandler.java | 9 +- .../integration/jms/PollableJmsChannel.java | 70 ++-- .../jms/SubscribableJmsChannel.java | 4 +- ...ssionEvaluatingParameterSourceFactory.java | 22 +- .../channel/SubscribableRedisChannel.java | 15 +- .../test/matcher/MessageMatcher.java | 14 +- .../test/predicate/MessagePredicate.java | 6 +- 33 files changed, 828 insertions(+), 750 deletions(-) diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/AbstractAmqpChannel.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/AbstractAmqpChannel.java index 32f3b9c2b1..950c99841b 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/AbstractAmqpChannel.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/AbstractAmqpChannel.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. @@ -219,7 +219,7 @@ public abstract class AbstractAmqpChannel extends AbstractMessageChannel } @Override - public void destroy() throws Exception { + public void destroy() { if (this.connectionFactory != null) { this.connectionFactory.removeConnectionListener(this); this.initialized = false; diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/AbstractSubscribableAmqpChannel.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/AbstractSubscribableAmqpChannel.java index 6033c9f573..8c6b68925b 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/AbstractSubscribableAmqpChannel.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/AbstractSubscribableAmqpChannel.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. @@ -240,7 +240,7 @@ abstract class AbstractSubscribableAmqpChannel extends AbstractAmqpChannel } @Override - public void destroy() throws Exception { + public void destroy() { super.destroy(); if (this.container != null) { this.container.destroy(); diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PollableAmqpChannel.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PollableAmqpChannel.java index 9239337dc1..34100a27e4 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PollableAmqpChannel.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PollableAmqpChannel.java @@ -20,6 +20,7 @@ import java.util.ArrayDeque; import java.util.Deque; import java.util.List; import java.util.Map; +import java.util.concurrent.atomic.AtomicBoolean; import org.springframework.amqp.core.AmqpAdmin; import org.springframework.amqp.core.AmqpTemplate; @@ -31,6 +32,7 @@ import org.springframework.integration.amqp.support.AmqpHeaderMapper; import org.springframework.integration.channel.ExecutorChannelInterceptorAware; import org.springframework.integration.support.management.PollableChannelManagement; import org.springframework.integration.support.management.metrics.CounterFacade; +import org.springframework.integration.support.management.metrics.MetricsCaptor; import org.springframework.lang.Nullable; import org.springframework.messaging.Message; import org.springframework.messaging.PollableChannel; @@ -176,80 +178,43 @@ public class PollableAmqpChannel extends AbstractAmqpChannel return doReceive(timeout); } - + @Nullable protected Message doReceive(Long timeout) { ChannelInterceptorList interceptorList = getIChannelInterceptorList(); Deque interceptorStack = null; - boolean counted = false; + AtomicBoolean counted = new AtomicBoolean(); boolean countsEnabled = isCountsEnabled(); + boolean traceEnabled = isLoggingEnabled() && logger.isTraceEnabled(); try { - if (isLoggingEnabled() && logger.isTraceEnabled()) { + if (traceEnabled) { logger.trace("preReceive on channel '" + this + "'"); } if (interceptorList.getInterceptors().size() > 0) { interceptorStack = new ArrayDeque<>(); - if (!interceptorList.preReceive(this, interceptorStack)) { return null; } } Object object = performReceive(timeout); - Message message = null; - if (object == null) { - if (isLoggingEnabled() && logger.isTraceEnabled()) { - logger.trace("postReceive on channel '" + this + "', message is null"); - } - } - else { - if (countsEnabled) { - if (getMetricsCaptor() != null) { - incrementReceiveCounter(); - } - getMetrics().afterReceive(); - counted = true; - } - if (object instanceof Message) { - message = (Message) object; - } - else { - message = getMessageBuilderFactory() - .withPayload(object) - .build(); - } - if (isLoggingEnabled() && logger.isDebugEnabled()) { - logger.debug("postReceive on channel '" + this + "', message: " + message); - } - } + Message message = buildMessageFromResult(object, traceEnabled, countsEnabled ? counted : null); - if (interceptorStack != null) { - if (message != null) { - message = interceptorList.postReceive(message, this); - } - interceptorList.afterReceiveCompletion(message, this, null, interceptorStack); + + if (message != null) { + message = interceptorList.postReceive(message, this); } + interceptorList.afterReceiveCompletion(message, this, null, interceptorStack); return message; } - catch (RuntimeException e) { - if (countsEnabled && !counted) { - if (getMetricsCaptor() != null) { - getMetricsCaptor().counterBuilder(RECEIVE_COUNTER_NAME) - .tag("name", getComponentName() == null ? "unknown" : getComponentName()) - .tag("type", "channel") - .tag("result", "failure") - .tag("exception", e.getClass().getSimpleName()) - .description("Messages received") - .build() - .increment(); - } - getMetrics().afterError(); + catch (RuntimeException ex) { + if (countsEnabled && !counted.get()) { + incrementReceiveErrorCounter(ex); } - if (interceptorStack != null) { - interceptorList.afterReceiveCompletion(null, this, e, interceptorStack); - } - throw e; + interceptorList.afterReceiveCompletion(null, this, ex, interceptorStack); + throw ex; } } + @Nullable protected Object performReceive(Long timeout) { if (!this.declared) { doDeclares(); @@ -289,17 +254,63 @@ public class PollableAmqpChannel extends AbstractAmqpChannel } } - private void incrementReceiveCounter() { - if (this.receiveCounter == null) { - this.receiveCounter = getMetricsCaptor().counterBuilder(RECEIVE_COUNTER_NAME) - .tag("name", getComponentName()) - .tag("type", "channel") - .tag("result", "success") - .tag("exception", "none") - .description("Messages received") - .build(); + private Message buildMessageFromResult(@Nullable Object object, boolean traceEnabled, + @Nullable AtomicBoolean counted) { + + Message message = null; + if (object != null) { + if (counted != null) { + incrementReceiveCounter(); + getMetrics().afterReceive(); + counted.set(true); + } + if (object instanceof Message) { + message = (Message) object; + } + else { + message = getMessageBuilderFactory() + .withPayload(object) + .build(); + } } - this.receiveCounter.increment(); + + if (traceEnabled) { + logger.trace("postReceive on channel '" + this + + "', message" + (message != null ? ": " + message : " is null")); + } + + return message; + } + + private void incrementReceiveCounter() { + MetricsCaptor metricsCaptor = getMetricsCaptor(); + if (metricsCaptor != null) { + if (this.receiveCounter == null) { + this.receiveCounter = buildReceiveCounter(metricsCaptor, null); + } + this.receiveCounter.increment(); + } + } + + private void incrementReceiveErrorCounter(Exception ex) { + MetricsCaptor metricsCaptor = getMetricsCaptor(); + if (metricsCaptor != null) { + buildReceiveCounter(metricsCaptor, ex).increment(); + } + getMetrics().afterError(); + } + + private CounterFacade buildReceiveCounter(MetricsCaptor metricsCaptor, @Nullable Exception ex) { + CounterFacade counterFacade = metricsCaptor + .counterBuilder(RECEIVE_COUNTER_NAME) + .tag("name", getComponentName() == null ? "unknown" : getComponentName()) + .tag("type", "channel") + .tag("result", ex == null ? "success" : "failure") + .tag("exception", ex == null ? "none" : ex.getClass().getSimpleName()) + .description("Messages received") + .build(); + this.meters.add(counterFacade); + return counterFacade; } @@ -339,6 +350,7 @@ public class PollableAmqpChannel extends AbstractAmqpChannel } @Override + @Nullable public ChannelInterceptor removeInterceptor(int index) { ChannelInterceptor interceptor = super.removeInterceptor(index); if (interceptor instanceof ExecutorChannelInterceptor) { @@ -353,7 +365,7 @@ public class PollableAmqpChannel extends AbstractAmqpChannel } @Override - public void destroy() throws Exception { + public void destroy() { super.destroy(); if (this.receiveCounter != null) { this.receiveCounter.remove(); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java index 0333a56a46..04d8f04bc2 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java @@ -386,13 +386,10 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP @Override public MessageChannel getDiscardChannel() { - if (this.discardChannelName != null) { - synchronized (this) { - if (this.discardChannelName != null) { - this.discardChannel = getChannelResolver().resolveDestination(this.discardChannelName); - this.discardChannelName = null; - } - } + String channelName = this.discardChannelName; + if (channelName != null) { + this.discardChannel = getChannelResolver().resolveDestination(channelName); + this.discardChannelName = null; } return this.discardChannel; } @@ -449,48 +446,7 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP boolean noOutput = true; lock.lockInterruptibly(); try { - ScheduledFuture scheduledFuture = this.expireGroupScheduledFutures.remove(groupIdUuid); - if (scheduledFuture != null) { - boolean canceled = scheduledFuture.cancel(true); - if (canceled && this.logger.isDebugEnabled()) { - this.logger.debug("Cancel 'ScheduledFuture' for MessageGroup with Correlation Key [ " - + correlationKey + "]."); - } - } - MessageGroup messageGroup = this.messageStore.getMessageGroup(correlationKey); - if (this.sequenceAware) { - messageGroup = new SequenceAwareMessageGroup(messageGroup); - } - - if (!messageGroup.isComplete() && messageGroup.canAdd(message)) { - if (this.logger.isTraceEnabled()) { - this.logger.trace("Adding message to group [ " + messageGroup + "]"); - } - messageGroup = this.store(correlationKey, message); - - if (this.releaseStrategy.canRelease(messageGroup)) { - Collection> completedMessages = null; - try { - noOutput = false; - completedMessages = completeGroup(message, correlationKey, messageGroup, lock); - } - finally { - // Possible clean (implementation dependency) up - // even if there was an exception processing messages - afterRelease(messageGroup, completedMessages); - } - if (!isExpireGroupsUponCompletion() && this.minimumTimeoutForEmptyGroups > 0) { - removeEmptyGroupAfterTimeout(messageGroup, this.minimumTimeoutForEmptyGroups); - } - } - else { - scheduleGroupToForceComplete(messageGroup); - } - } - else { - noOutput = false; - discardMessage(message, lock); - } + noOutput = processMessageForGroup(message, correlationKey, groupIdUuid, lock); } finally { if (noOutput || !this.releaseLockBeforeSend) { @@ -499,6 +455,57 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP } } + private boolean processMessageForGroup(Message message, Object correlationKey, UUID groupIdUuid, Lock lock) { + boolean noOutput = true; + cancelScheduledFutureIfAny(correlationKey, groupIdUuid, true); + MessageGroup messageGroup = this.messageStore.getMessageGroup(correlationKey); + if (this.sequenceAware) { + messageGroup = new SequenceAwareMessageGroup(messageGroup); + } + + if (!messageGroup.isComplete() && messageGroup.canAdd(message)) { + if (this.logger.isTraceEnabled()) { + this.logger.trace("Adding message to group [ " + messageGroup + "]"); + } + messageGroup = store(correlationKey, message); + + if (this.releaseStrategy.canRelease(messageGroup)) { + Collection> completedMessages = null; + try { + noOutput = false; + completedMessages = completeGroup(message, correlationKey, messageGroup, lock); + } + finally { + // Possible clean (implementation dependency) up + // even if there was an exception processing messages + afterRelease(messageGroup, completedMessages); + } + if (!isExpireGroupsUponCompletion() && this.minimumTimeoutForEmptyGroups > 0) { + removeEmptyGroupAfterTimeout(messageGroup, this.minimumTimeoutForEmptyGroups); + } + } + else { + scheduleGroupToForceComplete(messageGroup); + } + } + else { + noOutput = false; + discardMessage(message, lock); + } + return noOutput; + } + + private void cancelScheduledFutureIfAny(Object correlationKey, UUID groupIdUuid, boolean mayInterruptIfRunning) { + ScheduledFuture scheduledFuture = this.expireGroupScheduledFutures.remove(groupIdUuid); + if (scheduledFuture != null) { + boolean canceled = scheduledFuture.cancel(mayInterruptIfRunning); + if (canceled && this.logger.isDebugEnabled()) { + this.logger.debug("Cancel 'ScheduledFuture' for MessageGroup with Correlation Key [ " + + correlationKey + "]."); + } + } + } + protected boolean isExpireGroupsUponCompletion() { return false; } @@ -606,7 +613,10 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP } private void discardMessage(Message message) { - this.messagingTemplate.send(getDiscardChannel(), message); + MessageChannel messageChannel = getDiscardChannel(); + if (messageChannel != null) { + this.messagingTemplate.send(messageChannel, message); + } } /** @@ -630,20 +640,14 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP protected void forceComplete(MessageGroup group) { Object correlationKey = group.getGroupId(); // UUIDConverter is no-op if already converted - Lock lock = this.lockRegistry.obtain(UUIDConverter.getUUID(correlationKey).toString()); + UUID groupId = UUIDConverter.getUUID(correlationKey); + Lock lock = this.lockRegistry.obtain(groupId.toString()); boolean removeGroup = true; boolean noOutput = true; try { lock.lockInterruptibly(); try { - ScheduledFuture scheduledFuture = - this.expireGroupScheduledFutures.remove(UUIDConverter.getUUID(correlationKey)); - if (scheduledFuture != null) { - boolean canceled = scheduledFuture.cancel(false); - if (canceled && this.logger.isDebugEnabled()) { - this.logger.debug("Cancel 'forceComplete' scheduling for MessageGroup [ " + group + "]."); - } - } + cancelScheduledFutureIfAny(correlationKey, groupId, false); MessageGroup groupNow = group; /* * If the group argument is not already complete, diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/BarrierMessageHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/BarrierMessageHandler.java index e303b5ed9e..bd5c955197 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/BarrierMessageHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/BarrierMessageHandler.java @@ -1,5 +1,5 @@ /* - * Copyright 2015-2018 the original author or authors. + * Copyright 2015-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. @@ -45,6 +45,7 @@ import org.springframework.util.Assert; * The default output processor is a {@link DefaultAggregatingMessageGroupProcessor}. * * @author Gary Russell + * @author Artem Bilan * * @since 4.2 */ @@ -61,10 +62,10 @@ public class BarrierMessageHandler extends AbstractReplyProducingMessageHandler private final MessageGroupProcessor messageGroupProcessor; - private volatile MessageChannel discardChannel; - private String discardChannelName; + private MessageChannel discardChannel; + /** * Construct an instance with the provided timeout and default correlation and * output strategies. @@ -135,8 +136,10 @@ public class BarrierMessageHandler extends AbstractReplyProducingMessageHandler */ @Override public MessageChannel getDiscardChannel() { - if (this.discardChannel == null && this.discardChannelName != null && getChannelResolver() != null) { - this.discardChannel = getChannelResolver().resolveDestination(this.discardChannelName); + String channelName = this.discardChannelName; + if (channelName != null) { + this.discardChannel = getChannelResolver().resolveDestination(channelName); + this.discardChannelName = null; } return this.discardChannel; } @@ -221,8 +224,9 @@ public class BarrierMessageHandler extends AbstractReplyProducingMessageHandler if (!syncQueue.offer(message, this.timeout, TimeUnit.MILLISECONDS)) { this.logger.error("Suspending thread timed out or did not arrive within timeout for: " + message); this.suspensions.remove(key); - if (getDiscardChannel() != null) { - this.messagingTemplate.send(getDiscardChannel(), message); + MessageChannel messageChannel = getDiscardChannel(); + if (messageChannel != null) { + this.messagingTemplate.send(messageChannel, message); } } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/ExpressionEvaluatingReleaseStrategy.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/ExpressionEvaluatingReleaseStrategy.java index 60fc6b8680..08ab14de38 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/ExpressionEvaluatingReleaseStrategy.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/ExpressionEvaluatingReleaseStrategy.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. @@ -46,7 +46,7 @@ public class ExpressionEvaluatingReleaseStrategy extends AbstractExpressionEvalu * and return the result (must be boolean). */ public boolean canRelease(MessageGroup messages) { - return evaluateExpression(this.expression, messages, Boolean.class); + return Boolean.TRUE.equals(evaluateExpression(this.expression, messages, Boolean.class)); } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/MethodInvokingMessageListProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/MethodInvokingMessageListProcessor.java index 1bfaceea92..def45847dd 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/MethodInvokingMessageListProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/MethodInvokingMessageListProcessor.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,32 +34,33 @@ import org.springframework.messaging.Message; * @author Dave Syer * @author Artem Bilan * @author Gary Russell + * * @since 2.0 */ public class MethodInvokingMessageListProcessor extends AbstractExpressionEvaluator implements Lifecycle { - private final MessagingMethodInvokerHelper delegate; + private final MessagingMethodInvokerHelper delegate; public MethodInvokingMessageListProcessor(Object targetObject, Method method, Class expectedType) { - this.delegate = new MessagingMethodInvokerHelper(targetObject, method, expectedType, true); + this.delegate = new MessagingMethodInvokerHelper(targetObject, method, expectedType, true); } public MethodInvokingMessageListProcessor(Object targetObject, Method method) { - this.delegate = new MessagingMethodInvokerHelper(targetObject, method, true); + this.delegate = new MessagingMethodInvokerHelper(targetObject, method, true); } public MethodInvokingMessageListProcessor(Object targetObject, String methodName, Class expectedType) { - this.delegate = new MessagingMethodInvokerHelper(targetObject, methodName, + this.delegate = new MessagingMethodInvokerHelper(targetObject, methodName, expectedType, true); } public MethodInvokingMessageListProcessor(Object targetObject, String methodName) { - this.delegate = new MessagingMethodInvokerHelper(targetObject, methodName, true); + this.delegate = new MessagingMethodInvokerHelper(targetObject, methodName, true); } public MethodInvokingMessageListProcessor(Object targetObject, Class annotationType) { - this.delegate = new MessagingMethodInvokerHelper(targetObject, annotationType, Object.class, true); + this.delegate = new MessagingMethodInvokerHelper(targetObject, annotationType, Object.class, true); } @Override @@ -84,16 +85,9 @@ public class MethodInvokingMessageListProcessor extends AbstractExpressionEva return this.delegate.toString(); } + @SuppressWarnings("unchecked") public T process(Collection> messages, Map aggregateHeaders) { - try { - return this.delegate.process(messages, aggregateHeaders); - } - catch (RuntimeException e) { - throw e; - } - catch (Exception e) { - throw new IllegalStateException("Failed to process message list", e); - } + return (T) this.delegate.process(messages, aggregateHeaders); } @Override diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractExecutorChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractExecutorChannel.java index 0023355b7a..165c8c1bab 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractExecutorChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractExecutorChannel.java @@ -1,5 +1,5 @@ /* - * Copyright 2015-2018 the original author or authors. + * Copyright 2015-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. @@ -46,20 +46,23 @@ import org.springframework.util.CollectionUtils; * * @author Artem Bilan * @author Gary Russell + * + * @since 4.2 + * * @see ExecutorChannel * @see PublishSubscribeChannel - * @since 4.2 + * */ public abstract class AbstractExecutorChannel extends AbstractSubscribableChannel implements ExecutorChannelInterceptorAware { - protected volatile Executor executor; + protected volatile Executor executor; // NOSONAR - protected volatile AbstractDispatcher dispatcher; + protected volatile AbstractDispatcher dispatcher; // NOSONAR - protected volatile Integer maxSubscribers; + protected volatile Integer maxSubscribers; // NOSONAR - protected volatile int executorInterceptorsSize; + protected volatile int executorInterceptorsSize; // NOSONAR public AbstractExecutorChannel(@Nullable Executor executor) { this.executor = executor; @@ -68,7 +71,6 @@ public abstract class AbstractExecutorChannel extends AbstractSubscribableChanne /** * Specify the maximum number of subscribers supported by the * channel's dispatcher. - * * @param maxSubscribers The maximum number of subscribers allowed. */ public void setMaxSubscribers(int maxSubscribers) { @@ -112,6 +114,7 @@ public abstract class AbstractExecutorChannel extends AbstractSubscribableChanne } @Override + @Nullable public ChannelInterceptor removeInterceptor(int index) { ChannelInterceptor interceptor = super.removeInterceptor(index); if (interceptor instanceof ExecutorChannelInterceptor) { @@ -141,7 +144,7 @@ public abstract class AbstractExecutorChannel extends AbstractSubscribableChanne Deque interceptorStack = null; try { if (AbstractExecutorChannel.this.executorInterceptorsSize > 0) { - interceptorStack = new ArrayDeque(); + interceptorStack = new ArrayDeque<>(); message = applyBeforeHandle(message, interceptorStack); if (message == null) { return; @@ -156,7 +159,7 @@ public abstract class AbstractExecutorChannel extends AbstractSubscribableChanne if (!CollectionUtils.isEmpty(interceptorStack)) { triggerAfterMessageHandled(message, ex, interceptorStack); } - if (ex instanceof MessagingException) { + if (ex instanceof MessagingException) { // NOSONAR throw new MessagingExceptionWrapper(message, (MessagingException) ex); } String description = "Failed to handle " + message + " to " + this + " in " + messageHandler; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java index 7faa0fc73f..e75946fc29 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java @@ -28,6 +28,7 @@ import java.util.concurrent.CopyOnWriteArrayList; import org.apache.commons.logging.Log; +import org.springframework.beans.factory.BeanFactory; import org.springframework.core.OrderComparator; import org.springframework.integration.context.IntegrationContextUtils; import org.springframework.integration.context.IntegrationObjectSupport; @@ -71,7 +72,7 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport implements MessageChannel, TrackableComponent, ChannelInterceptorAware, MessageChannelMetrics, ConfigurableMetricsAware { - protected final ChannelInterceptorList interceptors; + protected final ChannelInterceptorList interceptors; // NOSONAR private final Comparator orderComparator = new OrderComparator(); @@ -120,6 +121,7 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport this.metricsCaptor = metricsCaptor; } + @Nullable protected MetricsCaptor getMetricsCaptor() { return this.metricsCaptor; } @@ -356,19 +358,20 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport protected void onInit() { super.onInit(); if (this.messageConverter == null) { - if (getBeanFactory() != null) { - if (getBeanFactory().containsBean( - IntegrationContextUtils.INTEGRATION_DATATYPE_CHANNEL_MESSAGE_CONVERTER_BEAN_NAME)) { - this.messageConverter = this.getBeanFactory().getBean( - IntegrationContextUtils.INTEGRATION_DATATYPE_CHANNEL_MESSAGE_CONVERTER_BEAN_NAME, - MessageConverter.class); - } + BeanFactory beanFactory = getBeanFactory(); + if (beanFactory != null && + beanFactory.containsBean( + IntegrationContextUtils.INTEGRATION_DATATYPE_CHANNEL_MESSAGE_CONVERTER_BEAN_NAME)) { + + this.messageConverter = + beanFactory.getBean( + IntegrationContextUtils.INTEGRATION_DATATYPE_CHANNEL_MESSAGE_CONVERTER_BEAN_NAME, + MessageConverter.class); } } if (this.statsEnabled) { this.channelMetrics.setFullStatsEnabled(true); } - this.fullChannelName = null; } @@ -420,7 +423,7 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport Assert.notNull(messageArg.getPayload(), "message payload must not be null"); Message message = messageArg; if (this.shouldTrack) { - message = MessageHistory.write(message, this, this.getMessageBuilderFactory()); + message = MessageHistory.write(message, this, getMessageBuilderFactory()); } Deque interceptorStack = null; @@ -432,9 +435,7 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport AbstractMessageChannelMetrics metrics = this.channelMetrics; SampleFacade sample = null; try { - if (this.datatypes.length > 0) { - message = this.convertPayloadIfNecessary(message); - } + message = convertPayloadIfNecessary(message); boolean debugEnabled = this.loggingEnabled && logger.isDebugEnabled(); if (debugEnabled) { logger.debug("preSend on channel '" + this + "', message: " + message); @@ -471,18 +472,18 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport } return sent; } - catch (Exception e) { + catch (Exception ex) { if (countsAreEnabled && !metricsProcessed) { if (sample != null) { - sample.stop(buildSendTimer(false, e.getClass().getSimpleName())); + sample.stop(buildSendTimer(false, ex.getClass().getSimpleName())); } metrics.afterSend(metricsContext, false); } if (interceptorStack != null) { - interceptorList.afterSendCompletion(message, this, sent, e, interceptorStack); + interceptorList.afterSendCompletion(message, this, sent, ex, interceptorStack); } throw IntegrationUtils.wrapInDeliveryExceptionIfNecessary(message, - () -> "failed to send Message to channel '" + this.getComponentName() + "'", e); + () -> "failed to send Message to channel '" + this.getComponentName() + "'", ex); } } @@ -514,33 +515,38 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport } private Message convertPayloadIfNecessary(Message message) { - // first pass checks if the payload type already matches any of the datatypes - for (Class datatype : this.datatypes) { - if (datatype.isAssignableFrom(message.getPayload().getClass())) { - return message; - } - } - if (this.messageConverter != null) { - // second pass applies conversion if possible, attempting datatypes in order + if (this.datatypes.length > 0) { + // first pass checks if the payload type already matches any of the datatypes for (Class datatype : this.datatypes) { - Object converted = this.messageConverter.fromMessage(message, datatype); - if (converted != null) { - if (converted instanceof Message) { - return (Message) converted; - } - else { - return getMessageBuilderFactory() - .withPayload(converted) - .copyHeaders(message.getHeaders()) - .build(); + if (datatype.isAssignableFrom(message.getPayload().getClass())) { + return message; + } + } + if (this.messageConverter != null) { + // second pass applies conversion if possible, attempting datatypes in order + for (Class datatype : this.datatypes) { + Object converted = this.messageConverter.fromMessage(message, datatype); + if (converted != null) { + if (converted instanceof Message) { + return (Message) converted; + } + else { + return getMessageBuilderFactory() + .withPayload(converted) + .copyHeaders(message.getHeaders()) + .build(); + } } } } + throw new MessageDeliveryException(message, "Channel '" + this.getComponentName() + + "' expected one of the following data types [" + + StringUtils.arrayToCommaDelimitedString(this.datatypes) + + "], but received [" + message.getPayload().getClass() + "]"); + } + else { + return message; } - throw new MessageDeliveryException(message, "Channel '" + this.getComponentName() + - "' expected one of the following datataypes [" + - StringUtils.arrayToCommaDelimitedString(this.datatypes) + - "], but received [" + message.getPayload().getClass() + "]"); } /** @@ -556,7 +562,7 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport protected abstract boolean doSend(Message message, long timeout); @Override - public void destroy() throws Exception { // NOSONAR TODO: remove throws in 5.2 + public void destroy() { this.meters.forEach(MeterFacade::remove); this.meters.clear(); } @@ -566,9 +572,9 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport */ protected static class ChannelInterceptorList { - private final Log logger; + protected final List interceptors = new CopyOnWriteArrayList<>(); // NOSONAR - protected final List interceptors = new CopyOnWriteArrayList(); + private final Log logger; private int size; @@ -671,15 +677,17 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport } public void afterReceiveCompletion(@Nullable Message message, MessageChannel channel, - @Nullable Exception ex, Deque interceptorStack) { + @Nullable Exception ex, @Nullable Deque interceptorStack) { - for (Iterator iterator = interceptorStack.descendingIterator(); iterator.hasNext(); ) { - ChannelInterceptor interceptor = iterator.next(); - try { - interceptor.afterReceiveCompletion(message, channel, ex); - } - catch (Exception ex2) { - this.logger.error("Exception from afterReceiveCompletion in " + interceptor, ex2); + if (interceptorStack != null) { + for (Iterator iterator = interceptorStack.descendingIterator(); iterator.hasNext(); ) { + ChannelInterceptor interceptor = iterator.next(); + try { + interceptor.afterReceiveCompletion(message, channel, ex); + } + catch (Exception ex2) { + this.logger.error("Exception from afterReceiveCompletion in " + interceptor, ex2); + } } } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractPollableChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractPollableChannel.java index fbef9ab5f0..e97accd755 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractPollableChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractPollableChannel.java @@ -22,6 +22,7 @@ import java.util.List; import org.springframework.integration.support.management.PollableChannelManagement; import org.springframework.integration.support.management.metrics.CounterFacade; +import org.springframework.integration.support.management.metrics.MetricsCaptor; import org.springframework.lang.Nullable; import org.springframework.messaging.Message; import org.springframework.messaging.PollableChannel; @@ -115,60 +116,61 @@ public abstract class AbstractPollableChannel extends AbstractMessageChannel } else { if (countsEnabled) { - if (getMetricsCaptor() != null) { - incrementReceiveCounter(); - } + incrementReceiveCounter(); getMetrics().afterReceive(); counted = true; } if (isLoggingEnabled() && logger.isDebugEnabled()) { - logger.debug("postReceive on channel '" + this + "', message: " + message); + logger.debug("postReceive on channel '" + this + "', message: " + message); } } - if (interceptorStack != null) { - if (message != null) { - message = interceptorList.postReceive(message, this); - } - interceptorList.afterReceiveCompletion(message, this, null, interceptorStack); + if (interceptorStack != null && message != null) { + message = interceptorList.postReceive(message, this); } + interceptorList.afterReceiveCompletion(message, this, null, interceptorStack); return message; } - catch (RuntimeException e) { + catch (RuntimeException ex) { if (countsEnabled && !counted) { - if (getMetricsCaptor() != null) { - CounterFacade counter = getMetricsCaptor().counterBuilder(RECEIVE_COUNTER_NAME) - .tag("name", getComponentName() == null ? "unknown" : getComponentName()) - .tag("type", "channel") - .tag("result", "failure") - .tag("exception", e.getClass().getSimpleName()) - .description("Messages received") - .build(); - this.meters.add(counter); - counter.increment(); - } - getMetrics().afterError(); + incrementReceiveErrorCounter(ex); } - if (interceptorStack != null) { - interceptorList.afterReceiveCompletion(null, this, e, interceptorStack); - } - throw e; + interceptorList.afterReceiveCompletion(null, this, ex, interceptorStack); + throw ex; } } private void incrementReceiveCounter() { - if (this.receiveCounter == null) { - this.receiveCounter = getMetricsCaptor().counterBuilder(RECEIVE_COUNTER_NAME) - .tag("name", getComponentName()) - .tag("type", "channel") - .tag("result", "success") - .tag("exception", "none") - .description("Messages received") - .build(); + MetricsCaptor metricsCaptor = getMetricsCaptor(); + if (metricsCaptor != null) { + if (this.receiveCounter == null) { + this.receiveCounter = buildReceiveCounter(metricsCaptor, null); + } + this.receiveCounter.increment(); } - this.receiveCounter.increment(); + } + + private void incrementReceiveErrorCounter(Exception ex) { + MetricsCaptor metricsCaptor = getMetricsCaptor(); + if (metricsCaptor != null) { + buildReceiveCounter(metricsCaptor, ex).increment(); + } + getMetrics().afterError(); + } + + private CounterFacade buildReceiveCounter(MetricsCaptor metricsCaptor, @Nullable Exception ex) { + CounterFacade counterFacade = metricsCaptor + .counterBuilder(RECEIVE_COUNTER_NAME) + .tag("name", getComponentName() == null ? "unknown" : getComponentName()) + .tag("type", "channel") + .tag("result", ex == null ? "success" : "failure") + .tag("exception", ex == null ? "none" : ex.getClass().getSimpleName()) + .description("Messages received") + .build(); + this.meters.add(counterFacade); + return counterFacade; } @Override @@ -207,6 +209,7 @@ public abstract class AbstractPollableChannel extends AbstractMessageChannel } @Override + @Nullable public ChannelInterceptor removeInterceptor(int index) { ChannelInterceptor interceptor = super.removeInterceptor(index); if (interceptor instanceof ExecutorChannelInterceptor) { @@ -233,7 +236,7 @@ public abstract class AbstractPollableChannel extends AbstractMessageChannel protected abstract Message doReceive(long timeout); @Override - public void destroy() throws Exception { // NOSONAR TODO: remove throws in 5.2 + public void destroy() { super.destroy(); if (this.receiveCounter != null) { this.receiveCounter.remove(); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageProducerSupport.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageProducerSupport.java index 19320c0604..0350d8180c 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageProducerSupport.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageProducerSupport.java @@ -16,7 +16,7 @@ package org.springframework.integration.endpoint; -import org.springframework.beans.factory.BeanInitializationException; +import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.SmartInitializingSingleton; import org.springframework.core.AttributeAccessor; import org.springframework.integration.core.MessageProducer; @@ -26,6 +26,7 @@ import org.springframework.integration.support.DefaultErrorMessageStrategy; import org.springframework.integration.support.ErrorMessageStrategy; import org.springframework.integration.support.ErrorMessageUtils; import org.springframework.integration.support.management.TrackableComponent; +import org.springframework.lang.Nullable; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessagingException; @@ -48,15 +49,15 @@ public abstract class MessageProducerSupport extends AbstractEndpoint implements private ErrorMessageStrategy errorMessageStrategy = new DefaultErrorMessageStrategy(); - private volatile MessageChannel outputChannel; + private MessageChannel outputChannel; - private volatile String outputChannelName; + private String outputChannelName; - private volatile MessageChannel errorChannel; + private MessageChannel errorChannel; - private volatile String errorChannelName; + private String errorChannelName; - private volatile boolean shouldTrack = false; + private boolean shouldTrack = false; protected MessageProducerSupport() { this.setPhase(Integer.MAX_VALUE / 2); @@ -81,13 +82,10 @@ public abstract class MessageProducerSupport extends AbstractEndpoint implements @Override public MessageChannel getOutputChannel() { - if (this.outputChannelName != null) { - synchronized (this) { - if (this.outputChannelName != null) { - this.outputChannel = getChannelResolver().resolveDestination(this.outputChannelName); - this.outputChannelName = null; - } - } + String channelName = this.outputChannelName; + if (channelName != null) { + this.outputChannel = getChannelResolver().resolveDestination(channelName); + this.outputChannelName = null; } return this.outputChannel; } @@ -114,14 +112,12 @@ public abstract class MessageProducerSupport extends AbstractEndpoint implements * @return the channel or null. * @since 4.3 */ + @Nullable public MessageChannel getErrorChannel() { - if (this.errorChannelName != null) { - synchronized (this) { - if (this.errorChannelName != null) { - this.errorChannel = getChannelResolver().resolveDestination(this.errorChannelName); - this.errorChannelName = null; - } - } + String channelName = this.errorChannelName; + if (channelName != null) { + this.errorChannel = getChannelResolver().resolveDestination(channelName); + this.errorChannelName = null; } return this.errorChannel; } @@ -164,15 +160,10 @@ public abstract class MessageProducerSupport extends AbstractEndpoint implements @Override protected void onInit() { - try { - super.onInit(); - } - catch (Exception e) { - throw new BeanInitializationException("Cannot initialize: " + this, e); - } - - if (this.getBeanFactory() != null) { - this.messagingTemplate.setBeanFactory(this.getBeanFactory()); + super.onInit(); + BeanFactory beanFactory = getBeanFactory(); + if (beanFactory != null) { + this.messagingTemplate.setBeanFactory(beanFactory); } } @@ -199,14 +190,16 @@ public abstract class MessageProducerSupport extends AbstractEndpoint implements throw new MessagingException("cannot send a null message"); } if (this.shouldTrack) { - message = MessageHistory.write(message, this, this.getMessageBuilderFactory()); + message = MessageHistory.write(message, this, getMessageBuilderFactory()); } try { - this.messagingTemplate.send(getOutputChannel(), message); + MessageChannel messageChannel = getOutputChannel(); + Assert.state(messageChannel != null, "The 'outputChannel' or `outputChannelName` must be configured"); + this.messagingTemplate.send(messageChannel, message); } - catch (RuntimeException e) { - if (!sendErrorMessageIfNecessary(message, e)) { - throw e; + catch (RuntimeException ex) { + if (!sendErrorMessageIfNecessary(message, ex)) { + throw ex; } } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/graph/IntegrationGraphServer.java b/spring-integration-core/src/main/java/org/springframework/integration/graph/IntegrationGraphServer.java index 1842d36d7e..82f8b7be0b 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/graph/IntegrationGraphServer.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/graph/IntegrationGraphServer.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. @@ -24,6 +24,7 @@ import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.Map.Entry; +import java.util.Objects; import java.util.concurrent.atomic.AtomicInteger; import java.util.function.Function; import java.util.stream.Collectors; @@ -155,7 +156,7 @@ public class IntegrationGraphServer implements ApplicationContextAware, Applicat if (implementationVersion == null) { implementationVersion = "unknown - is Spring Integration running from the distribution jar?"; } - Map descriptor = new HashMap(); + Map descriptor = new HashMap<>(); descriptor.put("provider", "spring-integration"); descriptor.put("providerVersion", implementationVersion); descriptor.put("providerFormatVersion", GRAPH_VERSION); @@ -351,61 +352,54 @@ public class IntegrationGraphServer implements ApplicationContextAware, Applicat } private MessageGatewayNode gatewayNode(String name, MessagingGatewaySupport gateway) { - MessageChannel gwErrorChannel = gateway.getErrorChannel(); - String errorChannel = gwErrorChannel != null ? gwErrorChannel.toString() : null; - MessageChannel gwRequestChannel = gateway.getRequestChannel(); - String requestChannel = gwRequestChannel != null ? gwRequestChannel.toString() : null; - return new MessageGatewayNode(this.nodeId.incrementAndGet(), name, gateway, - requestChannel, errorChannel); + String errorChannel = Objects.toString(gateway.getErrorChannel(), null); + String requestChannel = Objects.toString(gateway.getRequestChannel(), null); + return new MessageGatewayNode(this.nodeId.incrementAndGet(), name, gateway, requestChannel, errorChannel); } private MessageProducerNode producerNode(String name, MessageProducerSupport producer) { - String errorChannel = producer.getErrorChannel() != null ? producer.getErrorChannel().toString() : null; - String outputChannel = producer.getOutputChannel() != null ? producer.getOutputChannel().toString() : null; + String errorChannel = Objects.toString(producer.getErrorChannel(), null); + String outputChannel = Objects.toString(producer.getOutputChannel(), null); return new MessageProducerNode(this.nodeId.incrementAndGet(), name, producer, outputChannel, errorChannel); } private MessageSourceNode sourceNode(String name, SourcePollingChannelAdapter adapter) { - String errorChannel = adapter.getDefaultErrorChannel() != null - ? adapter.getDefaultErrorChannel().toString() : null; - String outputChannel = adapter.getOutputChannel() != null ? adapter.getOutputChannel().toString() : null; + String errorChannel = Objects.toString(adapter.getDefaultErrorChannel(), null); + String outputChannel = Objects.toString(adapter.getOutputChannel(), null); return new MessageSourceNode(this.nodeId.incrementAndGet(), name, adapter.getMessageSource(), outputChannel, errorChannel); } private MessageHandlerNode handlerNode(String name, IntegrationConsumer consumer) { - MessageChannel outputChannel = consumer.getOutputChannel(); - String outputChannelName = outputChannel == null ? null : outputChannel.toString(); + String outputChannelName = Objects.toString(consumer.getOutputChannel(), null); MessageHandler handler = consumer.getHandler(); if (handler instanceof CompositeMessageHandler) { return compositeHandler(name, consumer, (CompositeMessageHandler) handler, outputChannelName, null, - false); + false); } else if (handler instanceof DiscardingMessageHandler) { return discardingHandler(name, consumer, (DiscardingMessageHandler) handler, outputChannelName, null, - false); + false); } else if (handler instanceof MappingMessageRouterManagement) { return routingHandler(name, consumer, handler, (MappingMessageRouterManagement) handler, - outputChannelName, null, false); + outputChannelName, null, false); } else if (handler instanceof RecipientListRouterManagement) { return recipientListRoutingHandler(name, consumer, handler, (RecipientListRouterManagement) handler, - outputChannelName, null, false); + outputChannelName, null, false); } else { - String inputChannel = consumer.getInputChannel() != null ? consumer.getInputChannel().toString() : null; + String inputChannel = Objects.toString(consumer.getInputChannel(), null); return new MessageHandlerNode(this.nodeId.incrementAndGet(), name, handler, - inputChannel, outputChannelName); + inputChannel, outputChannelName); } } private MessageHandlerNode polledHandlerNode(String name, PollingConsumer consumer) { - MessageChannel outputChannel = consumer.getOutputChannel(); - String outputChannelName = outputChannel == null ? null : outputChannel.toString(); - String errorChannel = consumer.getDefaultErrorChannel() != null - ? consumer.getDefaultErrorChannel().toString() : null; + String outputChannelName = Objects.toString(consumer.getOutputChannel(), null); + String errorChannel = Objects.toString(consumer.getDefaultErrorChannel(), null); MessageHandler handler = consumer.getHandler(); if (handler instanceof CompositeMessageHandler) { return compositeHandler(name, consumer, (CompositeMessageHandler) handler, outputChannelName, @@ -424,7 +418,7 @@ public class IntegrationGraphServer implements ApplicationContextAware, Applicat outputChannelName, errorChannel, true); } else { - String inputChannel = consumer.getInputChannel() != null ? consumer.getInputChannel().toString() : null; + String inputChannel = Objects.toString(consumer.getInputChannel(), null); return new ErrorCapableMessageHandlerNode(this.nodeId.incrementAndGet(), name, handler, inputChannel, outputChannelName, errorChannel); } @@ -444,7 +438,7 @@ public class IntegrationGraphServer implements ApplicationContextAware, Applicat named.getComponentType())) .collect(Collectors.toList()); - String inputChannel = consumer.getInputChannel() != null ? consumer.getInputChannel().toString() : null; + String inputChannel = Objects.toString(consumer.getInputChannel(), null); return polled ? new ErrorCapableCompositeMessageHandlerNode(this.nodeId.incrementAndGet(), name, handler, inputChannel, output, errors, innerHandlers) @@ -455,8 +449,8 @@ public class IntegrationGraphServer implements ApplicationContextAware, Applicat private MessageHandlerNode discardingHandler(String name, IntegrationConsumer consumer, DiscardingMessageHandler handler, String output, String errors, boolean polled) { - String discards = handler.getDiscardChannel() != null ? handler.getDiscardChannel().toString() : null; - String inputChannel = consumer.getInputChannel() != null ? consumer.getInputChannel().toString() : null; + String discards = Objects.toString(handler.getDiscardChannel(), null); + String inputChannel = Objects.toString(consumer.getInputChannel(), null); return polled ? new ErrorCapableDiscardingMessageHandlerNode(this.nodeId.incrementAndGet(), name, handler, inputChannel, output, discards, errors) @@ -472,7 +466,7 @@ public class IntegrationGraphServer implements ApplicationContextAware, Applicat router.getDynamicChannelNames().stream()) .collect(Collectors.toList()); - String inputChannel = consumer.getInputChannel() != null ? consumer.getInputChannel().toString() : null; + String inputChannel = Objects.toString(consumer.getInputChannel(), null); return polled ? new ErrorCapableRoutingNode(this.nodeId.incrementAndGet(), name, handler, inputChannel, output, errors, routes) @@ -490,7 +484,7 @@ public class IntegrationGraphServer implements ApplicationContextAware, Applicat .map(recipient -> ((Recipient) recipient).getChannel().toString()) .collect(Collectors.toList()); - String inputChannel = consumer.getInputChannel() != null ? consumer.getInputChannel().toString() : null; + String inputChannel = Objects.toString(consumer.getInputChannel(), null); return polled ? new ErrorCapableRoutingNode(this.nodeId.incrementAndGet(), name, handler, inputChannel, output, errors, routes) 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 5296b06d3d..f2268cf143 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 @@ -338,7 +338,7 @@ public abstract class AbstractMessageHandler extends IntegrationObjectSupport } @Override - public void destroy() throws Exception { + public void destroy() { this.timers.forEach(MeterFacade::remove); } 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 cff14db384..97fa93e6b9 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 @@ -33,6 +33,7 @@ import org.springframework.integration.core.MessageProducer; import org.springframework.integration.core.MessagingTemplate; import org.springframework.integration.routingslip.RoutingSlipRouteStrategy; import org.springframework.integration.support.AbstractIntegrationMessageBuilder; +import org.springframework.integration.support.utils.IntegrationUtils; import org.springframework.lang.Nullable; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; @@ -64,7 +65,7 @@ import reactor.core.publisher.Mono; public abstract class AbstractMessageProducingHandler extends AbstractMessageHandler implements MessageProducer, HeaderPropagationAware { - protected final MessagingTemplate messagingTemplate = new MessagingTemplate(); + protected final MessagingTemplate messagingTemplate = new MessagingTemplate(); // NOSONAR final private boolean async; @@ -209,8 +210,9 @@ public abstract class AbstractMessageProducingHandler extends AbstractMessageHan @Override @Nullable public MessageChannel getOutputChannel() { - if (this.outputChannelName != null) { - this.outputChannel = getChannelResolver().resolveDestination(this.outputChannelName); + String channelName = this.outputChannelName; + if (channelName != null) { + this.outputChannel = getChannelResolver().resolveDestination(channelName); this.outputChannelName = null; } return this.outputChannel; @@ -272,12 +274,13 @@ public abstract class AbstractMessageProducingHandler extends AbstractMessageHan Object replyChannel) { if (this.async && (reply instanceof ListenableFuture || reply instanceof Publisher)) { + MessageChannel messageChannel = getOutputChannel(); if (reply instanceof ListenableFuture || - !(getOutputChannel() instanceof ReactiveStreamsSubscribableChannel)) { - asyncNonReactiveReply(requestMessage, requestHeaders, reply, replyChannel); + !(messageChannel instanceof ReactiveStreamsSubscribableChannel)) { + asyncNonReactiveReply(requestMessage, reply, replyChannel); } else { - ((ReactiveStreamsSubscribableChannel) getOutputChannel()) + ((ReactiveStreamsSubscribableChannel) messageChannel) .subscribeTo( Flux.from((Publisher) reply) .map(result -> createOutputMessage(result, requestHeaders))); @@ -307,8 +310,8 @@ public abstract class AbstractMessageProducingHandler extends AbstractMessageHan return builder; } - private void asyncNonReactiveReply(final Message requestMessage, final MessageHeaders requestHeaders, - Object reply, Object replyChannel) { + private void asyncNonReactiveReply(Message requestMessage, Object reply, Object replyChannel) { + ListenableFuture future; if (reply instanceof ListenableFuture) { future = (ListenableFuture) reply; @@ -322,35 +325,7 @@ public abstract class AbstractMessageProducingHandler extends AbstractMessageHan future = settableListenableFuture; } - Object theReplyChannel = replyChannel; - future.addCallback(new ListenableFutureCallback() { - - @Override - public void onSuccess(Object result) { - Message replyMessage = null; - try { - replyMessage = createOutputMessage(result, requestHeaders); - sendOutput(replyMessage, theReplyChannel, false); - } - catch (Exception e) { - Exception exceptionToLogAndSend = e; - if (!(e instanceof MessagingException)) { - exceptionToLogAndSend = new MessageHandlingException(requestMessage, e); - if (replyMessage != null) { - exceptionToLogAndSend = new MessagingException(replyMessage, exceptionToLogAndSend); - } - } - logger.error("Failed to send async reply: " + result.toString(), exceptionToLogAndSend); - onFailure(exceptionToLogAndSend); - } - } - - @Override - public void onFailure(Throwable ex) { - sendErrorMessage(requestMessage, ex); - } - - }); + future.addCallback(new ReplyFutureCallback(requestMessage, replyChannel)); } private Object getOutputChannelFromRoutingSlip(Object reply, Message requestMessage, List routingSlip, @@ -458,7 +433,7 @@ public abstract class AbstractMessageProducingHandler extends AbstractMessageHan return true; } - protected void sendErrorMessage(final Message requestMessage, Throwable ex) { + protected void sendErrorMessage(Message requestMessage, Throwable ex) { Object errorChannel = resolveErrorChannel(requestMessage.getHeaders()); Throwable result = ex; if (!(ex instanceof MessagingException)) { @@ -473,10 +448,8 @@ public abstract class AbstractMessageProducingHandler extends AbstractMessageHan sendOutput(new ErrorMessage(result), errorChannel, true); } catch (Exception e) { - Exception exceptionToLog = e; - if (!(e instanceof MessagingException)) { - exceptionToLog = new MessageHandlingException(requestMessage, e); - } + Exception exceptionToLog = + IntegrationUtils.wrapInHandlingExceptionIfNecessary(requestMessage, () -> null, e); logger.error("Failed to send async reply", exceptionToLog); } } @@ -495,4 +468,43 @@ public abstract class AbstractMessageProducingHandler extends AbstractMessageHan return errorChannel; } + private final class ReplyFutureCallback implements ListenableFutureCallback { + + private final Message requestMessage; + + private final Object replyChannel; + + ReplyFutureCallback(Message requestMessage, Object replyChannel) { + this.requestMessage = requestMessage; + this.replyChannel = replyChannel; + } + + + @Override + public void onSuccess(Object result) { + Message replyMessage = null; + try { + replyMessage = createOutputMessage(result, this.requestMessage.getHeaders()); + sendOutput(replyMessage, this.replyChannel, false); + } + catch (Exception ex) { + Exception exceptionToLogAndSend = ex; + if (!(ex instanceof MessagingException)) { // NOSONAR + exceptionToLogAndSend = new MessageHandlingException(this.requestMessage, ex); + if (replyMessage != null) { + exceptionToLogAndSend = new MessagingException(replyMessage, exceptionToLogAndSend); + } + } + logger.error("Failed to send async reply: " + result.toString(), exceptionToLogAndSend); + onFailure(exceptionToLogAndSend); + } + } + + @Override + public void onFailure(Throwable ex) { + sendErrorMessage(this.requestMessage, ex); + } + + } + } 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 b0d3f29664..12f8117e98 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 @@ -44,22 +44,22 @@ import org.springframework.messaging.Message; */ public class MethodInvokingMessageProcessor extends AbstractMessageProcessor implements Lifecycle { - private final MessagingMethodInvokerHelper delegate; + private final MessagingMethodInvokerHelper delegate; public MethodInvokingMessageProcessor(Object targetObject, Method method) { - this.delegate = new MessagingMethodInvokerHelper(targetObject, method, false); + this.delegate = new MessagingMethodInvokerHelper(targetObject, method, false); } public MethodInvokingMessageProcessor(Object targetObject, String methodName) { - this.delegate = new MessagingMethodInvokerHelper(targetObject, methodName, false); + this.delegate = new MessagingMethodInvokerHelper(targetObject, methodName, false); } public MethodInvokingMessageProcessor(Object targetObject, String methodName, boolean canProcessMessageList) { - this.delegate = new MessagingMethodInvokerHelper(targetObject, methodName, canProcessMessageList); + this.delegate = new MessagingMethodInvokerHelper(targetObject, methodName, canProcessMessageList); } public MethodInvokingMessageProcessor(Object targetObject, Class annotationType) { - this.delegate = new MessagingMethodInvokerHelper(targetObject, annotationType, false); + this.delegate = new MessagingMethodInvokerHelper(targetObject, annotationType, false); } @Override @@ -69,7 +69,7 @@ public class MethodInvokingMessageProcessor extends AbstractMessageProcessor< } @Override - public void setBeanFactory(@Nullable BeanFactory beanFactory) { + public void setBeanFactory(BeanFactory beanFactory) { super.setBeanFactory(beanFactory); this.delegate.setBeanFactory(beanFactory); } @@ -102,14 +102,15 @@ public class MethodInvokingMessageProcessor extends AbstractMessageProcessor< @Override @Nullable + @SuppressWarnings("unchecked") public T processMessage(Message message) { try { - return this.delegate.process(message); + return (T) this.delegate.process(message); } - catch (Exception e) { + catch (Exception ex) { throw IntegrationUtils.wrapInHandlingExceptionIfNecessary(message, - () -> "error occurred during processing message in 'MethodInvokingMessageProcessor' [" + this + "]", - e); + () -> "error occurred during processing message in 'MethodInvokingMessageProcessor' [" + this + + "]", ex); } } 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 c0ee267f30..e91c09c219 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 @@ -124,13 +124,13 @@ import org.springframework.util.StringUtils; * * @since 2.0 */ -public class MessagingMethodInvokerHelper extends AbstractExpressionEvaluator implements Lifecycle { +public class MessagingMethodInvokerHelper extends AbstractExpressionEvaluator implements Lifecycle { private static final String CANDIDATE_METHODS = "CANDIDATE_METHODS"; private static final String CANDIDATE_MESSAGE_METHODS = "CANDIDATE_MESSAGE_METHODS"; - private static final Log logger = LogFactory.getLog(MessagingMethodInvokerHelper.class); + private static final Log LOGGER = LogFactory.getLog(MessagingMethodInvokerHelper.class); // Number of times to try an InvocableHandlerMethod before giving up in favor of an expression. private static final int FAILED_ATTEMPTS_THRESHOLD = 100; @@ -151,16 +151,12 @@ public class MessagingMethodInvokerHelper extends AbstractExpressionEvaluator private static final Map SPEL_COMPILERS = new HashMap<>(); - private static final TypeDescriptor messageTypeDescriptor = TypeDescriptor.valueOf(Message.class); + private static final TypeDescriptor MESSAGE_TYPE_DESCRIPTOR = TypeDescriptor.valueOf(Message.class); - @SuppressWarnings("unused") - private static final Collection> dummyMessages = Collections.emptyList(); + private static final TypeDescriptor MESSAGE_LIST_TYPE_DESCRIPTOR = + TypeDescriptor.collection(Collection.class, TypeDescriptor.valueOf(Message.class)); - private static final TypeDescriptor messageListTypeDescriptor = - new TypeDescriptor(ReflectionUtils.findField(MessagingMethodInvokerHelper.class, // NOSONAR never null - "dummyMessages")); - - private static final TypeDescriptor messageArrayTypeDescriptor = TypeDescriptor.valueOf(Message[].class); + private static final TypeDescriptor MESSAGE_ARRAY_TYPE_DESCRIPTOR = TypeDescriptor.valueOf(Message[].class); static { SPEL_COMPILERS.put(SpelCompilerMode.OFF, EXPRESSION_PARSER_OFF); @@ -175,38 +171,38 @@ public class MessagingMethodInvokerHelper extends AbstractExpressionEvaluator private final JsonObjectMapper jsonObjectMapper; - private volatile String displayString; - - private volatile boolean requiresReply; - private final Map, HandlerMethod> handlerMethods; private final Map, HandlerMethod> handlerMessageMethods; private final List, HandlerMethod>> handlerMethodsList; - private HandlerMethod handlerMethod; - private final TypeDescriptor expectedType; private final boolean canProcessMessageList; - private Class annotationType; + private HandlerMethod handlerMethod; - private volatile boolean initialized; + private Class annotationType; private String methodName; private Method method; - private boolean useSpelInvoker; - private HandlerMethod defaultHandlerMethod; private BeanExpressionResolver resolver = new StandardBeanExpressionResolver(); private BeanExpressionContext expressionContext; + private volatile String displayString; + + private volatile boolean requiresReply; + + private volatile boolean initialized; + + private boolean useSpelInvoker; + public MessagingMethodInvokerHelper(Object targetObject, Method method, Class expectedType, boolean canProcessMessageList) { @@ -309,13 +305,13 @@ public class MessagingMethodInvokerHelper extends AbstractExpressionEvaluator } @Nullable - public T process(Message message) throws Exception { + public Object process(Message message) { ParametersWrapper parameters = new ParametersWrapper(message); return processInternal(parameters); } @Nullable - public T process(Collection> messages, Map headers) throws Exception { + public Object process(Collection> messages, Map headers) { ParametersWrapper parameters = new ParametersWrapper(messages, headers); return processInternal(parameters); } @@ -431,7 +427,7 @@ public class MessagingMethodInvokerHelper extends AbstractExpressionEvaluator this.displayString = sb.toString() + "]"; } - private void prepareEvaluationContext() throws Exception { + private void prepareEvaluationContext() { StandardEvaluationContext context = getEvaluationContext(false); Class targetType = AopUtils.getTargetClass(this.targetObject); if (this.method != null) { @@ -450,8 +446,13 @@ public class MessagingMethodInvokerHelper extends AbstractExpressionEvaluator context.registerMethodFilter(targetType, filter); } context.setVariable("target", this.targetObject); - context.registerFunction("requiredHeader", ParametersWrapper.class.getDeclaredMethod("getHeader", - Map.class, String.class)); + try { + context.registerFunction("requiredHeader", + ParametersWrapper.class.getDeclaredMethod("getHeader", Map.class, String.class)); + } + catch (NoSuchMethodException ex) { + throw new IllegalStateException(ex); + } } private boolean canReturnExpectedType(AnnotatedMethodFilter filter, Class targetType, @@ -469,9 +470,8 @@ public class MessagingMethodInvokerHelper extends AbstractExpressionEvaluator return false; } - @SuppressWarnings("unchecked") @Nullable - private T processInternal(ParametersWrapper parameters) throws Exception { + private Object processInternal(ParametersWrapper parameters) { if (!this.initialized) { initialize(); } @@ -485,7 +485,7 @@ public class MessagingMethodInvokerHelper extends AbstractExpressionEvaluator } Expression expression = candidate.expression; - T result; + Object result; if (this.useSpelInvoker || candidate.spelOnly) { result = invokeExpression(expression, parameters); } @@ -494,7 +494,7 @@ public class MessagingMethodInvokerHelper extends AbstractExpressionEvaluator } if (result != null && this.expectedType != null) { - return (T) getEvaluationContext(true) + return getEvaluationContext(true) .getTypeConverter() .convertValue(result, TypeDescriptor.forObject(result), this.expectedType); } @@ -520,11 +520,11 @@ public class MessagingMethodInvokerHelper extends AbstractExpressionEvaluator } @SuppressWarnings("deprecation") - private synchronized void initialize() throws Exception { + private synchronized void initialize() { if (!this.initialized) { BeanFactory beanFactory = getBeanFactory(); if (isProvidedMessageHandlerFactoryBean()) { - logger.info("Overriding default instance of MessageHandlerMethodFactory with provided one."); + LOGGER.info("Overriding default instance of MessageHandlerMethodFactory with provided one."); this.messageHandlerMethodFactory = beanFactory.getBean(IntegrationContextUtils.MESSAGE_HANDLER_FACTORY_BEAN_NAME, MessageHandlerMethodFactory.class); @@ -589,13 +589,19 @@ public class MessagingMethodInvokerHelper extends AbstractExpressionEvaluator } NullAwarePayloadArgumentResolver nullResolver = new NullAwarePayloadArgumentResolver(messageConverter); PayloadExpressionArgumentResolver payloadExpressionArgumentResolver = new PayloadExpressionArgumentResolver(); - payloadExpressionArgumentResolver.setBeanFactory(beanFactory); + if (beanFactory != null) { + payloadExpressionArgumentResolver.setBeanFactory(beanFactory); + } PayloadsArgumentResolver payloadsArgumentResolver = new PayloadsArgumentResolver(); - payloadsArgumentResolver.setBeanFactory(beanFactory); + if (beanFactory != null) { + payloadsArgumentResolver.setBeanFactory(beanFactory); + } MapArgumentResolver mapArgumentResolver = new MapArgumentResolver(); - mapArgumentResolver.setBeanFactory(beanFactory); + if (beanFactory != null) { + mapArgumentResolver.setBeanFactory(beanFactory); + } List customArgumentResolvers = new LinkedList<>(); customArgumentResolvers.add(payloadExpressionArgumentResolver); @@ -604,7 +610,9 @@ public class MessagingMethodInvokerHelper extends AbstractExpressionEvaluator if (this.canProcessMessageList) { CollectionArgumentResolver collectionArgumentResolver = new CollectionArgumentResolver(true); - collectionArgumentResolver.setBeanFactory(beanFactory); + if (beanFactory != null) { + collectionArgumentResolver.setBeanFactory(beanFactory); + } customArgumentResolvers.add(collectionArgumentResolver); } @@ -614,67 +622,83 @@ public class MessagingMethodInvokerHelper extends AbstractExpressionEvaluator .setCustomArgumentResolvers(customArgumentResolvers); } - @SuppressWarnings("unchecked") - private T invokeHandlerMethod(HandlerMethod handlerMethod, ParametersWrapper parameters) throws Exception { + private Object invokeHandlerMethod(HandlerMethod handlerMethod, ParametersWrapper parameters) { try { - return (T) handlerMethod.invoke(parameters); + return handlerMethod.invoke(parameters); } - catch (MethodArgumentResolutionException | MessageConversionException | IllegalStateException e) { - if (e instanceof MessageConversionException) { - if (e.getCause() instanceof ConversionFailedException && - !(e.getCause().getCause() instanceof ConverterNotFoundException)) { - throw e; - } - } - else if (e instanceof IllegalStateException) { - if (!(e.getCause() instanceof IllegalArgumentException) || - !e.getStackTrace()[0].getClassName().equals(InvocableHandlerMethod.class.getName()) || - (!"argument type mismatch".equals(e.getCause().getMessage()) && - // JVM generates GeneratedMethodAccessor### after several calls with less error - // checking - !e.getCause().getMessage().startsWith("java.lang.ClassCastException@"))) { - throw e; - } - } - - Expression expression = handlerMethod.expression; - - if (++handlerMethod.failedAttempts >= FAILED_ATTEMPTS_THRESHOLD) { - handlerMethod.spelOnly = true; - if (logger.isInfoEnabled()) { - logger.info("Failed to invoke [ " + handlerMethod.invocableHandlerMethod + - "] with provided arguments [ " + parameters + " ]. \n" + - "Falling back to SpEL invocation for expression [ " + - expression.getExpressionString() + " ]"); - } - } - - return invokeExpression(expression, parameters); + catch (MethodArgumentResolutionException | MessageConversionException | IllegalStateException ex) { + return processInvokeExceptionAndFallbackToExpressionIfAny(handlerMethod, parameters, ex); + } + catch (RuntimeException ex) { // NOSONAR no way to handle conditional catch according Sonar rules + throw ex; + } + catch (Exception ex) { + throw new IllegalStateException("HandlerMethod invocation error", ex); } } - @SuppressWarnings("unchecked") - private T invokeExpression(Expression expression, ParametersWrapper parameters) throws Exception { + private Object processInvokeExceptionAndFallbackToExpressionIfAny(HandlerMethod handlerMethod, + ParametersWrapper parameters, RuntimeException ex) { + + if (ex instanceof MessageConversionException) { + if (ex.getCause() instanceof ConversionFailedException && + !(ex.getCause().getCause() instanceof ConverterNotFoundException)) { + throw ex; + } + } + else if (ex instanceof IllegalStateException && // NOSONAR complex boolean expression + (!(ex.getCause() instanceof IllegalArgumentException) || + !ex.getStackTrace()[0].getClassName().equals(InvocableHandlerMethod.class.getName()) || + (!"argument type mismatch".equals(ex.getCause().getMessage()) && + // JVM generates GeneratedMethodAccessor### after several calls with less error + // checking + !ex.getCause().getMessage().startsWith("java.lang.ClassCastException@")))) { + throw ex; + } + + return fallbackToInvokeExpression(handlerMethod, parameters); + } + + private Object fallbackToInvokeExpression(HandlerMethod handlerMethod, ParametersWrapper parameters) { + Expression expression = handlerMethod.expression; + + if (++handlerMethod.failedAttempts >= FAILED_ATTEMPTS_THRESHOLD) { + handlerMethod.spelOnly = true; + if (LOGGER.isInfoEnabled()) { + LOGGER.info("Failed to invoke [ " + handlerMethod.invocableHandlerMethod + + "] with provided arguments [ " + parameters + " ]. \n" + + "Falling back to SpEL invocation for expression [ " + + expression.getExpressionString() + " ]"); + } + } + + return invokeExpression(expression, parameters); + } + + private Object invokeExpression(Expression expression, ParametersWrapper parameters) { try { convertJsonPayloadIfNecessary(parameters); - return (T) evaluateExpression(expression, parameters); + return evaluateExpression(expression, parameters); } - catch (Exception e) { - Throwable evaluationException = e; - if ((e instanceof EvaluationException || e instanceof MessageHandlingException) - && e.getCause() != null) { - evaluationException = e.getCause(); - } - if (evaluationException instanceof Exception) { - throw (Exception) evaluationException; - } - else { - throw new IllegalStateException("Cannot process message", evaluationException); - } + catch (Exception ex) { + throw processEvaluationException(ex); } } + private RuntimeException processEvaluationException(Exception ex) { + Throwable evaluationException = ex; + if ((ex instanceof EvaluationException || ex instanceof MessageHandlingException) + && ex.getCause() != null) { + evaluationException = ex.getCause(); + } + if (evaluationException instanceof RuntimeException) { + return (RuntimeException) evaluationException; + } + return new IllegalStateException("Cannot process message", evaluationException); + } + + /* * If there's a single method, it is SpEL only, the content is JSON, * the payload is a String or byte[], the parameter doesn't match the payload, @@ -687,33 +711,36 @@ public class MessagingMethodInvokerHelper extends AbstractExpressionEvaluator this.jsonObjectMapper != null) { Class type = this.handlerMethod.targetParameterType; - if ((parameters.getPayload() instanceof String && !type.equals(String.class) + if ((parameters.getPayload() instanceof String && !type.equals(String.class) // NOSONAR || parameters.getPayload() instanceof byte[] && !type.equals(byte[].class)) && contentTypeIsJson(parameters.message)) { - try { - Object targetPayload = this.jsonObjectMapper.fromJson(parameters.getPayload(), type); - - if (this.handlerMethod.targetParameterTypeDescriptor.isAssignableTo(messageTypeDescriptor)) { - parameters.message = - getMessageBuilderFactory() - .withPayload(targetPayload) - .copyHeaders(parameters.getHeaders()) - .build(); - } - else { - parameters.payload = targetPayload; - } - - - } - catch (Exception e) { - logger.debug("Failed to convert from JSON", e); - } + doConvertJsonPayload(parameters); } } } + private void doConvertJsonPayload(ParametersWrapper parameters) { + try { + Object targetPayload = + this.jsonObjectMapper.fromJson(parameters.getPayload(), this.handlerMethod.targetParameterType); + + if (this.handlerMethod.targetParameterTypeDescriptor.isAssignableTo(MESSAGE_TYPE_DESCRIPTOR)) { + parameters.message = + getMessageBuilderFactory() + .withPayload(targetPayload) + .copyHeaders(parameters.getHeaders()) + .build(); + } + else { + parameters.payload = targetPayload; + } + } + catch (Exception e) { + LOGGER.debug("Failed to convert from JSON", e); + } + } + private boolean contentTypeIsJson(Message message) { Object contentType = message.getHeaders().get(MessageHeaders.CONTENT_TYPE); return contentType != null && contentType.toString().contains("json"); @@ -789,15 +816,15 @@ public class MessagingMethodInvokerHelper extends AbstractExpressionEvaluator checkSpelInvokerRequired(targetClass, method1, handlerMethod1); } catch (IneligibleMethodException e) { - if (logger.isDebugEnabled()) { - logger.debug("Method [" + method1 + "] is not eligible for Message handling " + if (LOGGER.isDebugEnabled()) { + LOGGER.debug("Method [" + method1 + "] is not eligible for Message handling " + e.getMessage() + "."); } return; } catch (Exception e) { - if (logger.isDebugEnabled()) { - logger.debug("Method [" + method1 + "] is not eligible for Message handling.", e); + if (LOGGER.isDebugEnabled()) { + LOGGER.debug("Method [" + method1 + "] is not eligible for Message handling.", e); } return; } @@ -877,8 +904,8 @@ public class MessagingMethodInvokerHelper extends AbstractExpressionEvaluator try { if ("org.springframework.integration.gateway.RequestReplyExchanger".equals(iface.getName())) { frameworkMethods.add(targetClass.getMethod("exchange", Message.class)); - if (logger.isDebugEnabled()) { - logger.debug(targetObject.getClass() + + if (LOGGER.isDebugEnabled()) { + LOGGER.debug(targetObject.getClass() + ": Ambiguous fallback methods; using RequestReplyExchanger.exchange()"); } } @@ -996,8 +1023,9 @@ public class MessagingMethodInvokerHelper extends AbstractExpressionEvaluator } private String resolve(String value) { - if (getBeanFactory() != null && getBeanFactory() instanceof ConfigurableBeanFactory) { - return ((ConfigurableBeanFactory) getBeanFactory()).resolveEmbeddedValue(value); + BeanFactory beanFactory = getBeanFactory(); + if (beanFactory instanceof ConfigurableBeanFactory) { + return ((ConfigurableBeanFactory) beanFactory).resolveEmbeddedValue(value); } return value; } @@ -1015,7 +1043,7 @@ public class MessagingMethodInvokerHelper extends AbstractExpressionEvaluator } } catch (Exception e) { - logger.debug("Exception trying to extract interface", e); + LOGGER.debug("Exception trying to extract interface", e); } } } @@ -1066,7 +1094,7 @@ public class MessagingMethodInvokerHelper extends AbstractExpressionEvaluator } private static boolean isMethodDefinedOnObjectClass(Method method) { - return method != null && + return method != null && // NOSONAR (method.getDeclaringClass().equals(Object.class) || ReflectionUtils.isEqualsMethod(method) || ReflectionUtils.isHashCodeMethod(method) || ReflectionUtils.isToStringMethod(method) || AopUtils.isFinalizeMethod(method) || (method.getName().equals("clone") @@ -1113,13 +1141,20 @@ public class MessagingMethodInvokerHelper extends AbstractExpressionEvaluator } - @SuppressWarnings("unchecked") - public T invoke(ParametersWrapper parameters) throws Exception { + public Object invoke(ParametersWrapper parameters) { Message message = parameters.getMessage(); if (this.canProcessMessageList) { message = new MutableMessage<>(parameters.getMessages(), parameters.getHeaders()); } - return (T) this.invocableHandlerMethod.invoke(message); + try { + return this.invocableHandlerMethod.invoke(message); + } + catch (RuntimeException ex) { // NOSONAR no way to handle conditional catch according Sonar rules + throw ex; + } + catch (Exception ex) { + throw new IllegalStateException("InvocableHandlerMethod invoke error", ex); + } } Class getTargetParameterType() { @@ -1191,14 +1226,14 @@ public class MessagingMethodInvokerHelper extends AbstractExpressionEvaluator sb.append(this.determineHeaderExpression(mappingAnnotation, methodParameter)); } } - else if (parameterTypeDescriptor.isAssignableTo(messageTypeDescriptor)) { + else if (parameterTypeDescriptor.isAssignableTo(MESSAGE_TYPE_DESCRIPTOR)) { this.messageMethod = true; sb.append("message"); this.setExclusiveTargetParameterType(parameterTypeDescriptor, methodParameter); } else if (this.canProcessMessageList && - (parameterTypeDescriptor.isAssignableTo(messageListTypeDescriptor) - || parameterTypeDescriptor.isAssignableTo(messageArrayTypeDescriptor))) { + (parameterTypeDescriptor.isAssignableTo(MESSAGE_LIST_TYPE_DESCRIPTOR) + || parameterTypeDescriptor.isAssignableTo(MESSAGE_ARRAY_TYPE_DESCRIPTOR))) { sb.append("messages"); this.setExclusiveTargetParameterType(parameterTypeDescriptor, methodParameter); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/AbstractIntegrationMessageBuilder.java b/spring-integration-core/src/main/java/org/springframework/integration/support/AbstractIntegrationMessageBuilder.java index f4b0d40123..c296212697 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/support/AbstractIntegrationMessageBuilder.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/AbstractIntegrationMessageBuilder.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. @@ -150,7 +150,7 @@ public abstract class AbstractIntegrationMessageBuilder { * @see #copyHeadersIfAbsent(Map) */ public AbstractIntegrationMessageBuilder filterAndCopyHeadersIfAbsent(Map headersToCopy, - String... headerPatternsToFilter) { + @Nullable String... headerPatternsToFilter) { Map headers = headersToCopy; 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 825affd66a..4a8c3cafc4 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 @@ -184,9 +184,11 @@ public final class IntegrationUtils { 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); + runtimeException = new MessageHandlingException(message, text.get(), + (ex instanceof IllegalStateException && ex.getCause() != null) ? ex.getCause() : 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 e7118bc60e..4f04775f7f 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 @@ -47,7 +47,7 @@ import org.springframework.messaging.Message; */ public abstract class AbstractExpressionEvaluator implements BeanFactoryAware, InitializingBean { - protected final Log logger = LogFactory.getLog(this.getClass()); + protected final Log logger = LogFactory.getLog(this.getClass()); // NOSONAR final protected static final ExpressionParser EXPRESSION_PARSER = new SpelExpressionParser(); @@ -131,7 +131,7 @@ public abstract class AbstractExpressionEvaluator implements BeanFactoryAware, I catch (Exception ex) { this.logger.debug("SpEL Expression evaluation failed with Exception.", ex); Throwable cause = null; - if (ex instanceof EvaluationException) { + if (ex instanceof EvaluationException) { // NOSONAR cause = ex.getCause(); } throw IntegrationUtils.wrapInHandlingExceptionIfNecessary(message, diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/AggregatorParserTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/AggregatorParserTests.java index 4a3309ddd3..0ac6da90d6 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/AggregatorParserTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/AggregatorParserTests.java @@ -237,14 +237,13 @@ public class AggregatorParserTests { } @Test - @SuppressWarnings("unchecked") public void testAggregatorWithPojoReleaseStrategy() { MessageChannel input = this.context.getBean("aggregatorWithPojoReleaseStrategyInput", MessageChannel.class); EventDrivenConsumer endpoint = this.context.getBean("aggregatorWithPojoReleaseStrategy", EventDrivenConsumer.class); ReleaseStrategy releaseStrategy = TestUtils.getPropertyValue(endpoint, "handler.releaseStrategy", ReleaseStrategy.class); assertThat(releaseStrategy instanceof MethodInvokingReleaseStrategy).isTrue(); - MessagingMethodInvokerHelper methodInvokerHelper = + MessagingMethodInvokerHelper methodInvokerHelper = TestUtils.getPropertyValue(releaseStrategy, "adapter.delegate", MessagingMethodInvokerHelper.class); Object handlerMethods = TestUtils.getPropertyValue(methodInvokerHelper, "handlerMethods"); assertThat(handlerMethods).isNull(); 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 46c3c784aa..94b5f7f6a4 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 @@ -446,7 +446,7 @@ public class MethodInvokingMessageProcessorTests { processor.setBeanFactory(mock(BeanFactory.class)); assertThatExceptionOfType(MessageHandlingException.class) .isThrownBy(() -> processor.processMessage(new GenericMessage<>("foo"))) - .withCauseInstanceOf(CheckedException.class); + .withRootCauseInstanceOf(CheckedException.class); } @Test @@ -896,8 +896,7 @@ public class MethodInvokingMessageProcessorTests { processor.processMessage(new GenericMessage<>("foo")); } catch (Exception e) { - assertThat(e.getCause()).isInstanceOf(IllegalStateException.class); - assertThat(e.getCause().getCause()).isInstanceOf(IllegalArgumentException.class); + assertThat(e.getCause()).isInstanceOf(IllegalArgumentException.class); assertThat(e.getCause().getStackTrace()[0].getClassName()).isEqualTo(A.class.getName()); } @@ -976,8 +975,8 @@ public class MethodInvokingMessageProcessorTests { @Test public void testUseSpelInvoker() throws Exception { UseSpelInvokerBean bean = new UseSpelInvokerBean(); - MessagingMethodInvokerHelper helper = - new MessagingMethodInvokerHelper<>(bean, + MessagingMethodInvokerHelper helper = + new MessagingMethodInvokerHelper(bean, UseSpelInvokerBean.class.getDeclaredMethod("foo", String.class), false); helper.setBeanFactory(mock(BeanFactory.class)); Message message = new GenericMessage<>("Test"); @@ -985,28 +984,28 @@ public class MethodInvokingMessageProcessorTests { assertThat(TestUtils.getPropertyValue(helper, "handlerMethod.expression.configuration.compilerMode")) .isEqualTo(SpelCompilerMode.OFF); - helper = new MessagingMethodInvokerHelper<>(bean, + helper = new MessagingMethodInvokerHelper(bean, UseSpelInvokerBean.class.getDeclaredMethod("bar", String.class), false); helper.setBeanFactory(mock(BeanFactory.class)); helper.process(message); assertThat(TestUtils.getPropertyValue(helper, "handlerMethod.expression.configuration.compilerMode")) .isEqualTo(SpelCompilerMode.IMMEDIATE); - helper = new MessagingMethodInvokerHelper<>(bean, + helper = new MessagingMethodInvokerHelper(bean, UseSpelInvokerBean.class.getDeclaredMethod("baz", String.class), false); helper.setBeanFactory(mock(BeanFactory.class)); helper.process(message); assertThat(TestUtils.getPropertyValue(helper, "handlerMethod.expression.configuration.compilerMode")) .isEqualTo(SpelCompilerMode.MIXED); - helper = new MessagingMethodInvokerHelper<>(bean, + helper = new MessagingMethodInvokerHelper(bean, UseSpelInvokerBean.class.getDeclaredMethod("qux", String.class), false); helper.setBeanFactory(mock(BeanFactory.class)); helper.process(message); assertThat(TestUtils.getPropertyValue(helper, "handlerMethod.expression.configuration.compilerMode")) .isEqualTo(SpelCompilerMode.OFF); - helper = new MessagingMethodInvokerHelper<>(bean, + helper = new MessagingMethodInvokerHelper(bean, UseSpelInvokerBean.class.getDeclaredMethod("fiz", String.class), false); helper.setBeanFactory(mock(BeanFactory.class)); try { @@ -1017,7 +1016,7 @@ public class MethodInvokingMessageProcessorTests { .isEqualTo("No enum constant org.springframework.expression.spel.SpelCompilerMode.JUNK"); } - helper = new MessagingMethodInvokerHelper<>(bean, + helper = new MessagingMethodInvokerHelper(bean, UseSpelInvokerBean.class.getDeclaredMethod("buz", String.class), false); ConfigurableListableBeanFactory bf = mock(ConfigurableListableBeanFactory.class); willAnswer(returnsFirstArg()).given(bf).resolveEmbeddedValue(anyString()); @@ -1031,13 +1030,13 @@ public class MethodInvokingMessageProcessorTests { } // Check other CTORs - helper = new MessagingMethodInvokerHelper<>(bean, "bar", false); + helper = new MessagingMethodInvokerHelper(bean, "bar", false); helper.setBeanFactory(mock(BeanFactory.class)); helper.process(message); assertThat(TestUtils.getPropertyValue(helper, "handlerMethod.expression.configuration.compilerMode")) .isEqualTo(SpelCompilerMode.IMMEDIATE); - helper = new MessagingMethodInvokerHelper<>(bean, ServiceActivator.class, false); + helper = new MessagingMethodInvokerHelper(bean, ServiceActivator.class, false); helper.setBeanFactory(mock(BeanFactory.class)); helper.process(message); assertThat(TestUtils.getPropertyValue(helper, "handlerMethod.expression.configuration.compilerMode")) @@ -1047,8 +1046,8 @@ public class MethodInvokingMessageProcessorTests { @Test public void testSingleMethodJson() throws Exception { SingleMethodJsonWithSpELBean bean = new SingleMethodJsonWithSpELBean(); - MessagingMethodInvokerHelper helper = - new MessagingMethodInvokerHelper<>(bean, + MessagingMethodInvokerHelper helper = + new MessagingMethodInvokerHelper(bean, SingleMethodJsonWithSpELBean.class.getDeclaredMethod("foo", SingleMethodJsonWithSpELBean.Foo.class), false); @@ -1063,7 +1062,7 @@ public class MethodInvokingMessageProcessorTests { @Test public void testSingleMethodBadJson() throws Exception { SingleMethodJsonWithSpELMessageWildBean bean = new SingleMethodJsonWithSpELMessageWildBean(); - MessagingMethodInvokerHelper helper = new MessagingMethodInvokerHelper<>(bean, + MessagingMethodInvokerHelper helper = new MessagingMethodInvokerHelper(bean, SingleMethodJsonWithSpELMessageWildBean.class.getDeclaredMethod("foo", Message.class), false); helper.setBeanFactory(mock(BeanFactory.class)); Message message = new GenericMessage<>("baz", @@ -1075,7 +1074,7 @@ public class MethodInvokingMessageProcessorTests { @Test public void testSingleMethodJsonMessageFoo() throws Exception { SingleMethodJsonWithSpELMessageFooBean bean = new SingleMethodJsonWithSpELMessageFooBean(); - MessagingMethodInvokerHelper helper = new MessagingMethodInvokerHelper<>(bean, + MessagingMethodInvokerHelper helper = new MessagingMethodInvokerHelper(bean, SingleMethodJsonWithSpELMessageFooBean.class.getDeclaredMethod("foo", Message.class), false); helper.setBeanFactory(mock(BeanFactory.class)); @@ -1088,7 +1087,7 @@ public class MethodInvokingMessageProcessorTests { @Test public void testSingleMethodJsonMessageWild() throws Exception { SingleMethodJsonWithSpELMessageWildBean bean = new SingleMethodJsonWithSpELMessageWildBean(); - MessagingMethodInvokerHelper helper = new MessagingMethodInvokerHelper<>(bean, + MessagingMethodInvokerHelper helper = new MessagingMethodInvokerHelper(bean, SingleMethodJsonWithSpELMessageWildBean.class.getDeclaredMethod("foo", Message.class), false); helper.setBeanFactory(mock(BeanFactory.class)); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/resource/ResourceInboundChannelAdapterParserTests.java b/spring-integration-core/src/test/java/org/springframework/integration/resource/ResourceInboundChannelAdapterParserTests.java index 2b34d6b1b1..43006f376b 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/resource/ResourceInboundChannelAdapterParserTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/resource/ResourceInboundChannelAdapterParserTests.java @@ -27,7 +27,6 @@ import org.junit.BeforeClass; import org.junit.Test; import org.springframework.beans.factory.BeanCreationException; -import org.springframework.context.ApplicationContext; import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.core.io.Resource; import org.springframework.integration.channel.QueueChannel; @@ -69,8 +68,8 @@ public class ResourceInboundChannelAdapterParserTests { @Test public void testDefaultConfig() { - ApplicationContext context = new ClassPathXmlApplicationContext("ResourcePatternResolver-config.xml", - this.getClass()); + ClassPathXmlApplicationContext context = + new ClassPathXmlApplicationContext("ResourcePatternResolver-config.xml", getClass()); SourcePollingChannelAdapter resourceAdapter = context.getBean("resourceAdapterDefault", SourcePollingChannelAdapter.class); ResourceRetrievingMessageSource source = TestUtils.getPropertyValue(resourceAdapter, "source", @@ -81,17 +80,18 @@ public class ResourceInboundChannelAdapterParserTests { assertThat(TestUtils.getPropertyValue(source, "pattern")).isEqualTo("/**/*"); assertThat(TestUtils.getPropertyValue(source, "patternResolver")).isEqualTo(context); + context.close(); } @Test(expected = BeanCreationException.class) public void testDefaultConfigNoLocationPattern() { - new ClassPathXmlApplicationContext("ResourcePatternResolver-config-fail.xml", this.getClass()).close(); + new ClassPathXmlApplicationContext("ResourcePatternResolver-config-fail.xml", getClass()).close(); } @Test public void testCustomPatternResolver() { - ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( - "ResourcePatternResolver-config-custom.xml", this.getClass()); + ClassPathXmlApplicationContext context = + new ClassPathXmlApplicationContext("ResourcePatternResolver-config-custom.xml", getClass()); SourcePollingChannelAdapter resourceAdapter = context.getBean("resourceAdapterDefault", SourcePollingChannelAdapter.class); ResourceRetrievingMessageSource source = TestUtils.getPropertyValue(resourceAdapter, "source", diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileListFilterFactoryBean.java b/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileListFilterFactoryBean.java index 3c532f6e05..0e4bf75deb 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileListFilterFactoryBean.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileListFilterFactoryBean.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. @@ -91,7 +91,7 @@ public class FileListFilterFactoryBean implements FactoryBean getObject() throws Exception { + public FileListFilter getObject() { if (this.result == null) { synchronized (this.monitor) { this.initializeFileListFilter(); diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileReadingMessageSourceFactoryBean.java b/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileReadingMessageSourceFactoryBean.java index 7fa7e5803c..3e29d711ed 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileReadingMessageSourceFactoryBean.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileReadingMessageSourceFactoryBean.java @@ -19,6 +19,7 @@ package org.springframework.integration.file.config; import java.io.File; import java.util.Comparator; +import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.config.AbstractFactoryBean; import org.springframework.integration.file.DirectoryScanner; import org.springframework.integration.file.FileReadingMessageSource; @@ -114,10 +115,7 @@ public class FileReadingMessageSourceFactoryBean extends AbstractFactoryBean compositeFileListFilter = new CompositeFileListFilter<>(); - try { - compositeFileListFilter.addFilter(new FileListFilterFactoryBean().getObject()); - } - catch (Exception e) { - throw new IllegalStateException(e); - } + compositeFileListFilter.addFilter(new FileListFilterFactoryBean().getObject()); compositeFileListFilter.addFilter(this.locker); this.source.setFilter(compositeFileListFilter); this.source.setLocker(this.locker); } - if (this.scanEachPoll != null) { - this.source.setScanEachPoll(this.scanEachPoll); - } - if (this.autoCreateDirectory != null) { - this.source.setAutoCreateDirectory(this.autoCreateDirectory); - } - this.source.setBeanFactory(getBeanFactory()); - try { - this.source.afterPropertiesSet(); - } - catch (Exception e) { - throw new IllegalStateException(e); - } } } diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/gateway/AbstractRemoteFileOutboundGateway.java b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/gateway/AbstractRemoteFileOutboundGateway.java index 8fd74bdfc1..acf738288b 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/gateway/AbstractRemoteFileOutboundGateway.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/gateway/AbstractRemoteFileOutboundGateway.java @@ -22,6 +22,7 @@ import java.io.FileNotFoundException; import java.io.FileOutputStream; import java.io.IOException; import java.io.OutputStream; +import java.io.UncheckedIOException; import java.util.ArrayList; import java.util.Arrays; import java.util.Collection; @@ -31,6 +32,7 @@ import java.util.List; import java.util.Set; import java.util.stream.Collectors; +import org.springframework.beans.factory.BeanFactory; import org.springframework.expression.EvaluationContext; import org.springframework.expression.Expression; import org.springframework.expression.spel.standard.SpelExpressionParser; @@ -71,11 +73,11 @@ import org.springframework.util.StringUtils; public abstract class AbstractRemoteFileOutboundGateway extends AbstractReplyProducingMessageHandler { - protected final RemoteFileTemplate remoteFileTemplate; // NOSONAR + private final RemoteFileTemplate remoteFileTemplate; - protected final Command command; // NOSONAR + private final Command command; - protected final Set