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

Some Fixes and Improvements

Browse Source 
* Fix several typos in log messages. And some test on the matter as well
* Add comment to `AbstractPersistentAcceptOnceFileListFilter.rollback()` to clarify the reason of `rollingBack` variable
* Make `RemoteFileTemplate.StreamHolder` as `static` to avoid extra internal variable to outer class instance
* Replace `MessagingException` with `AbstractInboundFileSynchronizingMessageSource` in `init()` method of some components. It isn't Messaging yet in that phase
* Fix `SubscribableRedisChannel.MessageListenerDelegate` to handle `Object` not `String`, because with the `serializer` injection there is no guaranty that incoming is always `String`
* Move `JSch.setLogger(new JschLogger());` in the `DefaultSftpSessionFactory` to `static` block. It really should be done only once
* Remove `Assert.isTrue(this.port >= 0)` from the `DefaultSftpSessionFactory`. The subsequant `initJschSession()` convert it to default `22` port
* Change in the `JschProxyFactoryBean` `UnsupportedOperationException` to `IllegalArgumentException`. Wrong enum is wrong argument. That isn't a problem of operation
* Simplify `stop()` in the `CuratorFrameworkFactoryBean` and mark it as a `this.running = false`. Otherwise it wasn't able to be restarted
* Expose `leaderEventPublisher` in the `LeaderInitiatorFactoryBean`  and fix `stop(Runnable callback)` with propagation `callback` to delegate.

Fix `SubscribableRedisChannelTests` for new `handleMessage(Object)` signature
This commit is contained in:
Artem Bilan
2016-08-10 13:24:49 -04:00
committed by Gary Russell
parent 28216013dd
commit 7124136091
16 changed files with 91 additions and 64 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
spring-integration-file/src/main/java/org/springframework/integration/file/filters/AbstractPersistentAcceptOnceFileListFilter.java
View File

@@ -94,6 +94,7 @@ public abstract class AbstractPersistentAcceptOnceFileListFilter<F> extends Abst
*/
@Override
public void rollback(F file, List<F> 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)) {

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

@@ -572,7 +572,7 @@ public class RemoteFileTemplate<F> implements RemoteFileOperations<F>, Initializ
return directoryPath;
}
private final class StreamHolder {
private static final class StreamHolder {
private final InputStream stream;

6
spring-integration-file/src/main/java/org/springframework/integration/file/remote/synchronizer/AbstractInboundFileSynchronizingMessageSource.java
View File

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

4
spring-integration-ftp/src/test/java/org/springframework/integration/ftp/FtpParserInboundTests.java
View File

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

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

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

2
spring-integration-redis/src/test/java/org/springframework/integration/redis/channel/SubscribableRedisChannelTests.java
View File

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

8
spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/DefaultSftpSessionFactory.java
View File

@@ -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<LsEntry>, 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<LsEntry>, 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<LsEntry>, Share
}
private com.jcraft.jsch.Session initJschSession() throws Exception {
JSch.setLogger(new JschLogger());
if (this.port <= 0) {
this.port = 22;
}

4
spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/JschProxyFactoryBean.java
View File

@@ -64,7 +64,7 @@ public class JschProxyFactoryBean extends AbstractFactoryBean<Proxy> {
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<Proxy> {
httpProxy.setUserPasswd(this.user, this.password);
return httpProxy;
default:
throw new UnsupportedOperationException("Invalid type:" + this.type);
throw new IllegalArgumentException("Invalid type:" + this.type);
}
}

2
spring-integration-stomp/src/main/java/org/springframework/integration/stomp/inbound/StompInboundChannelAdapter.java
View File

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

4
spring-integration-syslog/src/main/java/org/springframework/integration/syslog/config/SyslogReceivingChannelAdapterFactoryBean.java
View File

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

4
spring-integration-syslog/src/test/java/org/springframework/integration/syslog/config/SyslogReceivingChannelAdapterParserTests.java
View File

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

2
spring-integration-xml/src/main/java/org/springframework/integration/xml/selector/RegexTestXPathMessageSelector.java
View File

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

4
spring-integration-xml/src/main/java/org/springframework/integration/xml/selector/XmlValidatingMessageSelector.java
View File

@@ -133,9 +133,7 @@ public class XmlValidatingMessageSelector implements MessageSelector {
new AggregatedXmlMessageValidationException(
Arrays.<Throwable>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;
}

11
spring-integration-zookeeper/src/main/java/org/springframework/integration/zookeeper/config/CuratorFrameworkFactoryBean.java
View File

@@ -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<CuratorFramework>, SmartLifecycle {
@@ -67,7 +67,7 @@ public class CuratorFrameworkFactoryBean implements FactoryBean<CuratorFramework
/**
* Construct an instance using the supplied connection string and retry policy.
* @param connectionString list of servers to connect to
* @param retryPolicy the retry policy
* @param retryPolicy the retry policy
*/
public CuratorFrameworkFactoryBean(String connectionString, RetryPolicy retryPolicy) {
Assert.notNull(connectionString, "'connectionString' cannot be null");
@@ -122,9 +122,8 @@ public class CuratorFrameworkFactoryBean implements FactoryBean<CuratorFramework
public void stop() {
synchronized (this.lifecycleLock) {
if (this.running) {
if (this.client.getState().equals(CuratorFrameworkState.STARTED)) {
CloseableUtils.closeQuietly(this.client);
}
CloseableUtils.closeQuietly(this.client);
this.running = false;
}
}
}

74
spring-integration-zookeeper/src/main/java/org/springframework/integration/zookeeper/config/LeaderInitiatorFactoryBean.java
View File

@@ -28,6 +28,7 @@ import org.springframework.context.SmartLifecycle;
import org.springframework.integration.leader.Candidate;
import org.springframework.integration.leader.DefaultCandidate;
import org.springframework.integration.leader.event.DefaultLeaderEventPublisher;
import org.springframework.integration.leader.event.LeaderEventPublisher;
import org.springframework.integration.zookeeper.leader.LeaderInitiator;
/**
@@ -36,7 +37,6 @@ import org.springframework.integration.zookeeper.leader.LeaderInitiator;
* @author Gary Russell
* @author Artem Bilan
* @since 4.2
*
*/
public class LeaderInitiatorFactoryBean
implements FactoryBean<LeaderInitiator>, 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));
}

2
spring-integration-zookeeper/src/main/java/org/springframework/integration/zookeeper/leader/LeaderInitiator.java
View File

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