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

IN PROGRESS - BATCH-710: updated tests to use SpringJUnit4ClassRunner and SimpleJdbcTemplate

Browse Source
This commit is contained in:
trisberg
2008-07-31 18:01:17 +00:00
parent c5de58c898
commit e9ea3dd42c
9 changed files with 346 additions and 687 deletions
Download Patch File Download Diff File Expand all files Collapse all files

33
spring-batch-infrastructure-tests/src/test/java/org/springframework/batch/config/DatasourceTests.java
View File

@@ -16,23 +16,36 @@
package org.springframework.batch.config;
import org.springframework.batch.jms.ExternalRetryInBatchTests;
import org.springframework.test.AbstractTransactionalDataSourceSpringContextTests;
import org.springframework.util.ClassUtils;
import static org.junit.Assert.*;
public class DatasourceTests extends AbstractTransactionalDataSourceSpringContextTests {
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jdbc.core.simple.SimpleJdbcTemplate;
import org.springframework.transaction.annotation.Transactional;
import org.junit.runner.RunWith;
import org.junit.Test;
protected String[] getConfigLocations() {
return new String[] { ClassUtils.addResourcePathToPackagePath(ExternalRetryInBatchTests.class, "jms-context.xml") };
import javax.sql.DataSource;
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations = "/org/springframework/batch/jms/jms-context.xml")
public class DatasourceTests {
private SimpleJdbcTemplate simpleJdbcTemplate;
@Autowired
public void setDataSource(DataSource dataSource) {
this.simpleJdbcTemplate = new SimpleJdbcTemplate(dataSource);
}
@Transactional @Test
public void testTemplate() throws Exception {
System.err.println(System.getProperty("java.class.path"));
jdbcTemplate.execute("delete from T_FOOS");
int count = jdbcTemplate.queryForInt("select count(*) from T_FOOS");
simpleJdbcTemplate.getJdbcOperations().execute("delete from T_FOOS");
int count = simpleJdbcTemplate.queryForInt("select count(*) from T_FOOS");
assertEquals(0, count);
jdbcTemplate.update("INSERT into T_FOOS (id,name,foo_date) values (?,?,null)", new Object[] { new Integer(0),
"foo" });
simpleJdbcTemplate.update("INSERT into T_FOOS (id,name,foo_date) values (?,?,null)", 0, "foo");
}
}

30
spring-batch-infrastructure-tests/src/test/java/org/springframework/batch/config/MessagingTests.java
View File

@@ -16,35 +16,35 @@
package org.springframework.batch.config;
import static org.junit.Assert.*;
import java.util.ArrayList;
import java.util.List;
import org.springframework.batch.jms.ExternalRetryInBatchTests;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.test.AbstractDependencyInjectionSpringContextTests;
import org.springframework.util.ClassUtils;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.beans.factory.annotation.Autowired;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
public class MessagingTests extends AbstractDependencyInjectionSpringContextTests {
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations = "/org/springframework/batch/jms/jms-context.xml")
public class MessagingTests {
@Autowired
private JmsTemplate jmsTemplate;
public void setJmsTemplate(JmsTemplate jmsTemplate) {
this.jmsTemplate = jmsTemplate;
}
protected String[] getConfigLocations() {
return new String[] { ClassUtils.addResourcePathToPackagePath(ExternalRetryInBatchTests.class,
"jms-context.xml") };
}
protected void onSetUp() throws Exception {
super.onSetUp();
@Before
public void onSetUp() throws Exception {
Thread.sleep(100L);
getMessages(); // drain queue
jmsTemplate.convertAndSend("queue", "foo");
jmsTemplate.convertAndSend("queue", "bar");
}
@Test
public void testMessaging() throws Exception {
List<String> list = getMessages();
System.err.println(list);

65
spring-batch-infrastructure-tests/src/test/java/org/springframework/batch/container/jms/BatchMessageListenerContainerIntegrationTests.java
View File

@@ -15,10 +15,11 @@
*/
package org.springframework.batch.container.jms;
import static org.junit.Assert.*;
import javax.jms.Message;
import javax.jms.MessageListener;
import org.springframework.batch.jms.ExternalRetryInBatchTests;
import org.springframework.batch.retry.RecoveryCallback;
import org.springframework.batch.retry.RetryCallback;
import org.springframework.batch.retry.RetryContext;
@@ -27,72 +28,53 @@ import org.springframework.batch.retry.policy.NeverRetryPolicy;
import org.springframework.batch.retry.policy.RecoveryCallbackRetryPolicy;
import org.springframework.batch.retry.support.RetryTemplate;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.test.AbstractDependencyInjectionSpringContextTests;
import org.springframework.util.ClassUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.test.context.ContextConfiguration;
import org.junit.Before;
import org.junit.After;
import org.junit.Test;
import org.junit.runner.RunWith;
/**
* @author Dave Syer
*
*/
public class BatchMessageListenerContainerIntegrationTests extends AbstractDependencyInjectionSpringContextTests {
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations = "/org/springframework/batch/jms/jms-context.xml")
public class BatchMessageListenerContainerIntegrationTests {
@Autowired
private JmsTemplate jmsTemplate;
@Autowired
private BatchMessageListenerContainer container;
private int recovered;
private volatile int count;
/**
* Public setter for the {@link BatchMessageListenerContainer}.
* @param container the container to set
*/
public void setContainer(BatchMessageListenerContainer container) {
this.container = container;
}
/**
* Public setter for the JmsTemplate.
* @param jmsTemplate the jmsTemplate to set
*/
public void setJmsTemplate(JmsTemplate jmsTemplate) {
this.jmsTemplate = jmsTemplate;
}
/*
* (non-Javadoc)
* @see org.springframework.test.AbstractSingleSpringContextTests#getConfigLocations()
*/
protected String[] getConfigLocations() {
// Share config with other test so that ActiveMQ only starts up once.
return new String[] { ClassUtils.addResourcePathToPackagePath(ExternalRetryInBatchTests.class, "jms-context.xml") };
}
/* (non-Javadoc)
* @see org.springframework.test.AbstractSingleSpringContextTests#onSetUp()
*/
protected void onSetUp() throws Exception {
@Before
public void onSetUp() throws Exception {
while(jmsTemplate.receiveAndConvert("queue")!=null) {
// do nothing
}
}
/*
* (non-Javadoc)
* @see org.springframework.test.AbstractSingleSpringContextTests#onTearDown()
*/
protected void onTearDown() throws Exception {
@After
public void onTearDown() throws Exception {
container.stop();
while(jmsTemplate.receiveAndConvert("queue")!=null) {
// do nothing
}
}
@Test
public void testConfiguration() throws Exception {
assertNotNull(container);
}
@Test
public void testSendAndReceive() throws Exception {
container.setMessageListener(new MessageListener() {
public void onMessage(Message msg) {
@@ -112,10 +94,10 @@ public class BatchMessageListenerContainerIntegrationTests extends AbstractDepen
}
}
@Test
public void testFailureAndRepresent() throws Exception {
container.setMessageListener(new MessageListener() {
public void onMessage(Message msg) {
logger.debug("Message: "+msg);
count++;
throw new RuntimeException("planned failure for represent: " + msg);
}
@@ -128,11 +110,11 @@ public class BatchMessageListenerContainerIntegrationTests extends AbstractDepen
Thread.sleep(100L);
}
if (count < 2) {
logger.debug("Count: "+count);
fail("Expected message to be processed twice.");
}
}
@Test
public void testFailureAndRecovery() throws Exception {
final RetryTemplate retryTemplate = new RetryTemplate();
retryTemplate.setRetryPolicy(new RecoveryCallbackRetryPolicy(new NeverRetryPolicy()));
@@ -141,7 +123,6 @@ public class BatchMessageListenerContainerIntegrationTests extends AbstractDepen
try {
RecoveryRetryCallback callback = new RecoveryRetryCallback(msg, new RetryCallback() {
public Object doWithRetry(RetryContext context) throws Throwable {
logger.debug("Message: "+msg);
count++;
throw new RuntimeException("planned failure: " + msg);
}
@@ -149,7 +130,6 @@ public class BatchMessageListenerContainerIntegrationTests extends AbstractDepen
callback.setRecoveryCallback(new RecoveryCallback() {
public Object recover(RetryContext context) {
recovered++;
logger.debug("Recovered: " + msg);
return msg;
}
});
@@ -170,4 +150,5 @@ public class BatchMessageListenerContainerIntegrationTests extends AbstractDepen
assertEquals(1, count);
assertEquals(1, recovered);
}
}

59
spring-batch-infrastructure-tests/src/test/java/org/springframework/batch/jms/ExternalRetryInBatchTests.java
View File

@@ -16,6 +16,12 @@
package org.springframework.batch.jms;
import static org.junit.Assert.*;
import org.junit.Before;
import org.junit.After;
import org.junit.Test;
import org.junit.runner.RunWith;
import java.util.ArrayList;
import java.util.List;
@@ -35,52 +41,44 @@ import org.springframework.batch.retry.callback.RecoveryRetryCallback;
import org.springframework.batch.retry.policy.RecoveryCallbackRetryPolicy;
import org.springframework.batch.retry.policy.SimpleRetryPolicy;
import org.springframework.batch.retry.support.RetryTemplate;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.core.simple.SimpleJdbcTemplate;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.test.AbstractDependencyInjectionSpringContextTests;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.support.TransactionCallback;
import org.springframework.transaction.support.TransactionTemplate;
import org.springframework.util.ClassUtils;
import org.springframework.beans.factory.annotation.Autowired;
public class ExternalRetryInBatchTests extends AbstractDependencyInjectionSpringContextTests {
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations = "/org/springframework/batch/jms/jms-context.xml")
public class ExternalRetryInBatchTests {
@Autowired
private JmsTemplate jmsTemplate;
private RetryTemplate retryTemplate;
@Autowired
private RepeatTemplate repeatTemplate;
private ItemReaderRecoverer provider;
private JdbcTemplate jdbcTemplate;
private SimpleJdbcTemplate jdbcTemplate;
@Autowired
private PlatformTransactionManager transactionManager;
@Autowired
public void setDataSource(DataSource dataSource) {
jdbcTemplate = new JdbcTemplate(dataSource);
jdbcTemplate = new SimpleJdbcTemplate(dataSource);
}
public void setTransactionManager(PlatformTransactionManager transactionManager) {
this.transactionManager = transactionManager;
}
public void setRepeatTemplate(RepeatTemplate repeatTemplate) {
this.repeatTemplate = repeatTemplate;
}
public void setJmsTemplate(JmsTemplate jmsTemplate) {
this.jmsTemplate = jmsTemplate;
}
protected String[] getConfigLocations() {
return new String[] { ClassUtils.addResourcePathToPackagePath(getClass(), "jms-context.xml" )};
}
protected void onSetUp() throws Exception {
super.onSetUp();
@Before
public void onSetUp() throws Exception {
getMessages(); // drain queue
jdbcTemplate.execute("delete from T_FOOS");
jdbcTemplate.getJdbcOperations().execute("delete from T_FOOS");
jmsTemplate.convertAndSend("queue", "foo");
jmsTemplate.convertAndSend("queue", "bar");
provider = new ItemReaderRecoverer() {
@@ -98,9 +96,10 @@ public class ExternalRetryInBatchTests extends AbstractDependencyInjectionSpring
retryTemplate = new RetryTemplate();
}
protected void onTearDown() throws Exception {
@After
public void onTearDown() throws Exception {
getMessages(); // drain queue
jdbcTemplate.execute("delete from T_FOOS");
jdbcTemplate.getJdbcOperations().execute("delete from T_FOOS");
}
private void assertInitialState() {
@@ -112,6 +111,7 @@ public class ExternalRetryInBatchTests extends AbstractDependencyInjectionSpring
private List<Object> recovered = new ArrayList<Object>();
@Test
public void testExternalRetryRecoveryInBatch() throws Exception {
assertInitialState();
@@ -142,8 +142,9 @@ public class ExternalRetryInBatchTests extends AbstractDependencyInjectionSpring
// No need for transaction here: the whole batch will roll
// back. When it comes back for recovery this code is not
// executed...
jdbcTemplate.update("INSERT into T_FOOS (id,name,foo_date) values (?,?,null)", new Object[] {
new Integer(list.size()), item });
jdbcTemplate.update(
"INSERT into T_FOOS (id,name,foo_date) values (?,?,null)",
list.size(), item);
throw new RuntimeException("Rollback!");
}
});

65
spring-batch-infrastructure-tests/src/test/java/org/springframework/batch/repeat/jms/AsynchronousTests.java
View File

@@ -16,6 +16,8 @@
package org.springframework.batch.repeat.jms;
import static org.junit.Assert.*;
import java.util.ArrayList;
import java.util.List;
@@ -23,49 +25,53 @@ import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.sql.DataSource;
import org.springframework.batch.container.jms.BatchMessageListenerContainer;
import org.springframework.batch.jms.ExternalRetryInBatchTests;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.core.simple.SimpleJdbcTemplate;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.listener.SessionAwareMessageListener;
import org.springframework.test.AbstractDependencyInjectionSpringContextTests;
import org.springframework.util.ClassUtils;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.beans.factory.annotation.Autowired;
import org.junit.runner.RunWith;
import org.junit.Before;
import org.junit.After;
import org.junit.Test;
public class AsynchronousTests extends AbstractDependencyInjectionSpringContextTests {
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations = "/org/springframework/batch/jms/jms-context.xml")
public class AsynchronousTests {
protected String[] getConfigLocations() {
return new String[] { ClassUtils.addResourcePathToPackagePath(ExternalRetryInBatchTests.class,
"jms-context.xml") };
}
@Autowired
private BatchMessageListenerContainer container;
@Autowired
private JmsTemplate jmsTemplate;
private JdbcTemplate jdbcTemplate;
private SimpleJdbcTemplate simpleJdbcTemplate;
public void setJdbcTemplate(JdbcTemplate jdbcTemplate) {
this.jdbcTemplate = jdbcTemplate;
@Autowired
public void setDataSource(DataSource dataSource) {
this.simpleJdbcTemplate = new SimpleJdbcTemplate(dataSource);
}
public void setJmsTemplate(JmsTemplate jmsTemplate) {
this.jmsTemplate = jmsTemplate;
}
public void setContainer(BatchMessageListenerContainer container) {
this.container = container;
}
protected void onSetUp() throws Exception {
super.onSetUp();
@Before
public void onSetUp() throws Exception {
String foo = "";
int count = 0;
while (foo != null && count < 100) {
foo = (String) jmsTemplate.receiveAndConvert("queue");
count++;
}
jdbcTemplate.execute("delete from T_FOOS");
simpleJdbcTemplate.getJdbcOperations().execute("delete from T_FOOS");
// Queue is now drained...
assertNull(foo);
@@ -76,8 +82,8 @@ public class AsynchronousTests extends AbstractDependencyInjectionSpringContextT
}
protected void onTearDown() throws Exception {
super.onTearDown();
@After
public void onTearDown() throws Exception {
container.stop();
// Need to give the container time to shutdown
Thread.sleep(1000L);
@@ -92,10 +98,11 @@ public class AsynchronousTests extends AbstractDependencyInjectionSpringContextT
List<String> list = new ArrayList<String>();
private void assertInitialState() {
int count = jdbcTemplate.queryForInt("select count(*) from T_FOOS");
int count = simpleJdbcTemplate.queryForInt("select count(*) from T_FOOS");
assertEquals(0, count);
}
@Test
public void testSunnyDay() throws Exception {
assertInitialState();
@@ -104,8 +111,7 @@ public class AsynchronousTests extends AbstractDependencyInjectionSpringContextT
public void onMessage(Message message, Session session) throws JMSException {
list.add(message.toString());
String text = ((TextMessage) message).getText();
jdbcTemplate.update("INSERT into T_FOOS (id,name,foo_date) values (?,?,null)", new Object[] {
new Integer(list.size()), text });
simpleJdbcTemplate.update("INSERT into T_FOOS (id,name,foo_date) values (?,?,null)", list.size(), text);
}
});
@@ -116,18 +122,19 @@ public class AsynchronousTests extends AbstractDependencyInjectionSpringContextT
// Need to sleep for at least a second here...
Thread.sleep(1000L);
System.err.println(jdbcTemplate.queryForList("select * from T_FOOS"));
System.err.println(simpleJdbcTemplate.queryForList("select * from T_FOOS"));
assertEquals(2, list.size());
String foo = (String) jmsTemplate.receiveAndConvert("queue");
assertEquals(null, foo);
int count = jdbcTemplate.queryForInt("select count(*) from T_FOOS");
int count = simpleJdbcTemplate.queryForInt("select count(*) from T_FOOS");
assertEquals(2, count);
}
@Test
public void testRollback() throws Exception {
assertInitialState();
@@ -139,9 +146,7 @@ public class AsynchronousTests extends AbstractDependencyInjectionSpringContextT
public void onMessage(Message message, Session session) throws JMSException {
list.add(message.toString());
final String text = ((TextMessage) message).getText();
logger.debug("Processing message: " + message);
jdbcTemplate.update("INSERT into T_FOOS (id,name,foo_date) values (?,?,null)", new Object[] {
new Integer(list.size()), text });
simpleJdbcTemplate.update("INSERT into T_FOOS (id,name,foo_date) values (?,?,null)", list.size(), text);
// This causes the DB to rollback but not the message
if (text.equals("bar")) {
throw new RuntimeException("Rollback!");
@@ -161,20 +166,18 @@ public class AsynchronousTests extends AbstractDependencyInjectionSpringContextT
// We rolled back so the messages might come in many times...
assertTrue(list.size() >= 1);
logger.debug("T_FOOS: "+jdbcTemplate.queryForList("select * from T_FOOS"));
String text = "";
List<String> msgs = new ArrayList<String>();
while (text != null) {
text = (String) jmsTemplate.receiveAndConvert("queue");
msgs.add(text);
}
logger.debug("Messages: "+msgs);
int count = jdbcTemplate.queryForInt("select count(*) from T_FOOS");
int count = simpleJdbcTemplate.queryForInt("select count(*) from T_FOOS");
assertEquals(0, count);
assertTrue("Foo not on queue", msgs.contains("foo"));
}
}

151
spring-batch-infrastructure-tests/src/test/java/org/springframework/batch/repeat/jms/SynchronousTests.java
View File

@@ -16,14 +16,16 @@
package org.springframework.batch.repeat.jms;
import static org.junit.Assert.*;
import java.util.ArrayList;
import java.util.List;
import javax.jms.ConnectionFactory;
import javax.jms.JMSException;
import javax.jms.Session;
import javax.sql.DataSource;
import org.springframework.batch.jms.ExternalRetryInBatchTests;
import org.springframework.batch.repeat.ExitStatus;
import org.springframework.batch.repeat.RepeatCallback;
import org.springframework.batch.repeat.RepeatContext;
@@ -31,54 +33,69 @@ import org.springframework.batch.repeat.support.RepeatTemplate;
import org.springframework.jms.connection.SessionProxy;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.SessionCallback;
import org.springframework.test.AbstractTransactionalDataSourceSpringContextTests;
import org.springframework.transaction.support.TransactionSynchronizationAdapter;
import org.springframework.transaction.support.TransactionSynchronizationManager;
import org.springframework.util.ClassUtils;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.transaction.BeforeTransaction;
import org.springframework.transaction.support.TransactionCallback;
import org.springframework.transaction.support.TransactionTemplate;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.BeansException;
import org.springframework.jdbc.core.simple.SimpleJdbcTemplate;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.ApplicationContext;
import org.junit.runner.RunWith;
import org.junit.Test;
public class SynchronousTests extends AbstractTransactionalDataSourceSpringContextTests {
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations = "/org/springframework/batch/jms/jms-context.xml")
public class SynchronousTests implements ApplicationContextAware {
@Autowired
private JmsTemplate jmsTemplate;
@Autowired
private RepeatTemplate repeatTemplate;
public void setJmsTemplate(JmsTemplate jmsTemplate) {
this.jmsTemplate = jmsTemplate;
@Autowired
private PlatformTransactionManager transactionManager;
private SimpleJdbcTemplate simpleJdbcTemplate;
private ApplicationContext applicationContext;
public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
this.applicationContext = applicationContext;
}
public void setRepeatTemplate(RepeatTemplate repeatTemplate) {
this.repeatTemplate = repeatTemplate;
@Autowired
public void setDataSource(DataSource dataSource) {
this.simpleJdbcTemplate = new SimpleJdbcTemplate(dataSource);
}
protected String[] getConfigLocations() {
return new String[] { ClassUtils.addResourcePathToPackagePath(ExternalRetryInBatchTests.class,
"jms-context.xml") };
}
protected void onSetUpBeforeTransaction() throws Exception {
super.onSetUpBeforeTransaction();
@BeforeTransaction
public void onSetUpBeforeTransaction() throws Exception {
String foo = "";
int count = 0;
while (foo != null && count < 100) {
foo = (String) jmsTemplate.receiveAndConvert("queue");
count++;
}
jdbcTemplate.execute("delete from T_FOOS");
simpleJdbcTemplate.getJdbcOperations().execute("delete from T_FOOS");
jmsTemplate.convertAndSend("queue", "foo");
jmsTemplate.convertAndSend("queue", "bar");
}
protected void onSetUpInTransaction() throws Exception {
super.onSetUpInTransaction();
}
private void assertInitialState() {
int count = jdbcTemplate.queryForInt("select count(*) from T_FOOS");
int count = simpleJdbcTemplate.queryForInt("select count(*) from T_FOOS");
assertEquals(0, count);
}
List<String> list = new ArrayList<String>();
@Transactional
@Test
public void testCommit() throws Exception {
assertInitialState();
@@ -87,47 +104,40 @@ public class SynchronousTests extends AbstractTransactionalDataSourceSpringConte
public ExitStatus doInIteration(RepeatContext context) throws Exception {
String text = (String) jmsTemplate.receiveAndConvert("queue");
list.add(text);
jdbcTemplate.update("INSERT into T_FOOS (id,name,foo_date) values (?,?,null)", new Object[] {
new Integer(list.size()), text });
simpleJdbcTemplate.update("INSERT into T_FOOS (id,name,foo_date) values (?,?,null)", list.size(), text);
return new ExitStatus(text != null);
}
});
// force commit...
setComplete();
endTransaction();
startNewTransaction();
System.err.println(jdbcTemplate.queryForList("select * from T_FOOS"));
// Database committed so this record should be there...
int count = jdbcTemplate.queryForInt("select count(*) from T_FOOS");
int count = simpleJdbcTemplate.queryForInt("select count(*) from T_FOOS");
assertEquals(2, count);
// ... the commit should also have cleared the queue, so this should now
// be null
String text = (String) jmsTemplate.receiveAndConvert("queue");
assertEquals(null, text);
}
@Test
public void testFullRollback() throws Exception {
assertInitialState();
repeatTemplate.iterate(new RepeatCallback() {
public ExitStatus doInIteration(RepeatContext context) throws Exception {
String text = (String) jmsTemplate.receiveAndConvert("queue");
list.add(text);
jdbcTemplate.update("INSERT into T_FOOS (id,name,foo_date) values (?,?,null)", new Object[] {
new Integer(list.size()), text });
return new ExitStatus(text != null);
new TransactionTemplate(transactionManager).execute(new TransactionCallback() {
public Object doInTransaction(org.springframework.transaction.TransactionStatus status) {
repeatTemplate.iterate(new RepeatCallback() {
public ExitStatus doInIteration(RepeatContext context) throws Exception {
String text = (String) jmsTemplate.receiveAndConvert("queue");
list.add(text);
simpleJdbcTemplate.update("INSERT into T_FOOS (id,name,foo_date) values (?,?,null)", list.size(), text);
return new ExitStatus(text != null);
}
});
// force rollback...
status.setRollbackOnly();
return null;
}
});
// force rollback...
endTransaction();
startNewTransaction();
String text = "";
List<String> msgs = new ArrayList<String>();
while (text != null) {
@@ -136,7 +146,7 @@ public class SynchronousTests extends AbstractTransactionalDataSourceSpringConte
}
// The database portion rolled back...
int count = jdbcTemplate.queryForInt("select count(*) from T_FOOS");
int count = simpleJdbcTemplate.queryForInt("select count(*) from T_FOOS");
assertEquals(0, count);
// ... and so did the message session. The rollback should have restored
@@ -144,31 +154,33 @@ public class SynchronousTests extends AbstractTransactionalDataSourceSpringConte
assertTrue("Foo not on queue", msgs.contains("foo"));
}
@Transactional @Test
public void testPartialRollback() throws Exception {
// The JmsTemplate is used elsewhere outside a transaction, so
// we need to use one here that is transaction aware.
final JmsTemplate jmsTemplate = new JmsTemplate((ConnectionFactory) applicationContext
.getBean("txAwareConnectionFactory"));
jmsTemplate.setReceiveTimeout(100L);
jmsTemplate.setSessionTransacted(true);
final JmsTemplate txJmsTemplate = new JmsTemplate(
(ConnectionFactory) applicationContext.getBean("txAwareConnectionFactory"));
txJmsTemplate.setReceiveTimeout(100L);
txJmsTemplate.setSessionTransacted(true);
assertInitialState();
repeatTemplate.iterate(new RepeatCallback() {
public ExitStatus doInIteration(RepeatContext context) throws Exception {
String text = (String) jmsTemplate.receiveAndConvert("queue");
list.add(text);
jdbcTemplate.update("INSERT into T_FOOS (id,name,foo_date) values (?,?,null)", new Object[] {
new Integer(list.size()), text });
return new ExitStatus(text != null);
}
});
TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronizationAdapter() {
public void beforeCommit(boolean readOnly) {
new TransactionTemplate(transactionManager).execute(new TransactionCallback() {
public Object doInTransaction(org.springframework.transaction.TransactionStatus status) {
repeatTemplate.iterate(new RepeatCallback() {
public ExitStatus doInIteration(RepeatContext context) throws Exception {
String text = (String) txJmsTemplate.receiveAndConvert("queue");
list.add(text);
simpleJdbcTemplate.update("INSERT into T_FOOS (id,name,foo_date) values (?,?,null)", list.size(), text);
return new ExitStatus(text != null);
}
});
// Simulate a message system failure before the main transaction
// commits...
jmsTemplate.execute(new SessionCallback() {
txJmsTemplate.execute(new SessionCallback() {
public Object doInJms(Session session) throws JMSException {
try {
assertTrue("Not a SessionProxy - wrong spring version?", session instanceof SessionProxy);
@@ -184,22 +196,20 @@ public class SynchronousTests extends AbstractTransactionalDataSourceSpringConte
return null;
}
});
return null;
}
});
// force commit...
setComplete();
endTransaction();
startNewTransaction();
String text = "";
List<String> msgs = new ArrayList<String>();
while (text != null) {
text = (String) jmsTemplate.receiveAndConvert("queue");
text = (String) txJmsTemplate.receiveAndConvert("queue");
msgs.add(text);
}
// The database portion committed...
int count = jdbcTemplate.queryForInt("select count(*) from T_FOOS");
int count = simpleJdbcTemplate.queryForInt("select count(*) from T_FOOS");
assertEquals(2, count);
// ...but the JMS session rolled back, so the message is still there
@@ -207,4 +217,5 @@ public class SynchronousTests extends AbstractTransactionalDataSourceSpringConte
assertTrue("Bar not on queue", msgs.contains("bar"));
}
}

65
spring-batch-infrastructure-tests/src/test/java/org/springframework/batch/retry/jms/ExternalRetryTests.java
View File

@@ -16,6 +16,8 @@
package org.springframework.batch.retry.jms;
import static org.junit.Assert.*;
import java.util.ArrayList;
import java.util.List;
@@ -24,55 +26,50 @@ import javax.sql.DataSource;
import org.springframework.batch.item.ItemRecoverer;
import org.springframework.batch.item.support.AbstractItemReader;
import org.springframework.batch.item.support.AbstractItemWriter;
import org.springframework.batch.jms.ExternalRetryInBatchTests;
import org.springframework.batch.retry.RecoveryCallback;
import org.springframework.batch.retry.RetryCallback;
import org.springframework.batch.retry.RetryContext;
import org.springframework.batch.retry.callback.RecoveryRetryCallback;
import org.springframework.batch.retry.policy.RecoveryCallbackRetryPolicy;
import org.springframework.batch.retry.support.RetryTemplate;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.core.simple.SimpleJdbcTemplate;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.test.AbstractDependencyInjectionSpringContextTests;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.support.TransactionCallback;
import org.springframework.transaction.support.TransactionTemplate;
import org.springframework.util.ClassUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.test.context.ContextConfiguration;
import org.junit.runner.RunWith;
import org.junit.Before;
import org.junit.Test;
public class ExternalRetryTests extends AbstractDependencyInjectionSpringContextTests {
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations = "/org/springframework/batch/jms/jms-context.xml")
public class ExternalRetryTests {
@Autowired
private JmsTemplate jmsTemplate;
private RetryTemplate retryTemplate;
private ItemReaderRecoverer<String> provider;
private JdbcTemplate jdbcTemplate;
private SimpleJdbcTemplate simpleJdbcTemplate;
@Autowired
private PlatformTransactionManager transactionManager;
@Autowired
public void setDataSource(DataSource dataSource) {
jdbcTemplate = new JdbcTemplate(dataSource);
simpleJdbcTemplate = new SimpleJdbcTemplate(dataSource);
}
public void setTransactionManager(PlatformTransactionManager transactionManager) {
this.transactionManager = transactionManager;
}
public void setJmsTemplate(JmsTemplate jmsTemplate) {
this.jmsTemplate = jmsTemplate;
}
protected String[] getConfigLocations() {
return new String[] { ClassUtils.addResourcePathToPackagePath(ExternalRetryInBatchTests.class,
"jms-context.xml") };
}
protected void onSetUp() throws Exception {
super.onSetUp();
@Before
public void onSetUp() throws Exception {
getMessages(); // drain queue
jdbcTemplate.execute("delete from T_FOOS");
simpleJdbcTemplate.getJdbcOperations().execute("delete from T_FOOS");
jmsTemplate.convertAndSend("queue", "foo");
provider = new ItemReaderRecoverer<String>() {
public String read() {
@@ -90,7 +87,7 @@ public class ExternalRetryTests extends AbstractDependencyInjectionSpringContext
}
private void assertInitialState() {
int count = jdbcTemplate.queryForInt("select count(*) from T_FOOS");
int count = simpleJdbcTemplate.queryForInt("select count(*) from T_FOOS");
assertEquals(0, count);
}
@@ -98,12 +95,11 @@ public class ExternalRetryTests extends AbstractDependencyInjectionSpringContext
private List<Object> recovered = new ArrayList<Object>();
/**
/*
* Message processing is successful on the second attempt but must receive
* the message again.
*
* @throws Exception
*/
@Test
public void testExternalRetrySuccessOnSecondAttempt() throws Exception {
assertInitialState();
@@ -112,8 +108,7 @@ public class ExternalRetryTests extends AbstractDependencyInjectionSpringContext
final AbstractItemWriter<Object> writer = new AbstractItemWriter<Object>() {
public void write(final Object text) {
jdbcTemplate.update("INSERT into T_FOOS (id,name,foo_date) values (?,?,null)", new Object[] {
new Integer(list.size()), text });
simpleJdbcTemplate.update("INSERT into T_FOOS (id,name,foo_date) values (?,?,null)", list.size(), text);
if (list.size() == 1) {
throw new RuntimeException("Rollback!");
}
@@ -171,18 +166,17 @@ public class ExternalRetryTests extends AbstractDependencyInjectionSpringContext
List<String> msgs = getMessages();
// The database portion committed once...
int count = jdbcTemplate.queryForInt("select count(*) from T_FOOS");
int count = simpleJdbcTemplate.queryForInt("select count(*) from T_FOOS");
assertEquals(1, count);
// ... and so did the message session.
assertEquals("[]", msgs.toString());
}
/**
/*
* Message processing fails on both attempts.
*
* @throws Exception
*/
@Test
public void testExternalRetryWithRecovery() throws Exception {
assertInitialState();
@@ -192,8 +186,7 @@ public class ExternalRetryTests extends AbstractDependencyInjectionSpringContext
final Object item = provider.read();
final RecoveryRetryCallback callback = new RecoveryRetryCallback(item, new RetryCallback() {
public Object doWithRetry(RetryContext context) throws Throwable {
jdbcTemplate.update("INSERT into T_FOOS (id,name,foo_date) values (?,?,null)", new Object[] {
new Integer(list.size()), item });
simpleJdbcTemplate.update("INSERT into T_FOOS (id,name,foo_date) values (?,?,null)", list.size(), item);
throw new RuntimeException("Rollback!");
}
});
@@ -239,7 +232,7 @@ public class ExternalRetryTests extends AbstractDependencyInjectionSpringContext
assertEquals(1, recovered.size());
// The database portion committed once...
int count = jdbcTemplate.queryForInt("select count(*) from T_FOOS");
int count = simpleJdbcTemplate.queryForInt("select count(*) from T_FOOS");
assertEquals(0, count);
// ... and so did the message session.

200
spring-batch-infrastructure-tests/src/test/java/org/springframework/batch/retry/jms/SynchronousTests.java
View File

@@ -16,6 +16,8 @@
package org.springframework.batch.retry.jms;
import static org.junit.Assert.*;
import java.util.ArrayList;
import java.util.List;
@@ -26,20 +28,41 @@ import org.springframework.batch.retry.RetryContext;
import org.springframework.batch.retry.callback.RecoveryRetryCallback;
import org.springframework.batch.retry.support.RetryTemplate;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.test.AbstractTransactionalDataSourceSpringContextTests;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.transaction.BeforeTransaction;
import org.springframework.test.context.transaction.AfterTransaction;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.transaction.support.TransactionCallback;
import org.springframework.transaction.support.TransactionTemplate;
import org.springframework.util.ClassUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jdbc.core.simple.SimpleJdbcTemplate;
import org.junit.runner.RunWith;
import org.junit.Before;
import org.junit.Test;
public class SynchronousTests extends AbstractTransactionalDataSourceSpringContextTests {
import javax.sql.DataSource;
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations = "/org/springframework/batch/jms/jms-context.xml")
public class SynchronousTests {
@Autowired
private JmsTemplate jmsTemplate;
@Autowired
private PlatformTransactionManager transactionManager;
private RetryTemplate retryTemplate;
public void setJmsTemplate(JmsTemplate jmsTemplate) {
this.jmsTemplate = jmsTemplate;
private SimpleJdbcTemplate simpleJdbcTemplate;
@Autowired
public void setDataSource(DataSource dataSource) {
this.simpleJdbcTemplate = new SimpleJdbcTemplate(dataSource);
}
protected String[] getConfigLocations() {
@@ -47,39 +70,43 @@ public class SynchronousTests extends AbstractTransactionalDataSourceSpringConte
"jms-context.xml") };
}
protected void onSetUpBeforeTransaction() throws Exception {
super.onSetUpBeforeTransaction();
@BeforeTransaction
public void onSetUpBeforeTransaction() throws Exception {
simpleJdbcTemplate.getJdbcOperations().execute("delete from T_FOOS");
jmsTemplate.convertAndSend("queue", "foo");
jmsTemplate.convertAndSend("queue", "foo");
final String text = (String) jmsTemplate.receiveAndConvert("queue");
assertNotNull(text);
}
@Before
public void onSetUpInTransaction() throws Exception {
retryTemplate = new RetryTemplate();
}
@AfterTransaction
public void afterTransaction() {
String foo = "";
int count = 0;
while (foo != null && count < 100) {
foo = (String) jmsTemplate.receiveAndConvert("queue");
count++;
}
jdbcTemplate.execute("delete from T_FOOS");
jmsTemplate.convertAndSend("queue", "foo");
jmsTemplate.convertAndSend("queue", "foo");
final String text = (String) jmsTemplate.receiveAndConvert("queue");
assertNotNull(text);
retryTemplate = new RetryTemplate();
}
protected void onSetUpInTransaction() throws Exception {
super.onSetUpInTransaction();
simpleJdbcTemplate.getJdbcOperations().execute("delete from T_FOOS");
}
private void assertInitialState() {
int count = jdbcTemplate.queryForInt("select count(*) from T_FOOS");
int count = simpleJdbcTemplate.queryForInt("select count(*) from T_FOOS");
assertEquals(0, count);
}
List<Object> list = new ArrayList<Object>();
/**
/*
* Message processing is successful on the second attempt without having to
* receive the message again.
*
* @throws Exception
*/
@Transactional @Test
public void testInternalRetrySuccessOnSecondAttempt() throws Exception {
assertInitialState();
@@ -105,8 +132,7 @@ public class SynchronousTests extends AbstractTransactionalDataSourceSpringConte
list.add(text);
System.err.println("Inserting: [" + list.size() + "," + text + "]");
jdbcTemplate.update("INSERT into T_FOOS (id,name,foo_date) values (?,?,null)", new Object[] {
new Integer(list.size()), text });
simpleJdbcTemplate.update("INSERT into T_FOOS (id,name,foo_date) values (?,?,null)", list.size(), text);
if (list.size() == 1) {
throw new RuntimeException("Rollback!");
}
@@ -118,28 +144,23 @@ public class SynchronousTests extends AbstractTransactionalDataSourceSpringConte
}
});
// force commit...
setComplete();
endTransaction();
startNewTransaction();
// Verify the state after stransactional processing is complete
List<String> msgs = getMessages();
// The database portion committed once...
int count = jdbcTemplate.queryForInt("select count(*) from T_FOOS");
int count = simpleJdbcTemplate.queryForInt("select count(*) from T_FOOS");
assertEquals(1, count);
// ... and so did the message session.
assertEquals("[]", msgs.toString());
}
/**
/*
* Message processing is successful on the second attempt without having to
* receive the message again - uses JmsItemProvider internally.
*
* @throws Exception
*/
@Transactional @Test
public void testInternalRetrySuccessOnSecondAttemptWithItemProvider() throws Exception {
assertInitialState();
@@ -150,6 +171,7 @@ public class SynchronousTests extends AbstractTransactionalDataSourceSpringConte
jmsTemplate.setDefaultDestinationName("queue");
final Object item = provider.read();
retryTemplate.execute(new RecoveryRetryCallback(item, new RetryCallback() {
public Object doWithRetry(RetryContext context) throws Throwable {
@@ -160,8 +182,7 @@ public class SynchronousTests extends AbstractTransactionalDataSourceSpringConte
list.add(item);
System.err.println("Inserting: [" + list.size() + "," + item + "]");
jdbcTemplate.update("INSERT into T_FOOS (id,name,foo_date) values (?,?,null)", new Object[] {
new Integer(list.size()), item });
simpleJdbcTemplate.update("INSERT into T_FOOS (id,name,foo_date) values (?,?,null)", list.size(), item);
if (list.size() == 1) {
throw new RuntimeException("Rollback!");
}
@@ -174,28 +195,23 @@ public class SynchronousTests extends AbstractTransactionalDataSourceSpringConte
}
}));
// force commit...
setComplete();
endTransaction();
startNewTransaction();
// Verify the state after stransactional processing is complete
List<String> msgs = getMessages();
// The database portion committed once...
int count = jdbcTemplate.queryForInt("select count(*) from T_FOOS");
int count = simpleJdbcTemplate.queryForInt("select count(*) from T_FOOS");
assertEquals(1, count);
// ... and so did the message session.
assertEquals("[]", msgs.toString());
}
/**
/*
* Message processing is successful on the second attempt without having to
* receive the message again.
*
* @throws Exception
*/
@Transactional @Test
public void testInternalRetrySuccessOnFirstAttemptRollbackOuter() throws Exception {
assertInitialState();
@@ -208,66 +224,76 @@ public class SynchronousTests extends AbstractTransactionalDataSourceSpringConte
* PROPAGATION_REQUIRES_NEW is wrong because it doesn't allow the outer
* transaction to fail and rollback the inner one.
*/
final String text = (String) jmsTemplate.receiveAndConvert("queue");
retryTemplate.execute(new RetryCallback() {
public Object doWithRetry(RetryContext status) throws Throwable {
TransactionTemplate outerTxTemplate = new TransactionTemplate(transactionManager);
outerTxTemplate.setPropagationBehavior(TransactionTemplate.PROPAGATION_REQUIRES_NEW);
outerTxTemplate.execute(new TransactionCallback() {
public Object doInTransaction(TransactionStatus outerStatus) {
TransactionTemplate transactionTemplate = new TransactionTemplate(transactionManager);
transactionTemplate.setPropagationBehavior(TransactionTemplate.PROPAGATION_NESTED);
return transactionTemplate.execute(new TransactionCallback() {
public Object doInTransaction(TransactionStatus status) {
final String text = (String) jmsTemplate.receiveAndConvert("queue");
list.add(text);
System.err.println("Inserting: [" + list.size() + "," + text + "]");
jdbcTemplate.update("INSERT into T_FOOS (id,name,foo_date) values (?,?,null)", new Object[] {
new Integer(list.size()), text });
return text;
try {
retryTemplate.execute(new RetryCallback() {
public Object doWithRetry(RetryContext status) throws Throwable {
}
});
TransactionTemplate nestedTxTemplate = new TransactionTemplate(transactionManager);
nestedTxTemplate.setPropagationBehavior(TransactionTemplate.PROPAGATION_NESTED);
return nestedTxTemplate.execute(new TransactionCallback() {
public Object doInTransaction(TransactionStatus nestedStatus) {
list.add(text);
System.err.println("Inserting: [" + list.size() + "," + text + "]");
simpleJdbcTemplate.update("INSERT into T_FOOS (id,name,foo_date) values (?,?,null)", list.size(), text);
return text;
}
});
}
});
} catch (Exception e) {
throw new RuntimeException(e);
}
// The nested database transaction has committed...
int count = simpleJdbcTemplate.queryForInt("select count(*) from T_FOOS");
assertEquals(1, count);
// force rollback...
outerStatus.setRollbackOnly();
return null;
}
});
// The database transaction has committed...
int count = jdbcTemplate.queryForInt("select count(*) from T_FOOS");
assertEquals(1, count);
// force rollback...
endTransaction();
startNewTransaction();
// Verify the state after stransactional processing is complete
List<String> msgs = getMessages();
// The database portion rolled back...
count = jdbcTemplate.queryForInt("select count(*) from T_FOOS");
int count = simpleJdbcTemplate.queryForInt("select count(*) from T_FOOS");
assertEquals(0, count);
// ... and so did the message session.
assertEquals("[foo]", msgs.toString());
}
/**
/*
* Message processing is successful on the second attempt but must receive
* the message again.
*
* @throws Exception
*/
@Test
public void testExternalRetrySuccessOnSecondAttempt() throws Exception {
assertInitialState();
// force commit so that the retry executes in its own transaction (not
// nested)...
setComplete();
endTransaction();
retryTemplate.execute(new RetryCallback() {
public Object doWithRetry(RetryContext status) throws Throwable {
// use REQUIRES_NEW so that the retry executes in its own transaction
TransactionTemplate transactionTemplate = new TransactionTemplate(transactionManager);
transactionTemplate.setPropagationBehavior(TransactionTemplate.PROPAGATION_REQUIRES_NEW);
return transactionTemplate.execute(new TransactionCallback() {
public Object doInTransaction(TransactionStatus status) {
@@ -275,8 +301,7 @@ public class SynchronousTests extends AbstractTransactionalDataSourceSpringConte
// transaction...
final String text = (String) jmsTemplate.receiveAndConvert("queue");
list.add(text);
jdbcTemplate.update("INSERT into T_FOOS (id,name,foo_date) values (?,?,null)", new Object[] {
new Integer(list.size()), text });
simpleJdbcTemplate.update("INSERT into T_FOOS (id,name,foo_date) values (?,?,null)", list.size(), text);
if (list.size() == 1) {
throw new RuntimeException("Rollback!");
}
@@ -288,38 +313,35 @@ public class SynchronousTests extends AbstractTransactionalDataSourceSpringConte
}
});
startNewTransaction();
// Verify the state after stransactional processing is complete
List<String> msgs = getMessages();
// The database portion committed once...
int count = jdbcTemplate.queryForInt("select count(*) from T_FOOS");
int count = simpleJdbcTemplate.queryForInt("select count(*) from T_FOOS");
assertEquals(1, count);
// ... and so did the message session.
assertEquals("[]", msgs.toString());
}
/**
/*
* Message processing fails.
*
* @throws Exception
*/
@Transactional @Test
public void testExternalRetryFailOnSecondAttempt() throws Exception {
assertInitialState();
// force commit so that the retry executes in its own transaction (not
// nested)...
setComplete();
endTransaction();
try {
retryTemplate.execute(new RetryCallback() {
public Object doWithRetry(RetryContext status) throws Throwable {
// use REQUIRES_NEW so that the retry executes in its own transaction
TransactionTemplate transactionTemplate = new TransactionTemplate(transactionManager);
transactionTemplate.setPropagationBehavior(TransactionTemplate.PROPAGATION_REQUIRES_NEW);
return transactionTemplate.execute(new TransactionCallback() {
public Object doInTransaction(TransactionStatus status) {
@@ -327,8 +349,7 @@ public class SynchronousTests extends AbstractTransactionalDataSourceSpringConte
// transaction...
final String text = (String) jmsTemplate.receiveAndConvert("queue");
list.add(text);
jdbcTemplate.update("INSERT into T_FOOS (id,name,foo_date) values (?,?,null)",
new Object[] { new Integer(list.size()), text });
simpleJdbcTemplate.update("INSERT into T_FOOS (id,name,foo_date) values (?,?,null)", list.size(), text);
throw new RuntimeException("Rollback!");
}
@@ -349,12 +370,12 @@ public class SynchronousTests extends AbstractTransactionalDataSourceSpringConte
// expected
}
startNewTransaction();
// Verify the state after stransactional processing is complete
List<String> msgs = getMessages();
// The database portion rolled back...
int count = jdbcTemplate.queryForInt("select count(*) from T_FOOS");
int count = simpleJdbcTemplate.queryForInt("select count(*) from T_FOOS");
assertEquals(0, count);
// ... and so did the message session.
@@ -371,4 +392,5 @@ public class SynchronousTests extends AbstractTransactionalDataSourceSpringConte
}
return msgs;
}
}

365
spring-batch-infrastructure-tests/src/test/java/org/springframework/retry/jms/SynchronousTests.java
View File

@@ -1,365 +0,0 @@
/*
* 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.retry.jms;
import java.util.ArrayList;
import java.util.List;
import org.springframework.batch.item.jms.JmsItemReader;
import org.springframework.batch.jms.ExternalRetryInBatchTests;
import org.springframework.batch.retry.RetryCallback;
import org.springframework.batch.retry.RetryContext;
import org.springframework.batch.retry.callback.RecoveryRetryCallback;
import org.springframework.batch.retry.support.RetryTemplate;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.test.AbstractTransactionalDataSourceSpringContextTests;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.support.TransactionCallback;
import org.springframework.transaction.support.TransactionTemplate;
import org.springframework.util.ClassUtils;
public class SynchronousTests extends AbstractTransactionalDataSourceSpringContextTests {
private JmsTemplate jmsTemplate;
private RetryTemplate retryTemplate;
public void setJmsTemplate(JmsTemplate jmsTemplate) {
this.jmsTemplate = jmsTemplate;
}
protected String[] getConfigLocations() {
return new String[] { ClassUtils.addResourcePathToPackagePath(ExternalRetryInBatchTests.class,
"jms-context.xml") };
}
protected void onSetUpBeforeTransaction() throws Exception {
super.onSetUpBeforeTransaction();
String foo = "";
int count = 0;
while (foo != null && count < 100) {
logger.debug("Drained message: "+count+": "+foo);
foo = (String) jmsTemplate.receiveAndConvert("queue");
count++;
}
jdbcTemplate.execute("delete from T_FOOS");
jmsTemplate.convertAndSend("queue", "foo");
retryTemplate = new RetryTemplate();
}
private void assertInitialState() {
int count = jdbcTemplate.queryForInt("select count(*) from T_FOOS");
assertEquals(0, count);
}
List<Object> list = new ArrayList<Object>();
/**
* Message processing is successful on the second attempt without having to
* receive the message again.
*
* @throws Exception
*/
public void testInternalRetrySuccessOnSecondAttempt() throws Exception {
assertInitialState();
/*
* We either want the JMS receive to be outside a transaction, or we
* need the database transaction in the retry to be PROPAGATION_NESTED.
* Otherwise JMS will roll back when the retry callback is eventually
* successful because of the previous exception.
* PROPAGATION_REQUIRES_NEW is wrong because it doesn't allow the outer
* transaction to fail and rollback the inner one.
*/
final String text = (String) jmsTemplate.receiveAndConvert("queue");
retryTemplate.execute(new RetryCallback() {
public Object doWithRetry(RetryContext status) throws Throwable {
TransactionTemplate transactionTemplate = new TransactionTemplate(transactionManager);
transactionTemplate.setPropagationBehavior(TransactionTemplate.PROPAGATION_NESTED);
return transactionTemplate.execute(new TransactionCallback() {
public Object doInTransaction(TransactionStatus status) {
list.add(text);
logger.debug("Inserting: [" + list.size() + "," + text + "]");
jdbcTemplate.update("INSERT into T_FOOS (id,name,foo_date) values (?,?,null)", new Object[] {
new Integer(list.size()), text });
if (list.size() == 1) {
throw new RuntimeException("Rollback!");
}
return text;
}
});
}
});
// force commit...
setComplete();
endTransaction();
startNewTransaction();
List<String> msgs = getMessages();
// The database portion committed once...
int count = jdbcTemplate.queryForInt("select count(*) from T_FOOS");
assertEquals(1, count);
// ... and so did the message session.
assertEquals("[]", msgs.toString());
}
/**
* Message processing is successful on the second attempt without having to
* receive the message again - uses JmsItemProvider internally.
*
* @throws Exception
*/
public void testInternalRetrySuccessOnSecondAttemptWithItemProvider() throws Exception {
assertInitialState();
JmsItemReader<Object> provider = new JmsItemReader<Object>();
// provider.setItemType(Message.class);
provider.setJmsTemplate(jmsTemplate);
jmsTemplate.setDefaultDestinationName("queue");
final Object text = provider.read();
retryTemplate.execute(new RecoveryRetryCallback(text, new RetryCallback() {
public Object doWithRetry(RetryContext context) throws Throwable {
TransactionTemplate transactionTemplate = new TransactionTemplate(transactionManager);
transactionTemplate.setPropagationBehavior(TransactionTemplate.PROPAGATION_NESTED);
return transactionTemplate.execute(new TransactionCallback() {
public Object doInTransaction(TransactionStatus status) {
list.add(text);
logger.debug("Inserting: [" + list.size() + "," + text + "]");
jdbcTemplate.update("INSERT into T_FOOS (id,name,foo_date) values (?,?,null)", new Object[] {
new Integer(list.size()), text });
if (list.size() == 1) {
throw new RuntimeException("Rollback!");
}
return text;
}
});
}
}));
// force commit...
setComplete();
endTransaction();
startNewTransaction();
List<String> msgs = getMessages();
// The database portion committed once...
int count = jdbcTemplate.queryForInt("select count(*) from T_FOOS");
assertEquals(1, count);
// ... and so did the message session.
assertEquals("[]", msgs.toString());
}
/**
* Message processing is successful on the second attempt without having to
* receive the message again.
*
* @throws Exception
*/
public void testInternalRetrySuccessOnFirstAttemptRollbackOuter() throws Exception {
assertInitialState();
/*
* We either want the JMS receive to be outside a transaction, or we
* need the database transaction in the retry to be PROPAGATION_NESTED.
* Otherwise JMS will roll back when the retry callback is eventually
* successful because of the previous exception.
* PROPAGATION_REQUIRES_NEW is wrong because it doesn't allow the outer
* transaction to fail and rollback the inner one.
*/
final String text = (String) jmsTemplate.receiveAndConvert("queue");
retryTemplate.execute(new RetryCallback() {
public Object doWithRetry(RetryContext status) throws Throwable {
TransactionTemplate transactionTemplate = new TransactionTemplate(transactionManager);
transactionTemplate.setPropagationBehavior(TransactionTemplate.PROPAGATION_NESTED);
return transactionTemplate.execute(new TransactionCallback() {
public Object doInTransaction(TransactionStatus status) {
list.add(text);
logger.debug("Inserting: [" + list.size() + "," + text + "]");
jdbcTemplate.update("INSERT into T_FOOS (id,name,foo_date) values (?,?,null)", new Object[] {
new Integer(list.size()), text });
return text;
}
});
}
});
// The database transaction has committed...
int count = jdbcTemplate.queryForInt("select count(*) from T_FOOS");
assertEquals(1, count);
// force rollback...
endTransaction();
startNewTransaction();
List<String> msgs = getMessages();
// The database portion rolled back...
count = jdbcTemplate.queryForInt("select count(*) from T_FOOS");
assertEquals(0, count);
// ... and so did the message session.
assertEquals("[foo]", msgs.toString());
}
/**
* Message processing is successful on the second attempt but must receive
* the message again.
*
* @throws Exception
*/
public void testExternalRetrySuccessOnSecondAttempt() throws Exception {
assertInitialState();
// force commit so that the retry executes in its own transaction (not
// nested)...
setComplete();
endTransaction();
retryTemplate.execute(new RetryCallback() {
public Object doWithRetry(RetryContext status) throws Throwable {
TransactionTemplate transactionTemplate = new TransactionTemplate(transactionManager);
return transactionTemplate.execute(new TransactionCallback() {
public Object doInTransaction(TransactionStatus status) {
// The receieve is inside the retry and the
// transaction...
final String text = (String) jmsTemplate.receiveAndConvert("queue");
list.add(text);
logger.debug("Processing Foo: "+text);
jdbcTemplate.update("INSERT into T_FOOS (id,name,foo_date) values (?,?,null)", new Object[] {
new Integer(list.size()), text });
if (list.size() == 1) {
throw new RuntimeException("Rollback!");
}
return text;
}
});
}
});
startNewTransaction();
List<String> msgs = getMessages();
// The database portion committed once...
int count = jdbcTemplate.queryForInt("select count(*) from T_FOOS");
assertEquals(1, count);
// ... and so did the message session.
assertEquals("[]", msgs.toString());
}
/**
* Message processing fails.
*
* @throws Exception
*/
public void testExternalRetryFailOnSecondAttempt() throws Exception {
assertInitialState();
// force commit so that the retry executes in its own transaction (not
// nested)...
setComplete();
endTransaction();
try {
retryTemplate.execute(new RetryCallback() {
public Object doWithRetry(RetryContext status) throws Throwable {
TransactionTemplate transactionTemplate = new TransactionTemplate(transactionManager);
return transactionTemplate.execute(new TransactionCallback() {
public Object doInTransaction(TransactionStatus status) {
// The receive is inside the retry and the
// transaction...
final String text = (String) jmsTemplate.receiveAndConvert("queue");
list.add(text);
logger.debug("Processing Foo: "+text);
jdbcTemplate.update("INSERT into T_FOOS (id,name,foo_date) values (?,?,null)",
new Object[] { new Integer(list.size()), text });
throw new RuntimeException("Rollback!");
}
});
}
});
fail("Expected RuntimeException");
}
catch (RuntimeException e) {
assertEquals("Rollback!", e.getMessage());
// expected
}
startNewTransaction();
List<String> msgs = getMessages();
// The database portion rolled back...
int count = jdbcTemplate.queryForInt("select count(*) from T_FOOS");
assertEquals(0, count);
// ... and so did the message session.
assertTrue(msgs.contains("foo"));
}
private List<String> getMessages() {
String next = "";
List<String> msgs = new ArrayList<String>();
while (next != null) {
next = (String) jmsTemplate.receiveAndConvert("queue");
if (next != null)
msgs.add(next);
}
return msgs;
}
}