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

Question

Using Publish subscribe ,I'm Pushing local file to multiple remote folders (remoteDirectory1,remoteDirectory2(Copy to remote success and archive folder)) at the same time I need to delete the same file in another remoteDirectory3 (ex: delete from Remote processing folder)

Using Below code code I'm able to transfer the files to remote but not able to delete the file in another remote directory .

I'm getting confused with SPEL expression for expression in outboundGateway ...

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();
}

The value of string sftpProperties.getRemoteRootDir() = > /home/hari, remoteDirectory3 = /Sample/Processing

....

I think Im missing the file name in expression

if i give hard code value "'/home/hari/Sample/Processing/sample.xml'" it deleting the file but Im getting no reply channel error

Getting following error

        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

Answer 1

Please provide version information when asking questions like this; it's a pain having to look through all versions to line up line numbers.

This expression sftpProperties.getRemoteRootDir() + remoteDirectory3) is not an expression; it needs to be made into a literal

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