From a6b4188b807dce93ee1a54958bd92de4142a6d48 Mon Sep 17 00:00:00 2001 From: Martin Ritchie Date: Tue, 16 Jan 2007 12:13:19 +0000 Subject: 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 --- qpid/java/broker/pom.xml | 23 + .../java/broker/src/main/grammar/SelectorParser.jj | 598 +++++++++++++++++++++ .../java/org/apache/qpid/server/AMQChannel.java | 40 +- .../src/main/java/org/apache/qpid/server/Main.java | 4 +- .../qpid/server/filter/ArithmeticExpression.java | 216 ++++++++ .../qpid/server/filter/BinaryExpression.java | 100 ++++ .../qpid/server/filter/BooleanExpression.java | 45 ++ .../qpid/server/filter/ComparisonExpression.java | 465 ++++++++++++++++ .../qpid/server/filter/ConstantExpression.java | 170 ++++++ .../org/apache/qpid/server/filter/Expression.java | 42 ++ .../apache/qpid/server/filter/FilterManager.java | 37 ++ .../qpid/server/filter/FilterManagerFactory.java | 86 +++ .../qpid/server/filter/JMSSelectorFilter.java | 79 +++ .../apache/qpid/server/filter/LogicExpression.java | 95 ++++ .../apache/qpid/server/filter/MessageFilter.java | 30 ++ .../qpid/server/filter/NoConsumerFilter.java | 48 ++ .../qpid/server/filter/PropertyExpression.java | 305 +++++++++++ .../qpid/server/filter/SimpleFilterManager.java | 77 +++ .../apache/qpid/server/filter/UnaryExpression.java | 263 +++++++++ .../apache/qpid/server/filter/XPathExpression.java | 130 +++++ .../qpid/server/filter/XQueryExpression.java | 56 ++ .../qpid/server/filter/XalanXPathEvaluator.java | 99 ++++ .../server/handler/BasicConsumeMethodHandler.java | 23 +- .../handler/ConnectionStartOkMethodHandler.java | 9 +- .../qpid/server/management/MBeanIntrospector.java | 1 - .../qpid/server/message/MessageDecorator.java | 25 + .../apache/qpid/server/message/jms/JMSMessage.java | 306 +++++++++++ .../server/protocol/AMQMinaProtocolSession.java | 67 ++- .../qpid/server/protocol/AMQProtocolSession.java | 6 + .../org/apache/qpid/server/queue/AMQMessage.java | 88 ++- .../org/apache/qpid/server/queue/AMQQueue.java | 46 +- .../server/queue/ConcurrentDeliveryManager.java | 6 +- .../queue/ConcurrentSelectorDeliveryManager.java | 388 +++++++++++++ .../apache/qpid/server/queue/DeliveryManager.java | 2 + .../org/apache/qpid/server/queue/Subscription.java | 18 + .../qpid/server/queue/SubscriptionFactory.java | 10 +- .../apache/qpid/server/queue/SubscriptionImpl.java | 228 +++++++- .../qpid/server/queue/SubscriptionManager.java | 3 + .../apache/qpid/server/queue/SubscriptionSet.java | 66 ++- .../server/queue/SynchronizedDeliveryManager.java | 8 +- .../security/auth/plain/PlainSaslServer.java | 2 +- .../apache/qpid/server/util/CircularBuffer.java | 9 +- .../java/org/apache/qpid/client/AMQConnection.java | 9 +- .../org/apache/qpid/client/AMQQueueBrowser.java | 128 +++++ .../java/org/apache/qpid/client/AMQSession.java | 173 ++++-- .../apache/qpid/client/BasicMessageConsumer.java | 119 +++- .../apache/qpid/client/BasicMessageProducer.java | 3 + .../org/apache/qpid/client/CustomJMXProperty.java | 47 ++ .../apache/qpid/client/QpidConnectionMetaData.java | 131 +++-- .../client/handler/BasicCancelOkMethodHandler.java | 35 ++ .../client/handler/ChannelCloseMethodHandler.java | 23 +- .../handler/ConnectionOpenOkMethodHandler.java | 8 - .../handler/ConnectionStartMethodHandler.java | 11 +- .../qpid/client/message/AbstractJMSMessage.java | 18 +- .../qpid/client/message/JMSBytesMessage.java | 18 +- .../qpid/client/message/JMSObjectMessage.java | 12 +- .../qpid/client/message/JMSStreamMessage.java | 11 +- .../apache/qpid/client/message/JMSTextMessage.java | 1 + .../qpid/client/protocol/AMQProtocolSession.java | 8 + .../protocol/BlockingMethodFrameListener.java | 2 +- .../apache/qpid/client/state/AMQStateManager.java | 1 + .../transport/SocketTransportConnection.java | 2 +- .../org/apache/qpid/test/unit/ack/RecoverTest.java | 2 +- .../test/unit/basic/MultipleConnectionTest.java | 5 +- .../qpid/test/unit/basic/PropertyValueTest.java | 10 +- .../apache/qpid/test/unit/basic/SelectorTest.java | 141 +++++ .../client/channelclose/ChannelCloseOkTest.java | 4 +- .../channelclose/CloseWithBlockingReceiveTest.java | 3 +- .../unit/client/connection/ConnectionTest.java | 3 +- .../test/unit/client/message/MapMessageTest.java | 2 +- .../qpid/test/unit/message/JMSDestinationTest.java | 70 +++ .../qpid/test/unit/message/StreamMessageTest.java | 4 +- .../qpid/test/unit/topic/TopicSessionTest.java | 2 +- .../server/queue/ClusteredSubscriptionManager.java | 7 + .../server/queue/NestedSubscriptionManager.java | 25 +- .../qpid/server/queue/RemoteSubscriptionImpl.java | 50 +- qpid/java/common/src/main/java/log4j.properties | 28 + .../apache/qpid/AMQInvalidSelectorException.java | 31 ++ .../org/apache/qpid/common/AMQPFilterTypes.java | 40 ++ .../org/apache/qpid/common/ClientProperties.java | 29 + .../org/apache/qpid/common/QpidProperties.java | 46 ++ .../apache/qpid/framing/PropertyFieldTable.java | 15 +- .../java/org/apache/qpid/protocol/AMQConstant.java | 4 +- .../qpid/server/exchange/ExchangeMBeanTest.java | 123 +++++ .../qpid/server/exchange/HeadersExchangeTest.java | 9 +- .../ReturnUnroutableMandatoryMessageTest.java | 48 +- .../protocol/AMQProtocolSessionMBeanTest.java | 118 ++++ .../apache/qpid/server/protocol/MockIoSession.java | 3 +- .../qpid/server/queue/AMQQueueMBeanTest.java | 180 +++++++ .../apache/qpid/server/queue/AMQueueMBeanTest.java | 179 ------ .../qpid/server/queue/MockProtocolSession.java | 10 + .../qpid/server/queue/SubscriptionTestHelper.java | 36 ++ 92 files changed, 5819 insertions(+), 487 deletions(-) create mode 100644 qpid/java/broker/src/main/grammar/SelectorParser.jj create mode 100644 qpid/java/broker/src/main/java/org/apache/qpid/server/filter/ArithmeticExpression.java create mode 100644 qpid/java/broker/src/main/java/org/apache/qpid/server/filter/BinaryExpression.java create mode 100644 qpid/java/broker/src/main/java/org/apache/qpid/server/filter/BooleanExpression.java create mode 100644 qpid/java/broker/src/main/java/org/apache/qpid/server/filter/ComparisonExpression.java create mode 100644 qpid/java/broker/src/main/java/org/apache/qpid/server/filter/ConstantExpression.java create mode 100644 qpid/java/broker/src/main/java/org/apache/qpid/server/filter/Expression.java create mode 100644 qpid/java/broker/src/main/java/org/apache/qpid/server/filter/FilterManager.java create mode 100644 qpid/java/broker/src/main/java/org/apache/qpid/server/filter/FilterManagerFactory.java create mode 100644 qpid/java/broker/src/main/java/org/apache/qpid/server/filter/JMSSelectorFilter.java create mode 100644 qpid/java/broker/src/main/java/org/apache/qpid/server/filter/LogicExpression.java create mode 100644 qpid/java/broker/src/main/java/org/apache/qpid/server/filter/MessageFilter.java create mode 100644 qpid/java/broker/src/main/java/org/apache/qpid/server/filter/NoConsumerFilter.java create mode 100644 qpid/java/broker/src/main/java/org/apache/qpid/server/filter/PropertyExpression.java create mode 100644 qpid/java/broker/src/main/java/org/apache/qpid/server/filter/SimpleFilterManager.java create mode 100644 qpid/java/broker/src/main/java/org/apache/qpid/server/filter/UnaryExpression.java create mode 100644 qpid/java/broker/src/main/java/org/apache/qpid/server/filter/XPathExpression.java create mode 100644 qpid/java/broker/src/main/java/org/apache/qpid/server/filter/XQueryExpression.java create mode 100644 qpid/java/broker/src/main/java/org/apache/qpid/server/filter/XalanXPathEvaluator.java create mode 100644 qpid/java/broker/src/main/java/org/apache/qpid/server/message/MessageDecorator.java create mode 100644 qpid/java/broker/src/main/java/org/apache/qpid/server/message/jms/JMSMessage.java create mode 100644 qpid/java/broker/src/main/java/org/apache/qpid/server/queue/ConcurrentSelectorDeliveryManager.java create mode 100644 qpid/java/client/src/main/java/org/apache/qpid/client/AMQQueueBrowser.java create mode 100644 qpid/java/client/src/main/java/org/apache/qpid/client/CustomJMXProperty.java create mode 100644 qpid/java/client/src/main/java/org/apache/qpid/client/handler/BasicCancelOkMethodHandler.java create mode 100644 qpid/java/client/src/test/java/org/apache/qpid/test/unit/basic/SelectorTest.java create mode 100644 qpid/java/client/src/test/java/org/apache/qpid/test/unit/message/JMSDestinationTest.java create mode 100644 qpid/java/common/src/main/java/log4j.properties create mode 100644 qpid/java/common/src/main/java/org/apache/qpid/AMQInvalidSelectorException.java create mode 100644 qpid/java/common/src/main/java/org/apache/qpid/common/AMQPFilterTypes.java create mode 100644 qpid/java/common/src/main/java/org/apache/qpid/common/ClientProperties.java create mode 100644 qpid/java/common/src/main/java/org/apache/qpid/common/QpidProperties.java create mode 100644 qpid/java/systests/src/test/java/org/apache/qpid/server/exchange/ExchangeMBeanTest.java create mode 100644 qpid/java/systests/src/test/java/org/apache/qpid/server/protocol/AMQProtocolSessionMBeanTest.java create mode 100644 qpid/java/systests/src/test/java/org/apache/qpid/server/queue/AMQQueueMBeanTest.java delete mode 100644 qpid/java/systests/src/test/java/org/apache/qpid/server/queue/AMQueueMBeanTest.java (limited to 'qpid/java') diff --git a/qpid/java/broker/pom.xml b/qpid/java/broker/pom.xml index aea2d5878a..5f4c490fd4 100644 --- a/qpid/java/broker/pom.xml +++ b/qpid/java/broker/pom.xml @@ -54,6 +54,10 @@ commons-lang commons-lang + + org.apache.geronimo.specs + geronimo-jms_1.1_spec + org.apache.mina mina-filter-ssl @@ -82,6 +86,25 @@ + + org.codehaus.mojo + javacc-maven-plugin + 2.0 + + + generate-sources + + ${basedir}/src/main/grammar + ${basedir}/target/generated + org.apache.qpid.server.filter.jms.selector + + + javacc + + + + + org.apache.maven.plugins maven-surefire-plugin 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 + // + +// ---------------------------------------------------------------------------- +// 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: +{ + +} + +SKIP: +{ + +} + +/* 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"])* ()? // matches: 5.5 or 5. or 5.5E10 or 5.E10 + | "." (["0"-"9"])+ ()? // matches: .5 or .5E10 + | (["0"-"9"])+ // 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() + ( + right = andExpression() + { + left = LogicExpression.createOR(asBooleanExpression(left), asBooleanExpression(right)); + } + )* + ) + { + return left; + } + +} + + +Expression andExpression() : +{ + Expression left; + Expression right; +} +{ + ( + left = equalityExpression() + ( + 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) + + { + left = ComparisonExpression.createIsNull(left); + } + | + + { + 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; + } + t = stringLitteral() + [ u = stringLitteral() ] + { + left = ComparisonExpression.createLike(left, t, u); + } + | + LOOKAHEAD(2) + { + u=null; + } + t = stringLitteral() [ u = stringLitteral() ] + { + left = ComparisonExpression.createNotLike(left, t, u); + } + | + low = addExpression() high = addExpression() + { + left = ComparisonExpression.createBetween(left, low, high); + } + | + LOOKAHEAD(2) + low = addExpression() high = addExpression() + { + left = ComparisonExpression.createNotBetween(left, low, high); + } + | + + "(" + t = stringLitteral() + { + list = new ArrayList(); + list.add( t ); + } + ( + "," + t = stringLitteral() + { + list.add( t ); + } + + )* + ")" + { + left = ComparisonExpression.createInFilter(left, list); + } + | + LOOKAHEAD(2) + + "(" + 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); + } + | + left=unaryExpr() + { + left = UnaryExpression.createNOT( asBooleanExpression(left) ); + } + | + s=stringLitteral() + { + left = UnaryExpression.createXPath( s ); + } + | + 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 = + { + left = ConstantExpression.createFromDecimal(t.image); + } + ) + | + ( + t = + { + left = ConstantExpression.createFromHex(t.image); + } + ) + | + ( + t = + { + left = ConstantExpression.createFromOctal(t.image); + } + ) + | + ( + t = + { + left = ConstantExpression.createFloat(t.image); + } + ) + | + ( + + { + left = ConstantExpression.TRUE; + } + ) + | + ( + + { + left = ConstantExpression.FALSE; + } + ) + | + ( + + { + left = ConstantExpression.NULL; + } + ) + ) + { + return left; + } +} + +String stringLitteral() : +{ + Token t; + StringBuffer rc = new StringBuffer(); + boolean first=true; +} +{ + t = + { + // 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 = + { + 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 _returns = new LinkedList(); + private Set _browsedAcks = new HashSet(); 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 +// + + +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 +// + + + +/** + * 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 +// + + +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 +// + +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 +// + +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 +// + +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 +// + +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 +// + +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 +// + +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 _filters; + + public SimpleFilterManager() + { + _logger.debug("Creating SimpleFilterManager"); + _filters = new ConcurrentLinkedQueue(); + } + + 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 +// + +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 +// + +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 +// + +/** + * 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 +// + +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 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 evt = new AMQMethodEvent(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: *
  • any queue subscriptions (this may in turn remove queues if they are auto delete
    • *
+ * * @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 _tokens = new HashSet(); 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 _decodedMessages; + private AtomicBoolean _taken; public AMQMessage(MessageStore messageStore, BasicPublishBody publishBody) @@ -96,7 +104,9 @@ public class AMQMessage _publishBody = publishBody; _store = messageStore; _contentBodies = new LinkedList(); + _decodedMessages = new ConcurrentHashMap(); _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(); _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(_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 _messages = new ConcurrentLinkedQueueAtomicSize(); + //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. + *

+ * 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 getMessages() + { + return new ArrayList(_messages); + } + + public void populatePreDeliveryQueue(Subscription subscription) + { + if (_log.isTraceEnabled()) + { + _log.trace("Populating PreDeliveryQueue for Subscription(" + System.identityHashCode(subscription) + ")"); + } + + Iterator 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 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 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 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 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. *

@@ -48,29 +57,43 @@ public class SubscriptionImpl implements Subscription private final Object sessionKey; + private Queue _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); } } public SubscriptionImpl(int channelId, AMQProtocolSession protocolSession, 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(); + + + } + else + { + // Reference the DeliveryManager + _messages = null; + } } + public SubscriptionImpl(int channel, AMQProtocolSession protocolSession, String consumerTag) throws AMQException @@ -128,10 +204,51 @@ public class SubscriptionImpl implements Subscription public void send(AMQMessage msg, AMQQueue queue) throws FailedDequeueException { 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 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 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. - * + *

* 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 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 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(_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); } } } diff --git a/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnection.java b/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnection.java index 638aa555a2..820b8c3f83 100644 --- a/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnection.java +++ b/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnection.java @@ -139,6 +139,12 @@ public class AMQConnection extends Closeable implements Connection, QueueConnect */ private AMQException _lastAMQException = null; + + /* + * The connection meta data + */ + private QpidConnectionMetaData _connectionMetaData; + public AMQConnection(String broker, String username, String password, String clientName, String virtualHost) throws AMQException, URLSyntaxException { @@ -281,6 +287,7 @@ public class AMQConnection extends Closeable implements Connection, QueueConnect throw e; } + _connectionMetaData = new QpidConnectionMetaData(this); } protected boolean checkException(Throwable thrown) @@ -550,7 +557,7 @@ public class AMQConnection extends Closeable implements Connection, QueueConnect public ConnectionMetaData getMetaData() throws JMSException { checkNotClosed(); - return QpidConnectionMetaData.instance(); + return _connectionMetaData; } diff --git a/qpid/java/client/src/main/java/org/apache/qpid/client/AMQQueueBrowser.java b/qpid/java/client/src/main/java/org/apache/qpid/client/AMQQueueBrowser.java new file mode 100644 index 0000000000..5c753946a6 --- /dev/null +++ b/qpid/java/client/src/main/java/org/apache/qpid/client/AMQQueueBrowser.java @@ -0,0 +1,128 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + * + */ +package org.apache.qpid.client; + +import org.apache.log4j.Logger; + +import java.util.Enumeration; +import java.util.ArrayList; +import java.util.concurrent.atomic.AtomicBoolean; +import javax.jms.*; +import javax.jms.IllegalStateException; + +public class AMQQueueBrowser implements QueueBrowser +{ + private static final Logger _logger = Logger.getLogger(AMQQueueBrowser.class); + + + private AtomicBoolean _isClosed = new AtomicBoolean(); + private final AMQSession _session; + private final AMQQueue _queue; + private final ArrayList _consumers = new ArrayList(); + private final String _messageSelector; + + + AMQQueueBrowser(AMQSession session, AMQQueue queue, String messageSelector) throws JMSException + { + _session = session; + _queue = queue; + _messageSelector = (messageSelector == null) || (messageSelector.trim().length() == 0) ? null : messageSelector; + BasicMessageConsumer consumer = (BasicMessageConsumer) _session.createBrowserConsumer(_queue, _messageSelector, false); + consumer.close(); + } + + public Queue getQueue() throws JMSException + { + checkState(); + return _queue; + } + + private void checkState() throws JMSException + { + if (_isClosed.get()) + { + throw new IllegalStateException("Queue Browser"); + } + if (_session.isClosed()) + { + throw new IllegalStateException("Session is closed"); + } + + } + + public String getMessageSelector() throws JMSException + { + + checkState(); + return _messageSelector; + } + + public Enumeration getEnumeration() throws JMSException + { + checkState(); + final BasicMessageConsumer consumer = (BasicMessageConsumer) _session.createBrowserConsumer(_queue, _messageSelector, false); + _consumers.add(consumer); + + return new Enumeration() + { + + + Message _nextMessage = consumer.receive(); + + + public boolean hasMoreElements() + { + _logger.info("QB:hasMoreElements:" + (_nextMessage != null)); + return (_nextMessage != null); + } + + public Object nextElement() + { + Message msg = _nextMessage; + try + { + _logger.info("QB:nextElement about to receive"); + + _nextMessage = consumer.receive(); + _logger.info("QB:nextElement received:" + _nextMessage); + } + catch (JMSException e) + { + _logger.warn("Exception caught while queue browsing", e); + _nextMessage = null; + } + + return msg; + } + }; + } + + public void close() throws JMSException + { + for (BasicMessageConsumer consumer : _consumers) + { + consumer.close(); + } + _consumers.clear(); + } + + +} diff --git a/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession.java b/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession.java index c25eb1f2c3..2136d565f1 100644 --- a/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession.java +++ b/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession.java @@ -23,13 +23,15 @@ package org.apache.qpid.client; import org.apache.log4j.Logger; import org.apache.qpid.AMQException; import org.apache.qpid.AMQUndeliveredException; +import org.apache.qpid.AMQInvalidSelectorException; +import org.apache.qpid.common.AMQPFilterTypes; import org.apache.qpid.client.failover.FailoverSupport; import org.apache.qpid.client.message.AbstractJMSMessage; import org.apache.qpid.client.message.JMSStreamMessage; import org.apache.qpid.client.message.MessageFactoryRegistry; import org.apache.qpid.client.message.UnprocessedMessage; -import org.apache.qpid.client.protocol.AMQMethodEvent; import org.apache.qpid.client.protocol.AMQProtocolHandler; +import org.apache.qpid.client.protocol.AMQMethodEvent; import org.apache.qpid.client.util.FlowControllingBlockingQueue; import org.apache.qpid.exchange.ExchangeDefaults; import org.apache.qpid.framing.*; @@ -69,15 +71,15 @@ public class AMQSession extends Closeable implements Session, QueueSession, Topi private int _defaultPrefetchLowMark = DEFAULT_PREFETCH_LOW_MARK; /** - * Used to reference durable subscribers so they requests for unsubscribe can be handled - * correctly. Note this only keeps a record of subscriptions which have been created - * in the current instance. It does not remember subscriptions between executions of the - * client + * Used to reference durable subscribers so they requests for unsubscribe can be handled + * correctly. Note this only keeps a record of subscriptions which have been created + * in the current instance. It does not remember subscriptions between executions of the + * client */ private final ConcurrentHashMap _subscriptions = new ConcurrentHashMap(); private final ConcurrentHashMap _reverseSubscriptionMap = - new ConcurrentHashMap(); + new ConcurrentHashMap(); /** * Used in the consume method. We generate the consume tag on the client so that we can use the nowait @@ -143,6 +145,7 @@ public class AMQSession extends Closeable implements Session, QueueSession, Topi private boolean _inRecovery; + /** * Responsible for decoding a message fragment and passing it to the appropriate message consumer. */ @@ -176,7 +179,7 @@ public class AMQSession extends Closeable implements Session, QueueSession, Topi { if (message.deliverBody != null) { - final BasicMessageConsumer consumer = _consumers.get(message.deliverBody.consumerTag); + final BasicMessageConsumer consumer = (BasicMessageConsumer) _consumers.get(message.deliverBody.consumerTag); if (consumer == null) { @@ -210,17 +213,15 @@ public class AMQSession extends Closeable implements Session, QueueSession, Topi { _connection.exceptionReceived(new AMQNoConsumersException("Error: " + reason, bouncedMessage)); } + else if (errorCode == AMQConstant.NO_ROUTE.getCode()) + { + _connection.exceptionReceived(new AMQNoRouteException("Error: " + reason, bouncedMessage)); + } else { - if (errorCode == AMQConstant.NO_ROUTE.getCode()) - { - _connection.exceptionReceived(new AMQNoRouteException("Error: " + reason, bouncedMessage)); - } - else - { - _connection.exceptionReceived(new AMQUndeliveredException(errorCode, "Error: " + reason, bouncedMessage)); - } + _connection.exceptionReceived(new AMQUndeliveredException(errorCode, "Error: " + reason, bouncedMessage)); } + } catch (Exception e) { @@ -318,7 +319,7 @@ public class AMQSession extends Closeable implements Session, QueueSession, Topi public BytesMessage createBytesMessage() throws JMSException { - synchronized (_connection.getFailoverMutex()) + synchronized(_connection.getFailoverMutex()) { checkNotClosed(); try @@ -334,7 +335,7 @@ public class AMQSession extends Closeable implements Session, QueueSession, Topi public MapMessage createMapMessage() throws JMSException { - synchronized (_connection.getFailoverMutex()) + synchronized(_connection.getFailoverMutex()) { checkNotClosed(); try @@ -350,7 +351,7 @@ public class AMQSession extends Closeable implements Session, QueueSession, Topi public javax.jms.Message createMessage() throws JMSException { - synchronized (_connection.getFailoverMutex()) + synchronized(_connection.getFailoverMutex()) { checkNotClosed(); try @@ -366,7 +367,7 @@ public class AMQSession extends Closeable implements Session, QueueSession, Topi public ObjectMessage createObjectMessage() throws JMSException { - synchronized (_connection.getFailoverMutex()) + synchronized(_connection.getFailoverMutex()) { checkNotClosed(); try @@ -382,7 +383,7 @@ public class AMQSession extends Closeable implements Session, QueueSession, Topi public ObjectMessage createObjectMessage(Serializable object) throws JMSException { - synchronized (_connection.getFailoverMutex()) + synchronized(_connection.getFailoverMutex()) { checkNotClosed(); try @@ -400,7 +401,7 @@ public class AMQSession extends Closeable implements Session, QueueSession, Topi public StreamMessage createStreamMessage() throws JMSException { - synchronized (_connection.getFailoverMutex()) + synchronized(_connection.getFailoverMutex()) { checkNotClosed(); @@ -417,7 +418,7 @@ public class AMQSession extends Closeable implements Session, QueueSession, Topi public TextMessage createTextMessage() throws JMSException { - synchronized (_connection.getFailoverMutex()) + synchronized(_connection.getFailoverMutex()) { checkNotClosed(); @@ -434,7 +435,7 @@ public class AMQSession extends Closeable implements Session, QueueSession, Topi public TextMessage createTextMessage(String text) throws JMSException { - synchronized (_connection.getFailoverMutex()) + synchronized(_connection.getFailoverMutex()) { checkNotClosed(); try @@ -504,7 +505,7 @@ public class AMQSession extends Closeable implements Session, QueueSession, Topi { // We must close down all producers and consumers in an orderly fashion. This is the only method // that can be called from a different thread of control from the one controlling the session - synchronized (_connection.getFailoverMutex()) + synchronized(_connection.getFailoverMutex()) { //Ensure we only try and close an open session. if (!_closed.getAndSet(true)) @@ -569,7 +570,7 @@ public class AMQSession extends Closeable implements Session, QueueSession, Topi */ public void closed(Throwable e) { - synchronized (_connection.getFailoverMutex()) + synchronized(_connection.getFailoverMutex()) { // An AMQException has an error code and message already and will be passed in when closure occurs as a // result of a channel close request @@ -721,11 +722,11 @@ public class AMQSession extends Closeable implements Session, QueueSession, Topi public void acknowledge() throws JMSException { - if(isClosed()) + if (isClosed()) { throw new IllegalStateException("Session is already closed"); } - for(BasicMessageConsumer consumer : _consumers.values()) + for (BasicMessageConsumer consumer : _consumers.values()) { consumer.acknowledge(); } @@ -734,7 +735,6 @@ public class AMQSession extends Closeable implements Session, QueueSession, Topi } - public MessageListener getMessageListener() throws JMSException { checkNotClosed(); @@ -843,7 +843,9 @@ public class AMQSession extends Closeable implements Session, QueueSession, Topi false, false, null, - null); + null, + false, + false); } public MessageConsumer createConsumer(Destination destination, String messageSelector) throws JMSException @@ -855,7 +857,9 @@ public class AMQSession extends Closeable implements Session, QueueSession, Topi false, false, messageSelector, - null); + null, + false, + false); } public MessageConsumer createConsumer(Destination destination, String messageSelector, boolean noLocal) @@ -868,7 +872,26 @@ public class AMQSession extends Closeable implements Session, QueueSession, Topi noLocal, false, messageSelector, - null); + null, + false, + false); + } + + public MessageConsumer createBrowserConsumer(Destination destination, + String messageSelector, + boolean noLocal) + throws JMSException + { + checkValidDestination(destination); + return createConsumerImpl(destination, + _defaultPrefetchHighMark, + _defaultPrefetchLowMark, + noLocal, + false, + messageSelector, + null, + true, + true); } public MessageConsumer createConsumer(Destination destination, @@ -878,7 +901,7 @@ public class AMQSession extends Closeable implements Session, QueueSession, Topi String selector) throws JMSException { checkValidDestination(destination); - return createConsumerImpl(destination, prefetch, prefetch, noLocal, exclusive, selector, null); + return createConsumerImpl(destination, prefetch, prefetch, noLocal, exclusive, selector, null, false, false); } @@ -890,7 +913,7 @@ public class AMQSession extends Closeable implements Session, QueueSession, Topi String selector) throws JMSException { checkValidDestination(destination); - return createConsumerImpl(destination, prefetchHigh, prefetchLow, noLocal, exclusive, selector, null); + return createConsumerImpl(destination, prefetchHigh, prefetchLow, noLocal, exclusive, selector, null, false, false); } public MessageConsumer createConsumer(Destination destination, @@ -902,7 +925,7 @@ public class AMQSession extends Closeable implements Session, QueueSession, Topi { checkValidDestination(destination); return createConsumerImpl(destination, prefetch, prefetch, noLocal, exclusive, - selector, rawSelector); + selector, rawSelector, false, false); } public MessageConsumer createConsumer(Destination destination, @@ -915,7 +938,7 @@ public class AMQSession extends Closeable implements Session, QueueSession, Topi { checkValidDestination(destination); return createConsumerImpl(destination, prefetchHigh, prefetchLow, noLocal, exclusive, - selector, rawSelector); + selector, rawSelector, false, false); } protected MessageConsumer createConsumerImpl(final Destination destination, @@ -924,7 +947,9 @@ public class AMQSession extends Closeable implements Session, QueueSession, Topi final boolean noLocal, final boolean exclusive, final String selector, - final FieldTable rawSelector) throws JMSException + final FieldTable rawSelector, + final boolean noConsume, + final boolean autoClose) throws JMSException { checkTemporaryDestination(destination); @@ -948,12 +973,18 @@ public class AMQSession extends Closeable implements Session, QueueSession, Topi BasicMessageConsumer consumer = new BasicMessageConsumer(_channelId, _connection, amqd, selector, noLocal, _messageFactoryRegistry, AMQSession.this, protocolHandler, ft, prefetchHigh, prefetchLow, exclusive, - _acknowledgeMode); + _acknowledgeMode, noConsume, autoClose); try { registerConsumer(consumer, false); } + catch (AMQInvalidSelectorException ise) + { + JMSException ex = new InvalidSelectorException(ise.getMessage()); + ex.setLinkedException(ise); + throw ex; + } catch (AMQException e) { JMSException ex = new JMSException("Error registering consumer: " + e); @@ -963,7 +994,7 @@ public class AMQSession extends Closeable implements Session, QueueSession, Topi synchronized(destination) { - _destinationConsumerCount.putIfAbsent(destination,new AtomicInteger()); + _destinationConsumerCount.putIfAbsent(destination, new AtomicInteger()); _destinationConsumerCount.get(destination).incrementAndGet(); } @@ -975,16 +1006,16 @@ public class AMQSession extends Closeable implements Session, QueueSession, Topi private void checkTemporaryDestination(Destination destination) throws JMSException { - if((destination instanceof TemporaryDestination)) + if ((destination instanceof TemporaryDestination)) { _logger.debug("destination is temporary"); final TemporaryDestination tempDest = (TemporaryDestination) destination; - if(tempDest.getSession() != this) + if (tempDest.getSession() != this) { _logger.debug("destination is on different session"); throw new JMSException("Cannot consume from a temporary destination created onanother session"); } - if(tempDest.isDeleted()) + if (tempDest.isDeleted()) { _logger.debug("destination is deleted"); throw new JMSException("Cannot consume from a deleted destination"); @@ -1065,12 +1096,26 @@ public class AMQSession extends Closeable implements Session, QueueSession, Topi * @return the consumer tag generated by the broker */ private void consumeFromQueue(BasicMessageConsumer consumer, String queueName, AMQProtocolHandler protocolHandler, - boolean nowait) throws AMQException + boolean nowait, String messageSelector) throws AMQException { //fixme prefetch values are not used here. Do we need to have them as parametsrs? //need to generate a consumer tag on the client so we can exploit the nowait flag String tag = Integer.toString(_nextTag++); + FieldTable arguments = FieldTableFactory.newFieldTable(); + if (messageSelector != null && !messageSelector.equals("")) + { + arguments.put(AMQPFilterTypes.JMS_SELECTOR.getValue(), messageSelector); + } + if(consumer.isAutoClose()) + { + arguments.put(AMQPFilterTypes.AUTO_CLOSE.getValue(), Boolean.TRUE); + } + if(consumer.isNoConsume()) + { + arguments.put(AMQPFilterTypes.NO_CONSUME.getValue(), Boolean.TRUE); + } + consumer.setConsumerTag(tag); // we must register the consumer in the map before we actually start listening _consumers.put(tag, consumer); @@ -1080,7 +1125,7 @@ public class AMQSession extends Closeable implements Session, QueueSession, Topi AMQFrame jmsConsume = BasicConsumeBody.createAMQFrame(_channelId, 0, queueName, tag, consumer.isNoLocal(), consumer.getAcknowledgeMode() == Session.NO_ACKNOWLEDGE, - consumer.isExclusive(), nowait); + consumer.isExclusive(), nowait, arguments); if (nowait) { protocolHandler.writeFrame(jmsConsume); @@ -1220,7 +1265,7 @@ public class AMQSession extends Closeable implements Session, QueueSession, Topi { checkNotClosed(); checkValidTopic(topic); - AMQTopic dest = AMQTopic.createDurableTopic((AMQTopic)topic, name, _connection); + AMQTopic dest = AMQTopic.createDurableTopic((AMQTopic) topic, name, _connection); TopicSubscriberAdaptor subscriber = _subscriptions.get(name); if (subscriber != null) { @@ -1247,8 +1292,8 @@ public class AMQSession extends Closeable implements Session, QueueSession, Topi subscriber = new TopicSubscriberAdaptor(dest, (BasicMessageConsumer) createConsumer(dest)); - _subscriptions.put(name,subscriber); - _reverseSubscriptionMap.put(subscriber.getMessageConsumer(),name); + _subscriptions.put(name, subscriber); + _reverseSubscriptionMap.put(subscriber.getMessageConsumer(), name); return subscriber; } @@ -1278,8 +1323,8 @@ public class AMQSession extends Closeable implements Session, QueueSession, Topi AMQTopic dest = AMQTopic.createDurableTopic((AMQTopic) topic, name, _connection); BasicMessageConsumer consumer = (BasicMessageConsumer) createConsumer(dest, messageSelector, noLocal); TopicSubscriberAdaptor subscriber = new TopicSubscriberAdaptor(dest, consumer); - _subscriptions.put(name,subscriber); - _reverseSubscriptionMap.put(subscriber.getMessageConsumer(),name); + _subscriptions.put(name, subscriber); + _reverseSubscriptionMap.put(subscriber.getMessageConsumer(), name); return subscriber; } @@ -1291,16 +1336,14 @@ public class AMQSession extends Closeable implements Session, QueueSession, Topi public QueueBrowser createBrowser(Queue queue) throws JMSException { - checkNotClosed(); - checkValidQueue(queue); - throw new UnsupportedOperationException("Queue browsing not supported"); + return createBrowser(queue, null); } public QueueBrowser createBrowser(Queue queue, String messageSelector) throws JMSException { checkNotClosed(); checkValidQueue(queue); - throw new UnsupportedOperationException("Queue browsing not supported"); + return new AMQQueueBrowser(this, (AMQQueue) queue,messageSelector); } public TemporaryQueue createTemporaryQueue() throws JMSException @@ -1476,7 +1519,14 @@ public class AMQSession extends Closeable implements Session, QueueSession, Topi bindQueue(amqd, queueName, protocolHandler, consumer.getRawSelectorFieldTable()); - consumeFromQueue(consumer, queueName, protocolHandler, nowait); + try + { + consumeFromQueue(consumer, queueName, protocolHandler, nowait, consumer.getMessageSelector()); + } + catch (JMSException e) //thrown by getMessageSelector + { + throw new AMQException(e.getMessage(), e); + } } /** @@ -1489,7 +1539,7 @@ public class AMQSession extends Closeable implements Session, QueueSession, Topi { _consumers.remove(consumer.getConsumerTag()); String subscriptionName = _reverseSubscriptionMap.remove(consumer); - if(subscriptionName != null) + if (subscriptionName != null) { _subscriptions.remove(subscriptionName); } @@ -1497,7 +1547,7 @@ public class AMQSession extends Closeable implements Session, QueueSession, Topi Destination dest = consumer.getDestination(); synchronized(dest) { - if(_destinationConsumerCount.get(dest).decrementAndGet() == 0) + if (_destinationConsumerCount.get(dest).decrementAndGet() == 0) { _destinationConsumerCount.remove(dest); } @@ -1567,6 +1617,16 @@ public class AMQSession extends Closeable implements Session, QueueSession, Topi _connection.getProtocolHandler().writeFrame(channelFlowFrame); } + public void confirmConsumerCancelled(String consumerTag) + { + BasicMessageConsumer consumer = (BasicMessageConsumer) _consumers.get(consumerTag); + if((consumer != null) && (consumer.isAutoClose())) + { + consumer.closeWhenNoMessages(true); + } + } + + /* * I could have combined the last 3 methods, but this way it improves readability */ @@ -1576,7 +1636,7 @@ public class AMQSession extends Closeable implements Session, QueueSession, Topi { throw new javax.jms.InvalidDestinationException("Invalid Topic"); } - if((topic instanceof TemporaryDestination) && ((TemporaryDestination)topic).getSession() != this) + if ((topic instanceof TemporaryDestination) && ((TemporaryDestination) topic).getSession() != this) { throw new JMSException("Cannot create a subscription on a temporary topic created in another session"); } @@ -1597,4 +1657,5 @@ public class AMQSession extends Closeable implements Session, QueueSession, Topi throw new javax.jms.InvalidDestinationException("Invalid Queue"); } } + } diff --git a/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer.java b/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer.java index d3d9db3806..cefaca8d52 100644 --- a/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer.java +++ b/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer.java @@ -22,6 +22,8 @@ package org.apache.qpid.client; import org.apache.log4j.Logger; import org.apache.qpid.AMQException; +import org.apache.qpid.url.AMQBindingURL; +import org.apache.qpid.url.URLSyntaxException; import org.apache.qpid.client.message.AbstractJMSMessage; import org.apache.qpid.client.message.MessageFactoryRegistry; import org.apache.qpid.client.message.UnprocessedMessage; @@ -39,6 +41,7 @@ import javax.jms.MessageListener; import java.util.Iterator; import java.util.concurrent.ArrayBlockingQueue; import java.util.concurrent.ConcurrentLinkedQueue; +import javax.jms.Destination; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicReference; @@ -142,10 +145,19 @@ public class BasicMessageConsumer extends Closeable implements MessageConsumer */ private Thread _receivingThread; + /** + * autoClose denotes that the consumer will automatically cancel itself when there are no more messages to receive + * on the queue. This is used for queue browsing. + */ + private boolean _autoClose; + private boolean _closeWhenNoMessages; + + private boolean _noConsume; + protected BasicMessageConsumer(int channelId, AMQConnection connection, AMQDestination destination, String messageSelector, - boolean noLocal, MessageFactoryRegistry messageFactory, AMQSession session, - AMQProtocolHandler protocolHandler, FieldTable rawSelectorFieldTable, - int prefetchHigh, int prefetchLow, boolean exclusive, int acknowledgeMode) + boolean noLocal, MessageFactoryRegistry messageFactory, AMQSession session, + AMQProtocolHandler protocolHandler, FieldTable rawSelectorFieldTable, + int prefetchHigh, int prefetchLow, boolean exclusive, int acknowledgeMode, boolean noConsume, boolean autoClose) { _channelId = channelId; _connection = connection; @@ -161,6 +173,8 @@ public class BasicMessageConsumer extends Closeable implements MessageConsumer _exclusive = exclusive; _acknowledgeMode = acknowledgeMode; _synchronousQueue = new ArrayBlockingQueue(prefetchHigh, true); + _autoClose = autoClose; + _noConsume = noConsume; } public AMQDestination getDestination() @@ -241,6 +255,17 @@ public class BasicMessageConsumer extends Closeable implements MessageConsumer if(_session.getAcknowledgeMode() == Session.CLIENT_ACKNOWLEDGE) { _unacknowledgedDeliveryTags.add(jmsMsg.getDeliveryTag()); + String url = jmsMsg.getStringProperty(CustomJMXProperty.JMSX_QPID_JMSDESTINATIONURL.toString()); + try + { + Destination dest = AMQDestination.createDestination(new AMQBindingURL(url)); + jmsMsg.setJMSDestination(dest); + } + catch (URLSyntaxException e) + { + _logger.warn("Unable to parse the supplied destination header: " + url); + } + } _session.setInRecovery(false); } @@ -307,6 +332,10 @@ public class BasicMessageConsumer extends Closeable implements MessageConsumer try { + if(closeOnAutoClose()) + { + return null; + } Object o = null; if (l > 0) { @@ -336,6 +365,19 @@ public class BasicMessageConsumer extends Closeable implements MessageConsumer } } + private boolean closeOnAutoClose() throws JMSException + { + if(isAutoClose() && _closeWhenNoMessages && _synchronousQueue.isEmpty()) + { + close(false); + return true; + } + else + { + return false; + } + } + public Message receiveNoWait() throws JMSException { checkPreConditions(); @@ -344,6 +386,10 @@ public class BasicMessageConsumer extends Closeable implements MessageConsumer try { + if(closeOnAutoClose()) + { + return null; + } Object o = _synchronousQueue.poll(); final AbstractJMSMessage m = returnMessageOrThrow(o); if (m != null) @@ -388,22 +434,31 @@ public class BasicMessageConsumer extends Closeable implements MessageConsumer } } + public void close() throws JMSException + { + close(true); + } + + public void close(boolean sendClose) throws JMSException { synchronized(_connection.getFailoverMutex()) { if (!_closed.getAndSet(true)) { - final AMQFrame cancelFrame = BasicCancelBody.createAMQFrame(_channelId, _consumerTag, false); - - try + if(sendClose) { - _protocolHandler.syncWrite(cancelFrame, BasicCancelOkBody.class); - } - catch (AMQException e) - { - _logger.error("Error closing consumer: " + e, e); - throw new JMSException("Error closing consumer: " + e); + final AMQFrame cancelFrame = BasicCancelBody.createAMQFrame(_channelId, _consumerTag, false); + + try + { + _protocolHandler.syncWrite(cancelFrame, BasicCancelOkBody.class); + } + catch (AMQException e) + { + _logger.error("Error closing consumer: " + e, e); + throw new JMSException("Error closing consumer: " + e); + } } deregisterConsumer(); @@ -499,6 +554,12 @@ public class BasicMessageConsumer extends Closeable implements MessageConsumer msg.setJMSDestination(_destination); switch (_acknowledgeMode) { + case Session.CLIENT_ACKNOWLEDGE: + if (isNoConsume()) + { + _session.acknowledgeMessage(msg.getDeliveryTag(), false); + } + break; case Session.DUPS_OK_ACKNOWLEDGE: if (++_outstanding >= _prefetchHigh) { @@ -525,7 +586,14 @@ public class BasicMessageConsumer extends Closeable implements MessageConsumer } break; case Session.SESSION_TRANSACTED: - _lastDeliveryTag = msg.getDeliveryTag(); + if (isNoConsume()) + { + _session.acknowledgeMessage(msg.getDeliveryTag(), false); + } + else + { + _lastDeliveryTag = msg.getDeliveryTag(); + } break; } } @@ -616,4 +684,29 @@ public class BasicMessageConsumer extends Closeable implements MessageConsumer { _unacknowledgedDeliveryTags.clear(); } + + public boolean isAutoClose() + { + return _autoClose; + } + + + public boolean isNoConsume() + { + return _noConsume; + } + + public void closeWhenNoMessages(boolean b) + { + _closeWhenNoMessages = b; + + if(_closeWhenNoMessages + && _synchronousQueue.isEmpty() + && _receiving.get() + && _messageListener != null) + { + _receivingThread.interrupt(); + } + + } } diff --git a/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageProducer.java b/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageProducer.java index e11d70cf41..7a5fcbccf9 100644 --- a/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageProducer.java +++ b/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageProducer.java @@ -507,8 +507,11 @@ public class BasicMessageProducer extends Closeable implements org.apache.qpid.j long timeToLive, boolean mandatory, boolean immediate, boolean wait) throws JMSException { checkTemporaryDestination(destination); + origMessage.setJMSDestination(destination); + AbstractJMSMessage message = convertToNativeMessage(origMessage); + message.getJmsContentHeaderProperties().getJMSHeaders().setString(CustomJMXProperty.JMSX_QPID_JMSDESTINATIONURL.toString(), destination.toURL()); AMQFrame publishFrame = BasicPublishBody.createAMQFrame(_channelId, 0, destination.getExchangeName(), destination.getRoutingKey(), mandatory, immediate); diff --git a/qpid/java/client/src/main/java/org/apache/qpid/client/CustomJMXProperty.java b/qpid/java/client/src/main/java/org/apache/qpid/client/CustomJMXProperty.java new file mode 100644 index 0000000000..3a7b7a7b3d --- /dev/null +++ b/qpid/java/client/src/main/java/org/apache/qpid/client/CustomJMXProperty.java @@ -0,0 +1,47 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + * + */ +package org.apache.qpid.client; + +import java.util.*; + +public enum CustomJMXProperty +{ + JMSX_QPID_JMSDESTINATIONURL, + JMSXGroupID, + JMSXGroupSeq; + + private static Enumeration _names; + + public static synchronized Enumeration asEnumeration() + { + if(_names == null) + { + CustomJMXProperty[] properties = values(); + ArrayList nameList = new ArrayList(properties.length); + for(CustomJMXProperty property : properties) + { + nameList.add(property.toString()); + } + _names = Collections.enumeration(nameList); + } + return _names; + } +} diff --git a/qpid/java/client/src/main/java/org/apache/qpid/client/QpidConnectionMetaData.java b/qpid/java/client/src/main/java/org/apache/qpid/client/QpidConnectionMetaData.java index 10a65c2ad8..d9e946c397 100644 --- a/qpid/java/client/src/main/java/org/apache/qpid/client/QpidConnectionMetaData.java +++ b/qpid/java/client/src/main/java/org/apache/qpid/client/QpidConnectionMetaData.java @@ -1,50 +1,97 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + * + */ package org.apache.qpid.client; +import org.apache.qpid.common.QpidProperties; + import java.util.Enumeration; import javax.jms.ConnectionMetaData; import javax.jms.JMSException; -public class QpidConnectionMetaData implements ConnectionMetaData { - - private static QpidConnectionMetaData _instance = new QpidConnectionMetaData(); - - private QpidConnectionMetaData(){ - } - - public static QpidConnectionMetaData instance(){ - return _instance; - } - - public int getJMSMajorVersion() throws JMSException { - return 1; - } - - public int getJMSMinorVersion() throws JMSException { - return 1; - } - - public String getJMSProviderName() throws JMSException { - return "Apache Qpid"; - } - - public String getJMSVersion() throws JMSException { - return "1.1"; - } - - public Enumeration getJMSXPropertyNames() throws JMSException { - return null; - } - - public int getProviderMajorVersion() throws JMSException { - return 0; - } - - public int getProviderMinorVersion() throws JMSException { - return 9; - } - - public String getProviderVersion() throws JMSException { - return "Incubating-M1"; - } +public class QpidConnectionMetaData implements ConnectionMetaData +{ + + + QpidConnectionMetaData(AMQConnection conn) + { + } + + public int getJMSMajorVersion() throws JMSException + { + return 1; + } + + public int getJMSMinorVersion() throws JMSException + { + return 1; + } + + public String getJMSProviderName() throws JMSException + { + return "Apache " + QpidProperties.getProductName(); + } + + public String getJMSVersion() throws JMSException + { + return "1.1"; + } + + public Enumeration getJMSXPropertyNames() throws JMSException + { + return CustomJMXProperty.asEnumeration(); + } + + public int getProviderMajorVersion() throws JMSException + { + return 0; + } + + public int getProviderMinorVersion() throws JMSException + { + return 8; + } + + public String getProviderVersion() throws JMSException + { + return QpidProperties.getProductName() + " (Client: [" + getClientVersion() + "] ; Broker [" + getBrokerVersion() + "] ; Protocol: [ " + + getProtocolVersion() + "] )"; + } + + private String getProtocolVersion() + { + // TODO - Implement based on connection negotiated protocol + return "0.8"; + } + + public String getBrokerVersion() + { + // TODO - get broker version + return ""; + } + + public String getClientVersion() + { + return QpidProperties.getBuildVerision(); + } + + } diff --git a/qpid/java/client/src/main/java/org/apache/qpid/client/handler/BasicCancelOkMethodHandler.java b/qpid/java/client/src/main/java/org/apache/qpid/client/handler/BasicCancelOkMethodHandler.java new file mode 100644 index 0000000000..d855e97204 --- /dev/null +++ b/qpid/java/client/src/main/java/org/apache/qpid/client/handler/BasicCancelOkMethodHandler.java @@ -0,0 +1,35 @@ +package org.apache.qpid.client.handler; + +import org.apache.qpid.client.state.StateAwareMethodListener; +import org.apache.qpid.client.state.AMQStateManager; +import org.apache.qpid.client.protocol.AMQMethodEvent; +import org.apache.qpid.client.BasicMessageConsumer; +import org.apache.qpid.AMQException; +import org.apache.qpid.framing.ExchangeBoundOkBody; +import org.apache.qpid.framing.BasicCancelOkBody; +import org.apache.log4j.Logger; + +/** + * @author Apache Software Foundation + */ +public class BasicCancelOkMethodHandler implements StateAwareMethodListener +{ + private static final Logger _logger = Logger.getLogger(BasicCancelOkMethodHandler.class); + private static final BasicCancelOkMethodHandler _instance = new BasicCancelOkMethodHandler(); + + public static BasicCancelOkMethodHandler getInstance() + { + return _instance; + } + + private BasicCancelOkMethodHandler() + { + } + + public void methodReceived(AMQStateManager stateManager, AMQMethodEvent evt) throws AMQException + { + _logger.debug("New BasicCancelOk method received"); + BasicCancelOkBody body = (BasicCancelOkBody) evt.getMethod(); + evt.getProtocolSession().confirmConsumerCancelled(evt.getChannelId(), body.consumerTag); + } +} diff --git a/qpid/java/client/src/main/java/org/apache/qpid/client/handler/ChannelCloseMethodHandler.java b/qpid/java/client/src/main/java/org/apache/qpid/client/handler/ChannelCloseMethodHandler.java index 2bd93f1508..fd2968cdfd 100644 --- a/qpid/java/client/src/main/java/org/apache/qpid/client/handler/ChannelCloseMethodHandler.java +++ b/qpid/java/client/src/main/java/org/apache/qpid/client/handler/ChannelCloseMethodHandler.java @@ -23,6 +23,7 @@ package org.apache.qpid.client.handler; import org.apache.log4j.Logger; import org.apache.qpid.AMQChannelClosedException; import org.apache.qpid.AMQException; +import org.apache.qpid.AMQInvalidSelectorException; import org.apache.qpid.client.AMQNoConsumersException; import org.apache.qpid.client.AMQNoRouteException; import org.apache.qpid.protocol.AMQConstant; @@ -46,7 +47,7 @@ public class ChannelCloseMethodHandler implements StateAwareMethodListener public void methodReceived(AMQStateManager stateManager, AMQMethodEvent evt) throws AMQException { - _logger.debug("ChannelClose method received"); + _logger.debug("ChannelClose method received"); ChannelCloseBody method = (ChannelCloseBody) evt.getMethod(); int errorCode = method.replyCode; @@ -65,17 +66,21 @@ public class ChannelCloseMethodHandler implements StateAwareMethodListener { throw new AMQNoConsumersException("Error: " + reason, null); } + else if (errorCode == AMQConstant.NO_ROUTE.getCode()) + { + throw new AMQNoRouteException("Error: " + reason, null); + } + else if (errorCode == AMQConstant.INVALID_SELECTOR.getCode()) + { + _logger.info("Broker responded with Invalid Selector."); + + throw new AMQInvalidSelectorException(reason); + } else { - if (errorCode == AMQConstant.NO_ROUTE.getCode()) - { - throw new AMQNoRouteException("Error: " + reason, null); - } - else - { - throw new AMQChannelClosedException(errorCode, "Error: " + reason); - } + throw new AMQChannelClosedException(errorCode, "Error: " + reason); } + } evt.getProtocolSession().channelClosed(evt.getChannelId(), errorCode, reason); } diff --git a/qpid/java/client/src/main/java/org/apache/qpid/client/handler/ConnectionOpenOkMethodHandler.java b/qpid/java/client/src/main/java/org/apache/qpid/client/handler/ConnectionOpenOkMethodHandler.java index 8785e7d44e..b5001a6e64 100644 --- a/qpid/java/client/src/main/java/org/apache/qpid/client/handler/ConnectionOpenOkMethodHandler.java +++ b/qpid/java/client/src/main/java/org/apache/qpid/client/handler/ConnectionOpenOkMethodHandler.java @@ -20,20 +20,14 @@ */ package org.apache.qpid.client.handler; -import org.apache.log4j.Logger; import org.apache.qpid.AMQException; import org.apache.qpid.client.protocol.AMQMethodEvent; -import org.apache.qpid.client.protocol.AMQProtocolSession; import org.apache.qpid.client.state.AMQState; import org.apache.qpid.client.state.AMQStateManager; import org.apache.qpid.client.state.StateAwareMethodListener; -import org.apache.qpid.framing.ConnectionOpenOkBody; public class ConnectionOpenOkMethodHandler implements StateAwareMethodListener { - - private static final Logger _logger = Logger.getLogger(ConnectionOpenOkMethodHandler.class); - private static final ConnectionOpenOkMethodHandler _instance = new ConnectionOpenOkMethodHandler(); public static ConnectionOpenOkMethodHandler getInstance() @@ -47,8 +41,6 @@ public class ConnectionOpenOkMethodHandler implements StateAwareMethodListener public void methodReceived(AMQStateManager stateManager, AMQMethodEvent evt) throws AMQException { - AMQProtocolSession session = evt.getProtocolSession(); - ConnectionOpenOkBody method = (ConnectionOpenOkBody) evt.getMethod(); stateManager.changeState(AMQState.CONNECTION_OPEN); } diff --git a/qpid/java/client/src/main/java/org/apache/qpid/client/handler/ConnectionStartMethodHandler.java b/qpid/java/client/src/main/java/org/apache/qpid/client/handler/ConnectionStartMethodHandler.java index 9333df3fe4..f7b0cb5331 100644 --- a/qpid/java/client/src/main/java/org/apache/qpid/client/handler/ConnectionStartMethodHandler.java +++ b/qpid/java/client/src/main/java/org/apache/qpid/client/handler/ConnectionStartMethodHandler.java @@ -22,6 +22,8 @@ package org.apache.qpid.client.handler; import org.apache.log4j.Logger; import org.apache.qpid.AMQException; +import org.apache.qpid.common.ClientProperties; +import org.apache.qpid.common.QpidProperties; import org.apache.qpid.client.protocol.AMQMethodEvent; import org.apache.qpid.client.protocol.AMQProtocolSession; import org.apache.qpid.client.security.AMQCallbackHandler; @@ -119,10 +121,11 @@ public class ConnectionStartMethodHandler implements StateAwareMethodListener stateManager.changeState(AMQState.CONNECTION_NOT_TUNED); FieldTable clientProperties = FieldTableFactory.newFieldTable(); - clientProperties.put("instance", ps.getClientID()); - clientProperties.put("product", "Qpid"); - clientProperties.put("version", "1.0"); - clientProperties.put("platform", getFullSystemInfo()); + + clientProperties.put(ClientProperties.instance.toString(), ps.getClientID()); + clientProperties.put(ClientProperties.product.toString(), QpidProperties.getProductName()); + clientProperties.put(ClientProperties.version.toString(), QpidProperties.getReleaseVerision()); + clientProperties.put(ClientProperties.platform.toString(), getFullSystemInfo()); ps.writeFrame(ConnectionStartOkBody.createAMQFrame(evt.getChannelId(), clientProperties, mechanism, saslResponse, selectedLocale)); } diff --git a/qpid/java/client/src/main/java/org/apache/qpid/client/message/AbstractJMSMessage.java b/qpid/java/client/src/main/java/org/apache/qpid/client/message/AbstractJMSMessage.java index 23d6c0151e..40d8b28411 100644 --- a/qpid/java/client/src/main/java/org/apache/qpid/client/message/AbstractJMSMessage.java +++ b/qpid/java/client/src/main/java/org/apache/qpid/client/message/AbstractJMSMessage.java @@ -26,7 +26,8 @@ import org.apache.qpid.AMQException; import org.apache.qpid.url.BindingURL; import org.apache.qpid.url.AMQBindingURL; import org.apache.qpid.url.URLSyntaxException; -import org.apache.qpid.client.*; +import org.apache.qpid.client.AMQDestination; +import org.apache.qpid.client.BasicMessageConsumer; import org.apache.qpid.framing.BasicContentHeaderProperties; import org.apache.qpid.framing.FieldTable; @@ -46,7 +47,8 @@ public abstract class AbstractJMSMessage extends AMQMessage implements org.apach protected ByteBuffer _data; private boolean _readableProperties = false; - private boolean _readableMessage = false; + protected boolean _readableMessage = false; + protected boolean _changedData; private Destination _destination; private BasicMessageConsumer _consumer; @@ -60,6 +62,7 @@ public abstract class AbstractJMSMessage extends AMQMessage implements org.apach } _readableProperties = false; _readableMessage = (data != null); + _changedData = (data == null); } protected AbstractJMSMessage(long deliveryTag, BasicContentHeaderProperties contentHeader, ByteBuffer data) throws AMQException @@ -172,13 +175,12 @@ public abstract class AbstractJMSMessage extends AMQMessage implements org.apach public Destination getJMSDestination() throws JMSException { - // TODO: implement this once we have sorted out how to figure out the exchange class - return _destination; + return _destination; } public void setJMSDestination(Destination destination) throws JMSException { - _destination = destination; + _destination = destination; } public int getJMSDeliveryMode() throws JMSException @@ -522,16 +524,16 @@ public abstract class AbstractJMSMessage extends AMQMessage implements org.apach return !_readableMessage; } - public void reset() + public void reset() { - if (_readableMessage) + if (!_changedData) { _data.rewind(); } else { _data.flip(); - _readableMessage = true; + _changedData = false; } } diff --git a/qpid/java/client/src/main/java/org/apache/qpid/client/message/JMSBytesMessage.java b/qpid/java/client/src/main/java/org/apache/qpid/client/message/JMSBytesMessage.java index f5c9f7111a..d769300c69 100644 --- a/qpid/java/client/src/main/java/org/apache/qpid/client/message/JMSBytesMessage.java +++ b/qpid/java/client/src/main/java/org/apache/qpid/client/message/JMSBytesMessage.java @@ -59,6 +59,12 @@ public class JMSBytesMessage extends AbstractBytesMessage implements BytesMessag super(messageNbr, contentHeader, data); } + public void reset() + { + super.reset(); + _readableMessage = true; + } + public String getMimeType() { return MIME_TYPE; @@ -226,48 +232,56 @@ public class JMSBytesMessage extends AbstractBytesMessage implements BytesMessag public void writeBoolean(boolean b) throws JMSException { checkWritable(); + _changedData = true; _data.put(b ? (byte) 1 : (byte) 0); } public void writeByte(byte b) throws JMSException { checkWritable(); + _changedData = true; _data.put(b); } public void writeShort(short i) throws JMSException { checkWritable(); + _changedData = true; _data.putShort(i); } public void writeChar(char c) throws JMSException { checkWritable(); + _changedData = true; _data.putChar(c); } public void writeInt(int i) throws JMSException { checkWritable(); + _changedData = true; _data.putInt(i); } public void writeLong(long l) throws JMSException { checkWritable(); + _changedData = true; _data.putLong(l); } public void writeFloat(float v) throws JMSException { checkWritable(); + _changedData = true; _data.putFloat(v); } public void writeDouble(double v) throws JMSException { checkWritable(); + _changedData = true; _data.putDouble(v); } @@ -281,7 +295,7 @@ public class JMSBytesMessage extends AbstractBytesMessage implements BytesMessag _data.putShort((short)encodedString.limit()); _data.put(encodedString); - + _changedData = true; //_data.putString(string, Charset.forName("UTF-8").newEncoder()); // we must add the null terminator manually //_data.put((byte)0); @@ -298,12 +312,14 @@ public class JMSBytesMessage extends AbstractBytesMessage implements BytesMessag { checkWritable(); _data.put(bytes); + _changedData = true; } public void writeBytes(byte[] bytes, int offset, int length) throws JMSException { checkWritable(); _data.put(bytes, offset, length); + _changedData = true; } public void writeObject(Object object) throws JMSException diff --git a/qpid/java/client/src/main/java/org/apache/qpid/client/message/JMSObjectMessage.java b/qpid/java/client/src/main/java/org/apache/qpid/client/message/JMSObjectMessage.java index 4fb070d2ff..35c5377f14 100644 --- a/qpid/java/client/src/main/java/org/apache/qpid/client/message/JMSObjectMessage.java +++ b/qpid/java/client/src/main/java/org/apache/qpid/client/message/JMSObjectMessage.java @@ -112,7 +112,7 @@ public class JMSObjectMessage extends AbstractJMSMessage implements ObjectMessag } } - + public Serializable getObject() throws JMSException { ObjectInputStream in = null; @@ -123,18 +123,18 @@ public class JMSObjectMessage extends AbstractJMSMessage implements ObjectMessag try { - _data.rewind(); + _data.rewind(); in = new ObjectInputStream(_data.asInputStream()); return (Serializable) in.readObject(); } catch (IOException e) - { - e.printStackTrace(); - throw new MessageFormatException("Could not deserialize message: " + e); + { + e.printStackTrace(); + throw new MessageFormatException("Could not deserialize message: " + e); } catch (ClassNotFoundException e) { - e.printStackTrace(); + e.printStackTrace(); throw new MessageFormatException("Could not deserialize message: " + e); } finally diff --git a/qpid/java/client/src/main/java/org/apache/qpid/client/message/JMSStreamMessage.java b/qpid/java/client/src/main/java/org/apache/qpid/client/message/JMSStreamMessage.java index c2dfdc1b65..6709ff802d 100644 --- a/qpid/java/client/src/main/java/org/apache/qpid/client/message/JMSStreamMessage.java +++ b/qpid/java/client/src/main/java/org/apache/qpid/client/message/JMSStreamMessage.java @@ -86,6 +86,12 @@ public class JMSStreamMessage extends AbstractBytesMessage implements StreamMess super(messageNbr, contentHeader, data); } + public void reset() + { + super.reset(); + _readableMessage = true; + } + public String getMimeType() { return MIME_TYPE; @@ -103,6 +109,7 @@ public class JMSStreamMessage extends AbstractBytesMessage implements StreamMess { checkWritable(); _data.put(type); + _changedData = true; } public boolean readBoolean() throws JMSException @@ -693,7 +700,7 @@ public class JMSStreamMessage extends AbstractBytesMessage implements StreamMess { _data.putString(string, Charset.forName("UTF-8").newEncoder()); // we must write the null terminator ourselves - _data.put((byte)0); + _data.put((byte) 0); } catch (CharacterCodingException e) { @@ -706,7 +713,7 @@ public class JMSStreamMessage extends AbstractBytesMessage implements StreamMess public void writeBytes(byte[] bytes) throws JMSException { - writeBytes(bytes, 0, bytes == null?0:bytes.length); + writeBytes(bytes, 0, bytes == null ? 0 : bytes.length); } public void writeBytes(byte[] bytes, int offset, int length) throws JMSException diff --git a/qpid/java/client/src/main/java/org/apache/qpid/client/message/JMSTextMessage.java b/qpid/java/client/src/main/java/org/apache/qpid/client/message/JMSTextMessage.java index 76f8a1c32f..d8394b0489 100644 --- a/qpid/java/client/src/main/java/org/apache/qpid/client/message/JMSTextMessage.java +++ b/qpid/java/client/src/main/java/org/apache/qpid/client/message/JMSTextMessage.java @@ -117,6 +117,7 @@ public class JMSTextMessage extends AbstractJMSMessage implements javax.jms.Text { _data.put(text.getBytes(getJmsContentHeaderProperties().getEncoding())); } + _changedData=true; } _decodedValue = text; } diff --git a/qpid/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolSession.java b/qpid/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolSession.java index a4ed89719b..6a40fd3133 100644 --- a/qpid/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolSession.java +++ b/qpid/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolSession.java @@ -406,4 +406,12 @@ public class AMQProtocolSession implements ProtocolVersionList HeartbeatDiagnostics.init(delay, HeartbeatConfig.CONFIG.getTimeout(delay)); } } + + public void confirmConsumerCancelled(int channelId, String consumerTag) + { + final Integer chId = channelId; + final AMQSession session = (AMQSession) _channelId2SessionMap.get(chId); + + session.confirmConsumerCancelled(consumerTag); + } } diff --git a/qpid/java/client/src/main/java/org/apache/qpid/client/protocol/BlockingMethodFrameListener.java b/qpid/java/client/src/main/java/org/apache/qpid/client/protocol/BlockingMethodFrameListener.java index 492571b6af..21ae3fc71f 100644 --- a/qpid/java/client/src/main/java/org/apache/qpid/client/protocol/BlockingMethodFrameListener.java +++ b/qpid/java/client/src/main/java/org/apache/qpid/client/protocol/BlockingMethodFrameListener.java @@ -110,7 +110,7 @@ public abstract class BlockingMethodFrameListener implements AMQMethodListener } else { - throw new AMQException("Woken up due to exception", _error); // FIXME: This will wrap FailoverException and prevent it being caught. + throw new AMQException("Woken up due to " + _error.getClass(), _error); // FIXME: This will wrap FailoverException and prevent it being caught. } } diff --git a/qpid/java/client/src/main/java/org/apache/qpid/client/state/AMQStateManager.java b/qpid/java/client/src/main/java/org/apache/qpid/client/state/AMQStateManager.java index 887850c06e..50bd1667f9 100644 --- a/qpid/java/client/src/main/java/org/apache/qpid/client/state/AMQStateManager.java +++ b/qpid/java/client/src/main/java/org/apache/qpid/client/state/AMQStateManager.java @@ -103,6 +103,7 @@ public class AMQStateManager implements AMQMethodListener frame2handlerMap.put(ConnectionCloseBody.class, ConnectionCloseMethodHandler.getInstance()); frame2handlerMap.put(BasicDeliverBody.class, BasicDeliverMethodHandler.getInstance()); frame2handlerMap.put(BasicReturnBody.class, BasicReturnMethodHandler.getInstance()); + frame2handlerMap.put(BasicCancelOkBody.class, BasicCancelOkMethodHandler.getInstance()); frame2handlerMap.put(ChannelFlowOkBody.class, ChannelFlowOkMethodHandler.getInstance()); frame2handlerMap.put(QueueDeleteOkBody.class, QueueDeleteOkMethodHandler.getInstance()); frame2handlerMap.put(ExchangeBoundOkBody.class, ExchangeBoundOkMethodHandler.getInstance()); diff --git a/qpid/java/client/src/main/java/org/apache/qpid/client/transport/SocketTransportConnection.java b/qpid/java/client/src/main/java/org/apache/qpid/client/transport/SocketTransportConnection.java index 0de2850080..d6364f45b0 100644 --- a/qpid/java/client/src/main/java/org/apache/qpid/client/transport/SocketTransportConnection.java +++ b/qpid/java/client/src/main/java/org/apache/qpid/client/transport/SocketTransportConnection.java @@ -59,7 +59,7 @@ public class SocketTransportConnection implements ITransportConnection // once more testing of the performance of the simple allocator has been done if (!Boolean.getBoolean("amqj.enablePooledAllocator")) { - _logger.warn("Using SimpleByteBufferAllocator"); + _logger.info("Using SimpleByteBufferAllocator"); ByteBuffer.setAllocator(new SimpleByteBufferAllocator()); } diff --git a/qpid/java/client/src/test/java/org/apache/qpid/test/unit/ack/RecoverTest.java b/qpid/java/client/src/test/java/org/apache/qpid/test/unit/ack/RecoverTest.java index 9f31f7f010..d12ab01bdc 100644 --- a/qpid/java/client/src/test/java/org/apache/qpid/test/unit/ack/RecoverTest.java +++ b/qpid/java/client/src/test/java/org/apache/qpid/test/unit/ack/RecoverTest.java @@ -127,7 +127,7 @@ public class RecoverTest extends TestCase _logger.info("Starting connection"); con.start(); TextMessage tm = (TextMessage) consumer.receive(); - TextMessage tm2 = (TextMessage) consumer.receive(); + consumer.receive(); tm.acknowledge(); _logger.info("Received 2 messages, acknowledge() first message, should acknowledge both"); diff --git a/qpid/java/client/src/test/java/org/apache/qpid/test/unit/basic/MultipleConnectionTest.java b/qpid/java/client/src/test/java/org/apache/qpid/test/unit/basic/MultipleConnectionTest.java index c88024f39f..1e9de221d4 100644 --- a/qpid/java/client/src/test/java/org/apache/qpid/test/unit/basic/MultipleConnectionTest.java +++ b/qpid/java/client/src/test/java/org/apache/qpid/test/unit/basic/MultipleConnectionTest.java @@ -31,7 +31,7 @@ import javax.jms.*; public class MultipleConnectionTest extends TestCase { public static final String _defaultBroker = "vm://:1"; - public static String _connectionString = _defaultBroker; + public String _connectionString = _defaultBroker; private static class Receiver { @@ -176,9 +176,6 @@ public class MultipleConnectionTest extends TestCase { String broker = argv.length > 0 ? argv[0] : _defaultBroker; - int connections = 7; - int sessions = 2; - MultipleConnectionTest test = new MultipleConnectionTest(); test._connectionString = broker; test.test(); diff --git a/qpid/java/client/src/test/java/org/apache/qpid/test/unit/basic/PropertyValueTest.java b/qpid/java/client/src/test/java/org/apache/qpid/test/unit/basic/PropertyValueTest.java index 7f76baa157..17679788bd 100644 --- a/qpid/java/client/src/test/java/org/apache/qpid/test/unit/basic/PropertyValueTest.java +++ b/qpid/java/client/src/test/java/org/apache/qpid/test/unit/basic/PropertyValueTest.java @@ -116,7 +116,9 @@ public class PropertyValueTest extends TestCase implements MessageListener m.setIntProperty("Int", (int) Integer.MAX_VALUE); m.setJMSCorrelationID("Correlation"); - m.setJMSPriority(100); + //fixme the m.setJMSMessage has no effect + producer.setPriority(8); + m.setJMSPriority(3); // Queue Queue q; @@ -182,10 +184,8 @@ public class PropertyValueTest extends TestCase implements MessageListener (int) Integer.MAX_VALUE, m.getIntProperty("Int")); Assert.assertEquals("Check CorrelationID properties are correctly transported", "Correlation", m.getJMSCorrelationID()); - - _logger.warn("getJMSPriority not being verified."); -// Assert.assertEquals("Check Priority properties are correctly transported", -// 100, m.getJMSPriority()); + Assert.assertEquals("Check Priority properties are correctly transported", + 8, m.getJMSPriority()); // Queue Assert.assertEquals("Check ReplyTo properties are correctly transported", diff --git a/qpid/java/client/src/test/java/org/apache/qpid/test/unit/basic/SelectorTest.java b/qpid/java/client/src/test/java/org/apache/qpid/test/unit/basic/SelectorTest.java new file mode 100644 index 0000000000..27a2ccb32e --- /dev/null +++ b/qpid/java/client/src/test/java/org/apache/qpid/test/unit/basic/SelectorTest.java @@ -0,0 +1,141 @@ +/* + * + * 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.test.unit.basic; + +import org.apache.qpid.client.AMQConnection; +import org.apache.qpid.client.AMQDestination; +import org.apache.qpid.client.AMQQueue; +import org.apache.qpid.client.AMQSession; +import org.apache.qpid.client.BasicMessageProducer; +import org.apache.qpid.client.transport.TransportConnection; + +import org.apache.log4j.Logger; + +import javax.jms.JMSException; +import javax.jms.Message; +import javax.jms.MessageListener; +import javax.jms.DeliveryMode; + +import junit.framework.TestCase; + +public class SelectorTest extends TestCase implements MessageListener +{ + + private final static Logger _logger = org.apache.log4j.Logger.getLogger(SelectorTest.class); + + private AMQConnection _connection; + private AMQDestination _destination; + private AMQSession _session; + private int count; + public String _connectionString = "vm://:1"; + + protected void setUp() throws Exception + { + super.setUp(); + TransportConnection.createVMBroker(1); + init(new AMQConnection(_connectionString, "guest", "guest", randomize("Client"), "/test_path")); + } + + protected void tearDown() throws Exception + { + super.tearDown(); + TransportConnection.killAllVMBrokers(); + } + + private void init(AMQConnection connection) throws Exception + { + init(connection, new AMQQueue(randomize("SessionStartTest"), true)); + } + + private void init(AMQConnection connection, AMQDestination destination) throws Exception + { + _connection = connection; + _destination = destination; + connection.start(); + + + String selector = null; +// selector = "Cost = 2 AND JMSDeliveryMode=" + DeliveryMode.NON_PERSISTENT; +// selector = "JMSType = Special AND Cost = 2 AND AMQMessageID > 0 AND JMSDeliveryMode=" + DeliveryMode.NON_PERSISTENT; + + _session = (AMQSession) connection.createSession(false, AMQSession.NO_ACKNOWLEDGE); + //_session.createConsumer(destination).setMessageListener(this); + _session.createConsumer(destination, selector).setMessageListener(this); + } + + public synchronized void test() throws JMSException, InterruptedException + { + try + { + Message msg = _session.createTextMessage("Message"); + msg.setJMSPriority(1); + msg.setIntProperty("Cost", 2); + msg.setJMSType("Special"); + + _logger.info("Sending Message:" + msg); + + ((BasicMessageProducer) _session.createProducer(_destination)).send(msg, DeliveryMode.NON_PERSISTENT); + System.out.println("Message sent, waiting for response..."); + wait(1000); + + if (count > 0) + { + _logger.info("Got message"); + } + + if (count == 0) + { + fail("Did not get message!"); + //throw new RuntimeException("Did not get message!"); + } + } + finally + { + _session.close(); + _connection.close(); + } + } + + public synchronized void onMessage(Message message) + { + count++; + _logger.info("Got Message:" + message); + notify(); + } + + private static String randomize(String in) + { + return in + System.currentTimeMillis(); + } + + public static void main(String[] argv) throws Exception + { + SelectorTest test = new SelectorTest(); + test._connectionString = argv.length == 0 ? "localhost:5672" : argv[0]; + test.setUp(); + test.test(); + } + + public static junit.framework.Test suite() + { + return new junit.framework.TestSuite(SelectorTest.class); + } +} diff --git a/qpid/java/client/src/test/java/org/apache/qpid/test/unit/client/channelclose/ChannelCloseOkTest.java b/qpid/java/client/src/test/java/org/apache/qpid/test/unit/client/channelclose/ChannelCloseOkTest.java index 7d4c2d06d3..ac789eb915 100644 --- a/qpid/java/client/src/test/java/org/apache/qpid/test/unit/client/channelclose/ChannelCloseOkTest.java +++ b/qpid/java/client/src/test/java/org/apache/qpid/test/unit/client/channelclose/ChannelCloseOkTest.java @@ -121,7 +121,7 @@ public class ChannelCloseOkTest extends TestCase { if (_connection != null) { - System.out.println(">>>>>>>>>>>>>>.. closing"); + _log.info(">>>>>>>>>>>>>>.. closing"); _connection.close(); } } @@ -137,7 +137,7 @@ public class ChannelCloseOkTest extends TestCase { public void onException(JMSException jmsException) { - _log.error("onException - ", jmsException); + _log.warn("onException - "+jmsException.getMessage()); } }); diff --git a/qpid/java/client/src/test/java/org/apache/qpid/test/unit/client/channelclose/CloseWithBlockingReceiveTest.java b/qpid/java/client/src/test/java/org/apache/qpid/test/unit/client/channelclose/CloseWithBlockingReceiveTest.java index de517459df..0b3ed931f8 100644 --- a/qpid/java/client/src/test/java/org/apache/qpid/test/unit/client/channelclose/CloseWithBlockingReceiveTest.java +++ b/qpid/java/client/src/test/java/org/apache/qpid/test/unit/client/channelclose/CloseWithBlockingReceiveTest.java @@ -25,7 +25,6 @@ import org.apache.qpid.client.transport.TransportConnection; import javax.jms.Connection; import javax.jms.Session; import javax.jms.MessageConsumer; -import javax.jms.Message; /** * @author Apache Software Foundation @@ -72,7 +71,7 @@ public class CloseWithBlockingReceiveTest extends TestCase }; long startTime = System.currentTimeMillis(); new Thread(r).start(); - Message m = consumer.receive(10000); + consumer.receive(10000); assertTrue(System.currentTimeMillis() - startTime < 10000); } diff --git a/qpid/java/client/src/test/java/org/apache/qpid/test/unit/client/connection/ConnectionTest.java b/qpid/java/client/src/test/java/org/apache/qpid/test/unit/client/connection/ConnectionTest.java index 0da4147351..c5aa62032c 100644 --- a/qpid/java/client/src/test/java/org/apache/qpid/test/unit/client/connection/ConnectionTest.java +++ b/qpid/java/client/src/test/java/org/apache/qpid/test/unit/client/connection/ConnectionTest.java @@ -54,8 +54,7 @@ public class ConnectionTest extends TestCase { try { - Connection connection = new AMQConnection(_broker, "guest", "guest", - "fred", "/test"); + new AMQConnection(_broker, "guest", "guest", "fred", "/test"); } catch (Exception e) { diff --git a/qpid/java/client/src/test/java/org/apache/qpid/test/unit/client/message/MapMessageTest.java b/qpid/java/client/src/test/java/org/apache/qpid/test/unit/client/message/MapMessageTest.java index bd4b3b3987..cddd73898d 100644 --- a/qpid/java/client/src/test/java/org/apache/qpid/test/unit/client/message/MapMessageTest.java +++ b/qpid/java/client/src/test/java/org/apache/qpid/test/unit/client/message/MapMessageTest.java @@ -108,7 +108,7 @@ public class MapMessageTest extends TestCase JMSMapMessage mm = TestMessageHelper.newJMSMapMessage(); mm.setString("value", null); - char c = mm.getChar("value"); + mm.getChar("value"); fail("Expected NullPointerException"); } diff --git a/qpid/java/client/src/test/java/org/apache/qpid/test/unit/message/JMSDestinationTest.java b/qpid/java/client/src/test/java/org/apache/qpid/test/unit/message/JMSDestinationTest.java new file mode 100644 index 0000000000..27736ac473 --- /dev/null +++ b/qpid/java/client/src/test/java/org/apache/qpid/test/unit/message/JMSDestinationTest.java @@ -0,0 +1,70 @@ +package org.apache.qpid.test.unit.message; + +import junit.framework.TestCase; +import org.apache.log4j.Logger; +import org.apache.qpid.client.transport.TransportConnection; +import org.apache.qpid.client.AMQConnection; +import org.apache.qpid.client.AMQSession; +import org.apache.qpid.client.AMQHeadersExchange; +import org.apache.qpid.client.AMQQueue; +import org.apache.qpid.url.AMQBindingURL; +import org.apache.qpid.url.BindingURL; +import org.apache.qpid.exchange.ExchangeDefaults; +import org.apache.qpid.framing.FieldTable; +import org.apache.qpid.framing.PropertyFieldTable; + +import javax.jms.*; + +/** + * @author Apache Software Foundation + */ +public class JMSDestinationTest extends TestCase +{ + + private static final Logger _logger = Logger.getLogger(JMSDestinationTest.class); + + public String _connectionString = "vm://:1"; + + protected void setUp() throws Exception + { + super.setUp(); + TransportConnection.createVMBroker(1); + } + + + protected void tearDown() throws Exception + { + super.tearDown(); + TransportConnection.killAllVMBrokers(); + } + + public void testJMSDestination() throws Exception + { + Connection con = new AMQConnection("vm://:1", "guest", "guest", "consumer1", "/test"); + AMQSession consumerSession = (AMQSession) con.createSession(false, Session.CLIENT_ACKNOWLEDGE); + Queue queue = new AMQQueue("someQ", "someQ", false, true); + MessageConsumer consumer = consumerSession.createConsumer(queue); + + Connection con2 = new AMQConnection("vm://:1", "guest", "guest", "producer1", "/test"); + Session producerSession = con2.createSession(false, Session.CLIENT_ACKNOWLEDGE); + MessageProducer producer = producerSession.createProducer(queue); + + TextMessage sentMsg = producerSession.createTextMessage("hello"); + assertNull(sentMsg.getJMSDestination()); + + producer.send(sentMsg); + + assertEquals(sentMsg.getJMSDestination(), queue); + + con2.close(); + + con.start(); + + TextMessage rm = (TextMessage) consumer.receive(); + assertNotNull(rm); + + assertEquals(rm.getJMSDestination(),queue); + con.close(); + } + +} diff --git a/qpid/java/client/src/test/java/org/apache/qpid/test/unit/message/StreamMessageTest.java b/qpid/java/client/src/test/java/org/apache/qpid/test/unit/message/StreamMessageTest.java index 315ba6ae4c..83969822c4 100644 --- a/qpid/java/client/src/test/java/org/apache/qpid/test/unit/message/StreamMessageTest.java +++ b/qpid/java/client/src/test/java/org/apache/qpid/test/unit/message/StreamMessageTest.java @@ -89,10 +89,10 @@ public class StreamMessageTest extends TestCase StreamMessage msg2 = (StreamMessage) consumer.receive(); - byte b1 = msg2.readByte(); + msg2.readByte(); try { - byte b2 = msg2.readByte(); + msg2.readByte(); } catch (Exception e) { diff --git a/qpid/java/client/src/test/java/org/apache/qpid/test/unit/topic/TopicSessionTest.java b/qpid/java/client/src/test/java/org/apache/qpid/test/unit/topic/TopicSessionTest.java index 14ceaa75f1..794316d2f5 100644 --- a/qpid/java/client/src/test/java/org/apache/qpid/test/unit/topic/TopicSessionTest.java +++ b/qpid/java/client/src/test/java/org/apache/qpid/test/unit/topic/TopicSessionTest.java @@ -260,7 +260,7 @@ public class TopicSessionTest extends TestCase TopicSession session2 = conn.createTopicSession(false, Session.AUTO_ACKNOWLEDGE); try { - MessageConsumer consumer2 = session2.createConsumer(topic); + session2.createConsumer(topic); fail("Expected a JMSException when subscribing to a temporary topic created on adifferent session"); } catch (JMSException je) diff --git a/qpid/java/cluster/src/main/java/org/apache/qpid/server/queue/ClusteredSubscriptionManager.java b/qpid/java/cluster/src/main/java/org/apache/qpid/server/queue/ClusteredSubscriptionManager.java index fa20e9ab76..39ae7e3c3e 100644 --- a/qpid/java/cluster/src/main/java/org/apache/qpid/server/queue/ClusteredSubscriptionManager.java +++ b/qpid/java/cluster/src/main/java/org/apache/qpid/server/queue/ClusteredSubscriptionManager.java @@ -23,6 +23,8 @@ package org.apache.qpid.server.queue; import org.apache.log4j.Logger; import org.apache.qpid.server.cluster.util.LogMessage; +import java.util.List; + class ClusteredSubscriptionManager extends SubscriptionSet { private static final Logger _logger = Logger.getLogger(ClusteredSubscriptionManager.class); @@ -82,6 +84,11 @@ class ClusteredSubscriptionManager extends SubscriptionSet return ClusteredSubscriptionManager.this.getWeight(); } + public List getSubscriptions() + { + return ClusteredSubscriptionManager.super.getSubscriptions(); + } + public boolean hasActiveSubscribers() { return ClusteredSubscriptionManager.super.hasActiveSubscribers(); diff --git a/qpid/java/cluster/src/main/java/org/apache/qpid/server/queue/NestedSubscriptionManager.java b/qpid/java/cluster/src/main/java/org/apache/qpid/server/queue/NestedSubscriptionManager.java index d01ebb5ba2..0566c5203b 100644 --- a/qpid/java/cluster/src/main/java/org/apache/qpid/server/queue/NestedSubscriptionManager.java +++ b/qpid/java/cluster/src/main/java/org/apache/qpid/server/queue/NestedSubscriptionManager.java @@ -21,12 +21,12 @@ package org.apache.qpid.server.queue; import java.util.List; +import java.util.LinkedList; import java.util.concurrent.CopyOnWriteArrayList; /** * Distributes messages among a list of subsscription managers, using their * weighting. - * */ class NestedSubscriptionManager implements SubscriptionManager { @@ -44,11 +44,24 @@ class NestedSubscriptionManager implements SubscriptionManager _subscribers.remove(s); } + + public List getSubscriptions() + { + List allSubs = new LinkedList(); + + for (WeightedSubscriptionManager subMans : _subscribers) + { + allSubs.addAll(subMans.getSubscriptions()); + } + + return allSubs; + } + public boolean hasActiveSubscribers() { - for(WeightedSubscriptionManager s : _subscribers) + for (WeightedSubscriptionManager s : _subscribers) { - if(s.hasActiveSubscribers()) + if (s.hasActiveSubscribers()) { return true; } @@ -59,9 +72,9 @@ class NestedSubscriptionManager implements SubscriptionManager public Subscription nextSubscriber(AMQMessage msg) { WeightedSubscriptionManager start = current(); - for(WeightedSubscriptionManager s = start; s != null; s = next(start)) + for (WeightedSubscriptionManager s = start; s != null; s = next(start)) { - if(hasMore(s)) + if (hasMore(s)) { return nextSubscriber(s); } @@ -94,7 +107,7 @@ class NestedSubscriptionManager implements SubscriptionManager private WeightedSubscriptionManager next() { _iterations = 0; - if(++_index >= _subscribers.size()) + if (++_index >= _subscribers.size()) { _index = 0; } diff --git a/qpid/java/cluster/src/main/java/org/apache/qpid/server/queue/RemoteSubscriptionImpl.java b/qpid/java/cluster/src/main/java/org/apache/qpid/server/queue/RemoteSubscriptionImpl.java index 0268ff2171..c751e4a011 100644 --- a/qpid/java/cluster/src/main/java/org/apache/qpid/server/queue/RemoteSubscriptionImpl.java +++ b/qpid/java/cluster/src/main/java/org/apache/qpid/server/queue/RemoteSubscriptionImpl.java @@ -25,6 +25,9 @@ import org.apache.qpid.server.cluster.GroupManager; import org.apache.qpid.server.cluster.SimpleSendable; import org.apache.qpid.AMQException; +import java.util.Queue; +import java.util.List; + class RemoteSubscriptionImpl implements Subscription, WeightedSubscriptionManager { private final GroupManager _groupMgr; @@ -76,6 +79,11 @@ class RemoteSubscriptionImpl implements Subscription, WeightedSubscriptionManage return _count; } + public List getSubscriptions() + { + return null; //To change body of implemented methods use File | Settings | File Templates. + } + public boolean hasActiveSubscribers() { return getWeight() == 0; @@ -88,9 +96,49 @@ class RemoteSubscriptionImpl implements Subscription, WeightedSubscriptionManage public void queueDeleted(AMQQueue queue) { - if(queue instanceof ClusteredQueue) + if (queue instanceof ClusteredQueue) { ((ClusteredQueue) queue).removeAllRemoteSubscriber(_peer); } } + + public boolean hasFilters() + { + return false; + } + + public boolean hasInterest(AMQMessage msg) + { + return true; + } + + public Queue getPreDeliveryQueue() + { + return null; + } + + public void enqueueForPreDelivery(AMQMessage msg) + { + //no-op -- if selectors are implemented on RemoteSubscriptions then look at SubscriptionImpl + } + + public boolean isAutoClose() + { + return false; + } + + public void close() + { + //no-op + } + + public boolean isBrowser() + { + return false; //To change body of implemented methods use File | Settings | File Templates. + } + + public void sendNextMessage(AMQQueue queue) + { + + } } diff --git a/qpid/java/common/src/main/java/log4j.properties b/qpid/java/common/src/main/java/log4j.properties new file mode 100644 index 0000000000..6d596d1d19 --- /dev/null +++ b/qpid/java/common/src/main/java/log4j.properties @@ -0,0 +1,28 @@ +# +# 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. +# +log4j.rootLogger=${root.logging.level} + + +log4j.logger.org.apache.qpid=${amqj.logging.level}, console +log4j.additivity.org.apache.qpid=false + +log4j.appender.console=org.apache.log4j.ConsoleAppender +log4j.appender.console.Threshold=all +log4j.appender.console.layout=org.apache.log4j.PatternLayout +log4j.appender.console.layout.ConversionPattern=%t %d %p [%c{4}] %m%n diff --git a/qpid/java/common/src/main/java/org/apache/qpid/AMQInvalidSelectorException.java b/qpid/java/common/src/main/java/org/apache/qpid/AMQInvalidSelectorException.java new file mode 100644 index 0000000000..dcd039b789 --- /dev/null +++ b/qpid/java/common/src/main/java/org/apache/qpid/AMQInvalidSelectorException.java @@ -0,0 +1,31 @@ +/* + * 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; + +import org.apache.qpid.protocol.AMQConstant; + +public class AMQInvalidSelectorException extends AMQException +{ + public AMQInvalidSelectorException(String message) + { + super(AMQConstant.INVALID_SELECTOR.getCode(),message); + } +} diff --git a/qpid/java/common/src/main/java/org/apache/qpid/common/AMQPFilterTypes.java b/qpid/java/common/src/main/java/org/apache/qpid/common/AMQPFilterTypes.java new file mode 100644 index 0000000000..56219755a3 --- /dev/null +++ b/qpid/java/common/src/main/java/org/apache/qpid/common/AMQPFilterTypes.java @@ -0,0 +1,40 @@ +/* + * 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.common; + +public enum AMQPFilterTypes +{ + JMS_SELECTOR("x-filter-jms-selector"), + NO_CONSUME("x-filter-no-consume"), + AUTO_CLOSE("x-filter-auto-close"); + + private final String _value; + + AMQPFilterTypes(String value) + { + _value = value; + } + + public String getValue() + { + return _value; + } +} diff --git a/qpid/java/common/src/main/java/org/apache/qpid/common/ClientProperties.java b/qpid/java/common/src/main/java/org/apache/qpid/common/ClientProperties.java new file mode 100644 index 0000000000..07371b5182 --- /dev/null +++ b/qpid/java/common/src/main/java/org/apache/qpid/common/ClientProperties.java @@ -0,0 +1,29 @@ +/* + * 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.common; + +public enum ClientProperties +{ + instance, + product, + version, + platform +} diff --git a/qpid/java/common/src/main/java/org/apache/qpid/common/QpidProperties.java b/qpid/java/common/src/main/java/org/apache/qpid/common/QpidProperties.java new file mode 100644 index 0000000000..3a96821e93 --- /dev/null +++ b/qpid/java/common/src/main/java/org/apache/qpid/common/QpidProperties.java @@ -0,0 +1,46 @@ +/* + * 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.common; + +public class QpidProperties +{ + + static + { + //load values from property file. + } + + public static String getProductName() + { + return "Qpid"; + } + + public static String getReleaseVerision() + { + return "1.0"; + } + + + public static String getBuildVerision() + { + return "1"; + } +} diff --git a/qpid/java/common/src/main/java/org/apache/qpid/framing/PropertyFieldTable.java b/qpid/java/common/src/main/java/org/apache/qpid/framing/PropertyFieldTable.java index 1292ff2f6e..4b8f56e4e8 100644 --- a/qpid/java/common/src/main/java/org/apache/qpid/framing/PropertyFieldTable.java +++ b/qpid/java/common/src/main/java/org/apache/qpid/framing/PropertyFieldTable.java @@ -129,7 +129,7 @@ public class PropertyFieldTable implements FieldTable } catch (Exception e) { - _logger.warn("Unable to decode PropertyFieldTable format:" + textFormat, e); + _logger.warn("Unable to decode PropertyFieldTable format:" + textFormat); throw new IllegalArgumentException("Unable to decode PropertyFieldTable format:" + textFormat); } } @@ -483,7 +483,7 @@ public class PropertyFieldTable implements FieldTable { return _properties.containsKey(name) && (_properties.get(name) == null) && _propertyNamesTypeMap.get(name).equals(Prefix.AMQP_NULL_STRING_PROPERTY_PREFIX); - + } @@ -606,7 +606,8 @@ public class PropertyFieldTable implements FieldTable // AMQ start character if (!(Character.isLetter(propertyName.charAt(0)) || propertyName.charAt(0) == '$' - || propertyName.charAt(0) == '#')) + || propertyName.charAt(0) == '#' + || propertyName.charAt(0) == '_')) // Not official AMQP added for JMS. { throw new IllegalArgumentException("Identifier '" + propertyName + "' does not start with a valid AMQP start character"); } @@ -1156,9 +1157,9 @@ public class PropertyFieldTable implements FieldTable if (type == null) { String msg = "Field '" + key + "' - unsupported field table type: " + type + "."; - //some extra trace information... - msg += " (" + iType + "), length=" + length + ", sizeRead=" + sizeRead + ", sizeRemaining=" + sizeRemaining; - throw new AMQFrameDecodingException(msg); + //some extra trace information... + msg += " (" + iType + "), length=" + length + ", sizeRead=" + sizeRead + ", sizeRemaining=" + sizeRemaining; + throw new AMQFrameDecodingException(msg); } Object value; @@ -1203,7 +1204,7 @@ public class PropertyFieldTable implements FieldTable value = EncodingUtils.readBytes(buffer); break; default: - String msg = "Internal error, the following type identifier is not handled: " + type; + String msg = "Internal error, the following type identifier is not handled: " + type; throw new AMQFrameDecodingException(msg); } diff --git a/qpid/java/common/src/main/java/org/apache/qpid/protocol/AMQConstant.java b/qpid/java/common/src/main/java/org/apache/qpid/protocol/AMQConstant.java index fc83c0726d..a0d243ca30 100644 --- a/qpid/java/common/src/main/java/org/apache/qpid/protocol/AMQConstant.java +++ b/qpid/java/common/src/main/java/org/apache/qpid/protocol/AMQConstant.java @@ -55,7 +55,7 @@ public final class AMQConstant { return _name; } - + public static final AMQConstant FRAME_MIN_SIZE = new AMQConstant(4096, "frame min size", true); public static final AMQConstant FRAME_END = new AMQConstant(206, "frame end", true); @@ -74,6 +74,8 @@ public final class AMQConstant public static final AMQConstant CONTEXT_UNKNOWN = new AMQConstant(321, "context unknown", true); + public static final AMQConstant INVALID_SELECTOR = new AMQConstant(322, "selector invalid", true); + public static final AMQConstant INVALID_PATH = new AMQConstant(402, "invalid path", true); public static final AMQConstant ACCESS_REFUSED = new AMQConstant(403, "access refused", true); diff --git a/qpid/java/systests/src/test/java/org/apache/qpid/server/exchange/ExchangeMBeanTest.java b/qpid/java/systests/src/test/java/org/apache/qpid/server/exchange/ExchangeMBeanTest.java new file mode 100644 index 0000000000..728d1eb9aa --- /dev/null +++ b/qpid/java/systests/src/test/java/org/apache/qpid/server/exchange/ExchangeMBeanTest.java @@ -0,0 +1,123 @@ +/* + * + * Copyright (c) 2006 The Apache Software Foundation + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ +package org.apache.qpid.server.exchange; + +import junit.framework.TestCase; +import org.apache.qpid.server.management.ManagedObject; +import org.apache.qpid.server.queue.AMQQueue; +import org.apache.qpid.server.queue.QueueRegistry; +import org.apache.qpid.server.registry.ApplicationRegistry; + +import javax.management.openmbean.CompositeData; +import javax.management.openmbean.TabularData; +import java.util.ArrayList; + +/** + * Unit test class for testing different Exchange MBean operations + */ +public class ExchangeMBeanTest extends TestCase +{ + private AMQQueue _queue; + private QueueRegistry _queueRegistry = ApplicationRegistry.getInstance().getQueueRegistry(); + + /** + * Test for direct exchange mbean + * @throws Exception + */ + public void testDirectExchangeMBean() throws Exception + { + DestNameExchange exchange = new DestNameExchange(); + exchange.initialise("amq.direct", false, 0, true); + ManagedObject managedObj = exchange.getManagedObject(); + ManagedExchange mbean = (ManagedExchange)managedObj; + + mbean.createNewBinding(_queue.getName(), "binding1"); + mbean.createNewBinding(_queue.getName(), "binding2"); + + TabularData data = mbean.bindings(); + ArrayList list = new ArrayList(data.values()); + assertTrue(list.size() == 2); + + // test general exchange properties + assertEquals(mbean.getName(), "amq.direct"); + assertEquals(mbean.getExchangeType(), "direct"); + assertTrue(mbean.getTicketNo() == 0); + assertTrue(!mbean.isDurable()); + assertTrue(mbean.isAutoDelete()); + } + + /** + * Test for "topic" exchange mbean + * @throws Exception + */ + public void testTopicExchangeMBean() throws Exception + { + DestWildExchange exchange = new DestWildExchange(); + exchange.initialise("amq.topic", false, 0, true); + ManagedObject managedObj = exchange.getManagedObject(); + ManagedExchange mbean = (ManagedExchange)managedObj; + + mbean.createNewBinding(_queue.getName(), "binding1"); + mbean.createNewBinding(_queue.getName(), "binding2"); + + TabularData data = mbean.bindings(); + ArrayList list = new ArrayList(data.values()); + assertTrue(list.size() == 2); + + // test general exchange properties + assertEquals(mbean.getName(), "amq.topic"); + assertEquals(mbean.getExchangeType(), "topic"); + assertTrue(mbean.getTicketNo() == 0); + assertTrue(!mbean.isDurable()); + assertTrue(mbean.isAutoDelete()); + } + + /** + * Test for "Headers" exchange mbean + * @throws Exception + */ + public void testHeadersExchangeMBean() throws Exception + { + HeadersExchange exchange = new HeadersExchange(); + exchange.initialise("amq.headers", false, 0, true); + ManagedObject managedObj = exchange.getManagedObject(); + ManagedExchange mbean = (ManagedExchange)managedObj; + + mbean.createNewBinding(_queue.getName(), "key1=binding1,key2=binding2"); + mbean.createNewBinding(_queue.getName(), "key3=binding3"); + + TabularData data = mbean.bindings(); + ArrayList list = new ArrayList(data.values()); + assertTrue(list.size() == 2); + + // test general exchange properties + assertEquals(mbean.getName(), "amq.headers"); + assertEquals(mbean.getExchangeType(), "headers"); + assertTrue(mbean.getTicketNo() == 0); + assertTrue(!mbean.isDurable()); + assertTrue(mbean.isAutoDelete()); + } + + @Override + protected void setUp() throws Exception + { + super.setUp(); + _queue = new AMQQueue("testQueue", false, "ExchangeMBeanTest", false, _queueRegistry); + _queueRegistry.registerQueue(_queue); + } +} diff --git a/qpid/java/systests/src/test/java/org/apache/qpid/server/exchange/HeadersExchangeTest.java b/qpid/java/systests/src/test/java/org/apache/qpid/server/exchange/HeadersExchangeTest.java index c220442a78..c01241d11d 100644 --- a/qpid/java/systests/src/test/java/org/apache/qpid/server/exchange/HeadersExchangeTest.java +++ b/qpid/java/systests/src/test/java/org/apache/qpid/server/exchange/HeadersExchangeTest.java @@ -43,8 +43,6 @@ public class HeadersExchangeTest extends AbstractHeadersExchangeTestBase TestQueue q6 = bindDefault("F0000=Aardvark", "F0001=Bear"); TestQueue q7 = bindDefault("F0000", "F0001=Bear"); TestQueue q8 = bindDefault("F0000=Aardvark", "F0001"); - TestQueue q9 = bindDefault("F0000=Apple", "F0001=Banana"); - TestQueue q10 = bindDefault("F0000=Apple", "F0001"); routeAndTest(new Message("Message1", "F0000"), q1); routeAndTest(new Message("Message2", "F0000=Aardvark"), q1, q2); @@ -74,7 +72,6 @@ public class HeadersExchangeTest extends AbstractHeadersExchangeTestBase TestQueue q2 = bindDefault("F0000=Aardvark", "F0001=Bear", "X-match=any"); TestQueue q3 = bindDefault("F0000", "F0001=Bear", "X-match=any"); TestQueue q4 = bindDefault("F0000=Aardvark", "F0001", "X-match=any"); - TestQueue q5 = bindDefault("F0000=Apple", "F0001=Banana", "X-match=any"); TestQueue q6 = bindDefault("F0000=Apple", "F0001", "X-match=any"); routeAndTest(new Message("Message1", "F0000"), q1, q3); @@ -87,16 +84,14 @@ public class HeadersExchangeTest extends AbstractHeadersExchangeTestBase public void testMandatory() throws AMQException { - TestQueue q1 = bindDefault("F0000"); + bindDefault("F0000"); Message m1 = new Message("Message1", "XXXXX"); Message m2 = new Message("Message2", "F0000"); BasicPublishBody pb1 = m1.getPublishBody(); pb1.mandatory = true; - BasicPublishBody pb2 = m1.getPublishBody(); + BasicPublishBody pb2 = m2.getPublishBody(); pb2.mandatory = true; routeAndTest(m1,true); - - } public static junit.framework.Test suite() diff --git a/qpid/java/systests/src/test/java/org/apache/qpid/server/exchange/ReturnUnroutableMandatoryMessageTest.java b/qpid/java/systests/src/test/java/org/apache/qpid/server/exchange/ReturnUnroutableMandatoryMessageTest.java index 4dffe3e75f..b125bc1d4c 100644 --- a/qpid/java/systests/src/test/java/org/apache/qpid/server/exchange/ReturnUnroutableMandatoryMessageTest.java +++ b/qpid/java/systests/src/test/java/org/apache/qpid/server/exchange/ReturnUnroutableMandatoryMessageTest.java @@ -5,8 +5,8 @@ import org.apache.log4j.Logger; import org.apache.qpid.test.VMBrokerSetup; import org.apache.qpid.server.registry.ApplicationRegistry; import org.apache.qpid.server.util.TestApplicationRegistry; -import org.apache.qpid.server.store.TestableMemoryMessageStore; import org.apache.qpid.client.*; +import org.apache.qpid.client.transport.TransportConnection; import org.apache.qpid.url.AMQBindingURL; import org.apache.qpid.url.BindingURL; import org.apache.qpid.exchange.ExchangeDefaults; @@ -39,12 +39,14 @@ public class ReturnUnroutableMandatoryMessageTest extends TestCase implements Ex protected void setUp() throws Exception { super.setUp(); + TransportConnection.createVMBroker(1); ApplicationRegistry.initialise(new TestApplicationRegistry(), 1); } protected void tearDown() throws Exception { super.tearDown(); + TransportConnection.killAllVMBrokers(); } /** @@ -57,17 +59,14 @@ public class ReturnUnroutableMandatoryMessageTest extends TestCase implements Ex _bouncedMessageList.clear(); Connection con = new AMQConnection("vm://:1", "guest", "guest", "consumer1", "/test"); - TestableMemoryMessageStore store = (TestableMemoryMessageStore) ApplicationRegistry.getInstance().getMessageStore(); AMQSession consumerSession = (AMQSession) con.createSession(false, Session.CLIENT_ACKNOWLEDGE); - - AMQHeadersExchange queue = new AMQHeadersExchange(new AMQBindingURL(ExchangeDefaults.HEADERS_EXCHANGE_CLASS+"://"+ExchangeDefaults.HEADERS_EXCHANGE_NAME+"/test/queue1?"+ BindingURL.OPTION_ROUTING_KEY+"='F0000=1'")); + AMQHeadersExchange queue = new AMQHeadersExchange(new AMQBindingURL(ExchangeDefaults.HEADERS_EXCHANGE_CLASS + "://" + ExchangeDefaults.HEADERS_EXCHANGE_NAME + "/test/queue1?" + BindingURL.OPTION_ROUTING_KEY + "='F0000=1'")); FieldTable ft = new PropertyFieldTable(); - ft.setString("F1000","1"); - MessageConsumer consumer = consumerSession.createConsumer(queue, AMQSession.DEFAULT_PREFETCH_LOW_MARK, AMQSession.DEFAULT_PREFETCH_HIGH_MARK, false, false, (String)null, ft); + ft.setString("F1000", "1"); + MessageConsumer consumer = consumerSession.createConsumer(queue, AMQSession.DEFAULT_PREFETCH_LOW_MARK, AMQSession.DEFAULT_PREFETCH_HIGH_MARK, false, false, (String) null, ft); - //force synch to ensure the consumer has resulted in a bound queue ((AMQSession) consumerSession).declareExchangeSynch(ExchangeDefaults.HEADERS_EXCHANGE_NAME, ExchangeDefaults.HEADERS_EXCHANGE_CLASS); @@ -81,49 +80,45 @@ public class ReturnUnroutableMandatoryMessageTest extends TestCase implements Ex con2.start(); - MessageProducer nonMandatoryProducer = producerSession.createProducer(queue,false,false); + MessageProducer nonMandatoryProducer = producerSession.createProducer(queue, false, false); MessageProducer mandatoryProducer = producerSession.createProducer(queue); - // First test - should neither be bounced nor routed _logger.info("Sending non-routable non-mandatory message"); - TextMessage msg1 = producerSession.createTextMessage("msg1"); + TextMessage msg1 = producerSession.createTextMessage("msg1"); nonMandatoryProducer.send(msg1); // Second test - should be bounced _logger.info("Sending non-routable mandatory message"); - TextMessage msg2 = producerSession.createTextMessage("msg2"); + TextMessage msg2 = producerSession.createTextMessage("msg2"); mandatoryProducer.send(msg2); // Third test - should be routed _logger.info("Sending routable message"); - TextMessage msg3 = producerSession.createTextMessage("msg3"); - msg3.setStringProperty("F1000","1"); + TextMessage msg3 = producerSession.createTextMessage("msg3"); + msg3.setStringProperty("F1000", "1"); mandatoryProducer.send(msg3); - _logger.info("Starting consumer connection"); con.start(); TextMessage tm = (TextMessage) consumer.receive(1000L); - assertTrue("No message routed to receiver",tm != null); - assertTrue("Wrong message routed to receiver: "+tm.getText(),"msg3".equals(tm.getText())); + assertTrue("No message routed to receiver", tm != null); + assertTrue("Wrong message routed to receiver: " + tm.getText(), "msg3".equals(tm.getText())); try { Thread.sleep(1000L); } - catch(InterruptedException e) + catch (InterruptedException e) { ; } - assertTrue("Wrong number of messages bounced (expect 1): "+_bouncedMessageList.size(),_bouncedMessageList.size()==1); + assertTrue("Wrong number of messages bounced (expect 1): " + _bouncedMessageList.size(), _bouncedMessageList.size() == 1); Message m = _bouncedMessageList.get(0); - assertTrue("Wrong message bounced: "+m.toString(),m.toString().contains("msg2")); - - + assertTrue("Wrong message bounced: " + m.toString(), m.toString().contains("msg2")); con.close(); @@ -134,18 +129,23 @@ public class ReturnUnroutableMandatoryMessageTest extends TestCase implements Ex public static junit.framework.Test suite() { - return new VMBrokerSetup(new junit.framework.TestSuite(ReturnUnroutableMandatoryMessageTest.class)); + return new junit.framework.TestSuite(ReturnUnroutableMandatoryMessageTest.class); } public void onException(JMSException jmsException) { - _logger.warn("Caught exception on producer: ",jmsException); + Exception linkedException = jmsException.getLinkedException(); - if(linkedException instanceof AMQNoRouteException) + if (linkedException instanceof AMQNoRouteException) { AMQNoRouteException noRoute = (AMQNoRouteException) linkedException; Message bounced = (Message) noRoute.getUndeliveredMessage(); _bouncedMessageList.add(bounced); + _logger.info("Caught expected NoRouteException"); + } + else + { + _logger.warn("Caught exception on producer: ", jmsException); } } } diff --git a/qpid/java/systests/src/test/java/org/apache/qpid/server/protocol/AMQProtocolSessionMBeanTest.java b/qpid/java/systests/src/test/java/org/apache/qpid/server/protocol/AMQProtocolSessionMBeanTest.java new file mode 100644 index 0000000000..0a41d4166b --- /dev/null +++ b/qpid/java/systests/src/test/java/org/apache/qpid/server/protocol/AMQProtocolSessionMBeanTest.java @@ -0,0 +1,118 @@ +/* + * + * Copyright (c) 2006 The Apache Software Foundation + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ +package org.apache.qpid.server.protocol; + +import junit.framework.TestCase; +import org.apache.mina.common.IoSession; +import org.apache.qpid.codec.AMQCodecFactory; +import org.apache.qpid.server.AMQChannel; +import org.apache.qpid.server.exchange.DefaultExchangeFactory; +import org.apache.qpid.server.exchange.DefaultExchangeRegistry; +import org.apache.qpid.server.exchange.ExchangeRegistry; +import org.apache.qpid.server.queue.DefaultQueueRegistry; +import org.apache.qpid.server.queue.QueueRegistry; +import org.apache.qpid.server.store.MessageStore; +import org.apache.qpid.server.store.SkeletonMessageStore; + +import javax.management.openmbean.CompositeData; +import javax.management.openmbean.TabularData; +import javax.management.JMException; +import java.util.ArrayList; + +/** + * Test class to test MBean operations for AMQMinaProtocolSession. + */ +public class AMQProtocolSessionMBeanTest extends TestCase +{ + private IoSession _mockIOSession; + private MessageStore _messageStore = new SkeletonMessageStore(); + private AMQMinaProtocolSession _protocolSession; + private AMQChannel _channel; + private QueueRegistry _queueRegistry; + private ExchangeRegistry _exchangeRegistry; + private AMQProtocolSessionMBean _mbean; + + public void testChannels() throws Exception + { + // check the channel count is correct + int channelCount = _mbean.channels().size(); + assertTrue(channelCount == 1); + _protocolSession.addChannel(new AMQChannel(2, _messageStore, null)); + channelCount = _mbean.channels().size(); + assertTrue(channelCount == 2); + + // check the channel closing + _mbean.closeChannel(1); + TabularData channels = _mbean.channels(); + ArrayList list = new ArrayList(channels.values()); + channelCount = list.size(); + assertTrue(channelCount == 1); + CompositeData channelData = list.get(0); + assertEquals(channelData.get("Channel Id"), new Integer(2)); + + // general properties test + _mbean.setMaximumNumberOfChannels(1000L); + assertTrue(_mbean.getMaximumNumberOfChannels() == 1000L); + + // check if the rollback and commit APIs + AMQChannel channel3 = new AMQChannel(3, _messageStore, null); + channel3.setTransactional(true); + _protocolSession.addChannel(channel3); + _mbean.rollbackTransactions(2); + _mbean.rollbackTransactions(3); + _mbean.commitTransactions(2); + _mbean.commitTransactions(3); + + // This should throw exception, because the channel does't exist + try + { + _mbean.commitTransactions(4); + fail(); + } + catch (JMException ex) + { + System.out.println("expected exception is thrown :" + ex.getMessage()); + } + + // check if closing of session works + _protocolSession.addChannel(new AMQChannel(5, _messageStore, null)); + _mbean.closeConnection(); + try + { + _mbean.closeChannel(5); + fail(); + } + catch(JMException ex) + { + System.out.println("expected exception is thrown :" + ex.getMessage()); + } + } + + @Override + protected void setUp() throws Exception + { + super.setUp(); + _channel = new AMQChannel(1, _messageStore, null); + _queueRegistry = new DefaultQueueRegistry(); + _exchangeRegistry = new DefaultExchangeRegistry(new DefaultExchangeFactory()); + _mockIOSession = new MockIoSession(); + _protocolSession = new AMQMinaProtocolSession(_mockIOSession, _queueRegistry, _exchangeRegistry, new AMQCodecFactory(true)); + _protocolSession.addChannel(_channel); + _mbean = (AMQProtocolSessionMBean)_protocolSession.getManagedObject(); + } +} diff --git a/qpid/java/systests/src/test/java/org/apache/qpid/server/protocol/MockIoSession.java b/qpid/java/systests/src/test/java/org/apache/qpid/server/protocol/MockIoSession.java index 81dea32a76..cf6366b513 100644 --- a/qpid/java/systests/src/test/java/org/apache/qpid/server/protocol/MockIoSession.java +++ b/qpid/java/systests/src/test/java/org/apache/qpid/server/protocol/MockIoSession.java @@ -25,6 +25,7 @@ import org.apache.mina.common.support.DefaultCloseFuture; import org.apache.mina.common.support.DefaultWriteFuture; import java.net.SocketAddress; +import java.net.InetSocketAddress; import java.util.Set; public class MockIoSession implements IoSession @@ -151,7 +152,7 @@ public class MockIoSession implements IoSession public SocketAddress getRemoteAddress() { - return null; //To change body of implemented methods use File | Settings | File Templates. + return new InetSocketAddress("127.0.0.1", 1234); //To change body of implemented methods use File | Settings | File Templates. } public SocketAddress getLocalAddress() diff --git a/qpid/java/systests/src/test/java/org/apache/qpid/server/queue/AMQQueueMBeanTest.java b/qpid/java/systests/src/test/java/org/apache/qpid/server/queue/AMQQueueMBeanTest.java new file mode 100644 index 0000000000..8de3c8bf33 --- /dev/null +++ b/qpid/java/systests/src/test/java/org/apache/qpid/server/queue/AMQQueueMBeanTest.java @@ -0,0 +1,180 @@ +/* + * + * Copyright (c) 2006 The Apache Software Foundation + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ +package org.apache.qpid.server.queue; + +import junit.framework.TestCase; +import org.apache.qpid.AMQException; +import org.apache.qpid.framing.BasicPublishBody; +import org.apache.qpid.framing.ContentHeaderBody; +import org.apache.qpid.server.AMQChannel; +import org.apache.qpid.server.store.MessageStore; +import org.apache.qpid.server.store.SkeletonMessageStore; + +import javax.management.JMException; + +/** + * Test class to test AMQQueueMBean attribtues and operations + */ +public class AMQQueueMBeanTest extends TestCase +{ + private AMQQueue _queue; + private AMQQueueMBean _queueMBean; + private QueueRegistry _queueRegistry; + private MessageStore _messageStore = new SkeletonMessageStore(); + private MockProtocolSession _protocolSession; + private AMQChannel _channel; + + public void testMessageCount() throws Exception + { + int messageCount = 10; + sendMessages(messageCount); + assertTrue(_queueMBean.getMessageCount() == messageCount); + assertTrue(_queueMBean.getReceivedMessageCount() == messageCount); + assertTrue(_queueMBean.getQueueDepth() == 10); + + _queueMBean.deleteMessageFromTop(); + assertTrue(_queueMBean.getMessageCount() == messageCount - 1); + assertTrue(_queueMBean.getReceivedMessageCount() == messageCount); + + _queueMBean.clearQueue(); + assertTrue(_queueMBean.getMessageCount() == 0); + assertTrue(_queueMBean.getReceivedMessageCount() == messageCount); + } + + public void testConsumerCount() throws Exception + { + SubscriptionManager mgr = _queue.getSubscribers(); + assertFalse(mgr.hasActiveSubscribers()); + assertTrue(_queueMBean.getActiveConsumerCount() == 0); + + _channel = new AMQChannel(1, _messageStore, null); + _protocolSession = new MockProtocolSession(_messageStore); + _protocolSession.addChannel(_channel); + + _queue.registerProtocolSession(_protocolSession, 1, "test", false, null); + assertTrue(_queueMBean.getActiveConsumerCount() == 1); + + SubscriptionSet _subscribers = (SubscriptionSet) mgr; + SubscriptionTestHelper s1 = new SubscriptionTestHelper("S1"); + SubscriptionTestHelper s2 = new SubscriptionTestHelper("S2"); + _subscribers.addSubscriber(s1); + _subscribers.addSubscriber(s2); + assertTrue(_queueMBean.getActiveConsumerCount() == 3); + assertTrue(_queueMBean.getConsumerCount() == 3); + + s1.setSuspended(true); + assertTrue(_queueMBean.getActiveConsumerCount() == 2); + assertTrue(_queueMBean.getConsumerCount() == 3); + } + + public void testGeneralProperties() + { + _queueMBean.setMaximumMessageCount(50000); + _queueMBean.setMaximumMessageSize(2000l); + _queueMBean.setMaximumQueueDepth(1000l); + + assertTrue(_queueMBean.getMaximumMessageCount() == 50000); + assertTrue(_queueMBean.getMaximumMessageSize() == 2000); + assertTrue(_queueMBean.getMaximumQueueDepth() == 1000); + + assertTrue(_queueMBean.getName().equals("testQueue")); + assertTrue(_queueMBean.getOwner().equals("AMQueueMBeanTest")); + assertFalse(_queueMBean.isAutoDelete()); + assertFalse(_queueMBean.isDurable()); + } + + public void testExceptions() throws Exception + { + try + { + _queueMBean.viewMessages(0, 3); + fail(); + } + catch (JMException ex) + { + + } + + try + { + _queueMBean.viewMessages(2, 1); + fail(); + } + catch (JMException ex) + { + + } + + try + { + _queueMBean.viewMessages(-1, 1); + fail(); + } + catch (JMException ex) + { + + } + + AMQMessage msg = message(false); + long id = msg.getMessageId(); + _queue.clearQueue(); + _queue.deliver(msg); + _queueMBean.viewMessageContent(id); + try + { + _queueMBean.viewMessageContent(id + 1); + fail(); + } + catch (JMException ex) + { + + } + } + + private AMQMessage message(boolean immediate) throws AMQException + { + BasicPublishBody publish = new BasicPublishBody(); + publish.immediate = immediate; + ContentHeaderBody contentHeaderBody = new ContentHeaderBody(); + contentHeaderBody.bodySize = 1000; // in bytes + return new AMQMessage(_messageStore, publish, contentHeaderBody, null); + } + + @Override + protected void setUp() throws Exception + { + super.setUp(); + _queueRegistry = new DefaultQueueRegistry(); + _queue = new AMQQueue("testQueue", false, "AMQueueMBeanTest", false, _queueRegistry); + _queueMBean = new AMQQueueMBean(_queue); + } + + private void sendMessages(int messageCount) throws AMQException + { + AMQMessage[] messages = new AMQMessage[messageCount]; + for (int i = 0; i < messages.length; i++) + { + messages[i] = message(false); + ; + } + for (int i = 0; i < messageCount; i++) + { + _queue.deliver(messages[i]); + } + } +} diff --git a/qpid/java/systests/src/test/java/org/apache/qpid/server/queue/AMQueueMBeanTest.java b/qpid/java/systests/src/test/java/org/apache/qpid/server/queue/AMQueueMBeanTest.java deleted file mode 100644 index a9dacac43c..0000000000 --- a/qpid/java/systests/src/test/java/org/apache/qpid/server/queue/AMQueueMBeanTest.java +++ /dev/null @@ -1,179 +0,0 @@ -/* - * - * Copyright (c) 2006 The Apache Software Foundation - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - * - */ -package org.apache.qpid.server.queue; - -import junit.framework.TestCase; -import org.apache.qpid.AMQException; -import org.apache.qpid.framing.BasicPublishBody; -import org.apache.qpid.framing.ContentHeaderBody; -import org.apache.qpid.server.AMQChannel; -import org.apache.qpid.server.store.MessageStore; -import org.apache.qpid.server.store.SkeletonMessageStore; - -import javax.management.JMException; - -/** - * Test class to test AMQQueueMBean attribtues and operations - */ -public class AMQueueMBeanTest extends TestCase -{ - private AMQQueue _queue; - private AMQQueueMBean _queueMBean; - private QueueRegistry _queueRegistry; - private MessageStore _messageStore = new SkeletonMessageStore(); - private MockProtocolSession _protocolSession; - private AMQChannel _channel; - - public void testMessageCount() throws Exception - { - int messageCount = 10; - sendMessages(messageCount); - assertTrue(_queueMBean.getMessageCount() == messageCount); - assertTrue(_queueMBean.getReceivedMessageCount() == messageCount); - assertTrue(_queueMBean.getQueueDepth() == 10); - - _queueMBean.deleteMessageFromTop(); - assertTrue(_queueMBean.getMessageCount() == messageCount - 1); - assertTrue(_queueMBean.getReceivedMessageCount() == messageCount); - - _queueMBean.clearQueue(); - assertTrue(_queueMBean.getMessageCount() == 0); - assertTrue(_queueMBean.getReceivedMessageCount() == messageCount); - } - - public void testConsumerCount() throws Exception - { - SubscriptionManager mgr = _queue.getSubscribers(); - assertFalse(mgr.hasActiveSubscribers()); - assertTrue(_queueMBean.getActiveConsumerCount() == 0); - - _channel = new AMQChannel(1, _messageStore, null); - _protocolSession = new MockProtocolSession(_messageStore); - _protocolSession.addChannel(_channel); - - _queue.registerProtocolSession(_protocolSession, 1, "test", false); - assertTrue(_queueMBean.getActiveConsumerCount() == 1); - - SubscriptionSet _subscribers = (SubscriptionSet)mgr; - SubscriptionTestHelper s1 = new SubscriptionTestHelper("S1"); - SubscriptionTestHelper s2 = new SubscriptionTestHelper("S2"); - _subscribers.addSubscriber(s1); - _subscribers.addSubscriber(s2); - assertTrue(_queueMBean.getActiveConsumerCount() == 3); - assertTrue(_queueMBean.getConsumerCount() == 3); - - s1.setSuspended(true); - assertTrue(_queueMBean.getActiveConsumerCount() == 2); - assertTrue(_queueMBean.getConsumerCount() == 3); - } - - public void testGeneralProperties() - { - _queueMBean.setMaximumMessageCount(50000); - _queueMBean.setMaximumMessageSize(2000l); - _queueMBean.setMaximumQueueDepth(1000l); - - assertTrue(_queueMBean.getMaximumMessageCount() == 50000); - assertTrue(_queueMBean.getMaximumMessageSize() == 2000); - assertTrue(_queueMBean.getMaximumQueueDepth() == 1000); - - assertTrue(_queueMBean.getName().equals("testQueue")); - assertTrue(_queueMBean.getOwner().equals("AMQueueMBeanTest")); - assertFalse(_queueMBean.isAutoDelete()); - assertFalse(_queueMBean.isDurable()); - } - - public void testExceptions() throws Exception - { - try - { - _queueMBean.viewMessages(0, 3); - fail(); - } - catch (JMException ex) - { - - } - - try - { - _queueMBean.viewMessages(2, 1); - fail(); - } - catch (JMException ex) - { - - } - - try - { - _queueMBean.viewMessages(-1, 1); - fail(); - } - catch (JMException ex) - { - - } - - AMQMessage msg = message(false); - long id = msg.getMessageId(); - _queue.clearQueue(); - _queue.deliver(msg); - _queueMBean.viewMessageContent(id); - try - { - _queueMBean.viewMessageContent(id + 1); - fail(); - } - catch (JMException ex) - { - - } - } - - private AMQMessage message(boolean immediate) throws AMQException - { - BasicPublishBody publish = new BasicPublishBody(); - publish.immediate = immediate; - ContentHeaderBody contentHeaderBody = new ContentHeaderBody(); - contentHeaderBody.bodySize = 1000; // in bytes - return new AMQMessage(_messageStore, publish, contentHeaderBody, null); - } - - @Override - protected void setUp() throws Exception - { - super.setUp(); - _queueRegistry = new DefaultQueueRegistry(); - _queue = new AMQQueue("testQueue", false, "AMQueueMBeanTest", false, _queueRegistry); - _queueMBean = new AMQQueueMBean(_queue); - } - - private void sendMessages(int messageCount) throws AMQException - { - AMQMessage[] messages = new AMQMessage[messageCount]; - for (int i = 0; i < messages.length; i++) - { - messages[i] = message(false);; - } - for (int i = 0; i < messageCount; i++) - { - _queue.deliver(messages[i]); - } - } -} diff --git a/qpid/java/systests/src/test/java/org/apache/qpid/server/queue/MockProtocolSession.java b/qpid/java/systests/src/test/java/org/apache/qpid/server/queue/MockProtocolSession.java index bb8fd5bc19..87e5c43932 100644 --- a/qpid/java/systests/src/test/java/org/apache/qpid/server/queue/MockProtocolSession.java +++ b/qpid/java/systests/src/test/java/org/apache/qpid/server/queue/MockProtocolSession.java @@ -22,6 +22,7 @@ package org.apache.qpid.server.queue; import org.apache.qpid.AMQException; import org.apache.qpid.framing.AMQDataBlock; +import org.apache.qpid.framing.FieldTable; import org.apache.qpid.server.AMQChannel; import org.apache.qpid.server.protocol.AMQProtocolSession; import org.apache.qpid.server.store.MessageStore; @@ -121,4 +122,13 @@ public class MockProtocolSession implements AMQProtocolSession public void setSaslServer(SaslServer saslServer) { } + + public FieldTable getClientProperties() + { + return null; + } + + public void setClientProperties(FieldTable clientProperties) + { + } } diff --git a/qpid/java/systests/src/test/java/org/apache/qpid/server/queue/SubscriptionTestHelper.java b/qpid/java/systests/src/test/java/org/apache/qpid/server/queue/SubscriptionTestHelper.java index 2773c810d2..fea3c93280 100644 --- a/qpid/java/systests/src/test/java/org/apache/qpid/server/queue/SubscriptionTestHelper.java +++ b/qpid/java/systests/src/test/java/org/apache/qpid/server/queue/SubscriptionTestHelper.java @@ -22,6 +22,7 @@ package org.apache.qpid.server.queue; import java.util.ArrayList; import java.util.List; +import java.util.Queue; public class SubscriptionTestHelper implements Subscription { @@ -70,6 +71,41 @@ public class SubscriptionTestHelper implements Subscription { } + public boolean hasFilters() + { + return false; + } + + public boolean hasInterest(AMQMessage msg) + { + return true; + } + + public Queue getPreDeliveryQueue() + { + return null; + } + + public void enqueueForPreDelivery(AMQMessage msg) + { + //no-op + } + + public boolean isAutoClose() + { + return false; + } + + public void close() + { + //no-op + } + + public boolean isBrowser() + { + return false; //To change body of implemented methods use File | Settings | File Templates. + } + public int hashCode() { return key.hashCode(); -- cgit v1.2.1