diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/RabbitListenerEndpointRegistrar.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/RabbitListenerEndpointRegistrar.java index dd42208a..4450080e 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/RabbitListenerEndpointRegistrar.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/RabbitListenerEndpointRegistrar.java @@ -1,5 +1,5 @@ /* - * Copyright 2014 the original author or authors. + * Copyright 2014-2015 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. @@ -32,11 +32,15 @@ import org.springframework.util.Assert; * * @author Stephane Nicoll * @author Juergen Hoeller + * @author Artem Bilan * @since 1.4 * @see org.springframework.amqp.rabbit.annotation.RabbitListenerConfigurer */ public class RabbitListenerEndpointRegistrar implements BeanFactoryAware, InitializingBean { + private final List endpointDescriptors = + new ArrayList(); + private RabbitListenerEndpointRegistry endpointRegistry; private MessageHandlerMethodFactory messageHandlerMethodFactory; @@ -47,9 +51,7 @@ public class RabbitListenerEndpointRegistrar implements BeanFactoryAware, Initia private BeanFactory beanFactory; - private final List endpointDescriptors = - new ArrayList(); - + private boolean startImmediately; /** * Set the {@link RabbitListenerEndpointRegistry} instance to use. @@ -127,8 +129,12 @@ public class RabbitListenerEndpointRegistrar implements BeanFactoryAware, Initia } protected void registerAllEndpoints() { - for (AmqpListenerEndpointDescriptor descriptor : this.endpointDescriptors) { - this.endpointRegistry.registerListenerContainer(descriptor.endpoint, resolveContainerFactory(descriptor)); + synchronized (this.endpointDescriptors) { + for (AmqpListenerEndpointDescriptor descriptor : this.endpointDescriptors) { + this.endpointRegistry.registerListenerContainer( + descriptor.endpoint, resolveContainerFactory(descriptor)); + } + this.startImmediately = true; // trigger immediate startup } } @@ -164,7 +170,16 @@ public class RabbitListenerEndpointRegistrar implements BeanFactoryAware, Initia Assert.notNull(endpoint, "Endpoint must be set"); Assert.hasText(endpoint.getId(), "Endpoint id must be set"); // Factory may be null, we defer the resolution right before actually creating the container - this.endpointDescriptors.add(new AmqpListenerEndpointDescriptor(endpoint, factory)); + AmqpListenerEndpointDescriptor descriptor = new AmqpListenerEndpointDescriptor(endpoint, factory); + synchronized (this.endpointDescriptors) { + if (this.startImmediately) { // Register and start immediately + this.endpointRegistry.registerListenerContainer(descriptor.endpoint, + resolveContainerFactory(descriptor), true); + } + else { + this.endpointDescriptors.add(descriptor); + } + } } /** diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/RabbitListenerEndpointRegistry.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/RabbitListenerEndpointRegistry.java index fd0a7977..339941e8 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/RabbitListenerEndpointRegistry.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/RabbitListenerEndpointRegistry.java @@ -18,8 +18,8 @@ package org.springframework.amqp.rabbit.listener; import java.util.Collection; import java.util.Collections; -import java.util.LinkedHashMap; import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.atomic.AtomicInteger; import org.apache.commons.logging.Log; @@ -46,6 +46,7 @@ import org.springframework.util.Assert; * * @author Stephane Nicoll * @author Juergen Hoeller + * @author Artem Bilan * @since 1.4 * @see RabbitListenerEndpoint * @see MessageListenerContainer @@ -56,7 +57,7 @@ public class RabbitListenerEndpointRegistry implements DisposableBean, SmartLife protected final Log logger = LogFactory.getLog(getClass()); private final Map listenerContainers = - new LinkedHashMap(); + new ConcurrentHashMap(); private int phase = Integer.MAX_VALUE; @@ -80,27 +81,46 @@ public class RabbitListenerEndpointRegistry implements DisposableBean, SmartLife return Collections.unmodifiableCollection(this.listenerContainers.values()); } + /** + * Create a message listener container for the given {@link RabbitListenerEndpoint}. + *

This create the necessary infrastructure to honor that endpoint + * with regards to its configuration. + * @param endpoint the endpoint to add + * @param factory the listener factory to use + * @see #registerListenerContainer(RabbitListenerEndpoint, RabbitListenerContainerFactory, boolean) + */ + public void registerListenerContainer(RabbitListenerEndpoint endpoint, RabbitListenerContainerFactory factory) { + registerListenerContainer(endpoint, factory, false); + } /** * Create a message listener container for the given {@link RabbitListenerEndpoint}. *

This create the necessary infrastructure to honor that endpoint * with regards to its configuration. + *

The {@code startImmediately} flag determines if the container should be + * started immediately. * @param endpoint the endpoint to add. * @param factory the {@link RabbitListenerContainerFactory} to use. + * @param startImmediately start the container immediately if necessary * @see #getListenerContainers() * @see #getListenerContainer(String) */ - public void registerListenerContainer(RabbitListenerEndpoint endpoint, RabbitListenerContainerFactory factory) { + public void registerListenerContainer(RabbitListenerEndpoint endpoint, RabbitListenerContainerFactory factory, + boolean startImmediately) { Assert.notNull(endpoint, "Endpoint must not be null"); Assert.notNull(factory, "Factory must not be null"); String id = endpoint.getId(); Assert.hasText(id, "Endpoint id must not be empty"); - Assert.state(!this.listenerContainers.containsKey(id), - "Another endpoint is already registered with id '" + id + "'"); - - MessageListenerContainer container = createListenerContainer(endpoint, factory); - this.listenerContainers.put(id, container); + synchronized (this.listenerContainers) { + Assert.state(!this.listenerContainers.containsKey(id), + "Another endpoint is already registered with id '" + id + "'"); + MessageListenerContainer container = createListenerContainer(endpoint, factory); + this.listenerContainers.put(id, container); + if (startImmediately) { + startIfNecessary(container); + } + } } /** @@ -167,7 +187,7 @@ public class RabbitListenerEndpointRegistry implements DisposableBean, SmartLife public void start() { for (MessageListenerContainer listenerContainer : getListenerContainers()) { if (listenerContainer.isAutoStartup()) { - listenerContainer.start(); + startIfNecessary(listenerContainer); } } } @@ -198,6 +218,17 @@ public class RabbitListenerEndpointRegistry implements DisposableBean, SmartLife return false; } + /** + * Start the specified {@link MessageListenerContainer} if it should be started + * on startup. + * @see MessageListenerContainer#isAutoStartup() + */ + private static void startIfNecessary(MessageListenerContainer listenerContainer) { + if (listenerContainer.isAutoStartup()) { + listenerContainer.start(); + } + } + private static class AggregatingCallback implements Runnable { diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/EnableRabbitTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/EnableRabbitTests.java index 174317ce..498c0d63 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/EnableRabbitTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/EnableRabbitTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2014 the original author or authors. + * Copyright 2002-2015 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. @@ -17,6 +17,8 @@ package org.springframework.amqp.rabbit.annotation; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; import static org.mockito.Mockito.mock; import org.hamcrest.core.Is; @@ -25,6 +27,7 @@ import org.junit.Test; import org.junit.rules.ExpectedException; import org.springframework.amqp.core.MessageListener; +import org.springframework.amqp.rabbit.config.MessageListenerTestContainer; import org.springframework.amqp.rabbit.config.RabbitListenerContainerTestFactory; import org.springframework.amqp.rabbit.config.SimpleRabbitListenerEndpoint; import org.springframework.amqp.rabbit.core.RabbitAdmin; @@ -37,6 +40,7 @@ import org.springframework.context.ConfigurableApplicationContext; import org.springframework.context.annotation.AnnotationConfigApplicationContext; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; +import org.springframework.context.annotation.Lazy; import org.springframework.context.annotation.PropertySource; import org.springframework.context.support.PropertySourcesPlaceholderConfigurer; import org.springframework.messaging.handler.annotation.support.DefaultMessageHandlerMethodFactory; @@ -45,8 +49,8 @@ import org.springframework.messaging.handler.annotation.support.MethodArgumentNo import org.springframework.stereotype.Component; /** - * * @author Stephane Nicoll + * @author Artem Bilan */ public class EnableRabbitTests extends AbstractRabbitAnnotationDrivenTests { @@ -135,6 +139,23 @@ public class EnableRabbitTests extends AbstractRabbitAnnotationDrivenTests { EnableRabbitSampleConfig.class, InvalidPriorityBean.class).close(); } + @Test + public void lazyComponent() { + ConfigurableApplicationContext context = new AnnotationConfigApplicationContext( + EnableRabbitDefaultContainerFactoryConfig.class, LazyBean.class); + RabbitListenerContainerTestFactory defaultFactory = + context.getBean("rabbitListenerContainerFactory", RabbitListenerContainerTestFactory.class); + assertEquals(0, defaultFactory.getListenerContainers().size()); + + context.getBean(LazyBean.class); // trigger lazy resolution + assertEquals(1, defaultFactory.getListenerContainers().size()); + MessageListenerTestContainer container = defaultFactory.getListenerContainers().get(0); + assertTrue("Should have been started " + container, container.isStarted()); + context.close(); // Close and stop the listeners + assertTrue("Should have been stopped " + container, container.isStopped()); + } + + @EnableRabbit @Configuration static class EnableRabbitSampleConfig { @@ -276,6 +297,17 @@ public class EnableRabbitTests extends AbstractRabbitAnnotationDrivenTests { @RabbitListener(queues = "myQueue", priority = "NotANumber") public void customHandle(String msg) { } + + } + + @Component + @Lazy + static class LazyBean { + + @RabbitListener(queues = "myQueue") + public void handle(String msg) { + } + } }