From ef4f2900cc815c2e7a91d2a4a15a0d85490f725d Mon Sep 17 00:00:00 2001 From: dsyer Date: Wed, 20 Aug 2008 15:24:05 +0000 Subject: [PATCH] OPEN - issue BATCH-789: Remove mark/reset from ItemReader Shift chunk iteration into StepHandler - had to sacrifice interrupted check inside chunk --- .../step/item/AbstractStepFactoryBean.java | 324 ---------------- .../step/item/ItemOrientedStepHandler.java | 50 ++- .../item/RepeatOperationsStepFactoryBean.java | 46 +-- .../core/step/item/SimpleStepFactoryBean.java | 363 ++++++++++++++++-- .../step/item/SkipLimitStepFactoryBean.java | 30 +- .../batch/core/step/item/StepHandlerStep.java | 61 +-- .../item/ItemOrientedStepHandlerTests.java | 34 +- .../RepeatOperationsStepFactoryBeanTests.java | 2 +- .../step/item/SimpleStepFactoryBeanTests.java | 67 ++-- .../core/step/item/SimpleStepHandler.java | 23 +- .../StatefulRetryStepFactoryBeanTests.java | 2 +- .../item/StepExecutorInterruptionTests.java | 54 ++- .../item/StepHandlerStepIntegrationTests.java | 12 +- .../core/step/item/StepHandlerStepTests.java | 39 +- .../.settings/org.eclipse.jdt.core.prefs | 17 +- .../chunk/ChunkMessageChannelItemWriter.java | 17 +- ...hunkMessageItemWriterIntegrationTests.java | 3 +- .../item/MessageChannelItemWriterTests.java | 12 - .../job/MessageOrientedStepTests.java | 5 - 19 files changed, 524 insertions(+), 637 deletions(-) delete mode 100644 spring-batch-core/src/main/java/org/springframework/batch/core/step/item/AbstractStepFactoryBean.java rename spring-batch-core/src/{main => test}/java/org/springframework/batch/core/step/item/SimpleStepHandler.java (64%) diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/AbstractStepFactoryBean.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/AbstractStepFactoryBean.java deleted file mode 100644 index 2184ba169..000000000 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/AbstractStepFactoryBean.java +++ /dev/null @@ -1,324 +0,0 @@ -/* - * 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.core.step.item; - -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.item.ItemProcessor; -import org.springframework.batch.item.ItemReader; -import org.springframework.batch.item.ItemStream; -import org.springframework.batch.item.ItemWriter; -import org.springframework.batch.item.validator.Validator; -import org.springframework.beans.factory.BeanNameAware; -import org.springframework.beans.factory.FactoryBean; -import org.springframework.transaction.PlatformTransactionManager; -import org.springframework.transaction.interceptor.DefaultTransactionAttribute; -import org.springframework.transaction.interceptor.TransactionAttribute; -import org.springframework.util.Assert; - -/** - * Base class for factory beans for {@link StepHandlerStep}. Ensures that all - * the mandatory properties are set, and provides basic support for the - * {@link Step} interface responsibilities like start limit. Supports - * registration of {@link ItemStream}s and {@link StepListener}s. - * - * @see SimpleStepFactoryBean - * @see RepeatOperationsStepFactoryBean - * - * @author Dave Syer - * - */ -public abstract class AbstractStepFactoryBean implements FactoryBean, BeanNameAware { - - private String name; - - private int startLimit = Integer.MAX_VALUE; - - private boolean allowStartIfComplete; - - private ItemReader itemReader; - - private ItemWriter itemWriter; - - private PlatformTransactionManager transactionManager; - - private TransactionAttribute transactionAttribute; - - private JobRepository jobRepository; - - private boolean singleton = true; - - private Validator jobRepositoryValidator = new TransactionInterceptorValidator(1); - - private ItemStream[] streams = new ItemStream[0]; - - private StepListener[] listeners = new StepListener[0]; - - private ItemProcessor itemProcessor = new ItemProcessor() { - @SuppressWarnings("unchecked") - public S process(T item) throws Exception {return (S)item;} - }; - - /** - * - */ - public AbstractStepFactoryBean() { - super(); - } - - /** - * Set the bean name property, which will become the name of the - * {@link Step} when it is created. - * - * @see org.springframework.beans.factory.BeanNameAware#setBeanName(java.lang.String) - */ - public void setBeanName(String name) { - this.name = name; - } - - /** - * Public getter for the String. - * @return the name - */ - public String getName() { - return name; - } - - /** - * Public setter for the startLimit. - * - * @param startLimit the startLimit to set - */ - public void setStartLimit(int startLimit) { - this.startLimit = startLimit; - } - - /** - * Public setter for the shouldAllowStartIfComplete. - * - * @param allowStartIfComplete the shouldAllowStartIfComplete to set - */ - public void setAllowStartIfComplete(boolean allowStartIfComplete) { - this.allowStartIfComplete = allowStartIfComplete; - } - - /** - * @param itemReader the itemReader to set - */ - public void setItemReader(ItemReader itemReader) { - this.itemReader = itemReader; - } - - /** - * @param itemWriter the itemWriter to set - */ - public void setItemWriter(ItemWriter itemWriter) { - this.itemWriter = itemWriter; - } - - /** - * @param itemProcessor the itemProcessor to set - */ - public void setItemProcessor(ItemProcessor itemProcessor) { - this.itemProcessor = itemProcessor; - } - - /** - * 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; - } - - /** - * The listeners to inject into the {@link Step}. Any instance of - * {@link StepListener} can be used, and will then receive callbacks at the - * appropriate stage in the step. - * - * @param listeners an array of listeners - */ - public void setListeners(StepListener[] listeners) { - this.listeners = listeners; - } - - /** - * Protected getter for the {@link StepListener}s. - * @return the listeners - */ - protected StepListener[] getListeners() { - return listeners; - } - - /** - * Protected getter for the {@link ItemReader} for subclasses to use. - * @return the itemReader - */ - protected ItemReader getItemReader() { - return itemReader; - } - - /** - * Protected getter for the {@link ItemWriter} for subclasses to use - * @return the itemWriter - */ - protected ItemWriter getItemWriter() { - return itemWriter; - } - - /** - * Protected getter for the {@link ItemProcessor} for subclasses to use - * @return the itemProcessor - */ - protected ItemProcessor getItemProcessor() { - return itemProcessor; - } - - /** - * Public setter for {@link JobRepository}. - * - * @param jobRepository is a mandatory dependence (no default). - */ - public void setJobRepository(JobRepository jobRepository) { - this.jobRepository = jobRepository; - } - - /** - * Public setter for the {@link PlatformTransactionManager}. - * - * @param transactionManager the transaction manager to set - */ - public void setTransactionManager(PlatformTransactionManager transactionManager) { - this.transactionManager = transactionManager; - } - - /** - * Public setter for the {@link TransactionAttribute}. - * @param transactionAttribute the {@link TransactionAttribute} to set - */ - public void setTransactionAttribute(TransactionAttribute transactionAttribute) { - this.transactionAttribute = transactionAttribute; - } - - /** - * Protected getter for the {@link TransactionAttribute} for subclasses - * only. - * @return the transactionAttribute - */ - protected TransactionAttribute getTransactionAttribute() { - return transactionAttribute!=null?transactionAttribute:new DefaultTransactionAttribute(); - } - - /** - * Create a {@link Step} from the configuration provided. - * - * @see org.springframework.beans.factory.FactoryBean#getObject() - */ - public final Object getObject() throws Exception { - StepHandlerStep step = new StepHandlerStep(getName()); - applyConfiguration(step); - return step; - } - - /** - * @param step - * - */ - protected void applyConfiguration(StepHandlerStep step) { - - Assert.notNull(getItemReader(), "ItemReader must be provided"); - Assert.notNull(getItemWriter(), "ItemWriter must be provided"); - Assert.notNull(transactionManager, "TransactionManager must be provided"); - jobRepositoryValidator.validate(jobRepository); - - step.setTransactionManager(transactionManager); - if (transactionAttribute!=null) { - step.setTransactionAttribute(transactionAttribute); - } - step.setJobRepository(jobRepository); - step.setStartLimit(startLimit); - step.setAllowStartIfComplete(allowStartIfComplete); - - step.setStreams(streams); - - ItemReader itemReader = getItemReader(); - ItemWriter itemWriter = getItemWriter(); - ItemProcessor 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); - } - - 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); - step.setItemHandler(new ItemOrientedStepHandler(itemReader, itemProcessor, itemWriter)); - - } - - public Class getObjectType() { - return Step.class; - } - - /** - * Returns true by default, but in most cases a {@link Step} should not be - * treated as thread safe. Clients are recommended to create a new step for - * each job execution. - * - * @see org.springframework.beans.factory.FactoryBean#isSingleton() - */ - public boolean isSingleton() { - return this.singleton; - } - - /** - * Public setter for the singleton flag. - * @param singleton the value to set. Defaults to true. - */ - public void setSingleton(boolean singleton) { - this.singleton = singleton; - } - -} \ No newline at end of file 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 22228bcb5..82092fa79 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 @@ -26,6 +26,10 @@ import org.springframework.batch.item.ItemWriter; import org.springframework.batch.item.MarkFailedException; import org.springframework.batch.item.ResetFailedException; import org.springframework.batch.repeat.ExitStatus; +import org.springframework.batch.repeat.RepeatCallback; +import org.springframework.batch.repeat.RepeatContext; +import org.springframework.batch.repeat.RepeatOperations; +import org.springframework.batch.repeat.support.RepeatTemplate; /** * Simplest possible implementation of {@link StepHandler} with no skipping or @@ -43,45 +47,57 @@ public class ItemOrientedStepHandler implements StepHandler { protected final Log logger = LogFactory.getLog(getClass()); - private ItemReader itemReader; + private final ItemReader itemReader; - private ItemProcessor itemProcessor; + private final ItemProcessor itemProcessor; - private ItemWriter itemWriter; + private final ItemWriter itemWriter; + + private final RepeatOperations repeatOperations; /** * @param itemReader * @param itemProcessor * @param itemWriter + * @param repeatOperations */ public ItemOrientedStepHandler(ItemReader itemReader, - ItemProcessor itemProcessor, ItemWriter itemWriter) { + ItemProcessor itemProcessor, ItemWriter itemWriter, + RepeatOperations repeatOperations) { super(); this.itemReader = itemReader; this.itemProcessor = itemProcessor; this.itemWriter = itemWriter; + this.repeatOperations = repeatOperations; } /** * Get the next item from {@link #read(StepContribution)} and if not null * pass the item to {@link #write(Object, StepContribution)}. If the - * {@link ItemProcessor} returns null, the write is omitted and another - * item taken from the reader. + * {@link ItemProcessor} returns null, the write is omitted and another item + * taken from the reader. * * @see org.springframework.batch.core.step.item.StepHandler#handle(org.springframework.batch.core.StepContribution) */ - public ExitStatus handle(StepContribution contribution) throws Exception { - boolean processed = false; - while (!processed) { - T item = read(contribution); - if (item == null) { - return ExitStatus.FINISHED; + public ExitStatus handle(final StepContribution contribution) throws Exception { + + ExitStatus result = repeatOperations.iterate(new RepeatCallback() { + public ExitStatus doInIteration(final RepeatContext context) throws Exception { + boolean processed = false; + while (!processed) { + T item = read(contribution); + if (item == null) { + return ExitStatus.FINISHED; + } + // TODO: segregate read / write / filter count + contribution.incrementItemCount(); + processed = write(item, contribution); + } + return ExitStatus.CONTINUABLE; } - // TODO: segregate read / write / filter count - contribution.incrementItemCount(); - processed = write(item, contribution); - } - return ExitStatus.CONTINUABLE; + }); + + return result; } /** diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/RepeatOperationsStepFactoryBean.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/RepeatOperationsStepFactoryBean.java index e04008604..a192a96b0 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/RepeatOperationsStepFactoryBean.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/RepeatOperationsStepFactoryBean.java @@ -17,57 +17,17 @@ package org.springframework.batch.core.step.item; import org.springframework.batch.core.Step; import org.springframework.batch.repeat.RepeatOperations; -import org.springframework.batch.repeat.support.RepeatTemplate; /** * Factory bean for {@link Step} implementations allowing registration of * listeners and also direct injection of the {@link RepeatOperations} needed at * step and chunk level. * + * @deprecated use the {@link SimpleStepFactoryBean} instead + * * @author Dave Syer * */ -public class RepeatOperationsStepFactoryBean extends AbstractStepFactoryBean { - - private RepeatOperations chunkOperations = new RepeatTemplate(); - - private RepeatOperations stepOperations = new RepeatTemplate(); - - /** - * The {@link RepeatOperations} to use for the outer loop of the batch - * processing. Should be set up by the caller through a factory. Defaults to - * a plain {@link RepeatTemplate}. - * - * @param stepOperations a {@link RepeatOperations} instance. - */ - public void setStepOperations(RepeatOperations stepOperations) { - this.stepOperations = stepOperations; - } - - /** - * The {@link RepeatOperations} to use for the inner loop of the batch - * processing. should be set up by the caller through a factory. defaults to - * a plain {@link RepeatTemplate}. - * - * @param chunkOperations a {@link RepeatOperations} instance. - */ - public void setChunkOperations(RepeatOperations chunkOperations) { - this.chunkOperations = chunkOperations; - } - - /** - * @param step - * - */ - protected void applyConfiguration(StepHandlerStep step) { - - super.applyConfiguration(step); - - RepeatOperations chunkOperations = BatchListenerFactoryHelper.addChunkListeners(this.chunkOperations, getListeners()); - - step.setChunkOperations(chunkOperations); - step.setStepOperations(stepOperations); - - } +public class RepeatOperationsStepFactoryBean extends SimpleStepFactoryBean { } 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 0b6a7c28d..590743786 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 @@ -18,17 +18,31 @@ package org.springframework.batch.core.step.item; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; 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.item.ItemProcessor; +import org.springframework.batch.item.ItemReader; +import org.springframework.batch.item.ItemStream; +import org.springframework.batch.item.ItemWriter; +import org.springframework.batch.item.validator.Validator; import org.springframework.batch.repeat.CompletionPolicy; +import org.springframework.batch.repeat.RepeatOperations; import org.springframework.batch.repeat.exception.DefaultExceptionHandler; import org.springframework.batch.repeat.exception.ExceptionHandler; import org.springframework.batch.repeat.policy.SimpleCompletionPolicy; import org.springframework.batch.repeat.support.RepeatTemplate; import org.springframework.batch.repeat.support.TaskExecutorRepeatTemplate; +import org.springframework.beans.factory.BeanNameAware; +import org.springframework.beans.factory.FactoryBean; import org.springframework.core.task.TaskExecutor; +import org.springframework.transaction.PlatformTransactionManager; +import org.springframework.transaction.interceptor.DefaultTransactionAttribute; +import org.springframework.transaction.interceptor.TransactionAttribute; import org.springframework.util.Assert; /** - * Most common configuration options for simple steps should be found here. Use + * Most common configuration options for simple steps should be found here. Use * this factory bean instead of creating a {@link Step} implementation manually. * * This factory does not support configuration of fault-tolerant behavior, use @@ -39,21 +53,50 @@ import org.springframework.util.Assert; * @author Dave Syer * */ -public class SimpleStepFactoryBean extends AbstractStepFactoryBean { +public class SimpleStepFactoryBean implements FactoryBean, BeanNameAware { + + private static final int DEFAULT_COMMIT_INTERVAL = 1; + + private String name; + + private int startLimit = Integer.MAX_VALUE; + + private boolean allowStartIfComplete; + + private ItemReader itemReader; + + private ItemWriter itemWriter; + + private PlatformTransactionManager transactionManager; + + private TransactionAttribute transactionAttribute; + + private JobRepository jobRepository; + + private boolean singleton = true; + + private Validator jobRepositoryValidator = new TransactionInterceptorValidator(1); + + private ItemStream[] streams = new ItemStream[0]; + + private StepListener[] listeners = new StepListener[0]; protected final Log logger = LogFactory.getLog(getClass()); - private static final int DEFAULT_COMMIT_INTERVAL = 1; + private ItemProcessor itemProcessor = new ItemProcessor() { + @SuppressWarnings("unchecked") + public S process(T item) throws Exception {return (S)item;} + }; private int commitInterval = 0; private TaskExecutor taskExecutor; - private StepHandler itemHandler; + private StepHandler stepHandler; - private RepeatTemplate stepOperations; + private RepeatOperations stepOperations; - private RepeatTemplate chunkOperations; + private RepeatOperations chunkOperations; private ExceptionHandler exceptionHandler = new DefaultExceptionHandler(); @@ -61,6 +104,193 @@ public class SimpleStepFactoryBean extends AbstractStepFactoryBean { private int throttleLimit = TaskExecutorRepeatTemplate.DEFAULT_THROTTLE_LIMIT; + /** + * + */ + public SimpleStepFactoryBean() { + super(); + } + + /** + * Set the bean name property, which will become the name of the + * {@link Step} when it is created. + * + * @see org.springframework.beans.factory.BeanNameAware#setBeanName(java.lang.String) + */ + public void setBeanName(String name) { + this.name = name; + } + + /** + * Public getter for the String. + * @return the name + */ + public String getName() { + return name; + } + + /** + * Public setter for the startLimit. + * + * @param startLimit the startLimit to set + */ + public void setStartLimit(int startLimit) { + this.startLimit = startLimit; + } + + /** + * Public setter for the shouldAllowStartIfComplete. + * + * @param allowStartIfComplete the shouldAllowStartIfComplete to set + */ + public void setAllowStartIfComplete(boolean allowStartIfComplete) { + this.allowStartIfComplete = allowStartIfComplete; + } + + /** + * @param itemReader the itemReader to set + */ + public void setItemReader(ItemReader itemReader) { + this.itemReader = itemReader; + } + + /** + * @param itemWriter the itemWriter to set + */ + public void setItemWriter(ItemWriter itemWriter) { + this.itemWriter = itemWriter; + } + + /** + * @param itemProcessor the itemProcessor to set + */ + public void setItemProcessor(ItemProcessor itemProcessor) { + this.itemProcessor = itemProcessor; + } + + /** + * 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; + } + + /** + * The listeners to inject into the {@link Step}. Any instance of + * {@link StepListener} can be used, and will then receive callbacks at the + * appropriate stage in the step. + * + * @param listeners an array of listeners + */ + public void setListeners(StepListener[] listeners) { + this.listeners = listeners; + } + + /** + * Protected getter for the {@link StepListener}s. + * @return the listeners + */ + protected StepListener[] getListeners() { + return listeners; + } + + /** + * Protected getter for the {@link ItemReader} for subclasses to use. + * @return the itemReader + */ + protected ItemReader getItemReader() { + return itemReader; + } + + /** + * Protected getter for the {@link ItemWriter} for subclasses to use + * @return the itemWriter + */ + protected ItemWriter getItemWriter() { + return itemWriter; + } + + /** + * Protected getter for the {@link ItemProcessor} for subclasses to use + * @return the itemProcessor + */ + protected ItemProcessor getItemProcessor() { + return itemProcessor; + } + + /** + * Public setter for {@link JobRepository}. + * + * @param jobRepository is a mandatory dependence (no default). + */ + public void setJobRepository(JobRepository jobRepository) { + this.jobRepository = jobRepository; + } + + /** + * Public setter for the {@link PlatformTransactionManager}. + * + * @param transactionManager the transaction manager to set + */ + public void setTransactionManager(PlatformTransactionManager transactionManager) { + this.transactionManager = transactionManager; + } + + /** + * Public setter for the {@link TransactionAttribute}. + * @param transactionAttribute the {@link TransactionAttribute} to set + */ + public void setTransactionAttribute(TransactionAttribute transactionAttribute) { + this.transactionAttribute = transactionAttribute; + } + + /** + * Protected getter for the {@link TransactionAttribute} for subclasses + * only. + * @return the transactionAttribute + */ + protected TransactionAttribute getTransactionAttribute() { + return transactionAttribute!=null?transactionAttribute:new DefaultTransactionAttribute(); + } + + /** + * Create a {@link Step} from the configuration provided. + * + * @see org.springframework.beans.factory.FactoryBean#getObject() + */ + public final Object getObject() throws Exception { + StepHandlerStep step = new StepHandlerStep(getName()); + applyConfiguration(step); + return step; + } + + public Class getObjectType() { + return Step.class; + } + + /** + * Returns true by default, but in most cases a {@link Step} should not be + * treated as thread safe. Clients are recommended to create a new step for + * each job execution. + * + * @see org.springframework.beans.factory.FactoryBean#isSingleton() + */ + public boolean isSingleton() { + return this.singleton; + } + + /** + * Public setter for the singleton flag. + * @param singleton the value to set. Defaults to true. + */ + public void setSingleton(boolean singleton) { + this.singleton = singleton; + } + /** * Set the commit interval. Either set this or the chunkCompletionPolicy but * not both. @@ -88,16 +318,32 @@ public class SimpleStepFactoryBean extends AbstractStepFactoryBean { * subclasses. * @return the step operations */ - protected RepeatTemplate getStepOperations() { + protected RepeatOperations getStepOperations() { return stepOperations; } + + /** + * Public setter for the stepOperations. + * @param stepOperations the stepOperations to set + */ + public void setStepOperations(RepeatOperations stepOperations) { + this.stepOperations = stepOperations; + } + + /** + * Public setter for the chunkOperations. + * @param chunkOperations the chunkOperations to set + */ + public void setChunkOperations(RepeatOperations chunkOperations) { + this.chunkOperations = chunkOperations; + } /** * Protected getter for the chunk operations to make them available in * subclasses. * @return the step operations */ - protected RepeatTemplate getChunkOperations() { + protected RepeatOperations getChunkOperations() { return chunkOperations; } @@ -139,19 +385,19 @@ public class SimpleStepFactoryBean extends AbstractStepFactoryBean { } /** - * Public getter for the ItemHandler. - * @return the ItemHandler + * Public getter for the {@link StepHandler}. + * @return the {@link StepHandler} */ - protected StepHandler getItemHandler() { - return itemHandler; + protected StepHandler getStepHandler() { + return stepHandler; } /** - * Public setter for the ItemHandler. - * @param itemHandler the ItemHandler to set + * Public setter for the {@link StepHandler}. + * @param stepHandler the {@link StepHandler} to set */ - protected void setItemHandler(StepHandler itemHandler) { - this.itemHandler = itemHandler; + protected void setStepHandler(StepHandler stepHandler) { + this.stepHandler = stepHandler; } /** @@ -160,26 +406,83 @@ public class SimpleStepFactoryBean extends AbstractStepFactoryBean { */ protected void applyConfiguration(StepHandlerStep step) { - super.applyConfiguration(step); + Assert.notNull(getItemReader(), "ItemReader must be provided"); + Assert.notNull(getItemWriter(), "ItemWriter must be provided"); + Assert.notNull(transactionManager, "TransactionManager must be provided"); + jobRepositoryValidator.validate(jobRepository); - chunkOperations = new RepeatTemplate(); - chunkOperations.setCompletionPolicy(getChunkCompletionPolicy()); - BatchListenerFactoryHelper.addChunkListeners(chunkOperations, getListeners()); - step.setChunkOperations(chunkOperations); + step.setTransactionManager(transactionManager); + if (transactionAttribute!=null) { + step.setTransactionAttribute(transactionAttribute); + } + step.setJobRepository(jobRepository); + step.setStartLimit(startLimit); + step.setAllowStartIfComplete(allowStartIfComplete); - stepOperations = new RepeatTemplate(); + step.setStreams(streams); - if (taskExecutor != null) { - TaskExecutorRepeatTemplate repeatTemplate = new TaskExecutorRepeatTemplate(); - repeatTemplate.setTaskExecutor(taskExecutor); - repeatTemplate.setThrottleLimit(throttleLimit); - stepOperations = repeatTemplate; + ItemReader itemReader = getItemReader(); + ItemWriter itemWriter = getItemWriter(); + ItemProcessor 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); } - stepOperations.setExceptionHandler(exceptionHandler); + 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) { + RepeatTemplate repeatTemplate = new RepeatTemplate(); + repeatTemplate.setCompletionPolicy(getChunkCompletionPolicy()); + chunkOperations = repeatTemplate; + } + BatchListenerFactoryHelper.addChunkListeners(chunkOperations, getListeners()); + + if (stepOperations == null) { + + stepOperations = new RepeatTemplate(); + + if (taskExecutor != null) { + TaskExecutorRepeatTemplate repeatTemplate = new TaskExecutorRepeatTemplate(); + repeatTemplate.setTaskExecutor(taskExecutor); + repeatTemplate.setThrottleLimit(throttleLimit); + stepOperations = repeatTemplate; + } + + ((RepeatTemplate) stepOperations).setExceptionHandler(exceptionHandler); + + } step.setStepOperations(stepOperations); + step.setItemHandler(new ItemOrientedStepHandler(itemReader, itemProcessor, itemWriter, chunkOperations)); + } /** @@ -190,7 +493,7 @@ public class SimpleStepFactoryBean extends AbstractStepFactoryBean { Assert.state(!(chunkCompletionPolicy != null && commitInterval != 0), "You must specify either a chunkCompletionPolicy or a commitInterval but not both."); Assert.state(commitInterval >= 0, "The commitInterval must be positive or zero (for default value)."); - + if (chunkCompletionPolicy != null) { return chunkCompletionPolicy; } @@ -201,4 +504,4 @@ public class SimpleStepFactoryBean extends AbstractStepFactoryBean { return new SimpleCompletionPolicy(commitInterval); } -} +} \ No newline at end of file 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 ea08514a3..1a8ace076 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 @@ -16,6 +16,8 @@ import org.springframework.batch.item.ItemKeyGenerator; import org.springframework.batch.item.ItemProcessor; import org.springframework.batch.item.ItemReader; import org.springframework.batch.item.ItemWriter; +import org.springframework.batch.repeat.RepeatOperations; +import org.springframework.batch.repeat.support.RepeatTemplate; import org.springframework.batch.retry.RecoveryCallback; import org.springframework.batch.retry.RetryCallback; import org.springframework.batch.retry.RetryContext; @@ -52,7 +54,7 @@ import org.springframework.batch.support.SubclassExceptionClassifier; * @author Robert Kasanicky * */ -public class SkipLimitStepFactoryBean extends SimpleStepFactoryBean { +public class SkipLimitStepFactoryBean extends SimpleStepFactoryBean { private int skipLimit = 0; @@ -223,8 +225,11 @@ public class SkipLimitStepFactoryBean extends SimpleStepFactoryBean { } // Co-ordinate the retry policy with the exception handler: - getStepOperations().setExceptionHandler( - new SimpleRetryExceptionHandler(retryPolicy, getExceptionHandler(), fatalExceptionClasses)); + RepeatOperations stepOperations = getStepOperations(); + if (stepOperations instanceof RepeatTemplate) { + ((RepeatTemplate) stepOperations).setExceptionHandler(new SimpleRetryExceptionHandler(retryPolicy, + getExceptionHandler(), fatalExceptionClasses)); + } RecoveryCallbackRetryPolicy recoveryCallbackRetryPolicy = new RecoveryCallbackRetryPolicy(retryPolicy) { protected boolean recoverForException(Throwable ex) { @@ -268,17 +273,14 @@ public class SkipLimitStepFactoryBean extends SimpleStepFactoryBean { } } }); - StatefulRetryStepHandler itemHandler = new StatefulRetryStepHandler(getItemReader(), getItemProcessor(), getItemWriter(), - retryTemplate, itemKeyGenerator, readSkipPolicy, writeSkipPolicy); + StatefulRetryStepHandler itemHandler = new StatefulRetryStepHandler(getItemReader(), + getItemProcessor(), getItemWriter(), getChunkOperations(), retryTemplate, itemKeyGenerator, readSkipPolicy, + writeSkipPolicy); itemHandler.setSkipListeners(BatchListenerFactoryHelper.getSkipListeners(getListeners())); step.setItemHandler(itemHandler); } - else { - // This is the default in ItemOrientedStep anyway... - step.setItemHandler(new ItemOrientedStepHandler(getItemReader(), getItemProcessor(), getItemWriter())); - } } @@ -306,7 +308,7 @@ public class SkipLimitStepFactoryBean extends SimpleStepFactoryBean { * @author Dave Syer * */ - private static class StatefulRetryStepHandler extends ItemOrientedStepHandler { + private static class StatefulRetryStepHandler extends ItemOrientedStepHandler { final private RetryOperations retryOperations; @@ -324,10 +326,11 @@ public class SkipLimitStepFactoryBean extends SimpleStepFactoryBean { * @param retryTemplate * @param itemKeyGenerator */ - public StatefulRetryStepHandler(ItemReader itemReader, ItemProcessor itemProcessor, ItemWriter itemWriter, + public StatefulRetryStepHandler(ItemReader itemReader, + ItemProcessor itemProcessor, ItemWriter itemWriter, RepeatOperations chunkOperations, RetryOperations retryTemplate, ItemKeyGenerator itemKeyGenerator, ItemSkipPolicy readSkipPolicy, ItemSkipPolicy writeSkipPolicy) { - super(itemReader, itemProcessor, itemWriter); + super(itemReader, itemProcessor, itemWriter, chunkOperations); this.retryOperations = retryTemplate; this.itemKeyGenerator = itemKeyGenerator; this.readSkipPolicy = readSkipPolicy; @@ -409,9 +412,6 @@ public class SkipLimitStepFactoryBean extends SimpleStepFactoryBean { * {@link SkipListener} provided is called when retry attempts are * exhausted. The listener callback (on write failure) will happen in * the next transaction automatically.
- * - * @see org.springframework.batch.core.step.item.SimpleStepHandler#write(java.lang.Object, - * org.springframework.batch.core.StepContribution) */ protected boolean write(final T item, final StepContribution contribution) throws Exception { RecoveryRetryCallback retryCallback = new RecoveryRetryCallback(item, new RetryCallback() { diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/StepHandlerStep.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/StepHandlerStep.java index 88064aa95..345ed9404 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/StepHandlerStep.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/StepHandlerStep.java @@ -45,15 +45,9 @@ import org.springframework.transaction.interceptor.TransactionAttribute; /** * Simple implementation of executing the step as a set of chunks, each chunk - * surrounded by a transaction. The structure is therefore that of two nested - * loops, with transaction boundary around the whole inner loop. The outer loop - * is controlled by the step operations ( - * {@link #setStepOperations(RepeatOperations)}), and the inner loop by the - * chunk operations ({@link #setChunkOperations(RepeatOperations)}). The inner - * loop should always be executed in a single thread, so the chunk operations - * should not do any concurrent execution. N.B. usually that means that the - * chunk operations should be a {@link RepeatTemplate} (which is the - * default).
+ * surrounded by a transaction. The structure is therefore that of a loop with + * transaction boundary inside the loop. The loop is controlled by the step + * operations ( {@link #setStepOperations(RepeatOperations)}).
* * Clients can use interceptors in the step operations to intercept or listen to * the iteration on a step-wide basis, for instance to get a callback when the @@ -69,8 +63,6 @@ public class StepHandlerStep extends AbstractStep { private static final Log logger = LogFactory.getLog(StepHandlerStep.class); - private RepeatOperations chunkOperations = new RepeatTemplate(); - private RepeatOperations stepOperations = new RepeatTemplate(); // default to checking current thread for interruption. @@ -173,17 +165,6 @@ public class StepHandlerStep extends AbstractStep { this.stepOperations = stepOperations; } - /** - * The {@link RepeatOperations} to use for the inner loop of the batch - * processing. should be set up by the caller through a factory. defaults to - * a plain {@link RepeatTemplate}. - * - * @param chunkOperations a {@link RepeatOperations} instance. - */ - public void setChunkOperations(RepeatOperations chunkOperations) { - this.chunkOperations = chunkOperations; - } - /** * Setter for the {@link StepInterruptionPolicy}. The policy is used to * check whether an external request has been made to interrupt the job @@ -234,9 +215,6 @@ public class StepHandlerStep extends AbstractStep { final StepContribution contribution = stepExecution.createStepContribution(); // Before starting a new transaction, check for // interruption. - if (stepExecution.isTerminateOnly()) { - context.setTerminateOnly(); - } interruptionPolicy.checkInterrupted(stepExecution); ExitStatus exitStatus = ExitStatus.CONTINUABLE; @@ -248,7 +226,7 @@ public class StepHandlerStep extends AbstractStep { try { try { - exitStatus = processChunk(stepExecution, contribution); + exitStatus = itemHandler.handle(contribution); } catch (Error e) { if (transactionAttribute.rollbackOn(e)) { @@ -351,37 +329,6 @@ public class StepHandlerStep extends AbstractStep { } - /** - * Execute a bunch of identical business logic operations all within a - * transaction. The transaction is programmatically started and stopped - * outside this method, so subclasses that override do not need to create a - * transaction. - * @param execution the current {@link StepExecution} which should be - * treated as read-only for the purposes of this method. - * @param contribution the current {@link StepContribution} which can accept - * changes to be aggregated later into the step execution. - * - * @return true if there is more data to process. - */ - protected ExitStatus processChunk(final StepExecution execution, final StepContribution contribution) { - - ExitStatus result = chunkOperations.iterate(new RepeatCallback() { - public ExitStatus doInIteration(final RepeatContext context) throws Exception { - if (execution.isTerminateOnly()) { - context.setTerminateOnly(); - } - // check for interruption before each item as well - interruptionPolicy.checkInterrupted(execution); - ExitStatus exitStatus = itemHandler.handle(contribution); - // check for interruption after each item as well - interruptionPolicy.checkInterrupted(execution); - return exitStatus; - } - }); - - return result; - } - /** * @param stepExecution * @param contribution diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/ItemOrientedStepHandlerTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/ItemOrientedStepHandlerTests.java index 21fbac41a..3a5442450 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/ItemOrientedStepHandlerTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/ItemOrientedStepHandlerTests.java @@ -19,6 +19,7 @@ import static org.junit.Assert.assertEquals; import java.util.List; +import org.junit.Before; import org.junit.Test; import org.springframework.batch.core.JobExecution; import org.springframework.batch.core.JobInstance; @@ -32,6 +33,8 @@ import org.springframework.batch.item.ParseException; import org.springframework.batch.item.UnexpectedInputException; import org.springframework.batch.item.support.AbstractItemReader; import org.springframework.batch.item.support.PassthroughItemProcessor; +import org.springframework.batch.repeat.policy.SimpleCompletionPolicy; +import org.springframework.batch.repeat.support.RepeatTemplate; /** * @author Dave Syer @@ -42,22 +45,18 @@ public class ItemOrientedStepHandlerTests { private StubItemReader itemReader = new StubItemReader(); private StubItemWriter itemWriter = new StubItemWriter(); + + private RepeatTemplate repeatTemplate = new RepeatTemplate(); + + @Before + public void setUp() { + repeatTemplate.setCompletionPolicy(new SimpleCompletionPolicy(2)); + } @Test public void testHandle() throws Exception { ItemOrientedStepHandler handler = new ItemOrientedStepHandler(itemReader, - new PassthroughItemProcessor(), itemWriter); - StepContribution contribution = new StepContribution(new StepExecution("foo", new JobExecution(new JobInstance( - 123L, new JobParameters(), "job")))); - handler.handle(contribution); - assertEquals(1, itemReader.count); - assertEquals("1", itemWriter.values); - } - - @Test - public void testHandleCompositeItem() throws Exception { - ItemOrientedStepHandler handler = new ItemOrientedStepHandler(itemReader, - new AgrgegateItemProcessor(), itemWriter); + new PassthroughItemProcessor(), itemWriter, repeatTemplate); StepContribution contribution = new StepContribution(new StepExecution("foo", new JobExecution(new JobInstance( 123L, new JobParameters(), "job")))); handler.handle(contribution); @@ -65,6 +64,17 @@ public class ItemOrientedStepHandlerTests { assertEquals("12", itemWriter.values); } + @Test + public void testHandleCompositeItem() throws Exception { + ItemOrientedStepHandler handler = new ItemOrientedStepHandler(itemReader, + new AgrgegateItemProcessor(), itemWriter, repeatTemplate); + StepContribution contribution = new StepContribution(new StepExecution("foo", new JobExecution(new JobInstance( + 123L, new JobParameters(), "job")))); + handler.handle(contribution); + assertEquals(4, itemReader.count); + assertEquals("1234", itemWriter.values); + } + /** * @author Dave Syer * diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/RepeatOperationsStepFactoryBeanTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/RepeatOperationsStepFactoryBeanTests.java index a401b2a4c..b75f711ff 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/RepeatOperationsStepFactoryBeanTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/RepeatOperationsStepFactoryBeanTests.java @@ -39,7 +39,7 @@ import org.springframework.batch.support.transaction.ResourcelessTransactionMana */ public class RepeatOperationsStepFactoryBeanTests extends TestCase { - private RepeatOperationsStepFactoryBean factory = new RepeatOperationsStepFactoryBean(); + private SimpleStepFactoryBean factory = new SimpleStepFactoryBean(); private List list; 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 ca6adc1f8..256f044ff 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 @@ -16,13 +16,18 @@ package org.springframework.batch.core.step.item; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; + import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; import java.util.List; -import junit.framework.TestCase; - +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.JobExecution; @@ -53,7 +58,7 @@ import org.springframework.core.task.SimpleAsyncTaskExecutor; /** * Tests for {@link SimpleStepFactoryBean}. */ -public class SimpleStepFactoryBeanTests extends TestCase { +public class SimpleStepFactoryBeanTests { private List recovered = new ArrayList(); @@ -76,37 +81,15 @@ public class SimpleStepFactoryBeanTests extends TestCase { } }; - protected void setUp() throws Exception { - super.setUp(); + @Before + public void setUp() throws Exception { job.setJobRepository(repository); MapJobInstanceDao.clear(); MapJobExecutionDao.clear(); MapStepExecutionDao.clear(); } - private SimpleStepFactoryBean getStepFactory(String arg) throws Exception { - return getStepFactory(new String[] { arg }); - } - - private SimpleStepFactoryBean getStepFactory(String arg0, String arg1) throws Exception { - return getStepFactory(new String[] { arg0, arg1 }); - } - - private SimpleStepFactoryBean getStepFactory(String[] args) throws Exception { - SimpleStepFactoryBean factory = new SimpleStepFactoryBean(); - - List items = TransactionAwareProxyFactory.createTransactionalList(); - items.addAll(Arrays.asList(args)); - reader = new ListItemReader(items); - - factory.setItemReader(reader); - factory.setItemWriter(writer); - factory.setJobRepository(repository); - factory.setTransactionManager(new ResourcelessTransactionManager()); - factory.setBeanName("stepName"); - return factory; - } - + @Test public void testSimpleJob() throws Exception { job.setSteps(new ArrayList()); @@ -125,6 +108,7 @@ public class SimpleStepFactoryBeanTests extends TestCase { assertTrue(written.contains("foo")); } + @Test public void testSimpleConcurrentJob() throws Exception { job.setSteps(new ArrayList()); @@ -144,12 +128,13 @@ public class SimpleStepFactoryBeanTests extends TestCase { assertTrue(written.contains("foo")); } + @Test public void testSimpleJobWithItemListeners() throws Exception { final List throwables = new ArrayList(); RepeatTemplate chunkOperations = new RepeatTemplate(); - // Always handle the exception a check it is the right one... + // Always handle the exception to check it is the right one... chunkOperations.setExceptionHandler(new ExceptionHandler() { public void handleException(RepeatContext context, Throwable throwable) throws RuntimeException { throwables.add(throwable); @@ -179,8 +164,8 @@ public class SimpleStepFactoryBeanTests extends TestCase { } } }); + factory.setChunkOperations(chunkOperations); StepHandlerStep step = (StepHandlerStep) factory.getObject(); - step.setChunkOperations(chunkOperations); job.setSteps(Collections.singletonList((Step) step)); @@ -194,6 +179,7 @@ public class SimpleStepFactoryBeanTests extends TestCase { assertEquals(3, recovered.size()); } + @Test public void testExceptionTerminates() throws Exception { SimpleStepFactoryBean factory = getStepFactory(new String[] { "foo", "bar", "spam" }); factory.setBeanName("exceptionStep"); @@ -217,6 +203,7 @@ public class SimpleStepFactoryBeanTests extends TestCase { assertEquals(BatchStatus.FAILED, jobExecution.getStatus()); } + @Test public void testExceptionHandler() throws Exception { SimpleStepFactoryBean factory = getStepFactory(new String[] { "foo", "bar", "spam" }); factory.setBeanName("exceptionStep"); @@ -240,6 +227,7 @@ public class SimpleStepFactoryBeanTests extends TestCase { assertEquals(BatchStatus.COMPLETED, jobExecution.getStatus()); } + @Test public void testChunkListeners() throws Exception { String[] items = new String[] { "1", "2", "3", "4", "5", "6", "7" }; int commitInterval = 3; @@ -282,11 +270,13 @@ public class SimpleStepFactoryBeanTests extends TestCase { * Commit interval specified is not allowed to be zero or negative. * @throws Exception */ + @Test public void testCommitIntervalMustBeGreaterThanZero() throws Exception { SimpleStepFactoryBean factory = getStepFactory("foo"); // nothing wrong here factory.getObject(); + factory = getStepFactory("foo"); // but exception expected after setting commit interval to value < 0 factory.setCommitInterval(-1); try { @@ -302,6 +292,7 @@ public class SimpleStepFactoryBeanTests extends TestCase { * Commit interval specified is not allowed to be zero or negative. * @throws Exception */ + @Test public void testCommitIntervalAndCompletionPolicyBothSet() throws Exception { SimpleStepFactoryBean factory = getStepFactory("foo"); @@ -318,4 +309,20 @@ public class SimpleStepFactoryBeanTests extends TestCase { } } + + private SimpleStepFactoryBean getStepFactory(String... args) throws Exception { + SimpleStepFactoryBean factory = new SimpleStepFactoryBean(); + + List items = TransactionAwareProxyFactory.createTransactionalList(); + items.addAll(Arrays.asList(args)); + reader = new ListItemReader(items); + + factory.setItemReader(reader); + factory.setItemWriter(writer); + factory.setJobRepository(repository); + factory.setTransactionManager(new ResourcelessTransactionManager()); + factory.setBeanName("stepName"); + return factory; + } + } diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleStepHandler.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SimpleStepHandler.java similarity index 64% rename from spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleStepHandler.java rename to spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SimpleStepHandler.java index 3288a845e..7c2e660dd 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleStepHandler.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SimpleStepHandler.java @@ -18,6 +18,9 @@ package org.springframework.batch.core.step.item; import org.springframework.batch.item.ItemReader; import org.springframework.batch.item.ItemWriter; import org.springframework.batch.item.support.PassthroughItemProcessor; +import org.springframework.batch.repeat.RepeatOperations; +import org.springframework.batch.repeat.policy.SimpleCompletionPolicy; +import org.springframework.batch.repeat.support.RepeatTemplate; /** * Simplest possible implementation of {@link StepHandler} with no skipping or @@ -28,12 +31,30 @@ import org.springframework.batch.item.support.PassthroughItemProcessor; */ public class SimpleStepHandler extends ItemOrientedStepHandler { + /** + * + */ + private static final RepeatTemplate repeatTemplate = new RepeatTemplate(); + + static { + // It's only for testing, and we don't want any infinite loops... + repeatTemplate.setCompletionPolicy(new SimpleCompletionPolicy(6)); + } + /** * Creates a {@link PassthroughItemProcessor} and uses it to create an * instance of {@link ItemOrientedStepHandler}. */ public SimpleStepHandler(ItemReader itemReader, ItemWriter itemWriter) { - super(itemReader, new PassthroughItemProcessor(), itemWriter); + super(itemReader, new PassthroughItemProcessor(), itemWriter, repeatTemplate); + } + + /** + * Creates a {@link PassthroughItemProcessor} and uses it to create an + * instance of {@link ItemOrientedStepHandler}. + */ + public SimpleStepHandler(ItemReader itemReader, ItemWriter itemWriter, RepeatOperations repeatOperations) { + super(itemReader, new PassthroughItemProcessor(), itemWriter, repeatOperations); } } 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 0ad802a10..264fec95b 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 @@ -136,7 +136,7 @@ public class StatefulRetryStepFactoryBeanTests extends TestCase { factory.setItemReader(provider); factory.setRetryLimit(10); factory.setSkippableExceptionClasses(new Class[0]); - AbstractStep step = (AbstractStep) factory.getObject(); + Step step = (Step) factory.getObject(); StepExecution stepExecution = new StepExecution(step.getName(), jobExecution); step.execute(stepExecution); diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/StepExecutorInterruptionTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/StepExecutorInterruptionTests.java index 3c5001677..d16fbc7f5 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/StepExecutorInterruptionTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/StepExecutorInterruptionTests.java @@ -68,18 +68,33 @@ public class StepExecutorInterruptionTests extends TestCase { public void write(List item) throws Exception { } }; - step.setItemHandler(new SimpleStepHandler(new AbstractItemReader() { - public Object read() throws Exception { - return null; - } - }, itemWriter)); stepExecution = new StepExecution(step.getName(), jobExecution); } - public void testInterruptChunk() throws Exception { - + public void testInterruptStep() throws Exception { + Thread processingThread = createThread(stepExecution); + RepeatTemplate template = new RepeatTemplate(); + // N.B, If we don't set the completion policy it might run forever + template.setCompletionPolicy(new SimpleCompletionPolicy(2)); + step.setItemHandler(new SimpleStepHandler(new AbstractItemReader() { + public Object read() throws Exception { + // do something non-trivial (and not Thread.sleep()) + double foo = 1; + for (int i = 2; i < 250; i++) { + foo = foo * i; + } + + if (foo != 1) { + return new Double(foo); + } + else { + return null; + } + } + }, itemWriter, template)); + processingThread.start(); Thread.sleep(100); processingThread.interrupt(); @@ -96,14 +111,6 @@ public class StepExecutorInterruptionTests extends TestCase { } - public void testInterruptStep() throws Exception { - RepeatTemplate template = new RepeatTemplate(); - // N.B, If we don't set the completion policy it might run forever - template.setCompletionPolicy(new SimpleCompletionPolicy(2)); - step.setChunkOperations(template); - testInterruptChunk(); - } - public void testInterruptOnInterruptedException() throws Exception { Thread processingThread = createThread(stepExecution); @@ -176,23 +183,6 @@ public class StepExecutorInterruptionTests extends TestCase { * @return */ private Thread createThread(final StepExecution stepExecution) { - step.setItemHandler(new SimpleStepHandler(new AbstractItemReader() { - public Object read() throws Exception { - // do something non-trivial (and not Thread.sleep()) - double foo = 1; - for (int i = 2; i < 250; i++) { - foo = foo * i; - } - - if (foo != 1) { - return new Double(foo); - } - else { - return null; - } - } - }, itemWriter)); - Thread processingThread = new Thread() { public void run() { try { diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/StepHandlerStepIntegrationTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/StepHandlerStepIntegrationTests.java index c6eeb226e..135283fec 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/StepHandlerStepIntegrationTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/StepHandlerStepIntegrationTests.java @@ -72,6 +72,8 @@ public class StepHandlerStepIntegrationTests { private JobRepository jobRepository; + private RepeatTemplate chunkOperations; + private ItemReader getReader(String[] args) { return new ListItemReader(Arrays.asList(args)); } @@ -88,19 +90,17 @@ public class StepHandlerStepIntegrationTests { jobRepositoryFactoryBean.setTransactionManager(transactionManager); jobRepositoryFactoryBean.afterPropertiesSet(); jobRepository = (JobRepository) jobRepositoryFactoryBean.getObject(); - RepeatTemplate template; step = new StepHandlerStep("stepName"); step.setJobRepository(jobRepository); step.setTransactionManager(transactionManager); - template = new RepeatTemplate(); + RepeatTemplate template = new RepeatTemplate(); template.setCompletionPolicy(new SimpleCompletionPolicy(1)); step.setStepOperations(template); // Only process one item: - template = new RepeatTemplate(); - template.setCompletionPolicy(new SimpleCompletionPolicy(1)); - step.setChunkOperations(template); + chunkOperations = new RepeatTemplate(); + chunkOperations.setCompletionPolicy(new SimpleCompletionPolicy(1)); job = new JobSupport("FOO"); @@ -121,7 +121,7 @@ public class StepHandlerStepIntegrationTests { } }); } - })); + }, chunkOperations)); JobExecution jobExecution = jobRepository.createJobExecution(job, new JobParameters()); StepExecution stepExecution = new StepExecution(step.getName(), jobExecution); diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/StepHandlerStepTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/StepHandlerStepTests.java index 911a8424d..14b6b4f31 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/StepHandlerStepTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/StepHandlerStepTests.java @@ -29,7 +29,6 @@ import org.springframework.batch.core.JobExecution; import org.springframework.batch.core.JobInstance; import org.springframework.batch.core.JobInterruptedException; import org.springframework.batch.core.JobParameters; -import org.springframework.batch.core.StepContribution; import org.springframework.batch.core.StepExecution; import org.springframework.batch.core.StepExecutionListener; import org.springframework.batch.core.UnexpectedJobExecutionException; @@ -41,7 +40,6 @@ import org.springframework.batch.core.repository.dao.MapJobExecutionDao; import org.springframework.batch.core.repository.dao.MapJobInstanceDao; import org.springframework.batch.core.repository.dao.MapStepExecutionDao; import org.springframework.batch.core.repository.support.SimpleJobRepository; -import org.springframework.batch.core.step.AbstractStep; import org.springframework.batch.core.step.JobRepositorySupport; import org.springframework.batch.core.step.StepInterruptionPolicy; import org.springframework.batch.item.ExecutionContext; @@ -93,9 +91,12 @@ public class StepHandlerStepTests extends TestCase { return new ListItemReader(Arrays.asList(args)); } - private AbstractStep getStep(String[] strings) throws Exception { + private StepHandlerStep getStep(String[] strings) throws Exception { StepHandlerStep step = new StepHandlerStep("stepName"); - step.setItemHandler(new SimpleStepHandler(getReader(strings), itemWriter)); + // Only process one item: + RepeatTemplate template = new RepeatTemplate(); + template.setCompletionPolicy(new SimpleCompletionPolicy(1)); + step.setItemHandler(new SimpleStepHandler(getReader(strings), itemWriter, template)); step.setJobRepository(new JobRepositorySupport()); step.setTransactionManager(transactionManager); return step; @@ -108,16 +109,11 @@ public class StepHandlerStepTests extends TestCase { transactionManager = new ResourcelessTransactionManager(); - RepeatTemplate template; + RepeatTemplate template = new RepeatTemplate(); + template.setCompletionPolicy(new SimpleCompletionPolicy(1)); - step = (StepHandlerStep) getStep(new String[] { "foo", "bar", "spam" }); - template = new RepeatTemplate(); - template.setCompletionPolicy(new SimpleCompletionPolicy(1)); + step = getStep(new String[] { "foo", "bar", "spam" }); step.setStepOperations(template); - // Only process one item: - template = new RepeatTemplate(); - template.setCompletionPolicy(new SimpleCompletionPolicy(1)); - step.setChunkOperations(template); job = new JobSupport("FOO"); jobInstance = new JobInstance(new Long(0), new JobParameters(), job.getName()); @@ -181,25 +177,6 @@ public class StepHandlerStepTests extends TestCase { } - public void testChunkExecutor() throws Exception { - - RepeatTemplate template = new RepeatTemplate(); - - // Only process one item: - template.setCompletionPolicy(new SimpleCompletionPolicy(1)); - step.setChunkOperations(template); - - JobExecution jobExecution = new JobExecution(jobInstance); - - StepExecution stepExecution = new StepExecution(step.getName(), jobExecution); - StepContribution contribution = stepExecution.createStepContribution(); - step.processChunk(stepExecution, contribution); - assertEquals(1, processed.size()); - assertEquals(0, stepExecution.getItemCount()); - assertEquals(1, contribution.getItemCount()); - - } - public void testRepository() throws Exception { SimpleJobRepository repository = new SimpleJobRepository(new MapJobInstanceDao(), new MapJobExecutionDao(), diff --git a/spring-batch-integration/.settings/org.eclipse.jdt.core.prefs b/spring-batch-integration/.settings/org.eclipse.jdt.core.prefs index 4210aeae5..edb5db1ad 100644 --- a/spring-batch-integration/.settings/org.eclipse.jdt.core.prefs +++ b/spring-batch-integration/.settings/org.eclipse.jdt.core.prefs @@ -1,4 +1,4 @@ -#Thu Jul 24 14:51:08 CEST 2008 +#Wed Aug 20 13:59:46 BST 2008 eclipse.preferences.version=1 org.eclipse.jdt.core.compiler.codegen.inlineJsrBytecode=enabled org.eclipse.jdt.core.compiler.codegen.targetPlatform=1.5 @@ -7,12 +7,13 @@ org.eclipse.jdt.core.compiler.compliance=1.5 org.eclipse.jdt.core.compiler.debug.lineNumber=generate org.eclipse.jdt.core.compiler.debug.localVariable=generate org.eclipse.jdt.core.compiler.debug.sourceFile=generate +org.eclipse.jdt.core.compiler.doc.comment.support=enabled org.eclipse.jdt.core.compiler.problem.annotationSuperInterface=warning org.eclipse.jdt.core.compiler.problem.assertIdentifier=error org.eclipse.jdt.core.compiler.problem.autoboxing=ignore org.eclipse.jdt.core.compiler.problem.deprecation=warning org.eclipse.jdt.core.compiler.problem.deprecationInDeprecatedCode=disabled -org.eclipse.jdt.core.compiler.problem.deprecationWhenOverridingDeprecatedMethod=disabled +org.eclipse.jdt.core.compiler.problem.deprecationWhenOverridingDeprecatedMethod=enabled org.eclipse.jdt.core.compiler.problem.discouragedReference=warning org.eclipse.jdt.core.compiler.problem.emptyStatement=ignore org.eclipse.jdt.core.compiler.problem.enumIdentifier=error @@ -26,9 +27,21 @@ org.eclipse.jdt.core.compiler.problem.hiddenCatchBlock=warning org.eclipse.jdt.core.compiler.problem.incompatibleNonInheritedInterfaceMethod=warning org.eclipse.jdt.core.compiler.problem.incompleteEnumSwitch=ignore org.eclipse.jdt.core.compiler.problem.indirectStaticAccess=ignore +org.eclipse.jdt.core.compiler.problem.invalidJavadoc=error +org.eclipse.jdt.core.compiler.problem.invalidJavadocTags=enabled +org.eclipse.jdt.core.compiler.problem.invalidJavadocTagsDeprecatedRef=enabled +org.eclipse.jdt.core.compiler.problem.invalidJavadocTagsNotVisibleRef=enabled +org.eclipse.jdt.core.compiler.problem.invalidJavadocTagsVisibility=default org.eclipse.jdt.core.compiler.problem.localVariableHiding=ignore org.eclipse.jdt.core.compiler.problem.methodWithConstructorName=warning org.eclipse.jdt.core.compiler.problem.missingDeprecatedAnnotation=ignore +org.eclipse.jdt.core.compiler.problem.missingJavadocComments=ignore +org.eclipse.jdt.core.compiler.problem.missingJavadocCommentsOverriding=disabled +org.eclipse.jdt.core.compiler.problem.missingJavadocCommentsVisibility=public +org.eclipse.jdt.core.compiler.problem.missingJavadocTagDescription=return_tag +org.eclipse.jdt.core.compiler.problem.missingJavadocTags=ignore +org.eclipse.jdt.core.compiler.problem.missingJavadocTagsOverriding=disabled +org.eclipse.jdt.core.compiler.problem.missingJavadocTagsVisibility=public org.eclipse.jdt.core.compiler.problem.missingOverrideAnnotation=ignore org.eclipse.jdt.core.compiler.problem.missingSerialVersion=ignore org.eclipse.jdt.core.compiler.problem.noEffectAssignment=warning diff --git a/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkMessageChannelItemWriter.java b/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkMessageChannelItemWriter.java index ccfd91b29..07ecb8854 100644 --- a/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkMessageChannelItemWriter.java +++ b/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkMessageChannelItemWriter.java @@ -7,9 +7,7 @@ import org.apache.commons.logging.LogFactory; import org.springframework.batch.core.BatchStatus; import org.springframework.batch.core.StepExecution; import org.springframework.batch.core.listener.StepExecutionListenerSupport; -import org.springframework.batch.item.ClearFailedException; import org.springframework.batch.item.ExecutionContext; -import org.springframework.batch.item.FlushFailedException; import org.springframework.batch.item.ItemStream; import org.springframework.batch.item.ItemStreamException; import org.springframework.batch.item.ItemWriter; @@ -78,20 +76,6 @@ public class ChunkMessageChannelItemWriter extends StepExecutionListenerSuppo } - /** - * No-op. - * @see org.springframework.batch.item.ItemWriter#flush() - */ - public void flush() throws FlushFailedException { - } - - /** - * No-op. - * @see org.springframework.batch.item.ItemWriter#clear() - */ - public void clear() throws ClearFailedException { - } - @Override public void beforeStep(StepExecution stepExecution) { localState.setStepExecution(stepExecution); @@ -169,6 +153,7 @@ public class ChunkMessageChannelItemWriter extends StepExecutionListenerSuppo Assert.state(jobInstanceId.equals(localState.getJobId()), "Message contained wrong job instance id [" + jobInstanceId + "] should have been [" + localState.getJobId() + "]."); localState.actual++; + // TODO: apply the skip count ExitStatus result = payload.getExitStatus(); // TODO: check it can never be ExitStatus.FINISHED? if (!result.isContinuable()) { diff --git a/spring-batch-integration/src/test/java/org/springframework/batch/integration/chunk/ChunkMessageItemWriterIntegrationTests.java b/spring-batch-integration/src/test/java/org/springframework/batch/integration/chunk/ChunkMessageItemWriterIntegrationTests.java index edc62e8ec..2ecb4701d 100644 --- a/spring-batch-integration/src/test/java/org/springframework/batch/integration/chunk/ChunkMessageItemWriterIntegrationTests.java +++ b/spring-batch-integration/src/test/java/org/springframework/batch/integration/chunk/ChunkMessageItemWriterIntegrationTests.java @@ -53,7 +53,7 @@ public class ChunkMessageItemWriterIntegrationTests { @Qualifier("replies") private PollableChannel replies; - private SimpleStepFactoryBean factory; + private SimpleStepFactoryBean factory = new SimpleStepFactoryBean(); private SimpleJobRepository jobRepository; @@ -63,7 +63,6 @@ public class ChunkMessageItemWriterIntegrationTests { @Before public void setUp() { - factory = new SimpleStepFactoryBean(); jobRepository = new SimpleJobRepository(new MapJobInstanceDao(), new MapJobExecutionDao(), new MapStepExecutionDao(), new MapExecutionContextDao()); factory.setJobRepository(jobRepository); diff --git a/spring-batch-integration/src/test/java/org/springframework/batch/integration/item/MessageChannelItemWriterTests.java b/spring-batch-integration/src/test/java/org/springframework/batch/integration/item/MessageChannelItemWriterTests.java index cf0f3d8de..807b1dab9 100644 --- a/spring-batch-integration/src/test/java/org/springframework/batch/integration/item/MessageChannelItemWriterTests.java +++ b/spring-batch-integration/src/test/java/org/springframework/batch/integration/item/MessageChannelItemWriterTests.java @@ -53,10 +53,6 @@ public class MessageChannelItemWriterTests { assertEquals(Required.class, annotations[0].annotationType()); } - /** - * Test method for {@link org.springframework.batch.integration.item.MessageChannelItemWriter#write(java.lang.Object)}. - * @throws Exception - */ @Test public void testWrite() throws Exception { DirectChannel channel = new DirectChannel(); @@ -70,10 +66,6 @@ public class MessageChannelItemWriterTests { assertEquals("foo", message.getPayload()); } - /** - * Test method for {@link org.springframework.batch.integration.item.MessageChannelItemWriter#write(java.lang.Object)}. - * @throws Exception - */ @Test public void testWriteWithRollback() throws Exception { DirectChannel channel = new DirectChannel(); @@ -93,10 +85,6 @@ public class MessageChannelItemWriterTests { } } - /** - * Test method for {@link org.springframework.batch.integration.item.MessageChannelItemWriter#write(java.lang.Object)}. - * @throws Exception - */ @Test public void testWriteWithRollbackOnEndpoint() throws Exception { DirectChannel channel = new DirectChannel(); diff --git a/spring-batch-integration/src/test/java/org/springframework/batch/integration/job/MessageOrientedStepTests.java b/spring-batch-integration/src/test/java/org/springframework/batch/integration/job/MessageOrientedStepTests.java index 5ecd3a6d3..5e9b1ad3f 100644 --- a/spring-batch-integration/src/test/java/org/springframework/batch/integration/job/MessageOrientedStepTests.java +++ b/spring-batch-integration/src/test/java/org/springframework/batch/integration/job/MessageOrientedStepTests.java @@ -39,7 +39,6 @@ import org.springframework.integration.channel.ThreadLocalChannel; import org.springframework.integration.message.BlockingSource; import org.springframework.integration.message.GenericMessage; import org.springframework.integration.message.Message; -import org.springframework.integration.message.MessageSource; import org.springframework.integration.message.MessageTarget; import org.springframework.util.ReflectionUtils; @@ -85,10 +84,6 @@ public class MessageOrientedStepTests { assertEquals(Required.class, annotations[0].annotationType()); } - /** - * Test method for - * {@link org.springframework.batch.integration.job.MessageOrientedStep#setSource(MessageSource)}. - */ @Test public void testSetReplyChannel() { Method method = ReflectionUtils.findMethod(MessageOrientedStep.class, "setSource",