Skip to content

Add support for RabbitMQ AMQP 1.0 #46608

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 9 commits into
base: main
Choose a base branch
from
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion module/spring-boot-amqp/build.gradle
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,7 @@ description = "Spring Boot AMQP"
dependencies {
api(project(":core:spring-boot"))
api("org.springframework:spring-messaging")
api("org.springframework.amqp:spring-rabbit")
api("org.springframework.amqp:spring-rabbitmq-client")

compileOnly("com.fasterxml.jackson.core:jackson-annotations")

Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,39 @@
/*
* Copyright 2012-present 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.boot.amqp.autoconfigure;

import com.rabbitmq.client.amqp.Environment;
import com.rabbitmq.client.amqp.impl.AmqpEnvironmentBuilder;

/**
* Callback interface that can be implemented by beans wishing to customize the
* auto-configured {@link Environment} that is created by an
* {@link AmqpEnvironmentBuilder}.
*
* @author Eddú Meléndez
* @since 4.0.0
*/
@FunctionalInterface
public interface AmqpEnvironmentBuilderCustomizer {

/**
* Customize the {@code AmqpEnvironmentBuilder}.
* @param builder the builder to customize
*/
void customize(AmqpEnvironmentBuilder builder);

}
Original file line number Diff line number Diff line change
@@ -0,0 +1,136 @@
/*
* Copyright 2012-present 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.boot.amqp.autoconfigure;

import com.rabbitmq.client.amqp.Connection;
import com.rabbitmq.client.amqp.CredentialsProvider;
import com.rabbitmq.client.amqp.Environment;
import com.rabbitmq.client.amqp.impl.AmqpEnvironmentBuilder;
import com.rabbitmq.client.amqp.impl.AmqpEnvironmentBuilder.EnvironmentConnectionSettings;

import org.springframework.amqp.rabbit.config.ContainerCustomizer;
import org.springframework.amqp.rabbit.retry.MessageRecoverer;
import org.springframework.amqp.rabbitmq.client.AmqpConnectionFactory;
import org.springframework.amqp.rabbitmq.client.RabbitAmqpAdmin;
import org.springframework.amqp.rabbitmq.client.RabbitAmqpTemplate;
import org.springframework.amqp.rabbitmq.client.SingleAmqpConnectionFactory;
import org.springframework.amqp.rabbitmq.client.config.RabbitAmqpListenerContainerFactory;
import org.springframework.amqp.rabbitmq.client.listener.RabbitAmqpListenerContainer;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.beans.factory.ObjectProvider;
import org.springframework.boot.amqp.autoconfigure.RabbitConnectionDetails.Address;
import org.springframework.boot.autoconfigure.AutoConfiguration;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.boot.context.properties.PropertyMapper;
import org.springframework.boot.ssl.SslBundles;
import org.springframework.context.annotation.Bean;

/**
* {@link EnableAutoConfiguration Auto-configuration} for {@link RabbitAmqpTemplate}.
*
* @author Eddú Meléndez
* @since 4.0.0
*/
@AutoConfiguration(before = RabbitAutoConfiguration.class)
@ConditionalOnClass({ RabbitAmqpTemplate.class, Connection.class })
@EnableConfigurationProperties(RabbitProperties.class)
public final class RabbitAmqpAutoConfiguration {

private final RabbitProperties properties;

RabbitAmqpAutoConfiguration(RabbitProperties properties) {
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think we need to look into a separate RabbitAmqpProperties abstraction.
Even if some duplication is possible, a bunch of existing properties are not going to be used for AMQP 1.0 at all.
And might cause confusion.

this.properties = properties;
}

@Bean
@ConditionalOnMissingBean
RabbitConnectionDetails rabbitConnectionDetails(ObjectProvider<SslBundles> sslBundles) {
return new PropertiesRabbitConnectionDetails(this.properties, sslBundles.getIfAvailable());
}

@Bean(name = "rabbitListenerContainerFactory")
@ConditionalOnMissingBean(name = "rabbitListenerContainerFactory")
RabbitAmqpListenerContainerFactory rabbitAmqpListenerContainerFactory(AmqpConnectionFactory connectionFactory,
ObjectProvider<ContainerCustomizer<RabbitAmqpListenerContainer>> amqpContainerCustomizer,
ObjectProvider<RabbitRetryTemplateCustomizer> retryTemplateCustomizers,
ObjectProvider<MessageRecoverer> messageRecoverer) {
RabbitAmqpListenerContainerFactory factory = new RabbitAmqpListenerContainerFactory(connectionFactory);
amqpContainerCustomizer.ifUnique(factory::setContainerCustomizer);

RabbitProperties.AmqpContainer configuration = this.properties.getListener().getSimple();
factory.setObservationEnabled(configuration.isObservationEnabled());
return factory;
}

@Bean
@ConditionalOnMissingBean
Environment rabbitAmqpEnvironment(RabbitConnectionDetails connectionDetails,
ObjectProvider<AmqpEnvironmentBuilderCustomizer> customizers,
ObjectProvider<CredentialsProvider> credentialsProvider) {
PropertyMapper map = PropertyMapper.get();
EnvironmentConnectionSettings environmentConnectionSettings = new AmqpEnvironmentBuilder().connectionSettings();
Address address = connectionDetails.getFirstAddress();
map.from(address::host).whenNonNull().to(environmentConnectionSettings::host);
map.from(address::port).to(environmentConnectionSettings::port);
map.from(connectionDetails::getUsername).whenNonNull().to(environmentConnectionSettings::username);
map.from(connectionDetails::getPassword).whenNonNull().to(environmentConnectionSettings::password);
map.from(connectionDetails::getVirtualHost).whenNonNull().to(environmentConnectionSettings::virtualHost);
map.from(credentialsProvider::getIfAvailable)
.whenNonNull()
.to(environmentConnectionSettings::credentialsProvider);

AmqpEnvironmentBuilder builder = environmentConnectionSettings.environmentBuilder();
customizers.orderedStream().forEach((customizer) -> customizer.customize(builder));
return builder.build();
}

@Bean
@ConditionalOnMissingBean
AmqpConnectionFactory amqpConnectionFactory(Environment environment) {
return new SingleAmqpConnectionFactory(environment);
}

@Bean
@ConditionalOnMissingBean
RabbitAmqpTemplate rabbitAmqpTemplate(AmqpConnectionFactory connectionFactory,
ObjectProvider<RabbitAmqpTemplateCustomizer> customizers,
ObjectProvider<MessageConverter> messageConverter) {
RabbitAmqpTemplate rabbitAmqpTemplate = new RabbitAmqpTemplate(connectionFactory);
if (messageConverter.getIfAvailable() != null) {
rabbitAmqpTemplate.setMessageConverter(messageConverter.getIfAvailable());
}
RabbitProperties.Template templateProperties = this.properties.getTemplate();

PropertyMapper map = PropertyMapper.get();
map.from(templateProperties::getDefaultReceiveQueue).whenNonNull().to(rabbitAmqpTemplate::setReceiveQueue);
map.from(templateProperties::getExchange).whenNonNull().to(rabbitAmqpTemplate::setExchange);
map.from(templateProperties::getRoutingKey).to(rabbitAmqpTemplate::setRoutingKey);

customizers.orderedStream().forEach((customizer) -> customizer.customize(rabbitAmqpTemplate));
return rabbitAmqpTemplate;
}

@Bean
@ConditionalOnMissingBean
RabbitAmqpAdmin rabbitAmqpAdmin(AmqpConnectionFactory connectionFactory) {
return new RabbitAmqpAdmin(connectionFactory);
}

}
Original file line number Diff line number Diff line change
@@ -0,0 +1,36 @@
/*
* Copyright 2012-present 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.boot.amqp.autoconfigure;

import org.springframework.amqp.rabbitmq.client.RabbitAmqpTemplate;

/**
* Callback interface that can be used to customize a {@link RabbitAmqpTemplate}.
*
* @author Eddú Meléndez
* @since 4.0.0
*/
@FunctionalInterface
public interface RabbitAmqpTemplateCustomizer {

/**
* Callback to customize a {@link RabbitAmqpTemplate} instance.
* @param rabbitAmqpTemplate the rabbitAmqpTemplate to customize
*/
void customize(RabbitAmqpTemplate rabbitAmqpTemplate);

}
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,7 @@
import org.springframework.boot.autoconfigure.condition.ConditionalOnBooleanProperty;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingClass;
import org.springframework.boot.autoconfigure.condition.ConditionalOnSingleCandidate;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.boot.ssl.SslBundles;
Expand Down Expand Up @@ -71,10 +72,13 @@
* @author Moritz Halbritter
* @author Andy Wilkinson
* @author Scott Frederick
* @author Eddú Meléndez
* @since 4.0.0
*/
@AutoConfiguration
@ConditionalOnClass({ RabbitTemplate.class, Channel.class })
@ConditionalOnMissingClass({ "com.rabbitmq.client.amqp.Connection",
"org.springframework.amqp.rabbitmq.client.RabbitAmqpTemplate" })
@EnableConfigurationProperties(RabbitProperties.class)
@Import({ RabbitAnnotationDrivenConfiguration.class, RabbitStreamConfiguration.class })
public final class RabbitAutoConfiguration {
Expand Down
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
org.springframework.boot.amqp.autoconfigure.RabbitAmqpAutoConfiguration
org.springframework.boot.amqp.autoconfigure.RabbitAutoConfiguration
org.springframework.boot.amqp.autoconfigure.health.RabbitHealthContributorAutoConfiguration
org.springframework.boot.amqp.autoconfigure.metrics.RabbitMetricsAutoConfiguration
Loading
Loading