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

Sonar fixes

Browse Source 
Hidden fields, `o.s.i.a*` through `o.s.i.j*`.
This commit is contained in:
Gary Russell
2019-01-08 15:03:00 -05:00
committed by Artem Bilan
parent 6eeec50b4a
commit e40cfe101e
53 changed files with 486 additions and 467 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AbstractAmqpOutboundEndpoint.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.
@@ -486,19 +486,19 @@ public abstract class AbstractAmqpOutboundEndpoint extends AbstractReplyProducin
}
protected String generateExchangeName(Message<?> requestMessage) {
String exchangeName = this.exchangeName;
String exchange = this.exchangeName;
if (this.exchangeNameGenerator != null) {
exchangeName = this.exchangeNameGenerator.processMessage(requestMessage);
exchange = this.exchangeNameGenerator.processMessage(requestMessage);
}
return exchangeName;
return exchange;
}
protected String generateRoutingKey(Message<?> requestMessage) {
String routingKey = this.routingKey;
String key = this.routingKey;
if (this.routingKeyGenerator != null) {
routingKey = this.routingKeyGenerator.processMessage(requestMessage);
key = this.routingKeyGenerator.processMessage(requestMessage);
}
return routingKey;
return key;
}
protected void addDelayProperty(Message<?> message, org.springframework.amqp.core.Message amqpMessage) {

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

@@ -425,10 +425,10 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport
Deque<ChannelInterceptor> interceptorStack = null;
boolean sent = false;
boolean metricsProcessed = false;
MetricsContext metrics = null;
boolean countsEnabled = this.countsEnabled;
ChannelInterceptorList interceptors = this.interceptors;
AbstractMessageChannelMetrics channelMetrics = this.channelMetrics;
MetricsContext metricsContext = null;
boolean countsAreEnabled = this.countsEnabled;
ChannelInterceptorList interceptorList = this.interceptors;
AbstractMessageChannelMetrics metrics = this.channelMetrics;
SampleFacade sample = null;
try {
if (this.datatypes.length > 0) {
@@ -438,15 +438,15 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport
if (debugEnabled) {
logger.debug("preSend on channel '" + this + "', message: " + message);
}
if (interceptors.getSize() > 0) {
if (interceptorList.getSize() > 0) {
interceptorStack = new ArrayDeque<>();
message = interceptors.preSend(message, this, interceptorStack);
message = interceptorList.preSend(message, this, interceptorStack);
if (message == null) {
return false;
}
}
if (countsEnabled) {
metrics = channelMetrics.beforeSend();
if (countsAreEnabled) {
metricsContext = metrics.beforeSend();
if (this.metricsCaptor != null) {
sample = this.metricsCaptor.start();
}
@@ -454,7 +454,7 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport
if (sample != null) {
sample.stop(sendTimer(sent));
}
channelMetrics.afterSend(metrics, sent);
metrics.afterSend(metricsContext, sent);
metricsProcessed = true;
}
else {
@@ -465,20 +465,20 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport
logger.debug("postSend (sent=" + sent + ") on channel '" + this + "', message: " + message);
}
if (interceptorStack != null) {
interceptors.postSend(message, this, sent);
interceptors.afterSendCompletion(message, this, sent, null, interceptorStack);
interceptorList.postSend(message, this, sent);
interceptorList.afterSendCompletion(message, this, sent, null, interceptorStack);
}
return sent;
}
catch (Exception e) {
if (countsEnabled && !metricsProcessed) {
if (countsAreEnabled && !metricsProcessed) {
if (sample != null) {
sample.stop(buildSendTimer(false, e.getClass().getSimpleName()));
}
channelMetrics.afterSend(metrics, false);
metrics.afterSend(metricsContext, false);
}
if (interceptorStack != null) {
interceptors.afterSendCompletion(message, this, sent, e, interceptorStack);
interceptorList.afterSendCompletion(message, this, sent, e, interceptorStack);
}
throw IntegrationUtils.wrapInDeliveryExceptionIfNecessary(message,
() -> "failed to send Message to channel '" + this.getComponentName() + "'", e);

7
spring-integration-core/src/main/java/org/springframework/integration/channel/DirectChannel.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.
@@ -85,8 +85,9 @@ public class DirectChannel extends AbstractSubscribableChannel {
protected void onInit() {
super.onInit();
if (this.maxSubscribers == null) {
Integer maxSubscribers = this.getIntegrationProperty(IntegrationProperties.CHANNELS_MAX_UNICAST_SUBSCRIBERS, Integer.class);
this.setMaxSubscribers(maxSubscribers);
Integer max = this.getIntegrationProperty(IntegrationProperties.CHANNELS_MAX_UNICAST_SUBSCRIBERS,
Integer.class);
this.setMaxSubscribers(max);
}
}

10
spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.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.
@@ -316,17 +316,17 @@ public class ConsumerEndpointFactoryBean
if (this.autoStartup != null) {
this.endpoint.setAutoStartup(this.autoStartup);
}
int phase = this.phase;
int phaseToSet = this.phase;
if (!this.isPhaseSet) {
if (this.endpoint instanceof PollingConsumer) {
phase = Integer.MAX_VALUE / 2;
phaseToSet = Integer.MAX_VALUE / 2;
}
else {
phase = Integer.MIN_VALUE;
phaseToSet = Integer.MIN_VALUE;
}
}
this.endpoint.setPhase(phase);
this.endpoint.setPhase(phaseToSet);
this.endpoint.setRole(this.role);
if (this.taskScheduler != null) {
this.endpoint.setTaskScheduler(this.taskScheduler);

18
spring-integration-core/src/main/java/org/springframework/integration/config/annotation/AbstractMethodAnnotationPostProcessor.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.
@@ -117,24 +117,20 @@ public abstract class AbstractMethodAnnotationPostProcessor<T extends Annotation
Assert.notNull(beanFactory, "'beanFactory' must not be null");
this.messageHandlerAttributes.add(SEND_TIMEOUT_ATTRIBUTE);
this.beanFactory = beanFactory;
ConversionService conversionService = this.beanFactory.getConversionService();
if (conversionService != null) {
this.conversionService = conversionService;
}
else {
this.conversionService = DefaultConversionService.getSharedInstance();
}
this.conversionService = this.beanFactory.getConversionService() != null
? this.beanFactory.getConversionService()
: DefaultConversionService.getSharedInstance();
this.channelResolver = new BeanFactoryChannelResolver(beanFactory);
this.annotationType = (Class<T>) GenericTypeResolver.resolveTypeArgument(this.getClass(),
MethodAnnotationPostProcessor.class);
Disposables disposables = null;
Disposables disposablesBean = null;
try {
disposables = beanFactory.getBean(Disposables.class);
disposablesBean = beanFactory.getBean(Disposables.class);
}
catch (Exception e) {
// NOSONAR - only for test cases
}
this.disposables = disposables;
this.disposables = disposablesBean;
}

8
spring-integration-core/src/main/java/org/springframework/integration/dsl/EndpointSpec.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.
@@ -88,9 +88,9 @@ public abstract class EndpointSpec<S extends EndpointSpec<S, F, H>, F extends Be
* @see PollerSpec
*/
public S poller(PollerSpec pollerMetadataSpec) {
Map<Object, String> componentsToRegister = pollerMetadataSpec.getComponentsToRegister();
if (componentsToRegister != null) {
this.componentsToRegister.putAll(componentsToRegister);
Map<Object, String> components = pollerMetadataSpec.getComponentsToRegister();
if (components != null) {
this.componentsToRegister.putAll(components);
}
return poller(pollerMetadataSpec.get());
}

10
spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.java
View File

@@ -3095,14 +3095,14 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
private boolean isOutputChannelRequired() {
if (this.currentComponent != null) {
Object currentComponent = this.currentComponent;
Object currentElement = this.currentComponent;
if (AopUtils.isAopProxy(currentComponent)) {
currentComponent = extractProxyTarget(currentComponent);
if (AopUtils.isAopProxy(currentElement)) {
currentElement = extractProxyTarget(currentElement);
}
return currentComponent instanceof AbstractMessageProducingHandler
|| currentComponent instanceof SourcePollingChannelAdapterSpec;
return currentElement instanceof AbstractMessageProducingHandler
|| currentElement instanceof SourcePollingChannelAdapterSpec;
}
return false;
}

18
spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractPollingEndpoint.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.
@@ -250,23 +250,23 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint implement
.collect(Collectors.toList());
}
Callable<Message<?>> pollingTask = this::doPoll;
Callable<Message<?>> task = this::doPoll;
List<Advice> adviceChain = this.adviceChain;
if (!CollectionUtils.isEmpty(adviceChain)) {
ProxyFactory proxyFactory = new ProxyFactory(pollingTask);
if (!CollectionUtils.isEmpty(adviceChain)) {
adviceChain.stream()
List<Advice> advices = this.adviceChain;
if (!CollectionUtils.isEmpty(advices)) {
ProxyFactory proxyFactory = new ProxyFactory(task);
if (!CollectionUtils.isEmpty(advices)) {
advices.stream()
.filter(advice -> !isReceiveOnlyAdvice(advice))
.forEach(proxyFactory::addAdvice);
}
pollingTask = (Callable<Message<?>>) proxyFactory.getProxy(this.beanClassLoader);
task = (Callable<Message<?>>) proxyFactory.getProxy(this.beanClassLoader);
}
if (!CollectionUtils.isEmpty(receiveOnlyAdviceChain)) {
applyReceiveOnlyAdviceChain(receiveOnlyAdviceChain);
}
return pollingTask;
return task;
}
private Runnable createPoller() {

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

@@ -219,9 +219,9 @@ public abstract class MessageProducerSupport extends AbstractEndpoint implements
* @since 4.3.10
*/
protected final boolean sendErrorMessageIfNecessary(Message<?> message, RuntimeException exception) {
MessageChannel errorChannel = getErrorChannel();
if (errorChannel != null) {
this.messagingTemplate.send(errorChannel, buildErrorMessage(message, exception));
MessageChannel channel = getErrorChannel();
if (channel != null) {
this.messagingTemplate.send(channel, buildErrorMessage(message, exception));
return true;
}
return false;

12
spring-integration-core/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java
View File

@@ -145,17 +145,13 @@ public class SourcePollingChannelAdapter extends AbstractPollingEndpoint
protected void applyReceiveOnlyAdviceChain(Collection<Advice> chain) {
if (!CollectionUtils.isEmpty(chain)) {
if (AopUtils.isAopProxy(this.source)) {
Advised source = (Advised) this.source;
this.appliedAdvices.forEach(source::removeAdvice);
for (Advice advice : chain) {
source.addAdvisor(adviceToReceiveAdvisor(advice));
}
Advised advised = (Advised) this.source;
this.appliedAdvices.forEach(advised::removeAdvice);
chain.stream().forEach(advice -> advised.addAdvisor(adviceToReceiveAdvisor(advice)));
}
else {
ProxyFactory proxyFactory = new ProxyFactory(this.source);
for (Advice advice : chain) {
proxyFactory.addAdvisor(adviceToReceiveAdvisor(advice));
}
chain.stream().forEach(advice -> proxyFactory.addAdvisor(adviceToReceiveAdvisor(advice)));
this.source = (MessageSource<?>) proxyFactory.getProxy(getBeanClassLoader());
}
this.appliedAdvices.clear();

19
spring-integration-core/src/main/java/org/springframework/integration/expression/ExpressionEvalMap.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2013-2018 the original author or authors.
* Copyright 2013-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.
@@ -282,16 +282,21 @@ public final class ExpressionEvalMap extends AbstractMap<String, Object> {
private class ExpressionEvalMapFinalBuilderImpl implements ExpressionEvalMapFinalBuilder {
ExpressionEvalMapFinalBuilderImpl() {
super();
}
@Override
public ExpressionEvalMap build() {
if (ExpressionEvalMapBuilder.this.evaluationCallback != null) {
return new ExpressionEvalMap(ExpressionEvalMapBuilder.this.expressions,
ExpressionEvalMapBuilder.this.evaluationCallback);
}
ComponentsEvaluationCallback evaluationCallback =
new ComponentsEvaluationCallback(ExpressionEvalMapBuilder.this.context,
ExpressionEvalMapBuilder.this.root, ExpressionEvalMapBuilder.this.returnType);
return new ExpressionEvalMap(ExpressionEvalMapBuilder.this.expressions, evaluationCallback);
else {
return new ExpressionEvalMap(ExpressionEvalMapBuilder.this.expressions,
new ComponentsEvaluationCallback(ExpressionEvalMapBuilder.this.context,
ExpressionEvalMapBuilder.this.root, ExpressionEvalMapBuilder.this.returnType));
}
}
}
@@ -300,6 +305,10 @@ public final class ExpressionEvalMap extends AbstractMap<String, Object> {
private class ExpressionEvalMapComponentsBuilderImpl extends ExpressionEvalMapFinalBuilderImpl
implements ExpressionEvalMapComponentsBuilder {
ExpressionEvalMapComponentsBuilderImpl() {
super();
}
@Override
public ExpressionEvalMapComponentsBuilder usingEvaluationContext(EvaluationContext context) {
return ExpressionEvalMapBuilder.this.usingEvaluationContext(context);

8
spring-integration-core/src/main/java/org/springframework/integration/filter/MessageFilter.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.
@@ -173,9 +173,9 @@ public class MessageFilter extends AbstractReplyProducingPostProcessingMessageHa
@Override
public Object postProcess(Message<?> message, Object result) {
if (result == null) {
MessageChannel discardChannel = getDiscardChannel();
if (discardChannel != null) {
this.messagingTemplate.send(discardChannel, message);
MessageChannel channel = getDiscardChannel();
if (channel != null) {
this.messagingTemplate.send(channel, message);
}
if (this.throwExceptionOnRejection) {
throw new MessageRejectedException(message, "MessageFilter '" + this.getComponentName()

13
spring-integration-core/src/main/java/org/springframework/integration/gateway/AnnotationGatewayProxyFactoryBean.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2017 the original author or authors.
* Copyright 2017-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.
@@ -38,6 +38,7 @@ import org.springframework.util.StringUtils;
* Otherwise the default state is applied.
*
* @author Artem Bilan
* @author Gary Russell
*
* @since 5.0
*/
@@ -47,17 +48,17 @@ public class AnnotationGatewayProxyFactoryBean extends GatewayProxyFactoryBean {
public AnnotationGatewayProxyFactoryBean(Class<?> serviceInterface) {
super(serviceInterface);
AnnotationAttributes gatewayAttributes =
AnnotationAttributes annotationAttributes =
AnnotatedElementUtils.getMergedAnnotationAttributes(serviceInterface,
MessagingGateway.class.getName(), false, true);
if (gatewayAttributes == null) {
gatewayAttributes = AnnotationUtils.getAnnotationAttributes(
if (annotationAttributes == null) {
annotationAttributes = AnnotationUtils.getAnnotationAttributes(
AnnotationUtils.synthesizeAnnotation(MessagingGateway.class), false, true);
}
this.gatewayAttributes = gatewayAttributes;
this.gatewayAttributes = annotationAttributes;
String id = gatewayAttributes.getString("name");
String id = annotationAttributes.getString("name");
if (StringUtils.hasText(id)) {
setBeanName(id);
}

8
spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.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.
@@ -634,10 +634,10 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint
}
Map<String, Object> headers = null;
// We don't want to eagerly resolve the error channel here
Object errorChannel = this.errorChannel == null ? this.errorChannelName : this.errorChannel;
if (errorChannel != null && method.getReturnType().equals(void.class)) {
Object errorChannelForVoidReturn = this.errorChannel == null ? this.errorChannelName : this.errorChannel;
if (errorChannelForVoidReturn != null && method.getReturnType().equals(void.class)) {
headers = new HashMap<>();
headers.put(MessageHeaders.ERROR_CHANNEL, errorChannel);
headers.put(MessageHeaders.ERROR_CHANNEL, errorChannelForVoidReturn);
}
if (getMessageBuilderFactory() instanceof DefaultMessageBuilderFactory) {

88
spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java
View File

@@ -411,19 +411,19 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint
protected void send(Object object) {
this.initializeIfNecessary();
Assert.notNull(object, "request must not be null");
MessageChannel requestChannel = getRequestChannel();
Assert.state(requestChannel != null,
MessageChannel channel = getRequestChannel();
Assert.state(channel != null,
"send is not supported, because no request channel has been configured");
try {
if (this.countsEnabled) {
this.messageCount.incrementAndGet();
}
this.messagingTemplate.convertAndSend(requestChannel, object, this.historyWritingPostProcessor);
this.messagingTemplate.convertAndSend(channel, object, this.historyWritingPostProcessor);
}
catch (Exception e) {
MessageChannel errorChannel = getErrorChannel();
if (errorChannel != null) {
this.messagingTemplate.send(errorChannel, new ErrorMessage(e));
MessageChannel errorChan = getErrorChannel();
if (errorChan != null) {
this.messagingTemplate.send(errorChan, new ErrorMessage(e));
}
else {
this.rethrow(e, "failed to send message");
@@ -434,37 +434,37 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint
@Nullable
protected Object receive() {
this.initializeIfNecessary();
MessageChannel replyChannel = getReplyChannel();
Assert.state(replyChannel != null && (replyChannel instanceof PollableChannel),
MessageChannel channel = getReplyChannel();
Assert.state(channel != null && (channel instanceof PollableChannel),
"receive is not supported, because no pollable reply channel has been configured");
return this.messagingTemplate.receiveAndConvert(replyChannel, Object.class);
return this.messagingTemplate.receiveAndConvert(channel, Object.class);
}
@Nullable
protected Message<?> receiveMessage() {
initializeIfNecessary();
MessageChannel replyChannel = getReplyChannel();
Assert.state(replyChannel instanceof PollableChannel,
MessageChannel channel = getReplyChannel();
Assert.state(channel instanceof PollableChannel,
"receive is not supported, because no pollable reply channel has been configured");
return this.messagingTemplate.receive(replyChannel);
return this.messagingTemplate.receive(channel);
}
@Nullable
protected Object receive(long timeout) {
this.initializeIfNecessary();
MessageChannel replyChannel = getReplyChannel();
Assert.state(replyChannel != null && (replyChannel instanceof PollableChannel),
MessageChannel channel = getReplyChannel();
Assert.state(channel != null && (channel instanceof PollableChannel),
"receive is not supported, because no pollable reply channel has been configured");
return this.messagingTemplate.receiveAndConvert(replyChannel, timeout);
return this.messagingTemplate.receiveAndConvert(channel, timeout);
}
@Nullable
protected Message<?> receiveMessage(long timeout) {
initializeIfNecessary();
MessageChannel replyChannel = getReplyChannel();
Assert.state(replyChannel instanceof PollableChannel,
MessageChannel channel = getReplyChannel();
Assert.state(channel instanceof PollableChannel,
"receive is not supported, because no pollable reply channel has been configured");
return this.messagingTemplate.receive(replyChannel, timeout);
return this.messagingTemplate.receive(channel, timeout);
}
@Nullable
@@ -482,8 +482,8 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint
private Object doSendAndReceive(Object object, boolean shouldConvert) {
this.initializeIfNecessary();
Assert.notNull(object, "request must not be null");
MessageChannel requestChannel = getRequestChannel();
if (requestChannel == null) {
MessageChannel channel = getRequestChannel();
if (channel == null) {
throw new MessagingException("No request channel available. Cannot send request message.");
}
@@ -497,7 +497,7 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint
this.messageCount.incrementAndGet();
}
if (shouldConvert) {
reply = this.messagingTemplate.convertSendAndReceive(requestChannel, object, Object.class,
reply = this.messagingTemplate.convertSendAndReceive(channel, object, Object.class,
this.historyWritingPostProcessor);
if (reply instanceof Throwable) {
error = (Throwable) reply;
@@ -508,7 +508,7 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint
? (Message<?>) object : this.requestMapper.toMessage(object);
Assert.state(requestMessage != null, () -> "request mapper resulted in no message for " + object);
requestMessage = this.historyWritingPostProcessor.postProcessMessage(requestMessage);
reply = this.messagingTemplate.sendAndReceive(requestChannel, requestMessage);
reply = this.messagingTemplate.sendAndReceive(channel, requestMessage);
if (reply instanceof ErrorMessage) {
error = ((ErrorMessage) reply).getPayload();
}
@@ -530,12 +530,12 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint
}
if (error != null) {
MessageChannel errorChannel = getErrorChannel();
if (errorChannel != null) {
MessageChannel errorChan = getErrorChannel();
if (errorChan != null) {
ErrorMessage errorMessage = buildErrorMessage(requestMessage, error);
Message<?> errorFlowReply = null;
try {
errorFlowReply = this.messagingTemplate.sendAndReceive(errorChannel, errorMessage);
errorFlowReply = this.messagingTemplate.sendAndReceive(errorChan, errorMessage);
}
catch (Exception errorFlowFailure) {
throw new MessagingException(errorMessage, "failure occurred in error-handling flow",
@@ -572,14 +572,14 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint
protected Mono<Message<?>> sendAndReceiveMessageReactive(Object object) {
initializeIfNecessary();
Assert.notNull(object, "request must not be null");
MessageChannel requestChannel = getRequestChannel();
if (requestChannel == null) {
MessageChannel channel = getRequestChannel();
if (channel == null) {
throw new MessagingException("No request channel available. Cannot send request message.");
}
registerReplyMessageCorrelatorIfNecessary();
return doSendAndReceiveMessageReactive(requestChannel, object, false);
return doSendAndReceiveMessageReactive(channel, object, false);
}
@SuppressWarnings("unchecked")
@@ -603,13 +603,13 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint
Object originalReplyChannelHeader = message.getHeaders().getReplyChannel();
Object originalErrorChannelHeader = message.getHeaders().getErrorChannel();
FutureReplyChannel replyChannel = new FutureReplyChannel();
FutureReplyChannel replyChan = new FutureReplyChannel();
Message<?> requestMessage = MutableMessageBuilder.fromMessage(message)
.setReplyChannel(replyChannel)
.setReplyChannel(replyChan)
.setHeader(this.messagingTemplate.getSendTimeoutHeader(), null)
.setHeader(this.messagingTemplate.getReceiveTimeoutHeader(), null)
.setErrorChannel(replyChannel)
.setErrorChannel(replyChan)
.build();
if (requestChannel instanceof ReactiveStreamsSubscribableChannel) {
@@ -631,7 +631,7 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint
}
}
return Mono.fromFuture(replyChannel.messageFuture)
return Mono.fromFuture(replyChan.messageFuture)
.doOnSubscribe(s -> {
if (!error && this.countsEnabled) {
this.messageCount.incrementAndGet();
@@ -662,11 +662,11 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint
if (logger.isDebugEnabled()) {
logger.debug("failure occurred in gateway sendAndReceiveReactive: " + exception.getMessage());
}
MessageChannel errorChannel = getErrorChannel();
if (errorChannel != null) {
MessageChannel channel = getErrorChannel();
if (channel != null) {
ErrorMessage errorMessage = buildErrorMessage(requestMessage, exception);
try {
return doSendAndReceiveMessageReactive(errorChannel, errorMessage, true);
return doSendAndReceiveMessageReactive(channel, errorMessage, true);
}
catch (Exception errorFlowFailure) {
throw new MessagingException(errorMessage, "failure occurred in error-handling flow", errorFlowFailure);
@@ -741,8 +741,8 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint
}
protected void registerReplyMessageCorrelatorIfNecessary() {
MessageChannel replyChannel = getReplyChannel();
if (replyChannel != null && this.replyMessageCorrelator == null) {
MessageChannel replyChan = getReplyChannel();
if (replyChan != null && this.replyMessageCorrelator == null) {
boolean shouldStartCorrelator;
synchronized (this.replyMessageCorrelatorMonitor) {
if (this.replyMessageCorrelator != null) {
@@ -754,24 +754,24 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint
handler.setBeanFactory(getBeanFactory());
}
handler.afterPropertiesSet();
if (replyChannel instanceof SubscribableChannel) {
correlator = new EventDrivenConsumer((SubscribableChannel) replyChannel, handler);
if (replyChan instanceof SubscribableChannel) {
correlator = new EventDrivenConsumer((SubscribableChannel) replyChan, handler);
}
else if (replyChannel instanceof PollableChannel) {
PollingConsumer endpoint = new PollingConsumer((PollableChannel) replyChannel, handler);
else if (replyChan instanceof PollableChannel) {
PollingConsumer endpoint = new PollingConsumer((PollableChannel) replyChan, handler);
endpoint.setBeanFactory(getBeanFactory());
endpoint.setReceiveTimeout(this.replyTimeout);
endpoint.afterPropertiesSet();
correlator = endpoint;
}
else if (replyChannel instanceof ReactiveStreamsSubscribableChannel) {
else if (replyChan instanceof ReactiveStreamsSubscribableChannel) {
ReactiveStreamsConsumer endpoint =
new ReactiveStreamsConsumer(replyChannel, (Subscriber<Message<?>>) handler);
new ReactiveStreamsConsumer(replyChan, (Subscriber<Message<?>>) handler);
endpoint.afterPropertiesSet();
correlator = endpoint;
}
else {
throw new MessagingException("Unsupported 'replyChannel' type [" + replyChannel.getClass() + "]."
throw new MessagingException("Unsupported 'replyChannel' type [" + replyChan.getClass() + "]."
+ "SubscribableChannel or PollableChannel type are supported.");
}
this.replyMessageCorrelator = correlator;

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

@@ -147,23 +147,23 @@ public abstract class AbstractMessageHandler extends IntegrationObjectSupport
this.logger.debug(this + " received message: " + message);
}
MetricsContext start = null;
boolean countsEnabled = this.countsEnabled;
AbstractMessageHandlerMetrics handlerMetrics = this.handlerMetrics;
boolean countsAreEnabled = this.countsEnabled;
AbstractMessageHandlerMetrics metrics = this.handlerMetrics;
SampleFacade sample = null;
if (countsEnabled && this.metricsCaptor != null) {
if (countsAreEnabled && this.metricsCaptor != null) {
sample = this.metricsCaptor.start();
}
try {
if (this.shouldTrack) {
message = MessageHistory.write(message, this, getMessageBuilderFactory());
}
if (countsEnabled) {
start = handlerMetrics.beforeHandle();
if (countsAreEnabled) {
start = metrics.beforeHandle();
handleMessageInternal(message);
if (sample != null) {
sample.stop(sendTimer());
}
handlerMetrics.afterHandle(start, true);
metrics.afterHandle(start, true);
}
else {
handleMessageInternal(message);
@@ -173,8 +173,8 @@ public abstract class AbstractMessageHandler extends IntegrationObjectSupport
if (sample != null) {
sample.stop(buildSendTimer(false, e.getClass().getSimpleName()));
}
if (countsEnabled) {
handlerMetrics.afterHandle(start, false);
if (countsAreEnabled) {
metrics.afterHandle(start, false);
}
throw IntegrationUtils.wrapInHandlingExceptionIfNecessary(message,
() -> "error occurred in message handler [" + this + "]", e);

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

@@ -418,9 +418,9 @@ public abstract class AbstractMessageProducingHandler extends AbstractMessageHan
*/
protected void sendOutput(Object output, @Nullable Object replyChannelArg, boolean useArgChannel) {
Object replyChannel = replyChannelArg;
MessageChannel outputChannel = getOutputChannel();
if (!useArgChannel && outputChannel != null) {
replyChannel = outputChannel;
MessageChannel outChannel = getOutputChannel();
if (!useArgChannel && outChannel != null) {
replyChannel = outChannel;
}
if (replyChannel == null) {
throw new DestinationResolutionException("no output-channel or replyChannel header available");

8
spring-integration-core/src/main/java/org/springframework/integration/handler/DelayHandler.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.
@@ -299,16 +299,16 @@ public class DelayHandler extends AbstractReplyProducingMessageHandler implement
}
private MessageHandler createReleaseMessageTask() {
ReleaseMessageHandler releaseHandler = new ReleaseMessageHandler();
ReleaseMessageHandler handler = new ReleaseMessageHandler();
if (!CollectionUtils.isEmpty(this.delayedAdviceChain)) {
ProxyFactory proxyFactory = new ProxyFactory(releaseHandler);
ProxyFactory proxyFactory = new ProxyFactory(handler);
for (Advice advice : this.delayedAdviceChain) {
proxyFactory.addAdvice(advice);
}
return (MessageHandler) proxyFactory.getProxy(getApplicationContext().getClassLoader());
}
return releaseHandler;
return handler;
}
@Override

98
spring-integration-core/src/main/java/org/springframework/integration/handler/support/MessagingMethodInvokerHelper.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.
@@ -354,6 +354,7 @@ public class MessagingMethodInvokerHelper<T> extends AbstractExpressionEvaluator
private MessagingMethodInvokerHelper(Object targetObject, Class<? extends Annotation> annotationType,
String methodName, Class<?> expectedType, boolean canProcessMessageList) {
this.annotationType = annotationType;
this.methodName = methodName;
this.canProcessMessageList = canProcessMessageList;
@@ -367,15 +368,15 @@ public class MessagingMethodInvokerHelper<T> extends AbstractExpressionEvaluator
this.targetObject = targetObject;
Map<String, Map<Class<?>, HandlerMethod>> handlerMethodsForTarget =
findHandlerMethodsForTarget(targetObject, annotationType, methodName, expectedType != null);
Map<Class<?>, HandlerMethod> handlerMethods = handlerMethodsForTarget.get(CANDIDATE_METHODS);
Map<Class<?>, HandlerMethod> handlerMessageMethods = handlerMethodsForTarget.get(CANDIDATE_MESSAGE_METHODS);
if ((handlerMethods.size() == 1 && handlerMessageMethods.isEmpty()) ||
(handlerMessageMethods.size() == 1 && handlerMethods.isEmpty())) {
if (handlerMethods.size() == 1) {
this.handlerMethod = handlerMethods.values().iterator().next();
Map<Class<?>, HandlerMethod> methods = handlerMethodsForTarget.get(CANDIDATE_METHODS);
Map<Class<?>, HandlerMethod> messageMethods = handlerMethodsForTarget.get(CANDIDATE_MESSAGE_METHODS);
if ((methods.size() == 1 && messageMethods.isEmpty()) ||
(messageMethods.size() == 1 && methods.isEmpty())) {
if (methods.size() == 1) {
this.handlerMethod = methods.values().iterator().next();
}
else {
this.handlerMethod = handlerMessageMethods.values().iterator().next();
this.handlerMethod = messageMethods.values().iterator().next();
}
this.handlerMethods = null;
this.handlerMessageMethods = null;
@@ -383,8 +384,8 @@ public class MessagingMethodInvokerHelper<T> extends AbstractExpressionEvaluator
}
else {
this.handlerMethod = null;
this.handlerMethods = handlerMethods;
this.handlerMessageMethods = handlerMessageMethods;
this.handlerMethods = methods;
this.handlerMessageMethods = messageMethods;
this.handlerMethodsList = new LinkedList<>();
//TODO Consider to use global option to determine a precedence of methods
@@ -440,12 +441,13 @@ public class MessagingMethodInvokerHelper<T> extends AbstractExpressionEvaluator
private boolean canReturnExpectedType(AnnotatedMethodFilter filter, Class<?> targetType,
TypeConverter typeConverter) {
if (this.expectedType == null) {
return true;
}
List<Method> methods = filter.filter(Arrays.asList(ReflectionUtils.getAllDeclaredMethods(targetType)));
for (Method method : methods) {
if (typeConverter.canConvert(TypeDescriptor.valueOf(method.getReturnType()), this.expectedType)) {
for (Method candidate : methods) {
if (typeConverter.canConvert(TypeDescriptor.valueOf(candidate.getReturnType()), this.expectedType)) {
return true;
}
}
@@ -687,10 +689,10 @@ public class MessagingMethodInvokerHelper<T> extends AbstractExpressionEvaluator
}
private Map<String, Map<Class<?>, HandlerMethod>> findHandlerMethodsForTarget(final Object targetObject,
final Class<? extends Annotation> annotationType, final String methodNameToUse,
final Class<? extends Annotation> annotationType, final String methodNameArg,
final boolean requiresReply) {
Map<String, Map<Class<?>, HandlerMethod>> handlerMethods = new HashMap<>();
Map<String, Map<Class<?>, HandlerMethod>> methods = new HashMap<>();
final Map<Class<?>, HandlerMethod> candidateMethods = new HashMap<>();
final Map<Class<?>, HandlerMethod> candidateMessageMethods = new HashMap<>();
@@ -700,21 +702,21 @@ public class MessagingMethodInvokerHelper<T> extends AbstractExpressionEvaluator
final AtomicReference<Class<?>> ambiguousFallbackMessageGenericType = new AtomicReference<>();
final Class<?> targetClass = getTargetClass(targetObject);
final String methodName;
final String methodNameToUse;
if (methodNameToUse == null) {
if (methodNameArg == null) {
if (Function.class.isAssignableFrom(targetClass)) {
methodName = "apply";
methodNameToUse = "apply";
}
else if (Consumer.class.isAssignableFrom(targetClass)) {
methodName = "accept";
methodNameToUse = "accept";
}
else {
methodName = null;
methodNameToUse = null;
}
}
else {
methodName = methodNameToUse;
methodNameToUse = methodNameArg;
}
@@ -739,10 +741,10 @@ public class MessagingMethodInvokerHelper<T> extends AbstractExpressionEvaluator
if (requiresReply && void.class.equals(method1.getReturnType())) {
return;
}
if (methodName != null && !methodName.equals(method1.getName())) {
if (methodNameToUse != null && !methodNameToUse.equals(method1.getName())) {
return;
}
if (methodName == null
if (methodNameToUse == null
&& ObjectUtils.containsElement(new String[] { "start", "stop", "isRunning" }, method1.getName())) {
return;
}
@@ -820,14 +822,14 @@ public class MessagingMethodInvokerHelper<T> extends AbstractExpressionEvaluator
if (candidateMethods.isEmpty() && candidateMessageMethods.isEmpty() && fallbackMethods.isEmpty()
&& fallbackMessageMethods.isEmpty()) {
findSingleSpecifMethodOnInterfacesIfProxy(targetObject, methodName, candidateMessageMethods,
findSingleSpecifMethodOnInterfacesIfProxy(targetObject, methodNameToUse, candidateMessageMethods,
candidateMethods);
}
if (!candidateMethods.isEmpty() || !candidateMessageMethods.isEmpty()) {
handlerMethods.put(CANDIDATE_METHODS, candidateMethods);
handlerMethods.put(CANDIDATE_MESSAGE_METHODS, candidateMessageMethods);
return handlerMethods;
methods.put(CANDIDATE_METHODS, candidateMethods);
methods.put(CANDIDATE_MESSAGE_METHODS, candidateMessageMethods);
return methods;
}
if ((ambiguousFallbackType.get() != null
|| ambiguousFallbackMessageGenericType.get() != null)
@@ -855,16 +857,16 @@ public class MessagingMethodInvokerHelper<T> extends AbstractExpressionEvaluator
}
}
if (frameworkMethods.size() == 1) {
Method method = org.springframework.util.ClassUtils.getMostSpecificMethod(frameworkMethods.get(0),
targetObject.getClass());
Method frameworkMethod = org.springframework.util.ClassUtils.getMostSpecificMethod(
frameworkMethods.get(0), targetObject.getClass());
InvocableHandlerMethod invocableHandlerMethod =
this.messageHandlerMethodFactory.createInvocableHandlerMethod(targetObject,
method);
HandlerMethod handlerMethod = new HandlerMethod(invocableHandlerMethod, this.canProcessMessageList);
checkSpelInvokerRequired(targetClass, method, handlerMethod);
handlerMethods.put(CANDIDATE_METHODS, Collections.singletonMap(Object.class, handlerMethod));
handlerMethods.put(CANDIDATE_MESSAGE_METHODS, candidateMessageMethods);
return handlerMethods;
frameworkMethod);
HandlerMethod theHandlerMethod = new HandlerMethod(invocableHandlerMethod, this.canProcessMessageList);
checkSpelInvokerRequired(targetClass, frameworkMethod, theHandlerMethod);
methods.put(CANDIDATE_METHODS, Collections.singletonMap(Object.class, theHandlerMethod));
methods.put(CANDIDATE_MESSAGE_METHODS, candidateMessageMethods);
return methods;
}
}
@@ -880,9 +882,9 @@ public class MessagingMethodInvokerHelper<T> extends AbstractExpressionEvaluator
+ "] for method match: "
+ fallbackMethods.values());
handlerMethods.put(CANDIDATE_METHODS, fallbackMethods);
handlerMethods.put(CANDIDATE_MESSAGE_METHODS, fallbackMessageMethods);
return handlerMethods;
methods.put(CANDIDATE_METHODS, fallbackMethods);
methods.put(CANDIDATE_MESSAGE_METHODS, fallbackMessageMethods);
return methods;
}
private void findSingleSpecifMethodOnInterfacesIfProxy(final Object targetObject, final String methodName,
@@ -909,15 +911,15 @@ public class MessagingMethodInvokerHelper<T> extends AbstractExpressionEvaluator
.getMostSpecificMethod(theMethod, targetObject.getClass());
InvocableHandlerMethod invocableHandlerMethod =
this.messageHandlerMethodFactory.createInvocableHandlerMethod(targetObject, theMethod);
HandlerMethod handlerMethod = new HandlerMethod(invocableHandlerMethod, this.canProcessMessageList);
checkSpelInvokerRequired(targetClass.get(), theMethod, handlerMethod);
Class<?> targetParameterType = handlerMethod.getTargetParameterType();
if (handlerMethod.isMessageMethod()) {
HandlerMethod theHandlerMethod = new HandlerMethod(invocableHandlerMethod, this.canProcessMessageList);
checkSpelInvokerRequired(targetClass.get(), theMethod, theHandlerMethod);
Class<?> targetParameterType = theHandlerMethod.getTargetParameterType();
if (theHandlerMethod.isMessageMethod()) {
if (candidateMessageMethods.containsKey(targetParameterType)) {
throw new IllegalArgumentException("Found more than one method match for type " +
"[Message<" + targetParameterType + ">]");
}
candidateMessageMethods.put(targetParameterType, handlerMethod);
candidateMessageMethods.put(targetParameterType, theHandlerMethod);
}
else {
if (candidateMethods.containsKey(targetParameterType)) {
@@ -930,15 +932,15 @@ public class MessagingMethodInvokerHelper<T> extends AbstractExpressionEvaluator
}
throw new IllegalArgumentException(exceptionMessage);
}
candidateMethods.put(targetParameterType, handlerMethod);
candidateMethods.put(targetParameterType, theHandlerMethod);
}
}
}
}
private void checkSpelInvokerRequired(final Class<?> targetClass, Method methodArg, HandlerMethod handlerMethod) {
Method method = AopUtils.getMostSpecificMethod(methodArg, targetClass);
UseSpelInvoker useSpel = AnnotationUtils.findAnnotation(method, UseSpelInvoker.class);
UseSpelInvoker useSpel = AnnotationUtils.findAnnotation(AopUtils.getMostSpecificMethod(methodArg, targetClass),
UseSpelInvoker.class);
if (useSpel == null) {
useSpel = AnnotationUtils.findAnnotation(targetClass, UseSpelInvoker.class);
}
@@ -1019,14 +1021,14 @@ public class MessagingMethodInvokerHelper<T> extends AbstractExpressionEvaluator
}
private HandlerMethod findClosestMatch(Class<?> payloadType) {
for (Map<Class<?>, HandlerMethod> handlerMethods : this.handlerMethodsList) {
Set<Class<?>> candidates = handlerMethods.keySet();
for (Map<Class<?>, HandlerMethod> methods : this.handlerMethodsList) {
Set<Class<?>> candidates = methods.keySet();
Class<?> match = null;
if (!CollectionUtils.isEmpty(candidates)) {
match = ClassUtils.findClosestMatch(payloadType, candidates, true);
}
if (match != null) {
return handlerMethods.get(match);
return methods.get(match);
}
}
return null;

10
spring-integration-core/src/main/java/org/springframework/integration/history/MessageHistoryConfigurer.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.
@@ -118,16 +118,16 @@ public class MessageHistoryConfigurer implements SmartLifecycle, BeanFactoryAwar
Assert.notNull(componentNamePatternsSet, "'componentNamePatternsSet' must not be null");
Assert.state(!this.running, "'componentNamePatternsSet' cannot be changed without invoking stop() first");
for (String s : componentNamePatternsSet) {
String[] componentNamePatterns = StringUtils.delimitedListToStringArray(s, ",", " ");
Arrays.sort(componentNamePatterns);
String[] patterns = StringUtils.delimitedListToStringArray(s, ",", " ");
Arrays.sort(patterns);
if (this.componentNamePatternsExplicitlySet
&& !Arrays.equals(this.componentNamePatterns, componentNamePatterns)) {
&& !Arrays.equals(this.componentNamePatterns, patterns)) {
throw new BeanDefinitionValidationException("When more than one message history definition " +
"(@EnableMessageHistory or <message-history>)" +
" is found in the context, they all must have the same 'componentNamePatterns'");
}
else {
this.componentNamePatterns = componentNamePatterns;
this.componentNamePatterns = patterns;
this.componentNamePatternsExplicitlySet = true;
}
}

24
spring-integration-file/src/main/java/org/springframework/integration/file/FileWritingMessageHandler.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.
@@ -396,39 +396,39 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand
* representing the bits that were set in the chmod value.
*/
BitSet bits = BitSet.valueOf(new byte[] { (byte) chmod, (byte) (chmod >> 8) });
final Set<PosixFilePermission> permissions = new HashSet<>();
final Set<PosixFilePermission> posixPermissions = new HashSet<>();
bits.stream().forEach(b -> {
switch (b) {
case 0:
permissions.add(PosixFilePermission.OTHERS_EXECUTE);
posixPermissions.add(PosixFilePermission.OTHERS_EXECUTE);
break;
case 1:
permissions.add(PosixFilePermission.OTHERS_WRITE);
posixPermissions.add(PosixFilePermission.OTHERS_WRITE);
break;
case 2:
permissions.add(PosixFilePermission.OTHERS_READ);
posixPermissions.add(PosixFilePermission.OTHERS_READ);
break;
case 3:
permissions.add(PosixFilePermission.GROUP_EXECUTE);
posixPermissions.add(PosixFilePermission.GROUP_EXECUTE);
break;
case 4:
permissions.add(PosixFilePermission.GROUP_WRITE);
posixPermissions.add(PosixFilePermission.GROUP_WRITE);
break;
case 5:
permissions.add(PosixFilePermission.GROUP_READ);
posixPermissions.add(PosixFilePermission.GROUP_READ);
break;
case 6:
permissions.add(PosixFilePermission.OWNER_EXECUTE);
posixPermissions.add(PosixFilePermission.OWNER_EXECUTE);
break;
case 7:
permissions.add(PosixFilePermission.OWNER_WRITE);
posixPermissions.add(PosixFilePermission.OWNER_WRITE);
break;
case 8:
permissions.add(PosixFilePermission.OWNER_READ);
posixPermissions.add(PosixFilePermission.OWNER_READ);
break;
}
});
this.permissions = permissions;
this.permissions = posixPermissions;
}
/**

7
spring-integration-file/src/main/java/org/springframework/integration/file/HeadDirectoryScanner.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.
@@ -38,9 +38,8 @@ public class HeadDirectoryScanner extends DefaultDirectoryScanner {
private final HeadFilter headFilter;
public HeadDirectoryScanner(int maxNumberOfFiles) {
HeadFilter headFilter = new HeadFilter(maxNumberOfFiles);
this.headFilter = headFilter;
this.setFilter(headFilter);
this.headFilter = new HeadFilter(maxNumberOfFiles);
setFilter(this.headFilter);
}
@Override

28
spring-integration-file/src/main/java/org/springframework/integration/file/config/FileTailInboundChannelAdapterFactoryBean.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.
@@ -63,7 +63,7 @@ public class FileTailInboundChannelAdapterFactoryBean extends AbstractFactoryBea
private volatile Boolean reopen;
private volatile FileTailingMessageProducerSupport adapter;
private volatile FileTailingMessageProducerSupport tailAdapter;
private volatile String beanName;
@@ -158,40 +158,40 @@ public class FileTailInboundChannelAdapterFactoryBean extends AbstractFactoryBea
@Override
public void start() {
if (this.adapter != null) {
this.adapter.start();
if (this.tailAdapter != null) {
this.tailAdapter.start();
}
}
@Override
public void stop() {
if (this.adapter != null) {
this.adapter.stop();
if (this.tailAdapter != null) {
this.tailAdapter.stop();
}
}
@Override
public boolean isRunning() {
return this.adapter != null && this.adapter.isRunning();
return this.tailAdapter != null && this.tailAdapter.isRunning();
}
@Override
public int getPhase() {
if (this.adapter != null) {
return this.adapter.getPhase();
if (this.tailAdapter != null) {
return this.tailAdapter.getPhase();
}
return 0;
}
@Override
public boolean isAutoStartup() {
return this.adapter != null && this.adapter.isAutoStartup();
return this.tailAdapter != null && this.tailAdapter.isAutoStartup();
}
@Override
public void stop(Runnable callback) {
if (this.adapter != null) {
this.adapter.stop(callback);
if (this.tailAdapter != null) {
this.tailAdapter.stop(callback);
}
else {
callback.run();
@@ -200,7 +200,7 @@ public class FileTailInboundChannelAdapterFactoryBean extends AbstractFactoryBea
@Override
public Class<?> getObjectType() {
return this.adapter == null ? FileTailingMessageProducerSupport.class : this.adapter.getClass();
return this.tailAdapter == null ? FileTailingMessageProducerSupport.class : this.tailAdapter.getClass();
}
@Override
@@ -256,7 +256,7 @@ public class FileTailInboundChannelAdapterFactoryBean extends AbstractFactoryBea
adapter.setBeanFactory(getBeanFactory()); // NOSONAR never null
}
adapter.afterPropertiesSet();
this.adapter = adapter;
this.tailAdapter = adapter;
return adapter;
}

11
spring-integration-file/src/main/java/org/springframework/integration/file/remote/RemoteFileTemplate.java
View File

@@ -239,19 +239,18 @@ public class RemoteFileTemplate<F> implements RemoteFileOperations<F>, Initializ
@Override
public void afterPropertiesSet() {
BeanFactory beanFactory = this.beanFactory;
if (beanFactory != null) {
if (this.beanFactory != null) {
if (this.directoryExpressionProcessor != null) {
this.directoryExpressionProcessor.setBeanFactory(beanFactory);
this.directoryExpressionProcessor.setBeanFactory(this.beanFactory);
}
if (this.temporaryDirectoryExpressionProcessor != null) {
this.temporaryDirectoryExpressionProcessor.setBeanFactory(beanFactory);
this.temporaryDirectoryExpressionProcessor.setBeanFactory(this.beanFactory);
}
if (!this.fileNameGeneratorSet && this.fileNameGenerator instanceof BeanFactoryAware) {
((BeanFactoryAware) this.fileNameGenerator).setBeanFactory(beanFactory);
((BeanFactoryAware) this.fileNameGenerator).setBeanFactory(this.beanFactory);
}
if (this.fileNameProcessor != null) {
this.fileNameProcessor.setBeanFactory(beanFactory);
this.fileNameProcessor.setBeanFactory(this.beanFactory);
}
}
if (this.autoCreateDirectory) {

16
spring-integration-file/src/main/java/org/springframework/integration/file/remote/gateway/AbstractRemoteFileOutboundGateway.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.
@@ -891,11 +891,11 @@ public abstract class AbstractRemoteFileOutboundGateway<F> extends AbstractReply
}
File localFile =
new File(generateLocalDirectory(message, remoteDir), generateLocalFileName(message, remoteFilename));
FileExistsMode fileExistsMode = this.fileExistsMode;
boolean appending = FileExistsMode.APPEND.equals(fileExistsMode);
FileExistsMode existsMode = this.fileExistsMode;
boolean appending = FileExistsMode.APPEND.equals(existsMode);
boolean exists = localFile.exists();
boolean replacing = FileExistsMode.REPLACE.equals(fileExistsMode)
|| (exists && FileExistsMode.REPLACE_IF_MODIFIED.equals(fileExistsMode)
boolean replacing = FileExistsMode.REPLACE.equals(existsMode)
|| (exists && FileExistsMode.REPLACE_IF_MODIFIED.equals(existsMode)
&& localFile.lastModified() != getModified(fileInfo));
if (!exists || appending || replacing) {
OutputStream outputStream;
@@ -939,7 +939,7 @@ public abstract class AbstractRemoteFileOutboundGateway<F> extends AbstractReply
throw new MessagingException("Failed to rename local file");
}
if (this.options.contains(Option.PRESERVE_TIMESTAMP)
|| FileExistsMode.REPLACE_IF_MODIFIED.equals(fileExistsMode)) {
|| FileExistsMode.REPLACE_IF_MODIFIED.equals(existsMode)) {
localFile.setLastModified(getModified(fileInfo));
}
if (this.options.contains(Option.DELETE)) {
@@ -952,7 +952,7 @@ public abstract class AbstractRemoteFileOutboundGateway<F> extends AbstractReply
}
}
}
else if (FileExistsMode.REPLACE_IF_MODIFIED.equals(fileExistsMode)) {
else if (FileExistsMode.REPLACE_IF_MODIFIED.equals(existsMode)) {
if (logger.isDebugEnabled()) {
logger.debug("Local file '" + localFile + "' has the same modified timestamp, ignored");
}
@@ -960,7 +960,7 @@ public abstract class AbstractRemoteFileOutboundGateway<F> extends AbstractReply
localFile = null;
}
}
else if (!FileExistsMode.IGNORE.equals(fileExistsMode)) {
else if (!FileExistsMode.IGNORE.equals(existsMode)) {
throw new MessageHandlingException(message, "Local file " + localFile + " already exists");
}
else {

10
spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/CachingSessionFactory.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.
@@ -153,7 +153,7 @@ public class CachingSessionFactory<F> implements SessionFactory<F>, DisposableBe
if (this.isSharedSessionCapable && ((SharedSessionCapable) this.sessionFactory).isSharedSession()) {
((SharedSessionCapable) this.sessionFactory).resetSharedSession();
}
long sharedSessionEpoch = System.nanoTime();
long epoch = System.nanoTime();
/*
* Spin until we get a new value - nano precision but may be lower resolution.
* We reset the epoch AFTER resetting the shared session so there is no possibility
@@ -161,10 +161,10 @@ public class CachingSessionFactory<F> implements SessionFactory<F>, DisposableBe
* that a "new" session might appear in the old epoch and thus be closed when returned to
* the cache.
*/
while (sharedSessionEpoch == this.sharedSessionEpoch) {
sharedSessionEpoch = System.nanoTime();
while (epoch == this.sharedSessionEpoch) {
epoch = System.nanoTime();
}
this.sharedSessionEpoch = sharedSessionEpoch;
this.sharedSessionEpoch = epoch;
this.pool.removeAllIdleItems();
}

8
spring-integration-file/src/main/java/org/springframework/integration/file/splitter/FileSplitter.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.
@@ -78,7 +78,7 @@ public class FileSplitter extends AbstractMessageSplitter {
private static final JsonObjectMapper<?, ?> objectMapper =
JsonObjectMapperProvider.jsonAvailable() ? JsonObjectMapperProvider.newInstance() : null;
private final boolean iterator;
private final boolean returnIterator;
private final boolean markers;
@@ -135,7 +135,7 @@ public class FileSplitter extends AbstractMessageSplitter {
* @since 4.2.7
*/
public FileSplitter(boolean iterator, boolean markers, boolean markersJson) {
this.iterator = iterator;
this.returnIterator = iterator;
this.markers = markers;
if (markers) {
setApplySequence(false);
@@ -359,7 +359,7 @@ public class FileSplitter extends AbstractMessageSplitter {
};
if (this.iterator) {
if (this.returnIterator) {
return iterator;
}
else {

8
spring-integration-file/src/main/java/org/springframework/integration/file/tail/ApacheCommonsFileTailingMessageProducer.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.
@@ -71,9 +71,9 @@ public class ApacheCommonsFileTailingMessageProducer extends FileTailingMessageP
@Override
protected void doStart() {
super.doStart();
Tailer tailer = new Tailer(this.getFile(), this, this.pollingDelay, this.end, this.reopen);
this.getTaskExecutor().execute(tailer);
this.tailer = tailer;
Tailer theTailer = new Tailer(this.getFile(), this, this.pollingDelay, this.end, this.reopen);
this.getTaskExecutor().execute(theTailer);
this.tailer = theTailer;
}
@Override

8
spring-integration-file/src/main/java/org/springframework/integration/file/tail/FileTailingMessageProducerSupport.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.
@@ -149,11 +149,11 @@ public abstract class FileTailingMessageProducerSupport extends MessageProducerS
this.idleEventScheduledFuture = getTaskScheduler().scheduleWithFixedDelay(() -> {
long now = System.currentTimeMillis();
long lastAlertAt = this.lastNoMessageAlert.get();
long lastProduce = this.lastProduce;
if (now > lastProduce + this.idleEventInterval
long lastSend = this.lastProduce;
if (now > lastSend + this.idleEventInterval
&& now > lastAlertAt + this.idleEventInterval
&& this.lastNoMessageAlert.compareAndSet(lastAlertAt, now)) {
publishIdleEvent(now - lastProduce);
publishIdleEvent(now - lastSend);
}
}, this.idleEventInterval);
}

22
spring-integration-file/src/main/java/org/springframework/integration/file/tail/OSDelegatingFileTailingMessageProducer.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.
@@ -39,7 +39,7 @@ import org.springframework.util.Assert;
public class OSDelegatingFileTailingMessageProducer extends FileTailingMessageProducerSupport
implements SchedulingAwareRunnable {
private volatile Process process;
private volatile Process nativeTailProcess;
private volatile String options = "-F -n 0";
@@ -47,7 +47,7 @@ public class OSDelegatingFileTailingMessageProducer extends FileTailingMessagePr
private volatile boolean enableStatusReader = true;
private volatile BufferedReader reader;
private volatile BufferedReader stdOutReader;
public void setOptions(String options) {
if (options == null) {
@@ -103,10 +103,10 @@ public class OSDelegatingFileTailingMessageProducer extends FileTailingMessagePr
}
private void destroyProcess() {
Process process = this.process;
Process process = this.nativeTailProcess;
if (process != null) {
process.destroy();
this.process = null;
this.nativeTailProcess = null;
}
}
@@ -121,12 +121,12 @@ public class OSDelegatingFileTailingMessageProducer extends FileTailingMessagePr
try {
Process process = Runtime.getRuntime().exec(this.command);
BufferedReader reader = new BufferedReader(new InputStreamReader(process.getInputStream()));
this.process = process;
this.nativeTailProcess = process;
this.startProcessMonitor();
if (this.enableStatusReader) {
startStatusReader();
}
this.reader = reader;
this.stdOutReader = reader;
this.getTaskExecutor().execute(this);
}
catch (IOException e) {
@@ -140,7 +140,7 @@ public class OSDelegatingFileTailingMessageProducer extends FileTailingMessagePr
*/
private void startProcessMonitor() {
this.getTaskExecutor().execute(() -> {
Process process = OSDelegatingFileTailingMessageProducer.this.process;
Process process = OSDelegatingFileTailingMessageProducer.this.nativeTailProcess;
if (process == null) {
if (logger.isDebugEnabled()) {
logger.debug("Process destroyed before starting process monitor");
@@ -181,7 +181,7 @@ public class OSDelegatingFileTailingMessageProducer extends FileTailingMessagePr
* (file not available, rotations etc) are sent to stderr.
*/
private void startStatusReader() {
Process process = this.process;
Process process = this.nativeTailProcess;
if (process == null) {
if (logger.isDebugEnabled()) {
logger.debug("Process destroyed before starting stderr reader");
@@ -230,7 +230,7 @@ public class OSDelegatingFileTailingMessageProducer extends FileTailingMessagePr
if (logger.isDebugEnabled()) {
logger.debug("Reading stdout");
}
while ((line = this.reader.readLine()) != null) {
while ((line = this.stdOutReader.readLine()) != null) {
this.send(line);
}
}
@@ -239,7 +239,7 @@ public class OSDelegatingFileTailingMessageProducer extends FileTailingMessagePr
logger.debug("Exception on tail reader", e);
}
try {
this.reader.close();
this.stdOutReader.close();
}
catch (IOException e1) {
if (logger.isDebugEnabled()) {

4
spring-integration-file/src/test/java/org/springframework/integration/file/config/FileSplitterParserTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2015-2017 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.
@@ -60,7 +60,7 @@ public class FileSplitterParserTests {
@Test
public void testComplete() {
assertFalse(TestUtils.getPropertyValue(this.splitter, "iterator", Boolean.class));
assertFalse(TestUtils.getPropertyValue(this.splitter, "returnIterator", Boolean.class));
assertTrue(TestUtils.getPropertyValue(this.splitter, "markers", Boolean.class));
assertTrue(TestUtils.getPropertyValue(this.splitter, "markersJson", Boolean.class));
assertTrue(TestUtils.getPropertyValue(this.splitter, "requiresReply", Boolean.class));

4
spring-integration-file/src/test/java/org/springframework/integration/file/tail/FileTailingMessageProducerTests.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.
@@ -95,7 +95,7 @@ public class FileTailingMessageProducerTests {
public void testOS() throws Exception {
OSDelegatingFileTailingMessageProducer adapter = new OSDelegatingFileTailingMessageProducer();
adapter.setOptions(TAIL_OPTIONS_FOLLOW_NAME_ALL_LINES);
testGuts(adapter, "reader");
testGuts(adapter, "stdOutReader");
}
@Test

8
spring-integration-ftp/src/main/java/org/springframework/integration/ftp/gateway/FtpOutboundGateway.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.
@@ -261,15 +261,15 @@ public class FtpOutboundGateway extends AbstractRemoteFileOutboundGateway<FTPFil
private <V> V doInWorkingDirectory(Message<?> message, Session<FTPFile> session, Callable<V> task)
throws IOException {
Expression workingDirExpression = this.workingDirExpression;
Expression workDirExpression = this.workingDirExpression;
FTPClient ftpClient = (FTPClient) session.getClientInstance();
String currentWorkingDirectory = null;
boolean restoreWorkingDirectory = false;
try {
if (workingDirExpression != null) {
if (workDirExpression != null) {
currentWorkingDirectory = ftpClient.printWorkingDirectory();
String newWorkingDirectory =
workingDirExpression.getValue(this.evaluationContext, message, String.class);
workDirExpression.getValue(this.evaluationContext, message, String.class);
if (!Objects.equals(currentWorkingDirectory, newWorkingDirectory)) {
ftpClient.changeWorkingDirectory(newWorkingDirectory);
restoreWorkingDirectory = true;

12
spring-integration-groovy/src/main/java/org/springframework/integration/groovy/GroovyScriptExecutingMessageProcessor.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.
@@ -145,13 +145,13 @@ public class GroovyScriptExecutingMessageProcessor extends AbstractScriptExecuti
((ConfigurableListableBeanFactory) this.beanFactory).ignoreDependencyType(MetaClass.class);
}
CompilerConfiguration compilerConfiguration = this.compilerConfiguration;
if (compilerConfiguration == null && this.compileStatic) {
compilerConfiguration = new CompilerConfiguration();
compilerConfiguration.addCompilationCustomizers(new ASTTransformationCustomizer(CompileStatic.class));
CompilerConfiguration compilerConfig = this.compilerConfiguration;
if (compilerConfig == null && this.compileStatic) {
compilerConfig = new CompilerConfiguration();
compilerConfig.addCompilationCustomizers(new ASTTransformationCustomizer(CompileStatic.class));
}
this.groovyClassLoader = new GroovyClassLoader(this.beanClassLoader, compilerConfiguration);
this.groovyClassLoader = new GroovyClassLoader(this.beanClassLoader, compilerConfig);
}
@Override

8
spring-integration-http/src/main/java/org/springframework/integration/http/inbound/HttpRequestHandlingEndpointSupport.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.
@@ -222,12 +222,12 @@ public abstract class HttpRequestHandlingEndpointSupport extends BaseHttpInbound
BeanFactory beanFactory = getBeanFactory();
if (this.multipartResolver == null && beanFactory != null) {
try {
MultipartResolver multipartResolver = beanFactory.getBean(
MultipartResolver resolver = beanFactory.getBean(
DispatcherServlet.MULTIPART_RESOLVER_BEAN_NAME, MultipartResolver.class);
if (logger.isDebugEnabled()) {
logger.debug("Using MultipartResolver [" + multipartResolver + "]");
logger.debug("Using MultipartResolver [" + resolver + "]");
}
this.multipartResolver = multipartResolver;
this.multipartResolver = resolver;
}
catch (NoSuchBeanDefinitionException e) {
if (logger.isDebugEnabled()) {

47
spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpConnectionFactoryFactoryBean.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.
@@ -157,20 +157,19 @@ public class TcpConnectionFactoryFactoryBean extends AbstractFactoryBean<Abstrac
}
if (this.usingNio) {
if (isServer()) {
TcpNioServerConnectionFactory connectionFactory = new TcpNioServerConnectionFactory(this.port);
this.setCommonAttributes(connectionFactory);
this.setServerAttributes(connectionFactory);
connectionFactory.setUsingDirectBuffers(this.usingDirectBuffers);
connectionFactory.setTcpNioConnectionSupport(this.obtainNioConnectionSupport());
this.connectionFactory = connectionFactory;
TcpNioServerConnectionFactory factory = new TcpNioServerConnectionFactory(this.port);
this.setCommonAttributes(factory);
this.setServerAttributes(factory);
factory.setUsingDirectBuffers(this.usingDirectBuffers);
factory.setTcpNioConnectionSupport(this.obtainNioConnectionSupport());
this.connectionFactory = factory;
}
else {
TcpNioClientConnectionFactory connectionFactory = new TcpNioClientConnectionFactory(
this.host, this.port);
this.setCommonAttributes(connectionFactory);
connectionFactory.setUsingDirectBuffers(this.usingDirectBuffers);
connectionFactory.setTcpNioConnectionSupport(this.obtainNioConnectionSupport());
this.connectionFactory = connectionFactory;
TcpNioClientConnectionFactory factory = new TcpNioClientConnectionFactory(this.host, this.port);
this.setCommonAttributes(factory);
factory.setUsingDirectBuffers(this.usingDirectBuffers);
factory.setTcpNioConnectionSupport(this.obtainNioConnectionSupport());
this.connectionFactory = factory;
}
if (this.sslHandshakeTimeout != null) {
this.connectionFactory.setSslHandshakeTimeout(this.sslHandshakeTimeout);
@@ -178,20 +177,20 @@ public class TcpConnectionFactoryFactoryBean extends AbstractFactoryBean<Abstrac
}
else {
if (isServer()) {
TcpNetServerConnectionFactory connectionFactory = new TcpNetServerConnectionFactory(this.port);
this.setCommonAttributes(connectionFactory);
this.setServerAttributes(connectionFactory);
connectionFactory.setTcpSocketFactorySupport(this.obtainSocketFactorySupport());
connectionFactory.setTcpNetConnectionSupport(this.obtainNetConnectionSupport());
this.connectionFactory = connectionFactory;
TcpNetServerConnectionFactory factory = new TcpNetServerConnectionFactory(this.port);
this.setCommonAttributes(factory);
this.setServerAttributes(factory);
factory.setTcpSocketFactorySupport(this.obtainSocketFactorySupport());
factory.setTcpNetConnectionSupport(this.obtainNetConnectionSupport());
this.connectionFactory = factory;
}
else {
TcpNetClientConnectionFactory connectionFactory = new TcpNetClientConnectionFactory(
TcpNetClientConnectionFactory factory = new TcpNetClientConnectionFactory(
this.host, this.port);
this.setCommonAttributes(connectionFactory);
connectionFactory.setTcpSocketFactorySupport(this.obtainSocketFactorySupport());
connectionFactory.setTcpNetConnectionSupport(this.obtainNetConnectionSupport());
this.connectionFactory = connectionFactory;
this.setCommonAttributes(factory);
factory.setTcpSocketFactorySupport(this.obtainSocketFactorySupport());
factory.setTcpNetConnectionSupport(this.obtainNetConnectionSupport());
this.connectionFactory = factory;
}
}
return this.connectionFactory;

10
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractClientConnectionFactory.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.
@@ -72,7 +72,7 @@ public abstract class AbstractClientConnectionFactory extends AbstractConnection
protected TcpConnectionSupport obtainConnection() throws Exception {
if (!this.isSingleUse()) {
TcpConnectionSupport connection = this.obtainSharedConnection();
TcpConnectionSupport connection = obtainSharedConnection();
if (connection != null) {
return connection;
}
@@ -84,9 +84,9 @@ public abstract class AbstractClientConnectionFactory extends AbstractConnection
protected final TcpConnectionSupport obtainSharedConnection() throws InterruptedException {
this.theConnectionLock.readLock().lockInterruptibly();
try {
TcpConnectionSupport theConnection = this.getTheConnection();
if (theConnection != null && theConnection.isOpen()) {
return theConnection;
TcpConnectionSupport connection = this.getTheConnection();
if (connection != null && connection.isOpen()) {
return connection;
}
}
finally {

10
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/DefaultTcpNetConnectionSupport.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2017 the original author or authors.
* Copyright 2017-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.
@@ -63,11 +63,11 @@ public class DefaultTcpNetConnectionSupport extends AbstractTcpConnectionSupport
@Override
protected InputStream inputStream() throws IOException {
InputStream wrapped = super.inputStream();
InputStream wrappedStream = super.inputStream();
// It shouldn't be possible for the wrapped stream to change but, just in case...
if (this.pushbackStream == null || wrapped != this.wrapped) {
this.pushbackStream = new PushbackInputStream(wrapped, this.pushbackBufferSize);
this.wrapped = wrapped;
if (this.pushbackStream == null || !wrappedStream.equals(this.wrapped)) {
this.pushbackStream = new PushbackInputStream(wrappedStream, this.pushbackBufferSize);
this.wrapped = wrappedStream;
}
return this.pushbackStream;
}

10
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/DefaultTcpNioConnectionSupport.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.
@@ -70,11 +70,11 @@ public class DefaultTcpNioConnectionSupport extends AbstractTcpConnectionSupport
@Override
protected InputStream inputStream() {
InputStream wrapped = super.inputStream();
InputStream wrappedStream = super.inputStream();
// It shouldn't be possible for the wrapped stream to change but, just in case...
if (this.pushbackStream == null || wrapped != this.wrapped) {
this.pushbackStream = new PushbackInputStream(wrapped, this.pushbackBufferSize);
this.wrapped = wrapped;
if (this.pushbackStream == null || !wrappedStream.equals(this.wrapped)) {
this.pushbackStream = new PushbackInputStream(wrappedStream, this.pushbackBufferSize);
this.wrapped = wrappedStream;
}
return this.pushbackStream;
}

10
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/DefaultTcpNioSSLConnectionSupport.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.
@@ -128,11 +128,11 @@ public class DefaultTcpNioSSLConnectionSupport extends AbstractTcpConnectionSupp
@Override
protected InputStream inputStream() {
InputStream wrapped = super.inputStream();
InputStream wrappedStream = super.inputStream();
// It shouldn't be possible for the wrapped stream to change but, just in case...
if (this.pushbackStream == null || wrapped != this.wrapped) {
this.pushbackStream = new PushbackInputStream(wrapped, this.pushbackBufferSize);
this.wrapped = wrapped;
if (this.pushbackStream == null || !wrappedStream.equals(this.wrapped)) {
this.pushbackStream = new PushbackInputStream(wrappedStream, this.pushbackBufferSize);
this.wrapped = wrappedStream;
}
return this.pushbackStream;
}

8
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetServerConnectionFactory.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.
@@ -63,9 +63,9 @@ public class TcpNetServerConnectionFactory extends AbstractServerConnectionFacto
@Override
public int getPort() {
int port = super.getPort();
ServerSocket serverSocket = this.serverSocket;
if (port == 0 && serverSocket != null) {
port = serverSocket.getLocalPort();
ServerSocket socket = this.serverSocket;
if (port == 0 && socket != null) {
port = socket.getLocalPort();
}
return port;
}

16
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioServerConnectionFactory.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.
@@ -71,10 +71,10 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto
@Override
public int getPort() {
int port = super.getPort();
ServerSocketChannel serverChannel = this.serverChannel;
if (port == 0 && serverChannel != null) {
ServerSocketChannel channel = this.serverChannel;
if (port == 0 && channel != null) {
try {
SocketAddress address = serverChannel.getLocalAddress();
SocketAddress address = channel.getLocalAddress();
if (address instanceof InetSocketAddress) {
port = ((InetSocketAddress) address).getPort();
}
@@ -126,18 +126,18 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto
if (logger.isInfoEnabled()) {
logger.info(this + " Listening");
}
final Selector selector = Selector.open();
final Selector theSelector = Selector.open();
if (this.serverChannel == null) {
if (logger.isDebugEnabled()) {
logger.debug(this + " stopped before registering the server channel");
}
}
else {
this.serverChannel.register(selector, SelectionKey.OP_ACCEPT);
this.serverChannel.register(theSelector, SelectionKey.OP_ACCEPT);
setListening(true);
publishServerListeningEvent(getPort());
this.selector = selector;
doSelect(this.serverChannel, selector);
this.selector = theSelector;
doSelect(this.serverChannel, theSelector);
}
}
catch (IOException e) {

19
spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastReceivingChannelAdapter.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.
@@ -218,10 +218,9 @@ public class UnicastReceivingChannelAdapter extends AbstractInternetProtocolRece
}
protected DatagramPacket receive() throws Exception {
DatagramSocket socket = this.getSocket();
final byte[] buffer = new byte[this.getReceiveBufferSize()];
DatagramPacket packet = new DatagramPacket(buffer, buffer.length);
socket.receive(packet);
getSocket().receive(packet);
return packet;
}
@@ -240,18 +239,18 @@ public class UnicastReceivingChannelAdapter extends AbstractInternetProtocolRece
public synchronized DatagramSocket getSocket() {
if (this.socket == null) {
try {
DatagramSocket socket = null;
DatagramSocket datagramSocket = null;
String localAddress = this.getLocalAddress();
int port = super.getPort();
if (localAddress == null) {
socket = port == 0 ? new DatagramSocket() : new DatagramSocket(port);
datagramSocket = port == 0 ? new DatagramSocket() : new DatagramSocket(port);
}
else {
InetAddress whichNic = InetAddress.getByName(localAddress);
socket = new DatagramSocket(new InetSocketAddress(whichNic, port));
datagramSocket = new DatagramSocket(new InetSocketAddress(whichNic, port));
}
setSocketAttributes(socket);
this.socket = socket;
setSocketAttributes(datagramSocket);
this.socket = datagramSocket;
}
catch (IOException e) {
throw new MessagingException("failed to create DatagramSocket", e);
@@ -279,9 +278,9 @@ public class UnicastReceivingChannelAdapter extends AbstractInternetProtocolRece
protected void doStop() {
super.doStop();
try {
DatagramSocket socket = this.socket;
DatagramSocket datagramSocket = this.socket;
this.socket = null;
socket.close();
datagramSocket.close();
}
catch (Exception e) {
// ignore

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

@@ -1,5 +1,5 @@
/*
* Copyright 2001-2018 the original author or authors.
* Copyright 2001-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.
@@ -266,13 +266,13 @@ public class UnicastSendingMessageHandler extends
}
String messageId = id.toString();
try {
boolean waitForAck = this.waitForAck;
if (waitForAck) {
boolean waitAck = this.waitForAck;
if (waitAck) {
countdownLatch = new CountDownLatch(this.ackCounter);
this.ackControl.put(messageId, countdownLatch);
}
convertAndSend(message);
if (waitForAck) {
if (waitAck) {
try {
if (!countdownLatch.await(this.ackTimeout, TimeUnit.MILLISECONDS)) {
throw new MessagingException(message, "Failed to receive UDP Ack in "
@@ -322,12 +322,12 @@ public class UnicastSendingMessageHandler extends
}
protected void convertAndSend(Message<?> message) throws Exception {
DatagramSocket socket;
DatagramSocket datagramSocket;
if (this.socketExpression != null) {
socket = this.socketExpression.getValue(this.evaluationContext, message, DatagramSocket.class);
datagramSocket = this.socketExpression.getValue(this.evaluationContext, message, DatagramSocket.class);
}
else {
socket = getSocket();
datagramSocket = getSocket();
}
SocketAddress destinationAddress;
if (this.destinationExpression != null) {
@@ -353,7 +353,7 @@ public class UnicastSendingMessageHandler extends
DatagramPacket packet = this.mapper.fromMessage(message);
if (packet != null) {
packet.setSocketAddress(destinationAddress);
socket.send(packet);
datagramSocket.send(packet);
if (logger.isDebugEnabled()) {
logger.debug("Sent packet for message " + message + " to " + packet.getSocketAddress());
}
@@ -463,9 +463,9 @@ public class UnicastSendingMessageHandler extends
* @return the ackPort
*/
public int getAckPort() {
DatagramSocket socket = this.socket;
if (this.ackPort == 0 && socket != null) {
return socket.getLocalPort();
DatagramSocket datagramSocket = this.socket;
if (this.ackPort == 0 && datagramSocket != null) {
return datagramSocket.getLocalPort();
}
else {
return this.ackPort;

12
spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcPollingChannelAdapter.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.
@@ -101,10 +101,14 @@ public class JdbcPollingChannelAdapter extends AbstractMessageSource<Object> {
};
this.selectQuery = selectQuery;
this.rowMapper = new ColumnMapRowMapper();
}
public void setRowMapper(RowMapper<?> rowMapper) {
this.rowMapper = rowMapper;
if (rowMapper == null) {
this.rowMapper = new ColumnMapRowMapper();
}
}
public void setUpdateSql(String updateSql) {
@@ -187,13 +191,11 @@ public class JdbcPollingChannelAdapter extends AbstractMessageSource<Object> {
}
protected List<?> doPoll(SqlParameterSource sqlQueryParameterSource) {
final RowMapper<?> rowMapper = this.rowMapper == null ? new ColumnMapRowMapper() : this.rowMapper;
if (sqlQueryParameterSource != null) {
return this.jdbcOperations.query(this.selectQuery, sqlQueryParameterSource, rowMapper);
return this.jdbcOperations.query(this.selectQuery, sqlQueryParameterSource, this.rowMapper);
}
else {
return this.jdbcOperations.query(this.selectQuery, rowMapper);
return this.jdbcOperations.query(this.selectQuery, this.rowMapper);
}
}

10
spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsMessageDrivenEndpoint.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.
@@ -172,12 +172,12 @@ public class JmsMessageDrivenEndpoint extends MessageProducerSupport implements
if (!this.listenerContainer.isActive()) {
this.listenerContainer.afterPropertiesSet();
}
String sessionAcknowledgeMode = this.sessionAcknowledgeMode;
if (sessionAcknowledgeMode == null && !this.externalContainer
String sessionAckeMode = this.sessionAcknowledgeMode;
if (sessionAckeMode == null && !this.externalContainer
&& DefaultMessageListenerContainer.class.isAssignableFrom(this.listenerContainer.getClass())) {
sessionAcknowledgeMode = JmsAdapterUtils.SESSION_TRANSACTED_STRING;
sessionAckeMode = JmsAdapterUtils.SESSION_TRANSACTED_STRING;
}
Integer acknowledgeMode = JmsAdapterUtils.parseAcknowledgeMode(sessionAcknowledgeMode);
Integer acknowledgeMode = JmsAdapterUtils.parseAcknowledgeMode(sessionAckeMode);
if (acknowledgeMode != null) {
if (JmsAdapterUtils.SESSION_TRANSACTED == acknowledgeMode) {
this.listenerContainer.setSessionTransacted(true);

107
spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsOutboundGateway.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.
@@ -123,7 +123,7 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp
private volatile long timeToLive = javax.jms.Message.DEFAULT_TIME_TO_LIVE;
private volatile int priority = javax.jms.Message.DEFAULT_PRIORITY;
private volatile int defaultPriority = javax.jms.Message.DEFAULT_PRIORITY;
private volatile boolean explicitQosEnabled;
@@ -302,13 +302,30 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp
}
/**
* Specify the JMS priority to use when sending request Messages.
* Specify the default JMS priority to use when sending request Messages with
* no {@link IntegrationMessageHeaderAccessor#PRIORITY} header.
*
* The value should be within the range of 0-9.
*
* @param priority The priority.
* @deprecated in favor of {@link #setDefaultPriority(int)}.
*/
@Deprecated
public void setPriority(int priority) {
this.priority = priority;
this.defaultPriority = priority;
}
/**
* Specify the default JMS priority to use when sending request Messages with
* no {@link IntegrationMessageHeaderAccessor#PRIORITY} header.
*
* The value should be within the range of 0-9.
*
* @param priority The priority.
* @since 5.1.2
*/
public void setDefaultPriority(int priority) {
this.defaultPriority = priority;
}
/**
@@ -826,9 +843,9 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp
Integer priority = new IntegrationMessageHeaderAccessor(requestMessage).getPriority();
if (priority == null) {
priority = this.priority;
priority = this.defaultPriority;
}
Destination requestDestination = this.determineRequestDestination(requestMessage, session);
Destination destination = determineRequestDestination(requestMessage, session);
Object reply = null;
if (this.correlationKey == null) {
@@ -837,10 +854,10 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp
* (it will be restored in the reply by normal ARPMH header processing).
*/
jmsRequest.setJMSCorrelationID(null);
reply = doSendAndReceiveAsyncDefaultCorrelation(requestDestination, jmsRequest, session, priority);
reply = doSendAndReceiveAsyncDefaultCorrelation(destination, jmsRequest, session, priority);
}
else {
reply = doSendAndReceiveAsync(requestDestination, jmsRequest, session, priority);
reply = doSendAndReceiveAsync(destination, jmsRequest, session, priority);
}
/*
* Remove the gateway's internal correlation Id to avoid conflicts with an upstream
@@ -883,20 +900,20 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp
Integer priority = new IntegrationMessageHeaderAccessor(requestMessage).getPriority();
if (priority == null) {
priority = this.priority;
priority = this.defaultPriority;
}
javax.jms.Message replyMessage = null;
Destination requestDestination = this.determineRequestDestination(requestMessage, session);
Destination destination = this.determineRequestDestination(requestMessage, session);
if (this.correlationKey != null) {
replyMessage = doSendAndReceiveWithGeneratedCorrelationId(requestDestination, jmsRequest, replyTo,
replyMessage = doSendAndReceiveWithGeneratedCorrelationId(destination, jmsRequest, replyTo,
session, priority);
}
else if (replyTo instanceof TemporaryQueue || replyTo instanceof TemporaryTopic) {
replyMessage = doSendAndReceiveWithTemporaryReplyToDestination(requestDestination, jmsRequest, replyTo,
replyMessage = doSendAndReceiveWithTemporaryReplyToDestination(destination, jmsRequest, replyTo,
session, priority);
}
else {
replyMessage = doSendAndReceiveWithMessageIdCorrelation(requestDestination, jmsRequest, replyTo,
replyMessage = doSendAndReceiveWithMessageIdCorrelation(destination, jmsRequest, replyTo,
session, priority);
}
return replyMessage;
@@ -920,15 +937,15 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp
Assert.state(this.correlationKey != null, "correlationKey must not be null");
String messageSelector = null;
if (!this.correlationKey.equals("JMSCorrelationID*") || jmsRequest.getJMSCorrelationID() == null) {
String correlationId = UUID.randomUUID().toString().replaceAll("'", "''");
String correlation = UUID.randomUUID().toString().replaceAll("'", "''");
if (this.correlationKey.equals("JMSCorrelationID")) {
jmsRequest.setJMSCorrelationID(correlationId);
messageSelector = "JMSCorrelationID = '" + correlationId + "'";
jmsRequest.setJMSCorrelationID(correlation);
messageSelector = "JMSCorrelationID = '" + correlation + "'";
}
else {
jmsRequest.setStringProperty(this.correlationKey, correlationId);
jmsRequest.setStringProperty(this.correlationKey, correlation);
jmsRequest.setJMSCorrelationID(null);
messageSelector = this.correlationKey + " = '" + correlationId + "'";
messageSelector = this.correlationKey + " = '" + correlation + "'";
}
}
else {
@@ -1064,16 +1081,17 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp
private Object doSendAndReceiveAsync(Destination requestDestination, javax.jms.Message jmsRequest, Session session,
int priority) throws JMSException {
String correlationId = null;
String correlation = null;
MessageProducer messageProducer = null;
try {
messageProducer = session.createProducer(requestDestination);
correlationId = this.gatewayCorrelation + "_" + Long.toString(this.correlationId.incrementAndGet());
correlation = this.gatewayCorrelation + "_" + Long.toString(this.correlationId.incrementAndGet());
if (this.correlationKey.equals("JMSCorrelationID")) {
jmsRequest.setJMSCorrelationID(correlationId);
jmsRequest.setJMSCorrelationID(correlation);
}
else {
jmsRequest.setStringProperty(this.correlationKey, correlationId);
jmsRequest.setStringProperty(this.correlationKey, correlation);
/*
* Remove any existing correlation id that was mapped from the inbound message
* (it will be restored in the reply by normal ARPMH header processing).
@@ -1082,16 +1100,16 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp
}
LinkedBlockingQueue<javax.jms.Message> replyQueue = null;
if (logger.isDebugEnabled()) {
logger.debug(this.getComponentName() + " Sending message with correlationId " + correlationId);
logger.debug(this.getComponentName() + " Sending message with correlationId " + correlation);
}
SettableListenableFuture<AbstractIntegrationMessageBuilder<?>> future = null;
boolean async = isAsync();
if (!async) {
replyQueue = new LinkedBlockingQueue<javax.jms.Message>(1);
this.replies.put(correlationId, replyQueue);
this.replies.put(correlation, replyQueue);
}
else {
future = createFuture(correlationId);
future = createFuture(correlation);
}
this.sendRequestMessage(jmsRequest, messageProducer, priority);
@@ -1100,20 +1118,21 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp
return future;
}
else {
return obtainReplyFromContainer(correlationId, replyQueue);
return obtainReplyFromContainer(correlation, replyQueue);
}
}
finally {
JmsUtils.closeMessageProducer(messageProducer);
if (correlationId != null && !isAsync()) {
this.replies.remove(correlationId);
if (correlation != null && !isAsync()) {
this.replies.remove(correlation);
}
}
}
private javax.jms.Message doSendAndReceiveAsyncDefaultCorrelation(Destination requestDestination,
javax.jms.Message jmsRequest, Session session, int priority) throws JMSException {
String correlationId = null;
String correlation = null;
MessageProducer messageProducer = null;
try {
@@ -1122,32 +1141,32 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp
this.sendRequestMessage(jmsRequest, messageProducer, priority);
correlationId = jmsRequest.getJMSMessageID();
correlation = jmsRequest.getJMSMessageID();
if (logger.isDebugEnabled()) {
logger.debug(this.getComponentName() + " Sent message with correlationId " + correlationId);
logger.debug(this.getComponentName() + " Sent message with correlationId " + correlation);
}
this.replies.put(correlationId, replyQueue);
this.replies.put(correlation, replyQueue);
/*
* Check to see if the reply arrived before we obtained the correlationId
*/
synchronized (this.earlyOrLateReplies) {
TimedReply timedReply = this.earlyOrLateReplies.remove(correlationId);
TimedReply timedReply = this.earlyOrLateReplies.remove(correlation);
if (timedReply != null) {
if (logger.isDebugEnabled()) {
logger.debug("Found early reply with correlationId " + correlationId);
logger.debug("Found early reply with correlationId " + correlation);
}
replyQueue.add(timedReply.getReply());
}
}
return obtainReplyFromContainer(correlationId, replyQueue);
return obtainReplyFromContainer(correlation, replyQueue);
}
finally {
JmsUtils.closeMessageProducer(messageProducer);
if (correlationId != null) {
this.replies.remove(correlationId);
if (correlation != null) {
this.replies.remove(correlation);
}
}
}
@@ -1266,7 +1285,7 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp
@Override
public void onMessage(javax.jms.Message message) {
String correlationId = null;
String correlation = null;
try {
if (logger.isTraceEnabled()) {
logger.trace(this.getComponentName() + " Received " + message);
@@ -1274,22 +1293,22 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp
if (this.correlationKey == null ||
this.correlationKey.equals("JMSCorrelationID") ||
this.correlationKey.equals("JMSCorrelationID*")) {
correlationId = message.getJMSCorrelationID();
correlation = message.getJMSCorrelationID();
}
else {
correlationId = message.getStringProperty(this.correlationKey);
correlation = message.getStringProperty(this.correlationKey);
}
Assert.state(correlationId != null, "Message with no correlationId received");
Assert.state(correlation != null, "Message with no correlationId received");
if (isAsync()) {
onMessageAsync(message, correlationId);
onMessageAsync(message, correlation);
}
else {
onMessageSync(message, correlationId);
onMessageSync(message, correlation);
}
}
catch (Exception e) {
if (logger.isWarnEnabled()) {
logger.warn("Failed to consume reply with correlationId " + correlationId, e);
logger.warn("Failed to consume reply with correlationId " + correlation, e);
}
}
}

5
spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsSendingMessageHandler.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.
@@ -157,7 +157,6 @@ public class JmsSendingMessageHandler extends AbstractMessageHandler {
@Override
protected void handleMessageInternal(final Message<?> message) {
Object destination = this.determineDestination(message);
Object objectToSend = (this.extractPayload) ? message.getPayload() : message;
MessagePostProcessor messagePostProcessor = new HeaderMappingMessagePostProcessor(message, this.headerMapper);
@@ -182,7 +181,7 @@ public class JmsSendingMessageHandler extends AbstractMessageHandler {
}
}
try {
send(destination, objectToSend, messagePostProcessor);
send(determineDestination(message), objectToSend, messagePostProcessor);
}
finally {
DynamicJmsTemplateProperties.clearPriority();

8
spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsChannelFactoryBean.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.
@@ -64,7 +64,7 @@ public class JmsChannelFactoryBean extends AbstractFactoryBean<AbstractJmsChanne
private final JmsTemplate jmsTemplate = new DynamicJmsTemplate();
private volatile AbstractMessageListenerContainer container;
private volatile AbstractMessageListenerContainer listenerContainer;
private volatile Class<? extends AbstractMessageListenerContainer> containerType;
@@ -375,8 +375,8 @@ public class JmsChannelFactoryBean extends AbstractFactoryBean<AbstractJmsChanne
protected AbstractJmsChannel createInstance() throws Exception {
this.initializeJmsTemplate();
if (this.messageDriven) {
this.container = createContainer();
SubscribableJmsChannel subscribableJmsChannel = new SubscribableJmsChannel(this.container, this.jmsTemplate);
this.listenerContainer = createContainer();
SubscribableJmsChannel subscribableJmsChannel = new SubscribableJmsChannel(this.listenerContainer, this.jmsTemplate);
subscribableJmsChannel.setMaxSubscribers(this.maxSubscribers);
this.channel = subscribableJmsChannel;
}

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

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2017 the original author or authors.
* Copyright 2016-2019 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -243,7 +243,7 @@ public class JmsOutboundGatewaySpec extends MessageHandlerSpec<JmsOutboundGatewa
* @see JmsOutboundGateway#setPriority(int)
*/
public JmsOutboundGatewaySpec priority(int priority) {
this.target.setPriority(priority);
this.target.setDefaultPriority(priority);
return _this();
}

14
spring-integration-jmx/src/main/java/org/springframework/integration/jmx/NotificationListeningMessageProducer.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.
@@ -62,7 +62,7 @@ public class NotificationListeningMessageProducer extends MessageProducerSupport
private volatile MBeanServerConnection server;
private volatile ObjectName[] objectNames;
private volatile ObjectName[] mBeanObjectNames;
private volatile NotificationFilter filter;
@@ -88,7 +88,7 @@ public class NotificationListeningMessageProducer extends MessageProducerSupport
*/
public void setObjectName(ObjectName... objectNames) {
Assert.isTrue(!ObjectUtils.isEmpty(objectNames), "'objectNames' must contain at least one ObjectName");
this.objectNames = objectNames;
this.mBeanObjectNames = objectNames;
}
/**
@@ -158,11 +158,11 @@ public class NotificationListeningMessageProducer extends MessageProducerSupport
this.logger.debug("Registering to receive notifications");
try {
Assert.notNull(this.server, "MBeanServer is required.");
Assert.notNull(this.objectNames, "An ObjectName is required.");
Assert.notNull(this.mBeanObjectNames, "An ObjectName is required.");
Collection<ObjectName> objectNames = this.retrieveMBeanNames();
if (objectNames.size() < 1) {
this.logger.error("No MBeans found matching ObjectName pattern(s): " +
Arrays.asList(this.objectNames));
Arrays.asList(this.mBeanObjectNames));
}
for (ObjectName objectName : objectNames) {
this.server.addNotificationListener(objectName, this, this.filter, this.handback);
@@ -182,7 +182,7 @@ public class NotificationListeningMessageProducer extends MessageProducerSupport
@Override
protected void doStop() {
this.logger.debug("Unregistering notifications");
if (this.server != null && this.objectNames != null) {
if (this.server != null && this.mBeanObjectNames != null) {
Collection<ObjectName> objectNames = this.retrieveMBeanNames();
for (ObjectName objectName : objectNames) {
try {
@@ -203,7 +203,7 @@ public class NotificationListeningMessageProducer extends MessageProducerSupport
protected Collection<ObjectName> retrieveMBeanNames() {
List<ObjectName> objectNames = new ArrayList<ObjectName>();
for (ObjectName pattern : this.objectNames) {
for (ObjectName pattern : this.mBeanObjectNames) {
Set<ObjectInstance> mBeanInfos;
try {
mBeanInfos = this.server.queryMBeans(pattern, null);

32
spring-integration-jmx/src/main/java/org/springframework/integration/jmx/OperationInvokingMessageHandler.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.
@@ -65,7 +65,7 @@ public class OperationInvokingMessageHandler extends AbstractReplyProducingMessa
private volatile MBeanServerConnection server;
private volatile ObjectName objectName;
private volatile ObjectName defaultObjectName;
private volatile String operationName;
@@ -88,7 +88,7 @@ public class OperationInvokingMessageHandler extends AbstractReplyProducingMessa
public void setObjectName(String objectName) {
try {
if (objectName != null) {
this.objectName = ObjectNameManager.getInstance(objectName);
this.defaultObjectName = ObjectNameManager.getInstance(objectName);
}
}
catch (MalformedObjectNameException e) {
@@ -118,15 +118,15 @@ public class OperationInvokingMessageHandler extends AbstractReplyProducingMessa
@Override
protected Object handleRequestMessage(Message<?> requestMessage) {
ObjectName objectName = this.resolveObjectName(requestMessage);
String operationName = this.resolveOperationName(requestMessage);
ObjectName objectName = resolveObjectName(requestMessage);
String operation = resolveOperationName(requestMessage);
Map<String, Object> paramsFromMessage = this.resolveParameters(requestMessage);
try {
MBeanInfo mbeanInfo = this.server.getMBeanInfo(objectName);
MBeanOperationInfo[] opInfoArray = mbeanInfo.getOperations();
boolean hasNoArgOption = false;
for (MBeanOperationInfo opInfo : opInfoArray) {
if (operationName.equals(opInfo.getName())) {
if (operation.equals(opInfo.getName())) {
MBeanParameterInfo[] paramInfoArray = opInfo.getSignature();
if (paramInfoArray.length == 0) {
hasNoArgOption = true;
@@ -152,21 +152,21 @@ public class OperationInvokingMessageHandler extends AbstractReplyProducingMessa
}
}
if (index == paramInfoArray.length) {
return this.server.invoke(objectName, operationName, values, signature);
return this.server.invoke(objectName, operation, values, signature);
}
}
}
}
if (hasNoArgOption) {
return this.server.invoke(objectName, operationName, null, null);
return this.server.invoke(objectName, operation, null, null);
}
throw new MessagingException(requestMessage, "failed to find JMX operation '"
+ operationName + "' on MBean [" + objectName + "] of type [" + mbeanInfo.getClassName()
+ operation + "' on MBean [" + objectName + "] of type [" + mbeanInfo.getClassName()
+ "] with " + paramsFromMessage.size() + " parameters: " + paramsFromMessage);
}
catch (JMException e) {
throw new MessageHandlingException(requestMessage, "failed to invoke JMX operation '" +
operationName + "' on MBean [" + objectName + "]" + " with " +
operation + "' on MBean [" + objectName + "]" + " with " +
paramsFromMessage.size() + " parameters: " + paramsFromMessage, e);
}
catch (IOException e) {
@@ -189,7 +189,7 @@ public class OperationInvokingMessageHandler extends AbstractReplyProducingMessa
* First checks if defaultObjectName is set, otherwise falls back on {@link JmxHeaders#OBJECT_NAME} header.
*/
private ObjectName resolveObjectName(Message<?> message) {
ObjectName objectName = this.objectName;
ObjectName objectName = this.defaultObjectName;
if (objectName == null) {
Object objectNameHeader = message.getHeaders().get(JmxHeaders.OBJECT_NAME);
if (objectNameHeader instanceof ObjectName) {
@@ -212,12 +212,12 @@ public class OperationInvokingMessageHandler extends AbstractReplyProducingMessa
* First checks if defaultOperationName is set, otherwise falls back on {@link JmxHeaders#OPERATION_NAME} header.
*/
private String resolveOperationName(Message<?> message) {
String operationName = this.operationName;
if (operationName == null) {
operationName = message.getHeaders().get(JmxHeaders.OPERATION_NAME, String.class);
String operation = this.operationName;
if (operation == null) {
operation = message.getHeaders().get(JmxHeaders.OPERATION_NAME, String.class);
}
Assert.notNull(operationName, "Failed to resolve operation name.");
return operationName;
Assert.notNull(operation, "Failed to resolve operation name.");
return operation;
}
@SuppressWarnings({ "unchecked", "rawtypes" })

26
spring-integration-jpa/src/main/java/org/springframework/integration/jpa/core/JpaExecutor.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.
@@ -429,18 +429,18 @@ public class JpaExecutor implements InitializingBean, BeanFactoryAware {
final Object result;
ParameterSource parameterSource = null;
ParameterSource paramSource = null;
if (this.jpaQuery != null || this.nativeQuery != null || this.namedQuery != null) {
parameterSource = determineParameterSource(message);
paramSource = determineParameterSource(message);
}
if (this.jpaQuery != null) {
result = this.jpaOperations.executeUpdate(this.jpaQuery, parameterSource);
result = this.jpaOperations.executeUpdate(this.jpaQuery, paramSource);
}
else if (this.nativeQuery != null) {
result = this.jpaOperations.executeUpdateWithNativeQuery(this.nativeQuery, parameterSource);
result = this.jpaOperations.executeUpdateWithNativeQuery(this.nativeQuery, paramSource);
}
else if (this.namedQuery != null) {
result = this.jpaOperations.executeUpdateWithNamedQuery(this.namedQuery, parameterSource);
result = this.jpaOperations.executeUpdateWithNamedQuery(this.namedQuery, paramSource);
}
else {
switch (this.persistMode) {
@@ -492,11 +492,11 @@ public class JpaExecutor implements InitializingBean, BeanFactoryAware {
if (this.idExpression != null) {
Object id = this.idExpression.getValue(this.evaluationContext, requestMessage); // NOSONAR It can be null
Assert.state(id != null, "The 'idExpression' cannot evaluate to null.");
Class<?> entityClass = this.entityClass;
if (entityClass == null && requestMessage != null) {
entityClass = requestMessage.getPayload().getClass();
Class<?> entityClazz = this.entityClass;
if (entityClazz == null && requestMessage != null) {
entityClazz = requestMessage.getPayload().getClass();
}
payload = this.jpaOperations.find(entityClass, id);
payload = this.jpaOperations.find(entityClazz, id);
}
else {
final List<?> result;
@@ -624,14 +624,12 @@ public class JpaExecutor implements InitializingBean, BeanFactoryAware {
}
private ParameterSource determineParameterSource(final Message<?> requestMessage) {
ParameterSource parameterSource;
if (this.usePayloadAsParameterSource) {
parameterSource = this.parameterSourceFactory.createParameterSource(requestMessage.getPayload());
return this.parameterSourceFactory.createParameterSource(requestMessage.getPayload());
}
else {
parameterSource = this.parameterSourceFactory.createParameterSource(requestMessage);
return this.parameterSourceFactory.createParameterSource(requestMessage);
}
return parameterSource;
}
}