diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/listener/RepeatListenerItemReadListenerAdapter.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/listener/RepeatListenerItemReadListenerAdapter.java deleted file mode 100644 index 97672b49d..000000000 --- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/listener/RepeatListenerItemReadListenerAdapter.java +++ /dev/null @@ -1,100 +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.execution.listener; - -import org.springframework.batch.core.domain.ItemReadListener; -import org.springframework.batch.repeat.RepeatContext; -import org.springframework.batch.repeat.RepeatListener; -import org.springframework.batch.repeat.synch.RepeatSynchronizationManager; -import org.springframework.util.ObjectUtils; - -/** - * Adapts a {@link RepeatListener} to the {@link ItemReadListener} interface. - * The {@link RepeatListener} is assumed to be targeted at the chunk operations - * in a step, so after an item corresponds to the after method in - * {@link RepeatListener}.
- * - * The open and close methods on the {@link RepeatListener} are also invoked. - * The open method is called on the first call to {@link #beforeRead()}, and - * the close is registered as a destruction callback in the - * {@link RepeatContext}.
- * - * A {@link RepeatContext} is obtained as needed from the - * {@link RepeatSynchronizationManager}. - * - * @author Dave Syer - * - */ -public class RepeatListenerItemReadListenerAdapter implements ItemReadListener { - - private RepeatListener delegate; - - /** - * @param delegate - */ - public RepeatListenerItemReadListenerAdapter(RepeatListener delegate) { - super(); - this.delegate = delegate; - } - - /** - * Does nothing. - * - * @see org.springframework.batch.core.domain.ItemReadListener#afterRead(java.lang.Object) - */ - public void afterRead(Object item) { - // NO-OP - - } - - /** - * Calls the delegate {@link RepeatListener#before}. Also calls - * {@link RepeatListener#open} if it hasn't been called yet on this context. - * - * @see org.springframework.batch.core.domain.ItemReadListener#beforeRead() - */ - public void beforeRead() { - RepeatContext context = RepeatSynchronizationManager.getContext(); - maybeOpen(context); - delegate.before(context); - } - - /** - * @param context - */ - private void maybeOpen(final RepeatContext context) { - String identity = ObjectUtils.identityToString(delegate); - if (!context.hasAttribute(identity)) { - context.setAttribute(identity, Boolean.TRUE); - delegate.open(context); - context.registerDestructionCallback(identity, new Runnable() { - public void run() { - delegate.close(context); - } - }); - } - } - - /** - * Calls the delegate {@link RepeatListener#onError}. - * - * @see org.springframework.batch.core.domain.ItemReadListener#onReadError(java.lang.Exception) - */ - public void onReadError(Exception ex) { - delegate.onError(RepeatSynchronizationManager.getContext(), ex); - } - -} diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/listener/RepeatListenerItemWriteListenerAdapter.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/listener/RepeatListenerItemWriteListenerAdapter.java deleted file mode 100644 index 00b8ad68c..000000000 --- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/listener/RepeatListenerItemWriteListenerAdapter.java +++ /dev/null @@ -1,77 +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.execution.listener; - -import org.springframework.batch.core.domain.ItemWriteListener; -import org.springframework.batch.repeat.ExitStatus; -import org.springframework.batch.repeat.RepeatContext; -import org.springframework.batch.repeat.RepeatListener; -import org.springframework.batch.repeat.synch.RepeatSynchronizationManager; - -/** - * Adapts a {@link RepeatListener} to the {@link ItemWriteListener} interface. - * The {@link RepeatListener} is assumed to be targeted at the chunk operations - * in a step, so after an item corresponds to the after method in - * {@link RepeatListener}.
- * - * A {@link RepeatContext} is obtained as needed from the - * {@link RepeatSynchronizationManager}. - * - * @author Dave Syer - * - */ -public class RepeatListenerItemWriteListenerAdapter implements ItemWriteListener { - - private RepeatListener delegate; - - /** - * @param delegate - */ - public RepeatListenerItemWriteListenerAdapter(RepeatListener delegate) { - super(); - this.delegate = delegate; - } - - /** - * Calls the delegate {@link RepeatListener#after} with - * {@link ExitStatus#CONTINUABLE}. - * - * @see org.springframework.batch.core.domain.ItemWriteListener#afterWrite() - */ - public void afterWrite() { - delegate.after(RepeatSynchronizationManager.getContext(), ExitStatus.CONTINUABLE); - } - - /** - * Does nothing. - * - * @see org.springframework.batch.core.domain.ItemWriteListener#beforeWrite(java.lang.Object) - */ - public void beforeWrite(Object item) { - // NO-OP - } - - /** - * Calls the delegate {@link RepeatListener#onError} ignoring the item. - * - * @see org.springframework.batch.core.domain.ItemWriteListener#onWriteError(java.lang.Exception, - * java.lang.Object) - */ - public void onWriteError(Exception ex, Object item) { - delegate.onError(RepeatSynchronizationManager.getContext(), ex); - } - -} diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/DefaultStepFactoryBean.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/DefaultStepFactoryBean.java new file mode 100644 index 000000000..b42c330f9 --- /dev/null +++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/DefaultStepFactoryBean.java @@ -0,0 +1,45 @@ +/* + * 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.execution.step.support; + +import org.springframework.batch.execution.step.ItemOrientedStep; + +/** + * Adds listeners to {@link SimpleStepFactoryBean}. + * + * @author Dave Syer + * + */ +public class DefaultStepFactoryBean extends SimpleStepFactoryBean { + + private Object[] listeners = new Object[0]; + + /** + * @param listeners + */ + public void setListeners(Object[] listeners) { + this.listeners = listeners; + } + + /** + * @param step + * + */ + protected void applyConfiguration(ItemOrientedStep step) { + super.applyConfiguration(step); + step.setListeners(listeners); + } +} diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/KitchenSinkStepFactoryBean.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/KitchenSinkStepFactoryBean.java index 144b84fca..a849d2b2e 100644 --- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/KitchenSinkStepFactoryBean.java +++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/KitchenSinkStepFactoryBean.java @@ -16,16 +16,18 @@ package org.springframework.batch.execution.step.support; import org.springframework.batch.core.domain.ItemSkipPolicy; +import org.springframework.batch.core.domain.Step; import org.springframework.batch.execution.step.ItemOrientedStep; import org.springframework.batch.item.ItemKeyGenerator; -import org.springframework.batch.repeat.RepeatOperations; import org.springframework.batch.repeat.exception.handler.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.batch.retry.RetryPolicy; import org.springframework.batch.retry.callback.ItemReaderRetryCallback; import org.springframework.batch.retry.policy.ItemReaderRetryPolicy; import org.springframework.batch.retry.support.RetryTemplate; +import org.springframework.core.task.TaskExecutor; /** * @author Dave Syer @@ -39,9 +41,9 @@ public class KitchenSinkStepFactoryBean extends AbstractStepFactoryBean { private ItemSkipPolicy itemSkipPolicy = new NeverSkipItemSkipPolicy(); - private RepeatOperations stepOperations = new RepeatTemplate(); + private TaskExecutor taskExecutor; - private RetryPolicy retryPolicy = null; + private RetryPolicy retryPolicy; private ExceptionHandler exceptionHandler; @@ -71,11 +73,13 @@ public class KitchenSinkStepFactoryBean extends AbstractStepFactoryBean { } /** - * Public setter for the RepeatOperations. - * @param stepOperations the stepOperations to set + * 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 setStepOperations(RepeatOperations stepOperations) { - this.stepOperations = stepOperations; + public void setTaskExecutor(TaskExecutor taskExecutor) { + this.taskExecutor = taskExecutor; } /** @@ -111,9 +115,9 @@ public class KitchenSinkStepFactoryBean extends AbstractStepFactoryBean { * */ protected void applyConfiguration(ItemOrientedStep step) { - + super.applyConfiguration(step); - + step.setListeners(listeners); step.setItemSkipPolicy(itemSkipPolicy); @@ -133,7 +137,15 @@ public class KitchenSinkStepFactoryBean extends AbstractStepFactoryBean { step.setChunkOperations(chunkOperations); } - if (exceptionHandler != null && stepOperations instanceof RepeatTemplate) { + RepeatTemplate stepOperations = new RepeatTemplate(); + + if (taskExecutor != null) { + TaskExecutorRepeatTemplate repeatTemplate = new TaskExecutorRepeatTemplate(); + repeatTemplate.setTaskExecutor(taskExecutor); + stepOperations = repeatTemplate; + } + + if (exceptionHandler != null) { ((RepeatTemplate) stepOperations).setExceptionHandler(exceptionHandler); step.setStepOperations(stepOperations); } diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/ListenerMulticaster.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/ListenerMulticaster.java index 2e512ceb7..1e5b371a0 100644 --- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/ListenerMulticaster.java +++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/ListenerMulticaster.java @@ -25,15 +25,11 @@ import org.springframework.batch.execution.listener.CompositeChunkListener; import org.springframework.batch.execution.listener.CompositeItemReadListener; import org.springframework.batch.execution.listener.CompositeItemWriteListener; import org.springframework.batch.execution.listener.CompositeStepListener; -import org.springframework.batch.execution.listener.RepeatListenerItemReadListenerAdapter; -import org.springframework.batch.execution.listener.RepeatListenerItemWriteListenerAdapter; import org.springframework.batch.item.ExecutionContext; import org.springframework.batch.item.ItemStream; import org.springframework.batch.item.exception.StreamException; import org.springframework.batch.item.stream.CompositeItemStream; import org.springframework.batch.repeat.ExitStatus; -import org.springframework.batch.repeat.RepeatListener; -import org.springframework.batch.repeat.listener.CompositeRepeatListener; /** * @author Dave Syer @@ -52,15 +48,11 @@ public class ListenerMulticaster implements ItemStream, StepListener, ChunkListe private CompositeItemWriteListener itemWriteListener = new CompositeItemWriteListener(); - private CompositeRepeatListener repeatListener = new CompositeRepeatListener(); - /** * Initialise the listener instance. */ public ListenerMulticaster() { super(); - itemWriteListener.register(new RepeatListenerItemWriteListenerAdapter(repeatListener)); - itemReadListener.register(new RepeatListenerItemReadListenerAdapter(repeatListener)); } /** @@ -79,7 +71,7 @@ public class ListenerMulticaster implements ItemStream, StepListener, ChunkListe /** * Register the listener for callbacks on the appropriate interfaces * implemented. Any {@link BatchListener} can be provided, or an - * {@link ItemStream}. Other types will be ignored. + * {@link ItemStream}. Other types will be ignored. */ public void register(Object listener) { if (listener instanceof StepListener) { @@ -97,9 +89,6 @@ public class ListenerMulticaster implements ItemStream, StepListener, ChunkListe if (listener instanceof ItemWriteListener) { this.itemWriteListener.register((ItemWriteListener) listener); } - if (listener instanceof RepeatListener) { - this.repeatListener.register((RepeatListener) listener); - } } /** diff --git a/spring-batch-execution/src/test/resources/org/springframework/batch/execution/configuration/test-context.xml b/spring-batch-execution/src/test/resources/org/springframework/batch/execution/configuration/test-context.xml index a46d69055..dcb20a8a6 100644 --- a/spring-batch-execution/src/test/resources/org/springframework/batch/execution/configuration/test-context.xml +++ b/spring-batch-execution/src/test/resources/org/springframework/batch/execution/configuration/test-context.xml @@ -20,7 +20,7 @@ - + diff --git a/spring-batch-samples/src/main/resources/jobs/beanWrapperMapperSampleJob.xml b/spring-batch-samples/src/main/resources/jobs/beanWrapperMapperSampleJob.xml index dbbda53e9..12b643ee4 100644 --- a/spring-batch-samples/src/main/resources/jobs/beanWrapperMapperSampleJob.xml +++ b/spring-batch-samples/src/main/resources/jobs/beanWrapperMapperSampleJob.xml @@ -12,7 +12,7 @@ - + - + - + - + + + + + + + + diff --git a/spring-batch-samples/src/main/resources/jobs/multilineJob.xml b/spring-batch-samples/src/main/resources/jobs/multilineJob.xml index 8c0980121..a93a1576f 100644 --- a/spring-batch-samples/src/main/resources/jobs/multilineJob.xml +++ b/spring-batch-samples/src/main/resources/jobs/multilineJob.xml @@ -9,7 +9,7 @@ - + - + diff --git a/spring-batch-samples/src/main/resources/jobs/parallelJob.xml b/spring-batch-samples/src/main/resources/jobs/parallelJob.xml index 6cf76d077..b775c6801 100644 --- a/spring-batch-samples/src/main/resources/jobs/parallelJob.xml +++ b/spring-batch-samples/src/main/resources/jobs/parallelJob.xml @@ -8,14 +8,14 @@ http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.0.xsd http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop-2.0.xsd http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-2.0.xsd"> - + - - - - + + + + @@ -41,15 +41,10 @@ - - + + - - - - + class="org.springframework.core.task.SimpleAsyncTaskExecutor" /> - + - - - - - diff --git a/spring-batch-samples/src/main/resources/jobs/retrySample.xml b/spring-batch-samples/src/main/resources/jobs/retrySample.xml index f75d75a37..c2e7c6c19 100644 --- a/spring-batch-samples/src/main/resources/jobs/retrySample.xml +++ b/spring-batch-samples/src/main/resources/jobs/retrySample.xml @@ -8,19 +8,26 @@ http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.0.xsd http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop-2.0.xsd http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-2.0.xsd"> - + - - - - + + + + - + + + + @@ -30,7 +37,8 @@ - + diff --git a/spring-batch-samples/src/main/resources/jobs/rollbackJob.xml b/spring-batch-samples/src/main/resources/jobs/rollbackJob.xml index 0f6fb6b06..759643060 100644 --- a/spring-batch-samples/src/main/resources/jobs/rollbackJob.xml +++ b/spring-batch-samples/src/main/resources/jobs/rollbackJob.xml @@ -7,21 +7,19 @@ http://www.springframework.org/schema/beans/spring-beans-2.0.xsd http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop-2.0.xsd"> - + - - - - + + class="org.springframework.batch.repeat.exception.handler.SimpleLimitExceptionHandler" + p:limit="5" /> diff --git a/spring-batch-samples/src/main/resources/jobs/tradeJob.xml b/spring-batch-samples/src/main/resources/jobs/tradeJob.xml index ba500f397..54b85b224 100644 --- a/spring-batch-samples/src/main/resources/jobs/tradeJob.xml +++ b/spring-batch-samples/src/main/resources/jobs/tradeJob.xml @@ -13,7 +13,7 @@ - + - - + + - - - - - - - - - - + - + + @@ -161,7 +154,7 @@ -