0

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

@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;
}

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

@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;
}

然后通过@ServiceActivator监听来处理解锁unlockChannel。

4

1 回答 1

1

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

对于您的FileLocker用例，请参阅ExpressionEvaluatingRequestHandlerAdvice您的fileWritingMessageHandler服务激活器上的一个。它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

于 2021-04-26T13:35:57.870 回答