From 7f0d06fcc667d7fcd0b42077870919fa078f1ab6 Mon Sep 17 00:00:00 2001 From: Glenn Renfro Date: Thu, 3 Feb 2022 17:24:38 -0500 Subject: [PATCH] Add sample for AMQP Reader and writer using java configuration Issue #3663 --- spring-batch-samples/README.md | 10 ++ .../batch/sample/amqp/AmqpConfiguration.java | 139 ++++++++++++++++++ .../batch/sample/AMQPJobFunctionalTests.java | 40 ++++- 3 files changed, 187 insertions(+), 2 deletions(-) create mode 100644 spring-batch-samples/src/main/java/org/springframework/batch/sample/amqp/AmqpConfiguration.java diff --git a/spring-batch-samples/README.md b/spring-batch-samples/README.md index 90b851f94..ec763cbcc 100644 --- a/spring-batch-samples/README.md +++ b/spring-batch-samples/README.md @@ -129,6 +129,16 @@ and running. The standard dashboard can be used to see the traffic from the `MessageProducer` to the `AmqpItemWriter`. Make sure you launch the `MessageProducer` before launching the test. +You can run the sample from the command line as following: + +``` +cd spring-batch-samples +# Launch the test using the XML configuration +../mvnw -Dtest=AMQPJobFunctionalTests#testLaunchJobWithXmlConfig test +# Launch the test using the Java configuration +../mvnw -Dtest=AMQPJobFunctionalTests#testLaunchJobWithJavaConfig test +``` + ### BeanWrapperMapper Sample This sample shows the use of automatic mapping from fields in a file diff --git a/spring-batch-samples/src/main/java/org/springframework/batch/sample/amqp/AmqpConfiguration.java b/spring-batch-samples/src/main/java/org/springframework/batch/sample/amqp/AmqpConfiguration.java new file mode 100644 index 000000000..9939103d8 --- /dev/null +++ b/spring-batch-samples/src/main/java/org/springframework/batch/sample/amqp/AmqpConfiguration.java @@ -0,0 +1,139 @@ +/* + * Copyright 2023 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.sample.amqp; + +import javax.sql.DataSource; + +import org.springframework.amqp.core.AmqpTemplate; +import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; +import org.springframework.amqp.rabbit.core.RabbitTemplate; +import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter; +import org.springframework.batch.core.Job; +import org.springframework.batch.core.Step; +import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing; +import org.springframework.batch.core.job.builder.JobBuilder; +import org.springframework.batch.core.repository.JobRepository; +import org.springframework.batch.core.step.builder.StepBuilder; +import org.springframework.batch.item.ItemReader; +import org.springframework.batch.item.ItemWriter; +import org.springframework.batch.item.amqp.builder.AmqpItemReaderBuilder; +import org.springframework.batch.item.amqp.builder.AmqpItemWriterBuilder; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseBuilder; +import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseType; +import org.springframework.jdbc.support.JdbcTransactionManager; + +/** + * Sample Configuration to demonstrate a simple reader and writer for AMQP. + * + * @author Glenn Renfro + */ +@Configuration +@EnableBatchProcessing +public class AmqpConfiguration { + + public final static String QUEUE_NAME = "rabbitmq.test.queue"; + + public final static String EXCHANGE_NAME = "rabbitmq.test.exchange"; + + private final static int amqpPort = 5672; + + private final static String host = "127.0.0.1"; + + @Bean + public Job job(JobRepository jobRepository, Step step) { + return new JobBuilder("amqp-config-job", jobRepository).start(step).build(); + } + + @Bean + public Step step(JobRepository jobRepository, JdbcTransactionManager transactionManager, + RabbitTemplate rabbitInputTemplate, RabbitTemplate rabbitOutputTemplate) { + return new StepBuilder("step", jobRepository).chunk(1, transactionManager) + .reader(amqpItemReader(rabbitInputTemplate)) + .writer(amqpItemWriter(rabbitOutputTemplate)) + .build(); + } + + @Bean + public DataSource dataSource() { + return new EmbeddedDatabaseBuilder().setType(EmbeddedDatabaseType.HSQL) + .addScript("/org/springframework/batch/core/schema-hsqldb.sql") + .addScript("/business-schema-hsqldb.sql") + .generateUniqueName(true) + .build(); + } + + @Bean + public JdbcTransactionManager transactionManager(DataSource dataSource) { + return new JdbcTransactionManager(dataSource); + } + + /** + * Reads from the designated queue. + * @param template the template to be used by the {@link ItemReader}. + * @return instance of {@link ItemReader}. + */ + @Bean + public ItemReader amqpItemReader(RabbitTemplate template) { + AmqpItemReaderBuilder builder = new AmqpItemReaderBuilder<>(); + return builder.amqpTemplate(template).build(); + } + + /** + * Reads from the designated destination. + * @param template the template to be used by the {@link ItemWriter}. + * @return instance of {@link ItemWriter}. + */ + @Bean + public ItemWriter amqpItemWriter(RabbitTemplate template) { + AmqpItemWriterBuilder builder = new AmqpItemWriterBuilder<>(); + return builder.amqpTemplate(template).build(); + } + + /** + * @return {@link CachingConnectionFactory} to be used by the {@link AmqpTemplate} + */ + @Bean + public CachingConnectionFactory connectionFactory() { + return new CachingConnectionFactory(host, amqpPort); + } + + /** + * @return {@link AmqpTemplate} to be used for the {@link ItemWriter} + */ + @Bean + public AmqpTemplate rabbitOutputTemplate(CachingConnectionFactory connectionFactory) { + RabbitTemplate template = new RabbitTemplate(connectionFactory); + template.setMessageConverter(new Jackson2JsonMessageConverter()); + template.setExchange(EXCHANGE_NAME); + return template; + } + + /** + * @return {@link AmqpTemplate} to be used for the {@link ItemReader}. + */ + @Bean + public RabbitTemplate rabbitInputTemplate() { + CachingConnectionFactory connectionFactory = new CachingConnectionFactory(host, amqpPort); + RabbitTemplate template = new RabbitTemplate(connectionFactory); + template.setMessageConverter(new Jackson2JsonMessageConverter()); + template.setDefaultReceiveQueue(QUEUE_NAME); + return template; + } + +} diff --git a/spring-batch-samples/src/test/java/org/springframework/batch/sample/AMQPJobFunctionalTests.java b/spring-batch-samples/src/test/java/org/springframework/batch/sample/AMQPJobFunctionalTests.java index c283a9ab2..e56c1e530 100644 --- a/spring-batch-samples/src/test/java/org/springframework/batch/sample/AMQPJobFunctionalTests.java +++ b/spring-batch-samples/src/test/java/org/springframework/batch/sample/AMQPJobFunctionalTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2012-2022 the original author or authors. + * Copyright 2012-2023 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. @@ -17,12 +17,24 @@ package org.springframework.batch.sample; import org.junit.jupiter.api.Test; +import org.springframework.amqp.core.AmqpAdmin; +import org.springframework.amqp.core.Binding; +import org.springframework.amqp.core.Queue; +import org.springframework.amqp.core.TopicExchange; +import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; +import org.springframework.amqp.rabbit.core.RabbitAdmin; +import org.springframework.batch.core.Job; +import org.springframework.batch.core.JobParameters; import org.springframework.batch.core.explore.JobExplorer; +import org.springframework.batch.core.launch.JobLauncher; +import org.springframework.batch.sample.amqp.AmqpConfiguration; import org.springframework.batch.test.JobLauncherTestUtils; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; import static org.junit.jupiter.api.Assertions.assertTrue; +import org.springframework.context.ApplicationContext; +import org.springframework.context.annotation.AnnotationConfigApplicationContext; /** *

@@ -49,7 +61,7 @@ class AMQPJobFunctionalTests { private JobExplorer jobExplorer; @Test - void testLaunchJob() throws Exception { + void testLaunchJobWithXmlConfig() throws Exception { // given this.jobLauncherTestUtils.launchJob(); @@ -58,7 +70,31 @@ class AMQPJobFunctionalTests { // then assertTrue(count > 0); + } + @Test + public void testLaunchJobWithJavaConfig() throws Exception { + // given + ApplicationContext context = new AnnotationConfigApplicationContext(AmqpConfiguration.class); + initializeExchange(context.getBean(CachingConnectionFactory.class)); + JobLauncher jobLauncher = context.getBean(JobLauncher.class); + Job job = context.getBean(Job.class); + + // when + jobLauncher.run(job, new JobParameters()); + + // then + JobExplorer localJobExplorer = context.getBean(JobExplorer.class); + int count = localJobExplorer.getJobInstances("amqp-config-job", 0, 1).size(); + assertTrue(count > 0); + } + + private void initializeExchange(CachingConnectionFactory connectionFactory) { + AmqpAdmin admin = new RabbitAdmin(connectionFactory); + admin.declareQueue(new Queue(AmqpConfiguration.QUEUE_NAME)); + admin.declareExchange(new TopicExchange(AmqpConfiguration.EXCHANGE_NAME)); + admin.declareBinding(new Binding(AmqpConfiguration.QUEUE_NAME, Binding.DestinationType.QUEUE, + AmqpConfiguration.EXCHANGE_NAME, "#", null)); } }