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

Make the outputChannel and messagingTemplate mutually exclusive

Browse Source 
This commit changes the remote chunking and partitioning master step
builders to either accept an output channel or a messaging template
but not both.

See BATCH-2687
This commit is contained in:
Mahmoud Ben Hassine
2018-07-12 11:54:35 +02:00
parent 39676a45c4
commit d72cd67cad
5 changed files with 66 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files

44
spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/RemoteChunkingMasterStepBuilder.java
View File

@@ -33,8 +33,8 @@ import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.repeat.CompletionPolicy;
import org.springframework.batch.repeat.RepeatOperations;
import org.springframework.batch.repeat.exception.ExceptionHandler;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.core.MessagingTemplate;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.PollableChannel;
import org.springframework.retry.RetryPolicy;
import org.springframework.retry.backoff.BackOffPolicy;
@@ -47,12 +47,14 @@ import org.springframework.util.Assert;
* Builder for a master step in a remote chunking setup. This builder creates and
* sets a {@link ChunkMessageChannelItemWriter} on the master step.
*
* If no <code>messagingTemplate</code> is provided through
* <p>If no {@code messagingTemplate} is provided through
* {@link RemoteChunkingMasterStepBuilder#messagingTemplate(MessagingTemplate)},
* this builder will create one. The <code>outputChannel</code> set with
* {@link RemoteChunkingMasterStepBuilder#outputChannel(DirectChannel)} will be
* used as a default channel of the messaging template and will override any default
* channel already set on the messaging template.
* this builder will create one and set its default channel to the {@code outputChannel}
* provided through {@link RemoteChunkingMasterStepBuilder#outputChannel(MessageChannel)}.</p>
*
* <p>If a {@code messagingTemplate} is provided, it is assumed that it is fully configured
* and that its default channel is set to an output channel on which requests to workers
* will be sent.</p>
*
* @param <I> type of input items
* @param <O> type of output items
@@ -64,7 +66,7 @@ public class RemoteChunkingMasterStepBuilder<I, O> extends FaultTolerantStepBuil
private MessagingTemplate messagingTemplate;
private PollableChannel inputChannel;
private DirectChannel outputChannel;
private MessageChannel outputChannel;
private final int DEFAULT_MAX_WAIT_TIMEOUTS = 40;
private static final long DEFAULT_THROTTLE_LIMIT = 6;
@@ -97,17 +99,18 @@ public class RemoteChunkingMasterStepBuilder<I, O> extends FaultTolerantStepBuil
}
/**
* Set the output channel on which requests to workers will be sent.
* The output channel will be set as a default channel on the provided
* {@link MessagingTemplate} trough {@link RemoteChunkingMasterStepBuilder#messagingTemplate(MessagingTemplate)}
* (or the one created by this builder if no messaging template is provided).
* Set the output channel on which requests to workers will be sent. By using
* this setter, a default messaging template will be created and the output
* channel will be set as its default channel.
* <p>Use either this setter or {@link RemoteChunkingMasterStepBuilder#messagingTemplate(MessagingTemplate)}
* to provide a fully configured messaging template.</p>
*
* @param outputChannel the output channel.
* @return this builder instance for fluent chaining
*
* @see RemoteChunkingMasterStepBuilder#messagingTemplate(MessagingTemplate)
*/
public RemoteChunkingMasterStepBuilder<I, O> outputChannel(DirectChannel outputChannel) {
public RemoteChunkingMasterStepBuilder<I, O> outputChannel(MessageChannel outputChannel) {
Assert.notNull(outputChannel, "outputChannel must not be null");
this.outputChannel = outputChannel;
return this;
@@ -115,13 +118,14 @@ public class RemoteChunkingMasterStepBuilder<I, O> extends FaultTolerantStepBuil
/**
* Set the {@link MessagingTemplate} to use to send data to workers.
*
* <p><strong>The default destination of the messaging template will be
* overridden by the output channel provided through
* {@link RemoteChunkingMasterStepBuilder#outputChannel(DirectChannel)}</strong>.</p>
* <strong>The default channel of the messaging template must be set</strong>.
* <p>Use either this setter to provide a fully configured messaging template or
* provide an output channel through {@link RemoteChunkingMasterStepBuilder#outputChannel(MessageChannel)}
* and a default messaging template will be created.</p>
*
* @param messagingTemplate the messaging template to use
* @return this builder instance for fluent chaining
* @see RemoteChunkingMasterStepBuilder#outputChannel(MessageChannel)
*/
public RemoteChunkingMasterStepBuilder<I, O> messagingTemplate(MessagingTemplate messagingTemplate) {
Assert.notNull(messagingTemplate, "messagingTemplate must not be null");
@@ -164,13 +168,17 @@ public class RemoteChunkingMasterStepBuilder<I, O> extends FaultTolerantStepBuil
*/
public TaskletStep build() {
Assert.notNull(this.inputChannel, "An InputChannel must be provided");
Assert.notNull(this.outputChannel, "An OutputChannel must be provided");
Assert.state(this.outputChannel == null || this.messagingTemplate == null,
"You must specify either an outputChannel or a messagingTemplate but not both.");
// configure messaging template
if (this.messagingTemplate == null) {
this.messagingTemplate = new MessagingTemplate();
this.messagingTemplate.setDefaultChannel(this.outputChannel);
if (this.logger.isDebugEnabled()) {
this.logger.debug("No messagingTemplate was provided, using a default one");
}
}
this.messagingTemplate.setDefaultChannel(this.outputChannel);
// configure item writer
ChunkMessageChannelItemWriter<O> chunkMessageChannelItemWriter = new ChunkMessageChannelItemWriter<>();

10
spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/RemoteChunkingWorkerBuilder.java
View File

@@ -19,9 +19,9 @@ import org.springframework.batch.core.step.item.SimpleChunkProcessor;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.support.PassThroughItemProcessor;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.IntegrationFlows;
import org.springframework.messaging.MessageChannel;
import org.springframework.util.Assert;
/**
@@ -49,8 +49,8 @@ public class RemoteChunkingWorkerBuilder<I, O> {
private ItemProcessor<I, O> itemProcessor;
private ItemWriter<O> itemWriter;
private DirectChannel inputChannel;
private DirectChannel outputChannel;
private MessageChannel inputChannel;
private MessageChannel outputChannel;
/**
* Set the {@link ItemProcessor} to use to process items sent by the master
@@ -83,7 +83,7 @@ public class RemoteChunkingWorkerBuilder<I, O> {
* @param inputChannel the input channel
* @return this builder instance for fluent chaining
*/
public RemoteChunkingWorkerBuilder<I, O> inputChannel(DirectChannel inputChannel) {
public RemoteChunkingWorkerBuilder<I, O> inputChannel(MessageChannel inputChannel) {
Assert.notNull(inputChannel, "inputChannel must not be null");
this.inputChannel = inputChannel;
return this;
@@ -95,7 +95,7 @@ public class RemoteChunkingWorkerBuilder<I, O> {
* @param outputChannel the output channel
* @return this builder instance for fluent chaining
*/
public RemoteChunkingWorkerBuilder<I, O> outputChannel(DirectChannel outputChannel) {
public RemoteChunkingWorkerBuilder<I, O> outputChannel(MessageChannel outputChannel) {
Assert.notNull(outputChannel, "outputChannel must not be null");
this.outputChannel = outputChannel;
return this;

38
spring-batch-integration/src/main/java/org/springframework/batch/integration/partition/RemotePartitioningMasterStepBuilder.java
View File

@@ -44,10 +44,12 @@ import org.springframework.util.Assert;
*
* <p>If no {@code messagingTemplate} is provided through
* {@link RemotePartitioningMasterStepBuilder#messagingTemplate(MessagingTemplate)},
* this builder will create one. The {@code outputChannel} set with
* {@link RemotePartitioningMasterStepBuilder#outputChannel(MessageChannel)} will be
* used as a default channel of the messaging template and will <strong>override any default
* channel already set on the messaging template</strong>.
* this builder will create one and set its default channel to the {@code outputChannel}
* provided through {@link RemotePartitioningMasterStepBuilder#outputChannel(MessageChannel)}.</p>
*
* <p>If a {@code messagingTemplate} is provided, it is assumed that it is fully configured
* and that its default channel is set to an output channel on which requests to workers
* will be sent.</p>
*
* @since 4.1
* @author Mahmoud Ben Hassine
@@ -85,14 +87,14 @@ public class RemotePartitioningMasterStepBuilder extends PartitionStepBuilder {
}
/**
* Set the output channel on which requests to workers will be sent.
* The output channel will be set as a default channel on the provided
* {@link MessagingTemplate} trough
* {@link RemotePartitioningMasterStepBuilder#messagingTemplate(MessagingTemplate)}
* (or the one created by this builder if no messaging template is provided).
* Set the output channel on which requests to workers will be sent. By using
* this setter, a default messaging template will be created and the output
* channel will be set as its default channel.
* <p>Use either this setter or {@link RemotePartitioningMasterStepBuilder#messagingTemplate(MessagingTemplate)}
* to provide a fully configured messaging template.</p>
*
* @param outputChannel the output channel.
* @return this builder instance for fluent chaining
*
* @see RemotePartitioningMasterStepBuilder#messagingTemplate(MessagingTemplate)
*/
public RemotePartitioningMasterStepBuilder outputChannel(MessageChannel outputChannel) {
@@ -103,12 +105,14 @@ public class RemotePartitioningMasterStepBuilder extends PartitionStepBuilder {
/**
* Set the {@link MessagingTemplate} to use to send data to workers.
* <p><strong>The default destination of the messaging template will be
* overridden by the output channel provided through
* {@link RemotePartitioningMasterStepBuilder#outputChannel(MessageChannel)}</strong>.</p>
* <strong>The default channel of the messaging template must be set</strong>.
* <p>Use either this setter to provide a fully configured messaging template or
* provide an output channel through {@link RemotePartitioningMasterStepBuilder#outputChannel(MessageChannel)}
* and a default messaging template will be created.</p>
*
* @param messagingTemplate the messaging template to use
* @return this builder instance for fluent chaining
* @see RemotePartitioningMasterStepBuilder#outputChannel(MessageChannel)
*/
public RemotePartitioningMasterStepBuilder messagingTemplate(MessagingTemplate messagingTemplate) {
Assert.notNull(messagingTemplate, "messagingTemplate must not be null");
@@ -159,13 +163,17 @@ public class RemotePartitioningMasterStepBuilder extends PartitionStepBuilder {
}
public Step build() {
Assert.notNull(this.outputChannel, "An outputChannel must be provided");
Assert.state(this.outputChannel == null || this.messagingTemplate == null,
"You must specify either an outputChannel or a messagingTemplate but not both.");
// configure messaging template
if (this.messagingTemplate == null) {
this.messagingTemplate = new MessagingTemplate();
this.messagingTemplate.setDefaultChannel(this.outputChannel);
if (this.logger.isDebugEnabled()) {
this.logger.debug("No messagingTemplate was provided, using a default one");
}
}
this.messagingTemplate.setDefaultChannel(this.outputChannel);
// Configure the partition handler
final MessageChannelPartitionHandler partitionHandler = new MessageChannelPartitionHandler();