Spring/spring-batch
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Refactor retry tests in integration to work with new adviceChain param in @POller

Browse Source
This commit is contained in:
dsyer
2008-10-14 13:54:37 +00:00
parent 5495f9b769
commit 2abeefba71
18 changed files with 541 additions and 381 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/interceptor/RetryOperationsInterceptor.java
View File

@@ -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;
}

24
spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/interceptor/StatefulRetryOperationsInterceptor.java
View File

@@ -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 + ")");

4
spring-batch-infrastructure/src/test/java/org/springframework/batch/retry/interceptor/RetryOperationsInterceptorTests.java
View File

@@ -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.");

29
spring-batch-infrastructure/src/test/java/org/springframework/batch/retry/interceptor/StatefulRetryOperationsInterceptorTests.java
View File

@@ -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<Collection<String>>() {
public Collection<String> recover(Object[] data, Throwable cause) {
count++;
return Collections.singleton((String)data[0]);
return Collections.singleton((String) data[0]);
}
});
Collection<String> result = transformer.transform("foo");

4
spring-batch-integration/.springBeans
View File

@@ -15,6 +15,10 @@
<config>src/test/resources/org/springframework/batch/integration/job/MessageOrientedStepIntegrationTests-context.xml</config>
<config>src/test/resources/org/springframework/batch/integration/file/ResourceSplitterIntegrationTests-context.xml</config>
<config>src/test/resources/org/springframework/batch/integration/retry/TransactionalPollingIntegrationTests-context.xml</config>
<config>src/test/resources/org/springframework/batch/integration/retry/RepeatTransactionalPollingIntegrationTests-context.xml</config>
<config>src/test/resources/org/springframework/batch/integration/retry/RetryRepeatTransactionalPollingIntegrationTests-context.xml</config>
<config>src/test/resources/org/springframework/batch/integration/SmokeTests-context.xml</config>
<config>src/test/resources/org/springframework/batch/integration/retry/RetryTransactionalPollingIntegrationTests-context.xml</config>
</configs>
<configSets>
</configSets>

11
spring-batch-integration/pom.xml
View File

@@ -66,7 +66,16 @@
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>4.4</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>aspectj</groupId>
<artifactId>aspectjrt</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>aspectj</groupId>
<artifactId>aspectjweaver</artifactId>
<scope>test</scope>
</dependency>
<dependency>

2
spring-batch-integration/src/test/java/org/springframework/batch/integration/item/MessageChannelItemWriterIntegrationTests.java
View File

@@ -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

353
spring-batch-integration/src/test/java/org/springframework/batch/integration/retry/PollableSourceRetryTests.java
View File

@@ -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<String> processed = new ArrayList<String>();
protected List<String> recovered = new ArrayList<String>();
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<Object>) item[0]).getPayload();
}
};
private PlatformTransactionManager transactionManager = new ResourcelessTransactionManager();;
// @Test
// public void testTransactionalHandlingWithRollback() throws Exception {
//
// List<String> 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<Object> 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<String> 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<Object> 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<String> 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<Object> 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<String> 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<Object> 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<Object> 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<Object> getPollableSource(List<String> list) {
// final ItemReader<String> reader = new ListItemReader<String>(list) {
// public String read() {
// String item = super.read();
// logger.debug("Reading: " + item);
// return item;
// }
// };
// MessageSource<Object> source = new MessageSource<Object>() {
// public Message<Object> receive() {
// try {
// String payload = reader.read();
// if (payload == null)
// return null;
// return new GenericMessage<Object>(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<Boolean>() {
@SuppressWarnings("unchecked")
public Boolean recover(Object[] data, Throwable cause) {
if (data == null) {
return false;
}
String payload = ((Message<String>) 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();
}
}

98
spring-batch-integration/src/test/java/org/springframework/batch/integration/retry/RepeatTransactionalPollingIntegrationTests.java Normal file
View File

@@ -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<String> processed = new ArrayList<String>();
private List<String> list = new ArrayList<String>();
@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();
}
}

93
spring-batch-integration/src/test/java/org/springframework/batch/integration/retry/RetryRepeatTransactionalPollingIntegrationTests.java Normal file
View File

@@ -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<String> list = new ArrayList<String>();
@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();
}
}

93
spring-batch-integration/src/test/java/org/springframework/batch/integration/retry/RetryTransactionalPollingIntegrationTests.java Normal file
View File

@@ -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<String> list = new ArrayList<String>();
@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();
}
}

26
spring-batch-integration/src/test/java/org/springframework/batch/integration/retry/Service.java Normal file
View File

@@ -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);
}

37
spring-batch-integration/src/test/java/org/springframework/batch/integration/retry/SimpleRecoverer.java Normal file
View File

@@ -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<String> {
private Log logger = LogFactory.getLog(getClass());
private final List<String> recovered = new ArrayList<String>();
/**
* Public getter for the recovered.
* @return the recovered
*/
public List<String> 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;
}
}

39
spring-batch-integration/src/test/java/org/springframework/batch/integration/retry/SimpleService.java Normal file
View File

@@ -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<String> processed = new CopyOnWriteArrayList<String>();
private int count = 0;
/**
* Public getter for the processed.
* @return the processed
*/
public List<String> 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;
}
}

27
spring-batch-integration/src/test/resources/org/springframework/batch/integration/retry/RepeatTransactionalPollingIntegrationTests-context.xml Normal file
View File

@@ -0,0 +1,27 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans" xmlns:integration="http://www.springframework.org/schema/integration"
xmlns:context="http://www.springframework.org/schema/context" xmlns:aop="http://www.springframework.org/schema/aop"
xmlns:tx="http://www.springframework.org/schema/tx" xmlns:p="http://www.springframework.org/schema/p" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop-2.5.xsd
http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-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/tx http://www.springframework.org/schema/tx/spring-tx-2.5.xsd">
<import resource="TransactionalPollingIntegrationTests-context.xml"/>
<tx:advice id="txAdvice">
<tx:attributes><tx:method name="*"/></tx:attributes>
</tx:advice>
<bean id="repeatAdvice" class="org.springframework.batch.repeat.interceptor.RepeatOperationsInterceptor">
<property name="repeatOperations">
<bean class="org.springframework.batch.repeat.support.RepeatTemplate">
<property name="completionPolicy">
<bean class="org.springframework.batch.repeat.policy.SimpleCompletionPolicy">
<property name="chunkSize" value="2" />
</bean>
</property>
</bean>
</property>
</bean>
</beans>

47
spring-batch-integration/src/test/resources/org/springframework/batch/integration/retry/RetryRepeatTransactionalPollingIntegrationTests-context.xml Normal file
View File

@@ -0,0 +1,47 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans" xmlns:integration="http://www.springframework.org/schema/integration"
xmlns:context="http://www.springframework.org/schema/context" xmlns:aop="http://www.springframework.org/schema/aop"
xmlns:tx="http://www.springframework.org/schema/tx" xmlns:p="http://www.springframework.org/schema/p" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop-2.5.xsd
http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-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/tx http://www.springframework.org/schema/tx/spring-tx-2.5.xsd">
<import resource="TransactionalPollingIntegrationTests-context.xml"/>
<tx:advice id="txAdvice">
<tx:attributes>
<tx:method name="*" />
</tx:attributes>
</tx:advice>
<bean id="repeatAdvice" class="org.springframework.batch.repeat.interceptor.RepeatOperationsInterceptor">
<property name="repeatOperations">
<bean class="org.springframework.batch.repeat.support.RepeatTemplate">
<property name="completionPolicy">
<bean class="org.springframework.batch.repeat.policy.SimpleCompletionPolicy">
<property name="chunkSize" value="2" />
</bean>
</property>
</bean>
</property>
</bean>
<bean id="service" class="org.springframework.batch.integration.retry.SimpleService" />
<bean id="recoverer" class="org.springframework.batch.integration.retry.SimpleRecoverer" />
<bean id="retryAdvice" class="org.springframework.batch.retry.interceptor.StatefulRetryOperationsInterceptor">
<property name="retryOperations">
<bean class="org.springframework.batch.retry.support.RetryTemplate">
<property name="retryPolicy">
<bean class="org.springframework.batch.retry.policy.SimpleRetryPolicy">
<property name="maxAttempts" value="2"/>
</bean>
</property>
</bean>
</property>
<property name="recoverer" ref="recoverer" />
</bean>
<aop:config proxy-target-class="true">
<aop:advisor advice-ref="retryAdvice" pointcut="execution(* org.springframework.batch.integration.retry.Service+.process(..))" />
</aop:config>
</beans>

31
spring-batch-integration/src/test/resources/org/springframework/batch/integration/retry/RetryTransactionalPollingIntegrationTests-context.xml Normal file
View File

@@ -0,0 +1,31 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans" xmlns:integration="http://www.springframework.org/schema/integration"
xmlns:context="http://www.springframework.org/schema/context" xmlns:aop="http://www.springframework.org/schema/aop"
xmlns:tx="http://www.springframework.org/schema/tx" xmlns:p="http://www.springframework.org/schema/p" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop-2.5.xsd
http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-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/tx http://www.springframework.org/schema/tx/spring-tx-2.5.xsd">
<import resource="TransactionalPollingIntegrationTests-context.xml" />
<bean id="service" class="org.springframework.batch.integration.retry.SimpleService" />
<bean id="recoverer" class="org.springframework.batch.integration.retry.SimpleRecoverer" />
<bean id="retryAdvice" class="org.springframework.batch.retry.interceptor.StatefulRetryOperationsInterceptor">
<property name="retryOperations">
<bean class="org.springframework.batch.retry.support.RetryTemplate">
<property name="retryPolicy">
<bean class="org.springframework.batch.retry.policy.SimpleRetryPolicy">
<property name="maxAttempts" value="2" />
</bean>
</property>
</bean>
</property>
<property name="recoverer" ref="recoverer" />
</bean>
<aop:config proxy-target-class="true">
<aop:advisor advice-ref="retryAdvice" pointcut="execution(* org.springframework.batch.integration.retry.Service+.process(..))" />
</aop:config>
</beans>

2
spring-batch-integration/src/test/resources/org/springframework/batch/integration/retry/TransactionalPollingIntegrationTests-context.xml
View File

@@ -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">
<import resource="classpath:/simple-job-launcher-context.xml" />
<integration:message-bus auto-startup="false" enable-annotations="true"/>
<integration:channel id="requests"/>
<integration:channel id="replies" />
<bean id="transactionManager" class="org.springframework.batch.support.transaction.ResourcelessTransactionManager" />
</beans>