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

INT-3606: Implement Closeable/Flushable

Browse Source 
JIRA: https://jira.spring.io/browse/INT-3606

Close the store when an (S)FTP synchronizing MessageSource is stopped.

Will need another commit on 4.2.WIP to change to `Lifecycle`.

Add dirty Flag to Properties MetadataStore

Avoid unnecessary persists.

Remove unnecessary implementation of `AbstractInboundFileSynchronizingMessageSource#getComponentType()`
This commit is contained in:
Gary Russell
2015-01-13 15:35:43 +02:00
committed by Artem Bilan
parent 8896d500ed
commit 2581388cc2
8 changed files with 122 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files

29
spring-integration-core/src/main/java/org/springframework/integration/metadata/PropertiesPersistingMetadataStore.java
View File

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

6
spring-integration-core/src/test/java/org/springframework/integration/metadata/PropertiesPersistingMetadataStoreTests.java
View File

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

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

@@ -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<F> extends AbstractFileListFilter<F>
implements ReversibleFileListFilter<F> {
implements ReversibleFileListFilter<F>, Closeable {
protected final ConcurrentMetadataStore store;
@@ -78,6 +80,13 @@ public abstract class AbstractPersistentAcceptOnceFileListFilter<F> 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.

15
spring-integration-file/src/main/java/org/springframework/integration/file/filters/CompositeFileListFilter.java
View File

@@ -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 <F> The type that will be filtered.
*/
public class CompositeFileListFilter<F> implements FileListFilter<F> {
public class CompositeFileListFilter<F> implements FileListFilter<F>, Closeable {
private final Set<FileListFilter<F>> fileFilters;
@@ -51,6 +53,15 @@ public class CompositeFileListFilter<F> implements FileListFilter<F> {
}
@Override
public void close() throws IOException {
for (FileListFilter<F> filter : this.fileFilters) {
if (filter instanceof Closeable) {
((Closeable) filter).close();
}
}
}
public CompositeFileListFilter<F> addFilter(FileListFilter<F> filter) {
return this.addFilters(Collections.singletonList(filter));
}

12
spring-integration-file/src/main/java/org/springframework/integration/file/remote/synchronizer/AbstractInboundFileSynchronizer.java
View File

@@ -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<F> implements InboundFileSynchronizer,
InitializingBean, IntegrationEvaluationContextAware {
InitializingBean, IntegrationEvaluationContextAware, Closeable {
protected final Log logger = LogFactory.getLog(this.getClass());
@@ -161,6 +162,13 @@ public abstract class AbstractInboundFileSynchronizer<F> 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 {

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

@@ -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<F> extends AbstractMessageSource<File> {
public abstract class AbstractInboundFileSynchronizingMessageSource<F> extends AbstractMessageSource<File>
implements LifecycleMessageSource<File> {
private volatile boolean running;
/**
* Should the endpoint attempt to create the local directory? True by default.
@@ -153,6 +159,29 @@ public abstract class AbstractInboundFileSynchronizingMessageSource<F> 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.

17
spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/FtpInboundRemoteFileSystemSynchronizerTests.java
View File

@@ -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<FileListFilter<FTPFile>> filters = new ArrayList<FileListFilter<FTPFile>>();
@@ -117,6 +120,11 @@ public class FtpInboundRemoteFileSystemSynchronizerTests {
ms.setLocalDirectory(localDirectoy);
ms.setBeanFactory(mock(BeanFactory.class));
CompositeFileListFilter<File> localFileListFilter = new CompositeFileListFilter<File>();
localFileListFilter.addFilter(new RegexPatternFileListFilter(".*\\.TEST\\.a$"));
AcceptOnceFileListFilter<File> localAcceptOnceFilter = new AcceptOnceFileListFilter<File>();
localFileListFilter.addFilter(localAcceptOnceFilter);
ms.setLocalFilter(localFileListFilter);
ms.afterPropertiesSet();
Message<File> 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();
}

18
spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/SftpInboundRemoteFileSystemSynchronizerTests.java
View File

@@ -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<FileListFilter<LsEntry>> filters = new ArrayList<FileListFilter<LsEntry>>();
@@ -107,11 +110,15 @@ public class SftpInboundRemoteFileSystemSynchronizerTests {
CompositeFileListFilter<LsEntry> filter = new CompositeFileListFilter<LsEntry>(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<File> localFileListFilter = new CompositeFileListFilter<File>();
localFileListFilter.addFilter(new RegexPatternFileListFilter(".*\\.test$"));
AcceptOnceFileListFilter<File> localAcceptOnceFilter = new AcceptOnceFileListFilter<File>();
localFileListFilter.addFilter(localAcceptOnceFilter);
ms.setLocalFilter(localFileListFilter);
ms.afterPropertiesSet();
Message<File> 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 {