Spring 集成 - AbstractInboundFileSynchronizer 未更新文件
Spring integration - AbstractInboundFileSynchronizer not updating file
我原以为 ftp 同步机制会更新已更改的文件。但是，从我在这里看到的情况来看，只有在文件尚不存在时才会下载该文件。至于现在，即使时间戳/内容发生变化，文件也不会保存在本地。
这是我目前的发现：
classorg.springframework.integration.file.remote.synchronizer.AbstractInboundFileSynchronizer
@Override
public void synchronizeToLocalDirectory(final File localDirectory) {
final String remoteDirectory = this.remoteDirectoryExpression.getValue(this.evaluationContext, String.class);
try {
int transferred = this.remoteFileTemplate.execute(new SessionCallback<F, Integer>() {
@Override
public Integer doInSession(Session<F> session) throws IOException {
F[] files = session.list(remoteDirectory);
if (!ObjectUtils.isEmpty(files)) {
List<F> filteredFiles = filterFiles(files);
for (F file : filteredFiles) {
try {
if (file != null) {
copyFileToLocalDirectory(
remoteDirectory, file, localDirectory,
session);
}
}
catch (RuntimeException e) {
if (AbstractInboundFileSynchronizer.this.filter instanceof ReversibleFileListFilter) {
((ReversibleFileListFilter<F>) AbstractInboundFileSynchronizer.this.filter)
.rollback(file, filteredFiles);
}
throw e;
}
catch (IOException e) {
if (AbstractInboundFileSynchronizer.this.filter instanceof ReversibleFileListFilter) {
((ReversibleFileListFilter<F>) AbstractInboundFileSynchronizer.this.filter)
.rollback(file, filteredFiles);
}
throw e;
}
}
return filteredFiles.size();
}
else {
return 0;
}
}
});
if (this.logger.isDebugEnabled()) {
this.logger.debug(transferred + " files transferred");
}
}
catch (Exception e) {
throw new MessagingException("Problem occurred while synchronizing remote to local directory", e);
}
}
过滤要下载的文件。我想使用 org.springframework.integration.ftp.filters.FtpPersistentAcceptOnceFileListFilter，它比较 文件名 和最后 修改日期 .
然后它会调用 copyFileToLocalDirectory 函数来过滤文件（要复制）。
protected void copyFileToLocalDirectory(String remoteDirectoryPath, F remoteFile, File localDirectory,
Session<F> session) throws IOException {
String remoteFileName = this.getFilename(remoteFile);
String localFileName = this.generateLocalFileName(remoteFileName);
String remoteFilePath = remoteDirectoryPath != null
? (remoteDirectoryPath + this.remoteFileSeparator + remoteFileName)
: remoteFileName;
if (!this.isFile(remoteFile)) {
if (this.logger.isDebugEnabled()) {
this.logger.debug("cannot copy, not a file: " + remoteFilePath);
}
return;
}
File localFile = new File(localDirectory, localFileName);
if (!localFile.exists()) {
String tempFileName = localFile.getAbsolutePath() + this.temporaryFileSuffix;
File tempFile = new File(tempFileName);
OutputStream outputStream = new BufferedOutputStream(new FileOutputStream(tempFile));
try {
session.read(remoteFilePath, outputStream);
}
catch (Exception e) {
if (e instanceof RuntimeException) {
throw (RuntimeException) e;
}
else {
throw new MessagingException("Failure occurred while copying from remote to local directory", e);
}
}
finally {
try {
outputStream.close();
}
catch (Exception ignored2) {
}
}
if (tempFile.renameTo(localFile)) {
if (this.deleteRemoteFiles) {
session.remove(remoteFilePath);
if (this.logger.isDebugEnabled()) {
this.logger.debug("deleted " + remoteFilePath);
}
}
}
if (this.preserveTimestamp) {
localFile.setLastModified(getModified(remoteFile));
}
}
}
但是，此方法会检查（仅基于文件名）文件是否已存在于本地磁盘上，并且仅在不存在时才下载。所以基本上没有机会下载带有（带有新时间戳）的更新文件。
我试过尝试更改 FtpInboundFileSynchronizer，但它变得太复杂了。 "customize" synchronize- / copyToLocalDirectory 方法的最佳方法是什么？
您可以在构造期间使用远程过滤器自定义 FtpInboundFileSynchonizer。
将过滤器设置为 FtpPersistentAcceptOnceFileListFilter 以配置基于时间戳更改的过滤。
设置后的远程过滤器将在synchronizeToLocalDirectory中用于过滤远程查看和下载文件。
这将在 doInSession 的这一行中使用
List<F> filteredFiles = filterFiles(files);
您还需要将本地过滤器设置为 FileSystemPersistentAcceptOnceFileListFilter 以根据远程文件更改相应地处理本地端更改。
此外，您需要提供您自己的元数据存储或使用 PropertiesPersistingMetadataStore 以在简单情况下持久保存本地和远程文件元数据更改，以便它可以在服务器重新启动后继续存在。
这里有更多
http://docs.spring.io/spring-integration/reference/html/ftp.html#ftp-inbound
直接的方法是extends AbstractInboundFileSynchronizerand override copyFileToLocalDirectory方法来比较你想要的条件比如修改时间，如果你不想这样做，那么你有另一种方法使用 AOP：
@Before("org.springframework.integration.file.remote.synchronizer.AbstractInboundFileSynchronizer.copyFileToLocalDirectory()")
然后从 joinPoint 获取参数并让你比较如果它是真的并且你想更新文件所以删除本地文件。
可以更新 AbstractInboundFileSynchronizer 以识别更新的文件，但它很脆弱，您 运行 会遇到其他问题。
更新 13/Nov/2016: 了解如何以秒为单位获取修改时间戳。
更新 AbstractInboundFileSynchronizer 的主要问题是它有 setter-方法但没有（受保护的）getter-方法。如果将来 setter 方法做一些聪明的事情，这里介绍的更新版本将会崩溃。
在本地目录中更新文件的主要问题是并发性：如果您在接收更新的同时处理本地文件，您可能 运行 陷入各种麻烦。简单的方法是将本地文件移动到（临时）处理目录，以便可以将更新作为新文件接收，从而无需更新 AbstractInboundFileSynchronizer。另见 Camel 的时间戳 remarks.
默认情况下 FTP 服务器以分钟为单位提供修改时间戳。为了进行测试，我更新了 FTP 客户端以使用 MLSD 命令，该命令以秒为单位提供修改时间戳（如果幸运的话，可以提供毫秒），但并非所有 FTP 服务器都支持此功能。
如 Spring FTP reference 中所述，本地文件过滤器需要 FileSystemPersistentAcceptOnceFileListFilter 以确保在修改时间戳更改时选择本地文件。
下面是我更新后的版本AbstractInboundFileSynchronizer，后面是我用过的一些测试classes。
public class FtpUpdatingFileSynchronizer extends FtpInboundFileSynchronizer {
protected final Log logger = LogFactory.getLog(this.getClass());
private volatile Expression localFilenameGeneratorExpression;
private volatile EvaluationContext evaluationContext;
private volatile boolean deleteRemoteFiles;
private volatile String remoteFileSeparator = "/";
private volatile boolean preserveTimestamp;
public FtpUpdatingFileSynchronizer(SessionFactory<FTPFile> sessionFactory) {
super(sessionFactory);
setPreserveTimestamp(true);
}
@Override
public void setLocalFilenameGeneratorExpression(Expression localFilenameGeneratorExpression) {
super.setLocalFilenameGeneratorExpression(localFilenameGeneratorExpression);
this.localFilenameGeneratorExpression = localFilenameGeneratorExpression;
}
@Override
public void setIntegrationEvaluationContext(EvaluationContext evaluationContext) {
super.setIntegrationEvaluationContext(evaluationContext);
this.evaluationContext = evaluationContext;
}
@Override
public void setDeleteRemoteFiles(boolean deleteRemoteFiles) {
super.setDeleteRemoteFiles(deleteRemoteFiles);
this.deleteRemoteFiles = deleteRemoteFiles;
}
@Override
public void setRemoteFileSeparator(String remoteFileSeparator) {
super.setRemoteFileSeparator(remoteFileSeparator);
this.remoteFileSeparator = remoteFileSeparator;
}
@Override
public void setPreserveTimestamp(boolean preserveTimestamp) {
// updated
Assert.isTrue(preserveTimestamp, "for updating timestamps must be preserved");
super.setPreserveTimestamp(preserveTimestamp);
this.preserveTimestamp = preserveTimestamp;
}
@Override
protected void copyFileToLocalDirectory(String remoteDirectoryPath, FTPFile remoteFile, File localDirectory,
Session<FTPFile> session) throws IOException {
String remoteFileName = this.getFilename(remoteFile);
String localFileName = this.generateLocalFileName(remoteFileName);
String remoteFilePath = (remoteDirectoryPath != null
? (remoteDirectoryPath + this.remoteFileSeparator + remoteFileName)
: remoteFileName);
if (!this.isFile(remoteFile)) {
if (this.logger.isDebugEnabled()) {
this.logger.debug("cannot copy, not a file: " + remoteFilePath);
}
return;
}
// start update
File localFile = new File(localDirectory, localFileName);
boolean update = false;
if (localFile.exists()) {
if (this.getModified(remoteFile) > localFile.lastModified()) {
this.logger.info("Updating local file " + localFile);
update = true;
} else {
this.logger.info("File already exists: " + localFile);
return;
}
}
// end update
String tempFileName = localFile.getAbsolutePath() + this.getTemporaryFileSuffix();
File tempFile = new File(tempFileName);
OutputStream outputStream = new BufferedOutputStream(new FileOutputStream(tempFile));
try {
session.read(remoteFilePath, outputStream);
} catch (Exception e) {
if (e instanceof RuntimeException) {
throw (RuntimeException) e;
}
else {
throw new MessagingException("Failure occurred while copying from remote to local directory", e);
}
} finally {
try {
outputStream.close();
}
catch (Exception ignored2) {
}
}
// updated
if (update && !localFile.delete()) {
throw new MessagingException("Unable to delete local file [" + localFile + "] for update.");
}
if (tempFile.renameTo(localFile)) {
if (this.deleteRemoteFiles) {
session.remove(remoteFilePath);
if (this.logger.isDebugEnabled()) {
this.logger.debug("deleted " + remoteFilePath);
}
}
// updated
this.logger.info("Stored file locally: " + localFile);
} else {
// updated
throw new MessagingException("Unable to rename temporary file [" + tempFile + "] to [" + localFile + "]");
}
if (this.preserveTimestamp) {
localFile.setLastModified(getModified(remoteFile));
}
}
private String generateLocalFileName(String remoteFileName) {
if (this.localFilenameGeneratorExpression != null) {
return this.localFilenameGeneratorExpression.getValue(this.evaluationContext, remoteFileName, String.class);
}
return remoteFileName;
}
}
根据我使用的一些测试 classes。
我使用了依赖项 org.springframework.integration:spring-integration-ftp:4.3.5.RELEASE 和 org.apache.ftpserver:ftpserver-core:1.0.6（加上通常的日志记录和测试依赖项）。
public class TestFtpSync {
static final Logger log = LoggerFactory.getLogger(TestFtpSync.class);
static final String FTP_ROOT_DIR = "target" + File.separator + "ftproot";
// org.apache.ftpserver:ftpserver-core:1.0.6
static FtpServer server;
@BeforeClass
public static void startServer() throws FtpException {
File ftpRoot = new File (FTP_ROOT_DIR);
ftpRoot.mkdirs();
TestUserManager userManager = new TestUserManager(ftpRoot.getAbsolutePath());
FtpServerFactory serverFactory = new FtpServerFactory();
serverFactory.setUserManager(userManager);
ListenerFactory factory = new ListenerFactory();
factory.setPort(4444);
serverFactory.addListener("default", factory.createListener());
server = serverFactory.createServer();
server.start();
}
@AfterClass
public static void stopServer() {
if (server != null) {
server.stop();
}
}
File ftpFile = Paths.get(FTP_ROOT_DIR, "test1.txt").toFile();
File ftpFile2 = Paths.get(FTP_ROOT_DIR, "test2.txt").toFile();
@Test
public void syncDir() {
// org.springframework.integration:spring-integration-ftp:4.3.5.RELEASE
AnnotationConfigApplicationContext ctx = new AnnotationConfigApplicationContext();
try {
ctx.register(FtpSyncConf.class);
ctx.refresh();
PollableChannel msgChannel = ctx.getBean("inputChannel", PollableChannel.class);
for (int j = 0; j < 2; j++) {
for (int i = 0; i < 2; i++) {
storeFtpFile();
}
for (int i = 0; i < 4; i++) {
fetchMessage(msgChannel);
}
}
} catch (Exception e) {
throw new AssertionError("FTP test failed.", e);
} finally {
ctx.close();
cleanup();
}
}
boolean tswitch = true;
void storeFtpFile() throws IOException, InterruptedException {
File f = (tswitch ? ftpFile : ftpFile2);
tswitch = !tswitch;
log.info("Writing message " + f.getName());
Files.write(f.toPath(), ("Hello " + System.currentTimeMillis()).getBytes());
}
Message<?> fetchMessage(PollableChannel msgChannel) {
log.info("Fetching message.");
Message<?> msg = msgChannel.receive(1000L);
if (msg == null) {
log.info("No message.");
} else {
log.info("Have a message: " + msg);
}
return msg;
}
void cleanup() {
delFile(ftpFile);
delFile(ftpFile2);
File d = new File(FtpSyncConf.LOCAL_DIR);
if (d.isDirectory()) {
for (File f : d.listFiles()) {
delFile(f);
}
}
log.info("Finished cleanup");
}
void delFile(File f) {
if (f.isFile()) {
if (f.delete()) {
log.info("Deleted " + f);
} else {
log.error("Cannot delete file " + f);
}
}
}
}
public class MlistFtpSessionFactory extends AbstractFtpSessionFactory<MlistFtpClient> {
@Override
protected MlistFtpClient createClientInstance() {
return new MlistFtpClient();
}
}
public class MlistFtpClient extends FTPClient {
@Override
public FTPFile[] listFiles(String pathname) throws IOException {
return super.mlistDir(pathname);
}
}
@EnableIntegration
@Configuration
public class FtpSyncConf {
private static final Logger log = LoggerFactory.getLogger(FtpSyncConf.class);
public static final String LOCAL_DIR = "/tmp/received";
@Bean(name = "ftpMetaData")
public ConcurrentMetadataStore ftpMetaData() {
return new SimpleMetadataStore();
}
@Bean(name = "localMetaData")
public ConcurrentMetadataStore localMetaData() {
return new SimpleMetadataStore();
}
@Bean(name = "ftpFileSyncer")
public FtpUpdatingFileSynchronizer ftpFileSyncer(
@Qualifier("ftpMetaData") ConcurrentMetadataStore metadataStore) {
MlistFtpSessionFactory ftpSessionFactory = new MlistFtpSessionFactory();
ftpSessionFactory.setHost("localhost");
ftpSessionFactory.setPort(4444);
ftpSessionFactory.setUsername("demo");
ftpSessionFactory.setPassword("demo");
FtpPersistentAcceptOnceFileListFilter fileFilter = new FtpPersistentAcceptOnceFileListFilter(metadataStore, "ftp");
fileFilter.setFlushOnUpdate(true);
FtpUpdatingFileSynchronizer ftpFileSync = new FtpUpdatingFileSynchronizer(ftpSessionFactory);
ftpFileSync.setFilter(fileFilter);
// ftpFileSync.setDeleteRemoteFiles(true);
return ftpFileSync;
}
@Bean(name = "syncFtp")
@InboundChannelAdapter(value = "inputChannel", poller = @Poller(fixedDelay = "500", maxMessagesPerPoll = "1"))
public MessageSource<File> syncChannel(
@Qualifier("localMetaData") ConcurrentMetadataStore metadataStore,
@Qualifier("ftpFileSyncer") FtpUpdatingFileSynchronizer ftpFileSync) throws Exception {
FtpInboundFileSynchronizingMessageSource messageSource = new FtpInboundFileSynchronizingMessageSource(ftpFileSync);
File receiveDir = new File(LOCAL_DIR);
receiveDir.mkdirs();
messageSource.setLocalDirectory(receiveDir);
messageSource.setLocalFilter(new FileSystemPersistentAcceptOnceFileListFilter(metadataStore, "local"));
log.info("Message source bean created.");
return messageSource;
}
@Bean(name = "inputChannel")
public PollableChannel inputChannel() {
QueueChannel channel = new QueueChannel();
log.info("Message channel bean created.");
return channel;
}
}
/**
* Copied from https://github.com/spring-projects/spring-integration-samples/tree/master/basic/ftp/src/test/java/org/springframework/integration/samples/ftp/support
* @author Gunnar Hillert
*
*/
public class TestUserManager extends AbstractUserManager {
private BaseUser testUser;
private BaseUser anonUser;
private static final String TEST_USERNAME = "demo";
private static final String TEST_PASSWORD = "demo";
public TestUserManager(String homeDirectory) {
super("admin", new ClearTextPasswordEncryptor());
testUser = new BaseUser();
testUser.setAuthorities(Arrays.asList(new Authority[] {new ConcurrentLoginPermission(1, 1), new WritePermission()}));
testUser.setEnabled(true);
testUser.setHomeDirectory(homeDirectory);
testUser.setMaxIdleTime(10000);
testUser.setName(TEST_USERNAME);
testUser.setPassword(TEST_PASSWORD);
anonUser = new BaseUser(testUser);
anonUser.setName("anonymous");
}
public User getUserByName(String username) throws FtpException {
if(TEST_USERNAME.equals(username)) {
return testUser;
} else if(anonUser.getName().equals(username)) {
return anonUser;
}
return null;
}
public String[] getAllUserNames() throws FtpException {
return new String[] {TEST_USERNAME, anonUser.getName()};
}
public void delete(String username) throws FtpException {
throw new UnsupportedOperationException("Deleting of FTP Users is not supported.");
}
public void save(User user) throws FtpException {
throw new UnsupportedOperationException("Saving of FTP Users is not supported.");
}
public boolean doesExist(String username) throws FtpException {
return (TEST_USERNAME.equals(username) || anonUser.getName().equals(username)) ? true : false;
}
public User authenticate(Authentication authentication) throws AuthenticationFailedException {
if(UsernamePasswordAuthentication.class.isAssignableFrom(authentication.getClass())) {
UsernamePasswordAuthentication upAuth = (UsernamePasswordAuthentication) authentication;
if(TEST_USERNAME.equals(upAuth.getUsername()) && TEST_PASSWORD.equals(upAuth.getPassword())) {
return testUser;
}
if(anonUser.getName().equals(upAuth.getUsername())) {
return anonUser;
}
} else if(AnonymousAuthentication.class.isAssignableFrom(authentication.getClass())) {
return anonUser;
}
return null;
}
}
更新 15/Nov/2016: xml-配置注意事项。
xml-元素 inbound-channel-adapter 通过 org.springframework.integration.ftp.config.FtpInboundChannelAdapterParser 通过 FtpNamespaceHandler 通过 spring-integration-ftp-4.3.5.RELEASE.jar!/META-INF/spring.handlers 直接链接到 FtpInboundFileSynchronizer。
按照 xml-custom 参考指南，在本地 META-INF/spring.handlers 文件中指定自定义 FtpNamespaceHandler 应该允许您使用 FtpUpdatingFileSynchronizer 而不是 FtpInboundFileSynchronizer.虽然它对我的单元测试不起作用，但正确的解决方案可能涉及创建 extra/modified xsd 文件，以便常规 inbound-channel-adapter 使用常规 FtpInboundFileSynchronizer 和一个特殊的 inbound-updating-channel-adapter 正在使用 FtpUpdatingFileSynchronizer。正确地做到这一点有点超出了这个答案的范围。
不过，快速破解可以让您入门。您可以通过在本地项目中创建包 org.springframework.integration.ftp.config 和 class FtpNamespaceHandler 来覆盖默认值 FtpNamespaceHandler。内容如下：
package org.springframework.integration.ftp.config;
public class FtpNamespaceHandler extends org.springframework.integration.config.xml.AbstractIntegrationNamespaceHandler {
@Override
public void init() {
System.out.println("Initializing FTP updating file synchronizer.");
// one updated line below, rest copied from original FtpNamespaceHandler
registerBeanDefinitionParser("inbound-channel-adapter", new MyFtpInboundChannelAdapterParser());
registerBeanDefinitionParser("inbound-streaming-channel-adapter",
new FtpStreamingInboundChannelAdapterParser());
registerBeanDefinitionParser("outbound-channel-adapter", new FtpOutboundChannelAdapterParser());
registerBeanDefinitionParser("outbound-gateway", new FtpOutboundGatewayParser());
}
}
package org.springframework.integration.ftp.config;
import org.springframework.integration.file.remote.synchronizer.InboundFileSynchronizer;
import org.springframework.integration.ftp.config.FtpInboundChannelAdapterParser;
public class MyFtpInboundChannelAdapterParser extends FtpInboundChannelAdapterParser {
@Override
protected Class<? extends InboundFileSynchronizer> getInboundFileSynchronizerClass() {
System.out.println("Returning updating file synchronizer.");
return FtpUpdatingFileSynchronizer.class;
}
}
同时将 preserve-timestamp="true" 添加到 xml 文件以防止新的 IllegalArgumentException: for updating timestamps must be preserved.
我原以为 ftp 同步机制会更新已更改的文件。但是，从我在这里看到的情况来看，只有在文件尚不存在时才会下载该文件。至于现在，即使时间戳/内容发生变化，文件也不会保存在本地。
这是我目前的发现：
classorg.springframework.integration.file.remote.synchronizer.AbstractInboundFileSynchronizer
@Override
public void synchronizeToLocalDirectory(final File localDirectory) {
final String remoteDirectory = this.remoteDirectoryExpression.getValue(this.evaluationContext, String.class);
try {
int transferred = this.remoteFileTemplate.execute(new SessionCallback<F, Integer>() {
@Override
public Integer doInSession(Session<F> session) throws IOException {
F[] files = session.list(remoteDirectory);
if (!ObjectUtils.isEmpty(files)) {
List<F> filteredFiles = filterFiles(files);
for (F file : filteredFiles) {
try {
if (file != null) {
copyFileToLocalDirectory(
remoteDirectory, file, localDirectory,
session);
}
}
catch (RuntimeException e) {
if (AbstractInboundFileSynchronizer.this.filter instanceof ReversibleFileListFilter) {
((ReversibleFileListFilter<F>) AbstractInboundFileSynchronizer.this.filter)
.rollback(file, filteredFiles);
}
throw e;
}
catch (IOException e) {
if (AbstractInboundFileSynchronizer.this.filter instanceof ReversibleFileListFilter) {
((ReversibleFileListFilter<F>) AbstractInboundFileSynchronizer.this.filter)
.rollback(file, filteredFiles);
}
throw e;
}
}
return filteredFiles.size();
}
else {
return 0;
}
}
});
if (this.logger.isDebugEnabled()) {
this.logger.debug(transferred + " files transferred");
}
}
catch (Exception e) {
throw new MessagingException("Problem occurred while synchronizing remote to local directory", e);
}
}
过滤要下载的文件。我想使用 org.springframework.integration.ftp.filters.FtpPersistentAcceptOnceFileListFilter，它比较 文件名 和最后 修改日期 .
然后它会调用 copyFileToLocalDirectory 函数来过滤文件（要复制）。
protected void copyFileToLocalDirectory(String remoteDirectoryPath, F remoteFile, File localDirectory,
Session<F> session) throws IOException {
String remoteFileName = this.getFilename(remoteFile);
String localFileName = this.generateLocalFileName(remoteFileName);
String remoteFilePath = remoteDirectoryPath != null
? (remoteDirectoryPath + this.remoteFileSeparator + remoteFileName)
: remoteFileName;
if (!this.isFile(remoteFile)) {
if (this.logger.isDebugEnabled()) {
this.logger.debug("cannot copy, not a file: " + remoteFilePath);
}
return;
}
File localFile = new File(localDirectory, localFileName);
if (!localFile.exists()) {
String tempFileName = localFile.getAbsolutePath() + this.temporaryFileSuffix;
File tempFile = new File(tempFileName);
OutputStream outputStream = new BufferedOutputStream(new FileOutputStream(tempFile));
try {
session.read(remoteFilePath, outputStream);
}
catch (Exception e) {
if (e instanceof RuntimeException) {
throw (RuntimeException) e;
}
else {
throw new MessagingException("Failure occurred while copying from remote to local directory", e);
}
}
finally {
try {
outputStream.close();
}
catch (Exception ignored2) {
}
}
if (tempFile.renameTo(localFile)) {
if (this.deleteRemoteFiles) {
session.remove(remoteFilePath);
if (this.logger.isDebugEnabled()) {
this.logger.debug("deleted " + remoteFilePath);
}
}
}
if (this.preserveTimestamp) {
localFile.setLastModified(getModified(remoteFile));
}
}
}
但是，此方法会检查（仅基于文件名）文件是否已存在于本地磁盘上，并且仅在不存在时才下载。所以基本上没有机会下载带有（带有新时间戳）的更新文件。
我试过尝试更改 FtpInboundFileSynchronizer，但它变得太复杂了。 "customize" synchronize- / copyToLocalDirectory 方法的最佳方法是什么？
您可以在构造期间使用远程过滤器自定义 FtpInboundFileSynchonizer。
将过滤器设置为 FtpPersistentAcceptOnceFileListFilter 以配置基于时间戳更改的过滤。
设置后的远程过滤器将在synchronizeToLocalDirectory中用于过滤远程查看和下载文件。
这将在 doInSession 的这一行中使用
List<F> filteredFiles = filterFiles(files);
您还需要将本地过滤器设置为 FileSystemPersistentAcceptOnceFileListFilter 以根据远程文件更改相应地处理本地端更改。
此外，您需要提供您自己的元数据存储或使用 PropertiesPersistingMetadataStore 以在简单情况下持久保存本地和远程文件元数据更改，以便它可以在服务器重新启动后继续存在。
这里有更多 http://docs.spring.io/spring-integration/reference/html/ftp.html#ftp-inbound
直接的方法是extends AbstractInboundFileSynchronizerand override copyFileToLocalDirectory方法来比较你想要的条件比如修改时间，如果你不想这样做，那么你有另一种方法使用 AOP：
@Before("org.springframework.integration.file.remote.synchronizer.AbstractInboundFileSynchronizer.copyFileToLocalDirectory()")
然后从 joinPoint 获取参数并让你比较如果它是真的并且你想更新文件所以删除本地文件。
可以更新 AbstractInboundFileSynchronizer 以识别更新的文件，但它很脆弱，您 运行 会遇到其他问题。
更新 13/Nov/2016: 了解如何以秒为单位获取修改时间戳。
更新 AbstractInboundFileSynchronizer 的主要问题是它有 setter-方法但没有（受保护的）getter-方法。如果将来 setter 方法做一些聪明的事情，这里介绍的更新版本将会崩溃。
在本地目录中更新文件的主要问题是并发性：如果您在接收更新的同时处理本地文件，您可能 运行 陷入各种麻烦。简单的方法是将本地文件移动到（临时）处理目录，以便可以将更新作为新文件接收，从而无需更新 AbstractInboundFileSynchronizer。另见 Camel 的时间戳 remarks.
默认情况下 FTP 服务器以分钟为单位提供修改时间戳。为了进行测试，我更新了 FTP 客户端以使用 MLSD 命令，该命令以秒为单位提供修改时间戳（如果幸运的话，可以提供毫秒），但并非所有 FTP 服务器都支持此功能。
如 Spring FTP reference 中所述，本地文件过滤器需要 FileSystemPersistentAcceptOnceFileListFilter 以确保在修改时间戳更改时选择本地文件。
下面是我更新后的版本AbstractInboundFileSynchronizer，后面是我用过的一些测试classes。
public class FtpUpdatingFileSynchronizer extends FtpInboundFileSynchronizer {
protected final Log logger = LogFactory.getLog(this.getClass());
private volatile Expression localFilenameGeneratorExpression;
private volatile EvaluationContext evaluationContext;
private volatile boolean deleteRemoteFiles;
private volatile String remoteFileSeparator = "/";
private volatile boolean preserveTimestamp;
public FtpUpdatingFileSynchronizer(SessionFactory<FTPFile> sessionFactory) {
super(sessionFactory);
setPreserveTimestamp(true);
}
@Override
public void setLocalFilenameGeneratorExpression(Expression localFilenameGeneratorExpression) {
super.setLocalFilenameGeneratorExpression(localFilenameGeneratorExpression);
this.localFilenameGeneratorExpression = localFilenameGeneratorExpression;
}
@Override
public void setIntegrationEvaluationContext(EvaluationContext evaluationContext) {
super.setIntegrationEvaluationContext(evaluationContext);
this.evaluationContext = evaluationContext;
}
@Override
public void setDeleteRemoteFiles(boolean deleteRemoteFiles) {
super.setDeleteRemoteFiles(deleteRemoteFiles);
this.deleteRemoteFiles = deleteRemoteFiles;
}
@Override
public void setRemoteFileSeparator(String remoteFileSeparator) {
super.setRemoteFileSeparator(remoteFileSeparator);
this.remoteFileSeparator = remoteFileSeparator;
}
@Override
public void setPreserveTimestamp(boolean preserveTimestamp) {
// updated
Assert.isTrue(preserveTimestamp, "for updating timestamps must be preserved");
super.setPreserveTimestamp(preserveTimestamp);
this.preserveTimestamp = preserveTimestamp;
}
@Override
protected void copyFileToLocalDirectory(String remoteDirectoryPath, FTPFile remoteFile, File localDirectory,
Session<FTPFile> session) throws IOException {
String remoteFileName = this.getFilename(remoteFile);
String localFileName = this.generateLocalFileName(remoteFileName);
String remoteFilePath = (remoteDirectoryPath != null
? (remoteDirectoryPath + this.remoteFileSeparator + remoteFileName)
: remoteFileName);
if (!this.isFile(remoteFile)) {
if (this.logger.isDebugEnabled()) {
this.logger.debug("cannot copy, not a file: " + remoteFilePath);
}
return;
}
// start update
File localFile = new File(localDirectory, localFileName);
boolean update = false;
if (localFile.exists()) {
if (this.getModified(remoteFile) > localFile.lastModified()) {
this.logger.info("Updating local file " + localFile);
update = true;
} else {
this.logger.info("File already exists: " + localFile);
return;
}
}
// end update
String tempFileName = localFile.getAbsolutePath() + this.getTemporaryFileSuffix();
File tempFile = new File(tempFileName);
OutputStream outputStream = new BufferedOutputStream(new FileOutputStream(tempFile));
try {
session.read(remoteFilePath, outputStream);
} catch (Exception e) {
if (e instanceof RuntimeException) {
throw (RuntimeException) e;
}
else {
throw new MessagingException("Failure occurred while copying from remote to local directory", e);
}
} finally {
try {
outputStream.close();
}
catch (Exception ignored2) {
}
}
// updated
if (update && !localFile.delete()) {
throw new MessagingException("Unable to delete local file [" + localFile + "] for update.");
}
if (tempFile.renameTo(localFile)) {
if (this.deleteRemoteFiles) {
session.remove(remoteFilePath);
if (this.logger.isDebugEnabled()) {
this.logger.debug("deleted " + remoteFilePath);
}
}
// updated
this.logger.info("Stored file locally: " + localFile);
} else {
// updated
throw new MessagingException("Unable to rename temporary file [" + tempFile + "] to [" + localFile + "]");
}
if (this.preserveTimestamp) {
localFile.setLastModified(getModified(remoteFile));
}
}
private String generateLocalFileName(String remoteFileName) {
if (this.localFilenameGeneratorExpression != null) {
return this.localFilenameGeneratorExpression.getValue(this.evaluationContext, remoteFileName, String.class);
}
return remoteFileName;
}
}
根据我使用的一些测试 classes。
我使用了依赖项 org.springframework.integration:spring-integration-ftp:4.3.5.RELEASE 和 org.apache.ftpserver:ftpserver-core:1.0.6（加上通常的日志记录和测试依赖项）。
public class TestFtpSync {
static final Logger log = LoggerFactory.getLogger(TestFtpSync.class);
static final String FTP_ROOT_DIR = "target" + File.separator + "ftproot";
// org.apache.ftpserver:ftpserver-core:1.0.6
static FtpServer server;
@BeforeClass
public static void startServer() throws FtpException {
File ftpRoot = new File (FTP_ROOT_DIR);
ftpRoot.mkdirs();
TestUserManager userManager = new TestUserManager(ftpRoot.getAbsolutePath());
FtpServerFactory serverFactory = new FtpServerFactory();
serverFactory.setUserManager(userManager);
ListenerFactory factory = new ListenerFactory();
factory.setPort(4444);
serverFactory.addListener("default", factory.createListener());
server = serverFactory.createServer();
server.start();
}
@AfterClass
public static void stopServer() {
if (server != null) {
server.stop();
}
}
File ftpFile = Paths.get(FTP_ROOT_DIR, "test1.txt").toFile();
File ftpFile2 = Paths.get(FTP_ROOT_DIR, "test2.txt").toFile();
@Test
public void syncDir() {
// org.springframework.integration:spring-integration-ftp:4.3.5.RELEASE
AnnotationConfigApplicationContext ctx = new AnnotationConfigApplicationContext();
try {
ctx.register(FtpSyncConf.class);
ctx.refresh();
PollableChannel msgChannel = ctx.getBean("inputChannel", PollableChannel.class);
for (int j = 0; j < 2; j++) {
for (int i = 0; i < 2; i++) {
storeFtpFile();
}
for (int i = 0; i < 4; i++) {
fetchMessage(msgChannel);
}
}
} catch (Exception e) {
throw new AssertionError("FTP test failed.", e);
} finally {
ctx.close();
cleanup();
}
}
boolean tswitch = true;
void storeFtpFile() throws IOException, InterruptedException {
File f = (tswitch ? ftpFile : ftpFile2);
tswitch = !tswitch;
log.info("Writing message " + f.getName());
Files.write(f.toPath(), ("Hello " + System.currentTimeMillis()).getBytes());
}
Message<?> fetchMessage(PollableChannel msgChannel) {
log.info("Fetching message.");
Message<?> msg = msgChannel.receive(1000L);
if (msg == null) {
log.info("No message.");
} else {
log.info("Have a message: " + msg);
}
return msg;
}
void cleanup() {
delFile(ftpFile);
delFile(ftpFile2);
File d = new File(FtpSyncConf.LOCAL_DIR);
if (d.isDirectory()) {
for (File f : d.listFiles()) {
delFile(f);
}
}
log.info("Finished cleanup");
}
void delFile(File f) {
if (f.isFile()) {
if (f.delete()) {
log.info("Deleted " + f);
} else {
log.error("Cannot delete file " + f);
}
}
}
}
public class MlistFtpSessionFactory extends AbstractFtpSessionFactory<MlistFtpClient> {
@Override
protected MlistFtpClient createClientInstance() {
return new MlistFtpClient();
}
}
public class MlistFtpClient extends FTPClient {
@Override
public FTPFile[] listFiles(String pathname) throws IOException {
return super.mlistDir(pathname);
}
}
@EnableIntegration
@Configuration
public class FtpSyncConf {
private static final Logger log = LoggerFactory.getLogger(FtpSyncConf.class);
public static final String LOCAL_DIR = "/tmp/received";
@Bean(name = "ftpMetaData")
public ConcurrentMetadataStore ftpMetaData() {
return new SimpleMetadataStore();
}
@Bean(name = "localMetaData")
public ConcurrentMetadataStore localMetaData() {
return new SimpleMetadataStore();
}
@Bean(name = "ftpFileSyncer")
public FtpUpdatingFileSynchronizer ftpFileSyncer(
@Qualifier("ftpMetaData") ConcurrentMetadataStore metadataStore) {
MlistFtpSessionFactory ftpSessionFactory = new MlistFtpSessionFactory();
ftpSessionFactory.setHost("localhost");
ftpSessionFactory.setPort(4444);
ftpSessionFactory.setUsername("demo");
ftpSessionFactory.setPassword("demo");
FtpPersistentAcceptOnceFileListFilter fileFilter = new FtpPersistentAcceptOnceFileListFilter(metadataStore, "ftp");
fileFilter.setFlushOnUpdate(true);
FtpUpdatingFileSynchronizer ftpFileSync = new FtpUpdatingFileSynchronizer(ftpSessionFactory);
ftpFileSync.setFilter(fileFilter);
// ftpFileSync.setDeleteRemoteFiles(true);
return ftpFileSync;
}
@Bean(name = "syncFtp")
@InboundChannelAdapter(value = "inputChannel", poller = @Poller(fixedDelay = "500", maxMessagesPerPoll = "1"))
public MessageSource<File> syncChannel(
@Qualifier("localMetaData") ConcurrentMetadataStore metadataStore,
@Qualifier("ftpFileSyncer") FtpUpdatingFileSynchronizer ftpFileSync) throws Exception {
FtpInboundFileSynchronizingMessageSource messageSource = new FtpInboundFileSynchronizingMessageSource(ftpFileSync);
File receiveDir = new File(LOCAL_DIR);
receiveDir.mkdirs();
messageSource.setLocalDirectory(receiveDir);
messageSource.setLocalFilter(new FileSystemPersistentAcceptOnceFileListFilter(metadataStore, "local"));
log.info("Message source bean created.");
return messageSource;
}
@Bean(name = "inputChannel")
public PollableChannel inputChannel() {
QueueChannel channel = new QueueChannel();
log.info("Message channel bean created.");
return channel;
}
}
/**
* Copied from https://github.com/spring-projects/spring-integration-samples/tree/master/basic/ftp/src/test/java/org/springframework/integration/samples/ftp/support
* @author Gunnar Hillert
*
*/
public class TestUserManager extends AbstractUserManager {
private BaseUser testUser;
private BaseUser anonUser;
private static final String TEST_USERNAME = "demo";
private static final String TEST_PASSWORD = "demo";
public TestUserManager(String homeDirectory) {
super("admin", new ClearTextPasswordEncryptor());
testUser = new BaseUser();
testUser.setAuthorities(Arrays.asList(new Authority[] {new ConcurrentLoginPermission(1, 1), new WritePermission()}));
testUser.setEnabled(true);
testUser.setHomeDirectory(homeDirectory);
testUser.setMaxIdleTime(10000);
testUser.setName(TEST_USERNAME);
testUser.setPassword(TEST_PASSWORD);
anonUser = new BaseUser(testUser);
anonUser.setName("anonymous");
}
public User getUserByName(String username) throws FtpException {
if(TEST_USERNAME.equals(username)) {
return testUser;
} else if(anonUser.getName().equals(username)) {
return anonUser;
}
return null;
}
public String[] getAllUserNames() throws FtpException {
return new String[] {TEST_USERNAME, anonUser.getName()};
}
public void delete(String username) throws FtpException {
throw new UnsupportedOperationException("Deleting of FTP Users is not supported.");
}
public void save(User user) throws FtpException {
throw new UnsupportedOperationException("Saving of FTP Users is not supported.");
}
public boolean doesExist(String username) throws FtpException {
return (TEST_USERNAME.equals(username) || anonUser.getName().equals(username)) ? true : false;
}
public User authenticate(Authentication authentication) throws AuthenticationFailedException {
if(UsernamePasswordAuthentication.class.isAssignableFrom(authentication.getClass())) {
UsernamePasswordAuthentication upAuth = (UsernamePasswordAuthentication) authentication;
if(TEST_USERNAME.equals(upAuth.getUsername()) && TEST_PASSWORD.equals(upAuth.getPassword())) {
return testUser;
}
if(anonUser.getName().equals(upAuth.getUsername())) {
return anonUser;
}
} else if(AnonymousAuthentication.class.isAssignableFrom(authentication.getClass())) {
return anonUser;
}
return null;
}
}
更新 15/Nov/2016: xml-配置注意事项。
xml-元素 inbound-channel-adapter 通过 org.springframework.integration.ftp.config.FtpInboundChannelAdapterParser 通过 FtpNamespaceHandler 通过 spring-integration-ftp-4.3.5.RELEASE.jar!/META-INF/spring.handlers 直接链接到 FtpInboundFileSynchronizer。
按照 xml-custom 参考指南，在本地 META-INF/spring.handlers 文件中指定自定义 FtpNamespaceHandler 应该允许您使用 FtpUpdatingFileSynchronizer 而不是 FtpInboundFileSynchronizer.虽然它对我的单元测试不起作用，但正确的解决方案可能涉及创建 extra/modified xsd 文件，以便常规 inbound-channel-adapter 使用常规 FtpInboundFileSynchronizer 和一个特殊的 inbound-updating-channel-adapter 正在使用 FtpUpdatingFileSynchronizer。正确地做到这一点有点超出了这个答案的范围。
不过，快速破解可以让您入门。您可以通过在本地项目中创建包 org.springframework.integration.ftp.config 和 class FtpNamespaceHandler 来覆盖默认值 FtpNamespaceHandler。内容如下：
package org.springframework.integration.ftp.config;
public class FtpNamespaceHandler extends org.springframework.integration.config.xml.AbstractIntegrationNamespaceHandler {
@Override
public void init() {
System.out.println("Initializing FTP updating file synchronizer.");
// one updated line below, rest copied from original FtpNamespaceHandler
registerBeanDefinitionParser("inbound-channel-adapter", new MyFtpInboundChannelAdapterParser());
registerBeanDefinitionParser("inbound-streaming-channel-adapter",
new FtpStreamingInboundChannelAdapterParser());
registerBeanDefinitionParser("outbound-channel-adapter", new FtpOutboundChannelAdapterParser());
registerBeanDefinitionParser("outbound-gateway", new FtpOutboundGatewayParser());
}
}
package org.springframework.integration.ftp.config;
import org.springframework.integration.file.remote.synchronizer.InboundFileSynchronizer;
import org.springframework.integration.ftp.config.FtpInboundChannelAdapterParser;
public class MyFtpInboundChannelAdapterParser extends FtpInboundChannelAdapterParser {
@Override
protected Class<? extends InboundFileSynchronizer> getInboundFileSynchronizerClass() {
System.out.println("Returning updating file synchronizer.");
return FtpUpdatingFileSynchronizer.class;
}
}
同时将 preserve-timestamp="true" 添加到 xml 文件以防止新的 IllegalArgumentException: for updating timestamps must be preserved.