JudgingNotJudging JudgingNotJudging - 3 months ago 64
Java Question

Spring Integration DSL - Outbound Gateway with access to Headers

I'm having an issue with Spring Integration. I'm using Spring Boot 1.4.0.RELEASE, Spring Integration 4.3.1.RELEASE, Spring Integration DSL 1.2.0.M1.

What I'm trying to do:

I'm writing an application that will read files from FTP and local file system (using inbound channel adapters), transfer the files to a local working directory (using file outbound gateways), process, then move them to a final destination (file outbound gateway/adapters).

EDIT: The original issue stemmed from incorrect usage of the

OutboundGateway
. For details of what I was doing wrong, look at the edit history.

I'm having a problem with wiring an
OutboundGateway
that needs information from the message header.

My basic flow is:

@Bean
@SuppressWarnings("unchecked")
public IntegrationFlow fileSystemReadingFlow() {
LOGGER.debug("Building fileSystemReadingFlow.");
// Create a FileInboundChannelAdapter

return IntegrationFlows.from(s -> s.file(new File(StringUtils.cleanPath(Paths.get(directoryProperties.getLocal() , UNKNOWN).toString())))
// Add a Filter to restrict which files are retrieved
.filter(fileListFilterBuilder.buildFileListFilter(File.class))
// Add a poller to continually watch the directory
, endpointConfigurer -> endpointConfigurer.poller(poller)
)
.enrichHeaders(h -> h.header(FOLDER_NAME, LOCAL))
.handle((message, headers) -> Files.outboundGateway(new File( StringUtils.cleanPath(Paths.get(directoryProperties.getWorking()
, (String) headers.get(FOLDER_NAME)).toString()))).deleteSourceFiles(true).autoCreateDirectory(true) )
// Send the files to the aggregatingChannel
.channel("aggregatingFileChannel")
.get();
}


The issue is the
handle((message, headers) -> Files.outboundGateway(...)
section. Using the lambda method above spits out this error over and over.

ErrorMessage [payload=org.springframework.messaging.MessagingException: failed to resolve channel name 'org.springframework.integration.dsl.file.FileWritingMessageHandlerSpec'; nested exception is org.springframework.messaging.core.DestinationResolutionException: failed to look up MessageChannel with name 'org.springframework.integration.dsl.file.FileWritingMessageHandlerSpec' in the BeanFactory.; nested exception is org.springframework.beans.factory.NoSuchBeanDefinitionException: No bean named 'org.springframework.integration.dsl.file.FileWritingMessageHandlerSpec' is defined, headers={id=e3f5f341-6379-0dd0-11e6-e4bff8f455b0, timestamp=1471619276894}]


The documentation and my testing show that defining it like below works. But, how do I get access to the headers?

.handle(Files.outboundGateway(new File( "someDir"))).deleteSourceFiles(true).autoCreateDirectory(true) )


I've tried using a SpEl expression, which may be the right method, but I didn't get it to work.

.handle(Files.outboundGateway("StringUtils.cleanPath(Paths.get(directoryProperties.getWorking()" +
", headers[FOLDER_NAME]).toString())").deleteSourceFiles(true).autoCreateDirectory(true) )

Answer

After re-reading the SpEL reference, I was able to figure out my problem. I needed to explicitly reference the static methods I needed to call and then escape the literals properly.

.handle(Files.outboundGateway("T(org.springframework.util.StringUtils).cleanPath(T(java.nio.file.Paths).get('" +directoryProperties.getWorking() +"'" +
                                    ", headers['"+FOLDER_NAME+"']).toString())").deleteSourceFiles(true).autoCreateDirectory(true) )
Comments