Add a remote chunking sample
This commit adds a self contained remote chunking sample that uses an embedded JMS broker (for simplicity). Even though the broker is embedded, communication between the master and workers is still done through JMS queues and Spring Integration channels and messages are sent over the wire through a TCP port. This commit also fixes a few typos in the documentation section about remote chunking. Resolves BATCH-2721
This commit is contained in:
Mahmoud Ben Hassine
committed by
Michael Minella
Michael Minella
parent
c2b0a208c3
commit
026bbe056b
@@ -0,0 +1,139 @@
|
||||
/*
|
||||
* 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.sample.remotechunking;
|
||||
|
||||
import java.util.Arrays;
|
||||
|
||||
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.item.support.ListItemReader;
|
||||
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.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;
|
||||
|
||||
/**
|
||||
* This configuration class is for the master side of the remote chunking sample.
|
||||
* The master step reads numbers from 1 to 6 and sends 2 chunks {1, 2, 3} and
|
||||
* {4, 5, 6} to workers for processing and writing.
|
||||
*
|
||||
* @author Mahmoud Ben Hassine
|
||||
*/
|
||||
@Configuration
|
||||
@EnableBatchProcessing
|
||||
@EnableIntegration
|
||||
@PropertySource("classpath:remote-chunking.properties")
|
||||
public class MasterConfiguration {
|
||||
|
||||
@Value("${broker.url}")
|
||||
private String brokerUrl;
|
||||
|
||||
@Autowired
|
||||
private JobBuilderFactory jobBuilderFactory;
|
||||
|
||||
@Autowired
|
||||
private StepBuilderFactory stepBuilderFactory;
|
||||
|
||||
@Bean
|
||||
public ActiveMQConnectionFactory connectionFactory() {
|
||||
ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory();
|
||||
connectionFactory.setBrokerURL(this.brokerUrl);
|
||||
connectionFactory.setTrustAllPackages(true);
|
||||
return connectionFactory;
|
||||
}
|
||||
|
||||
/*
|
||||
* Configure outbound flow (requests going to workers)
|
||||
*/
|
||||
@Bean
|
||||
public DirectChannel requests() {
|
||||
return new DirectChannel();
|
||||
}
|
||||
|
||||
@Bean
|
||||
public IntegrationFlow outboundFlow(ActiveMQConnectionFactory connectionFactory) {
|
||||
return IntegrationFlows
|
||||
.from(requests())
|
||||
.handle(Jms.outboundAdapter(connectionFactory).destination("requests"))
|
||||
.get();
|
||||
}
|
||||
|
||||
/*
|
||||
* Configure inbound flow (replies coming from workers)
|
||||
*/
|
||||
@Bean
|
||||
public QueueChannel replies() {
|
||||
return new QueueChannel();
|
||||
}
|
||||
|
||||
@Bean
|
||||
public IntegrationFlow inboundFlow(ActiveMQConnectionFactory connectionFactory) {
|
||||
return IntegrationFlows
|
||||
.from(Jms.messageDrivenChannelAdapter(connectionFactory).destination("replies"))
|
||||
.channel(replies())
|
||||
.get();
|
||||
}
|
||||
|
||||
/*
|
||||
* Configure master step components
|
||||
*/
|
||||
@Bean
|
||||
public ListItemReader<Integer> itemReader() {
|
||||
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")
|
||||
.<Integer, Integer>chunk(3)
|
||||
.reader(itemReader())
|
||||
.writer(itemWriter())
|
||||
.build();
|
||||
}
|
||||
|
||||
@Bean
|
||||
public Job remoteChunkingJob() {
|
||||
return this.jobBuilderFactory.get("remoteChunkingJob")
|
||||
.start(masterStep())
|
||||
.build();
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,126 @@
|
||||
/*
|
||||
* 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.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.item.ItemProcessor;
|
||||
import org.springframework.batch.item.ItemWriter;
|
||||
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;
|
||||
import org.springframework.integration.dsl.IntegrationFlows;
|
||||
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:
|
||||
* <ul>
|
||||
* <li>receive requests from the master</li>
|
||||
* <li>process chunks with the configured item processor and writer</li>
|
||||
* <li>send replies to the master</li>
|
||||
* </ul>
|
||||
*
|
||||
* @author Mahmoud Ben Hassine
|
||||
*/
|
||||
@Configuration
|
||||
@EnableBatchProcessing
|
||||
@EnableIntegration
|
||||
@PropertySource("classpath:remote-chunking.properties")
|
||||
public class WorkerConfiguration {
|
||||
|
||||
@Value("${broker.url}")
|
||||
private String brokerUrl;
|
||||
|
||||
@Bean
|
||||
public ActiveMQConnectionFactory connectionFactory() {
|
||||
ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory();
|
||||
connectionFactory.setBrokerURL(this.brokerUrl);
|
||||
connectionFactory.setTrustAllPackages(true);
|
||||
return connectionFactory;
|
||||
}
|
||||
|
||||
/*
|
||||
* Configure inbound flow (requests coming from the master)
|
||||
*/
|
||||
@Bean
|
||||
public DirectChannel requests() {
|
||||
return new DirectChannel();
|
||||
}
|
||||
|
||||
@Bean
|
||||
public IntegrationFlow inboundFlow(ActiveMQConnectionFactory connectionFactory) {
|
||||
return IntegrationFlows
|
||||
.from(Jms.messageDrivenChannelAdapter(connectionFactory).destination("requests"))
|
||||
.channel(requests())
|
||||
.get();
|
||||
}
|
||||
|
||||
/*
|
||||
* Configure outbound flow (replies going to the master)
|
||||
*/
|
||||
@Bean
|
||||
public DirectChannel replies() {
|
||||
return new DirectChannel();
|
||||
}
|
||||
|
||||
@Bean
|
||||
public IntegrationFlow outboundFlow(ActiveMQConnectionFactory connectionFactory) {
|
||||
return IntegrationFlows
|
||||
.from(replies())
|
||||
.handle(Jms.outboundAdapter(connectionFactory).destination("replies"))
|
||||
.get();
|
||||
}
|
||||
|
||||
/*
|
||||
* Configure worker components
|
||||
*/
|
||||
@Bean
|
||||
public ItemProcessor<Integer, Integer> itemProcessor() {
|
||||
return item -> {
|
||||
System.out.println("processing item " + item);
|
||||
return item;
|
||||
};
|
||||
}
|
||||
|
||||
@Bean
|
||||
public ItemWriter<Integer> itemWriter() {
|
||||
return items -> {
|
||||
for (Integer item : items) {
|
||||
System.out.println("writing item " + item);
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
@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;
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1 @@
|
||||
broker.url=tcp://localhost:61616
|
||||
@@ -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.sample;
|
||||
|
||||
import org.apache.activemq.broker.BrokerService;
|
||||
import org.junit.After;
|
||||
import org.junit.Assert;
|
||||
import org.junit.Before;
|
||||
import org.junit.Test;
|
||||
import org.junit.runner.RunWith;
|
||||
|
||||
import org.springframework.batch.core.ExitStatus;
|
||||
import org.springframework.batch.core.JobExecution;
|
||||
import org.springframework.batch.sample.config.JobRunnerConfiguration;
|
||||
import org.springframework.batch.sample.remotechunking.MasterConfiguration;
|
||||
import org.springframework.batch.sample.remotechunking.WorkerConfiguration;
|
||||
import org.springframework.batch.test.JobLauncherTestUtils;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.beans.factory.annotation.Value;
|
||||
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
|
||||
import org.springframework.context.annotation.PropertySource;
|
||||
import org.springframework.test.context.ContextConfiguration;
|
||||
import org.springframework.test.context.junit4.SpringRunner;
|
||||
|
||||
/**
|
||||
* The master step of the job under test will read data and send chunks to the worker
|
||||
* (started in {@link RemoteChunkingJobFunctionalTests#setUp()}) for processing and writing.
|
||||
*
|
||||
* @author Mahmoud Ben Hassine
|
||||
*/
|
||||
@RunWith(SpringRunner.class)
|
||||
@ContextConfiguration(classes = {JobRunnerConfiguration.class, MasterConfiguration.class})
|
||||
@PropertySource("classpath:remote-chunking.properties")
|
||||
public class RemoteChunkingJobFunctionalTests {
|
||||
|
||||
private static final String BROKER_DATA_DIRECTORY = "build/activemq-data";
|
||||
|
||||
@Value("${broker.url}")
|
||||
private String brokerUrl;
|
||||
|
||||
@Autowired
|
||||
private JobLauncherTestUtils jobLauncherTestUtils;
|
||||
|
||||
private BrokerService brokerService;
|
||||
|
||||
private AnnotationConfigApplicationContext workerApplicationContext;
|
||||
|
||||
@Before
|
||||
public void setUp() throws Exception {
|
||||
this.brokerService = new BrokerService();
|
||||
this.brokerService.addConnector(this.brokerUrl);
|
||||
this.brokerService.setDataDirectory(BROKER_DATA_DIRECTORY);
|
||||
this.brokerService.start();
|
||||
this.workerApplicationContext = new AnnotationConfigApplicationContext(WorkerConfiguration.class);
|
||||
}
|
||||
|
||||
@After
|
||||
public void tearDown() throws Exception {
|
||||
this.workerApplicationContext.close();
|
||||
this.brokerService.stop();
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testRemoteChunkingJob() throws Exception {
|
||||
// when
|
||||
JobExecution jobExecution = this.jobLauncherTestUtils.launchJob();
|
||||
|
||||
// then
|
||||
Assert.assertEquals(ExitStatus.COMPLETED.getExitCode(), jobExecution.getExitStatus().getExitCode());
|
||||
Assert.assertEquals(
|
||||
"Waited for 2 results.", // the master sent 2 chunks ({1, 2, 3} and {4, 5, 6}) to workers
|
||||
jobExecution.getExitStatus().getExitDescription());
|
||||
}
|
||||
|
||||
}
|
||||
Reference in New Issue
Block a user