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

BATCH-1684 - Allow serializer to be injected into JobRepository

Browse Source 
* Updated the job-repository tag to allow for a serializer and deserializer to be injected
* Updated the XStreamExecutionContextStringSerializer to implement Serializer and Deserializer
* Replaced the ExecutionContextStringSerializer with the ExecutionContextSerializer interface
* Updated the XSD and parser to accept the injection of an implementation of the ExecutionContextSerializer from the job-repository tag
* Updated the JdbcExecutionContextDao to use injected ExecutionContextSerializer implementation

Reference: https://jira.springsource.org/browse/BATCH-1684
This commit is contained in:
Michael Minella
2012-10-03 18:14:22 -05:00
committed by Gunnar Hillert
parent 38ef37b882
commit 64ca2608ad
15 changed files with 747 additions and 341 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
spring-batch-core-tests/pom.xml
View File

@@ -83,7 +83,7 @@
<dependency>
<groupId>org.springframework.batch</groupId>
<artifactId>spring-batch-core</artifactId>
<version>${project.version}</version>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>commons-dbcp</groupId>
@@ -132,6 +132,10 @@
<groupId>org.springframework</groupId>
<artifactId>spring-jdbc</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.retry</groupId>
<artifactId>spring-retry</artifactId>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-test</artifactId>

4
spring-batch-core-tests/src/main/resources/META-INF/batch/footballSkipJob.xml
View File

@@ -23,7 +23,7 @@
</step>
<step id="playerSummarization" parent="summarizationStep" />
</job>
<bean id="skipPolicy" class="org.springframework.batch.core.step.skip.LimitCheckingItemSkipPolicy" scope="step">
<property name="skipLimit" value="#{jobParameters['skip.limit']}" />
<property name="skippableExceptionMap">
@@ -33,7 +33,7 @@
</property>
</bean>
<bean id="retryPolicy" class="org.springframework.batch.retry.policy.SimpleRetryPolicy" scope="step">
<bean id="retryPolicy" class="org.springframework.retry.policy.SimpleRetryPolicy" scope="step">
<property name="maxAttempts" value="#{jobParameters['retry.limit']}" />
<property name="retryableExceptions">
<map key-type="java.lang.Class">

10
spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/JobRepositoryParser.java
View File

@@ -29,13 +29,14 @@ import org.w3c.dom.Element;
/**
* Parser for the lt;job-repository/gt; element in the Batch namespace. Sets up
* and returns a JobRepositoryFactoryBean.
*
*
* @author Thomas Risberg
* @since 2.0
*
*
*/
public class JobRepositoryParser extends AbstractSingleBeanDefinitionParser {
@Override
protected String getBeanClassName(Element element) {
return "org.springframework.batch.core.repository.support.JobRepositoryFactoryBean";
}
@@ -75,6 +76,8 @@ public class JobRepositoryParser extends AbstractSingleBeanDefinitionParser {
String lobHandler = element.getAttribute("lob-handler");
String serializer = element.getAttribute("serializer");
RuntimeBeanReference ds = new RuntimeBeanReference(dataSource);
builder.addPropertyValue("dataSource", ds);
RuntimeBeanReference tx = new RuntimeBeanReference(transactionManager);
@@ -92,6 +95,9 @@ public class JobRepositoryParser extends AbstractSingleBeanDefinitionParser {
if (StringUtils.hasText(maxVarCharLength)) {
builder.addPropertyValue("maxVarCharLength", maxVarCharLength);
}
if (StringUtils.hasText(serializer)) {
builder.addPropertyReference("serializer", serializer);
}
builder.setRole(BeanDefinition.ROLE_SUPPORT);

34
spring-batch-core/src/main/java/org/springframework/batch/core/repository/ExecutionContextSerializer.java Normal file
View File

@@ -0,0 +1,34 @@
/*
* Copyright 2006-2012 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.batch.core.repository;
import org.springframework.core.serializer.Deserializer;
import org.springframework.core.serializer.Serializer;
/**
* A composite interface that combines both serialization and deserialization
* of an execution context into a single implementation. Implementations of this
* interface are used to serialize the execution context for persistence during
* the execution of a job.
*
* @author Michael Minella
* @since 2.2
* @see Serializer
* @see Deserializer
*/
public interface ExecutionContextSerializer extends Serializer, Deserializer {
}

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

@@ -0,0 +1,55 @@
/**
*
*/
package org.springframework.batch.core.repository.dao;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import org.springframework.batch.core.repository.ExecutionContextSerializer;
import org.springframework.core.serializer.DefaultDeserializer;
import org.springframework.core.serializer.DefaultSerializer;
import org.springframework.core.serializer.Deserializer;
import org.springframework.core.serializer.Serializer;
import org.springframework.util.Assert;
/**
* An implementation of the {@link ExecutionContextSerializer} using the default
* serialization implementations from Spring ({@link DefaultSerializer} and
* {@link DefaultDeserializer}).
*
* @author Michael Minella
* @since 2.2
*/
public class DefaultExecutionContextSerializer implements ExecutionContextSerializer {
private Serializer serializer = new DefaultSerializer();
private Deserializer deserializer = new DefaultDeserializer();
/**
* Serializes an execution context to the provided {@link OutputStream}. The
* stream is not closed prior to it's return.
*
* @param context
* @param out
*/
@SuppressWarnings("unchecked")
public void serialize(Object context, OutputStream out) throws IOException {
Assert.notNull(context);
Assert.notNull(out);
serializer.serialize(context, out);
}
/**
* Deserializes an execution context from the provided {@link InputStream}.
*
* @param inputStream
* @return the object serialized in the provided {@link InputStream}
*/
public Object deserialize(InputStream inputStream) throws IOException {
return deserializer.deserialize(inputStream);
}
}

45
spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/ExecutionContextStringSerializer.java
View File

@@ -1,45 +0,0 @@
/*
* Copyright 2006-2008 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.batch.core.repository.dao;
import java.util.Map;
/**
* Interface defining serialization support for execution context Map in the form of a String.
*
* @author Thomas Risberg
* @since 2.0
*/
public interface ExecutionContextStringSerializer {
/**
* Serialize the context to a string representation
*
* @param context the object that should be serialized
* @return the serialization string
*/
String serialize(Map<String, Object> context);
/**
* De-serialize the context from a string representation
*
* @param context the serialization string that should be de-serialized
* @return the de-serialized context map
*/
Map<String, Object> deserialize(String context);
}

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

@@ -1,230 +1,261 @@
/*
* Copyright 2006-2008 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.batch.core.repository.dao;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.jdbc.core.PreparedStatementSetter;
import org.springframework.jdbc.core.simple.ParameterizedRowMapper;
import org.springframework.jdbc.support.lob.DefaultLobHandler;
import org.springframework.jdbc.support.lob.LobHandler;
import org.springframework.util.Assert;
/**
* JDBC DAO for {@link ExecutionContext}.
*
* Stores execution context data related to both Step and Job using
* a different table for each.
*
* @author Lucas Ward
* @author Robert Kasanicky
* @author Thomas Risberg
*/
public class JdbcExecutionContextDao extends AbstractJdbcBatchMetadataDao implements ExecutionContextDao {
private static final String FIND_JOB_EXECUTION_CONTEXT = "SELECT SHORT_CONTEXT, SERIALIZED_CONTEXT "
+ "FROM %PREFIX%JOB_EXECUTION_CONTEXT WHERE JOB_EXECUTION_ID = ?";
private static final String INSERT_JOB_EXECUTION_CONTEXT = "INSERT INTO %PREFIX%JOB_EXECUTION_CONTEXT "
+ "(SHORT_CONTEXT, SERIALIZED_CONTEXT, JOB_EXECUTION_ID) " + "VALUES(?, ?, ?)";
private static final String UPDATE_JOB_EXECUTION_CONTEXT = "UPDATE %PREFIX%JOB_EXECUTION_CONTEXT "
+ "SET SHORT_CONTEXT = ?, SERIALIZED_CONTEXT = ? " + "WHERE JOB_EXECUTION_ID = ?";
private static final String FIND_STEP_EXECUTION_CONTEXT = "SELECT SHORT_CONTEXT, SERIALIZED_CONTEXT "
+ "FROM %PREFIX%STEP_EXECUTION_CONTEXT WHERE STEP_EXECUTION_ID = ?";
private static final String INSERT_STEP_EXECUTION_CONTEXT = "INSERT INTO %PREFIX%STEP_EXECUTION_CONTEXT "
+ "(SHORT_CONTEXT, SERIALIZED_CONTEXT, STEP_EXECUTION_ID) " + "VALUES(?, ?, ?)";
private static final String UPDATE_STEP_EXECUTION_CONTEXT = "UPDATE %PREFIX%STEP_EXECUTION_CONTEXT "
+ "SET SHORT_CONTEXT = ?, SERIALIZED_CONTEXT = ? " + "WHERE STEP_EXECUTION_ID = ?";
private static final int DEFAULT_MAX_VARCHAR_LENGTH = 2500;
private int shortContextLength = DEFAULT_MAX_VARCHAR_LENGTH;
private LobHandler lobHandler = new DefaultLobHandler();
private ExecutionContextStringSerializer serializer;
/**
* The maximum size that an execution context can have and still be stored
* completely in short form in the column <code>SHORT_CONTEXT</code>.
* Anything longer than this will overflow into large-object storage, and
* the first part only will be retained in the short form for readability.
* Default value is 2500. Clients using multi-bytes charsets on the database
* server may need to reduce this value to as little as half the value of
* the column size.
* @param shortContextLength
*/
public void setShortContextLength(int shortContextLength) {
this.shortContextLength = shortContextLength;
}
public ExecutionContext getExecutionContext(JobExecution jobExecution) {
Long executionId = jobExecution.getId();
Assert.notNull(executionId, "ExecutionId must not be null.");
List<ExecutionContext> results = getJdbcTemplate().query(getQuery(FIND_JOB_EXECUTION_CONTEXT),
new ExecutionContextRowMapper(), executionId);
if (results.size() > 0) {
return results.get(0);
}
else {
return new ExecutionContext();
}
}
public ExecutionContext getExecutionContext(StepExecution stepExecution) {
Long executionId = stepExecution.getId();
Assert.notNull(executionId, "ExecutionId must not be null.");
List<ExecutionContext> results = getJdbcTemplate().query(getQuery(FIND_STEP_EXECUTION_CONTEXT),
new ExecutionContextRowMapper(), executionId);
if (results.size() > 0) {
return results.get(0);
}
else {
return new ExecutionContext();
}
}
public void updateExecutionContext(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.");
String serializedContext = serializeContext(executionContext);
persistSerializedContext(executionId, serializedContext, UPDATE_JOB_EXECUTION_CONTEXT);
}
public void updateExecutionContext(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.");
String serializedContext = serializeContext(executionContext);
persistSerializedContext(executionId, serializedContext, UPDATE_STEP_EXECUTION_CONTEXT);
}
public void saveExecutionContext(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.");
String serializedContext = serializeContext(executionContext);
persistSerializedContext(executionId, serializedContext, INSERT_JOB_EXECUTION_CONTEXT);
}
public void saveExecutionContext(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.");
String serializedContext = serializeContext(executionContext);
persistSerializedContext(executionId, serializedContext, INSERT_STEP_EXECUTION_CONTEXT);
}
public void setLobHandler(LobHandler lobHandler) {
this.lobHandler = lobHandler;
}
@Override
public void afterPropertiesSet() throws Exception {
super.afterPropertiesSet();
serializer = new XStreamExecutionContextStringSerializer();
((XStreamExecutionContextStringSerializer) serializer).afterPropertiesSet();
}
/**
* @param executionId
* @param serializedContext
* @param sql with parameters (shortContext, longContext, executionId)
*/
private void persistSerializedContext(final Long executionId, String serializedContext, String sql) {
final String shortContext;
final String longContext;
if (serializedContext.length() > shortContextLength) {
// Overestimate length of ellipsis to be on the safe side with
// 2-byte chars
shortContext = serializedContext.substring(0, shortContextLength - 8) + " ...";
longContext = serializedContext;
}
else {
shortContext = serializedContext;
longContext = null;
}
getJdbcTemplate().update(getQuery(sql), new PreparedStatementSetter() {
public void setValues(PreparedStatement ps) throws SQLException {
ps.setString(1, shortContext);
if (longContext != null) {
lobHandler.getLobCreator().setClobAsString(ps, 2, longContext);
}
else {
ps.setNull(2, getClobTypeToUse());
}
ps.setLong(3, executionId);
}
});
}
private String serializeContext(ExecutionContext ctx) {
Map<String, Object> m = new HashMap<String, Object>();
for (Entry<String, Object> me : ctx.entrySet()) {
m.put(me.getKey(), me.getValue());
}
return serializer.serialize(m);
}
private class ExecutionContextRowMapper implements ParameterizedRowMapper<ExecutionContext> {
public ExecutionContext mapRow(ResultSet rs, int i) throws SQLException {
ExecutionContext executionContext = new ExecutionContext();
String serializedContext = rs.getString("SERIALIZED_CONTEXT");
if (serializedContext == null) {
serializedContext = rs.getString("SHORT_CONTEXT");
}
Map<String, Object> map = serializer.deserialize(serializedContext);
for (Map.Entry<String, Object> entry : map.entrySet()) {
executionContext.put(entry.getKey(), entry.getValue());
}
return executionContext;
}
}
}
/*
* Copyright 2006-2008 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.batch.core.repository.dao;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.repository.ExecutionContextSerializer;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.core.serializer.Serializer;
import org.springframework.jdbc.core.PreparedStatementSetter;
import org.springframework.jdbc.core.simple.ParameterizedRowMapper;
import org.springframework.jdbc.support.lob.DefaultLobHandler;
import org.springframework.jdbc.support.lob.LobHandler;
import org.springframework.util.Assert;
/**
* JDBC DAO for {@link ExecutionContext}.
*
* Stores execution context data related to both Step and Job using
* a different table for each.
*
* @author Lucas Ward
* @author Robert Kasanicky
* @author Thomas Risberg
* @author Michael Minella
*/
public class JdbcExecutionContextDao extends AbstractJdbcBatchMetadataDao implements ExecutionContextDao {
private static final String FIND_JOB_EXECUTION_CONTEXT = "SELECT SHORT_CONTEXT, SERIALIZED_CONTEXT "
+ "FROM %PREFIX%JOB_EXECUTION_CONTEXT WHERE JOB_EXECUTION_ID = ?";
private static final String INSERT_JOB_EXECUTION_CONTEXT = "INSERT INTO %PREFIX%JOB_EXECUTION_CONTEXT "
+ "(SHORT_CONTEXT, SERIALIZED_CONTEXT, JOB_EXECUTION_ID) " + "VALUES(?, ?, ?)";
private static final String UPDATE_JOB_EXECUTION_CONTEXT = "UPDATE %PREFIX%JOB_EXECUTION_CONTEXT "
+ "SET SHORT_CONTEXT = ?, SERIALIZED_CONTEXT = ? " + "WHERE JOB_EXECUTION_ID = ?";
private static final String FIND_STEP_EXECUTION_CONTEXT = "SELECT SHORT_CONTEXT, SERIALIZED_CONTEXT "
+ "FROM %PREFIX%STEP_EXECUTION_CONTEXT WHERE STEP_EXECUTION_ID = ?";
private static final String INSERT_STEP_EXECUTION_CONTEXT = "INSERT INTO %PREFIX%STEP_EXECUTION_CONTEXT "
+ "(SHORT_CONTEXT, SERIALIZED_CONTEXT, STEP_EXECUTION_ID) " + "VALUES(?, ?, ?)";
private static final String UPDATE_STEP_EXECUTION_CONTEXT = "UPDATE %PREFIX%STEP_EXECUTION_CONTEXT "
+ "SET SHORT_CONTEXT = ?, SERIALIZED_CONTEXT = ? " + "WHERE STEP_EXECUTION_ID = ?";
private static final int DEFAULT_MAX_VARCHAR_LENGTH = 2500;
private int shortContextLength = DEFAULT_MAX_VARCHAR_LENGTH;
private LobHandler lobHandler = new DefaultLobHandler();
private ExecutionContextSerializer serializer;
/**
* Setter for {@link Serializer} implementation
*
* @param serializer
*/
public void setSerializer(ExecutionContextSerializer serializer) {
this.serializer = serializer;
}
/**
* The maximum size that an execution context can have and still be stored
* completely in short form in the column <code>SHORT_CONTEXT</code>.
* Anything longer than this will overflow into large-object storage, and
* the first part only will be retained in the short form for readability.
* Default value is 2500. Clients using multi-bytes charsets on the database
* server may need to reduce this value to as little as half the value of
* the column size.
* @param shortContextLength
*/
public void setShortContextLength(int shortContextLength) {
this.shortContextLength = shortContextLength;
}
public ExecutionContext getExecutionContext(JobExecution jobExecution) {
Long executionId = jobExecution.getId();
Assert.notNull(executionId, "ExecutionId must not be null.");
List<ExecutionContext> results = getJdbcTemplate().query(getQuery(FIND_JOB_EXECUTION_CONTEXT),
new ExecutionContextRowMapper(), executionId);
if (results.size() > 0) {
return results.get(0);
}
else {
return new ExecutionContext();
}
}
public ExecutionContext getExecutionContext(StepExecution stepExecution) {
Long executionId = stepExecution.getId();
Assert.notNull(executionId, "ExecutionId must not be null.");
List<ExecutionContext> results = getJdbcTemplate().query(getQuery(FIND_STEP_EXECUTION_CONTEXT),
new ExecutionContextRowMapper(), executionId);
if (results.size() > 0) {
return results.get(0);
}
else {
return new ExecutionContext();
}
}
public void updateExecutionContext(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.");
String serializedContext = serializeContext(executionContext);
persistSerializedContext(executionId, serializedContext, UPDATE_JOB_EXECUTION_CONTEXT);
}
public void updateExecutionContext(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.");
String serializedContext = serializeContext(executionContext);
persistSerializedContext(executionId, serializedContext, UPDATE_STEP_EXECUTION_CONTEXT);
}
public void saveExecutionContext(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.");
String serializedContext = serializeContext(executionContext);
persistSerializedContext(executionId, serializedContext, INSERT_JOB_EXECUTION_CONTEXT);
}
public void saveExecutionContext(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.");
String serializedContext = serializeContext(executionContext);
persistSerializedContext(executionId, serializedContext, INSERT_STEP_EXECUTION_CONTEXT);
}
public void setLobHandler(LobHandler lobHandler) {
this.lobHandler = lobHandler;
}
@Override
public void afterPropertiesSet() throws Exception {
super.afterPropertiesSet();
}
/**
* @param executionId
* @param serializedContext
* @param sql with parameters (shortContext, longContext, executionId)
*/
private void persistSerializedContext(final Long executionId, String serializedContext, String sql) {
final String shortContext;
final String longContext;
if (serializedContext.length() > shortContextLength) {
// Overestimate length of ellipsis to be on the safe side with
// 2-byte chars
shortContext = serializedContext.substring(0, shortContextLength - 8) + " ...";
longContext = serializedContext;
}
else {
shortContext = serializedContext;
longContext = null;
}
getJdbcTemplate().update(getQuery(sql), new PreparedStatementSetter() {
public void setValues(PreparedStatement ps) throws SQLException {
ps.setString(1, shortContext);
if (longContext != null) {
lobHandler.getLobCreator().setClobAsString(ps, 2, longContext);
}
else {
ps.setNull(2, getClobTypeToUse());
}
ps.setLong(3, executionId);
}
});
}
private String serializeContext(ExecutionContext ctx) {
Map<String, Object> m = new HashMap<String, Object>();
for (Entry<String, Object> me : ctx.entrySet()) {
m.put(me.getKey(), me.getValue());
}
ByteArrayOutputStream out = new ByteArrayOutputStream();
try {
serializer.serialize(m, out);
}
catch (IOException ioe) {
throw new IllegalArgumentException("Could not serialize the execution context", ioe);
}
return out.toString();
}
@SuppressWarnings("unchecked")
private class ExecutionContextRowMapper implements ParameterizedRowMapper<ExecutionContext> {
public ExecutionContext mapRow(ResultSet rs, int i) throws SQLException {
ExecutionContext executionContext = new ExecutionContext();
String serializedContext = rs.getString("SERIALIZED_CONTEXT");
if (serializedContext == null) {
serializedContext = rs.getString("SHORT_CONTEXT");
}
ByteArrayInputStream in = new ByteArrayInputStream(serializedContext.getBytes());
Map<String, Object> map;
try {
map = (Map<String, Object>) serializer.deserialize(in);
}
catch (IOException ioe) {
throw new IllegalArgumentException("Unable to deserialize the execution context", ioe);
}
for (Map.Entry<String, Object> entry : map.entrySet()) {
executionContext.put(entry.getKey(), entry.getValue());
}
return executionContext;
}
}
}

68
spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/XStreamExecutionContextStringSerializer.java
View File

@@ -16,22 +16,32 @@
package org.springframework.batch.core.repository.dao;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.OutputStream;
import org.springframework.batch.core.repository.ExecutionContextSerializer;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.core.serializer.Deserializer;
import org.springframework.core.serializer.Serializer;
import org.springframework.util.Assert;
import com.thoughtworks.xstream.XStream;
import com.thoughtworks.xstream.converters.reflection.ReflectionProvider;
import com.thoughtworks.xstream.io.HierarchicalStreamDriver;
import com.thoughtworks.xstream.io.json.JettisonMappedXmlDriver;
import com.thoughtworks.xstream.XStream;
import java.util.Map;
import org.springframework.beans.factory.InitializingBean;
/**
* Implementation that uses XStream and Jettison to provide serialization.
*
*
* @author Thomas Risberg
* @author Michael Minella
* @since 2.0
* @see ExecutionContextSerializer
*/
public class XStreamExecutionContextStringSerializer implements ExecutionContextStringSerializer, InitializingBean {
public class XStreamExecutionContextStringSerializer implements ExecutionContextSerializer, InitializingBean {
private ReflectionProvider reflectionProvider = null;
@@ -39,15 +49,6 @@ public class XStreamExecutionContextStringSerializer implements ExecutionContext
private XStream xstream;
public String serialize(Map<String, Object> context) {
return xstream.toXML(context);
}
@SuppressWarnings("unchecked")
public Map<String, Object> deserialize(String context) {
return (Map<String, Object>) xstream.fromXML(context);
}
public void setReflectionProvider(ReflectionProvider reflectionProvider) {
this.reflectionProvider = reflectionProvider;
}
@@ -71,4 +72,39 @@ public class XStreamExecutionContextStringSerializer implements ExecutionContext
xstream = new XStream(reflectionProvider, hierarchicalStreamDriver);
}
}
/**
* Serializes the passed execution context to the supplied OutputStream.
*
* @param context
* @param out
* @see Serializer#serialize(Object, OutputStream)
*/
public void serialize(Object context, OutputStream out) throws IOException {
Assert.notNull(context);
Assert.notNull(out);
out.write(xstream.toXML(context).getBytes());
}
/**
* Deserializes the supplied input stream into a new execution context.
*
* @param in
* @return a reconstructed execution context
* @see Deserializer#deserialize(InputStream)
*/
@SuppressWarnings("unchecked")
public Object deserialize(InputStream in) throws IOException {
BufferedReader br = new BufferedReader(new InputStreamReader(in));
StringBuilder sb = new StringBuilder();
String line;
while ((line = br.readLine()) != null) {
sb.append(line);
}
return xstream.fromXML(sb.toString());
}
}

38
spring-batch-core/src/main/java/org/springframework/batch/core/repository/support/JobRepositoryFactoryBean.java
View File

@@ -24,6 +24,7 @@ import javax.sql.DataSource;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.batch.core.repository.ExecutionContextSerializer;
import org.springframework.batch.core.repository.dao.AbstractJdbcBatchMetadataDao;
import org.springframework.batch.core.repository.dao.ExecutionContextDao;
import org.springframework.batch.core.repository.dao.JdbcExecutionContextDao;
@@ -33,6 +34,7 @@ import org.springframework.batch.core.repository.dao.JdbcStepExecutionDao;
import org.springframework.batch.core.repository.dao.JobExecutionDao;
import org.springframework.batch.core.repository.dao.JobInstanceDao;
import org.springframework.batch.core.repository.dao.StepExecutionDao;
import org.springframework.batch.core.repository.dao.XStreamExecutionContextStringSerializer;
import org.springframework.batch.item.database.support.DataFieldMaxValueIncrementerFactory;
import org.springframework.batch.item.database.support.DefaultDataFieldMaxValueIncrementerFactory;
import org.springframework.batch.support.DatabaseType;
@@ -50,10 +52,11 @@ import org.springframework.util.StringUtils;
* {@link SimpleJobRepository} using JDBC DAO implementations which persist
* batch metadata in database. Requires the user to describe what kind of
* database they are using.
*
*
* @author Ben Hale
* @author Lucas Ward
* @author Dave Syer
* @author Michael Minella
*/
public class JobRepositoryFactoryBean extends AbstractJobRepositoryFactoryBean implements InitializingBean {
@@ -73,13 +76,26 @@ public class JobRepositoryFactoryBean extends AbstractJobRepositoryFactoryBean i
private LobHandler lobHandler;
private ExecutionContextSerializer serializer;
/**
* A custom implementation of the {@link ExecutionContextSerializer}.
* The default, if not injected, is the {@link XStreamExecutionContextStringSerializer}.
*
* @param serializer
* @see ExecutionContextSerializer
*/
public void setSerializer(ExecutionContextSerializer serializer) {
this.serializer = serializer;
}
/**
* A special handler for large objects. The default is usually fine, except
* for some (usually older) versions of Oracle. The default is determined
* from the data base type.
*
*
* @param lobHandler the {@link LobHandler} to set
*
*
* @see LobHandler
*/
public void setLobHandler(LobHandler lobHandler) {
@@ -95,7 +111,7 @@ public class JobRepositoryFactoryBean extends AbstractJobRepositoryFactoryBean i
* multi-byte character sets this number can be smaller (by up to a factor
* of 2 for 2-byte characters) than the declaration of the column length in
* the DDL for the tables.
*
*
* @param maxVarCharLength the exitMessageLength to set
*/
public void setMaxVarCharLength(int maxVarCharLength) {
@@ -131,6 +147,7 @@ public class JobRepositoryFactoryBean extends AbstractJobRepositoryFactoryBean i
this.incrementerFactory = incrementerFactory;
}
@Override
public void afterPropertiesSet() throws Exception {
Assert.notNull(dataSource, "DataSource must not be null.");
@@ -150,6 +167,13 @@ public class JobRepositoryFactoryBean extends AbstractJobRepositoryFactoryBean i
lobHandler = new OracleLobHandler();
}
if(serializer == null) {
XStreamExecutionContextStringSerializer defaultSerializer = new XStreamExecutionContextStringSerializer();
defaultSerializer.afterPropertiesSet();
serializer = defaultSerializer;
}
Assert.isTrue(incrementerFactory.isSupportedIncrementerType(databaseType), "'" + databaseType
+ "' is an unsupported database type. The supported database types are "
+ StringUtils.arrayToCommaDelimitedString(incrementerFactory.getSupportedIncrementerTypes()));
@@ -200,9 +224,15 @@ public class JobRepositoryFactoryBean extends AbstractJobRepositoryFactoryBean i
dao.setJdbcTemplate(jdbcTemplate);
dao.setTablePrefix(tablePrefix);
dao.setClobTypeToUse(determineClobTypeToUse(this.databaseType));
if (lobHandler != null) {
dao.setLobHandler(lobHandler);
}
if(serializer != null) {
dao.setSerializer(serializer);
}
dao.afterPropertiesSet();
// Assume the same length.
dao.setShortContextLength(maxVarCharLength);

48
spring-batch-core/src/main/resources/org/springframework/batch/core/configuration/xml/spring-batch-2.1.xsd
View File

@@ -71,9 +71,9 @@
<xsd:attributeGroup ref="jobRepositoryAttribute" />
<xsd:attribute name="incrementer" type="xsd:string" use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[
A reference to a JobParametersIncrementer bean definition. This will be
used to provide new parameters to a Job instance that is starting in a
<xsd:documentation><![CDATA[
A reference to a JobParametersIncrementer bean definition. This will be
used to provide new parameters to a Job instance that is starting in a
sequence.
]]>
</xsd:documentation>
@@ -174,7 +174,7 @@
<xsd:element name="job-repository">
<xsd:annotation>
<xsd:documentation><![CDATA[
Configures a JobRepository using a relational database. This is
Configures a JobRepository using a relational database. This is
needed by many other components (principally Job and Step implementations).
]]></xsd:documentation>
</xsd:annotation>
@@ -197,7 +197,7 @@ ref" is not required, and only needs to be specified explicitly
<xsd:attribute name="transaction-manager" type="xsd:string" default="transactionManager">
<xsd:annotation>
<xsd:documentation source="java:org.springframework.transaction.PlatformTransactionManager"><![CDATA[
The bean name of the TransactionManager that is to be used. This attribute
The bean name of the TransactionManager that is to be used. This attribute
is not required, and only needs to be specified explicitly
if the bean name of the desired TransactionManager is not 'transactionManager'.
]]></xsd:documentation>
@@ -212,8 +212,8 @@ ref" is not required, and only needs to be specified explicitly
<xsd:annotation>
<xsd:documentation><![CDATA[
The isolation level to use for creation of job execution entities.
The default is SERIALIZABLE, which prevents accidental
concurrent execution of the same job (REPEATABLE_READ
The default is SERIALIZABLE, which prevents accidental
concurrent execution of the same job (REPEATABLE_READ
would work as well).
]]></xsd:documentation>
</xsd:annotation>
@@ -235,6 +235,20 @@ ref" is not required, and only needs to be specified explicitly
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="serializer" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
The bean name of the Serializer that is to be used. This attribute
is not required. Consideration should be given between the type of
serialization used and the max-varchar-length.
]]></xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.batch.core.repository.ExecutionContextSerializer" />
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="lob-handler" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
@@ -339,7 +353,7 @@ ref" is not required, and only needs to be specified explicitly
<xsd:attribute name="parent" type="xsd:string" use="required">
<xsd:annotation>
<xsd:documentation source="java:org.springframework.batch.core.job.flow.Flow"><![CDATA[
The flow that will execute at this point in the job specified as a
The flow that will execute at this point in the job specified as a
parent bean definition id.
]]></xsd:documentation>
<xsd:appinfo>
@@ -567,7 +581,7 @@ ref" is not required, and only needs to be specified explicitly
List of exception classes that should not cause rollback if possible. This list
is only a hint and has to be interpreted by the step to make sense in context (e.g.
it might not be possible to honour the hint during a write operation, so consider moving
code that throws these exceptions to a processor or validator).
code that throws these exceptions to a processor or validator).
]]>
</xsd:documentation>
</xsd:annotation>
@@ -619,7 +633,7 @@ ref" is not required, and only needs to be specified explicitly
<xsd:attribute name="transaction-manager" type="xsd:string">
<xsd:annotation>
<xsd:documentation source="java:org.springframework.transaction.PlatformTransactionManager"><![CDATA[
The bean name of the TransactionManager that is to be used. This attribute
The bean name of the TransactionManager that is to be used. This attribute
is not required, and only needs to be specified explicitly
if the bean name of the desired TransactionManager is not 'transactionManager'.
]]></xsd:documentation>
@@ -645,8 +659,8 @@ ref" is not required, and only needs to be specified explicitly
<xsd:attribute name="throttle-limit" type="xsd:string" use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[
This limits the number of tasks queued for concurrent
processing to prevent thread pools from being overwhelmed.
This limits the number of tasks queued for concurrent
processing to prevent thread pools from being overwhelmed.
Default is 4.
]]></xsd:documentation>
</xsd:annotation>
@@ -814,7 +828,7 @@ ref" is not required, and only needs to be specified explicitly
<xsd:annotation>
<xsd:documentation><![CDATA[
List of exception classes that are skippable. Exceptions (and their subclasses) that
are declared as included take precedence over the same value if it is also excluded.
are declared as included take precedence over the same value if it is also excluded.
Exceptions that are already marked as no-rollback
are automatically skippable (but it doesn't hurt to add them again here).
]]>
@@ -843,7 +857,7 @@ ref" is not required, and only needs to be specified explicitly
<xsd:attribute name="commit-interval" type="xsd:string" use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[
The number of items that will be processed before commit is called for the transaction.
The number of items that will be processed before commit is called for the transaction.
Either set this or the chunk-completion-policy but not both. Can be specified as an expression
that will be evaluated in the scope of the step (e.g. "#{jobParameters['commit.interval']}").
]]></xsd:documentation>
@@ -928,7 +942,7 @@ ref" is not required, and only needs to be specified explicitly
<xsd:attribute name="reader-transactional-queue" type="xsd:string" use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[
Whether the reader is a transactional queue. If it is then items read should not be cached
Whether the reader is a transactional queue. If it is then items read should not be cached
in the event of a rollback since they will be returned to the queue. Default is false.
]]></xsd:documentation>
</xsd:annotation>
@@ -938,7 +952,7 @@ ref" is not required, and only needs to be specified explicitly
<xsd:documentation><![CDATA[
Whether the processor is transaction aware. If it is then processed items should not be
cached in between transactions in case of a rollback. N.B. if reader-transactional-queue
is true then so should this be. Default is true. If false then the processor is only called
is true then so should this be. Default is true. If false then the processor is only called
once per item per chunk, even if there are rollbacks with retries and skips.
]]></xsd:documentation>
</xsd:annotation>
@@ -1217,7 +1231,7 @@ ref" is not required, and only needs to be specified explicitly
<xsd:attribute name="job-repository" type="xsd:string">
<xsd:annotation>
<xsd:documentation source="java:org.springframework.batch.core.repository.JobRepository"><![CDATA[
The bean name of the JobRepository that is to be used. This attribute
The bean name of the JobRepository that is to be used. This attribute
is not required, and only needs to be specified explicitly
if the bean name of the desired JobRepository is not 'jobRepository'.
]]></xsd:documentation>

169
spring-batch-core/src/test/java/org/springframework/batch/core/repository/dao/DefaultExecutionContextSerializerTests.java Normal file
View File

@@ -0,0 +1,169 @@
/**
*
*/
package org.springframework.batch.core.repository.dao;
import static org.junit.Assert.assertEquals;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.Serializable;
import java.math.BigDecimal;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import org.junit.Before;
import org.junit.Test;
/**
* @author Michael Minella
*
*/
public class DefaultExecutionContextSerializerTests {
private DefaultExecutionContextSerializer serializer;
/**
* @throws java.lang.Exception
*/
@Before
public void setUp() throws Exception {
serializer = new DefaultExecutionContextSerializer();
}
@Test
public void testSerializeAMap() throws Exception {
Map<String, Object> m1 = new HashMap<String, Object>();
m1.put("object1", Long.valueOf(12345L));
m1.put("object2", "OBJECT TWO");
// Use a date after 1971 (otherwise daylight saving screws up)...
m1.put("object3", new Date(123456790123L));
m1.put("object4", new Double(1234567.1234D));
Map<String, Object> m2 = serializationRoundTrip(m1);
compareContexts(m1, m2);
}
@Test
public void testComplexObject() throws Exception {
Map<String, Object> m1 = new HashMap<String, Object>();
ComplexObject o1 = new ComplexObject();
o1.setName("02345");
Map<String, Object> m = new HashMap<String, Object>();
m.put("object1", Long.valueOf(12345L));
m.put("object2", "OBJECT TWO");
o1.setMap(m);
o1.setNumber(new BigDecimal("12345.67"));
ComplexObject o2 = new ComplexObject();
o2.setName("Inner Object");
o2.setMap(m);
o2.setNumber(new BigDecimal("98765.43"));
o1.setObj(o2);
m1.put("co", o1);
Map<String, Object> m2 = serializationRoundTrip(m1);
compareContexts(m1, m2);
}
@Test (expected=IllegalArgumentException.class)
public void testNullSerialization() throws Exception {
serializer.serialize(null, null);
}
private void compareContexts(Map<String, Object> m1, Map<String, Object> m2) {
for (String key : m1.keySet()) {
System.out.println("m1 = " + m1 + " m2 = " + m2);
assertEquals("Bad key/value for " + key, m1.get(key), m2.get(key));
}
}
@SuppressWarnings("unchecked")
private Map<String, Object> serializationRoundTrip(Map<String, Object> m1) throws IOException {
ByteArrayOutputStream out = new ByteArrayOutputStream();
serializer.serialize(m1, out);
String s = out.toString();
InputStream in = new ByteArrayInputStream(s.getBytes());
Map<String, Object> m2 = (Map<String, Object>) serializer.deserialize(in);
return m2;
}
@SuppressWarnings("unused")
private static class ComplexObject implements Serializable {
private static final long serialVersionUID = 1L;
private String name;
private BigDecimal number;
private ComplexObject obj;
private Map<String,Object> map;
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
public BigDecimal getNumber() {
return number;
}
public void setNumber(BigDecimal number) {
this.number = number;
}
public ComplexObject getObj() {
return obj;
}
public void setObj(ComplexObject obj) {
this.obj = obj;
}
public Map<String,Object> getMap() {
return map;
}
public void setMap(Map<String,Object> map) {
this.map = map;
}
@Override
public boolean equals(Object o) {
if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
ComplexObject that = (ComplexObject) o;
if (map != null ? !map.equals(that.map) : that.map != null) return false;
if (name != null ? !name.equals(that.name) : that.name != null) return false;
if (number != null ? !number.equals(that.number) : that.number != null) return false;
if (obj != null ? !obj.equals(that.obj) : that.obj != null) return false;
return true;
}
@Override
public int hashCode() {
int result;
result = (name != null ? name.hashCode() : 0);
result = 31 * result + (number != null ? number.hashCode() : 0);
result = 31 * result + (obj != null ? obj.hashCode() : 0);
result = 31 * result + (map != null ? map.hashCode() : 0);
return result;
}
@Override
public String toString() {
return "ComplexObject [name=" + name + ", number=" + number + "]";
}
}
}

50
spring-batch-core/src/test/java/org/springframework/batch/core/repository/dao/XStreamExecutionContextStringSerializerTests.java
View File

@@ -2,6 +2,9 @@ package org.springframework.batch.core.repository.dao;
import static org.junit.Assert.assertEquals;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.math.BigDecimal;
import java.util.Date;
import java.util.HashMap;
@@ -9,22 +12,26 @@ import java.util.Map;
import org.junit.Before;
import org.junit.Test;
import org.springframework.batch.core.repository.ExecutionContextSerializer;
/**
* @author Thomas Risberg
* @author Michael Minella
*/
public class XStreamExecutionContextStringSerializerTests {
ExecutionContextStringSerializer serializer;
ExecutionContextSerializer serializer;
@Before
public void onSetUp() throws Exception {
serializer = new XStreamExecutionContextStringSerializer();
((XStreamExecutionContextStringSerializer)serializer).afterPropertiesSet();
XStreamExecutionContextStringSerializer serializerDeserializer = new XStreamExecutionContextStringSerializer();
(serializerDeserializer).afterPropertiesSet();
serializer = serializerDeserializer;
}
@Test
public void testSerializeAMap() {
public void testSerializeAMap() throws Exception {
Map<String, Object> m1 = new HashMap<String, Object>();
m1.put("object1", Long.valueOf(12345L));
m1.put("object2", "OBJECT TWO");
@@ -32,15 +39,13 @@ public class XStreamExecutionContextStringSerializerTests {
m1.put("object3", new Date(123456790123L));
m1.put("object4", new Double(1234567.1234D));
String s = serializer.serialize(m1);
Map<String, Object> m2 = serializer.deserialize(s);
Map<String, Object> m2 = serializationRoundTrip(m1);
compareContexts(m1, m2);
}
@Test
public void testComplexObject() {
public void testComplexObject() throws Exception {
Map<String, Object> m1 = new HashMap<String, Object>();
ComplexObject o1 = new ComplexObject();
o1.setName("02345");
@@ -56,19 +61,36 @@ public class XStreamExecutionContextStringSerializerTests {
o1.setObj(o2);
m1.put("co", o1);
String s = serializer.serialize(m1);
Map<String, Object> m2 = serializer.deserialize(s);
Map<String, Object> m2 = serializationRoundTrip(m1);
compareContexts(m1, m2);
}
@Test (expected=IllegalArgumentException.class)
@SuppressWarnings("unchecked")
public void testNullSerialization() throws Exception {
serializer.serialize(null, null);
}
private void compareContexts(Map<String, Object> m1, Map<String, Object> m2) {
for (String key : m1.keySet()) {
System.out.println("m1 = " + m1 + " m2 = " + m2);
assertEquals("Bad key/value for " + key, m1.get(key), m2.get(key));
}
}
@SuppressWarnings("unchecked")
private Map<String, Object> serializationRoundTrip(Map<String, Object> m1) throws IOException {
ByteArrayOutputStream out = new ByteArrayOutputStream();
serializer.serialize(m1, out);
String s = out.toString();
ByteArrayInputStream in = new ByteArrayInputStream(s.getBytes());
Map<String, Object> m2 = (Map<String, Object>) serializer.deserialize(in);
return m2;
}
@SuppressWarnings("unused")
private static class ComplexObject {
private String name;
@@ -109,6 +131,7 @@ public class XStreamExecutionContextStringSerializerTests {
}
@Override
public boolean equals(Object o) {
if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
@@ -123,6 +146,7 @@ public class XStreamExecutionContextStringSerializerTests {
return true;
}
@Override
public int hashCode() {
int result;
result = (name != null ? name.hashCode() : 0);
@@ -136,6 +160,6 @@ public class XStreamExecutionContextStringSerializerTests {
public String toString() {
return "ComplexObject [name=" + name + ", number=" + number + "]";
}
}
}

64
spring-batch-core/src/test/java/org/springframework/batch/core/repository/support/JobRepositoryFactoryBeanTests.java
View File

@@ -26,14 +26,19 @@ import static org.easymock.EasyMock.verify;
import java.sql.Connection;
import java.sql.DatabaseMetaData;
import java.util.Map;
import javax.sql.DataSource;
import org.junit.Before;
import org.junit.Test;
import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.repository.ExecutionContextSerializer;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.repository.dao.DefaultExecutionContextSerializer;
import org.springframework.batch.core.repository.dao.XStreamExecutionContextStringSerializer;
import org.springframework.batch.item.database.support.DataFieldMaxValueIncrementerFactory;
import org.springframework.core.serializer.Serializer;
import org.springframework.dao.DataAccessException;
import org.springframework.jdbc.support.incrementer.DataFieldMaxValueIncrementer;
import org.springframework.jdbc.support.lob.DefaultLobHandler;
@@ -45,7 +50,7 @@ import org.springframework.transaction.support.DefaultTransactionDefinition;
/**
* @author Lucas Ward
*
*
*/
public class JobRepositoryFactoryBeanTests {
@@ -81,26 +86,26 @@ public class JobRepositoryFactoryBeanTests {
expect(dataSource.getConnection()).andReturn(con);
expect(con.getMetaData()).andReturn(dmd);
expect(dmd.getDatabaseProductName()).andReturn("Oracle");
expect(incrementerFactory.isSupportedIncrementerType("ORACLE")).andReturn(true);
expect(incrementerFactory.getSupportedIncrementerTypes()).andReturn(new String[0]);
expect(incrementerFactory.getIncrementer("ORACLE", tablePrefix + "JOB_SEQ")).andReturn(new StubIncrementer());
expect(incrementerFactory.getIncrementer("ORACLE", tablePrefix + "JOB_EXECUTION_SEQ")).andReturn(new StubIncrementer());
expect(incrementerFactory.getIncrementer("ORACLE", tablePrefix + "STEP_EXECUTION_SEQ")).andReturn(new StubIncrementer());
replay(dataSource,con,dmd, incrementerFactory);
factory.afterPropertiesSet();
factory.getObject();
verify(incrementerFactory);
}
@Test
public void testOracleLobHandler() throws Exception {
factory.setDatabaseType("ORACLE");
incrementerFactory = createNiceMock(DataFieldMaxValueIncrementerFactory.class);
expect(incrementerFactory.isSupportedIncrementerType("ORACLE")).andReturn(true);
expect(incrementerFactory.getIncrementer("ORACLE", tablePrefix + "JOB_SEQ")).andReturn(new StubIncrementer());
@@ -112,14 +117,14 @@ public class JobRepositoryFactoryBeanTests {
factory.afterPropertiesSet();
LobHandler lobHandler = (LobHandler) ReflectionTestUtils.getField(factory, "lobHandler");
assertTrue(lobHandler instanceof OracleLobHandler);
}
@Test
public void testCustomLobHandler() throws Exception {
factory.setDatabaseType("ORACLE");
incrementerFactory = createNiceMock(DataFieldMaxValueIncrementerFactory.class);
expect(incrementerFactory.isSupportedIncrementerType("ORACLE")).andReturn(true);
expect(incrementerFactory.getIncrementer("ORACLE", tablePrefix + "JOB_SEQ")).andReturn(new StubIncrementer());
@@ -127,13 +132,52 @@ public class JobRepositoryFactoryBeanTests {
expect(incrementerFactory.getIncrementer("ORACLE", tablePrefix + "STEP_EXECUTION_SEQ")).andReturn(new StubIncrementer());
replay(dataSource,incrementerFactory);
factory.setIncrementerFactory(incrementerFactory);
LobHandler lobHandler = new DefaultLobHandler();
factory.setLobHandler(lobHandler);
factory.afterPropertiesSet();
assertEquals(lobHandler, ReflectionTestUtils.getField(factory, "lobHandler"));
}
@Test
@SuppressWarnings("unchecked")
public void tesDefaultSerializer() throws Exception {
factory.setDatabaseType("ORACLE");
incrementerFactory = createNiceMock(DataFieldMaxValueIncrementerFactory.class);
expect(incrementerFactory.isSupportedIncrementerType("ORACLE")).andReturn(true);
expect(incrementerFactory.getIncrementer("ORACLE", tablePrefix + "JOB_SEQ")).andReturn(new StubIncrementer());
expect(incrementerFactory.getIncrementer("ORACLE", tablePrefix + "JOB_EXECUTION_SEQ")).andReturn(new StubIncrementer());
expect(incrementerFactory.getIncrementer("ORACLE", tablePrefix + "STEP_EXECUTION_SEQ")).andReturn(new StubIncrementer());
replay(dataSource,incrementerFactory);
factory.setIncrementerFactory(incrementerFactory);
factory.afterPropertiesSet();
Serializer<Map<String, Object>> serializer = (Serializer<Map<String,Object>>) ReflectionTestUtils.getField(factory, "serializer");
assertTrue(serializer instanceof XStreamExecutionContextStringSerializer);
}
@Test
public void testCustomSerializer() throws Exception {
factory.setDatabaseType("ORACLE");
incrementerFactory = createNiceMock(DataFieldMaxValueIncrementerFactory.class);
expect(incrementerFactory.isSupportedIncrementerType("ORACLE")).andReturn(true);
expect(incrementerFactory.getIncrementer("ORACLE", tablePrefix + "JOB_SEQ")).andReturn(new StubIncrementer());
expect(incrementerFactory.getIncrementer("ORACLE", tablePrefix + "JOB_EXECUTION_SEQ")).andReturn(new StubIncrementer());
expect(incrementerFactory.getIncrementer("ORACLE", tablePrefix + "STEP_EXECUTION_SEQ")).andReturn(new StubIncrementer());
replay(dataSource,incrementerFactory);
factory.setIncrementerFactory(incrementerFactory);
ExecutionContextSerializer customSerializer = new DefaultExecutionContextSerializer();
factory.setSerializer(customSerializer);
factory.afterPropertiesSet();
assertEquals(customSerializer, ReflectionTestUtils.getField(factory, "serializer"));
}
@Test
@@ -161,7 +205,7 @@ public class JobRepositoryFactoryBeanTests {
expect(incrementerFactory.isSupportedIncrementerType("mockDb")).andReturn(true);
expect(incrementerFactory.getSupportedIncrementerTypes()).andReturn(new String[0]);
replay(incrementerFactory);
factory.afterPropertiesSet();
fail();
}

3
spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/xml/JobRepositoryParserTests-context.xml
View File

@@ -12,7 +12,8 @@
<beans:property name="dataSource" ref="dataSource"/>
</beans:bean>
<beans:bean id="lobHandler" class="org.springframework.jdbc.support.lob.DefaultLobHandler"/>
<beans:bean id="serializer" class="org.springframework.batch.core.repository.dao.DefaultExecutionContextSerializer"/>
<job-repository id="jobRepo1" data-source="dataSource" transaction-manager="transactionManager" lob-handler="lobHandler" max-varchar-length="100"/>
<job-repository id="jobRepo1" data-source="dataSource" transaction-manager="transactionManager" lob-handler="lobHandler" max-varchar-length="100" serializer="serializer"/>
</beans:beans>

3
spring-batch-core/src/test/resources/org/springframework/batch/core/repository/dao/sql-dao-test.xml
View File

@@ -13,6 +13,7 @@
<bean id="executionContextDao" class="org.springframework.batch.core.repository.dao.JdbcExecutionContextDao">
<property name="jdbcTemplate" ref="jdbcTemplate" />
<property name="serializer" ref="serializer"/>
</bean>
<bean id="jobInstanceDao" class="org.springframework.batch.core.repository.dao.JdbcJobInstanceDao">
@@ -33,4 +34,6 @@
<bean id="jdbcTemplate" class="org.springframework.jdbc.core.JdbcTemplate">
<constructor-arg ref="dataSource" />
</bean>
<bean id="serializer" class="org.springframework.batch.core.repository.dao.XStreamExecutionContextStringSerializer"/>
</beans>