diff --git a/docs/src/reference/docbook/amqp.xml b/docs/src/reference/docbook/amqp.xml index ec7a58dac8..58fe3f44d9 100644 --- a/docs/src/reference/docbook/amqp.xml +++ b/docs/src/reference/docbook/amqp.xml @@ -18,6 +18,10 @@ Outbound Gateway + + Spring Integration also provides a point-to-point Message Channel as well as a + publish/subscribe Message Channel backed by AMQP Exchanges and Queues. + In order to provide AMQP support, Spring Integration relies on Spring AMQP (http://www.springsource.org/spring-amqp) @@ -27,7 +31,7 @@ Whereas the provided AMQP Channel Adapters are intended for unidirectional - Messaging (send or receive), only, Spring Integration also provides inbound + Messaging (send or receive) only, Spring Integration also provides inbound and outbound AMQP Gateways for request/reply operations. @@ -250,48 +254,48 @@ A configuration sample for an AMQP Outbound Channel Adapter is shown below with all available parameters. - ]]> + ]]> - + Unique ID for this adapter. Optional. - + Message Channel to which Messages should be sent in order to have them converted and published to an AMQP Exchange. Required. - + Bean Reference to the configured AMQP Template Optional (Defaults to "amqpTemplate"). - + The name of the AMQP Exchange to which Messages should be sent. If not provided, Messages will be sent to the default, no-name Exchange. Optional. - + The order for this consumer when multiple consumers are registered thereby enabling load- balancing and/or failover. Optional (Defaults to Ordered.LOWEST_PRECEDENCE [=Integer.MAX_VALUE]). - + The fixed routing-key to use when sending Messages. By default, this will be an empty String. Optional. - + The routing-key to use when sending Messages evaluated as an expression on the message (e.g. 'payload.key'). By default, this will be an empty String. @@ -419,15 +423,30 @@ AMQP Backed Message Channels - This feature is not currently available, yet, but it is planned for the - Spring Integration 2.1 M2 release. In order to follow progress, please - visit the Spring Integration issue tracker at - - https://jira.springsource.org/browse/INT-1878 - + There are two Message Channel implementations available. One is point-to-point, and the other is publish/subscribe. + Both of these channels provide a wide range of configuration attributes for the underlying AmqpTemplate and + SimpleMessageListenerContainer as you have seen on the Channel Adapters and Gateways. However, the examples we'll + show here are going to have minimal configuration. Explore the XML schema to view the available attributes. - + + A point-to-point channel would look like this: + ]]> + Under the covers a Queue named "si.p2pChannel" would be declared, and this channel will send to + that Queue (technically by sending to the no-name Direct Exchange with a routing key that matches this Queue's name). + This channel will also register a consumer on that Queue. If for some reason, you want the Queue to be "pollable" + instead of message-driven, then simply provide the "message-driven" flag with a value of false: + ]]> + + + A publish/subscribe channel would look like this: + ]]> + Under the covers a Fanout Exchange named "si.fanout.pubSubChannel" would be declared, and this channel will send + to that Fanout Exchange. This channel will also declare a server-named exclusive, autodelete, non-durable Queue + and bind that to the Fanout Exchange while registering a consumer on that Queue to receive Messages. There is no + "pollable" option for a publish-subscribe-channel; it must be message-driven. + +
AMQP Samples 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 new file mode 100644 index 0000000000..dab69d5eb9 --- /dev/null +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/AbstractAmqpChannel.java @@ -0,0 +1,49 @@ +/* + * 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.AmqpTemplate; +import org.springframework.integration.Message; +import org.springframework.integration.channel.AbstractMessageChannel; +import org.springframework.util.Assert; + +/** + * @author Mark Fisher + * @since 2.1 + */ +public abstract class AbstractAmqpChannel extends AbstractMessageChannel { + + private final AmqpTemplate amqpTemplate; + + + AbstractAmqpChannel(AmqpTemplate amqpTemplate) { + Assert.notNull(amqpTemplate, "amqpTemplate must not be null"); + this.amqpTemplate = amqpTemplate; + } + + + AmqpTemplate getAmqpTemplate() { + return this.amqpTemplate; + } + + @Override + protected boolean doSend(Message message, long timeout) { + this.amqpTemplate.convertAndSend(message); + return true; + } + +} 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 new file mode 100644 index 0000000000..0965976b29 --- /dev/null +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PollableAmqpChannel.java @@ -0,0 +1,85 @@ +/* + * 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.AmqpTemplate; +import org.springframework.amqp.core.Queue; +import org.springframework.amqp.rabbit.core.RabbitAdmin; +import org.springframework.amqp.rabbit.core.RabbitTemplate; +import org.springframework.integration.Message; +import org.springframework.integration.core.PollableChannel; +import org.springframework.integration.support.MessageBuilder; +import org.springframework.util.Assert; + +/** + * @author Mark Fisher + * @since 2.1 + */ +public class PollableAmqpChannel extends AbstractAmqpChannel implements PollableChannel { + + private final String channelName; + + + public PollableAmqpChannel(String channelName, AmqpTemplate amqpTemplate) { + super(amqpTemplate); + Assert.hasText(channelName, "channel name must not be empty"); + this.channelName = channelName; + } + + + @Override + protected void onInit() throws Exception { + 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()); + String queueName = "si." + this.channelName; + Queue queue = new Queue(queueName); + admin.declareQueue(queue); + rabbitTemplate.setRoutingKey(queueName); + rabbitTemplate.setQueue(queueName); + } + + public Message receive() { + if (!this.getInterceptors().preReceive(this)) { + return null; + } + Object object = this.getAmqpTemplate().receiveAndConvert(); + if (object == null) { + return null; + } + Message replyMessage = null; + if (object instanceof Message) { + replyMessage = (Message) object; + } + else { + replyMessage = MessageBuilder.withPayload(object).build(); + } + return this.getInterceptors().postReceive(replyMessage, this) ; + } + + public Message receive(long timeout) { + if (logger.isInfoEnabled()) { + logger.info("Calling receive with a timeout value on PollableAmqpChannel. " + + "The timeout will be ignored since no receive timeout is supported."); + } + return this.receive(); + } + +} 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/SubscribableAmqpChannel.java new file mode 100644 index 0000000000..16bc2ad2dd --- /dev/null +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/SubscribableAmqpChannel.java @@ -0,0 +1,196 @@ +/* + * 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.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +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.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; + +/** + * @author Mark Fisher + * @since 2.1 + */ +public class SubscribableAmqpChannel 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) { + 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; + } + + + public boolean subscribe(MessageHandler handler) { + return this.dispatcher.addHandler(handler); + } + + public boolean unsubscribe(MessageHandler handler) { + return this.dispatcher.removeHandler(handler); + } + + @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.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; + } + } + + + private static class DispatchingMessageListener implements MessageListener { + + private final Log logger = LogFactory.getLog(this.getClass()); + + private final RabbitTemplate rabbitTemplate; + + private final MessageDispatcher dispatcher; + + + private DispatchingMessageListener(RabbitTemplate rabbitTemplate, MessageDispatcher dispatcher) { + this.rabbitTemplate = rabbitTemplate; + this.dispatcher = dispatcher; + } + + + public void onMessage(org.springframework.amqp.core.Message message) { + try { + Object converted = this.rabbitTemplate.getMessageConverter().fromMessage(message); + if (converted != null) { + Message messageToSend = (converted instanceof Message) ? (Message) converted + : MessageBuilder.withPayload(converted).build(); + this.dispatcher.dispatch(messageToSend); + } + else if (this.logger.isWarnEnabled()) { + logger.warn("MessageConverter returned null, no Message to dispatch"); + } + } + catch (Exception e) { + throw new MessagingException("Failure occured in AMQP listener while attempting to convert and dispatch Message.", e); + } + } + } + + + /* + * SmartLifecycle implementation (delegates to the MessageListener container) + */ + + public boolean isAutoStartup() { + return (this.container != null) ? this.container.isAutoStartup() : false; + } + + public int getPhase() { + return (this.container != null) ? this.container.getPhase() : 0; + } + + public boolean isRunning() { + return (this.container != null) ? this.container.isRunning() : false; + } + + public void start() { + if (this.container != null) { + this.container.start(); + } + } + + public void stop() { + if (this.container != null) { + this.container.stop(); + } + } + + public void stop(Runnable callback) { + if (this.container != null) { + this.container.stop(callback); + } + } + + public void destroy() throws Exception { + if (this.container != null) { + this.container.destroy(); + } + } + +} 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 new file mode 100644 index 0000000000..c7a47f4c2c --- /dev/null +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpChannelFactoryBean.java @@ -0,0 +1,350 @@ +/* + * 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.config; + +import java.util.List; +import java.util.concurrent.Executor; + +import org.aopalliance.aop.Advice; + +import org.springframework.amqp.core.AcknowledgeMode; +import org.springframework.amqp.rabbit.connection.ConnectionFactory; +import org.springframework.amqp.rabbit.core.RabbitTemplate; +import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer; +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.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.PollableAmqpChannel; +import org.springframework.integration.amqp.channel.SubscribableAmqpChannel; +import org.springframework.integration.channel.ChannelInterceptor; +import org.springframework.transaction.PlatformTransactionManager; +import org.springframework.transaction.interceptor.TransactionAttribute; +import org.springframework.util.Assert; +import org.springframework.util.CollectionUtils; +import org.springframework.util.ErrorHandler; +import org.springframework.util.ObjectUtils; + +/** + * 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 + * 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 + * bound to that exchange. + * + * @author Mark Fisher + * @since 2.1 + */ +public class AmqpChannelFactoryBean extends AbstractFactoryBean implements SmartLifecycle, DisposableBean, BeanNameAware { + + private volatile AbstractAmqpChannel channel; + + private volatile List interceptors; + + private final boolean messageDriven; + + private final RabbitTemplate rabbitTemplate = new RabbitTemplate(); + + private volatile SimpleMessageListenerContainer container; + + private volatile boolean autoStartup = true; + + private volatile Advice[] adviceChain; + + private volatile Integer concurrentConsumers; + + private volatile ConnectionFactory connectionFactory; + + private volatile MessagePropertiesConverter messagePropertiesConverter; + + private volatile ErrorHandler errorHandler; + + private volatile Boolean exposeListenerChannel; + + private volatile Integer phase; + + private volatile Integer prefetchCount; + + private volatile Boolean isPubSub; + + private volatile Long receiveTimeout; + + private volatile Long recoveryInterval; + + private volatile Long shutdownTimeout; + + private volatile String beanName; + + private volatile AcknowledgeMode acknowledgeMode; + + /** + * This value differs from the container implementations' default (which is false). + */ + private volatile boolean channelTransacted = true; + + private volatile Executor taskExecutor; + + private volatile PlatformTransactionManager transactionManager; + + private volatile TransactionAttribute transactionAttribute; + + private volatile Integer txSize; + + + public AmqpChannelFactoryBean() { + this(true); + } + + public AmqpChannelFactoryBean(boolean messageDriven) { + this.messageDriven = messageDriven; + } + + + public void setBeanName(String name) { + this.beanName = name; + } + + public void setInterceptors(List interceptors) { + this.interceptors = interceptors; + } + + /* + * Template-only properties + */ + + public void setEncoding(String encoding) { + this.rabbitTemplate.setEncoding(encoding); + } + + public void setMessageConverter(MessageConverter messageConverter) { + this.rabbitTemplate.setMessageConverter(messageConverter); + } + + /* + * Template and Container properties + */ + + public void setChannelTransacted(boolean channelTransacted) { + this.channelTransacted = channelTransacted; + this.rabbitTemplate.setChannelTransacted(channelTransacted); + } + + public void setConnectionFactory(ConnectionFactory connectionFactory) { + this.connectionFactory = connectionFactory; + this.rabbitTemplate.setConnectionFactory(this.connectionFactory); + } + + public void setMessagePropertiesConverter(MessagePropertiesConverter messagePropertiesConverter) { + this.rabbitTemplate.setMessagePropertiesConverter(messagePropertiesConverter); + this.messagePropertiesConverter = messagePropertiesConverter; + } + + /* + * Container-only properties + */ + + public void setAcknowledgeMode(AcknowledgeMode acknowledgeMode) { + this.acknowledgeMode = acknowledgeMode; + } + + public void setAdviceChain(Advice[] adviceChain) { + this.adviceChain = adviceChain; + } + + public void setAutoStartup(boolean autoStartup) { + this.autoStartup = autoStartup; + } + + public void setConcurrentConsumers(int concurrentConsumers) { + this.concurrentConsumers = concurrentConsumers; + } + + public void setErrorHandler(ErrorHandler errorHandler) { + this.errorHandler = errorHandler; + } + + public void setExposeListenerChannel(boolean exposeListenerChannel) { + this.exposeListenerChannel = exposeListenerChannel; + } + + public void setPhase(int phase) { + this.phase = phase; + } + + public void setPrefetchCount(int prefetchCount) { + this.prefetchCount = prefetchCount; + } + + public void setPubSub(boolean pubSub) { + this.isPubSub = pubSub; + } + + public void setReceiveTimeout(long receiveTimeout) { + this.receiveTimeout = receiveTimeout; + } + + public void setRecoveryInterval(long recoveryInterval) { + this.recoveryInterval = recoveryInterval; + } + + public void setShutdownTimeout(long shutdownTimeout) { + this.shutdownTimeout = shutdownTimeout; + } + + public void setTaskExecutor(Executor taskExecutor) { + this.taskExecutor = taskExecutor; + } + + public void setTransactionAttribute(TransactionAttribute transactionAttribute) { + this.transactionAttribute = transactionAttribute; + } + + public void setTransactionManager(PlatformTransactionManager transactionManager) { + this.transactionManager = transactionManager; + } + + public void setTxSize(int txSize) { + this.txSize = this.txSize; + } + + @Override + public Class getObjectType() { + return (this.channel != null) ? this.channel.getClass() : AbstractAmqpChannel.class; + } + + @Override + protected AbstractAmqpChannel createInstance() throws Exception { + if (this.messageDriven) { + this.container = this.createContainer(); + this.channel = new SubscribableAmqpChannel(this.beanName, this.container, this.rabbitTemplate, this.isPubSub); + } + else { + Assert.isTrue(!Boolean.TRUE.equals(this.isPubSub), + "An AMQP 'publish-subscribe-channel' must be message-driven."); + this.channel = new PollableAmqpChannel(this.beanName, this.rabbitTemplate); + } + if (!CollectionUtils.isEmpty(this.interceptors)) { + this.channel.setInterceptors(this.interceptors); + } + this.channel.afterPropertiesSet(); + this.channel.setBeanName(this.beanName); + return this.channel; + } + + private SimpleMessageListenerContainer createContainer() throws Exception { + //if (!messageDriven) TODO: no container attributes would apply if not message-driven + SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(); + if (this.acknowledgeMode != null) { + container.setAcknowledgeMode(this.acknowledgeMode); + } + if (!ObjectUtils.isEmpty(this.adviceChain)) { + container.setAdviceChain(this.adviceChain); + } + container.setAutoStartup(this.autoStartup); + container.setChannelTransacted(this.channelTransacted); + if (this.concurrentConsumers != null) { + container.setConcurrentConsumers(this.concurrentConsumers); + } + container.setConnectionFactory(this.connectionFactory); + if (this.errorHandler != null) { + container.setErrorHandler(this.errorHandler); + } + if (this.exposeListenerChannel != null) { + container.setExposeListenerChannel(this.exposeListenerChannel); + } + if (this.messagePropertiesConverter != null) { + container.setMessagePropertiesConverter(this.messagePropertiesConverter); + } + if (this.phase != null) { + container.setPhase(this.phase); + } + if (this.prefetchCount != null) { + container.setPrefetchCount(this.prefetchCount); + } + if (this.receiveTimeout != null) { + container.setReceiveTimeout(this.receiveTimeout); + } + if (this.recoveryInterval != null) { + container.setRecoveryInterval(this.recoveryInterval); + } + if (this.shutdownTimeout != null) { + container.setShutdownTimeout(this.shutdownTimeout); + } + if (this.taskExecutor != null) { + container.setTaskExecutor(this.taskExecutor); + } + if (this.transactionAttribute != null) { + container.setTransactionAttribute(this.transactionAttribute); + } + if (this.transactionManager != null) { + container.setTransactionManager(this.transactionManager); + } + if (this.txSize != null) { + container.setTxSize(this.txSize); + } + return container; + } + + /* + * SmartLifecycle implementation (delegates to the created channel if message-driven) + */ + + public boolean isAutoStartup() { + return (this.channel instanceof SmartLifecycle) ? + ((SmartLifecycle) this.channel).isAutoStartup() : false; + } + + public int getPhase() { + return (this.channel instanceof SmartLifecycle) ? + ((SmartLifecycle) this.channel).getPhase() : 0; + } + + public boolean isRunning() { + return (this.channel instanceof Lifecycle) ? + ((Lifecycle) this.channel).isRunning() : false; + } + + public void start() { + if (this.channel instanceof Lifecycle) { + ((Lifecycle) this.channel).start(); + } + } + + public void stop() { + if (this.channel instanceof Lifecycle) { + ((Lifecycle) this.channel).stop(); + } + } + + public void stop(Runnable callback) { + if (this.channel instanceof SmartLifecycle) { + ((SmartLifecycle) this.channel).stop(callback); + } + } + + protected void destroyInstance(AbstractAmqpChannel instance) throws Exception { + if (instance instanceof DisposableBean) { + ((DisposableBean) this.channel).destroy(); + } + } + +} diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpChannelParser.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpChannelParser.java new file mode 100644 index 0000000000..93b347b653 --- /dev/null +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpChannelParser.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.config; + +import org.w3c.dom.Element; + +import org.springframework.beans.factory.support.BeanDefinitionBuilder; +import org.springframework.beans.factory.xml.ParserContext; +import org.springframework.integration.config.xml.AbstractChannelParser; +import org.springframework.integration.config.xml.IntegrationNamespaceUtils; +import org.springframework.util.StringUtils; + +/** + * Parser for the 'channel' and 'publish-subscribe-channel' elements of the + * Spring Integration AMQP namespace. + * + * @author Mark Fisher + * @since 2.1 + */ +public class AmqpChannelParser extends AbstractChannelParser { + + @Override + protected BeanDefinitionBuilder buildBeanDefinition(Element element, ParserContext parserContext) { + BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(AmqpChannelFactoryBean.class); + String messageDriven = element.getAttribute("message-driven"); + if (StringUtils.hasText(messageDriven)) { + builder.addConstructorArgValue(messageDriven); + } + String connectionFactory = element.getAttribute("connection-factory"); + if (!StringUtils.hasText(connectionFactory)) { + connectionFactory = "rabbitConnectionFactory"; + } + builder.addPropertyReference("connectionFactory", connectionFactory); + if ("channel".equals(element.getLocalName())) { + builder.addPropertyValue("pubSub", false); + } + else if ("publish-subscribe-channel".equals(element.getLocalName())) { + builder.addPropertyValue("pubSub", true); + } + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "acknowledge-mode"); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "advice-chain"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "auto-startup"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "channel-transacted"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "concurrent-consumers"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "encoding"); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "error-handler"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "expose-listener-channel"); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "message-converter"); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "message-properties-converter"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "phase"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "prefetch-count"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "receive-timeout"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "recovery-interval"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "shutdown-timeout"); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "task-executor"); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "transaction-attribute"); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "transaction-manager"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "tx-size"); + return builder; + } + +} diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpNamespaceHandler.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpNamespaceHandler.java index 876566bc40..a5f854f725 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpNamespaceHandler.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpNamespaceHandler.java @@ -27,6 +27,8 @@ import org.springframework.integration.config.xml.AbstractIntegrationNamespaceHa public class AmqpNamespaceHandler extends AbstractIntegrationNamespaceHandler { public void init() { + this.registerBeanDefinitionParser("channel", new AmqpChannelParser()); + this.registerBeanDefinitionParser("publish-subscribe-channel", new AmqpChannelParser()); this.registerBeanDefinitionParser("inbound-channel-adapter", new AmqpInboundChannelAdapterParser()); this.registerBeanDefinitionParser("inbound-gateway", new AmqpInboundGatewayParser()); this.registerBeanDefinitionParser("outbound-channel-adapter", new AmqpOutboundChannelAdapterParser()); diff --git a/spring-integration-amqp/src/main/resources/org/springframework/integration/amqp/config/spring-integration-amqp-2.1.xsd b/spring-integration-amqp/src/main/resources/org/springframework/integration/amqp/config/spring-integration-amqp-2.1.xsd index 3525f85962..c6726ff496 100644 --- a/spring-integration-amqp/src/main/resources/org/springframework/integration/amqp/config/spring-integration-amqp-2.1.xsd +++ b/spring-integration-amqp/src/main/resources/org/springframework/integration/amqp/config/spring-integration-amqp-2.1.xsd @@ -242,6 +242,60 @@ + + + + Creates a point-to-point channel that is backed by an AMQP Queue. + + + + + + + + + Indicate whether this channel should be message-driven (subscribable) or not (pollable). + + + + + + + + + + + + Creates a publish-subscribe-channel that is backed by an AMQP FanoutExchange. + Always message-driven (subscribable). + + + + + + + + Base type for 'channel' and 'publish-subscribe-channel'. + + + + + + Unique ID for this Message Channel. + + + + + + + Flag to indicate whether this Message Channel should start automatically. + This only applies to a message-driven channel. Default is true. + + + + + + @@ -255,58 +309,7 @@ - - - - Acknowledge Mode for the MessageListenerContainer. - - - - - - - - - - - - - - - - - - Array of AOP Advice instances to be applied to the MessageListener. - - - - - - - - Flag to indicate that AMQP channels created by this component will be transactional. - - - - - - - Specify the number of concurrent consumers to create. Default is 1. - Raising the number of concurrent consumers is recommended in order to scale the consumption of messages coming in - from a queue. However, note that any ordering guarantees are lost once multiple consumers are registered. In - general, stick with 1 consumer for low-volume queues. - - - - - - - - - - - - + @@ -319,28 +322,18 @@ - + - ErrorHandler to be configured on the underlying MessageListener container. + MessageConverter to use when receiving AMQP Messages. - + - - - - - Set whether to expose the listener Rabbit Channel to a registered ChannelAwareMessageListener as well as - to RabbitTemplate calls. - - - - @@ -369,14 +362,47 @@ - + - MessageConverter to use when receiving AMQP Messages. + Names of the AMQP Queues from which Messages should be consumed (comma-separated list). + + + + + + + + + + + + + + + + + + + + + + + + + + Flag to indicate that channels created by this component will be transactional. + + + + + + + Reference to the Rabbit ConnectionFactory to be used by this component. - + @@ -393,6 +419,105 @@ + + + + + + Attributes for a RabbitTemplate. This does not include the exchange, queue, or routingKey properties + since those may or may not be exposed for configuration depending on what type of component uses this + attribute group. This group also does not include any of the properties that are shared with the + SimpleMessageListenerContainer, such as channelTransacted, connectionFactory, and messagePropertiesConverter. + + + + + + The encoding to use when converting between byte arrays and Strings in message properties. + + + + + + + Reference to a MessageConverter to be used by this RabbitTemplate. + + + + + + + + + + + + + + Attributes for a SimpleMesssageListenerContainer's properties other than queues, queueNames, messageListener, and + autoStartup which may or may not be exposed for configuration depending on what type of component uses this attribute group. + This group also does not include any of the properties that are shared with RabbitTemplate, such as channelTransacted, + connectionFactory, and messsagePropertiesConverter. + + + + + + Acknowledge Mode for the MessageListenerContainer. + + + + + + + + + + + + + + + + + + Array of AOP Advice instances to be applied to the MessageListener. + + + + + + + + Specify the number of concurrent consumers to create. Default is 1. + Raising the number of concurrent consumers is recommended in order to scale the consumption of messages coming in + from a queue. However, note that any ordering guarantees are lost once multiple consumers are registered. In + general, stick with 1 consumer for low-volume queues. + + + + + + + ErrorHandler to be configured on the underlying MessageListener container. + + + + + + + + + + + + + Set whether to expose the listener Rabbit Channel to a registered ChannelAwareMessageListener as well as + to RabbitTemplate calls. + + + + @@ -412,13 +537,6 @@ - - - - Names of the AMQP Queues from which Messages should be consumed (comma-separated list). - - - @@ -446,16 +564,6 @@ - - - - - How many messages to process in a single transaction (if the channel is transactional). For best results it should be - less than or equal to the prefetch count. - - - - @@ -493,6 +601,16 @@ - + + + + + How many messages to process in a single transaction (if the channel is transactional). For best results it should be + less than or equal to the prefetch count. + + + + + diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/ChannelSample-context.xml b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/ChannelSample-context.xml new file mode 100644 index 0000000000..41101975c5 --- /dev/null +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/ChannelSample-context.xml @@ -0,0 +1,28 @@ + + + + + + + + + + + + + + + + + + diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/ChannelSample.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/ChannelSample.java new file mode 100644 index 0000000000..af63d8dca1 --- /dev/null +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/ChannelSample.java @@ -0,0 +1,31 @@ +/* + * 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.config; + +import org.springframework.context.support.ClassPathXmlApplicationContext; + +/** + * @author Mark Fisher + * @since 2.1 + */ +public class ChannelSample { + + public static void main(String[] args) { + new ClassPathXmlApplicationContext("ChannelSample-context.xml", ChannelSample.class); + } + +} diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/PubSubChannelSample-context.xml b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/PubSubChannelSample-context.xml new file mode 100644 index 0000000000..b6f398009a --- /dev/null +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/PubSubChannelSample-context.xml @@ -0,0 +1,28 @@ + + + + + + + + + + + + + + + + + + diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/PubSubChannelSample.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/PubSubChannelSample.java new file mode 100644 index 0000000000..afdf37d904 --- /dev/null +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/PubSubChannelSample.java @@ -0,0 +1,31 @@ +/* + * 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.config; + +import org.springframework.context.support.ClassPathXmlApplicationContext; + +/** + * @author Mark Fisher + * @since 2.1 + */ +public class PubSubChannelSample { + + public static void main(String[] args) { + new ClassPathXmlApplicationContext("PubSubChannelSample-context.xml", PubSubChannelSample.class); + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/AbstractDispatcher.java b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/AbstractDispatcher.java index d57ffec6b2..70e711d9b9 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/AbstractDispatcher.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/AbstractDispatcher.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2009 the original author or authors. + * 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. @@ -25,6 +25,7 @@ import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.springframework.integration.core.MessageHandler; +import org.springframework.util.Assert; /** * Base class for {@link MessageDispatcher} implementations. @@ -61,6 +62,7 @@ public abstract class AbstractDispatcher implements MessageDispatcher { * @return the result of {@link Set#add(Object)} */ public boolean addHandler(MessageHandler handler) { + Assert.notNull(handler, "handler must not be null"); return this.handlers.add(handler); } @@ -70,6 +72,7 @@ public abstract class AbstractDispatcher implements MessageDispatcher { * @return the result of {@link Set#remove(Object)} */ public boolean removeHandler(MessageHandler handler) { + Assert.notNull(handler, "handler must not be null"); return this.handlers.remove(handler); }