如何设置 ExpressionEvaluatingRequestHandlerAdvice 以阻止任务被取消答案

【问题标题】：How to set up ExpressionEvaluatingRequestHandlerAdvice to stop task from being cancelled如何设置 ExpressionEvaluatingRequestHandlerAdvice 以阻止任务被取消
【发布时间】：2023-01-11 22:52:20
【问题描述】：

我有一个场景，我不确定我是否在 Integration Flow 中设置正确。需要的是：

轮询文件的 SFTP 位置
传输所有新的/已更改的信息并使用SftpPersistentAcceptOnceFileListFilter存储该信息
失败时继续下一个可用文件

对于最后一点，我在另一个answer 中发现我可以尝试ExpressionEvaluatingRequestHandlerAdvice。我想出了以下配置，但是添加 Advice 已经完全中断了流程（没有消息流过）

ExpressionEvaluatingRequestHandlerAdvice advice = new ExpressionEvaluatingRequestHandlerAdvice();
advice.setSuccessChannel(out);
advice.setFailureChannel(err);
advice.setTrapException(true);

IntegrationFlow integrationFlow = IntegrationFlows
        .from(Sftp.inboundAdapter(cachingSessionFactory)
        .remoteDirectory("sftpSource")
        .deleteRemoteFiles(false)
        .preserveTimestamp(true)
        .localDirectory(getTargetLocalDirectory()), e -> e.id("sftpInboundAdapter")
                .poller(Pollers.fixedDelay(100)
                        .maxMessagesPerPoll(3)
                        .advice(advice)))
        .channel(out)
        .get();

跳过已归档文件传输的要求来自现实世界的场景，我们的 SFTP 服务器拒绝传输空文件。为了模拟这一点，我在SessionFactory 中添加了间谍：

CachingSessionFactory<ChannelSftp.LsEntry> cachingSessionFactory = Mockito.spy(sessionFactory());
CachingSessionFactory.CachedSession session = (CachingSessionFactory.CachedSession) Mockito.spy(cachingSessionFactory.getSession());

doReturn(session).when(cachingSessionFactory).getSession();
doThrow(new RuntimeException("test exception")).when(session).read(contains("sftpSource2.txt"), any(OutputStream.class));

和测试代码：

Message<?> message = out.receive(1000);
assertThat(message).isNotNull();
Object payload = message.getPayload();
assertThat(payload).isInstanceOf(File.class);
File file = (File) payload;
assertThat(file.getName()).isEqualTo(" sftpSource1.txt");
assertThat(file.getAbsolutePath()).contains("localTarget");

message = out.receive(1000);
assertThat(message).isNull();

message = err.receive(1000);
System.out.println("error was:" + message.getPayload());

message = out.receive(1000);
assertThat(message).isNotNull();
file = (File) message.getPayload();
assertThat(file.getName()).isIn("sftpSource3.txt");
assertThat(file.getAbsolutePath()).contains("localTarget");

我感到困惑的是 - 当我将 advice 添加到 Poller 时，我应该从 Poller 中删除 .errorChannel(err) 吗？但是如果建议是处理消息结束的地方，我不应该也删除 integrationFlow 上的 .channel(out) 吗？没有它，IntegrationFlow 将无法构建，并出现错误 outputChannel is required。

我的第二个担心是 - 如果 advice.setTrapException(true); 是否意味着 SftpPersistentAcceptOnceFileListFilter 会将文件标记为已成功处理？（过滤器不在示例代码中，但我会在实际代码中需要它）。

【问题讨论】：

标签： java spring-integration spring-integration-dsl spring-integration-sftp

【解决方案1】：

ExpressionEvaluatingRequestHandlerAdvice 用于消费者端点，而不是源轮询通道适配器。因此，您无法使用该建议处理来自轮询器的错误。

事实上，你不能只有一个 IntegrationFlow 和一个 SourcePollingChannelAdapter - 你需要以任何一种方式添加 channel(out) 。

您当前的配置不清楚您将这些文件传输到哪里。所以，您可能正在谈论处理这些文件的错误，ExpressionEvaluatingRequestHandlerAdvice 在那里有意义。通过这种方式，您所有的远程文件实际上都会传输到本地目录并在 SftpPersistentAcceptOnceFileListFilter 中标记为已观看。只是因为您不会将异常从处理程序抛回给 SFTP 入站通道适配器。

【讨论】：