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

IN PROGRESS - BATCH-505: Job-level ExecutionContext

Browse Source 
JobExecutionContext now works
This commit is contained in:
robokaso
2008-05-22 10:44:43 +00:00
parent 5dea8a0144
commit 69d62dd9dd
7 changed files with 427 additions and 202 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
spring-batch-core/src/main/java/org/springframework/batch/core/JobExecution.java
View File

@@ -21,6 +21,7 @@ import java.util.Date;
import java.util.HashSet;
import java.util.Iterator;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.batch.repeat.ExitStatus;
/**
@@ -42,6 +43,8 @@ public class JobExecution extends Entity {
private volatile Date endTime = null;
private volatile ExitStatus exitStatus = ExitStatus.UNKNOWN;
private ExecutionContext executionContext;
/**
* Because a JobExecution isn't valid unless the job is set, this
@@ -178,4 +181,22 @@ public class JobExecution extends Entity {
}
status = BatchStatus.STOPPING;
}
/**
* Sets the {@link ExecutionContext} for this execution
*
* @param executionContext the context
*/
public void setExecutionContext(ExecutionContext executionContext) {
this.executionContext = executionContext;
}
/**
* Returns the {@link ExecutionContext} for this execution
*
* @return the context
*/
public ExecutionContext getExecutionContext() {
return executionContext;
}
}

310
spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/JdbcExecutionContextDao.java Normal file
View File

@@ -0,0 +1,310 @@
package org.springframework.batch.core.repository.dao;
import java.io.Serializable;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.Iterator;
import java.util.Map.Entry;
import org.apache.commons.lang.SerializationUtils;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.UnexpectedJobExecutionException;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.dao.DataAccessException;
import org.springframework.jdbc.core.PreparedStatementCallback;
import org.springframework.jdbc.core.RowCallbackHandler;
import org.springframework.jdbc.core.support.AbstractLobCreatingPreparedStatementCallback;
import org.springframework.jdbc.support.lob.DefaultLobHandler;
import org.springframework.jdbc.support.lob.LobCreator;
import org.springframework.jdbc.support.lob.LobHandler;
import org.springframework.util.Assert;
/**
* JDBC DAO for {@link ExecutionContext}.
*
*/
class JdbcExecutionContextDao extends AbstractJdbcBatchMetadataDao {
private static final String STEP_DISCRIMINATOR = "S";
private static final String JOB_DISCRIMINATOR = "J";
private static final String FIND_EXECUTION_CONTEXT = "SELECT TYPE_CD, KEY_NAME, STRING_VAL, DOUBLE_VAL, LONG_VAL, OBJECT_VAL "
+ "from %PREFIX%EXECUTION_CONTEXT where EXECUTION_ID = ? and DISCRIMINATOR = ?";
private static final String INSERT_STEP_EXECUTION_CONTEXT = "INSERT into %PREFIX%EXECUTION_CONTEXT(EXECUTION_ID, DISCRIMINATOR, TYPE_CD,"
+ " KEY_NAME, STRING_VAL, DOUBLE_VAL, LONG_VAL, OBJECT_VAL) values(?,?,?,?,?,?,?,?)";
private static final String UPDATE_STEP_EXECUTION_CONTEXT = "UPDATE %PREFIX%EXECUTION_CONTEXT set "
+ "TYPE_CD = ?, STRING_VAL = ?, DOUBLE_VAL = ?, LONG_VAL = ?, OBJECT_VAL = ? where EXECUTION_ID = ? and KEY_NAME = ?";
private LobHandler lobHandler = new DefaultLobHandler();
public ExecutionContext getExecutionContext(JobExecution jobExecution) {
final Long executionId = jobExecution.getId();
Assert.notNull(executionId, "ExecutionId must not be null.");
final ExecutionContext executionContext = new ExecutionContext();
RowCallbackHandler callback = new RowCallbackHandler() {
public void processRow(ResultSet rs) throws SQLException {
String typeCd = rs.getString("TYPE_CD");
AttributeType type = AttributeType.getType(typeCd);
String key = rs.getString("KEY_NAME");
if (type == AttributeType.STRING) {
executionContext.putString(key, rs.getString("STRING_VAL"));
}
else if (type == AttributeType.LONG) {
executionContext.putLong(key, rs.getLong("LONG_VAL"));
}
else if (type == AttributeType.DOUBLE) {
executionContext.putDouble(key, rs.getDouble("DOUBLE_VAL"));
}
else if (type == AttributeType.OBJECT) {
executionContext.put(key, SerializationUtils.deserialize(rs.getBinaryStream("OBJECT_VAL")));
}
else {
throw new UnexpectedJobExecutionException("Invalid type found: [" + typeCd
+ "] for execution id: [" + executionId + "]");
}
}
};
getJdbcTemplate().query(getQuery(FIND_EXECUTION_CONTEXT), new Object[] { executionId, JOB_DISCRIMINATOR }, callback);
return executionContext;
}
public ExecutionContext getExecutionContext(StepExecution stepExecution) {
final Long executionId = stepExecution.getId();
Assert.notNull(executionId, "ExecutionId must not be null.");
final ExecutionContext executionContext = new ExecutionContext();
RowCallbackHandler callback = new RowCallbackHandler() {
public void processRow(ResultSet rs) throws SQLException {
String typeCd = rs.getString("TYPE_CD");
AttributeType type = AttributeType.getType(typeCd);
String key = rs.getString("KEY_NAME");
if (type == AttributeType.STRING) {
executionContext.putString(key, rs.getString("STRING_VAL"));
}
else if (type == AttributeType.LONG) {
executionContext.putLong(key, rs.getLong("LONG_VAL"));
}
else if (type == AttributeType.DOUBLE) {
executionContext.putDouble(key, rs.getDouble("DOUBLE_VAL"));
}
else if (type == AttributeType.OBJECT) {
executionContext.put(key, SerializationUtils.deserialize(rs.getBinaryStream("OBJECT_VAL")));
}
else {
throw new UnexpectedJobExecutionException("Invalid type found: [" + typeCd
+ "] for execution id: [" + executionId + "]");
}
}
};
getJdbcTemplate().query(getQuery(FIND_EXECUTION_CONTEXT), new Object[] { executionId, STEP_DISCRIMINATOR }, callback);
return executionContext;
}
public void saveOrUpdateExecutionContext(final JobExecution jobExecution) {
Long executionId = jobExecution.getId();
ExecutionContext executionContext = jobExecution.getExecutionContext();
Assert.notNull(executionId, "ExecutionId must not be null.");
Assert.notNull(executionContext, "The ExecutionContext must not be null.");
for (Iterator it = executionContext.entrySet().iterator(); it.hasNext();) {
Entry entry = (Entry) it.next();
final String key = entry.getKey().toString();
final Object value = entry.getValue();
if (value instanceof String) {
updateExecutionAttribute(executionId, JOB_DISCRIMINATOR, key, value, AttributeType.STRING);
}
else if (value instanceof Double) {
updateExecutionAttribute(executionId, JOB_DISCRIMINATOR, key, value, AttributeType.DOUBLE);
}
else if (value instanceof Long) {
updateExecutionAttribute(executionId, JOB_DISCRIMINATOR, key, value, AttributeType.LONG);
}
else {
updateExecutionAttribute(executionId, JOB_DISCRIMINATOR, key, value, AttributeType.OBJECT);
}
}
}
/**
* Save or update execution attributes. A lob creator must be used, since
* any attributes that don't match a provided type must be serialized into a
* blob.
*
* @see LobCreator
*/
public void saveOrUpdateExecutionContext(final StepExecution stepExecution) {
Long executionId = stepExecution.getId();
ExecutionContext executionContext = stepExecution.getExecutionContext();
Assert.notNull(executionId, "ExecutionId must not be null.");
Assert.notNull(executionContext, "The ExecutionContext must not be null.");
for (Iterator it = executionContext.entrySet().iterator(); it.hasNext();) {
Entry entry = (Entry) it.next();
final String key = entry.getKey().toString();
final Object value = entry.getValue();
if (value instanceof String) {
updateExecutionAttribute(executionId, STEP_DISCRIMINATOR, key, value, AttributeType.STRING);
}
else if (value instanceof Double) {
updateExecutionAttribute(executionId, STEP_DISCRIMINATOR, key, value, AttributeType.DOUBLE);
}
else if (value instanceof Long) {
updateExecutionAttribute(executionId, STEP_DISCRIMINATOR, key, value, AttributeType.LONG);
}
else {
updateExecutionAttribute(executionId, STEP_DISCRIMINATOR, key, value, AttributeType.OBJECT);
}
}
}
private void updateExecutionAttribute(final Long executionId, final String discriminator, final String key, final Object value,
final AttributeType type) {
PreparedStatementCallback callback = new AbstractLobCreatingPreparedStatementCallback(lobHandler) {
protected void setValues(PreparedStatement ps, LobCreator lobCreator) throws SQLException,
DataAccessException {
ps.setLong(6, executionId.longValue());
ps.setString(7, key);
if (type == AttributeType.STRING) {
ps.setString(1, AttributeType.STRING.toString());
ps.setString(2, value.toString());
ps.setDouble(3, 0.0);
ps.setLong(4, 0);
lobCreator.setBlobAsBytes(ps, 5, null);
}
else if (type == AttributeType.DOUBLE) {
ps.setString(1, AttributeType.DOUBLE.toString());
ps.setString(2, null);
ps.setDouble(3, ((Double) value).doubleValue());
ps.setLong(4, 0);
lobCreator.setBlobAsBytes(ps, 5, null);
}
else if (type == AttributeType.LONG) {
ps.setString(1, AttributeType.LONG.toString());
ps.setString(2, null);
ps.setDouble(3, 0.0);
ps.setLong(4, ((Long) value).longValue());
lobCreator.setBlobAsBytes(ps, 5, null);
}
else {
ps.setString(1, AttributeType.OBJECT.toString());
ps.setString(2, null);
ps.setDouble(3, 0.0);
ps.setLong(4, 0);
lobCreator.setBlobAsBytes(ps, 5, SerializationUtils.serialize((Serializable) value));
}
}
};
// LobCreating callbacks always return the affect row count for SQL DML
// statements, if less than 1 row
// is affected, then this row is new and should be inserted.
Integer affectedRows = (Integer) getJdbcTemplate().execute(getQuery(UPDATE_STEP_EXECUTION_CONTEXT), callback);
if (affectedRows.intValue() < 1) {
insertExecutionAttribute(executionId, discriminator, key, value, type);
}
}
private void insertExecutionAttribute(final Long executionId, final String discriminator, final String key, final Object value,
final AttributeType type) {
PreparedStatementCallback callback = new AbstractLobCreatingPreparedStatementCallback(lobHandler) {
protected void setValues(PreparedStatement ps, LobCreator lobCreator) throws SQLException,
DataAccessException {
ps.setLong(1, executionId.longValue());
ps.setString(2, discriminator);
ps.setString(4, key);
if (type == AttributeType.STRING) {
ps.setString(3, AttributeType.STRING.toString());
ps.setString(5, value.toString());
ps.setDouble(6, 0.0);
ps.setLong(7, 0);
lobCreator.setBlobAsBytes(ps, 8, null);
}
else if (type == AttributeType.DOUBLE) {
ps.setString(3, AttributeType.DOUBLE.toString());
ps.setString(5, null);
ps.setDouble(6, ((Double) value).doubleValue());
ps.setLong(7, 0);
lobCreator.setBlobAsBytes(ps, 8, null);
}
else if (type == AttributeType.LONG) {
ps.setString(3, AttributeType.LONG.toString());
ps.setString(5, null);
ps.setDouble(6, 0.0);
ps.setLong(7, ((Long) value).longValue());
lobCreator.setBlobAsBytes(ps, 8, null);
}
else {
ps.setString(3, AttributeType.OBJECT.toString());
ps.setString(5, null);
ps.setDouble(6, 0.0);
ps.setLong(7, 0);
lobCreator.setBlobAsBytes(ps, 8, SerializationUtils.serialize((Serializable) value));
}
}
};
getJdbcTemplate().execute(getQuery(INSERT_STEP_EXECUTION_CONTEXT), callback);
}
public void setLobHandler(LobHandler lobHandler) {
this.lobHandler = lobHandler;
}
public static class AttributeType {
private final String type;
private AttributeType(String type) {
this.type = type;
}
public String toString() {
return type;
}
public static final AttributeType STRING = new AttributeType("STRING");
public static final AttributeType LONG = new AttributeType("LONG");
public static final AttributeType OBJECT = new AttributeType("OBJECT");
public static final AttributeType DOUBLE = new AttributeType("DOUBLE");
private static final AttributeType[] VALUES = { STRING, OBJECT, LONG, DOUBLE };
public static AttributeType getType(String typeAsString) {
for (int i = 0; i < VALUES.length; i++) {
if (VALUES[i].toString().equals(typeAsString)) {
return (AttributeType) VALUES[i];
}
}
return null;
}
}
}

22
spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/JdbcJobExecutionDao.java
View File

@@ -10,10 +10,13 @@ import org.apache.commons.logging.LogFactory;
import org.springframework.batch.core.BatchStatus;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobInstance;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.batch.repeat.ExitStatus;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.jdbc.core.RowMapper;
import org.springframework.jdbc.support.incrementer.DataFieldMaxValueIncrementer;
import org.springframework.jdbc.support.lob.DefaultLobHandler;
import org.springframework.jdbc.support.lob.LobHandler;
import org.springframework.util.Assert;
/**
@@ -53,6 +56,10 @@ public class JdbcJobExecutionDao extends AbstractJdbcBatchMetadataDao implements
+ " where JOB_INSTANCE_ID = ? and START_TIME = (SELECT max(START_TIME) from %PREFIX%JOB_EXECUTION where JOB_INSTANCE_ID = ?)";
private DataFieldMaxValueIncrementer jobExecutionIncrementer;
private LobHandler lobHandler = new DefaultLobHandler();
private JdbcExecutionContextDao ecDao = new JdbcExecutionContextDao();
public List findJobExecutions(final JobInstance job) {
@@ -174,6 +181,9 @@ public class JdbcJobExecutionDao extends AbstractJdbcBatchMetadataDao implements
public void afterPropertiesSet() throws Exception {
super.afterPropertiesSet();
Assert.notNull(jobExecutionIncrementer);
ecDao.setJdbcTemplate(getJdbcTemplate());
ecDao.setLobHandler(lobHandler);
ecDao.afterPropertiesSet();
}
/**
@@ -220,4 +230,16 @@ public class JdbcJobExecutionDao extends AbstractJdbcBatchMetadataDao implements
}
}
public ExecutionContext findExecutionContext(JobExecution jobExecution) {
return ecDao.getExecutionContext(jobExecution);
}
public void saveOrUpdateExecutionContext(JobExecution jobExecution) {
ecDao.saveOrUpdateExecutionContext(jobExecution);
}
public void setLobHandler(LobHandler lobHandler) {
this.lobHandler = lobHandler;
}
}

210
spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/JdbcStepExecutionDao.java
View File

@@ -1,31 +1,21 @@
package org.springframework.batch.core.repository.dao;
import java.io.Serializable;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Types;
import java.util.Iterator;
import java.util.List;
import java.util.Map.Entry;
import org.apache.commons.lang.SerializationUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.batch.core.BatchStatus;
import org.springframework.batch.core.UnexpectedJobExecutionException;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.batch.repeat.ExitStatus;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.dao.DataAccessException;
import org.springframework.dao.OptimisticLockingFailureException;
import org.springframework.jdbc.core.PreparedStatementCallback;
import org.springframework.jdbc.core.RowCallbackHandler;
import org.springframework.jdbc.core.RowMapper;
import org.springframework.jdbc.core.support.AbstractLobCreatingPreparedStatementCallback;
import org.springframework.jdbc.support.incrementer.DataFieldMaxValueIncrementer;
import org.springframework.jdbc.support.lob.DefaultLobHandler;
import org.springframework.jdbc.support.lob.LobCreator;
@@ -55,19 +45,10 @@ public class JdbcStepExecutionDao extends AbstractJdbcBatchMetadataDao implement
private static final Log logger = LogFactory.getLog(JdbcStepExecutionDao.class);
private static final String FIND_STEP_EXECUTION_CONTEXT = "SELECT TYPE_CD, KEY_NAME, STRING_VAL, DOUBLE_VAL, LONG_VAL, OBJECT_VAL "
+ "from %PREFIX%EXECUTION_CONTEXT where EXECUTION_ID = ? and DISCRIMINATOR = 'S'";
private static final String INSERT_STEP_EXECUTION_CONTEXT = "INSERT into %PREFIX%EXECUTION_CONTEXT(EXECUTION_ID, DISCRIMINATOR, TYPE_CD,"
+ " KEY_NAME, STRING_VAL, DOUBLE_VAL, LONG_VAL, OBJECT_VAL) values(?,'S',?,?,?,?,?,?)";
private static final String SAVE_STEP_EXECUTION = "INSERT into %PREFIX%STEP_EXECUTION(STEP_EXECUTION_ID, VERSION, STEP_NAME, JOB_EXECUTION_ID, START_TIME, "
+ "END_TIME, STATUS, COMMIT_COUNT, ITEM_COUNT, CONTINUABLE, EXIT_CODE, EXIT_MESSAGE, READ_SKIP_COUNT, WRITE_SKIP_COUNT, ROLLBACK_COUNT) "
+ "values(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)";
private static final String UPDATE_STEP_EXECUTION_CONTEXT = "UPDATE %PREFIX%EXECUTION_CONTEXT set "
+ "TYPE_CD = ?, STRING_VAL = ?, DOUBLE_VAL = ?, LONG_VAL = ?, OBJECT_VAL = ? where EXECUTION_ID = ? and KEY_NAME = ?";
private static final String UPDATE_STEP_EXECUTION = "UPDATE %PREFIX%STEP_EXECUTION set START_TIME = ?, END_TIME = ?, "
+ "STATUS = ?, COMMIT_COUNT = ?, ITEM_COUNT = ?, CONTINUABLE = ? , EXIT_CODE = ?, "
+ "EXIT_MESSAGE = ?, VERSION = ?, READ_SKIP_COUNT = ?, WRITE_SKIP_COUNT = ?, ROLLBACK_COUNT = ? where STEP_EXECUTION_ID = ? and VERSION = ?";
@@ -82,85 +63,12 @@ public class JdbcStepExecutionDao extends AbstractJdbcBatchMetadataDao implement
private LobHandler lobHandler = new DefaultLobHandler();
private DataFieldMaxValueIncrementer stepExecutionIncrementer;
private JdbcExecutionContextDao ecDao = new JdbcExecutionContextDao();
public ExecutionContext findExecutionContext(final StepExecution stepExecution) {
final Long executionId = stepExecution.getId();
Assert.notNull(executionId, "ExecutionId must not be null.");
final ExecutionContext executionContext = new ExecutionContext();
RowCallbackHandler callback = new RowCallbackHandler() {
public void processRow(ResultSet rs) throws SQLException {
String typeCd = rs.getString("TYPE_CD");
AttributeType type = AttributeType.getType(typeCd);
String key = rs.getString("KEY_NAME");
if (type == AttributeType.STRING) {
executionContext.putString(key, rs.getString("STRING_VAL"));
}
else if (type == AttributeType.LONG) {
executionContext.putLong(key, rs.getLong("LONG_VAL"));
}
else if (type == AttributeType.DOUBLE) {
executionContext.putDouble(key, rs.getDouble("DOUBLE_VAL"));
}
else if (type == AttributeType.OBJECT) {
executionContext.put(key, SerializationUtils.deserialize(rs.getBinaryStream("OBJECT_VAL")));
}
else {
throw new UnexpectedJobExecutionException("Invalid type found: [" + typeCd
+ "] for execution id: [" + executionId + "]");
}
}
};
getJdbcTemplate().query(getQuery(FIND_STEP_EXECUTION_CONTEXT), new Object[] { executionId }, callback);
return executionContext;
}
private void insertExecutionAttribute(final Long executionId, final String key, final Object value,
final AttributeType type) {
PreparedStatementCallback callback = new AbstractLobCreatingPreparedStatementCallback(lobHandler) {
protected void setValues(PreparedStatement ps, LobCreator lobCreator) throws SQLException,
DataAccessException {
ps.setLong(1, executionId.longValue());
ps.setString(3, key);
if (type == AttributeType.STRING) {
ps.setString(2, AttributeType.STRING.toString());
ps.setString(4, value.toString());
ps.setDouble(5, 0.0);
ps.setLong(6, 0);
lobCreator.setBlobAsBytes(ps, 7, null);
}
else if (type == AttributeType.DOUBLE) {
ps.setString(2, AttributeType.DOUBLE.toString());
ps.setString(4, null);
ps.setDouble(5, ((Double) value).doubleValue());
ps.setLong(6, 0);
lobCreator.setBlobAsBytes(ps, 7, null);
}
else if (type == AttributeType.LONG) {
ps.setString(2, AttributeType.LONG.toString());
ps.setString(4, null);
ps.setDouble(5, 0.0);
ps.setLong(6, ((Long) value).longValue());
lobCreator.setBlobAsBytes(ps, 7, null);
}
else {
ps.setString(2, AttributeType.OBJECT.toString());
ps.setString(4, null);
ps.setDouble(5, 0.0);
ps.setLong(6, 0);
lobCreator.setBlobAsBytes(ps, 7, SerializationUtils.serialize((Serializable) value));
}
}
};
getJdbcTemplate().execute(getQuery(INSERT_STEP_EXECUTION_CONTEXT), callback);
return ecDao.getExecutionContext(stepExecution);
}
/**
@@ -220,80 +128,10 @@ public class JdbcStepExecutionDao extends AbstractJdbcBatchMetadataDao implement
*/
public void saveOrUpdateExecutionContext(final StepExecution stepExecution) {
Long executionId = stepExecution.getId();
ExecutionContext executionContext = stepExecution.getExecutionContext();
Assert.notNull(executionId, "ExecutionId must not be null.");
Assert.notNull(executionContext, "The ExecutionContext must not be null.");
for (Iterator it = executionContext.entrySet().iterator(); it.hasNext();) {
Entry entry = (Entry) it.next();
final String key = entry.getKey().toString();
final Object value = entry.getValue();
if (value instanceof String) {
updateExecutionAttribute(executionId, key, value, AttributeType.STRING);
}
else if (value instanceof Double) {
updateExecutionAttribute(executionId, key, value, AttributeType.DOUBLE);
}
else if (value instanceof Long) {
updateExecutionAttribute(executionId, key, value, AttributeType.LONG);
}
else {
updateExecutionAttribute(executionId, key, value, AttributeType.OBJECT);
}
}
ecDao.saveOrUpdateExecutionContext(stepExecution);
}
private void updateExecutionAttribute(final Long executionId, final String key, final Object value,
final AttributeType type) {
PreparedStatementCallback callback = new AbstractLobCreatingPreparedStatementCallback(lobHandler) {
protected void setValues(PreparedStatement ps, LobCreator lobCreator) throws SQLException,
DataAccessException {
ps.setLong(6, executionId.longValue());
ps.setString(7, key);
if (type == AttributeType.STRING) {
ps.setString(1, AttributeType.STRING.toString());
ps.setString(2, value.toString());
ps.setDouble(3, 0.0);
ps.setLong(4, 0);
lobCreator.setBlobAsBytes(ps, 5, null);
}
else if (type == AttributeType.DOUBLE) {
ps.setString(1, AttributeType.DOUBLE.toString());
ps.setString(2, null);
ps.setDouble(3, ((Double) value).doubleValue());
ps.setLong(4, 0);
lobCreator.setBlobAsBytes(ps, 5, null);
}
else if (type == AttributeType.LONG) {
ps.setString(1, AttributeType.LONG.toString());
ps.setString(2, null);
ps.setDouble(3, 0.0);
ps.setLong(4, ((Long) value).longValue());
lobCreator.setBlobAsBytes(ps, 5, null);
}
else {
ps.setString(1, AttributeType.OBJECT.toString());
ps.setString(2, null);
ps.setDouble(3, 0.0);
ps.setLong(4, 0);
lobCreator.setBlobAsBytes(ps, 5, SerializationUtils.serialize((Serializable) value));
}
}
};
// LobCreating callbacks always return the affect row count for SQL DML
// statements, if less than 1 row
// is affected, then this row is new and should be inserted.
Integer affectedRows = (Integer) getJdbcTemplate().execute(getQuery(UPDATE_STEP_EXECUTION_CONTEXT), callback);
if (affectedRows.intValue() < 1) {
insertExecutionAttribute(executionId, key, value, type);
}
}
/*
* (non-Javadoc)
@@ -396,41 +234,11 @@ public class JdbcStepExecutionDao extends AbstractJdbcBatchMetadataDao implement
}
public void afterPropertiesSet() throws Exception {
super.afterPropertiesSet();
Assert.notNull(stepExecutionIncrementer, "StepExecutionIncrementer cannot be null.");
}
public static class AttributeType {
private final String type;
private AttributeType(String type) {
this.type = type;
}
public String toString() {
return type;
}
public static final AttributeType STRING = new AttributeType("STRING");
public static final AttributeType LONG = new AttributeType("LONG");
public static final AttributeType OBJECT = new AttributeType("OBJECT");
public static final AttributeType DOUBLE = new AttributeType("DOUBLE");
private static final AttributeType[] VALUES = { STRING, OBJECT, LONG, DOUBLE };
public static AttributeType getType(String typeAsString) {
for (int i = 0; i < VALUES.length; i++) {
if (VALUES[i].toString().equals(typeAsString)) {
return (AttributeType) VALUES[i];
}
}
return null;
}
ecDao.setJdbcTemplate(getJdbcTemplate());
ecDao.setLobHandler(lobHandler);
ecDao.afterPropertiesSet();
}
public StepExecution getStepExecution(JobExecution jobExecution, Step step) {

17
spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/JobExecutionDao.java
View File

@@ -4,6 +4,7 @@ import java.util.List;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobInstance;
import org.springframework.batch.item.ExecutionContext;
/**
* Data Access Object for job executions.
@@ -51,5 +52,21 @@ public interface JobExecutionDao {
* @return last JobExecution for given JobInstance.
*/
JobExecution getLastJobExecution(JobInstance jobInstance);
/**
* Find the {@link ExecutionContext} for the given {@link JobExecution}.
*
* @throws IllegalArgumentException if the id is null.
*/
ExecutionContext findExecutionContext(JobExecution jobExecution);
/**
* Save the {@link ExecutionContext} of the given {@link JobExecution}.
*
* @param jobExecution the {@link JobExecution} containing the
* {@link ExecutionContext} to be saved.
* @throws IllegalArgumentException if the attributes are null.
*/
void saveOrUpdateExecutionContext(JobExecution jobExecution);
}

15
spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/MapJobExecutionDao.java
View File

@@ -7,6 +7,7 @@ import java.util.Map;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobInstance;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.batch.support.transaction.TransactionAwareProxyFactory;
import org.springframework.util.Assert;
@@ -17,11 +18,14 @@ import org.springframework.util.Assert;
public class MapJobExecutionDao implements JobExecutionDao {
private static Map executionsById = TransactionAwareProxyFactory.createTransactionalMap();
private static Map contextsByJobExecutionId = TransactionAwareProxyFactory.createTransactionalMap();
private static long currentId;
private static long currentId = 0;
public static void clear() {
executionsById.clear();
contextsByJobExecutionId.clear();
}
public int getJobExecutionCount(JobInstance jobInstance) {
@@ -78,4 +82,13 @@ public class MapJobExecutionDao implements JobExecutionDao {
}
return lastExec;
}
public ExecutionContext findExecutionContext(JobExecution jobExecution) {
return (ExecutionContext) contextsByJobExecutionId.get(jobExecution.getId());
}
public void saveOrUpdateExecutionContext(JobExecution jobExecution) {
contextsByJobExecutionId.put(jobExecution.getId(), jobExecution.getExecutionContext());
}
}