在 SFTP 中将文件传输到一个远程文件夹并需要删除另一个远程文件夹中的相同文件

Question

使用发布订阅，我将本地文件推送到多个远程文件夹（remoteDirectory1、remoteDirectory2（复制到远程成功和存档文件夹））同时我需要删除另一个 remoteDirectory3 中的相同文件（例如：从远程处理文件夹中删除）

使用下面的代码我可以将文件传输到远程但不能删除另一个远程目录中的文件。

我对 outboundGateway 中表达式的 SPEL 表达式感到困惑...

private IntegrationFlow pushRemoteFolder(final String localDirectory, String remoteDirectory1, String remoteDirectory2,String remoteDirectory3,String adapterName, String filePattern, String fileRenameExpression) {

    return IntegrationFlows
            .from(Files.inboundAdapter(Paths.get(localDirectory).toFile())
                            .regexFilter(filePattern)
                            .preventDuplicates(false),
                    e -> {
                        e.poller(Pollers.fixedDelay(SftpProperties.getPollerIntervalMs())
                                .maxMessagesPerPoll(SftpProperties.getMaxFetchSize())
                                .errorChannel("errorChannel")
                                .transactional(transactionManager)
                                .transactionSynchronizationFactory(PushSftpSyncFactory()) // moves processed files
                        ).id(adapterName);
                    })
            .publishSubscribeChannel(s -> s
                    .subscribe(f -> f
                            .handle(Sftp.outboundAdapter(PushSftpSessionFactory())
                                    .remoteDirectory(sftpProperties.getRemoteRootDir() + remoteDirectory1)
                                    .fileNameExpression(fileRenameExpression) //.fileNameExpression("payload.getName().replace('A','B')")
                                    .temporaryFileSuffix(".tmp")))
                    .subscribe(f -> f
                            .handle(Sftp.outboundAdapter(PushSftpSessionFactory())
                                    .remoteDirectory(sftpProperties.getRemoteRootDir() + remoteDirectory2)
                                    .fileNameExpression(fileRenameExpression)
                                    .temporaryFileSuffix(".tmp")))
                    .subscribe(f -> f
                            .handle(Sftp.outboundGateway(PushSftpSessionFactory()
                                    ,AbstractRemoteFileOutboundGateway.Coand.RM
                                    ,"'" + mmSftpProperties.getRemoteRootDir() + remoteDirectory3 + "/' + payload.name")))

            ).get();
}

字符串的值 sftpProperties.getRemoteRootDir() = > /home/hari, remoteDirectory3 = /Sample/Processing

....

我想我在表达式中缺少文件名

如果我给出硬代码值“'/home/hari/Sample/Processing/sample.xml'”，它会删除文件，但我没有收到回复通道错误

出现以下错误

        16:57:13.447 [task-scheduler-5] ERROR o.s.i.handler.LoggingHandler - org.springframework.messaging.MessagingException: Failed to handle Message; nested exception is org.springframework.messaging.core.DestinationResolutionException: no output-channel or replyChannel header available, failedMessage=GenericMessage [payload=local\outbound\Sample.xml, headers={id=4cdd9962-cb42-7bc1-d30e-b999d693b22e, timestamp=1548780989522}]
    at org.springframework.integration.dispatcher.BroadcastingDispatcher.invokeHandler(BroadcastingDispatcher.java:242)
    at org.springframework.integration.dispatcher.BroadcastingDispatcher.dispatch(BroadcastingDispatcher.java:185)
    at org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:89)
    at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:425)
    at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:375)
    at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:115)
    at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:45)
    at org.springframework.messaging.core.AbstractMessageSendingTemplate.send(AbstractMessageSendingTemplate.java:105)
    at org.springframework.integration.endpoint.SourcePollingChannelAdapter.handleMessage(SourcePollingChannelAdapter.java:210)
    at org.springframework.integration.endpoint.AbstractPollingEndpoint.doPoll(AbstractPollingEndpoint.java:272)
    at org.springframework.integration.endpoint.AbstractPollingEndpoint.access[=12=]0(AbstractPollingEndpoint.java:58)
    at org.springframework.integration.endpoint.AbstractPollingEndpoint.call(AbstractPollingEndpoint.java:190)
    at org.springframework.integration.endpoint.AbstractPollingEndpoint.call(AbstractPollingEndpoint.java:186)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:498)
    at org.springframework.aop.support.AopUtils.invokeJoinpointUsingReflection(AopUtils.java:333)
    at org.springframework.aop.framework.ReflectiveMethodInvocation.invokeJoinpoint(ReflectiveMethodInvocation.java:190)
    at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:157)
    at org.springframework.transaction.interceptor.TransactionInterceptor.proceedWithInvocation(TransactionInterceptor.java:99)
    at org.springframework.transaction.interceptor.TransactionAspectSupport.invokeWithinTransaction(TransactionAspectSupport.java:282)
    at org.springframework.transaction.interceptor.TransactionInterceptor.invoke(TransactionInterceptor.java:96)
    at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:179)
    at org.springframework.aop.framework.JdkDynamicAopProxy.invoke(JdkDynamicAopProxy.java:213)
    at com.sun.proxy.$Proxy94.call(Unknown Source)
    at org.springframework.integration.endpoint.AbstractPollingEndpoint$Poller.run(AbstractPollingEndpoint.java:353)
    at org.springframework.integration.util.ErrorHandlingTaskExecutor.run(ErrorHandlingTaskExecutor.java:55)
    at org.springframework.core.task.SyncTaskExecutor.execute(SyncTaskExecutor.java:50)
    at org.springframework.integration.util.ErrorHandlingTaskExecutor.execute(ErrorHandlingTaskExecutor.java:51)
    at org.springframework.integration.endpoint.AbstractPollingEndpoint$Poller.run(AbstractPollingEndpoint.java:344)
    at org.springframework.scheduling.support.DelegatingErrorHandlingRunnable.run(DelegatingErrorHandlingRunnable.java:54)
    at org.springframework.scheduling.concurrent.ReschedulingRunnable.run(ReschedulingRunnable.java:81)
    at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
    at java.util.concurrent.FutureTask.run$$$capture(FutureTask.java:266)
    at java.util.concurrent.FutureTask.run(FutureTask.java)
    at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access1(ScheduledThreadPoolExecutor.java:180)
    at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    at java.lang.Thread.run(Thread.java:745)
Caused by: org.springframework.messaging.core.DestinationResolutionException: no output-channel or replyChannel header available
    at org.springframework.integration.handler.AbstractMessageProducingHandler.sendOutput(AbstractMessageProducingHandler.java:355)
    at org.springframework.integration.handler.AbstractMessageProducingHandler.produceOutput(AbstractMessageProducingHandler.java:271)
    at org.springframework.integration.handler.AbstractMessageProducingHandler.sendOutputs(AbstractMessageProducingHandler.java:188)
    at org.springframework.integration.handler.AbstractReplyProducingMessageHandler.handleMessageInternal(AbstractReplyProducingMessageHandler.java:115)
    at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:127)
    at org.springframework.integration.dispatcher.BroadcastingDispatcher.invokeHandler(BroadcastingDispatcher.java:236)
    ... 40 more

Answer 1

此类问题请提供版本信息；必须浏览所有版本以排列行号是一件痛苦的事情。

这个表达式 sftpProperties.getRemoteRootDir() + remoteDirectory3) 不是一个表达式；需要做成文字

"'" + sftpProperties.getRemoteRootDir() + remoteDirectory3) + "'"

在 SFTP 中将文件传输到一个远程文件夹并需要删除另一个远程文件夹中的相同文件

In SFTP Transfer a file to one remote folder and need to delete the same file in another remote folder

spring-integration

spring-integration-dsl

spring-integration-sftp