diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/AbstractAmqpChannel.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/AbstractAmqpChannel.java index 213c991422..2922040650 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/AbstractAmqpChannel.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/AbstractAmqpChannel.java @@ -16,10 +16,15 @@ package org.springframework.integration.amqp.channel; +import org.springframework.amqp.core.AmqpAdmin; import org.springframework.amqp.core.AmqpTemplate; import org.springframework.amqp.core.MessageDeliveryMode; +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.support.AmqpHeaders; +import org.springframework.beans.factory.DisposableBean; import org.springframework.integration.amqp.support.AmqpHeaderMapper; import org.springframework.integration.amqp.support.DefaultAmqpHeaderMapper; import org.springframework.integration.amqp.support.MappingUtils; @@ -29,9 +34,12 @@ import org.springframework.util.Assert; /** * @author Mark Fisher + * @author Artem Bilan + * * @since 2.1 */ -public abstract class AbstractAmqpChannel extends AbstractMessageChannel { +public abstract class AbstractAmqpChannel extends AbstractMessageChannel + implements DisposableBean, ConnectionListener { private final AmqpTemplate amqpTemplate; @@ -41,14 +49,20 @@ public abstract class AbstractAmqpChannel extends AbstractMessageChannel { private final AmqpHeaderMapper inboundHeaderMapper; - private volatile boolean extractPayload; + private AmqpAdmin admin; - private volatile boolean loggingEnabled = true; + private ConnectionFactory connectionFactory; + + private boolean extractPayload; + + private boolean loggingEnabled = true; private MessageDeliveryMode defaultDeliveryMode; private boolean headersMappedLast; + private volatile boolean initialized; + /** * Construct an instance with the supplied template and default header mappers * used if the template is a {@link RabbitTemplate} and the message is mapped. @@ -149,7 +163,6 @@ public abstract class AbstractAmqpChannel extends AbstractMessageChannel { /** * Subclasses may override this method to return an Exchange name. * By default, Messages will be sent to the no-name Direct Exchange. - * * @return The exchange name. */ protected String getExchangeName() { @@ -159,7 +172,6 @@ public abstract class AbstractAmqpChannel extends AbstractMessageChannel { /** * Subclasses may override this method to return a routing key. * By default, there will be no routing key (empty string). - * * @return The routing key. */ protected String getRoutingKey() { @@ -178,6 +190,41 @@ public abstract class AbstractAmqpChannel extends AbstractMessageChannel { return this.rabbitTemplate; } + protected final void setAdmin(AmqpAdmin admin) { + this.admin = admin; + } + + protected final void setConnectionFactory(ConnectionFactory connectionFactory) { + this.connectionFactory = connectionFactory; + } + + protected AmqpAdmin getAdmin() { + return this.admin; + } + + protected ConnectionFactory getConnectionFactory() { + return this.connectionFactory; + } + + @Override + protected void onInit() throws Exception { + super.onInit(); + if (!this.initialized && this.rabbitTemplate != null) { + if (this.connectionFactory != null) { + this.connectionFactory.addConnectionListener(this); + } + } + this.initialized = true; + } + + @Override + public void destroy() throws Exception { + if (this.connectionFactory != null) { + this.connectionFactory.removeConnectionListener(this); + this.initialized = false; + } + } + @Override protected boolean doSend(Message message, long timeout) { if (this.extractPayload) { @@ -191,4 +238,15 @@ public abstract class AbstractAmqpChannel extends AbstractMessageChannel { return true; } + @Override + public void onCreate(Connection connection) { + doDeclares(); + } + + @Override + public void onClose(Connection connection) { + } + + protected abstract void doDeclares(); + } 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 782ca27e38..44191c99ed 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 @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-2017 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. @@ -21,19 +21,18 @@ import java.util.Map; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; -import org.springframework.amqp.core.AmqpAdmin; +import org.springframework.amqp.AmqpConnectException; import org.springframework.amqp.core.AmqpTemplate; import org.springframework.amqp.core.MessageListener; -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.AbstractMessageListenerContainer; import org.springframework.amqp.support.converter.MessageConverter; import org.springframework.amqp.support.converter.SimpleMessageConverter; -import org.springframework.beans.factory.DisposableBean; import org.springframework.context.SmartLifecycle; import org.springframework.integration.MessageDispatchingException; import org.springframework.integration.amqp.support.AmqpHeaderMapper; +import org.springframework.integration.amqp.support.DefaultAmqpHeaderMapper; import org.springframework.integration.context.IntegrationProperties; import org.springframework.integration.dispatcher.AbstractDispatcher; import org.springframework.integration.dispatcher.MessageDispatcher; @@ -49,10 +48,11 @@ import org.springframework.util.Assert; * @author Mark Fisher * @author Gary Russell * @author Artem Bilan + * * @since 2.1 */ abstract class AbstractSubscribableAmqpChannel extends AbstractAmqpChannel - implements SubscribableChannel, SmartLifecycle, DisposableBean { + implements SubscribableChannel, SmartLifecycle { private final String channelName; @@ -64,9 +64,7 @@ abstract class AbstractSubscribableAmqpChannel extends AbstractAmqpChannel private volatile Integer maxSubscribers; - private final AmqpAdmin admin; - - private final ConnectionFactory connectionFactory; + private volatile boolean declared; /** * Construct an instance with the supplied name, container and template; default header @@ -109,14 +107,8 @@ abstract class AbstractSubscribableAmqpChannel extends AbstractAmqpChannel protected AbstractSubscribableAmqpChannel(String channelName, AbstractMessageListenerContainer container, AmqpTemplate amqpTemplate, boolean isPubSub) { - super(amqpTemplate); - Assert.notNull(container, "container must not be null"); - Assert.hasText(channelName, "channel name must not be empty"); - this.channelName = channelName; - this.container = container; - this.isPubSub = isPubSub; - this.connectionFactory = container.getConnectionFactory(); - this.admin = new RabbitAdmin(this.connectionFactory); + this(channelName, container, amqpTemplate, isPubSub, + DefaultAmqpHeaderMapper.outboundMapper(), DefaultAmqpHeaderMapper.inboundMapper()); } /** @@ -141,8 +133,8 @@ abstract class AbstractSubscribableAmqpChannel extends AbstractAmqpChannel this.channelName = channelName; this.container = container; this.isPubSub = isPubSub; - this.connectionFactory = container.getConnectionFactory(); - this.admin = new RabbitAdmin(this.connectionFactory); + setConnectionFactory(container.getConnectionFactory()); + setAdmin(new RabbitAdmin(getConnectionFactory())); } /** @@ -157,14 +149,6 @@ abstract class AbstractSubscribableAmqpChannel extends AbstractAmqpChannel } } - protected AmqpAdmin getAdmin() { - return this.admin; - } - - protected ConnectionFactory getConnectionFactory() { - return this.connectionFactory; - } - @Override public boolean subscribe(MessageHandler handler) { return this.dispatcher.addHandler(handler); @@ -181,12 +165,12 @@ abstract class AbstractSubscribableAmqpChannel extends AbstractAmqpChannel this.dispatcher = this.createDispatcher(); if (this.maxSubscribers == null) { this.maxSubscribers = this.getIntegrationProperty(this.isPubSub ? - IntegrationProperties.CHANNELS_MAX_BROADCAST_SUBSCRIBERS : - IntegrationProperties.CHANNELS_MAX_UNICAST_SUBSCRIBERS, + IntegrationProperties.CHANNELS_MAX_BROADCAST_SUBSCRIBERS : + IntegrationProperties.CHANNELS_MAX_UNICAST_SUBSCRIBERS, Integer.class); } setMaxSubscribers(this.maxSubscribers); - String queue = this.obtainQueueName(this.admin, this.channelName); + String queue = obtainQueueName(this.channelName); this.container.setQueueNames(queue); MessageConverter converter = (this.getAmqpTemplate() instanceof RabbitTemplate) ? ((RabbitTemplate) this.getAmqpTemplate()).getMessageConverter() @@ -221,6 +205,16 @@ abstract class AbstractSubscribableAmqpChannel extends AbstractAmqpChannel @Override public void start() { + if (!this.declared) { + try { + doDeclares(); + this.declared = true; + } + catch (AmqpConnectException e) { + logger.info("Broker not available; cannot check queue declarations. " + + "Postponed to the next connection create..."); + } + } if (this.container != null) { this.container.start(); } @@ -230,6 +224,7 @@ abstract class AbstractSubscribableAmqpChannel extends AbstractAmqpChannel public void stop() { if (this.container != null) { this.container.stop(); + this.declared = false; } } @@ -237,19 +232,22 @@ abstract class AbstractSubscribableAmqpChannel extends AbstractAmqpChannel public void stop(Runnable callback) { if (this.container != null) { this.container.stop(callback); + this.declared = false; } } @Override public void destroy() throws Exception { + super.destroy(); if (this.container != null) { this.container.destroy(); + this.declared = false; } } protected abstract AbstractDispatcher createDispatcher(); - protected abstract String obtainQueueName(AmqpAdmin admin, String channelName); + protected abstract String obtainQueueName(String channelName); private static final class DispatchingMessageListener implements MessageListener { diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PointToPointSubscribableAmqpChannel.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PointToPointSubscribableAmqpChannel.java index 3965b24900..858fd3a08d 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PointToPointSubscribableAmqpChannel.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PointToPointSubscribableAmqpChannel.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-2017 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. @@ -28,12 +28,12 @@ import org.springframework.integration.dispatcher.UnicastingDispatcher; /** * @author Mark Fisher * @author Artem Bilan + * * @since 2.1 */ public class PointToPointSubscribableAmqpChannel extends AbstractSubscribableAmqpChannel { - private volatile String queueName; - + private volatile Queue queue; /** * Construct an instance with the supplied name, container and template; default header @@ -72,18 +72,16 @@ public class PointToPointSubscribableAmqpChannel extends AbstractSubscribableAmq * @param queueName The queue name. */ public void setQueueName(String queueName) { - this.queueName = queueName; + this.queue = new Queue(queueName); } @Override - protected String obtainQueueName(AmqpAdmin admin, String channelName) { - if (this.queueName == null) { - this.queueName = channelName; + protected String obtainQueueName(String channelName) { + if (this.queue == null) { + this.queue = new Queue(channelName); } - if (admin.getQueueProperties(this.queueName) == null) { - admin.declareQueue(new Queue(this.queueName)); - } - return this.queueName; + + return this.queue.getName(); } @Override @@ -95,7 +93,15 @@ public class PointToPointSubscribableAmqpChannel extends AbstractSubscribableAmq @Override protected String getRoutingKey() { - return this.queueName; + return this.queue != null ? this.queue.getName() : super.getRoutingKey(); + } + + @Override + protected void doDeclares() { + AmqpAdmin admin = getAdmin(); + if (admin != null && this.queue != null && admin.getQueueProperties(this.queue.getName()) == null) { + admin.declareQueue(this.queue); + } } } diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PollableAmqpChannel.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PollableAmqpChannel.java index 726cf4d7de..28e806a47a 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PollableAmqpChannel.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PollableAmqpChannel.java @@ -24,6 +24,7 @@ import java.util.Map; import org.springframework.amqp.core.AmqpAdmin; import org.springframework.amqp.core.AmqpTemplate; 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.integration.amqp.support.AmqpHeaderMapper; @@ -51,12 +52,12 @@ public class PollableAmqpChannel extends AbstractAmqpChannel private final String channelName; - private volatile String queueName; - - private volatile AmqpAdmin amqpAdmin; + private volatile Queue queue; private volatile int executorInterceptorsSize; + private volatile boolean declared; + /** * Construct an instance with the supplied name, template and default header mappers * used if the template is a {@link RabbitTemplate} and the message is mapped. @@ -91,22 +92,20 @@ public class PollableAmqpChannel extends AbstractAmqpChannel * Provide an explicitly configured queue name. If this is not provided, then a Queue will be created * implicitly with the channelName as its name. The implicit creation will require that either an AmqpAdmin * instance has been provided or that the configured AmqpTemplate is an instance of RabbitTemplate. - * * @param queueName The queue name. */ public void setQueueName(String queueName) { - this.queueName = queueName; + this.queue = new Queue(queueName); } /** * Provide an instance of AmqpAdmin for implicitly declaring Queues if the queueName is not provided. * When providing a RabbitTemplate implementation, this is not strictly necessary since a RabbitAdmin * instance can be created from the template's ConnectionFactory reference. - * * @param amqpAdmin The amqp admin. */ public void setAmqpAdmin(AmqpAdmin amqpAdmin) { - this.amqpAdmin = amqpAdmin; + setAdmin(amqpAdmin); } @Override @@ -131,22 +130,33 @@ public class PollableAmqpChannel extends AbstractAmqpChannel @Override protected String getRoutingKey() { - return this.queueName; + return this.queue != null ? this.queue.getName() : super.getRoutingKey(); } @Override protected void onInit() throws Exception { - super.onInit(); - AmqpTemplate amqpTemplate = this.getAmqpTemplate(); - if (this.queueName == null) { - if (this.amqpAdmin == null && amqpTemplate instanceof RabbitTemplate) { - this.amqpAdmin = new RabbitAdmin(((RabbitTemplate) amqpTemplate).getConnectionFactory()); + AmqpTemplate amqpTemplate = getAmqpTemplate(); + if (this.queue == null) { + if (getAdmin() == null && amqpTemplate instanceof RabbitTemplate) { + ConnectionFactory connectionFactory = ((RabbitTemplate) amqpTemplate).getConnectionFactory(); + setAdmin(new RabbitAdmin(connectionFactory)); + setConnectionFactory(connectionFactory); } - Assert.notNull(this.amqpAdmin, + + Assert.notNull(getAdmin(), "If no queueName is configured explicitly, an AmqpAdmin instance must be provided, " + "or the AmqpTemplate must be a RabbitTemplate since the Queue needs to be declared."); - this.queueName = this.channelName; - this.amqpAdmin.declareQueue(new Queue(this.queueName)); + + this.queue = new Queue(this.channelName); + } + super.onInit(); + } + + @Override + protected void doDeclares() { + AmqpAdmin admin = getAdmin(); + if (admin != null && admin.getQueueProperties(this.queue.getName()) == null) { + admin.declareQueue(this.queue); } } @@ -218,22 +228,27 @@ public class PollableAmqpChannel extends AbstractAmqpChannel } protected Object performReceive(Long timeout) { + if (!this.declared) { + doDeclares(); + this.declared = true; + } + if (!isExtractPayload()) { if (timeout == null) { - return getAmqpTemplate().receiveAndConvert(this.queueName); + return getAmqpTemplate().receiveAndConvert(this.queue.getName()); } else { - return getAmqpTemplate().receiveAndConvert(this.queueName, timeout); + return getAmqpTemplate().receiveAndConvert(this.queue.getName(), timeout); } } else { RabbitTemplate rabbitTemplate = getRabbitTemplate(); org.springframework.amqp.core.Message message; if (timeout == null) { - message = rabbitTemplate.receive(this.queueName); + message = rabbitTemplate.receive(this.queue.getName()); } else { - message = rabbitTemplate.receive(this.queueName, timeout); + message = rabbitTemplate.receive(this.queue.getName(), timeout); } if (message != null) { 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 361ddfbf65..768904f0aa 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 @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-2017 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. @@ -23,10 +23,6 @@ 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.AbstractMessageListenerContainer; import org.springframework.integration.amqp.support.AmqpHeaderMapper; import org.springframework.integration.dispatcher.AbstractDispatcher; @@ -35,9 +31,11 @@ import org.springframework.integration.dispatcher.BroadcastingDispatcher; /** * @author Mark Fisher * @author Gary Russell + * @author Artem Bilan + * * @since 2.1 */ -public class PublishSubscribeAmqpChannel extends AbstractSubscribableAmqpChannel implements ConnectionListener { +public class PublishSubscribeAmqpChannel extends AbstractSubscribableAmqpChannel { private volatile FanoutExchange exchange; @@ -45,8 +43,6 @@ public class PublishSubscribeAmqpChannel extends AbstractSubscribableAmqpChannel private volatile Binding binding; - private volatile boolean initialized; - /** * Construct an instance with the supplied name, container and template; default header * mappers will be used if the message is mapped. @@ -89,37 +85,20 @@ public class PublishSubscribeAmqpChannel extends AbstractSubscribableAmqpChannel } @Override - protected String obtainQueueName(AmqpAdmin admin, String channelName) { + protected String getExchangeName() { + return (this.exchange != null) ? this.exchange.getName() : ""; + } + + @Override + protected String obtainQueueName(String channelName) { if (this.exchange == null) { String exchangeName = "si.fanout." + channelName; this.exchange = new FanoutExchange(exchangeName); } - admin.declareExchange(this.exchange); - 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.getName(); - } - 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); - } - } - } + this.binding = BindingBuilder.bind(this.queue).to(this.exchange); + + return this.queue.getName(); } @Override @@ -130,32 +109,19 @@ public class PublishSubscribeAmqpChannel extends AbstractSubscribableAmqpChannel } @Override - protected String getExchangeName() { - 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; + protected void doDeclares() { + AmqpAdmin admin = getAdmin(); + if (admin != null) { + if (admin.getQueueProperties(this.queue.getName()) == null) { + admin.declareQueue(this.queue); + } + if (this.exchange != null) { + admin.declareExchange(this.exchange); + } + if (this.binding != null) { + admin.declareBinding(this.binding); + } } } - @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/channel/ChannelTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/channel/ChannelTests.java index d6e79a4a52..879c7a440d 100644 --- 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 @@ -100,7 +100,7 @@ public class ChannelTests { private PollableChannel out; @Autowired - private CachingConnectionFactory factory; + private CachingConnectionFactory connectionFactory; @Autowired private AmqpHeaderMapper mapperIn; @@ -115,7 +115,6 @@ public class ChannelTests { } @Test - @DirtiesContext public void pubSubLostConnectionTest() throws Exception { final CyclicBarrier latch = new CyclicBarrier(2); channel.subscribe(message -> { @@ -130,13 +129,15 @@ public class ChannelTests { latch.reset(); BlockingQueueConsumer consumer = (BlockingQueueConsumer) TestUtils.getPropertyValue(this.channel, "container.consumers", Set.class).iterator().next(); - factory.destroy(); + connectionFactory.destroy(); waitForNewConsumer(this.channel, consumer); this.channel.send(new GenericMessage("bar")); latch.await(10, TimeUnit.SECONDS); this.channel.destroy(); this.pubSubWithEP.destroy(); - assertEquals(0, TestUtils.getPropertyValue(factory, "connectionListener.delegates", Collection.class).size()); + this.withEP.destroy(); + this.pollableWithEP.destroy(); + assertEquals(0, TestUtils.getPropertyValue(connectionFactory, "connectionListener.delegates", Collection.class).size()); } @SuppressWarnings("unchecked") @@ -166,25 +167,31 @@ public class ChannelTests { */ @Test public void channelDeclarationTests() { - RabbitAdmin admin = new RabbitAdmin(this.factory); + RabbitAdmin admin = new RabbitAdmin(this.connectionFactory); admin.deleteQueue("implicit"); - SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(this.factory); + SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(this.connectionFactory); + container.setAutoStartup(false); AmqpTemplate amqpTemplate = mock(AmqpTemplate.class); PointToPointSubscribableAmqpChannel channel = new PointToPointSubscribableAmqpChannel("implicit", container, amqpTemplate); channel.setBeanFactory(mock(BeanFactory.class)); channel.afterPropertiesSet(); + channel.onCreate(null); + assertNotNull(admin.getQueueProperties("implicit")); - admin.deleteQueue("implicit"); admin.deleteQueue("explicit"); channel.setQueueName("explicit"); channel.afterPropertiesSet(); + channel.onCreate(null); + assertNotNull(admin.getQueueProperties("explicit")); admin.deleteQueue("explicit"); admin.declareQueue(new Queue("explicit", false)); // verify no declaration if exists with non-standard props channel.afterPropertiesSet(); + channel.onCreate(null); + assertNotNull(admin.getQueueProperties("explicit")); admin.deleteQueue("explicit"); } @@ -193,7 +200,7 @@ public class ChannelTests { public void testAmqpChannelFactoryBean() throws Exception { AmqpChannelFactoryBean channelFactoryBean = new AmqpChannelFactoryBean(); channelFactoryBean.setBeanFactory(mock(BeanFactory.class)); - channelFactoryBean.setConnectionFactory(this.factory); + channelFactoryBean.setConnectionFactory(this.connectionFactory); channelFactoryBean.setBeanName("testChannel"); channelFactoryBean.afterPropertiesSet(); AbstractAmqpChannel channel = channelFactoryBean.getObject(); @@ -201,14 +208,14 @@ public class ChannelTests { channelFactoryBean = new AmqpChannelFactoryBean(); channelFactoryBean.setBeanFactory(mock(BeanFactory.class)); - channelFactoryBean.setConnectionFactory(this.factory); + channelFactoryBean.setConnectionFactory(this.connectionFactory); channelFactoryBean.setBeanName("testChannel"); channelFactoryBean.setPubSub(true); channelFactoryBean.afterPropertiesSet(); channel = channelFactoryBean.getObject(); assertThat(channel, instanceOf(PublishSubscribeAmqpChannel.class)); - RabbitAdmin rabbitAdmin = new RabbitAdmin(this.factory); + RabbitAdmin rabbitAdmin = new RabbitAdmin(this.connectionFactory); rabbitAdmin.deleteQueue("testChannel"); rabbitAdmin.deleteExchange("si.fanout.testChannel"); } @@ -241,11 +248,11 @@ public class ChannelTests { @Test public void messageConversionTests() throws Exception { - RabbitTemplate amqpTemplate = new RabbitTemplate(this.factory); + RabbitTemplate amqpTemplate = new RabbitTemplate(this.connectionFactory); MessageConverter messageConverter = mock(MessageConverter.class); amqpTemplate.setMessageConverter(messageConverter); PointToPointSubscribableAmqpChannel channel = new PointToPointSubscribableAmqpChannel("testConvertFail", - new SimpleMessageListenerContainer(this.factory), amqpTemplate); + new SimpleMessageListenerContainer(this.connectionFactory), amqpTemplate); channel.afterPropertiesSet(); MessageListener listener = TestUtils.getPropertyValue(channel, "container.messageListener", MessageListener.class); diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/channel/DispatcherHasNoSubscribersTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/channel/DispatcherHasNoSubscribersTests.java index 9867639595..ec769c067c 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/channel/DispatcherHasNoSubscribersTests.java +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/channel/DispatcherHasNoSubscribersTests.java @@ -35,11 +35,9 @@ import java.util.List; import org.apache.commons.logging.Log; import org.junit.Test; -import org.springframework.amqp.core.AmqpAdmin; import org.springframework.amqp.core.AmqpTemplate; import org.springframework.amqp.core.Message; import org.springframework.amqp.core.MessageListener; -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.listener.SimpleMessageListenerContainer; @@ -55,6 +53,7 @@ import com.rabbitmq.client.Channel; /** * @author Gary Russell * @author Artem Bilan + * * @since 2.1 * */ @@ -103,15 +102,8 @@ public class DispatcherHasNoSubscribersTests { SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(); container.setConnectionFactory(connectionFactory); AmqpTemplate amqpTemplate = mock(AmqpTemplate.class); - final Queue queue = new Queue("noSubscribersQueue"); PublishSubscribeAmqpChannel amqpChannel = new PublishSubscribeAmqpChannel("noSubscribersChannel", - container, amqpTemplate) { - @Override - protected String obtainQueueName(AmqpAdmin admin, - String channelName) { - return queue.getName(); - } - }; + container, amqpTemplate); amqpChannel.setBeanName("noSubscribersChannel"); amqpChannel.setBeanFactory(mock(BeanFactory.class)); amqpChannel.afterPropertiesSet(); diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpChannelParserTests-context.xml b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpChannelParserTests-context.xml index 58b9654ce4..f6d046422f 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpChannelParserTests-context.xml +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpChannelParserTests-context.xml @@ -1,12 +1,8 @@ @@ -18,26 +14,36 @@ + template-channel-transacted="true" consumers-per-queue="2" + recovery-interval="0" + shutdown-timeout="0"/> - + + inbound-header-mapper="inMapper" outbound-header-mapper="outMapper" + recovery-interval="0" + shutdown-timeout="0"/> + headers-last="true" + inbound-header-mapper="inMapper" outbound-header-mapper="outMapper" + recovery-interval="0" + shutdown-timeout="0"/> + inbound-header-mapper="inMapper" outbound-header-mapper="outMapper" + recovery-interval="0" + shutdown-timeout="0"/> - + - +