Используя опцию «Публикация подписки», я помещаю локальный файл в несколько удаленных папок (remoteDirectory1, remoteDirectory2 (Копирование в удаленную папку успешно и в архивную папку)), одновременно мне нужно удалить этот же файл в другом remoteDirectory3 (например, удалить изПапка удаленной обработки)
Использование приведенного ниже кода кода Я могу перенести файлы на удаленный сервер, но не могу удалить файл в другом удаленном каталоге.
Я запутался с выражением SPELдля выражения в 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();
}
Значение строки 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