使用 Spring Integration 在 sftp 中复制远程文件

Question

我需要在 Sftp 服务器中复制/复制远程文件，在复制时也要重命名，我读到 here 不支持在 Sftp 中复制远程文件，所以我唯一可用的选项是 GET 文件到本地，然后 PUT 再次到 Sftp 并删除本地文件，我已经成功实现了我的目标，但问题是有一个来自org.springframework.core.log.LogAccessor: 的日志打印我不知道它来自哪里来了。

有助于复制远程文件的代码：

 @Bean
public IntegrationFlow copyRemoteFile() {
    return IntegrationFlows.from("integration.channel.copy")
            .handle(Sftp.outboundGateway(sftpSessionFactory(),
                    AbstractRemoteFileOutboundGateway.Command.GET,
                    "headers[" + COPY_SOURCE_PATH.value + "]+'/'+" +
                            "headers[" + COPY_SOURCE_FILENAME.value + "]")
                    .autoCreateLocalDirectory(true)
                    .fileExistsMode(FileExistsMode.REPLACE)
                    .localDirectory(new File(localPath)))
            .log(LoggingHandler.Level.INFO, "SftpCopyService")
            .handle(Sftp.outboundGateway(sftpSessionFactory(),
                    AbstractRemoteFileOutboundGateway.Command.PUT,
                    "payload")
                    .remoteDirectoryExpression("headers[" + COPY_DEST_PATH.value + "]")
                    .fileNameGenerator(n -> (String)n.getHeaders().get(COPY_DEST_FILENAME.value))
                    .fileExistsMode(FileExistsMode.REPLACE))
            .log(LoggingHandler.Level.INFO, "SftpCopyService")
            .handle((p, h) -> {
                try {
                      return Files.deleteIfExists(
                            Paths.get(localPath + File.separator + h.get(COPY_SOURCE_FILENAME.value)));
                } catch (IOException e) {
                    e.printStackTrace();
                    return false;
                }
            })
            .get();

这是日志。

2021-02-16 18:10:22,577 WARN  [http-nio-9090-exec-1] org.springframework.core.log.LogAccessor: Failed to delete C:\Users\DELL\Desktop\GetTest\Spring Integration.txt
2021-02-16 18:10:22,784 INFO  [http-nio-9090-exec-1] org.springframework.core.log.LogAccessor: GenericMessage [payload=C:\Users\DELL\Desktop\GetTest\Spring Integration.txt, headers={file_remoteHostPort=X.X.X.X:22, replyChannel=nullChannel, sourceFileName=Spring Integration.txt, file_remoteDirectory=/uploads/, destFileName=Spring Integrat.txt, destPath=uploads/dest, id=5105bdd1-8180-1185-3661-2ed708e07ab9, sourcePath=/uploads, file_remoteFile=Spring Integration.txt, timestamp=1613479222779}]
2021-02-16 18:10:23,011 INFO  [http-nio-9090-exec-1] org.springframework.core.log.LogAccessor: GenericMessage [payload=uploads/dest/Spring Integrat.txt, headers={file_remoteHostPort=X.X.X.X:22, replyChannel=nullChannel, sourceFileName=Spring Integration.txt, file_remoteDirectory=/uploads/, destFileName=Spring Integrat.txt, destPath=uploads/dest, id=1bf83b0f-3b24-66bd-ffbf-2a9018b499fb, sourcePath=/uploads, file_remoteFile=Spring Integration.txt, timestamp=1613479223011}]

更令人惊讶的是，它甚至在流程执行之前就出现得很早，尽管我最后处理了文件删除。我怎样才能摆脱这个日志消息？虽然它不会影响我的过程，但日志消息具有误导性

还有没有更好的方法将远程文件复制到 sftp 中的另一个路径

编辑

就像你建议的那样，我尝试使用 SftpRemoteFileTemplate.execute() 方法在 sftp 中复制文件，但是当调用 session.write(InputStream stream,String path) 方法时，方法控件永远不会返回，它会永远保留控件

我试过调试，执行到这里就失去控制了：

for(_ackcount = this.seq - startid; _ackcount > ackcount && this.checkStatus((int[])null, header); ++ackcount) {
            }

这段代码位于ChannelSftp.class的_put方法中

这是我正在尝试的示例代码

 public boolean copy() {
   return remoteFileTemplate.execute(session -> {
        if (!session.exists("uploads/Spring Integration.txt")){
            return false;
        }
        if (!session.exists("uploads/dest")){
            session.mkdir("uploads/dest");
        }
        InputStream inputStream = session.readRaw("uploads/Spring Integration.txt");
        session.write(inputStream, "uploads/dest/spring.txt");
        session.finalizeRaw();
        return true;
    });
}

你能指出我在这里犯了什么错误吗？

Answer 1

我建议不要通过本地文件副本编写整个流程，而是查看SftpRemoteFileTemplate.execute(SessionCallback<F, T>) 的单个服务激活器。该回调中提供的SftpSession 可用于InputStream readRaw() 和write(InputStream inputStream, String destination)。最后你必须打电话给finalizeRaw()。

LogAccessor 问题尚不清楚。您使用什么 Spring Integration 版本？你会覆盖 Spring Core 版本吗？

我认为我们可以改进该 WARN 消息，如果没有 exists()，请不要调用 File.delete()。

随时提供这样的贡献！

更新

演示如何在 SFTP 服务器上执行复制的 JUnit 测试：

@Test
public void testSftpCopy() throws Exception {
    this.template.execute(session -> {
        PipedInputStream in = new PipedInputStream();
        PipedOutputStream out = new PipedOutputStream(in);
        session.read("sftpSource/sftpSource2.txt", out);
        session.write(in, "sftpTarget/sftpTarget2.txt");
        return null;
    });

    Session<?> session = this.sessionFactory.getSession();
    ByteArrayOutputStream baos = new ByteArrayOutputStream();
    FileCopyUtils.copy(session.readRaw("sftpTarget/sftpTarget2.txt"), baos);
    assertThat(session.finalizeRaw()).isTrue();
    assertThat(new String(baos.toByteArray())).isEqualTo("source2");

    baos = new ByteArrayOutputStream();
    FileCopyUtils.copy(session.readRaw("sftpSource/sftpSource2.txt"), baos);
    assertThat(session.finalizeRaw()).isTrue();
    assertThat(new String(baos.toByteArray())).isEqualTo("source2");

    session.close();
}