From 9005938ab5dc49bcda28ad5d6e28a91feb82034a Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Tue, 28 Jan 2014 19:54:05 +0200 Subject: [PATCH] INT-3278 AMQP PubSub Redeclare After Conn. Loss JIRA: https://jira.springsource.org/browse/INT-3278 A broker-named auto-delete queue is not suitable for use with a message listener container. If the connection is lost, the queue is deleted and the container cannot recover its consumer. Use an AnonymousQueue, which has the same semantics as the current queue, but can be redeclared after a connection loss. Register the channel as a ConnectionListener and redeclare the queue and binding as necessary. Note: You cannot currently unregister a listener (AMQP-367) so if the context is destroyed, the channel will remain a listener. However, the isRunning() flag can be used to avoid declaring the channels when in that state. INT-3278 Polishing * PR Comments * Remove the listener when the bean is destroy()ed. --- .../AbstractSubscribableAmqpChannel.java | 18 ++++- .../channel/PublishSubscribeAmqpChannel.java | 65 +++++++++++++-- .../amqp/StubRabbitConnectionFactory.java | 9 +++ .../amqp/channel/ChannelTests-context.xml | 16 ++++ .../amqp/channel/ChannelTests.java | 80 +++++++++++++++++++ 5 files changed, 181 insertions(+), 7 deletions(-) create mode 100644 spring-integration-amqp/src/test/java/org/springframework/integration/amqp/channel/ChannelTests-context.xml create mode 100644 spring-integration-amqp/src/test/java/org/springframework/integration/amqp/channel/ChannelTests.java diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/AbstractSubscribableAmqpChannel.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/AbstractSubscribableAmqpChannel.java index 03755f7efd..729ccc72ea 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/AbstractSubscribableAmqpChannel.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/AbstractSubscribableAmqpChannel.java @@ -23,6 +23,7 @@ import org.springframework.amqp.core.AmqpAdmin; import org.springframework.amqp.core.AmqpTemplate; import org.springframework.amqp.core.MessageListener; import org.springframework.amqp.core.Queue; +import org.springframework.amqp.rabbit.connection.ConnectionFactory; import org.springframework.amqp.rabbit.core.RabbitAdmin; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer; @@ -60,6 +61,10 @@ abstract class AbstractSubscribableAmqpChannel extends AbstractAmqpChannel imple private volatile Integer maxSubscribers; + private final AmqpAdmin admin; + + private final ConnectionFactory connectionFactory; + public AbstractSubscribableAmqpChannel(String channelName, SimpleMessageListenerContainer container, AmqpTemplate amqpTemplate) { this(channelName, container, amqpTemplate, false); } @@ -73,6 +78,8 @@ abstract class AbstractSubscribableAmqpChannel extends AbstractAmqpChannel imple this.channelName = channelName; this.container = container; this.isPubSub = isPubSub; + this.connectionFactory = container.getConnectionFactory(); + this.admin = new RabbitAdmin(this.connectionFactory); } /** @@ -87,6 +94,14 @@ abstract class AbstractSubscribableAmqpChannel extends AbstractAmqpChannel imple } } + protected AmqpAdmin getAdmin() { + return admin; + } + + protected ConnectionFactory getConnectionFactory() { + return connectionFactory; + } + @Override public boolean subscribe(MessageHandler handler) { return this.dispatcher.addHandler(handler); @@ -108,8 +123,7 @@ abstract class AbstractSubscribableAmqpChannel extends AbstractAmqpChannel imple Integer.class); } this.setMaxSubscribers(this.maxSubscribers); - AmqpAdmin admin = new RabbitAdmin(this.container.getConnectionFactory()); - Queue queue = this.initializeQueue(admin, this.channelName); + Queue queue = this.initializeQueue(this.admin, this.channelName); this.container.setQueues(queue); MessageConverter converter = (this.getAmqpTemplate() instanceof RabbitTemplate) ? ((RabbitTemplate) this.getAmqpTemplate()).getMessageConverter() diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PublishSubscribeAmqpChannel.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PublishSubscribeAmqpChannel.java index 30da510713..d95cde7047 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PublishSubscribeAmqpChannel.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PublishSubscribeAmqpChannel.java @@ -18,10 +18,15 @@ package org.springframework.integration.amqp.channel; import org.springframework.amqp.core.AmqpAdmin; import org.springframework.amqp.core.AmqpTemplate; +import org.springframework.amqp.core.AnonymousQueue; import org.springframework.amqp.core.Binding; import org.springframework.amqp.core.BindingBuilder; import org.springframework.amqp.core.FanoutExchange; import org.springframework.amqp.core.Queue; +import org.springframework.amqp.rabbit.connection.Connection; +import org.springframework.amqp.rabbit.connection.ConnectionFactory; +import org.springframework.amqp.rabbit.connection.ConnectionListener; +import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer; import org.springframework.integration.dispatcher.AbstractDispatcher; import org.springframework.integration.dispatcher.BroadcastingDispatcher; @@ -31,10 +36,15 @@ import org.springframework.integration.dispatcher.BroadcastingDispatcher; * @author Gary Russell * @since 2.1 */ -public class PublishSubscribeAmqpChannel extends AbstractSubscribableAmqpChannel { +public class PublishSubscribeAmqpChannel extends AbstractSubscribableAmqpChannel implements ConnectionListener { private volatile FanoutExchange exchange; + private final Queue queue = new AnonymousQueue(); + + private volatile Binding binding; + + private volatile boolean initialized; public PublishSubscribeAmqpChannel(String channelName, SimpleMessageListenerContainer container, AmqpTemplate amqpTemplate) { super(channelName, container, amqpTemplate, true); @@ -60,10 +70,31 @@ public class PublishSubscribeAmqpChannel extends AbstractSubscribableAmqpChannel this.exchange = new FanoutExchange(exchangeName); } admin.declareExchange(this.exchange); - Queue queue = admin.declareQueue(); - Binding binding = BindingBuilder.bind(queue).to(this.exchange); - admin.declareBinding(binding); - return queue; + admin.declareQueue(this.queue); + this.binding = BindingBuilder.bind(this.queue).to(this.exchange); + admin.declareBinding(this.binding); + if (!this.initialized && this.getAmqpTemplate() instanceof RabbitTemplate) { + ConnectionFactory connectionFactory = this.getConnectionFactory(); + if (connectionFactory != null) { + connectionFactory.addConnectionListener(this); + } + } + this.initialized = true; + return this.queue; + } + + private void doDeclares() { + if (this.isRunning()) { + AmqpAdmin admin = this.getAdmin(); + if (admin != null) { + if (this.queue != null) { + admin.declareQueue(this.queue); + } + if (this.binding != null) { + admin.declareBinding(this.binding); + } + } + } } @Override @@ -76,4 +107,28 @@ public class PublishSubscribeAmqpChannel extends AbstractSubscribableAmqpChannel return (this.exchange != null) ? this.exchange.getName() : ""; } + @Override + public void destroy() throws Exception { + super.destroy(); + if (this.getConnectionFactory() != null) { + this.getConnectionFactory().removeConnectionListener(this); + this.initialized = false; + } + } + + @Override + public void start() { + this.doDeclares(); // connection may have been lost while we were stopped + super.start(); + } + + @Override + public void onCreate(Connection connection) { + doDeclares(); + } + + @Override + public void onClose(Connection connection) { + } + } diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/StubRabbitConnectionFactory.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/StubRabbitConnectionFactory.java index 18f5446adb..0dd062d57b 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/StubRabbitConnectionFactory.java +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/StubRabbitConnectionFactory.java @@ -78,6 +78,15 @@ public class StubRabbitConnectionFactory implements ConnectionFactory { public void addConnectionListener(ConnectionListener listener) { } + @Override + public boolean removeConnectionListener(ConnectionListener listener) { + return false; + } + + @Override + public void clearConnectionListeners() { + } + private static class StubConnection implements Connection { diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/channel/ChannelTests-context.xml b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/channel/ChannelTests-context.xml new file mode 100644 index 0000000000..c1797d3203 --- /dev/null +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/channel/ChannelTests-context.xml @@ -0,0 +1,16 @@ + + + + + + + + diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/channel/ChannelTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/channel/ChannelTests.java new file mode 100644 index 0000000000..46cfc7dabf --- /dev/null +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/channel/ChannelTests.java @@ -0,0 +1,80 @@ +/* + * Copyright 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. + * 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.integration.amqp.channel; + +import static org.junit.Assert.assertEquals; + +import java.util.Collection; +import java.util.concurrent.CyclicBarrier; +import java.util.concurrent.TimeUnit; + +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.ConfigurableApplicationContext; +import org.springframework.integration.test.util.TestUtils; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageHandler; +import org.springframework.messaging.MessagingException; +import org.springframework.messaging.support.GenericMessage; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +/** + * @author Gary Russell + * @since 4.0 + * + */ +@ContextConfiguration +@RunWith(SpringJUnit4ClassRunner.class) +public class ChannelTests { + + @Autowired + private PublishSubscribeAmqpChannel channel; + + @Autowired + private CachingConnectionFactory factory; + + @Autowired + private ConfigurableApplicationContext context; + + @Test + public void pubSubLostConnectionTest() throws Exception { + final CyclicBarrier latch = new CyclicBarrier(2); + channel.subscribe(new MessageHandler() { + + @Override + public void handleMessage(Message message) throws MessagingException { + try { + latch.await(10, TimeUnit.SECONDS); + } + catch (Exception e) { + } + } + }); + channel.send(new GenericMessage("foo")); + latch.await(10, TimeUnit.SECONDS); + latch.reset(); + factory.destroy(); + channel.send(new GenericMessage("bar")); + latch.await(10, TimeUnit.SECONDS); + context.close(); + assertEquals(0, TestUtils.getPropertyValue(factory, "connectionListener.delegates", Collection.class).size()); + } + +}