diff options
Diffstat (limited to 'qpid/java/client/src/main')
8 files changed, 171 insertions, 41 deletions
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 + * <p> + * 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<C extends BasicMessageConsumer, P extends Basic /** Holds the highest received delivery tag. */ protected final AtomicLong _highestDeliveryTag = new AtomicLong(-1); private final AtomicLong _rollbackMark = new AtomicLong(-1); - + + /** Pre-fetched message tags */ + protected ConcurrentLinkedQueue<Long> _prefetchedMessageTags = new ConcurrentLinkedQueue<Long>(); + /** All the not yet acknowledged message tags */ protected ConcurrentLinkedQueue<Long> _unacknowledgedMessageTags = new ConcurrentLinkedQueue<Long>(); @@ -2925,11 +2928,6 @@ public abstract class AMQSession<C extends BasicMessageConsumer, P extends Basic _producers.put(new Long(producerId), producer); } - private void rejectAllMessages(boolean requeue) - { - rejectMessagesForConsumerTag(0, requeue, true); - } - /** * @param consumerTag The consumerTag to prune from queue or all if null * @param requeue Should the removed messages be requeued (or discarded. Possibly to DLQ) @@ -3235,7 +3233,7 @@ public abstract class AMQSession<C extends BasicMessageConsumer, P extends Basic for (C consumer : _consumers.values()) { List<Long> tags = consumer.drainReceiverQueueAndRetrieveDeliveryTags(); - _unacknowledgedMessageTags.addAll(tags); + _prefetchedMessageTags.addAll(tags); } setConnectionStopped(isStopped); @@ -3345,7 +3343,7 @@ public abstract class AMQSession<C extends BasicMessageConsumer, P extends Basic } else if (_usingDispatcherForCleanup) { - _unacknowledgedMessageTags.add(deliveryTag); + _prefetchedMessageTags.add(deliveryTag); } else { @@ -3548,4 +3546,5 @@ public abstract class AMQSession<C extends BasicMessageConsumer, P extends Basic _logger.debug("Rollback mark is set to " + _rollbackMark.get()); } } + } diff --git a/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession_0_8.java b/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession_0_8.java index e33410f5fe..96df463481 100644 --- a/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession_0_8.java +++ b/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession_0_8.java @@ -21,6 +21,7 @@ package org.apache.qpid.client; +import java.util.ArrayList; import java.util.Map; import javax.jms.Destination; @@ -40,7 +41,6 @@ import org.apache.qpid.client.protocol.AMQProtocolHandler; import org.apache.qpid.client.state.AMQState; import org.apache.qpid.client.state.AMQStateManager; import org.apache.qpid.client.state.listener.SpecificMethodFrameListener; -import org.apache.qpid.common.AMQPFilterTypes; import org.apache.qpid.filter.MessageFilter; import org.apache.qpid.framing.AMQFrame; import org.apache.qpid.framing.AMQMethodBody; @@ -62,7 +62,6 @@ import org.apache.qpid.framing.ExchangeBoundOkBody; import org.apache.qpid.framing.ExchangeDeclareBody; import org.apache.qpid.framing.ExchangeDeclareOkBody; import org.apache.qpid.framing.FieldTable; -import org.apache.qpid.framing.FieldTableFactory; import org.apache.qpid.framing.ProtocolVersion; import org.apache.qpid.framing.QueueBindOkBody; import org.apache.qpid.framing.QueueDeclareBody; @@ -223,6 +222,8 @@ public final class AMQSession_0_8 extends AMQSession<BasicMessageConsumer_0_8, B public void sendRecover() throws AMQException, FailoverException { + enforceRejectBehaviourDuringRecover(); + _prefetchedMessageTags.clear(); _unacknowledgedMessageTags.clear(); if (isStrictAMQP()) @@ -259,6 +260,49 @@ public final class AMQSession_0_8 extends AMQSession<BasicMessageConsumer_0_8, B } } + private void enforceRejectBehaviourDuringRecover() + { + if (_logger.isDebugEnabled()) + { + _logger.debug("Prefetched message: _unacknowledgedMessageTags :" + _unacknowledgedMessageTags); + } + ArrayList<BasicMessageConsumer_0_8> consumersToCheck = new ArrayList<BasicMessageConsumer_0_8>(_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<BasicMessageConsumer_0_8, B // Otherwise messages will be able to arrive out of order to a second // consumer on the queue. Whilst this is within the JMS spec it is not // user friendly and avoidable. + boolean normalRejectBehaviour = true; + for (BasicMessageConsumer_0_8 consumer : _consumers.values()) + { + if(RejectBehaviour.SERVER.equals(consumer.getRejectBehaviour())) + { + normalRejectBehaviour = false; + //no need to consult other consumers now, found server behaviour. + break; + } + } + while (true) { Long tag = _deliveredMessageTags.poll(); @@ -275,13 +330,14 @@ public final class AMQSession_0_8 extends AMQSession<BasicMessageConsumer_0_8, B break; } - rejectMessage(tag, true); + rejectMessage(tag, normalRejectBehaviour); } } public void rejectMessage(long deliveryTag, boolean requeue) { - if ((_acknowledgeMode == CLIENT_ACKNOWLEDGE) || (_acknowledgeMode == SESSION_TRANSACTED)) + if ((_acknowledgeMode == CLIENT_ACKNOWLEDGE) || (_acknowledgeMode == SESSION_TRANSACTED)|| + ((_acknowledgeMode == AUTO_ACKNOWLEDGE || _acknowledgeMode == DUPS_OK_ACKNOWLEDGE ) && hasMessageListeners())) { if (_logger.isDebugEnabled()) { diff --git a/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer.java b/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer.java index 7bb400fada..c6e5fbb019 100644 --- a/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer.java +++ b/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer.java @@ -147,7 +147,6 @@ public abstract class BasicMessageConsumer<U> extends Closeable implements Messa private List<StackTraceElement> _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<U> 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<U> 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 BasicMessageConsumer<UnprocessedMe { protected final Logger _logger = LoggerFactory.getLogger(getClass()); + private final RejectBehaviour _rejectBehaviour; + protected BasicMessageConsumer_0_8(int channelId, AMQConnection connection, AMQDestination destination, String messageSelector, boolean noLocal, MessageFactoryRegistry messageFactory, AMQSession session, AMQProtocolHandler protocolHandler, FieldTable rawSelector, int prefetchHigh, int prefetchLow, @@ -55,6 +60,25 @@ public class BasicMessageConsumer_0_8 extends BasicMessageConsumer<UnprocessedMe consumerArguments.put(AMQPFilterTypes.NO_CONSUME.getValue(), Boolean.TRUE); } + if (destination.getRejectBehaviour() != null) + { + _rejectBehaviour = destination.getRejectBehaviour(); + } + else + { + ConnectionURL connectionURL = connection.getConnectionURL(); + String rejectBehaviour = connectionURL.getOption(ConnectionURL.OPTIONS_REJECT_BEHAVIOUR); + if (rejectBehaviour != null) + { + _rejectBehaviour = RejectBehaviour.valueOf(rejectBehaviour.toUpperCase()); + } + else + { + // use the default value for all connections, if not set + rejectBehaviour = System.getProperty(ClientProperties.REJECT_BEHAVIOUR_PROP_NAME, RejectBehaviour.NORMAL.toString()); + _rejectBehaviour = RejectBehaviour.valueOf( rejectBehaviour.toUpperCase()); + } + } } void sendCancel() throws AMQException, FailoverException @@ -89,4 +113,9 @@ public class BasicMessageConsumer_0_8 extends BasicMessageConsumer<UnprocessedMe { } + + public RejectBehaviour getRejectBehaviour() + { + return _rejectBehaviour; + } } diff --git a/qpid/java/client/src/main/java/org/apache/qpid/client/RejectBehaviour.java b/qpid/java/client/src/main/java/org/apache/qpid/client/RejectBehaviour.java new file mode 100644 index 0000000000..e3c958044e --- /dev/null +++ b/qpid/java/client/src/main/java/org/apache/qpid/client/RejectBehaviour.java @@ -0,0 +1,32 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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.apache.qpid.client; + +/** + * This enum can be used only with for 0-8/0-9/0-9-1 protocols connections to notify + * the client to delegate the requeue/DLQ decision to the server + * if <code>SERVER</server> 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 + * <p> + * 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"; |