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 @@ + + + + + + + + +