Add AMQPItemReader
This adds autoconfiguration for an AMQPItemReader to the single step batch job starter. resolves TASK-680A
This commit is contained in:
Glenn Renfro
committed by
Michael Minella
Michael Minella
parent
7223f60ee7
commit
5fa4a2028d
@@ -9,6 +9,12 @@
|
||||
|
||||
<artifactId>spring-cloud-starter-single-step-batch-job</artifactId>
|
||||
|
||||
<properties>
|
||||
<test.containers.version>1.14.3</test.containers.version>
|
||||
<test.rabbit.containers.version>1.14.3</test.rabbit.containers.version>
|
||||
<test.ducttape.version>1.0.8</test.ducttape.version>
|
||||
</properties>
|
||||
|
||||
<dependencies>
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
@@ -41,6 +47,40 @@
|
||||
<artifactId>h2</artifactId>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.amqp</groupId>
|
||||
<artifactId>spring-amqp</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.amqp</groupId>
|
||||
<artifactId>spring-rabbit</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.testcontainers</groupId>
|
||||
<artifactId>testcontainers</artifactId>
|
||||
<version>${test.containers.version}</version>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.testcontainers</groupId>
|
||||
<artifactId>rabbitmq</artifactId>
|
||||
<version>${test.rabbit.containers.version}</version>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.rnorth.duct-tape</groupId>
|
||||
<artifactId>duct-tape</artifactId>
|
||||
<version>${test.ducttape.version}</version>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>com.fasterxml.jackson.core</groupId>
|
||||
<artifactId>jackson-core</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>com.fasterxml.jackson.core</groupId>
|
||||
<artifactId>jackson-annotations</artifactId>
|
||||
</dependency>
|
||||
</dependencies>
|
||||
|
||||
</project>
|
||||
|
||||
@@ -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<Map<Object, Object>> amqpItemReader(AmqpTemplate amqpTemplate,
|
||||
@Autowired(required = false) Class itemType) {
|
||||
AmqpItemReaderBuilder<Map<Object, Object>> builder = new AmqpItemReaderBuilder<Map<Object, Object>>()
|
||||
.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();
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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;
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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
|
||||
|
||||
@@ -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<Object, Object> 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<Map<Object, Object>> 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<Map<Object, Object>> itemWriter() {
|
||||
return new ListItemWriter<>();
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
Reference in New Issue
Block a user