diff --git a/spring-batch-integration/.springBeans b/spring-batch-integration/.springBeans index 8acabca68..3fd911e10 100644 --- a/spring-batch-integration/.springBeans +++ b/spring-batch-integration/.springBeans @@ -1,7 +1,7 @@ 1 - + @@ -20,6 +20,7 @@ src/test/resources/org/springframework/batch/integration/retry/RetryTransactionalPollingIntegrationTests-context.xml src/test/resources/org/springframework/batch/integration/item/MessagingGatewayIntegrationTests-context.xml src/test/resources/org/springframework/batch/integration/partition/VanillaIntegrationTests-context.xml + src/test/resources/org/springframework/batch/integration/async/AsyncItemProcessorMessagingGatewayTests-context.xml diff --git a/spring-batch-integration/src/main/java/org/springframework/batch/integration/async/AsyncItemProcessor.java b/spring-batch-integration/src/main/java/org/springframework/batch/integration/async/AsyncItemProcessor.java new file mode 100644 index 000000000..87a5ee7cf --- /dev/null +++ b/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 the input object type + * @param the output object type (will be wrapped in a Future) + */ +public class AsyncItemProcessor implements ItemProcessor>, InitializingBean { + + private ItemProcessor 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 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 process(final I item) throws Exception { + FutureTask task = new FutureTask(new Callable() { + public O call() throws Exception { + return delegate.process(item); + } + }); + taskExecutor.execute(task); + return task; + } + +} diff --git a/spring-batch-integration/src/main/java/org/springframework/batch/integration/async/AsyncItemWriter.java b/spring-batch-integration/src/main/java/org/springframework/batch/integration/async/AsyncItemWriter.java new file mode 100644 index 000000000..af5ea416b --- /dev/null +++ b/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 implements ItemWriter>, InitializingBean { + + private ItemWriter delegate; + + public void afterPropertiesSet() throws Exception { + Assert.notNull(delegate, "A delegate ItemWriter must be provided."); + } + + /** + * @param delegate + */ + public void setDelegate(ItemWriter delegate) { + this.delegate = delegate; + } + + public void write(List> items) throws Exception { + List list = new ArrayList(); + for (Future future : items) { + list.add(future.get()); + } + delegate.write(list); + } + +} diff --git a/spring-batch-integration/src/site/apt/index.apt b/spring-batch-integration/src/site/apt/index.apt index c5fa252f5..5b6a23462 100644 --- a/spring-batch-integration/src/site/apt/index.apt +++ b/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. *---+---+---+---+---+ |<>|<>|<>|<>|<>| *---- -|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 + diff --git a/spring-batch-integration/src/test/java/org/springframework/batch/integration/async/AsyncItemProcessorMessagingGatewayTests.java b/spring-batch-integration/src/test/java/org/springframework/batch/integration/async/AsyncItemProcessorMessagingGatewayTests.java new file mode 100644 index 000000000..207223fd0 --- /dev/null +++ b/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 processor = new AsyncItemProcessor(); + + @Autowired + private ItemProcessor delegate; + + @Test + public void testMultiExecution() throws Exception { + processor.setDelegate(delegate); + processor.setTaskExecutor(new SimpleAsyncTaskExecutor()); + List> list = new ArrayList>(); + for (int count = 0; count < 10; count++) { + list.add(processor.process("foo" + count)); + } + for (Future 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; + } + } + +} diff --git a/spring-batch-integration/src/test/java/org/springframework/batch/integration/async/AsyncItemProcessorTests.java b/spring-batch-integration/src/test/java/org/springframework/batch/integration/async/AsyncItemProcessorTests.java new file mode 100644 index 000000000..c6093d310 --- /dev/null +++ b/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 processor = new AsyncItemProcessor(); + + private ItemProcessor delegate = new ItemProcessor() { + 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 result = processor.process("foo"); + assertEquals("foofoo", result.get()); + } + + @Test + public void testMultiExecution() throws Exception { + processor.setDelegate(delegate); + processor.setTaskExecutor(new SimpleAsyncTaskExecutor()); + List> list = new ArrayList>(); + for (int count = 0; count < 10; count++) { + list.add(processor.process("foo" + count)); + } + for (Future future : list) { + assertTrue(future.get().matches("foo.*foo.*")); + } + } + +} diff --git a/spring-batch-integration/src/test/resources/org/springframework/batch/integration/async/AsyncItemProcessorMessagingGatewayTests-context.xml b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/async/AsyncItemProcessorMessagingGatewayTests-context.xml new file mode 100644 index 000000000..9139dcf0e --- /dev/null +++ b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/async/AsyncItemProcessorMessagingGatewayTests-context.xml @@ -0,0 +1,25 @@ + + + + + + + + + + + \ No newline at end of file diff --git a/spring-batch-integration/src/test/resources/org/springframework/batch/integration/partition/VanillaIntegrationTests-context.xml b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/partition/VanillaIntegrationTests-context.xml index 2f6d0b4fb..8af3d7dca 100644 --- a/spring-batch-integration/src/test/resources/org/springframework/batch/integration/partition/VanillaIntegrationTests-context.xml +++ b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/partition/VanillaIntegrationTests-context.xml @@ -23,13 +23,13 @@ - + - +