file:inbound-channel-adapter with nio-locker and file-to-string-transformer

[pwanner]

Hi everybody,

Using inbound-channel-adapter, locking the file with nio-locker cause file-to-string-transformer not able to access the file anymore.

I thought that the file locked would be still accessible by the thread that has taken the lock to transform it and then delete it.

I am wrong?

Here is my config:

<int-file:inbound-channel-adapter 
  channel="fileToStringChannel" 
  directory="file:c:/temp/scm" 
  prevent-duplicates="true">
  <int:poller id="poller" fixed-delay="1000">
    <int:transactional transaction-manager="txManager"/>
  </int:poller>
  <int-file:nio-locker/>
</int-file:inbound-channel-adapter>

<int-file:file-to-string-transformer
  input-channel="fileToStringChannel" 
  output-channel="stringToEnricherChannel"
  charset="UTF-8"
  delete-files="false"/>

<int:header-enricher 
  input-channel="stringToEnricherChannel" 
  output-channel="dispatchChannel">
  <int:header name="correlationId" ref="messageCoreKeyExtractor" method="extractKey"/>
  <int:header name="JMSXGroupID" ref="messageCoreKeyExtractor" method="extractKey"/>
</int:header-enricher>

<int-jms:outbound-channel-adapter order="1"
  channel="dispatchChannel"
  connection-factory="cachedConnectionFactory" 
  destination-name="outQueue" 
  header-mapper="defaultJmsHeaderMapper"/>

<!-- If JMS Send was successful, remove the file (within the transaction)
     http://forum.springsource.org/showthread.php?102004-File-Channel-Adapter-issues-->
<int:service-activator order="2"
  input-channel="dispatchChannel" 
  output-channel="nullChannel"
  expression="headers.file_originalFile.delete()"/>
  
<!-- Channels -->
<int:channel id="fileToStringChannel"/>
<int:channel id="stringToEnricherChannel"/>
<int:publish-subscribe-channel id="dispatchChannel"/>

<!-- Beans-->
<bean id="messageCoreKeyExtractor" class="com.xxx.integration.MessageCoreKeyExtractor"/>
<bean id="defaultJmsHeaderMapper" class="org.springframework.integration.jms.DefaultJmsHeaderMapper"/>

<!-- JMS connection factory and tx manager-->
<bean id="cachedConnectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory" 
  p:targetConnectionFactory-ref="connectionFactory"
  p:sessionCacheSize="5"
  p:cacheProducers="false"/>

<bean id="connectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory" 
  p:brokerURL="failover://(${broker.url})?jms.prefetchPolicy.all=1" />
	
<bean id="txManager" class="org.springframework.jms.connection.JmsTransactionManager" 
  p:connectionFactory-ref="connectionFactory"/>

[Mark Fisher]

Can you please show the config excerpt for 'dispatchChannel'?

[pwanner]

I have edited the above config.

[oleg.zhurakousky]

Can you show us your config so we can see the channel configuration?

[pwanner]

I have updated my first post so that the config is now complete and reflects the whole flow.

[oleg.zhurakousky]

Sorry, i see it now. I was afraid that one of those channels is async channel but they all seem to be sync channels so let's try something else. Put a logging channel adapter after int-file:inbound-channel-adapter
For example:

<int-file:inbound-channel-adapter 
  channel="fileToStringChannel" 
  directory="file:c:/temp/scm" 
  prevent-duplicates="true">
  <int:poller id="poller" fixed-delay="1000">
    <int:transactional transaction-manager="txManager"/>
  </int:poller>
  <int-file:nio-locker/>
</int-file:inbound-channel-adapter>
<int:logging-channel-adapter channel="fileToStringChannel" level="WARN"/>

And see if the message comes out of inbound-channel-adapter. If so than put logging channel adapter after file-to-string-transformer and so on until you see where the problem is.
You can also add channel interceptors to the channels and track at what point the message does not reach its destination.

[pwanner]

This is really weird, when I add the logging-channel-adapter there is no more exception thrown but the message is not consumed and not delivered.

Below the log with and without the logging-channel-adapter.
The ONLY thing that I have changed in the config

<int:logging-channel-adapter channel="fileToStringChannel" level="DEBUG"/>

Stacktrace without the logging-channel-adapter

INFO : org.springframework.integration.endpoint.SourcePollingChannelAdapter - started org.springframework.integration.config.SourcePollingChannelAdapterFactoryBean#0
INFO : org.apache.activemq.transport.failover.FailoverTransport - Successfully connected to tcp://localhost:61616
DEBUG: org.springframework.integration.file.FileReadingMessageSource - Added to queue: [c:\temp\scm\messageA2.xml]
INFO : org.springframework.integration.file.FileReadingMessageSource - Created message: [[Payload=c:\temp\scm\messageA2.xml][Headers={timestamp=1336397994640, id=1537f333-895c-437e-92a8-c8d20cb5c693}]]
DEBUG: org.springframework.integration.endpoint.SourcePollingChannelAdapter - Poll resulted in Message: [Payload=c:\temp\scm\messageA2.xml][Headers={timestamp=1336397994640, id=1537f333-895c-437e-92a8-c8d20cb5c693}]
DEBUG: org.springframework.integration.channel.DirectChannel - preSend on channel 'fileToStringChannel', message: [Payload=c:\temp\scm\messageA2.xml][Headers={timestamp=1336397994640, id=1537f333-895c-437e-92a8-c8d20cb5c693}]
DEBUG: org.springframework.integration.transformer.MessageTransformingHandler - org.springframework.integration.transformer.MessageTransformingHandler#0 received message: [Payload=c:\temp\scm\messageA2.xml][Headers={timestamp=1336397994640, id=1537f333-895c-437e-92a8-c8d20cb5c693}]
DEBUG: org.springframework.integration.channel.PublishSubscribeChannel - preSend on channel 'errorChannel', message: [Payload=org.springframework.integration.transformer.MessageTransformationException: org.springframework.integration.MessagingException: failed to transform File Message][Headers={timestamp=1336397994835, id=53d30828-4635-4fac-9612-e361801460ae}]
DEBUG: org.springframework.integration.handler.LoggingHandler - (inner bean)#4 received message: [Payload=org.springframework.integration.transformer.MessageTransformationException: org.springframework.integration.MessagingException: failed to transform File Message][Headers={timestamp=1336397994835, id=53d30828-4635-4fac-9612-e361801460ae}]
ERROR: org.springframework.integration.handler.LoggingHandler - org.springframework.integration.transformer.MessageTransformationException: org.springframework.integration.MessagingException: failed to transform File Message
	at org.springframework.integration.transformer.MessageTransformingHandler.handleRequestMessage(MessageTransformingHandler.java:73)
	at org.springframework.integration.handler.AbstractReplyProducingMessageHandler.handleMessageInternal(AbstractReplyProducingMessageHandler.java:97)
	at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:73)
	at org.springframework.integration.dispatcher.UnicastingDispatcher.doDispatch(UnicastingDispatcher.java:115)
	at org.springframework.integration.dispatcher.UnicastingDispatcher.dispatch(UnicastingDispatcher.java:102)
	at org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:77)
	at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:157)
	at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:128)
	at org.springframework.integration.core.MessagingTemplate.doSend(MessagingTemplate.java:288)
	at org.springframework.integration.core.MessagingTemplate.send(MessagingTemplate.java:149)
	at org.springframework.integration.endpoint.SourcePollingChannelAdapter.doPoll(SourcePollingChannelAdapter.java:97)
	at org.springframework.integration.endpoint.AbstractPollingEndpoint$1.call(AbstractPollingEndpoint.java:146)
	at org.springframework.integration.endpoint.AbstractPollingEndpoint$1.call(AbstractPollingEndpoint.java:144)
  ...
	at $Proxy9.call(Unknown Source)
	at org.springframework.integration.endpoint.AbstractPollingEndpoint$Poller$1.run(AbstractPollingEndpoint.java:207)
	at org.springframework.integration.util.ErrorHandlingTaskExecutor$1.run(ErrorHandlingTaskExecutor.java:52)
	at org.springframework.core.task.SyncTaskExecutor.execute(SyncTaskExecutor.java:48)
	at org.springframework.integration.util.ErrorHandlingTaskExecutor.execute(ErrorHandlingTaskExecutor.java:49)
	at org.springframework.integration.endpoint.AbstractPollingEndpoint$Poller.run(AbstractPollingEndpoint.java:202)
	at org.springframework.scheduling.support.DelegatingErrorHandlingRunnable.run(DelegatingErrorHandlingRunnable.java:53)
	at org.springframework.scheduling.concurrent.ReschedulingRunnable.run(ReschedulingRunnable.java:81)
  ...
	at java.lang.Thread.run(Thread.java:662)
Caused by: org.springframework.integration.MessagingException: failed to transform File Message
	at org.springframework.integration.file.transformer.AbstractFilePayloadTransformer.transform(AbstractFilePayloadTransformer.java:71)
	at org.springframework.integration.transformer.MessageTransformingHandler.handleRequestMessage(MessageTransformingHandler.java:67)
	... 38 more
Caused by: java.io.IOException: Le processus ne peut pas accéder au fichier car un autre processus en a verrouillé une partie
	at java.io.FileInputStream.readBytes(Native Method)
	at java.io.FileInputStream.read(FileInputStream.java:220)
	at sun.nio.cs.StreamDecoder.readBytes(StreamDecoder.java:264)
	at sun.nio.cs.StreamDecoder.implRead(StreamDecoder.java:306)
	at sun.nio.cs.StreamDecoder.read(StreamDecoder.java:158)
	at java.io.InputStreamReader.read(InputStreamReader.java:167)
	at java.io.Reader.read(Reader.java:123)
	at org.springframework.util.FileCopyUtils.copy(FileCopyUtils.java:188)
	at org.springframework.util.FileCopyUtils.copyToString(FileCopyUtils.java:240)
	at org.springframework.integration.file.transformer.FileToStringTransformer.transformFile(FileToStringTransformer.java:49)
	at org.springframework.integration.file.transformer.FileToStringTransformer.transformFile(FileToStringTransformer.java:32)
	at org.springframework.integration.file.transformer.AbstractFilePayloadTransformer.transform(AbstractFilePayloadTransformer.java:58)
	... 39 more

DEBUG: org.springframework.integration.channel.PublishSubscribeChannel - postSend (sent=true) on channel 'errorChannel', message: [Payload=org.springframework.integration.transformer.MessageTransformationException: org.springframework.integration.MessagingException: failed to transform File Message][Headers={timestamp=1336397994835, id=53d30828-4635-4fac-9612-e361801460ae}]

Stacktrace with the logging-channel-adapter

INFO : org.springframework.integration.endpoint.SourcePollingChannelAdapter - started org.springframework.integration.config.SourcePollingChannelAdapterFactoryBean#0
INFO : org.apache.activemq.transport.failover.FailoverTransport - Successfully connected to tcp://localhost:61616
DEBUG: org.springframework.integration.file.FileReadingMessageSource - Added to queue: [c:\temp\scm\messageA2.xml]
INFO : org.springframework.integration.file.FileReadingMessageSource - Created message: [[Payload=c:\temp\scm\messageA2.xml][Headers={timestamp=1336397894635, id=4c487c7c-0932-4a03-86c6-02b25403a071}]]
DEBUG: org.springframework.integration.endpoint.SourcePollingChannelAdapter - Poll resulted in Message: [Payload=c:\temp\scm\messageA2.xml][Headers={timestamp=1336397894635, id=4c487c7c-0932-4a03-86c6-02b25403a071}]
DEBUG: org.springframework.integration.channel.DirectChannel - preSend on channel 'fileToStringChannel', message: [Payload=c:\temp\scm\messageA2.xml][Headers={timestamp=1336397894635, id=4c487c7c-0932-4a03-86c6-02b25403a071}]
DEBUG: org.springframework.integration.transformer.MessageTransformingHandler - org.springframework.integration.transformer.MessageTransformingHandler#0 received message: [Payload=c:\temp\scm\messageA2.xml][Headers={timestamp=1336397894635, id=4c487c7c-0932-4a03-86c6-02b25403a071}]
DEBUG: org.springframework.integration.handler.LoggingHandler - org.springframework.integration.handler.LoggingHandler#0 received message: [Payload=c:\temp\scm\messageA2.xml][Headers={timestamp=1336397894635, id=4c487c7c-0932-4a03-86c6-02b25403a071}]
INFO : org.springframework.integration.handler.LoggingHandler - c:\temp\scm\messageA2.xml
DEBUG: org.springframework.integration.channel.DirectChannel - postSend (sent=true) on channel 'fileToStringChannel', message: [Payload=c:\temp\scm\messageA2.xml][Headers={timestamp=1336397894635, id=4c487c7c-0932-4a03-86c6-02b25403a071}]
INFO : org.apache.activemq.transport.failover.FailoverTransport - Successfully connected to tcp://localhost:61616
DEBUG: org.springframework.integration.endpoint.SourcePollingChannelAdapter - Poll resulted in Message: null
DEBUG: org.springframework.integration.endpoint.SourcePollingChannelAdapter - Received no Message during the poll, returning 'false'

[oleg.zhurakousky]

Hmm, strange since I see that you do have exception in 'file-to-string-transformer'
Let's try this. I just tried a subset of your sample and it works fine so try it as well and see what happens. So modify your configuration to look like this (only this)

<int-file:inbound-channel-adapter
		channel="fileToStringChannel" directory="file:c:/temp/scm"
		prevent-duplicates="true">
		<int:poller  fixed-rate="5000" max-messages-per-poll="1">
<!-- 			<int:transactional transaction-manager="txManager" /> -->
		</int:poller>
		<int-file:nio-locker />
	</int-file:inbound-channel-adapter>

	<int-file:file-to-string-transformer
		input-channel="fileToStringChannel" output-channel="loggingChannel"
		charset="UTF-8" delete-files="false" />
		
    <int:logging-channel-adapter id="loggingChannel" level="WARN"/>

[pwanner]

There is no difference as the error occurs in the file-to-string-transformer that cannot open the file locked by the nio-locker.

INFO : org.springframework.integration.endpoint.SourcePollingChannelAdapter - started org.springframework.integration.config.SourcePollingChannelAdapterFactoryBean#0
DEBUG: org.springframework.integration.file.FileReadingMessageSource - Added to queue: [c:\temp\scm\messageA2.xml]
INFO : org.springframework.integration.file.FileReadingMessageSource - Created message: [[Payload=c:\temp\scm\messageA2.xml][Headers={timestamp=1336401294594, id=5aedf9e8-c2eb-48e7-b425-5973d65d7edd}]]
DEBUG: org.springframework.integration.endpoint.SourcePollingChannelAdapter - Poll resulted in Message: [Payload=c:\temp\scm\messageA2.xml][Headers={timestamp=1336401294594, id=5aedf9e8-c2eb-48e7-b425-5973d65d7edd}]
DEBUG: org.springframework.integration.channel.DirectChannel - preSend on channel 'fileToStringChannel', message: [Payload=c:\temp\scm\messageA2.xml][Headers={timestamp=1336401294594, id=5aedf9e8-c2eb-48e7-b425-5973d65d7edd}]
DEBUG: org.springframework.integration.transformer.MessageTransformingHandler - org.springframework.integration.transformer.MessageTransformingHandler#0 received message: [Payload=c:\temp\scm\messageA2.xml][Headers={timestamp=1336401294594, id=5aedf9e8-c2eb-48e7-b425-5973d65d7edd}]
DEBUG: org.springframework.integration.channel.PublishSubscribeChannel - preSend on channel 'errorChannel', message: [Payload=org.springframework.integration.transformer.MessageTransformationException: org.springframework.integration.MessagingException: failed to transform File Message][Headers={timestamp=1336401294688, id=b44ecb56-2e1d-4a5f-a3b9-7eece16342dc}]
DEBUG: org.springframework.integration.handler.LoggingHandler - (inner bean)#4 received message: [Payload=org.springframework.integration.transformer.MessageTransformationException: org.springframework.integration.MessagingException: failed to transform File Message][Headers={timestamp=1336401294688, id=b44ecb56-2e1d-4a5f-a3b9-7eece16342dc}]
ERROR: org.springframework.integration.handler.LoggingHandler - org.springframework.integration.transformer.MessageTransformationException: org.springframework.integration.MessagingException: failed to transform File Message

[oleg.zhurakousky]

But that is what I am asking since the configuration i posted works for me here and that's why I wanted you to try that exact configuration.
So, are you saying you did try the configuration I attached and it fails in the transformer?
If so, can you also post your SI version, Spring, Java etc.