From 159e21465d3fac45a2acd3c6ff67e58050979cbb Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Tue, 1 Nov 2011 17:03:07 -0400 Subject: [PATCH] Refactored AMQP channels added namespace support moved afterPropertiesSet call into FactoryBean --- .../amqp/channel/AbstractAmqpChannel.java | 18 +++- ...a => AbstractSubscribableAmqpChannel.java} | 75 +++++-------- .../PointToPointSubscribableAmqpChannel.java | 71 ++++++++++++ .../amqp/channel/PollableAmqpChannel.java | 52 +++++++-- .../channel/PublishSubscribeAmqpChannel.java | 76 +++++++++++++ .../amqp/config/AmqpChannelFactoryBean.java | 102 ++++++++++++++++-- .../amqp/config/AmqpChannelParser.java | 3 + .../config/spring-integration-amqp-2.1.xsd | 50 ++++++++- 8 files changed, 374 insertions(+), 73 deletions(-) rename spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/{SubscribableAmqpChannel.java => AbstractSubscribableAmqpChannel.java} (63%) create mode 100644 spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PointToPointSubscribableAmqpChannel.java create mode 100644 spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PublishSubscribeAmqpChannel.java 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 dab69d5eb9..dc5f2a8c33 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 @@ -36,13 +36,29 @@ 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. + */ + protected String getExchangeName() { + return ""; + } + + /** + * Subclasses may override this method to return a routing key. + * By default, there will be no routing key (empty string). + */ + protected String getRoutingKey() { + return ""; + } + AmqpTemplate getAmqpTemplate() { return this.amqpTemplate; } @Override protected boolean doSend(Message message, long timeout) { - this.amqpTemplate.convertAndSend(message); + this.amqpTemplate.convertAndSend(this.getExchangeName(), this.getRoutingKey(), message); return true; } diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/SubscribableAmqpChannel.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/AbstractSubscribableAmqpChannel.java similarity index 63% rename from spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/SubscribableAmqpChannel.java rename to spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/AbstractSubscribableAmqpChannel.java index 16bc2ad2dd..cd3f14d798 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/SubscribableAmqpChannel.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/AbstractSubscribableAmqpChannel.java @@ -18,25 +18,23 @@ package org.springframework.integration.amqp.channel; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; + +import org.springframework.amqp.core.AmqpAdmin; import org.springframework.amqp.core.AmqpTemplate; -import org.springframework.amqp.core.Binding; -import org.springframework.amqp.core.BindingBuilder; -import org.springframework.amqp.core.FanoutExchange; import org.springframework.amqp.core.MessageListener; import org.springframework.amqp.core.Queue; import org.springframework.amqp.rabbit.core.RabbitAdmin; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer; +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.Message; import org.springframework.integration.MessagingException; import org.springframework.integration.core.MessageHandler; import org.springframework.integration.core.SubscribableChannel; -import org.springframework.integration.dispatcher.BroadcastingDispatcher; import org.springframework.integration.dispatcher.MessageDispatcher; -import org.springframework.integration.dispatcher.RoundRobinLoadBalancingStrategy; -import org.springframework.integration.dispatcher.UnicastingDispatcher; import org.springframework.integration.support.MessageBuilder; import org.springframework.util.Assert; @@ -44,24 +42,21 @@ import org.springframework.util.Assert; * @author Mark Fisher * @since 2.1 */ -public class SubscribableAmqpChannel extends AbstractAmqpChannel implements SubscribableChannel, SmartLifecycle, DisposableBean { +abstract class AbstractSubscribableAmqpChannel extends AbstractAmqpChannel implements SubscribableChannel, SmartLifecycle, DisposableBean { private final String channelName; private final SimpleMessageListenerContainer container; - private final boolean isPubSub; - private volatile MessageDispatcher dispatcher; - public SubscribableAmqpChannel(String channelName, SimpleMessageListenerContainer container, AmqpTemplate amqpTemplate, boolean isPubSub) { + public AbstractSubscribableAmqpChannel(String channelName, SimpleMessageListenerContainer container, AmqpTemplate amqpTemplate) { 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; } @@ -76,67 +71,45 @@ public class SubscribableAmqpChannel extends AbstractAmqpChannel implements Subs @Override public void onInit() throws Exception { super.onInit(); - this.configureDispatcher(); - AmqpTemplate amqpTemplate = this.getAmqpTemplate(); - if (!(amqpTemplate instanceof RabbitTemplate)) { - throw new IllegalArgumentException("AmqpTemplate must be a RabbitTemplate"); - } - RabbitTemplate rabbitTemplate = (RabbitTemplate) amqpTemplate; - RabbitAdmin admin = new RabbitAdmin(rabbitTemplate.getConnectionFactory()); - if (this.isPubSub) { - FanoutExchange exchange = new FanoutExchange("si.fanout." + this.channelName); - admin.declareExchange(exchange); - Queue queue = admin.declareQueue(); - Binding binding = BindingBuilder.bind(queue).to(exchange); - admin.declareBinding(binding); - this.container.setQueues(queue); - rabbitTemplate.setExchange(exchange.getName()); - } - else { - String queueName = "si." + this.channelName; - Queue queue = new Queue(queueName); - admin.declareQueue(queue); - this.container.setQueues(queue); - rabbitTemplate.setRoutingKey(queueName); - } - MessageListener listener = new DispatchingMessageListener(rabbitTemplate, this.dispatcher); + this.dispatcher = this.createDispatcher(); + AmqpAdmin admin = new RabbitAdmin(this.container.getConnectionFactory()); + Queue queue = this.initializeQueue(admin, this.channelName); + this.container.setQueues(queue); + MessageConverter converter = (this.getAmqpTemplate() instanceof RabbitTemplate) + ? ((RabbitTemplate) this.getAmqpTemplate()).getMessageConverter() + : new SimpleMessageConverter(); + MessageListener listener = new DispatchingMessageListener(converter, this.dispatcher); this.container.setMessageListener(listener); if (!this.container.isActive()) { this.container.afterPropertiesSet(); } - rabbitTemplate.afterPropertiesSet(); } - private void configureDispatcher() { - if (this.isPubSub) { - this.dispatcher = new BroadcastingDispatcher(); - } - else { - UnicastingDispatcher unicastingDispatcher = new UnicastingDispatcher(); - unicastingDispatcher.setLoadBalancingStrategy(new RoundRobinLoadBalancingStrategy()); - this.dispatcher = unicastingDispatcher; - } - } + protected abstract MessageDispatcher createDispatcher(); + + protected abstract Queue initializeQueue(AmqpAdmin admin, String channelName); private static class DispatchingMessageListener implements MessageListener { private final Log logger = LogFactory.getLog(this.getClass()); - private final RabbitTemplate rabbitTemplate; - private final MessageDispatcher dispatcher; + private final MessageConverter converter; - private DispatchingMessageListener(RabbitTemplate rabbitTemplate, MessageDispatcher dispatcher) { - this.rabbitTemplate = rabbitTemplate; + + private DispatchingMessageListener(MessageConverter converter, MessageDispatcher dispatcher) { + Assert.notNull(converter, "MessageConverter must not be null"); + Assert.notNull(dispatcher, "MessageDispatcher must not be null"); + this.converter = converter; this.dispatcher = dispatcher; } public void onMessage(org.springframework.amqp.core.Message message) { try { - Object converted = this.rabbitTemplate.getMessageConverter().fromMessage(message); + Object converted = this.converter.fromMessage(message); if (converted != null) { Message messageToSend = (converted instanceof Message) ? (Message) converted : MessageBuilder.withPayload(converted).build(); 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 new file mode 100644 index 0000000000..ded7d382e2 --- /dev/null +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PointToPointSubscribableAmqpChannel.java @@ -0,0 +1,71 @@ +/* + * Copyright 2002-2011 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 org.springframework.amqp.core.AmqpAdmin; +import org.springframework.amqp.core.AmqpTemplate; +import org.springframework.amqp.core.Queue; +import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer; +import org.springframework.integration.dispatcher.MessageDispatcher; +import org.springframework.integration.dispatcher.RoundRobinLoadBalancingStrategy; +import org.springframework.integration.dispatcher.UnicastingDispatcher; + +/** + * @author Mark Fisher + * @since 2.1 + */ +public class PointToPointSubscribableAmqpChannel extends AbstractSubscribableAmqpChannel { + + private volatile String queueName; + + + public PointToPointSubscribableAmqpChannel(String channelName, SimpleMessageListenerContainer container, AmqpTemplate amqpTemplate) { + super(channelName, container, amqpTemplate); + } + + + /** + * Provide a Queue name to be used. If this is not provided, + * the Queue's name will be the same as the channel name. + */ + public void setQueueName(String queueName) { + this.queueName = queueName; + } + + @Override + protected Queue initializeQueue(AmqpAdmin admin, String channelName) { + if (this.queueName == null) { + this.queueName = channelName; + } + Queue queue = new Queue(this.queueName); + admin.declareQueue(queue); + return queue; + } + + @Override + protected MessageDispatcher createDispatcher() { + UnicastingDispatcher unicastingDispatcher = new UnicastingDispatcher(); + unicastingDispatcher.setLoadBalancingStrategy(new RoundRobinLoadBalancingStrategy()); + return unicastingDispatcher; + } + + @Override + protected String getRoutingKey() { + return this.queueName; + } + +} 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 0965976b29..f806e139b2 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 @@ -16,6 +16,7 @@ package org.springframework.integration.amqp.channel; +import org.springframework.amqp.core.AmqpAdmin; import org.springframework.amqp.core.AmqpTemplate; import org.springframework.amqp.core.Queue; import org.springframework.amqp.rabbit.core.RabbitAdmin; @@ -26,6 +27,10 @@ import org.springframework.integration.support.MessageBuilder; import org.springframework.util.Assert; /** + * A {@link PollableChannel} implementation that is backed by an AMQP Queue. + * Messages will be sent to the default (no-name) exchange with that Queue's + * name as the routing key. + * * @author Mark Fisher * @since 2.1 */ @@ -33,6 +38,10 @@ public class PollableAmqpChannel extends AbstractAmqpChannel implements Pollable private final String channelName; + private volatile String queueName; + + private volatile AmqpAdmin amqpAdmin; + public PollableAmqpChannel(String channelName, AmqpTemplate amqpTemplate) { super(amqpTemplate); @@ -41,26 +50,49 @@ public class PollableAmqpChannel extends AbstractAmqpChannel implements Pollable } + /** + * 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. + */ + public void setQueueName(String queueName) { + this.queueName = 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. + */ + public void setAmqpAdmin(AmqpAdmin amqpAdmin) { + this.amqpAdmin = amqpAdmin; + } + @Override protected void onInit() throws Exception { AmqpTemplate amqpTemplate = this.getAmqpTemplate(); - if (!(amqpTemplate instanceof RabbitTemplate)) { - throw new IllegalArgumentException("AmqpTemplate must be a RabbitTemplate"); + if (this.queueName == null) { + if (this.amqpAdmin == null && amqpTemplate instanceof RabbitTemplate) { + this.amqpAdmin = new RabbitAdmin(((RabbitTemplate) amqpTemplate).getConnectionFactory()); + } + Assert.notNull(this.amqpAdmin, + "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)); } - RabbitTemplate rabbitTemplate = (RabbitTemplate) amqpTemplate; - RabbitAdmin admin = new RabbitAdmin(rabbitTemplate.getConnectionFactory()); - String queueName = "si." + this.channelName; - Queue queue = new Queue(queueName); - admin.declareQueue(queue); - rabbitTemplate.setRoutingKey(queueName); - rabbitTemplate.setQueue(queueName); + } + + @Override + protected String getRoutingKey() { + return this.queueName; } public Message receive() { if (!this.getInterceptors().preReceive(this)) { return null; } - Object object = this.getAmqpTemplate().receiveAndConvert(); + Object object = this.getAmqpTemplate().receiveAndConvert(this.queueName); if (object == null) { return 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 new file mode 100644 index 0000000000..adf5e2de6f --- /dev/null +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PublishSubscribeAmqpChannel.java @@ -0,0 +1,76 @@ +/* + * Copyright 2002-2011 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 org.springframework.amqp.core.AmqpAdmin; +import org.springframework.amqp.core.AmqpTemplate; +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.listener.SimpleMessageListenerContainer; +import org.springframework.integration.dispatcher.BroadcastingDispatcher; +import org.springframework.integration.dispatcher.MessageDispatcher; + +/** + * @author Mark Fisher + * @since 2.1 + */ +public class PublishSubscribeAmqpChannel extends AbstractSubscribableAmqpChannel { + + private volatile FanoutExchange exchange; + + + public PublishSubscribeAmqpChannel(String channelName, SimpleMessageListenerContainer container, AmqpTemplate amqpTemplate) { + super(channelName, container, amqpTemplate); + } + + + /** + * Configure the FanoutExchange instance. If this is not provided, then a + * FanoutExchange will be declared implicitly, and its name will be the same + * as the channel name prefixed by "si.fanout.". In either case, an effectively + * anonymous Queue will be declared automatically. + */ + public void setExchange(FanoutExchange exchange) { + this.exchange = exchange; + } + + @Override + protected Queue initializeQueue(AmqpAdmin admin, String channelName) { + if (this.exchange == null) { + String exchangeName = "si.fanout." + channelName; + 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; + } + + @Override + protected MessageDispatcher createDispatcher() { + return new BroadcastingDispatcher(); + } + + @Override + protected String getExchangeName() { + return (this.exchange != null) ? this.exchange.getName() : ""; + } + +} diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpChannelFactoryBean.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpChannelFactoryBean.java index c7a47f4c2c..7d8710d4eb 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpChannelFactoryBean.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpChannelFactoryBean.java @@ -22,6 +22,9 @@ import java.util.concurrent.Executor; import org.aopalliance.aop.Advice; import org.springframework.amqp.core.AcknowledgeMode; +import org.springframework.amqp.core.AmqpAdmin; +import org.springframework.amqp.core.AmqpTemplate; +import org.springframework.amqp.core.FanoutExchange; import org.springframework.amqp.rabbit.connection.ConnectionFactory; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer; @@ -29,12 +32,14 @@ import org.springframework.amqp.rabbit.support.MessagePropertiesConverter; import org.springframework.amqp.support.converter.MessageConverter; import org.springframework.beans.factory.BeanNameAware; import org.springframework.beans.factory.DisposableBean; +import org.springframework.beans.factory.InitializingBean; import org.springframework.beans.factory.config.AbstractFactoryBean; import org.springframework.context.Lifecycle; import org.springframework.context.SmartLifecycle; import org.springframework.integration.amqp.channel.AbstractAmqpChannel; +import org.springframework.integration.amqp.channel.PointToPointSubscribableAmqpChannel; import org.springframework.integration.amqp.channel.PollableAmqpChannel; -import org.springframework.integration.amqp.channel.SubscribableAmqpChannel; +import org.springframework.integration.amqp.channel.PublishSubscribeAmqpChannel; import org.springframework.integration.channel.ChannelInterceptor; import org.springframework.transaction.PlatformTransactionManager; import org.springframework.transaction.interceptor.TransactionAttribute; @@ -42,10 +47,11 @@ import org.springframework.util.Assert; import org.springframework.util.CollectionUtils; import org.springframework.util.ErrorHandler; import org.springframework.util.ObjectUtils; +import org.springframework.util.StringUtils; /** * If point-to-point, we send to the default exchange with the routing key - * equal to "si.[beanName]" and we declare that same Queue and register a listener + * equal to "[beanName]" and we declare that same Queue and register a listener * if message-driven or poll explicitly otherwise. If publish-subscribe, we declare * a FanoutExchange named "si.fanout.[beanName]" and we send to that without any * routing key, and on the receiving side, we create an anonymous Queue that is @@ -62,10 +68,16 @@ public class AmqpChannelFactoryBean extends AbstractFactoryBean + + + + Provide an explicitly configured queue name. If this is not provided, then a Queue will be created + implicitly with the same name as the channel itself (the "id" of this element). If this channel is + not message-driven, the implicit creation will require that either an AmqpAdmin instance has been + provided via the "amqp-admin" attribute or that the configured AmqpTemplate is an instance of RabbitTemplate. + If the channel is message-driven, the AmqpAdmin will be created using the underlying listener container's + ConnectionFactory. + + + + + + + An AmqpAdmin instance to use when declaring a Queue implicitly. This is only needed if an explicit + "queue-name" is not provided and the channel is not message-driven. Even then, if the referenced + AmqpTemplate is an instance of RabbitTemplate, the AmqpAdmin can be constructed from that template's + ConnectionFactory. + + + + + + + + - + Creates a publish-subscribe-channel that is backed by an AMQP FanoutExchange. Always message-driven (subscribable). + + + + + + + Reference to a FanoutExchange instance to which this channel should send Messages. If not provided, + a FanoutExchange will be declared with this channel's name prefixed by "si.fanout.". + A Queue will be declared automatically and bound to that exchange to handle the consumer role + of this channel. + + + + + + + + + + +