使用Spring集成SFTP遞歸讀取文件DSL

Question

我試圖從所有子文件夾遞歸地從遠程服務器上使Spring集成SFTP讀取文件（.txt）。遠程文件夾類似於「/ tmp/remoteFolder」，所有子文件夾都是日期文件夾，如「/ tmp/remoteFolder/20180830」，「/ tmp/remoteFolder/20170902」。

這是我到現在爲止

@Bean 
@InboundChannelAdapter(value = "sftpMgetInputChannel", 
    poller = @Poller(fixedDelay = "5000")) 
public IntegrationFlow sftpMGetFlow() { 
    return IntegrationFlows.from("sftpMgetInputChannel") 
      .handleWithAdapter(h -> h.sftpGateway(this.sftpSessionFactory, 
      Command.MGET, "'/tmp/remoteDirectory/*'") 
      .options(Option.RECURSIVE) 
      .regexFileNameFilter("((\\d{8})|*\\.txt)") 
      .localDirectoryExpression("sftp-inbound" + "/" + "#remoteDirectory")) 
      .channel(remoteFileOutputChannel()) 
      .get(); 
} 

@Bean 
public MessageChannel sftpMgetInboundChannel(){ 
    return new DirectChannel(); 
} 

@Bean 
public PollableChannel remoteFileOutputChannel() { 
    return new QueueChannel(); 
} 

如何指定根遠程目錄SFTP MGET是的/ tmp/remoteFolder的代碼？爲什麼這不起作用？爲什麼我需要指定輸出通道？

更新：而不是調用的通道（remoteFileOutputChannel（））我這樣調用

@Bean 
public MessageHandler messageHandler(){ 
return new MessageHandler() { ... } 
} 

代碼的處理程序更新：

 

    @InboundChannelAdapter(value = "sftpMgetInputChannel", 
     poller = @Poller(fixedDelay = "5000")) 
    public String filesForMGET(){ 
     return "'/tmp/input/remoteDirectory/*'"; 
    } 

    @Bean 
    public IntegrationFlow sftpMGetFlow() { 
     return IntegrationFlows.from("sftpMgetInputChannel") 
       .handleWithAdapter(h -> h.sftpGateway(this.sftpSessionFactory, 
       Command.MGET, "payload") 
       .options(Option.RECURSIVE) 
       .regexFileNameFilter("((\\d{8})|*\\.txt)") 
       .localDirectoryExpression("'sftp-inbound/'" + "#remoteDirectory")) 
       .handler(messageHandler()) 
       .get(); 
    } 

    @Bean 
    public MessageChannel sftpMgetInboundChannel(){ 
     return new DirectChannel(); 
    } 

    @Bean 
    public MessageHandler messageHandler(){ 
    return new MessageHandler() { ... } 
    } 

有了這個更新的代碼，我得到以下錯誤：

 

    rg.springframework.core.NestedIOException: failed to read file; nested exception is 2: No such file 
     at org.springframework.integration.sftp.session.SftpSession.read(SftpSession.java:100) 
     at org.springframework.integration.file.remote.session.CachingSessionFactory$CachedSession.read(CachingSessionFactory.java:137) 
     at org.springframework.integration.file.remote.synchronizer.AbstractInboundFileSynchronizer.copyFileToLocalDirectory(AbstractInboundFileSynchronizer.java:176) 
     at org.springframework.integration.file.remote.synchronizer.AbstractInboundFileSynchronizer.synchronizeToLocalDirectory(AbstractInboundFileSynchronizer.java:138) 
     at org.springframework.integration.file.remote.synchronizer.AbstractInboundFileSynchronizingMessageSource.receive(AbstractInboundFileSynchronizingMessageSource.java:144) 
     at org.springframework.integration.endpoint.SourcePollingChannelAdapter.doPoll(SourcePollingChannelAdapter.java:89) 
     at org.springframework.integration.endpoint.AbstractPollingEndpoint$1.call(AbstractPollingEndpoint.java:146) 
     at org.springframework.integration.endpoint.AbstractPollingEndpoint$1.call(AbstractPollingEndpoint.java:144) 
     at org.springframework.integration.endpoint.AbstractPollingEndpoint$Poller$1.run(AbstractPollingEndpoint.java:207) 
     at org.springframework.integration.util.ErrorHandlingTaskExecutor$1.run(ErrorHandlingTaskExecutor.java:52) 
     at org.springframework.core.task.SyncTaskExecutor.execute(SyncTaskExecutor.java:48) 
     at org.springframework.integration.util.ErrorHandlingTaskExecutor.execute(ErrorHandlingTaskExecutor.java:49) 
     at org.springframework.integration.endpoint.AbstractPollingEndpoint$Poller.run(AbstractPollingEndpoint.java:202) 
     at org.springframework.scheduling.support.DelegatingErrorHandlingRunnable.run(DelegatingErrorHandlingRunnable.java:51) 
     at org.springframework.scheduling.concurrent.ReschedulingRunnable.run(ReschedulingRunnable.java:81) 
     at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:441) 
     at java.util.concurrent.FutureTask$Sync.innerRun(FutureTask.java:303) 
     at java.util.concurrent.FutureTask.run(FutureTask.java:138) 
     at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$301(ScheduledThreadPoolExecutor.java:98) 
     at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:206) 
     at java.util.concurrent.ThreadPoolExecutor$Worker.runTask(ThreadPoolExecutor.java:886) 
     at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:908) 
     at java.lang.Thread.run(Thread.java:680) 
    Caused by: 2: No such file 
     at com.jcraft.jsch.ChannelSftp.throwStatusError(ChannelSftp.java:2289) 
     at com.jcraft.jsch.ChannelSftp._stat(ChannelSftp.java:1741) 
     at com.jcraft.jsch.ChannelSftp.get(ChannelSftp.java:1011) 
     at com.jcraft.jsch.ChannelSftp.get(ChannelSftp.java:986) 
     at org.springframework.integration.sftp.session.SftpSession.read(SftpSession.java:96) 
     ... 22 more 

Answer 1

將表達式設置爲​​（就像您的情況一樣在編輯之前的問題），發送到網關的消息有效載荷應該是/tmp/remoteFolder/*，它在內部被分成遠程目錄和遠程文件名（*）。

Why do I need to specifiy the output channel?

MGET（檢索到的文件列表）的結果需要去某處。

編輯

你misunderstond;您不能將@InboundChannelAdapter註釋添加到流程中;你需要這樣的東西...

@InboundChannelAdapter(value = "sftpMgetInputChannel", 
    poller = @Poller(fixedDelay = "5000")) 
public String filesForMGET() { 
    return "/tmp/remoteDirectory/"; 
} 

@Bean 
public IntegrationFlow sftpMGetFlow() { 
    return IntegrationFlows.from("sftpMgetInputChannel") 
      .handleWithAdapter(h -> h.sftpGateway(this.sftpSessionFactory, 
      Command.MGET, "payload") 
      .options(Option.RECURSIVE) 
      .regexFileNameFilter("((\\d{8})|*\\.txt)") 
      .localDirectoryExpression("sftp-inbound" + "/" + "#remoteDirectory")) 
      .channel(remoteFileOutputChannel()) 
      .get(); 
}