From 2abeefba7135e6ed3ece6eb36bd616f8fe1c97b5 Mon Sep 17 00:00:00 2001 From: dsyer Date: Tue, 14 Oct 2008 13:54:37 +0000 Subject: [PATCH] Refactor retry tests in integration to work with new adviceChain param in @POller --- .../RetryOperationsInterceptor.java | 2 +- .../StatefulRetryOperationsInterceptor.java | 24 +- .../RetryOperationsInterceptorTests.java | 4 +- ...atefulRetryOperationsInterceptorTests.java | 29 +- spring-batch-integration/.springBeans | 4 + spring-batch-integration/pom.xml | 11 +- ...sageChannelItemWriterIntegrationTests.java | 2 + .../retry/PollableSourceRetryTests.java | 353 ------------------ ...tTransactionalPollingIntegrationTests.java | 98 +++++ ...tTransactionalPollingIntegrationTests.java | 93 +++++ ...yTransactionalPollingIntegrationTests.java | 93 +++++ .../batch/integration/retry/Service.java | 26 ++ .../integration/retry/SimpleRecoverer.java | 37 ++ .../integration/retry/SimpleService.java | 39 ++ ...ctionalPollingIntegrationTests-context.xml | 27 ++ ...ctionalPollingIntegrationTests-context.xml | 47 +++ ...ctionalPollingIntegrationTests-context.xml | 31 ++ ...ctionalPollingIntegrationTests-context.xml | 2 +- 18 files changed, 541 insertions(+), 381 deletions(-) delete mode 100644 spring-batch-integration/src/test/java/org/springframework/batch/integration/retry/PollableSourceRetryTests.java create mode 100644 spring-batch-integration/src/test/java/org/springframework/batch/integration/retry/RepeatTransactionalPollingIntegrationTests.java create mode 100644 spring-batch-integration/src/test/java/org/springframework/batch/integration/retry/RetryRepeatTransactionalPollingIntegrationTests.java create mode 100644 spring-batch-integration/src/test/java/org/springframework/batch/integration/retry/RetryTransactionalPollingIntegrationTests.java create mode 100644 spring-batch-integration/src/test/java/org/springframework/batch/integration/retry/Service.java create mode 100644 spring-batch-integration/src/test/java/org/springframework/batch/integration/retry/SimpleRecoverer.java create mode 100644 spring-batch-integration/src/test/java/org/springframework/batch/integration/retry/SimpleService.java create mode 100644 spring-batch-integration/src/test/resources/org/springframework/batch/integration/retry/RepeatTransactionalPollingIntegrationTests-context.xml create mode 100644 spring-batch-integration/src/test/resources/org/springframework/batch/integration/retry/RetryRepeatTransactionalPollingIntegrationTests-context.xml create mode 100644 spring-batch-integration/src/test/resources/org/springframework/batch/integration/retry/RetryTransactionalPollingIntegrationTests-context.xml diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/interceptor/RetryOperationsInterceptor.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/interceptor/RetryOperationsInterceptor.java index 5cdbd0fe7..b09caf6be 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/interceptor/RetryOperationsInterceptor.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/interceptor/RetryOperationsInterceptor.java @@ -45,7 +45,7 @@ public class RetryOperationsInterceptor implements MethodInterceptor { private RetryOperations retryOperations = new RetryTemplate(); - public void setRetryTemplate(RetryOperations retryTemplate) { + public void setRetryOperations(RetryOperations retryTemplate) { Assert.notNull(retryTemplate, "'retryOperations' cannot be null."); this.retryOperations = retryTemplate; } diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/interceptor/StatefulRetryOperationsInterceptor.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/interceptor/StatefulRetryOperationsInterceptor.java index d2ac9a9fb..df712ade5 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/interceptor/StatefulRetryOperationsInterceptor.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/interceptor/StatefulRetryOperationsInterceptor.java @@ -21,12 +21,11 @@ import org.aopalliance.intercept.MethodInvocation; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.springframework.batch.retry.ExhaustedRetryException; -import org.springframework.batch.retry.RetryState; import org.springframework.batch.retry.RecoveryCallback; import org.springframework.batch.retry.RetryCallback; import org.springframework.batch.retry.RetryContext; import org.springframework.batch.retry.RetryOperations; -import org.springframework.batch.retry.RetryPolicy; +import org.springframework.batch.retry.RetryState; import org.springframework.batch.retry.policy.NeverRetryPolicy; import org.springframework.batch.retry.support.DefaultRetryState; import org.springframework.batch.retry.support.RetryTemplate; @@ -61,14 +60,21 @@ public class StatefulRetryOperationsInterceptor implements MethodInterceptor { private NewMethodArgumentsIdentifier newMethodArgumentsIdentifier; - private final RetryTemplate retryTemplate = new RetryTemplate(); + private RetryOperations retryOperations; + + public void setRetryOperations(RetryOperations retryTemplate) { + Assert.notNull(retryTemplate, "'retryOperations' cannot be null."); + this.retryOperations = retryTemplate; + } /** * */ public StatefulRetryOperationsInterceptor() { super(); + RetryTemplate retryTemplate = new RetryTemplate(); retryTemplate.setRetryPolicy(new NeverRetryPolicy()); + retryOperations = retryTemplate; } /** @@ -90,16 +96,6 @@ public class StatefulRetryOperationsInterceptor implements MethodInterceptor { this.keyGenerator = keyGenerator; } - /** - * Public setter for the retryPolicy. The value provided should be a normal - * stateless policy, which is wrapped into a stateful policy inside this - * method. - * @param retryPolicy the retryPolicy to set - */ - public void setRetryPolicy(RetryPolicy retryPolicy) { - retryTemplate.setRetryPolicy(retryPolicy); - } - /** * Public setter for the {@link NewMethodArgumentsIdentifier}. Only set this if the * arguments to the intercepted method can be inspected to find out if they @@ -141,7 +137,7 @@ public class StatefulRetryOperationsInterceptor implements MethodInterceptor { RetryState retryState = new DefaultRetryState(keyGenerator != null ? keyGenerator.getKey(args) : item, newMethodArgumentsIdentifier != null ? newMethodArgumentsIdentifier.isNew(args) : false ); - Object result = retryTemplate.execute(new MethodInvocationRetryCallback(invocation), new ItemRecovererCallback(args, recoverer), retryState); + Object result = retryOperations.execute(new MethodInvocationRetryCallback(invocation), new ItemRecovererCallback(args, recoverer), retryState); logger.debug("Exiting proxied method in stateful retry with result: (" + result + ")"); diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/retry/interceptor/RetryOperationsInterceptorTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/retry/interceptor/RetryOperationsInterceptorTests.java index 657e4df1e..737d9418c 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/retry/interceptor/RetryOperationsInterceptorTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/retry/interceptor/RetryOperationsInterceptorTests.java @@ -76,7 +76,7 @@ public class RetryOperationsInterceptorTests extends TestCase { }); RetryTemplate template = new RetryTemplate(); template.setRetryPolicy(new SimpleRetryPolicy(2)); - interceptor.setRetryTemplate(template); + interceptor.setRetryOperations(template); service.service(); assertEquals(2, count); assertEquals(2, list.size()); @@ -86,7 +86,7 @@ public class RetryOperationsInterceptorTests extends TestCase { ((Advised) service).addAdvice(interceptor); RetryTemplate template = new RetryTemplate(); template.setRetryPolicy(new NeverRetryPolicy()); - interceptor.setRetryTemplate(template); + interceptor.setRetryOperations(template); try { service.service(); fail("Expected Exception."); diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/retry/interceptor/StatefulRetryOperationsInterceptorTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/retry/interceptor/StatefulRetryOperationsInterceptorTests.java index a42ba9729..ee4a9dd6f 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/retry/interceptor/StatefulRetryOperationsInterceptorTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/retry/interceptor/StatefulRetryOperationsInterceptorTests.java @@ -31,6 +31,7 @@ import org.springframework.batch.retry.ExhaustedRetryException; import org.springframework.batch.retry.policy.AlwaysRetryPolicy; import org.springframework.batch.retry.policy.NeverRetryPolicy; import org.springframework.batch.retry.policy.SimpleRetryPolicy; +import org.springframework.batch.retry.support.RetryTemplate; /** * @author Dave Syer @@ -40,6 +41,8 @@ public class StatefulRetryOperationsInterceptorTests extends TestCase { private StatefulRetryOperationsInterceptor interceptor; + private RetryTemplate retryTemplate = new RetryTemplate(); + private Service service; private Transformer transformer; @@ -49,7 +52,8 @@ public class StatefulRetryOperationsInterceptorTests extends TestCase { public void setUp() throws Exception { interceptor = new StatefulRetryOperationsInterceptor(); service = (Service) ProxyFactory.getProxy(Service.class, new SingletonTargetSource(new ServiceImpl())); - transformer = (Transformer) ProxyFactory.getProxy(Transformer.class, new SingletonTargetSource(new TransformerImpl())); + transformer = (Transformer) ProxyFactory.getProxy(Transformer.class, new SingletonTargetSource( + new TransformerImpl())); count = 0; } @@ -80,7 +84,8 @@ public class StatefulRetryOperationsInterceptorTests extends TestCase { } public void testDefaultInterceptorAlwaysRetry() throws Exception { - interceptor.setRetryPolicy(new AlwaysRetryPolicy()); + retryTemplate.setRetryPolicy(new AlwaysRetryPolicy()); + interceptor.setRetryOperations(retryTemplate); ((Advised) service).addAdvice(interceptor); try { service.service("foo"); @@ -102,7 +107,8 @@ public class StatefulRetryOperationsInterceptorTests extends TestCase { return invocation.proceed(); } }); - interceptor.setRetryPolicy(new SimpleRetryPolicy(2)); + interceptor.setRetryOperations(retryTemplate); + retryTemplate.setRetryPolicy(new SimpleRetryPolicy(2)); try { service.service("foo"); fail("Expected Exception."); @@ -119,7 +125,8 @@ public class StatefulRetryOperationsInterceptorTests extends TestCase { public void testTransformerWithSuccessfulRetry() throws Exception { ((Advised) transformer).addAdvice(interceptor); - interceptor.setRetryPolicy(new SimpleRetryPolicy(2)); + interceptor.setRetryOperations(retryTemplate); + retryTemplate.setRetryPolicy(new SimpleRetryPolicy(2)); try { transformer.transform("foo"); fail("Expected Exception."); @@ -136,7 +143,8 @@ public class StatefulRetryOperationsInterceptorTests extends TestCase { public void testRetryExceptionAfterTooManyAttemptsWithNoRecovery() throws Exception { ((Advised) service).addAdvice(interceptor); - interceptor.setRetryPolicy(new NeverRetryPolicy()); + interceptor.setRetryOperations(retryTemplate); + retryTemplate.setRetryPolicy(new NeverRetryPolicy()); try { service.service("foo"); fail("Expected Exception."); @@ -149,7 +157,8 @@ public class StatefulRetryOperationsInterceptorTests extends TestCase { try { service.service("foo"); fail("Expected ExhaustedRetryException"); - } catch (ExhaustedRetryException e) { + } + catch (ExhaustedRetryException e) { // expected String message = e.getMessage(); assertTrue("Wrong message: " + message, message.startsWith("Retry was exhausted but there was no recover")); @@ -159,7 +168,8 @@ public class StatefulRetryOperationsInterceptorTests extends TestCase { public void testRecoveryAfterTooManyAttempts() throws Exception { ((Advised) service).addAdvice(interceptor); - interceptor.setRetryPolicy(new NeverRetryPolicy()); + interceptor.setRetryOperations(retryTemplate); + retryTemplate.setRetryPolicy(new NeverRetryPolicy()); try { service.service("foo"); fail("Expected Exception."); @@ -181,7 +191,8 @@ public class StatefulRetryOperationsInterceptorTests extends TestCase { public void testTransformerRecoveryAfterTooManyAttempts() throws Exception { ((Advised) transformer).addAdvice(interceptor); - interceptor.setRetryPolicy(new NeverRetryPolicy()); + interceptor.setRetryOperations(retryTemplate); + retryTemplate.setRetryPolicy(new NeverRetryPolicy()); try { transformer.transform("foo"); fail("Expected Exception."); @@ -194,7 +205,7 @@ public class StatefulRetryOperationsInterceptorTests extends TestCase { interceptor.setRecoverer(new MethodInvocationRecoverer>() { public Collection recover(Object[] data, Throwable cause) { count++; - return Collections.singleton((String)data[0]); + return Collections.singleton((String) data[0]); } }); Collection result = transformer.transform("foo"); diff --git a/spring-batch-integration/.springBeans b/spring-batch-integration/.springBeans index 93b42a00f..c9a3cf7f9 100644 --- a/spring-batch-integration/.springBeans +++ b/spring-batch-integration/.springBeans @@ -15,6 +15,10 @@ src/test/resources/org/springframework/batch/integration/job/MessageOrientedStepIntegrationTests-context.xml src/test/resources/org/springframework/batch/integration/file/ResourceSplitterIntegrationTests-context.xml src/test/resources/org/springframework/batch/integration/retry/TransactionalPollingIntegrationTests-context.xml + src/test/resources/org/springframework/batch/integration/retry/RepeatTransactionalPollingIntegrationTests-context.xml + src/test/resources/org/springframework/batch/integration/retry/RetryRepeatTransactionalPollingIntegrationTests-context.xml + src/test/resources/org/springframework/batch/integration/SmokeTests-context.xml + src/test/resources/org/springframework/batch/integration/retry/RetryTransactionalPollingIntegrationTests-context.xml diff --git a/spring-batch-integration/pom.xml b/spring-batch-integration/pom.xml index 5e3f59bdc..ab2e5a0b5 100644 --- a/spring-batch-integration/pom.xml +++ b/spring-batch-integration/pom.xml @@ -66,7 +66,16 @@ junit junit - 4.4 + test + + + aspectj + aspectjrt + test + + + aspectj + aspectjweaver test diff --git a/spring-batch-integration/src/test/java/org/springframework/batch/integration/item/MessageChannelItemWriterIntegrationTests.java b/spring-batch-integration/src/test/java/org/springframework/batch/integration/item/MessageChannelItemWriterIntegrationTests.java index f216fe98c..23b9a6478 100644 --- a/spring-batch-integration/src/test/java/org/springframework/batch/integration/item/MessageChannelItemWriterIntegrationTests.java +++ b/spring-batch-integration/src/test/java/org/springframework/batch/integration/item/MessageChannelItemWriterIntegrationTests.java @@ -24,6 +24,7 @@ import org.junit.Test; import org.junit.runner.RunWith; import org.springframework.batch.item.ItemWriter; import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.integration.channel.PollableChannel; import org.springframework.integration.message.Message; import org.springframework.test.context.ContextConfiguration; @@ -38,6 +39,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; public class MessageChannelItemWriterIntegrationTests { @Autowired + @Qualifier("requests") private PollableChannel channel; @Autowired diff --git a/spring-batch-integration/src/test/java/org/springframework/batch/integration/retry/PollableSourceRetryTests.java b/spring-batch-integration/src/test/java/org/springframework/batch/integration/retry/PollableSourceRetryTests.java deleted file mode 100644 index 24abf906b..000000000 --- a/spring-batch-integration/src/test/java/org/springframework/batch/integration/retry/PollableSourceRetryTests.java +++ /dev/null @@ -1,353 +0,0 @@ -package org.springframework.batch.integration.retry; - -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertFalse; - -import java.util.ArrayList; -import java.util.Arrays; -import java.util.List; - -import org.aopalliance.aop.Advice; -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; -import org.junit.Test; -import org.springframework.aop.framework.ProxyFactory; -import org.springframework.aop.support.DefaultPointcutAdvisor; -import org.springframework.aop.support.NameMatchMethodPointcut; -import org.springframework.batch.item.ItemReader; -import org.springframework.batch.item.support.ListItemReader; -import org.springframework.batch.repeat.interceptor.RepeatOperationsInterceptor; -import org.springframework.batch.repeat.policy.SimpleCompletionPolicy; -import org.springframework.batch.repeat.support.RepeatTemplate; -import org.springframework.batch.retry.interceptor.MethodArgumentsKeyGenerator; -import org.springframework.batch.retry.interceptor.MethodInvocationRecoverer; -import org.springframework.batch.retry.interceptor.StatefulRetryOperationsInterceptor; -import org.springframework.batch.support.transaction.ResourcelessTransactionManager; -import org.springframework.batch.support.transaction.TransactionAwareProxyFactory; -import org.springframework.context.Lifecycle; -import org.springframework.core.task.SimpleAsyncTaskExecutor; -import org.springframework.integration.channel.DirectChannel; -import org.springframework.integration.channel.MessageChannel; -import org.springframework.integration.message.GenericMessage; -import org.springframework.integration.message.Message; -import org.springframework.integration.message.MessageConsumer; -import org.springframework.integration.message.MessageSource; -import org.springframework.integration.scheduling.IntervalTrigger; -import org.springframework.integration.scheduling.SimpleTaskScheduler; -import org.springframework.integration.scheduling.TaskScheduler; -import org.springframework.transaction.PlatformTransactionManager; -import org.springframework.transaction.interceptor.MatchAlwaysTransactionAttributeSource; -import org.springframework.transaction.interceptor.TransactionInterceptor; -import org.springframework.util.StringUtils; - - -public class PollableSourceRetryTests { - - private Log logger = LogFactory.getLog(getClass()); - - private List processed = new ArrayList(); - - protected List recovered = new ArrayList(); - - public void add(String str) { - logger.debug("Adding: " + str); - processed.add(str); - } - - MethodArgumentsKeyGenerator methodArgumentsKeyGenerator = new MethodArgumentsKeyGenerator() { - @SuppressWarnings("unchecked") - public Object getKey(Object[] item) { - if (item == null) { - return "NULL"; - } - return ((Message) item[0]).getPayload(); - } - }; - - private PlatformTransactionManager transactionManager = new ResourcelessTransactionManager();; - -// @Test -// public void testTransactionalHandlingWithRollback() throws Exception { -// -// List list = TransactionAwareProxyFactory.createTransactionalList(); -// list.addAll(Arrays.asList(StringUtils.commaDelimitedListToStringArray("a,b,fail,d,e,f,g,h,j,k"))); -// int beforeCount = list.size(); -// -// MessageConsumer handler = new MessageConsumer() { -// public void onMessage(Message message) { -// Object payload = message.getPayload(); -// logger.debug("Handling: " + payload); -// processed.add((String) payload); -// if ("fail".equals(payload)) { -// throw new RuntimeException("Planned failure: " + payload); -// } -// } -// }; -// -// MessageSource source = getPollableSource(list); -// MessageChannel target = getChannel(handler); -// SourcePoller trigger = getSourcePoller(source, target, transactionManager, 1); -// TaskScheduler scheduler = getSchedulerWithErrorHandler(trigger); -// -// waitForResults(scheduler, 5, 50); -// -// assertEquals(5, processed.size()); -// assertFalse("No messages got to processor", processed.isEmpty()); -// // First two TX succeed, and the rest rolled back so list has had two -// // elements popped off -// assertEquals(beforeCount - 2, list.size()); -// assertEquals("a", processed.get(0)); -// assertEquals("b", processed.get(1)); -// // stuck in effectively an infinite loop - it fails every time... -// assertEquals("fail", processed.get(2)); -// assertEquals("fail", processed.get(3)); -// assertEquals("fail", processed.get(4)); -// -// } -// -// @Test -// public void testTransactionalHandlingWithRepeat() throws Exception { -// -// List list = TransactionAwareProxyFactory.createTransactionalList(); -// list.addAll(Arrays.asList(StringUtils.commaDelimitedListToStringArray("a,b,fail,d,e,f,g,h,j,k"))); -// int beforeCount = list.size(); -// -// MessageConsumer handler = new MessageConsumer() { -// public void onMessage(Message message) { -// Object payload = message.getPayload(); -// logger.debug("Handling: " + payload); -// processed.add((String) payload); -// if ("fail".equals(payload)) { -// throw new RuntimeException("Planned failure: " + payload); -// } -// } -// }; -// -// MessageSource source = getPollableSource(list); -// MessageChannel target = getChannel(handler); -// SourcePoller trigger = getSourcePoller(source, target, null, 1); -// SourcePoller task = (SourcePoller) getProxy(trigger, SourcePoller.class, new Advice[] { -// new TransactionInterceptor(transactionManager, new MatchAlwaysTransactionAttributeSource()), -// getRepeatOperationsInterceptor(3) }, "run"); -// TaskScheduler scheduler = getSchedulerWithErrorHandler(task); -// -// waitForResults(scheduler, 6, 100); -// -// assertEquals(6, processed.size()); -// assertFalse("No messages got to processor", processed.isEmpty()); -// // Two TX rolled back so list is same size as when it started -// assertEquals(beforeCount, list.size()); -// assertEquals("a", processed.get(0)); -// assertEquals("b", processed.get(1)); -// // stuck in effectively an infinite loop - it fails every time with the -// // same 3 records... -// assertEquals("fail", processed.get(2)); -// assertEquals("a", processed.get(3)); -// assertEquals("b", processed.get(4)); -// -// } -// -// @Test -// public void testTransactionalHandlingWithRetry() throws Exception { -// -// List list = TransactionAwareProxyFactory.createTransactionalList(); -// list.addAll(Arrays.asList(StringUtils.commaDelimitedListToStringArray("a,b,fail,d,e,f,g,h,j,k"))); -// int beforeCount = list.size(); -// -// MessageConsumer handler = new MessageConsumer() { -// public void onMessage(Message message) { -// if (message == null) { -// return; -// } -// Object payload = message.getPayload(); -// logger.debug("Handling: " + payload); -// processed.add((String) payload); -// // INT-184 this won't work if it is a "real" handler that throws -// // MessageHandlingException -// if ("fail".equals(payload)) { -// throw new RuntimeException("Planned failure: " + payload); -// } -// } -// }; -// -// MessageSource source = getPollableSource(list); -// MessageChannel target = getChannel(handler); -// // this was the old dispatch advice chain -// target = (MessageChannel) getProxy(target, MessageChannel.class, -// new Advice[] { getRetryOperationsInterceptor(methodArgumentsKeyGenerator) }, "send"); -// SourcePoller trigger = getSourcePoller(source, target, transactionManager, 1); -// -// TaskScheduler scheduler = getSchedulerWithErrorHandler(trigger); -// -// waitForResults(scheduler, 4, 40); -// -// assertEquals(4, processed.size()); -// assertEquals(1, recovered.size()); -// assertFalse("No messages got to processor", processed.isEmpty()); -// // 4 items from list should have been processed (with no repeats, since -// // the failed item was recovered with no retry - NeverRetryPolicy) -// assertEquals(beforeCount - 4, list.size()); -// assertEquals("a", processed.get(0)); -// assertEquals("b", processed.get(1)); -// // retry makes it fail once then recover... -// assertEquals("fail", processed.get(2)); -// assertEquals("d", processed.get(3)); -// -// } -// -// @Test -// public void testTransactionalHandlingWithRepeatAndRetry() throws Exception { -// -// List list = TransactionAwareProxyFactory.createTransactionalList(); -// list.addAll(Arrays.asList(StringUtils.commaDelimitedListToStringArray("a,fail,c,d,e,f,g,h,j,k"))); -// int beforeCount = list.size(); -// -// MessageConsumer handler = new MessageConsumer() { -// public void onMessage(Message message) { -// Object payload = message.getPayload(); -// logger.debug("Handling: " + payload); -// processed.add((String) payload); -// if ("fail".equals(payload)) { -// throw new RuntimeException("Planned failure: " + payload); -// } -// } -// }; -// -// MessageSource source = getPollableSource(list); -// MessageChannel target = getChannel(handler); -// -// // this was the old dispatch advice chain -// target = (MessageChannel) getProxy(target, MessageChannel.class, -// new Advice[] { getRetryOperationsInterceptor(methodArgumentsKeyGenerator) }, "send"); -// SourcePoller trigger = getSourcePoller(source, target, null, 1); -// SourcePoller task = (SourcePoller) getProxy(trigger, SourcePoller.class, new Advice[] { -// new TransactionInterceptor(transactionManager, new MatchAlwaysTransactionAttributeSource()), -// getRepeatOperationsInterceptor(3) }, "run"); -// TaskScheduler scheduler = getSchedulerWithErrorHandler(task); -// -// waitForResults(scheduler, 6, 100); -// System.err.println(processed); -// System.err.println(list); -// -// assertFalse("No messages got to processor", processed.isEmpty()); -// assertEquals(7, processed.size()); -// // 6 items were removed from the list -// assertEquals(beforeCount - 6, list.size()); -// assertEquals("a", processed.get(0)); -// assertEquals("fail", processed.get(1)); -// // retry makes it fail once then recover... -// assertEquals("a", processed.get(2)); -// assertEquals("c", processed.get(3)); -// assertEquals("d", processed.get(4)); -// -// } -// -// private SourcePoller getSourcePoller(MessageSource source, MessageChannel channel, -// PlatformTransactionManager transactionManager, int maxMessagesPerPoll) { -// SourcePoller poller = new SourcePoller(source, channel, new IntervalTrigger(100)); -// poller.setTransactionManager(transactionManager); -// poller.setMaxMessagesPerPoll(maxMessagesPerPoll); -// return poller; -// } -// -// private DirectChannel getChannel(MessageConsumer handler) { -// DirectChannel channel = new DirectChannel(); -// channel.setBeanName("input"); -// channel.subscribe(handler); -// return channel; -// } -// -// private void waitForResults(Lifecycle lifecycle, int count, int maxTries) throws InterruptedException { -// lifecycle.start(); -// int timeout = 0; -// while (processed.size() < count && timeout++ < maxTries) { -// Thread.sleep(10); -// } -// lifecycle.stop(); -// } -// -// private MessageSource getPollableSource(List list) { -// final ItemReader reader = new ListItemReader(list) { -// public String read() { -// String item = super.read(); -// logger.debug("Reading: " + item); -// return item; -// } -// }; -// MessageSource source = new MessageSource() { -// public Message receive() { -// try { -// String payload = reader.read(); -// if (payload == null) -// return null; -// return new GenericMessage(payload); -// } -// catch (RuntimeException e) { -// throw e; -// } -// catch (Exception e) { -// throw new IllegalStateException(e); -// } -// } -// }; -// return source; -// } -// -// private TaskScheduler getSchedulerWithErrorHandler(SourcePoller task) { -// SimpleAsyncTaskExecutor executor = new SimpleAsyncTaskExecutor(); -// executor.setConcurrencyLimit(1); -// TaskScheduler scheduler = new SimpleTaskScheduler(executor); -// scheduler.schedule(task, task.getTrigger()); -// return scheduler; -// } -// - /** - * @param methodArgumentsKeyGenerator - * @return - */ - private StatefulRetryOperationsInterceptor getRetryOperationsInterceptor( - MethodArgumentsKeyGenerator methodArgumentsKeyGenerator) { - StatefulRetryOperationsInterceptor advice = new StatefulRetryOperationsInterceptor(); - advice.setRecoverer(new MethodInvocationRecoverer() { - @SuppressWarnings("unchecked") - public Boolean recover(Object[] data, Throwable cause) { - if (data == null) { - return false; - } - String payload = ((Message) data[0]).getPayload(); - logger.debug("Recovering: " + payload); - recovered.add(payload); - return true; - } - }); - advice.setKeyGenerator(methodArgumentsKeyGenerator); - return advice; - } - - private RepeatOperationsInterceptor getRepeatOperationsInterceptor(int commitInterval) { - RepeatOperationsInterceptor advice = new RepeatOperationsInterceptor(); - RepeatTemplate repeatTemplate = new RepeatTemplate(); - repeatTemplate.setCompletionPolicy(new SimpleCompletionPolicy(commitInterval)); - advice.setRepeatOperations(repeatTemplate); - return advice; - } - - /** - * @param commitInterval - * @return - */ - private Object getProxy(Object target, Class intf, Advice[] advices, String methodName) { - ProxyFactory factory = new ProxyFactory(target); - for (int i = 0; i < advices.length; i++) { - DefaultPointcutAdvisor advisor = new DefaultPointcutAdvisor(advices[i]); - NameMatchMethodPointcut pointcut = new NameMatchMethodPointcut(); - pointcut.addMethodName(methodName); - advisor.setPointcut(pointcut); - factory.addAdvisor(advisor); - } - factory.setProxyTargetClass(false); - factory.addInterface(intf); - return factory.getProxy(); - } - -} diff --git a/spring-batch-integration/src/test/java/org/springframework/batch/integration/retry/RepeatTransactionalPollingIntegrationTests.java b/spring-batch-integration/src/test/java/org/springframework/batch/integration/retry/RepeatTransactionalPollingIntegrationTests.java new file mode 100644 index 000000000..a5d3c6837 --- /dev/null +++ b/spring-batch-integration/src/test/java/org/springframework/batch/integration/retry/RepeatTransactionalPollingIntegrationTests.java @@ -0,0 +1,98 @@ +package org.springframework.batch.integration.retry; + +import static org.junit.Assert.assertEquals; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.springframework.batch.support.transaction.TransactionAwareProxyFactory; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.Lifecycle; +import org.springframework.integration.annotation.ChannelAdapter; +import org.springframework.integration.annotation.MessageEndpoint; +import org.springframework.integration.annotation.Poller; +import org.springframework.integration.annotation.ServiceActivator; +import org.springframework.integration.bus.MessageBus; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; +import org.springframework.util.StringUtils; + +@ContextConfiguration +@RunWith(SpringJUnit4ClassRunner.class) +@MessageEndpoint +public class RepeatTransactionalPollingIntegrationTests { + + private Log logger = LogFactory.getLog(getClass()); + + private List processed = new ArrayList(); + + private List list = new ArrayList(); + + @Autowired + private MessageBus bus; + + private volatile int count = 0; + + @ServiceActivator(inputChannel = "requests", outputChannel = "replies") + public String process(String message) { + String result = message + ": " + count; + logger.debug("Handling: " + message); + processed.add(message); + if ("fail".equals(message)) { + throw new RuntimeException("Planned failure"); + } + return result; + } + + @ChannelAdapter("requests") + @Poller(interval=10,adviceChain={"txAdvice","repeatAdvice"}) + public String input() { + logger.debug("Polling: " + count); + if (list.isEmpty()) { + return null; + } + return list.remove(0); + } + + @ChannelAdapter("replies") + public void output(String message) { + count++; + logger.debug("Handled: " + message); + } + + @Test + @DirtiesContext + public void testSunnyDay() throws Exception { + list = TransactionAwareProxyFactory.createTransactionalList(Arrays.asList(StringUtils + .commaDelimitedListToStringArray("a,b,c,d,e,f,g,h,j,k"))); + waitForResults(bus, 4, 60); + assertEquals(4,processed.size()); // a,b,c,d + assertEquals(4,count); + } + + @Test + @DirtiesContext + public void testRollback() throws Exception { + list = TransactionAwareProxyFactory.createTransactionalList(Arrays.asList(StringUtils + .commaDelimitedListToStringArray("a,b,fail,d,e,f,g,h,j,k"))); + waitForResults(bus, 4, 30); // (a,b), (fail), (fail) + assertEquals(4,processed.size()); // a,b,fail,fail + assertEquals(2,count); // a,b + } + + private void waitForResults(Lifecycle lifecycle, int count, int maxTries) throws InterruptedException { + lifecycle.start(); + int timeout = 0; + while (processed.size() < count && timeout++ < maxTries) { + Thread.sleep(10); + } + lifecycle.stop(); + } + +} diff --git a/spring-batch-integration/src/test/java/org/springframework/batch/integration/retry/RetryRepeatTransactionalPollingIntegrationTests.java b/spring-batch-integration/src/test/java/org/springframework/batch/integration/retry/RetryRepeatTransactionalPollingIntegrationTests.java new file mode 100644 index 000000000..43a804ebd --- /dev/null +++ b/spring-batch-integration/src/test/java/org/springframework/batch/integration/retry/RetryRepeatTransactionalPollingIntegrationTests.java @@ -0,0 +1,93 @@ +package org.springframework.batch.integration.retry; + +import static org.junit.Assert.assertEquals; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.springframework.batch.support.transaction.TransactionAwareProxyFactory; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.Lifecycle; +import org.springframework.integration.annotation.ChannelAdapter; +import org.springframework.integration.annotation.MessageEndpoint; +import org.springframework.integration.annotation.Poller; +import org.springframework.integration.bus.MessageBus; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; +import org.springframework.util.StringUtils; + +@ContextConfiguration +@RunWith(SpringJUnit4ClassRunner.class) +@MessageEndpoint +public class RetryRepeatTransactionalPollingIntegrationTests { + + private Log logger = LogFactory.getLog(getClass()); + + private List list = new ArrayList(); + + @Autowired + private SimpleRecoverer recoverer; + + @Autowired + private SimpleService service; + + @Autowired + private MessageBus bus; + + private volatile int count = 0; + + @ChannelAdapter("requests") + @Poller(interval=10,adviceChain={"txAdvice","repeatAdvice"}) + public String input() { + logger.debug("Polling: " + count); + if (list.isEmpty()) { + return null; + } + return list.remove(0); + } + + @ChannelAdapter("replies") + public void output(String message) { + count++; + logger.debug("Handled: " + message); + } + + @Test + @DirtiesContext + public void testSunnyDay() throws Exception { + list = TransactionAwareProxyFactory.createTransactionalList(Arrays.asList(StringUtils + .commaDelimitedListToStringArray("a,b,c,d,e,f,g,h,j,k"))); + waitForResults(bus, 4, 60); + assertEquals(4,service.getProcessed().size()); // a,b,c,d + assertEquals(4,count); + } + + @Test + @DirtiesContext + public void testRollback() throws Exception { + list = TransactionAwareProxyFactory.createTransactionalList(Arrays.asList(StringUtils + .commaDelimitedListToStringArray("a,b,fail,d,e,f,g,h,j,k"))); + waitForResults(bus, 6, 100); // (a,b), (fail), (fail), ([fail],d), (e,f) + System.err.println(service.getProcessed()); + System.err.println(recoverer.getRecovered()); + assertEquals(7,service.getProcessed().size()); // a,b,fail,fail,d,e,f + assertEquals(1,recoverer.getRecovered().size()); // fail + assertEquals(5,count); // a,b,d,e,f + } + + private void waitForResults(Lifecycle lifecycle, int count, int maxTries) throws InterruptedException { + lifecycle.start(); + int timeout = 0; + while (service.getProcessed().size() < count && timeout++ < maxTries) { + Thread.sleep(10); + } + lifecycle.stop(); + } + +} diff --git a/spring-batch-integration/src/test/java/org/springframework/batch/integration/retry/RetryTransactionalPollingIntegrationTests.java b/spring-batch-integration/src/test/java/org/springframework/batch/integration/retry/RetryTransactionalPollingIntegrationTests.java new file mode 100644 index 000000000..93def00a8 --- /dev/null +++ b/spring-batch-integration/src/test/java/org/springframework/batch/integration/retry/RetryTransactionalPollingIntegrationTests.java @@ -0,0 +1,93 @@ +package org.springframework.batch.integration.retry; + +import static org.junit.Assert.assertEquals; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.springframework.batch.support.transaction.TransactionAwareProxyFactory; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.Lifecycle; +import org.springframework.integration.annotation.ChannelAdapter; +import org.springframework.integration.annotation.MessageEndpoint; +import org.springframework.integration.annotation.Poller; +import org.springframework.integration.bus.MessageBus; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; +import org.springframework.util.StringUtils; + +@ContextConfiguration +@RunWith(SpringJUnit4ClassRunner.class) +@MessageEndpoint +public class RetryTransactionalPollingIntegrationTests { + + private Log logger = LogFactory.getLog(getClass()); + + private List list = new ArrayList(); + + @Autowired + private SimpleRecoverer recoverer; + + @Autowired + private SimpleService service; + + @Autowired + private MessageBus bus; + + private volatile int count = 0; + + @ChannelAdapter("requests") + @Poller(interval=10, transactionManager="transactionManager") + public String input() { + logger.debug("Polling: " + count); + if (list.isEmpty()) { + return null; + } + return list.remove(0); + } + + @ChannelAdapter("replies") + public void output(String message) { + count++; + logger.debug("Handled: " + message); + } + + @Test + @DirtiesContext + public void testSunnyDay() throws Exception { + list = TransactionAwareProxyFactory.createTransactionalList(Arrays.asList(StringUtils + .commaDelimitedListToStringArray("a,b,c,d,e,f,g,h,j,k"))); + waitForResults(bus, 4, 60); + assertEquals(4,service.getProcessed().size()); // a,b,c,d + assertEquals(4,count); + } + + @Test + @DirtiesContext + public void testRollback() throws Exception { + list = TransactionAwareProxyFactory.createTransactionalList(Arrays.asList(StringUtils + .commaDelimitedListToStringArray("a,b,fail,d,e,f,g,h,j,k"))); + waitForResults(bus, 6, 200); // (a), (b), (fail), (fail), ... + System.err.println(service.getProcessed()); + System.err.println(recoverer.getRecovered()); + assertEquals(6,service.getProcessed().size()); // a,b,fail,fail,d,e + assertEquals(1,recoverer.getRecovered().size()); // fail + assertEquals(4,count); // a,b,d,e + } + + private void waitForResults(Lifecycle lifecycle, int count, int maxTries) throws InterruptedException { + lifecycle.start(); + int timeout = 0; + while (service.getProcessed().size() < count && timeout++ < maxTries) { + Thread.sleep(10); + } + lifecycle.stop(); + } + +} diff --git a/spring-batch-integration/src/test/java/org/springframework/batch/integration/retry/Service.java b/spring-batch-integration/src/test/java/org/springframework/batch/integration/retry/Service.java new file mode 100644 index 000000000..711f1cc50 --- /dev/null +++ b/spring-batch-integration/src/test/java/org/springframework/batch/integration/retry/Service.java @@ -0,0 +1,26 @@ +/* + * Copyright 2006-2007 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.batch.integration.retry; + +/** + * @author Dave Syer + * + */ +public interface Service { + + String process(String message); + +} diff --git a/spring-batch-integration/src/test/java/org/springframework/batch/integration/retry/SimpleRecoverer.java b/spring-batch-integration/src/test/java/org/springframework/batch/integration/retry/SimpleRecoverer.java new file mode 100644 index 000000000..f9be6fd76 --- /dev/null +++ b/spring-batch-integration/src/test/java/org/springframework/batch/integration/retry/SimpleRecoverer.java @@ -0,0 +1,37 @@ +package org.springframework.batch.integration.retry; + +import java.util.ArrayList; +import java.util.List; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.springframework.batch.retry.interceptor.MethodInvocationRecoverer; + +/** + * @author Dave Syer + * + */ +public final class SimpleRecoverer implements MethodInvocationRecoverer { + + private Log logger = LogFactory.getLog(getClass()); + + private final List recovered = new ArrayList(); + + /** + * Public getter for the recovered. + * @return the recovered + */ + public List getRecovered() { + return recovered; + } + + public String recover(Object[] data, Throwable cause) { + if (data == null) { + return null; + } + String payload = (String) data[0]; + logger.debug("Recovering: " + payload); + recovered.add(payload); + return null; + } +} \ No newline at end of file diff --git a/spring-batch-integration/src/test/java/org/springframework/batch/integration/retry/SimpleService.java b/spring-batch-integration/src/test/java/org/springframework/batch/integration/retry/SimpleService.java new file mode 100644 index 000000000..6a72fb424 --- /dev/null +++ b/spring-batch-integration/src/test/java/org/springframework/batch/integration/retry/SimpleService.java @@ -0,0 +1,39 @@ +package org.springframework.batch.integration.retry; + +import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.springframework.integration.annotation.MessageEndpoint; +import org.springframework.integration.annotation.ServiceActivator; + +@MessageEndpoint +public class SimpleService implements Service { + + private Log logger = LogFactory.getLog(getClass()); + + private List processed = new CopyOnWriteArrayList(); + + private int count = 0; + + /** + * Public getter for the processed. + * @return the processed + */ + public List getProcessed() { + return processed; + } + + @ServiceActivator(inputChannel = "requests", outputChannel = "replies") + public String process(String message) { + String result = message + ": " + (count++); + logger.debug("Handling: " + message); + processed.add(message); + if ("fail".equals(message)) { + throw new RuntimeException("Planned failure"); + } + return result; + } + +} diff --git a/spring-batch-integration/src/test/resources/org/springframework/batch/integration/retry/RepeatTransactionalPollingIntegrationTests-context.xml b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/retry/RepeatTransactionalPollingIntegrationTests-context.xml new file mode 100644 index 000000000..969b0e672 --- /dev/null +++ b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/retry/RepeatTransactionalPollingIntegrationTests-context.xml @@ -0,0 +1,27 @@ + + + + + + + + + + + + + + + + + + \ No newline at end of file diff --git a/spring-batch-integration/src/test/resources/org/springframework/batch/integration/retry/RetryRepeatTransactionalPollingIntegrationTests-context.xml b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/retry/RetryRepeatTransactionalPollingIntegrationTests-context.xml new file mode 100644 index 000000000..e1ca2836c --- /dev/null +++ b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/retry/RetryRepeatTransactionalPollingIntegrationTests-context.xml @@ -0,0 +1,47 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + \ No newline at end of file diff --git a/spring-batch-integration/src/test/resources/org/springframework/batch/integration/retry/RetryTransactionalPollingIntegrationTests-context.xml b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/retry/RetryTransactionalPollingIntegrationTests-context.xml new file mode 100644 index 000000000..c753ddecb --- /dev/null +++ b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/retry/RetryTransactionalPollingIntegrationTests-context.xml @@ -0,0 +1,31 @@ + + + + + + + + + + + + + + + + + + + + + + \ No newline at end of file diff --git a/spring-batch-integration/src/test/resources/org/springframework/batch/integration/retry/TransactionalPollingIntegrationTests-context.xml b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/retry/TransactionalPollingIntegrationTests-context.xml index 722072c2a..06c400cf5 100644 --- a/spring-batch-integration/src/test/resources/org/springframework/batch/integration/retry/TransactionalPollingIntegrationTests-context.xml +++ b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/retry/TransactionalPollingIntegrationTests-context.xml @@ -9,8 +9,8 @@ http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration-1.0.xsd http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-2.5.xsd"> - + \ No newline at end of file