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

RESOLVED - issue BATCH-343: close() is called twice on ItemReaders/ItemWriters

Browse Source 
http://jira.springframework.org/browse/BATCH-343

ItemStream implementations no longer implement DisposableBean
This commit is contained in:
robokaso
2008-02-06 14:20:21 +00:00
parent 242970dd29
commit 37cf297899
14 changed files with 195 additions and 249 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
spring-batch-infrastructure/src/test/java/org/springframework/batch/io/cursor/HibernateCursorItemReaderIntegrationTests.java
View File

@@ -45,7 +45,7 @@ public class HibernateCursorItemReaderIntegrationTests extends AbstractDataSourc
* called only in uninitialized state.
*/
public void testSetUseStatelessSession() {
HibernateCursorItemReader inputSource = ((HibernateCursorItemReader) source);
HibernateCursorItemReader inputSource = ((HibernateCursorItemReader) reader);
// initialize and call setter => error
inputSource.open();

16
spring-batch-infrastructure/src/test/java/org/springframework/batch/io/driving/FooInputSource.java
View File

@@ -8,9 +8,10 @@ import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.jdbc.core.JdbcTemplate;
class FooItemReader extends AbstractItemReader implements ItemStream, ItemReader, DisposableBean, InitializingBean{
class FooItemReader extends AbstractItemReader implements ItemStream, ItemReader, DisposableBean, InitializingBean {
DrivingQueryItemReader inputSource;
FooDao fooDao = new SingleKeyFooDao();
public FooItemReader(DrivingQueryItemReader inputSource, JdbcTemplate jdbcTemplate) {
@@ -20,9 +21,10 @@ class FooItemReader extends AbstractItemReader implements ItemStream, ItemReader
public Object read() {
Object key = inputSource.read();
if(key != null){
if (key != null) {
return fooDao.getFoo(key);
}else{
}
else {
return null;
}
}
@@ -36,7 +38,7 @@ class FooItemReader extends AbstractItemReader implements ItemStream, ItemReader
}
public void destroy() throws Exception {
inputSource.destroy();
inputSource.close();
}
public void setFooDao(FooDao fooDao) {
@@ -60,14 +62,16 @@ class FooItemReader extends AbstractItemReader implements ItemStream, ItemReader
return true;
}
/* (non-Javadoc)
/*
* (non-Javadoc)
* @see org.springframework.batch.item.ItemStream#mark(org.springframework.batch.item.StreamContext)
*/
public void mark() {
inputSource.mark();
}
/* (non-Javadoc)
/*
* (non-Javadoc)
* @see org.springframework.batch.item.ItemStream#reset(org.springframework.batch.item.StreamContext)
*/
public void reset() {

202
spring-batch-infrastructure/src/test/java/org/springframework/batch/io/support/AbstractDataSourceItemReaderIntegrationTests.java
View File

@@ -2,47 +2,49 @@ package org.springframework.batch.io.support;
import org.springframework.batch.io.Skippable;
import org.springframework.batch.io.sample.domain.Foo;
import org.springframework.batch.item.ExecutionAttributes;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemStream;
import org.springframework.batch.item.ExecutionAttributes;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.test.AbstractTransactionalDataSourceSpringContextTests;
import org.springframework.util.Assert;
/**
* Common scenarios for testing {@link ItemReader} implementations which read data from database.
* Common scenarios for testing {@link ItemReader} implementations which read
* data from database.
*
* @author Lucas Ward
* @author Robert Kasanicky
*/
public abstract class AbstractDataSourceItemReaderIntegrationTests extends AbstractTransactionalDataSourceSpringContextTests {
public abstract class AbstractDataSourceItemReaderIntegrationTests extends
AbstractTransactionalDataSourceSpringContextTests {
protected ItemReader source;
protected ItemReader reader;
/**
* @return configured input source ready for use
*/
protected abstract ItemReader createItemReader() throws Exception;
protected String[] getConfigLocations(){
return new String[] { "org/springframework/batch/io/sql/data-source-context.xml"};
protected String[] getConfigLocations() {
return new String[] { "org/springframework/batch/io/sql/data-source-context.xml" };
}
/* (non-Javadoc)
/*
* (non-Javadoc)
* @see org.springframework.test.AbstractTransactionalSpringContextTests#onSetUpInTransaction()
*/
protected void onSetUpInTransaction() throws Exception {
super.onSetUpInTransaction();
source = createItemReader();
reader = createItemReader();
}
/* (non-Javadoc)
/*
* (non-Javadoc)
* @see org.springframework.test.AbstractTransactionalSpringContextTests#onTearDownAfterTransaction()
*/
protected void onTearDownAfterTransaction() throws Exception {
getAsDisposableBean(source).destroy();
getAsItemStream(reader).close();
super.onTearDownAfterTransaction();
}
@@ -50,47 +52,47 @@ public abstract class AbstractDataSourceItemReaderIntegrationTests extends Abstr
* Regular scenario - read all rows and eventually return null.
*/
public void testNormalProcessing() throws Exception {
getAsInitializingBean(source).afterPropertiesSet();
Foo foo1 = (Foo) source.read();
getAsInitializingBean(reader).afterPropertiesSet();
Foo foo1 = (Foo) reader.read();
assertEquals(1, foo1.getValue());
Foo foo2 = (Foo) source.read();
Foo foo2 = (Foo) reader.read();
assertEquals(2, foo2.getValue());
Foo foo3 = (Foo) source.read();
Foo foo3 = (Foo) reader.read();
assertEquals(3, foo3.getValue());
Foo foo4 = (Foo) source.read();
Foo foo4 = (Foo) reader.read();
assertEquals(4, foo4.getValue());
Foo foo5 = (Foo) source.read();
Foo foo5 = (Foo) reader.read();
assertEquals(5, foo5.getValue());
assertNull(source.read());
assertNull(reader.read());
}
/**
* Restart scenario - read records, save restart data, create new input source
* and restore from restart data - the new input source should continue where
* the old one finished.
* Restart scenario - read records, save restart data, create new input
* source and restore from restart data - the new input source should
* continue where the old one finished.
*/
public void testRestart() throws Exception {
Foo foo1 = (Foo) source.read();
Foo foo1 = (Foo) reader.read();
assertEquals(1, foo1.getValue());
Foo foo2 = (Foo) source.read();
Foo foo2 = (Foo) reader.read();
assertEquals(2, foo2.getValue());
ExecutionAttributes streamContext = getAsRestartable(source).getExecutionAttributes();
ExecutionAttributes streamContext = getAsItemStream(reader).getExecutionAttributes();
// create new input source
source = createItemReader();
reader = createItemReader();
getAsRestartable(source).restoreFrom(streamContext);
getAsItemStream(reader).restoreFrom(streamContext);
Foo fooAfterRestart = (Foo) source.read();
Foo fooAfterRestart = (Foo) reader.read();
assertEquals(3, fooAfterRestart.getValue());
}
@@ -99,149 +101,147 @@ public abstract class AbstractDataSourceItemReaderIntegrationTests extends Abstr
*/
public void testInvalidRestore() throws Exception {
Foo foo1 = (Foo) source.read();
Foo foo1 = (Foo) reader.read();
assertEquals(1, foo1.getValue());
Foo foo2 = (Foo) source.read();
Foo foo2 = (Foo) reader.read();
assertEquals(2, foo2.getValue());
ExecutionAttributes streamContext = getAsRestartable(source).getExecutionAttributes();
ExecutionAttributes streamContext = getAsItemStream(reader).getExecutionAttributes();
// create new input source
source = createItemReader();
reader = createItemReader();
Foo foo = (Foo) source.read();
Foo foo = (Foo) reader.read();
assertEquals(1, foo.getValue());
try {
getAsRestartable(source).restoreFrom(streamContext);
getAsItemStream(reader).restoreFrom(streamContext);
fail();
}
catch (IllegalStateException ex) {
// expected
}
}
/**
* Empty restart data should be handled gracefully.
* @throws Exception
* @throws Exception
*/
public void testRestoreFromEmptyData() throws Exception {
ExecutionAttributes streamContext = new ExecutionAttributes();
getAsRestartable(source).restoreFrom(streamContext);
Foo foo = (Foo) source.read();
getAsItemStream(reader).restoreFrom(streamContext);
Foo foo = (Foo) reader.read();
assertEquals(1, foo.getValue());
}
/**
* Rollback scenario - input source rollbacks to last commit point.
* @throws Exception
* @throws Exception
*/
public void testRollback() throws Exception {
Foo foo1 = (Foo) source.read();
Foo foo1 = (Foo) reader.read();
commit();
Foo foo2 = (Foo) source.read();
Foo foo2 = (Foo) reader.read();
Assert.state(!foo2.equals(foo1));
Foo foo3 = (Foo) source.read();
Foo foo3 = (Foo) reader.read();
Assert.state(!foo2.equals(foo3));
rollback();
assertEquals(foo2, source.read());
assertEquals(foo2, reader.read());
}
/**
* Rollback scenario with skip - input source rollbacks to last commit point.
* @throws Exception
* Rollback scenario with skip - input source rollbacks to last commit
* point.
* @throws Exception
*/
public void testRollbackAndSkip() throws Exception {
if (!(source instanceof Skippable)) {
if (!(reader instanceof Skippable)) {
return;
}
Foo foo1 = (Foo) source.read();
Foo foo1 = (Foo) reader.read();
commit();
Foo foo2 = (Foo) source.read();
Foo foo2 = (Foo) reader.read();
Assert.state(!foo2.equals(foo1));
Foo foo3 = (Foo) source.read();
Foo foo3 = (Foo) reader.read();
Assert.state(!foo2.equals(foo3));
getAsSkippable(source).skip();
getAsSkippable(reader).skip();
rollback();
assertEquals(foo2, source.read());
Foo foo4 = (Foo) source.read();
assertEquals(foo2, reader.read());
Foo foo4 = (Foo) reader.read();
assertEquals(4, foo4.getValue());
}
/**
* Rollback scenario with skip and restart - input source rollbacks to last commit point.
* @throws Exception
* Rollback scenario with skip and restart - input source rollbacks to last
* commit point.
* @throws Exception
*/
public void testRollbackSkipAndRestart() throws Exception {
if (!(source instanceof Skippable)) {
if (!(reader instanceof Skippable)) {
return;
}
Foo foo1 = (Foo) source.read();
Foo foo1 = (Foo) reader.read();
commit();
Foo foo2 = (Foo) source.read();
Foo foo2 = (Foo) reader.read();
Assert.state(!foo2.equals(foo1));
Foo foo3 = (Foo) source.read();
Foo foo3 = (Foo) reader.read();
Assert.state(!foo2.equals(foo3));
getAsSkippable(source).skip();
getAsSkippable(reader).skip();
rollback();
ExecutionAttributes streamContext = getAsRestartable(source).getExecutionAttributes();
ExecutionAttributes streamContext = getAsItemStream(reader).getExecutionAttributes();
// create new input source
source = createItemReader();
reader = createItemReader();
getAsRestartable(source).restoreFrom(streamContext);
getAsItemStream(reader).restoreFrom(streamContext);
assertEquals(foo2, source.read());
Foo foo4 = (Foo) source.read();
assertEquals(foo2, reader.read());
Foo foo4 = (Foo) reader.read();
assertEquals(4, foo4.getValue());
}
private void commit() {
((ItemStream) source).mark();
((ItemStream) reader).mark();
}
private void rollback() {
((ItemStream) source).reset();
((ItemStream) reader).reset();
}
private Skippable getAsSkippable(ItemReader source) {
return (Skippable) source;
}
private ItemStream getAsRestartable(ItemReader source) {
private ItemStream getAsItemStream(ItemReader source) {
return (ItemStream) source;
}
private InitializingBean getAsInitializingBean(ItemReader source) {
return (InitializingBean) source;
}
private DisposableBean getAsDisposableBean(ItemReader source) {
return (DisposableBean) source;
}
}

4
spring-batch-infrastructure/src/test/java/org/springframework/batch/io/xml/StaxEventItemReaderTests.java
View File

@@ -88,7 +88,7 @@ public class StaxEventItemReaderTests extends TestCase {
assertNotNull(source.read());
assertNull(source.read()); // there are only two fragments
source.destroy();
source.close();
}
/**
@@ -228,7 +228,7 @@ public class StaxEventItemReaderTests extends TestCase {
assertNotNull(item);
assertTrue(newSource.isOpenCalled());
newSource.destroy(); // includes close()
newSource.close();
newSource.setOpenCalled(false);
// calling read again should require re-initialization because of close
item = newSource.read();

21
spring-batch-infrastructure/src/test/java/org/springframework/batch/io/xml/StaxEventWriterItemWriterTests.java
View File

@@ -42,7 +42,6 @@ public class StaxEventWriterItemWriterTests extends TestCase {
private static final int NOT_FOUND = -1;
protected void setUp() throws Exception {
resource = new FileSystemResource(File.createTempFile("StaxEventWriterOutputSourceTests", "xml"));
writer = createItemWriter();
@@ -94,9 +93,10 @@ public class StaxEventWriterItemWriterTests extends TestCase {
writer = createItemWriter();
writer.restoreFrom(streamContext);
writer.write(record);
writer.destroy();
writer.close();
// check the output is concatenation of 'before restart' and 'after restart' writes.
// check the output is concatenation of 'before restart' and 'after
// restart' writes.
String outputFile = outputFileContent();
int firstRecord = outputFile.indexOf(TEST_STRING);
int secondRecord = outputFile.indexOf(TEST_STRING, firstRecord + TEST_STRING.length());
@@ -115,8 +115,7 @@ public class StaxEventWriterItemWriterTests extends TestCase {
final int NUMBER_OF_RECORDS = 10;
for (int i = 1; i <= NUMBER_OF_RECORDS; i++) {
writer.write(record);
long writeStatistics =
writer.getExecutionAttributes().getLong(StaxEventItemWriter.WRITE_STATISTICS_NAME);
long writeStatistics = writer.getExecutionAttributes().getLong(StaxEventItemWriter.WRITE_STATISTICS_NAME);
assertEquals(i, writeStatistics);
}
@@ -127,9 +126,11 @@ public class StaxEventWriterItemWriterTests extends TestCase {
*/
public void testOpenAndClose() throws IOException {
writer.setRootTagName("testroot");
writer.setRootElementAttributes(new HashMap() {{
put("attribute", "value");
}});
writer.setRootElementAttributes(new HashMap() {
{
put("attribute", "value");
}
});
writer.open();
writer.mark();
@@ -191,12 +192,12 @@ public class StaxEventWriterItemWriterTests extends TestCase {
Marshaller marshaller = new SimpleMarshaller();
MarshallingEventWriterSerializer serializer = new MarshallingEventWriterSerializer(marshaller);
source.setSerializer(serializer);
source.setEncoding("UTF-8");
source.setRootTagName("root");
source.setVersion("1.0");
source.setOverwriteOutput(true);
source.afterPropertiesSet();
return source;