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

BATCHADM-58: first cut with all features and SI 2.0

Browse Source
This commit is contained in:
David Syer
2010-08-11 12:46:40 +00:00
committed by Michael Minella
parent 19b8263623
commit bced2c5595
20 changed files with 159 additions and 123 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -34,7 +34,10 @@ import org.springframework.batch.item.ExecutionContext;
import org.springframework.batch.item.ItemStream;
import org.springframework.batch.item.ItemStreamException;
import org.springframework.batch.item.ItemWriter;
import org.springframework.integration.gateway.MessagingGateway;
import org.springframework.integration.Message;
import org.springframework.integration.core.GenericMessage;
import org.springframework.integration.core.MessagingOperations;
import org.springframework.integration.core.PollableChannel;
import org.springframework.util.Assert;
public class ChunkMessageChannelItemWriter<T> extends StepExecutionListenerSupport implements ItemWriter<T>,
@@ -48,7 +51,7 @@ public class ChunkMessageChannelItemWriter<T> extends StepExecutionListenerSuppo
private static final long DEFAULT_THROTTLE_LIMIT = 6;
private MessagingGateway messagingGateway;
private MessagingOperations messagingGateway;
private LocalState localState = new LocalState();
@@ -58,6 +61,8 @@ public class ChunkMessageChannelItemWriter<T> extends StepExecutionListenerSuppo
private int maxWaitTimeouts = DEFAULT_MAX_WAIT_TIMEOUTS;
private PollableChannel replyChannel;
/**
* The maximum number of times to wait at the end of a step for a non-null result from the remote workers. This is a
* multiplier on the receive timeout set separately on the gateway. The ideal value is a compromise between allowing
@@ -78,10 +83,14 @@ public class ChunkMessageChannelItemWriter<T> extends StepExecutionListenerSuppo
this.throttleLimit = throttleLimit;
}
public void setMessagingGateway(MessagingGateway messagingGateway) {
public void setMessagingOperations(MessagingOperations messagingGateway) {
this.messagingGateway = messagingGateway;
}
public void setReplyChannel(PollableChannel replyChannel) {
this.replyChannel = replyChannel;
}
public void write(List<? extends T> items) throws Exception {
// Block until expecting <= throttle limit
@@ -94,7 +103,7 @@ public class ChunkMessageChannelItemWriter<T> extends StepExecutionListenerSuppo
logger.debug("Dispatching chunk: " + items);
ChunkRequest<T> request = new ChunkRequest<T>(items, localState.getJobId(), localState
.createStepContribution());
messagingGateway.send(request);
messagingGateway.send(new GenericMessage<ChunkRequest<T>>(request));
localState.incrementExpected();
}
@@ -192,8 +201,9 @@ public class ChunkMessageChannelItemWriter<T> extends StepExecutionListenerSuppo
* and we shouldn't be)
*/
private void getNextResult() throws AsynchronousFailureException {
ChunkResponse payload = (ChunkResponse) messagingGateway.receive();
if (payload != null) {
Message<ChunkResponse> message = messagingGateway.receive(replyChannel);
if (message != null) {
ChunkResponse payload = message.getPayload();
if (logger.isDebugEnabled()) {
logger.debug("Found result: " + payload);
}

3
spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/JmsRedeliveredExtractor.java
View File

@@ -18,6 +18,7 @@ package org.springframework.batch.integration.chunk;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.integration.annotation.Header;
import org.springframework.integration.jms.JmsHeaders;
/**
* @author Dave Syer
@@ -27,7 +28,7 @@ public class JmsRedeliveredExtractor {
private static final Log logger = LogFactory.getLog(JmsRedeliveredExtractor.class);
public ChunkResponse extract(ChunkResponse input, @Header("springintegration_jms_redelivered") boolean redelivered) {
public ChunkResponse extract(ChunkResponse input, @Header(JmsHeaders.REDELIVERED) boolean redelivered) {
logger.debug("Extracted redelivered flag for response, value="+redelivered);
return new ChunkResponse(input, redelivered);
}

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

@@ -3,13 +3,14 @@ package org.springframework.batch.integration.chunk;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.integration.Message;
import org.springframework.integration.channel.ChannelInterceptor;
import org.springframework.integration.channel.interceptor.ChannelInterceptorAdapter;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.message.MessageSource;
import org.springframework.integration.core.MessageSource;
import org.springframework.util.Assert;
/**
* A {@link ChannelInterceptor} that turns a pollable channel into a "pass-thru channel": if a client calls
* <code>receive()</code> on the channel it will delegate to a {@link MessageSource} to pull the message directly from

31
spring-batch-integration/src/main/java/org/springframework/batch/integration/partition/MessageChannelPartitionHandler.java
View File

@@ -9,12 +9,13 @@ import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.partition.PartitionHandler;
import org.springframework.batch.core.partition.StepExecutionSplitter;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.integration.Message;
import org.springframework.integration.annotation.Aggregator;
import org.springframework.integration.annotation.MessageEndpoint;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessageBuilder;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.gateway.MessagingGateway;
import org.springframework.integration.message.MessageBuilder;
import org.springframework.integration.core.MessagingOperations;
import org.springframework.integration.core.PollableChannel;
import org.springframework.util.Assert;
/**
@@ -38,13 +39,15 @@ public class MessageChannelPartitionHandler implements PartitionHandler {
private int gridSize = 1;
private MessagingGateway messagingGateway;
private MessagingOperations messagingGateway;
private String stepName;
private PollableChannel replyChannel;
public void afterPropertiesSet() throws Exception {
Assert.notNull(stepName, "A step name must be provided for the remote workers.");
Assert.state(messagingGateway != null, "The MessagingGateway must be set");
Assert.state(messagingGateway != null, "The MessagingOperations must be set");
}
/**
@@ -60,12 +63,16 @@ public class MessageChannelPartitionHandler implements PartitionHandler {
* The timeout for the repoy should be set sufficiently long that the remote
* steps have time to complete.
*
* @param messagingGateway the {@link MessagingGateway} to set
* @param messagingGateway the {@link MessagingOperations} to set
*/
public void setMessagingGateway(MessagingGateway messagingGateway) {
public void setMessagingOperations(MessagingOperations messagingGateway) {
this.messagingGateway = messagingGateway;
}
public void setReplyChannel(PollableChannel replyChannel) {
this.replyChannel = replyChannel;
}
/**
* Passed to the {@link StepExecutionSplitter} in the
* {@link #handle(StepExecutionSplitter, StepExecution)} method, instructing
@@ -97,18 +104,18 @@ public class MessageChannelPartitionHandler implements PartitionHandler {
* @param messages the messages to be aggregated
* @return the list as it was passed in
*/
@Aggregator(sendPartialResultsOnTimeout = true)
@Aggregator(sendPartialResultsOnExpiry = true)
public List<?> aggregate(List<?> messages) {
return messages;
}
/**
* Sends {@link StepExecutionRequest} objects to the request channel of the
* {@link MessagingGateway}, and then receives the result back as a list of
* {@link MessagingOperations}, and then receives the result back as a list of
* {@link StepExecution} on a reply channel. Use the
* {@link #aggregate(List)} method as an aggregator of the individual remote
* replies. The receive timeout needs to be set realistically in the
* {@link MessagingGateway} <b>and</b> the aggregator, so that there is a
* {@link MessagingOperations} <b>and</b> the aggregator, so that there is a
* good chance of all work being done.
*
* @see PartitionHandler#handle(StepExecutionSplitter, StepExecution)
@@ -123,8 +130,8 @@ public class MessageChannelPartitionHandler implements PartitionHandler {
.getJobExecutionId(), stepExecution.getId())));
}
@SuppressWarnings("unchecked")
Collection<StepExecution> result = (Collection<StepExecution>) messagingGateway.receive();
Message<Collection<StepExecution>> message = messagingGateway.receive(replyChannel);
Collection<StepExecution> result = message.getPayload();
return result;
}