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

INT-3451 AMQP - 'missing-queues-fatal' Attribute

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

Add `missing-queues-fatal` attribute to the AMQP
inbound endpoints and AMQP-backed channel.
This commit is contained in:
Gary Russell
2014-06-21 11:10:47 -04:00
committed by Artem Bilan
parent 3c063a265b
commit 24680deeb4
13 changed files with 59 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
build.gradle
View File

@@ -109,7 +109,7 @@ subprojects { subproject ->
slf4jVersion = "1.7.6"
smack3Version = '3.2.1'
smackVersion = '4.0.0'
springAmqpVersion = project.hasProperty('springAmqpVersion') ? project.springAmqpVersion : '1.3.4.RELEASE'
springAmqpVersion = project.hasProperty('springAmqpVersion') ? project.springAmqpVersion : '1.3.5.BUILD-SNAPSHOT'
springDataMongoVersion = '1.5.0.RELEASE'
springDataRedisVersion = '1.3.0.RELEASE'
springGemfireVersion = '1.4.0.RELEASE'

3
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AbstractAmqpInboundAdapterParser.java
View File

@@ -53,7 +53,8 @@ abstract class AbstractAmqpInboundAdapterParser extends AbstractSingleBeanDefini
"recovery-interval",
"receive-timeout",
"shutdown-timeout",
"tx-size"
"tx-size",
"missing-queues-fatal"
};
private static final String[] CONTAINER_REFERENCE_ATTRIBUTES = {

9
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpChannelFactoryBean.java
View File

@@ -124,6 +124,8 @@ public class AmqpChannelFactoryBean extends AbstractFactoryBean<AbstractAmqpChan
private volatile Integer maxSubscribers;
private volatile Boolean missingQueuesFatal;
public AmqpChannelFactoryBean() {
this(true);
@@ -297,6 +299,10 @@ public class AmqpChannelFactoryBean extends AbstractFactoryBean<AbstractAmqpChan
this.maxSubscribers = maxSubscribers;
}
public void setMissingQueuesFatal(Boolean missingQueuesFatal) {
this.missingQueuesFatal = missingQueuesFatal;
}
@Override
public Class<?> getObjectType() {
return (this.channel != null) ? this.channel.getClass() : AbstractAmqpChannel.class;
@@ -406,6 +412,9 @@ public class AmqpChannelFactoryBean extends AbstractFactoryBean<AbstractAmqpChan
if (this.txSize != null) {
container.setTxSize(this.txSize);
}
if (this.missingQueuesFatal != null) {
container.setMissingQueuesFatal(this.missingQueuesFatal);
}
return container;
}

3
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpChannelParser.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2013 the original author or authors.
* Copyright 2002-2014 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.
@@ -68,6 +68,7 @@ public class AmqpChannelParser extends AbstractChannelParser {
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "queue-name");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "receive-timeout");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "recovery-interval");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "missing-queues-fatal");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "shutdown-timeout");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "task-executor");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "transaction-attribute");

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

@@ -833,6 +833,21 @@ standard headers to also be mapped.
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="missing-queues-fatal">
<xsd:annotation>
<xsd:appinfo>
<xsd:documentation>
If 'true', and none of the queues are available on the broker, the container will throw a fatal exception during
startup and will stop if the queues are deleted when the container is running (after making 3 attempts to
passively declare the queues). If false, the container will not throw an exception and go into recovery mode,
attempting to restart according to the 'revcovery-interval'. Default 'true'.
</xsd:documentation>
</xsd:appinfo>
</xsd:annotation>
<xsd:simpleType>
<xsd:union memberTypes="xsd:boolean xsd:string" />
</xsd:simpleType>
</xsd:attribute>
<xsd:attributeGroup ref="integration:smartLifeCycleAttributeGroup"/>
</xsd:attributeGroup>

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

@@ -17,7 +17,7 @@
<bean id="rabbitConnectionFactory" class="org.springframework.integration.amqp.StubRabbitConnectionFactory"/>
<amqp:channel id="channelWithSubscriberLimit" max-subscribers="1" />
<amqp:channel id="channelWithSubscriberLimit" max-subscribers="1" missing-queues-fatal="false" />
<amqp:publish-subscribe-channel id="pubSub" />

4
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpChannelParserTests.java
View File

@@ -17,7 +17,9 @@
package org.springframework.integration.amqp.config;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertTrue;
import java.util.List;
@@ -57,6 +59,7 @@ public class AmqpChannelParserTests {
Object mbf = context.getBean(IntegrationUtils.INTEGRATION_MESSAGE_BUILDER_FACTORY_BEAN_NAME);
assertSame(mbf, TestUtils.getPropertyValue(channel, "dispatcher.messageBuilderFactory"));
assertSame(mbf, TestUtils.getPropertyValue(channel, "container.messageListener.messageBuilderFactory"));
assertTrue(TestUtils.getPropertyValue(channel, "container.missingQueuesFatal", Boolean.class));
}
@Test
@@ -64,6 +67,7 @@ public class AmqpChannelParserTests {
MessageChannel channel = context.getBean("channelWithSubscriberLimit", MessageChannel.class);
assertEquals(1, TestUtils.getPropertyValue(
TestUtils.getPropertyValue(channel, "dispatcher"), "maxSubscribers", Integer.class).intValue());
assertFalse(TestUtils.getPropertyValue(channel, "container.missingQueuesFatal", Boolean.class));
}

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

@@ -22,7 +22,7 @@
<amqp:inbound-channel-adapter id="rabbitInbound" queue-names="inboundchanneladapter.test.1"/>
<amqp:inbound-channel-adapter id="autoStartFalse" queue-names="inboundchanneladapter.test.2"
auto-startup="false" phase="123" acknowledge-mode="${ackMode}"/>
auto-startup="false" phase="123" acknowledge-mode="${ackMode}" missing-queues-fatal="false" />
<amqp:inbound-channel-adapter id="withHeaderMapperStandardAndCustomHeaders" channel="requestChannel" queue-names="inboundchanneladapter.test.2"
auto-startup="false" phase="123"

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

@@ -17,6 +17,7 @@
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.assertTrue;
@@ -62,6 +63,7 @@ public class AmqpInboundChannelAdapterParserTests {
assertEquals(AmqpInboundChannelAdapter.class, adapter.getClass());
assertEquals(Boolean.TRUE, TestUtils.getPropertyValue(adapter, "autoStartup"));
assertEquals(0, TestUtils.getPropertyValue(adapter, "phase"));
assertTrue(TestUtils.getPropertyValue(adapter, "messageListenerContainer.missingQueuesFatal", Boolean.class));
}
@Test
@@ -70,6 +72,7 @@ public class AmqpInboundChannelAdapterParserTests {
assertEquals(Boolean.FALSE, TestUtils.getPropertyValue(adapter, "autoStartup"));
assertEquals(123, TestUtils.getPropertyValue(adapter, "phase"));
assertEquals(AcknowledgeMode.NONE, TestUtils.getPropertyValue(adapter, "messageListenerContainer.acknowledgeMode"));
assertFalse(TestUtils.getPropertyValue(adapter, "messageListenerContainer.missingQueuesFatal", Boolean.class));
}
@Test

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

@@ -22,6 +22,7 @@
<si-amqp:inbound-gateway id="autoStartFalseGateway" request-channel="requests" queue-names="test"
connection-factory="rabbitConnectionFactory" message-converter="testConverter"
missing-queues-fatal="false"
auto-startup="false" phase="123"/>
<si-amqp:inbound-gateway id="withHeaderMapper" request-channel="requestChannel" queue-names="inboundchanneladapter.test.2"

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

@@ -17,6 +17,7 @@
package org.springframework.integration.amqp.config;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertTrue;
@@ -78,6 +79,7 @@ public class AmqpInboundGatewayParserTests {
assertEquals(0, TestUtils.getPropertyValue(gateway, "phase"));
assertEquals(Long.valueOf(1234L), TestUtils.getPropertyValue(gateway, "replyTimeout", Long.class));
assertEquals(Long.valueOf(1234L), TestUtils.getPropertyValue(gateway, "messagingTemplate.receiveTimeout", Long.class));
assertTrue(TestUtils.getPropertyValue(gateway, "messageListenerContainer.missingQueuesFatal", Boolean.class));
}
@Test
@@ -85,6 +87,7 @@ public class AmqpInboundGatewayParserTests {
Object gateway = context.getBean("autoStartFalseGateway");
assertEquals(Boolean.FALSE, TestUtils.getPropertyValue(gateway, "autoStartup"));
assertEquals(123, TestUtils.getPropertyValue(gateway, "phase"));
assertFalse(TestUtils.getPropertyValue(gateway, "messageListenerContainer.missingQueuesFatal", Boolean.class));
}
@SuppressWarnings("rawtypes")

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

@@ -72,6 +72,7 @@
prefetch-count=""]]><co id="amqp-inbound-channel-adapter-xml-18-co" linkends="amqp-inbound-channel-adapter-xml-18" /><![CDATA[
receive-timeout=""]]><co id="amqp-inbound-channel-adapter-xml-19-co" linkends="amqp-inbound-channel-adapter-xml-19" /><![CDATA[
recovery-interval=""]]><co id="amqp-inbound-channel-adapter-xml-20-co" linkends="amqp-inbound-channel-adapter-xml-20" /><![CDATA[
missing-queues-fatal=""]]><co id="amqp-inbound-channel-adapter-xml-20d-co" linkends="amqp-inbound-channel-adapter-xml-20d" /><![CDATA[
shutdown-timeout=""]]><co id="amqp-inbound-channel-adapter-xml-21-co" linkends="amqp-inbound-channel-adapter-xml-21" /><![CDATA[
task-executor=""]]><co id="amqp-inbound-channel-adapter-xml-22-co" linkends="amqp-inbound-channel-adapter-xml-22" /><![CDATA[
transaction-attribute=""]]><co id="amqp-inbound-channel-adapter-xml-23-co" linkends="amqp-inbound-channel-adapter-xml-23" /><![CDATA[
@@ -213,6 +214,14 @@
milliseconds).
<emphasis>Optional (Defaults to 5000)</emphasis>.</para>
</callout>
<callout arearefs="amqp-inbound-channel-adapter-xml-20d-co" id="amqp-inbound-channel-adapter-xml-20d">
<para> If 'true', and none of the queues are available on the broker, the container
will throw a fatal exception during startup and will stop if the queues are deleted when
the container is running (after making 3 attempts to passively declare the queues). If false,
the container will not throw an exception and go into recovery mode, attempting to restart according
to the <code>revcovery-interval</code>.
<emphasis>Optional (Defaults to <code>true</code>)</emphasis>.</para>
</callout>
<callout arearefs="amqp-inbound-channel-adapter-xml-21-co" id="amqp-inbound-channel-adapter-xml-21">
<para>The time to wait for workers in milliseconds after the
underlying <interface>SimpleMessageListenerContainer</interface>

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

@@ -11,7 +11,14 @@
</para>
<section id="4.1-general">
<title>General Changes</title>
<section id="4.1-amqp-outbound-lazy-connect">
<section id="4.1-amqp-inbound-missing-queues">
<title>AMQP Inbound Endpoints, Channel</title>
<para>
Elements that utilize a message listener container (inbound endpoints, channel)
now support the <code>missing-queues-fatal</code> attribute.
See <xref linkend="amqp"/> for more information.
</para>
</section>
<title>AMQP Outbound Endpoints</title>
<para>
The AMQP outbound endpoints support a new property <code>lazy-connect</code>