实现 spring-integration-aws 的问题

Question

我正在使用 spring 集成 aws 来轮询 S3 资源并从 S3 存储桶中获取文件并使用 spring 集成处理它们。 以下是我所拥有的：

AmazonS3 amazonS3 = new AmazonS3Client(new BasicAWSCredentials(accessKey, secretKey));

@Bean
IntegrationFlow fileReadingFlow() {
    return IntegrationFlows
              .from(s3InboundFileSynchronizingMessageSource(),
                      e -> e.poller(p -> p.fixedDelay(30, TimeUnit.SECONDS)))
            .handle(receiptProcessor())
            .get();
}
@Bean
public S3InboundFileSynchronizer s3InboundFileSynchronizer() {
    S3InboundFileSynchronizer synchronizer = new S3InboundFileSynchronizer(amazonS3);
    synchronizer.setDeleteRemoteFiles(false);
    synchronizer.setPreserveTimestamp(true);
    synchronizer.setRemoteDirectory(s3BucketName.concat("/").concat(s3InboundFolder));
    synchronizer.setFilter(new S3RegexPatternFileListFilter(".*\\.dat\\.{0,1}\\d{0,2}"));
    return synchronizer;
}

@Bean
public S3InboundFileSynchronizingMessageSource s3InboundFileSynchronizingMessageSource() {
    S3InboundFileSynchronizingMessageSource messageSource =
            new S3InboundFileSynchronizingMessageSource(s3InboundFileSynchronizer());
    messageSource.setAutoCreateLocalDirectory(false);
    messageSource.setLocalDirectory(new File(inboundDir));
    messageSource.setLocalFilter(new AcceptOnceFileListFilter<File>());
    return messageSource;
}

我的 S3 存储桶和密钥是：

bucketName = shipmentReceipts
key = receipts/originalReceipts/inbound/receipt1.dat

所以我在这个实现中面临 2 个问题：
1. inboundDir 文件夹名称被重命名为不同的路径名，并附加了 s3key，从而导致FileNotFoundException。我将此追溯到AbstractInboundFileSynchronizer.java 文件中的以下代码：

    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()) {........

所以它最终会寻找一个文件路径 C:\SpringAws\S3inbound\receipts\originalReceipts\inbound\receipt1.dat，它没有找到并给出 FileNotFoundExceptionerror。相反，它应该只是复制到本地文件夹 C:\SpringAws\S3inbound\receipt1.dat

1. 拉动 S3 对象时，我注意到它拉动了 shipmentReceipts/receipts 下的所有对象，而不是 shipmentReceipts/receipts/originalReceipts/inbound 在进一步调试时，我发现S3Session.java 中的以下代码 sn-p 负责它：

   @Override
   public S3ObjectSummary[] list(String path) throws IOException {
   Assert.hasText(path, "'path' must not be empty String.");
   String[] bucketPrefix = path.split("/");
   Assert.state(bucketPrefix.length > 0 && bucketPrefix[0].length() >= 3,
           "S3 bucket name must be at least 3 characters long.");
   
   String bucket = resolveBucket(bucketPrefix[0]);
   
   ListObjectsRequest listObjectsRequest = new ListObjectsRequest()
           .withBucketName(bucket);
   if (bucketPrefix.length > 1) {
       **listObjectsRequest.setPrefix(bucketPrefix[1]);**
   }
   
   /*
   For listing objects, Amazon S3 returns up to 1,000 keys in the response.
   If you have more than 1,000 keys in your bucket, the response will be truncated.
   You should always check for if the response is truncated.
   */
   ObjectListing objectListing;
   List<S3ObjectSummary> objectSummaries = new ArrayList<>();
   do {......
   

它将前缀设置为它遇到的第一个正斜杠/ 之后的所有内容。 我如何减轻这些？谢谢！

Answer 1

嵌套路径的第一个问题是已知问题，已在最新的5.0 M3:https://spring.io/blog/2017/04/05/spring-integration-5-0-milestone-3-available 和RecursiveDirectoryScanner 中修复。

同时您必须将LocalFilenameGeneratorExpression 指定为：

Expression expression = PARSER.parseExpression("#this.contains('/') ? #this.substring(#this.lastIndexOf('/') + 1) : #this");
synchronizer.setLocalFilenameGeneratorExpression(expression);

S3ObjectSummary 包含 key 作为没有bucket 的完整路径。

第二个“嵌套路径”问题已通过：https://github.com/spring-projects/spring-integration-aws/issues/45 修复。该修复程序可在1.1.0.M1：https://spring.io/blog/2017/03/09/spring-integration-extension-for-aws-1-1-0-m1-available

Answer 2

根据 Artem，我确实使用了 spring-integration-aws 的最新里程碑版本，但发现编写一个扩展 AbstractInboundFileSynchronizer 的自定义类来解决我的问题更容易。 这是我创建的类：

public class MyAbstractInboundFileSynchronizer extends AbstractInboundFileSynchronizer<S3ObjectSummary> {

private volatile String remoteFileSeparator = "/";
private volatile String temporaryFileSuffix = ".writing";
private volatile boolean deleteRemoteFiles;
private volatile boolean  preserveTimestamp;
private volatile FileListFilter<S3ObjectSummary> filter;
private volatile Expression localFilenameGeneratorExpression;
private volatile EvaluationContext evaluationContext;

@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 setRemoteFileSeparator(String remoteFileSeparator) {
    super.setRemoteFileSeparator(remoteFileSeparator);
    this.remoteFileSeparator = remoteFileSeparator;
}

public MyAbstractInboundFileSynchronizer() {
    this(new S3SessionFactory());
}

public MyAbstractInboundFileSynchronizer(AmazonS3 amazonS3) {
    this(new S3SessionFactory(amazonS3));
}

/**
 * Create a synchronizer with the {@link SessionFactory} used to acquire {@link Session} instances.
 * @param sessionFactory The session factory.
 */
public MyAbstractInboundFileSynchronizer(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);
}

@Override
protected boolean isFile(S3ObjectSummary file) {
    return true;
}

@Override
protected String getFilename(S3ObjectSummary file) {
    if(file != null){
        String key = file.getKey();
        String fileName = key.substring(key.lastIndexOf('/')+1);
        return fileName;
    }
    else return null;
}

@Override
protected long getModified(S3ObjectSummary file) {
    return file.getLastModified().getTime();
}

@Override
protected void copyFileToLocalDirectory(String remoteDirectoryPath, S3ObjectSummary remoteFile, File localDirectory,
                                        Session<S3ObjectSummary> session) throws IOException {
    String remoteFileName = this.getFilename(remoteFile);
    //String localFileName = this.generateLocalFileName(remoteFileName);
    String localFileName = remoteFileName;
    String remoteFilePath = remoteDirectoryPath != null
            ? (remoteDirectoryPath + 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));
        }
    }
}
}

我还根据 Artem 更新了 LocalFilenameGeneratorExpression。谢谢！

Answer 3

@user5758361 你用嵌套路径描述的第一个问题也可以通过覆盖S3FileInfo来解决：

public class S3FileInfo extends org.springframework.integration.aws.support.S3FileInfo {
    private static final ObjectWriter OBJECT_WRITER = new ObjectMapper().writerFor(S3ObjectSummary.class);

    public S3FileInfo(S3ObjectSummary s3ObjectSummary) {
        super(s3ObjectSummary);
    }

    @Override
    public String getFilename() {
        return FilenameUtils.getName(super.getFilename());
    }

    @Override
    public String toJson() {
        try {
            return OBJECT_WRITER.writeValueAsString(super.getFileInfo());
        } catch (JsonProcessingException e) {
            throw new UncheckedIOException(e);
        }
    }
}

toJson 被覆盖以避免某些对象的 NPE。

将其用于流式传输：

public class S3StreamingMessageSource extends org.springframework.integration.aws.inbound.S3StreamingMessageSource {
    public S3StreamingMessageSource(RemoteFileTemplate<S3ObjectSummary> template) {
        super(template, null);
    }

    public S3StreamingMessageSource(RemoteFileTemplate<S3ObjectSummary> template,
                                    Comparator<AbstractFileInfo<S3ObjectSummary>> comparator) {
        super(template, comparator);
    }

    @Override
    protected List<AbstractFileInfo<S3ObjectSummary>> asFileInfoList(Collection<S3ObjectSummary> collection) {
        return collection.stream()
                .map(S3FileInfo::new)
                .collect(toList());
    }
}

顺便说一句，我正在使用 Spring Integration 5.0.0.M4 和 Spring Integration AWS 1.1.0.M2，但在使用像 abc/def/ 这样的存储桶名称时仍然遇到同样的问题