时间戳

首页 > 解决方案 > 在 SFTP 中将文件传输到一个远程文件夹并需要删除另一个远程文件夹中的相同文件

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

问题描述

使用 Publish subscribe ，我将本地文件推送到多个远程文件夹（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$000(AbstractPollingEndpoint.java:58)
    at org.springframework.integration.endpoint.AbstractPollingEndpoint$1.call(AbstractPollingEndpoint.java:190)
    at org.springframework.integration.endpoint.AbstractPollingEndpoint$1.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$1.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$1.run(AbstractPollingEndpoint.java:353)
    at org.springframework.integration.util.ErrorHandlingTaskExecutor$1.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.access$201(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

标签： spring-integrationspring-integration-dslspring-integration-sftp

解决方案

提出此类问题时请提供版本信息；必须查看所有版本来排列行号是一件很痛苦的事情。

这个表达sftpProperties.getRemoteRootDir() + remoteDirectory3)不是表达；它需要变成文字

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

推荐阅读