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

AMQP-481: Register @RabbitListener for @Lazy

Browse Source 
JIRA: https://jira.spring.io/browse/AMQP-481

Previously the `@Lazy` components haven't been registered with `MessageListenerContainer`,
because `RabbitListenerEndpointRegistrar` has done that from its `afterPropertiesSet()`.

Add logic to `registerListenerContainer` on demand from the `RabbitListenerEndpointRegistrar#registerEndpoint`

**Cherry-pick to 1.4.x**
This commit is contained in:
Artem Bilan
2015-05-06 16:40:06 +03:00
committed by Gary Russell
parent b87f36263a
commit e20037bb65
3 changed files with 96 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files

29
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/RabbitListenerEndpointRegistrar.java
View File

@@ -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<AmqpListenerEndpointDescriptor> endpointDescriptors =
new ArrayList<AmqpListenerEndpointDescriptor>();
private RabbitListenerEndpointRegistry endpointRegistry;
private MessageHandlerMethodFactory messageHandlerMethodFactory;
@@ -47,9 +51,7 @@ public class RabbitListenerEndpointRegistrar implements BeanFactoryAware, Initia
private BeanFactory beanFactory;
private final List<AmqpListenerEndpointDescriptor> endpointDescriptors =
new ArrayList<AmqpListenerEndpointDescriptor>();
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);
}
}
}
/**

49
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/RabbitListenerEndpointRegistry.java
View File

@@ -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<String, MessageListenerContainer> listenerContainers =
new LinkedHashMap<String, MessageListenerContainer>();
new ConcurrentHashMap<String, MessageListenerContainer>();
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}.
* <p>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}.
* <p>This create the necessary infrastructure to honor that endpoint
* with regards to its configuration.
* <p>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 {

36
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/EnableRabbitTests.java
View File

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