0

我们正在使用与 S3 的弹簧集成。我们有 s3-inbound-streaming-channel-adapter 可以从 S3 读取。发生的情况是，如果“get”失败，s3-inbound-streaming-channel-adapter 将文件名放在“acceptOnceFilter”中，并且不会重试失败。

Q1。我们想要的是当 s3-inbound-streaming-channel-adapter 从 S3“获取”一个文件并说由于某种原因这个“获取”失败......我们如何让 s3-inbound-streaming-channel-adapter 到再次为同一文件重试此“获取”请求？

Q2。失败时，会从 s3-inbound-streaming-channel-adapter 向默认的“errorChannel”发送异常。异常中的消息是否会包含失败的“文件名”？

<int:channel id="s3FileProcessingChannel">
  <int:queue capacity="15"/>
</int:channel>

<bean id="metadataStore" class="org.springframework.integration.metadata.SimpleMetadataStore"/>

<bean id="acceptOnceFilter"
  class="org.springframework.integration.aws.support.filters.S3PersistentAcceptOnceFileListFilter">
  <constructor-arg index="0" ref="metadataStore"/>
  <constructor-arg index="1" value="streaming"/>
</bean>

<int-aws:s3-inbound-streaming-channel-adapter id="s3Region1"
channel="s3FileProcessingChannel" 
session-factory="s3SessionFactory"
filter="acceptOnceFilter"
remotedirectoryexpression="'${s3.sourceBucket}/emm'">

  <int:poller fixed-delay="1000" max-messages-per-poll="15"/>
</int-aws:s3-inbound-streaming-channel-adapter>

谢谢总经理

4

1 回答 1

1

S3PersistentAcceptOnceFileListFilter工具：

/**
 * A {@link FileListFilter} that can be reset by removing a specific file from its
 * state.
 * @author Gary Russell
 * @since 4.1.7
 *
 */
public interface ResettableFileListFilter<F> extends FileListFilter<F> {

    /**
     * Remove the specified file from the filter so it will pass on the next attempt.
     * @param f the element to remove.
     * @return true if the file was removed as a result of this call.
     */
    boolean remove(F f);

}

并S3StreamingMessageSource填充如下标题：

return getMessageBuilderFactory()
                    .withPayload(session.readRaw(remotePath))
                    .setHeader(IntegrationMessageHeaderAccessor.CLOSEABLE_RESOURCE, session)
                    .setHeader(FileHeaders.REMOTE_DIRECTORY, file.getRemoteDirectory())
                    .setHeader(FileHeaders.REMOTE_FILE, file.getFilename())
                    .setHeader(FileHeaders.REMOTE_FILE_INFO,
                            this.fileInfoJson ? file.toJson() : file);

当错误发生时，您只需要使用它FileHeaders.REMOTE_FILE来调用上面提到的内容，remove()并且您的失败文件将在下一个轮询周期从 S3 中提取。

于 2018-09-24T16:08:39.480 回答