Service activaor behaving erratically

[anirbanzeus]

Hi Guys,

I am trying to copy files from the landing directory to a destination directory via inbound/outbound channel adapters. In the process I am using two service activators, one for the movement between inbound and outbound channel and the second one for delegating from the outbound channel to a seperate class that is meant inform about the completion of the copy and trigger further processing on the file.
The problem is that the second activator is picking up files that has not moved from source to destination and triggering further processing thus resulting in exception.

Not sure if I am doing anything wrong in the wiring, any help would be highly appreciated

configuration file:

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
	   xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
	   xmlns:integration="http://www.springframework.org/schema/integration"
	   xmlns:file="http://www.springframework.org/schema/integration/file"
	xsi:schemaLocation="http://www.springframework.org/schema/beans
						http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
						http://www.springframework.org/schema/integration/file
						http://www.springframework.org/schema/integration/file/spring-integration-file-2.2.xsd 
						http://www.springframework.org/schema/integration
						http://www.springframework.org/schema/integration/spring-integration-2.2.xsd">
	
	
	<file:inbound-channel-adapter id="filesIn" directory="file:C:/test/tmpInbound" scanner="myscanner">
		<integration:poller id="poller" fixed-delay="1000" />
	</file:inbound-channel-adapter>	
	
	
	<integration:service-activator input-channel="filesIn" output-channel="filesOut" ref="handler" />	
	 <file:outbound-channel-adapter id="filesOut" directory="file:C:/test/DR" delete-source-files="false"	filename-generator="writer"/>	
	<integration:service-activator input-channel="filesOut" ref="complete"/>	
	
	 
	<bean class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer" />
	<bean id="myscanner" class="org.springframework.integration.file.RecursiveLeafOnlyDirectoryScanner" />
	<bean id="handler" class="com.watcher.Handler" />
	<bean id="writer" class="com.watcher.Writer" />
	<bean id="complete" class="com.watcher.Complete" />
</beans>

The output:

************start***************
generateFileName(C:\test\tmpInbound\Bitmap Image0.bmp)
{timestamp=1358314500737, id=575d8e7d-a959-4b8a-b5b6-19689588319c}
Bitmap Image0.bmp
Original ---> C:\test\tmpInbound\Bitmap Image0.bmp
Final --> \Bitmap Image0.bmp0.13921170863669796
*************done***************

Jan 16, 2013 12:35:01 AM org.springframework.integration.file.FileReadingMessageSource receive
INFO: Created message: [[Payload=C:\test\tmpInbound\Bitmap Image1.bmp][Headers={timestamp=1358314501783, id=6f807524-33a1-4ed5-8e86-907c9a63f6e2}]]
[COMPLETE] C:\test\tmpInbound\Bitmap Image1.bmp
Jan 16, 2013 12:35:02 AM org.springframework.integration.file.FileReadingMessageSource receive
INFO: Created message: [[Payload=C:\test\tmpInbound\Bitmap Image2.bmp][Headers={timestamp=1358314502799, id=33a7932c-0d2a-444f-bf5a-a85f0bfac6f7}]]

************start***************
generateFileName(C:\test\tmpInbound\Bitmap Image2.bmp)
{timestamp=1358314502799, id=8a2695bf-0b86-4cfa-9d22-f97e110258c0}
Bitmap Image2.bmp
Original ---> C:\test\tmpInbound\Bitmap Image2.bmp
Final --> \Bitmap Image2.bmp0.19998457677029258
*************done***************

As you can see in the result above Bitmap Image1.bmp is being picked up by the second activator eventhough the file has not been moved and also the activator is not all being called for the remaining two files that have been moved.

Thanks in advance
Anirban

[Gary Russell]

You have two subscribers to channel 'filesOut' (outbound channel adapter and second service activator). By default, the framework will round-robin messages to these subscribers.

[Gary Russell]

If you want the message to go to both the outbound adapter and the service-activator, define filesOut as a <publish-subscribe-channel/>.

The subscribers will be called in turn (the service activator after the adapter) but to be safe, you should add order="1" to the adapter and order="2" to the service activator.

[anirbanzeus]

Thank you so much Gary for the solution , the flow is now working smoothly... and I have also removed the duplicate post, the review post screen made me think something had gone wrong , so I reposted it.

However one quick question on the file copy part.
In the process of moving a file from the landing to the destination directory, is there anyway I can force a wait on the inbound-channel-adapter for the duration that a large file is landing in the input directory ?

Thanks
Anirban

[Gary Russell]

That is a common problem for this kind if application, regardless of whether you are using Spring Integration. A common approach (used by the outbound adapter) is to write the file with a temporary name (e.g. foo.txt.writing) and rename to (e.g. to foo.txt) when the copy is complete.

If you can't control the sender, you could write a custom FileListFilter that only exposes the file if the size hasn't changed for so many seconds/minutes, but the rename technique is better.

[anirbanzeus]

Thank you Gary, I am allready trying to ensure a temp file policy to negate this issue.

Incidentally there seems to be another issue with setting custom header values on teh message payload. I have set in some user-defined values from within the transformer component. But when this message payload reached the next inline pub-sub channel, the header is getting reset to the old value and the custom paratemer vanishes.

The snippet below shows the two payload details. Do I need to set-in any additional parameter to ensure persistence of the custom vaule xml=2122311111 ?

[Payload=C:\test\tmpInbound\Bitmap Image1.bmp][Headers={timestamp=1358439618627, xml=2122311111, id=e46e4914-cdbf-4e93-b027-256bd6f6c119}]

[Payload=C:\test\tmpInbound\Bitmap Image1.bmp][Headers={timestamp=1358439618627, id=a45a81b4-a7f6-468c-930e-290e52439ea2}]

Custom value have been set via

Message<Object> message1 = MessageBuilder.withPayload(messageTmp.getPayload()).setHeader("xml", new String("2122311111")).copyHeadersIfAbsent(messageTmp.getHeaders()).build();

[Gary Russell]

Headers cannot be dropped by a channel.

Please show your configuration, and any transformers you are using. Note that if your transformer is a POJO (that just transforms the payload), the framework will take care of propagating headers for you. However, if the signature of your <transformer /> returns a Message object for example...

 public Message<?> transform(Message<?> message)

then you are responsible for propagating headers (see MessageBuilder.copyHeaders()). The assumption is (with a transformer) that if you are dealing with Message objects, you have the responsibility. This is unique to transformers.

We generally recommend using POJOs when possible; users should avoid exposing the messaging infrastructure to their code.

[anirbanzeus]

here are teh configuration details that i am leveraging

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
	   xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
	   xmlns:integration="http://www.springframework.org/schema/integration"
	   xmlns:file="http://www.springframework.org/schema/integration/file"
	xsi:schemaLocation="http://www.springframework.org/schema/beans
						http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
						http://www.springframework.org/schema/integration/file
						http://www.springframework.org/schema/integration/file/spring-integration-file-2.2.xsd 
						http://www.springframework.org/schema/integration
						http://www.springframework.org/schema/integration/spring-integration-2.2.xsd">
	
	
	<file:inbound-channel-adapter id="filesIn" directory="file:C:/test/tmpInbound" scanner="myscanner">
		<integration:poller id="poller" fixed-delay="1000" />
	</file:inbound-channel-adapter>	
	
	<integration:header-enricher input-channel="filesIn" output-channel="filesRouter" method="addNewHeader" ref="transform"/>
	
	<integration:publish-subscribe-channel id="filesRouter" />
	
	<file:outbound-channel-adapter id="filesOut" channel="filesRouter" directory="file:C:/test/DR" delete-source-files="false" filename-generator="writer" order="1"/>
	
	<integration:service-activator input-channel="filesRouter" ref="complete" order="2"/>
	<bean class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer" />
	<bean id="myscanner" class="org.springframework.integration.file.RecursiveLeafOnlyDirectoryScanner" />
	<bean id="handler" class="com.watcher.Handler" />
	<bean id="writer" class="com.watcher.Writer" />
	<bean id="complete" class="com.watcher.Complete" />
	<bean id="transform" class="com.watcher.TestTransformer" />
</beans>

Within the transformer its a mere addition of the custom attribute with following method signature

public Message<?> addNewHeader(Message<?> messageTmp) {}

[Gary Russell]

FYI, if all you are doing in addNewHeader is adding a header, you don't need any code at all, just add a <header/> element to your <header-enricher/>.

Please provide a debug log snippet showing the message flow, including the dropped header.