From 34b8e275d9bf816ee35534981a3b5afbf905651a Mon Sep 17 00:00:00 2001 From: Keith Wall Date: Mon, 28 Nov 2011 09:19:15 +0000 Subject: QPID-3642,QPID-3643: Add Dead Letter Queue functionality for 0-8/0-9/0-9-1 paths, fixes isBound methods on FanoutExchange Applied patch from Keith Wall , Andrew MacBean and Oleksandr Rudyy git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@1207029 13f79535-47bb-0310-9956-ffa450edef68 --- .../java/org/apache/qpid/client/AMQConnection.java | 6 +- .../org/apache/qpid/client/AMQDestination.java | 28 +++++++++- .../java/org/apache/qpid/client/AMQSession.java | 15 +++-- .../org/apache/qpid/client/AMQSession_0_8.java | 64 ++++++++++++++++++++-- .../apache/qpid/client/BasicMessageConsumer.java | 27 +-------- .../qpid/client/BasicMessageConsumer_0_8.java | 29 ++++++++++ .../org/apache/qpid/client/RejectBehaviour.java | 32 +++++++++++ .../java/org/apache/qpid/jms/ConnectionURL.java | 11 +++- 8 files changed, 171 insertions(+), 41 deletions(-) create mode 100644 qpid/java/client/src/main/java/org/apache/qpid/client/RejectBehaviour.java (limited to 'qpid/java/client/src/main') diff --git a/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnection.java b/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnection.java index ad7885f195..6879fe0cfd 100644 --- a/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnection.java +++ b/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnection.java @@ -284,7 +284,10 @@ public class AMQConnection extends Closeable implements Connection, QueueConnect } String amqpVersion = System.getProperty((ClientProperties.AMQP_VERSION), "0-10"); - _logger.debug("AMQP version " + amqpVersion); + if (_logger.isDebugEnabled()) + { + _logger.debug("AMQP version " + amqpVersion); + } _failoverPolicy = new FailoverPolicy(connectionURL, this); BrokerDetails brokerDetails = _failoverPolicy.getCurrentBrokerDetails(); @@ -1485,4 +1488,5 @@ public class AMQConnection extends Closeable implements Connection, QueueConnect { return _lastFailoverTime; } + } diff --git a/qpid/java/client/src/main/java/org/apache/qpid/client/AMQDestination.java b/qpid/java/client/src/main/java/org/apache/qpid/client/AMQDestination.java index f9a38138ba..1df809c67c 100644 --- a/qpid/java/client/src/main/java/org/apache/qpid/client/AMQDestination.java +++ b/qpid/java/client/src/main/java/org/apache/qpid/client/AMQDestination.java @@ -75,6 +75,8 @@ public abstract class AMQDestination implements Destination, Referenceable private boolean _exchangeExistsChecked; + private RejectBehaviour _rejectBehaviour; + public static final int QUEUE_TYPE = 1; public static final int TOPIC_TYPE = 2; public static final int UNKNOWN_TYPE = 3; @@ -227,6 +229,8 @@ public abstract class AMQDestination implements Destination, Referenceable _queueName = binding.getQueueName() == null ? null : binding.getQueueName(); _routingKey = binding.getRoutingKey() == null ? null : binding.getRoutingKey(); _bindingKeys = binding.getBindingKeys() == null || binding.getBindingKeys().length == 0 ? new AMQShortString[0] : binding.getBindingKeys(); + final String rejectBehaviourValue = binding.getOption(BindingURL.OPTION_REJECT_BEHAVIOUR); + _rejectBehaviour = rejectBehaviourValue == null ? null : RejectBehaviour.valueOf(rejectBehaviourValue.toUpperCase()); } protected AMQDestination(AMQShortString exchangeName, AMQShortString exchangeClass, AMQShortString routingKey, AMQShortString queueName) @@ -294,7 +298,7 @@ public abstract class AMQDestination implements Destination, Referenceable _bindingKeys = bindingKeys == null || bindingKeys.length == 0 ? new AMQShortString[0] : bindingKeys; _destSyntax = DestSyntax.BURL; _browseOnly = browseOnly; - + _rejectBehaviour = null; if (_logger.isDebugEnabled()) { _logger.debug("Based on " + toString() + " the selected destination syntax is " + _destSyntax); @@ -499,6 +503,13 @@ public abstract class AMQDestination implements Destination, Referenceable sb.append(URLHelper.DEFAULT_OPTION_SEPERATOR); } + if (_rejectBehaviour != null) + { + sb.append(BindingURL.OPTION_REJECT_BEHAVIOUR); + sb.append("='" + _rejectBehaviour + "'"); + sb.append(URLHelper.DEFAULT_OPTION_SEPERATOR); + } + //removeKey the last char '?' if there is no options , ',' if there are. sb.deleteCharAt(sb.length() - 1); url = sb.toString(); @@ -842,4 +853,19 @@ public abstract class AMQDestination implements Destination, Referenceable { return _addressResolved.get() > time; } + + /** + * This option is only applicable for 0-8/0-9/0-9-1 protocols connection + *

+ * It tells the client to delegate the requeue/DLQ decision to the + * server .If this option is not specified, the messages won't be moved to + * the DLQ (or dropped) when delivery count exceeds the maximum. + * + * @return destination reject behaviour + */ + public RejectBehaviour getRejectBehaviour() + { + return _rejectBehaviour; + } + } diff --git a/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession.java b/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession.java index ef44221ec1..8984b7ca8c 100644 --- a/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession.java +++ b/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession.java @@ -310,7 +310,10 @@ public abstract class AMQSession _prefetchedMessageTags = new ConcurrentLinkedQueue(); + /** All the not yet acknowledged message tags */ protected ConcurrentLinkedQueue _unacknowledgedMessageTags = new ConcurrentLinkedQueue(); @@ -2925,11 +2928,6 @@ public abstract class AMQSession tags = consumer.drainReceiverQueueAndRetrieveDeliveryTags(); - _unacknowledgedMessageTags.addAll(tags); + _prefetchedMessageTags.addAll(tags); } setConnectionStopped(isStopped); @@ -3345,7 +3343,7 @@ public abstract class AMQSession consumersToCheck = new ArrayList(_consumers.values()); + boolean messageListenerFound = false; + boolean serverRejectBehaviourFound = false; + for(BasicMessageConsumer_0_8 consumer : consumersToCheck) + { + if (consumer.isMessageListenerSet()) + { + messageListenerFound = true; + } + if (RejectBehaviour.SERVER.equals(consumer.getRejectBehaviour())) + { + serverRejectBehaviourFound = true; + } + } + _logger.debug("about to pre-reject messages for " + consumersToCheck.size() + " consumer(s)"); + + if (serverRejectBehaviourFound) + { + //reject(false) any messages we don't want returned again + switch(_acknowledgeMode) + { + case Session.DUPS_OK_ACKNOWLEDGE: + case Session.AUTO_ACKNOWLEDGE: + if (!messageListenerFound) + { + break; + } + case Session.CLIENT_ACKNOWLEDGE: + for(Long tag : _unacknowledgedMessageTags) + { + rejectMessage(tag, false); + } + break; + } + } + } + public void releaseForRollback() { // Reject all the messages that have been received in this session and @@ -267,6 +311,17 @@ public final class AMQSession_0_8 extends AMQSession extends Closeable implements Messa private List _closedStack = null; - protected BasicMessageConsumer(int channelId, AMQConnection connection, AMQDestination destination, String messageSelector, boolean noLocal, MessageFactoryRegistry messageFactory, AMQSession session, AMQProtocolHandler protocolHandler, @@ -211,6 +210,7 @@ public abstract class BasicMessageConsumer extends Closeable implements Messa ft.put(AMQPFilterTypes.JMS_SELECTOR.getValue(), messageSelector == null ? "" : messageSelector); _arguments = ft; + } public AMQDestination getDestination() @@ -814,31 +814,6 @@ public abstract class BasicMessageConsumer extends Closeable implements Messa } } - - /** - * Acknowledge up to last message delivered (if any). Used when commiting. - * - * @return the lastDeliveryTag to acknowledge - */ - Long getLastDelivered() - { - if (!_receivedDeliveryTags.isEmpty()) - { - Long lastDeliveryTag = _receivedDeliveryTags.poll(); - - while (!_receivedDeliveryTags.isEmpty()) - { - lastDeliveryTag = _receivedDeliveryTags.poll(); - } - - assert _receivedDeliveryTags.isEmpty(); - - return lastDeliveryTag; - } - - return null; - } - void notifyError(Throwable cause) { // synchronized (_closed) diff --git a/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer_0_8.java b/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer_0_8.java index cf1d7cedeb..efcbfd5532 100644 --- a/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer_0_8.java +++ b/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer_0_8.java @@ -28,7 +28,10 @@ import org.apache.qpid.client.failover.FailoverException; import org.apache.qpid.client.message.*; import org.apache.qpid.client.protocol.AMQProtocolHandler; import org.apache.qpid.common.AMQPFilterTypes; +import org.apache.qpid.configuration.ClientProperties; import org.apache.qpid.framing.*; +import org.apache.qpid.jms.ConnectionURL; +import org.apache.qpid.url.BindingURL; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -36,6 +39,8 @@ public class BasicMessageConsumer_0_8 extends BasicMessageConsumerSERVER value is specified. Otherwise the messages won't be moved to + * the DLQ (or dropped) when delivery count exceeds the maximum. + */ +public enum RejectBehaviour +{ + NORMAL, SERVER; +} diff --git a/qpid/java/client/src/main/java/org/apache/qpid/jms/ConnectionURL.java b/qpid/java/client/src/main/java/org/apache/qpid/jms/ConnectionURL.java index 26641982d7..24d9360cfa 100644 --- a/qpid/java/client/src/main/java/org/apache/qpid/jms/ConnectionURL.java +++ b/qpid/java/client/src/main/java/org/apache/qpid/jms/ConnectionURL.java @@ -41,7 +41,16 @@ public interface ConnectionURL public static final String OPTIONS_USE_LEGACY_MAP_MESSAGE_FORMAT = "use_legacy_map_msg_format"; public static final String OPTIONS_BROKERLIST = "brokerlist"; public static final String OPTIONS_FAILOVER = "failover"; - public static final String OPTIONS_FAILOVER_CYCLE = "cyclecount"; + public static final String OPTIONS_FAILOVER_CYCLE = "cyclecount"; + + /** + * This option is only applicable for 0-8/0-9/0-9-1 protocols connection + *

+ * It tells the client to delegate the requeue/DLQ decision to the + * server .If this option is not specified, the messages won't be moved to + * the DLQ (or dropped) when delivery count exceeds the maximum. + */ + public static final String OPTIONS_REJECT_BEHAVIOUR = "rejectbehaviour"; public static final String OPTIONS_DEFAULT_TOPIC_EXCHANGE = "defaultTopicExchange"; public static final String OPTIONS_DEFAULT_QUEUE_EXCHANGE = "defaultQueueExchange"; public static final String OPTIONS_TEMPORARY_TOPIC_EXCHANGE = "temporaryTopicExchange"; -- cgit v1.2.1