List only enhancements

* Add metadata.store.type Property and idempotent SftpSupplier for list-only * Implement list-only for S3 source and Optimize metadastore access * Fixed build and READMEs * Change to ConditionalOnProperty * Change to ReactiveMessageProducer * Update cdc-debezium-source/README.adoc * Make all MetadataStoreProperties visible
2020-09-17 15:36:36 -04:00
parent 65782ab8d0
commit 1a593fc517
11 changed files with 383 additions and 114 deletions
--- a/common/metadata-store-common/README.adoc
+++ b/common/metadata-store-common/README.adoc
@@ -5,7 +5,7 @@ See Spring Integration "`https://docs.spring.io/spring-integration/docs/5.0.6.RE

 In addition to the standard Spring Boot configuration properties this module exposes a `MetadataStoreProperties` with the `metadata.store` prefix.

-To auto-configure particular `MetadataStore` you just need to bring respective dependencies into the target app starter:
+To auto-configure particular `MetadataStore` you need to set `metadata.store.type` and include the respective dependencies into the target app starter:

 ==== Redis

--- a/common/metadata-store-common/pom.xml
+++ b/common/metadata-store-common/pom.xml
@@ -31,6 +31,12 @@
            <artifactId>spring-boot-starter</artifactId>
        </dependency>

+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-configuration-processor</artifactId>
+            <scope>provided</scope>
+        </dependency>
+
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
--- a/common/metadata-store-common/src/main/java/org/springframework/cloud/fn/common/metadata/store/MetadataStoreAutoConfiguration.java
+++ b/common/metadata-store-common/src/main/java/org/springframework/cloud/fn/common/metadata/store/MetadataStoreAutoConfiguration.java
@@ -27,9 +27,9 @@ import org.apache.geode.cache.GemFireCache;
 import org.apache.geode.cache.Region;

 import org.springframework.beans.factory.ObjectProvider;
-import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
 import org.springframework.cloud.aws.core.region.RegionProvider;
 import org.springframework.context.annotation.Bean;
@@ -52,6 +52,7 @@ import org.springframework.jdbc.core.JdbcTemplate;

 /**
 * @author Artem Bilan
+ * @author David Turanski
 * @since 2.0.2
 */
@Configuration
@@ -60,47 +61,46 @@ import org.springframework.jdbc.core.JdbcTemplate;
 public class MetadataStoreAutoConfiguration {

 	@Bean
+	@ConditionalOnProperty(prefix = "metadata.store", name = "type", havingValue = "memory", matchIfMissing = true)
 	@ConditionalOnMissingBean
 	public ConcurrentMetadataStore simpleMetadataStore() {
 		return new SimpleMetadataStore();
 	}

-	@ConditionalOnClass(RedisMetadataStore.class)
-	@ConditionalOnBean(RedisTemplate.class)
+	@ConditionalOnProperty(prefix = "metadata.store", name = "type", havingValue = "redis")
 	static class Redis {

 		@Bean
 		@ConditionalOnMissingBean
 		public ConcurrentMetadataStore redisMetadataStore(RedisTemplate<String, ?> redisTemplate,
-														MetadataStoreProperties metadataStoreProperties) {
+				MetadataStoreProperties metadataStoreProperties) {

 			return new RedisMetadataStore(redisTemplate, metadataStoreProperties.getRedis().getKey());
 		}

 	}

-	@ConditionalOnClass(MongoDbMetadataStore.class)
-	@ConditionalOnBean(MongoTemplate.class)
+	@ConditionalOnProperty(prefix = "metadata.store", name = "type", havingValue = "mongodb")
 	static class Mongo {

 		@Bean
 		@ConditionalOnMissingBean
 		public ConcurrentMetadataStore mongoDbMetadataStore(MongoTemplate mongoTemplate,
-															MetadataStoreProperties metadataStoreProperties) {
+				MetadataStoreProperties metadataStoreProperties) {

 			return new MongoDbMetadataStore(mongoTemplate, metadataStoreProperties.getMongoDb().getCollection());
 		}

 	}

-	@ConditionalOnClass(GemfireMetadataStore.class)
+	@ConditionalOnProperty(prefix = "metadata.store", name = "type", havingValue = "gemfire")
 	@Import(ClientCacheAutoConfiguration.class)
 	static class Gemfire {

 		@Bean
 		@ConditionalOnMissingBean
 		public ClientRegionFactoryBean<?, ?> gemfireRegion(GemFireCache cache,
-														MetadataStoreProperties metadataStoreProperties) {
+				MetadataStoreProperties metadataStoreProperties) {

 			ClientRegionFactoryBean<?, ?> clientRegionFactoryBean = new ClientRegionFactoryBean<>();
 			clientRegionFactoryBean.setCache(cache);
@@ -111,7 +111,7 @@ public class MetadataStoreAutoConfiguration {
 		@Bean
 		@ConditionalOnMissingBean
 		public ConcurrentMetadataStore gemfireMetadataStore(Region<?, ?> region,
-															ObjectProvider<MetadataStoreListener> metadataStoreListenerObjectProvider) {
+				ObjectProvider<MetadataStoreListener> metadataStoreListenerObjectProvider) {

 			@SuppressWarnings("unchecked")
 			GemfireMetadataStore gemfireMetadataStore = new GemfireMetadataStore((Region<String, String>) region);
@@ -122,7 +122,7 @@ public class MetadataStoreAutoConfiguration {

 	}

-	@ConditionalOnClass(HazelcastMetadataStore.class)
+	@ConditionalOnProperty(prefix = "metadata.store", name = "type", havingValue = "hazelcast")
 	static class Hazelcast {

 		@Bean
@@ -134,7 +134,7 @@ public class MetadataStoreAutoConfiguration {
 		@Bean
 		@ConditionalOnMissingBean
 		public ConcurrentMetadataStore hazelcastMetadataStore(HazelcastInstance hazelcastInstance,
-															ObjectProvider<MetadataStoreListener> metadataStoreListenerObjectProvider) {
+				ObjectProvider<MetadataStoreListener> metadataStoreListenerObjectProvider) {

 			HazelcastMetadataStore hazelcastMetadataStore = new HazelcastMetadataStore(hazelcastInstance);
 			metadataStoreListenerObjectProvider.ifAvailable(hazelcastMetadataStore::addListener);
@@ -143,7 +143,7 @@ public class MetadataStoreAutoConfiguration {

 	}

-	@ConditionalOnClass({ZookeeperMetadataStore.class, CuratorFramework.class})
+	@ConditionalOnProperty(prefix = "metadata.store", name = "type", havingValue = "zookeeper")
 	static class Zookeeper {

 		@Bean(initMethod = "start")
@@ -157,8 +157,8 @@ public class MetadataStoreAutoConfiguration {
 		@Bean
 		@ConditionalOnMissingBean
 		public ConcurrentMetadataStore zookeeperMetadataStore(CuratorFramework curatorFramework,
-															MetadataStoreProperties metadataStoreProperties,
-															ObjectProvider<MetadataStoreListener> metadataStoreListenerObjectProvider) {
+				MetadataStoreProperties metadataStoreProperties,
+				ObjectProvider<MetadataStoreListener> metadataStoreListenerObjectProvider) {

 			MetadataStoreProperties.Zookeeper zookeeperProperties = metadataStoreProperties.getZookeeper();
 			ZookeeperMetadataStore zookeeperMetadataStore = new ZookeeperMetadataStore(curatorFramework);
@@ -170,14 +170,13 @@ public class MetadataStoreAutoConfiguration {

 	}

-	@ConditionalOnClass(DynamoDbMetadataStore.class)
-	@ConditionalOnBean({AWSCredentialsProvider.class, RegionProvider.class})
+	@ConditionalOnProperty(prefix = "metadata.store", name = "type", havingValue = "dynamodb")
 	static class DynamoDb {

 		@Bean
 		@ConditionalOnMissingBean
 		public AmazonDynamoDBAsync dynamoDB(AWSCredentialsProvider awsCredentialsProvider,
-											RegionProvider regionProvider) {
+				RegionProvider regionProvider) {

 			return AmazonDynamoDBAsyncClientBuilder.standard()
 					.withCredentials(awsCredentialsProvider)
@@ -190,12 +189,12 @@ public class MetadataStoreAutoConfiguration {
 		@Bean
 		@ConditionalOnMissingBean
 		public ConcurrentMetadataStore dynamoDbMetadataStore(AmazonDynamoDBAsync dynamoDB,
-															MetadataStoreProperties metadataStoreProperties) {
+				MetadataStoreProperties metadataStoreProperties) {

 			MetadataStoreProperties.DynamoDb dynamoDbProperties = metadataStoreProperties.getDynamoDb();

-			DynamoDbMetadataStore dynamoDbMetadataStore =
-					new DynamoDbMetadataStore(dynamoDB, dynamoDbProperties.getTable());
+			DynamoDbMetadataStore dynamoDbMetadataStore = new DynamoDbMetadataStore(dynamoDB,
+					dynamoDbProperties.getTable());

 			dynamoDbMetadataStore.setReadCapacity(dynamoDbProperties.getReadCapacity());
 			dynamoDbMetadataStore.setWriteCapacity(dynamoDbProperties.getWriteCapacity());
@@ -210,14 +209,12 @@ public class MetadataStoreAutoConfiguration {

 	}

-	@ConditionalOnClass(JdbcMetadataStore.class)
-	@ConditionalOnBean(JdbcTemplate.class)
+	@ConditionalOnProperty(prefix = "metadata.store", name = "type", havingValue = "jdbc")
 	static class Jdbc {
-
 		@Bean
 		@ConditionalOnMissingBean
 		public ConcurrentMetadataStore jdbcMetadataStore(JdbcTemplate jdbcTemplate,
-														MetadataStoreProperties metadataStoreProperties) {
+				MetadataStoreProperties metadataStoreProperties) {

 			MetadataStoreProperties.Jdbc jdbcProperties = metadataStoreProperties.getJdbc();

--- a/common/metadata-store-common/src/main/java/org/springframework/cloud/fn/common/metadata/store/MetadataStoreProperties.java
+++ b/common/metadata-store-common/src/main/java/org/springframework/cloud/fn/common/metadata/store/MetadataStoreProperties.java
@@ -27,10 +27,27 @@ import org.springframework.integration.redis.metadata.RedisMetadataStore;

 /**
 * @author Artem Bilan
+ * @author David Turanski
 * @since 2.0.2
 */
@ConfigurationProperties("metadata.store")
 public class MetadataStoreProperties {
+	enum StoreType {
+		mongodb,
+		gemfire,
+		redis,
+		dynamodb,
+		jdbc,
+		zookeeper,
+		hazelcast,
+		memory
+	}
+
+	/**
+	 * Indicates the type of metadata store to configure (default is 'memory').
+	 * You must include the corresponding Spring Integration dependency to use a persistent store.
+	 */
+	private StoreType type = StoreType.memory;

 	private final Mongo mongoDb = new Mongo();

@@ -44,6 +61,14 @@ public class MetadataStoreProperties {

 	private final Zookeeper zookeeper = new Zookeeper();

+	public StoreType getType() {
+		return this.type;
+	}
+
+	public void setType(StoreType type) {
+		this.type = type;
+	}
+
 	public Mongo getMongoDb() {
 		return this.mongoDb;
 	}
--- a/supplier/s3-supplier/pom.xml
+++ b/supplier/s3-supplier/pom.xml
@@ -28,6 +28,11 @@
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-validation</artifactId>
        </dependency>
+        <dependency>
+            <groupId>org.springframework.cloud.fn</groupId>
+            <artifactId>metadata-store-common</artifactId>
+            <version>${project.version}</version>
+        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-configuration-processor</artifactId>
--- a/supplier/s3-supplier/src/main/java/org/springframework/cloud/fn/supplier/s3/AwsS3SupplierConfiguration.java
+++ b/supplier/s3-supplier/src/main/java/org/springframework/cloud/fn/supplier/s3/AwsS3SupplierConfiguration.java
@@ -17,14 +17,17 @@
 package org.springframework.cloud.fn.supplier.s3;

 import java.io.File;
-import java.util.Arrays;
+import java.util.function.Predicate;
 import java.util.function.Supplier;
+import java.util.regex.Pattern;

 import com.amazonaws.services.s3.AmazonS3;
+import com.amazonaws.services.s3.model.ListObjectsRequest;
 import com.amazonaws.services.s3.model.S3ObjectSummary;
 import org.reactivestreams.Publisher;
 import reactor.core.publisher.Flux;

+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
 import org.springframework.cloud.aws.core.env.ResourceIdResolver;
 import org.springframework.cloud.fn.common.file.FileConsumerProperties;
@@ -37,80 +40,174 @@ import org.springframework.integration.aws.support.S3SessionFactory;
 import org.springframework.integration.aws.support.filters.S3PersistentAcceptOnceFileListFilter;
 import org.springframework.integration.aws.support.filters.S3RegexPatternFileListFilter;
 import org.springframework.integration.aws.support.filters.S3SimplePatternFileListFilter;
+import org.springframework.integration.core.GenericSelector;
 import org.springframework.integration.core.MessageSource;
 import org.springframework.integration.dsl.IntegrationFlows;
+import org.springframework.integration.endpoint.ReactiveMessageSourceProducer;
 import org.springframework.integration.file.filters.ChainFileListFilter;
-import org.springframework.integration.file.filters.FileListFilter;
-import org.springframework.integration.metadata.SimpleMetadataStore;
+import org.springframework.integration.metadata.ConcurrentMetadataStore;
 import org.springframework.integration.util.IntegrationReactiveUtils;
 import org.springframework.messaging.Message;
+import org.springframework.messaging.support.GenericMessage;
 import org.springframework.util.StringUtils;

 /**
 * @author Artem Bilan
+ * @author David Turanski
 */
@Configuration
-@EnableConfigurationProperties({AwsS3SupplierProperties.class, FileConsumerProperties.class})
-public class AwsS3SupplierConfiguration {
+@EnableConfigurationProperties({ AwsS3SupplierProperties.class, FileConsumerProperties.class })
+public abstract class AwsS3SupplierConfiguration {

-	private final AwsS3SupplierProperties awsS3SupplierProperties;
-	private final FileConsumerProperties fileConsumerProperties;
-	private final AmazonS3 amazonS3;
-	private final ResourceIdResolver resourceIdResolver;
+	protected static final String METADATA_STORE_PREFIX = "s3-metadata-";
+
+	protected final AwsS3SupplierProperties awsS3SupplierProperties;
+
+	protected final FileConsumerProperties fileConsumerProperties;
+
+	protected final AmazonS3 amazonS3;
+
+	protected final ResourceIdResolver resourceIdResolver;
+
+	protected final ConcurrentMetadataStore metadataStore;

 	public AwsS3SupplierConfiguration(AwsS3SupplierProperties awsS3SupplierProperties,
-									FileConsumerProperties fileConsumerProperties,
-									AmazonS3 amazonS3,
-									ResourceIdResolver resourceIdResolver) {
+			FileConsumerProperties fileConsumerProperties,
+			AmazonS3 amazonS3,
+			ResourceIdResolver resourceIdResolver, ConcurrentMetadataStore metadataStore) {
 		this.awsS3SupplierProperties = awsS3SupplierProperties;
 		this.fileConsumerProperties = fileConsumerProperties;
 		this.amazonS3 = amazonS3;
 		this.resourceIdResolver = resourceIdResolver;
+		this.metadataStore = metadataStore;
 	}

-	@Bean
-	public S3InboundFileSynchronizer s3InboundFileSynchronizer() {
-		S3SessionFactory s3SessionFactory = new S3SessionFactory(this.amazonS3, this.resourceIdResolver);
-		S3InboundFileSynchronizer synchronizer = new S3InboundFileSynchronizer(s3SessionFactory);
-		synchronizer.setDeleteRemoteFiles(this.awsS3SupplierProperties.isDeleteRemoteFiles());
-		synchronizer.setPreserveTimestamp(this.awsS3SupplierProperties.isPreserveTimestamp());
-		String remoteDir = this.awsS3SupplierProperties.getRemoteDir();
-		synchronizer.setRemoteDirectory(remoteDir);
-		synchronizer.setRemoteFileSeparator(this.awsS3SupplierProperties.getRemoteFileSeparator());
-		synchronizer.setTemporaryFileSuffix(this.awsS3SupplierProperties.getTmpFileSuffix());
+	@Configuration
+	@ConditionalOnProperty(prefix = "s3.supplier", name = "list-only", havingValue = "false", matchIfMissing = true)
+	static class SynchronizingConfiguration extends AwsS3SupplierConfiguration {

-		FileListFilter<S3ObjectSummary> filter = null;
-		if (StringUtils.hasText(this.awsS3SupplierProperties.getFilenamePattern())) {
-			filter = new S3SimplePatternFileListFilter(this.awsS3SupplierProperties.getFilenamePattern());
+		@Bean
+		public Supplier<Flux<Message<?>>> s3Supplier(Publisher<Message<Object>> s3SupplierFlow) {
+			return () -> Flux.from(s3SupplierFlow);
 		}
-		else if (this.awsS3SupplierProperties.getFilenameRegex() != null) {
-			filter = new S3RegexPatternFileListFilter(this.awsS3SupplierProperties.getFilenameRegex());
+
+		@Bean
+		public ChainFileListFilter<S3ObjectSummary> filter(ConcurrentMetadataStore metadataStore) {
+			ChainFileListFilter<S3ObjectSummary> chainFilter = new ChainFileListFilter<>();
+			if (StringUtils.hasText(this.awsS3SupplierProperties.getFilenamePattern())) {
+				chainFilter.addFilter(
+						new S3SimplePatternFileListFilter(this.awsS3SupplierProperties.getFilenamePattern()));
+			}
+			else if (this.awsS3SupplierProperties.getFilenameRegex() != null) {
+				chainFilter
+						.addFilter(new S3RegexPatternFileListFilter(this.awsS3SupplierProperties.getFilenameRegex()));
+			}
+
+			chainFilter.addFilter(new S3PersistentAcceptOnceFileListFilter(metadataStore, METADATA_STORE_PREFIX));
+			return chainFilter;
 		}
-		if (filter != null) {
-			synchronizer.setFilter(new ChainFileListFilter<>(Arrays.asList(filter,
-					new S3PersistentAcceptOnceFileListFilter(new SimpleMetadataStore(), "s3-metadata-"))));
+
+		SynchronizingConfiguration(AwsS3SupplierProperties awsS3SupplierProperties,
+				FileConsumerProperties fileConsumerProperties,
+				AmazonS3 amazonS3,
+				ResourceIdResolver resourceIdResolver,
+				ConcurrentMetadataStore concurrentMetadataStore) {
+			super(awsS3SupplierProperties, fileConsumerProperties, amazonS3, resourceIdResolver,
+					concurrentMetadataStore);
+		}
+
+		@Bean
+		public Publisher<Message<Object>> s3SupplierFlow(MessageSource<?> s3MessageSource) {
+			return FileUtils.enhanceFlowForReadingMode(IntegrationFlows
+					.from(IntegrationReactiveUtils.messageSourceToFlux(s3MessageSource)), fileConsumerProperties)
+					.toReactivePublisher();
+		}
+
+		@Bean
+		public S3InboundFileSynchronizer s3InboundFileSynchronizer(ChainFileListFilter<S3ObjectSummary> filter) {
+			S3SessionFactory s3SessionFactory = new S3SessionFactory(this.amazonS3, this.resourceIdResolver);
+			S3InboundFileSynchronizer synchronizer = new S3InboundFileSynchronizer(s3SessionFactory);
+			synchronizer.setDeleteRemoteFiles(this.awsS3SupplierProperties.isDeleteRemoteFiles());
+			synchronizer.setPreserveTimestamp(this.awsS3SupplierProperties.isPreserveTimestamp());
+			String remoteDir = this.awsS3SupplierProperties.getRemoteDir();
+			synchronizer.setRemoteDirectory(remoteDir);
+			synchronizer.setRemoteFileSeparator(this.awsS3SupplierProperties.getRemoteFileSeparator());
+			synchronizer.setTemporaryFileSuffix(this.awsS3SupplierProperties.getTmpFileSuffix());
+			synchronizer.setFilter(filter);
+
+			return synchronizer;
+		}
+
+		@Bean
+		public MessageSource<File> s3MessageSource(S3InboundFileSynchronizer s3InboundFileSynchronizer) {
+			S3InboundFileSynchronizingMessageSource s3MessageSource = new S3InboundFileSynchronizingMessageSource(
+					s3InboundFileSynchronizer);
+			s3MessageSource.setLocalDirectory(this.awsS3SupplierProperties.getLocalDir());
+			s3MessageSource.setAutoCreateLocalDirectory(this.awsS3SupplierProperties.isAutoCreateLocalDir());
+			return s3MessageSource;
 		}
-		return synchronizer;
 	}

-	@Bean
-	public MessageSource<File> s3MessageSource() {
-		S3InboundFileSynchronizingMessageSource s3MessageSource =
-				new S3InboundFileSynchronizingMessageSource(s3InboundFileSynchronizer());
-		s3MessageSource.setLocalDirectory(this.awsS3SupplierProperties.getLocalDir());
-		s3MessageSource.setAutoCreateLocalDirectory(this.awsS3SupplierProperties.isAutoCreateLocalDir());
-		return s3MessageSource;
-	}
+	@Configuration
+	@ConditionalOnProperty(prefix = "s3.supplier", name = "list-only", havingValue = "true")
+	static class ListOnlyConfiguration extends AwsS3SupplierConfiguration {
+		ListOnlyConfiguration(AwsS3SupplierProperties awsS3SupplierProperties,
+				FileConsumerProperties fileConsumerProperties,
+				AmazonS3 amazonS3,
+				ResourceIdResolver resourceIdResolver, ConcurrentMetadataStore metadataStore) {
+			super(awsS3SupplierProperties, fileConsumerProperties, amazonS3, resourceIdResolver, metadataStore);
+		}

-	@Bean
-	public Publisher<Message<Object>> s3SupplierFlow() {
-		return FileUtils.enhanceFlowForReadingMode(IntegrationFlows
-				.from(IntegrationReactiveUtils.messageSourceToFlux(s3MessageSource())), fileConsumerProperties)
-				.toReactivePublisher();
-	}
+		@Bean
+		public Supplier<Flux<Message<?>>> s3Supplier(Publisher<Message<Object>> s3SupplierFlow) {
+			return () -> Flux.from(s3SupplierFlow);
+		}

-	@Bean
-	public Supplier<Flux<Message<?>>> s3Supplier() {
-		return () -> Flux.from(s3SupplierFlow());
+		@Bean
+		public Publisher<Message<Object>> s3SupplierFlow(ReactiveMessageSourceProducer s3ListingProducer,
+				GenericSelector<S3ObjectSummary> listOnlyFilter) {
+			return IntegrationFlows
+					.from(s3ListingProducer)
+					.split()
+					.filter(listOnlyFilter)
+					.toReactivePublisher();
+		}
+
+		@Bean
+		GenericSelector<S3ObjectSummary> listOnlyFilter() {
+			Predicate<S3ObjectSummary> predicate = s -> true;
+			if (StringUtils.hasText(this.awsS3SupplierProperties.getFilenamePattern())) {
+				Pattern pattern = Pattern.compile(this.awsS3SupplierProperties.getFilenamePattern());
+				predicate = (S3ObjectSummary summary) -> pattern.matcher(summary.getKey()).matches();
+			}
+			else if (this.awsS3SupplierProperties.getFilenameRegex() != null) {
+				predicate = (S3ObjectSummary summary) -> this.awsS3SupplierProperties.getFilenameRegex()
+						.matcher(summary.getKey()).matches();
+			}
+			predicate = predicate.and((S3ObjectSummary summary) -> {
+				final String key = METADATA_STORE_PREFIX + summary.getBucketName() + "-" + summary.getKey();
+				final String lastModified = String.valueOf(summary.getLastModified().getTime());
+				final String storedLastModified = this.metadataStore.get(key);
+				boolean result = !lastModified.equals(storedLastModified);
+				if (result) {
+					metadataStore.put(key, lastModified);
+				}
+				return result;
+			});
+
+			GenericSelector<S3ObjectSummary> selector = predicate::test;
+
+			return selector;
+		}
+
+		@Bean
+		ReactiveMessageSourceProducer s3ListingMessageProducer(AmazonS3 amazonS3,
+				AwsS3SupplierProperties awsS3SupplierProperties) {
+			ListObjectsRequest listObjectsRequest = new ListObjectsRequest();
+			listObjectsRequest.setBucketName(awsS3SupplierProperties.getRemoteDir());
+			return new ReactiveMessageSourceProducer(
+					(MessageSource<Iterable<S3ObjectSummary>>) () -> new GenericMessage<>(
+							amazonS3.listObjects(listObjectsRequest).getObjectSummaries()));
+		}
 	}
 }
--- a/supplier/s3-supplier/src/main/java/org/springframework/cloud/fn/supplier/s3/AwsS3SupplierProperties.java
+++ b/supplier/s3-supplier/src/main/java/org/springframework/cloud/fn/supplier/s3/AwsS3SupplierProperties.java
@@ -80,6 +80,11 @@ public class AwsS3SupplierProperties {
 	 */
 	private boolean preserveTimestamp = true;

+	/**
+	 * Set to true to return s3 object metadata without copying file to a local directory.
+	 */
+	private boolean listOnly = false;
+
 	@Length(min = 3)
 	public String getRemoteDir() {
 		return this.remoteDir;
--- a/supplier/s3-supplier/src/test/java/org/springframework/cloud/fn/supplier/s3/AbstractAwsS3SupplierMockTests.java
+++ b/supplier/s3-supplier/src/test/java/org/springframework/cloud/fn/supplier/s3/AbstractAwsS3SupplierMockTests.java
@@ -130,16 +130,17 @@ public abstract class AbstractAwsS3SupplierMockTests {
 			Calendar calendar = Calendar.getInstance();
 			calendar.add(Calendar.DATE, 1);

+			ObjectListing objectListing = new ObjectListing();
+			List<S3ObjectSummary> objectSummaries = objectListing.getObjectSummaries();
+			for (S3Object s3Object : S3_OBJECTS) {
+				S3ObjectSummary s3ObjectSummary = new S3ObjectSummary();
+				s3ObjectSummary.setBucketName(S3_BUCKET);
+				s3ObjectSummary.setKey(s3Object.getKey());
+				s3ObjectSummary.setLastModified(calendar.getTime());
+				objectSummaries.add(s3ObjectSummary);
+			}
+
 			willAnswer(invocation -> {
-				ObjectListing objectListing = new ObjectListing();
-				List<S3ObjectSummary> objectSummaries = objectListing.getObjectSummaries();
-				for (S3Object s3Object : S3_OBJECTS) {
-					S3ObjectSummary s3ObjectSummary = new S3ObjectSummary();
-					s3ObjectSummary.setBucketName(S3_BUCKET);
-					s3ObjectSummary.setKey(s3Object.getKey());
-					s3ObjectSummary.setLastModified(calendar.getTime());
-					objectSummaries.add(s3ObjectSummary);
-				}
 				return objectListing;
 			}).given(amazonS3).listObjects(any(ListObjectsRequest.class));

--- a/supplier/s3-supplier/src/test/java/org/springframework/cloud/fn/supplier/s3/AmazonS3ListOnlyTests.java
+++ b/supplier/s3-supplier/src/test/java/org/springframework/cloud/fn/supplier/s3/AmazonS3ListOnlyTests.java
@@ -0,0 +1,69 @@
+/*
+ * Copyright 2020-2020 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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.fn.supplier.s3;
+
+import java.time.Duration;
+import java.util.HashSet;
+
+import com.amazonaws.services.s3.model.S3ObjectSummary;
+import org.junit.jupiter.api.Test;
+import reactor.core.publisher.Flux;
+import reactor.test.StepVerifier;
+
+import org.springframework.messaging.Message;
+import org.springframework.test.context.TestPropertySource;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+@TestPropertySource(properties = {
+		"s3.supplier.list-only=true"
+})
+public class AmazonS3ListOnlyTests extends AbstractAwsS3SupplierMockTests {
+
+	@Test
+	public void test() {
+		final Flux<Message<?>> messageFlux = s3Supplier.get();
+		final HashSet<String> keys = new HashSet<>();
+		keys.add("1.test");
+		keys.add("2.test");
+		keys.add("otherFile");
+		StepVerifier stepVerifier = StepVerifier.create(messageFlux)
+				.assertNext(message -> {
+					S3ObjectSummary summary = (S3ObjectSummary) message.getPayload();
+					assertThat(summary.getBucketName()).isEqualTo(S3_BUCKET);
+					assertThat(keys.contains(summary.getKey()));
+					keys.remove(summary.getKey());
+				})
+				.assertNext(message -> {
+					S3ObjectSummary summary = (S3ObjectSummary) message.getPayload();
+					assertThat(summary.getBucketName()).isEqualTo(S3_BUCKET);
+					assertThat(keys.contains(summary.getKey()));
+					keys.remove(summary.getKey());
+				})
+				.assertNext(message -> {
+					S3ObjectSummary summary = (S3ObjectSummary) message.getPayload();
+					assertThat(summary.getBucketName()).isEqualTo(S3_BUCKET);
+					assertThat(keys.contains(summary.getKey()));
+					keys.remove(summary.getKey());
+				})
+				.expectTimeout(Duration.ofSeconds(1))
+				.verifyLater();
+		standardIntegrationFlow.start();
+		stepVerifier.verify(Duration.ofSeconds(10));
+		standardIntegrationFlow.stop();
+	}
+}
--- a/supplier/sftp-supplier/src/main/java/org/springframework/cloud/fn/supplier/sftp/SftpSupplierConfiguration.java
+++ b/supplier/sftp-supplier/src/main/java/org/springframework/cloud/fn/supplier/sftp/SftpSupplierConfiguration.java
@@ -17,9 +17,10 @@
 package org.springframework.cloud.fn.supplier.sftp;

 import java.io.IOException;
-import java.util.Arrays;
 import java.util.List;
+import java.util.function.Predicate;
 import java.util.function.Supplier;
+import java.util.regex.Pattern;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;

@@ -41,8 +42,10 @@ import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Import;
 import org.springframework.context.annotation.Primary;
+import org.springframework.http.MediaType;
 import org.springframework.integration.aop.ReceiveMessageAdvice;
 import org.springframework.integration.channel.QueueChannel;
+import org.springframework.integration.core.GenericSelector;
 import org.springframework.integration.core.MessageSource;
 import org.springframework.integration.dsl.IntegrationFlow;
 import org.springframework.integration.dsl.IntegrationFlows;
@@ -66,9 +69,7 @@ import org.springframework.messaging.MessageHeaders;
 import org.springframework.messaging.MessagingException;
 import org.springframework.messaging.PollableChannel;
 import org.springframework.messaging.support.MessageBuilder;
-import org.springframework.util.Assert;
 import org.springframework.util.CollectionUtils;
-import org.springframework.util.MimeTypeUtils;
 import org.springframework.util.StringUtils;

 /**
@@ -87,6 +88,8 @@ public class SftpSupplierConfiguration {

 	private static final String METADATA_STORE_PREFIX = "sftpSource/";

+	private static final String FILE_MODIFIED_TIME_HEADER = "FILE_MODIFIED_TIME";
+
 	@Bean
 	public Supplier<Flux<? extends Message<?>>> sftpSupplier(MessageSource<?> sftpMessageSource,
 			@Nullable Publisher<Message<Object>> sftpReadingFlow,
@@ -130,7 +133,7 @@ public class SftpSupplierConfiguration {
 	 * Configure the standard filters for SFTP inbound adapters.
 	 */
 	@Bean
-	public ChainFileListFilter<LsEntry> chainFilter(SftpSupplierProperties sftpSupplierProperties,
+	public FileListFilter<LsEntry> chainFilter(SftpSupplierProperties sftpSupplierProperties,
 			ConcurrentMetadataStore metadataStore) {
 		ChainFileListFilter<LsEntry> chainFilter = new ChainFileListFilter<>();

@@ -142,9 +145,7 @@ public class SftpSupplierConfiguration {
 			chainFilter
 					.addFilter(new SftpRegexPatternFileListFilter(sftpSupplierProperties.getFilenameRegex()));
 		}
-		// TODO: Temporary work-around for
-		// https://github.com/spring-projects/spring-integration/issues/3315.
-		chainFilter.addFilter(Arrays::asList);
+
 		chainFilter.addFilter(new SftpPersistentAcceptOnceFileListFilter(metadataStore, METADATA_STORE_PREFIX));
 		return chainFilter;
 	}
@@ -279,7 +280,6 @@ public class SftpSupplierConfiguration {
 		}

 		@Bean
-		@SuppressWarnings("unchecked")
 		public MessageSource<?> targetMessageSource(PollableChannel listingChannel,
 				SftpListingMessageProducer sftpListingMessageProducer) {
 			return () -> {
@@ -299,29 +299,70 @@ public class SftpSupplierConfiguration {
 		}

 		@Bean
-		public IntegrationFlow listingFlow(MessageProducerSupport messageProducerSupport,
-				MessageChannel listingChannel, MessageProcessor<?> metadataWriter) {
+		GenericSelector<String> listOnlyFilter(SftpSupplierProperties sftpSupplierProperties) {
+			Predicate<String> predicate = s -> true;
+			if (StringUtils.hasText(sftpSupplierProperties.getFilenamePattern())) {
+				predicate = Pattern.compile(sftpSupplierProperties.getFilenamePattern()).asPredicate();
+			}
+			else if (sftpSupplierProperties.getFilenameRegex() != null) {
+				predicate = sftpSupplierProperties.getFilenameRegex().asPredicate();
+			}

-			return IntegrationFlows.from(messageProducerSupport)
+			GenericSelector<String> selector = predicate::test;
+
+			return selector;
+		}
+
+		@Bean
+		public IntegrationFlow listingFlow(MessageProducerSupport listingMessageProducer,
+				MessageChannel listingChannel, MessageProcessor<?> lsEntryToStringTransformer,
+				GenericSelector<Message<?>> duplicateFilter,
+				GenericSelector<String> listOnlyFilter) {
+
+			return IntegrationFlows.from(listingMessageProducer)
 					.split()
-					.transform(metadataWriter)
+					.transform(lsEntryToStringTransformer)
+					.filter(duplicateFilter)
+					.filter(listOnlyFilter)
 					.channel(listingChannel)
 					.get();
 		}

 		@Bean
-		public MessageProcessor<?> metadataWriter(ConcurrentMetadataStore metadataStore) {
-			return message -> {
-				MessageHeaders messageHeaders = message.getHeaders();
-				Assert.notNull(messageHeaders, "Cannot transform message with null headers");
-				Assert.isTrue(messageHeaders.containsKey(FileHeaders.REMOTE_DIRECTORY),
-						"Remote directory header not found");
-				Assert.hasText((String) message.getPayload(), "Payload must not be empty.");
+		public MessageProcessor<Message<?>> lsEntryToStringTransformer() {
+			return (Message<?> message) -> {

-				metadataStore.putIfAbsent(
-						message.getHeaders().get(FileHeaders.REMOTE_DIRECTORY).toString() + message.getPayload(),
-						String.valueOf(message.getHeaders().getTimestamp()));
-				return message;
+				LsEntry lsEntry = (LsEntry) message.getPayload();
+
+				String fileName = message.getHeaders().get(FileHeaders.REMOTE_DIRECTORY) + lsEntry.getFilename();
+
+				return MessageBuilder.withPayload(fileName)
+						.copyHeaders(message.getHeaders())
+						.setHeader(FILE_MODIFIED_TIME_HEADER, String.valueOf(lsEntry.getAttrs().getMTime()))
+						.setHeader(MessageHeaders.CONTENT_TYPE, MediaType.TEXT_PLAIN)
+						.build();
+			};
+
+		}
+
+		@Bean
+		GenericSelector<Message<?>> duplicateFilter(ConcurrentMetadataStore metadataStore) {
+			return new GenericSelector<Message<?>>() {
+				@Override
+				public boolean accept(Message<?> message) {
+
+					String lastModifiedTime = (String) message.getHeaders().get(FILE_MODIFIED_TIME_HEADER);
+					String storedLastModifiedTime = metadataStore.get(METADATA_STORE_PREFIX + message.getPayload());
+
+					boolean result = !lastModifiedTime.equals(storedLastModifiedTime);
+
+					if (result) {
+						metadataStore.put(
+								METADATA_STORE_PREFIX + message.getPayload(),
+								message.getHeaders().get(FILE_MODIFIED_TIME_HEADER).toString());
+					}
+					return result;
+				}
 			};
 		}

@@ -342,17 +383,19 @@ public class SftpSupplierConfiguration {
 			}

 			public void listNames() {
-				String[] names = {};
+				LsEntry[] entries = {};
 				try {
-					names = Stream.of(this.sessionFactory.getSession().listNames(this.remoteDirectory))
-							.map(name -> String.join(this.remoteFileSeparator, this.remoteDirectory, name))
-							.collect(Collectors.toList()).toArray(names);
+					entries = Stream.of(this.sessionFactory.getSession().list(this.remoteDirectory))
+							.filter(o -> {
+								LsEntry lsEntry = (LsEntry) o;
+								return !(lsEntry.getAttrs().isDir() || lsEntry.getAttrs().isLink());
+							})
+							.collect(Collectors.toList()).toArray(entries);
 				}
 				catch (IOException e) {
 					throw new MessagingException(e.getMessage(), e);
 				}
-				sendMessage(MessageBuilder.withPayload(names)
-						.setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.TEXT_PLAIN)
+				sendMessage(MessageBuilder.withPayload(entries)
 						.setHeader(FileHeaders.REMOTE_DIRECTORY, this.remoteDirectory + this.remoteFileSeparator)
 						.build());
 			}
--- a/supplier/sftp-supplier/src/test/java/org/springframework/cloud/fn/supplier/sftp/SftpSupplierApplicationTests.java
+++ b/supplier/sftp-supplier/src/test/java/org/springframework/cloud/fn/supplier/sftp/SftpSupplierApplicationTests.java
@@ -91,7 +91,28 @@ public class SftpSupplierApplicationTests extends SftpTestSupport {
 								assertThat(message.getHeaders().get(MessageHeaders.CONTENT_TYPE))
 										.isEqualTo(MediaType.TEXT_PLAIN);
 							})
-							.thenCancel()
+							.expectTimeout(Duration.ofMillis(1000))
+							.verify(Duration.ofSeconds(10));
+
+				});
+	}
+
+	@Test
+	void supplierForListOnlyWithPatternFilter() {
+		defaultApplicationContextRunner
+				.withPropertyValues("sftp.supplier.listOnly=true", "sftp.supplier.file-name-pattern=.*1.txt")
+				.run(context -> {
+					Supplier<Flux<Message<String>>> sftpSupplier = context.getBean("sftpSupplier",
+							Supplier.class);
+					SftpSupplierProperties properties = context.getBean(SftpSupplierProperties.class);
+
+					final AtomicReference<String> expectedFileName = new AtomicReference<>(
+							properties.getRemoteDir() + File.separator + "sftpSource1.txt");
+					StepVerifier.create(sftpSupplier.get())
+							.assertNext(message -> {
+								assertThat(expectedFileName.get()).contains(message.getPayload());
+							})
+							.expectTimeout(Duration.ofMillis(1000))
 							.verify(Duration.ofSeconds(10));

 				});
@@ -114,7 +135,7 @@ public class SftpSupplierApplicationTests extends SftpTestSupport {
 					final AtomicReference<Set<String>> expectedFileNames = new AtomicReference<>(fileNames);
 					StepVerifier.create(sftpSupplier.get())
 							.assertNext(message -> {
-								File file = (File) message.getPayload();
+								File file = message.getPayload();
 								assertThat(expectedFileNames.get()).contains(file.getAbsolutePath());
 								expectedFileNames.get().remove(file.getAbsolutePath());
 							})