Implemented topic pattern matching for the TopicExchange.

Corrected default bindings to use the exchange named "" rather than
"amqp.direct".
Added python and unit tests for all of the above.
Minor improvements to testlib.py, also some tests for testlib itself.



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

13 files changed, 407 insertions, 65 deletions

diff --git a/cpp/broker/Makefile b/cpp/broker/Makefile
index 58ba3a41b5..afe93c455a 100644
--- a/cpp/broker/Makefile
+++ b/cpp/broker/Makefile
@@ -27,11 +27,9 @@ LIB_OBJECTS= $(subst src/Broker.o,,$(OBJECTS))
 EXE_OBJECTS= src/Broker.o
 
 
-.PHONY: all clean test
+.PHONY: all clean 
 
 all: $(BROKER)
-
-test:
 	@$(MAKE) -C test all
 
 clean:
diff --git a/cpp/broker/inc/DirectExchange.h b/cpp/broker/inc/DirectExchange.h
index bf8c5f0b37..faf5a0b949 100644
--- a/cpp/broker/inc/DirectExchange.h
+++ b/cpp/broker/inc/DirectExchange.h
@@ -29,22 +29,19 @@
 namespace qpid {
 namespace broker {
     class DirectExchange : public virtual Exchange{
-        const string name;
         std::map<string, std::vector<Queue::shared_ptr> > bindings;
         qpid::concurrent::MonitorImpl lock;
 
     public:
         static const std::string typeName;
         
-        DirectExchange(const string& name);
+        DirectExchange(const std::string& name);
         
-        inline virtual const string& getName(){ return name; }
-        
-        virtual void bind(Queue::shared_ptr queue, const string& routingKey, qpid::framing::FieldTable* args);
+        virtual void bind(Queue::shared_ptr queue, const std::string& routingKey, qpid::framing::FieldTable* args);
 
-        virtual void unbind(Queue::shared_ptr queue, const string& routingKey, qpid::framing::FieldTable* args);
+        virtual void unbind(Queue::shared_ptr queue, const std::string& routingKey, qpid::framing::FieldTable* args);
 
-        virtual void route(Message::shared_ptr& msg, const string& routingKey, qpid::framing::FieldTable* args);
+        virtual void route(Message::shared_ptr& msg, const std::string& routingKey, qpid::framing::FieldTable* args);
 
         virtual ~DirectExchange();
     };
diff --git a/cpp/broker/inc/Exchange.h b/cpp/broker/inc/Exchange.h
index 5f5dc5ce71..4066f5ac20 100644
--- a/cpp/broker/inc/Exchange.h
+++ b/cpp/broker/inc/Exchange.h
@@ -25,12 +25,14 @@
 namespace qpid {
 namespace broker {
     class Exchange{
-    public:
-        virtual const string& getName() = 0;
+        const std::string name;
+      public:
+        explicit Exchange(const std::string& name) : name(name) {}
+        virtual ~Exchange(){}
+        std::string getName() { return name; }
         virtual void bind(Queue::shared_ptr queue, const string& routingKey, qpid::framing::FieldTable* args) = 0;
         virtual void unbind(Queue::shared_ptr queue, const string& routingKey, qpid::framing::FieldTable* args) = 0;
         virtual void route(Message::shared_ptr& msg, const string& routingKey, qpid::framing::FieldTable* args) = 0;
-        virtual ~Exchange(){}
     };
 }
 }
diff --git a/cpp/broker/inc/ExchangeRegistry.h b/cpp/broker/inc/ExchangeRegistry.h
index 0f0eaae0d0..a4a778482c 100644
--- a/cpp/broker/inc/ExchangeRegistry.h
+++ b/cpp/broker/inc/ExchangeRegistry.h
@@ -25,13 +25,15 @@
 namespace qpid {
 namespace broker {
     class ExchangeRegistry{
-        std::map<string, Exchange*> exchanges;
+        typedef std::map<string, Exchange*> ExchangeMap;
+        ExchangeMap exchanges;
         qpid::concurrent::Monitor* lock;
     public:
         ExchangeRegistry();
         void declare(Exchange* exchange);
         void destroy(const string& name);
         Exchange* get(const string& name);
+        Exchange* getDefault();
         inline qpid::concurrent::Monitor* getLock(){ return lock; }
         ~ExchangeRegistry();
     };
diff --git a/cpp/broker/inc/FanOutExchange.h b/cpp/broker/inc/FanOutExchange.h
index 9d0d32bbf8..1932e8429c 100644
--- a/cpp/broker/inc/FanOutExchange.h
+++ b/cpp/broker/inc/FanOutExchange.h
@@ -30,22 +30,19 @@ namespace qpid {
 namespace broker {
 
 class FanOutExchange : public virtual Exchange {
-    const string name;
     std::vector<Queue::shared_ptr> bindings;
     qpid::concurrent::MonitorImpl lock;
 
   public:
     static const std::string typeName;
         
-    FanOutExchange(const string& name);
+    FanOutExchange(const std::string& name);
         
-    inline virtual const string& getName(){ return name; }
+    virtual void bind(Queue::shared_ptr queue, const std::string& routingKey, qpid::framing::FieldTable* args);
 
-    virtual void bind(Queue::shared_ptr queue, const string& routingKey, qpid::framing::FieldTable* args);
+    virtual void unbind(Queue::shared_ptr queue, const std::string& routingKey, qpid::framing::FieldTable* args);
 
-    virtual void unbind(Queue::shared_ptr queue, const string& routingKey, qpid::framing::FieldTable* args);
-
-    virtual void route(Message::shared_ptr& msg, const string& routingKey, qpid::framing::FieldTable* args);
+    virtual void route(Message::shared_ptr& msg, const std::string& routingKey, qpid::framing::FieldTable* args);
 
     virtual ~FanOutExchange();
 };
diff --git a/cpp/broker/inc/TopicExchange.h b/cpp/broker/inc/TopicExchange.h
index d9ff62ecc6..68a4026ee7 100644
--- a/cpp/broker/inc/TopicExchange.h
+++ b/cpp/broker/inc/TopicExchange.h
@@ -18,7 +18,7 @@
 #ifndef _TopicExchange_
 #define _TopicExchange_
 
-#include <map>
+#include <tr1/unordered_map>
 #include <vector>
 #include "Exchange.h"
 #include "FieldTable.h"
@@ -28,28 +28,67 @@
 
 namespace qpid {
 namespace broker {
-    class TopicExchange : public virtual Exchange{
-        const string name;
-        std::map<string, std::vector<Queue::shared_ptr> > bindings;//NOTE: pattern matching not yet supported
-        qpid::concurrent::MonitorImpl lock;
 
-    public:
-        static const std::string typeName;
-        
-        TopicExchange(const string& name);
+/** A vector of string tokens */
+class Tokens : public std::vector<std::string> {
+  public:
+    Tokens() {};
+    // Default copy, assign, dtor are sufficient.
+
+    /** Tokenize s, provides automatic conversion of string to Tokens */
+    Tokens(const std::string& s) { operator=(s); }
+    /** Tokenize s */
+    Tokens & operator=(const std::string& s);
+
+    struct Hash { size_t operator()(const Tokens&) const; };
+    typedef std::equal_to<Tokens> Equal;
+};
+
+/**
+ * Tokens that have been normalized as a pattern and can be matched
+ * with topic Tokens.  Normalized meands all sequences of mixed * and
+ * # are reduced to a series of * followed by at most one #.
+ */
+class TopicPattern : public Tokens
+{
+  public:
+    TopicPattern() {}
+    // Default copy, assign, dtor are sufficient.
+    TopicPattern(const Tokens& tokens) { operator=(tokens); }
+    TopicPattern(const std::string& str) { operator=(str); }
+    TopicPattern& operator=(const Tokens&);
+    TopicPattern& operator=(const std::string& str) { operator=(Tokens(str)); }
+    
+    /** Match a topic */
+    bool match(const std::string& topic) { return match(Tokens(topic)); }
+    bool match(const Tokens& topic) const;
+
+  private:
+    void normalize();
+};
+
+class TopicExchange : public virtual Exchange{
+    typedef std::tr1::unordered_map<TopicPattern, Queue::vector, TopicPattern::Hash> BindingMap;
+    BindingMap bindings;
+    qpid::concurrent::MonitorImpl lock;
+
+  public:
+    static const std::string typeName;
+
+    TopicExchange(const string& name);
         
-        inline virtual const string& getName(){ return name; }
+    virtual void bind(Queue::shared_ptr queue, const string& routingKey, qpid::framing::FieldTable* args);
 
-        virtual void bind(Queue::shared_ptr queue, const string& routingKey, qpid::framing::FieldTable* args);
+    virtual void unbind(Queue::shared_ptr queue, const string& routingKey, qpid::framing::FieldTable* args);
+
+    virtual void route(Message::shared_ptr& msg, const string& routingKey, qpid::framing::FieldTable* args);
+
+    virtual ~TopicExchange();
+};
 
-        virtual void unbind(Queue::shared_ptr queue, const string& routingKey, qpid::framing::FieldTable* args);
 
-        virtual void route(Message::shared_ptr& msg, const string& routingKey, qpid::framing::FieldTable* args);
 
-        virtual ~TopicExchange();
-    };
 }
 }
 
-
 #endif
diff --git a/cpp/broker/src/DirectExchange.cpp b/cpp/broker/src/DirectExchange.cpp
index 70f7ee838f..ca29225bee 100644
--- a/cpp/broker/src/DirectExchange.cpp
+++ b/cpp/broker/src/DirectExchange.cpp
@@ -22,7 +22,7 @@
 using namespace qpid::broker;
 using namespace qpid::framing;
 
-DirectExchange::DirectExchange(const string& _name) : name(_name) {
+DirectExchange::DirectExchange(const string& name) : Exchange(name) {
 
 }
 
@@ -59,7 +59,7 @@ void DirectExchange::route(Message::shared_ptr& msg, const string& routingKey, F
         (*i)->deliver(msg);
     }
     if(!count){
-        std::cout << "WARNING: DirectExchange " << name << " could not route message with key " << routingKey << std::endl;
+        std::cout << "WARNING: DirectExchange " << getName() << " could not route message with key " << routingKey << std::endl;
     }
     lock.release();
 }
diff --git a/cpp/broker/src/ExchangeRegistry.cpp b/cpp/broker/src/ExchangeRegistry.cpp
index 0ee581af2f..05396382a7 100644
--- a/cpp/broker/src/ExchangeRegistry.cpp
+++ b/cpp/broker/src/ExchangeRegistry.cpp
@@ -24,6 +24,10 @@ using namespace qpid::concurrent;
 ExchangeRegistry::ExchangeRegistry() : lock(new MonitorImpl()){}
 
 ExchangeRegistry::~ExchangeRegistry(){
+    for (ExchangeMap::iterator i = exchanges.begin(); i != exchanges.end(); ++i)
+    {
+        delete i->second;
+    }
     delete lock;
 }
 
@@ -41,3 +45,13 @@ void ExchangeRegistry::destroy(const string& name){
 Exchange* ExchangeRegistry::get(const string& name){
     return exchanges[name];
 }
+
+namespace 
+{
+const std::string empty;
+}
+
+Exchange* ExchangeRegistry::getDefault()
+{
+    return get(empty);
+}
diff --git a/cpp/broker/src/FanOutExchange.cpp b/cpp/broker/src/FanOutExchange.cpp
index 7f261d5eda..4eb75cb920 100644
--- a/cpp/broker/src/FanOutExchange.cpp
+++ b/cpp/broker/src/FanOutExchange.cpp
@@ -23,7 +23,7 @@ using namespace qpid::broker;
 using namespace qpid::framing;
 using namespace qpid::concurrent;
 
-FanOutExchange::FanOutExchange(const string& _name) : name(_name) {}
+FanOutExchange::FanOutExchange(const std::string& name) : Exchange(name) {}
 
 void FanOutExchange::bind(Queue::shared_ptr queue, const string& routingKey, FieldTable* args){
     Locker locker(lock);
diff --git a/cpp/broker/src/SessionHandlerFactoryImpl.cpp b/cpp/broker/src/SessionHandlerFactoryImpl.cpp
index 661cb4ef81..280e89c475 100644
--- a/cpp/broker/src/SessionHandlerFactoryImpl.cpp
+++ b/cpp/broker/src/SessionHandlerFactoryImpl.cpp
@@ -22,10 +22,19 @@
 using namespace qpid::broker;
 using namespace qpid::io;
 
+namespace
+{
+const std::string empty;
+const std::string amq_direct("amq.direct");
+const std::string amq_topic("amq.topic");
+const std::string amq_fanout("amq.fanout");
+}
+
 SessionHandlerFactoryImpl::SessionHandlerFactoryImpl(u_int32_t _timeout) : timeout(_timeout), cleaner(&queues, timeout/10){
-    exchanges.declare(new DirectExchange("amq.direct"));
-    exchanges.declare(new TopicExchange("amq.topic"));
-    exchanges.declare(new FanOutExchange("amq.fanout"));
+    exchanges.declare(new DirectExchange(empty)); // Default exchange.
+    exchanges.declare(new DirectExchange(amq_direct));
+    exchanges.declare(new TopicExchange(amq_topic));
+    exchanges.declare(new FanOutExchange(amq_fanout));
     cleaner.start();
 }
 
@@ -35,6 +44,4 @@ SessionHandler* SessionHandlerFactoryImpl::create(SessionContext* ctxt){
 
 SessionHandlerFactoryImpl::~SessionHandlerFactoryImpl(){
     cleaner.stop();
-    exchanges.destroy("amq.direct");
-    exchanges.destroy("amq.topic");    
 }
diff --git a/cpp/broker/src/SessionHandlerImpl.cpp b/cpp/broker/src/SessionHandlerImpl.cpp
index a75b8fcf0f..872e6f124a 100644
--- a/cpp/broker/src/SessionHandlerImpl.cpp
+++ b/cpp/broker/src/SessionHandlerImpl.cpp
@@ -256,7 +256,7 @@ void SessionHandlerImpl::QueueHandlerImpl::declare(u_int16_t channel, u_int16_t
 	if (queue_created.second) { // This is a new queue
 	    parent->channels[channel]->setDefaultQueue(queue);
 	    //add default binding:
-	    parent->exchanges->get("amq.direct")->bind(queue, name, 0);
+	    parent->exchanges->getDefault()->bind(queue, name, 0);
 	    if(exclusive){
 		parent->exclusiveQueues.push_back(queue);
 	    } else if(autoDelete){
@@ -280,7 +280,7 @@ void SessionHandlerImpl::QueueHandlerImpl::bind(u_int16_t channel, u_int16_t tic
     Queue::shared_ptr queue = parent->getQueue(queueName, channel);
     Exchange* exchange = parent->exchanges->get(exchangeName);
     if(exchange){
-        if(routingKey.size() == 0 && queueName.size() == 0) routingKey = queue->getName();
+        if(routingKey.empty() && queueName.empty()) routingKey = queue->getName();
         exchange->bind(queue, routingKey, &arguments);
         if(!nowait) parent->client.getQueue().bindOk(channel);    
     }else{
@@ -361,7 +361,7 @@ void SessionHandlerImpl::BasicHandlerImpl::publish(u_int16_t channel, u_int16_t
                                                    string& exchange, string& routingKey, 
                                                    bool mandatory, bool immediate){
 
-    Message* msg = new Message(parent, exchange.length() ? exchange : "amq.direct", routingKey, mandatory, immediate);
+    Message* msg = new Message(parent, exchange, routingKey, mandatory, immediate);
     parent->channels[channel]->handlePublish(msg);
 } 
         
diff --git a/cpp/broker/src/TopicExchange.cpp b/cpp/broker/src/TopicExchange.cpp
index e0248958f9..287502bc88 100644
--- a/cpp/broker/src/TopicExchange.cpp
+++ b/cpp/broker/src/TopicExchange.cpp
@@ -17,46 +17,146 @@
  */
 #include "TopicExchange.h"
 #include "ExchangeBinding.h"
+#include <algorithm>
 
 using namespace qpid::broker;
 using namespace qpid::framing;
 
-TopicExchange::TopicExchange(const string& _name) : name(_name) {
 
+// TODO aconway 2006-09-20: More efficient matching algorithm.
+// Areas for improvement:
+// - excessive string copying: should be 0 copy, match from original buffer.
+// - match/lookup: use descision tree or other more efficient structure.
+
+Tokens& Tokens::operator=(const std::string& s) {
+    clear();
+    if (s.empty()) return *this;
+    std::string::const_iterator i = s.begin();
+    while (true) {
+        // Invariant: i is at the beginning of the next untokenized word.
+        std::string::const_iterator j = find(i, s.end(), '.');
+        push_back(std::string(i, j));
+        if (j == s.end()) return *this;
+        i = j + 1;
+    }
+    return *this;
+}
+    
+size_t Tokens::Hash::operator()(const Tokens& p) const {
+    size_t hash = 0;
+    for (Tokens::const_iterator i = p.begin(); i != p.end(); ++i) {
+        hash += std::tr1::hash<std::string>()(*i);
+    }
+}
+
+TopicPattern& TopicPattern::operator=(const Tokens& tokens) {
+    Tokens::operator=(tokens);
+    normalize();
+    return *this;
+}
+
+namespace {
+const std::string hashmark("#");
+const std::string star("*");
+}
+
+void TopicPattern::normalize() {
+    std::string word;
+    Tokens::iterator i = begin();
+    while (i != end()) {
+        if (*i == hashmark) {
+            ++i;
+            while (i != end()) {
+                // Invariant: *(i-1)==#, [begin()..i-1] is normalized.
+                if (*i == star) { // Move * before #.
+                    std::swap(*i, *(i-1));
+                    ++i;
+                } else if (*i == hashmark) {
+                    erase(i); // Remove extra #
+                } else {
+                    break;
+                }
+            }
+        } else {
+            i ++;
+        }
+    }
+}
+
+
+namespace {
+// TODO aconway 2006-09-20: Ineficient to convert every routingKey to a string.
+// Need more efficient Tokens impl that can operate on a string in place.
+// 
+bool do_match(Tokens::const_iterator pattern_begin,  Tokens::const_iterator pattern_end, Tokens::const_iterator target_begin,  Tokens::const_iterator target_end)
+{
+    // Invariant: [pattern_begin..p) matches [target_begin..t)
+    Tokens::const_iterator p = pattern_begin;
+    Tokens::const_iterator t = target_begin;
+    while (p != pattern_end && t != target_end)
+    {
+        if (*p == star || *p == *t) {
+            ++p, ++t;
+        } else if (*p == hashmark) {
+            ++p;
+            if (do_match(p, pattern_end, t, target_end)) return true;
+            while (t != target_end) {
+                ++t;
+                if (do_match(p, pattern_end, t, target_end)) return true;
+            }
+            return false;
+        } else {
+            return false;
+        }
+    }
+    while (p != pattern_end && *p == hashmark) ++p; // Ignore trailing #
+    return t == target_end && p == pattern_end;
+}
+}
+
+bool TopicPattern::match(const Tokens& target)  const
+{
+    return do_match(begin(), end(), target.begin(), target.end());
 }
 
+TopicExchange::TopicExchange(const string& name) : Exchange(name) { }
+
 void TopicExchange::bind(Queue::shared_ptr queue, const string& routingKey, FieldTable* args){
     lock.acquire();
-    bindings[routingKey].push_back(queue);
+    TopicPattern routingPattern(routingKey);
+    bindings[routingPattern].push_back(queue);
     queue->bound(new ExchangeBinding(this, queue, routingKey, args));
     lock.release();
 }
 
 void TopicExchange::unbind(Queue::shared_ptr queue, const string& routingKey, FieldTable* args){
     lock.acquire();
-    std::vector<Queue::shared_ptr>& queues(bindings[routingKey]);
-
-    std::vector<Queue::shared_ptr>::iterator i = find(queues.begin(), queues.end(), queue);
-    if(i < queues.end()){
-        queues.erase(i);
-        if(queues.empty()){
-            bindings.erase(routingKey);
-        }
-    }
+    BindingMap::iterator bi = bindings.find(TopicPattern(routingKey));
+    Queue::vector& qv(bi->second);
+    if (bi == bindings.end()) return;
+    Queue::vector::iterator q = find(qv.begin(), qv.end(), queue);
+    if(q == qv.end()) return;
+    qv.erase(q);
+    if(qv.empty()) bindings.erase(bi);
     lock.release();
 }
 
+
 void TopicExchange::route(Message::shared_ptr& msg, const string& routingKey, FieldTable* args){
     lock.acquire();
-    std::vector<Queue::shared_ptr>& queues(bindings[routingKey]);
-    for(std::vector<Queue::shared_ptr>::iterator i = queues.begin(); i != queues.end(); i++){
-        (*i)->deliver(msg);
+    for (BindingMap::iterator i = bindings.begin(); i != bindings.end(); ++i) {
+        if (i->first.match(routingKey)) {
+            Queue::vector& qv(i->second);
+            for(Queue::vector::iterator j = qv.begin(); j != qv.end(); j++){
+                (*j)->deliver(msg);
+            }
+        }
     }
     lock.release();
 }
 
-TopicExchange::~TopicExchange(){
-
-}
+TopicExchange::~TopicExchange() {}
 
 const std::string TopicExchange::typeName("topic");
+
+
diff --git a/cpp/broker/test/TopicExchangeTest.cpp b/cpp/broker/test/TopicExchangeTest.cpp
new file mode 100644
index 0000000000..4653540040
--- /dev/null
+++ b/cpp/broker/test/TopicExchangeTest.cpp
@@ -0,0 +1,186 @@
+#include "TopicExchange.h"
+#include <cppunit/TestCase.h>
+#include <cppunit/TextTestRunner.h>
+#include <cppunit/extensions/HelperMacros.h>
+#include <cppunit/plugin/TestPlugIn.h>
+
+using namespace qpid::broker;
+
+Tokens makeTokens(char** begin, char** end)
+{
+    Tokens t;
+    t.insert(t.end(), begin, end);
+    return t;
+}
+
+// Calculate size of an array. 
+#define LEN(a) (sizeof(a)/sizeof(a[0]))
+
+// Convert array to token vector
+#define TOKENS(a) makeTokens(a, a + LEN(a))
+
+// Allow CPPUNIT_EQUALS to print a Tokens.
+// TODO aconway 2006-09-19: Make it a template and put it in a shared test lib.
+//
+CppUnit::OStringStream& operator <<(CppUnit::OStringStream& out, const Tokens& v)
+{
+    out << "[ ";
+    for (Tokens::const_iterator i = v.begin();
+         i != v.end(); ++i)
+    {
+        out << '"' << *i << '"' << (i+1 == v.end() ? "]" : ", ");
+    }
+}
+
+
+class TokensTest : public CppUnit::TestCase
+{
+    CPPUNIT_TEST_SUITE(TokensTest);
+    CPPUNIT_TEST(testTokens);
+    CPPUNIT_TEST_SUITE_END();
+
+  public:
+    void testTokens() 
+    {
+        Tokens tokens("hello.world");
+        char* expect[] = {"hello", "world"};
+        CPPUNIT_ASSERT_EQUAL(TOKENS(expect), tokens);
+        
+        tokens = "a.b.c";
+        char* expect2[] = { "a", "b", "c" };
+        CPPUNIT_ASSERT_EQUAL(TOKENS(expect2), tokens);
+
+        tokens = "";
+        CPPUNIT_ASSERT(tokens.empty());
+
+        tokens = "x";
+        char* expect3[] = { "x" };
+        CPPUNIT_ASSERT_EQUAL(TOKENS(expect3), tokens);
+
+        tokens = (".x");
+        char* expect4[] = { "", "x" };
+        CPPUNIT_ASSERT_EQUAL(TOKENS(expect4), tokens);
+
+        tokens = ("x.");
+        char* expect5[] = { "x", "" };
+        CPPUNIT_ASSERT_EQUAL(TOKENS(expect5), tokens);
+
+        tokens = (".");
+        char* expect6[] = { "", "" };
+        CPPUNIT_ASSERT_EQUAL(TOKENS(expect6), tokens);        
+
+        tokens = ("..");
+        char* expect7[] = { "", "", "" };
+        CPPUNIT_ASSERT_EQUAL(TOKENS(expect7), tokens);        
+    }
+    
+};
+
+#define ASSERT_NORMALIZED(expect, pattern) \
+    CPPUNIT_ASSERT_EQUAL(Tokens(expect), static_cast<Tokens>(TopicPattern(pattern)))
+class TopicPatternTest : public CppUnit::TestCase 
+{
+    CPPUNIT_TEST_SUITE(TopicPatternTest);
+    CPPUNIT_TEST(testNormalize);
+    CPPUNIT_TEST(testPlain);
+    CPPUNIT_TEST(testStar);
+    CPPUNIT_TEST(testHash);
+    CPPUNIT_TEST(testMixed);
+    CPPUNIT_TEST(testCombo);
+    CPPUNIT_TEST_SUITE_END();
+
+  public:
+
+    void testNormalize() 
+    {
+        CPPUNIT_ASSERT(TopicPattern("").empty());
+        ASSERT_NORMALIZED("a.b.c", "a.b.c");
+        ASSERT_NORMALIZED("a.*.c", "a.*.c");
+        ASSERT_NORMALIZED("#", "#");
+        ASSERT_NORMALIZED("#", "#.#.#.#");
+        ASSERT_NORMALIZED("*.*.*.#", "#.*.#.*.#.#.*");
+        ASSERT_NORMALIZED("a.*.*.*.#", "a.*.#.*.#.*.#");
+        ASSERT_NORMALIZED("a.*.*.*.#", "a.*.#.*.#.*");
+    }
+    
+    void testPlain() {
+        TopicPattern p("ab.cd.e");
+        CPPUNIT_ASSERT(p.match("ab.cd.e"));
+        CPPUNIT_ASSERT(!p.match("abx.cd.e"));
+        CPPUNIT_ASSERT(!p.match("ab.cd"));
+        CPPUNIT_ASSERT(!p.match("ab.cd..e."));
+        CPPUNIT_ASSERT(!p.match("ab.cd.e."));
+        CPPUNIT_ASSERT(!p.match(".ab.cd.e"));
+
+        p = "";
+        CPPUNIT_ASSERT(p.match(""));
+
+        p = ".";
+        CPPUNIT_ASSERT(p.match("."));
+    }
+
+
+    void testStar() 
+    {
+        TopicPattern p("a.*.b");
+        CPPUNIT_ASSERT(p.match("a.xx.b"));
+        CPPUNIT_ASSERT(!p.match("a.b"));
+
+        p = "*.x";
+        CPPUNIT_ASSERT(p.match("y.x"));
+        CPPUNIT_ASSERT(p.match(".x"));
+        CPPUNIT_ASSERT(!p.match("x"));
+
+        p = "x.x.*";
+        CPPUNIT_ASSERT(p.match("x.x.y"));
+        CPPUNIT_ASSERT(p.match("x.x."));
+        CPPUNIT_ASSERT(!p.match("x.x"));
+        CPPUNIT_ASSERT(!p.match("q.x.y"));
+    }
+
+    void testHash() 
+    {
+        TopicPattern p("a.#.b");
+        CPPUNIT_ASSERT(p.match("a.b"));
+        CPPUNIT_ASSERT(p.match("a.x.b"));
+        CPPUNIT_ASSERT(p.match("a..x.y.zz.b"));
+        CPPUNIT_ASSERT(!p.match("a.b."));
+        CPPUNIT_ASSERT(!p.match("q.x.b"));
+
+        p = "a.#";
+        CPPUNIT_ASSERT(p.match("a"));
+        CPPUNIT_ASSERT(p.match("a.b"));
+        CPPUNIT_ASSERT(p.match("a.b.c"));
+
+        p = "#.a";
+        CPPUNIT_ASSERT(p.match("a"));
+        CPPUNIT_ASSERT(p.match("x.y.a"));
+    }
+
+    void testMixed() 
+    {
+        TopicPattern p("*.x.#.y");
+        CPPUNIT_ASSERT(p.match("a.x.y"));
+        CPPUNIT_ASSERT(p.match("a.x.p.qq.y"));
+        CPPUNIT_ASSERT(!p.match("a.a.x.y"));
+        CPPUNIT_ASSERT(!p.match("aa.x.b.c"));
+
+        p = "a.#.b.*";
+        CPPUNIT_ASSERT(p.match("a.b.x"));
+        CPPUNIT_ASSERT(p.match("a.x.x.x.b.x"));
+    }
+
+    void testCombo() {
+        TopicPattern p("*.#.#.*.*.#");
+        CPPUNIT_ASSERT(p.match("x.y.z"));
+        CPPUNIT_ASSERT(p.match("x.y.z.a.b.c"));
+        CPPUNIT_ASSERT(!p.match("x.y"));
+        CPPUNIT_ASSERT(!p.match("x"));
+    }
+};
+
+    
+// Make this test suite a plugin.
+CPPUNIT_PLUGIN_IMPLEMENT();
+CPPUNIT_TEST_SUITE_REGISTRATION(TopicPatternTest);
+CPPUNIT_TEST_SUITE_REGISTRATION(TokensTest);