Replace "Master/Slave" with "Manager/Worker"
Resolves BATCH-2834
This commit is contained in:
Mahmoud Ben Hassine
committed by
Michael Minella
Michael Minella
parent
3f126fe0ff
commit
3743894fa4
@@ -0,0 +1,420 @@
|
||||
/*
|
||||
* Copyright 2019 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
|
||||
*
|
||||
* https://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.ItemProcessor;
|
||||
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.core.MessagingTemplate;
|
||||
import org.springframework.messaging.MessageChannel;
|
||||
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 manager step in a remote chunking setup. This builder creates and
|
||||
* sets a {@link ChunkMessageChannelItemWriter} on the manager step.
|
||||
*
|
||||
* <p>If no {@code messagingTemplate} is provided through
|
||||
* {@link RemoteChunkingManagerStepBuilder#messagingTemplate(MessagingTemplate)},
|
||||
* this builder will create one and set its default channel to the {@code outputChannel}
|
||||
* provided through {@link RemoteChunkingManagerStepBuilder#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
|
||||
*
|
||||
* @since 4.2
|
||||
* @author Mahmoud Ben Hassine
|
||||
*/
|
||||
public class RemoteChunkingManagerStepBuilder<I, O> extends FaultTolerantStepBuilder<I, O> {
|
||||
|
||||
private MessagingTemplate messagingTemplate;
|
||||
private PollableChannel inputChannel;
|
||||
private MessageChannel 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 RemoteChunkingManagerStepBuilder}.
|
||||
*
|
||||
* @param stepName name of the manager step
|
||||
*/
|
||||
public RemoteChunkingManagerStepBuilder(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 RemoteChunkingManagerStepBuilder<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. 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 RemoteChunkingManagerStepBuilder#messagingTemplate(MessagingTemplate)}
|
||||
* to provide a fully configured messaging template.</p>
|
||||
*
|
||||
* @param outputChannel the output channel.
|
||||
* @return this builder instance for fluent chaining
|
||||
*
|
||||
* @see RemoteChunkingManagerStepBuilder#messagingTemplate(MessagingTemplate)
|
||||
*/
|
||||
public RemoteChunkingManagerStepBuilder<I, O> outputChannel(MessageChannel 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.
|
||||
* <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 RemoteChunkingManagerStepBuilder#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 RemoteChunkingManagerStepBuilder#outputChannel(MessageChannel)
|
||||
*/
|
||||
public RemoteChunkingManagerStepBuilder<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
|
||||
* @return this builder instance for fluent chaining
|
||||
* @see ChunkMessageChannelItemWriter#setMaxWaitTimeouts(int)
|
||||
*/
|
||||
public RemoteChunkingManagerStepBuilder<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
|
||||
* @return this builder instance for fluent chaining
|
||||
* @see ChunkMessageChannelItemWriter#setThrottleLimit(long)
|
||||
*/
|
||||
public RemoteChunkingManagerStepBuilder<I, O> throttleLimit(long throttleLimit) {
|
||||
Assert.isTrue(throttleLimit > 0, "throttleLimit must be greater than zero");
|
||||
this.throttleLimit = throttleLimit;
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* Build a manager {@link TaskletStep}.
|
||||
*
|
||||
* @return the configured manager step
|
||||
* @see RemoteChunkHandlerFactoryBean
|
||||
*/
|
||||
public TaskletStep build() {
|
||||
Assert.notNull(this.inputChannel, "An InputChannel 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");
|
||||
}
|
||||
}
|
||||
|
||||
// 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 RemoteChunkingManagerStepBuilder<I, O> reader(ItemReader<? extends I> reader) {
|
||||
super.reader(reader);
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public RemoteChunkingManagerStepBuilder<I, O> repository(JobRepository jobRepository) {
|
||||
super.repository(jobRepository);
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public RemoteChunkingManagerStepBuilder<I, O> transactionManager(PlatformTransactionManager transactionManager) {
|
||||
super.transactionManager(transactionManager);
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public RemoteChunkingManagerStepBuilder<I, O> listener(Object listener) {
|
||||
super.listener(listener);
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public RemoteChunkingManagerStepBuilder<I, O> listener(SkipListener<? super I, ? super O> listener) {
|
||||
super.listener(listener);
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public RemoteChunkingManagerStepBuilder<I, O> listener(ChunkListener listener) {
|
||||
super.listener(listener);
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public RemoteChunkingManagerStepBuilder<I, O> transactionAttribute(TransactionAttribute transactionAttribute) {
|
||||
super.transactionAttribute(transactionAttribute);
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public RemoteChunkingManagerStepBuilder<I, O> listener(org.springframework.retry.RetryListener listener) {
|
||||
super.listener(listener);
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public RemoteChunkingManagerStepBuilder<I, O> keyGenerator(KeyGenerator keyGenerator) {
|
||||
super.keyGenerator(keyGenerator);
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public RemoteChunkingManagerStepBuilder<I, O> retryLimit(int retryLimit) {
|
||||
super.retryLimit(retryLimit);
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public RemoteChunkingManagerStepBuilder<I, O> retryPolicy(RetryPolicy retryPolicy) {
|
||||
super.retryPolicy(retryPolicy);
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public RemoteChunkingManagerStepBuilder<I, O> backOffPolicy(BackOffPolicy backOffPolicy) {
|
||||
super.backOffPolicy(backOffPolicy);
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public RemoteChunkingManagerStepBuilder<I, O> retryContextCache(RetryContextCache retryContextCache) {
|
||||
super.retryContextCache(retryContextCache);
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public RemoteChunkingManagerStepBuilder<I, O> skipLimit(int skipLimit) {
|
||||
super.skipLimit(skipLimit);
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public RemoteChunkingManagerStepBuilder<I, O> noSkip(Class<? extends Throwable> type) {
|
||||
super.noSkip(type);
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public RemoteChunkingManagerStepBuilder<I, O> skip(Class<? extends Throwable> type) {
|
||||
super.skip(type);
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public RemoteChunkingManagerStepBuilder<I, O> skipPolicy(SkipPolicy skipPolicy) {
|
||||
super.skipPolicy(skipPolicy);
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public RemoteChunkingManagerStepBuilder<I, O> noRollback(Class<? extends Throwable> type) {
|
||||
super.noRollback(type);
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public RemoteChunkingManagerStepBuilder<I, O> noRetry(Class<? extends Throwable> type) {
|
||||
super.noRetry(type);
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public RemoteChunkingManagerStepBuilder<I, O> retry(Class<? extends Throwable> type) {
|
||||
super.retry(type);
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public RemoteChunkingManagerStepBuilder<I, O> stream(ItemStream stream) {
|
||||
super.stream(stream);
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public RemoteChunkingManagerStepBuilder<I, O> chunk(int chunkSize) {
|
||||
super.chunk(chunkSize);
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public RemoteChunkingManagerStepBuilder<I, O> chunk(CompletionPolicy completionPolicy) {
|
||||
super.chunk(completionPolicy);
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* This method will throw a {@link UnsupportedOperationException} since
|
||||
* the item writer of the manager step in a remote chunking setup will be
|
||||
* automatically set to an instance of {@link ChunkMessageChannelItemWriter}.
|
||||
*
|
||||
* When building a manager 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 RemoteChunkingManagerStepBuilder<I, O> writer(ItemWriter<? super O> writer) throws UnsupportedOperationException {
|
||||
throw new UnsupportedOperationException("When configuring a manager 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 RemoteChunkingManagerStepBuilder<I, O> readerIsTransactionalQueue() {
|
||||
super.readerIsTransactionalQueue();
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public RemoteChunkingManagerStepBuilder<I, O> listener(ItemReadListener<? super I> listener) {
|
||||
super.listener(listener);
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public RemoteChunkingManagerStepBuilder<I, O> listener(ItemWriteListener<? super O> listener) {
|
||||
super.listener(listener);
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public RemoteChunkingManagerStepBuilder<I, O> chunkOperations(RepeatOperations repeatTemplate) {
|
||||
super.chunkOperations(repeatTemplate);
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public RemoteChunkingManagerStepBuilder<I, O> exceptionHandler(ExceptionHandler exceptionHandler) {
|
||||
super.exceptionHandler(exceptionHandler);
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public RemoteChunkingManagerStepBuilder<I, O> stepOperations(RepeatOperations repeatTemplate) {
|
||||
super.stepOperations(repeatTemplate);
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public RemoteChunkingManagerStepBuilder<I, O> startLimit(int startLimit) {
|
||||
super.startLimit(startLimit);
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public RemoteChunkingManagerStepBuilder<I, O> listener(StepExecutionListener listener) {
|
||||
super.listener(listener);
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public RemoteChunkingManagerStepBuilder<I, O> allowStartIfComplete(boolean allowStartIfComplete) {
|
||||
super.allowStartIfComplete(allowStartIfComplete);
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public RemoteChunkingManagerStepBuilder<I, O> processor(ItemProcessor<? super I, ? extends O> itemProcessor) {
|
||||
super.processor(itemProcessor);
|
||||
return this;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,63 @@
|
||||
/*
|
||||
* Copyright 2019 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
|
||||
*
|
||||
* https://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 RemoteChunkingManagerStepBuilder} which sets
|
||||
* the {@link JobRepository} and {@link PlatformTransactionManager} automatically.
|
||||
*
|
||||
* @since 4.2
|
||||
* @author Mahmoud Ben Hassine
|
||||
*/
|
||||
public class RemoteChunkingManagerStepBuilderFactory {
|
||||
|
||||
private JobRepository jobRepository;
|
||||
|
||||
private PlatformTransactionManager transactionManager;
|
||||
|
||||
/**
|
||||
* Create a new {@link RemoteChunkingManagerStepBuilderFactory}.
|
||||
*
|
||||
* @param jobRepository the job repository to use
|
||||
* @param transactionManager the transaction manager to use
|
||||
*/
|
||||
public RemoteChunkingManagerStepBuilderFactory(
|
||||
JobRepository jobRepository,
|
||||
PlatformTransactionManager transactionManager) {
|
||||
|
||||
this.jobRepository = jobRepository;
|
||||
this.transactionManager = transactionManager;
|
||||
}
|
||||
|
||||
/**
|
||||
* Creates a {@link RemoteChunkingManagerStepBuilder} and initializes its job
|
||||
* repository and transaction manager.
|
||||
*
|
||||
* @param name the name of the step
|
||||
* @param <I> type of input items
|
||||
* @param <O> type of output items
|
||||
* @return a {@link RemoteChunkingManagerStepBuilder}
|
||||
*/
|
||||
public <I, O> RemoteChunkingManagerStepBuilder<I, O> get(String name) {
|
||||
return new RemoteChunkingManagerStepBuilder<I, O>(name)
|
||||
.repository(this.jobRepository)
|
||||
.transactionManager(this.transactionManager);
|
||||
}
|
||||
|
||||
}
|
||||
@@ -1,5 +1,5 @@
|
||||
/*
|
||||
* Copyright 2018 the original author or authors.
|
||||
* Copyright 2018-2019 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.
|
||||
@@ -59,9 +59,12 @@ import org.springframework.util.Assert;
|
||||
* @param <I> type of input items
|
||||
* @param <O> type of output items
|
||||
*
|
||||
* @deprecated Use {@link RemoteChunkingManagerStepBuilder} instead.
|
||||
*
|
||||
* @since 4.1
|
||||
* @author Mahmoud Ben Hassine
|
||||
*/
|
||||
@Deprecated
|
||||
public class RemoteChunkingMasterStepBuilder<I, O> extends FaultTolerantStepBuilder<I, O> {
|
||||
|
||||
private MessagingTemplate messagingTemplate;
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
/*
|
||||
* Copyright 2018 the original author or authors.
|
||||
* Copyright 2018-2019 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.
|
||||
@@ -22,9 +22,12 @@ import org.springframework.transaction.PlatformTransactionManager;
|
||||
* Convenient factory for a {@link RemoteChunkingMasterStepBuilder} which sets
|
||||
* the {@link JobRepository} and {@link PlatformTransactionManager} automatically.
|
||||
*
|
||||
* @deprecated Use {@link RemoteChunkingManagerStepBuilderFactory} instead.
|
||||
*
|
||||
* @since 4.1
|
||||
* @author Mahmoud Ben Hassine
|
||||
*/
|
||||
@Deprecated
|
||||
public class RemoteChunkingMasterStepBuilderFactory {
|
||||
|
||||
private JobRepository jobRepository;
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
/*
|
||||
* Copyright 2018 the original author or authors.
|
||||
* Copyright 2018-2019 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.
|
||||
@@ -18,8 +18,10 @@ package org.springframework.batch.integration.config.annotation;
|
||||
import org.springframework.batch.core.explore.JobExplorer;
|
||||
import org.springframework.batch.core.repository.JobRepository;
|
||||
import org.springframework.batch.integration.chunk.RemoteChunkingMasterStepBuilderFactory;
|
||||
import org.springframework.batch.integration.chunk.RemoteChunkingManagerStepBuilderFactory;
|
||||
import org.springframework.batch.integration.chunk.RemoteChunkingWorkerBuilder;
|
||||
import org.springframework.batch.integration.partition.RemotePartitioningMasterStepBuilderFactory;
|
||||
import org.springframework.batch.integration.partition.RemotePartitioningManagerStepBuilderFactory;
|
||||
import org.springframework.batch.integration.partition.RemotePartitioningWorkerStepBuilderFactory;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
@@ -52,23 +54,37 @@ public class BatchIntegrationConfiguration {
|
||||
this.transactionManager = transactionManager;
|
||||
}
|
||||
|
||||
@Deprecated
|
||||
@Bean
|
||||
public RemoteChunkingMasterStepBuilderFactory remoteChunkingMasterStepBuilderFactory() {
|
||||
return new RemoteChunkingMasterStepBuilderFactory(this.jobRepository,
|
||||
this.transactionManager);
|
||||
}
|
||||
|
||||
@Bean
|
||||
public RemoteChunkingManagerStepBuilderFactory remoteChunkingManagerStepBuilderFactory() {
|
||||
return new RemoteChunkingManagerStepBuilderFactory(this.jobRepository,
|
||||
this.transactionManager);
|
||||
}
|
||||
|
||||
@Bean
|
||||
public <I,O> RemoteChunkingWorkerBuilder<I, O> remoteChunkingWorkerBuilder() {
|
||||
return new RemoteChunkingWorkerBuilder<>();
|
||||
}
|
||||
|
||||
@Deprecated
|
||||
@Bean
|
||||
public RemotePartitioningMasterStepBuilderFactory remotePartitioningMasterStepBuilderFactory() {
|
||||
return new RemotePartitioningMasterStepBuilderFactory(this.jobRepository,
|
||||
this.jobExplorer, this.transactionManager);
|
||||
}
|
||||
|
||||
@Bean
|
||||
public RemotePartitioningManagerStepBuilderFactory remotePartitioningManagerStepBuilderFactory() {
|
||||
return new RemotePartitioningManagerStepBuilderFactory(this.jobRepository,
|
||||
this.jobExplorer, this.transactionManager);
|
||||
}
|
||||
|
||||
@Bean
|
||||
public RemotePartitioningWorkerStepBuilderFactory remotePartitioningWorkerStepBuilderFactory() {
|
||||
return new RemotePartitioningWorkerStepBuilderFactory(this.jobRepository,
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
/*
|
||||
* Copyright 2018 the original author or authors.
|
||||
* Copyright 2018-2019 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.
|
||||
@@ -22,8 +22,8 @@ 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.batch.integration.partition.RemotePartitioningMasterStepBuilderFactory;
|
||||
import org.springframework.batch.integration.chunk.RemoteChunkingManagerStepBuilderFactory;
|
||||
import org.springframework.batch.integration.partition.RemotePartitioningManagerStepBuilderFactory;
|
||||
import org.springframework.batch.integration.partition.RemotePartitioningWorkerStepBuilderFactory;
|
||||
import org.springframework.context.annotation.Import;
|
||||
import org.springframework.integration.config.EnableIntegration;
|
||||
@@ -36,13 +36,13 @@ import org.springframework.integration.config.EnableIntegration;
|
||||
* class, it will be possible to autowire the following beans:
|
||||
*
|
||||
* <ul>
|
||||
* <li>{@link RemoteChunkingMasterStepBuilderFactory}:
|
||||
* used to create a master step of a remote chunking setup by automatically
|
||||
* <li>{@link RemoteChunkingManagerStepBuilderFactory}:
|
||||
* used to create a manager step of a remote chunking setup by automatically
|
||||
* setting the job repository and transaction manager.</li>
|
||||
* <li>{@link RemoteChunkingWorkerBuilder}: used to create the integration
|
||||
* flow on the worker side of a remote chunking setup.</li>
|
||||
* <li>{@link RemotePartitioningMasterStepBuilderFactory}: used to create
|
||||
* a master step of a remote partitioning setup by automatically setting
|
||||
* <li>{@link RemotePartitioningManagerStepBuilderFactory}: used to create
|
||||
* a manager step of a remote partitioning setup by automatically setting
|
||||
* the job repository, job explorer, bean factory and transaction manager.</li>
|
||||
* <li>{@link RemotePartitioningWorkerStepBuilderFactory}: used to create
|
||||
* a worker step of a remote partitioning setup by automatically setting
|
||||
@@ -58,15 +58,15 @@ import org.springframework.integration.config.EnableIntegration;
|
||||
* public class RemoteChunkingAppConfig {
|
||||
*
|
||||
* @Autowired
|
||||
* private RemoteChunkingMasterStepBuilderFactory masterStepBuilderFactory;
|
||||
* private RemoteChunkingManagerStepBuilderFactory managerStepBuilderFactory;
|
||||
*
|
||||
* @Autowired
|
||||
* private RemoteChunkingWorkerBuilder workerBuilder;
|
||||
*
|
||||
* @Bean
|
||||
* public TaskletStep masterStep() {
|
||||
* return this.masterStepBuilderFactory
|
||||
* .get("masterStep")
|
||||
* public TaskletStep managerStep() {
|
||||
* return this.managerStepBuilderFactory
|
||||
* .get("managerStep")
|
||||
* .chunk(100)
|
||||
* .reader(itemReader())
|
||||
* .outputChannel(outgoingRequestsToWorkers())
|
||||
@@ -79,8 +79,8 @@ import org.springframework.integration.config.EnableIntegration;
|
||||
* return this.workerBuilder
|
||||
* .itemProcessor(itemProcessor())
|
||||
* .itemWriter(itemWriter())
|
||||
* .inputChannel(incomingRequestsFromMaster())
|
||||
* .outputChannel(outgoingRepliesToMaster())
|
||||
* .inputChannel(incomingRequestsFromManager())
|
||||
* .outputChannel(outgoingRepliesToManager())
|
||||
* .build();
|
||||
* }
|
||||
*
|
||||
@@ -98,15 +98,15 @@ import org.springframework.integration.config.EnableIntegration;
|
||||
* public class RemotePartitioningAppConfig {
|
||||
*
|
||||
* @Autowired
|
||||
* private RemotePartitioningMasterStepBuilderFactory masterStepBuilderFactory;
|
||||
* private RemotePartitioningManagerStepBuilderFactory managerStepBuilderFactory;
|
||||
*
|
||||
* @Autowired
|
||||
* private RemotePartitioningWorkerStepBuilderFactory workerStepBuilderFactory;
|
||||
*
|
||||
* @Bean
|
||||
* public Step masterStep() {
|
||||
* return this.masterStepBuilderFactory
|
||||
* .get("masterStep")
|
||||
* public Step managerStep() {
|
||||
* return this.managerStepBuilderFactory
|
||||
* .get("managerStep")
|
||||
* .partitioner("workerStep", partitioner())
|
||||
* .gridSize(10)
|
||||
* .outputChannel(outgoingRequestsToWorkers())
|
||||
@@ -118,8 +118,8 @@ import org.springframework.integration.config.EnableIntegration;
|
||||
* public Step workerStep() {
|
||||
* return this.workerStepBuilderFactory
|
||||
* .get("workerStep")
|
||||
* .inputChannel(incomingRequestsFromMaster())
|
||||
* .outputChannel(outgoingRepliesToMaster())
|
||||
* .inputChannel(incomingRequestsFromManager())
|
||||
* .outputChannel(outgoingRepliesToManager())
|
||||
* .chunk(100)
|
||||
* .reader(itemReader())
|
||||
* .processor(itemProcessor())
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
/*
|
||||
* Copyright 2002-2014 the original author or authors.
|
||||
* Copyright 2002-2019 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.
|
||||
@@ -22,6 +22,7 @@ import org.springframework.integration.config.xml.AbstractIntegrationNamespaceHa
|
||||
*
|
||||
* @author Gunnar Hillert
|
||||
* @author Chris Schaefer
|
||||
* @author Mahmoud Ben Hassine
|
||||
* @since 1.3
|
||||
*/
|
||||
public class BatchIntegrationNamespaceHandler extends AbstractIntegrationNamespaceHandler {
|
||||
@@ -30,7 +31,13 @@ public class BatchIntegrationNamespaceHandler extends AbstractIntegrationNamespa
|
||||
*/
|
||||
public void init() {
|
||||
this.registerBeanDefinitionParser("job-launching-gateway", new JobLaunchingGatewayParser());
|
||||
this.registerBeanDefinitionParser("remote-chunking-master", new RemoteChunkingMasterParser());
|
||||
this.registerBeanDefinitionParser("remote-chunking-slave", new RemoteChunkingSlaveParser());
|
||||
RemoteChunkingManagerParser remoteChunkingManagerParser = new RemoteChunkingManagerParser();
|
||||
this.registerBeanDefinitionParser("remote-chunking-manager", remoteChunkingManagerParser);
|
||||
RemoteChunkingWorkerParser remoteChunkingWorkerParser = new RemoteChunkingWorkerParser();
|
||||
this.registerBeanDefinitionParser("remote-chunking-worker", remoteChunkingWorkerParser);
|
||||
|
||||
// TODO remove the following when related deprecated APIs are removed
|
||||
this.registerBeanDefinitionParser("remote-chunking-master", remoteChunkingManagerParser);
|
||||
this.registerBeanDefinitionParser("remote-chunking-slave", remoteChunkingWorkerParser);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
/*
|
||||
* Copyright 2014 the original author or authors.
|
||||
* Copyright 2014-2019 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.
|
||||
@@ -28,13 +28,14 @@ import org.w3c.dom.Element;
|
||||
|
||||
/**
|
||||
* <p>
|
||||
* Parser for the remote-chunking-master namespace element.
|
||||
* Parser for the remote-chunking-manager namespace element.
|
||||
* </p>
|
||||
*
|
||||
* @author Chris Schaefer
|
||||
* @author Mahmoud Ben Hassine
|
||||
* @since 3.1
|
||||
*/
|
||||
public class RemoteChunkingMasterParser extends AbstractBeanDefinitionParser {
|
||||
public class RemoteChunkingManagerParser extends AbstractBeanDefinitionParser {
|
||||
private static final String MESSAGE_TEMPLATE_ATTRIBUTE = "message-template";
|
||||
private static final String STEP_ATTRIBUTE = "step";
|
||||
private static final String REPLY_CHANNEL_ATTRIBUTE = "reply-channel";
|
||||
@@ -1,5 +1,5 @@
|
||||
/*
|
||||
* Copyright 2014 the original author or authors.
|
||||
* Copyright 2014-2019 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.
|
||||
@@ -34,16 +34,17 @@ import org.springframework.util.StringUtils;
|
||||
|
||||
/**
|
||||
* <p>
|
||||
* Parser for the remote-chunking-slave namespace element. If an
|
||||
* Parser for the remote-chunking-worker namespace element. If an
|
||||
* {@link org.springframework.batch.item.ItemProcessor} is not provided, an
|
||||
* {@link org.springframework.batch.item.support.PassThroughItemProcessor} will be
|
||||
* configured.
|
||||
* </p>
|
||||
*
|
||||
* @author Chris Schaefer
|
||||
* @author Mahmoud Ben Hassine
|
||||
* @since 3.1
|
||||
*/
|
||||
public class RemoteChunkingSlaveParser extends AbstractBeanDefinitionParser {
|
||||
public class RemoteChunkingWorkerParser extends AbstractBeanDefinitionParser {
|
||||
private static final String INPUT_CHANNEL_ATTRIBUTE = "input-channel";
|
||||
private static final String OUTPUT_CHANNEL_ATTRIBUTE = "output-channel";
|
||||
private static final String ITEM_PROCESSOR_ATTRIBUTE = "item-processor";
|
||||
@@ -0,0 +1,305 @@
|
||||
/*
|
||||
* Copyright 2019 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
|
||||
*
|
||||
* https://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.partition;
|
||||
|
||||
import org.springframework.batch.core.Step;
|
||||
import org.springframework.batch.core.StepExecutionListener;
|
||||
import org.springframework.batch.core.explore.JobExplorer;
|
||||
import org.springframework.batch.core.partition.PartitionHandler;
|
||||
import org.springframework.batch.core.partition.StepExecutionSplitter;
|
||||
import org.springframework.batch.core.partition.support.Partitioner;
|
||||
import org.springframework.batch.core.partition.support.StepExecutionAggregator;
|
||||
import org.springframework.batch.core.repository.JobRepository;
|
||||
import org.springframework.batch.core.step.builder.PartitionStepBuilder;
|
||||
import org.springframework.batch.core.step.builder.StepBuilder;
|
||||
import org.springframework.beans.factory.BeanCreationException;
|
||||
import org.springframework.beans.factory.BeanFactory;
|
||||
import org.springframework.integration.channel.QueueChannel;
|
||||
import org.springframework.integration.core.MessagingTemplate;
|
||||
import org.springframework.integration.dsl.IntegrationFlows;
|
||||
import org.springframework.integration.dsl.StandardIntegrationFlow;
|
||||
import org.springframework.integration.dsl.context.IntegrationFlowContext;
|
||||
import org.springframework.messaging.MessageChannel;
|
||||
import org.springframework.messaging.PollableChannel;
|
||||
import org.springframework.transaction.PlatformTransactionManager;
|
||||
import org.springframework.util.Assert;
|
||||
|
||||
/**
|
||||
* Builder for a manager step in a remote partitioning setup. This builder creates and
|
||||
* sets a {@link MessageChannelPartitionHandler} on the manager step.
|
||||
*
|
||||
* <p>If no {@code messagingTemplate} is provided through
|
||||
* {@link RemotePartitioningManagerStepBuilder#messagingTemplate(MessagingTemplate)},
|
||||
* this builder will create one and set its default channel to the {@code outputChannel}
|
||||
* provided through {@link RemotePartitioningManagerStepBuilder#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.2
|
||||
* @author Mahmoud Ben Hassine
|
||||
*/
|
||||
public class RemotePartitioningManagerStepBuilder extends PartitionStepBuilder {
|
||||
|
||||
private static final long DEFAULT_POLL_INTERVAL = 10000L;
|
||||
private static final long DEFAULT_TIMEOUT = -1L;
|
||||
|
||||
private MessagingTemplate messagingTemplate;
|
||||
private MessageChannel inputChannel;
|
||||
private MessageChannel outputChannel;
|
||||
private JobExplorer jobExplorer;
|
||||
private BeanFactory beanFactory;
|
||||
private long pollInterval = DEFAULT_POLL_INTERVAL;
|
||||
private long timeout = DEFAULT_TIMEOUT;
|
||||
|
||||
/**
|
||||
* Create a new {@link RemotePartitioningManagerStepBuilder}.
|
||||
* @param stepName name of the manager step
|
||||
*/
|
||||
public RemotePartitioningManagerStepBuilder(String stepName) {
|
||||
super(new StepBuilder(stepName));
|
||||
}
|
||||
|
||||
/**
|
||||
* Set the input channel on which replies from workers will be received.
|
||||
* @param inputChannel the input channel
|
||||
* @return this builder instance for fluent chaining
|
||||
*/
|
||||
public RemotePartitioningManagerStepBuilder inputChannel(MessageChannel 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. 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 RemotePartitioningManagerStepBuilder#messagingTemplate(MessagingTemplate)}
|
||||
* to provide a fully configured messaging template.</p>
|
||||
*
|
||||
* @param outputChannel the output channel.
|
||||
* @return this builder instance for fluent chaining
|
||||
* @see RemotePartitioningManagerStepBuilder#messagingTemplate(MessagingTemplate)
|
||||
*/
|
||||
public RemotePartitioningManagerStepBuilder outputChannel(MessageChannel 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.
|
||||
* <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 RemotePartitioningManagerStepBuilder#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 RemotePartitioningManagerStepBuilder#outputChannel(MessageChannel)
|
||||
*/
|
||||
public RemotePartitioningManagerStepBuilder messagingTemplate(MessagingTemplate messagingTemplate) {
|
||||
Assert.notNull(messagingTemplate, "messagingTemplate must not be null");
|
||||
this.messagingTemplate = messagingTemplate;
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* Set the job explorer.
|
||||
* @param jobExplorer the job explorer to use.
|
||||
* @return this builder instance for fluent chaining
|
||||
*/
|
||||
public RemotePartitioningManagerStepBuilder jobExplorer(JobExplorer jobExplorer) {
|
||||
Assert.notNull(jobExplorer, "jobExplorer must not be null");
|
||||
this.jobExplorer = jobExplorer;
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* How often to poll the job repository for the status of the workers. Defaults to 10 seconds.
|
||||
* @param pollInterval the poll interval value in milliseconds
|
||||
* @return this builder instance for fluent chaining
|
||||
*/
|
||||
public RemotePartitioningManagerStepBuilder pollInterval(long pollInterval) {
|
||||
Assert.isTrue(pollInterval > 0, "The poll interval must be greater than zero");
|
||||
this.pollInterval = pollInterval;
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* When using job repository polling, the time limit to wait. Defaults to -1 (no timeout).
|
||||
* @param timeout the timeout value in milliseconds
|
||||
* @return this builder instance for fluent chaining
|
||||
*/
|
||||
public RemotePartitioningManagerStepBuilder timeout(long timeout) {
|
||||
this.timeout = timeout;
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* Set the bean factory.
|
||||
* @param beanFactory the bean factory to use
|
||||
* @return this builder instance for fluent chaining
|
||||
*/
|
||||
public RemotePartitioningManagerStepBuilder beanFactory(BeanFactory beanFactory) {
|
||||
this.beanFactory = beanFactory;
|
||||
return this;
|
||||
}
|
||||
|
||||
public Step build() {
|
||||
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");
|
||||
}
|
||||
}
|
||||
|
||||
// Configure the partition handler
|
||||
final MessageChannelPartitionHandler partitionHandler = new MessageChannelPartitionHandler();
|
||||
partitionHandler.setStepName(getStepName());
|
||||
partitionHandler.setGridSize(getGridSize());
|
||||
partitionHandler.setMessagingOperations(this.messagingTemplate);
|
||||
|
||||
if (isPolling()) {
|
||||
partitionHandler.setJobExplorer(this.jobExplorer);
|
||||
partitionHandler.setPollInterval(this.pollInterval);
|
||||
partitionHandler.setTimeout(this.timeout);
|
||||
}
|
||||
else {
|
||||
PollableChannel replies = new QueueChannel();
|
||||
partitionHandler.setReplyChannel(replies);
|
||||
StandardIntegrationFlow standardIntegrationFlow = IntegrationFlows
|
||||
.from(this.inputChannel)
|
||||
.aggregate(aggregatorSpec -> aggregatorSpec.processor(partitionHandler))
|
||||
.channel(replies)
|
||||
.get();
|
||||
IntegrationFlowContext integrationFlowContext = this.beanFactory.getBean(IntegrationFlowContext.class);
|
||||
integrationFlowContext.registration(standardIntegrationFlow)
|
||||
.autoStartup(false)
|
||||
.register();
|
||||
}
|
||||
|
||||
try {
|
||||
partitionHandler.afterPropertiesSet();
|
||||
super.partitionHandler(partitionHandler);
|
||||
}
|
||||
catch (Exception e) {
|
||||
throw new BeanCreationException("Unable to create a manager step for remote partitioning", e);
|
||||
}
|
||||
|
||||
return super.build();
|
||||
}
|
||||
|
||||
private boolean isPolling() {
|
||||
return this.inputChannel == null;
|
||||
}
|
||||
|
||||
@Override
|
||||
public RemotePartitioningManagerStepBuilder repository(JobRepository jobRepository) {
|
||||
super.repository(jobRepository);
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public RemotePartitioningManagerStepBuilder transactionManager(PlatformTransactionManager transactionManager) {
|
||||
super.transactionManager(transactionManager);
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public RemotePartitioningManagerStepBuilder partitioner(String slaveStepName, Partitioner partitioner) {
|
||||
super.partitioner(slaveStepName, partitioner);
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public RemotePartitioningManagerStepBuilder gridSize(int gridSize) {
|
||||
super.gridSize(gridSize);
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public RemotePartitioningManagerStepBuilder step(Step step) {
|
||||
super.step(step);
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public RemotePartitioningManagerStepBuilder splitter(StepExecutionSplitter splitter) {
|
||||
super.splitter(splitter);
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public RemotePartitioningManagerStepBuilder aggregator(StepExecutionAggregator aggregator) {
|
||||
super.aggregator(aggregator);
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public RemotePartitioningManagerStepBuilder startLimit(int startLimit) {
|
||||
super.startLimit(startLimit);
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public RemotePartitioningManagerStepBuilder listener(Object listener) {
|
||||
super.listener(listener);
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public RemotePartitioningManagerStepBuilder listener(StepExecutionListener listener) {
|
||||
super.listener(listener);
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public RemotePartitioningManagerStepBuilder allowStartIfComplete(boolean allowStartIfComplete) {
|
||||
super.allowStartIfComplete(allowStartIfComplete);
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* This method will throw a {@link UnsupportedOperationException} since
|
||||
* the partition handler of the manager step will be automatically set to an
|
||||
* instance of {@link MessageChannelPartitionHandler}.
|
||||
*
|
||||
* When building a manager step for remote partitioning using this builder,
|
||||
* no partition handler must be provided.
|
||||
*
|
||||
* @param partitionHandler a partition handler
|
||||
* @return this builder instance for fluent chaining
|
||||
* @throws UnsupportedOperationException if a partition handler is provided
|
||||
*/
|
||||
@Override
|
||||
public RemotePartitioningManagerStepBuilder partitionHandler(PartitionHandler partitionHandler) throws UnsupportedOperationException {
|
||||
throw new UnsupportedOperationException("When configuring a manager step " +
|
||||
"for remote partitioning using the RemotePartitioningManagerStepBuilder, " +
|
||||
"the partition handler will be automatically set to an instance " +
|
||||
"of MessageChannelPartitionHandler. The partition handler must " +
|
||||
"not be provided in this case.");
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,75 @@
|
||||
/*
|
||||
* Copyright 2019 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
|
||||
*
|
||||
* https://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.partition;
|
||||
|
||||
import org.springframework.batch.core.explore.JobExplorer;
|
||||
import org.springframework.batch.core.repository.JobRepository;
|
||||
import org.springframework.beans.BeansException;
|
||||
import org.springframework.beans.factory.BeanFactory;
|
||||
import org.springframework.beans.factory.BeanFactoryAware;
|
||||
import org.springframework.transaction.PlatformTransactionManager;
|
||||
|
||||
/**
|
||||
* Convenient factory for a {@link RemotePartitioningManagerStepBuilder} which sets
|
||||
* the {@link JobRepository}, {@link JobExplorer}, {@link BeanFactory} and
|
||||
* {@link PlatformTransactionManager} automatically.
|
||||
*
|
||||
* @since 4.2
|
||||
* @author Mahmoud Ben Hassine
|
||||
*/
|
||||
public class RemotePartitioningManagerStepBuilderFactory implements BeanFactoryAware {
|
||||
|
||||
private BeanFactory beanFactory;
|
||||
final private JobExplorer jobExplorer;
|
||||
final private JobRepository jobRepository;
|
||||
final private PlatformTransactionManager transactionManager;
|
||||
|
||||
|
||||
/**
|
||||
* Create a new {@link RemotePartitioningManagerStepBuilderFactory}.
|
||||
* @param jobRepository the job repository to use
|
||||
* @param jobExplorer the job explorer to use
|
||||
* @param transactionManager the transaction manager to use
|
||||
*/
|
||||
public RemotePartitioningManagerStepBuilderFactory(JobRepository jobRepository,
|
||||
JobExplorer jobExplorer, PlatformTransactionManager transactionManager) {
|
||||
|
||||
this.jobRepository = jobRepository;
|
||||
this.jobExplorer = jobExplorer;
|
||||
this.transactionManager = transactionManager;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
|
||||
this.beanFactory = beanFactory;
|
||||
}
|
||||
|
||||
/**
|
||||
* Creates a {@link RemotePartitioningManagerStepBuilder} and initializes its job
|
||||
* repository, job explorer, bean factory and transaction manager.
|
||||
* @param name the name of the step
|
||||
* @return a {@link RemotePartitioningManagerStepBuilder}
|
||||
*/
|
||||
public RemotePartitioningManagerStepBuilder get(String name) {
|
||||
return new RemotePartitioningManagerStepBuilder(name)
|
||||
.repository(this.jobRepository)
|
||||
.jobExplorer(this.jobExplorer)
|
||||
.beanFactory(this.beanFactory)
|
||||
.transactionManager(this.transactionManager);
|
||||
}
|
||||
|
||||
}
|
||||
@@ -1,5 +1,5 @@
|
||||
/*
|
||||
* Copyright 2018 the original author or authors.
|
||||
* Copyright 2018-2019 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.
|
||||
@@ -51,9 +51,12 @@ import org.springframework.util.Assert;
|
||||
* and that its default channel is set to an output channel on which requests to workers
|
||||
* will be sent.</p>
|
||||
*
|
||||
* @deprecated Use {@link RemotePartitioningManagerStepBuilder} instead.
|
||||
*
|
||||
* @since 4.1
|
||||
* @author Mahmoud Ben Hassine
|
||||
*/
|
||||
@Deprecated
|
||||
public class RemotePartitioningMasterStepBuilder extends PartitionStepBuilder {
|
||||
|
||||
private static final long DEFAULT_POLL_INTERVAL = 10000L;
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
/*
|
||||
* Copyright 2018 the original author or authors.
|
||||
* Copyright 2018-2019 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.
|
||||
@@ -28,9 +28,12 @@ import org.springframework.transaction.PlatformTransactionManager;
|
||||
* the {@link JobRepository}, {@link JobExplorer}, {@link BeanFactory} and
|
||||
* {@link PlatformTransactionManager} automatically.
|
||||
*
|
||||
* @deprecated Use {@link RemotePartitioningManagerStepBuilderFactory} instead
|
||||
*
|
||||
* @since 4.1
|
||||
* @author Mahmoud Ben Hassine
|
||||
*/
|
||||
@Deprecated
|
||||
public class RemotePartitioningMasterStepBuilderFactory implements BeanFactoryAware {
|
||||
|
||||
private BeanFactory beanFactory;
|
||||
|
||||
Reference in New Issue
Block a user