From 5944e34b6eaccc828e64e7a6005bc4e3af109955 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Wed, 11 Apr 2012 16:47:40 -0400 Subject: [PATCH] AMQP-213 Publisher Confirms/Returns Initial Commit Add support for multiple acks in one call. AMQP-213 Polishing Move confirmation state to PublisherCallbackChannel. Allows the RabbitTemplate to handle confirms from multiple channels. Remove unnecessary synchronization. Add tests for multiple threads - verifying the functionality when more than one thread concurrently uses the template. NOTES: The CF cache size must be large enough to cover the number of required channels - otherwise the channel is closed before the confirms are received. This could be alleviated by not closing channels if P.C. are turned on. Each PublisherCallbackConnection has two maps, the first map (seqToListener) identifies which listener gets the confirmation. The second map (pendingConfirms) is a map of maps where the key is the listener and the value is a map of pending confirms for that listener. This map is given back to the listener when registering. In this way, the Rabbit template does not need to hold a reference to every connection he has used. Instead, when a connection is closed, the listener is informed via a call to removePendingConfirmsReference(). The listener uses the pending confirms maps to determine if any confirms have expired. Add test for two templates sharing the same factory, ensuring the appropriate template gets the confirmation. AMQP-213 Polishing AMQP-213 Add Support for Basic.Return The rabbit template will be registered to receive return messages if it is provided with a ReturnCallback. Either mandatory or immediate must also be set for returns. mandator and immediate are ignored if no callback is provided. AMQP-213 Polishing The RabbitTemplate can only support one each of ConfirmCallback and ReturnCallback because there is no way to correlate back to the caller. In situations where more than one client is needed, each needs its own RabbitTemplate. The connection factory, and hence the connection and channels, can be shared across templates. AMQP-213 Add Namespace Support for Returns Attributes added to template for mandatory, immediate, return-callback. Also confirm-callback for confirms. AMQP-213 Return Reply Code Etc Pass the reply code, reply text, exchange, and routing key back to the client on returns. AMQP-213 Polishing Fix non-default exchange on convertAndSend. AMQP-213 Polishing PR Review --- .../config/ConnectionFactoryParser.java | 7 + .../amqp/rabbit/config/TemplateParser.java | 40 +- .../connection/CachingConnectionFactory.java | 37 +- .../amqp/rabbit/core/RabbitTemplate.java | 225 ++++++- .../amqp/rabbit/support/CorrelationData.java | 45 ++ .../amqp/rabbit/support/PendingConfirm.java | 55 ++ .../support/PublisherCallbackChannel.java | 104 ++++ .../support/PublisherCallbackChannelImpl.java | 559 ++++++++++++++++++ .../amqp/rabbit/config/spring-rabbit-1.0.xsd | 58 ++ .../rabbit/config/TemplateParserTests.java | 18 + .../core/RabbitTemplateHeaderTests.java | 1 + ...atePublisherCallbacksIntegrationTests.java | 351 +++++++++++ .../config/TemplateParserTests-context.xml | 10 + 13 files changed, 1474 insertions(+), 36 deletions(-) create mode 100644 spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/CorrelationData.java create mode 100644 spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/PendingConfirm.java create mode 100644 spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/PublisherCallbackChannel.java create mode 100644 spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/PublisherCallbackChannelImpl.java create mode 100644 spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests.java diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/ConnectionFactoryParser.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/ConnectionFactoryParser.java index b8f7fc55..e9fbc671 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/ConnectionFactoryParser.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/ConnectionFactoryParser.java @@ -43,6 +43,10 @@ class ConnectionFactoryParser extends AbstractSingleBeanDefinitionParser { private static final String EXECUTOR_ATTRIBUTE = "executor"; + private static final String PUBLISHER_CONFIRMS = "publisher-confirms"; + + private static final String PUBLISHER_RETURNS = "publisher-returns"; + @Override protected Class getBeanClass(Element element) { return CachingConnectionFactory.class; @@ -74,6 +78,9 @@ class ConnectionFactoryParser extends AbstractSingleBeanDefinitionParser { NamespaceUtils.setValueIfAttributeDefined(builder, element, VIRTUAL_HOST_ATTRIBUTE); NamespaceUtils.setReferenceIfAttributeDefined(builder, element, EXECUTOR_ATTRIBUTE); NamespaceUtils.setValueIfAttributeDefined(builder, element, ADDRESSES); + NamespaceUtils.setValueIfAttributeDefined(builder, element, PUBLISHER_CONFIRMS); + NamespaceUtils.setValueIfAttributeDefined(builder, element, PUBLISHER_RETURNS); + } } diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/TemplateParser.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/TemplateParser.java index 9a85f235..5e69171c 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/TemplateParser.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/TemplateParser.java @@ -13,6 +13,8 @@ package org.springframework.amqp.rabbit.config; +import java.util.List; + import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.beans.factory.config.BeanDefinition; import org.springframework.beans.factory.config.RuntimeBeanReference; @@ -20,6 +22,7 @@ import org.springframework.beans.factory.support.BeanDefinitionBuilder; import org.springframework.beans.factory.xml.AbstractSingleBeanDefinitionParser; import org.springframework.beans.factory.xml.ParserContext; import org.springframework.util.StringUtils; +import org.springframework.util.xml.DomUtils; import org.w3c.dom.Element; import org.w3c.dom.Node; import org.w3c.dom.NodeList; @@ -50,6 +53,14 @@ class TemplateParser extends AbstractSingleBeanDefinitionParser { private static final String LISTENER_ELEMENT = "reply-listener"; + private static final String MANDATORY_ATTRIBUTE = "mandatory"; + + private static final String IMMEDIATE_ATTRIBUTE = "immediate"; + + private static final String RETURN_CALLBACK_ATTRIBUTE = "return-callback"; + + private static final String CONFIRM_CALLBACK_ATTRIBUTE = "confirm-callback"; + @Override protected Class getBeanClass(Element element) { return RabbitTemplate.class; @@ -87,9 +98,17 @@ class TemplateParser extends AbstractSingleBeanDefinitionParser { NamespaceUtils.setValueIfAttributeDefined(builder, element, ENCODING_ATTRIBUTE); NamespaceUtils.setReferenceIfAttributeDefined(builder, element, MESSAGE_CONVERTER_ATTRIBUTE); NamespaceUtils.setReferenceIfAttributeDefined(builder, element, REPLY_QUEUE_ATTRIBUTE); + NamespaceUtils.setValueIfAttributeDefined(builder, element, MANDATORY_ATTRIBUTE); + NamespaceUtils.setValueIfAttributeDefined(builder, element, IMMEDIATE_ATTRIBUTE); + NamespaceUtils.setReferenceIfAttributeDefined(builder, element, RETURN_CALLBACK_ATTRIBUTE); + NamespaceUtils.setReferenceIfAttributeDefined(builder, element, CONFIRM_CALLBACK_ATTRIBUTE); BeanDefinition replyContainer = null; - Element childElement = getChildElement(element, parserContext); + Element childElement = null; + List childElements = DomUtils.getChildElementsByTagName(element, LISTENER_ELEMENT); + if (childElements.size() > 0) { + childElement = childElements.get(0); + } if (childElement != null) { replyContainer = parseListener(childElement, element, parserContext); @@ -116,27 +135,8 @@ class TemplateParser extends AbstractSingleBeanDefinitionParser { } } - private Element getChildElement(Element element, - ParserContext parserContext) { - Element childElement = null; - NodeList childNodes = element.getChildNodes(); - for (int i = 0; i < childNodes.getLength(); i++) { - Node child = childNodes.item(i); - if (child.getNodeType() == Node.ELEMENT_NODE) { - String localName = parserContext.getDelegate().getLocalName(child); - if (LISTENER_ELEMENT.equals(localName)) { - childElement = (Element) child; - } - } - } - return childElement; - } - private BeanDefinition parseListener(Element childElement, Element element, ParserContext parserContext) { - if (getChildElement(childElement, parserContext) != null) { - parserContext.getReaderContext().error(" is not allowed any child elements.", element); - } BeanDefinition replyContainer = RabbitNamespaceUtils.parseContainer(childElement, parserContext); if (replyContainer != null) { replyContainer.getPropertyValues().add( diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactory.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactory.java index 63f51861..062c9493 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactory.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactory.java @@ -13,6 +13,7 @@ package org.springframework.amqp.rabbit.connection; +import java.io.IOException; import java.lang.reflect.InvocationHandler; import java.lang.reflect.InvocationTargetException; import java.lang.reflect.Method; @@ -21,6 +22,8 @@ import java.util.LinkedList; import java.util.List; import org.springframework.amqp.AmqpException; +import org.springframework.amqp.rabbit.support.PublisherCallbackChannel; +import org.springframework.amqp.rabbit.support.PublisherCallbackChannelImpl; import org.springframework.util.Assert; import org.springframework.util.StringUtils; @@ -57,6 +60,8 @@ public class CachingConnectionFactory extends AbstractConnectionFactory { private ChannelCachingConnectionProxy connection; + private volatile boolean publisherConfirms; + /** Synchronization monitor for the shared Connection */ private final Object connectionMonitor = new Object(); @@ -118,6 +123,14 @@ public class CachingConnectionFactory extends AbstractConnectionFactory { return this.channelCacheSize; } + public boolean isPublisherConfirms() { + return publisherConfirms; + } + + public void setPublisherConfirms(boolean publisherConfirms) { + this.publisherConfirms = publisherConfirms; + } + public void setConnectionListeners(List listeners) { super.setConnectionListeners(listeners); // If the connection is already alive we assume that the new listeners want to be notified @@ -159,8 +172,15 @@ public class CachingConnectionFactory extends AbstractConnectionFactory { logger.debug("Creating cached Rabbit Channel from " + targetChannel); } getChannelListener().onCreate(targetChannel, transactional); + Class[] interfaces; + if (this.publisherConfirms) { + interfaces = new Class[] { ChannelProxy.class, PublisherCallbackChannel.class }; + } + else { + interfaces = new Class[] { ChannelProxy.class }; + } return (ChannelProxy) Proxy.newProxyInstance(ChannelProxy.class.getClassLoader(), - new Class[] { ChannelProxy.class }, new CachedChannelInvocationHandler(targetChannel, channelList, + interfaces, new CachedChannelInvocationHandler(targetChannel, channelList, transactional)); } @@ -170,7 +190,20 @@ public class CachingConnectionFactory extends AbstractConnectionFactory { // Use createConnection here not doCreateConnection so that the old one is properly disposed createConnection(); } - return this.connection.createBareChannel(transactional); + Channel channel = this.connection.createBareChannel(transactional); + if (this.publisherConfirms) { + try { + channel.confirmSelect(); + } catch (IOException e) { + // TODO Auto-generated catch block + e.printStackTrace(); + } + if (!(channel instanceof PublisherCallbackChannelImpl)) { + channel = new PublisherCallbackChannelImpl(channel); + } + } + // TODO returns + return channel; } public final Connection createConnection() throws AmqpException { diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitTemplate.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitTemplate.java index 705d23db..aecd106b 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitTemplate.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitTemplate.java @@ -14,7 +14,13 @@ package org.springframework.amqp.rabbit.core; import java.io.IOException; +import java.util.Collection; +import java.util.HashSet; +import java.util.Iterator; import java.util.Map; +import java.util.Map.Entry; +import java.util.Set; +import java.util.SortedMap; import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.LinkedBlockingQueue; @@ -33,14 +39,18 @@ import org.springframework.amqp.rabbit.connection.ConnectionFactoryUtils; import org.springframework.amqp.rabbit.connection.RabbitAccessor; import org.springframework.amqp.rabbit.connection.RabbitResourceHolder; import org.springframework.amqp.rabbit.connection.RabbitUtils; +import org.springframework.amqp.rabbit.support.CorrelationData; import org.springframework.amqp.rabbit.support.DefaultMessagePropertiesConverter; import org.springframework.amqp.rabbit.support.MessagePropertiesConverter; +import org.springframework.amqp.rabbit.support.PendingConfirm; +import org.springframework.amqp.rabbit.support.PublisherCallbackChannel; import org.springframework.amqp.support.converter.MessageConverter; import org.springframework.amqp.support.converter.SimpleMessageConverter; import org.springframework.util.Assert; import org.springframework.util.StringUtils; import com.rabbitmq.client.AMQP; +import com.rabbitmq.client.AMQP.BasicProperties; import com.rabbitmq.client.AMQP.Queue.DeclareOk; import com.rabbitmq.client.Channel; import com.rabbitmq.client.DefaultConsumer; @@ -82,7 +92,8 @@ import com.rabbitmq.client.GetResponse; * @author Gary Russell * @since 1.0 */ -public class RabbitTemplate extends RabbitAccessor implements RabbitOperations, MessageListener { +public class RabbitTemplate extends RabbitAccessor implements RabbitOperations, MessageListener, + PublisherCallbackChannel.Listener { private static final String DEFAULT_EXCHANGE = ""; // alias for amq.direct default exchange @@ -111,6 +122,18 @@ public class RabbitTemplate extends RabbitAccessor implements RabbitOperations, private final Map> replyHolder = new ConcurrentHashMap>(); + private volatile ConfirmCallback confirmCallback; + + private volatile ReturnCallback returnCallback; + + private final Map> pendingConfirms = new ConcurrentHashMap>(); + + private volatile boolean mandatory; + + private volatile boolean immediate; + + private final String uuid = UUID.randomUUID().toString(); + public static final String STACKED_CORRELATION_HEADER = "spring_reply_correlation"; public static final String STACKED_REPLY_TO_HEADER = "spring_reply_to"; @@ -239,6 +262,55 @@ public class RabbitTemplate extends RabbitAccessor implements RabbitOperations, return this.messageConverter; } + public void setConfirmCallback(ConfirmCallback confirmCallback) { + Assert.state(this.confirmCallback == null || this.confirmCallback == confirmCallback, + "Only one ConfirmCallback is supported by each RabbitTemplate"); + this.confirmCallback = confirmCallback; + } + + public void setReturnCallback(ReturnCallback returnCallback) { + Assert.state(this.returnCallback == null || this.returnCallback == returnCallback, + "Only one ReturnCallback is supported by each RabbitTemplate"); + this.returnCallback = returnCallback; + } + + public void setMandatory(boolean mandatory) { + this.mandatory = mandatory; + } + + public void setImmediate(boolean immediate) { + this.immediate = immediate; + } + + /** + * Gets unconfirmed correlatiom data older than age and removes them. + * @param age in millseconds + * @return the collection of correlation data for which confirms have + * not been received. + */ + public Collection getUnconfirmed(long age) { + Set unconfirmed = new HashSet(); + synchronized (this.pendingConfirms) { + long threshold = System.currentTimeMillis() - age; + for (Entry> channelPendingConfirmEntry : this.pendingConfirms.entrySet()) { + SortedMap channelPendingConfirms = channelPendingConfirmEntry.getValue(); + Iterator> iterator = channelPendingConfirms.entrySet().iterator(); + PendingConfirm pendingConfirm; + while (iterator.hasNext()) { + pendingConfirm = iterator.next().getValue(); + if (pendingConfirm.getTimestamp() < threshold) { + unconfirmed.add(pendingConfirm.getCorrelationData()); + iterator.remove(); + } + else { + break; + } + } + } + } + return unconfirmed.size() > 0 ? unconfirmed : null; + } + public void send(Message message) throws AmqpException { send(this.exchange, this.routingKey, message); } @@ -248,24 +320,42 @@ public class RabbitTemplate extends RabbitAccessor implements RabbitOperations, } public void send(final String exchange, final String routingKey, final Message message) throws AmqpException { + this.send(exchange, routingKey, message, null); + } + + public void send(final String exchange, final String routingKey, + final Message message, final CorrelationData correlationData) + throws AmqpException { execute(new ChannelCallback() { public Object doInRabbit(Channel channel) throws Exception { - doSend(channel, exchange, routingKey, message); + doSend(channel, exchange, routingKey, message, correlationData); return null; } }); } public void convertAndSend(Object object) throws AmqpException { - convertAndSend(this.exchange, this.routingKey, object); + convertAndSend(this.exchange, this.routingKey, object, (CorrelationData) null); + } + + public void correlationconvertAndSend(Object object, CorrelationData correlationData) throws AmqpException { + convertAndSend(this.exchange, this.routingKey, object, correlationData); } public void convertAndSend(String routingKey, final Object object) throws AmqpException { - convertAndSend(this.exchange, routingKey, object); + convertAndSend(this.exchange, routingKey, object, (CorrelationData) null); + } + + public void convertAndSend(String routingKey, final Object object, CorrelationData correlationData) throws AmqpException { + convertAndSend(this.exchange, routingKey, object, correlationData); } public void convertAndSend(String exchange, String routingKey, final Object object) throws AmqpException { - send(exchange, routingKey, getRequiredMessageConverter().toMessage(object, new MessageProperties())); + convertAndSend(exchange, routingKey, object, (CorrelationData) null); + } + + public void convertAndSend(String exchange, String routingKey, final Object object, CorrelationData corrationData) throws AmqpException { + send(exchange, routingKey, getRequiredMessageConverter().toMessage(object, new MessageProperties()), corrationData); } public void convertAndSend(Object message, MessagePostProcessor messagePostProcessor) throws AmqpException { @@ -274,14 +364,25 @@ public class RabbitTemplate extends RabbitAccessor implements RabbitOperations, public void convertAndSend(String routingKey, Object message, MessagePostProcessor messagePostProcessor) throws AmqpException { - convertAndSend(this.exchange, routingKey, message, messagePostProcessor); + convertAndSend(this.exchange, routingKey, message, messagePostProcessor, null); + } + + public void convertAndSend(String routingKey, Object message, MessagePostProcessor messagePostProcessor, + CorrelationData correlationData) + throws AmqpException { + convertAndSend(this.exchange, routingKey, message, messagePostProcessor, correlationData); } public void convertAndSend(String exchange, String routingKey, final Object message, final MessagePostProcessor messagePostProcessor) throws AmqpException { + convertAndSend(exchange, routingKey, message, messagePostProcessor, null); + } + + public void convertAndSend(String exchange, String routingKey, final Object message, + final MessagePostProcessor messagePostProcessor, CorrelationData correlationData) throws AmqpException { Message messageToSend = getRequiredMessageConverter().toMessage(message, new MessageProperties()); messageToSend = messagePostProcessor.postProcessMessage(messageToSend); - send(exchange, routingKey, messageToSend); + send(exchange, routingKey, messageToSend, correlationData); } public Message receive() throws AmqpException { @@ -425,7 +526,7 @@ public class RabbitTemplate extends RabbitAccessor implements RabbitOperations, } }; channel.basicConsume(replyTo, noAck, consumerTag, noLocal, exclusive, null, consumer); - doSend(channel, exchange, routingKey, message); + doSend(channel, exchange, routingKey, message, null); Message reply = (replyTimeout < 0) ? replyHandoff.take() : replyHandoff.poll(replyTimeout, TimeUnit.MILLISECONDS); channel.basicCancel(consumerTag); @@ -469,7 +570,7 @@ public class RabbitTemplate extends RabbitAccessor implements RabbitOperations, if (logger.isDebugEnabled()) { logger.debug("Sending message with tag " + messageTag); } - doSend(channel, exchange, routingKey, message); + doSend(channel, exchange, routingKey, message, null); Message reply = (replyTimeout < 0) ? replyHandoff.take() : replyHandoff.poll(replyTimeout, TimeUnit.MILLISECONDS); RabbitTemplate.this.replyHolder.remove(messageTag); @@ -483,6 +584,9 @@ public class RabbitTemplate extends RabbitAccessor implements RabbitOperations, Assert.notNull(action, "Callback object must not be null"); RabbitResourceHolder resourceHolder = getTransactionalResourceHolder(); Channel channel = resourceHolder.getChannel(); + if (this.confirmCallback != null || this.returnCallback != null) { + addListener(channel); + } try { if (logger.isDebugEnabled()) { logger.debug("Executing callback on RabbitMQ Channel: " + channel); @@ -507,7 +611,8 @@ public class RabbitTemplate extends RabbitAccessor implements RabbitOperations, * @param message the Message to send * @throws IOException if thrown by RabbitMQ API methods */ - protected void doSend(Channel channel, String exchange, String routingKey, Message message) throws Exception { + protected void doSend(Channel channel, String exchange, String routingKey, Message message, + CorrelationData correlationData) throws Exception { if (logger.isDebugEnabled()) { logger.debug("Publishing message on exchange [" + exchange + "], routingKey = [" + routingKey + "]"); } @@ -521,10 +626,21 @@ public class RabbitTemplate extends RabbitAccessor implements RabbitOperations, // try to send to configured routing key routingKey = this.routingKey; } - - channel.basicPublish(exchange, routingKey, false, false, - this.messagePropertiesConverter.fromMessageProperties(message.getMessageProperties(), encoding), - message.getBody()); + if (this.confirmCallback != null && channel instanceof PublisherCallbackChannel) { + PublisherCallbackChannel publisherCallbackChannel = (PublisherCallbackChannel) channel; + publisherCallbackChannel.addPendingConfirm(this, channel.getNextPublishSeqNo(), + new PendingConfirm(correlationData, System.currentTimeMillis())); + } + boolean mandatory = this.returnCallback == null ? false : this.mandatory; + boolean immediate = this.returnCallback == null ? false : this.immediate; + MessageProperties messageProperties = message.getMessageProperties(); + if (mandatory || immediate) { + messageProperties.getHeaders().put(PublisherCallbackChannel.RETURN_CORRELATION, this.uuid); + } + BasicProperties convertedMessageProperties = this.messagePropertiesConverter + .fromMessageProperties(messageProperties, encoding); + channel.basicPublish(exchange, routingKey, mandatory, immediate, + convertedMessageProperties, message.getBody()); // Check if commit needed if (isChannelLocallyTransacted(channel)) { // Transacted channel created by this template -> commit. @@ -562,6 +678,76 @@ public class RabbitTemplate extends RabbitAccessor implements RabbitOperations, return name; } + private void addListener(Channel channel) { + if (channel instanceof PublisherCallbackChannel) { + PublisherCallbackChannel publisherCallbackChannel = (PublisherCallbackChannel) channel; + SortedMap pendingConfirms = publisherCallbackChannel.addListener(this); + if (!this.pendingConfirms.containsKey(channel)) { + this.pendingConfirms.put(channel, pendingConfirms); + if (logger.isDebugEnabled()) { + logger.debug("Added pending confirms for " + channel + " to map, size now " + this.pendingConfirms.size()); + } + } + } + else { + throw new IllegalStateException("When using publisher confirms, channels must be wrapped in a PublisherCallbackChannelImpl"); + } + } + + public void handleConfirm(PendingConfirm pendingConfirm, boolean ack) { + if (this.confirmCallback != null) { + this.confirmCallback.confirm(pendingConfirm.getCorrelationData(), ack); + } + else { + if (logger.isDebugEnabled()) { + logger.warn("Confirm received but no callback available"); + } + } + } + + public void handleReturn(int replyCode, + String replyText, + String exchange, + String routingKey, + BasicProperties properties, + byte[] body) + throws IOException + { + if (this.returnCallback == null) { + if (logger.isWarnEnabled()) { + logger.warn("Returned message but no callback available"); + } + } + else { + properties.getHeaders().remove(PublisherCallbackChannel.RETURN_CORRELATION); + MessageProperties messageProperties = messagePropertiesConverter.toMessageProperties( + properties, null, this.encoding); + Message returnedMessage = new Message(body, messageProperties); + this.returnCallback.returnedMessage(returnedMessage, + replyCode, replyText, exchange, routingKey); + } + } + + public boolean isConfirmListener() { + return this.confirmCallback != null; + } + + public boolean isReturnListener() { + return this.returnCallback != null; + } + + public void removePendingConfirmsReference(Channel channel, + SortedMap unconfirmed) { + this.pendingConfirms.remove(channel); + if (logger.isDebugEnabled()) { + logger.debug("Removed pending confirms for " + channel + " from map, size now " + this.pendingConfirms.size()); + } + } + + public String getUUID() { + return this.uuid; + } + public void onMessage(Message message) { String messageTag = (String) message.getMessageProperties() .getHeaders().get(STACKED_CORRELATION_HEADER); @@ -637,4 +823,15 @@ public class RabbitTemplate extends RabbitAccessor implements RabbitOperations, return newValue; } } + + public static interface ConfirmCallback { + + void confirm(CorrelationData correlationData, boolean ack); + } + + public static interface ReturnCallback { + + void returnedMessage(Message message, int replyCode, String replyText, + String exchange, String routingKey); + } } diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/CorrelationData.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/CorrelationData.java new file mode 100644 index 00000000..f88517ba --- /dev/null +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/CorrelationData.java @@ -0,0 +1,45 @@ +/* + * Copyright 2002-2012 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.amqp.rabbit.support; + +import org.springframework.amqp.rabbit.core.RabbitTemplate; + +/** + * Base class for correlating publisher confirms to sent messages. + * Use the {@link RabbitTemplate} methods that include one of + * these as a parameter; when the publisher confirm is received, + * the CorrelationData is returned with the ack/nack. + * @author Gary Russell + * @since 1.0.1 + * + */ +public class CorrelationData { + + private String id; + + public CorrelationData(String id) { + this.id = id; + } + + public String getId() { + return id; + } + + @Override + public String toString() { + return "CorrelationData [id=" + id + "]"; + } +} diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/PendingConfirm.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/PendingConfirm.java new file mode 100644 index 00000000..4f87fa8f --- /dev/null +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/PendingConfirm.java @@ -0,0 +1,55 @@ +/* + * Copyright 2002-2012 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.amqp.rabbit.support; + +/** + * Instances of this object track pending publisher confirms. + * The timestamp allows the pending confirmation to be + * expired. It also holds {@link CorrelationData} for + * the client to correlate a confirm with a sent message. + * @author Gary Russell + * @since 1.0.1 + * + */ +public class PendingConfirm { + + private final CorrelationData correlationData; + + private final long timestamp; + + /** + * @param correlationId + * @param timestamp + */ + public PendingConfirm(CorrelationData correlationData, long timestamp) { + this.correlationData = correlationData; + this.timestamp = timestamp; + } + + public CorrelationData getCorrelationData() { + return correlationData; + } + + public long getTimestamp() { + return timestamp; + } + + @Override + public String toString() { + return "PendingConfirm [correlationData=" + correlationData + "]"; + } + +} diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/PublisherCallbackChannel.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/PublisherCallbackChannel.java new file mode 100644 index 00000000..ef47302a --- /dev/null +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/PublisherCallbackChannel.java @@ -0,0 +1,104 @@ +/* + * Copyright 2002-2012 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.amqp.rabbit.support; + +import java.io.IOException; +import java.util.SortedMap; + +import com.rabbitmq.client.AMQP; +import com.rabbitmq.client.Channel; + +/** + * Instances of this interface support a single listener being + * registered for publisher confirms with multiple channels, + * by adding context to the callbacks. + * @author Gary Russell + * @since 1.0.1 + * + */ +public interface PublisherCallbackChannel extends Channel { + + static String RETURN_CORRELATION = "spring_return_correlation"; + + /** + * Adds a {@link Listener} and returns a reference to + * the pending confirms map for that listener's pending + * confirms, allowing the Listener to + * assess unconfirmed sends at any point in time. + * The client must NOT modify the contents of + * this array, and must synchronize on it when + * iterating over its collections. + * @param listener The Listener. + * @return A reference to pending confirms for the listener + */ + SortedMap addListener(Listener listener); + + /** + * Gets a reference to the current listener, or null. + * @return the Listener. + */ + boolean removeListener(Listener listener); + + /** + * Adds a pending confirmation to this channel's map. + * @param seq The key to the map. + * @param pendingConfirm The PendingConfirm object. + */ + void addPendingConfirm(Listener listener, long seq, PendingConfirm pendingConfirm); + + /** + * Listeners implementing this interface can participate + * in publisher confirms received from multiple channels, + * by invoking addListener on each channel. Standard + * AMQP channels do not support a listener being + * registered on multiple channels. + */ + public static interface Listener { + + /** + * Invoked by the channel when a confirm is received. + * @param pendingConfirm The pending confirmation, containing + * correlation data. + * @param ack true when 'ack', false when 'nack'. + */ + void handleConfirm(PendingConfirm pendingConfirm, boolean ack); + + void handleReturn(int replyCode, + String replyText, + String exchange, + String routingKey, + AMQP.BasicProperties properties, + byte[] body) throws IOException; + + /** + * When called, this listener must remove all references to the + * pending confirm map. + * @param unconfirmed The pending confirm map. + */ + void removePendingConfirmsReference(Channel channel, SortedMap unconfirmed); + + /** + * Returns the UUID used to identify this Listener for returns. + * @return A string representation of the UUID. + */ + String getUUID(); + + boolean isConfirmListener(); + + boolean isReturnListener(); + } + +} \ No newline at end of file diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/PublisherCallbackChannelImpl.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/PublisherCallbackChannelImpl.java new file mode 100644 index 00000000..d3c9a2a5 --- /dev/null +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/PublisherCallbackChannelImpl.java @@ -0,0 +1,559 @@ +/* + * Copyright 2002-2012 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.amqp.rabbit.support; + +import java.io.IOException; +import java.util.Collections; +import java.util.Iterator; +import java.util.Map; +import java.util.Map.Entry; +import java.util.SortedMap; +import java.util.TreeMap; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.TimeoutException; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.springframework.util.Assert; + +import com.rabbitmq.client.AMQP; +import com.rabbitmq.client.AMQP.Basic.RecoverOk; +import com.rabbitmq.client.AMQP.BasicProperties; +import com.rabbitmq.client.AMQP.Channel.FlowOk; +import com.rabbitmq.client.AMQP.Exchange.BindOk; +import com.rabbitmq.client.AMQP.Exchange.DeclareOk; +import com.rabbitmq.client.AMQP.Exchange.DeleteOk; +import com.rabbitmq.client.AMQP.Exchange.UnbindOk; +import com.rabbitmq.client.AMQP.Queue.PurgeOk; +import com.rabbitmq.client.AMQP.Tx.CommitOk; +import com.rabbitmq.client.AMQP.Tx.RollbackOk; +import com.rabbitmq.client.AMQP.Tx.SelectOk; +import com.rabbitmq.client.Channel; +import com.rabbitmq.client.Command; +import com.rabbitmq.client.ConfirmListener; +import com.rabbitmq.client.Connection; +import com.rabbitmq.client.Consumer; +import com.rabbitmq.client.FlowListener; +import com.rabbitmq.client.GetResponse; +import com.rabbitmq.client.Method; +import com.rabbitmq.client.ReturnListener; +import com.rabbitmq.client.ShutdownListener; +import com.rabbitmq.client.ShutdownSignalException; + +/** + * Channel wrapper to allow a single listener able to handle + * confirms from multiple channels. + * + * @author Gary Russell + * @since 1.0.1 + * + */ +public class PublisherCallbackChannelImpl implements PublisherCallbackChannel, ConfirmListener, ReturnListener { + + private final Log logger = LogFactory.getLog(this.getClass()); + + private final Channel delegate; + + private final Map listeners = new ConcurrentHashMap(); + + private final Map> pendingConfirms + = new ConcurrentHashMap>(); + + private final Map listenerForSeq = new ConcurrentHashMap(); + + public PublisherCallbackChannelImpl(Channel delegate) { + this.delegate = delegate; + } + +////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////// +// BEGIN PURE DELEGATE METHODS +////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////// + + public void addShutdownListener(ShutdownListener listener) { + this.delegate.addShutdownListener(listener); + } + + public void removeShutdownListener(ShutdownListener listener) { + this.delegate.removeShutdownListener(listener); + } + + public ShutdownSignalException getCloseReason() { + return this.delegate.getCloseReason(); + } + + public void notifyListeners() { + this.delegate.notifyListeners(); + } + + public boolean isOpen() { + return this.delegate.isOpen(); + } + + public int getChannelNumber() { + return this.delegate.getChannelNumber(); + } + + public Connection getConnection() { + return this.delegate.getConnection(); + } + + public void close(int closeCode, String closeMessage) throws IOException { + this.delegate.close(closeCode, closeMessage); + } + + public FlowOk flow(boolean active) throws IOException { + return this.delegate.flow(active); + } + + public FlowOk getFlow() { + return this.delegate.getFlow(); + } + + public void abort() throws IOException { + this.delegate.abort(); + } + + public void abort(int closeCode, String closeMessage) throws IOException { + this.delegate.abort(closeCode, closeMessage); + } + + public void addFlowListener(FlowListener listener) { + this.delegate.addFlowListener(listener); + } + + public boolean removeFlowListener(FlowListener listener) { + return this.delegate.removeFlowListener(listener); + } + + public void clearFlowListeners() { + this.delegate.clearFlowListeners(); + } + + public Consumer getDefaultConsumer() { + return this.delegate.getDefaultConsumer(); + } + + public void setDefaultConsumer(Consumer consumer) { + this.delegate.setDefaultConsumer(consumer); + } + + public void basicQos(int prefetchSize, int prefetchCount, boolean global) + throws IOException { + this.delegate.basicQos(prefetchSize, prefetchCount, global); + } + + public void basicQos(int prefetchCount) throws IOException { + this.delegate.basicQos(prefetchCount); + } + + public void basicPublish(String exchange, String routingKey, + BasicProperties props, byte[] body) throws IOException { + this.delegate.basicPublish(exchange, routingKey, props, body); + } + + public void basicPublish(String exchange, String routingKey, + boolean mandatory, boolean immediate, BasicProperties props, + byte[] body) throws IOException { + this.delegate.basicPublish(exchange, routingKey, mandatory, immediate, + props, body); + } + + public DeclareOk exchangeDeclare(String exchange, String type) + throws IOException { + return this.delegate.exchangeDeclare(exchange, type); + } + + public DeclareOk exchangeDeclare(String exchange, String type, + boolean durable) throws IOException { + return this.delegate.exchangeDeclare(exchange, type, durable); + } + + public DeclareOk exchangeDeclare(String exchange, String type, + boolean durable, boolean autoDelete, Map arguments) + throws IOException { + return this.delegate.exchangeDeclare(exchange, type, durable, autoDelete, + arguments); + } + + public DeclareOk exchangeDeclare(String exchange, String type, + boolean durable, boolean autoDelete, boolean internal, + Map arguments) throws IOException { + return this.delegate.exchangeDeclare(exchange, type, durable, autoDelete, + internal, arguments); + } + + public DeclareOk exchangeDeclarePassive(String name) throws IOException { + return this.delegate.exchangeDeclarePassive(name); + } + + public DeleteOk exchangeDelete(String exchange, boolean ifUnused) + throws IOException { + return this.delegate.exchangeDelete(exchange, ifUnused); + } + + public DeleteOk exchangeDelete(String exchange) throws IOException { + return this.delegate.exchangeDelete(exchange); + } + + public BindOk exchangeBind(String destination, String source, + String routingKey) throws IOException { + return this.delegate.exchangeBind(destination, source, routingKey); + } + + public BindOk exchangeBind(String destination, String source, + String routingKey, Map arguments) + throws IOException { + return this.delegate + .exchangeBind(destination, source, routingKey, arguments); + } + + public UnbindOk exchangeUnbind(String destination, String source, + String routingKey) throws IOException { + return this.delegate.exchangeUnbind(destination, source, routingKey); + } + + public UnbindOk exchangeUnbind(String destination, String source, + String routingKey, Map arguments) + throws IOException { + return this.delegate.exchangeUnbind(destination, source, routingKey, + arguments); + } + + public com.rabbitmq.client.AMQP.Queue.DeclareOk queueDeclare() + throws IOException { + return this.delegate.queueDeclare(); + } + + public com.rabbitmq.client.AMQP.Queue.DeclareOk queueDeclare(String queue, + boolean durable, boolean exclusive, boolean autoDelete, + Map arguments) throws IOException { + return this.delegate.queueDeclare(queue, durable, exclusive, autoDelete, + arguments); + } + + public com.rabbitmq.client.AMQP.Queue.DeclareOk queueDeclarePassive( + String queue) throws IOException { + return this.delegate.queueDeclarePassive(queue); + } + + public com.rabbitmq.client.AMQP.Queue.DeleteOk queueDelete(String queue) + throws IOException { + return this.delegate.queueDelete(queue); + } + + public com.rabbitmq.client.AMQP.Queue.DeleteOk queueDelete(String queue, + boolean ifUnused, boolean ifEmpty) throws IOException { + return this.delegate.queueDelete(queue, ifUnused, ifEmpty); + } + + public com.rabbitmq.client.AMQP.Queue.BindOk queueBind(String queue, + String exchange, String routingKey) throws IOException { + return this.delegate.queueBind(queue, exchange, routingKey); + } + + public com.rabbitmq.client.AMQP.Queue.BindOk queueBind(String queue, + String exchange, String routingKey, Map arguments) + throws IOException { + return this.delegate.queueBind(queue, exchange, routingKey, arguments); + } + + public com.rabbitmq.client.AMQP.Queue.UnbindOk queueUnbind(String queue, + String exchange, String routingKey) throws IOException { + return this.delegate.queueUnbind(queue, exchange, routingKey); + } + + public com.rabbitmq.client.AMQP.Queue.UnbindOk queueUnbind(String queue, + String exchange, String routingKey, Map arguments) + throws IOException { + return this.delegate.queueUnbind(queue, exchange, routingKey, arguments); + } + + public PurgeOk queuePurge(String queue) throws IOException { + return this.delegate.queuePurge(queue); + } + + public GetResponse basicGet(String queue, boolean autoAck) + throws IOException { + return this.delegate.basicGet(queue, autoAck); + } + + public void basicAck(long deliveryTag, boolean multiple) throws IOException { + this.delegate.basicAck(deliveryTag, multiple); + } + + public void basicNack(long deliveryTag, boolean multiple, boolean requeue) + throws IOException { + this.delegate.basicNack(deliveryTag, multiple, requeue); + } + + public void basicReject(long deliveryTag, boolean requeue) + throws IOException { + this.delegate.basicReject(deliveryTag, requeue); + } + + public String basicConsume(String queue, Consumer callback) + throws IOException { + return this.delegate.basicConsume(queue, callback); + } + + public String basicConsume(String queue, boolean autoAck, Consumer callback) + throws IOException { + return this.delegate.basicConsume(queue, autoAck, callback); + } + + public String basicConsume(String queue, boolean autoAck, + String consumerTag, Consumer callback) throws IOException { + return this.delegate.basicConsume(queue, autoAck, consumerTag, callback); + } + + public String basicConsume(String queue, boolean autoAck, + String consumerTag, boolean noLocal, boolean exclusive, + Map arguments, Consumer callback) + throws IOException { + return this.delegate.basicConsume(queue, autoAck, consumerTag, noLocal, + exclusive, arguments, callback); + } + + public void basicCancel(String consumerTag) throws IOException { + this.delegate.basicCancel(consumerTag); + } + + public RecoverOk basicRecover() throws IOException { + return this.delegate.basicRecover(); + } + + public RecoverOk basicRecover(boolean requeue) throws IOException { + return this.delegate.basicRecover(requeue); + } + + @SuppressWarnings("deprecation") + public void basicRecoverAsync(boolean requeue) throws IOException { + this.delegate.basicRecoverAsync(requeue); + } + + public SelectOk txSelect() throws IOException { + return this.delegate.txSelect(); + } + + public CommitOk txCommit() throws IOException { + return this.delegate.txCommit(); + } + + public RollbackOk txRollback() throws IOException { + return this.delegate.txRollback(); + } + + public com.rabbitmq.client.AMQP.Confirm.SelectOk confirmSelect() + throws IOException { + return this.delegate.confirmSelect(); + } + + public long getNextPublishSeqNo() { + return this.delegate.getNextPublishSeqNo(); + } + + public boolean waitForConfirms() throws InterruptedException { + return this.delegate.waitForConfirms(); + } + + public boolean waitForConfirms(long timeout) throws InterruptedException, + TimeoutException { + return this.delegate.waitForConfirms(timeout); + } + + public void waitForConfirmsOrDie() throws IOException, InterruptedException { + this.delegate.waitForConfirmsOrDie(); + } + + public void waitForConfirmsOrDie(long timeout) throws IOException, + InterruptedException, TimeoutException { + this.delegate.waitForConfirmsOrDie(timeout); + } + + public void asyncRpc(Method method) throws IOException { + this.delegate.asyncRpc(method); + } + + public Command rpc(Method method) throws IOException { + return this.delegate.rpc(method); + } + + public void addConfirmListener(ConfirmListener listener) { + this.delegate.addConfirmListener(listener); + } + + public boolean removeConfirmListener(ConfirmListener listener) { + return this.delegate.removeConfirmListener(listener); + } + + public void clearConfirmListeners() { + this.delegate.clearConfirmListeners(); + } + + public void addReturnListener(ReturnListener listener) { + this.delegate.addReturnListener(listener); + } + + public boolean removeReturnListener(ReturnListener listener) { + return this.delegate.removeReturnListener(listener); + } + + public synchronized void clearReturnListeners() { + this.delegate.clearReturnListeners(); + } + +////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////// +// END PURE DELEGATE METHODS +////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////// + + public void close() throws IOException { + this.delegate.close(); + for (Entry> entry : this.pendingConfirms.entrySet()) { + Listener listener = entry.getKey(); + listener.removePendingConfirmsReference(this, entry.getValue()); + } + this.pendingConfirms.clear(); + this.listenerForSeq.clear(); + } + + public synchronized SortedMap addListener(Listener listener) { + Assert.notNull(listener, "Listener cannot be null"); + if (this.listeners.size() == 0) { + this.delegate.addConfirmListener(this); + this.delegate.addReturnListener(this); + } + if (!this.listeners.values().contains(listener)){ + this.listeners.put(listener.getUUID(), listener); + this.pendingConfirms.put(listener, Collections.synchronizedSortedMap(new TreeMap())); + if (logger.isDebugEnabled()) { + logger.debug("Added listener " + listener); + } + } + return this.pendingConfirms.get(listener); + } + + public synchronized boolean removeListener(Listener listener) { + Listener mappedListener = this.listeners.remove(listener.getUUID()); + boolean result = mappedListener != null; + if (result && this.listeners.size() == 0) { + this.delegate.removeConfirmListener(this); + this.delegate.removeReturnListener(this); + } + Iterator> iterator = this.listenerForSeq.entrySet().iterator(); + while (iterator.hasNext()) { + Entry entry = iterator.next(); + if (entry.getValue() == listener) { + iterator.remove(); + } + } + this.pendingConfirms.remove(listener); + return result; + } + + +// ConfirmListener + + public void handleAck(long seq, boolean multiple) + throws IOException { + if (logger.isDebugEnabled()) { + logger.debug(this.toString() + " PC:Ack:" + seq + ":" + multiple); + } + this.processAck(seq, true, multiple); + } + + public void handleNack(long seq, boolean multiple) + throws IOException { + if (logger.isDebugEnabled()) { + logger.debug(this.toString() + " PC:Nack:" + seq + ":" + multiple); + } + this.processAck(seq, false, multiple); + } + + private void processAck(long seq, boolean ack, boolean multiple) { + Listener listener = this.listenerForSeq.get(seq); + if (listener != null && listener.isConfirmListener()) { + if (multiple) { + Map headMap = this.pendingConfirms.get(listener).headMap(seq + 1); + synchronized(this.pendingConfirms) { + Iterator> iterator = headMap.entrySet().iterator(); + while (iterator.hasNext()) { + Entry entry = iterator.next(); + iterator.remove(); + listener.handleConfirm(entry.getValue(), ack); + } + } + } + else { + PendingConfirm pendingConfirm = this.pendingConfirms.get(listener).remove(seq); + if (pendingConfirm != null) { + listener.handleConfirm(pendingConfirm, ack); + } + } + } else { + logger.error("No listener for seq:" + seq); + } + } + + public void addPendingConfirm(Listener listener, long seq, PendingConfirm pendingConfirm) { + SortedMap pendingConfirmsForListener = this.pendingConfirms.get(listener); + Assert.notNull(pendingConfirmsForListener, "Listener not registered"); + pendingConfirmsForListener.put(seq, pendingConfirm); + this.listenerForSeq.put(seq, listener); + } + +// ReturnListener + + public void handleReturn(int replyCode, + String replyText, + String exchange, + String routingKey, + AMQP.BasicProperties properties, + byte[] body) throws IOException + { + Object uuidObject = properties.getHeaders().get(RETURN_CORRELATION).toString(); + Listener listener = this.listeners.get(uuidObject); + if (listener == null || !listener.isReturnListener()) { + if (logger.isWarnEnabled()) { + logger.warn("No Listener for returned message"); + } + } + else { + listener.handleReturn(replyCode, replyText, exchange, routingKey, properties, body); + } + } + +// Object + + @Override + public int hashCode() { + return this.delegate.hashCode(); + } + + + @Override + public boolean equals(Object obj) { + if (obj == this) { + return true; + } + return this.delegate.equals(obj); + } + + @Override + public String toString() { + return "PublisherCallbackChannelImpl: " + this.delegate.toString(); + } + +} diff --git a/spring-rabbit/src/main/resources/org/springframework/amqp/rabbit/config/spring-rabbit-1.0.xsd b/spring-rabbit/src/main/resources/org/springframework/amqp/rabbit/config/spring-rabbit-1.0.xsd index 664e1116..0b083f55 100644 --- a/spring-rabbit/src/main/resources/org/springframework/amqp/rabbit/config/spring-rabbit-1.0.xsd +++ b/spring-rabbit/src/main/resources/org/springframework/amqp/rabbit/config/spring-rabbit-1.0.xsd @@ -746,6 +746,50 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + @@ -839,6 +883,20 @@ + + + + + + + + + + diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/TemplateParserTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/TemplateParserTests.java index 714d0558..60d946f2 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/TemplateParserTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/TemplateParserTests.java @@ -13,7 +13,9 @@ package org.springframework.amqp.rabbit.config; +import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; import static org.junit.Assert.assertSame; import static org.junit.Assert.assertTrue; @@ -47,6 +49,22 @@ public final class TemplateParserTests { public void testTemplate() throws Exception { AmqpTemplate template = beanFactory.getBean("template", AmqpTemplate.class); assertNotNull(template); + DirectFieldAccessor dfa = new DirectFieldAccessor(template); + assertEquals(Boolean.FALSE, dfa.getPropertyValue("mandatory")); + assertEquals(Boolean.FALSE, dfa.getPropertyValue("immediate")); + assertNull(dfa.getPropertyValue("returnCallback")); + assertNull(dfa.getPropertyValue("confirmCallback")); + } + + @Test + public void testTemplateWithCallbacks() throws Exception { + AmqpTemplate template = beanFactory.getBean("withCallbacks", AmqpTemplate.class); + assertNotNull(template); + DirectFieldAccessor dfa = new DirectFieldAccessor(template); + assertEquals(Boolean.TRUE, dfa.getPropertyValue("mandatory")); + assertEquals(Boolean.TRUE, dfa.getPropertyValue("immediate")); + assertNotNull(dfa.getPropertyValue("returnCallback")); + assertNotNull(dfa.getPropertyValue("confirmCallback")); } @Test diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateHeaderTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateHeaderTests.java index f304c7be..f3543586 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateHeaderTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateHeaderTests.java @@ -111,6 +111,7 @@ public class RabbitTemplateHeaderTests { Mockito.any(String.class), Mockito.anyBoolean(), Mockito.anyBoolean(), Mockito.any(BasicProperties.class), Mockito.any(byte[].class)); Message reply = template.sendAndReceive(message); + assertNotNull(reply); assertEquals(1, props.size()); BasicProperties basicProperties = props.get(0); diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests.java new file mode 100644 index 00000000..bc28c161 --- /dev/null +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests.java @@ -0,0 +1,351 @@ +/* + * Copyright 2010-2012 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.amqp.rabbit.core; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.HashSet; +import java.util.Iterator; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; + +import org.junit.Before; +import org.junit.Rule; +import org.junit.Test; +import org.mockito.invocation.InvocationOnMock; +import org.mockito.stubbing.Answer; +import org.springframework.amqp.core.Message; +import org.springframework.amqp.core.MessageProperties; +import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; +import org.springframework.amqp.rabbit.connection.SingleConnectionFactory; +import org.springframework.amqp.rabbit.core.RabbitTemplate.ConfirmCallback; +import org.springframework.amqp.rabbit.core.RabbitTemplate.ReturnCallback; +import org.springframework.amqp.rabbit.support.CorrelationData; +import org.springframework.amqp.rabbit.support.PublisherCallbackChannelImpl; +import org.springframework.amqp.rabbit.test.BrokerRunning; +import org.springframework.amqp.rabbit.test.BrokerTestUtils; +import org.springframework.amqp.support.converter.SimpleMessageConverter; +import org.springframework.beans.DirectFieldAccessor; + +import com.rabbitmq.client.Channel; +import com.rabbitmq.client.Connection; +import com.rabbitmq.client.ConnectionFactory; + +public class RabbitTemplatePublisherCallbacksIntegrationTests { + + private static final String ROUTE = "test.queue"; + + private CachingConnectionFactory connectionFactory; + + private RabbitTemplate template; + + @Before + public void create() { + connectionFactory = new CachingConnectionFactory(); + // When using publisher confirms, the cache size needs to be large enough + // otherwise channels can be closed before confirms are received. + connectionFactory.setChannelCacheSize(10); + connectionFactory.setPort(BrokerTestUtils.getPort()); + connectionFactory.setPublisherConfirms(true); + template = new RabbitTemplate(connectionFactory); + } + + @Rule + public BrokerRunning brokerIsRunning = BrokerRunning.isRunningWithEmptyQueues(ROUTE); + + @Test + public void testPublisherConfirmReceived() throws Exception { + final CountDownLatch latch = new CountDownLatch(1); + template.setConfirmCallback(new ConfirmCallback() { + + public void confirm(CorrelationData correlationData, boolean ack) { + latch.countDown(); + } + }); + template.convertAndSend(ROUTE, (Object) "message", new CorrelationData("abc")); + assertTrue(latch.await(1000, TimeUnit.MILLISECONDS)); + assertNull(template.getUnconfirmed(0)); + } + + @Test + public void testPublisherConfirmReceivedConcurrentThreads() throws Exception { + final CountDownLatch latch = new CountDownLatch(2); + template.setConfirmCallback(new ConfirmCallback() { + + public void confirm(CorrelationData correlationData, boolean ack) { + latch.countDown(); + } + }); + + // Hold up the first thread so we get two channels + final CountDownLatch threadLatch = new CountDownLatch(1); + //Thread 1 + Executors.newSingleThreadExecutor().execute(new Runnable() { + + public void run() { + template.execute(new ChannelCallback() { + public Object doInRabbit(Channel channel) throws Exception { + try { + threadLatch.await(10, TimeUnit.SECONDS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + template.doSend(channel, "", ROUTE, + new SimpleMessageConverter().toMessage("message", new MessageProperties()), + new CorrelationData("def")); + return null; + } + }); + } + }); + + // Thread 2 + template.convertAndSend(ROUTE, (Object) "message", new CorrelationData("abc")); + threadLatch.countDown(); + assertTrue(latch.await(5000, TimeUnit.MILLISECONDS)); + assertNull(template.getUnconfirmed(0)); + } + + @Test + public void testPublisherConfirmReceivedTwoTemplates() throws Exception { + final CountDownLatch latch1 = new CountDownLatch(1); + final CountDownLatch latch2 = new CountDownLatch(1); + template.setConfirmCallback(new ConfirmCallback() { + + public void confirm(CorrelationData correlationData, boolean ack) { + latch1.countDown(); + } + }); + template.convertAndSend(ROUTE, (Object) "message", new CorrelationData("abc")); + RabbitTemplate secondTemplate = new RabbitTemplate(connectionFactory); + secondTemplate.setConfirmCallback(new ConfirmCallback() { + + public void confirm(CorrelationData correlationData, boolean ack) { + latch2.countDown(); + } + }); + secondTemplate.convertAndSend(ROUTE, (Object) "message", new CorrelationData("def")); + assertTrue(latch1.await(1000, TimeUnit.MILLISECONDS)); + assertTrue(latch2.await(1000, TimeUnit.MILLISECONDS)); + assertNull(template.getUnconfirmed(0)); + assertNull(secondTemplate.getUnconfirmed(0)); + } + + @Test + public void testPublisherReturns() throws Exception { + final CountDownLatch latch = new CountDownLatch(1); + final List returns = new ArrayList(); + template.setReturnCallback(new ReturnCallback() { + public void returnedMessage(Message message, int replyCode, + String replyText, String exchange, String routingKey) { + returns.add(message); + latch.countDown(); + } + }); + template.setMandatory(true); + template.setImmediate(true); + template.convertAndSend(ROUTE + "junk", (Object) "message", new CorrelationData("abc")); + assertTrue(latch.await(1000, TimeUnit.MILLISECONDS)); + assertEquals(1, returns.size()); + Message message = returns.get(0); + assertEquals("message", new String(message.getBody(), "utf-8")); + } + + @Test + public void testPublisherConfirmNotReceived() throws Exception { + ConnectionFactory mockConnectionFactory = mock(ConnectionFactory.class); + Connection mockConnection = mock(Connection.class); + Channel mockChannel = mock(Channel.class); + + when(mockConnectionFactory.newConnection()).thenReturn(mockConnection); + when(mockConnection.isOpen()).thenReturn(true); + when(mockConnection.createChannel()).thenReturn(new PublisherCallbackChannelImpl(mockChannel)); + + final RabbitTemplate template = new RabbitTemplate(new SingleConnectionFactory(mockConnectionFactory)); + + final AtomicBoolean confirmed = new AtomicBoolean(); + template.setConfirmCallback(new ConfirmCallback() { + + public void confirm(CorrelationData correlationData, boolean ack) { + confirmed.set(true); + } + }); + template.convertAndSend(ROUTE, (Object) "message", new CorrelationData("abc")); + Thread.sleep(5); + Collection unconfirmed = template.getUnconfirmed(0); + assertEquals(1, unconfirmed.size()); + assertEquals("abc", unconfirmed.iterator().next().getId()); + assertFalse(confirmed.get()); + } + + @Test + public void testPublisherConfirmNotReceivedMultiThreads() throws Exception { + ConnectionFactory mockConnectionFactory = mock(ConnectionFactory.class); + Connection mockConnection = mock(Connection.class); + Channel mockChannel = mock(Channel.class); + + when(mockConnectionFactory.newConnection()).thenReturn(mockConnection); + when(mockConnection.isOpen()).thenReturn(true); + PublisherCallbackChannelImpl channel1 = new PublisherCallbackChannelImpl(mockChannel); + PublisherCallbackChannelImpl channel2 = new PublisherCallbackChannelImpl(mockChannel); + when(mockConnection.createChannel()).thenReturn(channel1).thenReturn(channel2); + + final RabbitTemplate template = new RabbitTemplate(new SingleConnectionFactory(mockConnectionFactory)); + + final AtomicBoolean confirmed = new AtomicBoolean(); + template.setConfirmCallback(new ConfirmCallback() { + + public void confirm(CorrelationData correlationData, boolean ack) { + confirmed.set(true); + } + }); + + // Hold up the first thread so we get two channels + final CountDownLatch threadLatch = new CountDownLatch(1); + final CountDownLatch threadSentLatch = new CountDownLatch(1); + //Thread 1 + Executors.newSingleThreadExecutor().execute(new Runnable() { + + public void run() { + template.execute(new ChannelCallback() { + public Object doInRabbit(Channel channel) throws Exception { + try { + threadLatch.await(10, TimeUnit.SECONDS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + template.doSend(channel, "", ROUTE, + new SimpleMessageConverter().toMessage("message", new MessageProperties()), + new CorrelationData("def")); + threadSentLatch.countDown(); + return null; + } + }); + } + }); + + // Thread 2 + template.convertAndSend(ROUTE, (Object) "message", new CorrelationData("abc")); + threadLatch.countDown(); + assertTrue(threadSentLatch.await(5, TimeUnit.SECONDS)); + Thread.sleep(5); + Collection unconfirmed = template.getUnconfirmed(0); + assertEquals(2, unconfirmed.size()); + Set ids = new HashSet(); + Iterator iterator = unconfirmed.iterator(); + ids.add(iterator.next().getId()); + ids.add(iterator.next().getId()); + assertTrue(ids.remove("abc")); + assertTrue(ids.remove("def")); + assertFalse(confirmed.get()); + DirectFieldAccessor dfa = new DirectFieldAccessor(template); + Map pendingConfirms = (Map) dfa.getPropertyValue("pendingConfirms"); + assertEquals(2, pendingConfirms.size()); + channel1.close(); + assertEquals(1, pendingConfirms.size()); + channel2.close(); + assertEquals(0, pendingConfirms.size()); + } + + @Test + public void testPublisherConfirmNotReceivedAged() throws Exception { + ConnectionFactory mockConnectionFactory = mock(ConnectionFactory.class); + Connection mockConnection = mock(Connection.class); + Channel mockChannel = mock(Channel.class); + + when(mockConnectionFactory.newConnection()).thenReturn(mockConnection); + when(mockConnection.isOpen()).thenReturn(true); + when(mockConnection.createChannel()).thenReturn(new PublisherCallbackChannelImpl(mockChannel)); + + final AtomicInteger count = new AtomicInteger(); + doAnswer(new Answer(){ + public Object answer(InvocationOnMock invocation) throws Throwable { + return count.incrementAndGet(); + }}).when(mockChannel).getNextPublishSeqNo(); + + final RabbitTemplate template = new RabbitTemplate(new SingleConnectionFactory(mockConnectionFactory)); + + final AtomicBoolean confirmed = new AtomicBoolean(); + template.setConfirmCallback(new ConfirmCallback() { + + public void confirm(CorrelationData correlationData, boolean ack) { + confirmed.set(true); + } + }); + template.convertAndSend(ROUTE, (Object) "message", new CorrelationData("abc")); + Thread.sleep(100); + template.convertAndSend(ROUTE, (Object) "message", new CorrelationData("def")); + Collection unconfirmed = template.getUnconfirmed(50); + assertEquals(1, unconfirmed.size()); + assertEquals("abc", unconfirmed.iterator().next().getId()); + assertFalse(confirmed.get()); + Thread.sleep(100); + unconfirmed = template.getUnconfirmed(50); + assertEquals(1, unconfirmed.size()); + assertEquals("def", unconfirmed.iterator().next().getId()); + assertFalse(confirmed.get()); + } + + @Test + public void testPublisherConfirmMultiple() throws Exception { + ConnectionFactory mockConnectionFactory = mock(ConnectionFactory.class); + Connection mockConnection = mock(Connection.class); + Channel mockChannel = mock(Channel.class); + + when(mockConnectionFactory.newConnection()).thenReturn(mockConnection); + when(mockConnection.isOpen()).thenReturn(true); + PublisherCallbackChannelImpl callbackChannel = new PublisherCallbackChannelImpl(mockChannel); + when(mockConnection.createChannel()).thenReturn(callbackChannel); + + final AtomicInteger count = new AtomicInteger(); + doAnswer(new Answer(){ + public Object answer(InvocationOnMock invocation) throws Throwable { + return count.incrementAndGet(); + }}).when(mockChannel).getNextPublishSeqNo(); + + final RabbitTemplate template = new RabbitTemplate(new SingleConnectionFactory(mockConnectionFactory)); + + final List confirms = new ArrayList(); + final CountDownLatch latch = new CountDownLatch(2); + template.setConfirmCallback(new ConfirmCallback() { + + public void confirm(CorrelationData correlationData, boolean ack) { + if (ack) { + confirms.add(correlationData.getId()); + latch.countDown(); + } + } + }); + template.convertAndSend(ROUTE, (Object) "message", new CorrelationData("abc")); + template.convertAndSend(ROUTE, (Object) "message", new CorrelationData("def")); + callbackChannel.handleAck(2, true); + assertTrue(latch.await(1000, TimeUnit.MILLISECONDS)); + Collection unconfirmed = template.getUnconfirmed(0); + assertNull(unconfirmed); + } +} diff --git a/spring-rabbit/src/test/resources/org/springframework/amqp/rabbit/config/TemplateParserTests-context.xml b/spring-rabbit/src/test/resources/org/springframework/amqp/rabbit/config/TemplateParserTests-context.xml index 520f18ba..0a4b1a1b 100644 --- a/spring-rabbit/src/test/resources/org/springframework/amqp/rabbit/config/TemplateParserTests-context.xml +++ b/spring-rabbit/src/test/resources/org/springframework/amqp/rabbit/config/TemplateParserTests-context.xml @@ -23,4 +23,14 @@ + + + + + + + + +