SpringCloud/spring-functions-catalog
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Make RabbitMQ modules as auto-configuration

Browse Source 
* Fix all the Checkstyle violations and compiler warnings
* Add tests to these modules
This commit is contained in:
Artem Bilan
2024-01-08 19:10:22 -05:00
parent 80e0b80a30
commit f509b15cde
14 changed files with 178 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
consumer/spring-rabbit-consumer/README.adoc
View File

@@ -4,7 +4,7 @@ A consumer that allows you to send messages to RabbitMQ.
## Beans for injection
You can import `RabbitConsumerConfiguration` in the application and then inject the following bean.
The `RabbitConsumerConfiguration` auto-configuration provides the following bean:
`Function<Message<?>, Object> rabbitConsumer`
@@ -12,7 +12,7 @@ You can use `rabbitConsumer` as a qualifier when injecting.
## Configuration Options
All configuration properties are prefixed with `rabbit`.
All configuration properties are prefixed with `rabbit.consumer`.
For more information on the various options available, please see link:src/main/java/org/springframework/cloud/fn/consumer/rabbit/RabbitConsumerProperties.java[RabbitConsumerProperties].

3
consumer/spring-rabbit-consumer/build.gradle
View File

@@ -1,5 +1,6 @@
dependencies {
api 'org.springframework.integration:spring-integration-amqp'
api 'org.springframework.boot:spring-boot-starter-amqp'
api 'org.springframework.boot:spring-boot-starter-web'
testImplementation 'org.springframework.amqp:spring-rabbit-junit'
}

29
consumer/spring-rabbit-consumer/src/main/java/org/springframework/cloud/fn/consumer/rabbit/RabbitConsumerConfiguration.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2019-2022 the original author or authors.
* Copyright 2019-2024 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.
@@ -16,7 +16,7 @@
package org.springframework.cloud.fn.consumer.rabbit;
import java.util.function.Function;
import java.util.function.Consumer;
import com.rabbitmq.client.impl.CredentialsProvider;
import com.rabbitmq.client.impl.CredentialsRefreshService;
@@ -33,15 +33,16 @@ import org.springframework.beans.factory.ObjectProvider;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.AutoConfiguration;
import org.springframework.boot.autoconfigure.amqp.CachingConnectionFactoryConfigurer;
import org.springframework.boot.autoconfigure.amqp.ConnectionFactoryCustomizer;
import org.springframework.boot.autoconfigure.amqp.RabbitAutoConfiguration;
import org.springframework.boot.autoconfigure.amqp.RabbitConnectionFactoryBeanConfigurer;
import org.springframework.boot.autoconfigure.amqp.RabbitProperties;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.cloud.fn.common.config.ComponentCustomizer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.io.ResourceLoader;
import org.springframework.expression.Expression;
import org.springframework.integration.amqp.dsl.Amqp;
@@ -51,7 +52,7 @@ import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHandler;
/**
* A configuration for RabbitMQ Consumer function. Uses a
* Auto-configuration for RabbitMQ Consumer function. Uses a
* {@link AmqpOutboundChannelAdapterSpec} to save payload contents to RabbitMQ.
*
* @author Soby Chako
@@ -59,7 +60,7 @@ import org.springframework.messaging.MessageHandler;
* @author Chris Bono
*/
@EnableConfigurationProperties(RabbitConsumerProperties.class)
@Configuration
@AutoConfiguration(after = RabbitAutoConfiguration.class)
public class RabbitConsumerConfiguration implements DisposableBean {
@Autowired
@@ -86,11 +87,8 @@ public class RabbitConsumerConfiguration implements DisposableBean {
private CachingConnectionFactory ownConnectionFactory;
@Bean
public Function<Message<?>, Object> rabbitConsumer(@Qualifier("amqpChannelAdapter") MessageHandler messageHandler) {
return o -> {
messageHandler.handleMessage(o);
return "";
};
public Consumer<Message<?>> rabbitConsumer(@Qualifier("amqpChannelAdapter") MessageHandler messageHandler) {
return messageHandler::handleMessage;
}
@Bean
@@ -100,9 +98,9 @@ public class RabbitConsumerConfiguration implements DisposableBean {
AmqpOutboundChannelAdapterSpec handler = Amqp
.outboundAdapter(rabbitTemplate(
this.properties.isOwnConnection() ? buildLocalConnectionFactory() : rabbitConnectionFactory))
.mappedRequestHeaders(properties.getMappedRequestHeaders())
.defaultDeliveryMode(properties.getPersistentDeliveryMode() ? MessageDeliveryMode.PERSISTENT
(this.properties.isOwnConnection()) ? buildLocalConnectionFactory() : rabbitConnectionFactory))
.mappedRequestHeaders(this.properties.getMappedRequestHeaders())
.defaultDeliveryMode((this.properties.getPersistentDeliveryMode()) ? MessageDeliveryMode.PERSISTENT
: MessageDeliveryMode.NON_PERSISTENT)
.headersMappedLast(this.properties.isHeadersMappedLast());
@@ -129,8 +127,7 @@ public class RabbitConsumerConfiguration implements DisposableBean {
return handler;
}
@Bean
public RabbitTemplate rabbitTemplate(ConnectionFactory rabbitConnectionFactory) {
private RabbitTemplate rabbitTemplate(ConnectionFactory rabbitConnectionFactory) {
RabbitTemplate rabbitTemplate = new RabbitTemplate(rabbitConnectionFactory);
if (this.messageConverter != null) {
rabbitTemplate.setMessageConverter(this.messageConverter);
@@ -184,7 +181,7 @@ public class RabbitConsumerConfiguration implements DisposableBean {
CachingConnectionFactory cachingConnectionFactory = new CachingConnectionFactory(connectionFactory);
CachingConnectionFactoryConfigurer cachingConnectionFactoryConfigurer = new CachingConnectionFactoryConfigurer(
properties);
cachingConnectionFactoryConfigurer.setConnectionNameStrategy(cf -> "rabbit.sink.own.connection");
cachingConnectionFactoryConfigurer.setConnectionNameStrategy((cf) -> "rabbit.sink.own.connection");
cachingConnectionFactoryConfigurer.configure(cachingConnectionFactory);
cachingConnectionFactory.afterPropertiesSet();

11
consumer/spring-rabbit-consumer/src/main/java/org/springframework/cloud/fn/consumer/rabbit/RabbitConsumerProperties.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2019-2020 the original author or authors.
* Copyright 2019-2024 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.
@@ -22,7 +22,12 @@ import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.expression.Expression;
import org.springframework.validation.annotation.Validated;
@ConfigurationProperties("rabbit")
/**
* Configuration properties for RabbitMQ consumer.
*
* @author Soby Chacko
*/
@ConfigurationProperties("rabbit.consumer")
@Validated
public class RabbitConsumerProperties {
@@ -76,7 +81,7 @@ public class RabbitConsumerProperties {
/**
* When mapping headers for the outbound message, determine whether the headers are
* mapped before the message is converted, or afterwards.
* mapped before the message is converted, or afterward.
*/
private boolean headersMappedLast = true;

4
consumer/spring-rabbit-consumer/src/main/java/org/springframework/cloud/fn/consumer/rabbit/package-info.java Normal file
View File

@@ -0,0 +1,4 @@
/**
* The RabbitMQ consumer auto-configuration support.
*/
package org.springframework.cloud.fn.consumer.rabbit;

1
consumer/spring-rabbit-consumer/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports Normal file
View File

@@ -0,0 +1 @@
org.springframework.cloud.fn.consumer.rabbit.RabbitConsumerConfiguration

58
consumer/spring-rabbit-consumer/src/test/java/org/springframework/cloud/fn/consumer/rabbit/RabbitConsumerTests.java Normal file
View File

@@ -0,0 +1,58 @@
/*
* Copyright 2016-2024 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.fn.consumer.rabbit;
import java.util.function.Consumer;
import org.junit.jupiter.api.Test;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.junit.RabbitAvailable;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.GenericMessage;
import static org.assertj.core.api.Assertions.assertThat;
@RabbitAvailable(RabbitConsumerTests.TEST_QUEUE)
@SpringBootTest(properties = { "rabbit.consumer.routingKey=" + RabbitConsumerTests.TEST_QUEUE,
"rabbit.consumer.own-connection=true" })
public class RabbitConsumerTests {
static final String TEST_QUEUE = "test-consumer-queue";
@Test
void rabbitSupplierReceivesData(@Autowired RabbitTemplate rabbitTemplate,
@Autowired Consumer<Message<?>> rabbitConsumer) {
rabbitConsumer.accept(new GenericMessage<>("test data1"));
rabbitConsumer.accept(new GenericMessage<>("test data2"));
Object received = rabbitTemplate.receiveAndConvert(TEST_QUEUE, 10_000);
assertThat(received).isEqualTo("test data1");
received = rabbitTemplate.receiveAndConvert(TEST_QUEUE, 10_000);
assertThat(received).isEqualTo("test data2");
}
@SpringBootApplication
public static class TestConfiguration {
}
}

2
supplier/spring-rabbit-supplier/README.adoc
View File

@@ -8,7 +8,7 @@ Users have to subscribe to this `Flux` and receive the data.
## Beans for injection
You can import the `RabbitSupplierConfiguration` in the application and then inject the following bean.
The `RabbitSupplierConfiguration` auto-configuration provides the following bean:
`rabbitSupplier`

3
supplier/spring-rabbit-supplier/build.gradle
View File

@@ -1,5 +1,6 @@
dependencies {
api 'org.springframework.integration:spring-integration-amqp'
api 'org.springframework.boot:spring-boot-starter-amqp'
api 'org.springframework.boot:spring-boot-starter-web'
testImplementation 'org.springframework.amqp:spring-rabbit-junit'
}

19
supplier/spring-rabbit-supplier/src/main/java/org/springframework/cloud/fn/supplier/rabbit/RabbitSupplierConfiguration.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2023 the original author or authors.
* Copyright 2016-2024 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.
@@ -38,14 +38,15 @@ import org.springframework.amqp.rabbit.support.MessagePropertiesConverter;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.ObjectProvider;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.AutoConfiguration;
import org.springframework.boot.autoconfigure.amqp.CachingConnectionFactoryConfigurer;
import org.springframework.boot.autoconfigure.amqp.ConnectionFactoryCustomizer;
import org.springframework.boot.autoconfigure.amqp.RabbitAutoConfiguration;
import org.springframework.boot.autoconfigure.amqp.RabbitConnectionFactoryBeanConfigurer;
import org.springframework.boot.autoconfigure.amqp.RabbitProperties;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.cloud.fn.common.config.ComponentCustomizer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.io.ResourceLoader;
import org.springframework.integration.amqp.dsl.Amqp;
import org.springframework.integration.amqp.dsl.AmqpInboundChannelAdapterSMLCSpec;
@@ -56,7 +57,7 @@ import org.springframework.retry.interceptor.RetryOperationsInterceptor;
import org.springframework.util.Assert;
/**
* A source module that receives data from RabbitMQ.
* Auto-configuration for supplier that receives data from RabbitMQ.
*
* @author Gary Russell
* @author Chris Schaefer
@@ -64,11 +65,11 @@ import org.springframework.util.Assert;
* @author Chris Bono
* @author Artem Bilan
*/
@Configuration(proxyBeanMethods = false)
@AutoConfiguration(after = RabbitAutoConfiguration.class)
@EnableConfigurationProperties(RabbitSupplierProperties.class)
public class RabbitSupplierConfiguration implements DisposableBean {
private static final MessagePropertiesConverter inboundMessagePropertiesConverter = new DefaultMessagePropertiesConverter() {
private static final MessagePropertiesConverter INBOUND_MESSAGE_PROPERTIES_CONVERTER = new DefaultMessagePropertiesConverter() {
@Override
public MessageProperties toMessageProperties(AMQP.BasicProperties source, Envelope envelope, String charset) {
@@ -105,7 +106,7 @@ public class RabbitSupplierConfiguration implements DisposableBean {
@Bean
public SimpleMessageListenerContainer container(RetryOperationsInterceptor rabbitSourceRetryInterceptor) {
ConnectionFactory connectionFactory = this.properties.isOwnConnection() ? buildLocalConnectionFactory()
ConnectionFactory connectionFactory = (this.properties.isOwnConnection()) ? buildLocalConnectionFactory()
: this.rabbitConnectionFactory;
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
container.setAutoStartup(false);
@@ -140,7 +141,7 @@ public class RabbitSupplierConfiguration implements DisposableBean {
if (this.properties.isEnableRetry()) {
container.setAdviceChain(rabbitSourceRetryInterceptor);
}
container.setMessagePropertiesConverter(inboundMessagePropertiesConverter);
container.setMessagePropertiesConverter(INBOUND_MESSAGE_PROPERTIES_CONVERTER);
return container;
}
@@ -149,7 +150,7 @@ public class RabbitSupplierConfiguration implements DisposableBean {
@Nullable ComponentCustomizer<AmqpInboundChannelAdapterSMLCSpec> amqpMessageProducerCustomizer) {
AmqpInboundChannelAdapterSMLCSpec messageProducerSpec = Amqp.inboundAdapter(container)
.mappedRequestHeaders(properties.getMappedRequestHeaders());
.mappedRequestHeaders(this.properties.getMappedRequestHeaders());
if (amqpMessageProducerCustomizer != null) {
amqpMessageProducerCustomizer.customize(messageProducerSpec);
@@ -220,7 +221,7 @@ public class RabbitSupplierConfiguration implements DisposableBean {
CachingConnectionFactory cachingConnectionFactory = new CachingConnectionFactory(connectionFactory);
CachingConnectionFactoryConfigurer cachingConnectionFactoryConfigurer = new CachingConnectionFactoryConfigurer(
properties);
cachingConnectionFactoryConfigurer.setConnectionNameStrategy(cf -> "rabbit.supplier.own.connection");
cachingConnectionFactoryConfigurer.setConnectionNameStrategy((cf) -> "rabbit.supplier.own.connection");
cachingConnectionFactoryConfigurer.configure(cachingConnectionFactory);
cachingConnectionFactory.afterPropertiesSet();

25
supplier/spring-rabbit-supplier/src/main/java/org/springframework/cloud/fn/supplier/rabbit/RabbitSupplierProperties.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2019-2020 the original author or authors.
* Copyright 2019-2024 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.
@@ -22,6 +22,11 @@ import jakarta.validation.constraints.Size;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.validation.annotation.Validated;
/**
* The RabbitMQ supplier configuration properties.
*
* @author Gary Russell
*/
@ConfigurationProperties("rabbit.supplier")
@Validated
public class RabbitSupplierProperties {
@@ -77,7 +82,7 @@ public class RabbitSupplierProperties {
private boolean ownConnection;
public boolean getRequeue() {
return requeue;
return this.requeue;
}
public void setRequeue(boolean requeue) {
@@ -85,7 +90,7 @@ public class RabbitSupplierProperties {
}
public boolean getTransacted() {
return transacted;
return this.transacted;
}
public void setTransacted(boolean transacted) {
@@ -95,7 +100,7 @@ public class RabbitSupplierProperties {
@NotNull(message = "queue(s) are required")
@Size(min = 1, message = "At least one queue is required")
public String[] getQueues() {
return queues;
return this.queues;
}
public void setQueues(String[] queues) {
@@ -104,7 +109,7 @@ public class RabbitSupplierProperties {
@NotNull
public String[] getMappedRequestHeaders() {
return mappedRequestHeaders;
return this.mappedRequestHeaders;
}
public void setMappedRequestHeaders(String[] mappedRequestHeaders) {
@@ -112,7 +117,7 @@ public class RabbitSupplierProperties {
}
public int getInitialRetryInterval() {
return initialRetryInterval;
return this.initialRetryInterval;
}
public void setInitialRetryInterval(int initialRetryInterval) {
@@ -120,7 +125,7 @@ public class RabbitSupplierProperties {
}
public int getMaxRetryInterval() {
return maxRetryInterval;
return this.maxRetryInterval;
}
public void setMaxRetryInterval(int maxRetryInterval) {
@@ -128,7 +133,7 @@ public class RabbitSupplierProperties {
}
public double getRetryMultiplier() {
return retryMultiplier;
return this.retryMultiplier;
}
public void setRetryMultiplier(double retryMultiplier) {
@@ -136,7 +141,7 @@ public class RabbitSupplierProperties {
}
public int getMaxAttempts() {
return maxAttempts;
return this.maxAttempts;
}
public void setMaxAttempts(int maxAttempts) {
@@ -144,7 +149,7 @@ public class RabbitSupplierProperties {
}
public boolean isEnableRetry() {
return enableRetry;
return this.enableRetry;
}
public void setEnableRetry(boolean enableRetry) {

4
supplier/spring-rabbit-supplier/src/main/java/org/springframework/cloud/fn/supplier/rabbit/package-info.java Normal file
View File

@@ -0,0 +1,4 @@
/**
* The RabbitMQ supplier auto-configuration support.
*/
package org.springframework.cloud.fn.supplier.rabbit;

1
supplier/spring-rabbit-supplier/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports Normal file
View File

@@ -0,0 +1 @@
org.springframework.cloud.fn.supplier.rabbit.RabbitSupplierConfiguration

57
supplier/spring-rabbit-supplier/src/test/java/org/springframework/cloud/fn/supplier/rabbit/RabbitSupplierTests.java Normal file
View File

@@ -0,0 +1,57 @@
/*
* Copyright 2016-2024 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.fn.supplier.rabbit;
import java.time.Duration;
import java.util.function.Supplier;
import org.junit.jupiter.api.Test;
import reactor.core.publisher.Flux;
import reactor.test.StepVerifier;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.junit.RabbitAvailable;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.messaging.Message;
@RabbitAvailable(RabbitSupplierTests.TEST_QUEUE)
@SpringBootTest(properties = "rabbit.supplier.queues=" + RabbitSupplierTests.TEST_QUEUE)
public class RabbitSupplierTests {
static final String TEST_QUEUE = "test-supplier-queue";
@Test
void rabbitSupplierReceivesData(@Autowired RabbitTemplate rabbitTemplate,
@Autowired Supplier<Flux<Message<byte[]>>> rabbitSupplier) {
Flux<String> mapped = rabbitSupplier.get().map(Message::getPayload).map(String::new);
StepVerifier stepVerifier = StepVerifier.create(mapped).expectNext("test1", "test2").thenCancel().verifyLater();
rabbitTemplate.convertAndSend(TEST_QUEUE, "test1".getBytes());
rabbitTemplate.convertAndSend(TEST_QUEUE, "test2".getBytes());
stepVerifier.verify(Duration.ofSeconds(10));
}
@SpringBootApplication
public static class TestConfiguration {
}
}