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);