XS_PLUGIN and XS_FILTER implementation

This patch introduces following features:
- XS_PLUGIN context option to add plugins to libxs
- XS_FILTER option to switch between different filter types
- Automatic loading of plug-ins is *not* implemented.

From the implementation point of view:
- standard prefix filter is implemented as a pluggable filter
- trie_t and mtrie_t are joined into a single class
- the code for 0MQ/3.1 compatibility is left in in the form of comments
- new test for testing re-subscriptions is added
Signed-off-by: Martin Sustrik <sustrik@250bpm.com>

21 files changed, 1267 insertions, 1020 deletions

diff --git a/src/Makefile.am b/src/Makefile.am
index fdfa8d1..3c260be 100644
--- a/src/Makefile.am
+++ b/src/Makefile.am
@@ -23,6 +23,7 @@ libxs_la_SOURCES = \
     clock.hpp \
     command.hpp \
     config.hpp \
+    core.hpp \
     ctx.hpp \
     decoder.hpp \
     devpoll.hpp \
@@ -44,7 +45,6 @@ libxs_la_SOURCES = \
     likely.hpp \
     mailbox.hpp \
     msg.hpp \
-    mtrie.hpp \
     mutex.hpp \
     object.hpp \
     options.hpp \
@@ -56,6 +56,7 @@ libxs_la_SOURCES = \
     platform.hpp \
     poll.hpp \
     pair.hpp \
+    prefix_filter.hpp \
     pub.hpp \
     pull.hpp \
     push.hpp \
@@ -74,7 +75,6 @@ libxs_la_SOURCES = \
     tcp_connecter.hpp \
     tcp_listener.hpp \
     thread.hpp \
-    trie.hpp \
     upoll.hpp \
     windows.hpp \
     wire.hpp \
@@ -85,6 +85,7 @@ libxs_la_SOURCES = \
     ypipe.hpp \
     yqueue.hpp \
     clock.cpp \
+    core.cpp \
     ctx.cpp \
     decoder.cpp \
     devpoll.cpp \
@@ -103,7 +104,6 @@ libxs_la_SOURCES = \
     lb.cpp \
     mailbox.cpp \
     msg.cpp \
-    mtrie.cpp \
     object.cpp \
     options.cpp \
     own.cpp \
@@ -113,6 +113,7 @@ libxs_la_SOURCES = \
     pgm_socket.cpp \
     pipe.cpp \
     poll.cpp \
+    prefix_filter.cpp \
     pull.cpp \
     push.cpp \
     reaper.cpp \
@@ -130,7 +131,6 @@ libxs_la_SOURCES = \
     tcp_connecter.cpp \
     tcp_listener.cpp \
     thread.cpp \
-    trie.cpp \
     upoll.cpp \
     xpub.cpp \
     xrep.cpp \
diff --git a/src/core.cpp b/src/core.cpp
new file mode 100644
index 0000000..2d16167
--- /dev/null
+++ b/src/core.cpp
@@ -0,0 +1,68 @@
+/*
+    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 "../include/xs.h"
+
+#include "core.hpp"
+#include "err.hpp"
+
+xs::core_t::core_t ()
+{
+}
+
+xs::core_t::~core_t ()
+{
+}
+
+int xs::core_t::filter_subscribed (const unsigned char *data_, size_t size_)
+{
+    errno = ENOTSUP;
+    return -1;
+}
+
+int xs::core_t::filter_unsubscribed (const unsigned char *data_, size_t size_)
+{
+    errno = ENOTSUP;
+    return -1;
+}
+
+int xs::core_t::filter_matching (void *subscriber_)
+{
+    errno = ENOTSUP;
+    return -1;
+}
+
+int xs_filter_subscribed (void *core_,
+    const unsigned char *data_, size_t size_)
+{
+    return ((xs::core_t*) core_)->filter_subscribed (data_, size_);
+}
+
+int xs_filter_unsubscribed (void *core_,
+    const unsigned char *data_, size_t size_)
+{
+    return ((xs::core_t*) core_)->filter_unsubscribed (data_, size_);
+}
+
+int xs_filter_matching (void *core_, void *subscriber_)
+{
+    return ((xs::core_t*) core_)->filter_matching (subscriber_);
+}
+
diff --git a/src/core.hpp b/src/core.hpp
new file mode 100644
index 0000000..fb50646
--- /dev/null
+++ b/src/core.hpp
@@ -0,0 +1,53 @@
+/*
+    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/>.
+*/
+
+#ifndef __XS_CORE_HPP_INCLUDED__
+#define __XS_CORE_HPP_INCLUDED__
+
+#include <stddef.h>
+
+namespace xs
+{
+
+    //  This class is not a core of Crossroads. It's rather a callback interface
+    //  for extensions, ie. what's extensions see as Crossroads core.
+
+    class core_t
+    {
+    public:
+
+        core_t ();
+        virtual ~core_t ();
+
+        virtual int filter_subscribed (const unsigned char *data_,
+            size_t size_);
+        virtual int filter_unsubscribed (const unsigned char *data_,
+            size_t size_);
+        virtual int filter_matching (void *subscriber_);
+
+    private:
+
+        core_t (const core_t&);
+        const core_t &operator = (const core_t&);
+    };
+
+}
+
+#endif
diff --git a/src/ctx.cpp b/src/ctx.cpp
index 21dec58..46fa984 100644
--- a/src/ctx.cpp
+++ b/src/ctx.cpp
@@ -35,6 +35,7 @@
 #include "pipe.hpp"
 #include "err.hpp"
 #include "msg.hpp"
+#include "prefix_filter.hpp"
 
 xs::ctx_t::ctx_t () :
     tag (0xbadcafe0),
@@ -46,6 +47,9 @@ xs::ctx_t::ctx_t () :
     max_sockets (512),
     io_thread_count (1)
 {
+    //  Plug in the standard plugins.
+    int rc = plug (prefix_filter);
+    xs_assert (rc == 0);
 }
 
 bool xs::ctx_t::check_tag ()
@@ -124,6 +128,28 @@ int xs::ctx_t::terminate ()
     return 0;
 }
 
+int xs::ctx_t::plug (const void *ext_)
+{
+    if (!ext_) {
+        errno = EFAULT;
+        return -1;
+    }
+
+    //  The extension is a message filter plug-in.
+    xs_filter_t *filter = (xs_filter_t*) ext_;
+    if (filter->type == XS_PLUGIN_FILTER && filter->version == 1) {
+       opt_sync.lock ();
+       filters [filter->id (NULL)] = filter;
+       opt_sync.unlock ();
+       return 0;
+    }
+
+    //  Specified extension type is not supported by this version of
+    //  the library.
+    errno = ENOTSUP;
+    return -1;
+}
+
 int xs::ctx_t::setctxopt (int option_, const void *optval_, size_t optvallen_)
 {
     switch (option_) {
@@ -145,6 +171,8 @@ int xs::ctx_t::setctxopt (int option_, const void *optval_, size_t optvallen_)
         io_thread_count = *((int*) optval_);
         opt_sync.unlock ();
         break;
+    case XS_PLUGIN:
+        return plug (optval_);
     default:
         errno = EINVAL;
         return -1;
@@ -256,6 +284,17 @@ xs::object_t *xs::ctx_t::get_reaper ()
     return reaper;
 }
 
+xs_filter_t *xs::ctx_t::get_filter (int filter_id_)
+{
+    xs_filter_t *result = NULL;
+    opt_sync.lock ();
+    filters_t::iterator it = filters.find (filter_id_);
+    if (it != filters.end ())
+        result = it->second;
+    opt_sync.unlock ();
+    return result;
+}
+
 void xs::ctx_t::send_command (uint32_t tid_, const command_t &command_)
 {
     slots [tid_]->send (command_);
diff --git a/src/ctx.hpp b/src/ctx.hpp
index c4fa96a..54144ad 100644
--- a/src/ctx.hpp
+++ b/src/ctx.hpp
@@ -25,7 +25,8 @@
 #include <map>
 #include <vector>
 #include <string>
-#include <stdarg.h>
+
+#include "../include/xs.h"
 
 #include "mailbox.hpp"
 #include "array.hpp"
@@ -88,6 +89,10 @@ namespace xs
         //  Returns reaper thread object.
         xs::object_t *get_reaper ();
 
+        //  Get the filter associated with the specified filter ID or NULL
+        //  If such filter is not registered.
+        xs_filter_t *get_filter (int filter_id_);
+
         //  Management of inproc endpoints.
         int register_endpoint (const char *addr_, endpoint_t &endpoint_);
         void unregister_endpoints (xs::socket_base_t *socket_);
@@ -102,6 +107,9 @@ namespace xs
 
     private:
 
+        //  Plug in the extension specified.
+        int plug (const void *ext_);
+
         //  Used to check whether the object is a context.
         uint32_t tag;
 
@@ -161,6 +169,10 @@ namespace xs
         //  Synchronisation of access to context options.
         mutex_t opt_sync;
 
+        //  List of all filters plugged into the context.
+        typedef std::map <int, xs_filter_t*> filters_t;
+        filters_t filters;
+
         ctx_t (const ctx_t&);
         const ctx_t &operator = (const ctx_t&);
     };
diff --git a/src/mtrie.cpp b/src/mtrie.cpp
deleted file mode 100644
index eae34c2..0000000
--- a/src/mtrie.cpp
+++ /dev/null
@@ -1,436 +0,0 @@
-/*
-    Copyright (c) 2011-2012 250bpm s.r.o.
-    Copyright (c) 2011-2012 Spotify AB
-    Copyright (c) 2011 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 <stdlib.h>
-
-#include <new>
-#include <algorithm>
-
-#include "platform.hpp"
-#if defined XS_HAVE_WINDOWS
-#include "windows.hpp"
-#endif
-
-#include "err.hpp"
-#include "pipe.hpp"
-#include "mtrie.hpp"
-
-xs::mtrie_t::mtrie_t () :
-    pipes (0),
-    min (0),
-    count (0),
-    live_nodes (0)
-{
-}
-
-xs::mtrie_t::~mtrie_t ()
-{
-    if (pipes) {
-        delete pipes;
-        pipes = 0;
-    }
-
-    if (count == 1) {
-        xs_assert (next.node);
-        delete next.node;
-        next.node = 0;
-    }
-    else if (count > 1) {
-        for (unsigned short i = 0; i != count; ++i)
-            if (next.table [i])
-                delete next.table [i];
-        free (next.table);
-    }
-}
-
-bool xs::mtrie_t::add (unsigned char *prefix_, size_t size_, pipe_t *pipe_)
-{
-    return add_helper (prefix_, size_, pipe_);
-}
-
-bool xs::mtrie_t::add_helper (unsigned char *prefix_, size_t size_,
-    pipe_t *pipe_)
-{
-    //  We are at the node corresponding to the prefix. We are done.
-    if (!size_) {
-        bool result = !pipes;
-        if (!pipes)
-            pipes = new pipes_t;
-        pipes->insert (pipe_);
-        return result;
-    }
-
-    unsigned char c = *prefix_;
-    if (c < min || c >= min + count) {
-
-        //  The character is out of range of currently handled
-        //  charcters. We have to extend the table.
-        if (!count) {
-            min = c;
-            count = 1;
-            next.node = NULL;
-        }
-        else if (count == 1) {
-            unsigned char oldc = min;
-            mtrie_t *oldp = next.node;
-            count = (min < c ? c - min : min - c) + 1;
-            next.table = (mtrie_t**)
-                malloc (sizeof (mtrie_t*) * count);
-            xs_assert (next.table);
-            for (unsigned short i = 0; i != count; ++i)
-                next.table [i] = 0;
-            min = std::min (min, c);
-            next.table [oldc - min] = oldp;
-        }
-        else if (min < c) {
-
-            //  The new character is above the current character range.
-            unsigned short old_count = count;
-            count = c - min + 1;
-            next.table = (mtrie_t**) realloc ((void*) next.table,
-                sizeof (mtrie_t*) * count);
-            xs_assert (next.table);
-            for (unsigned short i = old_count; i != count; i++)
-                next.table [i] = NULL;
-        }
-        else {
-
-            //  The new character is below the current character range.
-            unsigned short old_count = count;
-            count = (min + old_count) - c;
-            next.table = (mtrie_t**) realloc ((void*) next.table,
-                sizeof (mtrie_t*) * count);
-            xs_assert (next.table);
-            memmove (next.table + min - c, next.table,
-                old_count * sizeof (mtrie_t*));
-            for (unsigned short i = 0; i != min - c; i++)
-                next.table [i] = NULL;
-            min = c;
-        }
-    }
-
-    //  If next node does not exist, create one.
-    if (count == 1) {
-        if (!next.node) {
-            next.node = new (std::nothrow) mtrie_t;
-            ++live_nodes;
-            xs_assert (next.node);
-        }
-        return next.node->add_helper (prefix_ + 1, size_ - 1, pipe_);
-    }
-    else {
-        if (!next.table [c - min]) {
-            next.table [c - min] = new (std::nothrow) mtrie_t;
-            ++live_nodes;
-            xs_assert (next.table [c - min]);
-        }
-        return next.table [c - min]->add_helper (prefix_ + 1, size_ - 1, pipe_);
-    }
-}
-
-
-void xs::mtrie_t::rm (pipe_t *pipe_,
-    void (*func_) (unsigned char *data_, size_t size_, void *arg_),
-    void *arg_)
-{
-    unsigned char *buff = NULL;
-    rm_helper (pipe_, &buff, 0, 0, func_, arg_);
-    free (buff);
-}
-
-void xs::mtrie_t::rm_helper (pipe_t *pipe_, unsigned char **buff_,
-    size_t buffsize_, size_t maxbuffsize_,
-    void (*func_) (unsigned char *data_, size_t size_, void *arg_),
-    void *arg_)
-{
-    //  Remove the subscription from this node.
-    if (pipes && pipes->erase (pipe_) && pipes->empty ()) {
-        func_ (*buff_, buffsize_, arg_);
-        delete pipes;
-        pipes = 0;
-    }
-
-    //  Adjust the buffer.
-    if (buffsize_ >= maxbuffsize_) {
-        maxbuffsize_ = buffsize_ + 256;
-        *buff_ = (unsigned char*) realloc (*buff_, maxbuffsize_);
-        alloc_assert (*buff_);
-    }
-
-    //  If there are no subnodes in the trie, return.
-    if (count == 0)
-        return;
-
-    //  If there's one subnode (optimisation).
-    if (count == 1) {
-        (*buff_) [buffsize_] = min;
-        buffsize_++;
-        next.node->rm_helper (pipe_, buff_, buffsize_, maxbuffsize_,
-            func_, arg_);
-
-        //  Prune the node if it was made redundant by the removal
-        if (next.node->is_redundant ()) {
-            delete next.node;
-            next.node = 0;
-            count = 0;
-            --live_nodes;
-            xs_assert (live_nodes == 0);
-        }
-        return;
-    }
-
-    //  If there are multiple subnodes.
-    //
-    //  New min non-null character in the node table after the removal
-    unsigned char new_min = min + count - 1;
-    //  New max non-null character in the node table after the removal
-    unsigned char new_max = min;
-    for (unsigned short c = 0; c != count; c++) {
-        (*buff_) [buffsize_] = min + c;
-        if (next.table [c]) {
-            next.table [c]->rm_helper (pipe_, buff_, buffsize_ + 1,
-                maxbuffsize_, func_, arg_);
-
-            //  Prune redundant nodes from the mtrie
-            if (next.table [c]->is_redundant ()) {
-                delete next.table [c];
-                next.table [c] = 0;
-
-                xs_assert (live_nodes > 0);
-                --live_nodes;
-            }
-            else {
-                //  The node is not redundant, so it's a candidate for being
-                //  the new min/max node.
-                //
-                //  We loop through the node array from left to right, so the
-                //  first non-null, non-redundant node encountered is the new
-                //  minimum index. Conversely, the last non-redundant, non-null
-                //  node encountered is the new maximum index.
-                if (c + min < new_min)
-                    new_min = c + min;
-                if (c + min > new_max)
-                    new_max = c + min;
-            }
-        }
-    }
-
-    xs_assert (count > 1);
-
-    //  Compact the node table if possible
-    if (live_nodes == 1) {
-        //  If there's only one live node in the table we can
-        //  switch to using the more compact single-node
-        //  representation
-        xs_assert (new_min == new_max);
-        xs_assert (new_min >= min && new_min < min + count);
-        mtrie_t *node = next.table [new_min - min];
-        xs_assert (node);
-        free (next.table);
-        next.node = node;
-        count = 1;
-        min = new_min;
-    }
-    else if (live_nodes > 1 && (new_min > min || new_max < min + count - 1)) {
-        xs_assert (new_max - new_min + 1 > 1);
-
-        mtrie_t **old_table = next.table;
-        xs_assert (new_min > min || new_max < min + count - 1);
-        xs_assert (new_min >= min);
-        xs_assert (new_max <= min + count - 1);
-        xs_assert (new_max - new_min + 1 < count);
-
-        count = new_max - new_min + 1;
-        next.table = (mtrie_t**) malloc (sizeof (mtrie_t*) * count);
-        xs_assert (next.table);
-
-        memmove (next.table, old_table + (new_min - min),
-                 sizeof (mtrie_t*) * count);
-        free (old_table);
-
-        min = new_min;
-    }
-}
-
-bool xs::mtrie_t::rm (unsigned char *prefix_, size_t size_, pipe_t *pipe_)
-{
-    return rm_helper (prefix_, size_, pipe_);
-}
-
-bool xs::mtrie_t::rm_helper (unsigned char *prefix_, size_t size_,
-    pipe_t *pipe_)
-{
-    if (!size_) {
-        if (pipes) {
-            pipes_t::size_type erased = pipes->erase (pipe_);
-            xs_assert (erased == 1);
-            if (pipes->empty ()) {
-                delete pipes;
-                pipes = 0;
-            }
-        }
-        return !pipes;
-    }
-
-    unsigned char c = *prefix_;
-    if (!count || c < min || c >= min + count)
-        return false;
-
-    mtrie_t *next_node =
-        count == 1 ? next.node : next.table [c - min];
-
-    if (!next_node)
-        return false;
-
-    bool ret = next_node->rm_helper (prefix_ + 1, size_ - 1, pipe_);
-
-    if (next_node->is_redundant ()) {
-        delete next_node;
-        xs_assert (count > 0);
-
-        if (count == 1) {
-            next.node = 0;
-            count = 0;
-            --live_nodes;
-            xs_assert (live_nodes == 0);
-        }
-        else {
-            next.table [c - min] = 0;
-            xs_assert (live_nodes > 1);
-            --live_nodes;
-
-            //  Compact the table if possible
-            if (live_nodes == 1) {
-                //  If there's only one live node in the table we can
-                //  switch to using the more compact single-node
-                //  representation
-                mtrie_t *node = 0;
-                for (unsigned short i = 0; i < count; ++i) {
-                    if (next.table [i]) {
-                        node = next.table [i];
-                        min = i + min;
-                        break;
-                    }
-                }
-
-                xs_assert (node);
-                free (next.table);
-                next.node = node;
-                count = 1;
-            }
-            else if (c == min) {
-                //  We can compact the table "from the left"
-                unsigned char new_min = min;
-                for (unsigned short i = 1; i < count; ++i) {
-                    if (next.table [i]) {
-                        new_min = i + min;
-                        break;
-                    }
-                }
-                xs_assert (new_min != min);
-
-                mtrie_t **old_table = next.table;
-                xs_assert (new_min > min);
-                xs_assert (count > new_min - min);
-
-                count = count - (new_min - min);
-                next.table = (mtrie_t**) malloc (sizeof (mtrie_t*) * count);
-                xs_assert (next.table);
-
-                memmove (next.table, old_table + (new_min - min),
-                         sizeof (mtrie_t*) * count);
-                free (old_table);
-
-                min = new_min;
-            }
-            else if (c == min + count - 1) {
-                //  We can compact the table "from the right"
-                unsigned short new_count = count;
-                for (unsigned short i = 1; i < count; ++i) {
-                    if (next.table [count - 1 - i]) {
-                        new_count = count - i;
-                        break;
-                    }
-                }
-                xs_assert (new_count != count);
-                count = new_count;
-
-                mtrie_t **old_table = next.table;
-                next.table = (mtrie_t**) malloc (sizeof (mtrie_t*) * count);
-                xs_assert (next.table);
-
-                memmove (next.table, old_table, sizeof (mtrie_t*) * count);
-                free (old_table);
-            }
-        }
-    }
-
-    return ret;
-}
-
-void xs::mtrie_t::match (unsigned char *data_, size_t size_,
-    void (*func_) (pipe_t *pipe_, void *arg_), void *arg_)
-{
-    mtrie_t *current = this;
-    while (true) {
-
-        //  Signal the pipes attached to this node.
-        if (current->pipes) {
-            for (pipes_t::iterator it = current->pipes->begin ();
-                  it != current->pipes->end (); ++it)
-                func_ (*it, arg_);
-        }
-
-        //  If we are at the end of the message, there's nothing more to match.
-        if (!size_)
-            break;
-
-        //  If there are no subnodes in the trie, return.
-        if (current->count == 0)
-            break;
-
-        //  If there's one subnode (optimisation).
-		if (current->count == 1) {
-            if (data_ [0] != current->min)
-                break;
-            current = current->next.node;
-            data_++;
-            size_--;
-		    continue;
-		}
-
-		//  If there are multiple subnodes.
-        if (data_ [0] < current->min || data_ [0] >=
-              current->min + current->count)
-            break;
-        if (!current->next.table [data_ [0] - current->min])
-            break;
-        current = current->next.table [data_ [0] - current->min];
-        data_++;
-        size_--;
-    }
-}
-
-bool xs::mtrie_t::is_redundant () const
-{
-    return !pipes && live_nodes == 0;
-}
-
diff --git a/src/mtrie.hpp b/src/mtrie.hpp
deleted file mode 100644
index 1e56b4e..0000000
--- a/src/mtrie.hpp
+++ /dev/null
@@ -1,93 +0,0 @@
-/*
-    Copyright (c) 2011-2012 250bpm s.r.o.
-    Copyright (c) 2011-2012 Spotify AB
-    Copyright (c) 2011 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/>.
-*/
-
-#ifndef __XS_MTRIE_HPP_INCLUDED__
-#define __XS_MTRIE_HPP_INCLUDED__
-
-#include <stddef.h>
-#include <set>
-
-#include "stdint.hpp"
-
-namespace xs
-{
-
-    class pipe_t;
-
-    //  Multi-trie. Each node in the trie is a set of pointers to pipes.
-
-    class mtrie_t
-    {
-    public:
-
-        mtrie_t ();
-        ~mtrie_t ();
-
-        //  Add key to the trie. Returns true if it's a new subscription
-        //  rather than a duplicate.
-        bool add (unsigned char *prefix_, size_t size_, xs::pipe_t *pipe_);
-
-        //  Remove all subscriptions for a specific peer from the trie.
-        //  If there are no subscriptions left on some topics, invoke the
-        //  supplied callback function.
-        void rm (xs::pipe_t *pipe_,
-            void (*func_) (unsigned char *data_, size_t size_, void *arg_),
-            void *arg_);
-
-        //  Remove specific subscription from the trie. Return true is it was
-        //  actually removed rather than de-duplicated.
-        bool rm (unsigned char *prefix_, size_t size_, xs::pipe_t *pipe_);
-
-        //  Signal all the matching pipes.
-        void match (unsigned char *data_, size_t size_,
-            void (*func_) (xs::pipe_t *pipe_, void *arg_), void *arg_);
-
-    private:
-
-        bool add_helper (unsigned char *prefix_, size_t size_,
-            xs::pipe_t *pipe_);
-        void rm_helper (xs::pipe_t *pipe_, unsigned char **buff_,
-            size_t buffsize_, size_t maxbuffsize_,
-            void (*func_) (unsigned char *data_, size_t size_, void *arg_),
-            void *arg_);
-        bool rm_helper (unsigned char *prefix_, size_t size_,
-            xs::pipe_t *pipe_);
-        bool is_redundant () const;
-
-        typedef std::set <xs::pipe_t*> pipes_t;
-        pipes_t *pipes;
-
-        unsigned char min;
-        unsigned short count;
-        unsigned short live_nodes;
-        union {
-            class mtrie_t *node;
-            class mtrie_t **table;
-        } next;
-
-        mtrie_t (const mtrie_t&);
-        const mtrie_t &operator = (const mtrie_t&);
-    };
-
-}
-
-#endif
-
diff --git a/src/object.cpp b/src/object.cpp
index 5c1ed84..4f04fe4 100644
--- a/src/object.cpp
+++ b/src/object.cpp
@@ -154,6 +154,11 @@ xs::io_thread_t *xs::object_t::choose_io_thread (uint64_t affinity_)
     return ctx->choose_io_thread (affinity_);
 }
 
+xs_filter_t *xs::object_t::get_filter (int filter_id_)
+{
+    return ctx->get_filter (filter_id_);
+}
+
 void xs::object_t::send_stop ()
 {
     //  'stop' command goes always from administrative thread to
diff --git a/src/object.hpp b/src/object.hpp
index 5b855a5..fabb156 100644
--- a/src/object.hpp
+++ b/src/object.hpp
@@ -22,6 +22,8 @@
 #ifndef __XS_OBJECT_HPP_INCLUDED__
 #define __XS_OBJECT_HPP_INCLUDED__
 
+#include "../include/xs.h"
+
 #include "stdint.hpp"
 
 namespace xs
@@ -64,6 +66,9 @@ namespace xs
         //  Chooses least loaded I/O thread.
         xs::io_thread_t *choose_io_thread (uint64_t affinity_);
 
+        //  Functions related to extensions.
+        xs_filter_t *get_filter (int filter_id_);
+
         //  Derived object can use these functions to send commands
         //  to other objects.
         void send_stop ();
diff --git a/src/options.cpp b/src/options.cpp
index c9cbaae..fdbffde 100644
--- a/src/options.cpp
+++ b/src/options.cpp
@@ -23,6 +23,8 @@
 #include <string.h>
 #include <limits>
 
+#include "../include/xs.h"
+
 #include "options.hpp"
 #include "err.hpp"
 
@@ -47,6 +49,7 @@ xs::options_t::options_t () :
     ipv4only (1),
     keepalive (0),
     protocol (0),
+    filter_id (XS_FILTER_PREFIX),
     delay_on_close (true),
     delay_on_disconnect (true),
     filter (false),
@@ -248,6 +251,14 @@ int xs::options_t::setsockopt (int option_, const void *optval_,
             return 0;
         }
 
+    case XS_FILTER:
+        if (optvallen_ != sizeof (int)) {
+            errno = EINVAL;
+            return -1;
+        }
+        filter_id = *((int*) optval_);
+        return 0;
+
     }
 
     errno = EINVAL;
@@ -438,6 +449,15 @@ int xs::options_t::getsockopt (int option_, void *optval_, size_t *optvallen_)
         *optvallen_ = sizeof (int);
         return 0;
 
+    case XS_FILTER:
+        if (*optvallen_ < sizeof (int)) {
+            errno = EINVAL;
+            return -1;
+        }
+        *((int*) optval_) = filter_id;
+        *optvallen_ = sizeof (int);
+        return 0;
+
     }
 
     errno = EINVAL;
diff --git a/src/options.hpp b/src/options.hpp
index c1e4dda..1288f72 100644
--- a/src/options.hpp
+++ b/src/options.hpp
@@ -95,6 +95,9 @@ names