Spring/spring-batch
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

OPEN - issue BATCH-378: RepeatListener is confusing and too generic to use for 'intercepting' a step

Browse Source 
http://jira.springframework.org/browse/BATCH-378

Slim down StremManager
This commit is contained in:
dsyer
2008-02-28 10:05:08 +00:00
parent 425c270da3
commit 707e94b3c8
8 changed files with 38 additions and 273 deletions
Download Patch File Download Diff File Expand all files Collapse all files

58
spring-batch-execution/src/main/java/org/springframework/batch/execution/step/AbstractStep.java
View File

@@ -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");

22
spring-batch-execution/src/main/java/org/springframework/batch/execution/step/ItemOrientedStep.java
View File

@@ -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;

34
spring-batch-execution/src/test/java/org/springframework/batch/execution/step/ItemOrientedStepTests.java
View File

@@ -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

3
spring-batch-execution/src/test/resources/org/springframework/batch/execution/bootstrap/support/job.xml
View File

@@ -27,9 +27,6 @@
<property name="jobRepository" >
<bean class="org.springframework.batch.execution.step.support.JobRepositorySupport" />
</property>
<property name="streamManager">
<bean class="org.springframework.batch.item.stream.SimpleStreamManager" />
</property>
</bean>
</property>
</bean>

3
spring-batch-execution/src/test/resources/org/springframework/batch/execution/configuration/test-context.xml
View File

@@ -24,9 +24,6 @@
<property name="itemReader" ref="itemReader" />
<property name="itemWriter" ref="itemWriter" />
<property name="jobRepository" ref="jobRepository" />
<property name="streamManager">
<bean class="org.springframework.batch.item.stream.SimpleStreamManager" />
</property>
</bean>
</property>
</bean>

48
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/stream/SimpleStreamManager.java
View File

@@ -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);
}
}

78
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/stream/StreamManager.java
View File

@@ -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);
}

65
spring-batch-infrastructure/src/test/java/org/springframework/batch/item/stream/SimpleStreamManagerTests.java
View File

@@ -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());
}
}