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

BATCHADM-49: added JMS integration test

Browse Source
This commit is contained in:
David Syer
2010-04-10 15:18:59 +00:00
committed by Michael Minella
parent c1616f0bbb
commit 6ca0512753
15 changed files with 329 additions and 116 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkHandler.java
View File

@@ -18,6 +18,6 @@ package org.springframework.batch.integration.chunk;
public interface ChunkHandler<T> {
ChunkResponse handleChunk(ChunkRequest<T> chunk);
ChunkResponse handleChunk(ChunkRequest<T> chunk) throws Exception;
}

5
spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkMessageChannelItemWriter.java
View File

@@ -195,9 +195,12 @@ public class ChunkMessageChannelItemWriter<T> extends StepExecutionListenerSuppo
* @throws IllegalStateException if the result contains the wrong job
* instance id (maybe we are sharing a channel and we shouldn't be)
*/
private void getNextResult() {
private void getNextResult() throws AsynchronousFailureException {
ChunkResponse payload = (ChunkResponse) messagingGateway.receive();
if (payload != null) {
if (logger.isDebugEnabled()) {
logger.debug("Found result: "+payload);
}
Long jobInstanceId = payload.getJobId();
Assert.state(jobInstanceId != null, "Message did not contain job instance id.");
Assert.state(jobInstanceId.equals(localState.getJobId()), "Message contained wrong job instance id ["

70
spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkProcessorChunkHandler.java
View File

@@ -22,7 +22,6 @@ import org.springframework.batch.core.JobInterruptedException;
import org.springframework.batch.core.StepContribution;
import org.springframework.batch.core.step.item.Chunk;
import org.springframework.batch.core.step.item.ChunkProcessor;
import org.springframework.batch.core.step.item.FaultTolerantChunkProcessor;
import org.springframework.batch.core.step.skip.NonSkippableReadException;
import org.springframework.batch.core.step.skip.SkipLimitExceededException;
import org.springframework.batch.core.step.skip.SkipListenerFailedException;
@@ -63,23 +62,39 @@ public class ChunkProcessorChunkHandler<S> implements ChunkHandler<S>, Initializ
* @see ChunkHandler#handleChunk(ChunkRequest)
*/
@ServiceActivator
public ChunkResponse handleChunk(ChunkRequest<S> chunkRequest) {
public ChunkResponse handleChunk(ChunkRequest<S> chunkRequest) throws Exception {
logger.debug("Handling chunk: " + chunkRequest);
StepContribution stepContribution = chunkRequest.getStepContribution();
Throwable failure = null;
try {
process(chunkRequest, stepContribution);
}
catch (SkipLimitExceededException e) {
failure = e;
}
catch (NonSkippableReadException e) {
failure = e;
}
catch (SkipListenerFailedException e) {
failure = e;
}
catch (RetryException e) {
failure = e;
}
catch (JobInterruptedException e) {
failure = e;
}
catch (Exception e) {
logger.debug("Failed chunk", e);
return new ChunkResponse(false, chunkRequest.getJobId(), stepContribution, e.getClass().getName() + ": "
+ e.getMessage());
} catch (Throwable e) {
// The handler might throw an Error or other non-exception
logger.debug("Failed chunk with non-exception", e);
return new ChunkResponse(false, chunkRequest.getJobId(), stepContribution, e.getClass().getName() + ": "
+ e.getMessage());
// try again...
throw e;
}
if (failure != null) {
logger.debug("Failed chunk", failure);
return new ChunkResponse(false, chunkRequest.getJobId(), stepContribution, failure.getClass().getName()
+ ": " + failure.getMessage());
}
logger.debug("Completed chunk handling with " + stepContribution);
@@ -95,40 +110,7 @@ public class ChunkProcessorChunkHandler<S> implements ChunkHandler<S>, Initializ
private void process(ChunkRequest<S> chunkRequest, StepContribution stepContribution) throws Exception {
Chunk<S> chunk = new Chunk<S>(chunkRequest.getItems());
if (chunkProcessor instanceof FaultTolerantChunkProcessor<?, ?>) {
boolean processed = false;
while (!processed) {
try {
chunkProcessor.process(stepContribution, chunk);
processed = true;
}
catch (SkipLimitExceededException e) {
throw e;
}
catch (NonSkippableReadException e) {
throw e;
}
catch (SkipListenerFailedException e) {
throw e;
}
catch (RetryException e) {
throw e;
}
catch (JobInterruptedException e) {
throw e;
}
catch (Exception e) {
// try again...
}
}
}
else {
chunkProcessor.process(stepContribution, chunk);
}
chunkProcessor.process(stepContribution, chunk);
}
}

11
spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/RemoteChunkHandlerFactoryBean.java
View File

@@ -25,6 +25,7 @@ import org.springframework.batch.core.StepExecutionListener;
import org.springframework.batch.core.step.item.Chunk;
import org.springframework.batch.core.step.item.ChunkOrientedTasklet;
import org.springframework.batch.core.step.item.ChunkProcessor;
import org.springframework.batch.core.step.item.FaultTolerantChunkProcessor;
import org.springframework.batch.core.step.item.SimpleChunkProcessor;
import org.springframework.batch.core.step.tasklet.Tasklet;
import org.springframework.batch.core.step.tasklet.TaskletStep;
@@ -114,6 +115,7 @@ public class RemoteChunkHandlerFactoryBean<T> implements FactoryBean<ChunkHandle
}
ChunkProcessorChunkHandler<T> handler = new ChunkProcessorChunkHandler<T>();
setNonBuffering(chunkProcessor);
handler.setChunkProcessor(chunkProcessor);
// TODO: create step context for the processor in case it has scope="step" dependencies
handler.afterPropertiesSet();
@@ -122,6 +124,15 @@ public class RemoteChunkHandlerFactoryBean<T> implements FactoryBean<ChunkHandle
}
/**
* @param chunkProcessor
*/
private void setNonBuffering(ChunkProcessor<T> chunkProcessor) {
if (chunkProcessor instanceof FaultTolerantChunkProcessor<?, ?>) {
((FaultTolerantChunkProcessor<?, ?>)chunkProcessor).setBuffering(false);
}
}
/**
* @param tasklet
* @param chunkWriter