diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/AbstractStepFactoryBean.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/AbstractStepFactoryBean.java deleted file mode 100644 index 07e2ac0d7..000000000 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/AbstractStepFactoryBean.java +++ /dev/null @@ -1,197 +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; - -import org.springframework.batch.core.Step; -import org.springframework.batch.core.repository.JobRepository; -import org.springframework.batch.item.ItemReader; -import org.springframework.batch.item.ItemWriter; -import org.springframework.beans.factory.BeanNameAware; -import org.springframework.beans.factory.FactoryBean; -import org.springframework.transaction.PlatformTransactionManager; -import org.springframework.util.Assert; - -/** - * Base class for factory beans for {@link ItemOrientedStep}. Ensures that all - * the mandatory properties are set, and provides basic support for the - * {@link Step} interface responsibilities like start limit. - * - * @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 JobRepository jobRepository; - - private boolean singleton = true; - - /** - * - */ - 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; - } - - /** - * 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; - } - - /** - * 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; - } - - /** - * Create a {@link Step} from the configuration provided. - * - * @see org.springframework.beans.factory.FactoryBean#getObject() - */ - public final Object getObject() throws Exception { - ItemOrientedStep step = new ItemOrientedStep(getName()); - applyConfiguration(step); - return step; - } - - /** - * @param step - * - */ - protected void applyConfiguration(ItemOrientedStep step) { - - Assert.notNull(getItemReader(), "ItemReader must be provided"); - Assert.notNull(getItemWriter(), "ItemWriter must be provided"); - Assert.notNull(jobRepository, "JobRepository must be provided"); - Assert.notNull(transactionManager, "TransactionManager must be provided"); - - step.setItemHandler(new SimpleItemHandler(itemReader, itemWriter)); - step.setTransactionManager(transactionManager); - step.setJobRepository(jobRepository); - step.setStartLimit(startLimit); - step.setAllowStartIfComplete(allowStartIfComplete); - - } - - 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/BatchListenerFactoryHelper.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/BatchListenerFactoryHelper.java deleted file mode 100644 index a7c182054..000000000 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/BatchListenerFactoryHelper.java +++ /dev/null @@ -1,173 +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; - -import java.util.ArrayList; -import java.util.List; - -import org.springframework.batch.core.BatchListener; -import org.springframework.batch.core.ChunkListener; -import org.springframework.batch.core.ItemReadListener; -import org.springframework.batch.core.ItemWriteListener; -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.reader.DelegatingItemReader; -import org.springframework.batch.item.writer.DelegatingItemWriter; -import org.springframework.batch.repeat.ExitStatus; -import org.springframework.batch.repeat.RepeatContext; -import org.springframework.batch.repeat.RepeatOperations; -import org.springframework.batch.repeat.listener.RepeatListenerSupport; -import org.springframework.batch.repeat.support.RepeatTemplate; -import org.springframework.util.Assert; - -/** - * Package private helper for step factory beans. - * - * @author Dave Syer - * - */ -class BatchListenerFactoryHelper { - - /** - * @param itemReader2 - * @param listeners - * @return - */ - public ItemReader getItemReader(ItemReader itemReader, BatchListener[] listeners) { - - final CompositeItemReadListener multicaster = new CompositeItemReadListener(); - - for (int i = 0; i < listeners.length; i++) { - BatchListener listener = listeners[i]; - if (listener instanceof ItemReadListener) { - multicaster.register((ItemReadListener) listener); - } - } - - itemReader = new DelegatingItemReader(itemReader) { - public Object read() throws Exception { - try { - multicaster.beforeRead(); - Object item = super.read(); - multicaster.afterRead(item); - return item; - } - catch (Exception e) { - multicaster.onReadError(e); - throw e; - } - } - }; - - return itemReader; - } - - /** - * @param itemWriter2 - * @param listeners - * @return - */ - public ItemWriter getItemWriter(ItemWriter itemWriter, BatchListener[] listeners) { - final CompositeItemWriteListener multicaster = new CompositeItemWriteListener(); - - for (int i = 0; i < listeners.length; i++) { - BatchListener listener = listeners[i]; - if (listener instanceof ItemWriteListener) { - multicaster.register((ItemWriteListener) listener); - } - } - - itemWriter = new DelegatingItemWriter(itemWriter) { - public void write(Object item) throws Exception { - try { - multicaster.beforeWrite(item); - super.write(item); - multicaster.afterWrite(); - } - catch (Exception e) { - multicaster.onWriteError(e, item); - throw e; - } - } - }; - - return itemWriter; - - } - - /** - * @param stepOperations - * @param listeners - * @return - */ - public RepeatOperations addChunkListeners(RepeatOperations stepOperations, BatchListener[] listeners) { - - final CompositeChunkListener multicaster = new CompositeChunkListener(); - - boolean hasChunkListener = false; - - for (int i = 0; i < listeners.length; i++) { - BatchListener listener = listeners[i]; - if (listener instanceof ChunkListener) { - hasChunkListener = true; - } - if (listener instanceof ChunkListener) { - multicaster.register((ChunkListener) listener); - } - } - - if (hasChunkListener) { - - Assert.state(stepOperations instanceof RepeatTemplate, - "Step operations is injected but not a RepeatTemplate, so chunk listeners cannot also be registered. " - + "Either inject a RepeatTemplate, or remove the ChunkListener."); - - RepeatTemplate stepTemplate = (RepeatTemplate) stepOperations; - stepTemplate.registerListener(new RepeatListenerSupport() { - public void before(RepeatContext context) { - multicaster.beforeChunk(); - } - public void after(RepeatContext context, ExitStatus result) { - multicaster.afterChunk(); - } - }); - - } - - return stepOperations; - - } - - /** - * @param listeners - * @return - */ - public StepListener[] getStepListeners(BatchListener[] listeners) { - List list = new ArrayList(); - for (int i = 0; i < listeners.length; i++) { - BatchListener listener = listeners[i]; - if (listener instanceof StepListener) { - list.add(listener); - } - } - return (StepListener[]) list.toArray(new StepListener[list.size()]); - } - -} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/DefaultStepFactoryBean.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/DefaultStepFactoryBean.java deleted file mode 100644 index 7ad022f4a..000000000 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/DefaultStepFactoryBean.java +++ /dev/null @@ -1,246 +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; - -import org.springframework.batch.core.BatchListener; -import org.springframework.batch.core.Step; -import org.springframework.batch.core.StepListener; -import org.springframework.batch.core.step.support.LimitCheckingItemSkipPolicy; -import org.springframework.batch.core.step.support.ListenerMulticaster; -import org.springframework.batch.core.step.support.NeverSkipItemSkipPolicy; -import org.springframework.batch.item.ItemReader; -import org.springframework.batch.item.ItemStream; -import org.springframework.batch.item.ItemWriter; -import org.springframework.batch.repeat.exception.ExceptionHandler; -import org.springframework.batch.repeat.exception.SimpleLimitExceptionHandler; -import org.springframework.batch.repeat.policy.SimpleCompletionPolicy; -import org.springframework.batch.repeat.support.RepeatTemplate; -import org.springframework.batch.repeat.support.TaskExecutorRepeatTemplate; -import org.springframework.core.task.TaskExecutor; - -/** - * Most common configuration options for simple steps should be found here. Use - * this factory bean instead of creating a {@link Step} implementation manually. - * - * @author Dave Syer - * - */ -public class DefaultStepFactoryBean extends AbstractStepFactoryBean { - - private int skipLimit = 0; - - private int commitInterval = 0; - - private ItemStream[] streams = new ItemStream[0]; - - private BatchListener[] listeners = new BatchListener[0]; - - private ListenerMulticaster listener = new ListenerMulticaster(); - - private TaskExecutor taskExecutor; - - private ItemHandler itemHandler; - - private RepeatTemplate stepOperations; - - private SimpleLimitExceptionHandler exceptionHandler; - - /** - * Set the commit interval. - * - * @param commitInterval - */ - public void setCommitInterval(int commitInterval) { - this.commitInterval = commitInterval; - } - - /** - * The streams to inject into the {@link Step}. Any instance of - * {@link ItemStream} can be used, and will then receive callbacks at the - * appropriate stage in the step. - * - * @param streams an array of listeners - */ - public void setStreams(ItemStream[] streams) { - this.streams = streams; - } - - /** - * Public setter for a limit that determines skip policy. If this value is - * positive then an exception in chunk processing will cause the item to be - * skipped and no exception propagated until the limit is reached. If it is - * zero then all exceptions will be propagated from the chunk and cause the - * step to abort. - * - * @param skipLimit the value to set. Default is 0 (never skip). - */ - public void setSkipLimit(int skipLimit) { - this.skipLimit = skipLimit; - } - - /** - * The listeners to inject into the {@link Step}. Any instance of - * {@link BatchListener} can be used, and will then receive callbacks at the - * appropriate stage in the step. - * - * @param listeners an array of listeners - */ - public void setListeners(BatchListener[] listeners) { - this.listeners = listeners; - } - - /** - * Protected getter for the step operations to make them available in - * subclasses. - * @return the step operations - */ - protected RepeatTemplate getStepOperations() { - return stepOperations; - } - - /** - * Public setter for the SimpleLimitExceptionHandler. - * @param exceptionHandler the exceptionHandler to set - */ - public void setExceptionHandler(SimpleLimitExceptionHandler exceptionHandler) { - this.exceptionHandler = exceptionHandler; - } - - /** - * Protected getter for the {@link ExceptionHandler}. - * @return the {@link ExceptionHandler} - */ - protected ExceptionHandler getExceptionHandler() { - return exceptionHandler; - } - - /** - * Public setter for the {@link TaskExecutor}. If this is set, then it will - * be used to execute the chunk processing inside the {@link Step}. - * - * @param taskExecutor the taskExecutor to set - */ - public void setTaskExecutor(TaskExecutor taskExecutor) { - this.taskExecutor = taskExecutor; - } - - /** - * Public getter for the ItemProcessor. - * @return the itemProcessor - */ - protected ItemHandler getItemHandler() { - return itemHandler; - } - - /** - * Public setter for the ItemProcessor. - * @param itemHandler the itemProcessor to set - */ - protected void setItemHandler(ItemHandler itemHandler) { - this.itemHandler = itemHandler; - } - - /** - * @param step - * - */ - protected void applyConfiguration(ItemOrientedStep step) { - - super.applyConfiguration(step); - - step.setStreams(streams); - - for (int i = 0; i < listeners.length; i++) { - BatchListener listener = listeners[i]; - if (listener instanceof StepListener) { - step.registerStepListener((StepListener) listener); - } - else { - this.listener.register(listener); - } - } - - ItemReader itemReader = getItemReader(); - ItemWriter itemWriter = getItemWriter(); - - // 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 StepListener) { - step.registerStepListener((StepListener) itemReader); - } - if (itemWriter instanceof ItemStream) { - step.registerStream((ItemStream) itemWriter); - } - if (itemWriter instanceof StepListener) { - step.registerStepListener((StepListener) itemWriter); - } - - BatchListenerFactoryHelper helper = new BatchListenerFactoryHelper(); - - if (commitInterval > 0) { - RepeatTemplate chunkOperations = new RepeatTemplate(); - chunkOperations.setCompletionPolicy(new SimpleCompletionPolicy(commitInterval)); - helper.addChunkListeners(chunkOperations, listeners); - step.setChunkOperations(chunkOperations); - } - - StepListener[] stepListeners = helper.getStepListeners(listeners); - itemReader = helper.getItemReader(itemReader, listeners); - itemWriter = helper.getItemWriter(itemWriter, listeners); - stepOperations = new RepeatTemplate(); - - // In case they are used by subclasses: - setItemReader(itemReader); - setItemWriter(itemWriter); - - step.setStepListeners(stepListeners); - - if (taskExecutor != null) { - TaskExecutorRepeatTemplate repeatTemplate = new TaskExecutorRepeatTemplate(); - repeatTemplate.setTaskExecutor(taskExecutor); - stepOperations = repeatTemplate; - } - - step.setStepOperations(stepOperations); - - ItemSkipPolicyItemHandler itemProcessor = new ItemSkipPolicyItemHandler(itemReader, itemWriter); - - if (skipLimit > 0) { - /* - * If there is a skip limit (not the default) then we are prepared - * to absorb exceptions at the step level because the failed items - * will never re-appear after a rollback. - */ - itemProcessor.setItemSkipPolicy(new LimitCheckingItemSkipPolicy(skipLimit)); - setExceptionHandler(new SimpleLimitExceptionHandler(skipLimit)); - stepOperations.setExceptionHandler(getExceptionHandler()); - step.setStepOperations(stepOperations); - } - else { - // This is the default in ItemOrientedStep anyway... - itemProcessor.setItemSkipPolicy(new NeverSkipItemSkipPolicy()); - } - - setItemHandler(itemProcessor); - step.setItemHandler(itemProcessor); - - } - -} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/ItemSkipPolicyItemHandler.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/ItemSkipPolicyItemHandler.java deleted file mode 100644 index 2a58ae03a..000000000 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/ItemSkipPolicyItemHandler.java +++ /dev/null @@ -1,103 +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; - -import org.springframework.batch.core.ItemSkipPolicy; -import org.springframework.batch.core.StepContribution; -import org.springframework.batch.core.step.support.NeverSkipItemSkipPolicy; -import org.springframework.batch.item.ItemReader; -import org.springframework.batch.item.ItemWriter; -import org.springframework.batch.item.Skippable; -import org.springframework.batch.repeat.ExitStatus; - -/** - * @author Dave Syer - * - */ -public class ItemSkipPolicyItemHandler extends SimpleItemHandler { - - private ItemSkipPolicy itemSkipPolicy = new NeverSkipItemSkipPolicy(); - - /** - * @param itemReader - * @param itemWriter - */ - public ItemSkipPolicyItemHandler(ItemReader itemReader, ItemWriter itemWriter) { - super(itemReader, itemWriter); - } - - /** - * @param itemSkipPolicy - */ - public void setItemSkipPolicy(ItemSkipPolicy itemSkipPolicy) { - this.itemSkipPolicy = itemSkipPolicy; - } - - /** - * Execute the business logic, delegating to the reader and writer. - * Subclasses could extend the behaviour as long as they always return the - * value of this method call in their superclass.
- * - * Read from the {@link ItemReader} and process (if not null) with the - * {@link ItemWriter}.
- * - * If there is an exception and the reader or writer implements - * {@link Skippable} then the skip method is called. - * - * @param contribution the current step - * @return {@link ExitStatus#CONTINUABLE} if there is more processing to do - * @throws Exception if there is an error - */ - public ExitStatus handle(StepContribution contribution) throws Exception { - ExitStatus exitStatus = ExitStatus.CONTINUABLE; - - try { - - exitStatus = super.handle(contribution); - - } - catch (Exception e) { - - if (itemSkipPolicy.shouldSkip(e, contribution.getSkipCount())) { - contribution.incrementSkipCount(); - skip(); - } - else { - // Rethrow so that outer transaction is rolled back properly - throw e; - } - - } - - return exitStatus; - } - - /** - * Mark the current item as skipped if possible. If the reader and / or - * writer are {@link Skippable} then delegate to them in that order. - * - * @see org.springframework.batch.item.Skippable#skip() - */ - private void skip() { - if (getItemReader() instanceof Skippable) { - ((Skippable) getItemReader()).skip(); - } - if (getItemWriter() instanceof Skippable) { - ((Skippable) getItemWriter()).skip(); - } - } - -} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/RepeatOperationsStepFactoryBean.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/RepeatOperationsStepFactoryBean.java deleted file mode 100644 index a70d69b55..000000000 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/RepeatOperationsStepFactoryBean.java +++ /dev/null @@ -1,139 +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; - -import org.springframework.batch.core.BatchListener; -import org.springframework.batch.core.Step; -import org.springframework.batch.core.StepListener; -import org.springframework.batch.item.ItemReader; -import org.springframework.batch.item.ItemStream; -import org.springframework.batch.item.ItemWriter; -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. - * - * @author Dave Syer - * - */ -public class RepeatOperationsStepFactoryBean extends AbstractStepFactoryBean { - - private ItemStream[] streams = new ItemStream[0]; - - private BatchListener[] listeners = new BatchListener[0]; - - private RepeatOperations chunkOperations = new RepeatTemplate(); - - private RepeatOperations stepOperations = new RepeatTemplate(); - - /** - * 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 BatchListener} can be used, and will then receive callbacks at the - * appropriate stage in the step. - * - * @param listeners an array of listeners - */ - public void setListeners(BatchListener[] listeners) { - this.listeners = listeners; - } - - /** - * 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(ItemOrientedStep step) { - - super.applyConfiguration(step); - - step.setStreams(streams); - - ItemReader itemReader = getItemReader(); - ItemWriter itemWriter = getItemWriter(); - - /* - * 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 StepListener) { - step.registerStepListener((StepListener) itemReader); - } - if (itemWriter instanceof ItemStream) { - step.registerStream((ItemStream) itemWriter); - } - if (itemWriter instanceof StepListener) { - step.registerStepListener((StepListener) itemWriter); - } - - BatchListenerFactoryHelper helper = new BatchListenerFactoryHelper(); - - StepListener[] stepListeners = helper.getStepListeners(listeners); - itemReader = helper.getItemReader(itemReader, listeners); - itemWriter = helper.getItemWriter(itemWriter, listeners); - RepeatOperations stepOperations = helper.addChunkListeners(this.stepOperations, listeners); - - // In case they are used by subclasses: - setItemReader(itemReader); - setItemWriter(itemWriter); - - step.setStepListeners(stepListeners); - step.setItemHandler(new SimpleItemHandler(itemReader, itemWriter)); - - step.setChunkOperations(chunkOperations); - step.setStepOperations(stepOperations); - - } - -} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/SimpleItemHandler.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/SimpleItemHandler.java deleted file mode 100644 index 0a1797c48..000000000 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/SimpleItemHandler.java +++ /dev/null @@ -1,114 +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; - -import org.springframework.batch.core.StepContribution; -import org.springframework.batch.item.ClearFailedException; -import org.springframework.batch.item.FlushFailedException; -import org.springframework.batch.item.ItemReader; -import org.springframework.batch.item.ItemWriter; -import org.springframework.batch.item.MarkFailedException; -import org.springframework.batch.item.ResetFailedException; -import org.springframework.batch.repeat.ExitStatus; - -/** - * Simplest possible implementation of {@link ItemHandler} with no skipping or - * recovering. Just delegates all calls to the provided {@link ItemReader} and - * {@link ItemWriter}. - * - * @author Dave Syer - * - */ -public class SimpleItemHandler implements ItemHandler { - - private ItemReader itemReader; - - private ItemWriter itemWriter; - - /** - * @param itemReader - * @param itemWriter - */ - public SimpleItemHandler(ItemReader itemReader, ItemWriter itemWriter) { - super(); - this.itemReader = itemReader; - this.itemWriter = itemWriter; - } - - /** - * Public getter for the ItemReader. - * @return the itemReader - */ - public ItemReader getItemReader() { - return itemReader; - } - - /** - * Public getter for the ItemWriter. - * @return the itemWriter - */ - public ItemWriter getItemWriter() { - return itemWriter; - } - - /** - * Read from the {@link ItemReader} and process (if not null) with the - * {@link ItemWriter}. - * - * @see org.springframework.batch.core.step.ItemHandler#handle(org.springframework.batch.core.StepContribution) - */ - public ExitStatus handle(StepContribution contribution) throws Exception { - Object item = itemReader.read(); - if (item == null) { - return ExitStatus.FINISHED; - } - itemWriter.write(item); - return ExitStatus.CONTINUABLE; - } - - /** - * @throws MarkFailedException - * @see org.springframework.batch.item.ItemReader#mark() - */ - public void mark() throws MarkFailedException { - itemReader.mark(); - } - - /** - * @throws ResetFailedException - * @see org.springframework.batch.item.ItemReader#reset() - */ - public void reset() throws ResetFailedException { - itemReader.reset(); - } - - /** - * @throws ClearFailedException - * @see org.springframework.batch.item.ItemWriter#clear() - */ - public void clear() throws ClearFailedException { - itemWriter.clear(); - } - - /** - * @throws FlushFailedException - * @see org.springframework.batch.item.ItemWriter#flush() - */ - public void flush() throws FlushFailedException { - itemWriter.flush(); - } - -} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/StatefulRetryStepFactoryBean.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/StatefulRetryStepFactoryBean.java deleted file mode 100644 index b615bbc20..000000000 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/StatefulRetryStepFactoryBean.java +++ /dev/null @@ -1,211 +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; - -import org.springframework.batch.core.Step; -import org.springframework.batch.core.StepContribution; -import org.springframework.batch.core.step.support.SimpleRetryExceptionHandler; -import org.springframework.batch.item.ItemKeyGenerator; -import org.springframework.batch.item.ItemReader; -import org.springframework.batch.item.ItemRecoverer; -import org.springframework.batch.item.ItemWriter; -import org.springframework.batch.repeat.ExitStatus; -import org.springframework.batch.retry.RetryListener; -import org.springframework.batch.retry.RetryOperations; -import org.springframework.batch.retry.RetryPolicy; -import org.springframework.batch.retry.backoff.BackOffPolicy; -import org.springframework.batch.retry.callback.ItemReaderRetryCallback; -import org.springframework.batch.retry.policy.ItemReaderRetryPolicy; -import org.springframework.batch.retry.policy.SimpleRetryPolicy; -import org.springframework.batch.retry.support.RetryTemplate; - -/** - * Factory bean for step that executes its item processing with a stateful - * retry. Failed items are never skipped, but always cause a rollback. Before a - * rollback, the {@link Step} makes a record of the failed item, caching it - * under a key given by the {@link ItemKeyGenerator}. Then when it is - * re-presented by the {@link ItemReader} it is recognised and retried up to a - * limit given by the {@link RetryPolicy}. When the retry is exhausted instead - * of the item being skipped it is handled by an {@link ItemRecoverer}.
- * - * The skipLimit property is still used to control the overall exception - * handling policy. Only exhausted retries count against the exception handler, - * instead of counting all exceptions. - * - * @author Dave Syer - * - */ -public class StatefulRetryStepFactoryBean extends DefaultStepFactoryBean { - - private ItemKeyGenerator itemKeyGenerator; - - private ItemRecoverer itemRecoverer; - - private int retryLimit; - - private Class[] retryableExceptionClasses; - - private BackOffPolicy backOffPolicy; - - private RetryListener[] retryListeners; - - /** - * Public setter for the retry limit. Each item can be retried up to this - * limit. - * @param retryLimit the retry limit to set - */ - public void setRetryLimit(int retryLimit) { - this.retryLimit = retryLimit; - } - - /** - * Public setter for the Class[]. - * @param retryableExceptionClasses the retryableExceptionClasses to set - */ - public void setRetryableExceptionClasses(Class[] retryableExceptionClasses) { - this.retryableExceptionClasses = retryableExceptionClasses; - } - - /** - * Public setter for the {@link BackOffPolicy}. - * @param backOffPolicy the {@link BackOffPolicy} to set - */ - public void setBackOffPolicy(BackOffPolicy backOffPolicy) { - this.backOffPolicy = backOffPolicy; - } - - /** - * Public setter for the {@link RetryListener}s. - * @param retryListeners the {@link RetryListener}s to set - */ - public void setRetryListeners(RetryListener[] retryListeners) { - this.retryListeners = retryListeners; - } - - /** - * Public setter for the {@link ItemKeyGenerator} which will be used to - * cache failed items between transactions. If it is not injected but the - * reader or writer implement {@link ItemKeyGenerator}, one of those will - * be used instead (preferring the reader to the writer if both would be - * appropriate). If neither can be used, then the default will be to just - * use the item itself as a cache key. - * - * @param itemKeyGenerator the {@link ItemKeyGenerator} to set - */ - public void setItemKeyGenerator(ItemKeyGenerator itemKeyGenerator) { - this.itemKeyGenerator = itemKeyGenerator; - } - - /** - * Public setter for the {@link ItemRecoverer}. If this is set the - * {@link ItemRecoverer#recover(Object, Throwable)} will be called when - * retry is exhausted, and within the business transaction (which will not - * roll back because of any other item-related errors). - * - * @param itemRecoverer the {@link ItemRecoverer} to set - */ - public void setItemRecoverer(ItemRecoverer itemRecoverer) { - this.itemRecoverer = itemRecoverer; - } - - /** - * @param step - * - */ - protected void applyConfiguration(ItemOrientedStep step) { - - super.applyConfiguration(step); - - if (retryLimit > 0) { - - SimpleRetryPolicy retryPolicy = new SimpleRetryPolicy(retryLimit); - if (retryableExceptionClasses != null) { - retryPolicy.setRetryableExceptionClasses(retryableExceptionClasses); - } - - // Co-ordinate the retry policy with the exception handler: - getStepOperations() - .setExceptionHandler(new SimpleRetryExceptionHandler(retryPolicy, getExceptionHandler())); - - ItemReaderRetryCallback retryCallback = new ItemReaderRetryCallback(getItemReader(), itemKeyGenerator, - getItemWriter()); - retryCallback.setRecoverer(itemRecoverer); - ItemReaderRetryPolicy itemProviderRetryPolicy = new ItemReaderRetryPolicy(retryPolicy); - - RetryTemplate retryTemplate = new RetryTemplate(); - if (retryListeners!=null) { - retryTemplate.setListeners(retryListeners); - } - retryTemplate.setRetryPolicy(itemProviderRetryPolicy); - if (backOffPolicy != null) { - retryTemplate.setBackOffPolicy(backOffPolicy); - } - - StatefulRetryItemHandler itemProcessor = new StatefulRetryItemHandler(getItemReader(), getItemWriter(), - retryTemplate, retryCallback); - - step.setItemHandler(itemProcessor); - - } - - } - - private static class StatefulRetryItemHandler extends SimpleItemHandler { - - final private RetryOperations retryOperations; - - final private ItemReaderRetryCallback retryCallback; - - /** - * @param itemReader - * @param itemWriter - * @param retryCallback - * @param retryTemplate - */ - public StatefulRetryItemHandler(ItemReader itemReader, ItemWriter itemWriter, RetryOperations retryTemplate, - ItemReaderRetryCallback retryCallback) { - super(itemReader, itemWriter); - this.retryOperations = retryTemplate; - this.retryCallback = retryCallback; - } - - /** - * Execute the business logic, delegating to the reader and writer. - * Subclasses could extend the behaviour as long as they always return - * the value of this method call in their superclass.
- * - * Read from the {@link ItemReader} and process (if not null) with the - * {@link ItemWriter}. The call to {@link ItemWriter} is wrapped in a - * stateful retry. In that case the {@link ItemRecoverer} is used (if - * provided) in the case of an exception to apply alternate processing - * to the item. If the stateful retry is in place then the recovery will - * happen in the next transaction automatically, otherwise it might be - * necessary for clients to make the recover method transactional with - * appropriate propagation behaviour (probably REQUIRES_NEW because the - * call will happen in the context of a transaction that is about to - * rollback).
- * - * @param contribution the current step - * @return {@link ExitStatus#CONTINUABLE} if there is more processing to - * do - * @throws Exception if there is an error - */ - public ExitStatus handle(StepContribution contribution) throws Exception { - return new ExitStatus(retryOperations.execute(retryCallback) != null); - } - - } -} diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/launch/support/CommandLineJobRunnerTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/launch/support/CommandLineJobRunnerTests.java index 52f13bc1d..17e4cf93b 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/launch/support/CommandLineJobRunnerTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/launch/support/CommandLineJobRunnerTests.java @@ -21,10 +21,7 @@ import org.springframework.batch.core.Job; import org.springframework.batch.core.JobExecution; import org.springframework.batch.core.JobParameters; import org.springframework.batch.core.launch.JobLauncher; -import org.springframework.batch.core.launch.support.CommandLineJobRunner; -import org.springframework.batch.core.launch.support.SystemExiter; import org.springframework.batch.core.repository.JobExecutionAlreadyRunningException; -import org.springframework.batch.core.runtime.ExitStatusExceptionClassifier; import org.springframework.batch.repeat.ExitStatus; import org.springframework.util.ClassUtils; @@ -61,7 +58,6 @@ public class CommandLineJobRunnerTests extends TestCase { ExitStatus exitStatus = ExitStatus.FINISHED; jobExecution.setExitStatus(exitStatus); StubJobLauncher.jobExecution = jobExecution; - StubExceptionClassifier.exception = null; } public void testMain() { @@ -69,10 +65,10 @@ public class CommandLineJobRunnerTests extends TestCase { assertEquals(0, StubSystemExiter.getStatus()); } - public void testJobAlreadyRunning() { + public void testJobAlreadyRunning() throws Throwable { StubJobLauncher.throwExecutionRunningException = true; CommandLineJobRunner.main(args); - assertTrue(StubExceptionClassifier.exception instanceof JobExecutionAlreadyRunningException); + assertEquals(1, StubSystemExiter.status); } // can't test because it will cause the system to exit. @@ -85,9 +81,6 @@ public class CommandLineJobRunnerTests extends TestCase { public void testWithNoParameters() throws Throwable { String[] args = new String[] { jobPath, jobName }; CommandLineJobRunner.main(args); - if (StubExceptionClassifier.exception != null) { - throw StubExceptionClassifier.exception; - } assertEquals(0, StubSystemExiter.status); assertEquals(new JobParameters(), StubJobLauncher.jobParameters); } @@ -137,22 +130,4 @@ public class CommandLineJobRunnerTests extends TestCase { } } - public static class StubExceptionClassifier implements ExitStatusExceptionClassifier { - - public static Throwable exception; - - public Object classify(Throwable throwable) { - return null; - } - - public Object getDefault() { - return null; - } - - public ExitStatus classifyForExitCode(Throwable throwable) { - exception = throwable; - return ExitStatus.FAILED; - } - - } } diff --git a/spring-batch-core/src/test/resources/org/springframework/batch/core/launch/support/test-environment.xml b/spring-batch-core/src/test/resources/org/springframework/batch/core/launch/support/test-environment.xml index 741a55dc8..bb6b43dd0 100644 --- a/spring-batch-core/src/test/resources/org/springframework/batch/core/launch/support/test-environment.xml +++ b/spring-batch-core/src/test/resources/org/springframework/batch/core/launch/support/test-environment.xml @@ -12,9 +12,6 @@ - - + class="org.springframework.batch.core.launch.support.CommandLineJobRunnerTests$StubSystemExiter" /> diff --git a/spring-batch-samples/src/main/resources/jobs/retrySample.xml b/spring-batch-samples/src/main/resources/jobs/retrySample.xml index e2c17c527..9b9705241 100644 --- a/spring-batch-samples/src/main/resources/jobs/retrySample.xml +++ b/spring-batch-samples/src/main/resources/jobs/retrySample.xml @@ -10,7 +10,7 @@ + class="org.springframework.batch.core.step.support.StatefulRetryStepFactoryBean"> diff --git a/spring-batch-samples/src/main/resources/simple-job-launcher-context.xml b/spring-batch-samples/src/main/resources/simple-job-launcher-context.xml index 993e42b49..ca46d8077 100644 --- a/spring-batch-samples/src/main/resources/simple-job-launcher-context.xml +++ b/spring-batch-samples/src/main/resources/simple-job-launcher-context.xml @@ -58,7 +58,7 @@ -