From a0ea47ce63d846fb941e32a9364efeab42cf6b53 Mon Sep 17 00:00:00 2001 From: dsyer Date: Tue, 26 Aug 2008 11:35:59 +0000 Subject: [PATCH] OPEN - issue BATCH-771: Refactor Listeners for chunk changes Move write listener into step handler --- .../step/item/BatchListenerFactoryHelper.java | 91 ------------------- .../step/item/ItemOrientedStepHandler.java | 58 +++++++++++- .../core/step/item/SimpleStepFactoryBean.java | 30 +----- .../step/item/SkipLimitStepFactoryBean.java | 38 ++------ .../step/item/SimpleStepFactoryBeanTests.java | 8 +- .../StatefulRetryStepFactoryBeanTests.java | 10 +- 6 files changed, 72 insertions(+), 163 deletions(-) diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/BatchListenerFactoryHelper.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/BatchListenerFactoryHelper.java index 5bda9fd88..18df6e832 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/BatchListenerFactoryHelper.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/BatchListenerFactoryHelper.java @@ -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 ItemReader getItemReader(ItemReader 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(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 ItemWriter getItemWriter(final ItemWriter 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() { - - public void write(List 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 list = new ArrayList(); - 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()]); - } - } diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ItemOrientedStepHandler.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ItemOrientedStepHandler.java index 0848193e8..5e0d37c9d 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ItemOrientedStepHandler.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ItemOrientedStepHandler.java @@ -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 implements StepHandler { private final RepeatOperations repeatOperations; + final private MulticasterBatchListener listener = new MulticasterBatchListener(); + /** * @param itemReader * @param itemProcessor @@ -85,6 +89,7 @@ public class ItemOrientedStepHandler 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 inputs = getInputBuffer(attributes); final Chunk outputs = getOutputBuffer(attributes); @@ -221,7 +226,16 @@ public class ItemOrientedStepHandler 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 implements StepHandler { * @throws Exception */ protected final void doWrite(List 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; } /** diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleStepFactoryBean.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleStepFactoryBean.java index 78cf20e16..74a6ca9f7 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleStepFactoryBean.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleStepFactoryBean.java @@ -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 implements FactoryBean, BeanNameAware { private TaskExecutor taskExecutor; - private StepHandler stepHandler; - private RepeatOperations stepOperations; private RepeatOperations chunkOperations; @@ -386,22 +383,6 @@ public class SimpleStepFactoryBean 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 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 implements FactoryBean, BeanNameAware { step.setStepOperations(stepOperations); - step.setStepHandler(new ItemOrientedStepHandler(itemReader, itemProcessor, itemWriter, chunkOperations)); + ItemOrientedStepHandler stepHandler = new ItemOrientedStepHandler(itemReader, itemProcessor, itemWriter, chunkOperations); + stepHandler.setListeners(getListeners()); + step.setStepHandler(stepHandler); } diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SkipLimitStepFactoryBean.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SkipLimitStepFactoryBean.java index 9255ce165..b19838425 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SkipLimitStepFactoryBean.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SkipLimitStepFactoryBean.java @@ -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 extends SimpleStepFactoryBean exceptions.addAll(new ArrayList>(retryableExceptionClasses)); ItemSkipPolicy writeSkipPolicy = new LimitCheckingItemSkipPolicy(skipLimit, exceptions, new ArrayList>(fatalExceptionClasses)); - StatefulRetryStepHandler itemHandler = new StatefulRetryStepHandler(getItemReader(), + ItemOrientedStepHandler stepHandler = new StatefulRetryStepHandler(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 extends SimpleStepFactoryBean * @author Dave Syer * */ - private static class StatefulRetryStepHandler extends ItemOrientedStepHandler { + static class StatefulRetryStepHandler extends ItemOrientedStepHandler { 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 extends SimpleStepFactoryBean 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 extends SimpleStepFactoryBean // 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 extends SimpleStepFactoryBean contribution.incrementWriteSkipCount(); S item = chunk.getSkippedItem(); try { - listener.onSkipInWrite(item, t); + getListener().onSkipInWrite(item, t); return null; } catch (RuntimeException ex) { diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SimpleStepFactoryBeanTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SimpleStepFactoryBeanTests.java index 25d517dae..87b216802 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SimpleStepFactoryBeanTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SimpleStepFactoryBeanTests.java @@ -56,7 +56,7 @@ import org.springframework.core.task.SimpleAsyncTaskExecutor; */ public class SimpleStepFactoryBeanTests { - private List recovered = new ArrayList(); + private List listened = new ArrayList(); 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 diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/StatefulRetryStepFactoryBeanTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/StatefulRetryStepFactoryBeanTests.java index 4f623bc0d..968441611 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/StatefulRetryStepFactoryBeanTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/StatefulRetryStepFactoryBeanTests.java @@ -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()); }