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

Merge pull request 126 from ddebree/BATCH-1966

Browse Source 
* BATCH-1966:
  BATCH-1966: Some minor cleanup
  Added method to replace old setName method.
  Fixed execution context name to match old code
  Added ExecutionContextUserSupport to the ItemStream Support class Changed child classes to use this shared object Changed some classes that previously extended ExecutionContextUserSupport to rather use the same common parent class as everything else. Changes all for issue BATCH-1966
This commit is contained in:
Michael Minella
2013-03-04 14:13:15 -06:00
parent b1bd3d3d1e aedf03fad0
commit 83c5fa2eaa
19 changed files with 193 additions and 146 deletions
Download Patch File Download Diff File Expand all files Collapse all files

17
spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ChunkMonitor.java
View File

@@ -21,7 +21,9 @@ import org.springframework.batch.item.ExecutionContext;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemStream;
import org.springframework.batch.item.ItemStreamException;
import org.springframework.batch.item.ItemStreamSupport;
import org.springframework.batch.item.support.CompositeItemStream;
import org.springframework.util.ClassUtils;
/**
* Manage the offset data between the last successful commit and updates made to
@@ -32,7 +34,7 @@ import org.springframework.batch.item.support.CompositeItemStream;
* @author Dave Syer
* @since 2.0
*/
public class ChunkMonitor implements ItemStream {
public class ChunkMonitor extends ItemStreamSupport {
private Log logger = LogFactory.getLog(getClass());
@@ -49,7 +51,7 @@ public class ChunkMonitor implements ItemStream {
}
}
private static final String OFFSET = ChunkMonitor.class.getName() + ".OFFSET";
private static final String OFFSET = "OFFSET";
private CompositeItemStream stream = new CompositeItemStream();
@@ -57,6 +59,10 @@ public class ChunkMonitor implements ItemStream {
private ItemReader<?> reader;
public ChunkMonitor() {
this.setExecutionContextName(ChunkMonitor.class.getName());
}
/**
* @param stream the stream to set
*/
@@ -95,6 +101,7 @@ public class ChunkMonitor implements ItemStream {
@Override
public void close() throws ItemStreamException {
super.close();
holder.set(null);
if (streamsRegistered) {
stream.close();
@@ -103,9 +110,10 @@ public class ChunkMonitor implements ItemStream {
@Override
public void open(ExecutionContext executionContext) throws ItemStreamException {
super.open(executionContext);
if (streamsRegistered) {
stream.open(executionContext);
ChunkMonitorData data = new ChunkMonitorData(executionContext.getInt(OFFSET, 0), 0);
ChunkMonitorData data = new ChunkMonitorData(executionContext.getInt(getExecutionContextKey(OFFSET), 0), 0);
holder.set(data);
if (reader == null) {
logger.warn("No ItemReader set (must be concurrent step), so ignoring offset data.");
@@ -124,6 +132,7 @@ public class ChunkMonitor implements ItemStream {
@Override
public void update(ExecutionContext executionContext) throws ItemStreamException {
super.update(executionContext);
if (streamsRegistered) {
ChunkMonitorData data = getData();
if (data.offset == 0) {
@@ -132,7 +141,7 @@ public class ChunkMonitor implements ItemStream {
stream.update(executionContext);
}
else {
executionContext.putInt(OFFSET, data.offset);
executionContext.putInt(getExecutionContextKey(OFFSET), data.offset);
}
}
}

6
spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/TestReader.java
View File

@@ -43,16 +43,16 @@ public class TestReader extends AbstractTestComponent implements ItemStreamReade
}
@Override
public void close() throws ItemStreamException {
public void close() {
}
@Override
public void open(ExecutionContext executionContext) throws ItemStreamException {
public void open(ExecutionContext executionContext) {
opened = true;
}
@Override
public void update(ExecutionContext executionContext) throws ItemStreamException {
public void update(ExecutionContext executionContext) {
}
}

18
spring-batch-core/src/test/java/org/springframework/batch/core/partition/ExampleItemReader.java
View File

@@ -5,11 +5,13 @@ import org.apache.commons.logging.LogFactory;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.batch.item.ItemStreamException;
import org.springframework.batch.item.ItemStreamReader;
import org.springframework.batch.item.support.AbstractItemStreamItemReader;
import org.springframework.util.ClassUtils;
/**
* {@link ItemStreamReader} with hard-coded input data.
*/
public class ExampleItemReader implements ItemStreamReader<String> {
public class ExampleItemReader extends AbstractItemStreamItemReader<String> {
private Log logger = LogFactory.getLog(getClass());
@@ -23,6 +25,10 @@ public class ExampleItemReader implements ItemStreamReader<String> {
public static volatile boolean fail = false;
public ExampleItemReader() {
this.setExecutionContextName(ClassUtils.getShortName(this.getClass()));
}
/**
* @param min the min to set
*/
@@ -61,18 +67,16 @@ public class ExampleItemReader implements ItemStreamReader<String> {
return input[index++];
}
@Override
public void close() throws ItemStreamException {
}
@Override
public void open(ExecutionContext executionContext) throws ItemStreamException {
index = (int) executionContext.getLong("POSITION", min);
super.open(executionContext);
index = (int) executionContext.getLong(getExecutionContextKey("POSITION"), min);
}
@Override
public void update(ExecutionContext executionContext) throws ItemStreamException {
executionContext.putLong("POSITION", index);
super.update(executionContext);
executionContext.putLong(getExecutionContextKey("POSITION"), index);
}
}

3
spring-batch-core/src/test/java/org/springframework/batch/core/step/item/ChunkMonitorTests.java
View File

@@ -54,7 +54,8 @@ public class ChunkMonitorTests {
});
monitor.registerItemStream(new ItemStreamSupport() {
@Override
public void close() throws ItemStreamException {
public void close() {
super.close();
closed = true;
}
});

15
spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanTests.java
View File

@@ -49,6 +49,7 @@ import org.springframework.batch.item.ParseException;
import org.springframework.batch.item.UnexpectedInputException;
import org.springframework.batch.item.WriteFailedException;
import org.springframework.batch.item.WriterNotOpenException;
import org.springframework.batch.item.support.AbstractItemStreamItemReader;
import org.springframework.batch.support.transaction.ResourcelessTransactionManager;
import org.springframework.beans.factory.FactoryBean;
import org.springframework.scheduling.concurrent.ConcurrentTaskExecutor;
@@ -830,23 +831,21 @@ public class FaultTolerantStepFactoryBeanTests {
public void testItemStreamOpenedEvenWithTaskExecutor() throws Exception {
writer.setFailures("4");
ItemStreamReader<String> reader = new ItemStreamReader<String>() {
ItemReader<String> reader = new AbstractItemStreamItemReader<String>() {
@Override
public void close() throws ItemStreamException {
public void close() {
super.close();
closed = true;
}
@Override
public void open(ExecutionContext executionContext) throws ItemStreamException {
public void open(ExecutionContext executionContext) {
super.open(executionContext);
opened = true;
}
@Override
public void update(ExecutionContext executionContext) throws ItemStreamException {
}
@Override
public String read() throws Exception, UnexpectedInputException, ParseException {
public String read() {
return null;
}
};

1
spring-batch-core/src/test/java/org/springframework/batch/core/step/item/TaskletStepExceptionTests.java
View File

@@ -184,6 +184,7 @@ public class TaskletStepExceptionTests {
taskletStep.setStreams(new ItemStream[] { new ItemStreamSupport() {
@Override
public void close() throws ItemStreamException {
super.close();
throw exception;
}
} });

1
spring-batch-core/src/test/java/org/springframework/batch/core/step/tasklet/AsyncTaskletStepTests.java
View File

@@ -103,6 +103,7 @@ public class AsyncTaskletStepTests {
@Override
public void update(ExecutionContext executionContext) {
super.update(executionContext);
executionContext.putInt("counter", count++);
}
});

16
spring-batch-core/src/test/java/org/springframework/batch/core/step/tasklet/TaskletStepTests.java
View File

@@ -429,12 +429,13 @@ public class TaskletStepTests {
public void testStreamManager() throws Exception {
MockRestartableItemReader reader = new MockRestartableItemReader() {
@Override
public String read() throws Exception {
public String read() {
return "foo";
}
@Override
public void update(ExecutionContext executionContext) {
super.update(executionContext);
executionContext.putString("foo", "bar");
}
};
@@ -457,6 +458,7 @@ public class TaskletStepTests {
step.setStreams(new ItemStream[] { new ItemStreamSupport() {
@Override
public void update(ExecutionContext executionContext) {
super.update(executionContext);
executionContext.putString("foo", "bar");
}
} });
@@ -500,6 +502,7 @@ public class TaskletStepTests {
@Override
public void open(ExecutionContext executionContext) throws ItemStreamException {
super.open(executionContext);
assertEquals(1, list.size());
}
};
@@ -547,7 +550,7 @@ public class TaskletStepTests {
});
step.setTasklet(new TestingChunkOrientedTasklet<String>(new MockRestartableItemReader() {
@Override
public String read() throws Exception {
public String read() throws RuntimeException {
throw new RuntimeException("FOO");
}
}, itemWriter));
@@ -562,12 +565,13 @@ public class TaskletStepTests {
public void testDirectlyInjectedStreamWhichIsAlsoReader() throws Exception {
MockRestartableItemReader reader = new MockRestartableItemReader() {
@Override
public String read() throws Exception {
public String read() {
return "foo";
}
@Override
public void update(ExecutionContext executionContext) {
super.update(executionContext);
executionContext.putString("foo", "bar");
}
};
@@ -740,6 +744,7 @@ public class TaskletStepTests {
step.setStreams(new ItemStream[] { new ItemStreamSupport() {
@Override
public void close() throws ItemStreamException {
super.close();
throw new RuntimeException("Bar");
}
} });
@@ -797,7 +802,7 @@ public class TaskletStepTests {
public void testRestartAfterFailureInFirstChunk() throws Exception {
MockRestartableItemReader reader = new MockRestartableItemReader() {
@Override
public String read() throws Exception {
public String read() throws RuntimeException {
// fail on the very first item
throw new RuntimeException("CRASH!");
}
@@ -927,12 +932,13 @@ public class TaskletStepTests {
private boolean restoreFromCalled = false;
@Override
public String read() throws Exception {
public String read() {
return "item";
}
@Override
public void update(ExecutionContext executionContext) {
super.update(executionContext);
getExecutionAttributesCalled = true;
executionContext.putString("spam", "bucket");
}

22
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ItemStreamSupport.java
View File

@@ -15,37 +15,49 @@
*/
package org.springframework.batch.item;
import org.springframework.batch.item.util.ExecutionContextUserSupport;
/**
* Empty method implementation of {@link ItemStream}.
*
* @author Dave Syer
* @author Dean de Bree
*
*/
public abstract class ItemStreamSupport implements ItemStream {
private final ExecutionContextUserSupport executionContextUserSupport = new ExecutionContextUserSupport();
/**
* No-op.
* @see org.springframework.batch.item.ItemStream#close()
*/
@Override
public void close() throws ItemStreamException {
@Override
public void close() {
}
/**
* No-op.
* @see org.springframework.batch.item.ItemStream#open(ExecutionContext)
*/
@Override
public void open(ExecutionContext executionContext) throws ItemStreamException {
@Override
public void open(ExecutionContext executionContext) {
}
/**
* Return empty {@link ExecutionContext}.
* @see org.springframework.batch.item.ItemStream#update(ExecutionContext)
*/
@Override
@Override
public void update(ExecutionContext executionContext) {
}
protected void setExecutionContextName(String name) {
executionContextUserSupport.setName(name);
}
public String getExecutionContextKey(String key) {
return executionContextUserSupport.getKey(key);
}
}

20
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/JdbcPagingItemReader.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2006-2012 the original author or authors.
* Copyright 2006-2013 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.
@@ -222,7 +222,7 @@ public class JdbcPagingItemReader<T> extends AbstractPagingItemReader<T> impleme
getParameterList(parameterValues, startAfterValues).toArray(), rowCallback);
}
}
Collection<T> result = (Collection<T>) query;
results.addAll(result);
}
@@ -231,7 +231,7 @@ public class JdbcPagingItemReader<T> extends AbstractPagingItemReader<T> impleme
public void update(ExecutionContext executionContext) throws ItemStreamException {
super.update(executionContext);
if (isSaveState() && startAfterValues != null) {
executionContext.put(getExecutionContextUserSupport().getKey(START_AFTER_VALUE), startAfterValues);
executionContext.put(getExecutionContextKey(START_AFTER_VALUE), startAfterValues);
}
}
@@ -239,13 +239,13 @@ public class JdbcPagingItemReader<T> extends AbstractPagingItemReader<T> impleme
@SuppressWarnings("unchecked")
public void open(ExecutionContext executionContext) {
if (isSaveState()) {
startAfterValues = (Map<String, Object>) executionContext.get(getExecutionContextUserSupport().getKey(START_AFTER_VALUE));
startAfterValues = (Map<String, Object>) executionContext.get(getExecutionContextKey(START_AFTER_VALUE));
if(startAfterValues == null) {
startAfterValues = new LinkedHashMap<String, Object>();
}
}
super.open(executionContext);
}
@@ -266,7 +266,7 @@ public class JdbcPagingItemReader<T> extends AbstractPagingItemReader<T> impleme
}
RowMapper startMapper = new RowMapper() {
@Override
@Override
public Object mapRow(ResultSet rs, int i) throws SQLException {
return rs.getObject(1);
}
@@ -316,7 +316,7 @@ public class JdbcPagingItemReader<T> extends AbstractPagingItemReader<T> impleme
parameterList.add(keys.get(i).getValue());
}
}
if (logger.isDebugEnabled()) {
logger.debug("Using parameterList:" + parameterList);
}
@@ -325,13 +325,13 @@ public class JdbcPagingItemReader<T> extends AbstractPagingItemReader<T> impleme
@SuppressWarnings("rawtypes")
private class PagingRowMapper implements RowMapper {
@Override
@Override
public Object mapRow(ResultSet rs, int rowNum) throws SQLException {
startAfterValues = new LinkedHashMap<String, Object>();
for (Map.Entry<String, Order> sortKey : queryProvider.getSortKeys().entrySet()) {
startAfterValues.put(sortKey.getKey(), rs.getObject(sortKey.getKey()));
}
return rowMapper.mapRow(rs, rowNum);
}
}

48
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/FlatFileItemWriter.java
View File

@@ -34,7 +34,7 @@ import org.springframework.batch.item.ItemStreamException;
import org.springframework.batch.item.WriteFailedException;
import org.springframework.batch.item.WriterNotOpenException;
import org.springframework.batch.item.file.transform.LineAggregator;
import org.springframework.batch.item.util.ExecutionContextUserSupport;
import org.springframework.batch.item.support.AbstractItemStreamItemWriter;
import org.springframework.batch.item.util.FileUtils;
import org.springframework.batch.support.transaction.TransactionAwareBufferedWriter;
import org.springframework.beans.factory.InitializingBean;
@@ -57,8 +57,8 @@ import org.springframework.util.ClassUtils;
* @author Dave Syer
* @author Michael Minella
*/
public class FlatFileItemWriter<T> extends ExecutionContextUserSupport implements ResourceAwareItemWriterItemStream<T>,
InitializingBean {
public class FlatFileItemWriter<T> extends AbstractItemStreamItemWriter<T> implements ResourceAwareItemWriterItemStream<T>,
InitializingBean {
private static final boolean DEFAULT_TRANSACTIONAL = true;
@@ -97,7 +97,7 @@ public class FlatFileItemWriter<T> extends ExecutionContextUserSupport implement
private boolean append = false;
public FlatFileItemWriter() {
setName(ClassUtils.getShortName(FlatFileItemWriter.class));
this.setExecutionContextName(ClassUtils.getShortName(FlatFileItemWriter.class));
}
/**
@@ -105,7 +105,7 @@ public class FlatFileItemWriter<T> extends ExecutionContextUserSupport implement
*
* @see org.springframework.beans.factory.InitializingBean#afterPropertiesSet()
*/
@Override
@Override
public void afterPropertiesSet() throws Exception {
Assert.notNull(lineAggregator, "A LineAggregator must be provided.");
if (append) {
@@ -150,7 +150,7 @@ public class FlatFileItemWriter<T> extends ExecutionContextUserSupport implement
*
* @param resource
*/
@Override
@Override
public void setResource(Resource resource) {
this.resource = resource;
}
@@ -248,7 +248,7 @@ public class FlatFileItemWriter<T> extends ExecutionContextUserSupport implement
* @throws Exception if the transformer or file output fail,
* WriterNotOpenException if the writer has not been initialized.
*/
@Override
@Override
public void write(List<? extends T> items) throws Exception {
if (!getOutputState().isInitialized()) {
@@ -279,8 +279,9 @@ public class FlatFileItemWriter<T> extends ExecutionContextUserSupport implement
/**
* @see ItemStream#close()
*/
@Override
@Override
public void close() {
super.close();
if (state != null) {
try {
if (footerCallback != null && state.outputBufferedWriter != null) {
@@ -312,8 +313,9 @@ public class FlatFileItemWriter<T> extends ExecutionContextUserSupport implement
*
* @see ItemStream#open(ExecutionContext)
*/
@Override
@Override
public void open(ExecutionContext executionContext) throws ItemStreamException {
super.open(executionContext);
Assert.notNull(resource, "The resource must be set");
@@ -324,7 +326,7 @@ public class FlatFileItemWriter<T> extends ExecutionContextUserSupport implement
private void doOpen(ExecutionContext executionContext) throws ItemStreamException {
OutputState outputState = getOutputState();
if (executionContext.containsKey(getKey(RESTART_DATA_NAME))) {
if (executionContext.containsKey(getExecutionContextKey(RESTART_DATA_NAME))) {
outputState.restoreFrom(executionContext);
}
try {
@@ -349,8 +351,9 @@ public class FlatFileItemWriter<T> extends ExecutionContextUserSupport implement
/**
* @see ItemStream#update(ExecutionContext)
*/
@Override
@Override
public void update(ExecutionContext executionContext) {
super.update(executionContext);
if (state == null) {
throw new ItemStreamException("ItemStream not open or already closed.");
}
@@ -360,13 +363,13 @@ public class FlatFileItemWriter<T> extends ExecutionContextUserSupport implement
if (saveState) {
try {
executionContext.putLong(getKey(RESTART_DATA_NAME), state.position());
executionContext.putLong(getExecutionContextKey(RESTART_DATA_NAME), state.position());
}
catch (IOException e) {
throw new ItemStreamException("ItemStream does not return current position properly", e);
}
executionContext.putLong(getKey(WRITTEN_STATISTICS_NAME), state.linesWritten);
executionContext.putLong(getExecutionContextKey(WRITTEN_STATISTICS_NAME), state.linesWritten);
}
}
@@ -386,7 +389,18 @@ public class FlatFileItemWriter<T> extends ExecutionContextUserSupport implement
state.setAppendAllowed(append);
state.setEncoding(encoding);
}
return (OutputState) state;
return state;
}
/**
* The name of the component which will be used as a stem for keys in the
* {@link ExecutionContext}. Subclasses should provide a default value, e.g.
* the short form of the class name.
*
* @param name the name for the component
*/
public void setName(String name) {
this.setExecutionContextName(name);
}
/**
@@ -454,7 +468,7 @@ public class FlatFileItemWriter<T> extends ExecutionContextUserSupport implement
* @param executionContext
*/
public void restoreFrom(ExecutionContext executionContext) {
lastMarkedByteOffsetPosition = executionContext.getLong(getKey(RESTART_DATA_NAME));
lastMarkedByteOffsetPosition = executionContext.getLong(getExecutionContextKey(RESTART_DATA_NAME));
restarted = true;
}
@@ -587,12 +601,12 @@ public class FlatFileItemWriter<T> extends ExecutionContextUserSupport implement
final FileChannel channel = fileChannel;
if (transactional) {
TransactionAwareBufferedWriter writer = new TransactionAwareBufferedWriter(channel, new Runnable() {
@Override
@Override
public void run() {
closeStream();
}
});
writer.setEncoding(encoding);
return writer;
}

50
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/MultiResourceItemReader.java
View File

@@ -22,12 +22,12 @@ import java.util.Comparator;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.batch.item.ItemStream;
import org.springframework.batch.item.ItemStreamException;
import org.springframework.batch.item.ItemStreamReader;
import org.springframework.batch.item.ParseException;
import org.springframework.batch.item.ResourceAware;
import org.springframework.batch.item.UnexpectedInputException;
import org.springframework.batch.item.util.ExecutionContextUserSupport;
import org.springframework.batch.item.support.AbstractItemStreamItemReader;
import org.springframework.core.io.Resource;
import org.springframework.util.Assert;
import org.springframework.util.ClassUtils;
@@ -43,14 +43,12 @@ import org.springframework.util.ClassUtils;
* @author Robert Kasanicky
* @author Lucas Ward
*/
public class MultiResourceItemReader<T> implements ItemStreamReader<T> {
public class MultiResourceItemReader<T> extends AbstractItemStreamItemReader<T> {
private static final Log logger = LogFactory.getLog(MultiResourceItemReader.class);
private static final String RESOURCE_KEY = "resourceIndex";
private final ExecutionContextUserSupport executionContextUserSupport = new ExecutionContextUserSupport();
private ResourceAwareItemReaderItemStream<? extends T> delegate;
private Resource[] resources;
@@ -78,7 +76,7 @@ public class MultiResourceItemReader<T> implements ItemStreamReader<T> {
/**
* Compares resource filenames.
*/
@Override
@Override
public int compare(Resource r1, Resource r2) {
return r1.getFilename().compareTo(r2.getFilename());
}
@@ -86,13 +84,13 @@ public class MultiResourceItemReader<T> implements ItemStreamReader<T> {
};
public MultiResourceItemReader() {
executionContextUserSupport.setName(ClassUtils.getShortName(MultiResourceItemReader.class));
this.setExecutionContextName(ClassUtils.getShortName(MultiResourceItemReader.class));
}
/**
* Reads the next item, jumping to next resource if necessary.
*/
@Override
@Override
public T read() throws Exception, UnexpectedInputException, ParseException {
if (noInput) {
@@ -132,25 +130,26 @@ public class MultiResourceItemReader<T> implements ItemStreamReader<T> {
delegate.setResource(resources[currentResource]);
delegate.open(new ExecutionContext());
item = readFromDelegate();
}
item = readFromDelegate();
}
return item;
}
private T readFromDelegate() throws Exception {
T item = delegate.read();
if(item instanceof ResourceAware){
((ResourceAware) item).setResource(getCurrentResource());
}
return item;
}
private T readFromDelegate() throws Exception {
T item = delegate.read();
if(item instanceof ResourceAware){
((ResourceAware) item).setResource(getCurrentResource());
}
return item;
}
/**
/**
* Close the {@link #setDelegate(ResourceAwareItemReaderItemStream)} reader and reset instance variable values.
*/
@Override
@Override
public void close() throws ItemStreamException {
super.close();
delegate.close();
noInput = false;
}
@@ -159,9 +158,9 @@ public class MultiResourceItemReader<T> implements ItemStreamReader<T> {
* Figure out which resource to start with in case of restart, open the delegate and restore delegate's position in
* the resource.
*/
@Override
@Override
public void open(ExecutionContext executionContext) throws ItemStreamException {
super.open(executionContext);
Assert.notNull(resources, "Resources must be set");
noInput = false;
@@ -179,8 +178,8 @@ public class MultiResourceItemReader<T> implements ItemStreamReader<T> {
Arrays.sort(resources, comparator);
if (executionContext.containsKey(executionContextUserSupport.getKey(RESOURCE_KEY))) {
currentResource = executionContext.getInt(executionContextUserSupport.getKey(RESOURCE_KEY));
if (executionContext.containsKey(getExecutionContextKey(RESOURCE_KEY))) {
currentResource = executionContext.getInt(getExecutionContextKey(RESOURCE_KEY));
// context could have been saved before reading anything
if (currentResource == -1) {
@@ -198,10 +197,11 @@ public class MultiResourceItemReader<T> implements ItemStreamReader<T> {
/**
* Store the current resource index and position in the resource.
*/
@Override
@Override
public void update(ExecutionContext executionContext) throws ItemStreamException {
super.update(executionContext);
if (saveState) {
executionContext.putInt(executionContextUserSupport.getKey(RESOURCE_KEY), currentResource);
executionContext.putInt(getExecutionContextKey(RESOURCE_KEY), currentResource);
delegate.update(executionContext);
}
}

20
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/MultiResourceItemWriter.java
View File

@@ -21,8 +21,7 @@ import java.io.IOException;
import java.util.List;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.batch.item.ItemStreamException;
import org.springframework.batch.item.ItemStreamWriter;
import org.springframework.batch.item.util.ExecutionContextUserSupport;
import org.springframework.batch.item.support.AbstractItemStreamItemWriter;
import org.springframework.core.io.FileSystemResource;
import org.springframework.core.io.Resource;
import org.springframework.util.Assert;
@@ -42,7 +41,7 @@ import org.springframework.util.ClassUtils;
*
* @author Robert Kasanicky
*/
public class MultiResourceItemWriter<T> extends ExecutionContextUserSupport implements ItemStreamWriter<T> {
public class MultiResourceItemWriter<T> extends AbstractItemStreamItemWriter<T> {
final static private String RESOURCE_INDEX_KEY = "resource.index";
@@ -65,7 +64,7 @@ public class MultiResourceItemWriter<T> extends ExecutionContextUserSupport impl
private boolean opened = false;
public MultiResourceItemWriter() {
setName(ClassUtils.getShortName(MultiResourceItemWriter.class));
this.setExecutionContextName(ClassUtils.getShortName(MultiResourceItemWriter.class));
}
@Override
@@ -128,6 +127,7 @@ public class MultiResourceItemWriter<T> extends ExecutionContextUserSupport impl
@Override
public void close() throws ItemStreamException {
super.close();
resourceIndex = 1;
currentResourceItemCount = 0;
if (opened) {
@@ -137,8 +137,9 @@ public class MultiResourceItemWriter<T> extends ExecutionContextUserSupport impl
@Override
public void open(ExecutionContext executionContext) throws ItemStreamException {
resourceIndex = executionContext.getInt(getKey(RESOURCE_INDEX_KEY), 1);
currentResourceItemCount = executionContext.getInt(getKey(CURRENT_RESOURCE_ITEM_COUNT), 0);
super.open(executionContext);
resourceIndex = executionContext.getInt(getExecutionContextKey(RESOURCE_INDEX_KEY), 1);
currentResourceItemCount = executionContext.getInt(getExecutionContextKey(CURRENT_RESOURCE_ITEM_COUNT), 0);
try {
setResourceToDelegate();
@@ -147,7 +148,7 @@ public class MultiResourceItemWriter<T> extends ExecutionContextUserSupport impl
throw new ItemStreamException("Couldn't assign resource", e);
}
if (executionContext.containsKey(getKey(CURRENT_RESOURCE_ITEM_COUNT))) {
if (executionContext.containsKey(getExecutionContextKey(CURRENT_RESOURCE_ITEM_COUNT))) {
// It's a restart
delegate.open(executionContext);
}
@@ -158,12 +159,13 @@ public class MultiResourceItemWriter<T> extends ExecutionContextUserSupport impl
@Override
public void update(ExecutionContext executionContext) throws ItemStreamException {
super.update(executionContext);
if (saveState) {
if (opened) {
delegate.update(executionContext);
}
executionContext.putInt(getKey(CURRENT_RESOURCE_ITEM_COUNT), currentResourceItemCount);
executionContext.putInt(getKey(RESOURCE_INDEX_KEY), resourceIndex);
executionContext.putInt(getExecutionContextKey(CURRENT_RESOURCE_ITEM_COUNT), currentResourceItemCount);
executionContext.putInt(getExecutionContextKey(RESOURCE_INDEX_KEY), resourceIndex);
}
}

19
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/ResourcesItemReader.java
View File

@@ -6,8 +6,7 @@ import java.util.concurrent.atomic.AtomicInteger;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemStreamException;
import org.springframework.batch.item.ItemStreamReader;
import org.springframework.batch.item.util.ExecutionContextUserSupport;
import org.springframework.batch.item.support.AbstractItemStreamItemReader;
import org.springframework.core.io.Resource;
import org.springframework.core.io.support.ResourceArrayPropertyEditor;
@@ -30,17 +29,17 @@ import org.springframework.core.io.support.ResourceArrayPropertyEditor;
*
* @since 2.1
*/
public class ResourcesItemReader extends ExecutionContextUserSupport implements ItemStreamReader<Resource> {
public class ResourcesItemReader extends AbstractItemStreamItemReader<Resource> {
private Resource[] resources = new Resource[0];
private AtomicInteger counter = new AtomicInteger(0);
{
public ResourcesItemReader() {
/*
* Initialize the name for the key in the execution context.
*/
setName(getClass().getName());
this.setExecutionContextName(getClass().getName());
}
/**
@@ -65,18 +64,16 @@ public class ResourcesItemReader extends ExecutionContextUserSupport implements
return resources[index];
}
@Override
public void close() throws ItemStreamException {
}
@Override
public void open(ExecutionContext executionContext) throws ItemStreamException {
counter.set(executionContext.getInt(getKey("COUNT"), 0));
super.open(executionContext);
counter.set(executionContext.getInt(getExecutionContextKey("COUNT"), 0));
}
@Override
public void update(ExecutionContext executionContext) throws ItemStreamException {
executionContext.putInt(getKey("COUNT"), counter.get());
super.update(executionContext);
executionContext.putInt(getExecutionContextKey("COUNT"), counter.get());
}
}

28
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/AbstractItemCountingItemStreamItemReader.java
View File

@@ -19,10 +19,8 @@ package org.springframework.batch.item.support;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemStreamException;
import org.springframework.batch.item.ItemStreamReader;
import org.springframework.batch.item.ParseException;
import org.springframework.batch.item.UnexpectedInputException;
import org.springframework.batch.item.util.ExecutionContextUserSupport;
import org.springframework.util.Assert;
/**
@@ -34,7 +32,7 @@ import org.springframework.util.Assert;
*
* @author Robert Kasanicky
*/
public abstract class AbstractItemCountingItemStreamItemReader<T> implements ItemStreamReader<T> {
public abstract class AbstractItemCountingItemStreamItemReader<T> extends AbstractItemStreamItemReader<T> {
private static final String READ_COUNT = "read.count";
@@ -44,8 +42,6 @@ public abstract class AbstractItemCountingItemStreamItemReader<T> implements Ite
private int maxItemCount = Integer.MAX_VALUE;
private ExecutionContextUserSupport ecSupport = new ExecutionContextUserSupport();
private boolean saveState = true;
/**
@@ -121,6 +117,7 @@ public abstract class AbstractItemCountingItemStreamItemReader<T> implements Ite
@Override
public void close() throws ItemStreamException {
super.close();
currentItemCount = 0;
try {
doClose();
@@ -132,7 +129,7 @@ public abstract class AbstractItemCountingItemStreamItemReader<T> implements Ite
@Override
public void open(ExecutionContext executionContext) throws ItemStreamException {
super.open(executionContext);
try {
doOpen();
}
@@ -143,12 +140,12 @@ public abstract class AbstractItemCountingItemStreamItemReader<T> implements Ite
return;
}
if (executionContext.containsKey(ecSupport.getKey(READ_COUNT_MAX))) {
maxItemCount = executionContext.getInt(ecSupport.getKey(READ_COUNT_MAX));
if (executionContext.containsKey(getExecutionContextKey(READ_COUNT_MAX))) {
maxItemCount = executionContext.getInt(getExecutionContextKey(READ_COUNT_MAX));
}
if (executionContext.containsKey(ecSupport.getKey(READ_COUNT))) {
int itemCount = executionContext.getInt(ecSupport.getKey(READ_COUNT));
if (executionContext.containsKey(getExecutionContextKey(READ_COUNT))) {
int itemCount = executionContext.getInt(getExecutionContextKey(READ_COUNT));
if (itemCount < maxItemCount) {
try {
@@ -166,20 +163,17 @@ public abstract class AbstractItemCountingItemStreamItemReader<T> implements Ite
@Override
public void update(ExecutionContext executionContext) throws ItemStreamException {
super.update(executionContext);
if (saveState) {
Assert.notNull(executionContext, "ExecutionContext must not be null");
executionContext.putInt(ecSupport.getKey(READ_COUNT), currentItemCount);
executionContext.putInt(getExecutionContextKey(READ_COUNT), currentItemCount);
if (maxItemCount < Integer.MAX_VALUE) {
executionContext.putInt(ecSupport.getKey(READ_COUNT_MAX), maxItemCount);
executionContext.putInt(getExecutionContextKey(READ_COUNT_MAX), maxItemCount);
}
}
}
protected ExecutionContextUserSupport getExecutionContextUserSupport() {
return ecSupport;
}
/**
* The name of the component which will be used as a stem for keys in the
* {@link ExecutionContext}. Subclasses should provide a default value, e.g.
@@ -188,7 +182,7 @@ public abstract class AbstractItemCountingItemStreamItemReader<T> implements Ite
* @param name the name for the component
*/
public void setName(String name) {
ecSupport.setName(name);
this.setExecutionContextName(name);
}
/**

22
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/xml/StaxEventItemWriter.java
View File

@@ -41,7 +41,7 @@ import org.springframework.batch.item.ItemStreamException;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.WriteFailedException;
import org.springframework.batch.item.file.ResourceAwareItemWriterItemStream;
import org.springframework.batch.item.util.ExecutionContextUserSupport;
import org.springframework.batch.item.support.AbstractItemStreamItemWriter;
import org.springframework.batch.item.util.FileUtils;
import org.springframework.batch.item.xml.stax.NoStartEndDocumentStreamWriter;
import org.springframework.batch.support.transaction.TransactionAwareBufferedWriter;
@@ -69,7 +69,7 @@ import org.springframework.util.StringUtils;
* @author Michael Minella
*
*/
public class StaxEventItemWriter<T> extends ExecutionContextUserSupport implements
public class StaxEventItemWriter<T> extends AbstractItemStreamItemWriter<T> implements
ResourceAwareItemWriterItemStream<T>, InitializingBean {
private static final Log log = LogFactory.getLog(StaxEventItemWriter.class);
@@ -143,7 +143,7 @@ public class StaxEventItemWriter<T> extends ExecutionContextUserSupport implemen
private boolean forceSync;
public StaxEventItemWriter() {
setName(ClassUtils.getShortName(StaxEventItemWriter.class));
setExecutionContextName(ClassUtils.getShortName(StaxEventItemWriter.class));
}
/**
@@ -343,7 +343,8 @@ public class StaxEventItemWriter<T> extends ExecutionContextUserSupport implemen
*/
@Override
public void open(ExecutionContext executionContext) {
super.open(executionContext);
Assert.notNull(resource, "The resource must be set");
long startAtPosition = 0;
@@ -351,8 +352,8 @@ public class StaxEventItemWriter<T> extends ExecutionContextUserSupport implemen
// if restart data is provided, restart from provided offset
// otherwise start from beginning
if (executionContext.containsKey(getKey(RESTART_DATA_NAME))) {
startAtPosition = executionContext.getLong(getKey(RESTART_DATA_NAME));
if (executionContext.containsKey(getExecutionContextKey(RESTART_DATA_NAME))) {
startAtPosition = executionContext.getLong(getExecutionContextKey(RESTART_DATA_NAME));
restarted = true;
}
@@ -576,7 +577,8 @@ public class StaxEventItemWriter<T> extends ExecutionContextUserSupport implemen
*/
@Override
public void close() {
super.close();
XMLEventFactory factory = createXmlEventFactory();
try {
delegateEventWriter.add(factory.createCharacters(""));
@@ -665,11 +667,11 @@ public class StaxEventItemWriter<T> extends ExecutionContextUserSupport implemen
*/
@Override
public void update(ExecutionContext executionContext) {
super.update(executionContext);
if (saveState) {
Assert.notNull(executionContext, "ExecutionContext must not be null");
executionContext.putLong(getKey(RESTART_DATA_NAME), getPosition());
executionContext.putLong(getKey(WRITE_STATISTICS_NAME), currentRecordCount);
executionContext.putLong(getExecutionContextKey(RESTART_DATA_NAME), getPosition());
executionContext.putLong(getExecutionContextKey(WRITE_STATISTICS_NAME), currentRecordCount);
}
}

8
spring-batch-infrastructure/src/test/java/org/springframework/batch/item/file/MultiResourceItemWriterFlatFileTests.java
View File

@@ -86,12 +86,12 @@ public class MultiResourceItemWriterFlatFileTests extends AbstractMultiResourceI
super.setUp(delegate);
tested.update(executionContext);
assertEquals(0, executionContext.getInt(tested.getKey("resource.item.count")));
assertEquals(1, executionContext.getInt(tested.getKey("resource.index")));
assertEquals(0, executionContext.getInt(tested.getExecutionContextKey("resource.item.count")));
assertEquals(1, executionContext.getInt(tested.getExecutionContextKey("resource.index")));
tested.write(Arrays.asList("1", "2", "3"));
tested.update(executionContext);
assertEquals(0, executionContext.getInt(tested.getKey("resource.item.count")));
assertEquals(2, executionContext.getInt(tested.getKey("resource.index")));
assertEquals(0, executionContext.getInt(tested.getExecutionContextKey("resource.item.count")));
assertEquals(2, executionContext.getInt(tested.getExecutionContextKey("resource.index")));
}

4
spring-batch-infrastructure/src/test/java/org/springframework/batch/item/file/ResourcesItemReaderTests.java
View File

@@ -34,7 +34,7 @@ public class ResourcesItemReaderTests {
@Test
public void testReadAfterOpen() throws Exception {
ExecutionContext executionContext = new ExecutionContext();
executionContext.putInt(reader.getKey("COUNT"), 1);
executionContext.putInt(reader.getExecutionContextKey("COUNT"), 1);
reader.open(executionContext);
assertNotNull(reader.read());
assertNull(reader.read());
@@ -46,7 +46,7 @@ public class ResourcesItemReaderTests {
assertNotNull(reader.read());
reader.update(executionContext);
assertEquals(1, executionContext.getInt(reader.getKey("COUNT")));
assertEquals(1, executionContext.getInt(reader.getExecutionContextKey("COUNT")));
}
}

21
spring-batch-infrastructure/src/test/java/org/springframework/batch/item/support/CompositeItemStreamTests.java
View File

@@ -38,8 +38,9 @@ public class CompositeItemStreamTests extends TestCase {
public void testRegisterAndOpen() {
ItemStreamSupport stream = new ItemStreamSupport() {
@Override
public void open(ExecutionContext executionContext) throws ItemStreamException {
@Override
public void open(ExecutionContext executionContext) {
super.open(executionContext);
list.add("bar");
}
};
@@ -51,7 +52,8 @@ public class CompositeItemStreamTests extends TestCase {
public void testRegisterTwice() {
ItemStreamSupport stream = new ItemStreamSupport() {
@Override
public void open(ExecutionContext executionContext) throws ItemStreamException {
public void open(ExecutionContext executionContext) {
super.open(executionContext);
list.add("bar");
}
};
@@ -63,8 +65,9 @@ public class CompositeItemStreamTests extends TestCase {
public void testMark() {
manager.register(new ItemStreamSupport() {
@Override
@Override
public void update(ExecutionContext executionContext) {
super.update(executionContext);
list.add("bar");
}
});
@@ -74,8 +77,9 @@ public class CompositeItemStreamTests extends TestCase {
public void testClose() {
manager.register(new ItemStreamSupport() {
@Override
public void close() throws ItemStreamException {
@Override
public void close() {
super.close();
list.add("bar");
}
});
@@ -85,8 +89,9 @@ public class CompositeItemStreamTests extends TestCase {
public void testCloseDoesNotUnregister() {
manager.setStreams(new ItemStream[] { new ItemStreamSupport() {
@Override
public void open(ExecutionContext executionContext) throws ItemStreamException {
@Override
public void open(ExecutionContext executionContext) {
super.open(executionContext);
list.add("bar");
}
} });