diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/filters/AbstractPersistentAcceptOnceFileListFilter.java b/spring-integration-file/src/main/java/org/springframework/integration/file/filters/AbstractPersistentAcceptOnceFileListFilter.java index 587d857de5..2fa5c7c6bf 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/filters/AbstractPersistentAcceptOnceFileListFilter.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/filters/AbstractPersistentAcceptOnceFileListFilter.java @@ -94,6 +94,7 @@ public abstract class AbstractPersistentAcceptOnceFileListFilter extends Abst */ @Override public void rollback(F file, List files) { + // If file must be removed all subsequent files should be removed as well boolean rollingBack = false; for (F fileToRollback : files) { if (fileToRollback.equals(file)) { diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/RemoteFileTemplate.java b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/RemoteFileTemplate.java index 58edb22d4d..46d1a1af3d 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/RemoteFileTemplate.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/RemoteFileTemplate.java @@ -572,7 +572,7 @@ public class RemoteFileTemplate implements RemoteFileOperations, Initializ return directoryPath; } - private final class StreamHolder { + private static final class StreamHolder { private final InputStream stream; diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/synchronizer/AbstractInboundFileSynchronizingMessageSource.java b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/synchronizer/AbstractInboundFileSynchronizingMessageSource.java index c1687bb663..7133da1136 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/synchronizer/AbstractInboundFileSynchronizingMessageSource.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/synchronizer/AbstractInboundFileSynchronizingMessageSource.java @@ -23,6 +23,7 @@ import java.util.Arrays; import java.util.Comparator; import java.util.regex.Pattern; +import org.springframework.beans.factory.BeanInitializationException; import org.springframework.context.Lifecycle; import org.springframework.integration.endpoint.AbstractMessageSource; import org.springframework.integration.file.FileReadingMessageSource; @@ -31,7 +32,6 @@ import org.springframework.integration.file.filters.CompositeFileListFilter; import org.springframework.integration.file.filters.FileListFilter; import org.springframework.integration.file.filters.RegexPatternFileListFilter; import org.springframework.messaging.Message; -import org.springframework.messaging.MessagingException; import org.springframework.util.Assert; /** @@ -153,8 +153,8 @@ public abstract class AbstractInboundFileSynchronizingMessageSource throw e; } catch (Exception e) { - throw new MessagingException( - "Failure during initialization of MessageSource for: " + this.getClass(), e); + throw new BeanInitializationException("Failure during initialization of MessageSource for: " + + this.getClass(), e); } } diff --git a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/FtpParserInboundTests.java b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/FtpParserInboundTests.java index 0fdb8c401d..1afd0b592e 100644 --- a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/FtpParserInboundTests.java +++ b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/FtpParserInboundTests.java @@ -31,8 +31,8 @@ import org.junit.Test; import org.springframework.beans.BeansException; import org.springframework.beans.factory.BeanCreationException; +import org.springframework.beans.factory.BeanInitializationException; import org.springframework.context.support.ClassPathXmlApplicationContext; -import org.springframework.messaging.MessagingException; /** * @author Oleg Zhurakousky @@ -63,7 +63,7 @@ public class FtpParserInboundTests { catch (BeansException e) { assertThat(e, Matchers.instanceOf(BeanCreationException.class)); Throwable cause = e.getCause(); - assertThat(cause, Matchers.instanceOf(MessagingException.class)); + assertThat(cause, Matchers.instanceOf(BeanInitializationException.class)); cause = cause.getCause(); assertThat(cause, Matchers.instanceOf(FileNotFoundException.class)); assertEquals("bar", cause.getMessage()); diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/channel/SubscribableRedisChannel.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/channel/SubscribableRedisChannel.java index be32942a11..b3b55a5aab 100644 --- a/spring-integration-redis/src/main/java/org/springframework/integration/redis/channel/SubscribableRedisChannel.java +++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/channel/SubscribableRedisChannel.java @@ -50,14 +50,19 @@ import org.springframework.util.StringUtils; /** * @author Oleg Zhurakousky * @author Gary Russell + * @author Artem Bilan * @since 2.0 */ @SuppressWarnings("rawtypes") -public class SubscribableRedisChannel extends AbstractMessageChannel implements SubscribableChannel, SmartLifecycle, DisposableBean { +public class SubscribableRedisChannel extends AbstractMessageChannel + implements SubscribableChannel, SmartLifecycle, DisposableBean { private final RedisMessageListenerContainer container = new RedisMessageListenerContainer(); + private final RedisConnectionFactory connectionFactory; + private final RedisTemplate redisTemplate; + private final String topicName; private final BroadcastingDispatcher dispatcher = new BroadcastingDispatcher(true); @@ -68,7 +73,9 @@ public class SubscribableRedisChannel extends AbstractMessageChannel implements // defaults private volatile Executor taskExecutor = new SimpleAsyncTaskExecutor(); + private volatile RedisSerializer serializer = new StringRedisSerializer(); + private volatile MessageConverter messageConverter = new SimpleMessageConverter(); public SubscribableRedisChannel(RedisConnectionFactory connectionFactory, String topicName) { @@ -130,7 +137,8 @@ public class SubscribableRedisChannel extends AbstractMessageChannel implements } super.onInit(); if (this.maxSubscribers == null) { - Integer maxSubscribers = this.getIntegrationProperty(IntegrationProperties.CHANNELS_MAX_BROADCAST_SUBSCRIBERS, Integer.class); + Integer maxSubscribers = + getIntegrationProperty(IntegrationProperties.CHANNELS_MAX_BROADCAST_SUBSCRIBERS, Integer.class); this.setMaxSubscribers(maxSubscribers); } if (this.messageConverter == null) { @@ -161,7 +169,7 @@ public class SubscribableRedisChannel extends AbstractMessageChannel implements @Override public boolean isAutoStartup() { - return (this.container != null) ? this.container.isAutoStartup() : false; + return (this.container != null) && this.container.isAutoStartup(); } @Override @@ -171,7 +179,7 @@ public class SubscribableRedisChannel extends AbstractMessageChannel implements @Override public boolean isRunning() { - return (this.container != null) ? this.container.isRunning() : false; + return (this.container != null) && this.container.isRunning(); } @Override @@ -205,8 +213,8 @@ public class SubscribableRedisChannel extends AbstractMessageChannel implements private class MessageListenerDelegate { @SuppressWarnings({ "unused", "unchecked" }) - public void handleMessage(String s) { - Message siMessage = SubscribableRedisChannel.this.messageConverter.toMessage(s, null); + public void handleMessage(Object payload) { + Message siMessage = SubscribableRedisChannel.this.messageConverter.toMessage(payload, null); try { SubscribableRedisChannel.this.dispatcher.dispatch(siMessage); } @@ -215,9 +223,10 @@ public class SubscribableRedisChannel extends AbstractMessageChannel implements topicName = StringUtils.hasText(topicName) ? topicName : "unknown"; throw new MessageDeliveryException(siMessage, e.getMessage() + " for redis-channel '" - + topicName + "' (" + SubscribableRedisChannel.this.getFullChannelName() - + ").", e); + + topicName + + "' (" + SubscribableRedisChannel.this.getFullChannelName() + ").", e); } } } + } diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/channel/SubscribableRedisChannelTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/channel/SubscribableRedisChannelTests.java index ae15addc4f..3913fd2535 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/channel/SubscribableRedisChannelTests.java +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/channel/SubscribableRedisChannelTests.java @@ -99,7 +99,7 @@ public class SubscribableRedisChannelTests extends RedisAvailableTests { MessageListenerAdapter listener = channelMapping.entrySet().iterator().next().getValue().iterator().next(); Object delegate = TestUtils.getPropertyValue(listener, "delegate"); try { - ReflectionUtils.findMethod(delegate.getClass(), "handleMessage", String.class).invoke(delegate, + ReflectionUtils.findMethod(delegate.getClass(), "handleMessage", Object.class).invoke(delegate, "Hello, world!"); fail("Exception expected"); } diff --git a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/DefaultSftpSessionFactory.java b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/DefaultSftpSessionFactory.java index 54292151ea..fc545f2c18 100644 --- a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/DefaultSftpSessionFactory.java +++ b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/DefaultSftpSessionFactory.java @@ -49,6 +49,7 @@ import com.jcraft.jsch.UserInfo; * @author Gary Russell * @author David Liu * @author Pat Turner + * @author Artem Bilan * * @since 2.0 */ @@ -56,6 +57,10 @@ public class DefaultSftpSessionFactory implements SessionFactory, Share private static final Log logger = LogFactory.getLog(DefaultSftpSessionFactory.class); + static { + JSch.setLogger(new JschLogger()); + } + private final ReadWriteLock sharedSessionLock = new ReentrantReadWriteLock(); private final UserInfo userInfoWrapper = new UserInfoWrapper(); @@ -343,7 +348,6 @@ public class DefaultSftpSessionFactory implements SessionFactory, Share public SftpSession getSession() { Assert.hasText(this.host, "host must not be empty"); Assert.hasText(this.user, "user must not be empty"); - Assert.isTrue(this.port >= 0, "port must be a positive number"); Assert.isTrue(StringUtils.hasText(this.userInfoWrapper.getPassword()) || this.privateKey != null, "either a password or a private key is required"); try { @@ -384,8 +388,6 @@ public class DefaultSftpSessionFactory implements SessionFactory, Share } private com.jcraft.jsch.Session initJschSession() throws Exception { - JSch.setLogger(new JschLogger()); - if (this.port <= 0) { this.port = 22; } diff --git a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/JschProxyFactoryBean.java b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/JschProxyFactoryBean.java index 52e72a3d7d..a71da71467 100644 --- a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/JschProxyFactoryBean.java +++ b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/JschProxyFactoryBean.java @@ -64,7 +64,7 @@ public class JschProxyFactoryBean extends AbstractFactoryBean { case HTTP: return ProxyHTTP.class; default: - throw new UnsupportedOperationException("Invalid type:" + this.type); + throw new IllegalArgumentException("Invalid type:" + this.type); } } @@ -84,7 +84,7 @@ public class JschProxyFactoryBean extends AbstractFactoryBean { httpProxy.setUserPasswd(this.user, this.password); return httpProxy; default: - throw new UnsupportedOperationException("Invalid type:" + this.type); + throw new IllegalArgumentException("Invalid type:" + this.type); } } diff --git a/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/inbound/StompInboundChannelAdapter.java b/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/inbound/StompInboundChannelAdapter.java index 2506981cd2..08a6756b7b 100644 --- a/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/inbound/StompInboundChannelAdapter.java +++ b/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/inbound/StompInboundChannelAdapter.java @@ -206,7 +206,7 @@ public class StompInboundChannelAdapter extends MessageProducerSupport implement } } catch (Exception e) { - logger.warn("The exception during unsubscribtion.", e); + logger.warn("The exception during unsubscription.", e); } this.subscriptions.clear(); } diff --git a/spring-integration-syslog/src/main/java/org/springframework/integration/syslog/config/SyslogReceivingChannelAdapterFactoryBean.java b/spring-integration-syslog/src/main/java/org/springframework/integration/syslog/config/SyslogReceivingChannelAdapterFactoryBean.java index 82298c46ef..6da977e9a1 100644 --- a/spring-integration-syslog/src/main/java/org/springframework/integration/syslog/config/SyslogReceivingChannelAdapterFactoryBean.java +++ b/spring-integration-syslog/src/main/java/org/springframework/integration/syslog/config/SyslogReceivingChannelAdapterFactoryBean.java @@ -181,7 +181,7 @@ public class SyslogReceivingChannelAdapterFactoryBean extends AbstractFactoryBea else if (this.applicationEventPublisher != null) { ((TcpSyslogReceivingChannelAdapter) adapter).setApplicationEventPublisher(this.applicationEventPublisher); } - Assert.isNull(this.udpAdapter, "Cannot specifiy 'udp-attributes' when the protocol is 'tcp'"); + Assert.isNull(this.udpAdapter, "Cannot specify 'udp-attributes' when the protocol is 'tcp'"); } else if (this.protocol == Protocol.udp) { adapter = new UdpSyslogReceivingChannelAdapter(); @@ -189,7 +189,7 @@ public class SyslogReceivingChannelAdapterFactoryBean extends AbstractFactoryBea Assert.isNull(this.port, "Cannot specify both 'port' and 'udpAdapter'"); ((UdpSyslogReceivingChannelAdapter) adapter).setUdpAdapter(this.udpAdapter); } - Assert.isNull(this.connectionFactory, "Cannot specifiy 'connection-factory' unless the protocol is 'tcp'"); + Assert.isNull(this.connectionFactory, "Cannot specify 'connection-factory' unless the protocol is 'tcp'"); } else { throw new IllegalStateException("Unsupported protocol: " + this.protocol.toString()); diff --git a/spring-integration-syslog/src/test/java/org/springframework/integration/syslog/config/SyslogReceivingChannelAdapterParserTests.java b/spring-integration-syslog/src/test/java/org/springframework/integration/syslog/config/SyslogReceivingChannelAdapterParserTests.java index 91ee776d2e..8a83bd446c 100644 --- a/spring-integration-syslog/src/test/java/org/springframework/integration/syslog/config/SyslogReceivingChannelAdapterParserTests.java +++ b/spring-integration-syslog/src/test/java/org/springframework/integration/syslog/config/SyslogReceivingChannelAdapterParserTests.java @@ -196,7 +196,7 @@ public class SyslogReceivingChannelAdapterParserTests { catch (BeanCreationException e) { e.printStackTrace(); - assertEquals("Cannot specifiy 'udp-attributes' when the protocol is 'tcp'", e.getCause().getMessage()); + assertEquals("Cannot specify 'udp-attributes' when the protocol is 'tcp'", e.getCause().getMessage()); } } @@ -208,7 +208,7 @@ public class SyslogReceivingChannelAdapterParserTests { fail("Expected exception"); } catch (BeanCreationException e) { - assertEquals("Cannot specifiy 'connection-factory' unless the protocol is 'tcp'", + assertEquals("Cannot specify 'connection-factory' unless the protocol is 'tcp'", e.getCause().getMessage()); } } diff --git a/spring-integration-xml/src/main/java/org/springframework/integration/xml/selector/RegexTestXPathMessageSelector.java b/spring-integration-xml/src/main/java/org/springframework/integration/xml/selector/RegexTestXPathMessageSelector.java index d2458ba75b..b6b6851ad0 100644 --- a/spring-integration-xml/src/main/java/org/springframework/integration/xml/selector/RegexTestXPathMessageSelector.java +++ b/spring-integration-xml/src/main/java/org/springframework/integration/xml/selector/RegexTestXPathMessageSelector.java @@ -98,7 +98,7 @@ public class RegexTestXPathMessageSelector extends AbstractXPathMessageSelector public boolean accept(Message message) { Node nodeToTest = getConverter().convertToNode(message.getPayload()); String xPathResult = getXPathExpresion().evaluateAsString(nodeToTest); - return StringUtils.hasText(xPathResult) ? xPathResult.matches(this.regex) : false; + return StringUtils.hasText(xPathResult) && xPathResult.matches(this.regex); } } diff --git a/spring-integration-xml/src/main/java/org/springframework/integration/xml/selector/XmlValidatingMessageSelector.java b/spring-integration-xml/src/main/java/org/springframework/integration/xml/selector/XmlValidatingMessageSelector.java index f296048a6a..8c462831f6 100644 --- a/spring-integration-xml/src/main/java/org/springframework/integration/xml/selector/XmlValidatingMessageSelector.java +++ b/spring-integration-xml/src/main/java/org/springframework/integration/xml/selector/XmlValidatingMessageSelector.java @@ -133,9 +133,7 @@ public class XmlValidatingMessageSelector implements MessageSelector { new AggregatedXmlMessageValidationException( Arrays.asList(validationExceptions))); } - if (this.logger.isDebugEnabled()) { - this.logger.debug("Message was rejected due to XML Validation errors"); - } + this.logger.debug("Message was rejected due to XML Validation errors"); } return validationSuccess; } diff --git a/spring-integration-zookeeper/src/main/java/org/springframework/integration/zookeeper/config/CuratorFrameworkFactoryBean.java b/spring-integration-zookeeper/src/main/java/org/springframework/integration/zookeeper/config/CuratorFrameworkFactoryBean.java index 297ab5e068..51d5820147 100644 --- a/spring-integration-zookeeper/src/main/java/org/springframework/integration/zookeeper/config/CuratorFrameworkFactoryBean.java +++ b/spring-integration-zookeeper/src/main/java/org/springframework/integration/zookeeper/config/CuratorFrameworkFactoryBean.java @@ -19,7 +19,6 @@ package org.springframework.integration.zookeeper.config; import org.apache.curator.RetryPolicy; import org.apache.curator.framework.CuratorFramework; import org.apache.curator.framework.CuratorFrameworkFactory; -import org.apache.curator.framework.imps.CuratorFrameworkState; import org.apache.curator.retry.ExponentialBackoffRetry; import org.apache.curator.utils.CloseableUtils; @@ -31,7 +30,8 @@ import org.springframework.util.Assert; * A spring-friendly way to build a {@link CuratorFramework} and implementing {@link SmartLifecycle}. * * @author Gary Russell - * + * @author Artem Bilan + * @since 4.2 */ public class CuratorFrameworkFactoryBean implements FactoryBean, SmartLifecycle { @@ -67,7 +67,7 @@ public class CuratorFrameworkFactoryBean implements FactoryBean, SmartLifecycle, InitializingBean, ApplicationEventPublisherAware { @@ -55,14 +55,16 @@ public class LeaderInitiatorFactoryBean private ApplicationEventPublisher applicationEventPublisher; + private LeaderEventPublisher leaderEventPublisher; + public LeaderInitiatorFactoryBean() { } /** * Construct the instance. * @param client the {@link CuratorFramework}. - * @param path the path in zookeeper. - * @param role the role of the leader. + * @param path the path in zookeeper. + * @param role the role of the leader. * @deprecated since {@literal 4.2.5} in favor of appropriate setters * to avoid {@code BeanCurrentlyInCreationException} * during {@code AbstractAutowireCapableBeanFactory.getSingletonFactoryBeanForTypeCheck()} @@ -89,6 +91,33 @@ public class LeaderInitiatorFactoryBean return this; } + /** + * A {@link LeaderEventPublisher} option for events from the {@link LeaderInitiator}. + * @param leaderEventPublisher the {@link LeaderEventPublisher} to use. + * @since 4.3.2 + */ + public void setLeaderEventPublisher(LeaderEventPublisher leaderEventPublisher) { + this.leaderEventPublisher = leaderEventPublisher; + } + + public void setPhase(int phase) { + this.phase = phase; + } + + public void setAutoStartup(boolean autoStartup) { + this.autoStartup = autoStartup; + } + + @Override + public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) { + this.applicationEventPublisher = applicationEventPublisher; + } + + @Override + public boolean isAutoStartup() { + return this.leaderInitiator != null && this.leaderInitiator.isAutoStartup(); + } + @Override public void start() { if (this.leaderInitiator != null) { @@ -103,6 +132,16 @@ public class LeaderInitiatorFactoryBean } } + @Override + public void stop(Runnable callback) { + if (this.leaderInitiator != null) { + this.leaderInitiator.stop(callback); + } + else { + callback.run(); + } + } + @Override public boolean isRunning() { return this.leaderInitiator != null && this.leaderInitiator.isRunning(); @@ -116,37 +155,16 @@ public class LeaderInitiatorFactoryBean return 0; } - public void setPhase(int phase) { - this.phase = phase; - } - - @Override - public boolean isAutoStartup() { - return this.leaderInitiator != null && this.leaderInitiator.isAutoStartup(); - } - - public void setAutoStartup(boolean autoStartup) { - this.autoStartup = autoStartup; - } - - @Override - public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) { - this.applicationEventPublisher = applicationEventPublisher; - } - - @Override - public void stop(Runnable callback) { - stop(); - callback.run(); - } - @Override public void afterPropertiesSet() throws Exception { if (this.leaderInitiator == null) { this.leaderInitiator = new LeaderInitiator(this.client, this.candidate, this.path); this.leaderInitiator.setPhase(this.phase); this.leaderInitiator.setAutoStartup(this.autoStartup); - if (this.applicationEventPublisher != null) { + if (this.leaderEventPublisher != null) { + this.leaderInitiator.setLeaderEventPublisher(this.leaderEventPublisher); + } + else if (this.applicationEventPublisher != null) { this.leaderInitiator.setLeaderEventPublisher( new DefaultLeaderEventPublisher(this.applicationEventPublisher)); } diff --git a/spring-integration-zookeeper/src/main/java/org/springframework/integration/zookeeper/leader/LeaderInitiator.java b/spring-integration-zookeeper/src/main/java/org/springframework/integration/zookeeper/leader/LeaderInitiator.java index cafb0c847a..95fd3f2db8 100644 --- a/spring-integration-zookeeper/src/main/java/org/springframework/integration/zookeeper/leader/LeaderInitiator.java +++ b/spring-integration-zookeeper/src/main/java/org/springframework/integration/zookeeper/leader/LeaderInitiator.java @@ -208,7 +208,7 @@ public class LeaderInitiator implements SmartLifecycle { if (!ns.endsWith("/")) { ns = ns + "/"; } - return String.format(ns + "%s", this.candidate.getRole()); + return ns + this.candidate.getRole(); } /**