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

Fix S3 Channel Adapters to use filter by default

Browse Source
This commit is contained in:
Artem Bilan
2018-06-26 13:14:25 -04:00
parent 6f39406309
commit 8247d0df1b
2 changed files with 13 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
src/main/java/org/springframework/integration/aws/inbound/S3InboundFileSynchronizer.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2017 the original author or authors.
* Copyright 2016-2018 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.
@@ -19,12 +19,10 @@ package org.springframework.integration.aws.inbound;
import java.io.File;
import java.io.IOException;
import org.springframework.expression.Expression;
import org.springframework.expression.common.LiteralExpression;
import org.springframework.integration.aws.support.S3Session;
import org.springframework.integration.aws.support.S3SessionFactory;
import org.springframework.integration.aws.support.filters.S3PersistentAcceptOnceFileListFilter;
import org.springframework.integration.file.filters.FileListFilter;
import org.springframework.integration.file.remote.session.Session;
import org.springframework.integration.file.remote.session.SessionFactory;
import org.springframework.integration.file.remote.synchronizer.AbstractInboundFileSynchronizer;
@@ -54,18 +52,8 @@ public class S3InboundFileSynchronizer extends AbstractInboundFileSynchronizer<S
*/
public S3InboundFileSynchronizer(SessionFactory<S3ObjectSummary> sessionFactory) {
super(sessionFactory);
setRemoteDirectoryExpression(new LiteralExpression(null));
setFilter(new S3PersistentAcceptOnceFileListFilter(new SimpleMetadataStore(), "s3MessageSource"));
}
@Override
public final void setRemoteDirectoryExpression(Expression remoteDirectoryExpression) {
super.setRemoteDirectoryExpression(remoteDirectoryExpression);
}
@Override
public final void setFilter(FileListFilter<S3ObjectSummary> filter) {
super.setFilter(filter);
doSetRemoteDirectoryExpression(new LiteralExpression(null));
doSetFilter(new S3PersistentAcceptOnceFileListFilter(new SimpleMetadataStore(), "s3MessageSource"));
}
@Override
@@ -86,6 +74,7 @@ public class S3InboundFileSynchronizer extends AbstractInboundFileSynchronizer<S
@Override
protected boolean copyFileToLocalDirectory(String remoteDirectoryPath, S3ObjectSummary remoteFile,
File localDirectory, Session<S3ObjectSummary> session) throws IOException {
return super.copyFileToLocalDirectory(((S3Session) session).normalizeBucketName(remoteDirectoryPath),
remoteFile, localDirectory, session);
}

16
src/main/java/org/springframework/integration/aws/inbound/S3StreamingMessageSource.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2017 the original author or authors.
* Copyright 2016-2018 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,16 +16,18 @@
package org.springframework.integration.aws.inbound;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Comparator;
import java.util.List;
import java.util.stream.Collectors;
import org.springframework.integration.aws.support.S3FileInfo;
import org.springframework.integration.aws.support.S3Session;
import org.springframework.integration.aws.support.filters.S3PersistentAcceptOnceFileListFilter;
import org.springframework.integration.file.remote.AbstractFileInfo;
import org.springframework.integration.file.remote.AbstractRemoteFileStreamingMessageSource;
import org.springframework.integration.file.remote.RemoteFileTemplate;
import org.springframework.integration.metadata.SimpleMetadataStore;
import com.amazonaws.services.s3.model.S3ObjectSummary;
@@ -47,15 +49,15 @@ public class S3StreamingMessageSource extends AbstractRemoteFileStreamingMessage
Comparator<AbstractFileInfo<S3ObjectSummary>> comparator) {
super(template, comparator);
doSetFilter(new S3PersistentAcceptOnceFileListFilter(new SimpleMetadataStore(), "s3StreamingMessageSource"));
}
@Override
protected List<AbstractFileInfo<S3ObjectSummary>> asFileInfoList(Collection<S3ObjectSummary> collection) {
List<AbstractFileInfo<S3ObjectSummary>> canonicalFiles = new ArrayList<AbstractFileInfo<S3ObjectSummary>>();
for (S3ObjectSummary s3ObjectSummary : collection) {
canonicalFiles.add(new S3FileInfo(s3ObjectSummary));
}
return canonicalFiles;
return collection.stream()
.map(S3FileInfo::new)
.collect(Collectors.toList());
}
@Override