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()); + } + +}