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

OPEN - issue BATCH-771: Refactor Listeners for chunk changes

Browse Source 
Move write listener into step handler
This commit is contained in:
dsyer
2008-08-26 11:35:59 +00:00
parent 207e882e37
commit a0ea47ce63
6 changed files with 72 additions and 163 deletions
Download Patch File Download Diff File Expand all files Collapse all files

91
spring-batch-core/src/main/java/org/springframework/batch/core/step/item/BatchListenerFactoryHelper.java
View File

@@ -16,21 +16,12 @@
package org.springframework.batch.core.step.item;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import org.springframework.batch.core.ChunkListener;
import org.springframework.batch.core.ItemReadListener;
import org.springframework.batch.core.ItemWriteListener;
import org.springframework.batch.core.SkipListener;
import org.springframework.batch.core.StepExecutionListener;
import org.springframework.batch.core.StepListener;
import org.springframework.batch.core.listener.CompositeChunkListener;
import org.springframework.batch.core.listener.CompositeItemReadListener;
import org.springframework.batch.core.listener.CompositeItemWriteListener;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.support.DelegatingItemReader;
import org.springframework.batch.repeat.RepeatContext;
import org.springframework.batch.repeat.RepeatOperations;
import org.springframework.batch.repeat.listener.RepeatListenerSupport;
@@ -45,74 +36,6 @@ import org.springframework.util.Assert;
*/
abstract class BatchListenerFactoryHelper {
/**
* @param itemReader
* @param listeners
*/
public static <T> ItemReader<T> getItemReader(ItemReader<T> itemReader, StepListener[] listeners) {
final CompositeItemReadListener multicaster = new CompositeItemReadListener();
for (int i = 0; i < listeners.length; i++) {
StepListener listener = listeners[i];
if (listener instanceof ItemReadListener) {
multicaster.register((ItemReadListener) listener);
}
}
itemReader = new DelegatingItemReader<T>(itemReader) {
public T read() throws Exception {
try {
multicaster.beforeRead();
T item = super.read();
multicaster.afterRead(item);
return item;
}
catch (Exception e) {
multicaster.onReadError(e);
throw e;
}
}
};
return itemReader;
}
/**
* @param itemWriter
* @param listeners
*/
public static <T> ItemWriter<T> getItemWriter(final ItemWriter<T> itemWriter, StepListener[] listeners) {
final CompositeItemWriteListener multicaster = new CompositeItemWriteListener();
for (int i = 0; i < listeners.length; i++) {
StepListener listener = listeners[i];
if (listener instanceof ItemWriteListener) {
multicaster.register((ItemWriteListener) listener);
}
}
return new ItemWriter<T>() {
public void write(List<? extends T> items) throws Exception {
for (T item : items) {
try {
multicaster.beforeWrite(item);
itemWriter.write(Collections.singletonList(item));
multicaster.afterWrite(item);
}
catch (Exception e) {
multicaster.onWriteError(e, item);
throw e;
}
}
}
};
}
/**
* @param chunkOperations
* @param listeners
@@ -170,18 +93,4 @@ abstract class BatchListenerFactoryHelper {
return list.toArray(new StepExecutionListener[list.size()]);
}
/**
* @param listeners
*/
public static SkipListener[] getSkipListeners(StepListener[] listeners) {
List<SkipListener> list = new ArrayList<SkipListener>();
for (int i = 0; i < listeners.length; i++) {
StepListener listener = listeners[i];
if (listener instanceof SkipListener) {
list.add((SkipListener) listener);
}
}
return list.toArray(new SkipListener[list.size()]);
}
}

58
spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ItemOrientedStepHandler.java
View File

@@ -20,6 +20,8 @@ import java.util.List;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.batch.core.StepContribution;
import org.springframework.batch.core.StepListener;
import org.springframework.batch.core.listener.MulticasterBatchListener;
import org.springframework.batch.core.step.handler.StepHandler;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemReader;
@@ -58,6 +60,8 @@ public class ItemOrientedStepHandler<T, S> implements StepHandler {
private final RepeatOperations repeatOperations;
final private MulticasterBatchListener listener = new MulticasterBatchListener();
/**
* @param itemReader
* @param itemProcessor
@@ -85,6 +89,7 @@ public class ItemOrientedStepHandler<T, S> implements StepHandler {
*/
public ExitStatus handle(final StepContribution contribution, AttributeAccessor attributes) throws Exception {
// TODO: check flags to see if these need to be saved or not (e.g. JMS not)
final Chunk<T> inputs = getInputBuffer(attributes);
final Chunk<S> outputs = getOutputBuffer(attributes);
@@ -221,7 +226,16 @@ public class ItemOrientedStepHandler<T, S> implements StepHandler {
* @throws Exception
*/
protected final T doRead() throws Exception {
return itemReader.read();
try {
listener.beforeRead();
T item = itemReader.read();
listener.afterRead(item);
return item;
}
catch (Exception e) {
listener.onReadError(e);
throw e;
}
}
/**
@@ -239,8 +253,46 @@ public class ItemOrientedStepHandler<T, S> implements StepHandler {
* @throws Exception
*/
protected final void doWrite(List<S> items) throws Exception {
itemWriter.write(items);
// TODO: increment write count
try {
listener.beforeWrite(items);
itemWriter.write(items);
// TODO: increment write count
listener.afterWrite(items);
}
catch (Exception e) {
listener.onWriteError(e, items);
throw e;
}
}
/**
* Register some {@link StepListener}s with the handler. Each will get
* the callbacks in the order specified at the correct stage.
*
* @param listeners
*/
public void setListeners(StepListener[] listeners) {
for (StepListener listener : listeners) {
registerListener(listener);
}
}
/**
* Register a listener for callbacks at the appropriate stages in a
* process.
*
* @param listener a {@link StepListener}
*/
public void registerListener(StepListener listener) {
this.listener.register(listener);
}
/**
* Public getter for the listener.
* @return the listener
*/
protected MulticasterBatchListener getListener() {
return listener;
}
/**

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

@@ -21,7 +21,6 @@ import org.springframework.batch.core.Step;
import org.springframework.batch.core.StepExecutionListener;
import org.springframework.batch.core.StepListener;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.step.handler.StepHandler;
import org.springframework.batch.core.step.handler.StepHandlerStep;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemReader;
@@ -94,8 +93,6 @@ public class SimpleStepFactoryBean<T,S> implements FactoryBean, BeanNameAware {
private TaskExecutor taskExecutor;
private StepHandler stepHandler;
private RepeatOperations stepOperations;
private RepeatOperations chunkOperations;
@@ -386,22 +383,6 @@ public class SimpleStepFactoryBean<T,S> implements FactoryBean, BeanNameAware {
this.throttleLimit = throttleLimit;
}
/**
* Public getter for the {@link StepHandler}.
* @return the {@link StepHandler}
*/
protected StepHandler getStepHandler() {
return stepHandler;
}
/**
* Public setter for the {@link StepHandler}.
* @param stepHandler the {@link StepHandler} to set
*/
protected void setStepHandler(StepHandler stepHandler) {
this.stepHandler = stepHandler;
}
/**
* @param step
*
@@ -450,13 +431,6 @@ public class SimpleStepFactoryBean<T,S> implements FactoryBean, BeanNameAware {
}
StepExecutionListener[] stepListeners = BatchListenerFactoryHelper.getStepListeners(listeners);
itemReader = BatchListenerFactoryHelper.getItemReader(itemReader, listeners);
itemWriter = BatchListenerFactoryHelper.getItemWriter(itemWriter, listeners);
// In case they are used by subclasses:
setItemReader(itemReader);
setItemWriter(itemWriter);
step.setStepExecutionListeners(stepListeners);
if (chunkOperations == null) {
@@ -483,7 +457,9 @@ public class SimpleStepFactoryBean<T,S> implements FactoryBean, BeanNameAware {
step.setStepOperations(stepOperations);
step.setStepHandler(new ItemOrientedStepHandler<T,S>(itemReader, itemProcessor, itemWriter, chunkOperations));
ItemOrientedStepHandler<T,S> stepHandler = new ItemOrientedStepHandler<T,S>(itemReader, itemProcessor, itemWriter, chunkOperations);
stepHandler.setListeners(getListeners());
step.setStepHandler(stepHandler);
}

38
spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SkipLimitStepFactoryBean.java
View File

@@ -8,7 +8,6 @@ import java.util.List;
import org.springframework.batch.core.SkipListener;
import org.springframework.batch.core.StepContribution;
import org.springframework.batch.core.listener.CompositeSkipListener;
import org.springframework.batch.core.step.handler.StepHandlerStep;
import org.springframework.batch.core.step.skip.ItemSkipPolicy;
import org.springframework.batch.core.step.skip.LimitCheckingItemSkipPolicy;
@@ -261,12 +260,12 @@ public class SkipLimitStepFactoryBean<T, S> extends SimpleStepFactoryBean<T, S>
exceptions.addAll(new ArrayList<Class<? extends Throwable>>(retryableExceptionClasses));
ItemSkipPolicy writeSkipPolicy = new LimitCheckingItemSkipPolicy(skipLimit, exceptions,
new ArrayList<Class<? extends Throwable>>(fatalExceptionClasses));
StatefulRetryStepHandler<T, S> itemHandler = new StatefulRetryStepHandler<T, S>(getItemReader(),
ItemOrientedStepHandler<T, S> stepHandler = new StatefulRetryStepHandler<T, S>(getItemReader(),
getItemProcessor(), getItemWriter(), getChunkOperations(), retryTemplate, readSkipPolicy,
writeSkipPolicy);
itemHandler.setSkipListeners(BatchListenerFactoryHelper.getSkipListeners(getListeners()));
stepHandler.setListeners(getListeners());
step.setStepHandler(itemHandler);
step.setStepHandler(stepHandler);
}
@@ -296,12 +295,10 @@ public class SkipLimitStepFactoryBean<T, S> extends SimpleStepFactoryBean<T, S>
* @author Dave Syer
*
*/
private static class StatefulRetryStepHandler<T, S> extends ItemOrientedStepHandler<T, S> {
static class StatefulRetryStepHandler<T, S> extends ItemOrientedStepHandler<T, S> {
final private RetryOperations retryOperations;
final private CompositeSkipListener listener = new CompositeSkipListener();
final private ItemSkipPolicy readSkipPolicy;
final private ItemSkipPolicy writeSkipPolicy;
@@ -321,29 +318,6 @@ public class SkipLimitStepFactoryBean<T, S> extends SimpleStepFactoryBean<T, S>
this.writeSkipPolicy = writeSkipPolicy;
}
/**
* Register some {@link SkipListener}s with the handler. Each will get
* the callbacks in the order specified at the correct stage if a skip
* occurs.
*
* @param listeners
*/
public void setSkipListeners(SkipListener[] listeners) {
for (SkipListener listener : listeners) {
registerSkipListener(listener);
}
}
/**
* Register a listener for callbacks at the appropriate stages in a skip
* process.
*
* @param listener a {@link SkipListener}
*/
public void registerSkipListener(SkipListener listener) {
this.listener.register(listener);
}
/**
* Tries to read the item from the reader, in case of exception skip the
* item if the skip policy allows, otherwise re-throw.
@@ -366,7 +340,7 @@ public class SkipLimitStepFactoryBean<T, S> extends SimpleStepFactoryBean<T, S>
// increment skip count and try again
try {
skipCount++;
listener.onSkipInRead(e);
getListener().onSkipInRead(e);
}
catch (RuntimeException ex) {
contribution.incrementReadSkipCount(skipCount);
@@ -456,7 +430,7 @@ public class SkipLimitStepFactoryBean<T, S> extends SimpleStepFactoryBean<T, S>
contribution.incrementWriteSkipCount();
S item = chunk.getSkippedItem();
try {
listener.onSkipInWrite(item, t);
getListener().onSkipInWrite(item, t);
return null;
}
catch (RuntimeException ex) {

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

@@ -56,7 +56,7 @@ import org.springframework.core.task.SimpleAsyncTaskExecutor;
*/
public class SimpleStepFactoryBeanTests {
private List<Exception> recovered = new ArrayList<Exception>();
private List<Exception> listened = new ArrayList<Exception>();
private SimpleJobRepository repository = new SimpleJobRepository(new MapJobInstanceDao(), new MapJobExecutionDao(),
new MapStepExecutionDao(), new MapExecutionContextDao());
@@ -136,11 +136,11 @@ public class SimpleStepFactoryBeanTests {
});
factory.setListeners(new StepListener[] { new ItemListenerSupport() {
public void onReadError(Exception ex) {
recovered.add(ex);
listened.add(ex);
}
public void onWriteError(Exception ex, Object item) {
recovered.add(ex);
listened.add(ex);
}
} });
@@ -161,7 +161,7 @@ public class SimpleStepFactoryBeanTests {
assertEquals(0, written.size());
// provider should be at second item
assertEquals("bar", reader.read());
assertEquals(1, recovered.size());
assertEquals(1, listened.size());
}
@Test

10
spring-batch-core/src/test/java/org/springframework/batch/core/step/item/StatefulRetryStepFactoryBeanTests.java
View File

@@ -314,11 +314,10 @@ public class StatefulRetryStepFactoryBeanTests {
assertEquals(2, stepExecution.getSkipCount());
assertEquals(2, stepExecution.getWriteSkipCount());
System.err.println(processed);
// [a, b, c, d, e, f, null]
assertEquals(7, provided.size());
// [a, b, a, b, b, b, b, b, b, c, a, c, d, d, d, d, d, e, f, e, f]
assertEquals(21, processed.size());
// [a, b, c, a, b, c, b, b, b, b, b, c, a, c, d, e, f, d, d, d, d, e, f, e, f]
assertEquals(25, processed.size());
// [b, d]
assertEquals(2, recovered.size());
}
@@ -511,11 +510,10 @@ public class StatefulRetryStepFactoryBeanTests {
// We added a bogus cache so no items are actually skipped
// because they aren't recognised as eligible
assertEquals(0, stepExecution.getSkipCount());
// only one item processed but three (the commit interval) were provided
// [0, 1, 2]
assertEquals(3, provided.size());
// [0]
assertEquals(1, processed.size());
// [0, 1, 2]
assertEquals(3, processed.size());
// []
assertEquals(0, recovered.size());
}