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

INT-3278 AMQP PubSub Redeclare After Conn. Loss

Browse Source 
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.
This commit is contained in:
Gary Russell
2014-01-28 19:54:05 +02:00
committed by Artem Bilan
parent 7e7fbe7247
commit 9005938ab5
5 changed files with 181 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/AbstractSubscribableAmqpChannel.java
View File

@@ -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()

65
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PublishSubscribeAmqpChannel.java
View File

@@ -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) {
}
}

9
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/StubRabbitConnectionFactory.java
View File

@@ -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 {

16
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/channel/ChannelTests-context.xml Normal file
View File

@@ -0,0 +1,16 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:int="http://www.springframework.org/schema/integration"
xmlns:int-amqp="http://www.springframework.org/schema/integration/amqp"
xmlns:rabbit="http://www.springframework.org/schema/rabbit"
xsi:schemaLocation="http://www.springframework.org/schema/integration/amqp http://www.springframework.org/schema/integration/amqp/spring-integration-amqp.xsd
http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/rabbit http://www.springframework.org/schema/rabbit/spring-rabbit.xsd
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
<int-amqp:publish-subscribe-channel id="foo" />
<rabbit:connection-factory id="rabbitConnectionFactory" />
</beans>

80
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/channel/ChannelTests.java Normal file
View File

@@ -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<String>("foo"));
latch.await(10, TimeUnit.SECONDS);
latch.reset();
factory.destroy();
channel.send(new GenericMessage<String>("bar"));
latch.await(10, TimeUnit.SECONDS);
context.close();
assertEquals(0, TestUtils.getPropertyValue(factory, "connectionListener.delegates", Collection.class).size());
}
}