我可以为 FileWritingMessageHandler 提供自定义 FileLocker 吗？

Question

我正在使用带有自定义 FileLocker 的 FileReadingMessageSource:

@Component
public class JdbcFileLocker implements FileLocker {
...

@Autowired
JdbcFileLocker jdbcFileLocker;

@Bean
@InboundChannelAdapter(value = "fileInputChannel", poller = @Poller(fixedDelay = "1000"))
public MessageSource<File> fileReadingMessageSource() {
    FileReadingMessageSource source = new FileReadingMessageSource();
    source.setDirectory(new File("/workspace/in"));
    source.setFilter(new SimplePatternFileListFilter("input"));
    source.getScanner().setLocker(jdbcFileLocker);
    return source;
}

现在我想在 FileWritingMessageHandler 中处理文件后使用 FileLocker jdbcFileLocker 解锁文件:

@Bean
@ServiceActivator(inputChannel = "fileInputChannel")
  public MessageHandler fileWritingMessageHandler() {
    FileWritingMessageHandler fileWritingMessageHandler = 
      new FileWritingMessageHandler(new File("/workspace/out"));

我是否必须在自己的 @ServiceActivator 中解锁文件，或者我可以将我的 FileLocker 以某种方式提供给 FileWritingMessageHandler 吗？

编辑：添加 ExpressionEvaluatingRequestHandlerAdvice

@Bean
ExpressionEvaluatingRequestHandlerAdvice unlockAdvice() {
  ExpressionEvaluatingRequestHandlerAdvice advice = 
    new ExpressionEvaluatingRequestHandlerAdvice();
  advice.setSuccessChannel(unlockChannel);
  return advice;
}

所以我可以将这个 Advice 添加到我的 FileWritingMessageHandler:

@Autowired
ExpressionEvaluatingRequestHandlerAdvice unlockAdvice;

@Bean
@ServiceActivator(inputChannel = "fileInputChannel")
public MessageHandler fileWritingMessageHandler() {
  FileWritingMessageHandler handler = 
    new FileWritingMessageHandler(new File("/workspace/out"));
  handler.setFileExistsMode(FileExistsMode.REPLACE);
  handler.setDeleteSourceFiles(true);
  handler.setExpectReply(false);
  handler.setFileNameGenerator(message -> "output");
  handler.setAdviceChain(List.of(unlockAdvice));
  return handler;
}

然后由监听 unlockChannel 的 @ServiceActivator 处理解锁。

Answer 1

首先考虑使用fileWritingMessageHandler.setExpectReply(false)。由于默认情况下 FileWritingMessageHandler 充当网关，并且将尝试生成未在您的流程中处理的回复，因此将向您的轮询器抛出一些错误。

对于您的 FileLocker 用例，请参阅 fileWritingMessageHandler 服务激活器上的 ExpressionEvaluatingRequestHandlerAdvice。它的 onSuccessExpression 可以进行解锁，这实际上将在 FileWritingMessageHandler 完成其过程后完成。有关详细信息，请参阅文档：https://docs.spring.io/spring-integration/docs/current/reference/html/messaging-endpoints.html#message-handler-advice-chain

另一种是将 fileInputChannel 作为 PublishSubscribeChannel，而不更改当前 fileWritingMessageHandler 的配置。但是您将另一个服务激活器订阅者添加到将要执行 FileLocker.unlock() 逻辑的频道。为确保第二个订阅者在第一个订阅者之后被调用，请考虑为您的解锁器服务激活器添加一个 @Order(100)。另请参阅文档：https://docs.spring.io/spring-integration/docs/current/reference/html/core.html#channel-implementations-publishsubscribechannel

我可以为 FileWritingMessageHandler 提供自定义 FileLocker 吗？

Can I provide a custom FileLocker for a FileWritingMessageHandler?

spring-integration

distributed-lock

spring-boot