File adapter in APPEND mode doesn't work.

[maxvor]

Hi,

I'm using newest Spring Integration 2.2.0.RELEASE and trying to write to file in APPEND mode. I generate message as following:

	private void generateMessagesFromFolder(File folder, String channelToSend) {
		MessageChannel channel = (MessageChannel) context.getBean(channelToSend);
		
		for (File file : folder.listFiles()) {
			Message<File> msg = MessageBuilder.withPayload(file)
					.build();
			
			channel.send(msg);
		}
	}

File adapter is configures as following:

 <int-file:outbound-channel-adapter id="eventsExport" 
 	channel="EventsExportChannel"
 	directory="file:C:/Test/Export"
 	mode="APPEND" />

Files are XML, therefore, before sending to the adapter, I do some pretty simple XSLT transformation.

 <int:chain input-channel="PreExportChannel" output-channel="EventsExportChannel">
   <int-file:file-to-string-transformer delete-files="true" />
   <int-xml:xslt-transformer xsl-resource="EventToCSV.xsl" />
 </int:chain>

When I do the same without APPEND mode, everything works fine, except that the result file contains only the info from the last message. I checked with debugger and the error seems to be raised by the attempt to get file lock.
What am I doing wrong?

As a side question: I also tried to workaround the problem with Aggregator (which I put as the last step in the chaing), therefore procuding the message with mannually set correlationId, sequence size and number, but didn't succeed, since my aggregate method never got called, resulting in TargetInvocationException.

Any help is much appreciated!

[Gary Russell]

the error seems to be raised

What error are you getting?

[maxvor]

What error are you getting?

Hi Gary,

It throws InterruptedException at line
lock.lockInterruptibly();
of WhileLockedProcessor class.

[maxvor]

I performed some additional investigation and found out that this error happens only if method

generateMessagesFromFolder(File folder, String channelToSend)

is called in a service activator as a result of incoming message generated by file adapter. When I do call that method directly from program's main method by generating an incoming message via MessageBuilder, it works as expected.

[Gary Russell]

Can you explain what you are trying to do? It doesn't make sense to me that you trigger this method (which generates a bunch of messages with File payloads) when it receives a message with a File payload.

How does 'folder' get populated?

Windows file locking is a pain; you have to be careful that the same file isn't being used in multiple places, even in the same process.

[maxvor]

Can you explain what you are trying to do? It doesn't make sense to me that you trigger this method (which generates a bunch of messages with File payloads) when it receives a message with a File payload.

How does 'folder' get populated?

Windows file locking is a pain; you have to be careful that the same file isn't being used in multiple places, even in the same process.

I'll try to explain what I'm doing. During the day number of events of different types are coming to the application. Those get routed and transformed according to a type, then saved to a dedicated folder as a separate file. At the end of the day, one End-Of-Day message per event type is sent to the application; as a reaction to this, the app is supposed to pick up all the processed events from corresponding folder and aggregate them into one file, while checking some constraints defined in the End-of-day message. So, the way I'm trying to accomplish this is by Service Activator, which internally invokes generateMessagesFromFolder method. The messages are sent to the channel, which is listened to by Appending file adapter. I also tried the same with Aggregator instead of the file adapter, but both seem to throw some kind of strange exceptions.

[Gary Russell]

You need to make sure that the file you are appending to does not have any other open references.

[maxvor]

Thanks for response, Gary, but the only source for messages in the channel, which the file adapter listens to, is this generateMessagesFromFolder method. It throws exception already on the first message sent to the channel. Also, I tried specifying alternative target file, still the same result. And once again, when I invoke the method from Main, it works as expected.

[maxvor]

Maybe actual configuration will help you to point to my problem.

spring-context-common.xml

<?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:int="http://www.springframework.org/schema/integration"
	xmlns:int-file="http://www.springframework.org/schema/integration/file"
    
...>
 
 <int:poller id="defaultPoller"  
 	max-messages-per-poll="10" 
	default="true" fixed-rate="3000" />
 
<!-- Channel adapters --> 
 
<!-- Chains --> 
 <int:chain input-channel="errorChannel">
  	<int:service-activator
    	ref="ErrorService" 
    	method="processError" />
    <int:object-to-string-transformer />
    <int-file:outbound-channel-adapter
 	  directory="file:C:/BASTest/Exceptions" /> 
 </int:chain>
 
<!-- Spring Beans -->  	
 <bean id="ErrorService"
 	class="org.bas.integration.common.ErrorService" />
 	 	
</beans>

spring-context-businessevents.xml

<?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:int="http://www.springframework.org/schema/integration"
	xmlns:jms="http://www.springframework.org/schema/integration/jms"
	xmlns:int-file="http://www.springframework.org/schema/integration/file"
	xmlns:int-xml="http://www.springframework.org/schema/integration/xml"
    
...>

 <import resource="classpath:spring-context-common.xml" />
 
 
<!-- Channels --> 
 <int:publish-subscribe-channel id="BusinessEventsFilesPubSubChannel" />
 <int:channel id="BusinessEventsInChannel" />
 <int:channel id="BusinessEventsPreExportChannel" />
 <int:channel id="BusinessEventsExportChannel" />
 <int:channel id="FXBusinessEventChannel" />
 <int:channel id="MMBusinessEventChannel" />
 <int:channel id="SAPBusinessEventChannel" />
 <int:publish-subscribe-channel id="ToProcessPubSubChannel" />
 <int:channel id="PostponedChannel" /> 
 <int:channel id="EodInChannel" />
 
<!-- Channel adapters --> 
 <int-file:inbound-channel-adapter id="businessEventsIn" 
 	channel="BusinessEventsFilesPubSubChannel"
 	prevent-duplicates="false" 
 	directory="file:C:/BASTest/BusinessEventsIn" />
 <int-file:outbound-channel-adapter id="businessEventsCopyOriginal"
 	channel="BusinessEventsFilesPubSubChannel" 
 	directory="file:C:/BASTest/BusinessEventsCopy" /> 	
 <int-file:outbound-channel-adapter id="businessEventsMM"
 	channel="MMBusinessEventChannel"
 	directory="file:C:/BASTest/BusinessEventsOut/Processed/MM" />
 <int-file:outbound-channel-adapter id="businessEventsFX"
 	channel="FXBusinessEventChannel" 
 	directory="file:C:/BASTest/BusinessEventsOut/Processed/FX" />
 <int-file:outbound-channel-adapter id="businessEventsSAP" 
 	channel="SAPBusinessEventChannel"
 	directory="file:C:/BASTest/BusinessEventsOut/Processed/SAP" />	
 
<!-- Chains -->
 <int:chain input-channel="BusinessEventsFilesPubSubChannel" output-channel="BusinessEventsInChannel">
   <int-file:file-to-string-transformer delete-files="true" />
   <int:transformer ref="BusinessEventUnmarshaller" />
 </int:chain> 
 
 <int:chain input-channel="ToProcessPubSubChannel">
  <int:transformer ref="BusinessEventMarshaller" />
  <int:header-value-router header-name="BusinessEventClass">
    <int:mapping value="org.bas.entities.businessevents.FXBusinessEvent" channel="FXBusinessEventChannel" />
    <int:mapping value="org.bas.entities.businessevents.MMBusinessEvent" channel="MMBusinessEventChannel" />
    <int:mapping value="org.bas.entities.businessevents.SAPBusinessEvent" channel="SAPBusinessEventChannel" />   
  </int:header-value-router>
 </int:chain>

 <int:chain input-channel="ToProcessPubSubChannel">
   <int:filter ref="SettlementFilter" method="accept" />
   <int:transformer ref="BusinessEventMarshaller" />
   <int-file:outbound-channel-adapter 
     directory="file:C:/BASTest/BusinessEventsOut/ForSettlement" /> 
 </int:chain>
 
 <int:chain input-channel="PostponedChannel">
   <int:transformer ref="BusinessEventMarshaller" />
   <int-file:outbound-channel-adapter
     directory="file:C:/BASTest/BusinessEventsOut/Postponed" />
 </int:chain>
 
 <int:chain input-channel="BusinessEventsPreExportChannel">
   <int-file:file-to-string-transformer delete-files="true" />
   <int:transformer ref="BusinessEventUnmarshaller" />
   <int:transformer ref="BusinessEventMarshaller" />
   <int-xml:xslt-transformer xsl-resource="BusinessEventToFinanceFormat.xsl" />
   <!-- <int:aggregator>
     <bean class="org.bas.integration.businessevents.BusinessEventsAggregator" />
   </int:aggregator>   --> 
   <int-file:outbound-channel-adapter
     directory="file:C:/BASTest/Export" mode="APPEND"
 	 filename-generator="businessEventFileNameGenerator" />	
 </int:chain> 
 
<!-- Routers -->
 <int:router method="route"	input-channel="BusinessEventsInChannel">
   <bean class="org.bas.integration.businessevents.BusinessEventProcessDateRouter" />
 </int:router>
 
<!-- Services -->
 <int:service-activator input-channel="EodInChannel"
    ref="BusinessEventsEodService" 
    method="export" />

<!-- Spring Beans --> 
 <bean id="BusinessEventsEodService"
 	 class="org.bas.integration.businessevents.BusinessEventsEodService">
   <constructor-arg value="C:/BASTest/BusinessEventsOut/Processed" />
   <constructor-arg value="C:/BASTest/BusinessEventsOut/Postponed" />
 </bean>
 
 <bean id="businessEventFileNameGenerator"
 	class="org.bas.integration.businessevents.FileNameGenerator" />
 	
 <bean id="HashFileCreator"
 	class="org.bas.integration.businessevents.HashFileCreator" />
 
 <bean id="BusinessEventMarshaller"
 	class="org.bas.integration.businessevents.BusinessEventMarshaller" />  
 
 <bean id="BusinessEventUnmarshaller"
 	class="org.bas.integration.businessevents.BusinessEventUnmarshaller" /> 	
 
 <bean id="SettlementFilter"
 	class="org.bas.integration.businessevents.SettlementFilter" /> 	 
 	
</beans>

Main.java

package org.bas.integration;

...

public class Main {
	
	@SuppressWarnings("resource")
	public static void main(final String... args) throws Exception {      
        final AbstractApplicationContext context = new ClassPathXmlApplicationContext(
                "classpath:spring-context-businessevents.xml");
        context.registerShutdownHook();
        
        /* This works as expected
        BusinessEventsEodService svc = (BusinessEventsEodService) 
        		context.getBean("BusinessEventsEodService");
		Map<String, String> headers = new HashMap<String, String>();
		headers.put("ProductFamily", "MM");
		headers.put("ProcessDate", "20130122");
		EodMessage msg = new EodMessage();
		msg.setProcessDate("20130122");
		msg.setDailyMessageCount(2);
		msg.setProductFamily("MM");
		svc.export(MessageBuilder.withPayload(msg).build());
		*/
	}
}

BusinessEventsEodService.java

package org.bas.integration.businessevents;
 ....
public class BusinessEventsEodService implements ApplicationContextAware {
	private ApplicationContext context;
	private String processedFolder;
	private String postponedFolder;
	
	public BusinessEventsEodService(String processedFolder, String postponedFolder) {
		this.processedFolder = processedFolder;
		this.postponedFolder = postponedFolder;
	}
	
	public void export(Message<EodMessage> message) throws Exception {
		SourcePollingChannelAdapter inAdapter = (SourcePollingChannelAdapter) 
				context.getBean("businessEventsIn");
		EodMessage eodMsg = message.getPayload();
		
		inAdapter.stop();
		
		File inFolder = new File(String.format("%s/%s", processedFolder, 
				eodMsg.getProductFamily()));
		
		validateEodMessage(eodMsg);
		
		if (inFolder.listFiles().length != eodMsg.getDailyMessageCount()) {
			throw new BasException(
					"Number of business events received is not the same as in EOD file", 
					DateTimeUtils.getCurrentBusinessDate(eodMsg.getProductFamily() + "BusinessEvent"));
		}

		Map<String, String> headers = new HashMap<String, String>();
		headers.put("ProductFamily", eodMsg.getProductFamily());
		headers.put("ProcessDate", eodMsg.getProcessDate());
		generateMessagesFromFolder(inFolder, "BusinessEventsPreExportChannel", headers);
		
		DateTimeUtils.closeBusinessDate(
				eodMsg.getProductFamily() + "BusinessEvent",
				DateTimeUtils.toDate(eodMsg.getProcessDate(), "yyyyMMdd"));
		
		inAdapter.start();
		generateMessagesFromFolder(new File(postponedFolder), "BusinessEventsFilesPubSubChannel");
	}

	public void generateMessagesFromFolder(File folder, String channelToSend, Map<String, String> headers) {
		MessageChannel channel = (MessageChannel) context.getBean(channelToSend);
		
		for (File file : folder.listFiles()) {
			Message<File> msg = MessageBuilder.withPayload(file)
					.copyHeaders(headers)
					.build();
			
			try {
				channel.send(msg);
			} catch (Exception e) {
				Logger.getRootLogger().error(e);
			}
		}
	}
	
	public void generateMessagesFromFolder(File folder, String channelToSend) {
		MessageChannel channel = (MessageChannel) context.getBean(channelToSend);
		
		for (File file : folder.listFiles()) {
			Message<File> msg = MessageBuilder.withPayload(file)
					.build();
			
			channel.send(msg);
		}
	}

....
	
	@Override
	public void setApplicationContext(ApplicationContext applicationContext)
			throws BeansException {
		context = applicationContext;
	}
}

[Gary Russell]

It's impossible for me to debug your application by looking at text only.

Running the same test from a main() is a completely different environment. The locking problem probably means that whatever created the file in the processed folder still holds a reference to it.