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

RESOLVED - issue BATCH-1111: ChunkListener called before WriteListener

Browse Source
This commit is contained in:
dsyer
2009-03-02 17:44:17 +00:00
parent 56d8b35be8
commit b215feb9e8
5 changed files with 57 additions and 68 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
spring-batch-core/src/main/java/org/springframework/batch/core/step/AbstractStep.java
View File

@@ -57,7 +57,7 @@ public abstract class AbstractStep implements Step, InitializingBean, BeanNameAw
private boolean allowStartIfComplete = false;
private CompositeStepExecutionListener listener = new CompositeStepExecutionListener();
private CompositeStepExecutionListener stepExecutionListener = new CompositeStepExecutionListener();
private JobRepository jobRepository;
@@ -192,7 +192,8 @@ public abstract class AbstractStep implements Step, InitializingBean, BeanNameAw
try {
doExecute(stepExecution);
} catch (RepeatException e) {
}
catch (RepeatException e) {
throw e.getCause();
}
exitStatus = stepExecution.getExitStatus();
@@ -281,7 +282,7 @@ public abstract class AbstractStep implements Step, InitializingBean, BeanNameAw
* @param listener a {@link StepExecutionListener}
*/
public void registerStepExecutionListener(StepExecutionListener listener) {
this.listener.register(listener);
this.stepExecutionListener.register(listener);
}
/**
@@ -299,7 +300,7 @@ public abstract class AbstractStep implements Step, InitializingBean, BeanNameAw
* @return composite listener that delegates to all registered listeners.
*/
protected StepExecutionListener getCompositeListener() {
return listener;
return stepExecutionListener;
}
/**

48
spring-batch-core/src/main/java/org/springframework/batch/core/step/item/BatchListenerFactoryHelper.java
View File

@@ -18,14 +18,7 @@ package org.springframework.batch.core.step.item;
import java.util.ArrayList;
import java.util.List;
import org.springframework.batch.core.ChunkListener;
import org.springframework.batch.core.StepListener;
import org.springframework.batch.core.listener.CompositeChunkListener;
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.
@@ -35,47 +28,6 @@ import org.springframework.util.Assert;
*/
abstract class BatchListenerFactoryHelper {
/**
* @param chunkOperations
* @param listeners
*/
public static RepeatOperations addChunkListeners(RepeatOperations chunkOperations, StepListener[] listeners) {
final CompositeChunkListener multicaster = new CompositeChunkListener();
boolean hasChunkListener = false;
for (int i = 0; i < listeners.length; i++) {
StepListener listener = listeners[i];
if (listener instanceof ChunkListener) {
hasChunkListener = true;
multicaster.register((ChunkListener) listener);
}
}
if (hasChunkListener) {
Assert.state(chunkOperations instanceof RepeatTemplate,
"Chunk 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) chunkOperations;
stepTemplate.registerListener(new RepeatListenerSupport() {
public void open(RepeatContext context) {
multicaster.beforeChunk();
}
public void close(RepeatContext context) {
multicaster.afterChunk();
}
});
}
return chunkOperations;
}
public static <T> List<T> getListeners(StepListener[] listeners, Class<? super T> cls) {
List<T> list = new ArrayList<T>();
for (int i = 0; i < listeners.length; i++) {

10
spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleStepFactoryBean.java
View File

@@ -15,10 +15,6 @@
*/
package org.springframework.batch.core.step.item;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.batch.core.ChunkListener;
@@ -524,7 +520,6 @@ public class SimpleStepFactoryBean<T, S> implements FactoryBean, BeanNameAware {
*/
private void registerStepListeners(TaskletStep step, RepeatOperations chunkOperations) {
List<StepListener> chunkListeners = new ArrayList<StepListener>(Arrays.asList(getListeners()));
for (Object itemHandler : new Object[] { itemReader, itemWriter, itemProcessor }) {
if (itemHandler instanceof ItemStream) {
step.registerStream((ItemStream) itemHandler);
@@ -535,14 +530,15 @@ public class SimpleStepFactoryBean<T, S> implements FactoryBean, BeanNameAware {
step.registerStepExecutionListener((StepExecutionListener) listener);
}
if (listener instanceof ChunkListener) {
chunkListeners.add((StepListener) listener);
step.registerChunkListener((ChunkListener) listener);
}
}
}
BatchListenerFactoryHelper.addChunkListeners(chunkOperations, chunkListeners.toArray(new StepListener[] {}));
step.setStepExecutionListeners(BatchListenerFactoryHelper.getListeners(listeners, StepExecutionListener.class)
.toArray(new StepExecutionListener[] {}));
step.setChunkListeners(BatchListenerFactoryHelper.getListeners(listeners, ChunkListener.class).toArray(
new ChunkListener[] {}));
}
/**

27
spring-batch-core/src/main/java/org/springframework/batch/core/step/tasklet/TaskletStep.java
View File

@@ -20,10 +20,12 @@ import java.util.concurrent.Semaphore;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.batch.core.BatchStatus;
import org.springframework.batch.core.ChunkListener;
import org.springframework.batch.core.JobInterruptedException;
import org.springframework.batch.core.StepContribution;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.StepExecutionListener;
import org.springframework.batch.core.listener.CompositeChunkListener;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.scope.context.ChunkContext;
import org.springframework.batch.core.scope.context.StepContextRepeatCallback;
@@ -69,6 +71,8 @@ public class TaskletStep extends AbstractStep {
private RepeatOperations stepOperations = new RepeatTemplate();
private CompositeChunkListener chunkListener = new CompositeChunkListener();
// default to checking current thread for interruption.
private StepInterruptionPolicy interruptionPolicy = new ThreadStepInterruptionPolicy();
@@ -143,17 +147,23 @@ public class TaskletStep extends AbstractStep {
}
/**
* Register each of the objects as listeners. If the {@link ItemReader} or
* {@link ItemWriter} themselves implements this interface they will be
* registered automatically, but their injected dependencies will not be.
* This is a good way to get access to job parameters and execution context
* if the tasklet is parameterised.
* Register a chunk listener for callbacks at the appropriate stages in a
* step execution.
*
* @param listener a {@link ChunkListener}
*/
public void registerChunkListener(ChunkListener listener) {
this.chunkListener.register(listener);
}
/**
* Register each of the objects as listeners.
*
* @param listeners an array of listener objects of known types.
*/
public void setStepExecutionListeners(StepExecutionListener[] listeners) {
public void setChunkListeners(ChunkListener[] listeners) {
for (int i = 0; i < listeners.length; i++) {
registerStepExecutionListener(listeners[i]);
registerChunkListener(listeners[i]);
}
}
@@ -242,6 +252,8 @@ public class TaskletStep extends AbstractStep {
RepeatStatus result = RepeatStatus.CONTINUABLE;
TransactionStatus transaction = transactionManager.getTransaction(transactionAttribute);
chunkListener.beforeChunk();
boolean locked = false;
@@ -249,6 +261,7 @@ public class TaskletStep extends AbstractStep {
try {
result = tasklet.execute(contribution, chunkContext);
chunkListener.afterChunk();
}
finally {
// Apply the contribution to the step

31
spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SimpleStepFactoryBeanTests.java
View File

@@ -39,6 +39,7 @@ import org.springframework.batch.core.Step;
import org.springframework.batch.core.StepListener;
import org.springframework.batch.core.job.SimpleJob;
import org.springframework.batch.core.listener.ItemListenerSupport;
import org.springframework.batch.core.listener.StepListenerSupport;
import org.springframework.batch.core.repository.dao.MapExecutionContextDao;
import org.springframework.batch.core.repository.dao.MapJobExecutionDao;
import org.springframework.batch.core.repository.dao.MapJobInstanceDao;
@@ -217,21 +218,46 @@ public class SimpleStepFactoryBeanTests {
int commitInterval = 3;
SimpleStepFactoryBean<String, String> factory = getStepFactory(items);
class AssertingWriteListener extends StepListenerSupport<Object, Object> {
String trail = "";
@Override
public void beforeWrite(List<? extends Object> items) {
trail = trail + "2";
}
@Override
public void afterWrite(List<? extends Object> items) {
trail = trail + "3";
}
}
class CountingChunkListener implements ChunkListener {
int beforeCount = 0;
int afterCount = 0;
private AssertingWriteListener writeListener;
public CountingChunkListener(AssertingWriteListener writeListener) {
super();
this.writeListener = writeListener;
}
public void afterChunk() {
writeListener.trail = writeListener.trail + "4";
afterCount++;
}
public void beforeChunk() {
writeListener.trail = writeListener.trail + "1";
beforeCount++;
}
}
CountingChunkListener chunkListener = new CountingChunkListener();
factory.setListeners(new StepListener[] { chunkListener });
AssertingWriteListener writeListener = new AssertingWriteListener();
CountingChunkListener chunkListener = new CountingChunkListener(writeListener);
factory.setListeners(new StepListener[] { chunkListener, writeListener });
factory.setCommitInterval(commitInterval);
AbstractStep step = (AbstractStep) factory.getObject();
@@ -248,6 +274,7 @@ public class SimpleStepFactoryBeanTests {
int expectedListenerCallCount = (items.length / commitInterval) + 1;
assertEquals(expectedListenerCallCount, chunkListener.afterCount);
assertEquals(expectedListenerCallCount, chunkListener.beforeCount);
assertTrue("Llistener order not as expected: " + writeListener.trail, writeListener.trail.startsWith("1234"));
}
/**