Spring/spring-batch
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Refactor AMQP sample to use TestContainers for RabbitMQ

Browse Source 
Issue #4052
This commit is contained in:
Mahmoud Ben Hassine
2023-09-21 12:54:35 +02:00
parent 008e582f9f
commit e708bf23a1
3 changed files with 84 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files

57
spring-batch-samples/src/main/java/org/springframework/batch/sample/amqp/AmqpConfiguration.java → spring-batch-samples/src/main/java/org/springframework/batch/sample/amqp/AmqpJobConfiguration.java
View File

@@ -18,10 +18,7 @@ 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;
@@ -42,18 +39,11 @@ import org.springframework.jdbc.support.JdbcTransactionManager;
* Sample Configuration to demonstrate a simple reader and writer for AMQP.
*
* @author Glenn Renfro
* @author Mahmoud Ben Hassine
*/
@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";
public class AmqpJobConfiguration {
@Bean
public Job job(JobRepository jobRepository, Step step) {
@@ -85,55 +75,24 @@ public class AmqpConfiguration {
/**
* Reads from the designated queue.
* @param template the template to be used by the {@link ItemReader}.
* @param rabbitInputTemplate the template to be used by the {@link ItemReader}.
* @return instance of {@link ItemReader}.
*/
@Bean
public ItemReader<String> amqpItemReader(RabbitTemplate template) {
public ItemReader<String> amqpItemReader(RabbitTemplate rabbitInputTemplate) {
AmqpItemReaderBuilder<String> builder = new AmqpItemReaderBuilder<>();
return builder.amqpTemplate(template).build();
return builder.amqpTemplate(rabbitInputTemplate).build();
}
/**
* Reads from the designated destination.
* @param template the template to be used by the {@link ItemWriter}.
* @param rabbitOutputTemplate the template to be used by the {@link ItemWriter}.
* @return instance of {@link ItemWriter}.
*/
@Bean
public ItemWriter<String> amqpItemWriter(RabbitTemplate template) {
public ItemWriter<String> amqpItemWriter(RabbitTemplate rabbitOutputTemplate) {
AmqpItemWriterBuilder<String> 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;
return builder.amqpTemplate(rabbitOutputTemplate).build();
}
}

73
spring-batch-samples/src/test/java/org/springframework/batch/sample/AMQPJobFunctionalTests.java
View File

@@ -16,20 +16,31 @@
package org.springframework.batch.sample;
import org.junit.jupiter.api.Test;
import org.testcontainers.containers.RabbitMQContainer;
import org.testcontainers.junit.jupiter.Container;
import org.testcontainers.junit.jupiter.Testcontainers;
import org.testcontainers.utility.DockerImageName;
import org.springframework.amqp.core.AmqpAdmin;
import org.springframework.amqp.core.AmqpTemplate;
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.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
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.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.sample.amqp.AmqpJobConfiguration;
import org.springframework.batch.test.JobLauncherTestUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -52,8 +63,14 @@ import org.springframework.context.annotation.AnnotationConfigApplicationContext
@SpringJUnitConfig(
locations = { "/simple-job-launcher-context.xml", "/jobs/amqp-example-job.xml", "/job-runner-context.xml" })
@Testcontainers(disabledWithoutDocker = true)
class AMQPJobFunctionalTests {
private static final DockerImageName RABBITMQ_IMAGE = DockerImageName.parse("rabbitmq:3");
@Container
public static RabbitMQContainer rabbitmq = new RabbitMQContainer(RABBITMQ_IMAGE);
@Autowired
private JobLauncherTestUtils jobLauncherTestUtils;
@@ -75,8 +92,8 @@ class AMQPJobFunctionalTests {
@Test
public void testLaunchJobWithJavaConfig() throws Exception {
// given
ApplicationContext context = new AnnotationConfigApplicationContext(AmqpConfiguration.class);
initializeExchange(context.getBean(CachingConnectionFactory.class));
ApplicationContext context = new AnnotationConfigApplicationContext(AmqpJobConfiguration.class,
AmqpConfiguration.class);
JobLauncher jobLauncher = context.getBean(JobLauncher.class);
Job job = context.getBean(Job.class);
@@ -89,12 +106,50 @@ class AMQPJobFunctionalTests {
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));
@Configuration
static class AmqpConfiguration {
public final static String QUEUE_NAME = "rabbitmq.test.queue";
public final static String EXCHANGE_NAME = "rabbitmq.test.exchange";
/**
* @return {@link CachingConnectionFactory} to be used by the {@link AmqpTemplate}
*/
@Bean
public CachingConnectionFactory connectionFactory() {
CachingConnectionFactory connectionFactory = new CachingConnectionFactory(rabbitmq.getHost(),
rabbitmq.getAmqpPort());
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));
return connectionFactory;
}
/**
* @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) {
RabbitTemplate template = new RabbitTemplate(connectionFactory);
template.setMessageConverter(new Jackson2JsonMessageConverter());
template.setDefaultReceiveQueue(QUEUE_NAME);
return template;
}
}
}