Chapter 13. Filtering Messages
Chapter 13. Filtering Messages AMQ Broker provides a powerful filter language based on a subset of the SQL 92 expression syntax. The filter language uses the same syntax as used for JMS selectors, but the predefined identifiers are different. The table below lists the identifiers that apply to a AMQ Broker message. Identifier Attribute AMQPriority The priority of a message. Message priorities are integers with valid values from 0 through 9 . 0 is the lowest priority and 9 is the highest. AMQExpiration The expiration time of a message. The value is a long integer. AMQDurable Whether a message is durable or not. The value is a string. Valid values are DURABLE or NON_DURABLE . AMQTimestamp The timestamp of when the message was created. The value is a long integer. AMQSize The value of the encodeSize property of the message. The value of encodeSize is the space, in bytes, that the message takes up in the journal. Because the broker uses a double-byte character set to encode messages, the actual size of the message is half the value of encodeSize . Any other identifiers used in core filter expressions are assumed to be properties of the message. For documentation on selector syntax for JMS Messages, see the Java EE API . 13.1. Configuring a Queue to Use a Filter You can add a filter to the queues you configure in <broker_instance_dir> /etc/broker.xml . Only messages that match the filter expression enter the queue. Procedure Add the filter element to the desired queue and include the filter you want to apply as the value of the element. In the example below, the filter NEWS='technology' is added to the queue technologyQueue . <configuration> <core> ... <addresses> <address name="myQueue"> <anycast> <queue name="myQueue"> <filter string="NEWS='technology'"/> </queue> </anycast> </address> </addresses> </core> </configuration> 13.2. Filtering JMS Message Properties The JMS specification states that a String property must not be converted to a numeric type when used in a selector. For example, if a message has the age property set to the String value 21 , the selector age > 18 must not match it. This restriction limits STOMP clients because they can send only messages with String properties. Configuring a Filter to Convert a String to a Number To convert String properties to a numeric type, add the prefix convert_string_expressions: to the value of the filter . Procedure Edit <broker_instance_dir> /etc/broker.xml by applying the prefix convert_string_expressions: to the desired filter . The example below edits the filter value from age > 18 to convert_string_expressions:age > 18 . <configuration> <core> ... <addresses> <address name="myQueue"> <anycast> <queue name="myQueue"> <filter string="convert_string_expressions='age > 18'"/> </queue> </anycast> </address> </addresses> </core> </configuration> 13.3. Filtering AMQP Messages Based on Properties on Annotations Before the broker moves an expired or undelivered AMQP message to an expiry or dead letter queue that you have configured, the broker applies annotations and properties to the message. A client can create a filter based on the properties or annotations, to select particular messages to consume from the expiry or dead letter queue. Note The properties that the broker applies are internal properties These properties are are not exposed to clients for regular use, but can be specified by a client in a filter. Shown below are examples of filters based on message properties and annotations. Filtering based on properties is the recommended approach, when possible, because this approach requires less processing by the broker. Filter based on message properties ConnectionFactory factory = new JmsConnectionFactory("amqp://localhost:5672"); Connection connection = factory.createConnection(); Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); connection.start(); javax.jms.Queue queue = session.createQueue("my_DLQ"); MessageConsumer consumer = session.createConsumer(queue, "_AMQ_ORIG_ADDRESS='original_address_name'"); Message message = consumer.receive(); Filter based on message annotations ConnectionFactory factory = new JmsConnectionFactory("amqp://localhost:5672"); Connection connection = factory.createConnection(); Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); connection.start(); javax.jms.Queue queue = session.createQueue("my_DLQ"); MessageConsumer consumer = session.createConsumer(queue, "\"m.x-opt-ORIG-ADDRESS\"='original_address_name'"); Message message = consumer.receive(); Note When consuming AMQP messages based on an annotation, the client must include append a m. prefix to the message annotation, as shown in the preceding example. Additional resources For more information about the annotations and properties that the broker applies to expired or undelivered AMQP messages, see Section 4.14, "Annotations and properties on expired or undelivered AMQP messages" . 13.4. Filtering XML Messages AMQ Broker provides a way of filtering Text messages that contain an XML body using XPath. XPath (XML Path Language) is a query language for selecting nodes from an XML document. Note Only text based messages are supported. Filtering large messages is not supported. To filter text based messages, you need to create a Message Selector of the form XPATH '<xpath-expression> . An example of a Message Body <root> <a key='first' num='1'/> <b key='second' num='2'>b</b> </root> Filter based on an XPath query Warning Since XPath applies to the body of the message and requires parsing of XML, filtering can be significantly slower than normal filters. XPath filters are supported with and between producers and consumers using the following protocols: OpenWire JMS Core (and Core JMS) STOMP AMQP Configuring the XML Parser By default the XML Parser used by the Broker is the Platform default DocumentBuilderFactory instance used by the JDK. The XML parser used for XPath default configuration includes the following settings: http://xml.org/sax/features/external-general-entities : false http://xml.org/sax/features/external-parameter-entities : false http://apache.org/xml/features/disallow-doctype-decl : true However, in order to deal with any implementation-specific issues the features can be customized by configuring System properties in the artemis.profile configuration file. org.apache.activemq.documentBuilderFactory.feature:prefix Example feature configuration
<configuration> <core> <addresses> <address name="myQueue"> <anycast> <queue name="myQueue"> <filter string="NEWS='technology'"/> </queue> </anycast> </address> </addresses> </core> </configuration> <configuration> <core> <addresses> <address name="myQueue"> <anycast> <queue name="myQueue"> <filter string="convert_string_expressions='age > 18'"/> </queue> </anycast> </address> </addresses> </core> </configuration> ConnectionFactory factory = new JmsConnectionFactory("amqp://localhost:5672"); Connection connection = factory.createConnection(); Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); connection.start(); javax.jms.Queue queue = session.createQueue("my_DLQ"); MessageConsumer consumer = session.createConsumer(queue, "_AMQ_ORIG_ADDRESS='original_address_name'"); Message message = consumer.receive(); ConnectionFactory factory = new JmsConnectionFactory("amqp://localhost:5672"); Connection connection = factory.createConnection(); Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); connection.start(); javax.jms.Queue queue = session.createQueue("my_DLQ"); MessageConsumer consumer = session.createConsumer(queue, "\"m.x-opt-ORIG-ADDRESS\"='original_address_name'"); Message message = consumer.receive(); <root> <a key='first' num='1'/> <b key='second' num='2'>b</b> </root> PATH 'root/a' -Dorg.apache.activemq.documentBuilderFactory.feature:http://xml.org/sax/features/external-general-entities=true