spring-integration- aws S3StreamingMessageSource как мне удалить удаленный файл?

Question

В настоящее время я работаю с S3StreamingMessageSource из Spring интеграции aws. Я передаю поток в поток интеграции.

public MessageSource<InputStream> s3InboundStreamingMessageSource() {
    S3StreamingMessageSource messageSource = new S3StreamingMessageSource(template());
    messageSource.setRemoteDirectory(bucketName);
    messageSource.setFilter(new S3PersistentAcceptOnceFileListFilter(new SimpleMetadataStore(),
            "streaming"));
    return messageSource;
}

@Bean
public IntegrationFlow s3IntegrationFlow() {
    return IntegrationFlows.from(s3InboundStreamingMessageSource(), spec -> spec.poller(Pollers.fixedDelay(10, TimeUnit.SECONDS)))
            .transform(new S3ObjectInputStreamToStringTransformer())
            .transform(Transformers.toJson())
            .handle(Http.outboundGateway("http://localhost:8099/create").httpMethod(HttpMethod.POST).extractPayload(true))
            .channel("nullChannel")
            .get();
}

Как я могу удалить полученный удаленный файл из S3?

В S3InboundFileSynchronizer есть метод для этого.

Вот так:

    @Bean
public S3InboundFileSynchronizer s3InboundFileSynchronizer() {
    S3InboundFileSynchronizer synchronizer = new S3InboundFileSynchronizer(factory);
    synchronizer.setDeleteRemoteFiles(true);
    synchronizer.setPreserveTimestamp(true);
    synchronizer.setRemoteDirectory(bucketName);
    return synchronizer;
}

Может ли кто-нибудь мне помочь или подсказать хорошее решение?

Answer 1

У адаптера потокового канала нет локальной копии удаленного файла, поэтому мы не можем угадать, что и как вы собираетесь делать с InputStream для удаленного файла. Вот почему на S3StreamingMessageSource нет setDeleteRemoteFiles().

Я вижу, вы делаете что-то вроде этого S3ObjectInputStreamToStringTransformer. Подскажите, пожалуйста, в чем причина такого нестандартного трансформатора. Уже существует StreamTransformer, и с его опцией charset InputStream для удаленного файла будет преобразовано в строку:

/**
 * Construct an instance with the charset to convert the stream to a
 * String; if null a {@code byte[]} will be produced instead.
 * @param charset the charset.
 */
public StreamTransformer(String charset) {

Также: необходимо иметь в виду, что StaticMessageHeaderAccessor.getCloseableResource(message) должен быть закрытым после чтения InputStream, чтобы избежать утечки ресурсов.

Вероятно, вместо .channel("nullChannel") вам следует подумать об использовании handle() для вызова AmazonS3.deleteObject(String bucketName, String key) API. bucketName сохраняется в FileHeaders.REMOTE_DIRECTORY и key в заголовках FileHeaders.REMOTE_FILE соответственно.