From 69d62dd9dd310315f167d635e82fa4dd61e02b87 Mon Sep 17 00:00:00 2001 From: robokaso Date: Thu, 22 May 2008 10:44:43 +0000 Subject: [PATCH] IN PROGRESS - BATCH-505: Job-level ExecutionContext JobExecutionContext now works --- .../batch/core/JobExecution.java | 21 ++ .../dao/JdbcExecutionContextDao.java | 310 ++++++++++++++++++ .../repository/dao/JdbcJobExecutionDao.java | 22 ++ .../repository/dao/JdbcStepExecutionDao.java | 210 +----------- .../core/repository/dao/JobExecutionDao.java | 17 + .../repository/dao/MapJobExecutionDao.java | 15 +- .../dao/AbstractJobExecutionDaoTests.java | 34 ++ 7 files changed, 427 insertions(+), 202 deletions(-) create mode 100644 spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/JdbcExecutionContextDao.java diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/JobExecution.java b/spring-batch-core/src/main/java/org/springframework/batch/core/JobExecution.java index 693b37d0a..b7f1ae656 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/JobExecution.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/JobExecution.java @@ -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; + } } diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/JdbcExecutionContextDao.java b/spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/JdbcExecutionContextDao.java new file mode 100644 index 000000000..08b34d28f --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/JdbcExecutionContextDao.java @@ -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; + } + } + +} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/JdbcJobExecutionDao.java b/spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/JdbcJobExecutionDao.java index 82e6211ae..10d40b6ff 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/JdbcJobExecutionDao.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/JdbcJobExecutionDao.java @@ -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; + } + } diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/JdbcStepExecutionDao.java b/spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/JdbcStepExecutionDao.java index 3c6984eb4..e39d439cb 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/JdbcStepExecutionDao.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/JdbcStepExecutionDao.java @@ -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) { diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/JobExecutionDao.java b/spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/JobExecutionDao.java index 4e4df06c8..0d2acd697 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/JobExecutionDao.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/JobExecutionDao.java @@ -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); } diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/MapJobExecutionDao.java b/spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/MapJobExecutionDao.java index 47f1b6cfe..1574b94e7 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/MapJobExecutionDao.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/MapJobExecutionDao.java @@ -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()); + + } } diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/repository/dao/AbstractJobExecutionDaoTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/repository/dao/AbstractJobExecutionDaoTests.java index 84645be9d..84efa550e 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/repository/dao/AbstractJobExecutionDaoTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/repository/dao/AbstractJobExecutionDaoTests.java @@ -1,12 +1,14 @@ package org.springframework.batch.core.repository.dao; import java.util.Date; +import java.util.HashMap; import java.util.List; import org.springframework.batch.core.BatchStatus; import org.springframework.batch.core.JobExecution; import org.springframework.batch.core.JobInstance; import org.springframework.batch.core.JobParameters; +import org.springframework.batch.item.ExecutionContext; import org.springframework.test.AbstractTransactionalDataSourceSpringContextTests; public abstract class AbstractJobExecutionDaoTests extends AbstractTransactionalDataSourceSpringContextTests { @@ -96,4 +98,36 @@ public abstract class AbstractJobExecutionDaoTests extends AbstractTransactional assertEquals(exec2, dao.getLastJobExecution(jobInstance)); } + + public void testSaveAndFindContext() { + dao.saveJobExecution(execution); + ExecutionContext ctx = new ExecutionContext(new HashMap() { + { + put("key", "value"); + } + }); + execution.setExecutionContext(ctx); + dao.saveOrUpdateExecutionContext(execution); + + ExecutionContext retrieved = dao.findExecutionContext(execution); + assertEquals(ctx, retrieved); + } + + public void testUpdateContext() { + dao.saveJobExecution(execution); + ExecutionContext ctx = new ExecutionContext(new HashMap() { + { + put("key", "value"); + } + }); + execution.setExecutionContext(ctx); + dao.saveOrUpdateExecutionContext(execution); + + ctx.putLong("longKey", 7); + dao.saveOrUpdateExecutionContext(execution); + + ExecutionContext retrieved = dao.findExecutionContext(execution); + assertEquals(ctx, retrieved); + assertEquals(7, retrieved.getLong("longKey")); + } }