diff --git a/spring-integration-core/src/main/java/org/springframework/integration/metadata/PropertiesPersistingMetadataStore.java b/spring-integration-core/src/main/java/org/springframework/integration/metadata/PropertiesPersistingMetadataStore.java index 89a0ac6dc9..8a6d81f4c3 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/metadata/PropertiesPersistingMetadataStore.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/metadata/PropertiesPersistingMetadataStore.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2014 the original author or authors. + * Copyright 2002-2015 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -18,9 +18,11 @@ package org.springframework.integration.metadata; import java.io.BufferedInputStream; import java.io.BufferedOutputStream; +import java.io.Closeable; import java.io.File; import java.io.FileInputStream; import java.io.FileOutputStream; +import java.io.Flushable; import java.io.IOException; import java.io.InputStream; import java.io.OutputStream; @@ -48,7 +50,8 @@ import org.springframework.util.DefaultPropertiesPersister; * @author Gary Russell * @since 2.0 */ -public class PropertiesPersistingMetadataStore implements ConcurrentMetadataStore, InitializingBean, DisposableBean { +public class PropertiesPersistingMetadataStore implements ConcurrentMetadataStore, InitializingBean, DisposableBean, + Closeable, Flushable { private final Log logger = LogFactory.getLog(getClass()); @@ -62,6 +65,8 @@ public class PropertiesPersistingMetadataStore implements ConcurrentMetadataStor private File file; + private volatile boolean dirty; + public void setBaseDirectory(String baseDirectory) { Assert.hasText(baseDirectory, "'baseDirectory' must be non-empty"); @@ -95,6 +100,7 @@ public class PropertiesPersistingMetadataStore implements ConcurrentMetadataStor this.metadata.setProperty(key, value); } finally { + this.dirty = true; lock.unlock(); } } @@ -121,6 +127,7 @@ public class PropertiesPersistingMetadataStore implements ConcurrentMetadataStor return (String) this.metadata.remove(key); } finally { + this.dirty = true; lock.unlock(); } } @@ -135,6 +142,7 @@ public class PropertiesPersistingMetadataStore implements ConcurrentMetadataStor String property = this.metadata.getProperty(key); if (property == null) { this.metadata.setProperty(key, value); + this.dirty = true; return null; } else { @@ -157,6 +165,7 @@ public class PropertiesPersistingMetadataStore implements ConcurrentMetadataStor String property = this.metadata.getProperty(key); if (oldValue.equals(property)) { this.metadata.setProperty(key, newValue); + this.dirty = true; return true; } else { @@ -168,12 +177,26 @@ public class PropertiesPersistingMetadataStore implements ConcurrentMetadataStor } } + @Override + public void close() throws IOException { + flush(); + } + + @Override + public void flush() { + saveMetadata(); + } + @Override public void destroy() throws Exception { - this.saveMetadata(); + flush(); } private void saveMetadata() { + if (this.file == null || !this.dirty) { + return; + } + this.dirty = false; OutputStream outputStream = null; try { outputStream = new BufferedOutputStream(new FileOutputStream(this.file)); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/metadata/PropertiesPersistingMetadataStoreTests.java b/spring-integration-core/src/test/java/org/springframework/integration/metadata/PropertiesPersistingMetadataStoreTests.java index 9437284663..c9ed5556f5 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/metadata/PropertiesPersistingMetadataStoreTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/metadata/PropertiesPersistingMetadataStoreTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2014 the original author or authors. + * Copyright 2002-2015 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -50,7 +50,7 @@ public class PropertiesPersistingMetadataStoreTests { assertNotNull(metadataStore.putIfAbsent("foo", "baz")); assertFalse(metadataStore.replace("foo", "xxx", "bar")); assertTrue(metadataStore.replace("foo", "baz", "bar")); - metadataStore.destroy(); + metadataStore.close(); Properties persistentProperties = PropertiesLoaderUtils.loadProperties(new FileSystemResource(file)); assertNotNull(persistentProperties); assertEquals(1, persistentProperties.size()); @@ -66,7 +66,7 @@ public class PropertiesPersistingMetadataStoreTests { metadataStore.setBaseDirectory("target/foo"); metadataStore.afterPropertiesSet(); metadataStore.put("foo", "bar"); - metadataStore.destroy(); + metadataStore.close(); assertTrue(file.exists()); Properties persistentProperties = PropertiesLoaderUtils.loadProperties(new FileSystemResource(file)); assertNotNull(persistentProperties); 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 7c0432303a..d07354ac12 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 @@ -1,5 +1,5 @@ /* - * Copyright 2013-2014 the original author or authors. + * Copyright 2013-2015 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -16,6 +16,8 @@ package org.springframework.integration.file.filters; +import java.io.Closeable; +import java.io.IOException; import java.util.List; import org.springframework.integration.metadata.ConcurrentMetadataStore; @@ -32,7 +34,7 @@ import org.springframework.util.Assert; * */ public abstract class AbstractPersistentAcceptOnceFileListFilter extends AbstractFileListFilter - implements ReversibleFileListFilter { + implements ReversibleFileListFilter, Closeable { protected final ConcurrentMetadataStore store; @@ -78,6 +80,13 @@ public abstract class AbstractPersistentAcceptOnceFileListFilter extends Abst } } + @Override + public void close() throws IOException { + if (this.store instanceof Closeable) { + ((Closeable) this.store).close(); + } + } + /** * The default value stored for the key is the last modified date. * @param file The file. diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/filters/CompositeFileListFilter.java b/spring-integration-file/src/main/java/org/springframework/integration/file/filters/CompositeFileListFilter.java index 3ddb686f94..7bfd635b3e 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/filters/CompositeFileListFilter.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/filters/CompositeFileListFilter.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2014 the original author or authors. + * Copyright 2002-2015 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -16,6 +16,8 @@ package org.springframework.integration.file.filters; +import java.io.Closeable; +import java.io.IOException; import java.util.ArrayList; import java.util.Arrays; import java.util.Collection; @@ -37,7 +39,7 @@ import org.springframework.util.Assert; * * @param The type that will be filtered. */ -public class CompositeFileListFilter implements FileListFilter { +public class CompositeFileListFilter implements FileListFilter, Closeable { private final Set> fileFilters; @@ -51,6 +53,15 @@ public class CompositeFileListFilter implements FileListFilter { } + @Override + public void close() throws IOException { + for (FileListFilter filter : this.fileFilters) { + if (filter instanceof Closeable) { + ((Closeable) filter).close(); + } + } + } + public CompositeFileListFilter addFilter(FileListFilter filter) { return this.addFilters(Collections.singletonList(filter)); } diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/synchronizer/AbstractInboundFileSynchronizer.java b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/synchronizer/AbstractInboundFileSynchronizer.java index e7297eab3b..c38e1c15be 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/synchronizer/AbstractInboundFileSynchronizer.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/synchronizer/AbstractInboundFileSynchronizer.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2014 the original author or authors. + * Copyright 2002-2015 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -17,6 +17,7 @@ package org.springframework.integration.file.remote.synchronizer; import java.io.BufferedOutputStream; +import java.io.Closeable; import java.io.File; import java.io.FileOutputStream; import java.io.IOException; @@ -57,7 +58,7 @@ import org.springframework.util.ObjectUtils; * @since 2.0 */ public abstract class AbstractInboundFileSynchronizer implements InboundFileSynchronizer, - InitializingBean, IntegrationEvaluationContextAware { + InitializingBean, IntegrationEvaluationContextAware, Closeable { protected final Log logger = LogFactory.getLog(this.getClass()); @@ -161,6 +162,13 @@ public abstract class AbstractInboundFileSynchronizer implements InboundFileS return temporaryFileSuffix; } + @Override + public void close() throws IOException { + if (this.filter instanceof Closeable) { + ((Closeable) this.filter).close(); + } + } + @Override public void synchronizeToLocalDirectory(final File localDirectory) { try { 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 deaaf34335..1bac6afa60 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 @@ -1,5 +1,5 @@ /* - * Copyright 2002-2014 the original author or authors. + * Copyright 2002-2015 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -16,12 +16,15 @@ package org.springframework.integration.file.remote.synchronizer; +import java.io.Closeable; import java.io.File; import java.io.FileNotFoundException; +import java.io.IOException; import java.util.Arrays; import java.util.Comparator; import java.util.regex.Pattern; +import org.springframework.integration.core.LifecycleMessageSource; import org.springframework.integration.endpoint.AbstractMessageSource; import org.springframework.integration.file.FileReadingMessageSource; import org.springframework.integration.file.filters.AcceptOnceFileListFilter; @@ -54,7 +57,10 @@ import org.springframework.util.Assert; * @author Oleg Zhurakousky * @author Gary Russell */ -public abstract class AbstractInboundFileSynchronizingMessageSource extends AbstractMessageSource { +public abstract class AbstractInboundFileSynchronizingMessageSource extends AbstractMessageSource + implements LifecycleMessageSource { + + private volatile boolean running; /** * Should the endpoint attempt to create the local directory? True by default. @@ -153,6 +159,29 @@ public abstract class AbstractInboundFileSynchronizingMessageSource extends A } } + @Override + public void start() { + this.running = true; + } + + @Override + public void stop() { + this.running = false; + if (this.synchronizer instanceof Closeable) { + try { + ((Closeable) this.synchronizer).close(); + } + catch (IOException e) { + logger.error("Error closing synchronizer", e); + } + } + } + + @Override + public boolean isRunning() { + return this.running; + } + /** * Polls from the file source. If the result is not null, it will be returned. * If the result is null, it attempts to sync up with the remote directory to populate the file source. diff --git a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/FtpInboundRemoteFileSystemSynchronizerTests.java b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/FtpInboundRemoteFileSystemSynchronizerTests.java index f244fde3fb..bfac31414c 100644 --- a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/FtpInboundRemoteFileSystemSynchronizerTests.java +++ b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/FtpInboundRemoteFileSystemSynchronizerTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2014 the original author or authors. + * Copyright 2002-2015 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -49,8 +49,10 @@ import org.springframework.expression.ExpressionParser; import org.springframework.expression.spel.SpelParserConfiguration; import org.springframework.expression.spel.standard.SpelExpressionParser; import org.springframework.integration.expression.ExpressionUtils; +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.RegexPatternFileListFilter; import org.springframework.integration.ftp.filters.FtpPersistentAcceptOnceFileListFilter; import org.springframework.integration.ftp.filters.FtpRegexPatternFileListFilter; import org.springframework.integration.ftp.session.AbstractFtpSessionFactory; @@ -96,8 +98,9 @@ public class FtpInboundRemoteFileSystemSynchronizerTests { synchronizer.setPreserveTimestamp(true); synchronizer.setRemoteDirectory("remote-test-dir"); FtpRegexPatternFileListFilter patternFilter = new FtpRegexPatternFileListFilter(".*\\.test$"); - PropertiesPersistingMetadataStore store = new PropertiesPersistingMetadataStore(); + PropertiesPersistingMetadataStore store = spy(new PropertiesPersistingMetadataStore()); store.setBaseDirectory("test"); + store.afterPropertiesSet(); FtpPersistentAcceptOnceFileListFilter persistFilter = new FtpPersistentAcceptOnceFileListFilter(store, "foo"); List> filters = new ArrayList>(); @@ -117,6 +120,11 @@ public class FtpInboundRemoteFileSystemSynchronizerTests { ms.setLocalDirectory(localDirectoy); ms.setBeanFactory(mock(BeanFactory.class)); + CompositeFileListFilter localFileListFilter = new CompositeFileListFilter(); + localFileListFilter.addFilter(new RegexPatternFileListFilter(".*\\.TEST\\.a$")); + AcceptOnceFileListFilter localAcceptOnceFilter = new AcceptOnceFileListFilter(); + localFileListFilter.addFilter(localAcceptOnceFilter); + ms.setLocalFilter(localFileListFilter); ms.afterPropertiesSet(); Message atestFile = ms.receive(); assertNotNull(atestFile); @@ -139,7 +147,7 @@ public class FtpInboundRemoteFileSystemSynchronizerTests { assertTrue(new File("test/A.TEST.a").exists()); assertTrue(new File("test/B.TEST.a").exists()); - TestUtils.getPropertyValue(ms, "localFileListFilter.seenSet", Collection.class).clear(); + TestUtils.getPropertyValue(localAcceptOnceFilter, "seenSet", Collection.class).clear(); new File("test/A.TEST.a").delete(); new File("test/B.TEST.a").delete(); @@ -147,6 +155,9 @@ public class FtpInboundRemoteFileSystemSynchronizerTests { nothing = ms.receive(); assertNull(nothing); + ms.stop(); + verify(synchronizer).close(); + verify(store).close(); } diff --git a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/SftpInboundRemoteFileSystemSynchronizerTests.java b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/SftpInboundRemoteFileSystemSynchronizerTests.java index d4aff7abba..3b8e7b9d5f 100644 --- a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/SftpInboundRemoteFileSystemSynchronizerTests.java +++ b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/SftpInboundRemoteFileSystemSynchronizerTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2014 the original author or authors. + * Copyright 2002-2015 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -43,8 +43,10 @@ import org.junit.Test; import org.springframework.beans.factory.BeanFactory; import org.springframework.integration.expression.ExpressionUtils; +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.RegexPatternFileListFilter; import org.springframework.integration.metadata.PropertiesPersistingMetadataStore; import org.springframework.integration.sftp.filters.SftpPersistentAcceptOnceFileListFilter; import org.springframework.integration.sftp.filters.SftpRegexPatternFileListFilter; @@ -97,8 +99,9 @@ public class SftpInboundRemoteFileSystemSynchronizerTests { synchronizer.setPreserveTimestamp(true); synchronizer.setRemoteDirectory("remote-test-dir"); SftpRegexPatternFileListFilter patternFilter = new SftpRegexPatternFileListFilter(".*\\.test$"); - PropertiesPersistingMetadataStore store = new PropertiesPersistingMetadataStore(); + PropertiesPersistingMetadataStore store = spy(new PropertiesPersistingMetadataStore()); store.setBaseDirectory("test"); + store.afterPropertiesSet(); SftpPersistentAcceptOnceFileListFilter persistFilter = new SftpPersistentAcceptOnceFileListFilter(store, "foo"); List> filters = new ArrayList>(); @@ -107,11 +110,15 @@ public class SftpInboundRemoteFileSystemSynchronizerTests { CompositeFileListFilter filter = new CompositeFileListFilter(filters); synchronizer.setFilter(filter); synchronizer.setIntegrationEvaluationContext(ExpressionUtils.createStandardEvaluationContext()); - SftpInboundFileSynchronizingMessageSource ms = new SftpInboundFileSynchronizingMessageSource(synchronizer); ms.setAutoCreateLocalDirectory(true); ms.setLocalDirectory(localDirectoy); ms.setBeanFactory(mock(BeanFactory.class)); + CompositeFileListFilter localFileListFilter = new CompositeFileListFilter(); + localFileListFilter.addFilter(new RegexPatternFileListFilter(".*\\.test$")); + AcceptOnceFileListFilter localAcceptOnceFilter = new AcceptOnceFileListFilter(); + localFileListFilter.addFilter(localAcceptOnceFilter); + ms.setLocalFilter(localFileListFilter); ms.afterPropertiesSet(); Message atestFile = ms.receive(); assertNotNull(atestFile); @@ -134,7 +141,7 @@ public class SftpInboundRemoteFileSystemSynchronizerTests { assertTrue(new File("test/a.test").exists()); assertTrue(new File("test/b.test").exists()); - TestUtils.getPropertyValue(ms, "localFileListFilter.seenSet", Collection.class).clear(); + TestUtils.getPropertyValue(localAcceptOnceFilter, "seenSet", Collection.class).clear(); new File("test/a.test").delete(); new File("test/b.test").delete(); @@ -142,6 +149,9 @@ public class SftpInboundRemoteFileSystemSynchronizerTests { nothing = ms.receive(); assertNull(nothing); + ms.stop(); + verify(synchronizer).close(); + verify(store).close(); } public static class TestSftpSessionFactory extends DefaultSftpSessionFactory {