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

OPEN - issue BATCH-404: FactoryBeans for step configuration

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

Switch back to strongly typed listeners and streams in ItemOrientedStep
This commit is contained in:
dsyer
2008-03-02 13:55:54 +00:00
parent e5cccef3d1
commit 81b9b5ebc8
15 changed files with 100 additions and 88 deletions
Download Patch File Download Diff File Expand all files Collapse all files

55
spring-batch-execution/src/main/java/org/springframework/batch/execution/step/ItemOrientedStep.java
View File

@@ -42,6 +42,7 @@ import org.springframework.batch.item.ItemRecoverer;
import org.springframework.batch.item.ItemStream;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.exception.CommitFailedException;
import org.springframework.batch.item.stream.CompositeItemStream;
import org.springframework.batch.repeat.ExitStatus;
import org.springframework.batch.repeat.RepeatCallback;
import org.springframework.batch.repeat.RepeatContext;
@@ -92,6 +93,8 @@ public class ItemOrientedStep extends AbstractStep {
private ItemReaderRetryCallback retryCallback;
private CompositeItemStream stream = new CompositeItemStream();
private ListenerMulticaster listener = new ListenerMulticaster();
private JobRepository jobRepository;
@@ -151,18 +154,34 @@ public class ItemOrientedStep extends AbstractStep {
}
/**
* Register each of the objects as listeners. The {@link ItemOrientedStep}
* accepts listeners of type {@link ItemStream} and {@link BatchListener}.
* The {@link ItemReader} and {@link ItemWriter} are automatically
* Register each of the streams for callbacks at the appropriate time in the
* step. The {@link ItemReader} and {@link ItemWriter} are automatically
* registered, but it doesn't hurt to also register them here. Injected
* dependencies of the reader and writer are not automatically registered,
* so if you implement {@link ItemWriter} using delegation to another object
* which itself is a {@link BatchListener}, you need to register the
* delegate here.
* which itself is a {@link ItemStream}, you need to register the delegate
* here.
*
* @param streams an array of {@link ItemStream} objects.
*/
public void setStreams(ItemStream[] streams) {
for (int i = 0; i < streams.length; i++) {
stream.register(streams[i]);
}
}
/**
* Register each of the objects as listeners. The {@link ItemOrientedStep}
* accepts listeners of type {@link BatchListener}. The {@link ItemReader}
* and {@link ItemWriter} are automatically registered, but it doesn't hurt
* to also register them here. Injected dependencies of the reader and
* writer are not automatically registered, so if you implement
* {@link ItemWriter} using delegation to another object which itself is a
* {@link BatchListener}, you need to register the delegate here.
*
* @param listeners an array of listener objects of known types.
*/
public void setListeners(Object[] listeners) {
public void setListeners(BatchListener[] listeners) {
for (int i = 0; i < listeners.length; i++) {
listener.register(listeners[i]);
}
@@ -275,7 +294,7 @@ public class ItemOrientedStep extends AbstractStep {
// fixed in the step. E.g. ItemStream instances need the the same
// reference to the ExecutionContext as the step execution.
listener.beforeStep(stepExecution);
listener.open(stepExecution.getExecutionContext());
stream.open(stepExecution.getExecutionContext());
status = stepOperations.iterate(new RepeatCallback() {
@@ -309,7 +328,7 @@ public class ItemOrientedStep extends AbstractStep {
// only if chunk was successful
stepExecution.apply(contribution);
listener.update(stepExecution.getExecutionContext());
stream.update(stepExecution.getExecutionContext());
try {
jobRepository.saveOrUpdateExecutionContext(stepExecution);
}
@@ -433,7 +452,7 @@ public class ItemOrientedStep extends AbstractStep {
}
try {
listener.close(stepExecution.getExecutionContext());
stream.close(stepExecution.getExecutionContext());
}
catch (RuntimeException e) {
String msg = "Fatal error detected during close of streams. "
@@ -455,12 +474,22 @@ public class ItemOrientedStep extends AbstractStep {
}
/**
* Register the item reader and writer as listeners. If they are manually
* registered anyway, it shouldn't matter.
* Register the item reader and writer as listeners and streams. If they are
* manually registered anyway, it shouldn't matter.
*/
private void possiblyRegisterStreams() {
listener.register(itemReader);
listener.register(itemWriter);
if (itemReader instanceof ItemStream) {
stream.register((ItemStream) itemReader);
}
if (itemReader instanceof BatchListener) {
listener.register((BatchListener) itemReader);
}
if (itemWriter instanceof ItemStream) {
stream.register((ItemStream) itemWriter);
}
if (itemWriter instanceof BatchListener) {
listener.register((BatchListener) itemWriter);
}
}
/**

11
spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/DefaultStepFactoryBean.java
View File

@@ -18,7 +18,6 @@ package org.springframework.batch.execution.step.support;
import org.springframework.batch.core.domain.BatchListener;
import org.springframework.batch.core.domain.Step;
import org.springframework.batch.execution.step.ItemOrientedStep;
import org.springframework.batch.item.ItemStream;
import org.springframework.batch.repeat.exception.handler.SimpleLimitExceptionHandler;
import org.springframework.batch.repeat.support.RepeatTemplate;
import org.springframework.batch.repeat.support.TaskExecutorRepeatTemplate;
@@ -34,7 +33,7 @@ public class DefaultStepFactoryBean extends SimpleStepFactoryBean {
private boolean alwaysSkip = false;
private Object[] listeners = new Object[0];
private BatchListener[] listeners = new BatchListener[0];
private TaskExecutor taskExecutor;
@@ -52,15 +51,15 @@ public class DefaultStepFactoryBean extends SimpleStepFactoryBean {
/**
* The listeners to inject into the {@link Step}. Any instance of
* {@link BatchListener} or {@link ItemStream} can be used, and will then
* receive callbacks at the appropriate stage in the step.
* {@link BatchListener} can be used, and will then receive callbacks at the
* appropriate stage in the step.
*
* @param listeners an array of listeners
*/
public void setListeners(Object[] listeners) {
public void setListeners(BatchListener[] listeners) {
this.listeners = listeners;
}
/**
* Public setter for the {@link TaskExecutor}. If this is set, then it will
* be used to execute the chunk processing inside the {@link Step}.

40
spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/ListenerMulticaster.java
View File

@@ -25,21 +25,16 @@ import org.springframework.batch.execution.listener.CompositeChunkListener;
import org.springframework.batch.execution.listener.CompositeItemReadListener;
import org.springframework.batch.execution.listener.CompositeItemWriteListener;
import org.springframework.batch.execution.listener.CompositeStepListener;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.batch.item.ItemStream;
import org.springframework.batch.item.exception.StreamException;
import org.springframework.batch.item.stream.CompositeItemStream;
import org.springframework.batch.repeat.ExitStatus;
/**
* @author Dave Syer
*
*/
public class ListenerMulticaster implements ItemStream, StepListener, ChunkListener, ItemReadListener,
public class ListenerMulticaster implements StepListener, ChunkListener, ItemReadListener,
ItemWriteListener {
private CompositeItemStream stream = new CompositeItemStream();
private CompositeStepListener stepListener = new CompositeStepListener();
private CompositeChunkListener chunkListener = new CompositeChunkListener();
@@ -62,7 +57,7 @@ public class ListenerMulticaster implements ItemStream, StepListener, ChunkListe
* @param listeners an array of listener objects of types known to the
* multicaster.
*/
public void setListeners(Object[] listeners) {
public void setListeners(BatchListener[] listeners) {
for (int i = 0; i < listeners.length; i++) {
register(listeners[i]);
}
@@ -73,13 +68,10 @@ public class ListenerMulticaster implements ItemStream, StepListener, ChunkListe
* implemented. Any {@link BatchListener} can be provided, or an
* {@link ItemStream}. Other types will be ignored.
*/
public void register(Object listener) {
public void register(BatchListener listener) {
if (listener instanceof StepListener) {
this.stepListener.register((StepListener) listener);
}
if (listener instanceof ItemStream) {
this.stream.register((ItemStream) listener);
}
if (listener instanceof ChunkListener) {
this.chunkListener.register((ChunkListener) listener);
}
@@ -116,32 +108,6 @@ public class ListenerMulticaster implements ItemStream, StepListener, ChunkListe
return stepListener.onErrorInStep(e);
}
/**
* @param executionContext
* @throws StreamException
* @see org.springframework.batch.item.stream.CompositeItemStream#close(org.springframework.batch.item.ExecutionContext)
*/
public void close(ExecutionContext executionContext) throws StreamException {
stream.close(executionContext);
}
/**
* @param executionContext
* @throws StreamException
* @see org.springframework.batch.item.stream.CompositeItemStream#open(org.springframework.batch.item.ExecutionContext)
*/
public void open(ExecutionContext executionContext) throws StreamException {
stream.open(executionContext);
}
/**
* @param executionContext
* @see org.springframework.batch.item.stream.CompositeItemStream#update(org.springframework.batch.item.ExecutionContext)
*/
public void update(ExecutionContext executionContext) {
stream.update(executionContext);
}
/**
*
* @see org.springframework.batch.execution.listener.CompositeChunkListener#afterChunk()

17
spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/SimpleStepFactoryBean.java
View File

@@ -15,7 +15,9 @@
*/
package org.springframework.batch.execution.step.support;
import org.springframework.batch.core.domain.Step;
import org.springframework.batch.execution.step.ItemOrientedStep;
import org.springframework.batch.item.ItemStream;
import org.springframework.batch.repeat.policy.SimpleCompletionPolicy;
import org.springframework.batch.repeat.support.RepeatTemplate;
@@ -27,6 +29,8 @@ public class SimpleStepFactoryBean extends AbstractStepFactoryBean {
private int commitInterval = 0;
private ItemStream[] streams = new ItemStream[0];
/**
* Set the commit interval.
*
@@ -36,6 +40,17 @@ public class SimpleStepFactoryBean extends AbstractStepFactoryBean {
this.commitInterval = commitInterval;
}
/**
* The streams to inject into the {@link Step}. Any instance of
* {@link ItemStream} can be used, and will then receive callbacks at the
* appropriate stage in the step.
*
* @param streams an array of listeners
*/
public void setStreams(ItemStream[] streams) {
this.streams = streams;
}
/**
* @param step
*
@@ -44,6 +59,8 @@ public class SimpleStepFactoryBean extends AbstractStepFactoryBean {
super.applyConfiguration(step);
step.setStreams(streams);
if (commitInterval > 0) {
RepeatTemplate chunkOperations = new RepeatTemplate();
chunkOperations.setCompletionPolicy(new SimpleCompletionPolicy(commitInterval));

7
spring-batch-execution/src/test/java/org/springframework/batch/execution/step/ItemOrientedStepTests.java
View File

@@ -42,6 +42,7 @@ import org.springframework.batch.execution.step.support.StepInterruptionPolicy;
import org.springframework.batch.io.exception.InfrastructureException;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemStream;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.exception.MarkFailedException;
import org.springframework.batch.item.exception.ResetFailedException;
@@ -341,7 +342,7 @@ public class ItemOrientedStepTests extends TestCase {
}
public void testDirectlyInjectedItemStream() throws Exception {
itemOrientedStep.setListeners(new Object[] {new ItemStreamSupport() {
itemOrientedStep.setStreams(new ItemStream[] {new ItemStreamSupport() {
public void update(ExecutionContext executionContext) {
executionContext.putString("foo", "bar");
}
@@ -387,7 +388,7 @@ public class ItemOrientedStepTests extends TestCase {
}
public void testDirectlyInjectedListenerOnError() throws Exception {
itemOrientedStep.setListeners(new Object[] {new StepListenerSupport() {
itemOrientedStep.setListeners(new StepListener[] {new StepListenerSupport() {
public ExitStatus onErrorInStep(Throwable e) {
list.add(e);
return null;
@@ -420,7 +421,7 @@ public class ItemOrientedStepTests extends TestCase {
}
};
itemOrientedStep.setItemReader(reader);
itemOrientedStep.setListeners(new Object[] {reader});
itemOrientedStep.setStreams(new ItemStream[] {reader});
JobExecution jobExecution = new JobExecution(jobInstance);
StepExecution stepExecution = new StepExecution(itemOrientedStep, jobExecution);

4
spring-batch-samples/src/main/resources/jobs/beanWrapperMapperSampleJob.xml
View File

@@ -12,8 +12,8 @@
<bean id="beanWrapperMapperJob" parent="simpleJob">
<property name="steps">
<list>
<bean id="step1" parent="defaultStep">
<property name="listeners" ref="fileInputTemplate" />
<bean id="step1" parent="simpleStep">
<property name="streams" ref="fileInputTemplate" />
<property name="itemReader">
<bean
class="org.springframework.batch.item.reader.ValidatingItemReader">

4
spring-batch-samples/src/main/resources/jobs/compositeProcessorSampleJob.xml
View File

@@ -11,8 +11,8 @@
<bean id="compositeProcessorJob" parent="simpleJob">
<property name="steps">
<bean id="step1" parent="defaultStep">
<property name="listeners" ref="fileInputTemplate" />
<bean id="step1" parent="simpleStep">
<property name="streams" ref="fileInputTemplate" />
<property name="itemReader">
<bean
class="org.springframework.batch.item.reader.ValidatingItemReader">

4
spring-batch-samples/src/main/resources/jobs/fixedLengthImportJob.xml
View File

@@ -11,8 +11,8 @@
<bean id="fixedLengthImportJob" parent="simpleJob">
<property name="steps">
<bean id="step1" parent="defaultStep">
<property name="listeners" ref="fileInputTemplate" />
<bean id="step1" parent="simpleStep">
<property name="streams" ref="fileInputTemplate" />
<property name="itemReader">
<bean
class="org.springframework.batch.item.reader.ValidatingItemReader">

4
spring-batch-samples/src/main/resources/jobs/multilineJob.xml
View File

@@ -9,8 +9,8 @@
<bean id="multilineJob" parent="simpleJob">
<property name="steps">
<bean id="step1" parent="defaultStep">
<property name="listeners" ref="fileItemReader"/>
<bean id="step1" parent="simpleStep">
<property name="streams" ref="fileItemReader"/>
<property name="itemReader">
<bean
class="org.springframework.batch.item.reader.AggregateItemReader">

22
spring-batch-samples/src/main/resources/jobs/multilineOrderIo.xml
View File

@@ -18,24 +18,24 @@
</property>
</bean>
<bean id="flatFileOutputSource"
<bean id="flatFileTransformerOutput"
class="org.springframework.batch.item.writer.ItemTransformerItemWriter">
<property name="delegate">
<bean
class="org.springframework.batch.io.file.FlatFileItemWriter">
<property name="resource" ref="fileOutputLocator" />
<property name="fieldSetUnmapper">
<bean
class="org.springframework.batch.io.file.mapping.PassThroughFieldSetMapper" />
</property>
</bean>
</property>
<property name="delegate" ref="flatFileOutput"/>
<property name="itemTransformer">
<bean
class="org.springframework.batch.io.file.transform.RecursiveCollectionItemTransformer" />
</property>
</bean>
<bean id="flatFileOutput"
class="org.springframework.batch.io.file.FlatFileItemWriter">
<property name="resource" ref="fileOutputLocator" />
<property name="fieldSetUnmapper">
<bean
class="org.springframework.batch.io.file.mapping.PassThroughFieldSetMapper" />
</property>
</bean>
<bean id="delimitedLineAggregator"
class="org.springframework.batch.io.file.transform.DelimitedLineTokenizer" />

8
spring-batch-samples/src/main/resources/jobs/multilineOrderJob.xml
View File

@@ -13,10 +13,10 @@
<bean id="multilineOrderJob" parent="simpleJob">
<property name="steps">
<bean id="step1" parent="defaultStep">
<property name="listeners">
<bean id="step1" parent="simpleStep">
<property name="streams">
<list>
<ref bean="orderWriter" />
<ref bean="flatFileOutput" />
<ref bean="fileInputTemplate" />
</list>
</property>
@@ -52,7 +52,7 @@
<bean id="orderWriter"
class="org.springframework.batch.sample.dao.FlatFileOrderWriter">
<property name="delegate" ref="flatFileOutputSource" />
<property name="delegate" ref="flatFileTransformerOutput" />
<property name="transformer">
<bean
class="org.springframework.batch.sample.dao.OrderTransformer">

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

@@ -15,7 +15,7 @@
<bean id="staging" parent="defaultStep">
<property name="commitInterval" value="2" />
<property name="startLimit" value="100" />
<property name="listeners" ref="fileInputTemplate" />
<property name="streams" ref="fileInputTemplate" />
<property name="itemReader">
<bean
class="org.springframework.batch.item.reader.ValidatingItemReader">

4
spring-batch-samples/src/main/resources/jobs/restartSample.xml
View File

@@ -11,8 +11,8 @@
<bean id="restartSampleJob" parent="simpleJob">
<property name="steps">
<bean id="step1" parent="defaultStep">
<property name="listeners" ref="fileItemReader"/>
<bean id="step1" parent="simpleStep">
<property name="streams" ref="fileItemReader"/>
<property name="itemReader">
<bean
class="org.springframework.batch.item.reader.ValidatingItemReader">

2
spring-batch-samples/src/main/resources/jobs/rollbackJob.xml
View File

@@ -16,7 +16,7 @@
<list>
<bean id="step1" parent="defaultStep">
<property name="alwaysSkip" value="true" />
<property name="listeners" ref="fileInputTemplate" />
<property name="streams" ref="fileInputTemplate" />
<property name="itemReader">
<bean
class="org.springframework.batch.item.reader.ValidatingItemReader">

4
spring-batch-samples/src/main/resources/jobs/tradeJob.xml
View File

@@ -13,8 +13,8 @@
<bean id="tradeJob" parent="simpleJob">
<property name="steps">
<list>
<bean id="step1" parent="defaultStep">
<property name="listeners" ref="fileInputTemplate" />
<bean id="step1" parent="simpleStep">
<property name="streams" ref="fileInputTemplate" />
<property name="itemReader">
<bean
class="org.springframework.batch.item.reader.ValidatingItemReader">