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

OPEN - issue BATCH-385: Merge user attributes in StepContext with ExecutionContext

Browse Source 
http://jira.springframework.org/browse/BATCH-385

Refactored existing StepContextAware to StepListener implementations.  The BatchResourceFactoryBean is now a ProxyResource with a hint in docos and exception messages about registering with a step.
This commit is contained in:
dsyer
2008-02-29 11:31:21 +00:00
parent 42904865de
commit a53f8f6cbe
10 changed files with 320 additions and 151 deletions
Download Patch File Download Diff File Expand all files Collapse all files

41
spring-batch-samples/src/main/java/org/springframework/batch/sample/item/reader/StagingItemReader.java
View File

@@ -9,12 +9,13 @@ import java.util.List;
import org.apache.commons.lang.SerializationUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.batch.execution.scope.StepContext;
import org.springframework.batch.execution.scope.StepContextAware;
import org.springframework.batch.core.domain.StepExecution;
import org.springframework.batch.core.domain.StepListener;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.batch.item.ItemStream;
import org.springframework.batch.item.KeyedItemReader;
import org.springframework.batch.item.exception.StreamException;
import org.springframework.batch.repeat.ExitStatus;
import org.springframework.batch.sample.item.writer.StagingItemWriter;
import org.springframework.dao.OptimisticLockingFailureException;
import org.springframework.jdbc.core.RowMapper;
@@ -24,14 +25,14 @@ import org.springframework.jdbc.support.lob.LobHandler;
import org.springframework.transaction.support.TransactionSynchronizationManager;
import org.springframework.util.Assert;
public class StagingItemReader extends JdbcDaoSupport implements ItemStream, KeyedItemReader, StepContextAware {
public class StagingItemReader extends JdbcDaoSupport implements ItemStream, KeyedItemReader, StepListener {
// Key for buffer in transaction synchronization manager
private static final String BUFFER_KEY = StagingItemReader.class.getName() + ".BUFFER";
private static Log logger = LogFactory.getLog(StagingItemReader.class);
private StepContext stepContext;
private StepExecution stepExecution;
private LobHandler lobHandler = new DefaultLobHandler();
@@ -76,15 +77,6 @@ public class StagingItemReader extends JdbcDaoSupport implements ItemStream, Key
}
}
/**
* Callback for injection of the step context.
*
* @param stepContext the stepContext to set
*/
public void setStepContext(StepContext stepContext) {
this.stepContext = stepContext;
}
private List retrieveKeys() {
synchronized (lock) {
@@ -93,7 +85,7 @@ public class StagingItemReader extends JdbcDaoSupport implements ItemStream, Key
"SELECT ID FROM BATCH_STAGING WHERE JOB_ID=? AND PROCESSED=? ORDER BY ID",
new Object[] { stepContext.getStepExecution().getJobExecution().getJobId(), StagingItemWriter.NEW },
new Object[] { stepExecution.getJobExecution().getJobId(), StagingItemWriter.NEW },
new RowMapper() {
public Object mapRow(ResultSet rs, int rowNum) throws SQLException {
@@ -234,4 +226,25 @@ public class StagingItemReader extends JdbcDaoSupport implements ItemStream, Key
public void update(ExecutionContext executionContext) {
}
/* (non-Javadoc)
* @see org.springframework.batch.core.domain.StepListener#afterStep()
*/
public ExitStatus afterStep() {
return null;
}
/* (non-Javadoc)
* @see org.springframework.batch.core.domain.StepListener#beforeStep(org.springframework.batch.core.domain.StepExecution)
*/
public void beforeStep(StepExecution stepExecution) {
this.stepExecution = stepExecution;
}
/* (non-Javadoc)
* @see org.springframework.batch.core.domain.StepListener#onErrorInStep(java.lang.Throwable)
*/
public ExitStatus onErrorInStep(Throwable e) {
return null;
}
}

41
spring-batch-samples/src/main/java/org/springframework/batch/sample/item/writer/StagingItemWriter.java
View File

@@ -5,11 +5,12 @@ import java.sql.PreparedStatement;
import java.sql.SQLException;
import org.apache.commons.lang.SerializationUtils;
import org.springframework.batch.execution.scope.StepContext;
import org.springframework.batch.execution.scope.StepContextAware;
import org.springframework.batch.core.domain.StepExecution;
import org.springframework.batch.core.domain.StepListener;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.exception.ClearFailedException;
import org.springframework.batch.item.exception.FlushFailedException;
import org.springframework.batch.repeat.ExitStatus;
import org.springframework.jdbc.core.PreparedStatementSetter;
import org.springframework.jdbc.core.support.JdbcDaoSupport;
import org.springframework.jdbc.support.incrementer.DataFieldMaxValueIncrementer;
@@ -18,7 +19,7 @@ import org.springframework.jdbc.support.lob.LobHandler;
import org.springframework.util.Assert;
import org.springframework.util.ClassUtils;
public class StagingItemWriter extends JdbcDaoSupport implements StepContextAware, ItemWriter {
public class StagingItemWriter extends JdbcDaoSupport implements StepListener, ItemWriter {
public static final String NEW = "N";
@@ -28,7 +29,7 @@ public class StagingItemWriter extends JdbcDaoSupport implements StepContextAwar
private DataFieldMaxValueIncrementer incrementer;
private StepContext stepContext;
private StepExecution stepExecution;
private LobHandler lobHandler = new DefaultLobHandler();
@@ -52,15 +53,6 @@ public class StagingItemWriter extends JdbcDaoSupport implements StepContextAwar
+ ClassUtils.getShortName(StagingItemWriter.class));
}
/**
* Callback for injection of the step context.
*
* @param stepContext the stepContext to set
*/
public void setStepContext(StepContext stepContext) {
this.stepContext = stepContext;
}
/**
* Setter for the key generator for the staging table.
*
@@ -77,7 +69,7 @@ public class StagingItemWriter extends JdbcDaoSupport implements StepContextAwar
*/
public void write(Object data) {
final long id = incrementer.nextLongValue();
final long jobId = stepContext.getStepExecution().getJobExecution().getJobId().longValue();
final long jobId = stepExecution.getJobExecution().getJobId().longValue();
final byte[] blob = SerializationUtils.serialize((Serializable) data);
getJdbcTemplate()
.update("INSERT into BATCH_STAGING (ID, JOB_ID, VALUE, PROCESSED) values (?,?,?,?)",
@@ -101,4 +93,25 @@ public class StagingItemWriter extends JdbcDaoSupport implements StepContextAwar
public void flush() throws FlushFailedException {
}
/* (non-Javadoc)
* @see org.springframework.batch.core.domain.StepListener#afterStep()
*/
public ExitStatus afterStep() {
return null;
}
/* (non-Javadoc)
* @see org.springframework.batch.core.domain.StepListener#beforeStep(org.springframework.batch.core.domain.StepExecution)
*/
public void beforeStep(StepExecution stepExecution) {
this.stepExecution = stepExecution;
}
/* (non-Javadoc)
* @see org.springframework.batch.core.domain.StepListener#onErrorInStep(java.lang.Throwable)
*/
public ExitStatus onErrorInStep(Throwable e) {
return null;
}
}

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

@@ -32,9 +32,7 @@
</property>
<property name="itemWriter">
<bean
class="org.springframework.batch.sample.item.writer.StagingItemWriter"
scope="step">
<aop:scoped-proxy />
class="org.springframework.batch.sample.item.writer.StagingItemWriter">
<property name="dataSource"
ref="dataSource" />
<property name="lobHandler"
@@ -60,9 +58,7 @@
</property>
<property name="itemReader">
<bean
class="org.springframework.batch.sample.item.reader.StagingItemReader"
scope="step">
<aop:scoped-proxy />
class="org.springframework.batch.sample.item.reader.StagingItemReader">
<property name="lobHandler"
ref="lobHandler" />
<property name="dataSource"
@@ -91,8 +87,7 @@
</bean>
<bean id="fileInputTemplate" parent="testInputTemplate"
autowire-candidate="false" scope="step">
<aop:scoped-proxy />
autowire-candidate="false">
</bean>
<bean id="fixedFileDescriptor"

57
spring-batch-samples/src/test/java/org/springframework/batch/sample/item/reader/StagingItemReaderTests.java
View File

@@ -4,12 +4,7 @@ import org.springframework.batch.core.domain.JobExecution;
import org.springframework.batch.core.domain.JobInstance;
import org.springframework.batch.core.domain.JobParameters;
import org.springframework.batch.core.domain.StepExecution;
import org.springframework.batch.execution.scope.SimpleStepContext;
import org.springframework.batch.execution.scope.StepContext;
import org.springframework.batch.execution.scope.StepSynchronizationManager;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.batch.repeat.context.RepeatContextSupport;
import org.springframework.batch.repeat.synch.RepeatSynchronizationManager;
import org.springframework.batch.sample.item.writer.StagingItemWriter;
import org.springframework.batch.sample.tasklet.JobSupport;
import org.springframework.batch.sample.tasklet.StepSupport;
@@ -20,7 +15,7 @@ public class StagingItemReaderTests extends AbstractTransactionalDataSourceSprin
private StagingItemWriter writer;
private StagingItemReader provider;
private StagingItemReader reader;
private Long jobId = new Long(11);
@@ -28,8 +23,8 @@ public class StagingItemReaderTests extends AbstractTransactionalDataSourceSprin
this.writer = writer;
}
public void setProvider(StagingItemReader provider) {
this.provider = provider;
public void setProvider(StagingItemReader reader) {
this.reader = reader;
}
protected String[] getConfigLocations() {
@@ -37,12 +32,14 @@ public class StagingItemReaderTests extends AbstractTransactionalDataSourceSprin
"staging-test-context.xml") };
}
protected void prepareTestInstance() throws Exception {
StepContext stepScopeContext = new SimpleStepContext(new StepExecution(new StepSupport("stepName"),
new JobExecution(new JobInstance(jobId, new JobParameters(), new JobSupport("testJob")))));
StepSynchronizationManager.register(stepScopeContext);
RepeatSynchronizationManager.register(new RepeatContextSupport(null));
super.prepareTestInstance();
/* (non-Javadoc)
* @see org.springframework.test.AbstractTransactionalSpringContextTests#onSetUpBeforeTransaction()
*/
protected void onSetUpBeforeTransaction() throws Exception {
StepExecution stepExecution = new StepExecution(new StepSupport("stepName"),
new JobExecution(new JobInstance(jobId, new JobParameters(), new JobSupport("testJob"))));
reader.beforeStep(stepExecution);
writer.beforeStep(stepExecution);
}
protected void onSetUpInTransaction() throws Exception {
@@ -50,11 +47,11 @@ public class StagingItemReaderTests extends AbstractTransactionalDataSourceSprin
writer.write("BAR");
writer.write("SPAM");
writer.write("BUCKET");
provider.open(new ExecutionContext());
reader.open(new ExecutionContext());
}
protected void onTearDownAfterTransaction() throws Exception {
provider.close(null);
reader.close(null);
getJdbcTemplate().update("DELETE FROM BATCH_STAGING");
}
@@ -66,7 +63,7 @@ public class StagingItemReaderTests extends AbstractTransactionalDataSourceSprin
new Object[] { new Long(id) }, String.class);
assertEquals(StagingItemWriter.NEW, before);
Object item = provider.read();
Object item = reader.read();
assertEquals("FOO", item);
String after = (String) getJdbcTemplate().queryForObject("SELECT PROCESSED from BATCH_STAGING where ID=?",
@@ -89,7 +86,7 @@ public class StagingItemReaderTests extends AbstractTransactionalDataSourceSprin
public void testProviderRollsBackMultipleTimes() throws Exception {
provider.mark();
reader.mark();
setComplete();
endTransaction();
startNewTransaction();
@@ -98,34 +95,34 @@ public class StagingItemReaderTests extends AbstractTransactionalDataSourceSprin
new Object[] { jobId, StagingItemWriter.NEW });
assertEquals(4, count);
Object item = provider.read();
Object item = reader.read();
assertEquals("FOO", item);
item = provider.read();
item = reader.read();
assertEquals("BAR", item);
provider.reset();
reader.reset();
endTransaction();
startNewTransaction();
item = provider.read();
item = reader.read();
assertEquals("FOO", item);
item = provider.read();
item = reader.read();
assertEquals("BAR", item);
item = provider.read();
item = reader.read();
assertEquals("SPAM", item);
provider.reset();
reader.reset();
endTransaction();
startNewTransaction();
item = provider.read();
item = reader.read();
assertEquals("FOO", item);
}
public void testProviderRollsBackProcessIndicator() throws Exception {
provider.mark();
reader.mark();
setComplete();
endTransaction();
startNewTransaction();
@@ -138,10 +135,10 @@ public class StagingItemReaderTests extends AbstractTransactionalDataSourceSprin
new Object[] { new Long(id) }, String.class);
assertEquals(StagingItemWriter.NEW, before);
Object item = provider.read();
Object item = reader.read();
assertEquals("FOO", item);
provider.reset();
reader.reset();
endTransaction();
startNewTransaction();
// After a rollback we have to resynchronize the TX to simulate a real
@@ -151,7 +148,7 @@ public class StagingItemReaderTests extends AbstractTransactionalDataSourceSprin
new Object[] { new Long(id) }, String.class);
assertEquals(StagingItemWriter.NEW, after);
item = provider.read();
item = reader.read();
assertEquals("FOO", item);
}
}

56
spring-batch-samples/src/test/java/org/springframework/batch/sample/item/writer/StagingItemWriterTests.java Normal file
View File

@@ -0,0 +1,56 @@
/*
* Copyright 2006-2007 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.sample.item.writer;
import org.springframework.batch.core.domain.JobExecution;
import org.springframework.batch.core.domain.JobInstance;
import org.springframework.batch.core.domain.JobParameters;
import org.springframework.batch.core.domain.StepExecution;
import org.springframework.batch.sample.tasklet.JobSupport;
import org.springframework.batch.sample.tasklet.StepSupport;
import org.springframework.test.AbstractTransactionalDataSourceSpringContextTests;
import org.springframework.util.ClassUtils;
public class StagingItemWriterTests extends AbstractTransactionalDataSourceSpringContextTests {
private StagingItemWriter writer;
public void setWriter(StagingItemWriter processor) {
this.writer = processor;
}
protected String[] getConfigLocations() {
return new String[] { ClassUtils.addResourcePathToPackagePath(StagingItemWriter.class,
"staging-test-context.xml") };
}
/* (non-Javadoc)
* @see org.springframework.test.AbstractTransactionalSpringContextTests#onSetUpBeforeTransaction()
*/
protected void onSetUpBeforeTransaction() throws Exception {
StepExecution stepExecution = new StepExecution(new StepSupport("stepName"),
new JobExecution(new JobInstance(new Long(12L), new JobParameters(), new JobSupport("testJob"))));
writer.beforeStep(stepExecution);
}
public void testProcessInsertsNewItem() throws Exception {
int before = getJdbcTemplate().queryForInt("SELECT COUNT(*) from BATCH_STAGING");
writer.write("FOO");
int after = getJdbcTemplate().queryForInt("SELECT COUNT(*) from BATCH_STAGING");
assertEquals(before + 1, after);
}
}

6
spring-batch-samples/src/test/resources/org/springframework/batch/sample/item/writer/staging-test-context.xml
View File

@@ -12,8 +12,7 @@
class="org.springframework.batch.execution.scope.StepScope" />
<bean id="processor"
class="org.springframework.batch.sample.item.writer.StagingItemWriter"
scope="step">
class="org.springframework.batch.sample.item.writer.StagingItemWriter">
<property name="incrementer">
<bean id="jobIncrementer" parent="incrementerParent">
<property name="incrementerName"
@@ -24,8 +23,7 @@
</bean>
<bean id="provider"
class="org.springframework.batch.sample.item.reader.StagingItemReader"
scope="step">
class="org.springframework.batch.sample.item.reader.StagingItemReader">
<property name="dataSource" ref="dataSource" />
</bean>