diff --git a/spring-cloud-starter-single-step-batch-job/pom.xml b/spring-cloud-starter-single-step-batch-job/pom.xml index 2867dcc1..8e223c1e 100644 --- a/spring-cloud-starter-single-step-batch-job/pom.xml +++ b/spring-cloud-starter-single-step-batch-job/pom.xml @@ -9,6 +9,12 @@ spring-cloud-starter-single-step-batch-job + + 1.14.3 + 1.14.3 + 1.0.8 + + org.springframework.boot @@ -41,6 +47,40 @@ h2 test + + org.springframework.amqp + spring-amqp + + + org.springframework.amqp + spring-rabbit + + + org.testcontainers + testcontainers + ${test.containers.version} + test + + + org.testcontainers + rabbitmq + ${test.rabbit.containers.version} + test + + + org.rnorth.duct-tape + duct-tape + ${test.ducttape.version} + test + + + com.fasterxml.jackson.core + jackson-core + + + com.fasterxml.jackson.core + jackson-annotations + diff --git a/spring-cloud-starter-single-step-batch-job/src/main/java/org/springframework/cloud/task/batch/autoconfigure/rabbit/AmqpItemReaderAutoConfiguration.java b/spring-cloud-starter-single-step-batch-job/src/main/java/org/springframework/cloud/task/batch/autoconfigure/rabbit/AmqpItemReaderAutoConfiguration.java new file mode 100644 index 00000000..90482a1e --- /dev/null +++ b/spring-cloud-starter-single-step-batch-job/src/main/java/org/springframework/cloud/task/batch/autoconfigure/rabbit/AmqpItemReaderAutoConfiguration.java @@ -0,0 +1,89 @@ +/* + * 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.core.Queue; +import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter; +import org.springframework.amqp.support.converter.MessageConverter; +import org.springframework.batch.item.amqp.AmqpItemReader; +import org.springframework.batch.item.amqp.builder.AmqpItemReaderBuilder; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.autoconfigure.AutoConfigureAfter; +import org.springframework.boot.autoconfigure.amqp.RabbitProperties; +import org.springframework.boot.autoconfigure.batch.BatchAutoConfiguration; +import org.springframework.boot.autoconfigure.condition.ConditionalOnBean; +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; +import org.springframework.util.StringUtils; + +/** + * Autconfiguration for a {@code AmqpItemReader}. + * + * @author Glenn Renfro + * @since 2.3 + */ +@Configuration +@EnableConfigurationProperties(AmqpItemReaderProperties.class) +@AutoConfigureAfter(BatchAutoConfiguration.class) +@ConditionalOnProperty(name = "spring.batch.job.amqpitemreader.enabled", + havingValue = "true", matchIfMissing = false) +public class AmqpItemReaderAutoConfiguration { + + @Autowired(required = false) + private RabbitProperties rabbitProperties; + + @Bean + public AmqpItemReaderProperties amqpItemReaderProperties() { + return new AmqpItemReaderProperties(); + } + + @ConditionalOnBean(RabbitProperties.class) + @Bean + public Queue defaultQueue() { + if (!StringUtils + .hasText(this.rabbitProperties.getTemplate().getDefaultReceiveQueue())) { + throw new IllegalArgumentException( + "DefaultReceiveQueue must not be empty nor null"); + } + return new Queue(this.rabbitProperties.getTemplate().getDefaultReceiveQueue(), + true); + } + + @Bean + public AmqpItemReader> amqpItemReader(AmqpTemplate amqpTemplate, + @Autowired(required = false) Class itemType) { + AmqpItemReaderBuilder> builder = new AmqpItemReaderBuilder>() + .amqpTemplate(amqpTemplate); + if (itemType != null) { + builder.itemType(itemType); + } + return builder.build(); + } + + @ConditionalOnProperty(name = "spring.batch.job.amqpitemreader.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/AmqpItemReaderProperties.java b/spring-cloud-starter-single-step-batch-job/src/main/java/org/springframework/cloud/task/batch/autoconfigure/rabbit/AmqpItemReaderProperties.java new file mode 100644 index 00000000..64e24e61 --- /dev/null +++ b/spring-cloud-starter-single-step-batch-job/src/main/java/org/springframework/cloud/task/batch/autoconfigure/rabbit/AmqpItemReaderProperties.java @@ -0,0 +1,70 @@ +/* + * 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; + +/** + * Properties to configure a {@code AmqpItemReader}. + * + * @author Glenn Renfro + * @since 2.3 + */ +@ConfigurationProperties(prefix = "spring.batch.job.amqpitemreader") +public class AmqpItemReaderProperties { + + private boolean enabled; + + private boolean jsonConverterEnabled = true; + + /** + * The state of the enabled flag. + * @return true if AmqpItemReader is enabled. Otherwise false. + */ + public boolean isEnabled() { + return enabled; + } + + /** + * Enables or disables the AmqpItemReader. + * @param enabled if true then AmqpItemReader 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 11068756..faf92ca8 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 @@ -3,4 +3,5 @@ org.springframework.boot.autoconfigure.EnableAutoConfiguration=org.springframewo org.springframework.cloud.task.batch.autoconfigure.SingleStepJobAutoConfiguration,\ 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.jdbc.JdbcCursorItemReaderAutoConfiguration, \ + org.springframework.cloud.task.batch.autoconfigure.rabbit.AmqpItemReaderAutoConfiguration diff --git a/spring-cloud-starter-single-step-batch-job/src/test/java/org/springframework/cloud/task/batch/autoconfigure/rabbit/AmqpItemReaderAutoConfigurationTests.java b/spring-cloud-starter-single-step-batch-job/src/test/java/org/springframework/cloud/task/batch/autoconfigure/rabbit/AmqpItemReaderAutoConfigurationTests.java new file mode 100644 index 00000000..c28486a9 --- /dev/null +++ b/spring-cloud-starter-single-step-batch-job/src/test/java/org/springframework/cloud/task/batch/autoconfigure/rabbit/AmqpItemReaderAutoConfigurationTests.java @@ -0,0 +1,251 @@ +/* + * 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.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.Queue; +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.support.ListItemWriter; +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 static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +public class AmqpItemReaderAutoConfigurationTests { + + private static int amqpPort; + + private static String host; + + private RabbitTemplate template; + + private ConnectionFactory connectionFactory; + + static { + GenericContainer rabbitmq = new GenericContainer("rabbitmq:3.5.3") + .withExposedPorts(5672); + rabbitmq.start(); + final Integer mappedPort = rabbitmq.getMappedPort(5672); + host = rabbitmq.getContainerIpAddress(); + amqpPort = mappedPort; + } + + @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("foo")); + + Map testMap = new HashMap<>(); + testMap.put("ITEM_NAME", "foo"); + this.template.convertAndSend("foo", testMap); + testMap = new HashMap<>(); + testMap.put("ITEM_NAME", "bar"); + this.template.convertAndSend("foo", testMap); + testMap = new HashMap<>(); + testMap.put("ITEM_NAME", "baz"); + this.template.convertAndSend("foo", testMap); + } + + @AfterEach + void teardownTest() { + AmqpAdmin admin = new RabbitAdmin(this.connectionFactory); + admin.deleteQueue("foo"); + this.template.destroy(); + } + + @Test + void basicTest() { + ApplicationContextRunner applicationContextRunner = new ApplicationContextRunner() + .withUserConfiguration(BaseConfiguration.class) + .withConfiguration( + AutoConfigurations.of(PropertyPlaceholderAutoConfiguration.class, + BatchAutoConfiguration.class, + SingleStepJobAutoConfiguration.class, + AmqpItemReaderAutoConfiguration.class, + RabbitAutoConfiguration.class)) + .withPropertyValues("spring.batch.job.jobName=integrationJob", + "spring.batch.job.stepName=step1", "spring.batch.job.chunkSize=5", + "spring.batch.job.amqpitemreader.enabled=true", + "spring.rabbitmq.template.default-receive-queue=foo", + "spring.rabbitmq.host=" + host, + "spring.rabbitmq.port=" + amqpPort); + + applicationContextRunner.run((context) -> { + JobExecution jobExecution = runJob(context); + + JobExplorer jobExplorer = context.getBean(JobExplorer.class); + + while (jobExplorer.getJobExecution(jobExecution.getJobId()).isRunning()) { + Thread.sleep(1000); + } + + validateBasicTest(context.getBean(ListItemWriter.class).getWrittenItems()); + }); + } + + @Test + void basicTestWithItemType() { + ApplicationContextRunner applicationContextRunner = new ApplicationContextRunner() + .withUserConfiguration(ItemTypeConfiguration.class) + .withConfiguration( + AutoConfigurations.of(PropertyPlaceholderAutoConfiguration.class, + BatchAutoConfiguration.class, + SingleStepJobAutoConfiguration.class, + AmqpItemReaderAutoConfiguration.class, + RabbitAutoConfiguration.class)) + .withPropertyValues("spring.batch.job.jobName=integrationJob", + "spring.batch.job.stepName=step1", "spring.batch.job.chunkSize=5", + "spring.batch.job.amqpitemreader.enabled=true", + "spring.rabbitmq.template.default-receive-queue=foo", + "spring.rabbitmq.host=" + host, + "spring.rabbitmq.port=" + amqpPort); + + applicationContextRunner.run((context) -> { + JobExecution jobExecution = runJob(context); + + JobExplorer jobExplorer = context.getBean(JobExplorer.class); + + while (jobExplorer.getJobExecution(jobExecution.getJobId()).isRunning()) { + Thread.sleep(1000); + } + validateBasicTest(context.getBean(ListItemWriter.class).getWrittenItems()); + }); + } + + @Test + void missingDefaultQueueTest() { + ApplicationContextRunner applicationContextRunner = new ApplicationContextRunner() + .withUserConfiguration(BaseConfiguration.class) + .withConfiguration( + AutoConfigurations.of(PropertyPlaceholderAutoConfiguration.class, + BatchAutoConfiguration.class, + SingleStepJobAutoConfiguration.class, + AmqpItemReaderAutoConfiguration.class, + RabbitAutoConfiguration.class)) + .withPropertyValues("spring.batch.job.jobName=integrationJob", + "spring.batch.job.stepName=step1", "spring.batch.job.chunkSize=5", + "spring.batch.job.amqpitemreader.enabled=true", + "spring.batch.job.amqpitemreader.jsonConverterEnabled=false", + "spring.rabbitmq.host=" + host, + "spring.rabbitmq.port=" + amqpPort); + + assertThatThrownBy(() -> { + applicationContextRunner.run((context) -> { + context.getBean(JobLauncher.class); + }); + }).isInstanceOf(IllegalStateException.class).getRootCause() + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("DefaultReceiveQueue must not be empty nor null"); + } + + @Test + void useAmqpTemplateTest() { + ApplicationContextRunner applicationContextRunner = new ApplicationContextRunner() + .withUserConfiguration(MockTemplateConfiguration.class) + .withConfiguration( + AutoConfigurations.of(PropertyPlaceholderAutoConfiguration.class, + BatchAutoConfiguration.class, + SingleStepJobAutoConfiguration.class, + AmqpItemReaderAutoConfiguration.class)) + .withPropertyValues("spring.batch.job.jobName=integrationJob", + "spring.batch.job.stepName=step1", "spring.batch.job.chunkSize=5", + "spring.batch.job.amqpitemreader.enabled=true", + "spring.rabbitmq.host=" + host, + "spring.rabbitmq.port=" + amqpPort); + + applicationContextRunner.run((context) -> { + runJob(context); + AmqpTemplate amqpTemplate = context.getBean(AmqpTemplate.class); + Mockito.verify(amqpTemplate, Mockito.times(1)).receiveAndConvert(); + }); + } + + 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()); + } + + private void validateBasicTest(List> items) { + assertThat(items.size()).isEqualTo(3); + assertThat(items.get(0).get("ITEM_NAME")).isEqualTo("foo"); + assertThat(items.get(1).get("ITEM_NAME")).isEqualTo("bar"); + assertThat(items.get(2).get("ITEM_NAME")).isEqualTo("baz"); + } + + public static class MockTemplateConfiguration extends BaseConfiguration { + + @Bean + AmqpTemplate amqpTemplateBean() { + return Mockito.mock(AmqpTemplate.class); + }; + + } + + public static class ItemTypeConfiguration extends BaseConfiguration { + + @Bean + Class itemTypeClass() { + return Map.class; + } + + } + + @EnableBatchProcessing + @Configuration + public static class BaseConfiguration { + + @Bean + public ListItemWriter> itemWriter() { + return new ListItemWriter<>(); + } + + } + +}