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

RESOLVED - BATCH-1009: Automatically register ItemReadListener, ItemWriteListener and ItemProcessListener

Browse Source 
applied patch + added autoregistration of skip and chunk listeners
This commit is contained in:
robokaso
2009-01-23 15:04:38 +00:00
parent 9518ddc3f5
commit 92890da76e
4 changed files with 206 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files

17
spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBean.java
View File

@@ -25,6 +25,7 @@ import org.springframework.batch.core.ItemProcessListener;
import org.springframework.batch.core.ItemReadListener;
import org.springframework.batch.core.ItemWriteListener;
import org.springframework.batch.core.SkipListener;
import org.springframework.batch.core.StepListener;
import org.springframework.batch.core.step.skip.LimitCheckingItemSkipPolicy;
import org.springframework.batch.core.step.skip.NonSkippableReadException;
import org.springframework.batch.core.step.skip.SkipLimitExceededException;
@@ -312,6 +313,22 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
chunkProcessor.setListeners(BatchListenerFactoryHelper.<SkipListener<T, S>> getListeners(getListeners(),
SkipListener.class));
for (Object itemHandler : new Object[] { getItemReader(), getItemWriter(), getItemProcessor() }) {
if (itemHandler instanceof SkipListener) {
chunkProvider.registerListener((StepListener) itemHandler);
chunkProcessor.registerListener((StepListener) itemHandler);
// already registered with both so avoid double-registering
continue;
}
if (itemHandler instanceof ItemReadListener) {
chunkProvider.registerListener((StepListener) itemHandler);
}
if (itemHandler instanceof ItemProcessListener || itemHandler instanceof ItemWriteListener) {
chunkProcessor.registerListener((StepListener) itemHandler);
}
}
ChunkOrientedTasklet<T> tasklet = new ChunkOrientedTasklet<T>(chunkProvider, chunkProcessor);
tasklet.setBuffering(!isReaderTransactionalQueue);

72
spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleStepFactoryBean.java
View File

@@ -15,10 +15,13 @@
*/
package org.springframework.batch.core.step.item;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.batch.core.ChunkListener;
import org.springframework.batch.core.ItemProcessListener;
import org.springframework.batch.core.ItemReadListener;
import org.springframework.batch.core.ItemWriteListener;
@@ -417,42 +420,11 @@ public class SimpleStepFactoryBean<T, S> implements FactoryBean, BeanNameAware {
ItemWriter<? super S> itemWriter = getItemWriter();
ItemProcessor<? super T, ? extends S> itemProcessor = getItemProcessor();
// Since we are going to wrap these things with listener callbacks we
// need to register them here because the step will not know we did
// that.
if (itemReader instanceof ItemStream) {
step.registerStream((ItemStream) itemReader);
}
if (itemReader instanceof StepExecutionListener) {
step.registerStepExecutionListener((StepExecutionListener) itemReader);
}
if (itemProcessor instanceof ItemStream) {
step.registerStream((ItemStream) itemProcessor);
}
if (itemProcessor instanceof StepExecutionListener) {
step.registerStepExecutionListener((StepExecutionListener) itemProcessor);
}
if (itemWriter instanceof ItemStream) {
step.registerStream((ItemStream) itemWriter);
}
if (itemWriter instanceof StepExecutionListener) {
step.registerStepExecutionListener((StepExecutionListener) itemWriter);
}
List<StepExecutionListener> array = BatchListenerFactoryHelper.getListeners(listeners,
StepExecutionListener.class);
StepExecutionListener[] stepListeners = new StepExecutionListener[array.size()];
for (int i = 0; i < stepListeners.length; i++) {
stepListeners[i] = array.get(i);
}
step.setStepExecutionListeners(stepListeners);
if (chunkOperations == null) {
RepeatTemplate repeatTemplate = new RepeatTemplate();
repeatTemplate.setCompletionPolicy(getChunkCompletionPolicy());
chunkOperations = repeatTemplate;
}
BatchListenerFactoryHelper.addChunkListeners(chunkOperations, getListeners());
if (stepOperations == null) {
@@ -471,15 +443,47 @@ public class SimpleStepFactoryBean<T, S> implements FactoryBean, BeanNameAware {
step.setStepOperations(stepOperations);
SimpleChunkProvider<T> chunkProvider = new SimpleChunkProvider<T>(itemReader, chunkOperations);
List<ItemReadListener<T>> readListeners = BatchListenerFactoryHelper.<ItemReadListener<T>>getListeners(getListeners(), ItemReadListener.class);
chunkProvider.setListeners(readListeners);
SimpleChunkProcessor<T, S> chunkProcessor = new SimpleChunkProcessor<T, S>(itemProcessor, itemWriter);
chunkProcessor.setListeners(BatchListenerFactoryHelper.<ItemProcessListener<T,S>>getListeners(getListeners(), ItemProcessListener.class));
chunkProcessor.setListeners(BatchListenerFactoryHelper.<ItemWriteListener<S>>getListeners(getListeners(), ItemWriteListener.class));
SimpleChunkProvider<T> chunkProvider = new SimpleChunkProvider<T>(itemReader, chunkOperations);
List<ItemReadListener<T>> readListeners = BatchListenerFactoryHelper.<ItemReadListener<T>>getListeners(getListeners(), ItemReadListener.class);
chunkProvider.setListeners(readListeners);
ChunkOrientedTasklet<T> tasklet = new ChunkOrientedTasklet<T>(chunkProvider, chunkProcessor);
// Since we are going to wrap these things with listener callbacks we
// need to register them here because the step will not know we did
// that.
List<StepListener> chunkListeners = new ArrayList<StepListener>(Arrays.asList(getListeners()));
for(Object itemHandler: new Object[]{itemReader, itemWriter, itemProcessor}){
if (itemHandler instanceof ItemStream) {
step.registerStream((ItemStream) itemHandler);
}
if (itemHandler instanceof StepExecutionListener) {
step.registerStepExecutionListener((StepExecutionListener) itemHandler);
}
if (itemHandler instanceof ChunkListener) {
chunkListeners.add((StepListener) itemHandler);
}
if (itemHandler instanceof ItemReadListener) {
chunkProvider.registerListener((StepListener) itemHandler);
}
if (itemHandler instanceof ItemProcessListener || itemHandler instanceof ItemWriteListener) {
chunkProcessor.registerListener((StepListener) itemHandler);
}
}
BatchListenerFactoryHelper.addChunkListeners(chunkOperations, chunkListeners.toArray(new StepListener[]{}));
List<StepExecutionListener> array = BatchListenerFactoryHelper.getListeners(listeners,
StepExecutionListener.class);
StepExecutionListener[] stepListeners = new StepExecutionListener[array.size()];
for (int i = 0; i < stepListeners.length; i++) {
stepListeners[i] = array.get(i);
}
step.setStepExecutionListeners(stepListeners);
step.setTasklet(tasklet);
}

80
spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanTests.java
View File

@@ -14,9 +14,14 @@ import org.apache.commons.logging.LogFactory;
import org.junit.Before;
import org.junit.Test;
import org.springframework.batch.core.BatchStatus;
import org.springframework.batch.core.ChunkListener;
import org.springframework.batch.core.ExitStatus;
import org.springframework.batch.core.ItemProcessListener;
import org.springframework.batch.core.ItemReadListener;
import org.springframework.batch.core.ItemWriteListener;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.SkipListener;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.StepListener;
@@ -602,6 +607,81 @@ public class FaultTolerantStepFactoryBeanTests {
}
@Test
public void testAutoRegisterItemListeners() throws Exception {
final List<Integer> listenerCalls = new ArrayList<Integer>();
class TestItemListenerWriter implements ItemWriter<String>, ItemReadListener<String>,
ItemWriteListener<String>, ItemProcessListener<String, String>, SkipListener<String, String>,
ChunkListener {
public void write(List<? extends String> items) throws Exception {
if (items.contains("4")) {
throw new SkippableException("skippable");
}
}
public void afterRead(String item) {
listenerCalls.add(1);
}
public void beforeRead() {
}
public void onReadError(Exception ex) {
}
public void afterWrite(List<? extends String> items) {
listenerCalls.add(2);
}
public void beforeWrite(List<? extends String> items) {
}
public void onWriteError(Exception exception, List<? extends String> items) {
}
public void afterProcess(String item, String result) {
listenerCalls.add(3);
}
public void beforeProcess(String item) {
}
public void onProcessError(String item, Exception e) {
}
public void afterChunk() {
listenerCalls.add(4);
}
public void beforeChunk() {
}
public void onSkipInProcess(String item, Throwable t) {
}
public void onSkipInRead(Throwable t) {
listenerCalls.add(6);
}
public void onSkipInWrite(String item, Throwable t) {
listenerCalls.add(5);
}
}
factory.setItemWriter(new TestItemListenerWriter());
Step step = (Step) factory.getObject();
step.execute(stepExecution);
assertEquals(BatchStatus.COMPLETED, stepExecution.getStatus());
for (int i = 1; i <= 6; i++) {
assertTrue("didn't call listener " + i, listenerCalls.contains(i));
}
}
private static class SkipProcessorStub implements ItemProcessor<String, String> {
private final Collection<String> failures;

72
spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SimpleStepFactoryBeanTests.java
View File

@@ -30,6 +30,9 @@ import org.junit.Before;
import org.junit.Test;
import org.springframework.batch.core.BatchStatus;
import org.springframework.batch.core.ChunkListener;
import org.springframework.batch.core.ItemProcessListener;
import org.springframework.batch.core.ItemReadListener;
import org.springframework.batch.core.ItemWriteListener;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.Step;
@@ -81,7 +84,7 @@ public class SimpleStepFactoryBeanTests {
MapStepExecutionDao.clear();
}
@Test(expected=IllegalArgumentException.class)
@Test(expected = IllegalArgumentException.class)
public void testMandatoryProperties() throws Exception {
new SimpleStepFactoryBean<String, String>().getObject();
}
@@ -291,6 +294,73 @@ public class SimpleStepFactoryBeanTests {
}
@Test
public void testAutoRegisterItemListeners() throws Exception {
SimpleStepFactoryBean<String, String> factory = getStepFactory(new String[] { "foo", "bar", "spam" });
final List<Integer> listenerCalls = new ArrayList<Integer>();
class TestItemListenerWriter implements ItemWriter<String>, ItemReadListener<String>,
ItemWriteListener<String>, ItemProcessListener<String, String>, ChunkListener {
public void write(List<? extends String> items) throws Exception {
}
public void afterRead(String item) {
listenerCalls.add(1);
}
public void beforeRead() {
}
public void onReadError(Exception ex) {
}
public void afterWrite(List<? extends String> items) {
listenerCalls.add(2);
}
public void beforeWrite(List<? extends String> items) {
}
public void onWriteError(Exception exception, List<? extends String> items) {
}
public void afterProcess(String item, String result) {
listenerCalls.add(3);
}
public void beforeProcess(String item) {
}
public void onProcessError(String item, Exception e) {
}
public void afterChunk() {
listenerCalls.add(4);
}
public void beforeChunk() {
}
}
factory.setItemWriter(new TestItemListenerWriter());
Step step = (Step) factory.getObject();
job.setSteps(Collections.singletonList(step));
JobExecution jobExecution = repository.createJobExecution(job.getName(), new JobParameters());
job.execute(jobExecution);
assertEquals(BatchStatus.COMPLETED, jobExecution.getStatus());
for (int i = 1; i <= 4; i++) {
assertTrue(listenerCalls.contains(i));
}
}
private SimpleStepFactoryBean<String, String> getStepFactory(String... args) throws Exception {
SimpleStepFactoryBean<String, String> factory = new SimpleStepFactoryBean<String, String>();