diff --git a/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/RemoteChunkingMasterStepBuilder.java b/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/RemoteChunkingMasterStepBuilder.java index a1d7cd921..cfd5b373b 100644 --- a/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/RemoteChunkingMasterStepBuilder.java +++ b/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/RemoteChunkingMasterStepBuilder.java @@ -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 messagingTemplate is provided through + *

If no {@code messagingTemplate} is provided through * {@link RemoteChunkingMasterStepBuilder#messagingTemplate(MessagingTemplate)}, - * this builder will create one. The outputChannel 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)}.

+ * + *

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.

* * @param type of input items * @param type of output items @@ -64,7 +66,7 @@ public class RemoteChunkingMasterStepBuilder 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 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. + *

Use either this setter or {@link RemoteChunkingMasterStepBuilder#messagingTemplate(MessagingTemplate)} + * to provide a fully configured messaging template.

* * @param outputChannel the output channel. * @return this builder instance for fluent chaining * * @see RemoteChunkingMasterStepBuilder#messagingTemplate(MessagingTemplate) */ - public RemoteChunkingMasterStepBuilder outputChannel(DirectChannel outputChannel) { + public RemoteChunkingMasterStepBuilder outputChannel(MessageChannel outputChannel) { Assert.notNull(outputChannel, "outputChannel must not be null"); this.outputChannel = outputChannel; return this; @@ -115,13 +118,14 @@ public class RemoteChunkingMasterStepBuilder extends FaultTolerantStepBuil /** * Set the {@link MessagingTemplate} to use to send data to workers. - * - *

The default destination of the messaging template will be - * overridden by the output channel provided through - * {@link RemoteChunkingMasterStepBuilder#outputChannel(DirectChannel)}.

+ * The default channel of the messaging template must be set. + *

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.

* * @param messagingTemplate the messaging template to use * @return this builder instance for fluent chaining + * @see RemoteChunkingMasterStepBuilder#outputChannel(MessageChannel) */ public RemoteChunkingMasterStepBuilder messagingTemplate(MessagingTemplate messagingTemplate) { Assert.notNull(messagingTemplate, "messagingTemplate must not be null"); @@ -164,13 +168,17 @@ public class RemoteChunkingMasterStepBuilder 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 chunkMessageChannelItemWriter = new ChunkMessageChannelItemWriter<>(); diff --git a/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/RemoteChunkingWorkerBuilder.java b/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/RemoteChunkingWorkerBuilder.java index c19e3f285..a30c4adad 100644 --- a/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/RemoteChunkingWorkerBuilder.java +++ b/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/RemoteChunkingWorkerBuilder.java @@ -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 { private ItemProcessor itemProcessor; private ItemWriter 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 { * @param inputChannel the input channel * @return this builder instance for fluent chaining */ - public RemoteChunkingWorkerBuilder inputChannel(DirectChannel inputChannel) { + public RemoteChunkingWorkerBuilder inputChannel(MessageChannel inputChannel) { Assert.notNull(inputChannel, "inputChannel must not be null"); this.inputChannel = inputChannel; return this; @@ -95,7 +95,7 @@ public class RemoteChunkingWorkerBuilder { * @param outputChannel the output channel * @return this builder instance for fluent chaining */ - public RemoteChunkingWorkerBuilder outputChannel(DirectChannel outputChannel) { + public RemoteChunkingWorkerBuilder outputChannel(MessageChannel outputChannel) { Assert.notNull(outputChannel, "outputChannel must not be null"); this.outputChannel = outputChannel; return this; diff --git a/spring-batch-integration/src/main/java/org/springframework/batch/integration/partition/RemotePartitioningMasterStepBuilder.java b/spring-batch-integration/src/main/java/org/springframework/batch/integration/partition/RemotePartitioningMasterStepBuilder.java index 84e323e19..481ca9c38 100644 --- a/spring-batch-integration/src/main/java/org/springframework/batch/integration/partition/RemotePartitioningMasterStepBuilder.java +++ b/spring-batch-integration/src/main/java/org/springframework/batch/integration/partition/RemotePartitioningMasterStepBuilder.java @@ -44,10 +44,12 @@ import org.springframework.util.Assert; * *

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 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 RemotePartitioningMasterStepBuilder#outputChannel(MessageChannel)}.

+ * + *

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.

* * @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. + *

Use either this setter or {@link RemotePartitioningMasterStepBuilder#messagingTemplate(MessagingTemplate)} + * to provide a fully configured messaging template.

+ * * @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. - *

The default destination of the messaging template will be - * overridden by the output channel provided through - * {@link RemotePartitioningMasterStepBuilder#outputChannel(MessageChannel)}.

+ * The default channel of the messaging template must be set. + *

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.

* * @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(); diff --git a/spring-batch-integration/src/test/java/org/springframework/batch/integration/chunk/RemoteChunkingMasterStepBuilderTest.java b/spring-batch-integration/src/test/java/org/springframework/batch/integration/chunk/RemoteChunkingMasterStepBuilderTest.java index 83ab092a7..89c03c93d 100644 --- a/spring-batch-integration/src/test/java/org/springframework/batch/integration/chunk/RemoteChunkingMasterStepBuilderTest.java +++ b/spring-batch-integration/src/test/java/org/springframework/batch/integration/chunk/RemoteChunkingMasterStepBuilderTest.java @@ -176,13 +176,15 @@ public class RemoteChunkingMasterStepBuilderTest { } @Test - public void testMandatoryOutputChannel() { + public void eitherOutputChannelOrMessagingTemplateMustBeProvided() { // given RemoteChunkingMasterStepBuilder builder = new RemoteChunkingMasterStepBuilder("step") - .inputChannel(this.inputChannel); + .inputChannel(this.inputChannel) + .outputChannel(new DirectChannel()) + .messagingTemplate(new MessagingTemplate()); - this.expectedException.expect(IllegalArgumentException.class); - this.expectedException.expectMessage("An OutputChannel must be provided"); + this.expectedException.expect(IllegalStateException.class); + this.expectedException.expectMessage("You must specify either an outputChannel or a messagingTemplate but not both."); // when TaskletStep step = builder.build(); diff --git a/spring-batch-integration/src/test/java/org/springframework/batch/integration/partition/RemotePartitioningMasterStepBuilderTests.java b/spring-batch-integration/src/test/java/org/springframework/batch/integration/partition/RemotePartitioningMasterStepBuilderTests.java index c104d0476..38b9ffad2 100644 --- a/spring-batch-integration/src/test/java/org/springframework/batch/integration/partition/RemotePartitioningMasterStepBuilderTests.java +++ b/spring-batch-integration/src/test/java/org/springframework/batch/integration/partition/RemotePartitioningMasterStepBuilderTests.java @@ -118,12 +118,14 @@ public class RemotePartitioningMasterStepBuilderTests { } @Test - public void testMandatoryOutputChannel() { + public void eitherOutputChannelOrMessagingTemplateMustBeProvided() { // given - RemotePartitioningMasterStepBuilder builder = new RemotePartitioningMasterStepBuilder("step"); + RemotePartitioningMasterStepBuilder builder = new RemotePartitioningMasterStepBuilder("step") + .outputChannel(new DirectChannel()) + .messagingTemplate(new MessagingTemplate()); - this.expectedException.expect(IllegalArgumentException.class); - this.expectedException.expectMessage("An outputChannel must be provided"); + this.expectedException.expect(IllegalStateException.class); + this.expectedException.expectMessage("You must specify either an outputChannel or a messagingTemplate but not both."); // when Step step = builder.build();