1 files changed, 221 insertions, 0 deletions
diff --git a/src/topic_filter.cpp b/src/topic_filter.cpp
new file mode 100644
index 0000000..8fda252
--- /dev/null
+++ b/src/topic_filter.cpp
@@ -0,0 +1,221 @@
+/*
+    Copyright (c) 2012 250bpm s.r.o.
+    Copyright (c) 2012 Other contributors as noted in the AUTHORS file
+
+    This file is part of Crossroads I/O project.
+
+    Crossroads I/O is free software; you can redistribute it and/or modify it
+    under the terms of the GNU Lesser General Public License as published by
+    the Free Software Foundation; either version 3 of the License, or
+    (at your option) any later version.
+
+    Crossroads is distributed in the hope that it will be useful,
+    but WITHOUT ANY WARRANTY; without even the implied warranty of
+    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+    GNU Lesser General Public License for more details.
+
+    You should have received a copy of the GNU Lesser General Public License
+    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+*/
+
+#include <vector>
+#include <string>
+#include <map>
+
+#include "../include/xs/xs.h"
+
+#include "topic_filter.hpp"
+#include "err.hpp"
+
+typedef std::vector <void*> subscribers_t;
+typedef std::map <std::string, subscribers_t> topic_t;
+
+static bool topic_match (const char *topic_,
+    const unsigned char *data_, size_t size_)
+{
+    while (true) {
+
+        //  Check whether matching is done.
+        if (*topic_ == 0)
+            return true;
+
+        // Match one element.
+        if (topic_ [0] == '*') {
+
+            //  March wildcard element.
+            ++topic_;
+            while (size_ && *data_ != 0 && *data_ != '.')
+                ++data_, --size_;
+        }
+        else {
+
+            //  Match literal element.
+            while (true) {
+                if (topic_ [0] == '.' || topic_ [0] == 0)
+                    break;
+                if (!size_)
+                    return false;
+                if (topic_ [0] != *data_)
+                    return false; 
+                ++data_;
+                --size_;
+                ++topic_;
+            }
+        }
+
+        //  Check whether matching is done.
+        if (topic_ [0] == 0)
+            return true;
+
+        //  Match dot.
+        if (topic_ [0] != '.')
+            return false;  //  Malformed subscription, e.g. "*abc"
+        if (!size_)
+            return false;
+        if (*data_ != '.')
+            return false;
+        ++data_;
+        --size_;
+        ++topic_;
+    }
+}
+
+static int id (void *core_)
+{
+    return XS_FILTER_TOPIC;
+}
+
+static void *pf_create (void *core_)
+{
+    topic_t *pf = new (std::nothrow) topic_t;
+    alloc_assert (pf);
+    return (void*) pf;
+}
+
+static void pf_destroy (void *core_, void *pf_)
+{
+    xs_assert (pf_);
+    delete (topic_t*) pf_;
+}
+
+static int pf_subscribe (void *core_, void *pf_, void *subscriber_,
+    const unsigned char *data_, size_t size_)
+{
+    topic_t *self = (topic_t*) pf_;
+    (*self) [std::string ((char*) data_, size_)].push_back (subscriber_);
+    return xs_filter_subscribed (core_, data_, size_);
+}
+
+static int pf_unsubscribe (void *core_, void *pf_, void *subscriber_,
+    const unsigned char *data_, size_t size_)
+{
+    topic_t *self = (topic_t*) pf_;
+
+    topic_t::iterator it = self->find (std::string ((char*) data_, size_));
+    if (it == self->end ()) {
+        errno = EINVAL;
+        return -1;
+    }
+
+    bool found = false;
+    for (subscribers_t::iterator its = it->second.begin ();
+          its != it->second.end (); ++its) {
+        if (*its == subscriber_) {
+            it->second.erase (its);
+            found = true;
+            break;
+        }
+    }
+    if (!found) {
+        errno = EINVAL;
+        return -1;
+    }
+
+    if (it->second.empty ())
+        self->erase (it);
+
+    return 0;
+}
+
+static void pf_unsubscribe_all (void *core_, void *pf_, void *subscriber_)
+{
+    topic_t *self = (topic_t*) pf_;
+
+    for (topic_t::iterator it = self->begin (); it != self->end ();) {
+        for (subscribers_t::size_type i = 0; i < it->second.size (); ++i) {
+            if (it->second [i] == subscriber_)
+                it->second.erase (it->second.begin () + i--);
+        }
+        if (it->second.empty ())
+            self->erase (it++);
+        else
+            ++it;
+    }
+}
+
+static void pf_match (void *core_, void *pf_,
+    const unsigned char *data_, size_t size_)
+{
+    topic_t *self = (topic_t*) pf_;
+    for (topic_t::iterator it = self->begin (); it != self->end (); ++it) {
+        if (topic_match (it->first.c_str (), data_, size_)) {
+            for (subscribers_t::iterator its = it->second.begin ();
+                  its != it->second.end (); ++its) {
+                int rc = xs_filter_matching (core_, *its);
+                errno_assert (rc == 0);
+            }
+        }
+    }
+}
+
+static void *sf_create (void *core_)
+{
+    return pf_create (core_);
+}
+
+static void sf_destroy (void *core_, void *sf_)
+{
+    pf_destroy (core_, sf_);
+}
+
+static int sf_subscribe (void *core_, void *sf_,
+    const unsigned char *data_, size_t size_)
+{
+    return pf_subscribe (core_, sf_, NULL, data_, size_);
+}
+
+static int sf_unsubscribe (void *core_, void *sf_,
+    const unsigned char *data_, size_t size_)
+{
+    return pf_unsubscribe (core_, sf_, NULL, data_, size_);
+}
+
+static int sf_match (void *core_, void *sf_,
+    const unsigned char *data_, size_t size_)
+{
+    topic_t *self = (topic_t*) sf_;
+    for (topic_t::iterator it = self->begin (); it != self->end (); ++it)
+       if (topic_match (it->first.c_str (), data_, size_))
+           return 1;
+    return 0;
+}
+
+static xs_filter_t rgxp_filter = {
+    XS_PLUGIN_FILTER,
+    1,
+    id,
+    pf_create,
+    pf_destroy,
+    pf_subscribe,
+    pf_unsubscribe,
+    pf_unsubscribe_all,
+    pf_match,
+    sf_create,
+    sf_destroy,
+    sf_subscribe,
+    sf_unsubscribe,
+    sf_match
+};
+
+void *xs::topic_filter = (void*) &rgxp_filter;
+