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

Added ExecutionContextUserSupport to the ItemStream Support class

Browse Source 
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:
Dean de Bree
2013-02-27 08:52:36 +02:00
committed by Michael Minella
parent b1bd3d3d1e
commit d1b84f3e2f
21 changed files with 168 additions and 142 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -15,21 +15,25 @@
*/
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 {
public void close() {
}
/**
@@ -37,7 +41,7 @@ public abstract class ItemStreamSupport implements ItemStream {
* @see org.springframework.batch.item.ItemStream#open(ExecutionContext)
*/
@Override
public void open(ExecutionContext executionContext) throws ItemStreamException {
public void open(ExecutionContext executionContext) {
}
/**
@@ -47,5 +51,17 @@ public abstract class ItemStreamSupport implements ItemStream {
@Override
public void update(ExecutionContext executionContext) {
}
private ExecutionContextUserSupport getExecutionContextUserSupport() {
return executionContextUserSupport;
}
protected void setExecutionContextName(String name) {
this.getExecutionContextUserSupport().setName(name);
}
public String getExecutionContextKey(String key) {
return this.getExecutionContextUserSupport().getKey(key);
}
}

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

@@ -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,7 +239,7 @@ 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>();

19
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,7 +57,7 @@ import org.springframework.util.ClassUtils;
* @author Dave Syer
* @author Michael Minella
*/
public class FlatFileItemWriter<T> extends ExecutionContextUserSupport implements ResourceAwareItemWriterItemStream<T>,
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));
}
/**
@@ -281,6 +281,7 @@ public class FlatFileItemWriter<T> extends ExecutionContextUserSupport implement
*/
@Override
public void close() {
super.close();
if (state != null) {
try {
if (footerCallback != null && state.outputBufferedWriter != null) {
@@ -314,7 +315,8 @@ public class FlatFileItemWriter<T> extends ExecutionContextUserSupport implement
*/
@Override
public void open(ExecutionContext executionContext) throws ItemStreamException {
super.open(executionContext);
Assert.notNull(resource, "The resource must be set");
if (!getOutputState().isInitialized()) {
@@ -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 {
@@ -351,6 +353,7 @@ public class FlatFileItemWriter<T> extends ExecutionContextUserSupport implement
*/
@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);
}
}
@@ -454,7 +457,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;
}

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

@@ -23,11 +23,10 @@ import org.apache.commons.logging.Log;
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.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 +42,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;
@@ -86,7 +83,7 @@ public class MultiResourceItemReader<T> implements ItemStreamReader<T> {
};
public MultiResourceItemReader() {
executionContextUserSupport.setName(ClassUtils.getShortName(MultiResourceItemReader.class));
this.setExecutionContextName(ClassUtils.getShortName(MultiResourceItemReader.class));
}
/**
@@ -151,6 +148,7 @@ public class MultiResourceItemReader<T> implements ItemStreamReader<T> {
*/
@Override
public void close() throws ItemStreamException {
super.close();
delegate.close();
noInput = false;
}
@@ -161,7 +159,7 @@ public class MultiResourceItemReader<T> implements ItemStreamReader<T> {
*/
@Override
public void open(ExecutionContext executionContext) throws ItemStreamException {
super.open(executionContext);
Assert.notNull(resources, "Resources must be set");
noInput = false;
@@ -179,8 +177,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) {
@@ -200,8 +198,9 @@ public class MultiResourceItemReader<T> implements ItemStreamReader<T> {
*/
@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);
}
/**

3
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/AbstractItemStreamItemReader.java
View File

@@ -17,6 +17,7 @@
package org.springframework.batch.item.support;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemStreamReader;
import org.springframework.batch.item.ItemStreamSupport;
@@ -25,6 +26,6 @@ import org.springframework.batch.item.ItemStreamSupport;
* @author Dave Syer
*
*/
public abstract class AbstractItemStreamItemReader<T> extends ItemStreamSupport implements ItemReader<T> {
public abstract class AbstractItemStreamItemReader<T> extends ItemStreamSupport implements ItemStreamReader<T> {
}

3
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/AbstractItemStreamItemWriter.java
View File

@@ -17,6 +17,7 @@
package org.springframework.batch.item.support;
import org.springframework.batch.item.ItemStreamSupport;
import org.springframework.batch.item.ItemStreamWriter;
import org.springframework.batch.item.ItemWriter;
@@ -25,6 +26,6 @@ import org.springframework.batch.item.ItemWriter;
* @author Dave Syer
*
*/
public abstract class AbstractItemStreamItemWriter<T> extends ItemStreamSupport implements ItemWriter<T> {
public abstract class AbstractItemStreamItemWriter<T> extends ItemStreamSupport implements ItemStreamWriter<T> {
}

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);
}
}