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

BATCH-219 and BATCH-330: Created ExecutionAttributes, which are stored in the table BATCH_STEP_EXECUTION_ATTRIBUTES.

Browse Source
This commit is contained in:
lucasward
2008-02-05 02:33:21 +00:00
parent 96a11b3bb7
commit 12345beaf7
38 changed files with 254 additions and 227 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
spring-batch-infrastructure/src/main/java/org/springframework/batch/io/cursor/HibernateCursorItemReader.java
View File

@@ -25,8 +25,7 @@ import org.hibernate.SessionFactory;
import org.hibernate.StatelessSession;
import org.springframework.batch.io.Skippable;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemStream;
import org.springframework.batch.item.StreamContext;
import org.springframework.batch.item.ExecutionAttributes;
import org.springframework.batch.item.reader.AbstractItemStreamItemReader;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
@@ -175,18 +174,21 @@ public class HibernateCursorItemReader extends AbstractItemStreamItemReader impl
/**
* @return the current row number wrapped as <code>StreamContext</code>
*/
public StreamContext getStreamContext() {
public ExecutionAttributes getStreamContext() {
Properties props = new Properties();
props.setProperty(RESTART_DATA_ROW_NUMBER_KEY, "" + currentProcessedRow);
ExecutionAttributes executionAttributes = new ExecutionAttributes();
executionAttributes.putString(RESTART_DATA_ROW_NUMBER_KEY, "" + currentProcessedRow);
String skipped = skippedRows.toString();
props.setProperty(SKIPPED_ROWS, skipped.substring(1, skipped.length() - 1));
return new StreamContext(props);
executionAttributes.putString(SKIPPED_ROWS, skipped.substring(1, skipped.length() - 1));
return executionAttributes;
}
/**
* Sets the cursor to the received row number.
*/
public void restoreFrom(StreamContext data) {
public void restoreFrom(ExecutionAttributes data) {
Assert.state(!initialized, "Cannot restore when already intialized. Call close() first before restore()");
Properties props = data.getProperties();
@@ -234,7 +236,7 @@ public class HibernateCursorItemReader extends AbstractItemStreamItemReader impl
* (non-Javadoc)
* @see org.springframework.batch.item.ItemStream#mark(org.springframework.batch.item.StreamContext)
*/
public void mark(StreamContext streamContext) {
public void mark(ExecutionAttributes streamContext) {
lastCommitRowNumber = currentProcessedRow;
if (!useStatelessSession) {
statefulSession.clear();
@@ -245,7 +247,7 @@ public class HibernateCursorItemReader extends AbstractItemStreamItemReader impl
* (non-Javadoc)
* @see org.springframework.batch.item.ItemStream#reset(org.springframework.batch.item.StreamContext)
*/
public void reset(StreamContext streamContext) {
public void reset(ExecutionAttributes streamContext) {
currentProcessedRow = lastCommitRowNumber;
if (lastCommitRowNumber == 0) {
cursor.beforeFirst();

15
spring-batch-infrastructure/src/main/java/org/springframework/batch/io/cursor/JdbcCursorItemReader.java
View File

@@ -32,7 +32,8 @@ import org.springframework.batch.io.Skippable;
import org.springframework.batch.io.support.AbstractTransactionalIoSource;
import org.springframework.batch.item.ItemStream;
import org.springframework.batch.item.KeyedItemReader;
import org.springframework.batch.item.StreamContext;
import org.springframework.batch.item.ExecutionAttributes;
import org.springframework.batch.item.stream.GenericStreamContext;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.dao.DataAccessException;
@@ -71,7 +72,7 @@ import org.springframework.util.StringUtils;
* </p>
*
* <p>
* {@link StreamContext}: The current row is returned as restart data, and when
* {@link ExecutionAttributes}: The current row is returned as restart data, and when
* restored from that same data, the cursor is opened and the current row set to
* the value within the restart data. There are also two statistics returned by
* this input source: the current line being processed and the number of lines
@@ -232,7 +233,7 @@ public class JdbcCursorItemReader extends AbstractTransactionalIoSource implemen
* Mark the current row. Calling reset will cause the result set to be set
* to the current row when mark was called.
*/
public void mark(StreamContext streamContext) {
public void mark(ExecutionAttributes streamContext) {
lastCommittedRow = currentProcessedRow;
skippedRows.clear();
}
@@ -242,7 +243,7 @@ public class JdbcCursorItemReader extends AbstractTransactionalIoSource implemen
*
* @throws DataAccessException
*/
public void reset(StreamContext streamContext) {
public void reset(ExecutionAttributes streamContext) {
try {
currentProcessedRow = lastCommittedRow;
if (currentProcessedRow > 0) {
@@ -390,9 +391,9 @@ public class JdbcCursorItemReader extends AbstractTransactionalIoSource implemen
*
* @see org.springframework.batch.restart.Restartable#getStreamContext()
*/
public StreamContext getStreamContext() {
public ExecutionAttributes getStreamContext() {
String skipped = skippedRows.toString();
StreamContext context = new StreamContext();
ExecutionAttributes context = new GenericStreamContext();
context.putString(SKIPPED_ROWS, skipped.substring(1, skipped.length() - 1));
context.putLong(CURRENT_PROCESSED_ROW, currentProcessedRow);
context.putLong(SKIP_COUNT, skipCount);
@@ -403,7 +404,7 @@ public class JdbcCursorItemReader extends AbstractTransactionalIoSource implemen
* (non-Javadoc)
* @see org.springframework.batch.item.ItemStream#restoreFrom(org.springframework.batch.item.StreamContext)
*/
public void restoreFrom(StreamContext data) {
public void restoreFrom(ExecutionAttributes data) {
Assert.state(!initialized);
if (data == null)

12
spring-batch-infrastructure/src/main/java/org/springframework/batch/io/driving/DrivingQueryItemReader.java
View File

@@ -21,7 +21,7 @@ import java.util.List;
import org.springframework.batch.io.support.AbstractTransactionalIoSource;
import org.springframework.batch.item.ItemStream;
import org.springframework.batch.item.KeyedItemReader;
import org.springframework.batch.item.StreamContext;
import org.springframework.batch.item.ExecutionAttributes;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.util.Assert;
@@ -159,7 +159,7 @@ public class DrivingQueryItemReader extends AbstractTransactionalIoSource
* been initialized before calling restore (meaning, read has been called)
* then an IllegalStateException will be thrown, since all input sources
* should be restored before being read from, otherwise already processed
* data could be returned. The {@link StreamContext} attempting to be restored from
* data could be returned. The {@link ExecutionAttributes} attempting to be restored from
* must have been obtained from the <strong>same input source as the one
* being restored from</strong> otherwise it is invalid.
*
@@ -168,7 +168,7 @@ public class DrivingQueryItemReader extends AbstractTransactionalIoSource
* @throws IllegalStateException
* if the input source has already been initialized.
*/
public final void restoreFrom(StreamContext data) {
public final void restoreFrom(ExecutionAttributes data) {
Assert.notNull(data, "StreamContext must not be null.");
Assert.notNull(data.getProperties(),
@@ -189,7 +189,7 @@ public class DrivingQueryItemReader extends AbstractTransactionalIoSource
}
}
public StreamContext getStreamContext() {
public ExecutionAttributes getStreamContext() {
return keyGenerator.getKeyAsStreamContext(getCurrentKey());
}
@@ -237,14 +237,14 @@ public class DrivingQueryItemReader extends AbstractTransactionalIoSource
/* (non-Javadoc)
* @see org.springframework.batch.item.ItemStream#mark(org.springframework.batch.item.StreamContext)
*/
public void mark(StreamContext streamContext) {
public void mark(ExecutionAttributes streamContext) {
lastCommitIndex = currentIndex;
}
/* (non-Javadoc)
* @see org.springframework.batch.item.ItemStream#reset(org.springframework.batch.item.StreamContext)
*/
public void reset(StreamContext streamContext) {
public void reset(ExecutionAttributes streamContext) {
keysIterator = keys.listIterator(lastCommitIndex);
}

6
spring-batch-infrastructure/src/main/java/org/springframework/batch/io/driving/KeyGenerator.java
View File

@@ -2,7 +2,7 @@ package org.springframework.batch.io.driving;
import java.util.List;
import org.springframework.batch.item.StreamContext;
import org.springframework.batch.item.ExecutionAttributes;
/**
* Strategy interface used to generate keys in driving query input.
@@ -24,7 +24,7 @@ public interface KeyGenerator {
* @return a list of keys.
* @throws IllegalArgumentException is streamContext is null.
*/
List restoreKeys(StreamContext streamContext);
List restoreKeys(ExecutionAttributes streamContext);
/**
* Return the provided key as restart data.
@@ -34,5 +34,5 @@ public interface KeyGenerator {
* @throws IllegalArgumentException if key is null.
* @throws IllegalArgumentException if key is an incompatible type.
*/
StreamContext getKeyAsStreamContext(Object key);
ExecutionAttributes getKeyAsStreamContext(Object key);
}

32
spring-batch-infrastructure/src/main/java/org/springframework/batch/io/driving/support/ColumnMapStreamContextRowMapper.java
View File

@@ -12,7 +12,7 @@ import java.util.Map;
import java.util.Properties;
import java.util.Map.Entry;
import org.springframework.batch.item.StreamContext;
import org.springframework.batch.item.ExecutionAttributes;
import org.springframework.core.CollectionFactory;
import org.springframework.jdbc.core.ColumnMapRowMapper;
import org.springframework.jdbc.core.PreparedStatementSetter;
@@ -23,27 +23,23 @@ import org.springframework.util.Assert;
import org.springframework.util.ClassUtils;
/**
* </p>
* Extension of the ColumnMapRowMapper that converts a column map to
* {@link StreamContext} and allows {@link StreamContext} to be converted back
* as a PreparedStatementSetter. This is useful in a restart scenario, as it
* allows for the standard functionality of the {@link ColumnMapRowMapper} to be
* used to create a map representing the columns returned by a query. It should
* be noted that this column ordering is preserved in the map using a link list
* version of Map.
* </p>Extension of the ColumnMapRowMapper that converts a column map to {@link ExecutionAttributes} and allows
* {@link ExecutionAttributes} to be converted back as a PreparedStatementSetter. This is useful in a restart
* scenario, as it allows for the standard functionality of the ColumnMapRowMapper to be used to
* create a map representing the columns returned by a query. It should be noted that this column ordering
* is preserved in the map using a link list version of Map.
*
*
* @author Lucas Ward
* @author Dave Syer
* @see StreamContextRowMapper
*/
public class ColumnMapStreamContextRowMapper extends ColumnMapRowMapper implements StreamContextRowMapper {
public static final String KEY_PREFIX = ClassUtils.getQualifiedName(ColumnMapStreamContextRowMapper.class)
+ ".KEY.";
public PreparedStatementSetter createSetter(StreamContext streamContext) {
public class ColumnMapStreamContextRowMapper extends ColumnMapRowMapper implements StreamContextRowMapper{
public static final String KEY_PREFIX = ClassUtils.getQualifiedName(ColumnMapStreamContextRowMapper.class) + ".KEY.";
public PreparedStatementSetter createSetter(ExecutionAttributes streamContext) {
ColumnMapStreamContext columnData = new ColumnMapStreamContext(streamContext.getProperties());
List columns = new ArrayList();
@@ -56,13 +52,13 @@ public class ColumnMapStreamContextRowMapper extends ColumnMapRowMapper implemen
return new ArgPreparedStatementSetter(columns.toArray());
}
public StreamContext createStreamContext(Object key) {
public ExecutionAttributes createStreamContext(Object key) {
Assert.isInstanceOf(Map.class, key, "Input to create StreamContext must be of type Map.");
Map keys = (Map) key;
return new ColumnMapStreamContext(keys);
}
private static class ColumnMapStreamContext extends StreamContext {
private static class ColumnMapStreamContext extends ExecutionAttributes {
private final Map keys;

13
spring-batch-infrastructure/src/main/java/org/springframework/batch/io/driving/support/IbatisKeyGenerator.java
View File

@@ -5,7 +5,7 @@ import java.util.Properties;
import org.springframework.batch.io.driving.DrivingQueryItemReader;
import org.springframework.batch.io.driving.KeyGenerator;
import org.springframework.batch.item.StreamContext;
import org.springframework.batch.item.ExecutionAttributes;
import org.springframework.orm.ibatis.SqlMapClientTemplate;
import org.springframework.util.Assert;
@@ -43,19 +43,20 @@ public class IbatisKeyGenerator implements KeyGenerator {
*
* @see org.springframework.batch.restart.Restartable#getRestartData()
*/
public StreamContext getKeyAsStreamContext(Object key) {
public ExecutionAttributes getKeyAsStreamContext(Object key) {
Properties props = new Properties();
props.setProperty(RESTART_KEY, key.toString());
return new StreamContext(props);
ExecutionAttributes executionAttributes = new ExecutionAttributes();
executionAttributes.putString(RESTART_KEY, key.toString());
return executionAttributes;
}
/**
* Restore the keys list given the provided restart data.
*
* @see org.springframework.batch.io.driving.DrivingQueryItemReader#restoreKeys(org.springframework.batch.item.StreamContext)
* @see org.springframework.batch.io.driving.DrivingQueryItemReader#restoreKeys(org.springframework.batch.item.ExecutionAttributes)
*/
public List restoreKeys(StreamContext data) {
public List restoreKeys(ExecutionAttributes data) {
Properties props = data.getProperties();
Object key = props.getProperty(RESTART_KEY);

6
spring-batch-infrastructure/src/main/java/org/springframework/batch/io/driving/support/MultipleColumnJdbcKeyGenerator.java
View File

@@ -20,7 +20,7 @@ import java.util.List;
import org.springframework.batch.io.driving.DrivingQueryItemReader;
import org.springframework.batch.io.driving.KeyGenerator;
import org.springframework.batch.item.StreamContext;
import org.springframework.batch.item.ExecutionAttributes;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
@@ -80,7 +80,7 @@ public class MultipleColumnJdbcKeyGenerator implements
/* (non-Javadoc)
* @see org.springframework.batch.io.driving.KeyGenerator#restoreKeys(org.springframework.batch.item.StreamContext)
*/
public List restoreKeys(StreamContext streamContext) {
public List restoreKeys(ExecutionAttributes streamContext) {
Assert.state(keyMapper != null, "KeyMapper must not be null.");
Assert.state(StringUtils.hasText(restartSql), "The RestartQuery must not be null or empty" +
@@ -96,7 +96,7 @@ public class MultipleColumnJdbcKeyGenerator implements
/* (non-Javadoc)
* @see org.springframework.batch.io.driving.KeyGenerator#getKeyAsStreamContext(java.lang.Object)
*/
public StreamContext getKeyAsStreamContext(Object key) {
public ExecutionAttributes getKeyAsStreamContext(Object key) {
Assert.state(keyMapper != null, "Kye mapper must not be null.");
return keyMapper.createStreamContext(key);
}

14
spring-batch-infrastructure/src/main/java/org/springframework/batch/io/driving/support/SingleColumnJdbcKeyGenerator.java
View File

@@ -20,7 +20,7 @@ import java.util.List;
import org.apache.commons.lang.ClassUtils;
import org.springframework.batch.io.driving.KeyGenerator;
import org.springframework.batch.item.StreamContext;
import org.springframework.batch.item.ExecutionAttributes;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.core.RowMapper;
import org.springframework.jdbc.core.SingleColumnRowMapper;
@@ -98,27 +98,27 @@ public class SingleColumnJdbcKeyGenerator implements KeyGenerator {
* @see KeyGenerator#getKeyAsStreamContext(Object)
* @throws IllegalArgumentException if key is null.
*/
public StreamContext getKeyAsStreamContext(Object key) {
public ExecutionAttributes getKeyAsStreamContext(Object key) {
Assert.notNull(key, "The key must not be null.");
StreamContext context = new StreamContext();
ExecutionAttributes context = new ExecutionAttributes();
context.putString(RESTART_KEY, key.toString());
return context;
}
/**
* Return the remaining to be processed for the provided
* {@link StreamContext}. The {@link StreamContext} attempting to be
* {@link ExecutionAttributes}. The {@link ExecutionAttributes} attempting to be
* restored from must have been obtained from the <strong>same
* KeyGenerationStrategy as the one being restored from</strong> otherwise
* it is invalid.
*
* @param StreamContext obtained by calling
* @param ExecutionAttributes obtained by calling
* {@link #getKeyAsStreamContext(Object)} during a previous run.
* @throws IllegalStateException if restart sql statement is null.
* @throws IllegalArgumentException if restart data is null.
* @see KeyGenerator#restoreKeys(org.springframework.batch.item.StreamContext)
* @see KeyGenerator#restoreKeys(org.springframework.batch.item.ExecutionAttributes)
*/
public List restoreKeys(StreamContext streamContext) {
public List restoreKeys(ExecutionAttributes streamContext) {
Assert.notNull(streamContext, "The restart data must not be null.");
Assert.state(StringUtils.hasText(restartSql), "The RestartQuery must not be null or empty"

6
spring-batch-infrastructure/src/main/java/org/springframework/batch/io/driving/support/StreamContextRowMapper.java
View File

@@ -15,7 +15,7 @@
*/
package org.springframework.batch.io.driving.support;
import org.springframework.batch.item.StreamContext;
import org.springframework.batch.item.ExecutionAttributes;
import org.springframework.jdbc.core.PreparedStatementSetter;
import org.springframework.jdbc.core.RowMapper;
@@ -40,7 +40,7 @@ public interface StreamContextRowMapper extends RowMapper {
* @return ResartData representing the composite key.
* @throws IllegalArgumentException if key is null or of an unsupported type.
*/
public StreamContext createStreamContext(Object key);
public ExecutionAttributes createStreamContext(Object key);
/**
* Given the provided restart data, return a PreparedStatementSeter that can
@@ -49,5 +49,5 @@ public interface StreamContextRowMapper extends RowMapper {
* @param streamContext
* @return an array of objects that can be used as arguments to a JdbcTemplate.
*/
public PreparedStatementSetter createSetter(StreamContext streamContext);
public PreparedStatementSetter createSetter(ExecutionAttributes streamContext);
}

14
spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/DefaultFlatFileItemReader.java
View File

@@ -24,7 +24,7 @@ import org.apache.commons.logging.LogFactory;
import org.springframework.batch.io.Skippable;
import org.springframework.batch.io.file.separator.LineReader;
import org.springframework.batch.item.ItemStream;
import org.springframework.batch.item.StreamContext;
import org.springframework.batch.item.ExecutionAttributes;
import org.springframework.batch.item.StreamException;
/**
@@ -61,9 +61,9 @@ public class DefaultFlatFileItemReader extends SimpleFlatFileItemReader implemen
* file and position the buffer reader according to information provided by
* the restart data
*
* @param data {@link StreamContext} information
* @param data {@link ExecutionAttributes} information
*/
public void restoreFrom(StreamContext data) {
public void restoreFrom(ExecutionAttributes data) {
if (data == null || data.getProperties() == null
|| data.getProperties().getProperty(READ_STATISTICS_NAME) == null || getReader() == null) {
@@ -90,11 +90,11 @@ public class DefaultFlatFileItemReader extends SimpleFlatFileItemReader implemen
* current Line Count which can be used to re initialise the batch job in
* case of restart.
*/
public StreamContext getStreamContext() {
public ExecutionAttributes getStreamContext() {
if (reader == null) {
throw new StreamException("ItemStream not open or already closed.");
}
StreamContext streamContext = new StreamContext();
ExecutionAttributes streamContext = new ExecutionAttributes();
streamContext.putLong(READ_STATISTICS_NAME, reader.getCurrentLineCount());
streamContext.putLong(SKIPPED_STATISTICS_NAME, skippedLines.size());
return streamContext;
@@ -116,7 +116,7 @@ public class DefaultFlatFileItemReader extends SimpleFlatFileItemReader implemen
* (non-Javadoc)
* @see org.springframework.batch.item.ItemStream#mark(org.springframework.batch.item.StreamContext)
*/
public void mark(StreamContext streamContext) {
public void mark(ExecutionAttributes streamContext) {
getReader().mark();
}
@@ -124,7 +124,7 @@ public class DefaultFlatFileItemReader extends SimpleFlatFileItemReader implemen
* (non-Javadoc)
* @see org.springframework.batch.item.ItemStream#reset(org.springframework.batch.item.StreamContext)
*/
public void reset(StreamContext streamContext) {
public void reset(ExecutionAttributes streamContext) {
getReader().reset();
}

14
spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/FlatFileItemWriter.java
View File

@@ -33,7 +33,7 @@ import org.springframework.batch.io.exception.BatchEnvironmentException;
import org.springframework.batch.io.support.AbstractTransactionalIoSource;
import org.springframework.batch.item.ItemStream;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.StreamContext;
import org.springframework.batch.item.ExecutionAttributes;
import org.springframework.batch.item.StreamException;
import org.springframework.batch.item.writer.ItemTransformer;
import org.springframework.beans.factory.DisposableBean;
@@ -71,7 +71,7 @@ public class FlatFileItemWriter extends AbstractTransactionalIoSource implements
private Resource resource;
private StreamContext streamContext = new StreamContext();
private ExecutionAttributes streamContext = new ExecutionAttributes();
private OutputState state = null;
@@ -246,7 +246,7 @@ public class FlatFileItemWriter extends AbstractTransactionalIoSource implements
/**
* @see ItemStream#getStreamContext()
*/
public StreamContext getStreamContext() {
public ExecutionAttributes getStreamContext() {
if (state == null) {
throw new StreamException("ItemStream not open or already closed.");
}
@@ -257,9 +257,9 @@ public class FlatFileItemWriter extends AbstractTransactionalIoSource implements
}
/**
* @see ItemStream#restoreFrom(StreamContext)
* @see ItemStream#restoreFrom(ExecutionAttributes)
*/
public void restoreFrom(StreamContext data) {
public void restoreFrom(ExecutionAttributes data) {
if (data == null)
return;
@@ -552,14 +552,14 @@ public class FlatFileItemWriter extends AbstractTransactionalIoSource implements
/* (non-Javadoc)
* @see org.springframework.batch.item.ItemStream#mark(org.springframework.batch.item.StreamContext)
*/
public void mark(StreamContext streamContext) {
public void mark(ExecutionAttributes streamContext) {
getOutputState().mark();
}
/* (non-Javadoc)
* @see org.springframework.batch.item.ItemStream#reset(org.springframework.batch.item.StreamContext)
*/
public void reset(StreamContext streamContext) {
public void reset(ExecutionAttributes streamContext) {
getOutputState().checkFileSize();
resetPositionForRestart();
}

6
spring-batch-infrastructure/src/main/java/org/springframework/batch/io/support/AbstractTransactionalIoSource.java
View File

@@ -17,7 +17,7 @@ package org.springframework.batch.io.support;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.StreamContext;
import org.springframework.batch.item.ExecutionAttributes;
import org.springframework.batch.item.stream.ItemStreamAdapter;
/**
@@ -39,14 +39,14 @@ public abstract class AbstractTransactionalIoSource extends ItemStreamAdapter {
*
* @see TransactionSynchronization#afterCompletion
*/
public abstract void mark(StreamContext streamContext);
public abstract void mark(ExecutionAttributes streamContext);
/*
* Called when a transaction has been rolled back.
*
* @see TransactionSynchronization#afterCompletion
*/
public abstract void reset(StreamContext streamContext);
public abstract void reset(ExecutionAttributes streamContext);
/* (non-Javadoc)
* @see org.springframework.batch.item.stream.ItemStreamAdapter#isMarkSupported()

14
spring-batch-infrastructure/src/main/java/org/springframework/batch/io/xml/StaxEventItemReader.java
View File

@@ -18,7 +18,7 @@ import org.springframework.batch.io.xml.stax.FragmentEventReader;
import org.springframework.batch.io.xml.stax.TransactionalEventReader;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemStream;
import org.springframework.batch.item.StreamContext;
import org.springframework.batch.item.ExecutionAttributes;
import org.springframework.batch.item.reader.AbstractItemReader;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
@@ -176,8 +176,8 @@ public class StaxEventItemReader extends AbstractItemReader implements ItemReade
* @return wrapped count of records read so far.
* @see ItemStream#getStreamContext()
*/
public StreamContext getStreamContext() {
StreamContext restartData = new StreamContext();
public ExecutionAttributes getStreamContext() {
ExecutionAttributes restartData = new ExecutionAttributes();
restartData.putLong(READ_COUNT_STATISTICS_NAME, currentRecordCount);
return restartData;
}
@@ -186,12 +186,12 @@ public class StaxEventItemReader extends AbstractItemReader implements ItemReade
* Restores the input source for the given restart data by rereading and
* skipping the number of records stored in the RestartData.
*
* @param StreamContext that holds the line count from the last commit.
* @param ExecutionAttributes that holds the line count from the last commit.
* @throws IllegalStateException if the ItemReader has already been
* initialized or if the number of records to read and skip exceeds the
* available records.
*/
public void restoreFrom(StreamContext data) {
public void restoreFrom(ExecutionAttributes data) {
Assert.state(!initialized);
if (data == null || data.getProperties() == null || !data.containsKey(READ_COUNT_STATISTICS_NAME)) {
return;
@@ -266,7 +266,7 @@ public class StaxEventItemReader extends AbstractItemReader implements ItemReade
/* (non-Javadoc)
* @see org.springframework.batch.item.ItemStream#mark(org.springframework.batch.item.StreamContext)
*/
public void mark(StreamContext streamContext) {
public void mark(ExecutionAttributes streamContext) {
lastCommitPointRecordCount = currentRecordCount;
txReader.onCommit();
skipRecords = new ArrayList();
@@ -275,7 +275,7 @@ public class StaxEventItemReader extends AbstractItemReader implements ItemReade
/* (non-Javadoc)
* @see org.springframework.batch.item.ItemStream#reset(org.springframework.batch.item.StreamContext)
*/
public void reset(StreamContext streamContext) {
public void reset(ExecutionAttributes streamContext) {
currentRecordCount = lastCommitPointRecordCount;
txReader.onRollback();
fragmentReader.reset();

14
spring-batch-infrastructure/src/main/java/org/springframework/batch/io/xml/StaxEventItemWriter.java
View File

@@ -17,7 +17,7 @@ import org.springframework.batch.io.support.FileUtils;
import org.springframework.batch.io.xml.stax.NoStartEndDocumentStreamWriter;
import org.springframework.batch.item.ItemStream;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.StreamContext;
import org.springframework.batch.item.ExecutionAttributes;
import org.springframework.batch.item.StreamException;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
@@ -363,11 +363,11 @@ public class StaxEventItemWriter implements ItemWriter, ItemStream, Initializing
* @return the restart data
* @see org.springframework.batch.item.ItemStream#getStreamContext()
*/
public StreamContext getStreamContext() {
public ExecutionAttributes getStreamContext() {
if (!initialized) {
throw new StreamException("ItemStream is not open, or may have been closed. Cannot access context.");
}
StreamContext context = new StreamContext();
ExecutionAttributes context = new ExecutionAttributes();
context.putLong(RESTART_DATA_NAME, getPosition());
context.putLong(WRITE_STATISTICS_NAME, currentRecordCount);
return context;
@@ -376,9 +376,9 @@ public class StaxEventItemWriter implements ItemWriter, ItemStream, Initializing
/**
* Restore processing from provided restart data.
* @param data the restart data
* @see org.springframework.batch.item.ItemStream#restoreFrom(org.springframework.batch.item.StreamContext)
* @see org.springframework.batch.item.ItemStream#restoreFrom(org.springframework.batch.item.ExecutionAttributes)
*/
public void restoreFrom(StreamContext data) {
public void restoreFrom(ExecutionAttributes data) {
long startAtPosition = 0;
@@ -450,7 +450,7 @@ public class StaxEventItemWriter implements ItemWriter, ItemStream, Initializing
/* (non-Javadoc)
* @see org.springframework.batch.item.ItemStream#mark(org.springframework.batch.item.StreamContext)
*/
public void mark(StreamContext streamContext) {
public void mark(ExecutionAttributes streamContext) {
lastCommitPointPosition = getPosition();
lastCommitPointRecordCount = currentRecordCount;
}
@@ -458,7 +458,7 @@ public class StaxEventItemWriter implements ItemWriter, ItemStream, Initializing
/* (non-Javadoc)
* @see org.springframework.batch.item.ItemStream#reset(org.springframework.batch.item.StreamContext)
*/
public void reset(StreamContext streamContext) {
public void reset(ExecutionAttributes streamContext) {
currentRecordCount = lastCommitPointRecordCount;
// close output
close();

23
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/StreamContext.java → spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ExecutionAttributes.java
View File

@@ -16,6 +16,7 @@
package org.springframework.batch.item;
import java.io.Serializable;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
@@ -31,17 +32,17 @@ import org.springframework.util.Assert;
* also allows for dirty checking by setting a 'dirty' flag whenever any put is
* called.
*/
public class StreamContext {
public class ExecutionAttributes {
private boolean dirty = false;
private final Map map;
public StreamContext() {
public ExecutionAttributes() {
map = new HashMap();
}
public StreamContext(Map map) {
public ExecutionAttributes(Map map){
this.map = map;
}
@@ -60,8 +61,9 @@ public class StreamContext {
put(key, new Double(value));
}
public void put(String key, Object value) {
public void put(String key, Object value){
Assert.isInstanceOf(Serializable.class, value, "Value: [ " + value + "must be serializable.");
dirty = true;
map.put(key, value);
}
@@ -133,17 +135,22 @@ public class StreamContext {
}
public boolean equals(Object obj) {
if(obj instanceof StreamContext == false){
if(obj instanceof ExecutionAttributes == false){
return false;
}
if(this == obj){
return true;
}
StreamContext rhs = (StreamContext)obj;
ExecutionAttributes rhs = (ExecutionAttributes)obj;
return this.entrySet().equals(rhs.entrySet());
}
public int hashCode() {
return map.hashCode();
}
public String toString() {
return map.toString();
}
}

14
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ItemStream.java
View File

@@ -23,11 +23,11 @@ package org.springframework.batch.item;
* <p>
*
* <p>
* The state that is stored is represented as {@link StreamContext} which
* The state that is stored is represented as {@link ExecutionAttributes} which
* enforces a requirement that any restart data can be represented by a
* Properties object. In general, the contract is that {@link StreamContext}
* Properties object. In general, the contract is that {@link ExecutionAttributes}
* that is returned via the {@link #getStreamContext()} method will be given
* back to the {@link #restoreFrom(StreamContext)} method, exactly as it was
* back to the {@link #restoreFrom(ExecutionAttributes)} method, exactly as it was
* provided.
* </p>
*
@@ -37,11 +37,11 @@ package org.springframework.batch.item;
public interface ItemStream extends StreamContextProvider {
/**
* Restart state given the provided {@link StreamContext}.
* Restart state given the provided {@link ExecutionAttributes}.
*
* @param context
*/
void restoreFrom(StreamContext context);
void restoreFrom(ExecutionAttributes context);
/**
* If any resources are needed for the stream to operate they need to be
@@ -75,7 +75,7 @@ public interface ItemStream extends StreamContextProvider {
*
* @throws UnsupportedOperationException if the operation is not supported
*/
void mark(StreamContext streamContext);
void mark(ExecutionAttributes streamContext);
/**
* Reset the stream to the last mark. After a reset the stream state will be
@@ -84,5 +84,5 @@ public interface ItemStream extends StreamContextProvider {
*
* @throws UnsupportedOperationException if the operation is not supported
*/
void reset(StreamContext streamContext);
void reset(ExecutionAttributes streamContext);
}

6
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/StreamContextProvider.java
View File

@@ -22,11 +22,11 @@ package org.springframework.batch.item;
public interface StreamContextProvider {
/**
* Get {@link StreamContext} representing this object's current state.
* Get {@link ExecutionAttributes} representing this object's current state.
* Should not return null even if there is no state.
*
* @return {@link StreamContext} representing current state.
* @return {@link ExecutionAttributes} representing current state.
*/
StreamContext getStreamContext();
ExecutionAttributes getStreamContext();
}

12
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/reader/DelegatingItemReader.java
View File

@@ -19,7 +19,7 @@ package org.springframework.batch.item.reader;
import org.springframework.batch.io.Skippable;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemStream;
import org.springframework.batch.item.StreamContext;
import org.springframework.batch.item.ExecutionAttributes;
import org.springframework.batch.item.StreamException;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.util.Assert;
@@ -52,18 +52,18 @@ public class DelegatingItemReader extends AbstractItemReader implements Skippabl
* @throws IllegalStateException if the parent template is not itself
* {@link ItemStream}.
*/
public StreamContext getStreamContext() {
public ExecutionAttributes getStreamContext() {
// TODO: this is not necessary...
Assert.state(inputSource instanceof ItemStream, "Input source is not ItemStream");
return ((ItemStream) inputSource).getStreamContext();
}
/**
* @see ItemStream#restoreFrom(StreamContext)
* @see ItemStream#restoreFrom(ExecutionAttributes)
* @throws IllegalStateException if the parent template is not itself
* {@link ItemStream}.
*/
public void restoreFrom(StreamContext data) {
public void restoreFrom(ExecutionAttributes data) {
Assert.state(inputSource instanceof ItemStream, "Input source is not ItemStream");
((ItemStream) inputSource).restoreFrom(data);
}
@@ -119,7 +119,7 @@ public class DelegatingItemReader extends AbstractItemReader implements Skippabl
/* (non-Javadoc)
* @see org.springframework.batch.item.ItemStream#mark(org.springframework.batch.item.StreamContext)
*/
public void mark(StreamContext streamContext) {
public void mark(ExecutionAttributes streamContext) {
if (inputSource instanceof ItemStream) {
((ItemStream) inputSource).mark(streamContext);
}
@@ -128,7 +128,7 @@ public class DelegatingItemReader extends AbstractItemReader implements Skippabl
/* (non-Javadoc)
* @see org.springframework.batch.item.ItemStream#reset(org.springframework.batch.item.StreamContext)
*/
public void reset(StreamContext streamContext) {
public void reset(ExecutionAttributes streamContext) {
if (inputSource instanceof ItemStream) {
((ItemStream) inputSource).reset(streamContext);
}

16
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/stream/ItemStreamAdapter.java
View File

@@ -16,7 +16,7 @@
package org.springframework.batch.item.stream;
import org.springframework.batch.item.ItemStream;
import org.springframework.batch.item.StreamContext;
import org.springframework.batch.item.ExecutionAttributes;
import org.springframework.batch.item.StreamException;
/**
@@ -41,17 +41,17 @@ public class ItemStreamAdapter implements ItemStream {
/**
* No-op.
* @see org.springframework.batch.item.ItemStream#restoreFrom(org.springframework.batch.item.StreamContext)
* @see org.springframework.batch.item.ItemStream#restoreFrom(org.springframework.batch.item.ExecutionAttributes)
*/
public void restoreFrom(StreamContext context) {
public void restoreFrom(ExecutionAttributes context) {
}
/**
* Return empty {@link StreamContext}.
* Return empty {@link ExecutionAttributes}.
* @see org.springframework.batch.item.StreamContextProvider#getStreamContext()
*/
public StreamContext getStreamContext() {
return new StreamContext();
public ExecutionAttributes getStreamContext() {
return new ExecutionAttributes();
}
/* (non-Javadoc)
@@ -64,14 +64,14 @@ public class ItemStreamAdapter implements ItemStream {
/* (non-Javadoc)
* @see org.springframework.batch.item.ItemStream#mark(org.springframework.batch.item.StreamContext)
*/
public void mark(StreamContext streamContext) {
public void mark(ExecutionAttributes streamContext) {
throw new UnsupportedOperationException("Mark operation not supported.");
}
/* (non-Javadoc)
* @see org.springframework.batch.item.ItemStream#reset(org.springframework.batch.item.StreamContext)
*/
public void reset(StreamContext streamContext) {
public void reset(ExecutionAttributes streamContext) {
throw new UnsupportedOperationException("Reset operation not supported.");
}

17
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/stream/SimpleStreamManager.java
View File

@@ -24,6 +24,7 @@ import java.util.Set;
import java.util.Map.Entry;
import org.springframework.batch.item.ItemStream;
import org.springframework.batch.item.ExecutionAttributes;
import org.springframework.batch.item.StreamContext;
import org.springframework.batch.item.StreamException;
import org.springframework.transaction.PlatformTransactionManager;
@@ -90,11 +91,11 @@ public class SimpleStreamManager implements StreamManager {
*
* @see org.springframework.batch.item.stream.StreamManager#getStreamContext(java.lang.Object)
*/
public StreamContext getStreamContext(Object key) {
final StreamContext result = new StreamContext();
public ExecutionAttributes getStreamContext(Object key) {
final ExecutionAttributes result = new ExecutionAttributes();
iterate(key, new Callback() {
public void execute(ItemStream stream) {
StreamContext context = stream.getStreamContext();
ExecutionAttributes context = stream.getStreamContext();
String prefix = ClassUtils.getQualifiedName(stream.getClass()) + ".";
if (!useClassNameAsPrefix) {
prefix = "";
@@ -102,7 +103,7 @@ public class SimpleStreamManager implements StreamManager {
for (Iterator iterator = context.entrySet().iterator(); iterator.hasNext();) {
Entry entry = (Entry) iterator.next();
String contextKey = prefix + entry.getKey();
result.put(contextKey, entry.getValue());
result.putString(contextKey, entry.getValue().toString());
}
}
});
@@ -116,7 +117,7 @@ public class SimpleStreamManager implements StreamManager {
* @see org.springframework.batch.item.stream.StreamManager#register(java.lang.Object,
* org.springframework.batch.item.ItemStream, StreamContext)
*/
public void register(Object key, ItemStream stream, StreamContext streamContext) {
public void register(Object key, ItemStream stream, ExecutionAttributes streamContext) {
synchronized (registry) {
Set set = (Set) registry.get(key);
if (set == null) {
@@ -135,8 +136,8 @@ public class SimpleStreamManager implements StreamManager {
* @param streamContext
* @return
*/
private StreamContext extract(ItemStream stream, StreamContext context) {
StreamContext result = new StreamContext();
private ExecutionAttributes extract(ItemStream stream, ExecutionAttributes context) {
ExecutionAttributes result = new ExecutionAttributes();
String prefix = ClassUtils.getQualifiedName(stream.getClass()) + ".";
if (!useClassNameAsPrefix) {
prefix = "";
@@ -145,7 +146,7 @@ public class SimpleStreamManager implements StreamManager {
Entry entry = (Entry) iterator.next();
String contextKey = (String) entry.getKey();
if (contextKey.startsWith(prefix)) {
result.put(contextKey.substring(prefix.length()), entry.getValue());
result.putString(contextKey.substring(prefix.length()), entry.getValue().toString());
}
}
return result;

10
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/stream/StreamManager.java
View File

@@ -16,7 +16,7 @@
package org.springframework.batch.item.stream;
import org.springframework.batch.item.ItemStream;
import org.springframework.batch.item.StreamContext;
import org.springframework.batch.item.ExecutionAttributes;
import org.springframework.batch.item.StreamException;
import org.springframework.transaction.TransactionStatus;
@@ -39,18 +39,18 @@ public interface StreamManager {
* @param stream an {@link ItemStream}
* @param streamContext the context (may be null) to restore from on registration
*/
void register(Object key, ItemStream stream, StreamContext streamContext);
void register(Object key, ItemStream stream, ExecutionAttributes streamContext);
/**
* Extract and aggregate the {@link StreamContext} from all streams under
* Extract and aggregate the {@link ExecutionAttributes} from all streams under
* this key.
*
* @param key the key under which {@link ItemStream} instances might have
* been registered.
* @return {@link StreamContext} aggregating the contexts of all providers
* @return {@link ExecutionAttributes} aggregating the contexts of all providers
* registered under this key, or empty otherwise.
*/
StreamContext getStreamContext(Object key);
ExecutionAttributes getStreamContext(Object key);
/**
* If any resources are needed for the stream to operate they need to be

10
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/writer/DelegatingItemWriter.java
View File

@@ -3,7 +3,7 @@ package org.springframework.batch.item.writer;
import org.springframework.batch.io.Skippable;
import org.springframework.batch.item.ItemStream;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.StreamContext;
import org.springframework.batch.item.ExecutionAttributes;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.util.Assert;
@@ -49,7 +49,7 @@ public class DelegatingItemWriter implements ItemWriter, Skippable, Initializing
/**
* @see ItemStream#getStreamContext()
*/
public StreamContext getStreamContext() {
public ExecutionAttributes getStreamContext() {
Assert.state(writer != null, "Source must not be null.");
@@ -57,14 +57,14 @@ public class DelegatingItemWriter implements ItemWriter, Skippable, Initializing
return ((ItemStream) writer).getStreamContext();
}
else {
return new StreamContext();
return new ExecutionAttributes();
}
}
/**
* @see ItemStream#restoreFrom(StreamContext)
* @see ItemStream#restoreFrom(ExecutionAttributes)
*/
public void restoreFrom(StreamContext data) {
public void restoreFrom(ExecutionAttributes data) {
Assert.state(writer != null, "Source must not be null.");