diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/AbstractStep.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/AbstractStep.java index a6c95525e..139054aeb 100644 --- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/AbstractStep.java +++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/AbstractStep.java @@ -26,7 +26,6 @@ import org.springframework.batch.execution.step.support.NeverSkipItemSkipPolicy; import org.springframework.batch.io.exception.BatchCriticalException; import org.springframework.batch.item.ItemReader; import org.springframework.batch.item.ItemWriter; -import org.springframework.batch.item.stream.StreamManager; import org.springframework.batch.repeat.exception.handler.ExceptionHandler; import org.springframework.batch.retry.RetryPolicy; import org.springframework.beans.factory.BeanNameAware; @@ -43,23 +42,21 @@ import org.springframework.util.Assert; public abstract class AbstractStep implements Step, InitializingBean, BeanNameAware { protected ExceptionHandler exceptionHandler; - + protected RetryPolicy retryPolicy; protected JobRepository jobRepository; protected PlatformTransactionManager transactionManager; - protected StreamManager streamManager; - protected ItemReader itemReader; protected ItemWriter itemWriter; - + protected ItemSkipPolicy itemSkipPolicy = new NeverSkipItemSkipPolicy(); - + protected ItemFailureHandler itemFailureHandler = new DefaultItemFailureHandler(); - + protected String name; protected int startLimit = Integer.MAX_VALUE; @@ -78,9 +75,11 @@ public abstract class AbstractStep implements Step, InitializingBean, BeanNameAw } /** - * Set the name property if it is not already set. Because of the order of the callbacks in a Spring container the - * name property will be set first if it is present. Care is needed with bean definition inheritance - if a parent - * bean has a name, then its children need an explicit name as well, otherwise they will not be unique. + * Set the name property if it is not already set. Because of the order of + * the callbacks in a Spring container the name property will be set first + * if it is present. Care is needed with bean definition inheritance - if a + * parent bean has a name, then its children need an explicit name as well, + * otherwise they will not be unique. * * @see org.springframework.beans.factory.BeanNameAware#setBeanName(java.lang.String) */ @@ -91,7 +90,8 @@ public abstract class AbstractStep implements Step, InitializingBean, BeanNameAw } /** - * Set the name property. Always overrides the default value if this object is a Spring bean. + * Set the name property. Always overrides the default value if this object + * is a Spring bean. * * @see #setBeanName(java.lang.String) */ @@ -134,14 +134,6 @@ public abstract class AbstractStep implements Step, InitializingBean, BeanNameAw this.name = name; } - /** - * Public getter for the {@link RetryPolicy}. - * @return the {@link RetryPolicy} - */ - public RetryPolicy getRetryPolicy() { - return retryPolicy; - } - /** * Public setter for the {@link RetryPolicy}. * @param retryPolicy the {@link RetryPolicy} to set @@ -150,10 +142,6 @@ public abstract class AbstractStep implements Step, InitializingBean, BeanNameAw this.retryPolicy = retryPolicy; } - public ExceptionHandler getExceptionHandler() { - return exceptionHandler; - } - public void setExceptionHandler(ExceptionHandler exceptionHandler) { this.exceptionHandler = exceptionHandler; } @@ -176,15 +164,6 @@ public abstract class AbstractStep implements Step, InitializingBean, BeanNameAw this.transactionManager = transactionManager; } - /** - * Public setter for the {@link StreamManager}. Set either this or the transaction manager, but not both. - * - * @param streamManager the {@link StreamManager} to set. - */ - public void setStreamManager(StreamManager streamManager) { - this.streamManager = streamManager; - } - /** * @param itemReader the itemReader to set */ @@ -202,18 +181,10 @@ public abstract class AbstractStep implements Step, InitializingBean, BeanNameAw public void setItemSkipPolicy(ItemSkipPolicy itemSkipPolicy) { this.itemSkipPolicy = itemSkipPolicy; } - - public ItemSkipPolicy getItemSkipPolicy() { - return itemSkipPolicy; - } - + public void setItemFailureHandler(ItemFailureHandler itemFailureHandler) { this.itemFailureHandler = itemFailureHandler; } - - public ItemFailureHandler getItemFailureHandler() { - return itemFailureHandler; - } /** * Assert that all mandatory properties are set (the {@link JobRepository}). @@ -226,10 +197,7 @@ public abstract class AbstractStep implements Step, InitializingBean, BeanNameAw protected void assertMandatoryProperties() { Assert.notNull(jobRepository, "JobRepository is mandatory"); - Assert.state(transactionManager != null || streamManager != null, - "Either StreamManager or TransactionManager must be set"); - Assert.state(transactionManager == null || streamManager == null, - "Only one of StreamManager or TransactionManager must be set"); + Assert.notNull(transactionManager, "TransactionManager must be set"); Assert.notNull(itemReader, "ItemReader must be provided"); Assert.notNull(itemWriter, "ItemWriter must be provided"); diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/ItemOrientedStep.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/ItemOrientedStep.java index 25d074523..2bdc8d5df 100644 --- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/ItemOrientedStep.java +++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/ItemOrientedStep.java @@ -52,6 +52,7 @@ import org.springframework.batch.retry.policy.ItemReaderRetryPolicy; import org.springframework.batch.retry.support.RetryTemplate; import org.springframework.beans.factory.InitializingBean; import org.springframework.transaction.TransactionStatus; +import org.springframework.transaction.support.DefaultTransactionDefinition; import org.springframework.util.Assert; /** @@ -95,6 +96,8 @@ public class ItemOrientedStep extends AbstractStep implements InitializingBean { private int commitInterval = 0; + private SimpleStreamManager streamManager; + /** * 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 @@ -180,12 +183,7 @@ public class ItemOrientedStep extends AbstractStep implements InitializingBean { retryCallback = new ItemReaderRetryCallback((KeyedItemReader) itemReader, itemWriter); } - if (streamManager == null && transactionManager != null) { - streamManager = new SimpleStreamManager(transactionManager); - } - else if (streamManager == null && transactionManager == null) { - throw new IllegalArgumentException("Either StreamManager or TransactionManager must be set"); - } + streamManager = new SimpleStreamManager(); if (this.chunkOperations instanceof RepeatTemplate && commitInterval > 0) { ((RepeatTemplate) chunkOperations).setCompletionPolicy(new SimpleCompletionPolicy(commitInterval)); @@ -257,7 +255,7 @@ public class ItemOrientedStep extends AbstractStep implements InitializingBean { ExitStatus result; - TransactionStatus transaction = streamManager.getTransaction(); + TransactionStatus transaction = transactionManager.getTransaction(new DefaultTransactionDefinition()); try { itemReader.mark(); @@ -290,7 +288,7 @@ public class ItemOrientedStep extends AbstractStep implements InitializingBean { try { itemReader.mark(); itemWriter.flush(); - streamManager.commit(transaction); + transactionManager.commit(transaction); } catch (Exception e) { fatalException.setException(e); @@ -316,7 +314,7 @@ public class ItemOrientedStep extends AbstractStep implements InitializingBean { try { itemReader.reset(); itemWriter.clear(); - streamManager.rollback(transaction); + transactionManager.rollback(transaction); } catch (Exception e) { fatalException.setException(e); @@ -475,7 +473,7 @@ public class ItemOrientedStep extends AbstractStep implements InitializingBean { } catch (Exception e) { - if (getItemSkipPolicy().shouldSkip(e, contribution.getSkipCount())) { + if (itemSkipPolicy.shouldSkip(e, contribution.getSkipCount())) { contribution.incrementSkipCount(); skip(); } @@ -510,7 +508,7 @@ public class ItemOrientedStep extends AbstractStep implements InitializingBean { item = itemReader.read(); } catch (Exception ex) { - getItemFailureHandler().handleReadFailure(ex); + itemFailureHandler.handleReadFailure(ex); throw ex; } if (item == null) { @@ -521,7 +519,7 @@ public class ItemOrientedStep extends AbstractStep implements InitializingBean { } catch (Exception e) { - getItemFailureHandler().handleWriteFailure(item, e); + itemFailureHandler.handleWriteFailure(item, e); // Re-throw the exception so that the surrounding transaction // rolls back if there is one throw e; diff --git a/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/ItemOrientedStepTests.java b/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/ItemOrientedStepTests.java index 5b2d12e51..2f9e9e228 100644 --- a/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/ItemOrientedStepTests.java +++ b/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/ItemOrientedStepTests.java @@ -46,7 +46,6 @@ import org.springframework.batch.item.exception.StreamException; import org.springframework.batch.item.reader.AbstractItemReader; import org.springframework.batch.item.reader.ListItemReader; import org.springframework.batch.item.stream.ItemStreamSupport; -import org.springframework.batch.item.stream.SimpleStreamManager; import org.springframework.batch.item.writer.AbstractItemWriter; import org.springframework.batch.repeat.ExitStatus; import org.springframework.batch.repeat.RepeatContext; @@ -56,7 +55,8 @@ import org.springframework.batch.repeat.policy.SimpleCompletionPolicy; import org.springframework.batch.repeat.support.RepeatTemplate; import org.springframework.batch.support.PropertiesConverter; import org.springframework.batch.support.transaction.ResourcelessTransactionManager; -import org.springframework.transaction.TransactionStatus; +import org.springframework.transaction.TransactionException; +import org.springframework.transaction.support.DefaultTransactionStatus; public class ItemOrientedStepTests extends TestCase { @@ -107,8 +107,7 @@ public class ItemOrientedStepTests extends TestCase { jobInstance = new JobInstance(new Long(0), new JobParameters(), new JobSupport("FOO")); - SimpleStreamManager streamManager = new SimpleStreamManager(transactionManager); - itemOrientedStep.setStreamManager(streamManager); + itemOrientedStep.setTransactionManager(transactionManager); } @@ -333,25 +332,20 @@ public class ItemOrientedStepTests extends TestCase { } public void testStreamManager() throws Exception { - itemOrientedStep.setItemReader(new AbstractItemReader() { + itemOrientedStep.setItemReader(new MockRestartableItemReader() { public Object read() throws Exception { return "foo"; } + public void update(ExecutionContext executionContext) { + // TODO Auto-generated method stub + executionContext.putString("foo", "bar"); + } }); JobExecution jobExecution = new JobExecution(jobInstance); StepExecution stepExecution = new StepExecution(itemOrientedStep, jobExecution); assertEquals(false, stepExecution.getExecutionContext().containsKey("foo")); - itemOrientedStep.setStreamManager(new SimpleStreamManager(new ResourcelessTransactionManager()) { - - public void update(ExecutionContext executionContext) { - // TODO Auto-generated method stub - executionContext.putString("foo", "bar"); - } - - }); - itemOrientedStep.execute(stepExecution); // At least once in that process the statistics service was asked for @@ -441,9 +435,8 @@ public class ItemOrientedStepTests extends TestCase { } }; itemOrientedStep.setItemReader(itemReader); - itemOrientedStep.setStreamManager(new SimpleStreamManager(transactionManager) { - public void rollback(TransactionStatus status) { - super.rollback(status); + itemOrientedStep.setTransactionManager(new ResourcelessTransactionManager() { + protected void doRollback(DefaultTransactionStatus status) throws TransactionException { // Simulate failure on rollback when stream resets throw new ResetFailedException("Bar"); } @@ -470,9 +463,8 @@ public class ItemOrientedStepTests extends TestCase { public void testStatusForCommitFailedException() throws Exception { - itemOrientedStep.setStreamManager(new SimpleStreamManager(transactionManager) { - public void commit(TransactionStatus status) { - super.commit(status); + itemOrientedStep.setTransactionManager(new ResourcelessTransactionManager() { + protected void doCommit(DefaultTransactionStatus status) throws TransactionException { // Simulate failure on rollback when stream resets throw new RuntimeException("Bar"); } @@ -531,7 +523,7 @@ public class ItemOrientedStepTests extends TestCase { public void testStatusForCloseFailedException() throws Exception { - itemOrientedStep.setStreamManager(new SimpleStreamManager(transactionManager) { + itemOrientedStep.setItemReader(new MockRestartableItemReader() { public void close(ExecutionContext executionContext) throws StreamException { super.close(executionContext); // Simulate failure on rollback when stream resets diff --git a/spring-batch-execution/src/test/resources/org/springframework/batch/execution/bootstrap/support/job.xml b/spring-batch-execution/src/test/resources/org/springframework/batch/execution/bootstrap/support/job.xml index 6158a414e..155587e8e 100644 --- a/spring-batch-execution/src/test/resources/org/springframework/batch/execution/bootstrap/support/job.xml +++ b/spring-batch-execution/src/test/resources/org/springframework/batch/execution/bootstrap/support/job.xml @@ -27,9 +27,6 @@ - - - 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 dbcb19c90..d7fcdb9a6 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 @@ -24,9 +24,6 @@ - - - diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/stream/SimpleStreamManager.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/stream/SimpleStreamManager.java index 1413c805d..cebbee542 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/stream/SimpleStreamManager.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/stream/SimpleStreamManager.java @@ -22,9 +22,6 @@ import java.util.List; import org.springframework.batch.item.ExecutionContext; import org.springframework.batch.item.ItemStream; import org.springframework.batch.item.exception.StreamException; -import org.springframework.transaction.PlatformTransactionManager; -import org.springframework.transaction.TransactionStatus; -import org.springframework.transaction.support.DefaultTransactionDefinition; /** * Simple {@link StreamManager} that tries to resolve conflicts between key @@ -33,20 +30,10 @@ import org.springframework.transaction.support.DefaultTransactionDefinition; * @author Dave Syer * */ -public class SimpleStreamManager implements StreamManager { +public class SimpleStreamManager implements ItemStream { private List streams = new ArrayList(); - private PlatformTransactionManager transactionManager; - - /** - * @param transactionManager a {@link PlatformTransactionManager} - */ - public SimpleStreamManager(PlatformTransactionManager transactionManager) { - this(); - this.transactionManager = transactionManager; - } - /** * */ @@ -54,13 +41,6 @@ public class SimpleStreamManager implements StreamManager { super(); } - /** - * Public setter for the {@link PlatformTransactionManager}. - * @param transactionManager the {@link PlatformTransactionManager} to set - */ - public void setTransactionManager(PlatformTransactionManager transactionManager) { - this.transactionManager = transactionManager; - } /** * Simple aggregate {@link ExecutionContext} provider for the contributions @@ -119,30 +99,4 @@ public class SimpleStreamManager implements StreamManager { } } - /** - * Delegate to the {@link PlatformTransactionManager} to create a new - * transaction. - * - * @see org.springframework.batch.item.stream.StreamManager#getTransaction() - */ - public TransactionStatus getTransaction() { - TransactionStatus transaction = transactionManager.getTransaction(new DefaultTransactionDefinition()); - return transaction; - } - - /* - * (non-Javadoc) - * @see org.springframework.batch.item.stream.StreamManager#commit(java.lang.Object) - */ - public void commit(TransactionStatus status) { - transactionManager.commit(status); - } - - /* - * (non-Javadoc) - * @see org.springframework.batch.item.stream.StreamManager#rollback(java.lang.Object) - */ - public void rollback(TransactionStatus status) { - transactionManager.rollback(status); - } } diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/stream/StreamManager.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/stream/StreamManager.java deleted file mode 100644 index 16f7ac872..000000000 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/stream/StreamManager.java +++ /dev/null @@ -1,78 +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.item.stream; - -import org.springframework.batch.item.ExecutionContext; -import org.springframework.batch.item.ItemStream; -import org.springframework.batch.item.exception.StreamException; -import org.springframework.transaction.TransactionStatus; - -/** - * Generalized stream management broadcast strategy. Clients register - * {@link ItemStream} instances under a well-known key, and then when they ask - * for {@link ItemStream} operations by that key, we call each one in turn that - * is registered under the key. - * - * @author Dave Syer - * - */ -public interface StreamManager { - - /** - * Register the {@link ItemStream} instance as one of possibly several that - * are associated with the given key. - * - * @param key the key under which to add the provider - * @param stream an {@link ItemStream} - */ - void register(ItemStream stream); - - /** - * Extract and aggregate the {@link ExecutionContext} from all streams under - * this key. - * - * @param key the key under which {@link ItemStream} instances might have - * been registered. - * @return {@link ExecutionContext} aggregating the contexts of all providers - * registered under this key, or empty otherwise. - */ - void update(ExecutionContext executionContext); - - /** - * If any resources are needed for the stream to operate they need to be - * destroyed here. - * - * @param key the key under which {@link ItemStream} instances might have - * been registered. - */ - void close(ExecutionContext executionContext) throws StreamException; - - /** - * If any resources are needed for the stream to operate they need to be - * opened here. - * - * @param key the key under which {@link ItemStream} instances might have - * been registered. - */ - void open(ExecutionContext executionContext) throws StreamException; - - TransactionStatus getTransaction(); - - void commit(TransactionStatus transaction); - - void rollback(TransactionStatus transaction); - -} diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/stream/SimpleStreamManagerTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/stream/SimpleStreamManagerTests.java index fd4eb0437..1b1b5995a 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/stream/SimpleStreamManagerTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/stream/SimpleStreamManagerTests.java @@ -22,9 +22,6 @@ import junit.framework.TestCase; import org.springframework.batch.item.ExecutionContext; import org.springframework.batch.item.exception.StreamException; -import org.springframework.batch.support.transaction.ResourcelessTransactionManager; -import org.springframework.transaction.TransactionException; -import org.springframework.transaction.TransactionStatus; /** * @author Dave Syer @@ -32,7 +29,7 @@ import org.springframework.transaction.TransactionStatus; */ public class SimpleStreamManagerTests extends TestCase { - private SimpleStreamManager manager = new SimpleStreamManager(new ResourcelessTransactionManager()); + private SimpleStreamManager manager = new SimpleStreamManager(); private List list = new ArrayList(); @@ -67,36 +64,6 @@ public class SimpleStreamManagerTests extends TestCase { assertEquals(1, list.size()); } - /** - * Test method for - * {@link org.springframework.batch.item.stream.SimpleStreamManager#SimpleStreamManager(org.springframework.transaction.PlatformTransactionManager)}. - */ - public void testSimpleStreamManagerPlatformTransactionManager() { - manager = new SimpleStreamManager(); - try { - manager.getTransaction(); - fail("Expected NullPointerException"); - } - catch (NullPointerException e) { - // expected; - } - } - - /** - * Test method for - * {@link org.springframework.batch.item.stream.SimpleStreamManager#setTransactionManager(org.springframework.transaction.PlatformTransactionManager)}. - */ - public void testSetTransactionManager() { - manager.setTransactionManager(new ResourcelessTransactionManager() { - protected Object doGetTransaction() throws TransactionException { - list.add("bar"); - return super.doGetTransaction(); - } - }); - manager.getTransaction(); - assertEquals("bar", list.get(0)); - } - /** * Test method for * {@link org.springframework.batch.item.stream.SimpleStreamManager#commit(org.springframework.transaction.TransactionStatus)}. @@ -141,34 +108,4 @@ public class SimpleStreamManagerTests extends TestCase { assertEquals(1, list.size()); } - /** - * Test method for - * {@link org.springframework.batch.item.stream.SimpleStreamManager#commit(org.springframework.transaction.TransactionStatus)}. - */ - public void testCommitWithoutMark() { - manager.register(new ItemStreamSupport() { - public void update(ExecutionContext executionContext) { - list.add("bar"); - } - }); - TransactionStatus status = manager.getTransaction(); - manager.commit(status); - assertEquals(0, list.size()); - } - - /** - * Test method for - * {@link org.springframework.batch.item.stream.SimpleStreamManager#rollback(org.springframework.transaction.TransactionStatus)}. - */ - public void testRollbackWithoutMark() { - manager.register( new ItemStreamSupport() { - public void update(ExecutionContext executionContext) { - list.add("bar"); - } - }); - TransactionStatus status = manager.getTransaction(); - manager.rollback(status); - assertEquals(0, list.size()); - } - }