From 8fa1eee8f595d347b5053fe60d987370180034f0 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Fri, 20 Jul 2018 16:58:37 -0400 Subject: [PATCH] Add SimplePropertyValueConnectionNameStrategy * Unused imports * Doc polishing * Polishing - PR comments; early use of environment --- ...lePropertyValueConnectionNameStrategy.java | 62 +++++++++++++++++++ .../EnableRabbitIntegrationTests.java | 21 +++++++ src/reference/asciidoc/amqp.adoc | 31 +++++++++- 3 files changed, 112 insertions(+), 2 deletions(-) create mode 100644 spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/SimplePropertyValueConnectionNameStrategy.java diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/SimplePropertyValueConnectionNameStrategy.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/SimplePropertyValueConnectionNameStrategy.java new file mode 100644 index 00000000..6a3c07fa --- /dev/null +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/SimplePropertyValueConnectionNameStrategy.java @@ -0,0 +1,62 @@ +/* + * Copyright 2018 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 + * + * http://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.amqp.rabbit.connection; + +import org.springframework.context.EnvironmentAware; +import org.springframework.core.env.Environment; +import org.springframework.util.Assert; + +/** + * A {@link ConnectionNameStrategy} that returns the value of a (required) property. If + * the property does not exist, the connection will be given the name of the property. + * + * @author Gary Russell + * @since 2.1 + * + */ +public class SimplePropertyValueConnectionNameStrategy implements ConnectionNameStrategy, EnvironmentAware { + + private final String propertyName; + + private String propertyValue; + + private Environment environment; + + public SimplePropertyValueConnectionNameStrategy(String propertyName) { + Assert.notNull(propertyName, "'propertyName' cannot be null"); + this.propertyName = propertyName; + } + + @Override + public void setEnvironment(Environment environment) { + this.environment = environment; + } + + @Override + public String obtainNewConnectionName(ConnectionFactory connectionFactory) { + if (this.propertyValue == null) { + if (this.environment != null) { + this.propertyValue = this.environment.getProperty(this.propertyName); + } + if (this.propertyValue == null) { + this.propertyValue = this.propertyName; + } + } + return this.propertyValue; + } + +} diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/EnableRabbitIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/EnableRabbitIntegrationTests.java index 6518612a..20234ac2 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/EnableRabbitIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/EnableRabbitIntegrationTests.java @@ -72,7 +72,10 @@ import org.springframework.amqp.rabbit.config.DirectRabbitListenerContainerFacto import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory; import org.springframework.amqp.rabbit.config.SimpleRabbitListenerEndpoint; import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; +import org.springframework.amqp.rabbit.connection.Connection; import org.springframework.amqp.rabbit.connection.ConnectionFactory; +import org.springframework.amqp.rabbit.connection.ConnectionNameStrategy; +import org.springframework.amqp.rabbit.connection.SimplePropertyValueConnectionNameStrategy; import org.springframework.amqp.rabbit.core.RabbitAdmin; import org.springframework.amqp.rabbit.core.RabbitManagementTemplate; import org.springframework.amqp.rabbit.core.RabbitTemplate; @@ -128,6 +131,7 @@ import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.TestContext; import org.springframework.test.context.TestExecutionListeners; import org.springframework.test.context.TestExecutionListeners.MergeMode; +import org.springframework.test.context.TestPropertySource; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; import org.springframework.test.context.support.AbstractTestExecutionListener; import org.springframework.transaction.PlatformTransactionManager; @@ -151,6 +155,7 @@ import com.rabbitmq.client.Channel; @DirtiesContext @TestExecutionListeners(mergeMode = MergeMode.MERGE_WITH_DEFAULTS, listeners = EnableRabbitIntegrationTests.DeleteQueuesExecutionListener.class) +@TestPropertySource(properties = "spring.application.name=testConnectionName") public class EnableRabbitIntegrationTests { @ClassRule @@ -209,6 +214,9 @@ public class EnableRabbitIntegrationTests { @Autowired private MetaListener metaListener; + @Autowired + private CachingConnectionFactory connectionFactory; + @BeforeClass public static void setUp() { System.setProperty(RabbitListenerAnnotationBeanPostProcessor.RABBIT_EMPTY_STRING_ARGUMENTS_PROPERTY, @@ -796,6 +804,13 @@ public class EnableRabbitIntegrationTests { assertThat(new String(config.message.getBody()), equalTo("bar")); } + @Test + public void connectionName() { + Connection conn = this.connectionFactory.createConnection(); + conn.close(); + assertThat(conn.getDelegate().getClientProvidedName(), equalTo("testConnectionName")); + } + interface TxService { @Transactional @@ -1243,6 +1258,11 @@ public class EnableRabbitIntegrationTests { private final CountDownLatch noListenerLatch = new CountDownLatch(1); + @Bean + public ConnectionNameStrategy cns() { + return new SimplePropertyValueConnectionNameStrategy("spring.application.name"); + } + @Bean public static ProxyListenerBPP listenerProxier() { // note static return new ProxyListenerBPP(); @@ -1449,6 +1469,7 @@ public class EnableRabbitIntegrationTests { executor.setThreadNamePrefix("rabbitClientThread-"); executor.afterPropertiesSet(); connectionFactory.setExecutor(executor); + connectionFactory.setConnectionNameStrategy(cns()); return connectionFactory; } diff --git a/src/reference/asciidoc/amqp.adoc b/src/reference/asciidoc/amqp.adoc index 8eeed76e..f7e22890 100644 --- a/src/reference/asciidoc/amqp.adoc +++ b/src/reference/asciidoc/amqp.adoc @@ -346,6 +346,8 @@ Here's an example with a custom thread factory that prefixes thread names with ` ---- +===== Naming Connections + Starting with _version 1.7_ a `ConnectionNameStrategy` is provided for the injection into the `AbstractionConnectionFactory`. The generated name is used for the application-specific identification of the target RabbitMQ connection. The connection name is displayed in the management UI if the RabbitMQ server supports it. @@ -361,6 +363,31 @@ The `ConnectionFactory` argument can be used to distinguish target connection na By default, the `beanName` of the `AbstractConnectionFactory`, a hex String representing the object, and an internal counter are used to generate the `connection_name`. The `` namespace component is also supplied with the `connection-name-strategy` attribute. +An implementation `SimplePropertyValueConnectionNameStrategy` is provided that sets the connection name to an application property. +Declare it as a `@Bean` and inject it into the connection factory: + +[source, java] +---- +@Bean +public ConnectionNameStrategy cns() { + return new SimplePropertyValueConnectionNameStrategy("spring.application.name"); +} + +@Bean +public ConnectionFactory rabbitConnectionFactory(ConnectionNameStrategy cns) { + CachingConnectionFactory connectionFactory = new CachingConnectionFactory(); + ... + connectionFactory.setConnectionNameStrategy(cns); + return connectionFactory; +} +---- + +The property must exist in the application context's `Environment`. + +NOTE: When using Spring Boot and it's autoconfigured connection factory, it is only necessary to declare the `ConnectionNameStrategy` `@Bean`. +Boot will auto-detect the bean and wire it into the factory. + +===== Blocked Connections and Resource Constraints The connection might be blocked for interaction from the Broker according to the https://www.rabbitmq.com/memory.html[Memory Alarm]. Starting with _version 2.0_, the `org.springframework.amqp.rabbit.connection.Connection` can be supplied with `com.rabbitmq.client.BlockedListener` s to to be notified for connection blocked and unblocked events. @@ -370,8 +397,8 @@ These allow you to provide application logic to react appropriately to problems IMPORTANT: When the application is configured with a single `CachingConnectionFactory`, as it is by default with Spring Boot auto-configuration, the application will stop working when the connection is blocked by the Broker. And when it is blocked by the Broker, any its clients stop to work. If we have producers and consumers in the same application, we may end up with a deadlock when producers are blocking the connection because there are no resources on the Broker anymore and consumers can't free them because the connection is blocked. -To mitigate the problem, there is just enough to have one more separate `CachingConnectionFactory` instance with the same options - one for producers and one for consumers. -The separate `CachingConnectionFactory` isn't recommended for transactional producers, since they should reuse a `Channel` associated with the consumer transactions. +To mitigate the problem, it is suggested to have one more separate `CachingConnectionFactory` instance with the same options - one for producers and one for consumers. +A separate `CachingConnectionFactory` isn't possible for transactional producers that execute on a consumer thread, since they should reuse the `Channel` associated with the consumer transactions. Starting with _version 2.0.2_, the `RabbitTemplate` has a configuration option to automatically use a second connection factory, unless transactions are being used. See <> for more information.