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

Fix unit tests in integration project (except retry)

Browse Source
This commit is contained in:
dsyer
2008-10-09 09:06:22 +00:00
parent a98d26b353
commit eae12dba78
18 changed files with 389 additions and 352 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
spring-batch-integration/.springBeans
View File

@@ -1,13 +1,12 @@
<?xml version="1.0" encoding="UTF-8"?>
<beansProjectDescription>
<version>1</version>
<pluginVersion><![CDATA[2.1.0.v200806081427]]></pluginVersion>
<pluginVersion><![CDATA[2.2.0.v200809261800]]></pluginVersion>
<configSuffixes>
<configSuffix><![CDATA[xml]]></configSuffix>
</configSuffixes>
<enableImports><![CDATA[true]]></enableImports>
<configs>
<config>src/test/resources/integration-context.xml</config>
<config>src/test/resources/job-execution-context.xml</config>
<config>src/test/resources/simple-job-launcher-context.xml</config>
<config>src/test/resources/org/springframework/batch/integration/chunk/ChunkMessageItemWriterIntegrationTests-context.xml</config>
@@ -15,6 +14,7 @@
<config>src/test/resources/org/springframework/batch/integration/item/MessageChannelItemWriterIntegrationTests-context.xml</config>
<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>
</configs>
<configSets>
</configSets>

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

@@ -17,7 +17,7 @@ import org.springframework.integration.message.Message;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@ContextConfiguration(locations = "/integration-context.xml")
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
@MessageEndpoint
public class SmokeTests {

7
spring-batch-integration/src/test/java/org/springframework/batch/integration/chunk/ChunkMessageItemWriterIntegrationTests.java
View File

@@ -30,6 +30,7 @@ import org.springframework.batch.repeat.ExitStatus;
import org.springframework.batch.support.transaction.ResourcelessTransactionManager;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.integration.bus.MessageBus;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.channel.PollableChannel;
import org.springframework.integration.message.GenericMessage;
@@ -44,6 +45,9 @@ public class ChunkMessageItemWriterIntegrationTests {
private ChunkMessageChannelItemWriter<Object> writer = new ChunkMessageChannelItemWriter<Object>();
@Autowired
private MessageBus bus;
@Autowired
@Qualifier("requests")
private MessageChannel requests;
@@ -80,6 +84,8 @@ public class ChunkMessageItemWriterIntegrationTests {
System.err.println(message);
message = replies.receive(10);
}
bus.start();
}
@@ -87,6 +93,7 @@ public class ChunkMessageItemWriterIntegrationTests {
public void tearDown() {
while (replies.receive(10L) != null) {
}
bus.stop();
}
@Test

15
spring-batch-integration/src/test/java/org/springframework/batch/integration/file/FileToMessagesJobFactoryBeanTests.java
View File

@@ -34,8 +34,7 @@ import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.integration.JobRepositorySupport;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.file.FlatFileItemReader;
import org.springframework.batch.item.file.mapping.FieldSet;
import org.springframework.batch.item.file.mapping.PassThroughFieldSetMapper;
import org.springframework.batch.item.file.mapping.PassThroughLineMapper;
import org.springframework.batch.support.transaction.ResourcelessTransactionManager;
import org.springframework.beans.factory.annotation.Required;
import org.springframework.core.annotation.AnnotationUtils;
@@ -53,9 +52,9 @@ import org.springframework.util.ReflectionUtils;
public class FileToMessagesJobFactoryBeanTests {
private static final String FILE_INPUT_PATH = ResourcePayloadAsJobParameterStrategy.FILE_INPUT_PATH;
private FileToMessagesJobFactoryBean<FieldSet> factory = new FileToMessagesJobFactoryBean<FieldSet>();
private FileToMessagesJobFactoryBean<String> factory = new FileToMessagesJobFactoryBean<String>();
private DirectChannel channel = new DirectChannel();
private List<FieldSet> receiver = new ArrayList<FieldSet>();
private List<String> receiver = new ArrayList<String>();
private JobRepositorySupport jobRepository;
@Before
@@ -63,14 +62,14 @@ public class FileToMessagesJobFactoryBeanTests {
jobRepository = new JobRepositorySupport();
factory.setJobRepository(jobRepository);
factory.setTransactionManager(new ResourcelessTransactionManager());
FlatFileItemReader<FieldSet> itemReader = new FlatFileItemReader<FieldSet>();
itemReader.setFieldSetMapper(new PassThroughFieldSetMapper());
FlatFileItemReader<String> itemReader = new FlatFileItemReader<String>();
itemReader.setLineMapper(new PassThroughLineMapper());
factory.setItemReader(itemReader);
factory.setChannel(channel);
channel.subscribe(new MessageConsumer() {
public void onMessage(Message<?> message) {
// TODO: Ask Mark: unsafe cast...
receiver.add((FieldSet) message.getPayload());
receiver.add((String) message.getPayload());
}
});
}
@@ -182,7 +181,7 @@ public class FileToMessagesJobFactoryBeanTests {
assertNotNull(jobExecution);
assertEquals(BatchStatus.COMPLETED, jobExecution.getStatus());
FieldSet payload;
String payload;
// first line from properties file
payload = receiver.get(0);

5
spring-batch-integration/src/test/java/org/springframework/batch/integration/file/ResourceSplitterIntegrationTests.java
View File

@@ -27,6 +27,7 @@ import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.core.io.Resource;
import org.springframework.integration.annotation.MessageEndpoint;
import org.springframework.integration.annotation.Splitter;
import org.springframework.integration.bus.MessageBus;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.channel.PollableChannel;
import org.springframework.integration.message.GenericMessage;
@@ -51,6 +52,9 @@ public class ResourceSplitterIntegrationTests {
@Qualifier("requests")
private PollableChannel requests;
@Autowired
private MessageBus bus;
/*
* This is so cool (but see INT-190)...<br/>
*
@@ -67,6 +71,7 @@ public class ResourceSplitterIntegrationTests {
@SuppressWarnings("unchecked")
@Test
public void testVanillaConversion() throws Exception {
bus.start();
resources.send(new GenericMessage<String>("classpath:*-context.xml"));
Message<Resource> message = (Message<Resource>) requests.receive(200L);
assertNotNull(message);

5
spring-batch-integration/src/test/java/org/springframework/batch/integration/job/MessageOrientedStepIntegrationTests.java
View File

@@ -26,6 +26,7 @@ import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.launch.JobLauncher;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.integration.bus.MessageBus;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@@ -37,6 +38,9 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@RunWith(SpringJUnit4ClassRunner.class)
public class MessageOrientedStepIntegrationTests {
@Autowired
private MessageBus bus;
@Autowired
private JobLauncher jobLauncher;
@@ -46,6 +50,7 @@ public class MessageOrientedStepIntegrationTests {
@Test
public void testLaunchJob() throws Exception {
bus.start();
JobExecution jobExecution = jobLauncher.run(job, new JobParameters());
assertEquals(BatchStatus.COMPLETED, jobExecution.getStatus());
}

5
spring-batch-integration/src/test/java/org/springframework/batch/integration/launch/JobLaunchingMessageHandlerIntegrationTests.java
View File

@@ -16,6 +16,7 @@ import org.springframework.batch.core.JobParametersBuilder;
import org.springframework.batch.integration.JobSupport;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.integration.bus.MessageBus;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.channel.PollableChannel;
import org.springframework.integration.message.GenericMessage;
@@ -30,6 +31,9 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@RunWith(SpringJUnit4ClassRunner.class)
public class JobLaunchingMessageHandlerIntegrationTests {
@Autowired
private MessageBus bus;
@Autowired
@Qualifier("requests")
private MessageChannel requestChannel;
@@ -43,6 +47,7 @@ public class JobLaunchingMessageHandlerIntegrationTests {
@Before
public void setUp() {
responseChannel.purge(null);
bus.start();
}
@Test

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

@@ -28,7 +28,6 @@ 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.endpoint.SourcePoller;
import org.springframework.integration.message.GenericMessage;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageConsumer;
@@ -41,6 +40,7 @@ import org.springframework.transaction.interceptor.MatchAlwaysTransactionAttribu
import org.springframework.transaction.interceptor.TransactionInterceptor;
import org.springframework.util.StringUtils;
public class PollableSourceRetryTests {
private Log logger = LogFactory.getLog(getClass());
@@ -66,333 +66,241 @@ public class PollableSourceRetryTests {
private PlatformTransactionManager transactionManager = new ResourcelessTransactionManager();;
@Test
public void testSimpleTransactionalPolling() throws Exception {
List<String> list = TransactionAwareProxyFactory.createTransactionalList();
list.addAll(Arrays.asList(StringUtils.commaDelimitedListToStringArray("a,b,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);
}
};
MessageSource<Object> source = getPollableSource(list);
MessageChannel target = getChannel(handler);
SourcePoller trigger = getSourcePoller(source, target, transactionManager, 1);
TaskScheduler scheduler = getSchedulerWithErrorHandler(trigger);
waitForResults(scheduler, 2, 40);
assertEquals(2, processed.size());
assertEquals(beforeCount - list.size(), processed.size());
assertEquals("a", processed.get(0));
}
@Test
public void testNonTransactionalPollingWithRollback() throws Exception {
List<String> list = TransactionAwareProxyFactory.createTransactionalList();
list.addAll(Arrays.asList(StringUtils.commaDelimitedListToStringArray("a,b,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);
throw new RuntimeException("Planned failure: " + payload);
}
};
MessageSource<Object> source = getPollableSource(list);
MessageChannel target = getChannel(handler);
SourcePoller trigger = getSourcePoller(source, target, null, 1);
TaskScheduler scheduler = getSchedulerWithErrorHandler(trigger);
waitForResults(scheduler, 2, 20);
assertEquals(2, processed.size());
// None rolled back because there was no transaction
assertEquals(beforeCount - list.size(), 2);
assertEquals("a", processed.get(0));
assertEquals("b", processed.get(1));
}
@Test
public void testTransactionalHandlingWithUnconditionalRollback() throws Exception {
List<String> list = TransactionAwareProxyFactory.createTransactionalList();
list.addAll(Arrays.asList(StringUtils.commaDelimitedListToStringArray("a,b,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);
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, 2, 40);
assertEquals(2, processed.size());
// TODO: this would fail if exception not propagated: INT-184.
// All rolled back
assertEquals(beforeCount - list.size(), 0);
assertEquals("a", processed.get(0));
// processed twice and rolled back both times
assertEquals("a", processed.get(1));
}
@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;
}
// @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

98
spring-batch-integration/src/test/java/org/springframework/batch/integration/retry/TransactionalPollingIntegrationTests.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 TransactionalPollingIntegrationTests {
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,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,count);
}
@Test
@DirtiesContext
public void testRollback() throws Exception {
// when @Poller accepts transactional=@Transactional(propagation=Propagation.REQUIRED)...
list = TransactionAwareProxyFactory.createTransactionalList(Arrays.asList(StringUtils
.commaDelimitedListToStringArray("a,b,fail,d,e,f,g,h,j,k")));
waitForResults(bus, 4, 30);
System.err.println(processed);
assertEquals(2,count);
}
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();
}
}

1
spring-batch-integration/src/test/resources/log4j.properties
View File

@@ -4,7 +4,6 @@ log4j.appender.stdout=org.apache.log4j.ConsoleAppender
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=%d %5p %t [%c] - <%m>%n
log4j.logger.org.springframework.integration.batch=DEBUG
log4j.logger.org.springframework.batch=DEBUG
log4j.category.org.springframework.integration=DEBUG
log4j.category.org.springframework.transaction=DEBUG

3
spring-batch-integration/src/test/resources/integration-context.xml → spring-batch-integration/src/test/resources/org/springframework/batch/integration/SmokeTests-context.xml
View File

@@ -10,8 +10,7 @@
http://www.springframework.org/schema/context/spring-context-2.5.xsd
http://www.springframework.org/schema/tx
http://www.springframework.org/schema/tx/spring-tx-2.5.xsd">
<message-bus/>
<annotation-driven />
<message-bus auto-startup="false" enable-annotations="true"/>
<channel id="smokein"/>
<channel id="smokeout">
<queue capacity="UNBOUNDED"/>

3
spring-batch-integration/src/test/resources/org/springframework/batch/integration/chunk/ChunkMessageItemWriterIntegrationTests-context.xml
View File

@@ -10,8 +10,7 @@
http://www.springframework.org/schema/context/spring-context-2.5.xsd
http://www.springframework.org/schema/tx
http://www.springframework.org/schema/tx/spring-tx-2.5.xsd">
<message-bus />
<annotation-driven />
<message-bus auto-startup="false" enable-annotations="true"/>
<channel id="requests" />
<channel id="replies">
<queue capacity="UNBOUNDED" />

3
spring-batch-integration/src/test/resources/org/springframework/batch/integration/file/ResourceSplitterIntegrationTests-context.xml
View File

@@ -11,8 +11,7 @@
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">
<integration:message-bus/>
<integration:annotation-driven/>
<integration:message-bus auto-startup="false" enable-annotations="true"/>
<integration:channel id="resources" />
<integration:channel id="requests">
<integration:queue capacity="UNBOUNDED"/>

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

@@ -11,7 +11,7 @@
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">
<integration:message-bus/>
<integration:message-bus auto-startup="false"/>
<integration:channel id="requests">
<integration:queue capacity="UNBOUNDED"/>
</integration:channel>

3
spring-batch-integration/src/test/resources/org/springframework/batch/integration/job/MessageOrientedStepIntegrationTests-context.xml
View File

@@ -10,8 +10,7 @@
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:annotation-driven />
<integration:message-bus />
<integration:message-bus auto-startup="false" enable-annotations="true"/>
<integration:channel id="requests" />
<integration:channel id="replies">
<integration:queue capacity="UNBOUNDED" />

3
spring-batch-integration/src/test/resources/org/springframework/batch/integration/launch/JobLaunchingMessageHandlerIntegrationTests-context.xml
View File

@@ -10,8 +10,7 @@
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 />
<integration:annotation-driven />
<integration:message-bus auto-startup="false" enable-annotations="true"/>
<integration:channel id="requests" />
<integration:channel id="response">
<integration:queue capacity="UNBOUNDED" />

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

@@ -0,0 +1,16 @@
<?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="classpath:/simple-job-launcher-context.xml" />
<integration:message-bus auto-startup="false" enable-annotations="true"/>
<integration:channel id="requests"/>
<integration:channel id="replies" />
</beans>

2
spring-batch-integration/src/test/resources/simple-job-launcher-context.xml
View File

@@ -41,7 +41,7 @@
<property name="commitInterval" value="1" />
</bean>
<bean id="skipLimitStep"
class="org.springframework.batch.core.step.item.SkipLimitStepFactoryBean"
class="org.springframework.batch.core.step.item.FaultTolerantStepFactoryBean"
parent="simpleStep" abstract="true">
<property name="skipLimit" value="0" />
</bean>