diff --git a/spring-cloud-starter-single-step-batch-job/pom.xml b/spring-cloud-starter-single-step-batch-job/pom.xml index 8e223c1e..58db850e 100644 --- a/spring-cloud-starter-single-step-batch-job/pom.xml +++ b/spring-cloud-starter-single-step-batch-job/pom.xml @@ -50,6 +50,7 @@ org.springframework.amqp spring-amqp + true org.springframework.amqp diff --git a/spring-cloud-starter-single-step-batch-job/src/main/java/org/springframework/cloud/task/batch/autoconfigure/rabbit/AmqpItemWriterAutoConfiguration.java b/spring-cloud-starter-single-step-batch-job/src/main/java/org/springframework/cloud/task/batch/autoconfigure/rabbit/AmqpItemWriterAutoConfiguration.java new file mode 100644 index 00000000..982a1612 --- /dev/null +++ b/spring-cloud-starter-single-step-batch-job/src/main/java/org/springframework/cloud/task/batch/autoconfigure/rabbit/AmqpItemWriterAutoConfiguration.java @@ -0,0 +1,64 @@ +/* + * Copyright 2020-2020 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.cloud.task.batch.autoconfigure.rabbit; + +import java.util.Map; + +import org.springframework.amqp.core.AmqpTemplate; +import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter; +import org.springframework.amqp.support.converter.MessageConverter; +import org.springframework.batch.item.amqp.AmqpItemWriter; +import org.springframework.batch.item.amqp.builder.AmqpItemWriterBuilder; +import org.springframework.boot.autoconfigure.AutoConfigureAfter; +import org.springframework.boot.autoconfigure.batch.BatchAutoConfiguration; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; + +/** + * Autconfiguration for a {@code AmqpItemWriter}. + * + * @author Glenn Renfro + * @since 2.3 + */ +@Configuration +@EnableConfigurationProperties(AmqpItemWriterProperties.class) +@AutoConfigureAfter(BatchAutoConfiguration.class) +@ConditionalOnProperty(name = "spring.batch.job.amqpitemwriter.enabled", + havingValue = "true", matchIfMissing = false) +public class AmqpItemWriterAutoConfiguration { + + @Bean + public AmqpItemWriter> amqpItemWriter(AmqpTemplate amqpTemplate) { + return new AmqpItemWriterBuilder>().amqpTemplate(amqpTemplate) + .build(); + } + + @Bean + public AmqpItemWriterProperties amqpItemWriterProperties() { + return new AmqpItemWriterProperties(); + } + + @ConditionalOnProperty(name = "spring.batch.job.amqpitemwriter.jsonConverterEnabled", + havingValue = "true", matchIfMissing = true) + @Bean + public MessageConverter messageConverter() { + return new Jackson2JsonMessageConverter(); + } + +} diff --git a/spring-cloud-starter-single-step-batch-job/src/main/java/org/springframework/cloud/task/batch/autoconfigure/rabbit/AmqpItemWriterProperties.java b/spring-cloud-starter-single-step-batch-job/src/main/java/org/springframework/cloud/task/batch/autoconfigure/rabbit/AmqpItemWriterProperties.java new file mode 100644 index 00000000..6e90fbb8 --- /dev/null +++ b/spring-cloud-starter-single-step-batch-job/src/main/java/org/springframework/cloud/task/batch/autoconfigure/rabbit/AmqpItemWriterProperties.java @@ -0,0 +1,68 @@ +/* + * Copyright 2020-2020 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.cloud.task.batch.autoconfigure.rabbit; + +import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter; +import org.springframework.boot.context.properties.ConfigurationProperties; + +/** + * @author Glenn Renfro + * @since 2.3 + */ +@ConfigurationProperties(prefix = "spring.batch.job.amqpitemwriter") +public class AmqpItemWriterProperties { + + private boolean enabled; + + private boolean jsonConverterEnabled = true; + + /** + * The state of the enabled flag. + * @return true if AmqpItemWriter is enabled. Otherwise false. + */ + public boolean isEnabled() { + return enabled; + } + + /** + * Enables or disables the AmqpItemReader. + * @param enabled if true then AmqpItemWriter will be enabled. Defaults to false. + */ + public void setEnabled(boolean enabled) { + this.enabled = enabled; + } + + /** + * States whether the {@link Jackson2JsonMessageConverter} is used as a message + * converter. + * @return true if enabled else false. + */ + public boolean isJsonConverterEnabled() { + return jsonConverterEnabled; + } + + /** + * Establishes whether the {@link Jackson2JsonMessageConverter} is to be used as a + * message converter. + * @param jsonConverterEnabled true if it is to be enabled else false. Defaults to + * true. + */ + public void setJsonConverterEnabled(boolean jsonConverterEnabled) { + this.jsonConverterEnabled = jsonConverterEnabled; + } + +} diff --git a/spring-cloud-starter-single-step-batch-job/src/main/resources/META-INF/spring.factories b/spring-cloud-starter-single-step-batch-job/src/main/resources/META-INF/spring.factories index faf92ca8..6938add2 100644 --- a/spring-cloud-starter-single-step-batch-job/src/main/resources/META-INF/spring.factories +++ b/spring-cloud-starter-single-step-batch-job/src/main/resources/META-INF/spring.factories @@ -4,4 +4,5 @@ org.springframework.boot.autoconfigure.EnableAutoConfiguration=org.springframewo org.springframework.cloud.task.batch.autoconfigure.flatfile.FlatFileItemWriterAutoConfiguration, \ org.springframework.cloud.task.batch.autoconfigure.jdbc.JdbcItemWriterAutoConfiguration, \ org.springframework.cloud.task.batch.autoconfigure.jdbc.JdbcCursorItemReaderAutoConfiguration, \ - org.springframework.cloud.task.batch.autoconfigure.rabbit.AmqpItemReaderAutoConfiguration + org.springframework.cloud.task.batch.autoconfigure.rabbit.AmqpItemReaderAutoConfiguration, \ + org.springframework.cloud.task.batch.autoconfigure.rabbit.AmqpItemWriterAutoConfiguration diff --git a/spring-cloud-starter-single-step-batch-job/src/test/java/org/springframework/cloud/task/batch/autoconfigure/rabbit/AmqpItemWriterAutoConfigurationTests.java b/spring-cloud-starter-single-step-batch-job/src/test/java/org/springframework/cloud/task/batch/autoconfigure/rabbit/AmqpItemWriterAutoConfigurationTests.java new file mode 100644 index 00000000..6750a9b6 --- /dev/null +++ b/spring-cloud-starter-single-step-batch-job/src/test/java/org/springframework/cloud/task/batch/autoconfigure/rabbit/AmqpItemWriterAutoConfigurationTests.java @@ -0,0 +1,219 @@ +/* + * Copyright 2020-2020 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.cloud.task.batch.autoconfigure.rabbit; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.mockito.Mockito; +import org.testcontainers.containers.GenericContainer; + +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.connection.ConnectionFactory; +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.JobExecution; +import org.springframework.batch.core.JobParameters; +import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing; +import org.springframework.batch.core.explore.JobExplorer; +import org.springframework.batch.core.launch.JobLauncher; +import org.springframework.batch.item.ItemReader; +import org.springframework.batch.item.support.ListItemReader; +import org.springframework.boot.autoconfigure.AutoConfigurations; +import org.springframework.boot.autoconfigure.amqp.RabbitAutoConfiguration; +import org.springframework.boot.autoconfigure.batch.BatchAutoConfiguration; +import org.springframework.boot.autoconfigure.context.PropertyPlaceholderAutoConfiguration; +import org.springframework.boot.test.context.assertj.AssertableApplicationContext; +import org.springframework.boot.test.context.runner.ApplicationContextRunner; +import org.springframework.cloud.task.batch.autoconfigure.SingleStepJobAutoConfiguration; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.jdbc.core.RowMapper; + +import static org.assertj.core.api.Assertions.assertThat; + +public class AmqpItemWriterAutoConfigurationTests { + + private final static String QUEUE_NAME = "foo"; + + private final static String EXCHANGE_NAME = "fooexchange"; + + private static int amqpPort; + + private static String host; + + private static List> sampleData; + + private RabbitTemplate template; + + private ConnectionFactory connectionFactory; + + private String[] configurations; + + static { + GenericContainer rabbitmq = new GenericContainer("rabbitmq:3.5.3") + .withExposedPorts(5672); + rabbitmq.start(); + final Integer mappedPort = rabbitmq.getMappedPort(5672); + host = rabbitmq.getContainerIpAddress(); + amqpPort = mappedPort; + sampleData = new ArrayList<>(5); + addNameToReaderList(sampleData, "Jane"); + addNameToReaderList(sampleData, "John"); + addNameToReaderList(sampleData, "Liz"); + addNameToReaderList(sampleData, "Cameron"); + addNameToReaderList(sampleData, "Judy"); + } + + private static void addNameToReaderList(List> itemReaderList, + String value) { + Map prepMap = new HashMap<>(); + prepMap.put("first_name", value); + itemReaderList.add(prepMap); + } + + @BeforeEach + void setupTest() { + this.connectionFactory = new CachingConnectionFactory(host, amqpPort); + this.template = new RabbitTemplate(this.connectionFactory); + this.template.setMessageConverter(new Jackson2JsonMessageConverter()); + AmqpAdmin admin = new RabbitAdmin(this.connectionFactory); + admin.declareQueue(new Queue(QUEUE_NAME)); + admin.declareExchange(new TopicExchange(EXCHANGE_NAME)); + admin.declareBinding(new Binding(QUEUE_NAME, Binding.DestinationType.QUEUE, + EXCHANGE_NAME, "#", null)); + this.configurations = new String[] { "spring.batch.job.jobName=integrationJob", + "spring.batch.job.stepName=step1", "spring.batch.job.chunkSize=5", + "spring.rabbitmq.template.exchange=" + EXCHANGE_NAME, + "spring.rabbitmq.host=" + host, + "spring.batch.job.amqpitemwriter.enabled=true", + "spring.rabbitmq.port=" + amqpPort }; + } + + @AfterEach + void teardownTest() { + AmqpAdmin admin = new RabbitAdmin(this.connectionFactory); + admin.deleteQueue(QUEUE_NAME); + this.template.destroy(); + } + + @Test + void basicTest() { + ApplicationContextRunner applicationContextRunner = new ApplicationContextRunner() + .withUserConfiguration(BaseConfiguration.class) + .withConfiguration( + AutoConfigurations.of(PropertyPlaceholderAutoConfiguration.class, + BatchAutoConfiguration.class, + SingleStepJobAutoConfiguration.class, + AmqpItemWriterAutoConfiguration.class, + RabbitAutoConfiguration.class)) + .withPropertyValues(this.configurations); + + applicationContextRunner.run((context) -> { + JobExecution jobExecution = runJob(context); + JobExplorer jobExplorer = context.getBean(JobExplorer.class); + + while (jobExplorer.getJobExecution(jobExecution.getJobId()).isRunning()) { + Thread.sleep(1000); + } + + for (Map sampleEntry : sampleData) { + Map map = (Map) template + .receiveAndConvert(QUEUE_NAME); + assertThat(map.get("first_name")) + .isEqualTo(sampleEntry.get("first_name")); + } + }); + } + + @Test + void useAmqpTemplateTest() { + ApplicationContextRunner applicationContextRunner = new ApplicationContextRunner() + .withUserConfiguration(MockConfiguration.class) + .withConfiguration( + AutoConfigurations.of(PropertyPlaceholderAutoConfiguration.class, + BatchAutoConfiguration.class, + SingleStepJobAutoConfiguration.class, + AmqpItemWriterAutoConfiguration.class)) + .withPropertyValues(this.configurations); + + applicationContextRunner.run((context) -> { + runJob(context); + AmqpTemplate amqpTemplate = context.getBean(AmqpTemplate.class); + Mockito.verify(amqpTemplate, Mockito.times(5)).convertAndSend(Mockito.any()); + }); + } + + private JobExecution runJob(AssertableApplicationContext context) throws Exception { + JobLauncher jobLauncher = context.getBean(JobLauncher.class); + + Job job = context.getBean(Job.class); + + return jobLauncher.run(job, new JobParameters()); + } + + @EnableBatchProcessing + @Configuration + public static class BaseConfiguration extends ItemWriterConfiguration { + + } + + @EnableBatchProcessing + @Configuration + public static class MockConfiguration extends ItemWriterConfiguration { + + @Bean + AmqpTemplate amqpTemplateBean() { + return Mockito.mock(AmqpTemplate.class); + } + + } + + public static class ItemWriterConfiguration { + + @Bean + public RowMapper> rowMapper() { + return (rs, rowNum) -> { + Map item = new HashMap<>(); + + item.put("item", rs.getString("item_name")); + + return item; + }; + } + + @Bean + public ItemReader> itemWriter() { + + return new ListItemReader<>(sampleData); + } + + } + +}