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

Fix new Sonar smells (#2768)

Browse Source 
* 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()`
This commit is contained in:
Artem Bilan
2019-02-27 15:17:51 -05:00
committed by Gary Russell
parent f741724656
commit d2e974a6de
33 changed files with 828 additions and 750 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/AbstractAmqpChannel.java
View File

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

4
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/AbstractSubscribableAmqpChannel.java
View File

@@ -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();

138
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PollableAmqpChannel.java
View File

@@ -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<ChannelInterceptor> 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();

122
spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java
View File

@@ -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<Message<?>> 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<Message<?>> 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,

18
spring-integration-core/src/main/java/org/springframework/integration/aggregator/BarrierMessageHandler.java
View File

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

4
spring-integration-core/src/main/java/org/springframework/integration/aggregator/ExpressionEvaluatingReleaseStrategy.java
View File

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

26
spring-integration-core/src/main/java/org/springframework/integration/aggregator/MethodInvokingMessageListProcessor.java
View File

@@ -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<T> extends AbstractExpressionEvaluator
implements Lifecycle {
private final MessagingMethodInvokerHelper<T> delegate;
private final MessagingMethodInvokerHelper delegate;
public MethodInvokingMessageListProcessor(Object targetObject, Method method, Class<T> expectedType) {
this.delegate = new MessagingMethodInvokerHelper<T>(targetObject, method, expectedType, true);
this.delegate = new MessagingMethodInvokerHelper(targetObject, method, expectedType, true);
}
public MethodInvokingMessageListProcessor(Object targetObject, Method method) {
this.delegate = new MessagingMethodInvokerHelper<T>(targetObject, method, true);
this.delegate = new MessagingMethodInvokerHelper(targetObject, method, true);
}
public MethodInvokingMessageListProcessor(Object targetObject, String methodName, Class<T> expectedType) {
this.delegate = new MessagingMethodInvokerHelper<T>(targetObject, methodName,
this.delegate = new MessagingMethodInvokerHelper(targetObject, methodName,
expectedType, true);
}
public MethodInvokingMessageListProcessor(Object targetObject, String methodName) {
this.delegate = new MessagingMethodInvokerHelper<T>(targetObject, methodName, true);
this.delegate = new MessagingMethodInvokerHelper(targetObject, methodName, true);
}
public MethodInvokingMessageListProcessor(Object targetObject, Class<? extends Annotation> annotationType) {
this.delegate = new MessagingMethodInvokerHelper<T>(targetObject, annotationType, Object.class, true);
this.delegate = new MessagingMethodInvokerHelper(targetObject, annotationType, Object.class, true);
}
@Override
@@ -84,16 +85,9 @@ public class MethodInvokingMessageListProcessor<T> extends AbstractExpressionEva
return this.delegate.toString();
}
@SuppressWarnings("unchecked")
public T process(Collection<Message<?>> messages, Map<String, Object> 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

21
spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractExecutorChannel.java
View File

@@ -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<ExecutorChannelInterceptor> interceptorStack = null;
try {
if (AbstractExecutorChannel.this.executorInterceptorsSize > 0) {
interceptorStack = new ArrayDeque<ExecutorChannelInterceptor>();
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;

108
spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java
View File

@@ -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<AbstractMessageChannelMetrics> {
protected final ChannelInterceptorList interceptors;
protected final ChannelInterceptorList interceptors; // NOSONAR
private final Comparator<Object> 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<ChannelInterceptor> 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<ChannelInterceptor> interceptors = new CopyOnWriteArrayList<>(); // NOSONAR
protected final List<ChannelInterceptor> interceptors = new CopyOnWriteArrayList<ChannelInterceptor>();
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<ChannelInterceptor> interceptorStack) {
@Nullable Exception ex, @Nullable Deque<ChannelInterceptor> interceptorStack) {
for (Iterator<ChannelInterceptor> 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<ChannelInterceptor> 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);
}
}
}
}

75
spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractPollableChannel.java
View File

@@ -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();

61
spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageProducerSupport.java
View File

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

56
spring-integration-core/src/main/java/org/springframework/integration/graph/IntegrationGraphServer.java
View File

@@ -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<String, Object> descriptor = new HashMap<String, Object>();
Map<String, Object> 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)

2
spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageHandler.java
View File

@@ -338,7 +338,7 @@ public abstract class AbstractMessageHandler extends IntegrationObjectSupport
}
@Override
public void destroy() throws Exception {
public void destroy() {
this.timers.forEach(MeterFacade::remove);
}

96
spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProducingHandler.java
View File

@@ -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<Object>() {
@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<Object> {
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);
}
}
}

21
spring-integration-core/src/main/java/org/springframework/integration/handler/MethodInvokingMessageProcessor.java
View File

@@ -44,22 +44,22 @@ import org.springframework.messaging.Message;
*/
public class MethodInvokingMessageProcessor<T> extends AbstractMessageProcessor<T> implements Lifecycle {
private final MessagingMethodInvokerHelper<T> delegate;
private final MessagingMethodInvokerHelper delegate;
public MethodInvokingMessageProcessor(Object targetObject, Method method) {
this.delegate = new MessagingMethodInvokerHelper<T>(targetObject, method, false);
this.delegate = new MessagingMethodInvokerHelper(targetObject, method, false);
}
public MethodInvokingMessageProcessor(Object targetObject, String methodName) {
this.delegate = new MessagingMethodInvokerHelper<T>(targetObject, methodName, false);
this.delegate = new MessagingMethodInvokerHelper(targetObject, methodName, false);
}
public MethodInvokingMessageProcessor(Object targetObject, String methodName, boolean canProcessMessageList) {
this.delegate = new MessagingMethodInvokerHelper<T>(targetObject, methodName, canProcessMessageList);
this.delegate = new MessagingMethodInvokerHelper(targetObject, methodName, canProcessMessageList);
}
public MethodInvokingMessageProcessor(Object targetObject, Class<? extends Annotation> annotationType) {
this.delegate = new MessagingMethodInvokerHelper<T>(targetObject, annotationType, false);
this.delegate = new MessagingMethodInvokerHelper(targetObject, annotationType, false);
}
@Override
@@ -69,7 +69,7 @@ public class MethodInvokingMessageProcessor<T> 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<T> 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);
}
}

275
spring-integration-core/src/main/java/org/springframework/integration/handler/support/MessagingMethodInvokerHelper.java
View File

@@ -124,13 +124,13 @@ import org.springframework.util.StringUtils;
*
* @since 2.0
*/
public class MessagingMethodInvokerHelper<T> 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<T> extends AbstractExpressionEvaluator
private static final Map<SpelCompilerMode, ExpressionParser> 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<Message<?>> 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<T> extends AbstractExpressionEvaluator
private final JsonObjectMapper<?, ?> jsonObjectMapper;
private volatile String displayString;
private volatile boolean requiresReply;
private final Map<Class<?>, HandlerMethod> handlerMethods;
private final Map<Class<?>, HandlerMethod> handlerMessageMethods;
private final List<Map<Class<?>, HandlerMethod>> handlerMethodsList;
private HandlerMethod handlerMethod;
private final TypeDescriptor expectedType;
private final boolean canProcessMessageList;
private Class<? extends Annotation> annotationType;
private HandlerMethod handlerMethod;
private volatile boolean initialized;
private Class<? extends Annotation> 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<T> 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<Message<?>> messages, Map<String, Object> headers) throws Exception {
public Object process(Collection<Message<?>> messages, Map<String, Object> headers) {
ParametersWrapper parameters = new ParametersWrapper(messages, headers);
return processInternal(parameters);
}
@@ -431,7 +427,7 @@ public class MessagingMethodInvokerHelper<T> 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<T> 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<T> 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<T> 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<T> 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<T> 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<T> 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<HandlerMethodArgumentResolver> customArgumentResolvers = new LinkedList<>();
customArgumentResolvers.add(payloadExpressionArgumentResolver);
@@ -604,7 +610,9 @@ public class MessagingMethodInvokerHelper<T> 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<T> 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<T> 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<T> 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<T> 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<T> 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<T> 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<T> 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<T> extends AbstractExpressionEvaluator
}
@SuppressWarnings("unchecked")
public <T> 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<T> 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);
}

4
spring-integration-core/src/main/java/org/springframework/integration/support/AbstractIntegrationMessageBuilder.java
View File

@@ -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<T> {
* @see #copyHeadersIfAbsent(Map)
*/
public AbstractIntegrationMessageBuilder<T> filterAndCopyHeadersIfAbsent(Map<String, ?> headersToCopy,
String... headerPatternsToFilter) {
@Nullable String... headerPatternsToFilter) {
Map<String, ?> headers = headersToCopy;

4
spring-integration-core/src/main/java/org/springframework/integration/support/utils/IntegrationUtils.java
View File

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

4
spring-integration-core/src/main/java/org/springframework/integration/util/AbstractExpressionEvaluator.java
View File

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

3
spring-integration-core/src/test/java/org/springframework/integration/config/AggregatorParserTests.java
View File

@@ -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<Long> methodInvokerHelper =
MessagingMethodInvokerHelper methodInvokerHelper =
TestUtils.getPropertyValue(releaseStrategy, "adapter.delegate", MessagingMethodInvokerHelper.class);
Object handlerMethods = TestUtils.getPropertyValue(methodInvokerHelper, "handlerMethods");
assertThat(handlerMethods).isNull();

33
spring-integration-core/src/test/java/org/springframework/integration/handler/MethodInvokingMessageProcessorTests.java
View File

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

12
spring-integration-core/src/test/java/org/springframework/integration/resource/ResourceInboundChannelAdapterParserTests.java
View File

@@ -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",

4
spring-integration-file/src/main/java/org/springframework/integration/file/config/FileListFilterFactoryBean.java
View File

@@ -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<FileListFilter<Fil
}
@Override
public FileListFilter<File> getObject() throws Exception {
public FileListFilter<File> getObject() {
if (this.result == null) {
synchronized (this.monitor) {
this.initializeFileListFilter();

41
spring-integration-file/src/main/java/org/springframework/integration/file/config/FileReadingMessageSourceFactoryBean.java
View File

@@ -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<Fil
return this.source;
}
private void initSource() { // NOSONAR
if (this.source != null) {
return;
}
private void initSource() {
boolean comparatorSet = this.comparator != null;
boolean queueSizeSet = this.queueSize != null;
if (comparatorSet) {
@@ -142,6 +140,21 @@ public class FileReadingMessageSourceFactoryBean extends AbstractFactoryBean<Fil
this.source.setWatchEvents(this.watchEvents);
}
}
configureFilterAndLockerOnSourceIfAny();
if (this.scanEachPoll != null) {
this.source.setScanEachPoll(this.scanEachPoll);
}
if (this.autoCreateDirectory != null) {
this.source.setAutoCreateDirectory(this.autoCreateDirectory);
}
BeanFactory beanFactory = getBeanFactory();
if (beanFactory != null) {
this.source.setBeanFactory(beanFactory);
}
this.source.afterPropertiesSet();
}
private void configureFilterAndLockerOnSourceIfAny() {
if (this.filter != null) {
if (this.locker == null) {
this.source.setFilter(this.filter);
@@ -156,29 +169,11 @@ public class FileReadingMessageSourceFactoryBean extends AbstractFactoryBean<Fil
}
else if (this.locker != null) {
CompositeFileListFilter<File> 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);
}
}
}

298
spring-integration-file/src/main/java/org/springframework/integration/file/remote/gateway/AbstractRemoteFileOutboundGateway.java
View File

@@ -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<F> extends AbstractReplyProducingMessageHandler {
protected final RemoteFileTemplate<F> remoteFileTemplate; // NOSONAR
private final RemoteFileTemplate<F> remoteFileTemplate;
protected final Command command; // NOSONAR
private final Command command;
protected final Set<Option> options = new HashSet<>();
private final Set<Option> options = new HashSet<>();
private final ExpressionEvaluatingMessageProcessor<String> fileNameProcessor;
@@ -370,7 +372,7 @@ public abstract class AbstractRemoteFileOutboundGateway<F> extends AbstractReply
*/
public void setChmodOctal(String chmod) {
Assert.notNull(chmod, "'chmod' cannot be null");
setChmod(Integer.parseInt(chmod, 8));
setChmod(Integer.parseInt(chmod, 8)); // NOSONAR octal radix
}
/**
@@ -388,6 +390,10 @@ public abstract class AbstractRemoteFileOutboundGateway<F> extends AbstractReply
return false;
}
protected final RemoteFileTemplate<F> getRemoteFileTemplate() {
return this.remoteFileTemplate;
}
@Override
protected void doInit() {
Assert.state(this.command != null || this.messageSessionCallback != null,
@@ -396,50 +402,57 @@ public abstract class AbstractRemoteFileOutboundGateway<F> extends AbstractReply
Command.GET.equals(this.command)) {
Assert.isNull(this.filter, "Filters are not supported with the rm and get commands");
}
if ((Command.GET.equals(this.command) && !this.options.contains(Option.STREAM))
|| Command.MGET.equals(this.command)) {
Assert.notNull(this.localDirectoryExpression, "localDirectory must not be null");
if (this.localDirectoryExpression instanceof ValueExpression) {
File localDirectory = ExpressionUtils.expressionToFile(this.localDirectoryExpression,
ExpressionUtils.createStandardEvaluationContext(getBeanFactory()), null,
"localDirectoryExpression");
try {
if (!localDirectory.exists()) {
if (this.autoCreateLocalDirectory) {
if (logger.isDebugEnabled()) {
logger.debug("The '" + localDirectory + "' directory doesn't exist; Will create.");
}
if (!localDirectory.mkdirs()) {
throw new IOException("Failed to make local directory: " + localDirectory);
}
}
else {
throw new FileNotFoundException(localDirectory.getName());
}
}
}
catch (RuntimeException e) {
throw e;
}
catch (Exception e) {
throw new MessagingException(
"Failure during initialization of: " + this.getComponentType(), e);
}
setupLocalDirectory();
}
}
if (Command.MGET.equals(this.command)) {
Assert.isTrue(!(this.options.contains(Option.SUBDIRS)),
"Cannot use " + Option.SUBDIRS.toString() + " when using 'mget' use " +
Option.RECURSIVE.toString() + " to obtain files in subdirectories");
}
if (getBeanFactory() != null) {
if (this.fileNameProcessor != null) {
this.fileNameProcessor.setBeanFactory(getBeanFactory());
}
populateBeanFactoryIntoComponentsIfAny();
}
this.renameProcessor.setBeanFactory(getBeanFactory());
this.remoteFileTemplate.setBeanFactory(getBeanFactory());
private void populateBeanFactoryIntoComponentsIfAny() {
BeanFactory beanFactory = getBeanFactory();
if (beanFactory != null) {
if (this.fileNameProcessor != null) {
this.fileNameProcessor.setBeanFactory(beanFactory);
}
this.renameProcessor.setBeanFactory(beanFactory);
this.remoteFileTemplate.setBeanFactory(beanFactory);
}
}
private void setupLocalDirectory() {
File localDirectory =
ExpressionUtils.expressionToFile(this.localDirectoryExpression,
ExpressionUtils.createStandardEvaluationContext(getBeanFactory()), null,
"localDirectoryExpression");
if (!localDirectory.exists()) {
try {
if (this.autoCreateLocalDirectory) {
if (logger.isDebugEnabled()) {
logger.debug("The '" + localDirectory + "' directory doesn't exist; Will create.");
}
if (!localDirectory.mkdirs()) {
throw new IOException("Failed to make local directory: " + localDirectory);
}
}
else {
throw new FileNotFoundException(localDirectory.getName());
}
}
catch (IOException ex) {
throw new UncheckedIOException(ex);
}
}
}
@@ -519,9 +532,9 @@ public abstract class AbstractRemoteFileOutboundGateway<F> extends AbstractReply
}
private Object doGet(final Message<?> requestMessage) {
final String remoteFilePath = this.fileNameProcessor.processMessage(requestMessage);
final String remoteFilename = getRemoteFilename(remoteFilePath);
final String remoteDir = getRemoteDirectory(remoteFilePath, remoteFilename);
String remoteFilePath = obtainRemoteFilePath(requestMessage);
String remoteFilename = getRemoteFilename(remoteFilePath);
String remoteDir = getRemoteDirectory(remoteFilePath, remoteFilename);
Session<F> session = null;
Object payload;
if (this.options.contains(Option.STREAM)) {
@@ -547,7 +560,7 @@ public abstract class AbstractRemoteFileOutboundGateway<F> extends AbstractReply
}
private Object doMget(final Message<?> requestMessage) {
final String remoteFilePath = this.fileNameProcessor.processMessage(requestMessage);
String remoteFilePath = obtainRemoteFilePath(requestMessage);
final String remoteFilename = getRemoteFilename(remoteFilePath);
final String remoteDir = getRemoteDirectory(remoteFilePath, remoteFilename);
List<File> payload = this.remoteFileTemplate.execute(session ->
@@ -559,7 +572,7 @@ public abstract class AbstractRemoteFileOutboundGateway<F> extends AbstractReply
}
private Object doRm(Message<?> requestMessage) {
final String remoteFilePath = this.fileNameProcessor.processMessage(requestMessage);
String remoteFilePath = obtainRemoteFilePath(requestMessage);
String remoteFilename = getRemoteFilename(remoteFilePath);
String remoteDir = getRemoteDirectory(remoteFilePath, remoteFilename);
@@ -587,7 +600,7 @@ public abstract class AbstractRemoteFileOutboundGateway<F> extends AbstractReply
}
private Object doMv(Message<?> requestMessage) {
String remoteFilePath = this.fileNameProcessor.processMessage(requestMessage);
String remoteFilePath = obtainRemoteFilePath(requestMessage);
String remoteFilename = getRemoteFilename(remoteFilePath);
String remoteDir = getRemoteDirectory(remoteFilePath, remoteFilename);
String remoteFileNewPath = this.renameProcessor.processMessage(requestMessage);
@@ -604,6 +617,13 @@ public abstract class AbstractRemoteFileOutboundGateway<F> extends AbstractReply
.setHeader(FileHeaders.RENAME_TO, remoteFileNewPath);
}
private String obtainRemoteFilePath(Message<?> requestMessage) {
String remoteFilePath = this.fileNameProcessor.processMessage(requestMessage);
Assert.state(remoteFilePath != null,
() -> "The 'fileNameProcessor' evaluated to null 'remoteFilePath' from message: " + requestMessage);
return remoteFilePath;
}
/**
* Move one remote path to another.
* The message can be consulted to determine some context;
@@ -618,6 +638,7 @@ public abstract class AbstractRemoteFileOutboundGateway<F> extends AbstractReply
*/
protected boolean mv(Message<?> message, Session<F> session, String remoteFilePath, String remoteFileNewPath)
throws IOException {
int lastSeparator = remoteFileNewPath.lastIndexOf(this.remoteFileTemplate.getRemoteFileSeparator());
if (lastSeparator > 0) {
String remoteFileDirectory = remoteFileNewPath.substring(0, lastSeparator + 1);
@@ -715,44 +736,36 @@ public abstract class AbstractRemoteFileOutboundGateway<F> extends AbstractReply
}
private List<String> putLocalDirectory(Message<?> requestMessage, File file, String subDirectory) {
File[] files = file.listFiles();
List<File> filteredFiles = this.filterMputFiles(files);
List<File> filteredFiles = filterMputFiles(file.listFiles());
List<String> replies = new ArrayList<>();
try {
for (File filteredFile : filteredFiles) {
if (!filteredFile.isDirectory()) {
String path = doPut(new MutableMessage<>(filteredFile, requestMessage.getHeaders()), subDirectory);
if (path == null) { //NOSONAR - false positive
if (logger.isDebugEnabled()) {
logger.debug("File " + filteredFile.getAbsolutePath()
+ " removed before transfer; ignoring");
}
}
else {
if (path != null) {
replies.add(path);
}
else if (logger.isDebugEnabled()) {
logger.debug("File " + filteredFile.getAbsolutePath() + " removed before transfer; ignoring");
}
}
else if (this.options.contains(Option.RECURSIVE)) {
String newSubDirectory = (StringUtils.hasText(subDirectory) ?
subDirectory + this.remoteFileTemplate.getRemoteFileSeparator() : "")
+ filteredFile.getName();
String newSubDirectory =
(StringUtils.hasText(subDirectory) ?
subDirectory + this.remoteFileTemplate.getRemoteFileSeparator()
: "") + filteredFile.getName();
replies.addAll(putLocalDirectory(requestMessage, filteredFile, newSubDirectory));
}
}
}
catch (Exception e) {
if (replies.size() > 0) {
catch (Exception ex) {
if (replies.size() > 0 || ex instanceof PartialSuccessException) { // NOSONAR
throw new PartialSuccessException(requestMessage,
"Partially successful 'mput' operation" +
(subDirectory == null ? "" : (" on " + subDirectory)), e, replies, filteredFiles);
}
else if (e instanceof PartialSuccessException) {
throw new PartialSuccessException(requestMessage,
"Partially successful 'mput' operation" +
(subDirectory == null ? "" : (" on " + subDirectory)), e, replies, filteredFiles);
(subDirectory == null ? "" : (" on " + subDirectory)), ex, replies, filteredFiles);
}
else {
throw e;
throw ex;
}
}
return replies;
@@ -976,6 +989,7 @@ public abstract class AbstractRemoteFileOutboundGateway<F> extends AbstractReply
protected List<File> mGet(Message<?> message, Session<F> session, String remoteDirectory,
String remoteFilename) throws IOException {
if (this.options.contains(Option.RECURSIVE)) {
if (logger.isWarnEnabled() && !("*".equals(remoteFilename))) {
logger.warn("File name pattern must be '*' when using recursion");
@@ -993,6 +1007,68 @@ public abstract class AbstractRemoteFileOutboundGateway<F> extends AbstractReply
List<File> files = new ArrayList<>();
String remotePath = buildRemotePath(remoteDirectory, remoteFilename);
List<AbstractFileInfo<F>> remoteFiles = lsRemoteFilesForMget(message, session, remoteDirectory,
remoteFilename, remotePath);
try {
for (AbstractFileInfo<F> lsEntry : remoteFiles) {
if (lsEntry.isDirectory()) {
continue;
}
File file = getRemoteFileForMget(message, session, remoteDirectory, lsEntry);
if (file != null) {
files.add(file);
}
}
}
catch (Exception ex) {
throw processMgetException(message, remoteDirectory, files, remoteFiles, ex);
}
return files;
}
private RuntimeException processMgetException(Message<?> message, String remoteDirectory, List<File> files,
List<AbstractFileInfo<F>> remoteFiles, Exception ex) {
if (files.size() > 0) {
return new PartialSuccessException(message,
"Partially successful recursive 'mget' operation on "
+ (remoteDirectory != null ? remoteDirectory : "Client Working Directory"),
ex, files, remoteFiles);
}
else if (ex instanceof MessagingException) {
return (MessagingException) ex;
}
else if (ex instanceof IOException) {
throw new UncheckedIOException((IOException) ex);
}
else {
return new MessagingException("Failed to process MGET", ex);
}
}
private List<File> mGetWithRecursion(Message<?> message, Session<F> session, String remoteDirectory,
String remoteFilename) throws IOException {
List<File> files = new ArrayList<>();
List<AbstractFileInfo<F>> fileNames = lsRemoteFilesForMget(message, session, remoteDirectory,
remoteFilename, remoteDirectory);
try {
for (AbstractFileInfo<F> lsEntry : fileNames) {
File file = getRemoteFileForMget(message, session, remoteDirectory, lsEntry);
if (file != null) {
files.add(file);
}
}
}
catch (Exception ex) {
throw processMgetException(message, remoteDirectory, files, fileNames, ex);
}
return files;
}
private List<AbstractFileInfo<F>> lsRemoteFilesForMget(Message<?> message, Session<F> session,
String remoteDirectory, String remoteFilename, String remotePath) throws IOException {
@SuppressWarnings("unchecked")
List<AbstractFileInfo<F>> remoteFiles = (List<AbstractFileInfo<F>>) ls(message, session, remotePath);
if (remoteFiles.size() == 0 && this.options.contains(Option.EXCEPTION_WHEN_EMPTY)) {
@@ -1000,91 +1076,23 @@ public abstract class AbstractRemoteFileOutboundGateway<F> extends AbstractReply
+ (remoteDirectory != null ? remoteDirectory : "Client Working Directory")
+ " with pattern " + remoteFilename);
}
try {
for (AbstractFileInfo<F> lsEntry : remoteFiles) {
if (lsEntry.isDirectory()) {
continue;
}
String fullFileName = remoteDirectory != null
? remoteDirectory + getFilename(lsEntry)
: getFilename(lsEntry);
/*
* With recursion, the filename might contain subdirectory information
* normalize each file separately.
*/
String fileName = this.getRemoteFilename(fullFileName);
String actualRemoteDirectory = this.getRemoteDirectory(fullFileName, fileName);
File file = get(message, session, actualRemoteDirectory, fullFileName, fileName,
lsEntry.getFileInfo());
if (file != null) {
files.add(file);
}
}
}
catch (Exception e) {
if (files.size() > 0) {
throw new PartialSuccessException(message,
"Partially successful recursive 'mget' operation on "
+ (remoteDirectory != null ? remoteDirectory : "Client Working Directory"),
e, files, remoteFiles);
}
else if (e instanceof MessagingException) {
throw (MessagingException) e;
}
else if (e instanceof IOException) {
throw (IOException) e;
}
}
return files;
return remoteFiles;
}
private List<File> mGetWithRecursion(Message<?> message, Session<F> session, String remoteDirectory,
String remoteFilename) throws IOException {
List<File> files = new ArrayList<>();
@SuppressWarnings("unchecked")
List<AbstractFileInfo<F>> fileNames = (List<AbstractFileInfo<F>>) ls(message, session, remoteDirectory);
if (fileNames.size() == 0 && this.options.contains(Option.EXCEPTION_WHEN_EMPTY)) {
throw new MessagingException("No files found at "
+ (remoteDirectory != null ? remoteDirectory : "Client Working Directory")
+ " with pattern " + remoteFilename);
}
try {
for (AbstractFileInfo<F> lsEntry : fileNames) {
String fullFileName =
remoteDirectory != null
? remoteDirectory + getFilename(lsEntry)
: getFilename(lsEntry);
/*
* With recursion, the filename might contain subdirectory information
* normalize each file separately.
*/
String fileName = this.getRemoteFilename(fullFileName);
String actualRemoteDirectory = this.getRemoteDirectory(fullFileName, fileName);
File file = get(message, session, actualRemoteDirectory, fullFileName, fileName,
lsEntry.getFileInfo());
if (file != null) {
files.add(file);
}
}
}
catch (Exception e) {
if (files.size() > 0) {
throw new PartialSuccessException(message,
"Partially successful recursive 'mget' operation on "
+ (remoteDirectory != null ? remoteDirectory : "Client Working Directory"),
e, files, fileNames);
}
else if (e instanceof MessagingException) {
throw (MessagingException) e;
}
else if (e instanceof IOException) {
throw (IOException) e;
}
else {
throw new MessagingException("Failed to process MGET on first file", e);
}
}
return files;
private File getRemoteFileForMget(Message<?> message, Session<F> session, String remoteDirectory,
AbstractFileInfo<F> lsEntry) throws IOException {
String fullFileName =
remoteDirectory != null
? remoteDirectory + getFilename(lsEntry)
: getFilename(lsEntry);
/*
* With recursion, the filename might contain subdirectory information
* normalize each file separately.
*/
String fileName = getRemoteFilename(fullFileName);
String actualRemoteDirectory = getRemoteDirectory(fullFileName, fileName);
return get(message, session, actualRemoteDirectory, fullFileName, fileName, lsEntry.getFileInfo());
}
private String getRemoteDirectory(String remoteFilePath, String remoteFilename) {

4
spring-integration-ftp/src/main/java/org/springframework/integration/ftp/gateway/FtpOutboundGateway.java
View File

@@ -64,7 +64,7 @@ public class FtpOutboundGateway extends AbstractRemoteFileOutboundGateway<FTPFil
public FtpOutboundGateway(SessionFactory<FTPFile> sessionFactory,
MessageSessionCallback<FTPFile, ?> messageSessionCallback) {
this(new FtpRemoteFileTemplate(sessionFactory), messageSessionCallback);
((FtpRemoteFileTemplate) this.remoteFileTemplate).setExistsMode(FtpRemoteFileTemplate.ExistsMode.NLST);
((FtpRemoteFileTemplate) getRemoteFileTemplate()).setExistsMode(FtpRemoteFileTemplate.ExistsMode.NLST);
}
/**
@@ -87,7 +87,7 @@ public class FtpOutboundGateway extends AbstractRemoteFileOutboundGateway<FTPFil
*/
public FtpOutboundGateway(SessionFactory<FTPFile> sessionFactory, String command, String expression) {
this(new FtpRemoteFileTemplate(sessionFactory), command, expression);
((FtpRemoteFileTemplate) this.remoteFileTemplate).setExistsMode(FtpRemoteFileTemplate.ExistsMode.NLST);
((FtpRemoteFileTemplate) getRemoteFileTemplate()).setExistsMode(FtpRemoteFileTemplate.ExistsMode.NLST);
}
/**

9
spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastSendingMessageHandler.java
View File

@@ -63,6 +63,8 @@ import org.springframework.util.StringUtils;
public class UnicastSendingMessageHandler extends
AbstractInternetProtocolSendingMessageHandler implements Runnable {
private static final int DEFAULT_ACK_TIMEOUT = 5000;
private final DatagramPacketMessageMapper mapper = new DatagramPacketMessageMapper();
private final Expression destinationExpression;
@@ -80,12 +82,11 @@ public class UnicastSendingMessageHandler extends
private volatile int ackPort;
private volatile int ackTimeout = 5000;
private volatile int ackTimeout = DEFAULT_ACK_TIMEOUT;
private volatile int ackCounter = 1;
private volatile Map<String, CountDownLatch> ackControl =
Collections.synchronizedMap(new HashMap<String, CountDownLatch>());
private volatile Map<String, CountDownLatch> ackControl = Collections.synchronizedMap(new HashMap<>());
private volatile int soReceiveBufferSize = -1;
@@ -283,7 +284,7 @@ public class UnicastSendingMessageHandler extends
}
}
catch (Exception ex) {
if (!(ex instanceof MessagingException)) {
if (!(ex instanceof MessagingException)) { // NOSONAR
closeSocketIfNeeded();
}
throw IntegrationUtils.wrapInHandlingExceptionIfNecessary(message,

70
spring-integration-jms/src/main/java/org/springframework/integration/jms/PollableJmsChannel.java
View File

@@ -23,6 +23,7 @@ import java.util.List;
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.jms.core.JmsTemplate;
import org.springframework.lang.Nullable;
import org.springframework.messaging.Message;
@@ -121,9 +122,7 @@ public class PollableJmsChannel extends AbstractJmsChannel
}
else {
if (countsEnabled) {
if (getMetricsCaptor() != null) {
incrementReceiveCounter();
}
incrementReceiveCounter();
getMetrics().afterReceive();
counted = true;
}
@@ -139,46 +138,50 @@ public class PollableJmsChannel extends AbstractJmsChannel
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) {
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();
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
@@ -217,6 +220,7 @@ public class PollableJmsChannel extends AbstractJmsChannel
}
@Override
@Nullable
public ChannelInterceptor removeInterceptor(int index) {
ChannelInterceptor interceptor = super.removeInterceptor(index);
if (interceptor instanceof ExecutorChannelInterceptor) {

4
spring-integration-jms/src/main/java/org/springframework/integration/jms/SubscribableJmsChannel.java
View File

@@ -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.
@@ -170,7 +170,7 @@ public class SubscribableJmsChannel extends AbstractJmsChannel
}
@Override
public void destroy() throws Exception {
public void destroy() {
if (this.container != null) {
this.container.destroy();
}

22
spring-integration-jpa/src/main/java/org/springframework/integration/jpa/support/parametersource/ExpressionEvaluatingParameterSourceFactory.java
View File

@@ -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.
@@ -44,7 +44,7 @@ import org.springframework.util.Assert;
*/
public class ExpressionEvaluatingParameterSourceFactory implements ParameterSourceFactory {
private static final Log logger = LogFactory.getLog(ExpressionEvaluatingParameterSourceFactory.class);
private static final Log LOGGER = LogFactory.getLog(ExpressionEvaluatingParameterSourceFactory.class);
private static final Object ERROR = new Object();
@@ -57,7 +57,9 @@ public class ExpressionEvaluatingParameterSourceFactory implements ParameterSour
}
public ExpressionEvaluatingParameterSourceFactory(@Nullable BeanFactory beanFactory) {
this.expressionEvaluator.setBeanFactory(beanFactory);
if (beanFactory != null) {
this.expressionEvaluator.setBeanFactory(beanFactory);
}
}
/**
@@ -136,8 +138,8 @@ public class ExpressionEvaluatingParameterSourceFactory implements ParameterSour
if (parameter.getName() != null) {
this.values.put(parameter.getName(), value);
}
if (logger.isDebugEnabled()) {
logger.debug("Resolved expression " + expression + " to " + value);
if (LOGGER.isDebugEnabled()) {
LOGGER.debug("Resolved expression " + expression + " to " + value);
}
return value;
@@ -174,8 +176,8 @@ public class ExpressionEvaluatingParameterSourceFactory implements ParameterSour
final Object value = this.expressionEvaluator.evaluateExpression(expression, this.input);
this.values.put(paramName, value);
if (logger.isDebugEnabled()) {
logger.debug("Resolved expression " + expression + " to " + value);
if (LOGGER.isDebugEnabled()) {
LOGGER.debug("Resolved expression " + expression + " to " + value);
}
return value;
}
@@ -184,13 +186,13 @@ public class ExpressionEvaluatingParameterSourceFactory implements ParameterSour
public boolean hasValue(String paramName) {
try {
final Object value = getValue(paramName);
if (value == ERROR) {
if (ERROR.equals(value)) {
return false;
}
}
catch (ExpressionException e) {
if (logger.isDebugEnabled()) {
logger.debug("Could not evaluate expression", e);
if (LOGGER.isDebugEnabled()) {
LOGGER.debug("Could not evaluate expression", e);
}
this.values.put(paramName, ERROR);
return false;

15
spring-integration-redis/src/main/java/org/springframework/integration/redis/channel/SubscribableRedisChannel.java
View File

@@ -197,8 +197,13 @@ public class SubscribableRedisChannel extends AbstractMessageChannel
}
@Override
public void destroy() throws Exception {
this.container.destroy();
public void destroy() {
try {
this.container.destroy();
}
catch (Exception ex) {
throw new IllegalStateException("Cannot destroy " + this.container, ex);
}
}
private class MessageListenerDelegate {
@@ -217,11 +222,11 @@ public class SubscribableRedisChannel extends AbstractMessageChannel
String exceptionMessage = e.getMessage();
throw new MessageDeliveryException(siMessage,
(exceptionMessage == null ? e.getClass().getSimpleName() : exceptionMessage)
+ " for redis-channel '"
+ (StringUtils.hasText(SubscribableRedisChannel.this.topicName)
+ " for redis-channel '"
+ (StringUtils.hasText(SubscribableRedisChannel.this.topicName)
? SubscribableRedisChannel.this.topicName
: "unknown")
+ "' (" + getFullChannelName() + ").", e); // NOSONAR false positive - never null
+ "' (" + getFullChannelName() + ").", e); // NOSONAR false positive - never null
}
}

14
spring-integration-test-support/src/main/java/org/springframework/integration/test/matcher/MessageMatcher.java
View File

@@ -59,13 +59,6 @@ public class MessageMatcher extends BaseMatcher<Message<?>> {
this.headers = getHeaders(operand);
}
private Map<String, Object> getHeaders(Message<?> operand) {
HashMap<String, Object> headers = new HashMap<>(operand.getHeaders());
headers.remove(MessageHeaders.ID);
headers.remove(MessageHeaders.TIMESTAMP);
return headers;
}
public boolean matches(Object arg) {
Message<?> input = (Message<?>) arg;
Map<String, Object> inputHeaders = getHeaders(input);
@@ -78,4 +71,11 @@ public class MessageMatcher extends BaseMatcher<Message<?>> {
.appendValue(this.headers);
}
private static Map<String, Object> getHeaders(Message<?> operand) {
HashMap<String, Object> headersToFilter = new HashMap<>(operand.getHeaders());
headersToFilter.remove(MessageHeaders.ID);
headersToFilter.remove(MessageHeaders.TIMESTAMP);
return headersToFilter;
}
}

6
spring-integration-test-support/src/main/java/org/springframework/integration/test/predicate/MessagePredicate.java
View File

@@ -68,9 +68,9 @@ public class MessagePredicate implements Predicate<Message<?>> {
}
private Map<String, Object> getHeaders(Message<?> operand) {
HashMap<String, Object> headers = new HashMap<>(operand.getHeaders());
this.ignoredHeaders.forEach(headers::remove);
return headers;
HashMap<String, Object> headersToFilter = new HashMap<>(operand.getHeaders());
this.ignoredHeaders.forEach(headersToFilter::remove);
return headersToFilter;
}
}