From 1d3445c53837d6336e575a53cd1d058e37d72498 Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Fri, 19 Nov 2010 08:11:26 -0500 Subject: [PATCH] INT-1614 removing scheduled task capabilities from synchronizers --- ...actInboundRemoteFileSystemSychronizer.java | 72 +++---------------- ...eFileSystemSynchronizingMessageSource.java | 14 ++-- ...tpInboundRemoteFileSystemSynchronizer.java | 31 +++----- ...eFileSystemSynchronizingMessageSource.java | 10 --- .../SftpInboundChannelAdapterParser.java | 1 - .../sftp/inbound/SftpInboundSynchronizer.java | 12 +--- ...SftpInboundSynchronizingMessageSource.java | 14 ++-- 7 files changed, 37 insertions(+), 117 deletions(-) diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/synchronization/AbstractInboundRemoteFileSystemSychronizer.java b/spring-integration-file/src/main/java/org/springframework/integration/file/synchronization/AbstractInboundRemoteFileSystemSychronizer.java index ea2c3c79cb..cd2de61b2d 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/synchronization/AbstractInboundRemoteFileSystemSychronizer.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/synchronization/AbstractInboundRemoteFileSystemSychronizer.java @@ -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 extends AbstractEndpoint { +public abstract class AbstractInboundRemoteFileSystemSychronizer implements InitializingBean { + + protected final Log logger = LogFactory.getLog(this.getClass()); + /** * Should we delete the source file? For an FTP @@ -54,11 +55,6 @@ public abstract class AbstractInboundRemoteFileSystemSychronizer extends Abst */ protected volatile FileListFilter filter = new AcceptAllFileListFilter(); - /** - * 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 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() { - 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 diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/synchronization/AbstractInboundRemoteFileSystemSynchronizingMessageSource.java b/spring-integration-file/src/main/java/org/springframework/integration/file/synchronization/AbstractInboundRemoteFileSystemSynchronizingMessageSource.java index 3e41d3564c..dbedbb8005 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/synchronization/AbstractInboundRemoteFileSystemSynchronizingMessageSource.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/synchronization/AbstractInboundRemoteFileSystemSynchronizingMessageSource.java @@ -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 receive() { + Assert.state(this.fileSource != null, "fileSource must not be null"); + Assert.state(this.synchronizer != null, "synchronizer must not be null"); Message message = this.fileSource.receive(); if (message == null) { this.synchronizer.syncRemoteToLocalFileSystem(); diff --git a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/inbound/FtpInboundRemoteFileSystemSynchronizer.java b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/inbound/FtpInboundRemoteFileSystemSynchronizer.java index b22bf7a33b..7795260500 100644 --- a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/inbound/FtpInboundRemoteFileSystemSynchronizer.java +++ b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/inbound/FtpInboundRemoteFileSystemSynchronizer.java @@ -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 { - 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(); diff --git a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/inbound/FtpInboundRemoteFileSystemSynchronizingMessageSource.java b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/inbound/FtpInboundRemoteFileSystemSynchronizingMessageSource.java index 86169619b7..9149f67d24 100644 --- a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/inbound/FtpInboundRemoteFileSystemSynchronizingMessageSource.java +++ b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/inbound/FtpInboundRemoteFileSystemSynchronizingMessageSource.java @@ -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(); - } - } diff --git a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/config/SftpInboundChannelAdapterParser.java b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/config/SftpInboundChannelAdapterParser.java index a67d7fa703..40bdad43c8 100644 --- a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/config/SftpInboundChannelAdapterParser.java +++ b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/config/SftpInboundChannelAdapterParser.java @@ -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"); diff --git a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/inbound/SftpInboundSynchronizer.java b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/inbound/SftpInboundSynchronizer.java index 53fdd4f71a..2be02b4311 100644 --- a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/inbound/SftpInboundSynchronizer.java +++ b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/inbound/SftpInboundSynchronizer.java @@ -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; } diff --git a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/inbound/SftpInboundSynchronizingMessageSource.java b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/inbound/SftpInboundSynchronizingMessageSource.java index 554bdd6cd3..996cad5bf4 100644 --- a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/inbound/SftpInboundSynchronizingMessageSource.java +++ b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/inbound/SftpInboundSynchronizingMessageSource.java @@ -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); } }