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

Upgrade to Spring Integration RC2

Browse Source
This commit is contained in:
dsyer
2008-12-02 16:17:54 +00:00
parent 9b758a59e3
commit 173e9a259c
28 changed files with 163 additions and 202 deletions
Download Patch File Download Diff File Expand all files Collapse all files

38
spring-batch-samples/src/main/java/org/springframework/batch/sample/common/StagingItemListener.java Normal file
View File

@@ -0,0 +1,38 @@
package org.springframework.batch.sample.common;
import javax.sql.DataSource;
import org.springframework.batch.core.listener.StepListenerSupport;
import org.springframework.batch.item.ItemReader;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.dao.OptimisticLockingFailureException;
import org.springframework.jdbc.core.simple.SimpleJdbcTemplate;
import org.springframework.util.Assert;
/**
* Thread-safe database {@link ItemReader} implementing the process indicator
* pattern.
*/
public class StagingItemListener extends StepListenerSupport<Long, Long> implements InitializingBean {
private SimpleJdbcTemplate jdbcTemplate;
public void setDataSource(DataSource dataSource) {
jdbcTemplate = new SimpleJdbcTemplate(dataSource);
}
public final void afterPropertiesSet() throws Exception {
Assert.notNull(jdbcTemplate, "You must provide a DataSource.");
}
@Override
public void afterRead(Long id) {
int count = jdbcTemplate.update("UPDATE BATCH_STAGING SET PROCESSED=? WHERE ID=? AND PROCESSED=?",
StagingItemWriter.DONE, id, StagingItemWriter.NEW);
if (count != 1) {
throw new OptimisticLockingFailureException("The staging record with ID=" + id
+ " was updated concurrently when trying to mark as complete (updated " + count + " records.");
}
}
}

145
spring-batch-samples/src/main/java/org/springframework/batch/sample/common/StagingItemReader.java
View File

@@ -1,96 +1,64 @@
package org.springframework.batch.sample.common;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.Iterator;
import java.util.List;
import javax.sql.DataSource;
import org.apache.commons.lang.SerializationUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.batch.core.ExitStatus;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.StepExecutionListener;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ReaderNotOpenException;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.batch.item.ItemStream;
import org.springframework.batch.item.ItemStreamException;
import org.springframework.batch.item.database.JdbcCursorItemReader;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.dao.DataAccessException;
import org.springframework.dao.OptimisticLockingFailureException;
import org.springframework.jdbc.core.PreparedStatementSetter;
import org.springframework.jdbc.core.RowMapper;
import org.springframework.jdbc.core.simple.ParameterizedRowMapper;
import org.springframework.jdbc.core.simple.SimpleJdbcTemplate;
import org.springframework.util.Assert;
/**
* Thread-safe database {@link ItemReader} implementing the process indicator
* pattern.
*/
public class StagingItemReader<T> implements ItemReader<T>, StepExecutionListener, InitializingBean, DisposableBean {
private static Log logger = LogFactory.getLog(StagingItemReader.class);
private StepExecution stepExecution;
private final Object lock = new Object();
private volatile boolean initialized = false;
private volatile Iterator<Long> keys;
public class StagingItemReader<T> implements ItemReader<T>, ItemStream, InitializingBean {
private JdbcCursorItemReader<Long> delegate;
private SimpleJdbcTemplate jdbcTemplate;
private long jobId;
public void setJobId(long jobId) {
this.jobId = jobId;
}
public void setDataSource(DataSource dataSource) {
jdbcTemplate = new SimpleJdbcTemplate(dataSource);
}
public void destroy() throws Exception {
initialized = false;
keys = null;
delegate = new JdbcCursorItemReader<Long>();
delegate.setDataSource(dataSource);
delegate.setSql("SELECT ID FROM BATCH_STAGING WHERE JOB_ID=? AND PROCESSED=? ORDER BY ID");
delegate.setPreparedStatementSetter(new PreparedStatementSetter() {
public void setValues(PreparedStatement ps) throws SQLException {
ps.setLong(1, jobId);
ps.setString(2, StagingItemWriter.NEW);
}
});
delegate.setMapper(new RowMapper() {
public Object mapRow(ResultSet rs, int rowNum) throws SQLException {
return rs.getLong(1);
}
});
}
public final void afterPropertiesSet() throws Exception {
Assert.notNull(jdbcTemplate, "You must provide a DataSource.");
delegate.afterPropertiesSet();
}
private List<Long> retrieveKeys() {
public T read() throws Exception {
synchronized (lock) {
return jdbcTemplate.query(
"SELECT ID FROM BATCH_STAGING WHERE JOB_ID=? AND PROCESSED=? ORDER BY ID",
new ParameterizedRowMapper<Long>() {
public Long mapRow(ResultSet rs, int rowNum) throws SQLException {
return rs.getLong(1);
}
},
stepExecution.getJobExecution().getJobId(), StagingItemWriter.NEW);
}
}
public T read() throws DataAccessException {
if (!initialized) {
throw new ReaderNotOpenException("ItemStream must be open before it can be read.");
}
Long id = null;
synchronized (lock) {
if (keys.hasNext()) {
id = keys.next();
}
}
logger.debug("Retrieved key from list: " + id);
if (id == null) {
return null;
}
Long id = delegate.read();
@SuppressWarnings("unchecked")
T result = (T) jdbcTemplate.queryForObject("SELECT VALUE FROM BATCH_STAGING WHERE ID=?",
new ParameterizedRowMapper<Object>() {
@@ -100,55 +68,20 @@ public class StagingItemReader<T> implements ItemReader<T>, StepExecutionListene
}
}, id);
// Update now - changes will rollback if there is a problem later.
int count = jdbcTemplate.update("UPDATE BATCH_STAGING SET PROCESSED=? WHERE ID=? AND PROCESSED=?",
StagingItemWriter.DONE, id, StagingItemWriter.NEW);
if (count != 1) {
throw new OptimisticLockingFailureException("The staging record with ID=" + id
+ " was updated concurrently when trying to mark as complete (updated " + count + " records.");
}
return result;
}
/*
* (non-Javadoc)
*
* @see
* org.springframework.batch.core.domain.StepListener#afterStep(StepExecution
* )
*/
public ExitStatus afterStep(StepExecution stepExecution) {
return null;
public void close(ExecutionContext executionContext) throws ItemStreamException {
delegate.close(executionContext);
}
/*
* (non-Javadoc)
*
* @seeorg.springframework.batch.core.domain.StepListener#beforeStep(org.
* springframework.batch.core.domain.StepExecution)
*/
public void beforeStep(StepExecution stepExecution) {
this.stepExecution = stepExecution;
synchronized (lock) {
if (keys == null) {
keys = retrieveKeys().iterator();
logger.info("Keys obtained for staging.");
initialized = true;
}
}
public void open(ExecutionContext executionContext) throws ItemStreamException {
delegate.open(executionContext);
}
/*
* (non-Javadoc)
*
* @see
* org.springframework.batch.core.domain.StepListener#onErrorInStep(java
* .lang.Throwable)
*/
public ExitStatus onErrorInStep(StepExecution stepExecution, Throwable e) {
return null;
public void update(ExecutionContext executionContext) throws ItemStreamException {
delegate.update(executionContext);
}
}

5
spring-batch-samples/src/main/resources/jobs/parallelJob.xml
View File

@@ -46,6 +46,11 @@
<property name="dao" ref="tradeDao" />
</bean>
</property>
<property name="listeners">
<bean class="org.springframework.batch.sample.common.StagingItemListener">
<property name="dataSource" ref="dataSource" />
</bean>
</property>
</bean>
</list>
</property>

12
spring-batch-samples/src/test/java/org/springframework/batch/sample/common/StagingItemReaderTests.java
View File

@@ -13,6 +13,7 @@ import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobInstance;
import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.UnexpectedJobExecutionException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jdbc.core.simple.SimpleJdbcTemplate;
import org.springframework.test.context.ContextConfiguration;
@@ -54,12 +55,11 @@ public class StagingItemReaderTests {
new JobParameters(), "testJob")));
writer.beforeStep(stepExecution);
writer.write(Arrays.asList(new String[] { "FOO", "BAR", "SPAM", "BUCKET" }));
reader.beforeStep(stepExecution);
reader.setJobId(jobId);
}
@AfterTransaction
public void onTearDownAfterTransaction() throws Exception {
reader.destroy();
simpleJdbcTemplate.update("DELETE FROM BATCH_STAGING");
}
@@ -118,7 +118,13 @@ public class StagingItemReaderTests {
String.class, id);
assertEquals(StagingItemWriter.NEW, before);
Object item = reader.read();
Object item;
try {
item = reader.read();
}
catch (Exception e) {
throw new UnexpectedJobExecutionException("Reader error", e);
}
assertEquals("FOO", item);
transactionStatus.setRollbackOnly();