Avoid throws Exception where possible - Phase I

* Polishing - PR Comments
This commit is contained in:
Gary Russell
2019-03-07 12:53:52 -05:00
committed by Artem Bilan
parent 005bc80680
commit b187bca36e
130 changed files with 992 additions and 784 deletions

View File

@@ -261,7 +261,7 @@ subprojects { subproject ->
}
test {
maxHeapSize = "1024m"
maxHeapSize = "1536m"
jacoco {
append = false
destinationFile = file("$buildDir/jacoco.exec")

View File

@@ -1 +1,2 @@
version=5.2.0.BUILD-SNAPSHOT
org.gradle.jvmargs=-Xmx1536m

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.
@@ -28,6 +28,7 @@ import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.amqp.core.FanoutExchange;
import org.springframework.amqp.core.MessageDeliveryMode;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.RabbitAccessor;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.DirectMessageListenerContainer;
@@ -35,8 +36,6 @@ import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.rabbit.support.MessagePropertiesConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.beans.factory.BeanNameAware;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.config.AbstractFactoryBean;
import org.springframework.context.Lifecycle;
import org.springframework.context.SmartLifecycle;
@@ -46,15 +45,13 @@ import org.springframework.integration.amqp.channel.PollableAmqpChannel;
import org.springframework.integration.amqp.channel.PublishSubscribeAmqpChannel;
import org.springframework.integration.amqp.support.AmqpHeaderMapper;
import org.springframework.integration.amqp.support.DefaultAmqpHeaderMapper;
import org.springframework.integration.util.JavaUtils;
import org.springframework.lang.Nullable;
import org.springframework.messaging.support.ChannelInterceptor;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.interceptor.TransactionAttribute;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;
import org.springframework.util.ErrorHandler;
import org.springframework.util.ObjectUtils;
import org.springframework.util.StringUtils;
/**
* If point-to-point, we send to the default exchange with the routing key
@@ -71,7 +68,7 @@ import org.springframework.util.StringUtils;
* @since 2.1
*/
public class AmqpChannelFactoryBean extends AbstractFactoryBean<AbstractAmqpChannel>
implements SmartLifecycle, DisposableBean, BeanNameAware {
implements SmartLifecycle, BeanNameAware {
private volatile AbstractAmqpChannel channel;
@@ -355,32 +352,26 @@ public class AmqpChannelFactoryBean extends AbstractFactoryBean<AbstractAmqpChan
}
@Override
protected AbstractAmqpChannel createInstance() throws Exception {
protected AbstractAmqpChannel createInstance() {
if (this.messageDriven) {
AbstractMessageListenerContainer container = this.createContainer();
if (this.amqpTemplate instanceof InitializingBean) {
((InitializingBean) this.amqpTemplate).afterPropertiesSet();
if (this.amqpTemplate instanceof RabbitAccessor) {
((RabbitAccessor) this.amqpTemplate).afterPropertiesSet();
}
if (this.isPubSub) {
PublishSubscribeAmqpChannel pubsub = new PublishSubscribeAmqpChannel(
this.beanName, container, this.amqpTemplate, this.outboundHeaderMapper, this.inboundHeaderMapper);
if (this.exchange != null) {
pubsub.setExchange(this.exchange);
}
if (this.maxSubscribers != null) {
pubsub.setMaxSubscribers(this.maxSubscribers);
}
JavaUtils.INSTANCE
.acceptIfNotNull(this.exchange, pubsub::setExchange)
.acceptIfNotNull(this.maxSubscribers, pubsub::setMaxSubscribers);
this.channel = pubsub;
}
else {
PointToPointSubscribableAmqpChannel p2p = new PointToPointSubscribableAmqpChannel(
this.beanName, container, this.amqpTemplate, this.outboundHeaderMapper, this.inboundHeaderMapper);
if (StringUtils.hasText(this.queueName)) {
p2p.setQueueName(this.queueName);
}
if (this.maxSubscribers != null) {
p2p.setMaxSubscribers(this.maxSubscribers);
}
JavaUtils.INSTANCE
.acceptIfHasText(this.queueName, p2p::setQueueName)
.acceptIfNotNull(this.maxSubscribers, p2p::setMaxSubscribers);
this.channel = p2p;
}
}
@@ -388,45 +379,31 @@ public class AmqpChannelFactoryBean extends AbstractFactoryBean<AbstractAmqpChan
Assert.isTrue(!this.isPubSub, "An AMQP 'publish-subscribe-channel' must be message-driven.");
PollableAmqpChannel pollable = new PollableAmqpChannel(this.beanName, this.amqpTemplate,
this.outboundHeaderMapper, this.inboundHeaderMapper);
if (this.amqpAdmin != null) {
pollable.setAmqpAdmin(this.amqpAdmin);
}
if (StringUtils.hasText(this.queueName)) {
pollable.setQueueName(this.queueName);
}
JavaUtils.INSTANCE
.acceptIfNotNull(this.amqpAdmin, pollable::setAmqpAdmin)
.acceptIfHasText(this.queueName, pollable::setQueueName);
this.channel = pollable;
}
if (!CollectionUtils.isEmpty(this.interceptors)) {
this.channel.setInterceptors(this.interceptors);
}
JavaUtils.INSTANCE
.acceptIfNotEmpty(this.interceptors, this.channel::setInterceptors);
this.channel.setBeanName(this.beanName);
if (getBeanFactory() != null) {
this.channel.setBeanFactory(getBeanFactory()); // NOSONAR never null
}
if (this.defaultDeliveryMode != null) {
this.channel.setDefaultDeliveryMode(this.defaultDeliveryMode);
}
if (this.extractPayload != null) {
this.channel.setExtractPayload(this.extractPayload);
}
JavaUtils.INSTANCE
.acceptIfNotNull(getBeanFactory(), this.channel::setBeanFactory)
.acceptIfNotNull(this.defaultDeliveryMode, this.channel::setDefaultDeliveryMode)
.acceptIfNotNull(this.extractPayload, this.channel::setExtractPayload);
this.channel.setHeadersMappedLast(this.headersLast);
this.channel.afterPropertiesSet();
return this.channel;
}
private AbstractMessageListenerContainer createContainer() throws Exception {
private AbstractMessageListenerContainer createContainer() {
AbstractMessageListenerContainer container;
if (this.consumersPerQueue == null) {
SimpleMessageListenerContainer smlc = new SimpleMessageListenerContainer();
if (this.concurrentConsumers != null) {
smlc.setConcurrentConsumers(this.concurrentConsumers);
}
if (this.receiveTimeout != null) {
smlc.setReceiveTimeout(this.receiveTimeout);
}
if (this.txSize != null) {
smlc.setTxSize(this.txSize);
}
JavaUtils.INSTANCE
.acceptIfNotNull(this.concurrentConsumers, smlc::setConcurrentConsumers)
.acceptIfNotNull(this.receiveTimeout, smlc::setReceiveTimeout)
.acceptIfNotNull(this.txSize, smlc::setTxSize);
container = smlc;
}
else {
@@ -434,48 +411,25 @@ public class AmqpChannelFactoryBean extends AbstractFactoryBean<AbstractAmqpChan
dmlc.setConsumersPerQueue(this.consumersPerQueue);
container = dmlc;
}
if (this.acknowledgeMode != null) {
container.setAcknowledgeMode(this.acknowledgeMode);
}
if (!ObjectUtils.isEmpty(this.adviceChain)) {
container.setAdviceChain(this.adviceChain);
}
JavaUtils.INSTANCE
.acceptIfNotNull(this.acknowledgeMode, container::setAcknowledgeMode)
.acceptIfNotEmpty(this.adviceChain, container::setAdviceChain);
container.setAutoStartup(this.autoStartup);
container.setChannelTransacted(this.channelTransacted);
container.setConnectionFactory(this.connectionFactory);
if (this.errorHandler != null) {
container.setErrorHandler(this.errorHandler);
}
if (this.exposeListenerChannel != null) {
container.setExposeListenerChannel(this.exposeListenerChannel);
}
if (this.messagePropertiesConverter != null) {
container.setMessagePropertiesConverter(this.messagePropertiesConverter);
}
if (this.phase != null) {
container.setPhase(this.phase);
}
if (this.prefetchCount != null) {
container.setPrefetchCount(this.prefetchCount);
}
if (this.recoveryInterval != null) {
container.setRecoveryInterval(this.recoveryInterval);
}
if (this.shutdownTimeout != null) {
container.setShutdownTimeout(this.shutdownTimeout);
}
if (this.taskExecutor != null) {
container.setTaskExecutor(this.taskExecutor);
}
if (this.transactionAttribute != null) {
container.setTransactionAttribute(this.transactionAttribute);
}
if (this.transactionManager != null) {
container.setTransactionManager(this.transactionManager);
}
if (this.missingQueuesFatal != null) {
container.setMissingQueuesFatal(this.missingQueuesFatal);
}
JavaUtils.INSTANCE
.acceptIfNotNull(this.errorHandler, container::setErrorHandler)
.acceptIfNotNull(this.exposeListenerChannel, container::setExposeListenerChannel)
.acceptIfNotNull(this.messagePropertiesConverter, container::setMessagePropertiesConverter)
.acceptIfNotNull(this.phase, container::setPhase)
.acceptIfNotNull(this.prefetchCount, container::setPrefetchCount)
.acceptIfNotNull(this.recoveryInterval, container::setRecoveryInterval)
.acceptIfNotNull(this.shutdownTimeout, container::setShutdownTimeout)
.acceptIfNotNull(this.taskExecutor, container::setTaskExecutor)
.acceptIfNotNull(this.transactionAttribute, container::setTransactionAttribute)
.acceptIfNotNull(this.transactionManager, container::setTransactionManager)
.acceptIfNotNull(this.missingQueuesFatal, container::setMissingQueuesFatal);
return container;
}
@@ -524,7 +478,7 @@ public class AmqpChannelFactoryBean extends AbstractFactoryBean<AbstractAmqpChan
}
@Override
protected void destroyInstance(AbstractAmqpChannel instance) throws Exception {
protected void destroyInstance(AbstractAmqpChannel instance) {
this.channel.destroy();
}

View File

@@ -197,7 +197,7 @@ public class AmqpInboundChannelAdapter extends MessageProducerSupport implements
@SuppressWarnings("unchecked")
@Override
public void onMessage(final Message message, final Channel channel) throws Exception {
public void onMessage(final Message message, final Channel channel) {
boolean retryDisabled = AmqpInboundChannelAdapter.this.retryTemplate == null;
try {
if (retryDisabled) {

View File

@@ -330,7 +330,7 @@ public class AmqpOutboundChannelAdapterParserTests {
public static class FooAdvice extends AbstractRequestHandlerAdvice {
@Override
protected Object doInvoke(ExecutionCallback callback, Object target, Message<?> message) throws Exception {
protected Object doInvoke(ExecutionCallback callback, Object target, Message<?> message) {
adviceCalled++;
return null;
}

View File

@@ -328,7 +328,7 @@ public class AmqpOutboundGatewayParserTests {
public static class FooAdvice extends AbstractRequestHandlerAdvice {
@Override
protected Object doInvoke(ExecutionCallback callback, Object target, Message<?> message) throws Exception {
protected Object doInvoke(ExecutionCallback callback, Object target, Message<?> message) {
adviceCalled++;
return callback.execute();
}

View File

@@ -55,6 +55,7 @@ import org.springframework.integration.util.UUIDConverter;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageDeliveryException;
import org.springframework.messaging.MessageHandlingException;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;
@@ -431,7 +432,7 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP
}
@Override
protected void handleMessageInternal(Message<?> message) throws InterruptedException {
protected void handleMessageInternal(Message<?> message) {
Object correlationKey = this.correlationStrategy.getCorrelationKey(message);
Assert.state(correlationKey != null,
"Null correlation not allowed. Maybe the CorrelationStrategy is failing?");
@@ -444,7 +445,13 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP
Lock lock = this.lockRegistry.obtain(groupIdUuid.toString());
boolean noOutput = true;
lock.lockInterruptibly();
try {
lock.lockInterruptibly();
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new MessageHandlingException(message, "Interrupted getting lock", e);
}
try {
noOutput = processMessageForGroup(message, correlationKey, groupIdUuid, lock);
}

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2016 the original author or authors.
* Copyright 2002-2019 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -91,7 +91,7 @@ public class CorrelatingMessageBarrier extends AbstractMessageHandler implements
}
@Override
protected void handleMessageInternal(Message<?> message) throws Exception {
protected void handleMessageInternal(Message<?> message) {
Object correlationKey = this.correlationStrategy.getCorrelationKey(message);
Object lock = getLock(correlationKey);
synchronized (lock) {

View File

@@ -281,7 +281,7 @@ public class NullChannel implements PollableChannel, MessageChannelMetrics,
}
@Override
public void destroy() throws Exception {
public void destroy() {
if (this.successTimer != null) {
this.successTimer.remove();
}

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2018 the original author or authors.
* Copyright 2018-2019 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -90,7 +90,7 @@ public abstract class AbstractEvaluationContextFactoryBean implements Applicatio
return this.functions;
}
protected void initialize(String beanName) throws Exception {
protected void initialize(String beanName) {
if (this.applicationContext != null) {
ConversionService conversionService = IntegrationUtils.getConversionService(getApplicationContext());
if (conversionService != null) {
@@ -115,7 +115,7 @@ public abstract class AbstractEvaluationContextFactoryBean implements Applicatio
}
}
}
catch (NoSuchBeanDefinitionException e) {
catch (@SuppressWarnings("unused") NoSuchBeanDefinitionException e) {
// There is no 'SpelPropertyAccessorRegistrar' bean in the application context.
}

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.
@@ -61,7 +61,7 @@ final class ChannelInitializer implements BeanFactoryAware, InitializingBean {
}
@Override
public void afterPropertiesSet() throws Exception {
public void afterPropertiesSet() {
Assert.notNull(this.beanFactory, "'beanFactory' must not be null");
if (!this.autoCreate) {
return;

View File

@@ -182,7 +182,7 @@ public class ConsumerEndpointFactoryBean
}
@Override
public void afterPropertiesSet() throws Exception {
public void afterPropertiesSet() {
if (this.beanName == null) {
logger.error("The MessageHandler [" + this.handler + "] will be created without a 'componentName'. " +
"Consider specifying the 'beanName' property on this ConsumerEndpointFactoryBean.");
@@ -248,7 +248,7 @@ public class ConsumerEndpointFactoryBean
}
@Override
public AbstractEndpoint getObject() throws Exception {
public AbstractEndpoint getObject() {
if (!this.initialized) {
this.initializeEndpoint();
}
@@ -263,8 +263,7 @@ public class ConsumerEndpointFactoryBean
return this.endpoint.getClass();
}
@SuppressWarnings("unchecked")
private void initializeEndpoint() throws Exception {
private void initializeEndpoint() {
synchronized (this.initializationMonitor) {
if (this.initialized) {
return;
@@ -381,7 +380,7 @@ public class ConsumerEndpointFactoryBean
}
@Override
public void destroy() throws Exception {
public void destroy() {
if (this.endpoint != null) {
this.endpoint.destroy();
}

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.
@@ -56,7 +56,7 @@ public class CorrelationStrategyFactoryBean implements FactoryBean<CorrelationSt
}
@Override
public void afterPropertiesSet() throws Exception {
public void afterPropertiesSet() {
if (this.target instanceof CorrelationStrategy && !StringUtils.hasText(this.methodName)) {
this.strategy = (CorrelationStrategy) this.target;
return;
@@ -75,14 +75,17 @@ public class CorrelationStrategyFactoryBean implements FactoryBean<CorrelationSt
}
}
public CorrelationStrategy getObject() throws Exception {
@Override
public CorrelationStrategy getObject() {
return this.strategy;
}
@Override
public Class<?> getObjectType() {
return CorrelationStrategy.class;
}
@Override
public boolean isSingleton() {
return true;
}

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.
@@ -79,7 +79,7 @@ public class IntegrationEvaluationContextFactoryBean extends AbstractEvaluationC
}
@Override
public void afterPropertiesSet() throws Exception {
public void afterPropertiesSet() {
if (getApplicationContext() != null) {
this.beanResolver = new BeanFactoryResolver(getApplicationContext());
}
@@ -87,7 +87,7 @@ public class IntegrationEvaluationContextFactoryBean extends AbstractEvaluationC
}
@Override
public StandardEvaluationContext getObject() throws Exception {
public StandardEvaluationContext getObject() {
StandardEvaluationContext evaluationContext = new StandardEvaluationContext();
if (this.typeLocator != null) {
evaluationContext.setTypeLocator(this.typeLocator);

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.
@@ -272,12 +272,12 @@ public class IntegrationManagementConfigurer
}
@Override
public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
public Object postProcessAfterInitialization(Object bean, String name) throws BeansException {
if (this.singletonsInstantiated) {
if (bean instanceof IntegrationManagement) {
((IntegrationManagement) bean).registerMetricsCaptor(this.metricsCaptor);
}
return doConfigureMetrics(bean, beanName);
return doConfigureMetrics(bean, name);
}
return bean;
}

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2018 the original author or authors.
* Copyright 2018-2019 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -68,12 +68,12 @@ public class IntegrationSimpleEvaluationContextFactoryBean extends AbstractEvalu
}
@Override
public void afterPropertiesSet() throws Exception {
public void afterPropertiesSet() {
initialize(IntegrationContextUtils.INTEGRATION_SIMPLE_EVALUATION_CONTEXT_BEAN_NAME);
}
@Override
public SimpleEvaluationContext getObject() throws Exception {
public SimpleEvaluationContext getObject() {
Collection<PropertyAccessor> accessors = getPropertyAccessors().values();
PropertyAccessor[] accessorArray = accessors.toArray(new PropertyAccessor[accessors.size() + 2]);
accessorArray[accessors.size()] = new MapAccessor();

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.
@@ -57,7 +57,7 @@ public class ReleaseStrategyFactoryBean implements FactoryBean<ReleaseStrategy>,
}
@Override
public void afterPropertiesSet() throws Exception {
public void afterPropertiesSet() {
if (this.target instanceof ReleaseStrategy && !StringUtils.hasText(this.methodName)) {
this.strategy = (ReleaseStrategy) this.target;
return;
@@ -88,7 +88,7 @@ public class ReleaseStrategyFactoryBean implements FactoryBean<ReleaseStrategy>,
}
@Override
public ReleaseStrategy getObject() throws Exception {
public ReleaseStrategy getObject() {
return this.strategy;
}

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.
@@ -137,7 +137,7 @@ public class SourcePollingChannelAdapterFactoryBean implements FactoryBean<Sourc
}
@Override
public void afterPropertiesSet() throws Exception {
public void afterPropertiesSet() {
if (this.channelResolver == null) {
this.channelResolver = new BeanFactoryMessageChannelDestinationResolver(this.beanFactory);
}
@@ -256,7 +256,7 @@ public class SourcePollingChannelAdapterFactoryBean implements FactoryBean<Sourc
}
@Override
public void destroy() throws Exception {
public void destroy() {
if (this.adapter != null) {
this.adapter.destroy();
}

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2013-2016 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.
@@ -61,7 +61,7 @@ public class SpelFunctionFactoryBean implements FactoryBean<Method>, Initializin
}
@Override
public void afterPropertiesSet() throws Exception {
public void afterPropertiesSet() {
this.method = BeanUtils.resolveSignature(this.functionMethodSignature, this.functionClass);
if (this.method == null) {
@@ -74,7 +74,7 @@ public class SpelFunctionFactoryBean implements FactoryBean<Method>, Initializin
}
@Override
public Method getObject() throws Exception {
public Method getObject() {
return this.method;
}

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2018 the original author or authors.
* Copyright 2018-2019 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -36,17 +36,19 @@ class Disposables implements DisposableBean {
private final List<DisposableBean> disposables = new ArrayList<>();
public void add(DisposableBean... disposables) {
this.disposables.addAll(Arrays.asList(disposables));
@SafeVarargs
@SuppressWarnings("varargs")
public final void add(DisposableBean... disposablesToAdd) {
this.disposables.addAll(Arrays.asList(disposablesToAdd));
}
@Override
public void destroy() throws Exception {
public void destroy() {
this.disposables.forEach(d -> {
try {
d.destroy();
}
catch (Exception e) {
catch (@SuppressWarnings("unused") Exception e) {
// NOSONAR
}
});

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.
@@ -54,7 +54,7 @@ class ConverterRegistrar implements InitializingBean, BeanFactoryAware {
}
@Override
public void afterPropertiesSet() throws Exception {
public void afterPropertiesSet() {
Assert.notNull(this.beanFactory, "BeanFactory is required");
ConversionService conversionService = IntegrationUtils.getConversionService(this.beanFactory);
if (conversionService instanceof GenericConversionService) {

View File

@@ -112,14 +112,14 @@ public abstract class AbstractEndpoint extends IntegrationObjectSupport
this.roleController.addLifecycleToRole(this.role, this);
}
catch (NoSuchBeanDefinitionException e) {
catch (@SuppressWarnings("unused") NoSuchBeanDefinitionException e) {
this.logger.trace("No LifecycleRoleController in the context");
}
}
}
@Override
public void destroy() throws Exception { // NOSONAR TODO: remove throws in 5.2
public void destroy() {
if (this.roleController != null) {
this.roleController.removeLifecycle(this);
}

View File

@@ -229,7 +229,7 @@ public abstract class AbstractMessageSource<T> extends AbstractExpressionEvaluat
protected abstract Object doReceive();
@Override
public void destroy() throws Exception {
public void destroy() {
if (this.receiveCounter != null) {
this.receiveCounter.remove();
}

View File

@@ -421,7 +421,7 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint
}
@Override
public Object getObject() throws Exception {
public Object getObject() {
if (this.serviceProxy == null) {
this.onInit();
Assert.notNull(this.serviceProxy, "failed to initialize proxy");
@@ -436,7 +436,7 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint
@Override
@Nullable
public Object invoke(final MethodInvocation invocation) throws Throwable {
public Object invoke(final MethodInvocation invocation) throws Throwable { // NOSONAR
final Class<?> returnType = invocation.getMethod().getReturnType();
if (this.asyncExecutor != null && !Object.class.equals(returnType)) {
Invoker invoker = new Invoker(invocation);
@@ -479,7 +479,7 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint
}
@Nullable
private Object invokeGatewayMethod(MethodInvocation invocation, boolean runningOnCallerThread) throws Exception {
private Object invokeGatewayMethod(MethodInvocation invocation, boolean runningOnCallerThread) {
if (!this.initialized) {
this.afterPropertiesSet();
}

View File

@@ -96,8 +96,8 @@ public abstract class AbstractMessageHandler extends IntegrationObjectSupport
}
@Override
public void registerMetricsCaptor(MetricsCaptor metricsCaptor) {
this.metricsCaptor = metricsCaptor;
public void registerMetricsCaptor(MetricsCaptor metricsCaptorToRegister) {
this.metricsCaptor = metricsCaptorToRegister;
}
@Override
@@ -222,7 +222,7 @@ public abstract class AbstractMessageHandler extends IntegrationObjectSupport
}
protected abstract void handleMessageInternal(Message<?> message) throws Exception;
protected abstract void handleMessageInternal(Message<?> message);
@Override
public void reset() {

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.
@@ -58,7 +58,7 @@ public class ExpressionEvaluatingMessageHandler extends AbstractMessageHandler {
}
@Override
protected void handleMessageInternal(Message<?> message) throws Exception {
protected void handleMessageInternal(Message<?> message) {
this.processor.processMessage(message);
}

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,8 +63,8 @@ import org.springframework.util.Assert;
* @author Gary Russell
* @author Artem Bilan
*/
public class MessageHandlerChain extends AbstractMessageProducingHandler implements MessageProducer,
CompositeMessageHandler, Lifecycle {
public class MessageHandlerChain extends AbstractMessageProducingHandler
implements CompositeMessageHandler, Lifecycle {
private volatile List<MessageHandler> handlers;
@@ -103,7 +103,7 @@ public class MessageHandlerChain extends AbstractMessageProducingHandler impleme
}
@Override
protected void handleMessageInternal(Message<?> message) throws Exception {
protected void handleMessageInternal(Message<?> message) {
if (!this.initialized) {
this.onInit();
}

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.
@@ -20,13 +20,12 @@ import java.lang.reflect.Method;
import org.aopalliance.intercept.MethodInterceptor;
import org.aopalliance.intercept.MethodInvocation;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.aop.ProxyMethodInvocation;
import org.springframework.integration.context.IntegrationObjectSupport;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;
/**
* Base class for {@link MessageHandler} advice classes. Subclasses should provide an
@@ -42,8 +41,6 @@ import org.springframework.messaging.MessageHandler;
public abstract class AbstractRequestHandlerAdvice extends IntegrationObjectSupport
implements MethodInterceptor {
protected final Log logger = LogFactory.getLog(this.getClass());
@Override
public final Object invoke(final MethodInvocation invocation) throws Throwable {
@@ -70,20 +67,17 @@ public abstract class AbstractRequestHandlerAdvice extends IntegrationObjectSupp
return doInvoke(new ExecutionCallback() {
@Override
public Object execute() throws Exception {
public Object execute() {
try {
return invocation.proceed();
}
catch (Exception e) { //NOSONAR - catch necessary so we can wrap Errors
throw e;
}
catch (Throwable e) { //NOSONAR - ok to catch; unwrapped and rethrown below
throw new ThrowableHolderException(e);
}
}
@Override
public Object cloneAndExecute() throws Exception {
public Object cloneAndExecute() {
try {
/*
* If we don't copy the invocation carefully it won't keep a reference to the other
@@ -99,7 +93,7 @@ public abstract class AbstractRequestHandlerAdvice extends IntegrationObjectSupp
}
}
catch (Exception e) { //NOSONAR - catch necessary so we can wrap Errors
throw e;
throw new MessagingException(message, "Failed to handle", e);
}
catch (Throwable e) { //NOSONAR - ok to catch; unwrapped and rethrown below
throw new ThrowableHolderException(e);
@@ -122,9 +116,8 @@ public abstract class AbstractRequestHandlerAdvice extends IntegrationObjectSupp
* @param target The target handler.
* @param message The message that will be sent to the handler.
* @return the result after invoking the {@link MessageHandler}.
* @throws Exception Any Exception.
*/
protected abstract Object doInvoke(ExecutionCallback callback, Object target, Message<?> message) throws Exception;
protected abstract Object doInvoke(ExecutionCallback callback, Object target, Message<?> message);
/**
* Unwrap the cause of a {@link AbstractRequestHandlerAdvice.ThrowableHolderException}.
@@ -165,9 +158,8 @@ public abstract class AbstractRequestHandlerAdvice extends IntegrationObjectSupp
* Call this for a normal invocation.proceed().
*
* @return The result of the execution.
* @throws Exception Any Exception.
*/
Object execute() throws Exception;
Object execute();
/**
* Call this when it is necessary to clone the invocation before
@@ -175,14 +167,13 @@ public abstract class AbstractRequestHandlerAdvice extends IntegrationObjectSupp
* multiple times - for example in a retry advice.
*
* @return The result of the execution.
* @throws Exception Any Exception.
*/
Object cloneAndExecute() throws Exception;
Object cloneAndExecute();
}
@SuppressWarnings("serial")
private static final class ThrowableHolderException extends RuntimeException {
protected static final class ThrowableHolderException extends RuntimeException {
ThrowableHolderException(Throwable cause) {
super(cause);

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.
@@ -79,7 +79,7 @@ public class ErrorMessageSendingRecoverer extends ErrorMessagePublisher implemen
}
@Override
public Object recover(RetryContext context) throws Exception {
public Object recover(RetryContext context) {
publish(context.getLastThrowable(), context);
return null;
}

View File

@@ -232,9 +232,7 @@ public class ExpressionEvaluatingRequestHandlerAdvice extends AbstractRequestHan
}
@Override
protected Object doInvoke(ExecutionCallback callback, Object target, Message<?> message)
throws Exception { // NOSONAR
protected Object doInvoke(ExecutionCallback callback, Object target, Message<?> message) {
try {
Object result = callback.execute();
if (this.onSuccessExpression != null) {
@@ -242,7 +240,7 @@ public class ExpressionEvaluatingRequestHandlerAdvice extends AbstractRequestHan
}
return result;
}
catch (Exception e) {
catch (RuntimeException e) {
Exception actualException = unwrapExceptionIfNecessary(e);
if (this.onFailureExpression != null) {
Object evalResult = evaluateFailureExpression(message, actualException);
@@ -251,13 +249,18 @@ public class ExpressionEvaluatingRequestHandlerAdvice extends AbstractRequestHan
}
}
if (!this.trapException) {
throw actualException;
if (e instanceof ThrowableHolderException) { // NOSONAR
throw (ThrowableHolderException) e;
}
else {
throw new ThrowableHolderException(actualException); // NOSONAR lost stack trace
}
}
return null;
}
}
private void evaluateSuccessExpression(Message<?> message) throws Exception { // NOSONAR
private void evaluateSuccessExpression(Message<?> message) {
Object evalResult;
try {
evalResult = this.onSuccessExpression.getValue(prepareEvaluationContextToUse(null), message);
@@ -274,7 +277,7 @@ public class ExpressionEvaluatingRequestHandlerAdvice extends AbstractRequestHan
this.messagingTemplate.send(this.successChannel, resultMessage);
}
if (evalResult instanceof Exception && this.propagateOnSuccessEvaluationFailures) {
throw (Exception) evalResult;
throw new ThrowableHolderException((Exception) evalResult);
}
}

View File

@@ -119,7 +119,7 @@ public class RateLimiterRequestHandlerAdvice extends AbstractRequestHandlerAdvic
}
/**
* Get a {@link RateLimiter} which is configured for this advice.
* Get the {@link RateLimiter} which is configured for this advice.
* @return the {@link RateLimiter} for this advice.
*/
public RateLimiter getRateLimiter() {

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.
@@ -51,7 +51,7 @@ public class RequestHandlerCircuitBreakerAdvice extends AbstractRequestHandlerAd
}
@Override
protected Object doInvoke(ExecutionCallback callback, Object target, Message<?> message) throws Exception {
protected Object doInvoke(ExecutionCallback callback, Object target, Message<?> message) {
AdvisedMetadata metadata = this.metadataMap.get(target);
if (metadata == null) {
this.metadataMap.putIfAbsent(target, new AdvisedMetadata());
@@ -72,7 +72,12 @@ public class RequestHandlerCircuitBreakerAdvice extends AbstractRequestHandlerAd
catch (Exception e) {
metadata.getFailures().incrementAndGet();
metadata.setLastFailure(System.currentTimeMillis());
throw this.unwrapExceptionIfNecessary(e);
if (e instanceof ThrowableHolderException) { // NOSONAR
throw (ThrowableHolderException) e;
}
else {
throw new ThrowableHolderException(e);
}
}
}

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.
@@ -42,15 +42,20 @@ import org.springframework.util.Assert;
public class RequestHandlerRetryAdvice extends AbstractRequestHandlerAdvice
implements RetryListener {
private volatile RetryTemplate retryTemplate = new RetryTemplate();
private RetryTemplate retryTemplate = new RetryTemplate();
private volatile RecoveryCallback<Object> recoveryCallback;
private RecoveryCallback<Object> recoveryCallback;
private static final ThreadLocal<Message<?>> messageHolder = new ThreadLocal<Message<?>>();
// Stateless unless a state generator is provided
private volatile RetryStateGenerator retryStateGenerator = message -> null;
/**
* Set the retry template. Cause traversal should be enabled in the retry policy
* because user exceptions may be wrapped in a {@link MessagingException}.
* @param retryTemplate the retry template.
*/
public void setRetryTemplate(RetryTemplate retryTemplate) {
Assert.notNull(retryTemplate, "'retryTemplate' cannot be null");
this.retryTemplate = retryTemplate;
@@ -72,8 +77,7 @@ public class RequestHandlerRetryAdvice extends AbstractRequestHandlerAdvice
}
@Override
protected Object doInvoke(final ExecutionCallback callback, Object target, final Message<?> message)
throws Exception {
protected Object doInvoke(final ExecutionCallback callback, Object target, final Message<?> message) {
RetryState retryState = null;
retryState = this.retryStateGenerator.determineRetryState(message);
messageHolder.set(message);
@@ -87,8 +91,11 @@ public class RequestHandlerRetryAdvice extends AbstractRequestHandlerAdvice
}
throw e;
}
catch (ThrowableHolderException e) { // NOSONAR catch and rethrow
throw e;
}
catch (Exception e) {
throw new MessagingException(message, "Failed to invoke handler", unwrapExceptionIfNecessary(e));
throw new ThrowableHolderException(e);
}
finally {
messageHolder.remove();

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2017-2018 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.
@@ -58,7 +58,7 @@ public class MapArgumentResolver extends AbstractExpressionEvaluator
@Override
@SuppressWarnings("unchecked")
public Object resolveArgument(MethodParameter parameter, Message<?> message) throws Exception {
public Object resolveArgument(MethodParameter parameter, Message<?> message) {
Object payload = message.getPayload();
if (Properties.class.isAssignableFrom(parameter.getParameterType())) {
Map<String, Object> map = message.getHeaders();

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2017-2018 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.
@@ -50,7 +50,7 @@ public class PayloadExpressionArgumentResolver extends AbstractExpressionEvaluat
}
@Override
public Object resolveArgument(MethodParameter parameter, Message<?> message) throws Exception {
public Object resolveArgument(MethodParameter parameter, Message<?> message) {
Expression expression = this.expressionCache.get(parameter);
if (expression == null) {
Payload ann = parameter.getParameterAnnotation(Payload.class);

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2017-2018 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.
@@ -55,7 +55,7 @@ public class PayloadsArgumentResolver extends AbstractExpressionEvaluator
@Override
@SuppressWarnings("unchecked")
public Object resolveArgument(MethodParameter parameter, Message<?> message) throws Exception {
public Object resolveArgument(MethodParameter parameter, Message<?> message) {
Object payload = message.getPayload();
Assert.state(payload instanceof Collection,
"This Argument Resolver support only messages with payload as Collection<Message<?>>");

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.
@@ -29,6 +29,6 @@ import org.springframework.messaging.Message;
public interface OutboundMessageMapper<T> {
@Nullable
T fromMessage(Message<?> message) throws Exception;
T fromMessage(Message<?> message) throws Exception; // NOSONAR
}

View File

@@ -16,8 +16,14 @@
package org.springframework.integration.util;
import java.util.List;
import java.util.function.BiConsumer;
import java.util.function.Consumer;
import org.springframework.util.CollectionUtils;
import org.springframework.util.ObjectUtils;
import org.springframework.util.StringUtils;
/**
* Chained utility methods to simplify some Java repetitive code. Obtain a reference to
* the singleton {@link #INSTANCE} and then chain calls to the utility methods.
@@ -67,4 +73,103 @@ public final class JavaUtils {
return this;
}
/**
* Invoke {@link Consumer#accept(Object)} with the value if it is not null or empty.
* @param value the value.
* @param consumer the consumer.
* @return this.
* @since 5.2
*/
public JavaUtils acceptIfHasText(String value, Consumer<String> consumer) {
if (StringUtils.hasText(value)) {
consumer.accept(value);
}
return this;
}
/**
* Invoke {@link Consumer#accept(Object)} with the value if it is not null or empty.
* @param value the value.
* @param consumer the consumer.
* @param <T> the value type.
* @return this.
* @since 5.2
*/
public <T> JavaUtils acceptIfNotEmpty(List<T> value, Consumer<List<T>> consumer) {
if (!CollectionUtils.isEmpty(value)) {
consumer.accept(value);
}
return this;
}
/**
* Invoke {@link Consumer#accept(Object)} with the value if it is not null or empty.
* @param value the value.
* @param consumer the consumer.
* @param <T> the value type.
* @return this.
* @since 5.2
*/
public <T> JavaUtils acceptIfNotEmpty(T[] value, Consumer<T[]> consumer) {
if (!ObjectUtils.isEmpty(value)) {
consumer.accept(value);
}
return this;
}
/**
* Invoke {@link BiConsumer#accept(Object, Object)} with the arguments if the
* condition is true.
* @param condition the condition.
* @param t1 the first consumer argument
* @param t2 the second consumer argument
* @param consumer the consumer.
* @param <T1> the first argument type.
* @param <T2> the second argument type.
* @return this.
* @since 5.2
*/
public <T1, T2> JavaUtils acceptIfCondition(boolean condition, T1 t1, T2 t2, BiConsumer<T1, T2> consumer) {
if (condition) {
consumer.accept(t1, t2);
}
return this;
}
/**
* Invoke {@link BiConsumer#accept(Object, Object)} with the arguments if the t2
* argument is not null.
* @param t1 the first argument
* @param t2 the second consumer argument
* @param consumer the consumer.
* @param <T1> the first argument type.
* @param <T2> the second argument type.
* @return this.
* @since 5.2
*/
public <T1, T2> JavaUtils acceptIfNotNull(T1 t1, T2 t2, BiConsumer<T1, T2> consumer) {
if (t2 != null) {
consumer.accept(t1, t2);
}
return this;
}
/**
* Invoke {@link BiConsumer#accept(Object, Object)} with the arguments if the value
* argument is not null or empty.
* @param t1 the first consumer argument.
* @param value the second consumer argument
* @param <T> the first argument type.
* @param consumer the consumer.
* @return this.
* @since 5.2
*/
public <T> JavaUtils acceptIfHasText(T t1, String value, BiConsumer<T, String> consumer) {
if (StringUtils.hasText(value)) {
consumer.accept(t1, value);
}
return this;
}
}

View File

@@ -190,7 +190,7 @@ public class FilterParserTests {
public static class FooFilter extends AbstractRequestHandlerAdvice {
@Override
protected Object doInvoke(ExecutionCallback callback, Object target, Message<?> message) throws Exception {
protected Object doInvoke(ExecutionCallback callback, Object target, Message<?> message) {
adviceCalled++;
return callback.execute();
}

View File

@@ -273,7 +273,7 @@ public class FilterAnnotationPostProcessorTests {
public static class TestAdvice extends AbstractRequestHandlerAdvice {
@Override
protected Object doInvoke(ExecutionCallback callback, Object target, Message<?> message) throws Exception {
protected Object doInvoke(ExecutionCallback callback, Object target, Message<?> message) {
return callback.execute();
}

View File

@@ -395,7 +395,7 @@ public class MessagingAnnotationPostProcessorTests {
public static class ServiceActivatorAdvice extends AbstractRequestHandlerAdvice {
@Override
protected Object doInvoke(ExecutionCallback callback, Object target, Message<?> message) throws Exception {
protected Object doInvoke(ExecutionCallback callback, Object target, Message<?> message) {
return callback.execute() + " advised";
}

View File

@@ -258,7 +258,7 @@ public class EnricherParserTests {
public static class FooAdvice extends AbstractRequestHandlerAdvice {
@Override
protected Object doInvoke(ExecutionCallback callback, Object target, Message<?> message) throws Exception {
protected Object doInvoke(ExecutionCallback callback, Object target, Message<?> message) {
adviceCalled++;
return callback.execute();
}

View File

@@ -146,7 +146,7 @@ public class EnricherParserTests4 {
public static class FooAdvice extends AbstractRequestHandlerAdvice {
@Override
protected Object doInvoke(ExecutionCallback callback, Object target, Message<?> message) throws Exception {
protected Object doInvoke(ExecutionCallback callback, Object target, Message<?> message) {
adviceCalled++;
return callback.execute();
}

View File

@@ -279,7 +279,7 @@ public class ServiceActivatorParserTests {
public static class BarAdvice extends AbstractRequestHandlerAdvice {
@Override
protected Object doInvoke(ExecutionCallback callback, Object target, Message<?> message) throws Exception {
protected Object doInvoke(ExecutionCallback callback, Object target, Message<?> message) {
callback.execute();
return "bar";
}

View File

@@ -375,9 +375,7 @@ public class TransformerTests {
return new AbstractRequestHandlerAdvice() {
@Override
protected Object doInvoke(ExecutionCallback callback, Object target, Message<?> message)
throws Exception {
protected Object doInvoke(ExecutionCallback callback, Object target, Message<?> message) {
adviceChannel().send(message);
return callback.execute();
}

View File

@@ -146,7 +146,7 @@ public class AdvisedMessageHandlerTests {
adviceChain.add(new AbstractRequestHandlerAdvice() {
@Override
protected Object doInvoke(ExecutionCallback callback, Object target, Message<?> message) throws Exception {
protected Object doInvoke(ExecutionCallback callback, Object target, Message<?> message) {
compName.set(((AbstractReplyProducingMessageHandler.RequestHandler) target).getAdvisedHandler()
.getComponentName());
return callback.execute();
@@ -734,7 +734,7 @@ public class AdvisedMessageHandlerTests {
adviceChain.add(new AbstractRequestHandlerAdvice() {
@Override
protected Object doInvoke(ExecutionCallback callback, Object target, Message<?> message) throws Exception {
protected Object doInvoke(ExecutionCallback callback, Object target, Message<?> message) {
outerCounter.incrementAndGet();
return callback.execute();
}
@@ -814,14 +814,18 @@ public class AdvisedMessageHandlerTests {
AbstractRequestHandlerAdvice advice = new AbstractRequestHandlerAdvice() {
@Override
protected Object doInvoke(ExecutionCallback callback, Object target, Message<?> message) throws Exception {
protected Object doInvoke(ExecutionCallback callback, Object target, Message<?> message) {
Object result;
try {
result = callback.execute();
}
catch (Exception e) {
// should not be unwrapped because the cause is a Throwable
throw this.unwrapExceptionIfNecessary(e);
if (e instanceof ThrowableHolderException) {
throw (ThrowableHolderException) e;
}
else {
throw new ThrowableHolderException(e);
}
}
return result;
}
@@ -882,7 +886,7 @@ public class AdvisedMessageHandlerTests {
Advice advice = new AbstractRequestHandlerAdvice() {
@Override
protected Object doInvoke(ExecutionCallback callback, Object target, Message<?> message) throws Exception {
protected Object doInvoke(ExecutionCallback callback, Object target, Message<?> message) {
called.set(true);
return callback.execute();
}
@@ -938,7 +942,7 @@ public class AdvisedMessageHandlerTests {
adviceChain.add(new AbstractRequestHandlerAdvice() {
@Override
protected Object doInvoke(ExecutionCallback callback, Object target, Message<?> message) throws Exception {
protected Object doInvoke(ExecutionCallback callback, Object target, Message<?> message) {
Object result = callback.execute();
discardedWithinAdvice.set(discardChannel.receive(0));
return result;
@@ -963,7 +967,7 @@ public class AdvisedMessageHandlerTests {
adviceChain.add(new AbstractRequestHandlerAdvice() {
@Override
protected Object doInvoke(ExecutionCallback callback, Object target, Message<?> message) throws Exception {
protected Object doInvoke(ExecutionCallback callback, Object target, Message<?> message) {
Object result = callback.execute();
discardedWithinAdvice.set(discardChannel.receive(0));
adviceCalled.set(true);
@@ -1015,9 +1019,8 @@ public class AdvisedMessageHandlerTests {
Map<Class<? extends Throwable>, Boolean> retryableExceptions = new HashMap<Class<? extends Throwable>, Boolean>();
retryableExceptions.put(MyException.class, retryForMyException);
retryableExceptions.put(MessagingException.class, true);
retryTemplate.setRetryPolicy(new SimpleRetryPolicy(3, retryableExceptions));
retryTemplate.setRetryPolicy(new SimpleRetryPolicy(3, retryableExceptions, true));
advice.setRetryTemplate(retryTemplate);

View File

@@ -161,7 +161,7 @@ public class IdempotentReceiverTests {
private int adviceCalled;
@Override
protected Object doInvoke(ExecutionCallback callback, Object target, Message<?> message) throws Exception {
protected Object doInvoke(ExecutionCallback callback, Object target, Message<?> message) {
adviceCalled++;
return callback.execute();
}

View File

@@ -96,7 +96,7 @@ public class TransformerContextTests {
public static class FooAdvice extends AbstractRequestHandlerAdvice {
@Override
protected Object doInvoke(ExecutionCallback callback, Object target, Message<?> message) throws Exception {
protected Object doInvoke(ExecutionCallback callback, Object target, Message<?> message) {
adviceCalled++;
return callback.execute();
}

View File

@@ -161,7 +161,7 @@ public class EventOutboundChannelAdapterParserTests {
public static class FooAdvice extends AbstractRequestHandlerAdvice {
@Override
protected Object doInvoke(ExecutionCallback callback, Object target, Message<?> message) throws Exception {
protected Object doInvoke(ExecutionCallback callback, Object target, Message<?> message) {
adviceCalled++;
return callback.execute();
}

View File

@@ -204,7 +204,7 @@ public class FileTailInboundChannelAdapterFactoryBean extends AbstractFactoryBea
}
@Override
protected FileTailingMessageProducerSupport createInstance() throws Exception {
protected FileTailingMessageProducerSupport createInstance() {
FileTailingMessageProducerSupport adapter;
if (this.delay == null && this.end == null && this.reopen == null) {
adapter = new OSDelegatingFileTailingMessageProducer();

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.
@@ -17,6 +17,7 @@
package org.springframework.integration.file.transformer;
import java.io.File;
import java.io.IOException;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@@ -107,8 +108,8 @@ public abstract class AbstractFilePayloadTransformer<T> implements Transformer,
*
* @param file The file.
* @return The result of the transformation.
* @throws Exception Any Exception.
* @throws IOException Any IOException.
*/
protected abstract T transformFile(File file) throws Exception;
protected abstract T transformFile(File file) throws IOException;
}

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2008 the original author or authors.
* Copyright 2002-2019 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -17,6 +17,7 @@
package org.springframework.integration.file.transformer;
import java.io.File;
import java.io.IOException;
import org.springframework.util.FileCopyUtils;
@@ -28,7 +29,7 @@ import org.springframework.util.FileCopyUtils;
public class FileToByteArrayTransformer extends AbstractFilePayloadTransformer<byte[]> {
@Override
protected final byte[] transformFile(File file) throws Exception {
protected final byte[] transformFile(File file) throws IOException {
return FileCopyUtils.copyToByteArray(file);
}

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2014 the original author or authors.
* Copyright 2002-2019 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -19,6 +19,7 @@ package org.springframework.integration.file.transformer;
import java.io.BufferedReader;
import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.Reader;
import java.nio.charset.Charset;
@@ -49,7 +50,7 @@ public class FileToStringTransformer extends AbstractFilePayloadTransformer<Stri
}
@Override
protected final String transformFile(File file) throws Exception {
protected final String transformFile(File file) throws IOException {
Reader reader = new BufferedReader(new InputStreamReader(new FileInputStream(file), this.charset));
return FileCopyUtils.copyToString(reader);
}

View File

@@ -368,7 +368,7 @@ public class FileOutboundChannelAdapterParserTests {
public static class FooAdvice extends AbstractRequestHandlerAdvice {
@Override
protected Object doInvoke(ExecutionCallback callback, Object target, Message<?> message) throws Exception {
protected Object doInvoke(ExecutionCallback callback, Object target, Message<?> message) {
adviceCalled++;
return callback.execute();
}

View File

@@ -173,7 +173,7 @@ public class FtpOutboundChannelAdapterParserTests {
public static class FooAdvice extends AbstractRequestHandlerAdvice {
@Override
protected Object doInvoke(ExecutionCallback callback, Object target, Message<?> message) throws Exception {
protected Object doInvoke(ExecutionCallback callback, Object target, Message<?> message) {
adviceCalled++;
return null;
}

View File

@@ -183,7 +183,7 @@ public class FtpOutboundGatewayParserTests {
public static class FooAdvice extends AbstractRequestHandlerAdvice {
@Override
protected Object doInvoke(ExecutionCallback callback, Object target, Message<?> message) throws Exception {
protected Object doInvoke(ExecutionCallback callback, Object target, Message<?> message) {
adviceCalled++;
return null;
}

View File

@@ -50,7 +50,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@DirtiesContext
public class GemfireOutboundChannelAdapterParserTests {
private GemfireOutboundChannelAdapterParser underTest = new GemfireOutboundChannelAdapterParser();
private final GemfireOutboundChannelAdapterParser underTest = new GemfireOutboundChannelAdapterParser();
private static final CountDownLatch adviceCalled = new CountDownLatch(1);
@@ -89,7 +89,7 @@ public class GemfireOutboundChannelAdapterParserTests {
public static class FooAdvice extends AbstractRequestHandlerAdvice {
@Override
protected Object doInvoke(ExecutionCallback callback, Object target, Message<?> message) throws Exception {
protected Object doInvoke(ExecutionCallback callback, Object target, Message<?> message) {
adviceCalled.countDown();
return null;
}

View File

@@ -196,6 +196,7 @@ public class GroovyControlBusTests {
private volatile boolean executed;
@Override
public void customize(GroovyObject goo) {
this.executed = true;
}
@@ -206,33 +207,41 @@ public class GroovyControlBusTests {
private final Map<String, Object> fakeRequest = new HashMap<>();
@Override
public Object getAttribute(String name, int scope) {
return fakeRequest.get(name);
}
@Override
public void setAttribute(String name, Object value, int scope) {
fakeRequest.put(name, value);
}
@Override
public void removeAttribute(String name, int scope) {
}
@Override
public String[] getAttributeNames(int scope) {
return null;
}
@Override
public void registerDestructionCallback(String name, Runnable callback, int scope) {
}
@Override
public Object resolveReference(String key) {
return null;
}
@Override
public String getSessionId() {
return null;
}
@Override
public Object getSessionMutex() {
return null;
}
@@ -242,7 +251,7 @@ public class GroovyControlBusTests {
public static class FooAdvice extends AbstractRequestHandlerAdvice {
@Override
protected Object doInvoke(ExecutionCallback callback, Object target, Message<?> message) throws Exception {
protected Object doInvoke(ExecutionCallback callback, Object target, Message<?> message) {
adviceCalled++;
return callback.execute();
}

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.
@@ -73,6 +73,8 @@ import org.springframework.util.StringUtils;
*/
public class DefaultHttpHeaderMapper implements HeaderMapper<HttpHeaders>, BeanFactoryAware, InitializingBean {
private static final String UNUSED = "unused";
protected final Log logger = LogFactory.getLog(getClass());
public static final String ACCEPT = "Accept";
@@ -486,7 +488,7 @@ public class DefaultHttpHeaderMapper implements HeaderMapper<HttpHeaders>, BeanF
}
@Override
public void afterPropertiesSet() throws Exception {
public void afterPropertiesSet() {
if (this.beanFactory != null) {
this.conversionService = IntegrationUtils.getConversionService(this.beanFactory);
}
@@ -763,7 +765,7 @@ public class DefaultHttpHeaderMapper implements HeaderMapper<HttpHeaders>, BeanF
try {
target.setDate(Long.parseLong((String) value));
}
catch (NumberFormatException e) {
catch (@SuppressWarnings(UNUSED) NumberFormatException e) {
target.setDate(this.getFirstDate((String) value, DATE));
}
}
@@ -794,7 +796,7 @@ public class DefaultHttpHeaderMapper implements HeaderMapper<HttpHeaders>, BeanF
try {
target.setExpires(Long.parseLong((String) value));
}
catch (NumberFormatException e) {
catch (@SuppressWarnings(UNUSED) NumberFormatException e) {
target.setExpires(this.getFirstDate((String) value, EXPIRES));
}
}
@@ -815,7 +817,7 @@ public class DefaultHttpHeaderMapper implements HeaderMapper<HttpHeaders>, BeanF
try {
target.setIfModifiedSince(Long.parseLong((String) value));
}
catch (NumberFormatException e) {
catch (@SuppressWarnings(UNUSED) NumberFormatException e) {
target.setIfModifiedSince(this.getFirstDate((String) value, IF_MODIFIED_SINCE));
}
}
@@ -838,7 +840,7 @@ public class DefaultHttpHeaderMapper implements HeaderMapper<HttpHeaders>, BeanF
try {
ifUnmodifiedSinceValue = this.formatDate(Long.parseLong((String) value));
}
catch (NumberFormatException e) {
catch (@SuppressWarnings(UNUSED) NumberFormatException e) {
long longValue = this.getFirstDate((String) value, IF_UNMODIFIED_SINCE);
ifUnmodifiedSinceValue = this.formatDate(longValue);
}
@@ -888,7 +890,7 @@ public class DefaultHttpHeaderMapper implements HeaderMapper<HttpHeaders>, BeanF
try {
target.setLastModified(Long.parseLong((String) value));
}
catch (NumberFormatException e) {
catch (@SuppressWarnings(UNUSED) NumberFormatException e) {
target.setLastModified(this.getFirstDate((String) value, LAST_MODIFIED));
}
}
@@ -1078,7 +1080,7 @@ public class DefaultHttpHeaderMapper implements HeaderMapper<HttpHeaders>, BeanF
.toInstant()
.toEpochMilli();
}
catch (DateTimeParseException ex) {
catch (@SuppressWarnings(UNUSED) DateTimeParseException ex) {
// ignore
}
}

View File

@@ -303,7 +303,7 @@ public class HttpOutboundChannelAdapterParserTests {
public static class FooAdvice extends AbstractRequestHandlerAdvice {
@Override
protected Object doInvoke(ExecutionCallback callback, Object target, Message<?> message) throws Exception {
protected Object doInvoke(ExecutionCallback callback, Object target, Message<?> message) {
adviceCalled++;
return null;
}

View File

@@ -237,7 +237,7 @@ public class HttpOutboundGatewayParserTests {
public static class FooAdvice extends AbstractRequestHandlerAdvice {
@Override
protected Object doInvoke(ExecutionCallback callback, Object target, Message<?> message) throws Exception {
protected Object doInvoke(ExecutionCallback callback, Object target, Message<?> message) {
adviceCalled++;
return null;
}

View File

@@ -19,7 +19,6 @@ package org.springframework.integration.ip.config;
import java.util.concurrent.Executor;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.beans.factory.BeanNameAware;
import org.springframework.beans.factory.config.AbstractFactoryBean;
import org.springframework.context.ApplicationEventPublisher;
@@ -58,7 +57,7 @@ import org.springframework.util.Assert;
* @since 2.0.5
*/
public class TcpConnectionFactoryFactoryBean extends AbstractFactoryBean<AbstractConnectionFactory>
implements Lifecycle, BeanNameAware, BeanFactoryAware, ApplicationEventPublisherAware {
implements Lifecycle, BeanNameAware, ApplicationEventPublisherAware {
private volatile AbstractConnectionFactory connectionFactory;
@@ -151,7 +150,7 @@ public class TcpConnectionFactoryFactoryBean extends AbstractFactoryBean<Abstrac
}
@Override
protected AbstractConnectionFactory createInstance() throws Exception {
protected AbstractConnectionFactory createInstance() {
if (!this.mapperSet) {
this.mapper.setBeanFactory(this.beanFactory);
}

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.
@@ -41,6 +41,7 @@ import org.springframework.integration.ip.tcp.connection.TcpListener;
import org.springframework.integration.ip.tcp.connection.TcpSender;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandlingException;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.support.ErrorMessage;
import org.springframework.util.Assert;
@@ -138,26 +139,9 @@ public class TcpOutboundGateway extends AbstractReplyProducingMessageHandler
TcpConnection connection = null;
String connectionId = null;
try {
if (!this.isSingleUse) {
logger.debug("trying semaphore");
if (!this.semaphore.tryAcquire(this.requestTimeout, TimeUnit.MILLISECONDS)) {
throw new MessageTimeoutException(requestMessage, "Timed out waiting for connection");
}
haveSemaphore = true;
if (logger.isDebugEnabled()) {
logger.debug("got semaphore");
}
}
haveSemaphore = acquireSemaphoreIfNeeded(requestMessage);
connection = this.connectionFactory.getConnection();
Long remoteTimeout = this.remoteTimeoutExpression.getValue(this.evaluationContext, requestMessage,
Long.class);
if (remoteTimeout == null) {
remoteTimeout = DEFAULT_REMOTE_TIMEOUT;
if (logger.isWarnEnabled()) {
logger.warn("remoteTimeoutExpression evaluated to null; falling back to default for message "
+ requestMessage);
}
}
Long remoteTimeout = getRemoteTimeout(requestMessage);
AsyncReply reply = new AsyncReply(remoteTimeout);
connectionId = connection.getConnectionId();
this.pendingReplies.put(connectionId, reply);
@@ -165,42 +149,83 @@ public class TcpOutboundGateway extends AbstractReplyProducingMessageHandler
logger.debug("Added pending reply " + connectionId);
}
connection.send(requestMessage);
Message<?> replyMessage = reply.getReply();
if (replyMessage == null) {
if (logger.isDebugEnabled()) {
logger.debug("Remote Timeout on " + connectionId);
}
// The connection is dirty - force it closed.
this.connectionFactory.forceClose(connection);
throw new MessageTimeoutException(requestMessage, "Timed out waiting for response");
}
if (logger.isDebugEnabled()) {
logger.debug("Response " + replyMessage);
}
return replyMessage;
return getReply(requestMessage, connection, connectionId, reply);
}
catch (Exception e) {
catch (RuntimeException e) {
logger.error("Tcp Gateway exception", e);
if (e instanceof MessagingException) {
throw (MessagingException) e;
}
throw new MessagingException("Failed to send or receive", e);
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new MessageHandlingException(requestMessage, "Interrupted", e);
}
finally {
if (connectionId != null) {
this.pendingReplies.remove(connectionId);
if (logger.isDebugEnabled()) {
logger.debug("Removed pending reply " + connectionId);
}
if (this.isSingleUse) {
connection.close();
}
cleanUp(haveSemaphore, connection, connectionId);
}
}
private boolean acquireSemaphoreIfNeeded(Message<?> requestMessage) throws InterruptedException {
if (!this.isSingleUse) {
logger.debug("trying semaphore");
if (!this.semaphore.tryAcquire(this.requestTimeout, TimeUnit.MILLISECONDS)) {
throw new MessageTimeoutException(requestMessage, "Timed out waiting for connection");
}
if (haveSemaphore) {
this.semaphore.release();
if (logger.isDebugEnabled()) {
logger.debug("released semaphore");
}
if (logger.isDebugEnabled()) {
logger.debug("got semaphore");
}
return true;
}
return false;
}
private Long getRemoteTimeout(Message<?> requestMessage) {
Long remoteTimeout = this.remoteTimeoutExpression.getValue(this.evaluationContext, requestMessage,
Long.class);
if (remoteTimeout == null) {
remoteTimeout = DEFAULT_REMOTE_TIMEOUT;
if (logger.isWarnEnabled()) {
logger.warn("remoteTimeoutExpression evaluated to null; falling back to default for message "
+ requestMessage);
}
}
return remoteTimeout;
}
private Message<?> getReply(Message<?> requestMessage, TcpConnection connection, String connectionId,
AsyncReply reply) {
Message<?> replyMessage = reply.getReply();
if (replyMessage == null) {
if (logger.isDebugEnabled()) {
logger.debug("Remote Timeout on " + connectionId);
}
// The connection is dirty - force it closed.
this.connectionFactory.forceClose(connection);
throw new MessageTimeoutException(requestMessage, "Timed out waiting for response");
}
if (logger.isDebugEnabled()) {
logger.debug("Response " + replyMessage);
}
return replyMessage;
}
private void cleanUp(boolean haveSemaphore, TcpConnection connection, String connectionId) {
if (connectionId != null) {
this.pendingReplies.remove(connectionId);
if (logger.isDebugEnabled()) {
logger.debug("Removed pending reply " + connectionId);
}
if (this.isSingleUse) {
connection.close();
}
}
if (haveSemaphore) {
this.semaphore.release();
if (logger.isDebugEnabled()) {
logger.debug("released semaphore");
}
}
}
@@ -334,13 +359,13 @@ public class TcpOutboundGateway extends AbstractReplyProducingMessageHandler
* @return The return message or null if we time out
* @throws Exception
*/
public Message<?> getReply() throws Exception {
public Message<?> getReply() {
try {
if (!this.latch.await(this.remoteTimeout, TimeUnit.MILLISECONDS)) {
return null;
}
}
catch (InterruptedException e) {
catch (@SuppressWarnings("unused") InterruptedException e) {
Thread.currentThread().interrupt();
}
boolean waitForMessageAfterError = true;
@@ -351,19 +376,31 @@ public class TcpOutboundGateway extends AbstractReplyProducingMessageHandler
* before the reply, on a different thread.
*/
logger.debug("second chance");
this.secondChanceLatch.await(TcpOutboundGateway.this.secondChanceDelay, TimeUnit.SECONDS); // NOSONAR
try {
this.secondChanceLatch.await(TcpOutboundGateway.this.secondChanceDelay, TimeUnit.SECONDS); // NOSONAR
}
catch (@SuppressWarnings("unused") InterruptedException e) {
Thread.currentThread().interrupt();
doThrowErrorMessagePayload();
}
waitForMessageAfterError = false;
}
else if (this.reply.getPayload() instanceof MessagingException) {
throw (MessagingException) this.reply.getPayload();
}
else {
throw new MessagingException("Exception while awaiting reply", (Throwable) this.reply.getPayload());
doThrowErrorMessagePayload();
}
}
return this.reply;
}
private void doThrowErrorMessagePayload() {
if (this.reply.getPayload() instanceof MessagingException) {
throw (MessagingException) this.reply.getPayload();
}
else {
throw new MessagingException("Exception while awaiting reply", (Throwable) this.reply.getPayload());
}
}
/**
* We have a race condition when a socket is closed right after the reply is received. The close "error"
* might arrive before the actual reply. Overwrite an error with a good reply, but not vice-versa.

View File

@@ -63,14 +63,15 @@ public abstract class AbstractClientConnectionFactory extends AbstractConnection
* Obtains a connection - if {@link #setSingleUse(boolean)} was called with
* true, a new connection is returned; otherwise a single connection is
* reused for all requests while the connection remains open.
* @throws InterruptedException if interrupted.
*/
@Override
public TcpConnectionSupport getConnection() throws Exception {
this.checkActive();
public TcpConnectionSupport getConnection() throws InterruptedException {
checkActive();
return obtainConnection();
}
protected TcpConnectionSupport obtainConnection() throws Exception {
protected TcpConnectionSupport obtainConnection() throws InterruptedException {
if (!this.isSingleUse()) {
TcpConnectionSupport connection = obtainSharedConnection();
if (connection != null) {
@@ -92,11 +93,10 @@ public abstract class AbstractClientConnectionFactory extends AbstractConnection
finally {
this.theConnectionLock.readLock().unlock();
}
return null;
}
protected final TcpConnectionSupport obtainNewConnection() throws Exception {
protected final TcpConnectionSupport obtainNewConnection() throws InterruptedException {
boolean singleUse = this.isSingleUse();
if (!singleUse) {
this.theConnectionLock.writeLock().lockInterruptibly();
@@ -115,14 +115,14 @@ public abstract class AbstractClientConnectionFactory extends AbstractConnection
logger.debug("Opening new socket connection to " + this.getHost() + ":" + this.getPort());
}
connection = this.buildNewConnection();
connection = buildNewConnection();
if (!singleUse) {
this.setTheConnection(connection);
}
connection.publishConnectionOpenEvent();
return connection;
}
catch (Exception e) {
catch (RuntimeException e) {
ApplicationEventPublisher applicationEventPublisher = getApplicationEventPublisher();
if (applicationEventPublisher != null) {
applicationEventPublisher.publishEvent(new TcpConnectionFailedEvent(this, e));
@@ -136,7 +136,7 @@ public abstract class AbstractClientConnectionFactory extends AbstractConnection
}
}
protected TcpConnectionSupport buildNewConnection() throws Exception {
protected TcpConnectionSupport buildNewConnection() {
throw new UnsupportedOperationException("Factories that don't override this class' obtainConnection() must implement this method");
}

View File

@@ -18,6 +18,7 @@ package org.springframework.integration.ip.tcp.connection;
import java.io.EOFException;
import java.io.IOException;
import java.io.UncheckedIOException;
import java.net.Socket;
import java.net.SocketException;
import java.net.SocketTimeoutException;
@@ -62,6 +63,8 @@ import org.springframework.util.Assert;
public abstract class AbstractConnectionFactory extends IntegrationObjectSupport
implements ConnectionFactory, ApplicationEventPublisherAware {
private static final String UNUSED = "unused";
protected static final int DEFAULT_REPLY_TIMEOUT = 10000;
private static final int DEFAULT_NIO_HARVEST_INTERVAL = 2000;
@@ -351,24 +354,24 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport
/**
* Registers a TcpListener to receive messages after
* the payload has been converted from the input data.
* @param listener the TcpListener.
* @param listenerToRegister the TcpListener.
*/
public void registerListener(TcpListener listener) {
public void registerListener(TcpListener listenerToRegister) {
Assert.isNull(this.listener, this.getClass().getName() +
" may only be used by one inbound adapter");
this.listener = listener;
this.listener = listenerToRegister;
}
/**
* Registers a TcpSender; for server sockets, used to
* provide connection information so a sender can be used
* to reply to incoming messages.
* @param sender The sender
* @param senderToRegister The sender
*/
public void registerSender(TcpSender sender) {
public void registerSender(TcpSender senderToRegister) {
Assert.isNull(this.sender, this.getClass().getName() +
" may only be used by one outbound adapter");
this.sender = sender;
this.sender = senderToRegister;
}
/**
@@ -557,7 +560,7 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport
}
}
}
catch (InterruptedException e) {
catch (@SuppressWarnings(UNUSED) InterruptedException e) {
executorService.shutdownNow();
Thread.currentThread().interrupt();
}
@@ -572,7 +575,7 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport
}
}
protected TcpConnectionSupport wrapConnection(TcpConnectionSupport connectionArg) throws Exception {
protected TcpConnectionSupport wrapConnection(TcpConnectionSupport connectionArg) {
TcpConnectionSupport connection = connectionArg;
try {
if (this.interceptorFactoryChain == null) {
@@ -611,12 +614,11 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport
* @param selectionCount Number of IO Events, if 0 we were probably woken up by a close.
* @param selector The selector.
* @param server The server socket channel.
* @param connections Map of connections.
* @throws IOException Any IOException.
* @param connectionMap Map of connections.
*/
protected void processNioSelections(int selectionCount, final Selector selector,
@Nullable ServerSocketChannel server,
Map<SocketChannel, TcpNioConnection> connections) throws IOException {
Map<SocketChannel, TcpNioConnection> connectionMap) {
final long now = System.currentTimeMillis();
rescheduleDelayedReads(selector, now);
@@ -624,7 +626,7 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport
now >= this.nextCheckForClosedNioConnections ||
selectionCount == 0) {
this.nextCheckForClosedNioConnections = now + this.nioHarvestInterval;
Iterator<Entry<SocketChannel, TcpNioConnection>> it = connections.entrySet().iterator();
Iterator<Entry<SocketChannel, TcpNioConnection>> it = connectionMap.entrySet().iterator();
while (it.hasNext()) {
SocketChannel channel = it.next().getKey();
if (!channel.isOpen()) {
@@ -632,7 +634,7 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport
it.remove();
}
else if (this.soTimeout > 0) {
TcpNioConnection connection = connections.get(channel);
TcpNioConnection connection = connectionMap.get(channel);
if (now - connection.getLastRead() >= this.soTimeout) {
/*
* For client connections, we have to wait for 2 timeouts if the last
@@ -650,7 +652,7 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport
if (logger.isWarnEnabled()) {
logger.warn("Timing out TcpNioConnection " + connection.getConnectionId());
}
SocketTimeoutException exception = new SocketTimeoutException("Timing out connection");
Exception exception = new SocketTimeoutException("Timing out connection");
connection.publishConnectionExceptionEvent(exception);
connection.timeout();
connection.sendExceptionToListener(exception);
@@ -689,7 +691,7 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport
try {
connection.readPacket();
}
catch (RejectedExecutionException e1) {
catch (@SuppressWarnings(UNUSED) RejectedExecutionException e1) {
delayRead(selector, now, key);
delayed = true;
}
@@ -715,7 +717,7 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport
}
});
}
catch (RejectedExecutionException e) {
catch (@SuppressWarnings(UNUSED) RejectedExecutionException e) {
delayRead(selector, now, key);
}
}
@@ -731,7 +733,7 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport
logger.error("Unexpected key: " + key);
}
}
catch (CancelledKeyException e) {
catch (@SuppressWarnings(UNUSED) CancelledKeyException e) {
if (logger.isDebugEnabled()) {
logger.debug("Selection key " + key + " cancelled");
}
@@ -787,7 +789,7 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport
}
}
}
catch (InterruptedException e) {
catch (@SuppressWarnings(UNUSED) InterruptedException e) {
Thread.currentThread().interrupt();
}
finally {
@@ -801,9 +803,8 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport
* @param selector The selector.
* @param server The server socket channel.
* @param now The current time.
* @throws IOException Any IOException.
*/
protected void doAccept(final Selector selector, ServerSocketChannel server, long now) throws IOException {
protected void doAccept(final Selector selector, ServerSocketChannel server, long now) {
throw new UnsupportedOperationException("Nio server factory must override this method");
}
@@ -874,9 +875,9 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport
this.active = active;
}
protected void checkActive() throws IOException {
protected void checkActive() {
if (!this.isActive()) {
throw new IOException(this + " connection factory has not been started");
throw new UncheckedIOException(new IOException(this + " connection factory has not been started"));
}
}

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.
@@ -214,7 +214,7 @@ public abstract class AbstractServerConnectionFactory extends AbstractConnection
try {
taskScheduler.schedule((Runnable) () -> eventPublisher.publishEvent(event), new Date());
}
catch (TaskRejectedException e) {
catch (@SuppressWarnings("unused") TaskRejectedException e) {
eventPublisher.publishEvent(event);
}
}

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.
@@ -133,7 +133,7 @@ public class CachingClientConnectionFactory extends AbstractClientConnectionFact
}
@Override
public TcpConnectionSupport obtainConnection() throws Exception {
public TcpConnectionSupport obtainConnection() {
return new CachedConnection(this.pool.getItem(), getListener());
}

View File

@@ -34,7 +34,7 @@ public class DefaultTcpNetConnectionSupport extends AbstractTcpConnectionSupport
@Override
public TcpNetConnection createNewConnection(Socket socket, boolean server, boolean lookupHost,
ApplicationEventPublisher applicationEventPublisher, String connectionFactoryName) throws Exception {
ApplicationEventPublisher applicationEventPublisher, String connectionFactoryName) {
if (isPushbackCapable()) {
return new PushBackTcpNetConnection(socket, server, lookupHost, applicationEventPublisher,
connectionFactoryName, getPushbackBufferSize());

View File

@@ -38,7 +38,7 @@ public class DefaultTcpNioConnectionSupport extends AbstractTcpConnectionSupport
@Override
public TcpNioConnection createNewConnection(SocketChannel socketChannel, boolean server, boolean lookupHost,
ApplicationEventPublisher applicationEventPublisher, String connectionFactoryName) throws Exception {
ApplicationEventPublisher applicationEventPublisher, String connectionFactoryName) {
if (isPushbackCapable()) {
return new PushBackTcpNioConnection(socketChannel, server, lookupHost, applicationEventPublisher,
connectionFactoryName, getPushbackBufferSize());
@@ -61,7 +61,7 @@ public class DefaultTcpNioConnectionSupport extends AbstractTcpConnectionSupport
PushBackTcpNioConnection(SocketChannel socketChannel, boolean server, boolean lookupHost,
ApplicationEventPublisher applicationEventPublisher, @Nullable String connectionFactoryName,
int bufferSize) throws Exception {
int bufferSize) {
super(socketChannel, server, lookupHost, applicationEventPublisher, connectionFactoryName);
this.pushbackBufferSize = bufferSize;

View File

@@ -75,7 +75,7 @@ public class DefaultTcpNioSSLConnectionSupport extends AbstractTcpConnectionSupp
*/
@Override
public TcpNioConnection createNewConnection(SocketChannel socketChannel, boolean server, boolean lookupHost,
ApplicationEventPublisher applicationEventPublisher, String connectionFactoryName) throws Exception {
ApplicationEventPublisher applicationEventPublisher, String connectionFactoryName) {
SSLEngine sslEngine = this.sslContext.createSSLEngine();
postProcessSSLEngine(sslEngine);
@@ -120,7 +120,8 @@ public class DefaultTcpNioSSLConnectionSupport extends AbstractTcpConnectionSupp
PushBackTcpNioSSLConnection(SocketChannel socketChannel, boolean server, boolean lookupHost,
ApplicationEventPublisher applicationEventPublisher, String connectionFactoryName, SSLEngine sslEngine,
int bufferSize) throws Exception {
int bufferSize) {
super(socketChannel, server, lookupHost, applicationEventPublisher, connectionFactoryName, sslEngine);
this.pushbackBufferSize = bufferSize;
this.connectionId = "pushback:" + super.getConnectionId();

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.
@@ -92,7 +92,7 @@ public class FailoverClientConnectionFactory extends AbstractClientConnectionFac
}
@Override
protected TcpConnectionSupport obtainConnection() throws Exception {
protected TcpConnectionSupport obtainConnection() throws InterruptedException {
TcpConnectionSupport connection = this.getTheConnection();
if (connection != null && connection.isOpen()) {
((FailoverTcpConnection) connection).incrementEpoch();
@@ -146,7 +146,7 @@ public class FailoverClientConnectionFactory extends AbstractClientConnectionFac
*/
private final class FailoverTcpConnection extends TcpConnectionSupport implements TcpListener {
private final List<AbstractClientConnectionFactory> factories;
private final List<AbstractClientConnectionFactory> connectionFactories;
private final String connectionId;
@@ -160,8 +160,8 @@ public class FailoverClientConnectionFactory extends AbstractClientConnectionFac
private final AtomicLong epoch = new AtomicLong();
private FailoverTcpConnection(List<AbstractClientConnectionFactory> factories) throws Exception {
this.factories = factories;
private FailoverTcpConnection(List<AbstractClientConnectionFactory> factories) throws InterruptedException {
this.connectionFactories = factories;
this.factoryIterator = factories.iterator();
findAConnection();
this.connectionId = UUID.randomUUID().toString();
@@ -177,14 +177,14 @@ public class FailoverClientConnectionFactory extends AbstractClientConnectionFac
* This allows for the condition where the current connection is closed,
* the current factory can serve up a new connection, but all other
* factories are down.
* @throws Exception if an exception occurs
* @throws InterruptedException if interrupted.
*/
private synchronized void findAConnection() throws Exception {
private synchronized void findAConnection() throws InterruptedException {
boolean success = false;
AbstractClientConnectionFactory lastFactoryToTry = this.currentFactory;
AbstractClientConnectionFactory nextFactory = null;
if (!this.factoryIterator.hasNext()) {
this.factoryIterator = this.factories.iterator();
this.factoryIterator = this.connectionFactories.iterator();
}
boolean retried = false;
while (!success) {
@@ -198,7 +198,7 @@ public class FailoverClientConnectionFactory extends AbstractClientConnectionFac
this.currentFactory = nextFactory;
success = this.delegate.isOpen();
}
catch (Exception e) {
catch (RuntimeException e) {
if (logger.isDebugEnabled()) {
logger.debug(nextFactory + " failed with "
+ e.toString()
@@ -213,7 +213,7 @@ public class FailoverClientConnectionFactory extends AbstractClientConnectionFac
this.open = false;
throw e;
}
this.factoryIterator = this.factories.iterator();
this.factoryIterator = this.connectionFactories.iterator();
retried = true;
}
}
@@ -237,7 +237,7 @@ public class FailoverClientConnectionFactory extends AbstractClientConnectionFac
* If send fails on a connection from every factory, we give up.
*/
@Override
public synchronized void send(Message<?> message) throws Exception {
public synchronized void send(Message<?> message) {
boolean success = false;
AbstractClientConnectionFactory lastFactoryToTry = this.currentFactory;
AbstractClientConnectionFactory lastFactoryTried = null;
@@ -248,7 +248,7 @@ public class FailoverClientConnectionFactory extends AbstractClientConnectionFac
this.delegate.send(message);
success = true;
}
catch (Exception e) {
catch (RuntimeException e) {
if (retried && lastFactoryTried == lastFactoryToTry) {
logger.error("All connection factories exhausted", e);
this.open = false;
@@ -259,7 +259,12 @@ public class FailoverClientConnectionFactory extends AbstractClientConnectionFac
logger.debug("Send to " + this.delegate.getConnectionId() + " failed; attempting failover", e);
}
this.delegate.close();
findAConnection();
try {
findAConnection();
}
catch (@SuppressWarnings("unused") InterruptedException e1) {
Thread.currentThread().interrupt();
}
if (logger.isDebugEnabled()) {
logger.debug("Failing over to " + this.delegate.getConnectionId());
}
@@ -268,7 +273,7 @@ public class FailoverClientConnectionFactory extends AbstractClientConnectionFac
}
@Override
public Object getPayload() throws Exception {
public Object getPayload() {
return this.delegate.getPayload();
}

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.
@@ -47,18 +47,16 @@ public interface TcpConnection extends Runnable {
/**
* Converts and sends the message.
* @param message The message,
* @throws Exception Any Exception.
*/
void send(Message<?> message) throws Exception;
void send(Message<?> message);
/**
* Uses the deserializer to obtain the message payload
* from the connection's input stream.
* @return The payload.
* @throws Exception Any Exception.
*/
@Nullable
Object getPayload() throws Exception;
Object getPayload();
/**
* @return the host name

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.
@@ -60,7 +60,7 @@ public abstract class TcpConnectionInterceptorSupport extends TcpConnectionSuppo
}
@Override
public Object getPayload() throws Exception {
public Object getPayload() {
return this.theConnection.getPayload();
}
@@ -165,7 +165,7 @@ public abstract class TcpConnectionInterceptorSupport extends TcpConnectionSuppo
}
@Override
public void send(Message<?> message) throws Exception {
public void send(Message<?> message) {
this.theConnection.send(message);
}

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.
@@ -17,8 +17,8 @@
package org.springframework.integration.ip.tcp.connection;
import java.io.IOException;
import java.io.UncheckedIOException;
import java.net.Socket;
import java.net.SocketException;
import org.springframework.util.Assert;
@@ -45,16 +45,21 @@ public class TcpNetClientConnectionFactory extends
}
@Override
protected TcpConnectionSupport buildNewConnection() throws IOException, SocketException, Exception {
Socket socket = createSocket(this.getHost(), this.getPort());
setSocketAttributes(socket);
TcpConnectionSupport connection = this.tcpNetConnectionSupport.createNewConnection(socket, false, isLookupHost(),
getApplicationEventPublisher(), getComponentName());
connection = wrapConnection(connection);
initializeConnection(connection, socket);
this.getTaskExecutor().execute(connection);
this.harvestClosedConnections();
return connection;
protected TcpConnectionSupport buildNewConnection() {
try {
Socket socket = createSocket(this.getHost(), this.getPort());
setSocketAttributes(socket);
TcpConnectionSupport connection = this.tcpNetConnectionSupport.createNewConnection(socket, false, isLookupHost(),
getApplicationEventPublisher(), getComponentName());
connection = wrapConnection(connection);
initializeConnection(connection, socket);
this.getTaskExecutor().execute(connection);
this.harvestClosedConnections();
return connection;
}
catch (IOException e) {
throw new UncheckedIOException(e);
}
}
/**

View File

@@ -20,9 +20,11 @@ import java.io.BufferedOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.io.UncheckedIOException;
import java.net.Socket;
import java.net.SocketException;
import java.net.SocketTimeoutException;
import java.util.function.Supplier;
import javax.net.ssl.SSLSession;
import javax.net.ssl.SSLSocket;
@@ -85,7 +87,7 @@ public class TcpNetConnection extends TcpConnectionSupport implements Scheduling
try {
this.socket.close();
}
catch (Exception e) {
catch (@SuppressWarnings("unused") Exception e) {
}
super.close();
}
@@ -97,23 +99,24 @@ public class TcpNetConnection extends TcpConnectionSupport implements Scheduling
@Override
@SuppressWarnings("unchecked")
public synchronized void send(Message<?> message) throws Exception {
if (this.socketOutputStream == null) {
int writeBufferSize = this.socket.getSendBufferSize();
this.socketOutputStream = new BufferedOutputStream(this.socket.getOutputStream(),
writeBufferSize > 0 ? writeBufferSize : 8192);
}
Object object = getMapper().fromMessage(message);
Assert.state(object != null, "Mapper mapped the message to 'null'.");
this.lastSend = System.currentTimeMillis();
public synchronized void send(Message<?> message) {
try {
if (this.socketOutputStream == null) {
int writeBufferSize = this.socket.getSendBufferSize();
this.socketOutputStream = new BufferedOutputStream(this.socket.getOutputStream(),
writeBufferSize > 0 ? writeBufferSize : 8192);
}
Object object = getMapper().fromMessage(message);
Assert.state(object != null, "Mapper mapped the message to 'null'.");
this.lastSend = System.currentTimeMillis();
((Serializer<Object>) getSerializer()).serialize(object, this.socketOutputStream);
this.socketOutputStream.flush();
}
catch (Exception e) {
publishConnectionExceptionEvent(new MessagingException(message, "Failed TCP serialization", e));
MessagingException mex = new MessagingException(message, "Send Failed", e);
publishConnectionExceptionEvent(mex);
closeConnection(true);
throw e;
throw mex;
}
if (logger.isDebugEnabled()) {
logger.debug(getConnectionId() + " Message sent " + message);
@@ -121,9 +124,14 @@ public class TcpNetConnection extends TcpConnectionSupport implements Scheduling
}
@Override
public Object getPayload() throws Exception {
return getDeserializer()
.deserialize(inputStream());
public Object getPayload() {
try {
return getDeserializer()
.deserialize(inputStream());
}
catch (IOException e) {
throw new UncheckedIOException(e);
}
}
@Override
@@ -137,7 +145,7 @@ public class TcpNetConnection extends TcpConnectionSupport implements Scheduling
try {
return inputStream();
}
catch (Exception e) {
catch (@SuppressWarnings("unused") Exception e) {
return null;
}
}
@@ -198,7 +206,7 @@ public class TcpNetConnection extends TcpConnectionSupport implements Scheduling
}
listener.onMessage(message);
}
catch (NoListenerException nle) { // could also be thrown by an interceptor
catch (@SuppressWarnings("unused") NoListenerException nle) { // could also be thrown by an interceptor
if (logger.isWarnEnabled()) {
logger.warn("Unexpected message - no endpoint registered with connection interceptor: "
+ getConnectionId()
@@ -213,12 +221,33 @@ public class TcpNetConnection extends TcpConnectionSupport implements Scheduling
}
}
protected boolean handleReadException(Exception e) {
protected boolean handleReadException(Exception exception) {
Exception e = exception instanceof UncheckedIOException ? (Exception) exception.getCause() : exception;
if (checkTimeout(e)) {
boolean readErrorOnClose = !isNoReadErrorOnClose();
closeConnection(true);
if (!(e instanceof SoftEndOfStreamException)) {
if (e instanceof SocketTimeoutException) {
if (logger.isDebugEnabled()) {
logger.debug("Closed socket after timeout:" + getConnectionId());
}
}
else {
logOtherExceptions(e, readErrorOnClose);
}
sendExceptionToListener(e);
}
return true;
}
return false;
}
/*
* For client connections, we have to wait for 2 timeouts if the last
* send was within the current timeout.
*/
private boolean checkTimeout(Exception e) {
boolean doClose = true;
/*
* For client connections, we have to wait for 2 timeouts if the last
* send was within the current timeout.
*/
if (!isServer() && e instanceof SocketTimeoutException) {
long now = System.currentTimeMillis();
try {
@@ -234,43 +263,26 @@ public class TcpNetConnection extends TcpConnectionSupport implements Scheduling
logger.error("Error accessing soTimeout", e1);
}
}
if (doClose) {
boolean noReadErrorOnClose = isNoReadErrorOnClose();
closeConnection(true);
if (!(e instanceof SoftEndOfStreamException)) {
if (e instanceof SocketTimeoutException) {
if (logger.isDebugEnabled()) {
logger.debug("Closed socket after timeout:" + getConnectionId());
}
}
else {
if (noReadErrorOnClose) {
if (logger.isTraceEnabled()) {
logger.trace("Read exception " +
getConnectionId(), e);
}
else if (logger.isDebugEnabled()) {
logger.debug("Read exception " +
getConnectionId() + " " +
e.getClass().getSimpleName() +
":" + (e.getCause() != null ? e.getCause() + ":" : "") + e.getMessage());
}
}
else if (logger.isTraceEnabled()) {
logger.error("Read exception " +
getConnectionId(), e);
}
else {
logger.error("Read exception " +
getConnectionId() + " " +
e.getClass().getSimpleName() +
":" + (e.getCause() != null ? e.getCause() + ":" : "") + e.getMessage());
}
}
sendExceptionToListener(e);
}
}
return doClose;
}
private void logOtherExceptions(Exception e, boolean readErrorOnClose) {
if (this.logger.isErrorEnabled()) {
String messagePrefix = "Read exception " + getConnectionId();
Supplier<String> summaryMessageSupplier = () -> messagePrefix + " " + e.getClass().getSimpleName() + ":"
+ (e.getCause() != null ? e.getCause() + ":" : "") + e.getMessage();
if (logger.isTraceEnabled()) {
logger.trace(messagePrefix, e);
}
else if (readErrorOnClose) {
logger.error(summaryMessageSupplier.get());
}
else {
if (logger.isDebugEnabled()) {
logger.debug(summaryMessageSupplier.get());
}
}
}
}
}

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.
@@ -41,11 +41,10 @@ public interface TcpNetConnectionSupport {
* @param connectionFactoryName the name of the connection factory creating this connection; used
* during event publishing, may be null, in which case "unknown" will be used.
* @return the TcpNetConnection
* @throws Exception Any exception.
*/
TcpNetConnection createNewConnection(Socket socket,
boolean server, boolean lookupHost,
ApplicationEventPublisher applicationEventPublisher,
String connectionFactoryName) throws Exception;
String connectionFactoryName);
}

View File

@@ -135,7 +135,7 @@ public class TcpNetServerConnectionFactory extends AbstractServerConnectionFacto
socket = this.serverSocket.accept();
}
}
catch (SocketTimeoutException ste) {
catch (@SuppressWarnings("unused") SocketTimeoutException ste) {
if (logger.isDebugEnabled()) {
logger.debug("Timed out on accept; continuing");
}
@@ -164,20 +164,20 @@ public class TcpNetServerConnectionFactory extends AbstractServerConnectionFacto
harvestClosedConnections();
connection.publishConnectionOpenEvent();
}
catch (Exception e) {
catch (RuntimeException e) {
this.logger.error("Failed to create and configure a TcpConnection for the new socket: "
+ socket.getInetAddress().getHostAddress() + ":" + socket.getPort(), e);
try {
socket.close();
}
catch (IOException e1) {
catch (@SuppressWarnings("unused") IOException e1) {
// empty
}
}
}
}
}
catch (Exception e) {
catch (IOException e) { // NOSONAR flow control via exceptions
// don't log an error if we had a good socket once and now it's closed
if (e instanceof SocketException && theServerSocket != null) {
logger.info("Server Socket closed");
@@ -224,7 +224,7 @@ public class TcpNetServerConnectionFactory extends AbstractServerConnectionFacto
try {
this.serverSocket.close();
}
catch (IOException e) {
catch (@SuppressWarnings("unused") IOException e) {
}
this.serverSocket = null;
super.stop();

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.
@@ -17,6 +17,7 @@
package org.springframework.integration.ip.tcp.connection;
import java.io.IOException;
import java.io.UncheckedIOException;
import java.net.InetSocketAddress;
import java.nio.channels.CancelledKeyException;
import java.nio.channels.ClosedChannelException;
@@ -63,43 +64,48 @@ public class TcpNioClientConnectionFactory extends
}
@Override
protected void checkActive() throws IOException {
protected void checkActive() {
super.checkActive();
int n = 0;
while (this.selector == null) {
try {
Thread.sleep(100);
}
catch (InterruptedException e) {
catch (@SuppressWarnings("unused") InterruptedException e) {
Thread.currentThread().interrupt();
}
if (n++ > 600) {
throw new IOException("Factory failed to start");
throw new UncheckedIOException(new IOException("Factory failed to start"));
}
}
}
@Override
protected TcpConnectionSupport buildNewConnection() throws Exception {
SocketChannel socketChannel = SocketChannel.open(new InetSocketAddress(getHost(), getPort()));
setSocketAttributes(socketChannel.socket());
TcpNioConnection connection = this.tcpNioConnectionSupport.createNewConnection(
socketChannel, false, this.isLookupHost(), this.getApplicationEventPublisher(), getComponentName());
connection.setUsingDirectBuffers(this.usingDirectBuffers);
connection.setTaskExecutor(this.getTaskExecutor());
if (getSslHandshakeTimeout() != null && connection instanceof TcpNioSSLConnection) {
((TcpNioSSLConnection) connection).setHandshakeTimeout(getSslHandshakeTimeout());
protected TcpConnectionSupport buildNewConnection() {
try {
SocketChannel socketChannel = SocketChannel.open(new InetSocketAddress(getHost(), getPort()));
setSocketAttributes(socketChannel.socket());
TcpNioConnection connection = this.tcpNioConnectionSupport.createNewConnection(
socketChannel, false, this.isLookupHost(), this.getApplicationEventPublisher(), getComponentName());
connection.setUsingDirectBuffers(this.usingDirectBuffers);
connection.setTaskExecutor(this.getTaskExecutor());
if (getSslHandshakeTimeout() != null && connection instanceof TcpNioSSLConnection) {
((TcpNioSSLConnection) connection).setHandshakeTimeout(getSslHandshakeTimeout());
}
TcpConnectionSupport wrappedConnection = wrapConnection(connection);
initializeConnection(wrappedConnection, socketChannel.socket());
socketChannel.configureBlocking(false);
if (this.getSoTimeout() > 0) {
connection.setLastRead(System.currentTimeMillis());
}
this.channelMap.put(socketChannel, connection);
this.newChannels.add(socketChannel);
this.selector.wakeup();
return wrappedConnection;
}
TcpConnectionSupport wrappedConnection = wrapConnection(connection);
initializeConnection(wrappedConnection, socketChannel.socket());
socketChannel.configureBlocking(false);
if (this.getSoTimeout() > 0) {
connection.setLastRead(System.currentTimeMillis());
catch (IOException e) {
throw new UncheckedIOException(e);
}
this.channelMap.put(socketChannel, connection);
this.newChannels.add(socketChannel);
this.selector.wakeup();
return wrappedConnection;
}
/**
@@ -164,7 +170,7 @@ public class TcpNioClientConnectionFactory extends
}
selectionCount = this.selector.select(timeout);
}
catch (CancelledKeyException cke) {
catch (@SuppressWarnings("unused") CancelledKeyException cke) {
if (logger.isDebugEnabled()) {
logger.debug("CancelledKeyException during Selector.select()");
}
@@ -173,7 +179,7 @@ public class TcpNioClientConnectionFactory extends
try {
newChannel.register(this.selector, SelectionKey.OP_READ, this.channelMap.get(newChannel));
}
catch (ClosedChannelException cce) {
catch (@SuppressWarnings("unused") ClosedChannelException cce) {
if (logger.isDebugEnabled()) {
logger.debug("Channel closed before registering with selector for reading");
}

View File

@@ -20,6 +20,7 @@ import java.io.BufferedOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.io.UncheckedIOException;
import java.net.SocketTimeoutException;
import java.nio.ByteBuffer;
import java.nio.channels.ClosedChannelException;
@@ -57,6 +58,10 @@ import org.springframework.util.Assert;
*/
public class TcpNioConnection extends TcpConnectionSupport {
private static final String UNUSED = "unused";
private static final int SIXTY = 60;
private static final long DEFAULT_PIPE_TIMEOUT = 60000;
private static final byte[] EOF = new byte[0]; // EOF marker buffer
@@ -123,12 +128,12 @@ public class TcpNioConnection extends TcpConnectionSupport {
try {
this.channelInputStream.close();
}
catch (IOException e) {
catch (@SuppressWarnings(UNUSED) IOException e) {
}
try {
this.socketChannel.close();
}
catch (Exception e) {
catch (@SuppressWarnings(UNUSED) Exception e) {
}
super.close();
}
@@ -140,24 +145,25 @@ public class TcpNioConnection extends TcpConnectionSupport {
@Override
@SuppressWarnings("unchecked")
public void send(Message<?> message) throws Exception {
public void send(Message<?> message) {
synchronized (this.socketChannel) {
if (this.bufferedOutputStream == null) {
int writeBufferSize = this.socketChannel.socket().getSendBufferSize();
this.bufferedOutputStream = new BufferedOutputStream(getChannelOutputStream(),
writeBufferSize > 0 ? writeBufferSize : 8192);
}
Object object = getMapper().fromMessage(message);
Assert.state(object != null, "Mapper mapped the message to 'null'.");
this.lastSend = System.currentTimeMillis();
try {
if (this.bufferedOutputStream == null) {
int writeBufferSize = this.socketChannel.socket().getSendBufferSize();
this.bufferedOutputStream = new BufferedOutputStream(getChannelOutputStream(),
writeBufferSize > 0 ? writeBufferSize : 8192);
}
Object object = getMapper().fromMessage(message);
Assert.state(object != null, "Mapper mapped the message to 'null'.");
this.lastSend = System.currentTimeMillis();
((Serializer<Object>) getSerializer()).serialize(object, this.bufferedOutputStream);
this.bufferedOutputStream.flush();
}
catch (Exception e) {
publishConnectionExceptionEvent(new MessagingException(message, "Failed TCP serialization", e));
MessagingException mex = new MessagingException(message, "Send Failed", e);
publishConnectionExceptionEvent(mex);
closeConnection(true);
throw e;
throw mex;
}
if (logger.isDebugEnabled()) {
logger.debug(getConnectionId() + " Message sent " + message);
@@ -166,9 +172,14 @@ public class TcpNioConnection extends TcpConnectionSupport {
}
@Override
public Object getPayload() throws Exception {
return getDeserializer()
.deserialize(inputStream());
public Object getPayload() {
try {
return getDeserializer()
.deserialize(inputStream());
}
catch (IOException e) {
throw new UncheckedIOException(e);
}
}
@Override
@@ -240,7 +251,7 @@ public class TcpNioConnection extends TcpConnectionSupport {
try {
this.taskExecutor.execute2(this);
}
catch (RejectedExecutionException e) {
catch (@SuppressWarnings(UNUSED) RejectedExecutionException e) {
this.executionControl.decrementAndGet();
if (logger.isInfoEnabled()) {
logger.info(getConnectionId()
@@ -288,40 +299,35 @@ public class TcpNioConnection extends TcpConnectionSupport {
// Final check in case new data came in and the
// timing was such that we were the last assembler and
// a new one wasn't run
try {
if (dataAvailable()) {
synchronized (this.executionControl) {
if (this.executionControl.incrementAndGet() <= 1) {
// only continue if we don't already have another assembler running
this.executionControl.set(1);
moreDataAvailable = true;
if (dataAvailable()) {
synchronized (this.executionControl) {
if (this.executionControl.incrementAndGet() <= 1) {
// only continue if we don't already have another assembler running
this.executionControl.set(1);
moreDataAvailable = true;
}
else {
this.executionControl.decrementAndGet();
}
}
}
if (moreDataAvailable) {
if (logger.isTraceEnabled()) {
logger.trace(getConnectionId() + " Nio message assembler continuing...");
}
}
else {
if (logger.isTraceEnabled()) {
logger.trace(getConnectionId() + " Nio message assembler exiting... avail: "
+ this.channelInputStream.available());
else {
this.executionControl.decrementAndGet();
}
}
}
catch (IOException e) {
logger.error("Exception when checking for assembler", e);
if (moreDataAvailable) {
if (logger.isTraceEnabled()) {
logger.trace(getConnectionId() + " Nio message assembler continuing...");
}
}
else {
if (logger.isTraceEnabled()) {
logger.trace(getConnectionId() + " Nio message assembler exiting... avail: "
+ this.channelInputStream.available());
}
}
}
}
}
private boolean dataAvailable() throws IOException {
private boolean dataAvailable() {
if (logger.isTraceEnabled()) {
logger.trace(getConnectionId() + " checking data avail: " + this.channelInputStream.available() +
" pending: " + (this.writingToPipe));
@@ -343,7 +349,7 @@ public class TcpNioConnection extends TcpConnectionSupport {
}
if (this.channelInputStream.available() <= 0) {
try {
if (this.writingLatch.await(60, TimeUnit.SECONDS)) {
if (this.writingLatch.await(SIXTY, TimeUnit.SECONDS)) {
if (this.channelInputStream.available() <= 0) {
return null;
}
@@ -352,7 +358,7 @@ public class TcpNioConnection extends TcpConnectionSupport {
throw new IOException("Timed out waiting for IO");
}
}
catch (InterruptedException e) {
catch (@SuppressWarnings(UNUSED) InterruptedException e) {
Thread.currentThread().interrupt();
throw new IOException("Interrupted waiting for IO");
}
@@ -451,13 +457,13 @@ public class TcpNioConnection extends TcpConnectionSupport {
}
}
protected void sendToPipe(ByteBuffer rawBuffer) throws IOException {
Assert.notNull(rawBuffer, "rawBuffer cannot be null");
protected void sendToPipe(ByteBuffer rawBufferToSend) throws IOException {
Assert.notNull(rawBufferToSend, "rawBuffer cannot be null");
if (logger.isTraceEnabled()) {
logger.trace(getConnectionId() + " Sending " + rawBuffer.limit() + " to pipe");
logger.trace(getConnectionId() + " Sending " + rawBufferToSend.limit() + " to pipe");
}
this.channelInputStream.write(rawBuffer);
rawBuffer.clear();
this.channelInputStream.write(rawBufferToSend);
rawBufferToSend.clear();
}
private void checkForAssembler() {
@@ -496,7 +502,7 @@ public class TcpNioConnection extends TcpConnectionSupport {
try {
doRead();
}
catch (ClosedChannelException cce) {
catch (@SuppressWarnings(UNUSED) ClosedChannelException cce) {
if (logger.isDebugEnabled()) {
logger.debug(getConnectionId() + " Channel is closed");
}
@@ -593,12 +599,12 @@ public class TcpNioConnection extends TcpConnectionSupport {
}
@Override
public void close() throws IOException {
public void close() {
doClose();
}
@Override
public void flush() throws IOException {
public void flush() {
}
@Override
@@ -767,7 +773,7 @@ public class TcpNioConnection extends TcpConnectionSupport {
try {
this.buffers.offer(EOF, TcpNioConnection.this.pipeTimeout, TimeUnit.SECONDS);
}
catch (InterruptedException e) {
catch (@SuppressWarnings(UNUSED) InterruptedException e) {
Thread.currentThread().interrupt();
}
}

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.
@@ -42,11 +42,10 @@ public interface TcpNioConnectionSupport {
* @param connectionFactoryName the name of the connection factory creating this connection; used
* during event publishing, may be null, in which case "unknown" will be used.
* @return the TcpNioConnection
* @throws Exception Any exception.
*/
TcpNioConnection createNewConnection(SocketChannel socketChannel,
boolean server, boolean lookupHost,
ApplicationEventPublisher applicationEventPublisher,
String connectionFactoryName) throws Exception;
String connectionFactoryName);
}

View File

@@ -17,6 +17,7 @@
package org.springframework.integration.ip.tcp.connection;
import java.io.IOException;
import java.io.UncheckedIOException;
import java.nio.ByteBuffer;
import java.nio.channels.SocketChannel;
import java.util.concurrent.Semaphore;
@@ -82,7 +83,7 @@ public class TcpNioSSLConnection extends TcpNioConnection {
public TcpNioSSLConnection(SocketChannel socketChannel, boolean server, boolean lookupHost,
ApplicationEventPublisher applicationEventPublisher, @Nullable String connectionFactoryName,
SSLEngine sslEngine) throws Exception {
SSLEngine sslEngine) {
super(socketChannel, server, lookupHost, applicationEventPublisher, connectionFactoryName);
this.sslEngine = sslEngine;
@@ -252,14 +253,17 @@ public class TcpNioSSLConnection extends TcpNioConnection {
/**
* Initializes the SSLEngine and sets up the encryption/decryption buffers.
*
* @throws IOException Any IOException.
*/
public void init() throws IOException {
public void init() {
if (this.decoded == null) {
this.decoded = allocateEncryptionBuffer(2048);
this.encoded = allocateEncryptionBuffer(2048);
initilizeEngine();
try {
initilizeEngine();
}
catch (IOException e) {
throw new UncheckedIOException(e);
}
}
}

View File

@@ -17,6 +17,7 @@
package org.springframework.integration.ip.tcp.connection;
import java.io.IOException;
import java.io.UncheckedIOException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.Socket;
@@ -80,6 +81,7 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto
}
}
catch (IOException e) {
logger.error("Error getting port", e);
}
}
return port;
@@ -93,6 +95,7 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto
return this.serverChannel.getLocalAddress();
}
catch (IOException e) {
logger.error("Error getting local address", e);
}
}
return null;
@@ -161,10 +164,10 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto
* schedules a call to doRead which reads all available data. When the read
* is complete, the socket is again registered for read interest.
* @param server the ServerSocketChannel to select
* @param selector the Selector multiplexor
* @param selectorToSelect the Selector multiplexor
* @throws IOException
*/
private void doSelect(ServerSocketChannel server, final Selector selector) throws IOException {
private void doSelect(ServerSocketChannel server, final Selector selectorToSelect) throws IOException {
while (isActive()) {
int soTimeout = getSoTimeout();
int selectionCount = 0;
@@ -176,10 +179,10 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto
if (logger.isTraceEnabled()) {
logger.trace("Delayed reads: " + getDelayedReads().size() + " timeout " + timeout);
}
selectionCount = selector.select(timeout);
processNioSelections(selectionCount, selector, server, this.channelMap);
selectionCount = selectorToSelect.select(timeout);
processNioSelections(selectionCount, selectorToSelect, server, this.channelMap);
}
catch (CancelledKeyException cke) {
catch (@SuppressWarnings("unused") CancelledKeyException cke) {
logger.debug("CancelledKeyException during Selector.select()");
}
catch (ClosedSelectorException cse) {
@@ -193,47 +196,51 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto
}
/**
* @param selector The selector.
* @param selectorForNewSocket The selector.
* @param server The server socket channel.
* @param now The current time.
* @throws IOException Any IOException.
*/
@Override
protected void doAccept(final Selector selector, ServerSocketChannel server, long now) throws IOException {
protected void doAccept(final Selector selectorForNewSocket, ServerSocketChannel server, long now) {
logger.debug("New accept");
SocketChannel channel = server.accept();
if (isShuttingDown()) {
if (logger.isInfoEnabled()) {
logger.info("New connection from " + channel.socket().getInetAddress().getHostAddress()
+ ":" + channel.socket().getPort()
+ " rejected; the server is in the process of shutting down.");
}
channel.close();
}
else {
try {
channel.configureBlocking(false);
Socket socket = channel.socket();
setSocketAttributes(socket);
TcpNioConnection connection = createTcpNioConnection(channel);
if (connection == null) {
return;
try {
SocketChannel channel = server.accept();
if (isShuttingDown()) {
if (logger.isInfoEnabled()) {
logger.info("New connection from " + channel.socket().getInetAddress().getHostAddress()
+ ":" + channel.socket().getPort()
+ " rejected; the server is in the process of shutting down.");
}
connection.setTaskExecutor(getTaskExecutor());
connection.setLastRead(now);
if (getSslHandshakeTimeout() != null && connection instanceof TcpNioSSLConnection) {
((TcpNioSSLConnection) connection).setHandshakeTimeout(getSslHandshakeTimeout());
}
this.channelMap.put(channel, connection);
channel.register(selector, SelectionKey.OP_READ, connection);
connection.publishConnectionOpenEvent();
}
catch (Exception e) {
logger.error("Exception accepting new connection from "
+ channel.socket().getInetAddress().getHostAddress()
+ ":" + channel.socket().getPort(), e);
channel.close();
}
else {
try {
channel.configureBlocking(false);
Socket socket = channel.socket();
setSocketAttributes(socket);
TcpNioConnection connection = createTcpNioConnection(channel);
if (connection == null) {
return;
}
connection.setTaskExecutor(getTaskExecutor());
connection.setLastRead(now);
if (getSslHandshakeTimeout() != null && connection instanceof TcpNioSSLConnection) {
((TcpNioSSLConnection) connection).setHandshakeTimeout(getSslHandshakeTimeout());
}
this.channelMap.put(channel, connection);
channel.register(selectorForNewSocket, SelectionKey.OP_READ, connection);
connection.publishConnectionOpenEvent();
}
catch (IOException e) {
logger.error("Exception accepting new connection from "
+ channel.socket().getInetAddress().getHostAddress()
+ ":" + channel.socket().getPort(), e);
channel.close();
}
}
}
catch (IOException e) {
throw new UncheckedIOException(e);
}
}

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.
@@ -61,7 +61,7 @@ public class ThreadAffinityClientConnectionFactory extends AbstractClientConnect
}
@Override
public TcpConnectionSupport getConnection() throws Exception {
public TcpConnectionSupport getConnection() throws InterruptedException {
TcpThreadConnection connection = this.connections.get();
if (connection == null || !connection.isOpen()) {
TcpConnectionSupport delegate = this.connectionFactory.getConnection();
@@ -380,7 +380,7 @@ public class ThreadAffinityClientConnectionFactory extends AbstractClientConnect
}
@Override
public void send(Message<?> message) throws Exception {
public void send(Message<?> message) {
this.connection.send(message);
}
@@ -390,7 +390,7 @@ public class ThreadAffinityClientConnectionFactory extends AbstractClientConnect
}
@Override
public Object getPayload() throws Exception {
public Object getPayload() {
return this.connection.getPayload();
}

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.
@@ -39,6 +39,7 @@ import org.springframework.lang.Nullable;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHandlingException;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.MessagingException;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
@@ -133,7 +134,7 @@ public class DatagramPacketMessageMapper implements InboundMessageMapper<Datagra
* Raw byte[] from message, possibly with a length field up front.
*/
@Override
public DatagramPacket fromMessage(Message<?> message) throws Exception {
public DatagramPacket fromMessage(Message<?> message) {
if (this.acknowledge) {
return fromMessageWithAck(message);
}
@@ -152,23 +153,28 @@ public class DatagramPacketMessageMapper implements InboundMessageMapper<Datagra
/**
* Prefix raw byte[] from message with 'acknowledge to' and 'message id' "headers".
*/
private DatagramPacket fromMessageWithAck(Message<?> message) throws Exception {
private DatagramPacket fromMessageWithAck(Message<?> message) {
Assert.state(StringUtils.hasText(this.ackAddress), "'ackAddress' must not be empty");
byte[] bytes = getPayloadAsBytes(message);
ByteBuffer buffer = ByteBuffer.allocate(100 + bytes.length);
if (this.lengthCheck) {
buffer.putInt(0); // placeholder for length
}
buffer.put(IpHeaders.ACK_ADDRESS.getBytes(this.charset));
buffer.put((byte) '=');
buffer.put(this.ackAddress.getBytes(this.charset));
buffer.put((byte) ';');
UUID id = message.getHeaders().getId();
if (id != null) {
buffer.put(MessageHeaders.ID.getBytes(this.charset));
try {
buffer.put(IpHeaders.ACK_ADDRESS.getBytes(this.charset));
buffer.put((byte) '=');
buffer.put(id.toString().getBytes(this.charset));
buffer.put(this.ackAddress.getBytes(this.charset));
buffer.put((byte) ';');
UUID id = message.getHeaders().getId();
if (id != null) {
buffer.put(MessageHeaders.ID.getBytes(this.charset));
buffer.put((byte) '=');
buffer.put(id.toString().getBytes(this.charset));
buffer.put((byte) ';');
}
}
catch (UnsupportedEncodingException e) {
throw new MessagingException(message, "Failed to get headers", e);
}
int headersLength = buffer.position() - 4;
buffer.put(bytes);
@@ -203,13 +209,13 @@ public class DatagramPacketMessageMapper implements InboundMessageMapper<Datagra
@Override
@Nullable
public Message<byte[]> toMessage(DatagramPacket object) throws Exception {
public Message<byte[]> toMessage(DatagramPacket object) {
return toMessage(object, null);
}
@Override
@Nullable
public Message<byte[]> toMessage(DatagramPacket packet, @Nullable Map<String, Object> headers) throws Exception {
public Message<byte[]> toMessage(DatagramPacket packet, @Nullable Map<String, Object> headers) {
int offset = packet.getOffset();
int length = packet.getLength();
byte[] payload;

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.
@@ -21,6 +21,7 @@ import java.net.DatagramSocket;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.MulticastSocket;
import java.net.URISyntaxException;
import org.springframework.expression.Expression;
import org.springframework.messaging.Message;
@@ -190,7 +191,7 @@ public class MulticastSendingMessageHandler extends UnicastSendingMessageHandler
}
@Override
protected void convertAndSend(Message<?> message) throws Exception {
protected void convertAndSend(Message<?> message) throws IOException, URISyntaxException {
super.convertAndSend(message);
if (logger.isDebugEnabled()) {
logger.debug("Sent packet to " + this.multicastSocket.getInterface());

View File

@@ -217,7 +217,7 @@ public class UnicastReceivingChannelAdapter extends AbstractInternetProtocolRece
}
}
protected DatagramPacket receive() throws Exception {
protected DatagramPacket receive() throws IOException {
final byte[] buffer = new byte[this.getReceiveBufferSize()];
DatagramPacket packet = new DatagramPacket(buffer, buffer.length);
getSocket().receive(packet);

View File

@@ -24,6 +24,7 @@ import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.net.SocketException;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
@@ -278,7 +279,7 @@ public class UnicastSendingMessageHandler extends
+ this.ackTimeout + " millis");
}
}
catch (InterruptedException e) {
catch (@SuppressWarnings("unused") InterruptedException e) {
Thread.currentThread().interrupt();
}
}
@@ -312,7 +313,7 @@ public class UnicastSendingMessageHandler extends
try {
this.ackLatch.await(10000, TimeUnit.MILLISECONDS);
}
catch (InterruptedException e) {
catch (@SuppressWarnings("unused") InterruptedException e) {
Thread.currentThread().interrupt();
}
}
@@ -320,7 +321,7 @@ public class UnicastSendingMessageHandler extends
}
}
protected void convertAndSend(Message<?> message) throws Exception {
protected void convertAndSend(Message<?> message) throws IOException, URISyntaxException {
DatagramSocket datagramSocket;
if (this.socketExpression != null) {
datagramSocket = this.socketExpression.getValue(this.evaluationContext, message, DatagramSocket.class);

View File

@@ -139,10 +139,12 @@ public final class TestingUtilities {
* of connections.
* @param factory The factory.
* @param n The required number of connections.
* @throws Exception IllegalStateException if the count does not match.
* @throws InterruptedException if interrupted.
* @throws IllegalStateException if the count does not match.
*/
public static void waitUntilFactoryHasThisNumberOfConnections(AbstractConnectionFactory factory, int n)
throws Exception {
throws InterruptedException {
int timer = 0;
while (timer < 10000) {
if (factory.getOpenConnectionIds().size() == n) {

View File

@@ -653,7 +653,7 @@ public class ParserUnitTests {
public static class FooAdvice extends AbstractRequestHandlerAdvice {
@Override
protected Object doInvoke(ExecutionCallback callback, Object target, Message<?> message) throws Exception {
protected Object doInvoke(ExecutionCallback callback, Object target, Message<?> message) {
adviceCalled.countDown();
return null;
}

View File

@@ -17,6 +17,7 @@
package org.springframework.integration.ip.tcp;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.catchThrowable;
import static org.assertj.core.api.Assertions.fail;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
@@ -27,6 +28,7 @@ import java.io.EOFException;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.io.UncheckedIOException;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketTimeoutException;
@@ -74,6 +76,8 @@ import org.springframework.integration.test.rule.Log4j2LevelAdjuster;
import org.springframework.integration.test.support.LongRunningIntegrationTest;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHandlingException;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.support.ErrorMessage;
import org.springframework.messaging.support.GenericMessage;
@@ -485,7 +489,8 @@ public class TcpOutboundGatewayTests {
AbstractClientConnectionFactory factory1 = mock(AbstractClientConnectionFactory.class);
TcpConnectionSupport mockConn1 = makeMockConnection();
when(factory1.getConnection()).thenReturn(mockConn1);
doThrow(new IOException("fail")).when(mockConn1).send(Mockito.any(Message.class));
doThrow(new UncheckedIOException(new IOException("fail")))
.when(mockConn1).send(Mockito.any(Message.class));
AbstractClientConnectionFactory factory2 = new TcpNetClientConnectionFactory("localhost",
serverSocket.get().getLocalPort());
@@ -567,7 +572,8 @@ public class TcpOutboundGatewayTests {
TcpConnectionSupport mockConn1 = makeMockConnection();
when(factory1.getConnection()).thenReturn(mockConn1);
when(factory1.isSingleUse()).thenReturn(true);
doThrow(new IOException("fail")).when(mockConn1).send(Mockito.any(Message.class));
doThrow(new UncheckedIOException(new IOException("fail")))
.when(mockConn1).send(Mockito.any(Message.class));
CachingClientConnectionFactory cachingFactory1 = new CachingClientConnectionFactory(factory1, 1);
AbstractClientConnectionFactory factory2 = new TcpNetClientConnectionFactory("localhost",
@@ -727,13 +733,10 @@ public class TcpOutboundGatewayTests {
gateway.setBeanFactory(mock(BeanFactory.class));
gateway.afterPropertiesSet();
gateway.start();
try {
gateway.handleMessage(MessageBuilder.withPayload("Test").build());
fail("expected failure");
}
catch (Exception e) {
assertThat(e.getCause().getCause()).isInstanceOf(EOFException.class);
}
Throwable thrown = catchThrowable(() -> gateway.handleMessage(MessageBuilder.withPayload("Test").build()));
assertThat(thrown).isInstanceOf(MessageHandlingException.class);
assertThat(thrown.getCause()).isInstanceOf(MessagingException.class);
assertThat(thrown.getCause().getCause()).isInstanceOf(EOFException.class);
assertThat(TestUtils.getPropertyValue(gateway, "pendingReplies", Map.class).size()).isEqualTo(0);
Message<?> reply = replyChannel.receive(0);
assertThat(reply).isNull();
@@ -837,13 +840,10 @@ public class TcpOutboundGatewayTests {
gateway.setBeanFactory(mock(BeanFactory.class));
gateway.afterPropertiesSet();
gateway.start();
try {
gateway.handleMessage(MessageBuilder.withPayload("Test").build());
fail("expected failure");
}
catch (Exception e) {
assertThat(e.getCause().getCause()).isInstanceOf(SocketTimeoutException.class);
}
Throwable thrown = catchThrowable(() -> gateway.handleMessage(MessageBuilder.withPayload("Test").build()));
assertThat(thrown).isInstanceOf(MessageHandlingException.class);
assertThat(thrown.getCause()).isInstanceOf(MessagingException.class);
assertThat(thrown.getCause().getCause()).isInstanceOf(SocketTimeoutException.class);
assertThat(TestUtils.getPropertyValue(gateway, "pendingReplies", Map.class).size()).isEqualTo(0);
Message<?> reply = replyChannel.receive(0);
assertThat(reply).isNull();

View File

@@ -17,7 +17,7 @@
package org.springframework.integration.ip.tcp.connection;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.fail;
import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyInt;
import static org.mockito.ArgumentMatchers.anyString;
@@ -32,6 +32,7 @@ import static org.mockito.Mockito.when;
import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.OutputStream;
import java.io.UncheckedIOException;
import java.net.Socket;
import java.nio.ByteBuffer;
import java.nio.channels.SocketChannel;
@@ -73,6 +74,7 @@ import org.springframework.integration.test.util.TestUtils;
import org.springframework.integration.util.PoolItemNotAvailableException;
import org.springframework.integration.util.SimplePool;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.SubscribableChannel;
import org.springframework.messaging.support.ErrorMessage;
@@ -363,13 +365,8 @@ public class CachingClientConnectionFactoryTests {
private void doTestCloseOnSendError(TcpConnection conn1, TcpConnection conn2,
CachingClientConnectionFactory cccf) throws Exception {
TcpConnection cached1 = cccf.getConnection();
try {
cached1.send(new GenericMessage<String>("foo"));
fail("Expected IOException");
}
catch (IOException e) {
assertThat(e.getMessage()).isEqualTo("Foo");
}
assertThatExceptionOfType(MessagingException.class)
.isThrownBy(() -> cached1.send(new GenericMessage<String>("foo")));
// Before INT-3163 this failed with a timeout - connection not returned to pool after failure on send()
TcpConnection cached2 = cccf.getConnection();
assertThat(cached1.getConnectionId().contains(conn1.getConnectionId())).isTrue();
@@ -550,7 +547,7 @@ public class CachingClientConnectionFactoryTests {
when(factory2.getConnection()).thenReturn(mockConn2);
when(factory1.isActive()).thenReturn(true);
when(factory2.isActive()).thenReturn(true);
doThrow(new IOException("fail")).when(mockConn1).send(Mockito.any(Message.class));
doThrow(new UncheckedIOException(new IOException("fail"))).when(mockConn1).send(Mockito.any(Message.class));
doAnswer(invocation -> null).when(mockConn2).send(Mockito.any(Message.class));
FailoverClientConnectionFactory failoverFactory = new FailoverClientConnectionFactory(factories);
failoverFactory.start();

View File

@@ -27,6 +27,7 @@ import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.verify;
import java.io.OutputStream;
import java.io.UncheckedIOException;
import java.net.BindException;
import java.net.ServerSocket;
import java.net.Socket;
@@ -258,13 +259,7 @@ public class ConnectionEventTests {
}
private void testServerExceptionGuts(AbstractServerConnectionFactory factory) throws Exception {
ServerSocket ss = null;
try {
ss = ServerSocketFactory.getDefault().createServerSocket(0);
}
catch (Exception e) {
fail("Failed to get a server socket");
}
ServerSocket ss = ServerSocketFactory.getDefault().createServerSocket(0);
factory.setPort(ss.getLocalPort());
final AtomicReference<TcpConnectionServerExceptionEvent> theEvent =
new AtomicReference<TcpConnectionServerExceptionEvent>();
@@ -315,8 +310,8 @@ public class ConnectionEventTests {
}
@Override
protected TcpConnectionSupport buildNewConnection() throws Exception {
throw new UnknownHostException("Mocking for test ");
protected TcpConnectionSupport buildNewConnection() {
throw new UncheckedIOException(new UnknownHostException("Mocking for test "));
}
};
@@ -340,7 +335,7 @@ public class ConnectionEventTests {
fail("expected exception");
}
catch (Exception e) {
assertThat(e).isInstanceOf(UnknownHostException.class);
assertThat(e.getCause()).isInstanceOf(UnknownHostException.class);
TcpConnectionFailedEvent event = (TcpConnectionFailedEvent) failEvent.get();
assertThat(event.getCause()).isSameAs(e);
}

View File

@@ -171,7 +171,7 @@ public class ConnectionTimeoutTests {
connection.send(MessageBuilder.withPayload("foo").build());
Thread.sleep(1400);
assertThat(connection.isOpen()).isTrue();
assertThat(clientCloseLatch.await(2000, TimeUnit.SECONDS)).isTrue();
assertThat(clientCloseLatch.await(5, TimeUnit.SECONDS)).isTrue();
assertThat(reply.get()).isNull();
assertThat(connection.isOpen()).isFalse();
server.stop();

View File

@@ -17,7 +17,7 @@
package org.springframework.integration.ip.tcp.connection;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.fail;
import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
@@ -25,6 +25,7 @@ import static org.mockito.Mockito.times;
import static org.mockito.Mockito.when;
import java.io.IOException;
import java.io.UncheckedIOException;
import java.net.Socket;
import java.nio.channels.SocketChannel;
import java.util.ArrayList;
@@ -100,7 +101,8 @@ public class FailoverClientConnectionFactoryTests {
when(factory2.getConnection()).thenReturn(conn2);
when(factory1.isActive()).thenReturn(true);
when(factory2.isActive()).thenReturn(true);
doThrow(new IOException("fail")).when(conn1).send(Mockito.any(Message.class));
doThrow(new UncheckedIOException(new IOException("fail")))
.when(conn1).send(Mockito.any(Message.class));
doAnswer(invocation -> null).when(conn2).send(Mockito.any(Message.class));
FailoverClientConnectionFactory failoverFactory = new FailoverClientConnectionFactory(factories);
failoverFactory.start();
@@ -109,7 +111,7 @@ public class FailoverClientConnectionFactoryTests {
Mockito.verify(conn2).send(message);
}
@Test(expected = IOException.class)
@Test(expected = UncheckedIOException.class)
public void testFailoverAllDead() throws Exception {
AbstractClientConnectionFactory factory1 = mock(AbstractClientConnectionFactory.class);
AbstractClientConnectionFactory factory2 = mock(AbstractClientConnectionFactory.class);
@@ -122,8 +124,10 @@ public class FailoverClientConnectionFactoryTests {
when(factory2.getConnection()).thenReturn(conn2);
when(factory1.isActive()).thenReturn(true);
when(factory2.isActive()).thenReturn(true);
doThrow(new IOException("fail")).when(conn1).send(Mockito.any(Message.class));
doThrow(new IOException("fail")).when(conn2).send(Mockito.any(Message.class));
doThrow(new UncheckedIOException(new IOException("fail")))
.when(conn1).send(Mockito.any(Message.class));
doThrow(new UncheckedIOException(new IOException("fail")))
.when(conn2).send(Mockito.any(Message.class));
FailoverClientConnectionFactory failoverFactory = new FailoverClientConnectionFactory(factories);
failoverFactory.start();
GenericMessage<String> message = new GenericMessage<String>("foo");
@@ -148,11 +152,12 @@ public class FailoverClientConnectionFactoryTests {
doAnswer(invocation -> {
if (!failedOnce.get()) {
failedOnce.set(true);
throw new IOException("fail");
throw new UncheckedIOException(new IOException("fail"));
}
return null;
}).when(conn1).send(Mockito.any(Message.class));
doThrow(new IOException("fail")).when(conn2).send(Mockito.any(Message.class));
doThrow(new UncheckedIOException(new IOException("fail")))
.when(conn2).send(Mockito.any(Message.class));
FailoverClientConnectionFactory failoverFactory = new FailoverClientConnectionFactory(factories);
failoverFactory.start();
GenericMessage<String> message = new GenericMessage<String>("foo");
@@ -161,15 +166,15 @@ public class FailoverClientConnectionFactoryTests {
Mockito.verify(conn1, times(2)).send(message);
}
@Test(expected = IOException.class)
@Test(expected = UncheckedIOException.class)
public void testFailoverConnectNone() throws Exception {
AbstractClientConnectionFactory factory1 = mock(AbstractClientConnectionFactory.class);
AbstractClientConnectionFactory factory2 = mock(AbstractClientConnectionFactory.class);
List<AbstractClientConnectionFactory> factories = new ArrayList<AbstractClientConnectionFactory>();
factories.add(factory1);
factories.add(factory2);
when(factory1.getConnection()).thenThrow(new IOException("fail"));
when(factory2.getConnection()).thenThrow(new IOException("fail"));
when(factory1.getConnection()).thenThrow(new UncheckedIOException(new IOException("fail")));
when(factory2.getConnection()).thenThrow(new UncheckedIOException(new IOException("fail")));
when(factory1.isActive()).thenReturn(true);
when(factory2.isActive()).thenReturn(true);
FailoverClientConnectionFactory failoverFactory = new FailoverClientConnectionFactory(factories);
@@ -187,8 +192,11 @@ public class FailoverClientConnectionFactoryTests {
factories.add(factory2);
TcpConnectionSupport conn1 = makeMockConnection();
doAnswer(invocation -> null).when(conn1).send(Mockito.any(Message.class));
when(factory1.getConnection()).thenThrow(new IOException("fail")).thenReturn(conn1);
when(factory2.getConnection()).thenThrow(new IOException("fail"));
when(factory1.getConnection())
.thenThrow(new UncheckedIOException(new IOException("fail")))
.thenReturn(conn1);
when(factory2.getConnection())
.thenThrow(new UncheckedIOException(new IOException("fail")));
when(factory1.isActive()).thenReturn(true);
when(factory2.isActive()).thenReturn(true);
FailoverClientConnectionFactory failoverFactory = new FailoverClientConnectionFactory(factories);
@@ -214,20 +222,17 @@ public class FailoverClientConnectionFactoryTests {
final AtomicInteger failCount = new AtomicInteger();
doAnswer(invocation -> {
if (failCount.incrementAndGet() < 3) {
throw new IOException("fail");
throw new UncheckedIOException(new IOException("fail"));
}
return null;
}).when(conn1).send(Mockito.any(Message.class));
doThrow(new IOException("fail")).when(conn2).send(Mockito.any(Message.class));
doThrow(new UncheckedIOException(new IOException("fail")))
.when(conn2).send(Mockito.any(Message.class));
FailoverClientConnectionFactory failoverFactory = new FailoverClientConnectionFactory(factories);
failoverFactory.start();
GenericMessage<String> message = new GenericMessage<String>("foo");
try {
failoverFactory.getConnection().send(message);
fail("ExpectedFailure");
}
catch (IOException e) {
}
assertThatExceptionOfType(UncheckedIOException.class)
.isThrownBy(() -> failoverFactory.getConnection().send(message));
failoverFactory.getConnection().send(message);
Mockito.verify(conn2).send(message);
Mockito.verify(conn1, times(3)).send(message);

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.
@@ -19,9 +19,6 @@ package org.springframework.integration.ip.tcp.connection;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
@@ -34,8 +31,6 @@ import org.springframework.messaging.MessagingException;
*/
public class HelloWorldInterceptor extends TcpConnectionInterceptorSupport {
Log logger = LogFactory.getLog(this.getClass());
private volatile boolean negotiated;
private final Semaphore negotiationSemaphore = new Semaphore(0);
@@ -109,14 +104,19 @@ public class HelloWorldInterceptor extends TcpConnectionInterceptorSupport {
}
@Override
public void send(Message<?> message) throws Exception {
public void send(Message<?> message) {
this.pendingSend = true;
try {
if (!this.negotiated) {
if (!this.isServer()) {
logger.debug(this.toString() + " Sending " + hello);
super.send(MessageBuilder.withPayload(hello).build());
this.negotiationSemaphore.tryAcquire(this.timeout, TimeUnit.MILLISECONDS);
try {
this.negotiationSemaphore.tryAcquire(this.timeout, TimeUnit.MILLISECONDS);
}
catch (@SuppressWarnings("unused") InterruptedException e) {
Thread.currentThread().interrupt();
}
if (!this.negotiated) {
throw new MessagingException("Negotiation error");
}

View File

@@ -17,18 +17,15 @@
package org.springframework.integration.ip.tcp.connection;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.fail;
import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import java.io.IOException;
import java.io.InputStream;
import java.net.InetAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketException;
import java.nio.channels.ClosedChannelException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
@@ -41,7 +38,6 @@ import java.util.concurrent.atomic.AtomicReference;
import javax.net.ServerSocketFactory;
import javax.net.SocketFactory;
import javax.net.ssl.SSLEngine;
import javax.net.ssl.SSLException;
import javax.net.ssl.SSLServerSocket;
import org.junit.Test;
@@ -51,6 +47,7 @@ import org.springframework.integration.ip.tcp.serializer.ByteArrayCrLfSerializer
import org.springframework.integration.ip.util.TestingUtilities;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.support.GenericMessage;
/**
@@ -366,13 +363,8 @@ public class SocketSupportTests {
@Test
public void testNetClientAndServerSSLDifferentContexts() throws Exception {
testNetClientAndServerSSLDifferentContexts(false);
try {
testNetClientAndServerSSLDifferentContexts(true);
fail("expected Exception");
}
catch (SSLException | SocketException e) {
// NOSONAR
}
assertThatExceptionOfType(MessagingException.class)
.isThrownBy(() -> testNetClientAndServerSSLDifferentContexts(true));
}
private void testNetClientAndServerSSLDifferentContexts(boolean badClient) throws Exception {
@@ -478,19 +470,10 @@ public class SocketSupportTests {
@Test
public void testNioClientAndServerSSLDifferentContexts() throws Exception {
testNioClientAndServerSSLDifferentContexts(false);
try {
testNioClientAndServerSSLDifferentContexts(true);
fail("expected Exception");
}
catch (IOException e) {
if (!(e instanceof ClosedChannelException)) {
assertThat(e.getMessage())
.satisfiesAnyOf(
s -> assertThat(s).contains("Socket closed during SSL Handshake"),
s -> assertThat(s).contains("Broken pipe"),
s -> assertThat(s).contains("Connection reset by peer"));
}
}
assertThatExceptionOfType(MessagingException.class)
.isThrownBy(() -> testNioClientAndServerSSLDifferentContexts(true))
.withMessageMatching(".*(Socket closed during SSL Handshake|Broken pipe"
+ "|Connection reset by peer|AsynchronousCloseException).*");
}
private void testNioClientAndServerSSLDifferentContexts(boolean badClient) throws Exception {

View File

@@ -146,7 +146,7 @@ public class TcpMessageMapperTests {
}
@Test(expected = IllegalArgumentException.class)
public void testToMessageWithBadContentType() throws Exception {
public void testToMessageWithBadContentType() {
TcpMessageMapper mapper = new TcpMessageMapper();
mapper.setAddContentTypeHeader(true);
try {
@@ -169,7 +169,7 @@ public class TcpMessageMapperTests {
}
@Override
public void send(Message<?> message) throws Exception {
public void send(Message<?> message) {
}
@Override
@@ -183,7 +183,7 @@ public class TcpMessageMapperTests {
}
@Override
public Object getPayload() throws Exception {
public Object getPayload() {
return TEST_PAYLOAD.getBytes();
}
@@ -252,7 +252,7 @@ public class TcpMessageMapperTests {
}
@Override
public void send(Message<?> message) throws Exception {
public void send(Message<?> message) {
}
@Override
@@ -266,7 +266,7 @@ public class TcpMessageMapperTests {
}
@Override
public Object getPayload() throws Exception {
public Object getPayload() {
return TEST_PAYLOAD.getBytes();
}

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2018 the original author or authors.
* Copyright 2018-2019 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -52,13 +52,13 @@ public class TcpNetConnectionSupportTests {
server.setTcpNetConnectionSupport(new DefaultTcpNetConnectionSupport() {
@Override
public TcpNetConnection createNewConnection(Socket socket, boolean server, boolean lookupHost,
ApplicationEventPublisher applicationEventPublisher, String connectionFactoryName)
throws Exception {
public TcpNetConnection createNewConnection(Socket socket, boolean isServer, boolean lookupHost,
ApplicationEventPublisher applicationEventPublisher, String connectionFactoryName) {
if (firstTime.getAndSet(false)) {
throw new RuntimeException("intended");
}
return super.createNewConnection(socket, server, lookupHost, applicationEventPublisher, connectionFactoryName);
return super.createNewConnection(socket, isServer, lookupHost, applicationEventPublisher, connectionFactoryName);
}
});

View File

@@ -17,6 +17,7 @@
package org.springframework.integration.ip.tcp.connection;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.BDDMockito.given;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
@@ -63,6 +64,7 @@ public class TcpNetConnectionTests {
connection.setDeserializer(new ByteArrayStxEtxSerializer());
final AtomicReference<Object> log = new AtomicReference<Object>();
Log logger = mock(Log.class);
given(logger.isErrorEnabled()).willReturn(true);
doAnswer(invocation -> {
log.set(invocation.getArguments()[0]);
return null;

View File

@@ -83,6 +83,7 @@ import org.springframework.integration.test.rule.Log4j2LevelAdjuster;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.integration.util.CompositeExecutor;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.support.ErrorMessage;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.util.ReflectionUtils;
@@ -143,8 +144,8 @@ public class TcpNioConnectionTests {
TcpConnection connection = factory.getConnection();
connection.send(MessageBuilder.withPayload(new byte[1000000]).build());
}
catch (Exception e) {
assertThat(e instanceof SocketTimeoutException)
catch (MessagingException e) {
assertThat(e.getCause() instanceof SocketTimeoutException)
.as("Expected SocketTimeoutException, got " + e.getClass().getSimpleName() +
":" + e.getMessage()).isTrue();
}

Some files were not shown because too many files have changed in this diff Show More