Merged Trunk Changes to version 489140.

The point where we achieved 100% JMS compilance.

Revision: 489140
Author: ritchiem
Date: 17:04:33, 20 December 2006
Message:
QPID-225
Applied Patch for queue browsing with client acknowledgement
----
Modified : /incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer.java


Revision: 489113
Author: bhupendrab
Date: 15:25:45, 20 December 2006
Message:
renamed the jar
----
Modified : /incubator/qpid/trunk/qpid/java/distribution/src/main/assembly/management-eclipse-plugin.xml


Revision: 489106
Author: ritchiem
Date: 14:54:01, 20 December 2006
Message:
QPID-101

Initial Implementation of Queue Browsing by Robert Godfrey and Martin Ritchie

AMQChannel.java - record messages browsed so not to discard them on ack.
FilterManagerFactory.java - Added a NoConsumerFilter
ConcurrentSelectorDeliveryManager.java - Update to send browsers messages without taking the message from other consumers
Subscription.java - Added autoClose and isBrowser methods
SubscriptionTestHelper.java / RemoteSubscriptionImpl.java / SubscriptionImpl.java - implemented new interface methods
Added NoConsumerFilter.java


Patches from Rob Godfrey for client implmentation
AMQSession.java - Added AUTO_CLOSE and NO_CONSUME properties to arguments FieldTable for consume method.
BasicMessageConsumer.java - updates to correctly close consumer when an BasicCancel is received from the broker.
AMQProtocolSession.java - method to allow cancellation of the client 
AMQStateManager.java - added handler for BasicCancelOkMethodHandler.java
Added new AMQQueueBrowser.java BasicCancelOkMethodHandler.java
----
Modified : /incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java
Modified : /incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/FilterManagerFactory.java
Added : /incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/NoConsumerFilter.java
Modified : /incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/ConcurrentSelectorDeliveryManager.java
Modified : /incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/Subscription.java
Modified : /incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SubscriptionImpl.java
Added : /incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQQueueBrowser.java
Modified : /incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession.java
Modified : /incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer.java
Added : /incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/handler/BasicCancelOkMethodHandler.java
Modified : /incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolSession.java
Modified : /incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/state/AMQStateManager.java
Modified : /incubator/qpid/trunk/qpid/java/cluster/src/main/java/org/apache/qpid/server/queue/RemoteSubscriptionImpl.java
Modified : /incubator/qpid/trunk/qpid/java/systests/src/test/java/org/apache/qpid/server/queue/SubscriptionTestHelper.java


Revision: 489083
Author: ritchiem
Date: 13:26:12, 20 December 2006
Message:
Updated FilterTypes to be more accurate NO_CONSUME and AUTO_CLOSE
----
Modified : /incubator/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/common/AMQPFilterTypes.java


Revision: 489082
Author: ritchiem
Date: 13:22:27, 20 December 2006
Message:
QPID-233

Applied patch from Rupert Smith
----
Modified : /incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession.java
Modified : /incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/AbstractJMSMessage.java
Modified : /incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/JMSBytesMessage.java
Modified : /incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/JMSObjectMessage.java
Modified : /incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/JMSStreamMessage.java
Modified : /incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/JMSTextMessage.java


Revision: 489078
Author: ritchiem
Date: 12:57:27, 20 December 2006
Message:
Added new enum for AMQP Filter types
----
Added : /incubator/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/common/AMQPFilterTypes.java


Revision: 489070
Author: ritchiem
Date: 12:46:20, 20 December 2006
Message:
QPID-21 outstanding issues:
Fixed an issue where a consumer with no_local set would not have its filters applied to messages.
Fixed problem where new consumers would start with an empty PDQ rather than checking the existing queue of messages for messages of interest.

AMQQueue.java - Added code check exisiting queue data for messages for the new subscriber with a filter.
DeliveryManager.java - added populatePreDeliveryQueue
SynchronizedDeliveryManager.java/ConcurrentDeliveryManager.java - implemented new DeliveryManager.java interface
SubscriptionImpl.java - fixed issue with no_local subscribers had their filters ignored.
----
Modified : /incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/AMQQueue.java
Modified : /incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/ConcurrentDeliveryManager.java
Modified : /incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/ConcurrentSelectorDeliveryManager.java
Modified : /incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/DeliveryManager.java
Modified : /incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SubscriptionImpl.java
Modified : /incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SynchronizedDeliveryManager.java


Revision: 488726
Author: ritchiem
Date: 17:02:19, 19 December 2006
Message:
QPID-222

ensured that the TXBuffer of a message is set to null when re queuing.
----
Modified : /incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java


Revision: 488715
Author: ritchiem
Date: 16:14:28, 19 December 2006
Message:
Maven output clean up.
Mainly removed exception stack traces from expected exceptions.
----
Modified : /incubator/qpid/trunk/qpid/java/systests/src/test/java/org/apache/qpid/server/exchange/ReturnUnroutableMandatoryMessageTest.java


Revision: 488713
Author: ritchiem
Date: 16:09:39, 19 December 2006
Message:
Maven output clean up.
Mainly removed exception stack traces from expected exceptions.
----
Modified : /incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/util/CircularBuffer.java
Modified : /incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/basic/PropertyValueTest.java
Modified : /incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/client/channelclose/ChannelCloseOkTest.java
Modified : /incubator/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/framing/PropertyFieldTable.java


Revision: 488712
Author: ritchiem
Date: 16:07:12, 19 December 2006
Message:
QPID-216
BasicConsumeMethodHandler.java - Pulled the nolocal param from the method body and passed down channel to subscription.
SubscriptionFactory.java / AMQQueue.java/AMQChannel.java - passed the nolocal parameter through to the Subscription
ConnectionStartOkMethodHandler.java - Saved the client properties so the client identifier can be used in comparison with the publisher id to implement no_local
AMQMinaProtocolSession.java - added _clientProperties to store the sent client properties.
AMQProtocolSession.java - interface changes to get/set ClientProperties
ConcurrentSelectorDeliveryManager.java - only need to do hasInterset as this will take care of the hasFilters optimisation check.
SubscriptionImpl.java - Added code to do comparison of client ids to determin insterest in a given message.
SubscriptionSet.java - tidied up code to use hasInterest as this is where the nolocal is implemented.

ConnectionStartMethodHandler.java - Moved literal values to a ClientProperties.java enumeration and a QpidProperties.java values.
QpidConnectionMetaData.java - updated to get values from QpidProperties.java

MockProtocolSession.java - null implementation of new get/set methods
----
Modified : /incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java
Modified : /incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/BasicConsumeMethodHandler.java
Modified : /incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionStartOkMethodHandler.java
Modified : /incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQMinaProtocolSession.java
Modified : /incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolSession.java
Modified : /incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/AMQQueue.java
Modified : /incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/ConcurrentSelectorDeliveryManager.java
Modified : /incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SubscriptionFactory.java
Modified : /incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SubscriptionImpl.java
Modified : /incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SubscriptionSet.java
Modified : /incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/QpidConnectionMetaData.java
Modified : /incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/handler/ConnectionStartMethodHandler.java
Added : /incubator/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/common
Added : /incubator/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/common/ClientProperties.java
Added : /incubator/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/common/QpidProperties.java
Modified : /incubator/qpid/trunk/qpid/java/systests/src/test/java/org/apache/qpid/server/queue/MockProtocolSession.java


Revision: 488711
Author: bhupendrab
Date: 16:00:13, 19 December 2006
Message:
QPID-188 
Unit test for Exchange MBeans
----
Added : /incubator/qpid/trunk/qpid/java/systests/src/test/java/org/apache/qpid/server/exchange/ExchangeMBeanTest.java


Revision: 488705
Author: bhupendrab
Date: 15:17:25, 19 December 2006
Message:
QPID-188 
Adding unit tests for Java broker JMX functionality
----
Added : /incubator/qpid/trunk/qpid/java/systests/src/test/java/org/apache/qpid/server/protocol/AMQProtocolSessionMBeanTest.java


Revision: 488701
Author: bhupendrab
Date: 15:09:50, 19 December 2006
Message:
QPID-188 
Adding unit tests for Java broker JMX functionality
----
Modified : /incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQMinaProtocolSession.java
Modified : /incubator/qpid/trunk/qpid/java/systests/src/test/java/org/apache/qpid/server/protocol/MockIoSession.java


Revision: 488624
Author: ritchiem
Date: 10:51:39, 19 December 2006
Message:
QPID-21
Added:
SelectorParser.jj - ActiveMQ selector javacc grammar used to generate SelectorParser.java
server/filter - Selector Filtering code from ActiveMQ project adjusted to suite our class and package structure.
server/message - Decorator classes to allow access to the JMSMessage inside the AMQMessage
ConcurrentSelectorDeliveryManager.java - A new DeliveryManager that utilises PreDeliveryQueues to implement selectors
AMQInvalidSelectorException.java - thrown on client and broker when the Selector text is invalid.
Common: log4j.properties to remove error log4j warnings on Common tests.

Modified:
broker/pom.xml - to generate SelectorParser.java
AMQChannel.java - Addition of argument fieldtable for filter setup.
BasicConsumeMethodHandler.java - writing of InvalidSelector channel close exception.
AMQMessage.java - Added decorator to get access to the enclosed JMSMessage
AMQQueue.java - Enhanced 'deliverymanager' property to allow the selection of the ConcurrentSelectorDeliveryManager.
Subscription.java - Enhanced interface to allow a subscription to state an 'interest' in a given message.
SubscriptionFactory.java - Added method to allow passing of filter arguments.
SubscriptionImpl.java - Implemented new Subscription.java methods.
SubscriptionManager.java - Added ability to get a list of current subscribers.
SubscriptionSet.java - augmented nextSubscriber to allow the subscriber to exert the new hasInterest feature.
SynchronizedDeliveryManager.java - fixed Logging class
AMQSession - Added filter extraction from consume call and pass it on to the registration.
ChannelCloseMethodHandler.java - Handle the reception and correct raising of the InvalidSelector Exception
AbstractJMSMessage.java - Expanded imports
BlockingMethodFrameListener.java - added extra info to a debug output line.
SocketTransportConnection.java - made output an info not a warn.
PropertiesFileInitialContextFactory.java - updated to allow the PROVIDER_URL to specify a property file to read in for the initial values.
ClusteredSubscriptionManager.java - Implementation of SubscriptionSet.java
NestedSubscriptionManager.java - Implementation of SubscriptionManager.java
RemoteSubscriptionImpl.java - Implementation Subscription.java
AMQConstant.java - Added '322' "Invalid Selector"
SubscriptionTestHelper.java - Implementation of Subscription.java

Edited specs/amqp-8.0.xml to add field table to consume method.

Thanks to the ActiveMQ project for writing the initial SelectorParser.jj and associated filter Expressions.
----
Modified : /incubator/qpid/trunk/qpid/java/broker/pom.xml
Added : /incubator/qpid/trunk/qpid/java/broker/src/main/grammar(Copy from path: /incubator/qpid/branches/jmsselectors/java/broker/src/main/grammar, Revision, 488302
Replacing : /incubator/qpid/trunk/qpid/java/broker/src/main/grammar/SelectorParser.jj(Copy from path: /incubator/qpid/branches/jmsselectors/java/broker/src/main/grammar/SelectorParser.jj, Revision, 488302
Modified : /incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java
Added : /incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/filter(Copy from path: /incubator/qpid/branches/jmsselectors/java/broker/src/main/java/org/apache/qpid/server/filter, Revision, 488302
Replacing : /incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/ArithmeticExpression.java(Copy from path: /incubator/qpid/branches/jmsselectors/java/broker/src/main/java/org/apache/qpid/server/filter/ArithmeticExpression.java, Revision, 488302
Replacing : /incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/BinaryExpression.java(Copy from path: /incubator/qpid/branches/jmsselectors/java/broker/src/main/java/org/apache/qpid/server/filter/BinaryExpression.java, Revision, 488302
Replacing : /incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/BooleanExpression.java(Copy from path: /incubator/qpid/branches/jmsselectors/java/broker/src/main/java/org/apache/qpid/server/filter/BooleanExpression.java, Revision, 488302
Replacing : /incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/ComparisonExpression.java(Copy from path: /incubator/qpid/branches/jmsselectors/java/broker/src/main/java/org/apache/qpid/server/filter/ComparisonExpression.java, Revision, 488302
Replacing : /incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/ConstantExpression.java(Copy from path: /incubator/qpid/branches/jmsselectors/java/broker/src/main/java/org/apache/qpid/server/filter/ConstantExpression.java, Revision, 488302
Replacing : /incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/Expression.java(Copy from path: /incubator/qpid/branches/jmsselectors/java/broker/src/main/java/org/apache/qpid/server/filter/Expression.java, Revision, 488302
Replacing : /incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/FilterManager.java(Copy from path: /incubator/qpid/branches/jmsselectors/java/broker/src/main/java/org/apache/qpid/server/filter/FilterManager.java, Revision, 488302
Replacing : /incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/FilterManagerFactory.java(Copy from path: /incubator/qpid/branches/jmsselectors/java/broker/src/main/java/org/apache/qpid/server/filter/FilterManagerFactory.java, Revision, 488302
Replacing : /incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/JMSSelectorFilter.java(Copy from path: /incubator/qpid/branches/jmsselectors/java/broker/src/main/java/org/apache/qpid/server/filter/JMSSelectorFilter.java, Revision, 488302
Replacing : /incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/LogicExpression.java(Copy from path: /incubator/qpid/branches/jmsselectors/java/broker/src/main/java/org/apache/qpid/server/filter/LogicExpression.java, Revision, 488302
Replacing : /incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/MessageFilter.java(Copy from path: /incubator/qpid/branches/jmsselectors/java/broker/src/main/java/org/apache/qpid/server/filter/MessageFilter.java, Revision, 488302
Replacing : /incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/PropertyExpression.java(Copy from path: /incubator/qpid/branches/jmsselectors/java/broker/src/main/java/org/apache/qpid/server/filter/PropertyExpression.java, Revision, 488302
Replacing : /incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/SimpleFilterManager.java(Copy from path: /incubator/qpid/branches/jmsselectors/java/broker/src/main/java/org/apache/qpid/server/filter/SimpleFilterManager.java, Revision, 488302
Replacing : /incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/UnaryExpression.java(Copy from path: /incubator/qpid/branches/jmsselectors/java/broker/src/main/java/org/apache/qpid/server/filter/UnaryExpression.java, Revision, 488302
Replacing : /incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/XPathExpression.java(Copy from path: /incubator/qpid/branches/jmsselectors/java/broker/src/main/java/org/apache/qpid/server/filter/XPathExpression.java, Revision, 488302
Replacing : /incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/XQueryExpression.java(Copy from path: /incubator/qpid/branches/jmsselectors/java/broker/src/main/java/org/apache/qpid/server/filter/XQueryExpression.java, Revision, 488302
Replacing : /incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/XalanXPathEvaluator.java(Copy from path: /incubator/qpid/branches/jmsselectors/java/broker/src/main/java/org/apache/qpid/server/filter/XalanXPathEvaluator.java, Revision, 488302
Modified : /incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/BasicConsumeMethodHandler.java
Added : /incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/message(Copy from path: /incubator/qpid/branches/jmsselectors/java/broker/src/main/java/org/apache/qpid/server/message, Revision, 488302
Replacing : /incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/message/MessageDecorator.java(Copy from path: /incubator/qpid/branches/jmsselectors/java/broker/src/main/java/org/apache/qpid/server/message/MessageDecorator.java, Revision, 488302
Replacing : /incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/message/jms(Copy from path: /incubator/qpid/branches/jmsselectors/java/broker/src/main/java/org/apache/qpid/server/message/jms, Revision, 488302
Replacing : /incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/message/jms/JMSMessage.java(Copy from path: /incubator/qpid/branches/jmsselectors/java/broker/src/main/java/org/apache/qpid/server/message/jms/JMSMessage.java, Revision, 488302
Modified : /incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/AMQMessage.java
Modified : /incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/AMQQueue.java
Added : /incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/ConcurrentSelectorDeliveryManager.java(Copy from path: /incubator/qpid/branches/jmsselectors/java/broker/src/main/java/org/apache/qpid/server/queue/ConcurrentSelectorDeliveryManager.java, Revision, 488302
Modified : /incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/Subscription.java
Modified : /incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SubscriptionFactory.java
Modified : /incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SubscriptionImpl.java
Modified : /incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SubscriptionManager.java
Modified : /incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SubscriptionSet.java
Modified : /incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SynchronizedDeliveryManager.java
Modified : /incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession.java
Modified : /incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/handler/ChannelCloseMethodHandler.java
Modified : /incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/AbstractJMSMessage.java
Modified : /incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/protocol/BlockingMethodFrameListener.java
Modified : /incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/transport/SocketTransportConnection.java
Modified : /incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/jndi/PropertiesFileInitialContextFactory.java
Added : /incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/basic/SelectorTest.java(Copy from path: /incubator/qpid/branches/jmsselectors/java/client/src/test/java/org/apache/qpid/test/unit/basic/SelectorTest.java, Revision, 488302
Modified : /incubator/qpid/trunk/qpid/java/cluster/src/main/java/org/apache/qpid/server/queue/ClusteredSubscriptionManager.java
Modified : /incubator/qpid/trunk/qpid/java/cluster/src/main/java/org/apache/qpid/server/queue/NestedSubscriptionManager.java
Modified : /incubator/qpid/trunk/qpid/java/cluster/src/main/java/org/apache/qpid/server/queue/RemoteSubscriptionImpl.java
Added : /incubator/qpid/trunk/qpid/java/common/src/main/java/log4j.properties
Added : /incubator/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/AMQInvalidSelectorException.java(Copy from path: /incubator/qpid/branches/jmsselectors/java/common/src/main/java/org/apache/qpid/AMQInvalidSelectorException.java, Revision, 488302
Modified : /incubator/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/framing/PropertyFieldTable.java
Modified : /incubator/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/protocol/AMQConstant.java
Modified : /incubator/qpid/trunk/qpid/java/systests/src/test/java/org/apache/qpid/server/queue/AMQQueueMBeanTest.java
Modified : /incubator/qpid/trunk/qpid/java/systests/src/test/java/org/apache/qpid/server/queue/SubscriptionTestHelper.java
Modified : /incubator/qpid/trunk/qpid/specs/amqp-8.0.xml


Revision: 488596
Author: rgreig
Date: 09:29:19, 19 December 2006
Message:
QPID-215 : Patch supplied by Rob Godfrey - Implement custom JMSX properties
----
Modified : /incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnection.java
Modified : /incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer.java
Modified : /incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageProducer.java
Added : /incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/CustomJMXProperty.java
Modified : /incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/QpidConnectionMetaData.java
Modified : /incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/AbstractJMSMessage.java
Added : /incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/message/JMSDestinationTest.java


Revision: 488594
Author: bhupendrab
Date: 09:13:29, 19 December 2006
Message:
Name corrected
----
Added : /incubator/qpid/trunk/qpid/java/systests/src/test/java/org/apache/qpid/server/queue/AMQQueueMBeanTest.java(Copy from path: /incubator/qpid/trunk/qpid/java/systests/src/test/java/org/apache/qpid/server/queue/AMQueueMBeanTest.java, Revision, 488281
Deleted : /incubator/qpid/trunk/qpid/java/systests/src/test/java/org/apache/qpid/server/queue/AMQueueMBeanTest.java


Revision: 488450
Author: vinoski
Date: 23:09:14, 18 December 2006
Message:
clean up warnings about unused variables

Remove all warnings in common, broker, client, and systests regarding
unused variables, as indicated by Eclipse builds.
----
Modified : /incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/Main.java
Modified : /incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/management/MBeanIntrospector.java
Modified : /incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/security/auth/plain/PlainSaslServer.java
Modified : /incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/handler/ConnectionOpenOkMethodHandler.java
Modified : /incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/ack/RecoverTest.java
Modified : /incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/basic/MultipleConnectionTest.java
Modified : /incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/client/channelclose/CloseWithBlockingReceiveTest.java
Modified : /incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/client/connection/ConnectionTest.java
Modified : /incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/client/message/MapMessageTest.java
Modified : /incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/message/StreamMessageTest.java
Modified : /incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/topic/TopicSessionTest.java
Modified : /incubator/qpid/trunk/qpid/java/systests/src/test/java/org/apache/qpid/server/exchange/HeadersExchangeTest.java
Modified : /incubator/qpid/trunk/qpid/java/systests/src/test/java/org/apache/qpid/server/exchange/ReturnUnroutableMandatoryMessageTest.java



git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/branches/perftesting@496666 13f79535-47bb-0310-9956-ffa450edef68

41 files changed, 4207 insertions, 89 deletions

diff --git a/qpid/java/broker/src/main/grammar/SelectorParser.jj b/qpid/java/broker/src/main/grammar/SelectorParser.jj
new file mode 100644
index 0000000000..5553a46e47
--- /dev/null
+++ b/qpid/java/broker/src/main/grammar/SelectorParser.jj
@@ -0,0 +1,598 @@
+/*

+ *

+ * 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.

+ *

+ */

+

+ //

+ // Original File from r450141 of the Apache ActiveMQ project <http://www.activemq.org/site/home.html>

+ //

+

+// ----------------------------------------------------------------------------

+// OPTIONS

+// ----------------------------------------------------------------------------

+options {

+  STATIC = false;

+  UNICODE_INPUT = true;

+

+  // some performance optimizations

+  OPTIMIZE_TOKEN_MANAGER = true;

+  ERROR_REPORTING = false;

+}

+

+// ----------------------------------------------------------------------------

+// PARSER

+// ----------------------------------------------------------------------------

+

+PARSER_BEGIN(SelectorParser)

+/*

+ *

+ * 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.server.filter.jms.selector;

+

+import java.io.*;

+import java.util.*;

+

+import javax.jms.InvalidSelectorException;

+

+import org.apache.qpid.server.filter.*;

+

+/**

+ * JMS Selector Parser generated by JavaCC

+ *

+ * Do not edit this .java file directly - it is autogenerated from SelectorParser.jj

+ */

+public class SelectorParser {

+

+    public SelectorParser() {

+        this(new StringReader(""));

+    }

+

+    public BooleanExpression parse(String sql) throws InvalidSelectorException {

+        this.ReInit(new StringReader(sql));

+

+        try {

+            return this.JmsSelector();

+        }

+        catch (Throwable e) {

+	        throw (InvalidSelectorException)new InvalidSelectorException(sql).initCause(e);

+        }

+

+    }

+

+    private BooleanExpression asBooleanExpression(Expression value) throws ParseException  {

+        if (value instanceof BooleanExpression) {

+            return (BooleanExpression) value;

+        }

+        if (value instanceof PropertyExpression) {

+            return UnaryExpression.createBooleanCast( value );

+        }

+        throw new ParseException("Expression will not result in a boolean value: " + value);

+    }

+

+

+}

+

+PARSER_END(SelectorParser)

+

+// ----------------------------------------------------------------------------

+// Tokens

+// ----------------------------------------------------------------------------

+

+/* White Space */

+SPECIAL_TOKEN :

+{

+  " " | "\t" | "\n" | "\r" | "\f"

+}

+

+/* Comments */

+SKIP:

+{

+  <LINE_COMMENT: "--" (~["\n","\r"])* ("\n"|"\r"|"\r\n") >

+}

+

+SKIP:

+{

+  <BLOCK_COMMENT: "/*" (~["*"])* "*" ("*" | (~["*","/"] (~["*"])* "*"))* "/">

+}

+

+/* Reserved Words */

+TOKEN [IGNORE_CASE] :

+{

+    <  NOT     : "NOT">

+  | <  AND     : "AND">

+  | <  OR      : "OR">

+  | <  BETWEEN : "BETWEEN">

+  | <  LIKE    : "LIKE">

+  | <  ESCAPE  : "ESCAPE">

+  | <  IN      : "IN">

+  | <  IS      : "IS">

+  | <  TRUE    : "TRUE" >

+  | <  FALSE   : "FALSE" >

+  | <  NULL    : "NULL" >

+  | <  XPATH   : "XPATH" >

+  | <  XQUERY  : "XQUERY" >

+}

+

+/* Literals */

+TOKEN [IGNORE_CASE] :

+{

+

+    < DECIMAL_LITERAL: ["1"-"9"] (["0"-"9"])* (["l","L"])? >

+  | < HEX_LITERAL: "0" ["x","X"] (["0"-"9","a"-"f","A"-"F"])+ >

+  | < OCTAL_LITERAL: "0" (["0"-"7"])* >

+  | < FLOATING_POINT_LITERAL:

+          (["0"-"9"])+ "." (["0"-"9"])* (<EXPONENT>)? // matches: 5.5 or 5. or 5.5E10 or 5.E10

+        | "." (["0"-"9"])+ (<EXPONENT>)?              // matches: .5 or .5E10

+        | (["0"-"9"])+ <EXPONENT>                     // matches: 5E10

+    >

+  | < #EXPONENT: "E" (["+","-"])? (["0"-"9"])+ >

+  | < STRING_LITERAL: "'" ( ("''") | ~["'"] )*  "'" >

+}

+

+TOKEN [IGNORE_CASE] :

+{

+    < ID : ["a"-"z", "_", "$"] (["a"-"z","0"-"9","_", "$"])* >

+}

+

+// ----------------------------------------------------------------------------

+// Grammer

+// ----------------------------------------------------------------------------

+BooleanExpression JmsSelector() :

+{

+    Expression left=null;

+}

+{

+    (

+        left = orExpression()

+    )

+    {

+        return asBooleanExpression(left);

+    }

+

+}

+

+Expression orExpression() :

+{

+    Expression left;

+    Expression right;

+}

+{

+    (

+        left = andExpression()

+        (

+            <OR> right = andExpression()

+            {

+                left = LogicExpression.createOR(asBooleanExpression(left), asBooleanExpression(right));

+            }

+        )*

+    )

+    {

+        return left;

+    }

+

+}

+

+

+Expression andExpression() :

+{

+    Expression left;

+    Expression right;

+}

+{

+    (

+        left = equalityExpression()

+        (

+            <AND> right = equalityExpression()

+            {

+                left = LogicExpression.createAND(asBooleanExpression(left), asBooleanExpression(right));

+            }

+        )*

+    )

+    {

+        return left;

+    }

+}

+

+Expression equalityExpression() :

+{

+    Expression left;

+    Expression right;

+}

+{

+    (

+        left = comparisonExpression()

+        (

+

+            "=" right = comparisonExpression()

+            {

+                left = ComparisonExpression.createEqual(left, right);

+            }

+            |

+            "<>" right = comparisonExpression()

+            {

+                left = ComparisonExpression.createNotEqual(left, right);

+            }

+            |

+            LOOKAHEAD(2)

+            <IS> <NULL>

+            {

+                left = ComparisonExpression.createIsNull(left);

+            }

+            |

+            <IS> <NOT> <NULL>

+            {

+                left = ComparisonExpression.createIsNotNull(left);

+            }

+        )*

+    )

+    {

+        return left;

+    }

+}

+

+Expression comparisonExpression() :

+{

+    Expression left;

+    Expression right;

+    Expression low;

+    Expression high;

+    String t, u;

+	boolean not;

+	ArrayList list;

+}

+{

+    (

+        left = addExpression()

+        (

+

+                ">" right = addExpression()

+                {

+                    left = ComparisonExpression.createGreaterThan(left, right);

+                }

+            |

+                ">=" right = addExpression()

+                {

+                    left = ComparisonExpression.createGreaterThanEqual(left, right);

+                }

+            |

+                "<" right = addExpression()

+                {

+                    left = ComparisonExpression.createLessThan(left, right);

+                }

+            |

+                "<=" right = addExpression()

+                {

+                    left = ComparisonExpression.createLessThanEqual(left, right);

+                }

+           |

+				{

+					u=null;

+				}

+		        <LIKE> t = stringLitteral()

+		        	[ <ESCAPE> u = stringLitteral() ]

+		        {

+                    left = ComparisonExpression.createLike(left, t, u);

+		        }

+           |

+	        	LOOKAHEAD(2)

+				{

+					u=null;

+				}

+		        <NOT> <LIKE> t = stringLitteral() [ <ESCAPE> u = stringLitteral() ]

+		        {

+                    left = ComparisonExpression.createNotLike(left, t, u);

+		        }

+            |

+		        <BETWEEN> low = addExpression() <AND> high = addExpression()

+		        {

+					left = ComparisonExpression.createBetween(left, low, high);

+		        }

+	        |

+	        	LOOKAHEAD(2)

+		        <NOT> <BETWEEN> low = addExpression() <AND> high = addExpression()

+		        {

+					left = ComparisonExpression.createNotBetween(left, low, high);

+		        }

+            |

+				<IN>

+		        "("

+		            t = stringLitteral()

+		            {

+			            list = new ArrayList();

+			            list.add( t );

+		            }

+			        (

+			        	","

+			            t = stringLitteral()

+			            {

+				            list.add( t );

+			            }

+

+			        )*

+		        ")"

+		        {

+		           left = ComparisonExpression.createInFilter(left, list);

+		        }

+            |

+	        	LOOKAHEAD(2)

+	            <NOT> <IN>

+		        "("

+		            t = stringLitteral()

+		            {

+			            list = new ArrayList();

+			            list.add( t );

+		            }

+			        (

+			        	","

+			            t = stringLitteral()

+			            {

+				            list.add( t );

+			            }

+

+			        )*

+		        ")"

+		        {

+		           left = ComparisonExpression.createNotInFilter(left, list);

+		        }

+

+        )*

+    )

+    {

+        return left;

+    }

+}

+

+Expression addExpression() :

+{

+    Expression left;

+    Expression right;

+}

+{

+    left = multExpr()

+    (

+	    LOOKAHEAD( ("+"|"-") multExpr())

+	    (

+	        "+" right = multExpr()

+	        {

+	            left = ArithmeticExpression.createPlus(left, right);

+	        }

+	        |

+	        "-" right = multExpr()

+	        {

+	            left = ArithmeticExpression.createMinus(left, right);

+	        }

+        )

+

+    )*

+    {

+        return left;

+    }

+}

+

+Expression multExpr() :

+{

+    Expression left;

+    Expression right;

+}

+{

+    left = unaryExpr()

+    (

+        "*" right = unaryExpr()

+        {

+	        left = ArithmeticExpression.createMultiply(left, right);

+        }

+        |

+        "/" right = unaryExpr()

+        {

+	        left = ArithmeticExpression.createDivide(left, right);

+        }

+        |

+        "%" right = unaryExpr()

+        {

+	        left = ArithmeticExpression.createMod(left, right);

+        }

+

+    )*

+    {

+        return left;

+    }

+}

+

+

+Expression unaryExpr() :

+{

+    String s=null;

+    Expression left=null;

+}

+{

+	(

+		LOOKAHEAD( "+" unaryExpr() )

+	    "+" left=unaryExpr()

+	    |

+	    "-" left=unaryExpr()

+	    {

+	        left = UnaryExpression.createNegate(left);

+	    }

+	    |

+	    <NOT> left=unaryExpr()

+	    {

+		    left = UnaryExpression.createNOT( asBooleanExpression(left) );

+	    }

+	    |

+	    <XPATH> s=stringLitteral()

+	    {

+		    left = UnaryExpression.createXPath( s );

+	    }

+	    |

+	    <XQUERY> s=stringLitteral()

+	    {

+		    left = UnaryExpression.createXQuery( s );

+	    }

+	    |

+	    left = primaryExpr()

+    )

+    {

+        return left;

+    }

+

+}

+

+Expression primaryExpr() :

+{

+    Expression left=null;

+}

+{

+    (

+        left = literal()

+        |

+        left = variable()

+        |

+        "(" left = orExpression() ")"

+    )

+    {

+        return left;

+    }

+}

+

+

+

+ConstantExpression literal() :

+{

+    Token t;

+    String s;

+    ConstantExpression left=null;

+}

+{

+    (

+        (

+            s = stringLitteral()

+            {

+                left = new ConstantExpression(s);

+            }

+        )

+        |

+        (

+            t = <DECIMAL_LITERAL>

+            {

+            	left = ConstantExpression.createFromDecimal(t.image);

+            }

+        )

+        |

+        (

+            t = <HEX_LITERAL>

+            {

+            	left = ConstantExpression.createFromHex(t.image);

+            }

+        )

+        |

+        (

+            t = <OCTAL_LITERAL>

+            {

+            	left = ConstantExpression.createFromOctal(t.image);

+            }

+        )

+        |

+        (

+            t = <FLOATING_POINT_LITERAL>

+            {

+            	left = ConstantExpression.createFloat(t.image);

+            }

+        )

+        |

+        (

+            <TRUE>

+            {

+                left = ConstantExpression.TRUE;

+            }

+        )

+        |

+        (

+            <FALSE>

+            {

+                left = ConstantExpression.FALSE;

+            }

+        )

+        |

+        (

+            <NULL>

+            {

+                left = ConstantExpression.NULL;

+            }

+        )

+    )

+    {

+        return left;

+    }

+}

+

+String stringLitteral() :

+{

+    Token t;

+    StringBuffer rc = new StringBuffer();

+    boolean first=true;

+}

+{

+    t = <STRING_LITERAL>

+    {

+    	// Decode the sting value.

+    	String image = t.image;

+    	for( int i=1; i < image.length()-1; i++ ) {

+    		char c = image.charAt(i);

+    		if( c == '\'' )

+    			i++;

+   			rc.append(c);

+    	}

+	    return rc.toString();

+    }

+}

+

+PropertyExpression variable() :

+{

+    Token t;

+    PropertyExpression left=null;

+}

+{

+    (

+        t = <ID>

+        {

+            left = new PropertyExpression(t.image);

+        }

+    )

+    {

+        return left;

+    }

+}

diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java
index b0fbafac56..c5b45659cf 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java
@@ -26,6 +26,7 @@ import org.apache.qpid.framing.AMQDataBlock;
 import org.apache.qpid.framing.BasicPublishBody;
 import org.apache.qpid.framing.ContentBody;
 import org.apache.qpid.framing.ContentHeaderBody;
+import org.apache.qpid.framing.FieldTable;
 import org.apache.qpid.server.ack.TxAck;
 import org.apache.qpid.server.ack.UnacknowledgedMessage;
 import org.apache.qpid.server.ack.UnacknowledgedMessageMap;
@@ -45,6 +46,8 @@ import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
 import java.util.TreeMap;
+import java.util.Set;
+import java.util.HashSet;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
 
@@ -109,6 +112,7 @@ public class AMQChannel
     private TxAck ackOp;
 
     private final List<AMQDataBlock> _returns = new LinkedList<AMQDataBlock>();
+    private Set<Long> _browsedAcks = new HashSet<Long>();
 
     public AMQChannel(int channelId, MessageStore messageStore, MessageRouter exchanges)
             throws AMQException
@@ -285,12 +289,14 @@ public class AMQChannel
      * @param tag     the tag chosen by the client (if null, server will generate one)
      * @param queue   the queue to subscribe to
      * @param session the protocol session of the subscriber
+     * @param noLocal
      * @return the consumer tag. This is returned to the subscriber and used in
      *         subsequent unsubscribe requests
      * @throws ConsumerTagNotUniqueException if the tag is not unique
      * @throws AMQException                  if something goes wrong
      */
-    public String subscribeToQueue(String tag, AMQQueue queue, AMQProtocolSession session, boolean acks) throws AMQException, ConsumerTagNotUniqueException
+    public String subscribeToQueue(String tag, AMQQueue queue, AMQProtocolSession session, boolean acks,
+                                   FieldTable filters, boolean noLocal) throws AMQException, ConsumerTagNotUniqueException
     {
         if (tag == null)
         {
@@ -301,7 +307,7 @@ public class AMQChannel
             throw new ConsumerTagNotUniqueException();
         }
 
-        queue.registerProtocolSession(session, _channelId, tag, acks);
+        queue.registerProtocolSession(session, _channelId, tag, acks, filters, noLocal);
         _consumerTag2QueueMap.put(tag, queue);
         return tag;
     }
@@ -379,6 +385,8 @@ public class AMQChannel
         {
             if (unacked.queue != null)
             {
+                unacked.message.setTxnBuffer(null);
+
                 unacked.queue.deliver(unacked.message);
             }
         }
@@ -498,7 +506,7 @@ public class AMQChannel
         if (_log.isDebugEnabled())
         {
             _log.debug("Handling acknowledgement for channel " + _channelId + " with delivery tag " + deliveryTag +
-                      " and multiple " + multiple);
+                       " and multiple " + multiple);
         }
         if (multiple)
         {
@@ -550,7 +558,14 @@ public class AMQChannel
 
             for (UnacknowledgedMessage msg : acked)
             {
-                msg.discard();
+                if (!_browsedAcks.contains(deliveryTag))
+                {
+                    msg.discard();
+                }
+                else
+                {
+                    _browsedAcks.remove(deliveryTag);
+                }
             }
 
         }
@@ -567,7 +582,16 @@ public class AMQChannel
                 _log.trace("Single ack on delivery tag " + deliveryTag + " not known for channel:" + _channelId);
                 throw new AMQException("Single ack on delivery tag " + deliveryTag + " not known for channel:" + _channelId);
             }
-            msg.discard();
+
+            if (!_browsedAcks.contains(deliveryTag))
+            {
+                msg.discard();
+            }
+            else
+            {
+                _browsedAcks.remove(deliveryTag);
+            }
+
             if (_log.isTraceEnabled())
             {
                 _log.trace("Received non-multiple ack for messaging with delivery tag " + deliveryTag);
@@ -688,6 +712,12 @@ public class AMQChannel
         _returns.clear();
     }
 
+    public void addUnacknowledgedBrowsedMessage(AMQMessage msg, long deliveryTag, String consumerTag, AMQQueue queue)
+    {
+        _browsedAcks.add(deliveryTag);
+        addUnacknowledgedMessage(msg, deliveryTag, consumerTag, queue);
+    }
+
     //we use this wrapper to ensure we are always using the correct
     //map instance (its not final unfortunately)
     private class AckMap implements UnacknowledgedMessageMap
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/Main.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/Main.java
index 3470a5bb08..553aecc217 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/Main.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/Main.java
@@ -522,10 +522,10 @@ public class Main implements ProtocolVersionList
          */
         public void unregisterExchange(String exchangeName) throws JMException
         {
-            boolean inUse = false;
             // TODO
             // Check if the exchange is in use.
-            // Check if there are queue-bindings with the exchnage and unregister
+        	// boolean inUse = false;
+            // Check if there are queue-bindings with the exchange and unregister
             // when there are no bindings.
             try
             {
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/ArithmeticExpression.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/ArithmeticExpression.java
new file mode 100644
index 0000000000..0aa5739c1c
--- /dev/null
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/ArithmeticExpression.java
@@ -0,0 +1,216 @@
+/**
+ *
+ * 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.server.filter;
+//
+// Based on like named file from r450141 of the Apache ActiveMQ project <http://www.activemq.org/site/home.html>
+//
+
+
+import org.apache.qpid.server.queue.AMQMessage;
+
+import javax.jms.JMSException;
+
+/**
+ * An expression which performs an operation on two expression values
+ * 
+ * @version $Revision$
+ */
+public abstract class ArithmeticExpression extends BinaryExpression {
+
+    protected static final int INTEGER = 1;
+    protected static final int LONG = 2;
+    protected static final int DOUBLE = 3;
+
+    /**
+     * @param left
+     * @param right
+     */
+    public ArithmeticExpression(Expression left, Expression right) {
+        super(left, right);
+    }
+
+    public static Expression createPlus(Expression left, Expression right) {
+        return new ArithmeticExpression(left, right) {
+            protected Object evaluate(Object lvalue, Object rvalue) {
+                if (lvalue instanceof String) {
+                    String text = (String) lvalue;
+                    String answer = text + rvalue;
+                    return answer;
+                }
+                else if (lvalue instanceof Number) {
+                    return plus((Number) lvalue, asNumber(rvalue));
+                }
+                throw new RuntimeException("Cannot call plus operation on: " + lvalue + " and: " + rvalue);
+            }
+
+            public String getExpressionSymbol() {
+                return "+";
+            }
+        };
+    }
+
+    public static Expression createMinus(Expression left, Expression right) {
+        return new ArithmeticExpression(left, right) {
+            protected Object evaluate(Object lvalue, Object rvalue) {
+                if (lvalue instanceof Number) {
+                    return minus((Number) lvalue, asNumber(rvalue));
+                }
+                throw new RuntimeException("Cannot call minus operation on: " + lvalue + " and: " + rvalue);
+            }
+
+            public String getExpressionSymbol() {
+                return "-";
+            }
+        };
+    }
+
+    public static Expression createMultiply(Expression left, Expression right) {
+        return new ArithmeticExpression(left, right) {
+
+            protected Object evaluate(Object lvalue, Object rvalue) {
+                if (lvalue instanceof Number) {
+                    return multiply((Number) lvalue, asNumber(rvalue));
+                }
+                throw new RuntimeException("Cannot call multiply operation on: " + lvalue + " and: " + rvalue);
+            }
+
+            public String getExpressionSymbol() {
+                return "*";
+            }
+        };
+    }
+
+    public static Expression createDivide(Expression left, Expression right) {
+        return new ArithmeticExpression(left, right) {
+
+            protected Object evaluate(Object lvalue, Object rvalue) {
+                if (lvalue instanceof Number) {
+                    return divide((Number) lvalue, asNumber(rvalue));
+                }
+                throw new RuntimeException("Cannot call divide operation on: " + lvalue + " and: " + rvalue);
+            }
+
+            public String getExpressionSymbol() {
+                return "/";
+            }
+        };
+    }
+
+    public static Expression createMod(Expression left, Expression right) {
+        return new ArithmeticExpression(left, right) {
+
+            protected Object evaluate(Object lvalue, Object rvalue) {
+                if (lvalue instanceof Number) {
+                    return mod((Number) lvalue, asNumber(rvalue));
+                }
+                throw new RuntimeException("Cannot call mod operation on: " + lvalue + " and: " + rvalue);
+            }
+
+            public String getExpressionSymbol() {
+                return "%";
+            }
+        };
+    }
+
+    protected Number plus(Number left, Number right) {
+        switch (numberType(left, right)) {
+            case INTEGER:
+                return new Integer(left.intValue() + right.intValue());
+            case LONG:
+                return new Long(left.longValue() + right.longValue());
+            default:
+                return new Double(left.doubleValue() + right.doubleValue());
+        }
+    }
+
+    protected Number minus(Number left, Number right) {
+        switch (numberType(left, right)) {
+            case INTEGER:
+                return new Integer(left.intValue() - right.intValue());
+            case LONG:
+                return new Long(left.longValue() - right.longValue());
+            default:
+                return new Double(left.doubleValue() - right.doubleValue());
+        }
+    }
+
+    protected Number multiply(Number left, Number right) {
+        switch (numberType(left, right)) {
+            case INTEGER:
+                return new Integer(left.intValue() * right.intValue());
+            case LONG:
+                return new Long(left.longValue() * right.longValue());
+            default:
+                return new Double(left.doubleValue() * right.doubleValue());
+        }
+    }
+
+    protected Number divide(Number left, Number right) {
+        return new Double(left.doubleValue() / right.doubleValue());
+    }
+
+    protected Number mod(Number left, Number right) {
+        return new Double(left.doubleValue() % right.doubleValue());
+    }
+
+    private int numberType(Number left, Number right) {
+        if (isDouble(left) || isDouble(right)) {
+            return DOUBLE;
+        }
+        else if (left instanceof Long || right instanceof Long) {
+            return LONG;
+        }
+        else {
+            return INTEGER;
+        }
+    }
+
+    private boolean isDouble(Number n) {
+        return n instanceof Float || n instanceof Double;
+    }
+
+    protected Number asNumber(Object value) {
+        if (value instanceof Number) {
+            return (Number) value;
+        }
+        else {
+            throw new RuntimeException("Cannot convert value: " + value + " into a number");
+        }
+    }
+
+    public Object evaluate(AMQMessage message) throws JMSException {
+        Object lvalue = left.evaluate(message);
+        if (lvalue == null) {
+            return null;
+        }
+        Object rvalue = right.evaluate(message);
+        if (rvalue == null) {
+            return null;
+        }
+        return evaluate(lvalue, rvalue);
+    }
+
+
+    /**
+     * @param lvalue
+     * @param rvalue
+     * @return
+     */
+    abstract protected Object evaluate(Object lvalue, Object rvalue);
+
+}
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/BinaryExpression.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/BinaryExpression.java
new file mode 100644
index 0000000000..4256ab9189
--- /dev/null
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/BinaryExpression.java
@@ -0,0 +1,100 @@
+/**
+ *
+ * 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.server.filter;
+//
+// Based on like named file from r450141 of the Apache ActiveMQ project <http://www.activemq.org/site/home.html>
+//
+
+
+
+/**
+ * An expression which performs an operation on two expression values.
+ * 
+ * @version $Revision$
+ */
+abstract public class BinaryExpression implements Expression {
+    protected Expression left;
+    protected Expression right;
+
+    public BinaryExpression(Expression left, Expression right) {
+        this.left = left;
+        this.right = right;
+    }
+
+    public Expression getLeft() {
+        return left;
+    }
+
+    public Expression getRight() {
+        return right;
+    }
+
+
+    /**
+     * @see java.lang.Object#toString()
+     */
+    public String toString() {
+        return "(" + left.toString() + " " + getExpressionSymbol() + " " + right.toString() + ")";
+    }
+
+    /**
+     * TODO: more efficient hashCode()
+     *
+     * @see java.lang.Object#hashCode()
+     */
+    public int hashCode() {
+        return toString().hashCode();
+    }
+
+    /**
+     * TODO: more efficient hashCode()
+     *
+     * @see java.lang.Object#equals(java.lang.Object)
+     */
+    public boolean equals(Object o) {
+
+        if (o == null || !this.getClass().equals(o.getClass())) {
+            return false;
+        }
+        return toString().equals(o.toString());
+
+    }
+
+    /**
+     * Returns the symbol that represents this binary expression.  For example, addition is
+     * represented by "+"
+     *
+     * @return
+     */
+    abstract public String getExpressionSymbol();
+
+    /**
+     * @param expression
+     */
+    public void setRight(Expression expression) {
+        right = expression;
+    }
+
+    /**
+     * @param expression
+     */
+    public void setLeft(Expression expression) {
+        left = expression;
+    }
+    
+}
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/BooleanExpression.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/BooleanExpression.java
new file mode 100644
index 0000000000..b66de3fbc5
--- /dev/null
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/BooleanExpression.java
@@ -0,0 +1,45 @@
+/**
+ *
+ * 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.server.filter;
+//
+// Based on like named file from r450141 of the Apache ActiveMQ project <http://www.activemq.org/site/home.html>
+//
+
+
+import org.apache.qpid.server.queue.AMQMessage;
+import org.apache.qpid.server.message.jms.JMSMessage;
+
+import javax.jms.JMSException;
+
+
+/**
+ * A BooleanExpression is an expression that always
+ * produces a Boolean result.
+ *
+ * @version $Revision$
+ */
+public interface BooleanExpression extends Expression {
+    
+    /**
+     * @param message
+     * @return true if the expression evaluates to Boolean.TRUE.
+     * @throws JMSException
+     */
+    public boolean matches(AMQMessage message) throws JMSException;
+
+}
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/ComparisonExpression.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/ComparisonExpression.java
new file mode 100644
index 0000000000..13d278cf65
--- /dev/null
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/ComparisonExpression.java
@@ -0,0 +1,465 @@
+/**
+ *
+ * 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.server.filter;
+//
+// Based on like named file from r450141 of the Apache ActiveMQ project <http://www.activemq.org/site/home.html>
+//
+
+import org.apache.qpid.server.queue.AMQMessage;
+import org.apache.qpid.server.message.jms.JMSMessage;
+
+import java.util.HashSet;
+import java.util.List;
+import java.util.regex.Pattern;
+
+import javax.jms.JMSException;
+
+/**
+ * A filter performing a comparison of two objects
+ * 
+ * @version $Revision$
+ */
+public abstract class ComparisonExpression extends BinaryExpression implements BooleanExpression {
+
+    public static BooleanExpression createBetween(Expression value, Expression left, Expression right) {
+        return LogicExpression.createAND(createGreaterThanEqual(value, left), createLessThanEqual(value, right));
+    }
+
+    public static BooleanExpression createNotBetween(Expression value, Expression left, Expression right) {
+        return LogicExpression.createOR(createLessThan(value, left), createGreaterThan(value, right));
+    }
+
+    static final private HashSet REGEXP_CONTROL_CHARS = new HashSet();
+
+    static {
+        REGEXP_CONTROL_CHARS.add(new Character('.'));
+        REGEXP_CONTROL_CHARS.add(new Character('\\'));
+        REGEXP_CONTROL_CHARS.add(new Character('['));
+        REGEXP_CONTROL_CHARS.add(new Character(']'));
+        REGEXP_CONTROL_CHARS.add(new Character('^'));
+        REGEXP_CONTROL_CHARS.add(new Character('$'));
+        REGEXP_CONTROL_CHARS.add(new Character('?'));
+        REGEXP_CONTROL_CHARS.add(new Character('*'));
+        REGEXP_CONTROL_CHARS.add(new Character('+'));
+        REGEXP_CONTROL_CHARS.add(new Character('{'));
+        REGEXP_CONTROL_CHARS.add(new Character('}'));
+        REGEXP_CONTROL_CHARS.add(new Character('|'));
+        REGEXP_CONTROL_CHARS.add(new Character('('));
+        REGEXP_CONTROL_CHARS.add(new Character(')'));
+        REGEXP_CONTROL_CHARS.add(new Character(':'));
+        REGEXP_CONTROL_CHARS.add(new Character('&'));
+        REGEXP_CONTROL_CHARS.add(new Character('<'));
+        REGEXP_CONTROL_CHARS.add(new Character('>'));
+        REGEXP_CONTROL_CHARS.add(new Character('='));
+        REGEXP_CONTROL_CHARS.add(new Character('!'));
+    }
+
+    static class LikeExpression extends UnaryExpression implements BooleanExpression {
+
+        Pattern likePattern;
+
+        /**
+         * @param right
+         */
+        public LikeExpression(Expression right, String like, int escape) {
+            super(right);
+
+            StringBuffer regexp = new StringBuffer(like.length() * 2);
+            regexp.append("\\A"); // The beginning of the input
+            for (int i = 0; i < like.length(); i++) {
+                char c = like.charAt(i);
+                if (escape == (0xFFFF & c)) {
+                    i++;
+                    if (i >= like.length()) {
+                        // nothing left to escape...
+                        break;
+                    }
+
+                    char t = like.charAt(i);
+                    regexp.append("\\x");
+                    regexp.append(Integer.toHexString(0xFFFF & t));
+                }
+                else if (c == '%') {
+                    regexp.append(".*?"); // Do a non-greedy match 
+                }
+                else if (c == '_') {
+                    regexp.append("."); // match one 
+                }
+                else if (REGEXP_CONTROL_CHARS.contains(new Character(c))) {
+                    regexp.append("\\x");
+                    regexp.append(Integer.toHexString(0xFFFF & c));
+                }
+                else {
+                    regexp.append(c);
+                }
+            }
+            regexp.append("\\z"); // The end of the input
+
+            likePattern = Pattern.compile(regexp.toString(), Pattern.DOTALL);
+        }
+
+        /**
+         *  org.apache.activemq.filter.UnaryExpression#getExpressionSymbol()
+         */
+        public String getExpressionSymbol() {
+            return "LIKE";
+        }
+
+        /**
+         *  org.apache.activemq.filter.Expression#evaluate(MessageEvaluationContext)
+         */
+        public Object evaluate(AMQMessage message) throws JMSException {
+
+            Object rv = this.getRight().evaluate(message);
+
+            if (rv == null) {
+                return null;
+            }
+
+            if (!(rv instanceof String)) {
+            	return Boolean.FALSE;
+                //throw new RuntimeException("LIKE can only operate on String identifiers.  LIKE attemped on: '" + rv.getClass());
+            }
+
+            return likePattern.matcher((String) rv).matches() ? Boolean.TRUE : Boolean.FALSE;
+        }
+        
+        public boolean matches(AMQMessage message) throws JMSException {
+            Object object = evaluate(message);
+            return object!=null && object==Boolean.TRUE;            
+        }
+    }
+
+    public static BooleanExpression createLike(Expression left, String right, String escape) {
+        if (escape != null && escape.length() != 1) {
+            throw new RuntimeException("The ESCAPE string litteral is invalid.  It can only be one character.  Litteral used: " + escape);
+        }
+        int c = -1;
+        if (escape != null) {
+            c = 0xFFFF & escape.charAt(0);
+        }
+
+        return new LikeExpression(left, right, c);
+    }
+
+    public static BooleanExpression createNotLike(Expression left, String right, String escape) {
+        return UnaryExpression.createNOT(createLike(left, right, escape));
+    }    
+
+    public static BooleanExpression createInFilter(Expression left, List elements) {
+    	
+    	if( !(left instanceof PropertyExpression) )
+    		throw new RuntimeException("Expected a property for In expression, got: "+left);    	
+    	return UnaryExpression.createInExpression((PropertyExpression)left, elements, false);
+    	
+    }
+
+    public static BooleanExpression createNotInFilter(Expression left, List elements) {
+    	
+    	if( !(left instanceof PropertyExpression) )
+    		throw new RuntimeException("Expected a property for In expression, got: "+left);    	
+    	return UnaryExpression.createInExpression((PropertyExpression)left, elements, true);
+
+    }
+
+    public static BooleanExpression createIsNull(Expression left) {
+        return doCreateEqual(left, ConstantExpression.NULL);
+    }
+
+    public static BooleanExpression createIsNotNull(Expression left) {
+        return UnaryExpression.createNOT(doCreateEqual(left, ConstantExpression.NULL));
+    }
+
+    public static BooleanExpression createNotEqual(Expression left, Expression right) {
+        return UnaryExpression.createNOT(createEqual(left, right));
+    }
+
+    public static BooleanExpression createEqual(Expression left, Expression right) {
+    	checkEqualOperand(left);
+    	checkEqualOperand(right);
+    	checkEqualOperandCompatability(left, right);
+    	return doCreateEqual(left, right);
+    }
+    
+	private static BooleanExpression doCreateEqual(Expression left, Expression right) {
+        return new ComparisonExpression(left, right) {
+
+            public Object evaluate(AMQMessage message) throws JMSException {
+                Object lv = left.evaluate(message);
+                Object rv = right.evaluate(message);
+                
+                // Iff one of the values is null
+                if (lv == null ^ rv == null) {
+                    return Boolean.FALSE;
+                }
+                if (lv == rv || lv.equals(rv)) {
+                    return Boolean.TRUE;
+                }                
+                if( lv instanceof Comparable && rv instanceof Comparable ) {
+                    return compare((Comparable)lv, (Comparable)rv);
+                }
+                return Boolean.FALSE;
+            }
+
+            protected boolean asBoolean(int answer) {
+                return answer == 0;
+            }
+
+            public String getExpressionSymbol() {
+                return "=";
+            }
+        };
+    }
+
+    public static BooleanExpression createGreaterThan(final Expression left, final Expression right) {
+    	checkLessThanOperand(left);
+    	checkLessThanOperand(right);
+        return new ComparisonExpression(left, right) {
+            protected boolean asBoolean(int answer) {
+                return answer > 0;
+            }
+
+            public String getExpressionSymbol() {
+                return ">";
+            }
+        };
+    }
+
+    public static BooleanExpression createGreaterThanEqual(final Expression left, final Expression right) {
+    	checkLessThanOperand(left);
+    	checkLessThanOperand(right);
+        return new ComparisonExpression(left, right) {
+            protected boolean asBoolean(int answer) {
+                return answer >= 0;
+            }
+
+            public String getExpressionSymbol() {
+                return ">=";
+            }
+        };
+    }
+
+    public static BooleanExpression createLessThan(final Expression left, final Expression right) {
+    	checkLessThanOperand(left);
+    	checkLessThanOperand(right);
+        return new ComparisonExpression(left, right) {
+
+            protected boolean asBoolean(int answer) {
+                return answer < 0;
+            }
+
+            public String getExpressionSymbol() {
+                return "<";
+            }
+
+        };
+    }
+
+	public static BooleanExpression createLessThanEqual(final Expression left, final Expression right) {
+    	checkLessThanOperand(left);
+    	checkLessThanOperand(right);
+        return new ComparisonExpression(left, right) {
+
+            protected boolean asBoolean(int answer) {
+                return answer <= 0;
+            }
+
+            public String getExpressionSymbol() {
+                return "<=";
+            }
+        };
+    }
+
+	/**
+     * Only Numeric expressions can be used in >, >=, < or <= expressions.s 
+     * 
+	 * @param expr
+	 */
+	public static void checkLessThanOperand(Expression expr ) {
+		if( expr instanceof ConstantExpression ) {
+			Object value = ((ConstantExpression)expr).getValue();
+			if( value instanceof Number )
+				return;
+			
+			// Else it's boolean or a String..  
+			throw new RuntimeException("Value '"+expr+"' cannot be compared.");
+		}
+		if( expr instanceof BooleanExpression ) {
+			throw new RuntimeException("Value '"+expr+"' cannot be compared.");
+		}		
+	}
+
+	/**
+     * Validates that the expression can be used in == or <> expression.  
+     * Cannot not be NULL TRUE or FALSE litterals.
+     * 
+	 * @param expr
+	 */
+	public static void checkEqualOperand(Expression expr ) {
+		if( expr instanceof ConstantExpression ) {
+			Object value = ((ConstantExpression)expr).getValue();
+			if( value == null )
+				throw new RuntimeException("'"+expr+"' cannot be compared.");
+		}
+	}
+
+	/**
+	 * 
+	 * @param left
+	 * @param right
+	 */
+	private static void checkEqualOperandCompatability(Expression left, Expression right) {
+		if( left instanceof ConstantExpression && right instanceof ConstantExpression ) {
+			if( left instanceof BooleanExpression && !(right instanceof BooleanExpression) )
+				throw new RuntimeException("'"+left+"' cannot be compared with '"+right+"'");
+		}
+	}
+
+	
+	
+    /**
+     * @param left
+     * @param right
+     */
+    public ComparisonExpression(Expression left, Expression right) {
+        super(left, right);
+    }
+
+    public Object evaluate(AMQMessage message) throws JMSException {
+        Comparable lv = (Comparable) left.evaluate(message);
+        if (lv == null) {
+            return null;
+        }
+        Comparable rv = (Comparable) right.evaluate(message);
+        if (rv == null) {
+            return null;
+        }
+        return compare(lv, rv);
+    }
+
+    protected Boolean compare(Comparable lv, Comparable rv) {
+        Class lc = lv.getClass();
+        Class rc = rv.getClass();
+        // If the the objects are not of the same type,
+        // try to convert up to allow the comparison.
+        if (lc != rc) {
+            if (lc == Byte.class) {
+                if (rc == Short.class) {
+                    lv = new Short(((Number) lv).shortValue());
+                }
+                else if (rc == Integer.class) {
+                    lv = new Integer(((Number) lv).intValue());
+                }
+                else if (rc == Long.class) {
+                    lv = new Long(((Number) lv).longValue());
+                }
+                else if (rc == Float.class) {
+                    lv = new Float(((Number) lv).floatValue());
+                }
+                else if (rc == Double.class) {
+                    lv = new Double(((Number) lv).doubleValue());
+                }
+                else {
+                    return Boolean.FALSE;
+                }
+             } else if (lc == Short.class) {
+                if (rc == Integer.class) {
+                    lv = new Integer(((Number) lv).intValue());
+                }
+                else if (rc == Long.class) {
+                    lv = new Long(((Number) lv).longValue());
+                }
+                else if (rc == Float.class) {
+                    lv = new Float(((Number) lv).floatValue());
+                }
+                else if (rc == Double.class) {
+                    lv = new Double(((Number) lv).doubleValue());
+                }
+                else {
+                    return Boolean.FALSE;
+                }
+            } else if (lc == Integer.class) {
+                if (rc == Long.class) {
+                    lv = new Long(((Number) lv).longValue());
+                }
+                else if (rc == Float.class) {
+                    lv = new Float(((Number) lv).floatValue());
+                }
+                else if (rc == Double.class) {
+                    lv = new Double(((Number) lv).doubleValue());
+                }
+                else {
+                    return Boolean.FALSE;
+                }
+            }
+            else if (lc == Long.class) {
+                if (rc == Integer.class) {
+                    rv = new Long(((Number) rv).longValue());
+                }
+                else if (rc == Float.class) {
+                    lv = new Float(((Number) lv).floatValue());
+                }
+                else if (rc == Double.class) {
+                    lv = new Double(((Number) lv).doubleValue());
+                }
+                else {
+                    return Boolean.FALSE;
+                }
+            }
+            else if (lc == Float.class) {
+                if (rc == Integer.class) {
+                    rv = new Float(((Number) rv).floatValue());
+                }
+                else if (rc == Long.class) {
+                    rv = new Float(((Number) rv).floatValue());
+                }
+                else if (rc == Double.class) {
+                    lv = new Double(((Number) lv).doubleValue());
+                }
+                else {
+                    return Boolean.FALSE;
+                }
+            } 
+            else if (lc == Double.class) {
+                if (rc == Integer.class) {
+                    rv = new Double(((Number) rv).doubleValue());
+                }
+                else if (rc == Long.class) {
+                    rv = new Double(((Number) rv).doubleValue());
+                }
+                else if (rc == Float.class) {
+                	rv = new Float(((Number) rv).doubleValue());
+                }
+                else {
+                    return Boolean.FALSE;
+                }
+            } 
+            else 
+                return Boolean.FALSE;
+        }
+        return asBoolean(lv.compareTo(rv)) ? Boolean.TRUE : Boolean.FALSE;
+    }
+
+    protected abstract boolean asBoolean(int answer);
+    
+    public boolean matches(AMQMessage message) throws JMSException {
+        Object object = evaluate(message);
+        return object!=null && object==Boolean.TRUE;            
+    }
+
+}
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/ConstantExpression.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/ConstantExpression.java
new file mode 100644
index 0000000000..9bde712da2
--- /dev/null
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/ConstantExpression.java
@@ -0,0 +1,170 @@
+/**
+ *
+ * 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.server.filter;
+//
+// Based on like named file from r450141 of the Apache ActiveMQ project <http://www.activemq.org/site/home.html>
+//
+
+import org.apache.qpid.server.queue.AMQMessage;
+import org.apache.qpid.server.message.jms.JMSMessage;
+
+import java.math.BigDecimal;
+
+import javax.jms.JMSException;
+
+/**
+ * Represents a constant expression
+ * 
+ * @version $Revision$
+ */
+public class ConstantExpression implements Expression {
+
+    static class BooleanConstantExpression extends ConstantExpression implements BooleanExpression {
+        public BooleanConstantExpression(Object value) {
+            super(value);
+        }
+        public boolean matches(AMQMessage message) throws JMSException {
+            Object object = evaluate(message);
+            return object!=null && object==Boolean.TRUE;            
+        }        
+    }
+
+    public static final BooleanConstantExpression NULL = new BooleanConstantExpression(null);
+    public static final BooleanConstantExpression TRUE = new BooleanConstantExpression(Boolean.TRUE);
+    public static final BooleanConstantExpression FALSE = new BooleanConstantExpression(Boolean.FALSE);
+
+    private Object value;
+
+    public static ConstantExpression createFromDecimal(String text) {
+    	    	
+    	// Strip off the 'l' or 'L' if needed.
+    	if( text.endsWith("l") || text.endsWith("L") )
+    		text = text.substring(0, text.length()-1);
+
+    	Number value;
+    	try {
+    		value = new Long(text);
+    	} catch ( NumberFormatException e) {
+    		// The number may be too big to fit in a long.
+        	value = new BigDecimal(text);    		
+    	}
+    	
+        long l = value.longValue();
+        if (Integer.MIN_VALUE <= l && l <= Integer.MAX_VALUE) {
+            value = new Integer(value.intValue());
+        }
+        return new ConstantExpression(value);
+    }
+
+    public static ConstantExpression createFromHex(String text) {
+        Number value = new Long(Long.parseLong(text.substring(2), 16));
+        long l = value.longValue();
+        if (Integer.MIN_VALUE <= l && l <= Integer.MAX_VALUE) {
+            value = new Integer(value.intValue());
+        }
+        return new ConstantExpression(value);
+    }
+
+    public static ConstantExpression createFromOctal(String text) {
+        Number value = new Long(Long.parseLong(text, 8));
+        long l = value.longValue();
+        if (Integer.MIN_VALUE <= l && l <= Integer.MAX_VALUE) {
+            value = new Integer(value.intValue());
+        }
+        return new ConstantExpression(value);
+    }
+
+    public static ConstantExpression createFloat(String text) {
+        Number value = new Double(text);
+        return new ConstantExpression(value);
+    }
+
+    public ConstantExpression(Object value) {
+        this.value = value;
+    }
+
+    public Object evaluate(AMQMessage message) throws JMSException {
+        return value;
+    }
+
+    public Object getValue() {
+        return value;
+    }    
+
+    /**
+     * @see java.lang.Object#toString()
+     */
+    public String toString() {
+        if (value == null) {
+            return "NULL";
+        }
+        if (value instanceof Boolean) {
+            return ((Boolean) value).booleanValue() ? "TRUE" : "FALSE";
+        }
+        if (value instanceof String) {
+            return encodeString((String) value);
+        }
+        return value.toString();
+    }
+
+    /**
+     * TODO: more efficient hashCode()
+     *
+     * @see java.lang.Object#hashCode()
+     */
+    public int hashCode() {
+        return toString().hashCode();
+    }
+
+    /**
+     * TODO: more efficient hashCode()
+     *
+     * @see java.lang.Object#equals(java.lang.Object)
+     */
+    public boolean equals(Object o) {
+
+        if (o == null || !this.getClass().equals(o.getClass())) {
+            return false;
+        }
+        return toString().equals(o.toString());
+
+    }
+
+
+    /**
+     * Encodes the value of string so that it looks like it would look like
+     * when it was provided in a selector.
+     *
+     * @param s
+     * @return
+     */
+    public static String encodeString(String s) {
+        StringBuffer b = new StringBuffer();
+        b.append('\'');
+        for (int i = 0; i < s.length(); i++) {
+            char c = s.charAt(i);
+            if (c == '\'') {
+                b.append(c);
+            }
+            b.append(c);
+        }
+        b.append('\'');
+        return b.toString();
+    }
+    
+}
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/Expression.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/Expression.java
new file mode 100644
index 0000000000..a15c15fb91
--- /dev/null
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/Expression.java
@@ -0,0 +1,42 @@
+/**
+ *
+ * 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.server.filter;
+//
+// Based on like named file from r450141 of the Apache ActiveMQ project <http://www.activemq.org/site/home.html>
+//
+
+import org.apache.qpid.server.queue.AMQMessage;
+import org.apache.qpid.server.message.jms.JMSMessage;
+
+import javax.jms.JMSException;
+
+
+/**
+ * Represents an expression
+ * 
+ * @version $Revision$
+ */
+public interface Expression {
+
+    /**
+     * @return the value of this expression
+     */
+    public Object evaluate(AMQMessage message) throws JMSException;
+    
+}
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/FilterManager.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/FilterManager.java
new file mode 100644
index 0000000000..c82de9fa15
--- /dev/null
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/FilterManager.java
@@ -0,0 +1,37 @@
+/*
+ *  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.server.filter;
+//
+// Based on like named file from r450141 of the Apache ActiveMQ project <http://www.activemq.org/site/home.html>
+//
+
+import org.apache.qpid.server.queue.AMQMessage;
+
+public interface FilterManager
+{
+    void add(MessageFilter filter);
+
+    void remove(MessageFilter filter);
+
+    boolean allAllow(AMQMessage msg);
+
+    boolean hasFilters();
+}
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/FilterManagerFactory.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/FilterManagerFactory.java
new file mode 100644
index 0000000000..49f99132ef
--- /dev/null
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/FilterManagerFactory.java
@@ -0,0 +1,86 @@
+/*
+ *  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.server.filter;
+
+import org.apache.qpid.framing.FieldTable;
+import org.apache.qpid.AMQException;
+import org.apache.qpid.common.AMQPFilterTypes;
+//import org.slf4j.Logger;
+//import org.slf4j.LoggerFactory;
+
+import java.util.Iterator;
+
+
+public class FilterManagerFactory
+{
+    //private final static Logger _logger = LoggerFactory.getLogger(FilterManagerFactory.class);
+    private final static org.apache.log4j.Logger _logger = org.apache.log4j.Logger.getLogger(FilterManagerFactory.class);
+
+    //fixme move to a common class so it can be refered to from client code.
+
+    public static FilterManager createManager(FieldTable filters) throws AMQException
+    {
+        FilterManager manager = null;
+
+        if (filters != null)
+        {
+
+            manager = new SimpleFilterManager();
+
+            Iterator it = filters.keySet().iterator();
+            _logger.info("Processing filters:");
+            while (it.hasNext())
+            {
+                String key = (String) it.next();
+                _logger.info("filter:" + key);
+                if (key.equals(AMQPFilterTypes.JMS_SELECTOR.getValue()))
+                {
+                    String selector = (String) filters.get(key);
+
+                    if (selector != null && !selector.equals(""))
+                    {
+                        manager.add(new JMSSelectorFilter(selector));
+                    }
+                }
+
+                if (key.equals(AMQPFilterTypes.NO_CONSUME.getValue()))
+                {
+                    manager.add(new NoConsumerFilter());
+                }
+
+            }
+
+            //If we added no filters don't bear the overhead of having an filter manager
+            if (!manager.hasFilters())
+            {
+                manager = null;
+            }
+        }
+        else
+        {
+            _logger.info("No Filters found.");
+        }
+
+
+        return manager;
+
+    }
+}
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/JMSSelectorFilter.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/JMSSelectorFilter.java
new file mode 100644
index 0000000000..4884067237
--- /dev/null
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/JMSSelectorFilter.java
@@ -0,0 +1,79 @@
+/*
+ *  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.server.filter;
+
+import org.apache.qpid.server.queue.AMQMessage;
+import org.apache.qpid.server.filter.jms.selector.SelectorParser;
+import org.apache.qpid.AMQException;
+import org.apache.qpid.AMQInvalidSelectorException;
+import org.apache.log4j.Logger;
+
+
+import javax.jms.InvalidSelectorException;
+import javax.jms.JMSException;
+
+public class JMSSelectorFilter implements MessageFilter
+{
+    private final static Logger _logger = org.apache.log4j.Logger.getLogger(JMSSelectorFilter.class);
+
+    private String _selector;
+    private BooleanExpression _matcher;
+
+    public JMSSelectorFilter(String selector) throws AMQException
+    {
+        _selector = selector;
+        _logger.info("Created JMSSelectorFilter with selector:" + _selector);
+
+
+        try
+        {
+            _matcher = new SelectorParser().parse(selector);
+        }
+        catch (InvalidSelectorException e)
+        {
+            // fixme
+            // Is this the correct way of throwing exception
+            throw new AMQInvalidSelectorException(e.getMessage());
+        }
+
+    }
+
+    public boolean matches(AMQMessage message)
+    {
+        try
+        {
+            boolean match = _matcher.matches(message);
+            _logger.info(message + " match(" + match + ") selector(" + System.identityHashCode(_selector) + "):" + _selector);
+            return match;
+        }
+        catch (JMSException e)
+        {
+            //fixme this needs to be sorted.. it shouldn't happen
+            e.printStackTrace();  //To change body of catch statement use File | Settings | File Templates.
+        }
+        return false;
+    }
+
+    public String getSelector()
+    {
+        return _selector;
+    }
+}
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/LogicExpression.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/LogicExpression.java
new file mode 100644
index 0000000000..714d8c23f5
--- /dev/null
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/LogicExpression.java
@@ -0,0 +1,95 @@
+/**
+ *
+ * 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.server.filter;
+//
+// Based on like named file from r450141 of the Apache ActiveMQ project <http://www.activemq.org/site/home.html>
+//
+
+import org.apache.qpid.server.queue.AMQMessage;
+import org.apache.qpid.server.message.jms.JMSMessage;
+
+import javax.jms.JMSException;
+
+/**
+ * A filter performing a comparison of two objects
+ * 
+ * @version $Revision$
+ */
+public abstract class LogicExpression extends BinaryExpression implements BooleanExpression {
+
+    public static BooleanExpression createOR(BooleanExpression lvalue, BooleanExpression rvalue) {
+        return new LogicExpression(lvalue, rvalue) {
+        	
+            public Object evaluate(AMQMessage message) throws JMSException {
+                
+            	Boolean lv = (Boolean) left.evaluate(message);
+                // Can we do an OR shortcut??
+            	if (lv !=null && lv.booleanValue()) {
+                    return Boolean.TRUE;
+                }
+            	
+                Boolean rv = (Boolean) right.evaluate(message);
+                return rv==null ? null : rv;
+            }
+
+            public String getExpressionSymbol() {
+                return "OR";
+            }
+        };
+    }
+
+    public static BooleanExpression createAND(BooleanExpression lvalue, BooleanExpression rvalue) {
+        return new LogicExpression(lvalue, rvalue) {
+
+            public Object evaluate(AMQMessage message) throws JMSException {
+
+                Boolean lv = (Boolean) left.evaluate(message);
+
+                // Can we do an AND shortcut??
+                if (lv == null)
+                    return null;
+                if (!lv.booleanValue()) {
+                    return Boolean.FALSE;
+                }
+
+                Boolean rv = (Boolean) right.evaluate(message);
+                return rv == null ? null : rv;
+            }
+
+            public String getExpressionSymbol() {
+                return "AND";
+            }
+        };
+    }
+
+    /**
+     * @param left
+     * @param right
+     */
+    public LogicExpression(BooleanExpression left, BooleanExpression right) {
+        super(left, right);
+    }
+
+    abstract public Object evaluate(AMQMessage message) throws JMSException;
+
+    public boolean matches(AMQMessage message) throws JMSException {
+        Object object = evaluate(message);
+        return object!=null && object==Boolean.TRUE;            
+    }
+
+}
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/MessageFilter.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/MessageFilter.java
new file mode 100644
index 0000000000..b8ca75d209
--- /dev/null
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/MessageFilter.java
@@ -0,0 +1,30 @@
+/*
+ *  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.server.filter;
+
+import org.apache.qpid.server.queue.AMQMessage;
+
+import javax.jms.JMSException;
+
+public interface MessageFilter
+{
+    boolean matches(AMQMessage message) throws JMSException;
+}
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/NoConsumerFilter.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/NoConsumerFilter.java
new file mode 100644
index 0000000000..283d324ff6
--- /dev/null
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/NoConsumerFilter.java
@@ -0,0 +1,48 @@
+/*
+ *  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.server.filter;
+
+import org.apache.qpid.server.queue.AMQMessage;
+import org.apache.qpid.server.filter.jms.selector.SelectorParser;
+import org.apache.qpid.AMQException;
+import org.apache.qpid.AMQInvalidSelectorException;
+import org.apache.log4j.Logger;
+
+
+import javax.jms.InvalidSelectorException;
+import javax.jms.JMSException;
+
+public class NoConsumerFilter implements MessageFilter
+{
+    private final static Logger _logger = org.apache.log4j.Logger.getLogger(NoConsumerFilter.class);
+
+
+    public NoConsumerFilter() throws AMQException
+    {        
+        _logger.info("Created NoConsumerFilter");
+    }
+
+    public boolean matches(AMQMessage message)
+    {
+       return true;
+    }
+
+}
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/PropertyExpression.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/PropertyExpression.java
new file mode 100644
index 0000000000..f3e9965c2e
--- /dev/null
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/PropertyExpression.java
@@ -0,0 +1,305 @@
+/**
+ *
+ * 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.server.filter;
+//
+// Based on like named file from r450141 of the Apache ActiveMQ project <http://www.activemq.org/site/home.html>
+//
+         
+import java.io.IOException;
+import java.util.HashMap;
+
+import javax.jms.DeliveryMode;
+import javax.jms.JMSException;
+
+//import org.apache.activemq.command.ActiveMQDestination;
+//import org.apache.activemq.command.Message;
+//import org.apache.activemq.command.TransactionId;
+//import org.apache.activemq.util.JMSExceptionSupport;
+import org.apache.qpid.server.queue.AMQMessage;
+import org.apache.qpid.server.message.jms.JMSMessage;
+import org.apache.qpid.AMQException;
+import org.apache.qpid.framing.BasicContentHeaderProperties;
+import org.apache.log4j.Logger;
+
+/**
+ * Represents a property  expression
+ *
+ * @version $Revision$
+ */
+public class PropertyExpression implements Expression
+{
+
+    interface SubExpression
+    {
+        public Object evaluate(AMQMessage message);
+    }
+
+    interface JMSExpression
+    {
+        public abstract Object evaluate(JMSMessage message);
+    }
+
+    static class SubJMSExpression implements SubExpression
+    {
+        JMSExpression _expression;
+
+        SubJMSExpression(JMSExpression expression)
+        {
+            _expression = expression;
+        }
+
+
+        public Object evaluate(AMQMessage message)
+        {
+            JMSMessage msg = (JMSMessage) message.getDecodedMessage(AMQMessage.JMS_MESSAGE);
+            if (msg != null)
+            {
+                return _expression.evaluate(msg);
+            }
+            else
+            {
+                return null;
+            }
+        }
+    }
+
+    private final static Logger _logger = org.apache.log4j.Logger.getLogger(PropertyExpression.class);
+
+
+    static final private HashMap JMS_PROPERTY_EXPRESSIONS = new HashMap();
+
+    static
+    {
+        JMS_PROPERTY_EXPRESSIONS.put("JMSDestination", new SubJMSExpression(
+                new JMSExpression()
+                {
+                    public Object evaluate(JMSMessage message)
+                    {
+                        return message.getJMSDestination();
+                    }
+                }
+        ));
+//
+//            public Object evaluate(AMQMessage message)
+//            {
+//                //fixme
+//
+//
+////                AMQDestination dest = message.getOriginalDestination();
+////                if (dest == null)
+////                {
+////                    dest = message.getDestination();
+////                }
+////                if (dest == null)
+////                {
+////                    return null;
+////                }
+////                return dest.toString();
+//                return "";
+//            }
+//        });
+        JMS_PROPERTY_EXPRESSIONS.put("JMSReplyTo", new SubJMSExpression(
+                new JMSExpression()
+                {
+                    public Object evaluate(JMSMessage message)
+                    {
+                        return message.getJMSReplyTo();
+                    }
+                })
+        );
+
+        JMS_PROPERTY_EXPRESSIONS.put("JMSType", new SubJMSExpression(
+                new JMSExpression()
+                {
+                    public Object evaluate(JMSMessage message)
+                    {
+                        return message.getJMSType();
+                    }
+                }
+        ));
+
+        JMS_PROPERTY_EXPRESSIONS.put("JMSDeliveryMode", new SubJMSExpression(
+                new JMSExpression()
+                {
+                    public Object evaluate(JMSMessage message)
+                    {
+                        try
+                        {
+                            Integer mode = new Integer(message.getAMQMessage().isPersistent() ? DeliveryMode.PERSISTENT : DeliveryMode.NON_PERSISTENT);
+                            _logger.info("JMSDeliveryMode is :" + mode);
+                            return mode;
+                        }
+                        catch (AMQException e)
+                        {
+                            //shouldn't happen
+                        }
+
+                        return DeliveryMode.NON_PERSISTENT;
+                    }
+                }));
+
+        JMS_PROPERTY_EXPRESSIONS.put("JMSPriority", new SubJMSExpression(
+                new JMSExpression()
+                {
+                    public Object evaluate(JMSMessage message)
+                    {
+                        return message.getJMSPriority();
+                    }
+                }
+        ));
+
+
+        JMS_PROPERTY_EXPRESSIONS.put("AMQMessageID", new SubJMSExpression(
+                new JMSExpression()
+                {
+                    public Object evaluate(JMSMessage message)
+                    {
+                        return message.getAMQMessage().getMessageId();
+                    }
+                }
+        ));
+
+        JMS_PROPERTY_EXPRESSIONS.put("JMSTimestamp", new SubJMSExpression(
+                new JMSExpression()
+                {
+                    public Object evaluate(JMSMessage message)
+                    {
+                        return message.getJMSTimestamp();
+                    }
+                }
+        ));
+
+        JMS_PROPERTY_EXPRESSIONS.put("JMSCorrelationID", new SubJMSExpression(
+                new JMSExpression()
+                {
+                    public Object evaluate(JMSMessage message)
+                    {
+                        return message.getJMSCorrelationID();
+                    }
+                }
+        ));
+
+        JMS_PROPERTY_EXPRESSIONS.put("JMSExpiration", new SubJMSExpression(
+                new JMSExpression()
+                {
+                    public Object evaluate(JMSMessage message)
+                    {
+                        return message.getJMSExpiration();
+                    }
+                }
+        ));
+
+        JMS_PROPERTY_EXPRESSIONS.put("JMSRedelivered", new SubJMSExpression(
+                new JMSExpression()
+                {
+                    public Object evaluate(JMSMessage message)
+                    {
+                        return message.getAMQMessage().isRedelivered();
+                    }
+                }
+        ));
+
+    }
+
+    private final String name;
+    private final SubExpression jmsPropertyExpression;
+
+    public PropertyExpression(String name)
+    {
+        this.name = name;
+        jmsPropertyExpression = (SubExpression) JMS_PROPERTY_EXPRESSIONS.get(name);
+    }
+
+    public Object evaluate(AMQMessage message) throws JMSException
+    {
+//        try
+//        {
+//            if (message.isDropped())
+//            {
+//                return null;
+//            }
+
+        if (jmsPropertyExpression != null)
+        {
+            return jmsPropertyExpression.evaluate(message);
+        }
+//            try
+        else
+        {
+
+            BasicContentHeaderProperties _properties = (BasicContentHeaderProperties) message.getContentHeaderBody().properties;
+
+            _logger.info("Looking up property:" + name);
+            _logger.info("Properties are:" + _properties.getHeaders().keySet());
+
+            return _properties.getHeaders().get(name);
+        }
+//            catch (IOException ioe)
+//            {
+//                JMSException exception = new JMSException("Could not get property: " + name + " reason: " + ioe.getMessage());
+//                exception.initCause(ioe);
+//                throw exception;
+//            }
+//        }
+//        catch (IOException e)
+//        {
+//            JMSException exception = new JMSException(e.getMessage());
+//            exception.initCause(e);
+//            throw exception;
+//        }
+
+    }
+
+    public String getName()
+    {
+        return name;
+    }
+
+
+    /**
+     * @see java.lang.Object#toString()
+     */
+    public String toString()
+    {
+        return name;
+    }
+
+    /**
+     * @see java.lang.Object#hashCode()
+     */
+    public int hashCode()
+    {
+        return name.hashCode();
+    }
+
+    /**
+     * @see java.lang.Object#equals(java.lang.Object)
+     */
+    public boolean equals(Object o)
+    {
+
+        if (o == null || !this.getClass().equals(o.getClass()))
+        {
+            return false;
+        }
+        return name.equals(((PropertyExpression) o).name);
+
+    }
+
+}
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/SimpleFilterManager.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/SimpleFilterManager.java
new file mode 100644
index 0000000000..dc2c2c0e6c
--- /dev/null
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/SimpleFilterManager.java
@@ -0,0 +1,77 @@
+/*
+ *  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.server.filter;
+
+import org.apache.qpid.server.queue.AMQMessage;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.jms.JMSException;
+import java.util.concurrent.ConcurrentLinkedQueue;
+
+public class SimpleFilterManager implements FilterManager
+{
+    private final Logger _logger = LoggerFactory.getLogger(SimpleFilterManager.class);
+
+    private final ConcurrentLinkedQueue<MessageFilter> _filters;
+
+    public SimpleFilterManager()
+    {
+        _logger.debug("Creating SimpleFilterManager");
+        _filters = new ConcurrentLinkedQueue<MessageFilter>();
+    }
+
+    public void add(MessageFilter filter)
+    {
+        _filters.add(filter);
+    }
+
+    public void remove(MessageFilter filter)
+    {
+        _filters.remove(filter);
+    }
+
+    public boolean allAllow(AMQMessage msg)
+    {
+        for (MessageFilter filter : _filters)
+        {
+            try
+            {
+                if (!filter.matches(msg))
+                {
+                    return false;
+                }
+            }
+            catch (JMSException e)
+            {
+                //fixme
+                e.printStackTrace();  //To change body of catch statement use File | Settings | File Templates.
+                return false;
+            }
+        }
+        return true;
+    }
+
+    public boolean hasFilters()
+    {
+        return !_filters.isEmpty();
+    }
+}
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/UnaryExpression.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/UnaryExpression.java
new file mode 100644
index 0000000000..49ff147411
--- /dev/null
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/UnaryExpression.java
@@ -0,0 +1,263 @@
+/**
+ *
+ * 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.server.filter;
+//
+// Based on like named file from r450141 of the Apache ActiveMQ project <http://www.activemq.org/site/home.html>
+//
+
+import org.apache.qpid.server.queue.AMQMessage;
+import org.apache.qpid.server.message.jms.JMSMessage;
+
+import java.math.BigDecimal;
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+
+import javax.jms.JMSException;
+
+/**
+ * An expression which performs an operation on two expression values
+ * 
+ * @version $Revision$
+ */
+public abstract class UnaryExpression implements Expression {
+
+    private static final BigDecimal BD_LONG_MIN_VALUE = BigDecimal.valueOf(Long.MIN_VALUE);
+    protected Expression right;
+
+    public static Expression createNegate(Expression left) {
+        return new UnaryExpression(left) {
+            public Object evaluate(AMQMessage message) throws JMSException {
+                Object rvalue = right.evaluate(message);
+                if (rvalue == null) {
+                    return null;
+                }
+                if (rvalue instanceof Number) {
+                    return negate((Number) rvalue);
+                }
+                return null;
+            }
+
+            public String getExpressionSymbol() {
+                return "-";
+            }
+        };
+    }
+
+    public static BooleanExpression createInExpression(PropertyExpression right, List elements, final boolean not) {
+    	
+    	// Use a HashSet if there are many elements.
+    	Collection t;
+		if( elements.size()==0 )
+    		t=null;
+    	else if( elements.size() < 5 )
+    		t = elements;
+    	else {
+    		t = new HashSet(elements);
+    	}
+    	final Collection inList = t;
+    	
+        return new BooleanUnaryExpression(right) {
+            public Object evaluate(AMQMessage message) throws JMSException {
+            	
+                Object rvalue = right.evaluate(message);
+                if (rvalue == null) {
+                    return null;
+                }
+                if( rvalue.getClass()!=String.class )
+                	return null;
+                
+                if( (inList!=null && inList.contains(rvalue)) ^ not ) {
+                	return Boolean.TRUE;
+                } else {
+                	return Boolean.FALSE;                	
+                }
+                
+            }
+
+            public String toString() {
+            	StringBuffer answer = new StringBuffer();
+            	answer.append(right);
+            	answer.append(" ");
+            	answer.append(getExpressionSymbol());
+            	answer.append(" ( ");
+
+            	int count=0;
+            	for (Iterator i = inList.iterator(); i.hasNext();) {
+					Object o = (Object) i.next();
+					if( count!=0 ) {
+		            	answer.append(", ");				
+					}
+	            	answer.append(o);				
+	            	count++;
+				}
+            	
+            	answer.append(" )");				
+                return answer.toString();
+			}
+			
+            public String getExpressionSymbol() {
+            	if( not )
+            		return "NOT IN";
+            	else 
+            		return "IN";
+            }
+        };
+    }
+
+    abstract static class BooleanUnaryExpression extends UnaryExpression implements BooleanExpression {
+        public BooleanUnaryExpression(Expression left) {        	
+            super(left);
+        }
+
+        public boolean matches(AMQMessage message) throws JMSException {
+            Object object = evaluate(message);
+            return object!=null && object==Boolean.TRUE;            
+        }
+    };
+
+        
+    public static BooleanExpression createNOT(BooleanExpression left) {
+        return new BooleanUnaryExpression(left) {
+            public Object evaluate(AMQMessage message) throws JMSException {
+                Boolean lvalue = (Boolean) right.evaluate(message);
+                if (lvalue == null) {
+                    return null;
+                }
+                return lvalue.booleanValue() ? Boolean.FALSE : Boolean.TRUE;
+            }
+
+            public String getExpressionSymbol() {
+                return "NOT";
+            }
+        };
+    }
+    
+    public static BooleanExpression createXPath(final String xpath) {
+        return new XPathExpression(xpath);
+    }
+
+    public static BooleanExpression createXQuery(final String xpath) {
+        return new XQueryExpression(xpath);
+    }
+
+    public static BooleanExpression createBooleanCast(Expression left) {
+        return new BooleanUnaryExpression(left) {
+            public Object evaluate(AMQMessage message) throws JMSException {
+                Object rvalue = right.evaluate(message);
+                if (rvalue == null) 
+                    return null;
+                if (!rvalue.getClass().equals(Boolean.class)) 
+                    return Boolean.FALSE;
+                return ((Boolean)rvalue).booleanValue() ? Boolean.TRUE : Boolean.FALSE;
+            }
+
+            public String toString() {
+                return right.toString();
+            }
+
+            public String getExpressionSymbol() {
+                return "";
+            }
+        };
+    }
+
+    private static Number negate(Number left) {
+    	Class clazz = left.getClass();
+        if (clazz == Integer.class) {
+            return new Integer(-left.intValue());
+        }
+        else if (clazz == Long.class) {
+            return new Long(-left.longValue());
+        }
+        else if (clazz ==  Float.class) {
+            return new Float(-left.floatValue());
+        }
+        else if (clazz == Double.class) {
+            return new Double(-left.doubleValue());
+        }
+        else if (clazz == BigDecimal.class) {
+        	// We ussually get a big deciamal when we have Long.MIN_VALUE constant in the 
+        	// Selector.  Long.MIN_VALUE is too big to store in a Long as a positive so we store it 
+        	// as a Big decimal.  But it gets Negated right away.. to here we try to covert it back
+        	// to a Long.        	
+        	BigDecimal bd = (BigDecimal)left;
+        	bd = bd.negate();
+        	
+        	if( BD_LONG_MIN_VALUE.compareTo(bd)==0  ) {
+        		return new Long(Long.MIN_VALUE);
+        	}
+            return bd;
+        }
+        else {
+            throw new RuntimeException("Don't know how to negate: "+left);
+        }
+    }
+
+    public UnaryExpression(Expression left) {
+        this.right = left;
+    }
+
+    public Expression getRight() {
+        return right;
+    }
+
+    public void setRight(Expression expression) {
+        right = expression;
+    }
+
+    /**
+     * @see java.lang.Object#toString()
+     */
+    public String toString() {
+        return "(" + getExpressionSymbol() + " " + right.toString() + ")";
+    }
+
+    /**
+     * TODO: more efficient hashCode()
+     *
+     * @see java.lang.Object#hashCode()
+     */
+    public int hashCode() {
+        return toString().hashCode();
+    }
+
+    /**
+     * TODO: more efficient hashCode()
+     *
+     * @see java.lang.Object#equals(java.lang.Object)
+     */
+    public boolean equals(Object o) {
+
+        if (o == null || !this.getClass().equals(o.getClass())) {
+            return false;
+        }
+        return toString().equals(o.toString());
+
+    }
+
+    /**
+     * Returns the symbol that represents this binary expression.  For example, addition is
+     * represented by "+"
+     *
+     * @return
+     */
+    abstract public String getExpressionSymbol();
+
+}
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/XPathExpression.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/XPathExpression.java
new file mode 100644
index 0000000000..ab952b6fea
--- /dev/null
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/XPathExpression.java
@@ -0,0 +1,130 @@
+/**
+ *
+ * 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.server.filter;
+//
+// Based on like named file from r450141 of the Apache ActiveMQ project <http://www.activemq.org/site/home.html>
+//
+
+import java.io.IOException;
+import java.lang.reflect.Constructor;
+import java.lang.reflect.InvocationTargetException;
+
+import javax.jms.JMSException;
+
+//import org.apache.activemq.command.Message;
+//import org.apache.activemq.util.JMSExceptionSupport;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.qpid.server.queue.AMQMessage;
+
+/**
+ * Used to evaluate an XPath Expression in a JMS selector.
+ */
+public final class XPathExpression implements BooleanExpression {
+
+    private static final Log log = LogFactory.getLog(XPathExpression.class);
+    private static final String EVALUATOR_SYSTEM_PROPERTY = "org.apache.qpid.server.filter.XPathEvaluatorClassName";
+    private static final String DEFAULT_EVALUATOR_CLASS_NAME=XalanXPathEvaluator.class.getName();
+    
+    private static final Constructor EVALUATOR_CONSTRUCTOR;
+    
+    static {
+        String cn = System.getProperty(EVALUATOR_SYSTEM_PROPERTY, DEFAULT_EVALUATOR_CLASS_NAME);        
+        Constructor m = null;
+        try {
+            try {
+                m = getXPathEvaluatorConstructor(cn);
+            } catch (Throwable e) {
+                log.warn("Invalid "+XPathEvaluator.class.getName()+" implementation: "+cn+", reason: "+e,e);
+                cn = DEFAULT_EVALUATOR_CLASS_NAME;
+                try {
+                    m = getXPathEvaluatorConstructor(cn);
+                } catch (Throwable e2) {
+                    log.error("Default XPath evaluator could not be loaded",e);
+                }
+            }
+        } finally {
+            EVALUATOR_CONSTRUCTOR = m;
+        }
+    }
+
+    private static Constructor getXPathEvaluatorConstructor(String cn) throws ClassNotFoundException, SecurityException, NoSuchMethodException {
+        Class c = XPathExpression.class.getClassLoader().loadClass(cn);
+        if( !XPathEvaluator.class.isAssignableFrom(c) ) {
+            throw new ClassCastException(""+c+" is not an instance of "+XPathEvaluator.class);
+        }
+        return c.getConstructor(new Class[]{String.class});
+    }
+    
+    private final String xpath;
+    private final XPathEvaluator evaluator;
+    
+    static public interface XPathEvaluator {
+        public boolean evaluate(AMQMessage message) throws JMSException;
+    }    
+    
+    XPathExpression(String xpath) {
+        this.xpath = xpath;
+        this.evaluator = createEvaluator(xpath);
+    }
+
+    private XPathEvaluator createEvaluator(String xpath2) {
+        try {
+            return (XPathEvaluator)EVALUATOR_CONSTRUCTOR.newInstance(new Object[]{xpath});
+        } catch (InvocationTargetException e) {
+            Throwable cause = e.getCause();
+            if( cause instanceof RuntimeException ) {
+                throw (RuntimeException)cause;
+            }
+            throw new RuntimeException("Invalid XPath Expression: "+xpath+" reason: "+e.getMessage(), e);
+        } catch (Throwable e) {
+            throw new RuntimeException("Invalid XPath Expression: "+xpath+" reason: "+e.getMessage(), e);
+        }
+    }
+
+    public Object evaluate(AMQMessage message) throws JMSException {
+//        try {
+//FIXME this is flow to disk work
+//            if( message.isDropped() )
+//                return null;
+            return evaluator.evaluate(message) ? Boolean.TRUE : Boolean.FALSE;
+//        } catch (IOException e) {
+//
+//            JMSException exception = new JMSException(e.getMessage());
+//            exception.initCause(e);
+//            throw exception;
+//
+//        }
+
+    }
+
+    public String toString() {
+        return "XPATH "+ConstantExpression.encodeString(xpath);
+    }
+    
+    /**
+     * @param message
+     * @return true if the expression evaluates to Boolean.TRUE.
+     * @throws JMSException
+     */
+    public boolean matches(AMQMessage message) throws JMSException {
+        Object object = evaluate(message);
+        return object!=null && object==Boolean.TRUE;            
+    }
+
+}
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/XQueryExpression.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/XQueryExpression.java
new file mode 100644
index 0000000000..53764cbf75
--- /dev/null
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/XQueryExpression.java
@@ -0,0 +1,56 @@
+/**
+ *
+ * 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.server.filter;
+
+import org.apache.qpid.server.queue.AMQMessage;
+
+import javax.jms.JMSException;
+//
+// Based on like named file from r450141 of the Apache ActiveMQ project <http://www.activemq.org/site/home.html>
+//
+
+/**
+ * Used to evaluate an XQuery Expression in a JMS selector.
+ */
+public final class XQueryExpression implements BooleanExpression {
+    private final String xpath;
+
+    XQueryExpression(String xpath) {
+        super();
+        this.xpath = xpath;
+    }
+
+    public Object evaluate(AMQMessage message) throws JMSException {
+        return Boolean.FALSE;
+    }
+
+    public String toString() {
+        return "XQUERY "+ConstantExpression.encodeString(xpath);
+    }
+    
+    /**
+     * @param message
+     * @return true if the expression evaluates to Boolean.TRUE.
+     * @throws JMSException
+     */
+    public boolean matches(AMQMessage message) throws JMSException {
+        Object object = evaluate(message);
+        return object!=null && object==Boolean.TRUE;            
+    }
+
+}
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/XalanXPathEvaluator.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/XalanXPathEvaluator.java
new file mode 100644
index 0000000000..4b78fd18df
--- /dev/null
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/XalanXPathEvaluator.java
@@ -0,0 +1,99 @@
+/**
+ *
+ * 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.server.filter;
+//
+// Based on like named file from r450141 of the Apache ActiveMQ project <http://www.activemq.org/site/home.html>
+//
+
+import java.io.StringReader;
+import java.io.ByteArrayInputStream;
+
+import javax.jms.BytesMessage;
+import javax.jms.JMSException;
+import javax.jms.TextMessage;
+import javax.xml.parsers.DocumentBuilder;
+import javax.xml.parsers.DocumentBuilderFactory;
+
+//import org.apache.activemq.command.Message;
+//import org.apache.activemq.util.ByteArrayInputStream;
+import org.apache.xpath.CachedXPathAPI;
+import org.apache.qpid.server.queue.AMQMessage;
+import org.w3c.dom.Document;
+import org.w3c.dom.traversal.NodeIterator;
+import org.xml.sax.InputSource;
+
+public class XalanXPathEvaluator implements XPathExpression.XPathEvaluator {
+    
+    private final String xpath;
+
+    public XalanXPathEvaluator(String xpath) {
+        this.xpath = xpath;
+    }
+    
+    public boolean evaluate(AMQMessage m) throws JMSException {
+        if( m instanceof TextMessage ) {
+            String text = ((TextMessage)m).getText();
+            return evaluate(text);                
+        } else if ( m instanceof BytesMessage ) {
+            BytesMessage bm = (BytesMessage) m;
+            byte data[] = new byte[(int) bm.getBodyLength()];
+            bm.readBytes(data);
+            return evaluate(data);
+        }            
+        return false;
+    }
+
+    private boolean evaluate(byte[] data) {
+        try {
+            
+            InputSource inputSource = new InputSource(new ByteArrayInputStream(data));
+            
+            DocumentBuilderFactory factory = DocumentBuilderFactory.newInstance();
+            factory.setNamespaceAware(true);
+            DocumentBuilder dbuilder = factory.newDocumentBuilder();
+            Document doc = dbuilder.parse(inputSource);
+            
+            CachedXPathAPI cachedXPathAPI = new CachedXPathAPI();
+            NodeIterator iterator = cachedXPathAPI.selectNodeIterator(doc,xpath);
+            return iterator.nextNode()!=null;
+            
+        } catch (Throwable e) {
+            return false;
+        }
+    }
+
+    private boolean evaluate(String text) {
+        try {
+            InputSource inputSource = new InputSource(new StringReader(text));
+            
+            DocumentBuilderFactory factory = DocumentBuilderFactory.newInstance();
+            factory.setNamespaceAware(true);
+            DocumentBuilder dbuilder = factory.newDocumentBuilder();
+            Document doc = dbuilder.parse(inputSource);
+            
+            // We should associated the cachedXPathAPI object with the message being evaluated
+            // since that should speedup subsequent xpath expressions.
+            CachedXPathAPI cachedXPathAPI = new CachedXPathAPI();
+            NodeIterator iterator = cachedXPathAPI.selectNodeIterator(doc,xpath);
+            return iterator.nextNode()!=null;
+        } catch (Throwable e) {
+            return false;
+        }
+    }
+}
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/BasicConsumeMethodHandler.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/BasicConsumeMethodHandler.java
index d4c94061a0..1e57c714ff 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/BasicConsumeMethodHandler.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/BasicConsumeMethodHandler.java
@@ -21,10 +21,12 @@
 package org.apache.qpid.server.handler;
 
 import org.apache.qpid.AMQException;
+import org.apache.qpid.AMQInvalidSelectorException;
 import org.apache.qpid.protocol.AMQConstant;
 import org.apache.qpid.framing.BasicConsumeBody;
 import org.apache.qpid.framing.BasicConsumeOkBody;
 import org.apache.qpid.framing.ConnectionCloseBody;
+import org.apache.qpid.framing.ChannelCloseBody;
 import org.apache.qpid.server.AMQChannel;
 import org.apache.qpid.server.ConsumerTagNotUniqueException;
 import org.apache.qpid.server.exchange.ExchangeRegistry;
@@ -32,6 +34,7 @@ import org.apache.qpid.server.protocol.AMQMethodEvent;
 import org.apache.qpid.server.protocol.AMQProtocolSession;
 import org.apache.qpid.server.queue.QueueRegistry;
 import org.apache.qpid.server.queue.AMQQueue;
+import org.apache.qpid.server.queue.AMQMessage;
 import org.apache.qpid.server.state.AMQStateManager;
 import org.apache.qpid.server.state.StateAwareMethodListener;
 import org.apache.log4j.Logger;
@@ -68,14 +71,15 @@ public class BasicConsumeMethodHandler implements StateAwareMethodListener<Basic
         {
             AMQQueue queue = body.queue == null ? channel.getDefaultQueue() : queueRegistry.getQueue(body.queue);
 
-            if(queue == null)
+            if (queue == null)
             {
                 _log.info("No queue for '" + body.queue + "'");
             }
             try
             {
-                String consumerTag = channel.subscribeToQueue(body.consumerTag, queue,  session, !body.noAck);
-                if(!body.nowait)
+                String consumerTag = channel.subscribeToQueue(body.consumerTag, queue, session, !body.noAck,
+                                                              body.arguments, body.noLocal);
+                if (!body.nowait)
                 {
                     session.writeFrame(BasicConsumeOkBody.createAMQFrame(channelId, consumerTag));
                 }
@@ -83,10 +87,19 @@ public class BasicConsumeMethodHandler implements StateAwareMethodListener<Basic
                 //now allow queue to start async processing of any backlog of messages
                 queue.deliverAsync();
             }
-            catch(ConsumerTagNotUniqueException e)
+            catch (AMQInvalidSelectorException ise)
+            {
+                _log.info("Closing connection due to invalid selector");
+                session.writeFrame(ChannelCloseBody.createAMQFrame(channelId, AMQConstant.INVALID_SELECTOR.getCode(),
+                                                                   ise.getMessage(), BasicConsumeBody.CLASS_ID,
+                                                                   BasicConsumeBody.METHOD_ID));
+            }
+            catch (ConsumerTagNotUniqueException e)
             {
                 String msg = "Non-unique consumer tag, '" + body.consumerTag + "'";
-                session.writeFrame(ConnectionCloseBody.createAMQFrame(channelId, AMQConstant.NOT_ALLOWED.getCode(), msg, BasicConsumeBody.CLASS_ID, BasicConsumeBody.METHOD_ID));
+                session.writeFrame(ConnectionCloseBody.createAMQFrame(channelId, AMQConstant.NOT_ALLOWED.getCode(), msg,
+                                                                      BasicConsumeBody.CLASS_ID,
+                                                                      BasicConsumeBody.METHOD_ID));
             }
         }
     }
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionStartOkMethodHandler.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionStartOkMethodHandler.java
index 00ae547683..79b2e11bca 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionStartOkMethodHandler.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionStartOkMethodHandler.java
@@ -78,12 +78,19 @@ public class ConnectionStartOkMethodHandler implements StateAwareMethodListener<
 
             AuthenticationResult authResult = authMgr.authenticate(ss, body.response);
 
+            //save clientProperties
+            if (protocolSession.getClientProperties() == null)
+            {
+                protocolSession.setClientProperties(body.clientProperties);
+            }
+
             switch (authResult.status)
             {
                 case ERROR:
                     throw new AMQException("Authentication failed");
                 case SUCCESS:
                     _logger.info("Connected as: " + ss.getAuthorizationID());
+
                     stateManager.changeState(AMQState.CONNECTION_NOT_TUNED);
                     AMQFrame tune = ConnectionTuneBody.createAMQFrame(0, Integer.MAX_VALUE, getConfiguredFrameSize(),
                                                                       HeartbeatConfig.getInstance().getDelay());
@@ -122,7 +129,7 @@ public class ConnectionStartOkMethodHandler implements StateAwareMethodListener<
     static int getConfiguredFrameSize()
     {
         final Configuration config = ApplicationRegistry.getInstance().getConfiguration();
-        final int framesize =  config.getInt("advanced.framesize", DEFAULT_FRAME_SIZE);
+        final int framesize = config.getInt("advanced.framesize", DEFAULT_FRAME_SIZE);
         _logger.info("Framesize set to " + framesize);
         return framesize;
     }
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/management/MBeanIntrospector.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/management/MBeanIntrospector.java
index b8235a0808..535122efb5 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/management/MBeanIntrospector.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/management/MBeanIntrospector.java
@@ -61,7 +61,6 @@ class MBeanIntrospector {
          */
         for (Method method : interfaceClass.getMethods())
         {
-            int    argCount = method.getParameterTypes().length;
             String name = method.getName();
             Class<?>  resultType = method.getReturnType();
             MBeanAttributeInfo attributeInfo = null;
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/message/MessageDecorator.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/message/MessageDecorator.java
new file mode 100644
index 0000000000..aba3b88a59
--- /dev/null
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/message/MessageDecorator.java
@@ -0,0 +1,25 @@
+/*
+ *  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.server.message;
+
+public interface MessageDecorator
+{
+}
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/message/jms/JMSMessage.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/message/jms/JMSMessage.java
new file mode 100644
index 0000000000..72e241ea0a
--- /dev/null
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/message/jms/JMSMessage.java
@@ -0,0 +1,306 @@
+/*
+ *  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.server.message.jms;
+
+import org.apache.qpid.server.message.MessageDecorator;
+import org.apache.qpid.server.queue.AMQMessage;
+import org.apache.qpid.framing.BasicContentHeaderProperties;
+import org.apache.qpid.framing.ContentHeaderBody;
+
+import javax.jms.Message;
+import javax.jms.JMSException;
+import javax.jms.Destination;
+import javax.jms.MessageNotWriteableException;
+import java.util.Enumeration;
+
+public class JMSMessage implements MessageDecorator
+{
+
+    private AMQMessage _message;
+    private BasicContentHeaderProperties _properties;
+
+    public JMSMessage(AMQMessage message)
+    {
+        _message = message;
+        ContentHeaderBody contentHeader = message.getContentHeaderBody();
+        _properties = (BasicContentHeaderProperties) contentHeader.properties;
+    }
+
+    protected void checkWriteable() throws MessageNotWriteableException
+    {
+        //The broker should not modify a message.
+//        if (_readableMessage)
+        {
+          throw new MessageNotWriteableException("The broker should not modify a message.");
+        }
+    }
+
+
+    public String getJMSMessageID()
+    {
+        return _properties.getMessageId();
+    }
+
+    public void setJMSMessageID(String string) throws MessageNotWriteableException
+    {
+        checkWriteable();
+        _properties.setMessageId(string);
+    }
+
+    public long getJMSTimestamp()
+    {
+        return _properties.getTimestamp();
+    }
+
+    public void setJMSTimestamp(long l) throws MessageNotWriteableException
+    {
+        checkWriteable();
+        _properties.setTimestamp(l);
+    }
+
+    public byte[] getJMSCorrelationIDAsBytes()
+    {
+        return _properties.getCorrelationId().getBytes();
+    }
+
+//    public void setJMSCorrelationIDAsBytes(byte[] bytes)
+//    {
+//    }
+
+    public void setJMSCorrelationID(String string) throws MessageNotWriteableException
+    {
+        checkWriteable();
+        _properties.setCorrelationId(string);
+    }
+
+    public String getJMSCorrelationID()
+    {
+        return _properties.getCorrelationId();
+    }
+
+    public String getJMSReplyTo()
+    {
+        return _properties.getReplyTo();
+    }
+
+    public void setJMSReplyTo(Destination destination) throws MessageNotWriteableException
+    {
+        checkWriteable();
+        _properties.setReplyTo(destination.toString());
+    }
+
+    public String getJMSDestination()
+    {
+        //fixme should be a deestination
+        return "";
+    }
+
+    public void setJMSDestination(Destination destination) throws MessageNotWriteableException
+    {
+        checkWriteable();
+        //_properties.setDestination(destination.toString());
+    }
+
+    public int getJMSDeliveryMode()
+    {
+        return _properties.getDeliveryMode();
+    }
+
+    public void setJMSDeliveryMode(byte i) throws MessageNotWriteableException
+    {
+        checkWriteable();
+        _properties.setDeliveryMode(i);
+    }
+
+    public boolean getJMSRedelivered()
+    {
+        return _message.isRedelivered();
+    }
+
+    public void setJMSRedelivered(boolean b) throws MessageNotWriteableException
+    {
+        checkWriteable();
+        _message.setRedelivered(b);
+    }
+
+    public String getJMSType()
+    {
+        return _properties.getType();
+    }
+
+    public void setJMSType(String string) throws MessageNotWriteableException
+    {
+        checkWriteable();
+        _properties.setType(string);
+    }
+
+    public long getJMSExpiration()
+    {
+        return _properties.getExpiration();
+    }
+
+    public void setJMSExpiration(long l) throws MessageNotWriteableException
+    {
+        checkWriteable();
+        _properties.setExpiration(l);
+    }
+
+    public int getJMSPriority()
+    {
+        return _properties.getPriority();
+    }
+
+    public void setJMSPriority(byte i) throws MessageNotWriteableException
+    {
+        checkWriteable();
+        _properties.setPriority(i);
+    }
+
+    public void clearProperties() throws MessageNotWriteableException
+    {
+        checkWriteable();
+        _properties.getJMSHeaders().clear();
+    }
+
+    public boolean propertyExists(String string)
+    {
+        return _properties.getJMSHeaders().propertyExists(string);
+    }
+
+    public boolean getBooleanProperty(String string) throws JMSException
+    {
+        return _properties.getJMSHeaders().getBoolean(string);
+    }
+
+    public byte getByteProperty(String string) throws JMSException
+    {
+        return _properties.getJMSHeaders().getByte(string);
+    }
+
+    public short getShortProperty(String string) throws JMSException
+    {
+        return _properties.getJMSHeaders().getShort(string);
+    }
+
+    public int getIntProperty(String string) throws JMSException
+    {
+        return _properties.getJMSHeaders().getInteger(string);
+    }
+
+    public long getLongProperty(String string) throws JMSException
+    {
+        return _properties.getJMSHeaders().getLong(string);
+    }
+
+    public float getFloatProperty(String string) throws JMSException
+    {
+        return _properties.getJMSHeaders().getFloat(string);
+    }
+
+    public double getDoubleProperty(String string) throws JMSException
+    {
+        return _properties.getJMSHeaders().getDouble(string);
+    }
+
+    public String getStringProperty(String string) throws JMSException
+    {
+        return _properties.getJMSHeaders().getString(string);
+    }
+
+    public Object getObjectProperty(String string) throws JMSException
+    {
+        return _properties.getJMSHeaders().getObject(string);
+    }
+
+    public Enumeration getPropertyNames()
+    {
+        return _properties.getJMSHeaders().getPropertyNames();
+    }
+
+    public void setBooleanProperty(String string, boolean b) throws JMSException
+    {
+        checkWriteable();
+        _properties.getJMSHeaders().setBoolean(string, b);
+    }
+
+    public void setByteProperty(String string, byte b) throws JMSException
+    {
+        checkWriteable();
+        _properties.getJMSHeaders().setByte(string, b);
+    }
+
+    public void setShortProperty(String string, short i) throws JMSException
+    {
+        checkWriteable();
+        _properties.getJMSHeaders().setShort(string, i);
+    }
+
+    public void setIntProperty(String string, int i) throws JMSException
+    {
+        checkWriteable();
+        _properties.getJMSHeaders().setInteger(string, i);
+    }
+
+    public void setLongProperty(String string, long l) throws JMSException
+    {
+        checkWriteable();
+        _properties.getJMSHeaders().setLong(string, l);
+    }
+
+    public void setFloatProperty(String string, float v) throws JMSException
+    {
+        checkWriteable();
+        _properties.getJMSHeaders().setFloat(string, v);
+    }
+
+    public void setDoubleProperty(String string, double v) throws JMSException
+    {
+        checkWriteable();
+        _properties.getJMSHeaders().setDouble(string, v);
+    }
+
+    public void setStringProperty(String string, String string1) throws JMSException
+    {
+        checkWriteable();
+        _properties.getJMSHeaders().setString(string, string1);
+    }
+
+    public void setObjectProperty(String string, Object object) throws JMSException
+    {
+        checkWriteable();
+        _properties.getJMSHeaders().setObject(string, object);
+    }
+
+    public void acknowledge() throws MessageNotWriteableException
+    {
+        checkWriteable();
+    }
+
+    public void clearBody() throws MessageNotWriteableException
+    {
+        checkWriteable();
+    }
+
+    public AMQMessage getAMQMessage()
+    {
+        return _message;
+    }
+}
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQMinaProtocolSession.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQMinaProtocolSession.java
index c38f7f630b..6ba78ba722 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQMinaProtocolSession.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQMinaProtocolSession.java
@@ -26,17 +26,19 @@ import org.apache.mina.common.IoSession;
 import org.apache.mina.transport.vmpipe.VmPipeAddress;
 import org.apache.qpid.AMQChannelException;
 import org.apache.qpid.AMQException;
-import org.apache.qpid.codec.AMQCodecFactory;
-import org.apache.qpid.codec.AMQDecoder;
 import org.apache.qpid.framing.AMQDataBlock;
+import org.apache.qpid.framing.ProtocolInitiation;
+import org.apache.qpid.framing.ConnectionStartBody;
 import org.apache.qpid.framing.AMQFrame;
+import org.apache.qpid.framing.FieldTable;
+import org.apache.qpid.framing.ProtocolVersionList;
 import org.apache.qpid.framing.AMQMethodBody;
-import org.apache.qpid.framing.ConnectionStartBody;
 import org.apache.qpid.framing.ContentBody;
-import org.apache.qpid.framing.ContentHeaderBody;
 import org.apache.qpid.framing.HeartbeatBody;
-import org.apache.qpid.framing.ProtocolInitiation;
-import org.apache.qpid.framing.ProtocolVersionList;
+import org.apache.qpid.framing.ContentHeaderBody;
+import org.apache.qpid.codec.AMQCodecFactory;
+import org.apache.qpid.codec.AMQDecoder;
+
 import org.apache.qpid.server.AMQChannel;
 import org.apache.qpid.server.RequiredDeliveryException;
 import org.apache.qpid.server.exchange.ExchangeRegistry;
@@ -89,10 +91,11 @@ public class AMQMinaProtocolSession implements AMQProtocolSession,
     private boolean _closed;
     // maximum number of channels this session should have
     private long _maxNoOfChannels = 1000;
-    
+
     /* AMQP Version for this session */
     private byte _major;
     private byte _minor;
+    private FieldTable _clientProperties;
 
     public ManagedObject getManagedObject()
     {
@@ -128,7 +131,7 @@ public class AMQMinaProtocolSession implements AMQProtocolSession,
         {
             return new AMQProtocolSessionMBean(this);
         }
-        catch(JMException ex)
+        catch (JMException ex)
         {
             _logger.error("AMQProtocolSession MBean creation has failed ", ex);
             throw new AMQException("AMQProtocolSession MBean creation has failed ", ex);
@@ -153,18 +156,21 @@ public class AMQMinaProtocolSession implements AMQProtocolSession,
         {
             ProtocolInitiation pi = (ProtocolInitiation) message;
             // this ensures the codec never checks for a PI message again
-            ((AMQDecoder)_codecFactory.getDecoder()).setExpectProtocolInitiation(false);
-            try {
+            ((AMQDecoder) _codecFactory.getDecoder()).setExpectProtocolInitiation(false);
+            try
+            {
                 pi.checkVersion(this); // Fails if not correct
                 // This sets the protocol version (and hence framing classes) for this session.
                 _major = pi.protocolMajor;
                 _minor = pi.protocolMinor;
                 String mechanisms = ApplicationRegistry.getInstance().getAuthenticationManager().getMechanisms();
                 String locales = "en_US";
-                AMQFrame response = ConnectionStartBody.createAMQFrame((short)0, pi.protocolMajor, pi.protocolMinor, null,
+                AMQFrame response = ConnectionStartBody.createAMQFrame((short) 0, pi.protocolMajor, pi.protocolMinor, null,
                                                                        mechanisms.getBytes(), locales.getBytes());
                 _minaProtocolSession.write(response);
-            } catch (AMQException e) {
+            }
+            catch (AMQException e)
+            {
                 _logger.error("Received incorrect protocol initiation", e);
                 /* Find last protocol version in protocol version list. Make sure last protocol version
                 listed in the build file (build-module.xml) is the latest version which will be used
@@ -211,7 +217,7 @@ public class AMQMinaProtocolSession implements AMQProtocolSession,
             _logger.debug("Method frame received: " + frame);
         }
         final AMQMethodEvent<AMQMethodBody> evt = new AMQMethodEvent<AMQMethodBody>(frame.channel,
-                                                                                    (AMQMethodBody)frame.bodyFrame);
+                                                                                    (AMQMethodBody) frame.bodyFrame);
         try
         {
             boolean wasAnyoneInterested = false;
@@ -266,7 +272,7 @@ public class AMQMinaProtocolSession implements AMQProtocolSession,
         {
             _logger.debug("Content header frame received: " + frame);
         }
-        getChannel(frame.channel).publishContentHeader((ContentHeaderBody)frame.bodyFrame);
+        getChannel(frame.channel).publishContentHeader((ContentHeaderBody) frame.bodyFrame);
     }
 
     private void contentBodyReceived(AMQFrame frame) throws AMQException
@@ -275,7 +281,7 @@ public class AMQMinaProtocolSession implements AMQProtocolSession,
         {
             _logger.debug("Content body frame received: " + frame);
         }
-        getChannel(frame.channel).publishContentBody((ContentBody)frame.bodyFrame);
+        getChannel(frame.channel).publishContentBody((ContentBody) frame.bodyFrame);
     }
 
     /**
@@ -355,6 +361,7 @@ public class AMQMinaProtocolSession implements AMQProtocolSession,
      * Close a specific channel. This will remove any resources used by the channel, including:
      * <ul><li>any queue subscriptions (this may in turn remove queues if they are auto delete</li>
      * </ul>
+     *
      * @param channelId id of the channel to close
      * @throws AMQException if an error occurs closing the channel
      * @throws IllegalArgumentException if the channel id is not valid
@@ -381,6 +388,7 @@ public class AMQMinaProtocolSession implements AMQProtocolSession,
 
     /**
      * In our current implementation this is used by the clustering code.
+     *
      * @param channelId
      */
     public void removeChannel(int channelId)
@@ -390,11 +398,12 @@ public class AMQMinaProtocolSession implements AMQProtocolSession,
 
     /**
      * Initialise heartbeats on the session.
+     *
      * @param delay delay in seconds (not ms)
      */
     public void initHeartbeats(int delay)
     {
-        if(delay > 0)
+        if (delay > 0)
         {
             _minaProtocolSession.setIdleTime(IdleStatus.WRITER_IDLE, delay);
             _minaProtocolSession.setIdleTime(IdleStatus.READER_IDLE, HeartbeatConfig.getInstance().getTimeout(delay));
@@ -404,6 +413,7 @@ public class AMQMinaProtocolSession implements AMQProtocolSession,
     /**
      * Closes all channels that were opened by this protocol session. This frees up all resources
      * used by the channel.
+     *
      * @throws AMQException if an error occurs while closing any channel
      */
     private void closeAllChannels() throws AMQException
@@ -412,6 +422,7 @@ public class AMQMinaProtocolSession implements AMQProtocolSession,
         {
             channel.close(this);
         }
+        _channelMap.clear();
     }
 
     /**
@@ -420,7 +431,7 @@ public class AMQMinaProtocolSession implements AMQProtocolSession,
      */
     public void closeSession() throws AMQException
     {
-        if(!_closed)
+        if (!_closed)
         {
             _closed = true;
             closeAllChannels();
@@ -462,11 +473,11 @@ public class AMQMinaProtocolSession implements AMQProtocolSession,
         // information is used by SASL primary.
         if (address instanceof InetSocketAddress)
         {
-            return ((InetSocketAddress)address).getHostName();
+            return ((InetSocketAddress) address).getHostName();
         }
         else if (address instanceof VmPipeAddress)
         {
-            return "vmpipe:" + ((VmPipeAddress)address).getPort();
+            return "vmpipe:" + ((VmPipeAddress) address).getPort();
         }
         else
         {
@@ -483,22 +494,32 @@ public class AMQMinaProtocolSession implements AMQProtocolSession,
     {
         _saslServer = saslServer;
     }
-    
+
+    public FieldTable getClientProperties()
+    {
+        return _clientProperties;
+    }
+
+    public void setClientProperties(FieldTable clientProperties)
+    {
+        _clientProperties = clientProperties;
+    }
+
     /**
      * Convenience methods for managing AMQP version.
      * NOTE: Both major and minor will be set to 0 prior to protocol initiation.
      */
-    
+
     public byte getAmqpMajor()
     {
         return _major;
     }
-    
+
     public byte getAmqpMinor()
     {
         return _minor;
     }
-    
+
     public boolean amqpVersionEquals(byte major, byte minor)
     {
         return _major == major && _minor == minor;
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolSession.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolSession.java
index acaf6b0d9b..03d0c50dac 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolSession.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolSession.java
@@ -21,6 +21,7 @@
 package org.apache.qpid.server.protocol;
 
 import org.apache.qpid.framing.AMQDataBlock;
+import org.apache.qpid.framing.FieldTable;
 import org.apache.qpid.server.AMQChannel;
 import org.apache.qpid.AMQException;
 
@@ -122,4 +123,9 @@ public interface AMQProtocolSession
      * @param saslServer
      */
     void setSaslServer(SaslServer saslServer);
+
+
+    FieldTable getClientProperties();
+
+    void setClientProperties(FieldTable clientProperties);
 }
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/AMQMessage.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/AMQMessage.java
index 12e06b31ed..b27cd807c0 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/AMQMessage.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/AMQMessage.java
@@ -25,6 +25,8 @@ import org.apache.qpid.framing.*;
 import org.apache.qpid.server.protocol.AMQProtocolSession;
 import org.apache.qpid.server.store.MessageStore;
 import org.apache.qpid.server.txn.TxnBuffer;
+import org.apache.qpid.server.message.MessageDecorator;
+import org.apache.qpid.server.message.jms.JMSMessage;
 import org.apache.qpid.AMQException;
 
 import java.util.ArrayList;
@@ -33,17 +35,21 @@ import java.util.LinkedList;
 import java.util.Set;
 import java.util.HashSet;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.ConcurrentHashMap;
 
 /**
  * Combines the information that make up a deliverable message into a more manageable form.
  */
 public class AMQMessage
 {
+    public static final String JMS_MESSAGE = "jms.message";
+
     private final Set<Object> _tokens = new HashSet<Object>();
 
     private AMQProtocolSession _publisher;
 
-    private final BasicPublishBody  _publishBody;
+    private final BasicPublishBody _publishBody;
 
     private ContentHeaderBody _contentHeaderBody;
 
@@ -83,6 +89,8 @@ public class AMQMessage
      * messages published with the 'immediate' flag.
      */
     private boolean _deliveredToConsumer;
+    private ConcurrentHashMap<String, MessageDecorator> _decodedMessages;
+    private AtomicBoolean _taken;
 
 
     public AMQMessage(MessageStore messageStore, BasicPublishBody publishBody)
@@ -96,7 +104,9 @@ public class AMQMessage
         _publishBody = publishBody;
         _store = messageStore;
         _contentBodies = new LinkedList<ContentBody>();
+        _decodedMessages = new ConcurrentHashMap<String, MessageDecorator>();
         _storeWhenComplete = storeWhenComplete;
+        _taken = new AtomicBoolean(false);
     }
 
     public AMQMessage(MessageStore store, long messageId, BasicPublishBody publishBody,
@@ -107,6 +117,7 @@ public class AMQMessage
         _publishBody = publishBody;
         _contentHeaderBody = contentHeaderBody;
         _contentBodies = contentBodies;
+        _decodedMessages = new ConcurrentHashMap<String, MessageDecorator>();
         _messageId = messageId;
         _store = store;
         storeMessage();
@@ -271,7 +282,7 @@ public class AMQMessage
             {
                 _store.removeMessage(_messageId);
             }
-            catch(AMQException e)
+            catch (AMQException e)
             {
                 //to maintain consistency, we revert the count
                 incrementReference();
@@ -292,7 +303,7 @@ public class AMQMessage
 
     public boolean checkToken(Object token)
     {
-        if(_tokens.contains(token))
+        if (_tokens.contains(token))
         {
             return true;
         }
@@ -308,7 +319,7 @@ public class AMQMessage
         //if the message is not persistent or the queue is not durable
         //we will not need to recover the association and so do not
         //need to record it
-        if(isPersistent() && queue.isDurable())
+        if (isPersistent() && queue.isDurable())
         {
             _store.enqueueMessage(queue.getName(), _messageId);
         }
@@ -318,7 +329,7 @@ public class AMQMessage
     {
         //only record associations where both queue and message will survive
         //a restart, so only need to remove association if this is the case
-        if(isPersistent() && queue.isDurable())
+        if (isPersistent() && queue.isDurable())
         {
             _store.dequeueMessage(queue.getName(), _messageId);
         }
@@ -326,14 +337,14 @@ public class AMQMessage
 
     public boolean isPersistent() throws AMQException
     {
-        if(_contentHeaderBody == null)
+        if (_contentHeaderBody == null)
         {
             throw new AMQException("Cannot determine delivery mode of message. Content header not found.");
         }
 
         //todo remove literal values to a constant file such as AMQConstants in common
         return _contentHeaderBody.properties instanceof BasicContentHeaderProperties
-                &&((BasicContentHeaderProperties) _contentHeaderBody.properties).getDeliveryMode() == 2;
+               && ((BasicContentHeaderProperties) _contentHeaderBody.properties).getDeliveryMode() == 2;
     }
 
     public void setTxnBuffer(TxnBuffer buffer)
@@ -352,8 +363,9 @@ public class AMQMessage
      * immediate delivery but has not been marked as delivered to a
      * consumer
      */
-    public void checkDeliveredToConsumer() throws NoConsumersException{
-        if(isImmediate() && !_deliveredToConsumer)
+    public void checkDeliveredToConsumer() throws NoConsumersException
+    {
+        if (isImmediate() && !_deliveredToConsumer)
         {
             throw new NoConsumersException(_publishBody, _contentHeaderBody, _contentBodies);
         }
@@ -362,8 +374,64 @@ public class AMQMessage
     /**
      * Called when this message is delivered to a consumer. (used to
      * implement the 'immediate' flag functionality).
+     * And by selectors to determin if the message has already been sent
      */
-    public void setDeliveredToConsumer(){
+    public void setDeliveredToConsumer()
+    {
         _deliveredToConsumer = true;
     }
+
+    /**
+     * Called selectors to determin if the message has already been sent
+     * @return   _deliveredToConsumer
+     */
+    public boolean getDeliveredToConsumer()
+    {
+        return _deliveredToConsumer;
+    }
+
+
+    public MessageDecorator getDecodedMessage(String type)
+    {
+        MessageDecorator msgtype = null;
+
+        if (_decodedMessages != null)
+        {
+            msgtype = _decodedMessages.get(type);
+
+            if (msgtype == null)
+            {
+                msgtype = decorateMessage(type);
+            }
+        }
+
+        return msgtype;
+    }
+
+    private MessageDecorator decorateMessage(String type)
+    {
+        MessageDecorator msgdec = null;
+
+        if (type.equals(JMS_MESSAGE))
+        {
+            msgdec = new JMSMessage(this);
+        }
+
+        if (msgdec != null)
+        {
+            _decodedMessages.put(type, msgdec);
+        }
+
+        return msgdec;
+    }
+
+    public boolean taken()
+    {
+        return _taken.getAndSet(true);
+    }
+
+    public void release()
+    {
+        _taken.set(false);
+    }
 }
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/AMQQueue.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/AMQQueue.java
index f2ef97cf9a..101a2833a0 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/AMQQueue.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/AMQQueue.java
@@ -22,6 +22,7 @@ package org.apache.qpid.server.queue;
 
 import org.apache.log4j.Logger;
 import org.apache.qpid.AMQException;
+import org.apache.qpid.framing.FieldTable;
 import org.apache.qpid.server.exchange.Exchange;
 import org.apache.qpid.server.management.Managable;
 import org.apache.qpid.server.management.ManagedObject;
@@ -95,7 +96,7 @@ public class AMQQueue implements Managable, Comparable
      * max allowed number of messages on a queue.
      */
     private Integer _maxMessageCount = 10000;
-    
+
     /**
      * max queue depth(KB) for the queue
      */
@@ -187,16 +188,29 @@ public class AMQQueue implements Managable, Comparable
         _subscribers = subscribers;
         _subscriptionFactory = subscriptionFactory;
 
-        //fixme - Pick one.
-        if (Boolean.getBoolean("concurrentdeliverymanager"))
+        //fixme - Make this configurable via the broker config.xml
+        if (System.getProperties().getProperty("deliverymanager") != null)
         {
-            _logger.info("Using ConcurrentDeliveryManager");
-            _deliveryMgr = new ConcurrentDeliveryManager(_subscribers, this);
+            if (System.getProperties().getProperty("deliverymanager").equals("ConcurrentSelectorDeliveryManager"))
+            {
+                _logger.info("Using ConcurrentSelectorDeliveryManager");
+                _deliveryMgr = new ConcurrentSelectorDeliveryManager(_subscribers, this);
+            }
+            else if (System.getProperties().getProperty("deliverymanager").equals("ConcurrentDeliveryManager"))
+            {
+                _logger.info("Using ConcurrentDeliveryManager");
+                _deliveryMgr = new ConcurrentDeliveryManager(_subscribers, this);
+            }
+            else
+            {
+                _logger.info("Using SynchronizedDeliveryManager");
+                _deliveryMgr = new SynchronizedDeliveryManager(_subscribers, this);
+            }
         }
         else
         {
-            _logger.info("Using SynchronizedDeliveryManager");
-            _deliveryMgr = new SynchronizedDeliveryManager(_subscribers, this);
+            _logger.info("Using Default DeliveryManager: ConcurrentSelectorDeliveryManager");
+            _deliveryMgr = new ConcurrentSelectorDeliveryManager(_subscribers, this);
         }
     }
 
@@ -348,12 +362,26 @@ public class AMQQueue implements Managable, Comparable
         _bindings.addBinding(routingKey, exchange);
     }
 
-    public void registerProtocolSession(AMQProtocolSession ps, int channel, String consumerTag, boolean acks)
+    public void registerProtocolSession(AMQProtocolSession ps, int channel, String consumerTag, boolean acks, FieldTable filters) throws AMQException
+    {
+        registerProtocolSession(ps, channel, consumerTag, acks, filters, false);
+    }
+
+    public void registerProtocolSession(AMQProtocolSession ps, int channel, String consumerTag, boolean acks, FieldTable filters, boolean noLocal)
             throws AMQException
     {
         debug("Registering protocol session {0} with channel {1} and consumer tag {2} with {3}", ps, channel, consumerTag, this);
 
-        Subscription subscription = _subscriptionFactory.createSubscription(channel, ps, consumerTag, acks);
+        Subscription subscription = _subscriptionFactory.createSubscription(channel, ps, consumerTag, acks, filters, noLocal);
+
+        if(subscription.hasFilters())
+        {
+            if (_deliveryMgr.hasQueuedMessages())
+            {
+                _deliveryMgr.populatePreDeliveryQueue(subscription);   
+            }
+        }
+
         _subscribers.addSubscriber(subscription);
     }
 
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/ConcurrentDeliveryManager.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/ConcurrentDeliveryManager.java
index f9c8898182..022d3b9635 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/ConcurrentDeliveryManager.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/ConcurrentDeliveryManager.java
@@ -198,6 +198,11 @@ public class ConcurrentDeliveryManager implements DeliveryManager
         return new ArrayList<AMQMessage>(_messages);
     }
 
+    public void populatePreDeliveryQueue(Subscription subscription)
+    {
+        //no-op . This DM has no PreDeliveryQueues
+    }
+
     public synchronized void removeAMessageFromTop() throws AMQException
     {
         AMQMessage msg = poll();
@@ -312,7 +317,6 @@ public class ConcurrentDeliveryManager implements DeliveryManager
                 else
                 {
                     s.send(msg, _queue);
-                    msg.setDeliveredToConsumer();
                 }
             }
             finally
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/ConcurrentSelectorDeliveryManager.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/ConcurrentSelectorDeliveryManager.java
new file mode 100644
index 0000000000..2100734ada
--- /dev/null
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/ConcurrentSelectorDeliveryManager.java
@@ -0,0 +1,388 @@
+/*
+ *
+ * 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.server.queue;
+
+import org.apache.log4j.Logger;
+import org.apache.qpid.AMQException;
+import org.apache.qpid.util.ConcurrentLinkedQueueAtomicSize;
+import org.apache.qpid.configuration.Configured;
+import org.apache.qpid.framing.ContentBody;
+import org.apache.qpid.server.configuration.Configurator;
+
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Queue;
+import java.util.concurrent.Executor;
+import java.util.concurrent.locks.ReentrantLock;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+
+/**
+ * Manages delivery of messages on behalf of a queue
+ */
+public class ConcurrentSelectorDeliveryManager implements DeliveryManager
+{
+    private static final Logger _log = Logger.getLogger(ConcurrentSelectorDeliveryManager.class);
+
+    @Configured(path = "advanced.compressBufferOnQueue",
+                defaultValue = "false")
+    public boolean compressBufferOnQueue;
+    /**
+     * Holds any queued messages
+     */
+    private final Queue<AMQMessage> _messages = new ConcurrentLinkedQueueAtomicSize<AMQMessage>();
+    //private int _messageCount;
+    /**
+     * Ensures that only one asynchronous task is running for this manager at
+     * any time.
+     */
+    private final AtomicBoolean _processing = new AtomicBoolean();
+    /**
+     * The subscriptions on the queue to whom messages are delivered
+     */
+    private final SubscriptionManager _subscriptions;
+
+    /**
+     * A reference to the queue we are delivering messages for. We need this to be able
+     * to pass the code that handles acknowledgements a handle on the queue.
+     */
+    private final AMQQueue _queue;
+
+
+    /**
+     * Lock used to ensure that an channel that becomes unsuspended during the start of the queueing process is forced
+     * to wait till the first message is added to the queue. This will ensure that the _queue has messages to be delivered
+     * via the async thread.
+     * <p/>
+     * Lock is used to control access to hasQueuedMessages() and over the addition of messages to the queue.
+     */
+    private ReentrantLock _lock = new ReentrantLock();
+
+
+    ConcurrentSelectorDeliveryManager(SubscriptionManager subscriptions, AMQQueue queue)
+    {
+
+        //Set values from configuration
+        Configurator.configure(this);
+
+        if (compressBufferOnQueue)
+        {
+            _log.warn("Compressing Buffers on queue.");
+        }
+
+        _subscriptions = subscriptions;
+        _queue = queue;
+    }
+
+
+    private boolean addMessageToQueue(AMQMessage msg)
+    {
+        // Shrink the ContentBodies to their actual size to save memory.
+        if (compressBufferOnQueue)
+        {
+            Iterator it = msg.getContentBodies().iterator();
+            while (it.hasNext())
+            {
+                ContentBody cb = (ContentBody) it.next();
+                cb.reduceBufferToFit();
+            }
+        }
+
+        _messages.offer(msg);
+
+        return true;
+    }
+
+
+    public boolean hasQueuedMessages()
+    {
+        _lock.lock();
+        try
+        {
+            return !_messages.isEmpty();
+        }
+        finally
+        {
+            _lock.unlock();
+        }
+    }
+
+    public int getQueueMessageCount()
+    {
+        return getMessageCount();
+    }
+
+    /**
+     * This is an EXPENSIVE opperation to perform with a ConcurrentLinkedQueue as it must run the queue to determine size.
+     * The ConcurrentLinkedQueueAtomicSize uses an AtomicInteger to record the number of elements on the queue.
+     *
+     * @return int the number of messages in the delivery queue.
+     */
+    private int getMessageCount()
+    {
+        return _messages.size();
+    }
+
+
+    public synchronized List<AMQMessage> getMessages()
+    {
+        return new ArrayList<AMQMessage>(_messages);
+    }
+
+    public void populatePreDeliveryQueue(Subscription subscription)
+    {
+        if (_log.isTraceEnabled())
+        {
+            _log.trace("Populating PreDeliveryQueue for Subscription(" + System.identityHashCode(subscription) + ")");
+        }
+
+        Iterator<AMQMessage> currentQueue = _messages.iterator();
+
+        while (currentQueue.hasNext())
+        {
+            AMQMessage message = currentQueue.next();
+            if (subscription.hasInterest(message))
+            {
+                subscription.enqueueForPreDelivery(message);
+            }
+        }
+    }
+
+    public synchronized void removeAMessageFromTop() throws AMQException
+    {
+        AMQMessage msg = poll();
+        if (msg != null)
+        {
+            msg.dequeue(_queue);
+        }
+    }
+
+    public synchronized void clearAllMessages() throws AMQException
+    {
+        AMQMessage msg = poll();
+        while (msg != null)
+        {
+            msg.dequeue(_queue);
+            msg = poll();
+        }
+    }
+
+
+    private AMQMessage getNextMessage(Queue<AMQMessage> messages, Subscription sub)
+    {
+        AMQMessage message = messages.peek();
+
+        while (message != null && (sub.isBrowser() || message.taken()))
+        {
+            //remove the already taken message
+            messages.poll();
+            // try the next message
+            message = messages.peek();
+        }
+        return message;
+    }
+
+    public void sendNextMessage(Subscription sub, Queue<AMQMessage> messageQueue)
+    {
+        AMQMessage message = null;
+        try
+        {
+            message = getNextMessage(messageQueue, sub);
+
+            // message will be null if we have no messages in the messageQueue.
+            if (message == null)
+            {
+                return;
+            }
+            _log.info("Async Delivery Message:" + message + " to :" + sub);
+
+            sub.send(message, _queue);
+
+            //remove sent message from our queue.
+            messageQueue.poll();
+        }
+        catch (FailedDequeueException e)
+        {
+            message.release();
+            _log.error("Unable to deliver message as dequeue failed: " + e, e);
+        }
+    }
+
+    /**
+     * Only one thread should ever execute this method concurrently, but
+     * it can do so while other threads invoke deliver().
+     */
+    private void processQueue()
+    {
+        // Continue to process delivery while we haveSubscribers and messages
+        boolean hasSubscribers = _subscriptions.hasActiveSubscribers();
+
+        while (hasSubscribers && hasQueuedMessages())
+        {
+            hasSubscribers = false;
+
+            for (Subscription sub : _subscriptions.getSubscriptions())
+            {
+                if (!sub.isSuspended())
+                {
+                    sendNextMessage(sub);
+
+                    hasSubscribers = true;
+                }
+            }
+        }
+    }
+
+    private void sendNextMessage(Subscription sub)
+    {
+        if (sub.hasFilters())
+        {
+            sendNextMessage(sub, sub.getPreDeliveryQueue());
+            if (sub.isAutoClose())
+            {
+                if (sub.getPreDeliveryQueue().isEmpty())
+                {
+                    sub.close();
+                }
+            }
+        }
+        else
+        {
+            sendNextMessage(sub, _messages);
+        }
+    }
+
+    private AMQMessage poll()
+    {
+        return _messages.poll();
+    }
+
+    public void deliver(String name, AMQMessage msg) throws FailedDequeueException
+    {
+        _log.info(id() + "deliver :" + System.identityHashCode(msg));
+
+        //Check if we have someone to deliver the message to.
+        _lock.lock();
+        try
+        {
+            Subscription s = _subscriptions.nextSubscriber(msg);
+
+            if (s == null) //no-one can take the message right now.
+            {
+                _log.info(id() + "Testing Message(" + System.identityHashCode(msg) + ") for Queued Delivery");
+                if (!msg.isImmediate())
+                {
+                    addMessageToQueue(msg);
+
+                    //release lock now message is on queue.
+                    _lock.unlock();
+
+                    //Pre Deliver to all subscriptions
+                    _log.info(id() + "We have " + _subscriptions.getSubscriptions().size() + " subscribers to give the message to.");
+                    for (Subscription sub : _subscriptions.getSubscriptions())
+                    {
+
+                        // stop if the message gets delivered whilst PreDelivering if we have a shared queue.
+                        if (_queue.isShared() && msg.getDeliveredToConsumer())
+                        {
+                            _log.info(id() + "Stopping PreDelivery as message(" + System.identityHashCode(msg) + ") is already delivered.");
+                            continue;
+                        }
+
+                        // Only give the message to those that want them.
+                        if (sub.hasInterest(msg))
+                        {
+                            _log.info(id() + "Queuing message(" + System.identityHashCode(msg) + ") for PreDelivery for subscriber(" + System.identityHashCode(sub) + ")");
+                            sub.enqueueForPreDelivery(msg);
+                        }
+                    }
+                }
+            }
+            else
+            {
+                //release lock now
+                _lock.unlock();
+
+                _log.info(id() + "Delivering Message:" + System.identityHashCode(msg) + " to(" + System.identityHashCode(s) + ") :" + s);
+                //Deliver the message
+                s.send(msg, _queue);
+            }
+        }
+        finally
+        {
+            //ensure lock is released
+            if (_lock.isLocked())
+            {
+                _lock.unlock();
+            }
+        }
+    }
+
+    //fixme remove
+    private final String id = "(" + String.valueOf(System.identityHashCode(this)) + ")";
+
+    private String id()
+    {
+        return id;
+    }
+
+    Runner asyncDelivery = new Runner();
+
+    private class Runner implements Runnable
+    {
+        public void run()
+        {
+            boolean running = true;
+            while (running)
+            {
+                processQueue();
+
+                //Check that messages have not been added since we did our last peek();
+                // Synchronize with the thread that adds to the queue.
+                // If the queue is still empty then we can exit
+
+                if (!(hasQueuedMessages() && _subscriptions.hasActiveSubscribers()))
+                {
+                    running = false;
+                    _processing.set(false);
+                }
+            }
+        }
+    }
+
+    public void processAsync(Executor executor)
+    {
+        _log.info("Processing Async. Queued:" + hasQueuedMessages() + "(" + getQueueMessageCount() + ")" +
+                  " Active:" + _subscriptions.hasActiveSubscribers() +
+                  " Processing:" + _processing.get());
+
+        if (hasQueuedMessages() && _subscriptions.hasActiveSubscribers())
+        {
+            //are we already running? if so, don't re-run
+            if (_processing.compareAndSet(false, true))
+            {
+                executor.execute(asyncDelivery);
+            }
+        }
+    }
+
+}
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/DeliveryManager.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/DeliveryManager.java
index dadf86c1d8..cac499587f 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/DeliveryManager.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/DeliveryManager.java
@@ -73,4 +73,6 @@ interface DeliveryManager
     void clearAllMessages() throws AMQException;
 
     List<AMQMessage> getMessages();
+
+    void populatePreDeliveryQueue(Subscription subscription);
 }
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/Subscription.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/Subscription.java
index 49f0a51bf2..a5672f2b19 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/Subscription.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/Subscription.java
@@ -20,6 +20,10 @@
  */
 package org.apache.qpid.server.queue;
 
+import org.apache.qpid.AMQException;
+
+import java.util.Queue;
+
 public interface Subscription
 {
     void send(AMQMessage msg, AMQQueue queue) throws FailedDequeueException;
@@ -27,4 +31,18 @@ public interface Subscription
     boolean isSuspended();
 
     void queueDeleted(AMQQueue queue);
+
+    boolean hasFilters();
+
+    boolean hasInterest(AMQMessage msg);
+
+    Queue<AMQMessage> getPreDeliveryQueue();
+
+    void enqueueForPreDelivery(AMQMessage msg);
+
+    boolean isAutoClose();
+
+    void close();
+
+    boolean isBrowser();   
 }
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SubscriptionFactory.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SubscriptionFactory.java
index 0fd44e4fbc..2bb77dc649 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SubscriptionFactory.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SubscriptionFactory.java
@@ -22,6 +22,7 @@ package org.apache.qpid.server.queue;
 
 import org.apache.qpid.server.protocol.AMQProtocolSession;
 import org.apache.qpid.AMQException;
+import org.apache.qpid.framing.FieldTable;
 
 /**
  * Allows the customisation of the creation of a subscription. This is typically done within an AMQQueue. This
@@ -32,9 +33,10 @@ import org.apache.qpid.AMQException;
  */
 public interface SubscriptionFactory
 {
-    Subscription createSubscription(int channel, AMQProtocolSession protocolSession, String consumerTag, boolean acks)
-        throws AMQException;
+    Subscription createSubscription(int channel, AMQProtocolSession protocolSession, String consumerTag, boolean acks,
+                                    FieldTable filters, boolean noLocal) throws AMQException;
 
-    Subscription createSubscription(int channel, AMQProtocolSession protocolSession,String consumerTag)
-        throws AMQException;
+
+    Subscription createSubscription(int channel, AMQProtocolSession protocolSession, String consumerTag)
+            throws AMQException;
 }
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SubscriptionImpl.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SubscriptionImpl.java
index 5cad28b80d..4272541298 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SubscriptionImpl.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SubscriptionImpl.java
@@ -23,12 +23,21 @@ package org.apache.qpid.server.queue;
 import org.apache.log4j.Logger;
 import org.apache.mina.common.ByteBuffer;
 import org.apache.qpid.AMQException;
+import org.apache.qpid.common.ClientProperties;
+import org.apache.qpid.common.AMQPFilterTypes;
+import org.apache.qpid.util.ConcurrentLinkedQueueAtomicSize;
 import org.apache.qpid.framing.AMQDataBlock;
 import org.apache.qpid.framing.AMQFrame;
 import org.apache.qpid.framing.BasicDeliverBody;
+import org.apache.qpid.framing.FieldTable;
+import org.apache.qpid.framing.BasicCancelOkBody;
 import org.apache.qpid.server.AMQChannel;
+import org.apache.qpid.server.filter.FilterManager;
+import org.apache.qpid.server.filter.FilterManagerFactory;
 import org.apache.qpid.server.protocol.AMQProtocolSession;
 
+import java.util.Queue;
+
 /**
  * Encapsulation of a supscription to a queue.
  * <p/>
@@ -48,23 +57,30 @@ public class SubscriptionImpl implements Subscription
 
     private final Object sessionKey;
 
+    private Queue<AMQMessage> _messages;
+
+    private final boolean _noLocal;
+
     /**
      * True if messages need to be acknowledged
      */
     private final boolean _acks;
+    private FilterManager _filters;
+    private final boolean _isBrowser;
+    private final Boolean _autoClose;
+    private boolean _closed = false;
 
     public static class Factory implements SubscriptionFactory
     {
-        public SubscriptionImpl createSubscription(int channel, AMQProtocolSession protocolSession, String consumerTag, boolean acks)
-                throws AMQException
+        public Subscription createSubscription(int channel, AMQProtocolSession protocolSession, String consumerTag, boolean acks, FieldTable filters, boolean noLocal) throws AMQException
         {
-            return new SubscriptionImpl(channel, protocolSession, consumerTag, acks);
+            return new SubscriptionImpl(channel, protocolSession, consumerTag, acks, filters, noLocal);
         }
 
         public SubscriptionImpl createSubscription(int channel, AMQProtocolSession protocolSession, String consumerTag)
                 throws AMQException
         {
-            return new SubscriptionImpl(channel, protocolSession, consumerTag);
+            return new SubscriptionImpl(channel, protocolSession, consumerTag, false, null, false);
         }
     }
 
@@ -72,6 +88,13 @@ public class SubscriptionImpl implements Subscription
                             String consumerTag, boolean acks)
             throws AMQException
     {
+        this(channelId, protocolSession, consumerTag, acks, null, false);
+    }
+
+    public SubscriptionImpl(int channelId, AMQProtocolSession protocolSession,
+                            String consumerTag, boolean acks, FieldTable filters, boolean noLocal)
+            throws AMQException
+    {
         AMQChannel channel = protocolSession.getChannel(channelId);
         if (channel == null)
         {
@@ -83,8 +106,61 @@ public class SubscriptionImpl implements Subscription
         this.consumerTag = consumerTag;
         sessionKey = protocolSession.getKey();
         _acks = acks;
+        _noLocal = noLocal;
+
+        _filters = FilterManagerFactory.createManager(filters);
+
+
+        if (_filters != null)
+        {
+            Object isBrowser = filters.get(AMQPFilterTypes.NO_CONSUME.getValue());
+            if (isBrowser != null)
+            {
+                _isBrowser = (Boolean) isBrowser;
+            }
+            else
+            {
+                _isBrowser = false;
+            }
+        }
+        else
+        {
+            _isBrowser = false;
+        }
+
+
+        if (_filters != null)
+        {
+            Object autoClose = filters.get(AMQPFilterTypes.AUTO_CLOSE.getValue());
+            if (autoClose != null)
+            {
+                _autoClose = (Boolean) autoClose;
+            }
+            else
+            {
+                _autoClose = false;
+            }
+        }
+        else
+        {
+            _autoClose = false;
+        }
+
+
+        if (_filters != null)
+        {
+            _messages = new ConcurrentLinkedQueueAtomicSize<AMQMessage>();
+
+
+        }
+        else
+        {
+            // Reference the DeliveryManager
+            _messages = null;
+        }
     }
 
+
     public SubscriptionImpl(int channel, AMQProtocolSession protocolSession,
                             String consumerTag)
             throws AMQException
@@ -129,9 +205,50 @@ public class SubscriptionImpl implements Subscription
     {
         if (msg != null)
         {
+            if (_isBrowser)
+            {
+                sendToBrowser(msg, queue);
+            }
+            else
+            {
+                sendToConsumer(msg, queue);
+            }
+        }
+        else
+        {
+            _logger.error("Attempt to send Null message", new NullPointerException());
+        }
+    }
+
+    private void sendToBrowser(AMQMessage msg, AMQQueue queue) throws FailedDequeueException
+    {
+        // We don't decrement the reference here as we don't want to consume the message
+        // but we do want to send it to the client.
+
+        synchronized(channel)
+        {
+            long deliveryTag = channel.getNextDeliveryTag();
+
+            // We don't need to add the message to the unacknowledgedMap as we don't need to know if the client
+            // received the message. If it is lost in transit that is not important.
+            if (_acks)
+            {
+                channel.addUnacknowledgedBrowsedMessage(msg, deliveryTag, consumerTag, queue);
+            }
+            ByteBuffer deliver = createEncodedDeliverFrame(deliveryTag, msg.getRoutingKey(), msg.getExchangeName());
+            AMQDataBlock frame = msg.getDataBlock(deliver, channel.getChannelId());
+
+            protocolSession.writeFrame(frame);
+        }
+    }
+
+    private void sendToConsumer(AMQMessage msg, AMQQueue queue) throws FailedDequeueException
+    {
+        try
+        {
             // if we do not need to wait for client acknowledgements
-            // we can decrement the reference count immediately. 
-            
+            // we can decrement the reference count immediately.
+
             // By doing this _before_ the send we ensure that it
             // doesn't get sent if it can't be dequeued, preventing
             // duplicate delivery on recovery.
@@ -157,9 +274,9 @@ public class SubscriptionImpl implements Subscription
                 protocolSession.writeFrame(frame);
             }
         }
-        else
+        finally
         {
-            _logger.error("Attempt to send Null message", new NullPointerException());
+            msg.setDeliveredToConsumer();
         }
     }
 
@@ -178,6 +295,101 @@ public class SubscriptionImpl implements Subscription
         channel.queueDeleted(queue);
     }
 
+    public boolean hasFilters()
+    {
+        return _filters != null;
+    }
+
+    public boolean hasInterest(AMQMessage msg)
+    {
+        if (_noLocal)
+        {
+            // We don't want local messages so check to see if message is one we sent
+            if (protocolSession.getClientProperties().get(ClientProperties.instance.toString()).equals(
+                    msg.getPublisher().getClientProperties().get(ClientProperties.instance.toString())))
+            {
+                if (_logger.isTraceEnabled())
+                {
+                    _logger.trace("(" + System.identityHashCode(this) + ") has no interest as it is a local message(" +
+                                  System.identityHashCode(msg) + ")");
+                }
+                return false;
+            }
+            else // if not then filter the message.
+            {
+                if (_logger.isTraceEnabled())
+                {
+                    _logger.trace("(" + System.identityHashCode(this) + ") local message(" + System.identityHashCode(msg) +
+                                  ") but not ours so filtering");
+                }
+                return checkFilters(msg);
+            }
+        }
+        else
+        {
+            if (_logger.isTraceEnabled())
+            {
+                _logger.trace("(" + System.identityHashCode(this) + ") checking filters for message (" + System.identityHashCode(msg));
+            }
+            return checkFilters(msg);
+        }
+    }
+
+    private boolean checkFilters(AMQMessage msg)
+    {
+        if (_filters != null)
+        {
+            if (_logger.isTraceEnabled())
+            {
+                _logger.trace("(" + System.identityHashCode(this) + ") has filters.");
+            }
+            return _filters.allAllow(msg);
+        }
+        else
+        {
+            if (_logger.isTraceEnabled())
+            {
+                _logger.trace("(" + System.identityHashCode(this) + ") has no filters");
+            }
+
+            return true;
+        }
+    }
+
+    public Queue<AMQMessage> getPreDeliveryQueue()
+    {
+        return _messages;
+    }
+
+    public void enqueueForPreDelivery(AMQMessage msg)
+    {
+        if (_messages != null)
+        {
+            _messages.offer(msg);
+        }
+    }
+
+    public boolean isAutoClose()
+    {
+        return _autoClose;
+    }
+
+    public void close()
+    {
+        if (!_closed)
+        {
+            _logger.info("Closing autoclose subscription:" + this);
+            protocolSession.writeFrame(BasicCancelOkBody.createAMQFrame(channel.getChannelId(), consumerTag));
+            _closed = true;
+        }
+    }
+
+    public boolean isBrowser()
+    {
+        return _isBrowser;
+    }
+
+
     private ByteBuffer createEncodedDeliverFrame(long deliveryTag, String routingKey, String exchange)
     {
         AMQFrame deliverFrame = BasicDeliverBody.createAMQFrame(channel.getChannelId(), consumerTag,
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SubscriptionManager.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SubscriptionManager.java
index 353b461c8d..4df88baebc 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SubscriptionManager.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SubscriptionManager.java
@@ -20,12 +20,15 @@
  */
 package org.apache.qpid.server.queue;
 
+import java.util.List;
+
 /**
  * Abstraction of actor that will determine the subscriber to whom
  * a message will be sent.
  */
 public interface SubscriptionManager
 {
+    public List<Subscription> getSubscriptions();
     public boolean hasActiveSubscribers();
     public Subscription nextSubscriber(AMQMessage msg);
 }
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SubscriptionSet.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SubscriptionSet.java
index 7cc3f5f719..91e720ea54 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SubscriptionSet.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SubscriptionSet.java
@@ -21,6 +21,8 @@
 package org.apache.qpid.server.queue;
 
 import org.apache.log4j.Logger;
+import org.apache.qpid.AMQException;
+
 import java.util.List;
 import java.util.ListIterator;
 import java.util.concurrent.CopyOnWriteArrayList;
@@ -58,6 +60,7 @@ class SubscriptionSet implements WeightedSubscriptionManager
 
     /**
      * Remove the subscription, returning it if it was found
+     *
      * @param subscription
      * @return null if no match was found
      */
@@ -90,7 +93,7 @@ class SubscriptionSet implements WeightedSubscriptionManager
 
     /**
      * Return the next unsuspended subscription or null if not found.
-     *
+     * <p/>
      * Performance note:
      * This method can scan all items twice when looking for a subscription that is not
      * suspended. The worst case occcurs when all subscriptions are suspended. However, it is does this
@@ -105,31 +108,51 @@ class SubscriptionSet implements WeightedSubscriptionManager
             return null;
         }
 
-        try {
-            final Subscription result = nextSubscriber();
-            if (result == null) {
+        try
+        {
+            final Subscription result = nextSubscriberImpl(msg);
+            if (result == null)
+            {
                 _currentSubscriber = 0;
-                return nextSubscriber();
-            } else {
+                return nextSubscriberImpl(msg);
+            }
+            else
+            {
                 return result;
             }
-        } catch (IndexOutOfBoundsException e) {
+        }
+        catch (IndexOutOfBoundsException e)
+        {
             _currentSubscriber = 0;
-            return nextSubscriber();
+            return nextSubscriber(msg);
         }
     }
 
-    private Subscription nextSubscriber()
+    private Subscription nextSubscriberImpl(AMQMessage msg)
     {
         final ListIterator<Subscription> iterator = _subscriptions.listIterator(_currentSubscriber);
-        while (iterator.hasNext()) {
+        while (iterator.hasNext())
+        {
             Subscription subscription = iterator.next();
             ++_currentSubscriber;
             subscriberScanned();
-            if (!subscription.isSuspended()) {
-                return subscription;
+
+            if (!subscription.isSuspended())
+            {
+                if (subscription.hasInterest(msg))
+                {
+                    // if the queue is not empty then this client is ready to receive a message.
+                    //FIXME the queue could be full of sent messages.
+                    // Either need to clean all PDQs after sending a message
+                    // OR have a clean up thread that runs the PDQs expunging the messages.
+                    if (!subscription.hasFilters() || subscription.getPreDeliveryQueue().isEmpty())
+                    {
+                        return subscription;
+                    }
+                }
             }
         }
+
         return null;
     }
 
@@ -145,11 +168,19 @@ class SubscriptionSet implements WeightedSubscriptionManager
         return _subscriptions.isEmpty();
     }
 
+    public List<Subscription> getSubscriptions()
+    {
+        return _subscriptions;
+    }
+
     public boolean hasActiveSubscribers()
     {
         for (Subscription s : _subscriptions)
         {
-            if (!s.isSuspended()) return true;
+            if (!s.isSuspended())
+            {
+                return true;
+            }
         }
         return false;
     }
@@ -159,7 +190,10 @@ class SubscriptionSet implements WeightedSubscriptionManager
         int count = 0;
         for (Subscription s : _subscriptions)
         {
-            if (!s.isSuspended()) count++;
+            if (!s.isSuspended())
+            {
+                count++;
+            }
         }
         return count;
     }
@@ -167,6 +201,7 @@ class SubscriptionSet implements WeightedSubscriptionManager
     /**
      * Notification that a queue has been deleted. This is called so that the subscription can inform the
      * channel, which in turn can update its list of unacknowledged messages.
+     *
      * @param queue
      */
     public void queueDeleted(AMQQueue queue)
@@ -177,7 +212,8 @@ class SubscriptionSet implements WeightedSubscriptionManager
         }
     }
 
-    int size() {
+    int size()
+    {
         return _subscriptions.size();
     }
 }
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SynchronizedDeliveryManager.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SynchronizedDeliveryManager.java
index ea64952bc7..c967ea2cde 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SynchronizedDeliveryManager.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SynchronizedDeliveryManager.java
@@ -35,7 +35,7 @@ import java.util.concurrent.atomic.AtomicBoolean;
  */
 class SynchronizedDeliveryManager implements DeliveryManager
 {
-    private static final Logger _log = Logger.getLogger(ConcurrentDeliveryManager.class);
+    private static final Logger _log = Logger.getLogger(SynchronizedDeliveryManager.class);
 
     /**
      * Holds any queued messages
@@ -122,6 +122,11 @@ class SynchronizedDeliveryManager implements DeliveryManager
         return new ArrayList<AMQMessage>(_messages);
     }
 
+    public void populatePreDeliveryQueue(Subscription subscription)
+    {
+        //no-op . This DM has no PreDeliveryQueues
+    }
+
     public synchronized void removeAMessageFromTop() throws AMQException
     {
         AMQMessage msg = poll();
@@ -243,7 +248,6 @@ class SynchronizedDeliveryManager implements DeliveryManager
                 else
                 {
                     s.send(msg, _queue);
-                    msg.setDeliveredToConsumer();
                 }
             }
         }
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/security/auth/plain/PlainSaslServer.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/security/auth/plain/PlainSaslServer.java
index fdf655c2d9..7f51260e45 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/security/auth/plain/PlainSaslServer.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/security/auth/plain/PlainSaslServer.java
@@ -62,7 +62,7 @@ public class PlainSaslServer implements SaslServer
             }
 
             // we do not currently support authcid in any meaningful way
-            String authcid = new String(response, 0, authzidNullPosition, "utf8");
+            // String authcid = new String(response, 0, authzidNullPosition, "utf8");
             String authzid = new String(response, authzidNullPosition + 1, authcidNullPosition - 1, "utf8");
 
             // we do not care about the prompt but it throws if null
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/util/CircularBuffer.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/util/CircularBuffer.java
index 4767844abe..b58d551226 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/util/CircularBuffer.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/util/CircularBuffer.java
@@ -20,10 +20,15 @@
  */
 package org.apache.qpid.server.util;
 
+import org.apache.log4j.Logger;
+
 import java.util.Iterator;
 
 public class CircularBuffer implements Iterable
 {
+
+    private static final Logger _logger = Logger.getLogger(CircularBuffer.class);
+
     private final Object[] _log;
     private int _size;
     private int _index;
@@ -102,7 +107,7 @@ public class CircularBuffer implements Iterable
     {
         for(Object o : this)
         {
-         System.out.println(o);   
+         _logger.info(o);
         }
     }
 
@@ -120,7 +125,7 @@ public class CircularBuffer implements Iterable
         for(String s : items)
         {
             buffer.add(s);
-            System.out.println(buffer);
+            _logger.info(buffer);
         }
     }
 }