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

INT-1614 removing scheduled task capabilities from synchronizers

Browse Source
This commit is contained in:
Mark Fisher
2010-11-19 08:11:26 -05:00
parent d03a6533cf
commit 1d3445c538
7 changed files with 37 additions and 117 deletions
Download Patch File Download Diff File Expand all files Collapse all files

72
spring-integration-file/src/main/java/org/springframework/integration/file/synchronization/AbstractInboundRemoteFileSystemSychronizer.java
View File

@@ -16,15 +16,13 @@
package org.springframework.integration.file.synchronization;
import java.util.concurrent.ScheduledFuture;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.core.io.Resource;
import org.springframework.integration.MessagingException;
import org.springframework.integration.endpoint.AbstractEndpoint;
import org.springframework.integration.file.filters.AcceptAllFileListFilter;
import org.springframework.integration.file.filters.FileListFilter;
import org.springframework.scheduling.Trigger;
import org.springframework.util.Assert;
/**
* Base class charged with knowing how to connect to a remote file system,
@@ -36,7 +34,10 @@ import org.springframework.util.Assert;
*
* @author Josh Long
*/
public abstract class AbstractInboundRemoteFileSystemSychronizer<F> extends AbstractEndpoint {
public abstract class AbstractInboundRemoteFileSystemSychronizer<F> implements InitializingBean {
protected final Log logger = LogFactory.getLog(this.getClass());
/**
* Should we <emphasis>delete</emphasis> the <b>source</b> file? For an FTP
@@ -54,11 +55,6 @@ public abstract class AbstractInboundRemoteFileSystemSychronizer<F> extends Abst
*/
protected volatile FileListFilter<F> filter = new AcceptAllFileListFilter<F>();
/**
* The {@link ScheduledFuture} instance we get when we schedule our SynchronizeTask.
*/
protected ScheduledFuture<?> scheduledFuture;
/**
* The {@link EntryAcknowledgmentStrategy} implementation.
*/
@@ -96,65 +92,17 @@ public abstract class AbstractInboundRemoteFileSystemSychronizer<F> extends Abst
* escape hatch exception, let the adapter deal with it.
*/
protected void acknowledge(Object usefulContextOrClientData, F file) throws Throwable {
Assert.notNull(this.entryAcknowledgmentStrategy != null,
"entryAcknowledgmentStrategy can't be null!");
this.entryAcknowledgmentStrategy.acknowledge(usefulContextOrClientData, file);
}
/**
* {@inheritDoc}
*/
protected void doStart() {
if (this.entryAcknowledgmentStrategy == null) {
this.entryAcknowledgmentStrategy = new EntryAcknowledgmentStrategy<F>() {
public void acknowledge(Object o, F msg) {
// no-op
}
};
}
this.scheduledFuture = this.getTaskScheduler().schedule(new SynchronizeTask(), this.getTrigger());
}
/**
* {@inheritDoc}
*/
protected void doStop() {
if (this.scheduledFuture != null) {
this.scheduledFuture.cancel(true);
if (this.entryAcknowledgmentStrategy != null) {
this.entryAcknowledgmentStrategy.acknowledge(usefulContextOrClientData, file);
}
}
/**
* Returns the {@link Trigger} that dictates how frequently the trigger should fire.
*/
protected abstract Trigger getTrigger();
/**
* This is the callback where we need the implementation to do some specific work
* This is the callback where the subclasses must synchronize.
*/
protected abstract void syncRemoteToLocalFileSystem();
/**
* This {@link Runnable} is launched as a background thread and is used to manage the
* {@link AbstractInboundRemoteFileSystemSychronizer#localDirectory} by queueing and
* delivering accumulated files as possible.
*/
class SynchronizeTask implements Runnable {
public void run() {
try {
syncRemoteToLocalFileSystem();
}
catch (RuntimeException e) {
throw e;
}
catch (Exception e) {
throw new MessagingException("failure occurred in synchronization task", e);
}
}
}
/**
* Strategy interface to expose a hook for dispatching, moving, or deleting
* the file once it has been delivered. This will typically be a NOOP for the

14
spring-integration-file/src/main/java/org/springframework/integration/file/synchronization/AbstractInboundRemoteFileSystemSynchronizingMessageSource.java
View File

@@ -31,6 +31,7 @@ import org.springframework.integration.file.filters.AcceptOnceFileListFilter;
import org.springframework.integration.file.filters.CompositeFileListFilter;
import org.springframework.integration.file.filters.FileListFilter;
import org.springframework.integration.file.filters.PatternMatchingFileListFilter;
import org.springframework.util.Assert;
/**
* Factors out the common logic between the FTP and SFTP adapters. Designed to
@@ -60,6 +61,7 @@ public abstract class AbstractInboundRemoteFileSystemSynchronizingMessageSource<
*/
public static final String INCOMPLETE_EXTENSION = ".INCOMPLETE";
/**
* Should the endpoint attempt to create the local directory and/or the remote directory?
*/
@@ -125,10 +127,10 @@ public abstract class AbstractInboundRemoteFileSystemSynchronizingMessageSource<
* Make sure the remote files get here.
*/
this.synchronizer.setLocalDirectory(this.localDirectory);
this.synchronizer.setTaskScheduler(this.getTaskScheduler());
this.synchronizer.setBeanFactory(this.getBeanFactory());
this.synchronizer.setPhase(this.getPhase());
this.synchronizer.setBeanName(this.getComponentName());
//this.synchronizer.setTaskScheduler(this.getTaskScheduler());
//this.synchronizer.setBeanFactory(this.getBeanFactory());
//this.synchronizer.setPhase(this.getPhase());
//this.synchronizer.setBeanName(this.getComponentName());
/**
* Forwards files once they ultimately appear in the {@link #localDirectory}.
@@ -137,7 +139,7 @@ public abstract class AbstractInboundRemoteFileSystemSynchronizingMessageSource<
this.fileSource.setFilter(this.buildFilter());
this.fileSource.setDirectory(this.localDirectory.getFile());
this.fileSource.afterPropertiesSet();
this.synchronizer.afterPropertiesSet();
//this.synchronizer.afterPropertiesSet();
}
catch (RuntimeException e) {
throw e;
@@ -154,6 +156,8 @@ public abstract class AbstractInboundRemoteFileSystemSynchronizingMessageSource<
* Then, it polls the file source again and returns the result, whether or not it is null.
*/
public final Message<File> receive() {
Assert.state(this.fileSource != null, "fileSource must not be null");
Assert.state(this.synchronizer != null, "synchronizer must not be null");
Message<File> message = this.fileSource.receive();
if (message == null) {
this.synchronizer.syncRemoteToLocalFileSystem();

31
spring-integration-ftp/src/main/java/org/springframework/integration/ftp/inbound/FtpInboundRemoteFileSystemSynchronizer.java
View File

@@ -16,23 +16,22 @@
package org.springframework.integration.ftp.inbound;
import org.apache.commons.net.ftp.FTPClient;
import org.apache.commons.net.ftp.FTPFile;
import org.springframework.core.io.Resource;
import org.springframework.integration.MessagingException;
import org.springframework.integration.file.synchronization.AbstractInboundRemoteFileSystemSychronizer;
import org.springframework.integration.file.synchronization.AbstractInboundRemoteFileSystemSynchronizingMessageSource;
import org.springframework.integration.ftp.client.FtpClientPool;
import org.springframework.scheduling.Trigger;
import org.springframework.scheduling.support.PeriodicTrigger;
import org.springframework.util.Assert;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.util.Collection;
import org.apache.commons.net.ftp.FTPClient;
import org.apache.commons.net.ftp.FTPFile;
import org.springframework.core.io.Resource;
import org.springframework.integration.MessagingException;
import org.springframework.integration.file.synchronization.AbstractInboundRemoteFileSystemSychronizer;
import org.springframework.integration.file.synchronization.AbstractInboundRemoteFileSystemSynchronizingMessageSource;
import org.springframework.integration.ftp.client.FtpClientPool;
import org.springframework.util.Assert;
/**
* An FTP-adapter implementation of {@link org.springframework.integration.file.synchronization.AbstractInboundRemoteFileSystemSychronizer}
*
@@ -41,16 +40,9 @@ import java.util.Collection;
*/
public class FtpInboundRemoteFileSystemSynchronizer extends AbstractInboundRemoteFileSystemSychronizer<FTPFile> {
private volatile Trigger trigger = new PeriodicTrigger(10 * 1000);
protected volatile FtpClientPool clientPool;
@Override
protected Trigger getTrigger() {
return this.trigger;
}
/**
* The {@link org.springframework.integration.ftp.client.FtpClientPool} that holds references to {@link org.apache.commons.net.ftp.FTPClient} instances
*
@@ -60,8 +52,7 @@ public class FtpInboundRemoteFileSystemSynchronizer extends AbstractInboundRemot
this.clientPool = clientPool;
}
@Override
protected void onInit() throws Exception {
public void afterPropertiesSet() throws Exception {
Assert.notNull(this.clientPool, "clientPool must not be null");
if (this.shouldDeleteSourceFile) {
this.entryAcknowledgmentStrategy = new DeletionEntryAcknowledgmentStrategy();

10
spring-integration-ftp/src/main/java/org/springframework/integration/ftp/inbound/FtpInboundRemoteFileSystemSynchronizingMessageSource.java
View File

@@ -47,14 +47,4 @@ public class FtpInboundRemoteFileSystemSynchronizingMessageSource
this.synchronizer.setClientPool(this.clientPool);
}
@Override
protected void doStart() {
//this.synchronizer.start();
}
@Override
protected void doStop() {
//this.synchronizer.stop();
}
}

1
spring-integration-sftp/src/main/java/org/springframework/integration/sftp/config/SftpInboundChannelAdapterParser.java
View File

@@ -58,7 +58,6 @@ public class SftpInboundChannelAdapterParser extends AbstractPollingInboundChann
BeanDefinitionBuilder synchronizerBuilder = BeanDefinitionBuilder.genericBeanDefinition(
"org.springframework.integration.sftp.inbound.SftpInboundSynchronizer");
synchronizerBuilder.addConstructorArgReference(sessionPollName);
synchronizerBuilder.addPropertyValue("autoStartup", autoStartup);
IntegrationNamespaceUtils.setValueIfAttributeDefined(synchronizerBuilder, element, "remote-directory", "remotePath");
IntegrationNamespaceUtils.setValueIfAttributeDefined(synchronizerBuilder, element, "local-directory");
IntegrationNamespaceUtils.setValueIfAttributeDefined(synchronizerBuilder, element, "auto-delete-remote-files-on-sync", "shouldDeleteSourceFile");

12
spring-integration-sftp/src/main/java/org/springframework/integration/sftp/inbound/SftpInboundSynchronizer.java
View File

@@ -29,7 +29,6 @@ import org.springframework.integration.file.synchronization.AbstractInboundRemot
import org.springframework.integration.file.synchronization.AbstractInboundRemoteFileSystemSynchronizingMessageSource;
import org.springframework.integration.sftp.session.SftpSession;
import org.springframework.integration.sftp.session.SftpSessionPool;
import org.springframework.scheduling.Trigger;
import org.springframework.util.Assert;
import com.jcraft.jsch.ChannelSftp;
@@ -71,13 +70,7 @@ public class SftpInboundSynchronizer extends AbstractInboundRemoteFileSystemSych
this.remotePath = remotePath;
}
@Override
protected Trigger getTrigger() {
throw new UnsupportedOperationException("This method curently is not implemented");
}
@Override
protected void onInit() throws Exception {
public void afterPropertiesSet() throws Exception {
Assert.notNull(this.remotePath, "'remotePath' must not be null");
if (this.shouldDeleteSourceFile) {
this.entryAcknowledgmentStrategy = new DeletionEntryAcknowledgmentStrategy();
@@ -94,9 +87,7 @@ public class SftpInboundSynchronizer extends AbstractInboundRemoteFileSystemSych
* existed.)
*/
private boolean checkThatRemotePathExists(String remotePath, SftpSession session) {
ChannelSftp channelSftp = session.getChannel();
try {
SftpATTRS attrs = channelSftp.stat(remotePath);
assert (attrs != null) && attrs.isDir() : "attrs can't be null, and should indicate that it's a directory!";
@@ -122,7 +113,6 @@ public class SftpInboundSynchronizer extends AbstractInboundRemoteFileSystemSych
}
}
return false;
}

14
spring-integration-sftp/src/main/java/org/springframework/integration/sftp/inbound/SftpInboundSynchronizingMessageSource.java
View File

@@ -65,19 +65,17 @@ public class SftpInboundSynchronizingMessageSource
this.fileSource = new FileReadingMessageSource();
this.fileSource.setDirectory(this.localDirectory.getFile());
this.fileSource.afterPropertiesSet();
if (this.filenamePattern != null) {
SftpPatternMatchingFileListFilter filter = new SftpPatternMatchingFileListFilter(this.filenamePattern);
this.synchronizer.setFilter(filter);
this.synchronizer.setAutoCreateDirectories(this.autoCreateDirectories);
}
}
catch (RuntimeException e) {
throw e;
}
catch (Exception e) {
throw new MessagingException("Failure during initialization of MessageSource for: "
+ this.getComponentType(), e);
}
if (this.filenamePattern != null) {
SftpPatternMatchingFileListFilter filter = new SftpPatternMatchingFileListFilter(this.filenamePattern);
this.synchronizer.setFilter(filter);
this.synchronizer.setAutoCreateDirectories(this.autoCreateDirectories);
throw new MessagingException("Failure during initialization of MessageSource for: " + this.getComponentType(), e);
}
}