Added async item processor (could be moved into infrastructure at some point).

spring-batch-integration/.springBeans

@@ -1,7 +1,7 @@
 <?xml version="1.0" encoding="UTF-8"?>
 <beansProjectDescription>
 	<version>1</version>
-	<pluginVersion><![CDATA[2.2.3.RELEASE]]></pluginVersion>
+	<pluginVersion><![CDATA[2.2.4.RELEASE]]></pluginVersion>
 	<configSuffixes>
 		<configSuffix><![CDATA[xml]]></configSuffix>
 	</configSuffixes>
@@ -20,6 +20,7 @@
 		<config>src/test/resources/org/springframework/batch/integration/retry/RetryTransactionalPollingIntegrationTests-context.xml</config>
 		<config>src/test/resources/org/springframework/batch/integration/item/MessagingGatewayIntegrationTests-context.xml</config>
 		<config>src/test/resources/org/springframework/batch/integration/partition/VanillaIntegrationTests-context.xml</config>
+		<config>src/test/resources/org/springframework/batch/integration/async/AsyncItemProcessorMessagingGatewayTests-context.xml</config>
 	</configs>
 	<configSets>
 	</configSets>

spring-batch-integration/src/main/java/org/springframework/batch/integration/async/AsyncItemProcessor.java

@@ -0,0 +1,77 @@
+package org.springframework.batch.integration.async;
+
+import java.util.concurrent.Callable;
+import java.util.concurrent.Future;
+import java.util.concurrent.FutureTask;
+
+import org.springframework.batch.item.ItemProcessor;
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.core.task.SyncTaskExecutor;
+import org.springframework.core.task.TaskExecutor;
+import org.springframework.util.Assert;
+
+/**
+ * An {@link ItemProcessor} that delegates to a nested processor and in the
+ * background. To allow for background processing the return value from the
+ * processor is a {@link Future} which needs to be unpacked before the item can
+ * be used by a client.
+ * 
+ * @author Dave Syer
+ * 
+ * @param <I> the input object type
+ * @param <O> the output object type (will be wrapped in a Future)
+ */
+public class AsyncItemProcessor<I, O> implements ItemProcessor<I, Future<O>>, InitializingBean {
+
+	private ItemProcessor<I, O> delegate;
+
+	private TaskExecutor taskExecutor = new SyncTaskExecutor();
+
+	/**
+	 * Check mandatory properties (the {@link #setDelegate(ItemProcessor)}).
+	 * 
+	 * @see InitializingBean#afterPropertiesSet()
+	 */
+	public void afterPropertiesSet() throws Exception {
+		Assert.notNull(delegate, "The delegate must be set.");
+	}
+
+	/**
+	 * The {@link ItemProcessor} to use to delegate processing to in a
+	 * background thread.
+	 * 
+	 * @param delegate the {@link ItemProcessor} to use as a delegate
+	 */
+	public void setDelegate(ItemProcessor<I, O> delegate) {
+		this.delegate = delegate;
+	}
+
+	/**
+	 * The {@link TaskExecutor} to use to allow the item processing to proceed
+	 * in the background. Defaults to a {@link SyncTaskExecutor} so no threads
+	 * are created unless this is overridden.
+	 * 
+	 * @param taskExecutor a {@link TaskExecutor}
+	 */
+	public void setTaskExecutor(TaskExecutor taskExecutor) {
+		this.taskExecutor = taskExecutor;
+	}
+
+	/**
+	 * Transform the input by delegating to the provided item processor. The
+	 * return value is wrapped in a {@link Future} so that clients can unpack it
+	 * later.
+	 * 
+	 * @see ItemProcessor#process(Object)
+	 */
+	public Future<O> process(final I item) throws Exception {
+		FutureTask<O> task = new FutureTask<O>(new Callable<O>() {
+			public O call() throws Exception {
+				return delegate.process(item);
+			}
+		});
+		taskExecutor.execute(task);
+		return task;
+	}
+
+}

spring-batch-integration/src/main/java/org/springframework/batch/integration/async/AsyncItemWriter.java

@@ -0,0 +1,34 @@
+package org.springframework.batch.integration.async;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.Future;
+
+import org.springframework.batch.item.ItemWriter;
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.util.Assert;
+
+public class AsyncItemWriter<T> implements ItemWriter<Future<T>>, InitializingBean {
+	
+	private ItemWriter<T> delegate;
+	
+	public void afterPropertiesSet() throws Exception {
+		Assert.notNull(delegate, "A delegate ItemWriter must be provided.");
+	}
+	
+	/**
+	 * @param delegate
+	 */
+	public void setDelegate(ItemWriter<T> delegate) {
+		this.delegate = delegate;
+	}
+
+	public void write(List<? extends Future<T>> items) throws Exception {
+		List<T> list = new ArrayList<T>();
+		for (Future<T> future : items) {
+			list.add(future.get());
+		}
+		delegate.write(list);
+	}
+
+}

spring-batch-integration/src/site/apt/index.apt

@@ -7,29 +7,31 @@
 
 Overview of the Spring Integration Batch Module
 
-  Many of the 2.0 Features we identified for Spring Batch look like they might be efficiently and concisely implemented in Spring Integration.  Here is a list of use cases.  These are features that can extend Spring Batch, or use Spring batch features in the context of Spring Integration.  Work in progress by Dave Syer and Jonas Partner.  Many issues to do with transactionality and synchronous execution have been raised and fixed in Spring Integration as a result of these use cases being prototyped.
+  Many use cases in Spring Batch look like they might be efficiently and concisely implemented in Spring Integration.  Here is a list.  These are features that can extend Spring Batch, or use Spring batch features in the context of Spring Integration.  Work in progress waiting for community feedback.  Many issues to do with transactionality and synchronous execution have been raised and fixed in Spring Integration as a result of these use cases being prototyped.
 
 *---+---+---+---+---+
 |<<ID>>|<<Description>>|<<Status>>|<<Sub-package>>|<<Comments>>|
 *----
-|1|{{{Triggers}Message triggers job}}|Prototype|launch|Relatively complete - maybe look at making handlers strongly typed.  Also lots of opportunities with monitoring progress.|
+|1|{{{Triggers}Message triggers job}}|Complete|launch|Complete.  Also lots of opportunities with monitoring progress.|
 *----
-|2|{{{Chunking}Chunking and multi-VM job execution}}|Prototype|chunk|Sunny day case works fine (but not packaged yet as a re-usable handler).  Failures might need some analysis.  Use of statefulStepExecutionListener might be improved on?|
+|2|{{{Chunking}Chunking and multi-VM job execution}}|Complete|chunk|Failures might need some analysis.  Use of stateful StepExecutionListener requires use of step scope.|
 *----
-|3|{{{Aggregator}Asynchronous Aggregator}}|Unstarted| |
+|3|{{{Aggregator}Asynchronous Aggregator}}|Unstarted| | |
 *----
-|4|{{{jobs}Stateful and non-linear jobs}} -> job = flow|Prototype|job|Failure cases need to be analysed - in particular, what happens on restart (after lights out) to messages from the middle of a job.|
+|4|{{{jobs}Stateful and non-linear jobs}} -> job = flow|Complete|job|Simple use cases work well with Spring Batch 2.0 and no Integration features.|
 *----
-|5|{{{Flexible}Flexible item processing model}} (as message flow) -> step = flow|Prototype|item|Complete (v. simple).|
+|5|{{{Flexible}Flexible item processing model}} (as message flow) -> step = flow|Complete|item|Complete (v. simple using MessagingGateway).  Unit tests only.|
 *----
-|6|{{{repeat}Automatic repeat / retry}}|Prototype|retry (unit test)|Works with patched PollingSourceAdapter.|
+|6|{{{repeat}Automatic repeat / retry}}|Complete|retry (unit test)|Unit tests only, since it just uses existing features.|
 *----
-|7|{{{files}Restartable file processing}}|Prototype|file|Seems to hang together.  Not tested thoroughly.|
+|7|{{{files}Restartable file processing}}|Complete|file|Seems to hang together.  Not tested thoroughly, but apparently someone is using it.|
+*----
+|8|{{{async}Asynchronous item processing}}|Complete|async|A general purpose ItemProcesor that returns a Future.|
 *----
 
   Numbers 2, 4, 5 have also been identified as high level Spring Batch 2.0 Features or themes.  If we implement 1, then we also don't need to do any more scheduling and triggering in Spring Batch.
 
-  Number 6 from the list (repeat/retry) is more of a Spring Integration pattern than a Spring Batch one.  We will try and implement it in Spring Batch first, and see about pushing it out into Spring Integration later (with probably a split of repeat/retry out of Batch at that time).
+  Number 6 from the list (repeat/retry) is more of a Spring Integration pattern than a Spring Batch one.  We implemented it in Spring Batch first, with an eye to seeing about pushing it out into Spring Integration later (with probably a split of repeat/retry out of Batch at that time).
 
 * Message {Triggers} Job
 
@@ -172,3 +174,14 @@ Overview of the Spring Integration Batch Module
     [[1]] System sends sucess message to reply channel
 
   Variation: send to asynchronous flow.  Same as main use case but item message is sent to asynchronous flow.  Not as robust because if the lights go out then meesages will be lost, but at least a large file can be split into smaller chunks.
+
+* Asynchronous item processing
+
+  This is actually a variation on {{{Flexible}flexible item processing model}}.
+
+  Description ({async}):
+
+    [[1]] ItemProcessor executes in background (non-transactionally)
+
+    [[1]] ItemWriter collects outputs from futures before phyically writing data
+

spring-batch-integration/src/test/java/org/springframework/batch/integration/async/AsyncItemProcessorMessagingGatewayTests.java

@@ -0,0 +1,59 @@
+package org.springframework.batch.integration.async;
+
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.Future;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.springframework.batch.item.ItemProcessor;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.core.task.SimpleAsyncTaskExecutor;
+import org.springframework.integration.annotation.MessageEndpoint;
+import org.springframework.integration.annotation.ServiceActivator;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+
+@RunWith(SpringJUnit4ClassRunner.class)
+@ContextConfiguration
+public class AsyncItemProcessorMessagingGatewayTests {
+
+	private AsyncItemProcessor<String, String> processor = new AsyncItemProcessor<String, String>();
+
+	@Autowired
+	private ItemProcessor<String, String> delegate;
+
+	@Test
+	public void testMultiExecution() throws Exception {
+		processor.setDelegate(delegate);
+		processor.setTaskExecutor(new SimpleAsyncTaskExecutor());
+		List<Future<String>> list = new ArrayList<Future<String>>();
+		for (int count = 0; count < 10; count++) {
+			list.add(processor.process("foo" + count));
+		}
+		for (Future<String> future : list) {
+			String value = future.get();
+			/**
+			 * TODO: this delegate is a Spring Integration MessagingGateway. It
+			 * can easily return null because of a timeout, but that will be
+			 * treated by Batch as a filtered item, whereas it is really more
+			 * like a skip. Maybe we should have an option to throw an exception
+			 * in the processor if an unexpected null value comes back?
+			 */
+			assertNotNull(value);
+			assertTrue(value.matches("foo.*foo.*"));
+		}
+	}
+
+	@MessageEndpoint
+	public static class Doubler {
+		@ServiceActivator
+		public String cat(String value) {
+			return value + value;
+		}
+	}
+
+}

spring-batch-integration/src/test/java/org/springframework/batch/integration/async/AsyncItemProcessorTests.java

@@ -0,0 +1,49 @@
+package org.springframework.batch.integration.async;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.Future;
+
+import org.junit.Test;
+import org.springframework.batch.item.ItemProcessor;
+import org.springframework.core.task.SimpleAsyncTaskExecutor;
+
+public class AsyncItemProcessorTests {
+
+	private AsyncItemProcessor<String, String> processor = new AsyncItemProcessor<String, String>();
+
+	private ItemProcessor<String, String> delegate = new ItemProcessor<String, String>() {
+		public String process(String item) throws Exception {
+			return item + item;
+		};
+	};
+
+	@Test(expected = IllegalArgumentException.class)
+	public void testNoDelegate() throws Exception {
+		processor.afterPropertiesSet();
+	}
+
+	@Test
+	public void testExecution() throws Exception {
+		processor.setDelegate(delegate);
+		Future<String> result = processor.process("foo");
+		assertEquals("foofoo", result.get());
+	}
+
+	@Test
+	public void testMultiExecution() throws Exception {
+		processor.setDelegate(delegate);
+		processor.setTaskExecutor(new SimpleAsyncTaskExecutor());
+		List<Future<String>> list = new ArrayList<Future<String>>();
+		for (int count = 0; count < 10; count++) {
+			list.add(processor.process("foo" + count));
+		}
+		for (Future<String> future : list) {
+			assertTrue(future.get().matches("foo.*foo.*"));
+		}
+	}
+
+}

spring-batch-integration/src/test/resources/org/springframework/batch/integration/async/AsyncItemProcessorMessagingGatewayTests-context.xml

@@ -0,0 +1,25 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans:beans xmlns="http://www.springframework.org/schema/integration"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:beans="http://www.springframework.org/schema/beans"
+	xmlns:context="http://www.springframework.org/schema/context" xmlns:tx="http://www.springframework.org/schema/tx"
+	xsi:schemaLocation="http://www.springframework.org/schema/beans
+                                 http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
+                                 http://www.springframework.org/schema/integration
+                                 http://www.springframework.org/schema/integration/spring-integration-1.0.xsd
+                                 http://www.springframework.org/schema/context
+                                 http://www.springframework.org/schema/context/spring-context-2.5.xsd
+                                 http://www.springframework.org/schema/tx
+                                 http://www.springframework.org/schema/tx/spring-tx-2.5.xsd">
+	<annotation-config />
+	<channel id="requests" />
+	<channel id="replies">
+		<queue />
+	</channel>
+	<gateway service-interface="org.springframework.batch.item.ItemProcessor"
+		id="processor" default-reply-timeout="1000" default-request-channel="requests"
+		default-reply-channel="replies" />
+	<service-activator input-channel="requests"
+		output-channel="replies" ref="doubler" />
+	<beans:bean id="doubler"
+		class="org.springframework.batch.integration.async.AsyncItemProcessorMessagingGatewayTests$Doubler" />
+</beans:beans>

spring-batch-integration/src/test/resources/org/springframework/batch/integration/partition/VanillaIntegrationTests-context.xml

@@ -23,13 +23,13 @@
 	<integration:service-activator ref="stepExecutionRequestHandler"
 		input-channel="requests" output-channel="staging">
 		<integration:poller>
-			<integration:interval-trigger interval="10"/>
+			<integration:interval-trigger interval="10" />
 		</integration:poller>
 	</integration:service-activator>
 	<integration:aggregator ref="partitionHandler"
 		timeout="10000" input-channel="staging" output-channel="replies">
 		<integration:poller>
-			<integration:interval-trigger interval="10"/>
+			<integration:interval-trigger interval="10" />
 		</integration:poller>
 	</integration:aggregator>