Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

INT-3430 AMQP Outbound: Add Eager Connect

Browse Source 
JIRA: https://jira.spring.io/browse/INT-3430

Add an option to eagerly connect to rabbit when
only using outbound endpoints. Log an ERROR
if the connection can not be established during
context initialization.

Polishing
This commit is contained in:
Gary Russell
2014-06-17 12:31:10 -04:00
committed by Artem Bilan
parent cf94a109b4
commit 1df6d872fa
11 changed files with 140 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParser.java
View File

@@ -50,6 +50,7 @@ public class AmqpOutboundChannelAdapterParser extends AbstractOutboundChannelAda
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "routing-key", true);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "routing-key-expression");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "default-delivery-mode");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "lazy-connect");
IntegrationNamespaceUtils.configureHeaderMapper(element, builder, parserContext, DefaultAmqpHeaderMapper.class, null);

2
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParser.java
View File

@@ -30,6 +30,7 @@ import org.springframework.util.StringUtils;
* @author Oleg Zhurakousky
* @author Gunnar Hillert
* @author Artem Bilan
* @author Gary Russell
*
* @since 2.1
*/
@@ -56,6 +57,7 @@ public class AmqpOutboundGatewayParser extends AbstractConsumerEndpointParser {
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "reply-timeout", "sendTimeout");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "requires-reply");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "default-delivery-mode");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "lazy-connect");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "reply-channel", "outputChannel");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "return-channel");

53
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpoint.java
View File

@@ -20,11 +20,15 @@ import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.amqp.core.MessageDeliveryMode;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.connection.Connection;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.core.RabbitTemplate.ReturnCallback;
import org.springframework.amqp.rabbit.support.CorrelationData;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.context.ApplicationListener;
import org.springframework.context.event.ContextRefreshedEvent;
import org.springframework.expression.Expression;
import org.springframework.expression.ExpressionParser;
import org.springframework.expression.spel.SpelParserConfiguration;
@@ -50,9 +54,10 @@ import org.springframework.util.Assert;
* @since 2.1
*/
public class AmqpOutboundEndpoint extends AbstractReplyProducingMessageHandler
implements RabbitTemplate.ConfirmCallback, ReturnCallback {
implements RabbitTemplate.ConfirmCallback, ReturnCallback, ApplicationListener<ContextRefreshedEvent> {
private static final ExpressionParser expressionParser = new SpelExpressionParser(new SpelParserConfiguration(true, true));
private static final ExpressionParser expressionParser =
new SpelExpressionParser(new SpelParserConfiguration(true, true));
private final AmqpTemplate amqpTemplate;
@@ -85,6 +90,8 @@ public class AmqpOutboundEndpoint extends AbstractReplyProducingMessageHandler
private volatile MessageDeliveryMode defaultDeliveryMode;
private volatile boolean lazyConnect = true;
public AmqpOutboundEndpoint(AmqpTemplate amqpTemplate) {
Assert.notNull(amqpTemplate, "amqpTemplate must not be null");
this.amqpTemplate = amqpTemplate;
@@ -137,6 +144,17 @@ public class AmqpOutboundEndpoint extends AbstractReplyProducingMessageHandler
this.defaultDeliveryMode = defaultDeliveryMode;
}
/**
* Set to {@code false} to attempt to connect during endpoint start;
* default {@code true}, meaning the connection will be attempted
* to be established on the arrival of the first message.
* @param lazyConnect the lazyConnect to set
* @since 4.1
*/
public void setLazyConnect(boolean lazyConnect) {
this.lazyConnect = lazyConnect;
}
@Override
public String getComponentType() {
return expectReply ? "amqp:outbound-gateway" : "amqp:outbound-channel-adapter";
@@ -188,6 +206,25 @@ public class AmqpOutboundEndpoint extends AbstractReplyProducingMessageHandler
}
}
@Override
public void onApplicationEvent(ContextRefreshedEvent event) {
if (!this.lazyConnect && event.getApplicationContext().equals(getApplicationContext())
&& this.amqpTemplate instanceof RabbitTemplate) {
ConnectionFactory connectionFactory = ((RabbitTemplate) this.amqpTemplate).getConnectionFactory();
if (connectionFactory != null) {
try {
Connection connection = connectionFactory.createConnection();
if (connection != null) {
connection.close();
}
}
catch (RuntimeException e) {
logger.error("Failed to eagerly establish the connection.", e);
}
}
}
}
@Override
protected Object handleRequestMessage(Message<?> requestMessage) {
String exchangeName = this.exchangeName;
@@ -228,7 +265,8 @@ public class AmqpOutboundEndpoint extends AbstractReplyProducingMessageHandler
@Override
public org.springframework.amqp.core.Message postProcessMessage(
org.springframework.amqp.core.Message message) throws AmqpException {
headerMapper.fromHeadersToRequest(requestMessage.getHeaders(), message.getMessageProperties());
headerMapper.fromHeadersToRequest(requestMessage.getHeaders(),
message.getMessageProperties());
checkDeliveryMode(requestMessage, message.getMessageProperties());
return message;
}
@@ -241,7 +279,8 @@ public class AmqpOutboundEndpoint extends AbstractReplyProducingMessageHandler
@Override
public org.springframework.amqp.core.Message postProcessMessage(
org.springframework.amqp.core.Message message) throws AmqpException {
headerMapper.fromHeadersToRequest(requestMessage.getHeaders(), message.getMessageProperties());
headerMapper.fromHeadersToRequest(requestMessage.getHeaders(),
message.getMessageProperties());
return message;
}
});
@@ -253,10 +292,12 @@ public class AmqpOutboundEndpoint extends AbstractReplyProducingMessageHandler
"RabbitTemplate implementation is required for publisher confirms");
MessageConverter converter = ((RabbitTemplate) this.amqpTemplate).getMessageConverter();
MessageProperties amqpMessageProperties = new MessageProperties();
org.springframework.amqp.core.Message amqpMessage = converter.toMessage(requestMessage.getPayload(), amqpMessageProperties);
org.springframework.amqp.core.Message amqpMessage =
converter.toMessage(requestMessage.getPayload(), amqpMessageProperties);
this.headerMapper.fromHeadersToRequest(requestMessage.getHeaders(), amqpMessageProperties);
checkDeliveryMode(requestMessage, amqpMessageProperties);
org.springframework.amqp.core.Message amqpReplyMessage = this.amqpTemplate.sendAndReceive(exchangeName, routingKey, amqpMessage);
org.springframework.amqp.core.Message amqpReplyMessage =
this.amqpTemplate.sendAndReceive(exchangeName, routingKey, amqpMessage);
if (amqpReplyMessage == null) {
return null;
}

12
spring-integration-amqp/src/main/resources/org/springframework/integration/amqp/config/spring-integration-amqp-4.1.xsd
View File

@@ -489,6 +489,18 @@ property set to TRUE.
<xsd:union memberTypes="deliveryModeEnumeration xsd:string" />
</xsd:simpleType>
</xsd:attribute>
<xsd:attribute name="lazy-connect" default="true">
<xsd:annotation>
<xsd:documentation>
By default, the connection is established lazily, when the first message is sent. If you wish to detect
connection configuration problems during application initialization, set this to 'false'.
If the eager connection fails, an ERROR log will be emitted.
</xsd:documentation>
</xsd:annotation>
<xsd:simpleType>
<xsd:union memberTypes="xsd:boolean xsd:string" />
</xsd:simpleType>
</xsd:attribute>
</xsd:complexType>
<xsd:simpleType name="deliveryModeEnumeration">

1
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParserTests-context.xml
View File

@@ -24,6 +24,7 @@
<amqp:outbound-channel-adapter id="withHeaderMapperCustomHeaders" channel="requestChannel"
exchange-name="outboundchanneladapter.test.1"
default-delivery-mode="NON_PERSISTENT"
lazy-connect="false"
mapped-request-headers="foo*"/>
<bean id="customHeaderMapper" class="org.mockito.Mockito" factory-method="mock">

33
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParserTests.java
View File

@@ -17,12 +17,18 @@
package org.springframework.integration.amqp.config;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import static org.mockito.Matchers.any;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import java.io.IOException;
@@ -32,8 +38,10 @@ import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.commons.logging.Log;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.Matchers;
import org.mockito.Mockito;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
@@ -47,10 +55,12 @@ import org.springframework.amqp.rabbit.support.CorrelationData;
import org.springframework.amqp.rabbit.support.PublisherCallbackChannel;
import org.springframework.amqp.rabbit.support.PublisherCallbackChannelImpl;
import org.springframework.beans.BeansException;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.parsing.BeanDefinitionParsingException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.event.ContextRefreshedEvent;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.amqp.AmqpHeaders;
import org.springframework.integration.amqp.outbound.AmqpOutboundEndpoint;
@@ -108,6 +118,7 @@ public class AmqpOutboundChannelAdapterParserTests {
assertEquals("amqp:outbound-channel-adapter", ((NamedComponent) handler).getComponentType());
handler.handleMessage(new GenericMessage<String>("foo"));
assertEquals(1, adviceCalled);
assertTrue(TestUtils.getPropertyValue(handler, "lazyConnect", Boolean.class));
}
@Test
@@ -116,6 +127,7 @@ public class AmqpOutboundChannelAdapterParserTests {
AmqpOutboundEndpoint endpoint = TestUtils.getPropertyValue(eventDrivenConsumer, "handler", AmqpOutboundEndpoint.class);
assertNotNull(TestUtils.getPropertyValue(endpoint, "defaultDeliveryMode"));
assertFalse(TestUtils.getPropertyValue(endpoint, "lazyConnect", Boolean.class));
Field amqpTemplateField = ReflectionUtils.findField(AmqpOutboundEndpoint.class, "amqpTemplate");
amqpTemplateField.setAccessible(true);
@@ -335,6 +347,27 @@ public class AmqpOutboundChannelAdapterParserTests {
assertSame(this.context.getBean("customHeaderMapper"), headerMapper);
}
@Test
public void testInt3430FailForNotLazyConnect() {
RabbitTemplate amqpTemplate = mock(RabbitTemplate.class);
ConnectionFactory connectionFactory = mock(ConnectionFactory.class);
RuntimeException toBeThrown = new RuntimeException("Test Connection Exception");
doThrow(toBeThrown).when(connectionFactory).createConnection();
when(amqpTemplate.getConnectionFactory()).thenReturn(connectionFactory);
AmqpOutboundEndpoint handler = new AmqpOutboundEndpoint(amqpTemplate);
Log logger = spy(TestUtils.getPropertyValue(handler, "logger", Log.class));
new DirectFieldAccessor(handler).setPropertyValue("logger", logger);
ApplicationContext context = mock(ApplicationContext.class);
handler.setApplicationContext(context);
handler.afterPropertiesSet();
ContextRefreshedEvent event = new ContextRefreshedEvent(context);
handler.onApplicationEvent(event);
verify(logger, never()).error(Matchers.anyString(), any(RuntimeException.class));
handler.setLazyConnect(false);
handler.onApplicationEvent(event);
verify(logger).error("Failed to eagerly establish the connection.", toBeThrown);
}
public static class FooAdvice extends AbstractRequestHandlerAdvice {

1
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParserTests-context.xml
View File

@@ -39,6 +39,7 @@
exchange-name="si.test.exchange"
routing-key="si.test.binding"
amqp-template="amqpTemplate"
lazy-connect="false"
order="5"
default-delivery-mode="NON_PERSISTENT"
requires-reply="false"

3
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParserTests.java
View File

@@ -76,6 +76,8 @@ public class AmqpOutboundGatewayParserTests {
Long sendTimeout = TestUtils.getPropertyValue(gateway, "messagingTemplate.sendTimeout", Long.class);
assertEquals(Long.valueOf(777), sendTimeout);
assertTrue(TestUtils.getPropertyValue(gateway, "lazyConnect", Boolean.class));
context.close();
}
@@ -88,6 +90,7 @@ public class AmqpOutboundGatewayParserTests {
AmqpOutboundEndpoint endpoint = TestUtils.getPropertyValue(eventDrivernConsumer, "handler", AmqpOutboundEndpoint.class);
assertNotNull(TestUtils.getPropertyValue(endpoint, "defaultDeliveryMode"));
assertFalse(TestUtils.getPropertyValue(endpoint, "lazyConnect", Boolean.class));
assertFalse(TestUtils.getPropertyValue(endpoint, "requiresReply", Boolean.class));

7
spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationObjectSupport.java
View File

@@ -192,6 +192,13 @@ public abstract class IntegrationObjectSupport implements BeanNameAware, NamedCo
return this.applicationContext == null ? null : this.applicationContext.getId();
}
/**
* @return the applicationContext
*/
protected ApplicationContext getApplicationContext() {
return applicationContext;
}
/**
* @see IntegrationContextUtils#getIntegrationProperties(BeanFactory)
* @return The global integration properties.

22
src/reference/docbook/amqp.xml
View File

@@ -418,7 +418,8 @@ public Object handle(@Payload String payload, @Header(AmqpHeaders.CHANNEL) Chann
confirm-correlation-expression=""]]><co id="amqp-outbound-channel-adapter-xml-8-co" linkends="amqp-outbound-channel-adapter-xml-8" /><![CDATA[
confirm-ack-channel=""]]><co id="amqp-outbound-channel-adapter-xml-9-co" linkends="amqp-outbound-channel-adapter-xml-9" /><![CDATA[
confirm-nack-channel=""]]><co id="amqp-outbound-channel-adapter-xml-10-co" linkends="amqp-outbound-channel-adapter-xml-10" /><![CDATA[
return-channel=""]]><co id="amqp-outbound-channel-adapter-xml-11-co" linkends="amqp-outbound-channel-adapter-xml-11" /><![CDATA[/>]]>
return-channel=""]]><co id="amqp-outbound-channel-adapter-xml-11-co" linkends="amqp-outbound-channel-adapter-xml-11" /><![CDATA[
lazy-connect="true"]]><co id="amqp-outbound-channel-adapter-xml-12-co" linkends="amqp-outbound-channel-adapter-xml-12" /><![CDATA[/>]]>
</programlisting>
<para>
<calloutlist>
@@ -505,6 +506,13 @@ public Object handle(@Payload String payload, @Header(AmqpHeaders.CHANNEL) Chann
for each endpoint.
</important>
</callout>
<callout arearefs="amqp-outbound-channel-adapter-xml-12-co" id="amqp-outbound-channel-adapter-xml-12">
<para>When set to <code>false</code>, the endpoint will attempt to connect to the
broker during application context initialization. This allows "fail fast" detection of
bad configuration, but will also cause initialization to fail if the broker is down.
When true (default), the connection is established (if it doesn't already exist because
some other component established it) when the first message is sent.</para>
</callout>
</calloutlist>
</para>
</section>
@@ -521,8 +529,9 @@ public Object handle(@Payload String payload, @Header(AmqpHeaders.CHANNEL) Chann
reply-channel=""]]><co id="amqp-outbound-gateway-adapter-xml-6-co" linkends="amqp-outbound-gateway-adapter-xml-6" /><![CDATA[
routing-key=""]]><co id="amqp-outbound-gateway-adapter-xml-7-co" linkends="amqp-outbound-gateway-adapter-xml-7" /><![CDATA[
routing-key-expression=""]]><co id="amqp-outbound-gateway-adapter-xml-8-co" linkends="amqp-outbound-gateway-adapter-xml-8" /><![CDATA[
default-delivery-mode""]]><co id="amqp-outbound-gateway-adapter-xml-8a-co" linkends="amqp-outbound-channel-adapter-xml-8a" /><![CDATA[
return-channel=""]]><co id="amqp-outbound-gateway-adapter-xml-9-co" linkends="amqp-outbound-gateway-adapter-xml-9" /><![CDATA[/>]]>
default-delivery-mode""]]><co id="amqp-outbound-gateway-adapter-xml-8a-co" linkends="amqp-outbound-gateway-adapter-xml-8a" /><![CDATA[
return-channel=""]]><co id="amqp-outbound-gateway-adapter-xml-9-co" linkends="amqp-outbound-gateway-adapter-xml-9" /><![CDATA[
lazy-connect="true"]]><co id="amqp-outbound-gateway-adapter-xml-10-co" linkends="amqp-outbound-gateway-adapter-xml-10" /><![CDATA[/>]]>
</programlisting>
<para>
<calloutlist>
@@ -593,6 +602,13 @@ public Object handle(@Payload String payload, @Header(AmqpHeaders.CHANNEL) Chann
for each endpoint.
</important>
</callout>
<callout arearefs="amqp-outbound-gateway-adapter-xml-10-co" id="amqp-outbound-gateway-adapter-xml-10">
<para>When set to <code>false</code>, the endpoint will attempt to connect to the
broker during application context initialization. This allows "fail fast" detection of
bad configuration, but will also cause initialization to fail if the broker is down.
When true (default), the connection is established (if it doesn't already exist because
some other component established it) when the first message is sent.</para>
</callout>
</calloutlist>
</para>
<note>

14
src/reference/docbook/whats-new.xml
View File

@@ -9,4 +9,18 @@
in more details, please see the Issue Tracker tickets that
were resolved as part of the 4.1 development process.
</para>
<section id="4.1-general">
<title>General Changes</title>
<section>
<title>AMQP Outbound Endpoints</title>
<para>
The AMQP outbound endpoints support a new property <code>lazy-connect</code>
(default true). When true, the connection to the broker is not established
until the first message arrives (assuming there are no inbound endpoints, which
always attempt to establish the connection during startup). When set the 'false' an
attempt to establish the connection is made during application startup.
See <xref linkend="amqp"/> for more information.
</para>
</section>
</section>
</chapter>