Add builders for master/worker beans in remote chunking setup
This commit adds a new annotation `@EnableBatchIntegration` that makes it possible to autowire two beans in the application context: * RemoteChunkingMasterStepBuilderFactory: used to create a master step * RemoteChunkingWorkerBuilder: used to create a worker integration flow The goal of these new APIs is to simplify the setup of remote chunking job by automatically configuring infrastructure beans. Resolves BATCH-2686
This commit is contained in:
Mahmoud Ben Hassine
committed by
Michael Minella
Michael Minella
parent
977741c294
commit
ee56a2320d
BIN
spring-batch-docs/asciidoc/images/remote-chunking-config.png
Normal file
BIN
spring-batch-docs/asciidoc/images/remote-chunking-config.png
Normal file
Binary file not shown.
|
After Width: | Height: | Size: 346 KiB |
@@ -771,7 +771,6 @@ public org.apache.activemq.ActiveMQConnectionFactory connectionFactory() {
|
||||
/*
|
||||
* Configure outbound flow (requests going to workers)
|
||||
*/
|
||||
|
||||
@Bean
|
||||
public DirectChannel requests() {
|
||||
return new DirectChannel();
|
||||
@@ -788,7 +787,6 @@ public IntegrationFlow outboundFlow(ActiveMQConnectionFactory connectionFactory)
|
||||
/*
|
||||
* Configure inbound flow (replies coming from workers)
|
||||
*/
|
||||
|
||||
@Bean
|
||||
public QueueChannel replies() {
|
||||
return new QueueChannel();
|
||||
@@ -805,7 +803,6 @@ public IntegrationFlow inboundFlow(ActiveMQConnectionFactory connectionFactory)
|
||||
/*
|
||||
* Configure the ChunkMessageChannelItemWriter
|
||||
*/
|
||||
|
||||
@Bean
|
||||
public ItemWriter<Integer> itemWriter() {
|
||||
MessagingTemplate messagingTemplate = new MessagingTemplate();
|
||||
@@ -883,7 +880,6 @@ public org.apache.activemq.ActiveMQConnectionFactory connectionFactory() {
|
||||
/*
|
||||
* Configure inbound flow (requests coming from the master)
|
||||
*/
|
||||
|
||||
@Bean
|
||||
public DirectChannel requests() {
|
||||
return new DirectChannel();
|
||||
@@ -900,7 +896,6 @@ public IntegrationFlow inboundFlow(ActiveMQConnectionFactory connectionFactory)
|
||||
/*
|
||||
* Configure outbound flow (replies going to the master)
|
||||
*/
|
||||
|
||||
@Bean
|
||||
public DirectChannel replies() {
|
||||
return new DirectChannel();
|
||||
@@ -917,7 +912,6 @@ public IntegrationFlow outboundFlow(ActiveMQConnectionFactory connectionFactory)
|
||||
/*
|
||||
* Configure the ChunkProcessorChunkHandler
|
||||
*/
|
||||
|
||||
@Bean
|
||||
@ServiceActivator(inputChannel = "requests", outputChannel = "replies")
|
||||
public ChunkProcessorChunkHandler<Integer> chunkProcessorChunkHandler() {
|
||||
@@ -944,6 +938,88 @@ when it receives chunks from the master.
|
||||
For more information, see the section of the "Scalability" chapter on
|
||||
link:$$http://docs.spring.io/spring-batch/reference/html/scalability.html#remoteChunking$$[Remote Chunking].
|
||||
|
||||
Starting from version 4.1, Spring Batch Integration introduces the `@EnableBatchIntegration`
|
||||
annotation that can be used to simplify remote chunking setup. This annotation provides
|
||||
two beans that can be autowired in the application context:
|
||||
|
||||
* `RemoteChunkingMasterStepBuilderFactory`: used to configure the master step
|
||||
* `RemoteChunkingWorkerBuilder`: used to configure the remote worker integration flow
|
||||
|
||||
These APIs will take care of configuring a number of components as described in the following diagram:
|
||||
|
||||
.Remote Chunking Configuration
|
||||
image::{batch-asciidoc}images/remote-chunking-config.png[Remote Chunking Configuration, scaledwidth="80%"]
|
||||
|
||||
On the master side, the `RemoteChunkingMasterStepBuilderFactory` allows you to
|
||||
configure a master step by declaring:
|
||||
|
||||
* the item reader to read items and send them to workers
|
||||
* the output channel ("Outgoing requests") to send requests to workers
|
||||
* the input channel ("Incoming replies") to receive replies from workers
|
||||
|
||||
There is no need anymore to explicitly configure the `ChunkMessageChannelItemWriter`
|
||||
and the `MessagingTemplate` (Those can still be explicitly configured if required).
|
||||
|
||||
On the worker side, the `RemoteChunkingWorkerBuilder` allows you to configure a worker to:
|
||||
|
||||
* listen to requests sent by the master on the input channel ("Incoming requests")
|
||||
* call the `handleChunk` method of `ChunkProcessorChunkHandler` for each request
|
||||
with the configured `ItemProcessor` and `ItemWriter`
|
||||
* send replies on the output channel ("Outgoing replies") to the master
|
||||
|
||||
There is no need anymore to explicitly configure the `SimpleChunkProcessor`
|
||||
and the `ChunkProcessorChunkHandler` (Those can still be explicitly configured if required).
|
||||
|
||||
The following example shows how to use these APIs:
|
||||
|
||||
[source, java]
|
||||
----
|
||||
@EnableBatchIntegration
|
||||
public class RemoteChunkingJobConfiguration {
|
||||
|
||||
@Configuration
|
||||
public static class MasterConfiguration {
|
||||
|
||||
@Autowired
|
||||
private RemoteChunkingMasterStepBuilderFactory masterStepBuilderFactory;
|
||||
|
||||
@Bean
|
||||
public TaskletStep masterStep() {
|
||||
return this.masterStepBuilderFactory.get("masterStep")
|
||||
.chunk(100)
|
||||
.reader(itemReader())
|
||||
.outputChannel(requests()) // requests sent to workers
|
||||
.inputChannel(replies()) // replies received from workers
|
||||
.build();
|
||||
}
|
||||
|
||||
// Middleware beans setup omitted
|
||||
|
||||
}
|
||||
|
||||
@Configuration
|
||||
public static class WorkerConfiguration {
|
||||
|
||||
@Autowired
|
||||
private RemoteChunkingWorkerBuilder workerBuilder;
|
||||
|
||||
@Bean
|
||||
public IntegrationFlow workerFlow() {
|
||||
return this.workerBuilder
|
||||
.itemProcessor(itemProcessor())
|
||||
.itemWriter(itemWriter())
|
||||
.inputChannel(requests()) // requests received from the master
|
||||
.outputChannel(replies()) // replies sent to the master
|
||||
.build();
|
||||
}
|
||||
|
||||
// Middleware beans setup omitted
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
----
|
||||
|
||||
You can find a complete example of a remote chunking job
|
||||
link:$$https://github.com/spring-projects/spring-batch/tree/master/spring-batch-samples#remote-chunking-sample$$[here].
|
||||
|
||||
|
||||
@@ -0,0 +1,403 @@
|
||||
/*
|
||||
* Copyright 2018 the original author or authors.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
package org.springframework.batch.integration.chunk;
|
||||
|
||||
import org.springframework.batch.core.ChunkListener;
|
||||
import org.springframework.batch.core.ItemReadListener;
|
||||
import org.springframework.batch.core.ItemWriteListener;
|
||||
import org.springframework.batch.core.SkipListener;
|
||||
import org.springframework.batch.core.StepExecutionListener;
|
||||
import org.springframework.batch.core.repository.JobRepository;
|
||||
import org.springframework.batch.core.step.builder.FaultTolerantStepBuilder;
|
||||
import org.springframework.batch.core.step.builder.StepBuilder;
|
||||
import org.springframework.batch.core.step.item.KeyGenerator;
|
||||
import org.springframework.batch.core.step.skip.SkipPolicy;
|
||||
import org.springframework.batch.core.step.tasklet.TaskletStep;
|
||||
import org.springframework.batch.item.ItemReader;
|
||||
import org.springframework.batch.item.ItemStream;
|
||||
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.PollableChannel;
|
||||
import org.springframework.retry.RetryPolicy;
|
||||
import org.springframework.retry.backoff.BackOffPolicy;
|
||||
import org.springframework.retry.policy.RetryContextCache;
|
||||
import org.springframework.transaction.PlatformTransactionManager;
|
||||
import org.springframework.transaction.interceptor.TransactionAttribute;
|
||||
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
|
||||
* {@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.
|
||||
*
|
||||
* @param <I> type of input items
|
||||
* @param <O> type of output items
|
||||
*
|
||||
* @since 4.1
|
||||
* @author Mahmoud Ben Hassine
|
||||
*/
|
||||
public class RemoteChunkingMasterStepBuilder<I, O> extends FaultTolerantStepBuilder<I, O> {
|
||||
|
||||
private MessagingTemplate messagingTemplate;
|
||||
private PollableChannel inputChannel;
|
||||
private DirectChannel outputChannel;
|
||||
|
||||
private final int DEFAULT_MAX_WAIT_TIMEOUTS = 40;
|
||||
private static final long DEFAULT_THROTTLE_LIMIT = 6;
|
||||
private int maxWaitTimeouts = DEFAULT_MAX_WAIT_TIMEOUTS;
|
||||
private long throttleLimit = DEFAULT_THROTTLE_LIMIT;
|
||||
|
||||
/**
|
||||
* Create a new {@link RemoteChunkingMasterStepBuilder}.
|
||||
*
|
||||
* @param stepName name of the master step
|
||||
*/
|
||||
public RemoteChunkingMasterStepBuilder(String stepName) {
|
||||
super(new StepBuilder(stepName));
|
||||
}
|
||||
|
||||
/**
|
||||
* Set the input channel on which replies from workers will be received.
|
||||
* The provided input channel will be set as a reply channel on the
|
||||
* {@link ChunkMessageChannelItemWriter} created by this builder.
|
||||
*
|
||||
* @param inputChannel the input channel
|
||||
* @return this builder instance for fluent chaining
|
||||
*
|
||||
* @see ChunkMessageChannelItemWriter#setReplyChannel
|
||||
*/
|
||||
public RemoteChunkingMasterStepBuilder<I, O> inputChannel(PollableChannel inputChannel) {
|
||||
Assert.notNull(inputChannel, "inputChannel must not be null");
|
||||
this.inputChannel = inputChannel;
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* 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).
|
||||
*
|
||||
* @param outputChannel the output channel.
|
||||
* @return this builder instance for fluent chaining
|
||||
*
|
||||
* @see RemoteChunkingMasterStepBuilder#messagingTemplate(MessagingTemplate)
|
||||
*/
|
||||
public RemoteChunkingMasterStepBuilder<I, O> outputChannel(DirectChannel outputChannel) {
|
||||
Assert.notNull(outputChannel, "outputChannel must not be null");
|
||||
this.outputChannel = outputChannel;
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* 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>
|
||||
*
|
||||
* @param messagingTemplate the messaging template to use
|
||||
* @return this builder instance for fluent chaining
|
||||
*/
|
||||
public RemoteChunkingMasterStepBuilder<I, O> messagingTemplate(MessagingTemplate messagingTemplate) {
|
||||
Assert.notNull(messagingTemplate, "messagingTemplate must not be null");
|
||||
this.messagingTemplate = messagingTemplate;
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* 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
|
||||
* slow workers time to finish, and responsiveness if there is a dead worker. Defaults to 40.
|
||||
*
|
||||
* @param maxWaitTimeouts the maximum number of wait timeouts
|
||||
* @see ChunkMessageChannelItemWriter#setMaxWaitTimeouts(int)
|
||||
*/
|
||||
public RemoteChunkingMasterStepBuilder<I, O> maxWaitTimeouts(int maxWaitTimeouts) {
|
||||
Assert.isTrue(maxWaitTimeouts > 0, "maxWaitTimeouts must be greater than zero");
|
||||
this.maxWaitTimeouts = maxWaitTimeouts;
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* Public setter for the throttle limit. This limits the number of pending requests for chunk processing to avoid
|
||||
* overwhelming the receivers.
|
||||
*
|
||||
* @param throttleLimit the throttle limit to set
|
||||
* @see ChunkMessageChannelItemWriter#setThrottleLimit(long)
|
||||
*/
|
||||
public RemoteChunkingMasterStepBuilder<I, O> throttleLimit(long throttleLimit) {
|
||||
Assert.isTrue(throttleLimit > 0, "throttleLimit must be greater than zero");
|
||||
this.throttleLimit = throttleLimit;
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* Build a master {@link TaskletStep}.
|
||||
*
|
||||
* @return the configured master step
|
||||
* @see RemoteChunkHandlerFactoryBean
|
||||
*/
|
||||
public TaskletStep build() {
|
||||
Assert.notNull(this.inputChannel, "An InputChannel must be provided");
|
||||
Assert.notNull(this.outputChannel, "An OutputChannel must be provided");
|
||||
|
||||
// configure messaging template
|
||||
if (this.messagingTemplate == null) {
|
||||
this.messagingTemplate = new MessagingTemplate();
|
||||
}
|
||||
this.messagingTemplate.setDefaultChannel(this.outputChannel);
|
||||
|
||||
// configure item writer
|
||||
ChunkMessageChannelItemWriter<O> chunkMessageChannelItemWriter = new ChunkMessageChannelItemWriter<>();
|
||||
chunkMessageChannelItemWriter.setMessagingOperations(this.messagingTemplate);
|
||||
chunkMessageChannelItemWriter.setMaxWaitTimeouts(this.maxWaitTimeouts);
|
||||
chunkMessageChannelItemWriter.setThrottleLimit(this.throttleLimit);
|
||||
chunkMessageChannelItemWriter.setReplyChannel(this.inputChannel);
|
||||
super.writer(chunkMessageChannelItemWriter);
|
||||
|
||||
return super.build();
|
||||
}
|
||||
|
||||
/*
|
||||
* The following methods override those from parent builders and return
|
||||
* the current builder type.
|
||||
* FIXME: Change parent builders to be generic and return current builder
|
||||
* type in each method.
|
||||
*/
|
||||
|
||||
@Override
|
||||
public RemoteChunkingMasterStepBuilder<I, O> reader(ItemReader<? extends I> reader) {
|
||||
super.reader(reader);
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public RemoteChunkingMasterStepBuilder<I, O> repository(JobRepository jobRepository) {
|
||||
super.repository(jobRepository);
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public RemoteChunkingMasterStepBuilder<I, O> transactionManager(PlatformTransactionManager transactionManager) {
|
||||
super.transactionManager(transactionManager);
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public RemoteChunkingMasterStepBuilder<I, O> listener(Object listener) {
|
||||
super.listener(listener);
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public RemoteChunkingMasterStepBuilder<I, O> listener(SkipListener<? super I, ? super O> listener) {
|
||||
super.listener(listener);
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public RemoteChunkingMasterStepBuilder<I, O> listener(ChunkListener listener) {
|
||||
super.listener(listener);
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public RemoteChunkingMasterStepBuilder<I, O> transactionAttribute(TransactionAttribute transactionAttribute) {
|
||||
super.transactionAttribute(transactionAttribute);
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public RemoteChunkingMasterStepBuilder<I, O> listener(org.springframework.retry.RetryListener listener) {
|
||||
super.listener(listener);
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public RemoteChunkingMasterStepBuilder<I, O> keyGenerator(KeyGenerator keyGenerator) {
|
||||
super.keyGenerator(keyGenerator);
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public RemoteChunkingMasterStepBuilder<I, O> retryLimit(int retryLimit) {
|
||||
super.retryLimit(retryLimit);
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public RemoteChunkingMasterStepBuilder<I, O> retryPolicy(RetryPolicy retryPolicy) {
|
||||
super.retryPolicy(retryPolicy);
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public RemoteChunkingMasterStepBuilder<I, O> backOffPolicy(BackOffPolicy backOffPolicy) {
|
||||
super.backOffPolicy(backOffPolicy);
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public RemoteChunkingMasterStepBuilder<I, O> retryContextCache(RetryContextCache retryContextCache) {
|
||||
super.retryContextCache(retryContextCache);
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public RemoteChunkingMasterStepBuilder<I, O> skipLimit(int skipLimit) {
|
||||
super.skipLimit(skipLimit);
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public RemoteChunkingMasterStepBuilder<I, O> noSkip(Class<? extends Throwable> type) {
|
||||
super.noSkip(type);
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public RemoteChunkingMasterStepBuilder<I, O> skip(Class<? extends Throwable> type) {
|
||||
super.skip(type);
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public RemoteChunkingMasterStepBuilder<I, O> skipPolicy(SkipPolicy skipPolicy) {
|
||||
super.skipPolicy(skipPolicy);
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public RemoteChunkingMasterStepBuilder<I, O> noRollback(Class<? extends Throwable> type) {
|
||||
super.noRollback(type);
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public RemoteChunkingMasterStepBuilder<I, O> noRetry(Class<? extends Throwable> type) {
|
||||
super.noRetry(type);
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public RemoteChunkingMasterStepBuilder<I, O> retry(Class<? extends Throwable> type) {
|
||||
super.retry(type);
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public RemoteChunkingMasterStepBuilder<I, O> stream(ItemStream stream) {
|
||||
super.stream(stream);
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public RemoteChunkingMasterStepBuilder<I, O> chunk(int chunkSize) {
|
||||
super.chunk(chunkSize);
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public RemoteChunkingMasterStepBuilder<I, O> chunk(CompletionPolicy completionPolicy) {
|
||||
super.chunk(completionPolicy);
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* This method will throw a {@link UnsupportedOperationException} since
|
||||
* the item writer of the master step in a remote chunking setup will be
|
||||
* automatically set to an instance of {@link ChunkMessageChannelItemWriter}.
|
||||
*
|
||||
* When building a master step for remote chunking, no item writer must be
|
||||
* provided.
|
||||
*
|
||||
* @throws UnsupportedOperationException if an item writer is provided
|
||||
* @see ChunkMessageChannelItemWriter
|
||||
* @see RemoteChunkHandlerFactoryBean#setChunkWriter(ItemWriter)
|
||||
*/
|
||||
@Override
|
||||
public RemoteChunkingMasterStepBuilder<I, O> writer(ItemWriter<? super O> writer) throws UnsupportedOperationException {
|
||||
throw new UnsupportedOperationException("When configuring a master step " +
|
||||
"for remote chunking, the item writer will be automatically set " +
|
||||
"to an instance of ChunkMessageChannelItemWriter. The item writer " +
|
||||
"must not be provided in this case.");
|
||||
}
|
||||
|
||||
@Override
|
||||
public RemoteChunkingMasterStepBuilder<I, O> readerIsTransactionalQueue() {
|
||||
super.readerIsTransactionalQueue();
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public RemoteChunkingMasterStepBuilder<I, O> listener(ItemReadListener<? super I> listener) {
|
||||
super.listener(listener);
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public RemoteChunkingMasterStepBuilder<I, O> listener(ItemWriteListener<? super O> listener) {
|
||||
super.listener(listener);
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public RemoteChunkingMasterStepBuilder<I, O> chunkOperations(RepeatOperations repeatTemplate) {
|
||||
super.chunkOperations(repeatTemplate);
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public RemoteChunkingMasterStepBuilder<I, O> exceptionHandler(ExceptionHandler exceptionHandler) {
|
||||
super.exceptionHandler(exceptionHandler);
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public RemoteChunkingMasterStepBuilder<I, O> stepOperations(RepeatOperations repeatTemplate) {
|
||||
super.stepOperations(repeatTemplate);
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public RemoteChunkingMasterStepBuilder<I, O> startLimit(int startLimit) {
|
||||
super.startLimit(startLimit);
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public RemoteChunkingMasterStepBuilder<I, O> listener(StepExecutionListener listener) {
|
||||
super.listener(listener);
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public RemoteChunkingMasterStepBuilder<I, O> allowStartIfComplete(boolean allowStartIfComplete) {
|
||||
super.allowStartIfComplete(allowStartIfComplete);
|
||||
return this;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,61 @@
|
||||
/*
|
||||
* Copyright 2018 the original author or authors.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
package org.springframework.batch.integration.chunk;
|
||||
|
||||
import org.springframework.batch.core.repository.JobRepository;
|
||||
import org.springframework.transaction.PlatformTransactionManager;
|
||||
|
||||
/**
|
||||
* Convenient factory for a {@link RemoteChunkingMasterStepBuilder} which sets
|
||||
* the {@link JobRepository} and {@link PlatformTransactionManager} automatically.
|
||||
*
|
||||
* @since 4.1
|
||||
* @author Mahmoud Ben Hassine
|
||||
*/
|
||||
public class RemoteChunkingMasterStepBuilderFactory {
|
||||
|
||||
private JobRepository jobRepository;
|
||||
|
||||
private PlatformTransactionManager transactionManager;
|
||||
|
||||
/**
|
||||
* Create a new {@link RemoteChunkingMasterStepBuilderFactory}.
|
||||
*
|
||||
* @param jobRepository the job repository to use
|
||||
* @param transactionManager the transaction manager to use
|
||||
*/
|
||||
public RemoteChunkingMasterStepBuilderFactory(
|
||||
JobRepository jobRepository,
|
||||
PlatformTransactionManager transactionManager) {
|
||||
|
||||
this.jobRepository = jobRepository;
|
||||
this.transactionManager = transactionManager;
|
||||
}
|
||||
|
||||
/**
|
||||
* Creates a {@link RemoteChunkingMasterStepBuilder} and initializes its job
|
||||
* repository and transaction manager.
|
||||
*
|
||||
* @param name the name of the step
|
||||
* @return a {@link RemoteChunkingMasterStepBuilder}
|
||||
*/
|
||||
public <I, O> RemoteChunkingMasterStepBuilder<I, O> get(String name) {
|
||||
return new RemoteChunkingMasterStepBuilder<I, O>(name)
|
||||
.repository(this.jobRepository)
|
||||
.transactionManager(this.transactionManager);
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,131 @@
|
||||
/*
|
||||
* Copyright 2018 the original author or authors.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
package org.springframework.batch.integration.chunk;
|
||||
|
||||
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.util.Assert;
|
||||
|
||||
/**
|
||||
* Builder for a worker in a remote chunking setup. This builder:
|
||||
*
|
||||
* <ul>
|
||||
* <li>creates a {@link ChunkProcessorChunkHandler} with the provided
|
||||
* item processor and writer. If no item processor is provided, a
|
||||
* {@link PassThroughItemProcessor} will be used</li>
|
||||
* <li>creates an {@link IntegrationFlow} with the
|
||||
* {@link ChunkProcessorChunkHandler} as a service activator which listens
|
||||
* to incoming requests on <code>inputChannel</code> and sends replies
|
||||
* on <code>outputChannel</code></li>
|
||||
* </ul>
|
||||
*
|
||||
* @param <I> type of input items
|
||||
* @param <O> type of output items
|
||||
*
|
||||
* @since 4.1
|
||||
* @author Mahmoud Ben Hassine
|
||||
*/
|
||||
public class RemoteChunkingWorkerBuilder<I, O> {
|
||||
|
||||
private static final String SERVICE_ACTIVATOR_METHOD_NAME = "handleChunk";
|
||||
|
||||
private ItemProcessor<I, O> itemProcessor;
|
||||
private ItemWriter<O> itemWriter;
|
||||
private DirectChannel inputChannel;
|
||||
private DirectChannel outputChannel;
|
||||
|
||||
/**
|
||||
* Set the {@link ItemProcessor} to use to process items sent by the master
|
||||
* step.
|
||||
*
|
||||
* @param itemProcessor to use
|
||||
* @return this builder instance for fluent chaining
|
||||
*/
|
||||
public RemoteChunkingWorkerBuilder<I, O> itemProcessor(ItemProcessor<I, O> itemProcessor) {
|
||||
Assert.notNull(itemProcessor, "itemProcessor must not be null");
|
||||
this.itemProcessor = itemProcessor;
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* Set the {@link ItemWriter} to use to write items sent by the master step.
|
||||
*
|
||||
* @param itemWriter to use
|
||||
* @return this builder instance for fluent chaining
|
||||
*/
|
||||
public RemoteChunkingWorkerBuilder<I, O> itemWriter(ItemWriter<O> itemWriter) {
|
||||
Assert.notNull(itemWriter, "itemWriter must not be null");
|
||||
this.itemWriter = itemWriter;
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* Set the input channel on which items sent by the master are received.
|
||||
*
|
||||
* @param inputChannel the input channel
|
||||
* @return this builder instance for fluent chaining
|
||||
*/
|
||||
public RemoteChunkingWorkerBuilder<I, O> inputChannel(DirectChannel inputChannel) {
|
||||
Assert.notNull(inputChannel, "inputChannel must not be null");
|
||||
this.inputChannel = inputChannel;
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* Set the output channel on which replies will be sent to the master step.
|
||||
*
|
||||
* @param outputChannel the output channel
|
||||
* @return this builder instance for fluent chaining
|
||||
*/
|
||||
public RemoteChunkingWorkerBuilder<I, O> outputChannel(DirectChannel outputChannel) {
|
||||
Assert.notNull(outputChannel, "outputChannel must not be null");
|
||||
this.outputChannel = outputChannel;
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* Create an {@link IntegrationFlow} with a {@link ChunkProcessorChunkHandler}
|
||||
* configured as a service activator listening to the input channel and replying
|
||||
* on the output channel.
|
||||
*
|
||||
* @return the integration flow
|
||||
*/
|
||||
public IntegrationFlow build() {
|
||||
Assert.notNull(this.itemWriter, "An ItemWriter must be provided");
|
||||
Assert.notNull(this.inputChannel, "An InputChannel must be provided");
|
||||
Assert.notNull(this.outputChannel, "An OutputChannel must be provided");
|
||||
|
||||
if(this.itemProcessor == null) {
|
||||
this.itemProcessor = new PassThroughItemProcessor();
|
||||
}
|
||||
SimpleChunkProcessor<I, O> chunkProcessor = new SimpleChunkProcessor<>(this.itemProcessor, this.itemWriter);
|
||||
|
||||
ChunkProcessorChunkHandler<I> chunkProcessorChunkHandler = new ChunkProcessorChunkHandler<>();
|
||||
chunkProcessorChunkHandler.setChunkProcessor(chunkProcessor);
|
||||
|
||||
return IntegrationFlows
|
||||
.from(this.inputChannel)
|
||||
.handle(chunkProcessorChunkHandler, SERVICE_ACTIVATOR_METHOD_NAME)
|
||||
.channel(this.outputChannel)
|
||||
.get();
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,59 @@
|
||||
/*
|
||||
* Copyright 2018 the original author or authors.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
package org.springframework.batch.integration.config.annotation;
|
||||
|
||||
import org.springframework.batch.core.repository.JobRepository;
|
||||
import org.springframework.batch.integration.chunk.RemoteChunkingMasterStepBuilderFactory;
|
||||
import org.springframework.batch.integration.chunk.RemoteChunkingWorkerBuilder;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.context.annotation.Configuration;
|
||||
import org.springframework.transaction.PlatformTransactionManager;
|
||||
|
||||
/**
|
||||
* Base configuration class for Spring Batch Integration factory beans.
|
||||
*
|
||||
* @since 4.1
|
||||
* @author Mahmoud Ben Hassine
|
||||
*/
|
||||
@Configuration
|
||||
public class BatchIntegrationConfiguration {
|
||||
|
||||
private JobRepository jobRepository;
|
||||
|
||||
private PlatformTransactionManager transactionManager;
|
||||
|
||||
@Autowired
|
||||
public BatchIntegrationConfiguration(
|
||||
JobRepository jobRepository,
|
||||
PlatformTransactionManager transactionManager) {
|
||||
|
||||
this.jobRepository = jobRepository;
|
||||
this.transactionManager = transactionManager;
|
||||
}
|
||||
|
||||
@Bean
|
||||
public RemoteChunkingMasterStepBuilderFactory remoteChunkingMasterStepBuilderFactory() {
|
||||
return new RemoteChunkingMasterStepBuilderFactory(this.jobRepository,
|
||||
this.transactionManager);
|
||||
}
|
||||
|
||||
@Bean
|
||||
public <I,O> RemoteChunkingWorkerBuilder<I, O> remoteChunkingWorkerBuilder() {
|
||||
return new RemoteChunkingWorkerBuilder<>();
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,88 @@
|
||||
/*
|
||||
* Copyright 2018 the original author or authors.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
package org.springframework.batch.integration.config.annotation;
|
||||
|
||||
import java.lang.annotation.Documented;
|
||||
import java.lang.annotation.ElementType;
|
||||
import java.lang.annotation.Retention;
|
||||
import java.lang.annotation.RetentionPolicy;
|
||||
import java.lang.annotation.Target;
|
||||
|
||||
import org.springframework.batch.integration.chunk.RemoteChunkingWorkerBuilder;
|
||||
import org.springframework.batch.integration.chunk.RemoteChunkingMasterStepBuilderFactory;
|
||||
import org.springframework.context.annotation.Import;
|
||||
|
||||
/**
|
||||
* Enable Spring Batch Integration features and provide a base configuration for
|
||||
* setting up remote chunking infrastructure beans.
|
||||
*
|
||||
* By adding this annotation on a {@link org.springframework.context.annotation.Configuration}
|
||||
* class, it will be possible to autowire the following beans:
|
||||
*
|
||||
* <ul>
|
||||
* <li>{@link RemoteChunkingMasterStepBuilderFactory}:
|
||||
* used to create a master step by automatically setting up the job repository
|
||||
* and transaction manager.</li>
|
||||
* <li>{@link RemoteChunkingWorkerBuilder}: used to create the integration
|
||||
* flow on the worker side.</li>
|
||||
* </ul>
|
||||
*
|
||||
* For example:
|
||||
*
|
||||
* <pre class="code">
|
||||
* @Configuration
|
||||
* @EnableBatchIntegration
|
||||
* public class RemoteChunkingAppConfig {
|
||||
*
|
||||
* @Autowired
|
||||
* private RemoteChunkingMasterStepBuilderFactory masterStepBuilderFactory;
|
||||
*
|
||||
* @Autowired
|
||||
* private RemoteChunkingWorkerBuilder workerBuilder;
|
||||
*
|
||||
* @Bean
|
||||
* public TaskletStep masterStep() {
|
||||
* return this.masterStepBuilderFactory
|
||||
* .get("masterStep")
|
||||
* .chunk(100)
|
||||
* .reader(itemReader())
|
||||
* .outputChannel(outgoingRequestsToWorkers())
|
||||
* .inputChannel(incomingRepliesFromWorkers())
|
||||
* .build();
|
||||
* }
|
||||
*
|
||||
* @Bean
|
||||
* public IntegrationFlow worker() {
|
||||
* return this.workerBuilder
|
||||
* .itemProcessor(itemProcessor())
|
||||
* .itemWriter(itemWriter())
|
||||
* .inputChannel(incomingRequestsFromMaster())
|
||||
* .outputChannel(outgoingRepliesToMaster())
|
||||
* .build();
|
||||
* }
|
||||
*
|
||||
* }
|
||||
* </pre>
|
||||
*
|
||||
* @since 4.1
|
||||
* @author Mahmoud Ben Hassine
|
||||
*/
|
||||
@Target(ElementType.TYPE)
|
||||
@Retention(RetentionPolicy.RUNTIME)
|
||||
@Documented
|
||||
@Import(BatchIntegrationConfiguration.class)
|
||||
public @interface EnableBatchIntegration {
|
||||
}
|
||||
@@ -0,0 +1,264 @@
|
||||
/*
|
||||
* Copyright 2018 the original author or authors.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
package org.springframework.batch.integration.chunk;
|
||||
|
||||
import java.util.Arrays;
|
||||
|
||||
import org.junit.Assert;
|
||||
import org.junit.Rule;
|
||||
import org.junit.Test;
|
||||
import org.junit.rules.ExpectedException;
|
||||
import org.junit.runner.RunWith;
|
||||
|
||||
import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
|
||||
import org.springframework.batch.core.listener.ChunkListenerSupport;
|
||||
import org.springframework.batch.core.listener.CompositeItemReadListener;
|
||||
import org.springframework.batch.core.listener.CompositeItemWriteListener;
|
||||
import org.springframework.batch.core.listener.SkipListenerSupport;
|
||||
import org.springframework.batch.core.listener.StepExecutionListenerSupport;
|
||||
import org.springframework.batch.core.repository.JobRepository;
|
||||
import org.springframework.batch.core.step.tasklet.TaskletStep;
|
||||
import org.springframework.batch.item.ItemReader;
|
||||
import org.springframework.batch.item.ItemStreamSupport;
|
||||
import org.springframework.batch.item.support.ListItemReader;
|
||||
import org.springframework.batch.repeat.exception.DefaultExceptionHandler;
|
||||
import org.springframework.batch.repeat.support.RepeatTemplate;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.context.annotation.Configuration;
|
||||
import org.springframework.integration.channel.DirectChannel;
|
||||
import org.springframework.integration.channel.QueueChannel;
|
||||
import org.springframework.messaging.PollableChannel;
|
||||
import org.springframework.retry.backoff.NoBackOffPolicy;
|
||||
import org.springframework.retry.listener.RetryListenerSupport;
|
||||
import org.springframework.retry.policy.MapRetryContextCache;
|
||||
import org.springframework.test.context.ContextConfiguration;
|
||||
import org.springframework.test.context.junit4.SpringRunner;
|
||||
import org.springframework.transaction.PlatformTransactionManager;
|
||||
import org.springframework.transaction.interceptor.DefaultTransactionAttribute;
|
||||
|
||||
/**
|
||||
* @author Mahmoud Ben Hassine
|
||||
*/
|
||||
@RunWith(SpringRunner.class)
|
||||
@ContextConfiguration(classes = {RemoteChunkingMasterStepBuilderTest.BatchConfiguration.class})
|
||||
public class RemoteChunkingMasterStepBuilderTest {
|
||||
|
||||
@Rule
|
||||
public ExpectedException expectedException = ExpectedException.none();
|
||||
|
||||
@Autowired
|
||||
private JobRepository jobRepository;
|
||||
@Autowired
|
||||
private PlatformTransactionManager transactionManager;
|
||||
|
||||
private PollableChannel inputChannel = new QueueChannel();
|
||||
private DirectChannel outputChannel = new DirectChannel();
|
||||
private ItemReader<String> itemReader = new ListItemReader<>(Arrays.asList("a", "b", "c"));
|
||||
|
||||
@Test
|
||||
public void inputChannelMustNotBeNull() {
|
||||
// given
|
||||
this.expectedException.expect(IllegalArgumentException.class);
|
||||
this.expectedException.expectMessage("inputChannel must not be null");
|
||||
|
||||
// when
|
||||
TaskletStep step = new RemoteChunkingMasterStepBuilder<String, String>("step")
|
||||
.inputChannel(null)
|
||||
.build();
|
||||
|
||||
// then
|
||||
// expected exception
|
||||
}
|
||||
|
||||
@Test
|
||||
public void outputChannelMustNotBeNull() {
|
||||
// given
|
||||
this.expectedException.expect(IllegalArgumentException.class);
|
||||
this.expectedException.expectMessage("outputChannel must not be null");
|
||||
|
||||
// when
|
||||
TaskletStep step = new RemoteChunkingMasterStepBuilder<String, String>("step")
|
||||
.outputChannel(null)
|
||||
.build();
|
||||
|
||||
// then
|
||||
// expected exception
|
||||
}
|
||||
|
||||
@Test
|
||||
public void messagingTemplateMustNotBeNull() {
|
||||
// given
|
||||
this.expectedException.expect(IllegalArgumentException.class);
|
||||
this.expectedException.expectMessage("messagingTemplate must not be null");
|
||||
|
||||
// when
|
||||
TaskletStep step = new RemoteChunkingMasterStepBuilder<String, String>("step")
|
||||
.messagingTemplate(null)
|
||||
.build();
|
||||
|
||||
// then
|
||||
// expected exception
|
||||
}
|
||||
|
||||
@Test
|
||||
public void maxWaitTimeoutsMustBeGreaterThanZero() {
|
||||
// given
|
||||
this.expectedException.expect(IllegalArgumentException.class);
|
||||
this.expectedException.expectMessage("maxWaitTimeouts must be greater than zero");
|
||||
|
||||
// when
|
||||
TaskletStep step = new RemoteChunkingMasterStepBuilder<String, String>("step")
|
||||
.maxWaitTimeouts(-1)
|
||||
.build();
|
||||
|
||||
// then
|
||||
// expected exception
|
||||
}
|
||||
|
||||
@Test
|
||||
public void throttleLimitMustNotBeGreaterThanZero() {
|
||||
// given
|
||||
this.expectedException.expect(IllegalArgumentException.class);
|
||||
this.expectedException.expectMessage("throttleLimit must be greater than zero");
|
||||
|
||||
// when
|
||||
TaskletStep step = new RemoteChunkingMasterStepBuilder<String, String>("step")
|
||||
.throttleLimit(-1L)
|
||||
.build();
|
||||
|
||||
// then
|
||||
// expected exception
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testMandatoryInputChannel() {
|
||||
// given
|
||||
RemoteChunkingMasterStepBuilder<String, String> builder = new RemoteChunkingMasterStepBuilder<>("step");
|
||||
|
||||
this.expectedException.expect(IllegalArgumentException.class);
|
||||
this.expectedException.expectMessage("An InputChannel must be provided");
|
||||
|
||||
// when
|
||||
TaskletStep step = builder.build();
|
||||
|
||||
// then
|
||||
// expected exception
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testMandatoryOutputChannel() {
|
||||
// given
|
||||
RemoteChunkingMasterStepBuilder<String, String> builder = new RemoteChunkingMasterStepBuilder<String, String>("step")
|
||||
.inputChannel(this.inputChannel);
|
||||
|
||||
this.expectedException.expect(IllegalArgumentException.class);
|
||||
this.expectedException.expectMessage("An OutputChannel must be provided");
|
||||
|
||||
// when
|
||||
TaskletStep step = builder.build();
|
||||
|
||||
// then
|
||||
// expected exception
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testUnsupportedOperationExceptionWhenSpecifyingAnItemWriter() {
|
||||
// given
|
||||
this.expectedException.expect(UnsupportedOperationException.class);
|
||||
this.expectedException.expectMessage("When configuring a master " +
|
||||
"step for remote chunking, the item writer will be automatically " +
|
||||
"set to an instance of ChunkMessageChannelItemWriter. " +
|
||||
"The item writer must not be provided in this case.");
|
||||
|
||||
// when
|
||||
TaskletStep step = new RemoteChunkingMasterStepBuilder<String, String>("step")
|
||||
.reader(this.itemReader)
|
||||
.writer(items -> { })
|
||||
.repository(this.jobRepository)
|
||||
.transactionManager(this.transactionManager)
|
||||
.inputChannel(this.inputChannel)
|
||||
.outputChannel(this.outputChannel)
|
||||
.build();
|
||||
|
||||
// then
|
||||
// expected exception
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testMasterStepCreation() {
|
||||
// when
|
||||
TaskletStep taskletStep = new RemoteChunkingMasterStepBuilder<String, String>("step")
|
||||
.reader(this.itemReader)
|
||||
.repository(this.jobRepository)
|
||||
.transactionManager(this.transactionManager)
|
||||
.inputChannel(this.inputChannel)
|
||||
.outputChannel(this.outputChannel)
|
||||
.build();
|
||||
|
||||
// then
|
||||
Assert.assertNotNull(taskletStep);
|
||||
}
|
||||
|
||||
/*
|
||||
* The following test is to cover setters that override those from parent builders.
|
||||
*/
|
||||
@Test
|
||||
public void testSetters() {
|
||||
// when
|
||||
TaskletStep taskletStep = new RemoteChunkingMasterStepBuilder<String, String>("step")
|
||||
.reader(this.itemReader)
|
||||
.readerIsTransactionalQueue()
|
||||
.repository(this.jobRepository)
|
||||
.transactionManager(this.transactionManager)
|
||||
.transactionAttribute(new DefaultTransactionAttribute())
|
||||
.inputChannel(this.inputChannel)
|
||||
.outputChannel(this.outputChannel)
|
||||
.listener(new Object())
|
||||
.listener(new SkipListenerSupport<>())
|
||||
.listener(new ChunkListenerSupport())
|
||||
.listener(new StepExecutionListenerSupport())
|
||||
.listener(new CompositeItemReadListener<>())
|
||||
.listener(new CompositeItemWriteListener<>())
|
||||
.listener(new RetryListenerSupport())
|
||||
.skip(Exception.class)
|
||||
.noSkip(RuntimeException.class)
|
||||
.skipLimit(10)
|
||||
.retry(Exception.class)
|
||||
.noRetry(RuntimeException.class)
|
||||
.retryLimit(10)
|
||||
.retryContextCache(new MapRetryContextCache())
|
||||
.noRollback(Exception.class)
|
||||
.chunk(10)
|
||||
.startLimit(3)
|
||||
.allowStartIfComplete(true)
|
||||
.exceptionHandler(new DefaultExceptionHandler())
|
||||
.stepOperations(new RepeatTemplate())
|
||||
.chunkOperations(new RepeatTemplate())
|
||||
.backOffPolicy(new NoBackOffPolicy())
|
||||
.stream(new ItemStreamSupport() {})
|
||||
.keyGenerator(Object::hashCode)
|
||||
.build();
|
||||
|
||||
// then
|
||||
Assert.assertNotNull(taskletStep);
|
||||
}
|
||||
|
||||
@Configuration
|
||||
@EnableBatchProcessing
|
||||
public static class BatchConfiguration {
|
||||
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,166 @@
|
||||
/*
|
||||
* Copyright 2018 the original author or authors.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
package org.springframework.batch.integration.chunk;
|
||||
|
||||
import org.junit.Assert;
|
||||
import org.junit.Rule;
|
||||
import org.junit.Test;
|
||||
import org.junit.rules.ExpectedException;
|
||||
|
||||
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;
|
||||
|
||||
/**
|
||||
* @author Mahmoud Ben Hassine
|
||||
*/
|
||||
public class RemoteChunkingWorkerBuilderTest {
|
||||
|
||||
@Rule
|
||||
public ExpectedException expectedException = ExpectedException.none();
|
||||
|
||||
private ItemProcessor<String, String> itemProcessor = new PassThroughItemProcessor<>();
|
||||
private ItemWriter<String> itemWriter = items -> { };
|
||||
|
||||
@Test
|
||||
public void itemProcessorMustNotBeNull() {
|
||||
// given
|
||||
this.expectedException.expect(IllegalArgumentException.class);
|
||||
this.expectedException.expectMessage("itemProcessor must not be null");
|
||||
|
||||
// when
|
||||
IntegrationFlow integrationFlow = new RemoteChunkingWorkerBuilder<String, String>()
|
||||
.itemProcessor(null)
|
||||
.build();
|
||||
|
||||
// then
|
||||
// expected exception
|
||||
}
|
||||
|
||||
@Test
|
||||
public void itemWriterMustNotBeNull() {
|
||||
// given
|
||||
this.expectedException.expect(IllegalArgumentException.class);
|
||||
this.expectedException.expectMessage("itemWriter must not be null");
|
||||
|
||||
// when
|
||||
IntegrationFlow integrationFlow = new RemoteChunkingWorkerBuilder<String, String>()
|
||||
.itemWriter(null)
|
||||
.build();
|
||||
|
||||
// then
|
||||
// expected exception
|
||||
}
|
||||
|
||||
@Test
|
||||
public void inputChannelMustNotBeNull() {
|
||||
// given
|
||||
this.expectedException.expect(IllegalArgumentException.class);
|
||||
this.expectedException.expectMessage("inputChannel must not be null");
|
||||
|
||||
// when
|
||||
IntegrationFlow integrationFlow = new RemoteChunkingWorkerBuilder<String, String>()
|
||||
.inputChannel(null)
|
||||
.build();
|
||||
|
||||
// then
|
||||
// expected exception
|
||||
}
|
||||
|
||||
@Test
|
||||
public void outputChannelMustNotBeNull() {
|
||||
// given
|
||||
this.expectedException.expect(IllegalArgumentException.class);
|
||||
this.expectedException.expectMessage("outputChannel must not be null");
|
||||
|
||||
// when
|
||||
IntegrationFlow integrationFlow = new RemoteChunkingWorkerBuilder<String, String>()
|
||||
.outputChannel(null)
|
||||
.build();
|
||||
|
||||
// then
|
||||
// expected exception
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testMandatoryItemWriter() {
|
||||
// given
|
||||
RemoteChunkingWorkerBuilder<String, String> builder = new RemoteChunkingWorkerBuilder<>();
|
||||
|
||||
this.expectedException.expect(IllegalArgumentException.class);
|
||||
this.expectedException.expectMessage("An ItemWriter must be provided");
|
||||
|
||||
// when
|
||||
builder.build();
|
||||
|
||||
// then
|
||||
// expected exception
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testMandatoryInputChannel() {
|
||||
// given
|
||||
RemoteChunkingWorkerBuilder<String, String> builder = new RemoteChunkingWorkerBuilder<String, String>()
|
||||
.itemWriter(items -> { });
|
||||
|
||||
this.expectedException.expect(IllegalArgumentException.class);
|
||||
this.expectedException.expectMessage("An InputChannel must be provided");
|
||||
|
||||
// when
|
||||
builder.build();
|
||||
|
||||
// then
|
||||
// expected exception
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testMandatoryOutputChannel() {
|
||||
// given
|
||||
RemoteChunkingWorkerBuilder<String, String> builder = new RemoteChunkingWorkerBuilder<String, String>()
|
||||
.itemWriter(items -> { })
|
||||
.inputChannel(new DirectChannel());
|
||||
|
||||
this.expectedException.expect(IllegalArgumentException.class);
|
||||
this.expectedException.expectMessage("An OutputChannel must be provided");
|
||||
|
||||
// when
|
||||
builder.build();
|
||||
|
||||
// then
|
||||
// expected exception
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testIntegrationFlowCreation() {
|
||||
// given
|
||||
DirectChannel inputChannel = new DirectChannel();
|
||||
DirectChannel outputChannel = new DirectChannel();
|
||||
RemoteChunkingWorkerBuilder<String, String> builder = new RemoteChunkingWorkerBuilder<String, String>()
|
||||
.itemProcessor(this.itemProcessor)
|
||||
.itemWriter(this.itemWriter)
|
||||
.inputChannel(inputChannel)
|
||||
.outputChannel(outputChannel);
|
||||
|
||||
// when
|
||||
IntegrationFlow integrationFlow = builder.build();
|
||||
|
||||
// then
|
||||
Assert.assertNotNull(integrationFlow);
|
||||
}
|
||||
|
||||
}
|
||||
@@ -651,11 +651,10 @@ This sample shows how to configure a remote chunking job. The master step will
|
||||
read numbers from 1 to 6 and send two chunks ({1, 2, 3} and {4, 5, 6}) to workers
|
||||
for processing and writing.
|
||||
|
||||
This example shows how to:
|
||||
This example shows how to use:
|
||||
|
||||
* configure a `ChunkMessageChannelItemWriter` on the master side to send chunks to workers
|
||||
* configure a `ChunkProcessorChunkHandler` on the worker side to process chunks and
|
||||
send replies back to the master
|
||||
* the `RemoteChunkingMasterStepBuilderFactory` to create a master step
|
||||
* the `RemoteChunkingWorkerBuilder` to configure an integration flow on the worker side.
|
||||
|
||||
The sample uses an embedded JMS broker as a communication middleware between the
|
||||
master and workers. The usage of an embedded broker is only for simplicity's sake,
|
||||
|
||||
@@ -22,9 +22,9 @@ import org.apache.activemq.ActiveMQConnectionFactory;
|
||||
import org.springframework.batch.core.Job;
|
||||
import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
|
||||
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
|
||||
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
|
||||
import org.springframework.batch.core.step.tasklet.TaskletStep;
|
||||
import org.springframework.batch.integration.chunk.ChunkMessageChannelItemWriter;
|
||||
import org.springframework.batch.integration.chunk.RemoteChunkingMasterStepBuilderFactory;
|
||||
import org.springframework.batch.integration.config.annotation.EnableBatchIntegration;
|
||||
import org.springframework.batch.item.support.ListItemReader;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.beans.factory.annotation.Value;
|
||||
@@ -34,7 +34,6 @@ import org.springframework.context.annotation.PropertySource;
|
||||
import org.springframework.integration.channel.DirectChannel;
|
||||
import org.springframework.integration.channel.QueueChannel;
|
||||
import org.springframework.integration.config.EnableIntegration;
|
||||
import org.springframework.integration.core.MessagingTemplate;
|
||||
import org.springframework.integration.dsl.IntegrationFlow;
|
||||
import org.springframework.integration.dsl.IntegrationFlows;
|
||||
import org.springframework.integration.jms.dsl.Jms;
|
||||
@@ -48,6 +47,7 @@ import org.springframework.integration.jms.dsl.Jms;
|
||||
*/
|
||||
@Configuration
|
||||
@EnableBatchProcessing
|
||||
@EnableBatchIntegration
|
||||
@EnableIntegration
|
||||
@PropertySource("classpath:remote-chunking.properties")
|
||||
public class MasterConfiguration {
|
||||
@@ -59,7 +59,7 @@ public class MasterConfiguration {
|
||||
private JobBuilderFactory jobBuilderFactory;
|
||||
|
||||
@Autowired
|
||||
private StepBuilderFactory stepBuilderFactory;
|
||||
private RemoteChunkingMasterStepBuilderFactory masterStepBuilderFactory;
|
||||
|
||||
@Bean
|
||||
public ActiveMQConnectionFactory connectionFactory() {
|
||||
@@ -109,23 +109,13 @@ public class MasterConfiguration {
|
||||
return new ListItemReader<>(Arrays.asList(1, 2, 3, 4, 5, 6));
|
||||
}
|
||||
|
||||
@Bean
|
||||
public ChunkMessageChannelItemWriter<Integer> itemWriter() {
|
||||
MessagingTemplate messagingTemplate = new MessagingTemplate();
|
||||
messagingTemplate.setDefaultChannel(requests());
|
||||
ChunkMessageChannelItemWriter<Integer> chunkMessageChannelItemWriter
|
||||
= new ChunkMessageChannelItemWriter<>();
|
||||
chunkMessageChannelItemWriter.setMessagingOperations(messagingTemplate);
|
||||
chunkMessageChannelItemWriter.setReplyChannel(replies());
|
||||
return chunkMessageChannelItemWriter;
|
||||
}
|
||||
|
||||
@Bean
|
||||
public TaskletStep masterStep() {
|
||||
return this.stepBuilderFactory.get("masterStep")
|
||||
return this.masterStepBuilderFactory.get("masterStep")
|
||||
.<Integer, Integer>chunk(3)
|
||||
.reader(itemReader())
|
||||
.writer(itemWriter())
|
||||
.outputChannel(requests())
|
||||
.inputChannel(replies())
|
||||
.build();
|
||||
}
|
||||
|
||||
|
||||
@@ -18,16 +18,15 @@ package org.springframework.batch.sample.remotechunking;
|
||||
import org.apache.activemq.ActiveMQConnectionFactory;
|
||||
|
||||
import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
|
||||
import org.springframework.batch.core.step.item.ChunkProcessor;
|
||||
import org.springframework.batch.core.step.item.SimpleChunkProcessor;
|
||||
import org.springframework.batch.integration.chunk.ChunkProcessorChunkHandler;
|
||||
import org.springframework.batch.integration.chunk.RemoteChunkingWorkerBuilder;
|
||||
import org.springframework.batch.integration.config.annotation.EnableBatchIntegration;
|
||||
import org.springframework.batch.item.ItemProcessor;
|
||||
import org.springframework.batch.item.ItemWriter;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.beans.factory.annotation.Value;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.context.annotation.Configuration;
|
||||
import org.springframework.context.annotation.PropertySource;
|
||||
import org.springframework.integration.annotation.ServiceActivator;
|
||||
import org.springframework.integration.channel.DirectChannel;
|
||||
import org.springframework.integration.config.EnableIntegration;
|
||||
import org.springframework.integration.dsl.IntegrationFlow;
|
||||
@@ -36,7 +35,8 @@ import org.springframework.integration.jms.dsl.Jms;
|
||||
|
||||
/**
|
||||
* This configuration class is for the worker side of the remote chunking sample.
|
||||
* It configures a {@link ChunkProcessorChunkHandler} as a service activator to:
|
||||
* It uses the {@link RemoteChunkingWorkerBuilder} to configure an
|
||||
* {@link IntegrationFlow} in order to:
|
||||
* <ul>
|
||||
* <li>receive requests from the master</li>
|
||||
* <li>process chunks with the configured item processor and writer</li>
|
||||
@@ -47,6 +47,7 @@ import org.springframework.integration.jms.dsl.Jms;
|
||||
*/
|
||||
@Configuration
|
||||
@EnableBatchProcessing
|
||||
@EnableBatchIntegration
|
||||
@EnableIntegration
|
||||
@PropertySource("classpath:remote-chunking.properties")
|
||||
public class WorkerConfiguration {
|
||||
@@ -54,6 +55,9 @@ public class WorkerConfiguration {
|
||||
@Value("${broker.url}")
|
||||
private String brokerUrl;
|
||||
|
||||
@Autowired
|
||||
private RemoteChunkingWorkerBuilder<Integer, Integer> remoteChunkingWorkerBuilder;
|
||||
|
||||
@Bean
|
||||
public ActiveMQConnectionFactory connectionFactory() {
|
||||
ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory();
|
||||
@@ -115,12 +119,13 @@ public class WorkerConfiguration {
|
||||
}
|
||||
|
||||
@Bean
|
||||
@ServiceActivator(inputChannel = "requests", outputChannel = "replies")
|
||||
public ChunkProcessorChunkHandler<Integer> chunkProcessorChunkHandler() {
|
||||
ChunkProcessor<Integer> chunkProcessor = new SimpleChunkProcessor<>(itemProcessor(), itemWriter());
|
||||
ChunkProcessorChunkHandler<Integer> chunkProcessorChunkHandler = new ChunkProcessorChunkHandler<>();
|
||||
chunkProcessorChunkHandler.setChunkProcessor(chunkProcessor);
|
||||
return chunkProcessorChunkHandler;
|
||||
public IntegrationFlow workerIntegrationFlow() {
|
||||
return this.remoteChunkingWorkerBuilder
|
||||
.itemProcessor(itemProcessor())
|
||||
.itemWriter(itemWriter())
|
||||
.inputChannel(requests())
|
||||
.outputChannel(replies())
|
||||
.build();
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user