org.springframework.batch.integration.partition

Class MessageChannelPartitionHandler

java.lang.Object

org.springframework.batch.integration.partition.MessageChannelPartitionHandler

All Implemented Interfaces:

org.springframework.batch.core.partition.PartitionHandler

@MessageEndpoint
public class MessageChannelPartitionHandler
extends Object
implements org.springframework.batch.core.partition.PartitionHandler

A PartitionHandler that uses MessageChannel instances to send instructions to remote workers and receive their responses. The MessageChannel provides a nice abstraction so that the location of the workers and the transport used to communicate with them can be changed at run time. The communication with the remote workers does not need to be transactional or have guaranteed delivery, so a local thread pool based implementation works as well as a remote web service or JMS implementation. If a remote worker fails or doesn't send a reply message, the job will fail and can be restarted to pick up missing messages and processing. The remote workers need access to the Spring Batch JobRepository so that the shared state across those restarts can be managed centrally.

Author:

Dave Syer, Will Schipp

Constructor Summary

Constructors

Constructor and Description
MessageChannelPartitionHandler()

Method Summary

Modifier and Type	Method and Description
void	afterPropertiesSet()
List<?>	aggregate(List<?> messages)
Collection<org.springframework.batch.core.StepExecution>	handle(org.springframework.batch.core.partition.StepExecutionSplitter stepExecutionSplitter, org.springframework.batch.core.StepExecution masterStepExecution) Sends StepExecutionRequest objects to the request channel of the MessagingOperations, and then receives the result back as a list of StepExecution on a reply channel.
void	setGridSize(int gridSize) Passed to the StepExecutionSplitter in the handle(StepExecutionSplitter, StepExecution) method, instructing it how many StepExecution instances are required, ideally.
void	setMessagingOperations(org.springframework.integration.core.MessagingOperations messagingGateway) A pre-configured gateway for sending and receiving messages to the remote workers.
void	setReplyChannel(org.springframework.integration.core.PollableChannel replyChannel)
void	setStepName(String stepName) The name of the Step that will be used to execute the partitioned StepExecution.

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Constructor Detail

MessageChannelPartitionHandler

public MessageChannelPartitionHandler()

Method Detail

afterPropertiesSet

public void afterPropertiesSet()
                        throws Exception

Throws:

Exception

setMessagingOperations

public void setMessagingOperations(org.springframework.integration.core.MessagingOperations messagingGateway)

A pre-configured gateway for sending and receiving messages to the remote workers. Using this property allows a large degree of control over the timeouts and other properties of the send. It should have channels set up internally:

• request channel capable of accepting StepExecutionRequest payloads
• reply channel that returns a list of StepExecution results

The timeout for the repoy should be set sufficiently long that the remote steps have time to complete.

Parameters:

messagingGateway - the MessagingOperations to set

setGridSize

public void setGridSize(int gridSize)

Passed to the StepExecutionSplitter in the handle(StepExecutionSplitter, StepExecution) method, instructing it how many StepExecution instances are required, ideally. The StepExecutionSplitter is allowed to ignore the grid size in the case of a restart, since the input data partitions must be preserved.

Parameters:

gridSize - the number of step executions that will be created

setStepName

public void setStepName(String stepName)

The name of the Step that will be used to execute the partitioned StepExecution. This is a regular Spring Batch step, with all the business logic required to complete an execution based on the input parameters in its StepExecution context. The name will be translated into a Step instance by the remote worker.

Parameters:

stepName - the name of the Step instance to execute business logic

aggregate

@Aggregator(sendPartialResultsOnExpiry=true)
public List<?> aggregate(@Payloads
                                                                      List<?> messages)

Parameters:

messages - the messages to be aggregated

Returns:

the list as it was passed in

setReplyChannel

public void setReplyChannel(org.springframework.integration.core.PollableChannel replyChannel)

handle

public Collection<org.springframework.batch.core.StepExecution> handle(org.springframework.batch.core.partition.StepExecutionSplitter stepExecutionSplitter,
                                                                       org.springframework.batch.core.StepExecution masterStepExecution)
                                                                throws Exception

Sends StepExecutionRequest objects to the request channel of the MessagingOperations, and then receives the result back as a list of StepExecution on a reply channel. Use the aggregate(List) method as an aggregator of the individual remote replies. The receive timeout needs to be set realistically in the MessagingOperations and the aggregator, so that there is a good chance of all work being done.

Specified by:

handle in interface org.springframework.batch.core.partition.PartitionHandler

Throws:

Exception

See Also:

PartitionHandler.handle(StepExecutionSplitter, StepExecution)