IN PROGRESS - issue BATCH-7: Remove transaction synchronization and state management from input/output sources (formerly buffering)
http://jira.springframework.org/browse/BATCH-7 Remove ResourceLifecycle and make existing implementations into ItemStream
This commit is contained in:
@@ -8,7 +8,7 @@ import org.springframework.beans.factory.DisposableBean;
|
||||
import org.springframework.beans.factory.InitializingBean;
|
||||
import org.springframework.jdbc.core.JdbcTemplate;
|
||||
|
||||
class FooItemReader extends AbstractItemReader implements ItemReader, ItemStream, DisposableBean, InitializingBean{
|
||||
class FooItemReader extends AbstractItemReader implements ItemStream, ItemReader, DisposableBean, InitializingBean{
|
||||
|
||||
DrivingQueryItemReader inputSource;
|
||||
FooDao fooDao = new SingleKeyFooDao();
|
||||
@@ -44,5 +44,8 @@ class FooItemReader extends AbstractItemReader implements ItemReader, ItemStream
|
||||
}
|
||||
|
||||
public void afterPropertiesSet() throws Exception {
|
||||
}
|
||||
|
||||
public void open() throws Exception {
|
||||
};
|
||||
}
|
||||
|
||||
@@ -22,7 +22,7 @@ import java.util.Map;
|
||||
|
||||
import junit.framework.TestCase;
|
||||
|
||||
import org.springframework.batch.item.writer.AbstractItemWriter;
|
||||
import org.springframework.batch.item.ItemWriter;
|
||||
import org.springframework.batch.repeat.ExitStatus;
|
||||
import org.springframework.batch.repeat.RepeatContext;
|
||||
import org.springframework.batch.repeat.RepeatInterceptor;
|
||||
@@ -41,13 +41,12 @@ public class HibernateAwareItemWriterTests extends TestCase {
|
||||
public void flush() throws DataAccessException {
|
||||
list.add("flush");
|
||||
}
|
||||
|
||||
public void clear() {
|
||||
list.add("clear");
|
||||
list.add("clear");
|
||||
};
|
||||
}
|
||||
|
||||
private class StubItemWriter extends AbstractItemWriter implements RepeatInterceptor {
|
||||
private class StubItemWriter implements ItemWriter, RepeatInterceptor {
|
||||
public void write(Object item) {
|
||||
list.add(item);
|
||||
}
|
||||
@@ -71,10 +70,13 @@ public class HibernateAwareItemWriterTests extends TestCase {
|
||||
public void open(RepeatContext context) {
|
||||
list.add(context);
|
||||
}
|
||||
|
||||
public void close() throws Exception {
|
||||
}
|
||||
}
|
||||
|
||||
HibernateAwareItemWriter writer = new HibernateAwareItemWriter();
|
||||
|
||||
|
||||
final List list = new ArrayList();
|
||||
|
||||
private RepeatContextSupport context;
|
||||
@@ -91,16 +93,15 @@ public class HibernateAwareItemWriterTests extends TestCase {
|
||||
writer.setHibernateTemplate(new HibernateTemplateWrapper());
|
||||
list.clear();
|
||||
}
|
||||
|
||||
/*
|
||||
* (non-Javadoc)
|
||||
|
||||
/* (non-Javadoc)
|
||||
* @see junit.framework.TestCase#tearDown()
|
||||
*/
|
||||
protected void tearDown() throws Exception {
|
||||
Map map = TransactionSynchronizationManager.getResourceMap();
|
||||
for (Iterator iterator = map.keySet().iterator(); iterator.hasNext();) {
|
||||
String key = (String) iterator.next();
|
||||
TransactionSynchronizationManager.unbindResource(key);
|
||||
TransactionSynchronizationManager.unbindResource(key);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -115,10 +116,10 @@ public class HibernateAwareItemWriterTests extends TestCase {
|
||||
try {
|
||||
writer.afterPropertiesSet();
|
||||
fail("Expected IllegalArgumentException");
|
||||
}
|
||||
catch (IllegalArgumentException e) {
|
||||
} catch (IllegalArgumentException e) {
|
||||
// expected
|
||||
assertTrue("Wrong message for exception: " + e.getMessage(), e.getMessage().indexOf("delegate") >= 0);
|
||||
assertTrue("Wrong message for exception: " + e.getMessage(), e
|
||||
.getMessage().indexOf("delegate") >= 0);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -135,7 +136,7 @@ public class HibernateAwareItemWriterTests extends TestCase {
|
||||
/**
|
||||
* Test method for
|
||||
* {@link org.springframework.batch.io.support.HibernateAwareItemWriter#write(java.lang.Object)}.
|
||||
* @throws Exception
|
||||
* @throws Exception
|
||||
*/
|
||||
public void testWrite() throws Exception {
|
||||
writer.write("foo");
|
||||
@@ -157,8 +158,7 @@ public class HibernateAwareItemWriterTests extends TestCase {
|
||||
try {
|
||||
writer.close(context);
|
||||
fail("Expected RuntimeException");
|
||||
}
|
||||
catch (RuntimeException e) {
|
||||
} catch (RuntimeException e) {
|
||||
assertEquals("bar", e.getMessage());
|
||||
}
|
||||
assertEquals(2, list.size());
|
||||
@@ -169,7 +169,7 @@ public class HibernateAwareItemWriterTests extends TestCase {
|
||||
/**
|
||||
* Test method for
|
||||
* {@link org.springframework.batch.io.support.HibernateAwareItemWriter#write(java.lang.Object)}.
|
||||
* @throws Exception
|
||||
* @throws Exception
|
||||
*/
|
||||
public void testWriteAndCloseWithFailure() throws Exception {
|
||||
final RuntimeException ex = new RuntimeException("bar");
|
||||
@@ -182,8 +182,7 @@ public class HibernateAwareItemWriterTests extends TestCase {
|
||||
try {
|
||||
writer.close(context);
|
||||
fail("Expected RuntimeException");
|
||||
}
|
||||
catch (RuntimeException e) {
|
||||
} catch (RuntimeException e) {
|
||||
assertEquals("bar", e.getMessage());
|
||||
}
|
||||
assertEquals(3, list.size());
|
||||
|
||||
@@ -8,10 +8,11 @@ import java.util.Properties;
|
||||
import junit.framework.TestCase;
|
||||
|
||||
import org.easymock.MockControl;
|
||||
import org.springframework.batch.item.ItemStream;
|
||||
import org.springframework.batch.item.ItemWriter;
|
||||
import org.springframework.batch.item.StreamContext;
|
||||
import org.springframework.batch.item.stream.GenericStreamContext;
|
||||
import org.springframework.batch.item.stream.ItemStreamAdapter;
|
||||
import org.springframework.batch.item.writer.CompositeItemWriter;
|
||||
import org.springframework.batch.statistics.StatisticsProvider;
|
||||
|
||||
/**
|
||||
@@ -48,7 +49,7 @@ public class CompositeItemWriterTests extends TestCase {
|
||||
controls.add(control);
|
||||
}
|
||||
|
||||
itemProcessor.setItemWriters(processors);
|
||||
itemProcessor.setDelegates(processors);
|
||||
itemProcessor.write(data);
|
||||
|
||||
for (Iterator iterator = controls.iterator(); iterator.hasNext();) {
|
||||
@@ -61,22 +62,27 @@ public class CompositeItemWriterTests extends TestCase {
|
||||
* All Restartable processors should be restarted, not-Restartable processors should be ignored.
|
||||
*/
|
||||
public void testRestart() {
|
||||
final ItemWriter p2 = new ItemWriterStub();
|
||||
final ItemWriter p3 = new ItemWriterStub();
|
||||
//this mock with undefined behaviour makes sure not-Restartable processor is ignored
|
||||
MockControl p1c = MockControl.createStrictControl(ItemWriter.class);
|
||||
final ItemWriter p1 = (ItemWriter) p1c.getMock();
|
||||
|
||||
final ItemWriter p2 = new StubItemWriter();
|
||||
final ItemWriter p3 = new StubItemWriter();
|
||||
List itemProcessors = new ArrayList(){{
|
||||
add(p1);
|
||||
add(p2);
|
||||
add(p3);
|
||||
}};
|
||||
itemProcessor.setItemWriters(itemProcessors);
|
||||
itemProcessor.setDelegates(itemProcessors);
|
||||
|
||||
StreamContext rd = itemProcessor.getRestartData();
|
||||
itemProcessor.restoreFrom(rd);
|
||||
|
||||
for (Iterator iterator = itemProcessors.iterator(); iterator.hasNext();) {
|
||||
ItemWriter processor = (ItemWriter) iterator.next();
|
||||
if (processor instanceof ItemWriterStub) {
|
||||
if (processor instanceof StubItemWriter) {
|
||||
assertTrue("Injected processors are restarted",
|
||||
((ItemWriterStub)processor).restarted);
|
||||
((StubItemWriter)processor).restarted);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -86,35 +92,34 @@ public class CompositeItemWriterTests extends TestCase {
|
||||
|
||||
final int NUMBER_OF_PROCESSORS = 10;
|
||||
|
||||
List controls = new ArrayList(NUMBER_OF_PROCESSORS);
|
||||
List processors = new ArrayList(NUMBER_OF_PROCESSORS);
|
||||
final List list = new ArrayList(NUMBER_OF_PROCESSORS);
|
||||
|
||||
for (int i = 0; i < NUMBER_OF_PROCESSORS; i++) {
|
||||
MockControl control = MockControl.createStrictControl(ItemWriter.class);
|
||||
ItemWriter processor = (ItemWriter) control.getMock();
|
||||
|
||||
ItemWriter processor = new AbstractItemWriter() {
|
||||
public void write(Object item) throws Exception {
|
||||
throw new IllegalStateException("No way!");
|
||||
}
|
||||
public void close() throws Exception {
|
||||
list.add(this);
|
||||
}
|
||||
};
|
||||
processor.close();
|
||||
control.setVoidCallable();
|
||||
control.replay();
|
||||
|
||||
processors.add(processor);
|
||||
controls.add(control);
|
||||
}
|
||||
|
||||
itemProcessor.setItemWriters(processors);
|
||||
itemProcessor.setDelegates(processors);
|
||||
itemProcessor.close();
|
||||
|
||||
for (Iterator iterator = controls.iterator(); iterator.hasNext();) {
|
||||
MockControl control = (MockControl) iterator.next();
|
||||
control.verify();
|
||||
}
|
||||
assertEquals(NUMBER_OF_PROCESSORS, list.size());
|
||||
|
||||
}
|
||||
|
||||
/**
|
||||
* Stub for testing restart. Checks the restart data received is the same that was returned by
|
||||
* <code>getRestartData()</code>
|
||||
*/
|
||||
private static class ItemWriterStub extends ItemStreamAdapter implements ItemWriter, StatisticsProvider {
|
||||
private static class StubItemWriter implements ItemWriter, ItemStream, StatisticsProvider {
|
||||
|
||||
private static final String RESTART_KEY = "restartData";
|
||||
private static final String STATS_KEY = "stats";
|
||||
@@ -149,9 +154,10 @@ public class CompositeItemWriterTests extends TestCase {
|
||||
}
|
||||
|
||||
public void close() throws Exception {
|
||||
|
||||
}
|
||||
|
||||
public void open() throws Exception {
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -19,9 +19,9 @@ package org.springframework.batch.repeat.support;
|
||||
import java.util.HashSet;
|
||||
import java.util.Set;
|
||||
|
||||
import org.springframework.batch.repeat.ExitStatus;
|
||||
import org.springframework.batch.repeat.RepeatCallback;
|
||||
import org.springframework.batch.repeat.RepeatContext;
|
||||
import org.springframework.batch.repeat.ExitStatus;
|
||||
import org.springframework.batch.repeat.callback.ItemReaderRepeatCallback;
|
||||
import org.springframework.core.task.SimpleAsyncTaskExecutor;
|
||||
|
||||
|
||||
Reference in New Issue
Block a user