QPID-4558: Selectors for C++ broker

- Added numeric and boolean values
* To literals and identifier values
* To the code that extracts values from message properties
- Added the full set of comparison operators
- Implemented full "unknown" semantics for all
  operators.
- Implemented extended "is null" and "is not null" operators
  that allow expressions as well as just identifiers.

git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk/qpid@1452525 13f79535-47bb-0310-9956-ffa450edef68

8 files changed, 720 insertions, 186 deletions

diff --git a/cpp/src/qpid/broker/Selector.cpp b/cpp/src/qpid/broker/Selector.cpp
index 0880285692..62d2c43005 100644
--- a/cpp/src/qpid/broker/Selector.cpp
+++ b/cpp/src/qpid/broker/Selector.cpp
@@ -23,56 +23,23 @@
 
 #include "qpid/broker/Message.h"
 #include "qpid/broker/SelectorExpression.h"
+#include "qpid/broker/SelectorValue.h"
 #include "qpid/log/Statement.h"
+#include "qpid/types/Variant.h"
 
 #include <string>
 #include <sstream>
+#include "qpid/sys/unordered_map.h"
 
 #include <boost/make_shared.hpp>
 #include <boost/lexical_cast.hpp>
+#include <boost/ptr_container/ptr_vector.hpp>
 
 namespace qpid {
 namespace broker {
 
 using std::string;
-
-Selector::Selector(const string& e) :
-    parse(parseTopBoolExpression(e)),
-    expression(e)
-{
-    bool debugOut;
-    QPID_LOG_TEST(debug, debugOut);
-    if (debugOut) {
-        std::stringstream ss;
-        parse->repr(ss);
-        QPID_LOG(debug, "Selector parsed[" << e << "] into: " << ss.str());
-    }
-}
-
-Selector::~Selector()
-{
-}
-
-bool Selector::eval(const SelectorEnv& env)
-{
-    return parse->eval(env);
-}
-
-bool Selector::filter(const Message& msg)
-{
-    return eval(MessageSelectorEnv(msg));
-}
-
-MessageSelectorEnv::MessageSelectorEnv(const Message& m) :
-    msg(m)
-{
-}
-
-bool MessageSelectorEnv::present(const string& identifier) const
-{
-    // If the value we get is void then most likely the property wasn't present
-    return !msg.getProperty(identifier).isVoid();
-}
+using qpid::sys::unordered_map;
 
 /**
  * Identifier  (amqp.)  | JMS...       | amqp 1.0 equivalent
@@ -89,48 +56,124 @@ bool MessageSelectorEnv::present(const string& identifier) const
  * creation_time        | Timestamp    | creation-time        properties section
  * jms_type             | Type         | jms-type             message-annotations section
  */
+const string EMPTY;
+const string PERSISTENT("PERSISTENT");
+const string NON_PERSISTENT("NON_PERSISTENT");
 
-string specialValue(const Message& msg, const string& id)
+const Value specialValue(const Message& msg, const string& id)
 {
     // TODO: Just use a simple if chain for now - improve this later
     if ( id=="delivery_mode" ) {
-        return msg.getEncoding().isPersistent() ? "PERSISTENT" : "NON_PERSISTENT";
+        return msg.getEncoding().isPersistent() ? PERSISTENT : NON_PERSISTENT;
     } else if ( id=="redelivered" ) {
-        return msg.getDeliveryCount()>0 ? "TRUE" : "FALSE";
+        return msg.getDeliveryCount()>0 ? true : false;
     } else if ( id=="priority" ) {
-        return boost::lexical_cast<string>(static_cast<uint32_t>(msg.getEncoding().getPriority()));
+        return int64_t(msg.getPriority());
     } else if ( id=="correlation_id" ) {
-        return ""; // Needs an indirection in getEncoding().
+        return EMPTY; // Needs an indirection in getEncoding().
     } else if ( id=="message_id" ) {
-        return ""; // Needs an indirection in getEncoding().
+        return EMPTY; // Needs an indirection in getEncoding().
     } else if ( id=="to" ) {
-        return msg.getRoutingKey(); // This is good for 0-10, not sure about 1.0
+        return EMPTY; // This is good for 0-10, not sure about 1.0
     } else if ( id=="reply_to" ) {
-        return ""; // Needs an indirection in getEncoding().
+        return EMPTY; // Needs an indirection in getEncoding().
     } else if ( id=="absolute_expiry_time" ) {
-        return ""; // Needs an indirection in getEncoding().
+        return EMPTY; // Needs an indirection in getEncoding().
     } else if ( id=="creation_time" ) {
-        return ""; // Needs an indirection in getEncoding().
+        return EMPTY; // Needs an indirection in getEncoding().
     } else if ( id=="jms_type" ) {
-        return msg.getAnnotation("jms-type");
-    } else return "";
+        return EMPTY;
+    } else return Value();
 }
 
-string MessageSelectorEnv::value(const string& identifier) const
+class MessageSelectorEnv : public SelectorEnv {
+    const Message& msg;
+    mutable boost::ptr_vector<string> returnedStrings;
+    mutable unordered_map<string, Value> returnedValues;
+
+    const Value& value(const string&) const;
+
+public:
+    MessageSelectorEnv(const Message&);
+};
+
+MessageSelectorEnv::MessageSelectorEnv(const Message& m) :
+msg(m)
+{
+}
+
+const Value& MessageSelectorEnv::value(const string& identifier) const
+{
+    if (returnedValues.find(identifier)==returnedValues.end()) {
+        Value v;
+
+        // Check for amqp prefix and strip it if present
+        if (identifier.substr(0, 5) == "amqp.") {
+            v = specialValue(msg, identifier.substr(5));
+        } else {
+            // Just return property as string
+            //v = &msg.getPropertyAsString(identifier);
+            qpid::types::Variant var = msg.getProperty(identifier);
+            switch (var.getType()) {
+            case types::VAR_VOID:
+                v = Value(); break;
+            case types::VAR_STRING: {
+                string& s = var.getString();
+                returnedStrings.push_back(new string(s));
+                v = returnedStrings[returnedStrings.size()-1];
+                break;
+            }
+            case types::VAR_UINT64:
+                // TODO: Need to take care of values too high to be int64_t
+            case types::VAR_UINT32:
+            case types::VAR_UINT16:
+            case types::VAR_UINT8:
+            case types::VAR_INT64:
+            case types::VAR_INT32:
+            case types::VAR_INT16:
+            case types::VAR_INT8:
+                v = var.asInt64(); break;
+            case types::VAR_FLOAT:
+            case types::VAR_DOUBLE:
+                v = var.asDouble(); break;
+            case types::VAR_BOOL:
+                v = var.asBool(); break;
+            default:
+                v = Value(); break;
+            }
+        }
+        QPID_LOG(debug, "Selector identifier: " << identifier << "->" << v);
+        returnedValues[identifier] = v;
+    }
+    return returnedValues[identifier];
+}
+
+Selector::Selector(const string& e) :
+    parse(TopExpression::parse(e)),
+    expression(e)
 {
-    string v;
-
-    // Check for amqp prefix and strip it if present
-    if (identifier.substr(0, 5) == "amqp.") {
-        v = specialValue(msg, identifier.substr(5));
-    } else {
-        // Just return property as string
-        v = msg.getPropertyAsString(identifier);
+    bool debugOut;
+    QPID_LOG_TEST(debug, debugOut);
+    if (debugOut) {
+        std::stringstream ss;
+        parse->repr(ss);
+        QPID_LOG(debug, "Selector parsed[" << e << "] into: " << ss.str());
     }
-    QPID_LOG(debug, "Selector identifier: " << identifier << "->" << v);
-    return v;
 }
 
+Selector::~Selector()
+{
+}
+
+bool Selector::eval(const SelectorEnv& env)
+{
+    return parse->eval(env);
+}
+
+bool Selector::filter(const Message& msg)
+{
+    return eval(MessageSelectorEnv(msg));
+}
 
 namespace {
 const boost::shared_ptr<Selector> NULL_SELECTOR = boost::shared_ptr<Selector>();
diff --git a/cpp/src/qpid/broker/Selector.h b/cpp/src/qpid/broker/Selector.h
index 352d9c8e02..0412e6fb8d 100644
--- a/cpp/src/qpid/broker/Selector.h
+++ b/cpp/src/qpid/broker/Selector.h
@@ -31,7 +31,8 @@ namespace qpid {
 namespace broker {
 
 class Message;
-class BoolExpression;
+class Value;
+class TopExpression;
 
 /**
  * Interface to provide values to a Selector evaluation
@@ -42,22 +43,11 @@ class SelectorEnv {
 public:
     virtual ~SelectorEnv() {};
 
-    virtual bool present(const std::string&) const = 0;
-    virtual std::string value(const std::string&) const = 0;
-};
-
-class MessageSelectorEnv : public SelectorEnv {
-    const Message& msg;
-
-    bool present(const std::string&) const;
-    std::string value(const std::string&) const;
-
-public:
-    MessageSelectorEnv(const Message&);
+    virtual const Value& value(const std::string&) const = 0;
 };
 
 class Selector {
-    boost::scoped_ptr<BoolExpression> parse;
+    boost::scoped_ptr<TopExpression> parse;
     const std::string expression;
 
 public:
diff --git a/cpp/src/qpid/broker/SelectorExpression.cpp b/cpp/src/qpid/broker/SelectorExpression.cpp
index 4593b21c83..36d3a7d8ec 100644
--- a/cpp/src/qpid/broker/SelectorExpression.cpp
+++ b/cpp/src/qpid/broker/SelectorExpression.cpp
@@ -23,12 +23,15 @@
 
 #include "qpid/broker/Selector.h"
 #include "qpid/broker/SelectorToken.h"
+#include "qpid/broker/SelectorValue.h"
+#include "qpid/sys/IntegerTypes.h"
 
 #include <string>
 #include <memory>
 #include <ostream>
 
 #include <boost/scoped_ptr.hpp>
+#include <boost/lexical_cast.hpp>
 
 /*
  * Syntax for JMS style selector expressions (informal):
@@ -42,17 +45,15 @@
  *
  * LiteralString ::= ("'" ~[']* "'")+ // Repeats to cope with embedded single quote
  *
- * // Currently no numerics at all
- * //LiteralExactNumeric ::= Digit+
- * //LiteralApproxNumeric ::= ( Digit "." Digit* [ "E" LiteralExactNumeric ] ) |
- * //                         ( "." Digit+ [ "E" LiteralExactNumeric ] ) |
- * //                         ( Digit+ "E" LiteralExactNumeric )
- * //LiteralBool ::= "TRUE" | "FALSE"
- * //
+ * LiteralExactNumeric ::= Digit+
+ * Exponent ::= ['+'|'-'] LiteralExactNumeric
+ * LiteralApproxNumeric ::= ( Digit "." Digit* [ "E" Exponent ] ) |
+ *                          ( "." Digit+ [ "E" Exponent ] ) |
+ *                          ( Digit+ "E" Exponent )
+ * LiteralBool ::= "TRUE" | "FALSE"
  *
  * Literal ::= LiteralBool | LiteralString | LiteralApproxNumeric | LiteralExactNumeric
  *
- * // Currently only simple string comparison expressions + IS NULL or IS NOT NULL
  * EqOps ::= "=" | "<>"
  *
  * ComparisonOps ::= EqOps | ">" | ">=" | "<" | "<="
@@ -63,38 +64,54 @@
  *
  * AndExpression :: = EqualityExpression ( "AND" EqualityExpression  )*
  *
- * EqualityExpression ::= Identifier "IS" "NULL" |
- *                        Identifier "IS "NOT" "NULL" |
- *                        PrimaryExpression EqOps PrimaryExpression |
- *                        "NOT" EqualityExpression |
+ * ComparisonExpression ::= PrimaryExpression "IS" "NULL" |
+ *                        PrimaryExpression "IS" "NOT" "NULL" |
+ *                        PrimaryExpression ComparisonOpsOps PrimaryExpression |
+ *                        "NOT" ComparisonExpression |
  *                        "(" OrExpression ")"
  *
  * PrimaryExpression :: = Identifier |
- *                        LiteralString
+ *                        Literal
  *
  */
 
 namespace qpid {
 namespace broker {
 
-class Expression;
-
 using std::string;
 using std::ostream;
 
+class Expression {
+public:
+    virtual ~Expression() {}
+    virtual void repr(std::ostream&) const = 0;
+    virtual Value eval(const SelectorEnv&) const = 0;
+};
+
+class BoolExpression : public Expression {
+public:
+    virtual ~BoolExpression() {}
+    virtual void repr(std::ostream&) const = 0;
+    virtual BoolOrNone eval_bool(const SelectorEnv&) const = 0;
+
+    Value eval(const SelectorEnv& env) const {
+        return eval_bool(env);
+    }
+};
+
 // Operators
 
-class EqualityOperator {
+class ComparisonOperator {
 public:
     virtual void repr(ostream&) const = 0;
-    virtual bool eval(Expression&, Expression&, const SelectorEnv&) const = 0;
+    virtual BoolOrNone eval(Expression&, Expression&, const SelectorEnv&) const = 0;
 };
 
 template <typename T>
 class UnaryBooleanOperator {
 public:
     virtual void repr(ostream&) const = 0;
-    virtual bool eval(T&, const SelectorEnv&) const = 0;
+    virtual BoolOrNone eval(T&, const SelectorEnv&) const = 0;
 };
 
 ////////////////////////////////////////////////////
@@ -107,13 +124,7 @@ ostream& operator<<(ostream& os, const Expression& e)
     return os;
 }
 
-ostream& operator<<(ostream& os, const BoolExpression& e)
-{
-    e.repr(os);
-    return os;
-}
-
-ostream& operator<<(ostream& os, const EqualityOperator& e)
+ostream& operator<<(ostream& os, const ComparisonOperator& e)
 {
     e.repr(os);
     return os;
@@ -128,13 +139,13 @@ ostream& operator<<(ostream& os, const UnaryBooleanOperator<T>& e)
 
 // Boolean Expression types...
 
-class EqualityExpression : public BoolExpression {
-    EqualityOperator* op;
+class ComparisonExpression : public BoolExpression {
+    ComparisonOperator* op;
     boost::scoped_ptr<Expression> e1;
     boost::scoped_ptr<Expression> e2;
 
 public:
-    EqualityExpression(EqualityOperator* o, Expression* e, Expression* e_):
+    ComparisonExpression(ComparisonOperator* o, Expression* e, Expression* e_):
         op(o),
         e1(e),
         e2(e_)
@@ -144,7 +155,7 @@ public:
         os << "(" << *e1 << *op << *e2 << ")";
     }
 
-    bool eval(const SelectorEnv& env) const {
+    BoolOrNone eval_bool(const SelectorEnv& env) const {
         return op->eval(*e1, *e2, env);
     }
 };
@@ -163,9 +174,19 @@ public:
         os << "(" << *e1 << " OR " << *e2 << ")";
     }
 
-    // We can use the regular C++ short-circuiting operator||
-    bool eval(const SelectorEnv& env) const {
-        return e1->eval(env) || e2->eval(env);
+    BoolOrNone eval_bool(const SelectorEnv& env) const {
+        BoolOrNone bn1(e1->eval_bool(env));
+        if (bn1==BN_TRUE) {
+            return BN_TRUE;
+        } else {
+            BoolOrNone bn2(e2->eval_bool(env));
+            if (bn2==BN_TRUE) {
+                return BN_TRUE;
+            } else {
+                if (bn1==BN_FALSE && bn2==BN_FALSE) return BN_FALSE;
+                else return BN_UNKNOWN;
+            }
+        }
     }
 };
 
@@ -183,9 +204,19 @@ public:
         os << "(" << *e1 << " AND " << *e2 << ")";
     }
 
-    // We can use the regular C++ short-circuiting operator&&
-    bool eval(const SelectorEnv& env) const {
-        return e1->eval(env) && e2->eval(env);
+    BoolOrNone eval_bool(const SelectorEnv& env) const {
+        BoolOrNone bn1(e1->eval_bool(env));
+        if (bn1==BN_FALSE) {
+            return BN_FALSE;
+        } else {
+            BoolOrNone bn2(e2->eval_bool(env));
+            if (bn2==BN_FALSE) {
+                return BN_FALSE;
+            } else {
+                if (bn1==BN_TRUE && bn2==BN_TRUE) return BN_TRUE;
+                else return BN_UNKNOWN;
+            }
+        }
     }
 };
 
@@ -204,7 +235,7 @@ public:
         os << *op << "(" << *e1 << ")";
     }
 
-    virtual bool eval(const SelectorEnv& env) const {
+    virtual BoolOrNone eval_bool(const SelectorEnv& env) const {
         return op->eval(*e1, env);
     }
 };
@@ -212,10 +243,28 @@ public:
 // Expression types...
 
 class Literal : public Expression {
-    string value;
+    const Value value;
+
+public:
+    template <typename T>
+    Literal(const T& v) :
+        value(v)
+    {}
+
+    void repr(ostream& os) const {
+        os << value;
+    }
+
+    Value eval(const SelectorEnv&) const {
+        return value;
+    }
+};
+
+class StringLiteral : public Expression {
+    const string value;
 
 public:
-    Literal(const string& v) :
+    StringLiteral(const string& v) :
         value(v)
     {}
 
@@ -223,7 +272,7 @@ public:
         os << "'" << value << "'";
     }
 
-    string eval(const SelectorEnv&) const {
+    Value eval(const SelectorEnv&) const {
         return value;
     }
 };
@@ -240,60 +289,113 @@ public:
         os << "I:" << identifier;
     }
 
-    string eval(const SelectorEnv& env) const {
+    Value eval(const SelectorEnv& env) const {
         return env.value(identifier);
     }
-
-    bool present(const SelectorEnv& env) const {
-        return env.present(identifier);
-    }
 };
 
 ////////////////////////////////////////////////////
 
 // Some operators...
 
+typedef bool BoolOp(const Value&, const Value&);
+
+BoolOrNone booleval(BoolOp* op, Expression& e1, Expression& e2, const SelectorEnv& env) {
+    const Value v1(e1.eval(env));
+    if (!unknown(v1)) {
+        const Value v2(e2.eval(env));
+        if (!unknown(v2)) {
+            return BoolOrNone(op(v1, v2));
+        }
+    }
+    return BN_UNKNOWN;
+}
+
 // "="
-class Eq : public EqualityOperator {
+class Eq : public ComparisonOperator {
     void repr(ostream& os) const {
         os << "=";
     }
 
-    bool eval(Expression& e1, Expression& e2, const SelectorEnv& env) const {
-        return e1.eval(env) == e2.eval(env);
+    BoolOrNone eval(Expression& e1, Expression& e2, const SelectorEnv& env) const {
+        return booleval(&operator==, e1, e2, env);
     }
 };
 
 // "<>"
-class Neq : public EqualityOperator {
+class Neq : public ComparisonOperator {
     void repr(ostream& os) const {
         os << "<>";
     }
 
-    bool eval(Expression& e1, Expression& e2, const SelectorEnv& env) const {
-        return e1.eval(env) != e2.eval(env);
+    BoolOrNone eval(Expression& e1, Expression& e2, const SelectorEnv& env) const {
+        return booleval(&operator!=, e1, e2, env);
+    }
+};
+
+// "<"
+class Ls : public ComparisonOperator {
+    void repr(ostream& os) const {
+        os << "<";
+    }
+
+    BoolOrNone eval(Expression& e1, Expression& e2, const SelectorEnv& env) const {
+        return booleval(&operator<, e1, e2, env);
+    }
+};
+
+// ">"
+class Gr : public ComparisonOperator {
+    void repr(ostream& os) const {
+        os << ">";
+    }
+
+    BoolOrNone eval(Expression& e1, Expression& e2, const SelectorEnv& env) const {
+        return booleval(&operator>, e1, e2, env);
+    }
+};
+
+// "<="
+class Lseq : public ComparisonOperator {
+    void repr(ostream& os) const {
+        os << "<=";
+    }
+
+    BoolOrNone eval(Expression& e1, Expression& e2, const SelectorEnv& env) const {
+        return booleval(&operator<=, e1, e2, env);
+    }
+};
+
+// ">="
+class Greq : public ComparisonOperator {
+    void repr(ostream& os) const {
+        os << ">=";
+    }
+
+    BoolOrNone eval(Expression& e1, Expression& e2, const SelectorEnv& env) const {
+        return booleval(&operator>=, e1, e2, env);
     }
 };
 
 // "IS NULL"
-class IsNull : public UnaryBooleanOperator<Identifier> {
+class IsNull : public UnaryBooleanOperator<Expression> {
     void repr(ostream& os) const {
         os << "IsNull";
     }
 
-    bool eval(Identifier& i, const SelectorEnv& env) const {
-        return !i.present(env);
+    BoolOrNone eval(Expression& e, const SelectorEnv& env) const {
+        return BoolOrNone(unknown(e.eval(env)));
     }
 };
 
 // "IS NOT NULL"
-class IsNonNull : public UnaryBooleanOperator<Identifier> {
+class IsNonNull : public UnaryBooleanOperator<Expression> {
     void repr(ostream& os) const {
         os << "IsNonNull";
     }
 
-    bool eval(Identifier& i, const SelectorEnv& env) const {
-        return i.present(env);
+    BoolOrNone eval(Expression& e, const SelectorEnv& env) const {
+        return BoolOrNone(!unknown(e.eval(env)));
     }
 };
 
@@ -303,21 +405,51 @@ class Not : public UnaryBooleanOperator<BoolExpression> {
         os << "NOT";
     }
 
-    bool eval(BoolExpression& e, const SelectorEnv& env) const {
-        return !e.eval(env);
+    BoolOrNone eval(BoolExpression& e, const SelectorEnv& env) const {
+        BoolOrNone bn = e.eval_bool(env);
+        if (bn==BN_UNKNOWN) return bn;
+        else return BoolOrNone(!bn);
     }
 };
 
 Eq eqOp;
 Neq neqOp;
+Ls lsOp;
+Gr grOp;
+Lseq lseqOp;
+Greq greqOp;
 IsNull isNullOp;
 IsNonNull isNonNullOp;
 Not notOp;
 
 ////////////////////////////////////////////////////
 
+BoolExpression* parseOrExpression(Tokeniser&);
+BoolExpression* parseAndExpression(Tokeniser&);
+BoolExpression* parseComparisonExpression(Tokeniser&);
+Expression* parsePrimaryExpression(Tokeniser&);
+
 // Top level parser
-BoolExpression* parseTopBoolExpression(const string& exp)
+class TopBoolExpression : public TopExpression {
+    boost::scoped_ptr<BoolExpression> expression;
+
+    void repr(ostream& os) const {
+        expression->repr(os);
+    }
+
+    bool eval(const SelectorEnv& env) const {
+        BoolOrNone bn = expression->eval_bool(env);
+        if (bn==BN_TRUE) return true;
+        else return false;
+    }
+
+public:
+    TopBoolExpression(BoolExpression* be) :
+        expression(be)
+    {}
+};
+
+TopExpression* TopExpression::parse(const string& exp)
 {
     string::const_iterator s = exp.begin();
     string::const_iterator e = exp.end();
@@ -325,7 +457,7 @@ BoolExpression* parseTopBoolExpression(const string& exp)
     std::auto_ptr<BoolExpression> b(parseOrExpression(tokeniser));
     if (!b.get()) throw std::range_error("Illegal selector: couldn't parse");
     if (tokeniser.nextToken().type != T_EOS) throw std::range_error("Illegal selector: too much input");
-    return b.release();
+    return new TopBoolExpression(b.release());
 }
 
 BoolExpression* parseOrExpression(Tokeniser& tokeniser)
@@ -344,11 +476,11 @@ BoolExpression* parseOrExpression(Tokeniser& tokeniser)
 
 BoolExpression* parseAndExpression(Tokeniser& tokeniser)
 {
-    std::auto_ptr<BoolExpression> e(parseEqualityExpression(tokeniser));
+    std::auto_ptr<BoolExpression> e(parseComparisonExpression(tokeniser));
     if (!e.get()) return 0;
     while ( tokeniser.nextToken().type==T_AND ) {
         std::auto_ptr<BoolExpression> e1(e);
-        std::auto_ptr<BoolExpression> e2(parseEqualityExpression(tokeniser));
+        std::auto_ptr<BoolExpression> e2(parseComparisonExpression(tokeniser));
         if (!e2.get()) return 0;
         e.reset(new AndExpression(e1.release(), e2.release()));
     }
@@ -356,31 +488,16 @@ BoolExpression* parseAndExpression(Tokeniser& tokeniser)
     return e.release();
 }
 
-BoolExpression* parseEqualityExpression(Tokeniser& tokeniser)
+BoolExpression* parseComparisonExpression(Tokeniser& tokeniser)
 {
     const Token t = tokeniser.nextToken();
-    if ( t.type==T_IDENTIFIER ) {
-        // Check for "IS NULL" and "IS NOT NULL"
-        if ( tokeniser.nextToken().type==T_IS ) {
-            // The rest must be T_NULL or T_NOT, T_NULL
-            switch (tokeniser.nextToken().type) {
-            case T_NULL:
-                return new UnaryBooleanExpression<Identifier>(&isNullOp, new Identifier(t.val));
-            case T_NOT:
-                if ( tokeniser.nextToken().type == T_NULL)
-                    return new UnaryBooleanExpression<Identifier>(&isNonNullOp, new Identifier(t.val));
-            default:
-                return 0;
-            }
-        }
-        tokeniser.returnTokens();
-    } else if ( t.type==T_LPAREN ) {
+    if ( t.type==T_LPAREN ) {
         std::auto_ptr<BoolExpression> e(parseOrExpression(tokeniser));
         if (!e.get()) return 0;
         if ( tokeniser.nextToken().type!=T_RPAREN ) return 0;
         return e.release();
     } else if ( t.type==T_NOT ) {
-        std::auto_ptr<BoolExpression> e(parseEqualityExpression(tokeniser));
+        std::auto_ptr<BoolExpression> e(parseComparisonExpression(tokeniser));
         if (!e.get()) return 0;
         return new UnaryBooleanExpression<BoolExpression>(&notOp, e.release());
     }
@@ -389,6 +506,21 @@ BoolExpression* parseEqualityExpression(Tokeniser& tokeniser)
     std::auto_ptr<Expression> e1(parsePrimaryExpression(tokeniser));
     if (!e1.get()) return 0;
 
+    // Check for "IS NULL" and "IS NOT NULL"
+    if ( tokeniser.nextToken().type==T_IS ) {
+        // The rest must be T_NULL or T_NOT, T_NULL
+        switch (tokeniser.nextToken().type) {
+            case T_NULL:
+                return new UnaryBooleanExpression<Expression>(&isNullOp, e1.release());
+            case T_NOT:
+                if ( tokeniser.nextToken().type == T_NULL)
+                    return new UnaryBooleanExpression<Expression>(&isNonNullOp, e1.release());
+            default:
+                return 0;
+        }
+    }
+    tokeniser.returnTokens();
+
     const Token op = tokeniser.nextToken();
     if (op.type != T_OPERATOR) {
         return 0;
@@ -397,8 +529,12 @@ BoolExpression* parseEqualityExpression(Tokeniser& tokeniser)
     std::auto_ptr<Expression> e2(parsePrimaryExpression(tokeniser));
     if (!e2.get()) return 0;
 
-    if (op.val == "=") return new EqualityExpression(&eqOp, e1.release(), e2.release());
-    if (op.val == "<>") return new EqualityExpression(&neqOp, e1.release(), e2.release());
+    if (op.val == "=") return new ComparisonExpression(&eqOp, e1.release(), e2.release());
+    if (op.val == "<>") return new ComparisonExpression(&neqOp, e1.release(), e2.release());
+    if (op.val == "<") return new ComparisonExpression(&lsOp, e1.release(), e2.release());
+    if (op.val == ">") return new ComparisonExpression(&grOp, e1.release(), e2.release());
+    if (op.val == "<=") return new ComparisonExpression(&lseqOp, e1.release(), e2.release());
+    if (op.val == ">=") return new ComparisonExpression(&greqOp, e1.release(), e2.release());
 
     return 0;
 }
@@ -410,7 +546,15 @@ Expression* parsePrimaryExpression(Tokeniser& tokeniser)
         case T_IDENTIFIER:
             return new Identifier(t.val);
         case T_STRING:
-            return new Literal(t.val);
+            return new StringLiteral(t.val);
+        case T_FALSE:
+            return new Literal(false);
+        case T_TRUE:
+            return new Literal(true);
+        case T_NUMERIC_EXACT:
+            return new Literal(boost::lexical_cast<int64_t>(t.val));
+        case T_NUMERIC_APPROX:
+            return new Literal(boost::lexical_cast<double>(t.val));
         default:
             return 0;
     }
diff --git a/cpp/src/qpid/broker/SelectorExpression.h b/cpp/src/qpid/broker/SelectorExpression.h
index be73d4876c..158f086287 100644
--- a/cpp/src/qpid/broker/SelectorExpression.h
+++ b/cpp/src/qpid/broker/SelectorExpression.h
@@ -29,29 +29,15 @@ namespace qpid {
 namespace broker {
 
 class SelectorEnv;
-class Tokeniser;
 
-class Expression {
+class TopExpression {
 public:
-    virtual ~Expression() {}
-    virtual void repr(std::ostream&) const = 0;
-    virtual std::string eval(const SelectorEnv&) const = 0;
-};
-
-class BoolExpression {
-public:
-    virtual ~BoolExpression() {};
+    virtual ~TopExpression() {};
     virtual void repr(std::ostream&) const = 0;
     virtual bool eval(const SelectorEnv&) const = 0;
-};
 
-BoolExpression* parseTopBoolExpression(const std::string& exp);
-BoolExpression* parseBoolExpression(Tokeniser&);
-BoolExpression* parseOrExpression(Tokeniser&);
-BoolExpression* parseAndExpression(Tokeniser&);
-BoolExpression* parseNotExpression(Tokeniser&);
-BoolExpression* parseEqualityExpression(Tokeniser&);
-Expression* parsePrimaryExpression(Tokeniser&);
+    static TopExpression* parse(const std::string& exp);
+};
 
 }}
 
diff --git a/cpp/src/qpid/broker/SelectorToken.cpp b/cpp/src/qpid/broker/SelectorToken.cpp
index 6215f169d3..509d56af4b 100644
--- a/cpp/src/qpid/broker/SelectorToken.cpp
+++ b/cpp/src/qpid/broker/SelectorToken.cpp
@@ -221,10 +221,77 @@ bool tokeniseOperator(std::string::const_iterator& s, std::string::const_iterato
     return true;
 }
 
-// Can't parse numerics yet
-bool tokeniseNumeric(std::string::const_iterator& /*s*/, std::string::const_iterator& /*e*/, Token& /*tok*/)
+bool tokeniseNumeric(std::string::const_iterator& s, std::string::const_iterator& e, Token& tok)
 {
-    return false;
+    std::string::const_iterator t = s;
+
+    // Hand constructed state machine recogniser
+    enum {
+        START,
+        REJECT,
+        DIGIT,
+        DECIMAL_START,
+        DECIMAL,
+        EXPONENT_SIGN,
+        EXPONENT_START,
+        EXPONENT,
+        ACCEPT_EXACT,
+        ACCEPT_INEXACT
+    } state = START;
+
+    while (true)
+    switch (state) {
+    case START:
+        if (t==e) {state = REJECT;}
+        else if (std::isdigit(*t)) {++t; state = DIGIT;}
+        else if (*t=='.') {++t; state = DECIMAL_START;}
+        else state = REJECT;
+        break;
+    case DECIMAL_START:
+        if (t==e) {state = REJECT;}
+        else if (std::isdigit(*t)) {++t; state = DECIMAL;}
+        else state = REJECT;
+        break;
+    case EXPONENT_SIGN:
+        if (t==e) {state = REJECT;}
+        else if (*t=='-' || *t=='+') {++t; state = EXPONENT_START;}
+        else if (std::isdigit(*t)) {++t; state = EXPONENT;}
+        else state = REJECT;
+        break;
+    case EXPONENT_START:
+        if (t==e) {state = REJECT;}
+        else if (std::isdigit(*t)) {++t; state = EXPONENT;}
+        else state = REJECT;
+        break;
+    case DIGIT:
+        if (t==e) {state = ACCEPT_EXACT;}
+        else if (std::isdigit(*t)) {++t; state = DIGIT;}
+        else if (*t=='.') {++t; state = DECIMAL;}
+        else if (*t=='e' || *t=='E') {++t; state = EXPONENT_SIGN;}
+        else state = ACCEPT_EXACT;
+        break;
+    case DECIMAL:
+        if (t==e) {state = ACCEPT_INEXACT;}
+        else if (std::isdigit(*t)) {++t; state = DECIMAL;}
+        else if (*t=='e' || *t=='E') {++t; state = EXPONENT_SIGN;}
+        else state = ACCEPT_INEXACT;
+        break;
+    case EXPONENT:
+        if (t==e) {state = ACCEPT_INEXACT;}
+        else if (std::isdigit(*t)) {++t; state = EXPONENT;}
+        else state = ACCEPT_INEXACT;
+        break;
+    case ACCEPT_EXACT:
+        tok = Token(T_NUMERIC_EXACT, s, t);
+        s = t;
+        return true;
+    case ACCEPT_INEXACT:
+        tok = Token(T_NUMERIC_APPROX, s, t);
+        s = t;
+        return true;
+    case REJECT:
+        return false;
+    };
 }
 
 Tokeniser::Tokeniser(const std::string::const_iterator& s, const std::string::const_iterator& e) :
diff --git a/cpp/src/qpid/broker/SelectorToken.h b/cpp/src/qpid/broker/SelectorToken.h
index 03a7ca63cc..0fb4125778 100644
--- a/cpp/src/qpid/broker/SelectorToken.h
+++ b/cpp/src/qpid/broker/SelectorToken.h
@@ -83,19 +83,12 @@ public:
 };
 
 bool tokeniseEos(std::string::const_iterator& s, std::string::const_iterator& e, Token& tok);
-
 bool tokeniseIdentifier(std::string::const_iterator& s, std::string::const_iterator& e, Token& tok);
-
 bool tokeniseReservedWord(std::string::const_iterator& s, std::string::const_iterator& e, Token& tok);
-
 bool tokeniseIdentifierOrReservedWord(std::string::const_iterator& s, std::string::const_iterator& e, Token& tok);
-
 bool tokeniseString(std::string::const_iterator& s, std::string::const_iterator& e, Token& tok);
-
 bool tokeniseParens(std::string::const_iterator& s, std::string::const_iterator& e, Token& tok);
-
 bool tokeniseOperator(std::string::const_iterator& s, std::string::const_iterator& e, Token& tok);
-
 bool tokeniseNumeric(std::string::const_iterator& s, std::string::const_iterator& e, Token& tok);
 
 class Tokeniser {
diff --git a/cpp/src/qpid/broker/SelectorValue.cpp b/cpp/src/qpid/broker/SelectorValue.cpp
new file mode 100644
index 0000000000..d35e4edd3e
--- /dev/null
+++ b/cpp/src/qpid/broker/SelectorValue.cpp
@@ -0,0 +1,196 @@
+/*
+ *
+ * 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.
+ *
+ */
+
+#include "qpid/broker/SelectorValue.h"
+
+#include <ostream>
+#include <boost/scoped_ptr.hpp>
+
+using std::ostream;
+
+namespace qpid {
+namespace broker {
+
+ostream& operator<<(ostream& os, const Value& v)
+{
+    switch (v.type) {
+    case Value::T_UNKNOWN: os << "UNKNOWN"; break;
+    case Value::T_BOOL: os << "BOOL:" << std::boolalpha << v.b; break;
+    case Value::T_EXACT: os << "EXACT:" << v.i; break;
+    case Value::T_INEXACT: os << "APPROX:" << v.x; break;
+    case Value::T_STRING: os << "STRING:'" << *v.s << "'"; break;
+    };
+    return os;
+}
+
+class NumericPairBase {
+public:
+    virtual Value add() = 0;
+    virtual Value sub() = 0;
+    virtual Value mul() = 0;
+    virtual Value div() = 0;
+
+    virtual bool eq() = 0;
+    virtual bool ne() = 0;
+    virtual bool ls() = 0;
+    virtual bool gr() = 0;
+    virtual bool le() = 0;
+    virtual bool ge() = 0;
+};
+
+template <typename T>
+class NumericPair : public NumericPairBase {
+    const T n1;
+    const T n2;
+
+    Value add() { return n1+n2; }
+    Value sub() { return n1-n2; }
+    Value mul() { return n1*n2; }
+    Value div() { return n1/n2; }
+
+    bool eq() { return n1==n2; }
+    bool ne() { return n1!=n2; }
+    bool ls() { return n1<n2; }
+    bool gr() { return n1>n2; }
+    bool le() { return n1<=n2; }
+    bool ge() { return n1>=n2; }
+
+public:
+    NumericPair(T x, T y) :
+        n1(x),
+        n2(y)
+    {}
+};
+
+NumericPairBase* promoteNumeric(const Value& v1, const Value& v2)
+{
+    if (!numeric(v1) || !numeric(v2)) return 0;
+
+    if (v1.type != v2.type) {
+        switch (v1.type) {
+        case Value::T_INEXACT: return new NumericPair<double>(v1.x, v2.i);
+        case Value::T_EXACT: return new NumericPair<double>(v1.i, v2.x);
+        default:
+            assert(false);
+        }
+    } else {
+        switch (v1.type) {
+        case Value::T_INEXACT: return new NumericPair<double>(v1.x, v2.x);
+        case Value::T_EXACT: return new NumericPair<int64_t>(v1.i, v2.i);
+        default:
+            assert(false);
+        }
+    }
+}
+
+bool operator==(const Value& v1, const Value& v2)
+{
+    boost::scoped_ptr<NumericPairBase> nbp(promoteNumeric(v1, v2));
+    if (nbp) return nbp->eq();
+
+    if (v1.type != v2.type) return false;
+    switch (v1.type) {
+    case Value::T_BOOL: return v1.b == v2.b;
+    case Value::T_STRING: return *v1.s == *v2.s;
+    default: // Cannot ever get here
+        return false;
+    }
+}
+
+bool operator!=(const Value& v1, const Value& v2)
+{
+    boost::scoped_ptr<NumericPairBase> nbp(promoteNumeric(v1, v2));
+    if (nbp) return nbp->ne();
+
+    if (v1.type != v2.type) return false;
+    switch (v1.type) {
+    case Value::T_BOOL: return v1.b != v2.b;
+    case Value::T_STRING: return *v1.s != *v2.s;
+    default: // Cannot ever get here
+        return false;
+    }
+}
+
+bool operator<(const Value& v1, const Value& v2)
+{
+    boost::scoped_ptr<NumericPairBase> nbp(promoteNumeric(v1, v2));
+    if (nbp) return nbp->ls();
+
+    return false;
+}
+
+bool operator>(const Value& v1, const Value& v2)
+{
+    boost::scoped_ptr<NumericPairBase> nbp(promoteNumeric(v1, v2));
+    if (nbp) return nbp->gr();
+
+    return false;
+}
+
+bool operator<=(const Value& v1, const Value& v2)
+{
+    boost::scoped_ptr<NumericPairBase> nbp(promoteNumeric(v1, v2));
+    if (nbp) return nbp->le();
+
+    return false;
+}
+
+bool operator>=(const Value& v1, const Value& v2)
+{
+    boost::scoped_ptr<NumericPairBase> nbp(promoteNumeric(v1, v2));
+    if (nbp) return nbp->ge();
+
+    return false;
+}
+
+Value operator+(const Value& v1, const Value& v2)
+{
+    boost::scoped_ptr<NumericPairBase> nbp(promoteNumeric(v1, v2));
+    if (nbp) return nbp->add();
+
+    return Value();
+}
+
+Value operator-(const Value& v1, const Value& v2)
+{
+    boost::scoped_ptr<NumericPairBase> nbp(promoteNumeric(v1, v2));
+    if (nbp) return nbp->sub();
+
+    return Value();
+}
+
+Value operator*(const Value& v1, const Value& v2)
+{
+    boost::scoped_ptr<NumericPairBase> nbp(promoteNumeric(v1, v2));
+    if (nbp) return nbp->mul();
+
+    return Value();
+}
+
+Value operator/(const Value& v1, const Value& v2)
+{
+    boost::scoped_ptr<NumericPairBase> nbp(promoteNumeric(v1, v2));
+    if (nbp) return nbp->div();
+
+    return Value();
+}
+
+}}
\ No newline at end of file
diff --git a/cpp/src/qpid/broker/SelectorValue.h b/cpp/src/qpid/broker/SelectorValue.h
new file mode 100644
index 0000000000..ec3fe29cf6
--- /dev/null
+++ b/cpp/src/qpid/broker/SelectorValue.h
@@ -0,0 +1,115 @@
+#ifndef QPID_BROKER_SELECTORVALUE_H
+#define QPID_BROKER_SELECTORVALUE_H
+
+/*
+ *
+ * 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.
+ *
+ */
+
+#include "qpid/sys/IntegerTypes.h"
+
+#include <iosfwd>
+#include <string>
+
+namespace qpid {
+namespace broker {
+
+enum BoolOrNone {
+    BN_FALSE = false,
+    BN_TRUE = true,
+    BN_UNKNOWN
+};
+
+// The user of the Value class for strings must ensure that
+// the string has a lifetime longer than the string used and
+// is responsible for managing its lifetime.
+class Value {
+public:
+    union {
+        bool               b;
+        int64_t            i;
+        double             x;
+        const std::string* s;
+    };
+    enum {
+        T_UNKNOWN,
+        T_BOOL,
+        T_STRING,
+        T_EXACT,
+        T_INEXACT
+    } type;
+
+    // Default copy contructor
+    // Default assignment operator
+    // Default destructor
+    Value() :
+        type(T_UNKNOWN)
+    {}
+
+    Value(const std::string& s0) :
+        s(&s0),
+        type(T_STRING)
+    {}
+
+    Value(const int64_t i0) :
+        i(i0),
+        type(T_EXACT)
+    {}
+
+    Value(const double x0) :
+        x(x0),
+        type(T_INEXACT)
+    {}
+
+    Value(bool b0) :
+        b(b0),
+        type(T_BOOL)
+    {}
+
+    Value(BoolOrNone bn) :
+        b(bn),
+        type(bn==BN_UNKNOWN ? T_UNKNOWN : T_BOOL)
+    {}
+};
+
+inline bool unknown(const Value& v) {
+    return v.type == Value::T_UNKNOWN;
+}
+
+inline bool numeric(const Value& v) {
+    return v.type == Value::T_EXACT || v.type == Value::T_INEXACT;
+}
+
+std::ostream& operator<<(std::ostream& os, const Value& v);
+
+bool operator==(const Value&, const Value&);
+bool operator!=(const Value&, const Value&);
+bool operator<(const Value&, const Value&);
+bool operator>(const Value&, const Value&);
+bool operator<=(const Value&, const Value&);
+bool operator>=(const Value&, const Value&);
+
+Value operator+(const Value&, const Value&);
+Value operator-(const Value&, const Value&);
+Value operator*(const Value&, const Value&);
+Value operator/(const Value&, const Value&);
+
+}}
+
+#endif