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

AMQP-122: defer declarations until first use of Connection

Browse Source
This commit is contained in:
Dave Syer
2011-03-22 12:39:59 +00:00
parent 663cbdffc2
commit 983dca42ff
15 changed files with 153 additions and 348 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/AbstractRabbitConfiguration.java
View File

@@ -46,7 +46,9 @@ public abstract class AbstractRabbitConfiguration extends AbstractAmqpConfigurat
@Bean
public AmqpAdmin amqpAdmin() {
return new RabbitAdmin(rabbitTemplate().getConnectionFactory());
RabbitAdmin rabbitAdmin = new RabbitAdmin(rabbitTemplate().getConnectionFactory());
rabbitAdmin.setAutoStartup(true);
return rabbitAdmin;
}
}

7
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/AdminParser.java
View File

@@ -27,8 +27,6 @@ class AdminParser extends AbstractSingleBeanDefinitionParser {
private static final String CONNECTION_FACTORY_ATTRIBUTE = "connection-factory";
private static final String PHASE_ATTRIBUTE = "phase";
private static final String AUTO_STARTUP_ATTRIBUTE = "auto-startup";
@Override
@@ -62,11 +60,6 @@ class AdminParser extends AbstractSingleBeanDefinitionParser {
}
String attributeValue;
attributeValue = element.getAttribute(PHASE_ATTRIBUTE);
if (StringUtils.hasText(attributeValue)) {
builder.addPropertyValue("phase", attributeValue);
}
attributeValue = element.getAttribute(AUTO_STARTUP_ATTRIBUTE);
if (StringUtils.hasText(attributeValue)) {
builder.addPropertyValue("autoStartup", attributeValue);

2
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/ConnectionFactory.java
View File

@@ -33,5 +33,7 @@ public interface ConnectionFactory {
int getPort();
String getVirtualHost();
void addConnectionListener(ConnectionListener listener);
}

104
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitAdmin.java
View File

@@ -15,6 +15,7 @@ package org.springframework.amqp.rabbit.core;
import java.io.IOException;
import java.util.Collection;
import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@@ -22,10 +23,12 @@ import org.springframework.amqp.core.AmqpAdmin;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.Exchange;
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.beans.factory.InitializingBean;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.SmartLifecycle;
import org.springframework.jmx.export.annotation.ManagedOperation;
import org.springframework.util.Assert;
@@ -39,24 +42,25 @@ import com.rabbitmq.client.Channel;
* @author Mark Fisher
* @author Dave Syer
*/
public class RabbitAdmin implements AmqpAdmin, ApplicationContextAware, SmartLifecycle {
public class RabbitAdmin implements AmqpAdmin, ApplicationContextAware, InitializingBean {
/** Logger available to subclasses */
protected final Log logger = LogFactory.getLog(getClass());
private final RabbitTemplate rabbitTemplate;
private volatile boolean running;
private volatile boolean running = false;
private volatile boolean autoStartup = true;
private volatile int phase = Integer.MIN_VALUE;
private volatile ApplicationContext applicationContext;
private final Object lifecycleMonitor = new Object();
private final ConnectionFactory connectionFactory;
public RabbitAdmin(ConnectionFactory connectionFactory) {
this.connectionFactory = connectionFactory;
Assert.notNull(connectionFactory, "ConnectionFactory must not be null");
this.rabbitTemplate = new RabbitTemplate(connectionFactory);
}
@@ -65,10 +69,6 @@ public class RabbitAdmin implements AmqpAdmin, ApplicationContextAware, SmartLif
this.autoStartup = autoStartup;
}
public void setPhase(int phase) {
this.phase = phase;
}
public void setApplicationContext(ApplicationContext applicationContext) {
this.applicationContext = applicationContext;
}
@@ -193,48 +193,72 @@ public class RabbitAdmin implements AmqpAdmin, ApplicationContextAware, SmartLif
return this.autoStartup;
}
public int getPhase() {
return this.phase;
}
/**
* If {@link #setAutoStartup(boolean) autoStartup} is set to true, registers a callback on the
* {@link ConnectionFactory} to declare all exchanges and queues in the enclosing application context. If the
* callback fails then it may cause other clients of the connection factory to fail, but since only exchanges,
* queues and bindings are declared failure is not expected.
*
* @see InitializingBean#afterPropertiesSet()
* @see #initialize()
*/
public void afterPropertiesSet() {
public boolean isRunning() {
return this.running;
}
public void start() {
synchronized (this.lifecycleMonitor) {
if (this.running) {
if (this.running || !this.autoStartup) {
return;
}
if (this.applicationContext == null) {
if (this.logger.isDebugEnabled()) {
this.logger
.debug("no ApplicationContext has been set, cannot auto-declare Exchanges, Queues, and Bindings");
}
return;
}
final Collection<Exchange> exchanges = this.applicationContext.getBeansOfType(Exchange.class).values();
final Collection<Queue> queues = this.applicationContext.getBeansOfType(Queue.class).values();
final Collection<Binding> bindings = this.applicationContext.getBeansOfType(Binding.class).values();
this.rabbitTemplate.execute(new ChannelCallback<Object>() {
public Object doInRabbit(Channel channel) throws Exception {
declareExchanges(channel, exchanges.toArray(new Exchange[exchanges.size()]));
declareQueues(channel, queues.toArray(new Queue[queues.size()]));
declareBindings(channel, bindings.toArray(new Binding[bindings.size()]));
return null;
// Prevent stack overflow...
final AtomicBoolean initializing = new AtomicBoolean(false);
connectionFactory.addConnectionListener(new ConnectionListener() {
private volatile boolean initialized = false;
public void onCreate(Connection connection) {
if (!initializing.compareAndSet(false, true) || initialized) {
return;
}
initialize();
initializing.compareAndSet(true, false);
initialized = true;
}
});
this.running = true;
}
}
public void stop() {
this.running = false;
}
/**
* Declares all the exchanges, queues and bindings in the enclosing application context, if any. It should be safe
* (but unnecessary) to call this method more than once.
*/
public void initialize() {
if (this.applicationContext == null) {
if (this.logger.isDebugEnabled()) {
this.logger
.debug("no ApplicationContext has been set, cannot auto-declare Exchanges, Queues, and Bindings");
}
return;
}
final Collection<Exchange> exchanges = applicationContext.getBeansOfType(Exchange.class).values();
final Collection<Queue> queues = applicationContext.getBeansOfType(Queue.class).values();
final Collection<Binding> bindings = applicationContext.getBeansOfType(Binding.class).values();
rabbitTemplate.execute(new ChannelCallback<Object>() {
public Object doInRabbit(Channel channel) throws Exception {
declareExchanges(channel, exchanges.toArray(new Exchange[exchanges.size()]));
declareQueues(channel, queues.toArray(new Queue[queues.size()]));
declareBindings(channel, bindings.toArray(new Binding[bindings.size()]));
return null;
}
});
public void stop(Runnable callback) {
this.stop();
callback.run();
}
// private methods for declaring Exchanges, Queues, and Bindings on a Channel

14
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/admin/PojoHandler.java
View File

@@ -1,14 +0,0 @@
package org.springframework.amqp.rabbit.admin;
import java.util.concurrent.atomic.AtomicInteger;
public class PojoHandler {
private final AtomicInteger messageCount = new AtomicInteger();
public void handleMessage(String textMessage) {
int msgCount = this.messageCount.incrementAndGet();
System.out.println("Thread [" + Thread.currentThread().getId() + "] PojoHandler Received Message " + msgCount + ", = " + textMessage);
}
}

23
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/admin/RabbitConsumerConfiguration.java
View File

@@ -1,23 +0,0 @@
package org.springframework.amqp.rabbit.admin;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class RabbitConsumerConfiguration extends TestRabbitConfiguration {
@Bean
public SimpleMessageListenerContainer simpleMessageListenerContainer() {
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
container.setConnectionFactory(connectionFactory());
container.setQueueName(TestConstants.QUEUE_NAME);
container.setConcurrentConsumers(5);
MessageListenerAdapter adapter = new MessageListenerAdapter();
adapter.setDelegate(new PojoHandler());
container.setMessageListener(adapter);
return container;
}
}

21
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/admin/RabbitProducerConfiguration.java
View File

@@ -1,21 +0,0 @@
package org.springframework.amqp.rabbit.admin;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class RabbitProducerConfiguration extends TestRabbitConfiguration {
@Bean
public Queue fooQueue() {
return new Queue(TestConstants.QUEUE_NAME);
}
@Bean
public Binding fooBinding() {
return new Binding(fooQueue(), defaultExchange(), TestConstants.ROUTING_KEY);
}
}

98
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/admin/RabbitTemplateConsumerExample.java
View File

@@ -1,98 +0,0 @@
/*
* Copyright 2002-2010 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.amqp.rabbit.admin;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageListener;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
/**
* @author Mark Fisher
* @author Mark Pollack
*/
public class RabbitTemplateConsumerExample {
private static Log log = LogFactory.getLog(RabbitTemplateConsumerExample.class);
public static void main(String[] args) throws Exception {
boolean sync = true;
if (sync) {
ConfigurableApplicationContext ctx = new AnnotationConfigApplicationContext(
TestRabbitConfiguration.class);
RabbitTemplate template = ctx.getBean(RabbitTemplate.class);
receiveSync(template, TestConstants.NUM_MESSAGES);
}
else {
ConfigurableApplicationContext ctx = new AnnotationConfigApplicationContext(
RabbitConsumerConfiguration.class);
receiveAsync(ctx);
}
}
private static void receiveAsync(ConfigurableApplicationContext ctx) throws InterruptedException {
ctx.getBean(SimpleMessageListenerContainer.class);
log.debug("Main execution thread sleeping 5 seconds...");
Thread.sleep(500000);
log.debug("Application exiting.");
System.exit(0);
}
private static void receiveSync(RabbitTemplate template, int numMessages) {
// receive response
for (int i = 0; i < numMessages; i++) {
Message message = template.receive(TestConstants.QUEUE_NAME);
if (message == null) {
System.out.println("Thread [" + Thread.currentThread().getId()
+ "] Received Null Message!");
}
else {
System.out.println("Thread [" + Thread.currentThread().getId()
+ "] Received Message = "
+ new String(message.getBody()));
Map<String, Object> headers = message.getMessageProperties()
.getHeaders();
Object objFloat = headers.get("float");
Object objcp = headers.get("object");
System.out.println("float header type = " + objFloat.getClass());
System.out.println("object header type = " + objcp.getClass());
}
}
}
public static class SimpleMessageListener implements MessageListener {
private final AtomicInteger messageCount = new AtomicInteger();
public void onMessage(Message message) {
int msgCount = this.messageCount.incrementAndGet();
System.out.println("Thread [" + Thread.currentThread().getId()
+ "] SimpleMessageListener Received Message " + msgCount
+ ", = " + new String(message.getBody()));
}
}
}

47
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/admin/RabbitTemplateProducerExample.java
View File

@@ -1,47 +0,0 @@
/*
* Copyright 2002-2010 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.amqp.rabbit.admin;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
/**
* @author Mark Fisher
*/
public class RabbitTemplateProducerExample {
private static Log log = LogFactory.getLog(RabbitTemplateProducerExample.class);
public static void main(String[] args) throws Exception {
ConfigurableApplicationContext ctx = new AnnotationConfigApplicationContext(RabbitProducerConfiguration.class);
RabbitTemplate template = ctx.getBean(RabbitTemplate.class);
for (int i = 1; i <= 10; i++) {
template.convertAndSend("test-" + i);
Thread.sleep(100);
}
log.debug("done sending");
ctx.stop();
System.exit(0);
}
}

32
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/admin/TestConstants.java
View File

@@ -1,32 +0,0 @@
/*
* Copyright 2002-2010 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.amqp.rabbit.admin;
/**
* Exchange, queue, and routing key constants for the testing code.
*/
public class TestConstants {
public static String EXCHANGE_NAME = "";
public static String QUEUE_NAME = "foo";
public static String ROUTING_KEY = "foo";
public static int NUM_MESSAGES = 500;
}

48
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/admin/TestRabbitConfiguration.java
View File

@@ -1,48 +0,0 @@
/*
* Copyright 2002-2010 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.amqp.rabbit.admin;
import org.springframework.amqp.rabbit.config.AbstractRabbitConfiguration;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.SingleConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.context.annotation.Bean;
/**
* @author Mark Pollack
* @author Mark Fisher
*/
public class TestRabbitConfiguration extends AbstractRabbitConfiguration {
@Bean
public RabbitTemplate rabbitTemplate() {
RabbitTemplate template = new RabbitTemplate(connectionFactory());
template.setExchange(TestConstants.EXCHANGE_NAME);
template.setRoutingKey(TestConstants.ROUTING_KEY);
return template;
}
@Bean
public ConnectionFactory connectionFactory() {
//TODO make it possible to customize in subclasses.
SingleConnectionFactory connectionFactory = new SingleConnectionFactory("localhost");
connectionFactory.setUsername("guest");
connectionFactory.setPassword("guest");
return connectionFactory;
}
}

4
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/AdminParserTests.java
View File

@@ -43,8 +43,6 @@ public final class AdminParserTests {
// <class-name>-<contextIndex>-context.xml.
private int contextIndex;
private int expectedPhase;
private boolean expectedAutoStartup;
private String adminBeanName;
@@ -62,7 +60,6 @@ public final class AdminParserTests {
public void testValid() throws Exception {
contextIndex = 2;
validContext = true;
expectedPhase = 12;
doTest();
}
@@ -81,7 +78,6 @@ public final class AdminParserTests {
} else {
admin = beanFactory.getBean(RabbitAdmin.class);
}
assertEquals(expectedPhase, admin.getPhase());
assertEquals(expectedAutoStartup, admin.isAutoStartup());
assertEquals(beanFactory.getBean(ConnectionFactory.class), admin.getRabbitTemplate().getConnectionFactory());

38
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitAdminIntegrationTests.java Normal file
View File

@@ -0,0 +1,38 @@
package org.springframework.amqp.rabbit.core;
import static org.junit.Assert.assertTrue;
import org.junit.Rule;
import org.junit.Test;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.test.BrokerRunning;
import org.springframework.amqp.rabbit.test.BrokerTestUtils;
import org.springframework.context.support.GenericApplicationContext;
public class RabbitAdminIntegrationTests {
private static Queue queue = new Queue("test.queue");
private CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
@Rule
public BrokerRunning brokerIsRunning = BrokerRunning.isRunning();
public RabbitAdminIntegrationTests() {
connectionFactory.setPort(BrokerTestUtils.getPort());
}
@Test
public void testStartupWithBroker() throws Exception {
GenericApplicationContext applicationContext = new GenericApplicationContext();
applicationContext.getBeanFactory().registerSingleton("foo", queue);
RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);
rabbitAdmin.setApplicationContext(applicationContext);
rabbitAdmin.setAutoStartup(true);
rabbitAdmin.deleteQueue(queue.getName());
rabbitAdmin.afterPropertiesSet();
assertTrue(rabbitAdmin.deleteQueue(queue.getName()));
}
}

39
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitAdminTests.java
View File

@@ -2,10 +2,18 @@ package org.springframework.amqp.rabbit.core;
import static org.junit.Assert.fail;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.ExpectedException;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.SingleConnectionFactory;
import org.springframework.context.support.GenericApplicationContext;
public class RabbitAdminTests {
@Rule
public ExpectedException exception = ExpectedException.none();
@Test
public void testSettingOfNullRabbitTemplate() {
@@ -13,10 +21,35 @@ public class RabbitAdminTests {
try {
new RabbitAdmin(connectionFactory);
fail("should have thrown IllegalStateException when RabbitTemplate is not set.");
}
catch (IllegalArgumentException e) {
} catch (IllegalArgumentException e) {
}
}
@Test
public void testNoFailOnStartupWithMissingBroker() throws Exception {
SingleConnectionFactory connectionFactory = new SingleConnectionFactory("foo");
connectionFactory.setPort(434343);
GenericApplicationContext applicationContext = new GenericApplicationContext();
applicationContext.getBeanFactory().registerSingleton("foo", new Queue("queue"));
RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);
rabbitAdmin.setApplicationContext(applicationContext);
rabbitAdmin.setAutoStartup(true);
rabbitAdmin.afterPropertiesSet();
}
@Test
public void testFailOnFirstUseWithMissingBroker() throws Exception {
SingleConnectionFactory connectionFactory = new SingleConnectionFactory("foo");
connectionFactory.setPort(434343);
GenericApplicationContext applicationContext = new GenericApplicationContext();
applicationContext.getBeanFactory().registerSingleton("foo", new Queue("queue"));
RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);
rabbitAdmin.setApplicationContext(applicationContext);
rabbitAdmin.setAutoStartup(true);
rabbitAdmin.afterPropertiesSet();
exception.expect(IllegalArgumentException.class);
rabbitAdmin.declareQueue();
}
}

20
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/test/BrokerRunning.java
View File

@@ -57,7 +57,7 @@ public class BrokerRunning extends TestWatchman {
private Queue queue;
private int port = BrokerTestUtils.DEFAULT_PORT;
private int port = BrokerTestUtils.getPort();
private String hostName = null;
@@ -71,7 +71,7 @@ public class BrokerRunning extends TestWatchman {
}
/**
* Ensure the broker is running and has an empty queue in the default exchange.
* Ensure the broker is running and has an empty queue (which can be addressed via the default exchange).
*
* @return a new rule that assumes an existing running broker
*/
@@ -106,14 +106,14 @@ public class BrokerRunning extends TestWatchman {
private BrokerRunning(boolean assumeOnline) {
this(assumeOnline, new Queue(DEFAULT_QUEUE_NAME));
}
/**
* @param port the port to set
*/
public void setPort(int port) {
this.port = port;
}
/**
* @param hostName the hostName to set
*/
@@ -148,12 +148,12 @@ public class BrokerRunning extends TestWatchman {
admin.deleteQueue(queueName);
}
if (isDefaultQueue(queueName)) {
// Just for test probe.
admin.deleteQueue(queueName);
} else {
admin.declareQueue(queue);
}
if (isDefaultQueue(queueName)) {
// Just for test probe.
admin.deleteQueue(queueName);
} else {
admin.declareQueue(queue);
}
brokerOffline = false;
if (!assumeOnline) {
Assume.assumeTrue(brokerOffline);