diff options
| -rw-r--r-- | java/client/src/main/java/org/apache/qpid/client/AMQSession_0_10.java | 22 |
1 files changed, 19 insertions, 3 deletions
diff --git a/java/client/src/main/java/org/apache/qpid/client/AMQSession_0_10.java b/java/client/src/main/java/org/apache/qpid/client/AMQSession_0_10.java index b2dce0404c..74cd0c8b78 100644 --- a/java/client/src/main/java/org/apache/qpid/client/AMQSession_0_10.java +++ b/java/client/src/main/java/org/apache/qpid/client/AMQSession_0_10.java @@ -565,9 +565,25 @@ public class AMQSession_0_10 extends AMQSession<BasicMessageConsumer_0_10, Basic try { - preAcquire = ( ! consumer.isNoConsume() && - (consumer.getMessageSelector() == null || consumer.getMessageSelector().equals("")) ) - || !(consumer.getDestination() instanceof AMQQueue); + boolean isTopic; + + if (consumer.getDestination().getDestSyntax() == AMQDestination.DestSyntax.BURL) + { + isTopic = consumer.getDestination() instanceof AMQTopic || + consumer.getDestination().getExchangeClass().equals(ExchangeDefaults.TOPIC_EXCHANGE_CLASS) ; + + preAcquire = isTopic || (!consumer.isNoConsume() && + (consumer.getMessageSelector() == null || consumer.getMessageSelector().equals(""))); + } + else + { + isTopic = consumer.getDestination().getAddressType() == AMQDestination.TOPIC_TYPE; + + preAcquire = !consumer.isNoConsume() && + (isTopic || consumer.getMessageSelector() == null || + consumer.getMessageSelector().equals("")); + } + getQpidSession().messageSubscribe (queueName.toString(), String.valueOf(tag), getAcknowledgeMode() == NO_ACKNOWLEDGE ? MessageAcceptMode.NONE : MessageAcceptMode.EXPLICIT, |