clean up - session/socket/engine stuff removed

73 files changed, 109 insertions, 4856 deletions

diff --git a/src/Makefile.am b/src/Makefile.am
index 27f4412..bde9c39 100644
--- a/src/Makefile.am
+++ b/src/Makefile.am
@@ -7,53 +7,30 @@ libzmq_la_SOURCES = \
     atomic_ptr.hpp \
     command.hpp \
     config.hpp \
-    connecter.hpp \
     context.hpp \
-    data_distributor.hpp \
     decoder.hpp \
     devpoll.hpp \
-    dummy_aggregator.hpp \
-    dummy_distributor.hpp \
     encoder.hpp \
     epoll.hpp \
     err.hpp \
-    fair_aggregator.hpp \
     fd.hpp \
     fd_signaler.hpp \
-    io_object.hpp \
     io_thread.hpp \
     ip.hpp \
     i_api.hpp \
-    i_demux.hpp \
-    i_mux.hpp \
     i_poller.hpp \
     i_poll_events.hpp \
-    i_session.hpp \
     i_signaler.hpp \
-    i_engine.hpp \
-    i_thread.hpp \
-    listener.hpp \
+    i_socket.hpp \
     kqueue.hpp \
-    load_balancer.hpp \
     msg.hpp \
     mutex.hpp \
     object.hpp \
-    p2p.hpp \
     pipe.hpp \
-    pipe_reader.hpp \
-    pipe_writer.hpp \
     platform.hpp \
     poll.hpp \
-    pub.hpp \
-    rep.hpp \
-    req.hpp \
-    safe_object.hpp \
     select.hpp \
-    session.hpp \
-    session_stub.hpp \
     simple_semaphore.hpp \
-    socket_base.hpp \
-    sub.hpp \
     stdint.hpp \
     tcp_connecter.hpp \
     tcp_listener.hpp \
@@ -65,50 +42,24 @@ libzmq_la_SOURCES = \
     ypipe.hpp \
     ypollset.hpp \
     yqueue.hpp \
-    zmq_decoder.hpp \
-    zmq_encoder.hpp \
-    zmq_tcp_engine.hpp \
     app_thread.cpp \
-    connecter.cpp \
     context.cpp \
-    data_distributor.cpp \
     devpoll.hpp \
-    dummy_aggregator.cpp \
-    dummy_distributor.cpp \
     epoll.cpp \
     err.cpp \
-    fair_aggregator.cpp \
     fd_signaler.cpp \
-    io_object.cpp \
     io_thread.cpp \
     ip.cpp \
     kqueue.cpp \
-    listener.cpp \
-    load_balancer.cpp \
     object.cpp \
-    p2p.cpp \
-    pipe.cpp \
-    pipe_reader.cpp \
-    pipe_writer.cpp \
     poll.cpp \
-    pub.cpp \
-    rep.cpp \
-    req.cpp \
-    safe_object.cpp \
     select.cpp \
-    session.cpp \
-    session_stub.cpp \
-    socket_base.cpp \
-    sub.cpp \
     tcp_connecter.cpp \
     tcp_listener.cpp \
     tcp_socket.cpp \
     thread.cpp \
     uuid.cpp \
     ypollset.cpp \
-    zmq_decoder.cpp \
-    zmq_encoder.cpp \
-    zmq_tcp_engine.cpp \
     zmq.cpp
 
 libzmq_la_LDFLAGS = -version-info 0:0:0
diff --git a/src/app_thread.cpp b/src/app_thread.cpp
index 9cc61c7..23a055a 100644
--- a/src/app_thread.cpp
+++ b/src/app_thread.cpp
@@ -28,20 +28,8 @@
 #include "app_thread.hpp"
 #include "context.hpp"
 #include "err.hpp"
-#include "session.hpp"
 #include "pipe.hpp"
 #include "config.hpp"
-#include "i_api.hpp"
-#include "dummy_aggregator.hpp"
-#include "fair_aggregator.hpp"
-#include "dummy_distributor.hpp"
-#include "data_distributor.hpp"
-#include "load_balancer.hpp"
-#include "p2p.hpp"
-#include "pub.hpp"
-#include "sub.hpp"
-#include "req.hpp"
-#include "rep.hpp"
 
 //  If the RDTSC is available we use it to prevent excessive
 //  polling for commands. The nice thing here is that it will work on any
@@ -58,37 +46,15 @@ zmq::app_thread_t::app_thread_t (context_t *context_, int thread_slot_) :
 {
 }
 
-void zmq::app_thread_t::shutdown ()
-{
-    //  Deallocate all the sessions associated with the thread.
-    while (!sessions.empty ())
-        sessions [0]->shutdown ();
-
-    delete this;
-}
-
 zmq::app_thread_t::~app_thread_t ()
 {
-}
+    //  Ask all the sockets to start termination, then wait till it is complete.
+    for (sockets_t::iterator it = sockets.begin (); it != sockets.end (); it ++)
+        (*it)->stop ();
+    for (sockets_t::iterator it = sockets.begin (); it != sockets.end (); it ++)
+        delete *it;
 
-void zmq::app_thread_t::attach_session (session_t *session_)
-{
-    session_->set_index (sessions.size ());
-    sessions.push_back (session_); 
-}
-
-void zmq::app_thread_t::detach_session (session_t *session_)
-{
-    //  O(1) removal of the session from the list.
-    sessions_t::size_type i = session_->get_index ();
-    sessions [i] = sessions [sessions.size () - 1];
-    sessions [i]->set_index (i);
-    sessions.pop_back ();
-}
-
-zmq::i_poller *zmq::app_thread_t::get_poller ()
-{
-    zmq_assert (false);
+    delete this;
 }
 
 zmq::i_signaler *zmq::app_thread_t::get_signaler ()
@@ -98,76 +64,20 @@ zmq::i_signaler *zmq::app_thread_t::get_signaler ()
 
 bool zmq::app_thread_t::is_current ()
 {
-    return !sessions.empty () && tid == getpid ();
+    return !sockets.empty () && tid == getpid ();
 }
 
 bool zmq::app_thread_t::make_current ()
 {
     //  If there are object managed by this slot we cannot assign the slot
     //  to a different thread.
-    if (!sessions.empty ())
+    if (!sockets.empty ())
         return false;
 
     tid = getpid ();
     return true;
 }
 
-zmq::i_api *zmq::app_thread_t::create_socket (int type_)
-{
-    i_mux *mux = NULL;
-    i_demux *demux = NULL;
-    session_t *session = NULL;
-    i_api *api = NULL;
-
-    switch (type_) {
-    case ZMQ_P2P:
-        mux = new dummy_aggregator_t;
-        zmq_assert (mux);
-        demux = new dummy_distributor_t;
-        zmq_assert (demux);
-        session = new session_t (this, this, mux, demux, true, false);
-        zmq_assert (session);
-        api = new p2p_t (this, session);
-        zmq_assert (api);
-        break;
-    case ZMQ_PUB:
-        demux = new data_distributor_t;
-        zmq_assert (demux);
-        session = new session_t (this, this, mux, demux, true, false);
-        zmq_assert (session);
-        api = new pub_t (this, session);
-        zmq_assert (api);
-        break;
-    case ZMQ_SUB:
-        mux = new fair_aggregator_t;
-        zmq_assert (mux);
-        session = new session_t (this, this, mux, demux, true, false);
-        zmq_assert (session);
-        api = new sub_t (this, session);
-        zmq_assert (api);
-        break;
-    case ZMQ_REQ:
-        //  TODO
-        zmq_assert (false);
-        api = new req_t (this, session);
-        zmq_assert (api);
-        break;
-    case ZMQ_REP:
-        //  TODO
-        zmq_assert (false);
-        api = new rep_t (this, session);
-        zmq_assert (api);
-        break;
-    default:
-        errno = EINVAL;
-        return NULL;
-    }
-
-    attach_session (session);
-    
-    return api;
-}
-
 void zmq::app_thread_t::process_commands (bool block_)
 {
     ypollset_t::signals_t signals;
diff --git a/src/app_thread.hpp b/src/app_thread.hpp
index 8295c2f..31679b8 100644
--- a/src/app_thread.hpp
+++ b/src/app_thread.hpp
@@ -22,7 +22,7 @@
 
 #include <vector>
 
-#include "i_thread.hpp"
+#include "i_socket.hpp"
 #include "stdint.hpp"
 #include "object.hpp"
 #include "ypollset.hpp"
@@ -30,23 +30,18 @@
 namespace zmq
 {
 
-    class app_thread_t : public object_t, public i_thread
+    class app_thread_t : public object_t
     {
     public:
 
         app_thread_t (class context_t *context_, int thread_slot_);
 
-        //  To be called when the whole infrastrucure is being closed.
-        void shutdown ();
+        ~app_thread_t ();
 
         //  Returns signaler associated with this application thread.
         i_signaler *get_signaler ();
 
-        //  Create socket engine in this thread. Return false if the calling
-        //  thread doesn't match the thread handled by this app thread object.
-        struct i_api *create_socket (int type_);
-
-        //  Nota bene: The following two functions are accessed from different
+        //  Nota bene: Following two functions are accessed from different
         //  threads. The caller (context) is responsible for synchronisation
         //  of accesses.
 
@@ -61,25 +56,17 @@ namespace zmq
         //  set to true, returns only after at least one command was processed.
         void process_commands (bool block_);
 
-        //  i_thread implementation.
-        void attach_session (class session_t *session_);
-        void detach_session (class session_t *session_);
-        struct i_poller *get_poller ();
-
     private:
 
-        //  Clean-up.
-        ~app_thread_t ();
+        //  All the sockets created from this application thread.
+        typedef std::vector <i_socket*> sockets_t;
+        sockets_t sockets;
 
         //  Thread ID associated with this slot.
         //  TODO: Virtualise pid_t!
         //  TODO: Check whether getpid returns unique ID for each thread.
         int tid;
 
-        //  Vector of all sessionss associated with this app thread.
-        typedef std::vector <class session_t*> sessions_t;
-        sessions_t sessions;
-
         //  App thread's signaler object.
         ypollset_t pollset;
 
diff --git a/src/connecter.cpp b/src/connecter.cpp
deleted file mode 100644
index 970dcf7..0000000
--- a/src/connecter.cpp
+++ /dev/null
@@ -1,189 +0,0 @@
-/*
-    Copyright (c) 2007-2009 FastMQ Inc.
-
-    This file is part of 0MQ.
-
-    0MQ is free software; you can redistribute it and/or modify it under
-    the terms of the Lesser GNU General Public License as published by
-    the Free Software Foundation; either version 3 of the License, or
-    (at your option) any later version.
-
-    0MQ 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
-    Lesser GNU General Public License for more details.
-
-    You should have received a copy of the Lesser GNU General Public License
-    along with this program.  If not, see <http://www.gnu.org/licenses/>.
-*/
-
-#include "connecter.hpp"
-#include "io_thread.hpp"
-#include "session.hpp"
-#include "err.hpp"
-#include "simple_semaphore.hpp"
-#include "zmq_tcp_engine.hpp"
-
-zmq::connecter_t::connecter_t (io_thread_t *thread_, const char *addr_,
-      session_t *session_) :
-    io_object_t (thread_),
-    state (idle),
-    poller (NULL),
-    session (session_),
-    addr (addr_),
-    identity ("abcde"),
-    engine (NULL)
-{
-}
-
-void zmq::connecter_t::terminate ()
-{
-    delete this;
-}
-
-void zmq::connecter_t::shutdown ()
-{
-    delete this;
-}
-
-zmq::connecter_t::~connecter_t ()
-{
-}
-
-void zmq::connecter_t::process_reg (simple_semaphore_t *smph_)
-{
-    //  Fet poller pointer for further use.
-    zmq_assert (!poller);
-    poller = get_poller ();
-
-    //  Ask the session to register itself with the I/O thread. Note that
-    //  the session is living in the same I/O thread, thus this results
-    //  in a synchronous call.
-    session->inc_seqnum ();
-    send_reg (session, NULL);
-
-    //  Unlock the application thread that created the connecter.
-    if (smph_)
-        smph_->post ();
-
-    //  Manually trigger timer event which will launch asynchronous connect.
-    state = waiting;
-    timer_event ();
-}
-
-void zmq::connecter_t::process_unreg (simple_semaphore_t *smph_)
-{
-    //  Unregister connecter/engine from the poller.
-    zmq_assert (poller);
-    if (state == connecting)
-        poller->rm_fd (handle);
-    else if (state == waiting)
-        poller->cancel_timer (this);
-    else if (state == sending)
-        engine->terminate ();
-
-    //  Unlock the application thread closing the connecter.
-    if (smph_)
-        smph_->post ();
-}
-
-void zmq::connecter_t::in_event ()
-{
-    //  Error occured in asynchronous connect. Retry to connect after a while.
-    if (state == connecting) {
-        fd_t fd = tcp_connecter.connect ();
-        zmq_assert (fd == retired_fd);
-        poller->rm_fd (handle);
-        poller->add_timer (this);
-        state = waiting;
-        return;
-    }
-
-    zmq_assert (false);
-}
-
-void zmq::connecter_t::out_event ()
-{
-    if (state == connecting) {
-
-        fd_t fd = tcp_connecter.connect ();
-        if (fd == retired_fd) {
-            poller->rm_fd (handle);
-            poller->add_timer (this);
-            state = waiting;
-            return;
-        }
-
-        poller->rm_fd (handle);
-        engine = new zmq_tcp_engine_t (fd);
-        zmq_assert (engine);
-        engine->attach (poller, this);
-        state = sending;
-        return;
-    }
-
-    zmq_assert (false);
-}
-
-void zmq::connecter_t::timer_event ()
-{
-    zmq_assert (state == waiting);
-
-    //  Initiate async connect and start polling for its completion. If async
-    //  connect fails instantly, try to reconnect after a while.
-    int rc = tcp_connecter.open (addr.c_str ());
-    if (rc == 0) {
-        state = connecting;
-        in_event ();
-    }
-    else if (rc == 1) {
-        handle = poller->add_fd (tcp_connecter.get_fd (), this);
-        poller->set_pollout (handle);
-        state = connecting;
-    }
-    else {
-        poller->add_timer (this);
-        state = waiting;
-    }
-}
-
-void zmq::connecter_t::set_engine (struct i_engine *engine_)
-{
-    engine = engine_;
-}
-
-bool zmq::connecter_t::read (zmq_msg *msg_)
-{
-    zmq_assert (state == sending);
-
-    //  Deallocate old content of the message just in case.
-    zmq_msg_close (msg_);
-
-    //  Send the identity.
-    zmq_msg_init_size (msg_, identity.size ());
-    memcpy (zmq_msg_data (msg_), identity.c_str (), identity.size ());
-
-    //  Ask engine to unregister from the poller.
-    i_engine *e = engine;
-    engine->detach ();
-
-    //  Attach the engine to the session. (Note that this is actually
-    //  a synchronous call.
-    session->inc_seqnum ();
-    send_engine (session, e);
-
-    state = idle;
-
-    return true;    
-}
-
-bool zmq::connecter_t::write (struct zmq_msg *msg_)
-{
-    //  No incoming messages are accepted till identity is sent.
-    return false;
-}
-
-void zmq::connecter_t::flush ()
-{
-    //  No incoming messages are accepted till identity is sent.
-}
diff --git a/src/connecter.hpp b/src/connecter.hpp
deleted file mode 100644
index 1f11c63..0000000
--- a/src/connecter.hpp
+++ /dev/null
@@ -1,99 +0,0 @@
-/*
-    Copyright (c) 2007-2009 FastMQ Inc.
-
-    This file is part of 0MQ.
-
-    0MQ is free software; you can redistribute it and/or modify it under
-    the terms of the Lesser GNU General Public License as published by
-    the Free Software Foundation; either version 3 of the License, or
-    (at your option) any later version.
-
-    0MQ 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
-    Lesser GNU General Public License for more details.
-
-    You should have received a copy of the Lesser GNU General Public License
-    along with this program.  If not, see <http://www.gnu.org/licenses/>.
-*/
-
-#ifndef __ZMQ_CONNECTER_HPP_INCLUDED__
-#define __ZMQ_CONNECTER_HPP_INCLUDED__
-
-#include <string>
-
-#include "../include/zmq.h"
-
-#include "i_poller.hpp"
-#include "io_object.hpp"
-#include "i_poll_events.hpp"
-#include "i_session.hpp"
-#include "tcp_connecter.hpp"
-
-namespace zmq
-{
-
-    class connecter_t : public io_object_t, public i_poll_events,
-        public i_session
-    {
-    public:
-
-        connecter_t (class io_thread_t *thread_, const char *addr_,
-            class session_t *session_);
-
-        void terminate ();
-        void shutdown ();
-
-        void process_reg (class simple_semaphore_t *smph_);
-        void process_unreg (class simple_semaphore_t *smph_);
-
-        //  i_poll_events implementation.
-        void in_event ();
-        void out_event ();
-        void timer_event ();
-
-        //  i_session implementation
-        void set_engine (struct i_engine *engine_);
-        //  void shutdown ();
-        bool read (struct zmq_msg *msg_);
-        bool write (struct zmq_msg *msg_);
-        void flush ();
-
-    private:
-
-        //  Clean-up.
-        ~connecter_t ();
-
-         enum {
-             idle,
-             waiting,
-             connecting,
-             sending
-         } state;
-
-        //  Cached pointer to the poller.
-        struct i_poller *poller;
-
-        //  Handle of the connecting socket.
-        handle_t handle;
-
-        //  Associated session. It lives in the same I/O thread.
-        class session_t *session;
-
-        //  Address to connect to.
-        std::string addr;
-
-        //  Identity of the connection.
-        std::string identity;
-
-        tcp_connecter_t tcp_connecter;
-
-        struct i_engine *engine;
-
-        connecter_t (const connecter_t&);
-        void operator = (const connecter_t&);
-    };
-
-}
-
-#endif
diff --git a/src/context.cpp b/src/context.cpp
index ab4643e..6b071cf 100644
--- a/src/context.cpp
+++ b/src/context.cpp
@@ -20,15 +20,12 @@
 #include "../include/zmq.h"
 
 #include "context.hpp"
+#include "i_api.hpp"
 #include "app_thread.hpp"
 #include "io_thread.hpp"
 #include "platform.hpp"
 #include "err.hpp"
 #include "pipe.hpp"
-#include "pipe_reader.hpp"
-#include "pipe_writer.hpp"
-#include "session.hpp"
-#include "i_api.hpp"
 
 #if defined ZMQ_HAVE_WINDOWS
 #include "windows.h"
@@ -72,37 +69,23 @@ zmq::context_t::context_t (int app_threads_, int io_threads_)
         io_threads [i]->start ();
 }
 
-void zmq::context_t::shutdown ()
-{
-     delete this;
-}
-
 zmq::context_t::~context_t ()
 {
-    //  Ask I/O threads to terminate.
+    //  Close all application theads, sockets, io_objects etc.
+    for (app_threads_t::size_type i = 0; i != app_threads.size (); i++)
+        delete app_threads [i];
+
+    //  Ask I/O threads to terminate. If stop signal wasn't sent to I/O
+    //  thread subsequent invocation of destructor would hang-up.
     for (io_threads_t::size_type i = 0; i != io_threads.size (); i++)
         io_threads [i]->stop ();
 
     //  Wait till I/O threads actually terminate.
     for (io_threads_t::size_type i = 0; i != io_threads.size (); i++)
-        io_threads [i]->join ();
-
-    //  At this point the current thread is the only thread with access to
-    //  our internal data. Deallocation will be done exclusively in this thread.
-    for (app_threads_t::size_type i = 0; i != app_threads.size (); i++)
-        app_threads [i]->shutdown ();
-    for (io_threads_t::size_type i = 0; i != io_threads.size (); i++)
-        io_threads [i]->shutdown ();
+        delete io_threads [i];
 
     delete [] command_pipes;
 
-    //  Deallocate all the pipes, pipe readers and pipe writers.
-    for (pipes_t::iterator it = pipes.begin (); it != pipes.end (); it++) {
-        delete it->pipe;
-        delete it->reader;
-        delete it->writer;
-    }
-
 #ifdef ZMQ_HAVE_WINDOWS
     //  On Windows, uninitialise socket layer.
     int rc = WSACleanup ();
@@ -123,7 +106,11 @@ zmq::i_api *zmq::context_t::create_socket (int type_)
         threads_sync.unlock ();
         return NULL;
     }
-    i_api *s = thread->create_socket (type_);
+
+    zmq_assert (false);
+    i_api *s = NULL;
+    //i_api *s = thread->create_socket (type_);
+
     threads_sync.unlock ();
     return s;
 }
@@ -164,103 +151,3 @@ zmq::io_thread_t *zmq::context_t::choose_io_thread (uint64_t taskset_)
 
     return io_threads [result];
 }
-
-void zmq::context_t::create_pipe (object_t *reader_parent_,
-    object_t *writer_parent_, uint64_t hwm_, uint64_t lwm_,
-    pipe_reader_t **reader_, pipe_writer_t **writer_)
-{
-    //  Create the pipe, reader & writer triple.
-    pipe_t *pipe = new pipe_t;
-    zmq_assert (pipe);
-    pipe_reader_t *reader = new pipe_reader_t (reader_parent_, pipe,
-        hwm_, lwm_);
-    zmq_assert (reader);
-    pipe_writer_t *writer = new pipe_writer_t (writer_parent_, pipe, reader,
-        hwm_, lwm_);
-    zmq_assert (writer);
-    reader->set_peer (writer);
-
-    //  Store the pipe in the repository.
-    pipe_info_t info = {pipe, reader, writer};
-    pipes_sync.lock ();
-    pipe->set_index (pipes.size ());
-    pipes.push_back (info);
-    pipes_sync.unlock ();
-
-    *reader_ = reader;
-    *writer_ = writer;
-}
-
-void zmq::context_t::destroy_pipe (pipe_t *pipe_)
-{
-    //  Remove the pipe from the repository.
-    pipe_info_t info;
-    pipes_sync.lock ();
-    pipes_t::size_type i = pipe_->get_index ();
-    info = pipes [i];
-    pipes [i] = pipes.back ();
-    pipes.pop_back ();
-    pipes_sync.unlock ();
-
-    //  Deallocate the pipe and associated pipe reader & pipe writer.
-    zmq_assert (info.pipe == pipe_);
-    delete info.pipe;
-    delete info.reader;
-    delete info.writer;
-}
-
-int zmq::context_t::register_inproc_endpoint (const char *endpoint_,
-    session_t *session_)
-{
-   inproc_endpoint_sync.lock ();
-   inproc_endpoints_t::iterator it = inproc_endpoints.find (endpoint_);
-
-   if (it != inproc_endpoints.end ()) {
-       inproc_endpoint_sync.unlock ();
-       errno = EADDRINUSE;
-       return -1;
-   }
-
-   inproc_endpoints.insert (std::make_pair (endpoint_, session_));
-
-   inproc_endpoint_sync.unlock ();
-   return 0;
-}
-
-zmq::object_t *zmq::context_t::get_inproc_endpoint (const char *endpoint_)
-{
-    inproc_endpoint_sync.lock ();
-    inproc_endpoints_t::iterator it = inproc_endpoints.find (endpoint_);
-
-    if (it == inproc_endpoints.end ()) {
-        inproc_endpoint_sync.unlock ();
-        errno = EADDRNOTAVAIL;
-        return NULL;
-    }
-
-    it->second->inc_seqnum ();
-    object_t *session = it->second;
-
-    inproc_endpoint_sync.unlock ();
-    return session;
-}
-
-void zmq::context_t::unregister_inproc_endpoints (session_t *session_)
-{
-    inproc_endpoint_sync.lock ();
-
-    //  Remove the connection from the repository.
-    //  TODO: Yes, the algorithm has O(n^2) complexity. Should be O(log n).
-    for (inproc_endpoints_t::iterator it = inproc_endpoints.begin ();
-          it != inproc_endpoints.end ();) {
-        if (it->second == session_) {
-            inproc_endpoints.erase (it);
-            it = inproc_endpoints.begin ();
-        }
-        else
-            it++;
-    }
-        
-    inproc_endpoint_sync.unlock ();
-}
-
diff --git a/src/context.hpp b/src/context.hpp
index 7701ef7..f2eab1c 100644
--- a/src/context.hpp
+++ b/src/context.hpp
@@ -52,9 +52,9 @@ namespace zmq
         context_t (int app_threads_, int io_threads_);
 
         //  To be called to terminate the whole infrastructure (zmq_term).
-        void shutdown ();
+        ~context_t ();
 
-        //  Create a socket engine.
+        //  Create a socket.
         struct i_api *create_socket (int type_);
 
         //  Returns number of thread slots in the context. To be used by
@@ -81,37 +81,12 @@ namespace zmq
                 destination_].read (command_);
         }
 
-        //  Creates new pipe.
-        void create_pipe (class object_t *reader_parent_,
-            class object_t *writer_parent_, uint64_t hwm_, uint64_t lwm_,
-            class pipe_reader_t **reader_, class pipe_writer_t **writer_);
-
-        //  Deallocates the pipe.
-        void destroy_pipe (class pipe_t *pipe_);
-
-        //  Registers existing session object as an inproc endpoint.
-        int register_inproc_endpoint (const char *endpoint_,
-            class session_t *session_);
-
-        //  Retrieves an inproc endpoint. Increments the command sequence number
-        //  of the object by one. Caller is thus bound to send the command
-        //  to the connection after invoking this function. Returns NULL if
-        //  the endpoint doesn't exist.
-        class object_t *get_inproc_endpoint (const char *endpoint_);
-
-        //  Removes all the inproc endpoints associated with the given session
-        //  object from the global repository.
-        void unregister_inproc_endpoints (class session_t *session_);
-
         //  Returns the I/O thread that is the least busy at the moment.
         //  Taskset specifies which I/O threads are eligible (0 = all).
         class io_thread_t *choose_io_thread (uint64_t taskset_);
 
     private:
 
-        //  Clean-up.
-        ~context_t ();
-
         //  Returns the app thread associated with the current thread.
         //  NULL if we are out of app thread slots.
         class app_thread_t *choose_app_thread ();
@@ -137,29 +112,6 @@ namespace zmq
         //  Synchronisation of accesses to shared thread data.
         mutex_t threads_sync;
 
-        //  Global repository of pipes. It's used only on terminal shutdown
-        //  to deallocate all the pipes irrespective of whether they are
-        //  referenced from pipe_reader, pipe_writer or both.
-        struct pipe_info_t
-        {
-            class pipe_t *pipe;
-            class pipe_reader_t *reader;
-            class pipe_writer_t *writer;
-        };
-        typedef std::vector <pipe_info_t> pipes_t;
-        pipes_t pipes;
-
-        //  Synchronisation of access to global repository of pipes.
-        mutex_t pipes_sync;
-
-        //  Global repository of available inproc endpoints.
-        typedef std::map <std::string, class session_t*> inproc_endpoints_t;
-        inproc_endpoints_t inproc_endpoints;
-
-        //  Synchronisation of access to the global repository
-        //  of inproc endpoints.
-        mutex_t inproc_endpoint_sync;
-
         context_t (const context_t&);
         void operator = (const context_t&);
     };
diff --git a/src/data_distributor.cpp b/src/data_distributor.cpp
deleted file mode 100644
index 971edce..0000000
--- a/src/data_distributor.cpp
+++ /dev/null
@@ -1,155 +0,0 @@
-/*
-    Copyright (c) 2007-2009 FastMQ Inc.
-
-    This file is part of 0MQ.
-
-    0MQ is free software; you can redistribute it and/or modify it under
-    the terms of the Lesser GNU General Public License as published by
-    the Free Software Foundation; either version 3 of the License, or
-    (at your option) any later version.
-
-    0MQ 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
-    Lesser GNU General Public License for more details.
-
-    You should have received a copy of the Lesser GNU General Public License
-    along with this program.  If not, see <http://www.gnu.org/licenses/>.
-*/
-
-#include "../include/zmq.h"
-
-#include "data_distributor.hpp"
-#include "pipe_writer.hpp"
-#include "err.hpp"
-#include "session.hpp"
-#include "msg.hpp"
-
-zmq::data_distributor_t::data_distributor_t () :
-    session (NULL)
-{
-}
-
-void zmq::data_distributor_t::set_session (session_t *session_)
-{
-    zmq_assert (!session);
-    session = session_;
-}
-
-void zmq::data_distributor_t::shutdown ()
-{
-    //  No need to deallocate pipes here. They'll be deallocated during the
-    //  shutdown of the dispatcher.
-    delete this;
-}
-
-void zmq::data_distributor_t::terminate ()
-{
-    //  Pipe unregisters itself during the call to terminate, so the pipes
-    //  list shinks by one in each iteration.
-    while (!pipes.empty ())
-        pipes [0]->terminate ();
-
-   delete this;
-}
-
-zmq::data_distributor_t::~data_distributor_t ()
-{
-}
-
-void zmq::data_distributor_t::attach_pipe (pipe_writer_t *pipe_)
-{
-    //  Associate demux with a new pipe.
-    pipe_->set_demux (this);
-    pipe_->set_index (pipes.size ());
-    pipes.push_back (pipe_);
-}
-
-void zmq::data_distributor_t::detach_pipe (pipe_writer_t *pipe_)
-{
-    //  Release the reference to the pipe.
-    int index = pipe_->get_index ();
-    pipe_->set_index (-1);
-    pipes [index] = pipes.back ();
-    pipes [index]->set_index (index);
-    pipes.pop_back ();
-}
-
-bool zmq::data_distributor_t::empty ()
-{
-    return pipes.empty ();
-}
-
-bool zmq::data_distributor_t::send (zmq_msg *msg_)
-{
-    int pipes_count = pipes.size ();
-
-    //  If there are no pipes available, simply drop the message.
-    if (pipes_count == 0) {
-        zmq_msg_close (msg_);
-        zmq_msg_init (msg_);
-        return true;
-    }
-
-    //  TODO: ???
-    //  First check whether all pipes are available for writing.
-//    for (pipes_t::iterator it = pipes.begin (); it != pipes.end (); it ++)
-//        if (!(*it)->check_write (msg_))
-//            return false;
-
-    //  For VSMs the copying is straighforward.
-    if (msg_->content == (zmq_msg_content*) ZMQ_VSM) {
-        for (pipes_t::iterator it = pipes.begin (); it != pipes.end (); it ++)
-            write_to_pipe (*it, msg_);
-        zmq_msg_init (msg_);
-        return true;
-    }
-
-    //  Optimisation for the case when there's only a single pipe
-    //  to send the message to - no refcount adjustment (i.e. atomic
-    //  operations) needed.
-    if (pipes_count == 1) {
-        write_to_pipe (*pipes.begin (), msg_);
-        zmq_msg_init (msg_);
-        return true;
-    }
-
-    //  There are at least 2 destinations for the message. That means we have
-    //  to deal with reference counting. First add N-1 references to
-    //  the content (we are holding one reference anyway, that's why the -1).
-    if (msg_->shared)
-        msg_->content->refcnt.add (pipes_count - 1);
-    else {
-        msg_->shared = true;
-        // TODO: Add memory barrier here.
-        msg_->content->refcnt.set (pipes_count);
-    }
-
-    //  Push the message to all destinations.
-    for (pipes_t::iterator it = pipes.begin (); it != pipes.end (); it ++)
-        write_to_pipe (*it, msg_);
-
-    //  Detach the original message from the data buffer.
-    zmq_msg_init (msg_);
-
-    return true;
-}
-
-void zmq::data_distributor_t::flush ()
-{
-    //  Flush all pipes. If there's large number of pipes, it can be pretty
-    //  inefficient (especially if there's new message only in a single pipe).
-    //  Can it be improved?
-    for (pipes_t::iterator it = pipes.begin (); it != pipes.end (); it ++)
-        (*it)->flush ();
-}
-
-void zmq::data_distributor_t::write_to_pipe (class pipe_writer_t *pipe_,
-    struct zmq_msg *msg_)
-{
-    if (!pipe_->write (msg_)) {
-        //  TODO: Push gap notification to the pipe.
-        zmq_assert (false);
-    }
-}
-
diff --git a/src/data_distributor.hpp b/src/data_distributor.hpp
deleted file mode 100644
index 5bde2e8..0000000
--- a/src/data_distributor.hpp
+++ /dev/null
@@ -1,70 +0,0 @@
-/*
-    Copyright (c) 2007-2009 FastMQ Inc.
-
-    This file is part of 0MQ.
-
-    0MQ is free software; you can redistribute it and/or modify it under
-    the terms of the Lesser GNU General Public License as published by
-    the Free Software Foundation; either version 3 of the License, or
-    (at your option) any later version.
-
-    0MQ 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
-    Lesser GNU General Public License for more details.
-
-    You should have received a copy of the Lesser GNU General Public License
-    along with this program.  If not, see <http://www.gnu.org/licenses/>.
-*/
-
-#ifndef __ZMQ_DATA_DISTRIBUTOR_HPP_INCLUDED__
-#define __ZMQ_DATA_DISTRIBUTOR_HPP_INCLUDED__
-
-#include <vector>
-
-#include <i_demux.hpp>
-
-namespace zmq
-{
-
-    //  Object to distribute messages to outbound pipes.
-
-    class data_distributor_t : public i_demux
-    {
-    public:
-
-        data_distributor_t ();
-
-        //  i_demux implementation.
-        void set_session (class session_t *session_);
-        void shutdown ();
-        void terminate ();
-        void attach_pipe (class pipe_writer_t *pipe_);
-        void detach_pipe (class pipe_writer_t *pipe_);
-        bool empty ();
-        bool send (struct zmq_msg *msg_);
-        void flush ();
-
-    private:
-
-        //  Clean-up.
-        ~data_distributor_t ();
-
-        //  Reference to the owner session object.
-        class session_t *session;
-
-        //  Writes the message to the pipe if possible. If it isn't, writes
-        //  a gap notification to the pipe.
-        void write_to_pipe (class pipe_writer_t *pipe_, struct zmq_msg *msg_);
-
-        //  The list of outbound pipes.
-        typedef std::vector <class pipe_writer_t*> pipes_t;
-        pipes_t pipes;
-
-        data_distributor_t (const data_distributor_t&);
-        void operator = (const data_distributor_t&);
-    };
-
-}
-
-#endif
diff --git a/src/devpoll.cpp b/src/devpoll.cpp
index 8fb0877..b7d153c 100644
--- a/src/devpoll.cpp
+++ b/src/devpoll.cpp
@@ -52,6 +52,10 @@ zmq::devpoll_t::devpoll_t ()
 
 zmq::devpoll_t::~devpoll_t ()
 {
+    //  Make sure there are no fds registered on shutdown.
+    zmq_assert (load.get () == 0);
+
+    worker.stop ();
     close (devpoll_fd);
 }
 
@@ -152,11 +156,6 @@ void zmq::devpoll_t::stop ()
     stopping = true;
 }
 
-void zmq::devpoll_t::join ()
-{
-    worker.stop ();
-}
-
 bool zmq::devpoll_t::loop ()
 {
     //  According to the poll(7d) man page, we can retrieve
diff --git a/src/devpoll.hpp b/src/devpoll.hpp
index 28274c0..57f0156 100644
--- a/src/devpoll.hpp
+++ b/src/devpoll.hpp
@@ -42,7 +42,7 @@ namespace zmq
     public:
 
         devpoll_t ();
-        virtual ~devpoll_t ();
+        ~devpoll_t ();
 
         //  i_poller implementation.
         handle_t add_fd (fd_t fd_, i_poll_events *events_);
@@ -56,7 +56,6 @@ namespace zmq
         int get_load ();
         void start ();
         void stop ();
-        void join ();
 
     private:
 
diff --git a/src/dummy_aggregator.cpp b/src/dummy_aggregator.cpp
deleted file mode 100644
index 0b27fab..0000000
--- a/src/dummy_aggregator.cpp
+++ /dev/null
@@ -1,111 +0,0 @@
-/*
-    Copyright (c) 2007-2009 FastMQ Inc.
-
-    This file is part of 0MQ.
-
-    0MQ is free software; you can redistribute it and/or modify it under
-    the terms of the Lesser GNU General Public License as published by
-    the Free Software Foundation; either version 3 of the License, or
-    (at your option) any later version.
-
-    0MQ 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
-    Lesser GNU General Public License for more details.
-
-    You should have received a copy of the Lesser GNU General Public License
-    along with this program.  If not, see <http://www.gnu.org/licenses/>.
-*/
-
-#include "../include/zmq.h"
-
-#include "dummy_aggregator.hpp"
-#include "err.hpp"
-#include "pipe_reader.hpp"
-#include "session.hpp"
-
-//  Swaps pipes at specified indices. 
-#define swap_pipes(i1_, i2_) \
-    std::swap (pipes [i1_], pipes [i2_]);\
-    pipes [i1_]->set_index (i1_);\
-    pipes [i2_]->set_index (i2_);
-
-zmq::dummy_aggregator_t::dummy_aggregator_t () :
-    session (NULL),
-    pipe (NULL),
-    active (false)
-{
-}
-
-void zmq::dummy_aggregator_t::set_session (session_t *session_)
-{
-    zmq_assert (!session);
-    session = session_;
-}
-
-void zmq::dummy_aggregator_t::shutdown ()
-{
-    //  No need to deallocate the pipe here. It'll be deallocated during the
-    //  shutdown of the dispatcher.
-    delete this;
-}
-
-void zmq::dummy_aggregator_t::terminate ()
-{
-    if (pipe)
-        pipe->terminate ();
-
-   delete this;
-}
-
-zmq::dummy_aggregator_t::~dummy_aggregator_t ()
-{
-}
-
-void zmq::dummy_aggregator_t::attach_pipe (pipe_reader_t *pipe_)
-{
-    zmq_assert (!pipe);
-    pipe = pipe_;
-    active = true;
-
-    //  Associate new pipe with the mux object.
-    pipe_->set_mux (this);
-    session->revive ();
-}
-
-void zmq::dummy_aggregator_t::detach_pipe (pipe_reader_t *pipe_)
-{
-    zmq_assert (pipe == pipe_);
-    deactivate (pipe_);
-    pipe = NULL;
-}
-
-bool zmq::dummy_aggregator_t::empty ()
-{
-    return pipe == NULL;
-}
-
-bool zmq::dummy_aggregator_t::recv (zmq_msg *msg_)
-{
-    //  Deallocate old content of the message.
-    zmq_msg_close (msg_);
-        
-    //  Try to read from the pipe.
-    if (pipe && pipe->read (msg_))
-        return true;
-
-    //  No message is available. Initialise the output parameter
-    //  to be a 0-byte message.
-    zmq_msg_init (msg_);
-    return false;
-}
-
-void zmq::dummy_aggregator_t::deactivate (pipe_reader_t *pipe_)
-{
-    active = false;
-}
-
-void zmq::dummy_aggregator_t::reactivate (pipe_reader_t *pipe_)
-{
-    active = true;
-}
diff --git a/src/dummy_aggregator.hpp b/src/dummy_aggregator.hpp
deleted file mode 100644
index 6a9e9db..0000000
--- a/src/dummy_aggregator.hpp
+++ /dev/null
@@ -1,73 +0,0 @@
-/*
-    Copyright (c) 2007-2009 FastMQ Inc.
-
-    This file is part of 0MQ.
-
-    0MQ is free software; you can redistribute it and/or modify it under
-    the terms of the Lesser GNU General Public License as published by
-    the Free Software Foundation; either version 3 of the License, or
-    (at your option) any later version.
-
-    0MQ 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
-    Lesser GNU General Public License for more details.
-
-    You should have received a copy of the Lesser GNU General Public License
-    along with this program.  If not, see <http://www.gnu.org/licenses/>.
-*/
-
-#ifndef __ZMQ_DUMMY_AGGREGATOR_HPP_INCLUDED__
-#define __ZMQ_DUMMY_AGGREGATOR_HPP_INCLUDED__
-
-#include <vector>
-
-#include "i_mux.hpp"
-
-namespace zmq
-{
-
-    //  Fake message aggregator. There can be at most one pipe bound to it,
-    //  so there's no real aggregation going on. However, it is more efficient
-    //  than a real aggregator. It's intended to be used in the contexts
-    //  where business logic ensures there'll be at most one pipe bound.
-
-    class dummy_aggregator_t : public i_mux
-    {
-    public:
-
-        dummy_aggregator_t ();
-
-        //  i_mux interface implementation.
-        void set_session (session_t *session_);
-        void shutdown ();
-        void terminate ();
-        void attach_pipe (class pipe_reader_t *pipe_);
-        void detach_pipe (class pipe_reader_t *pipe_);
-        bool empty ();
-        void deactivate (class pipe_reader_t *pipe_);
-        void reactivate (class pipe_reader_t *pipe_);
-        bool recv (struct zmq_msg *msg_);
-
-
-    private:
-
-        //  Clean-up.
-        ~dummy_aggregator_t ();
-
-        //  Reference to the owner session object.
-        class session_t *session;
-
-        //  The single pipe bound.
-        class pipe_reader_t *pipe;
-
-        //  If true, the pipe is active.
-        bool active;
-
-        dummy_aggregator_t (const dummy_aggregator_t&);
-        void operator = (const dummy_aggregator_t&);
-    };
-
-}
-
-#endif
diff --git a/src/dummy_distributor.cpp b/src/dummy_distributor.cpp
deleted file mode 100644
index 62e2b88..0000000
--- a/src/dummy_distributor.cpp
+++ /dev/null
@@ -1,85 +0,0 @@
-/*
-    Copyright (c) 2007-2009 FastMQ Inc.
-
-    This file is part of 0MQ.
-
-    0MQ is free software; you can redistribute it and/or modify it under
-    the terms of the Lesser GNU General Public License as published by
-    the Free Software Foundation; either version 3 of the License, or
-    (at your option) any later version.
-
-    0MQ 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
-    Lesser GNU General Public License for more details.
-
-    You should have received a copy of the Lesser GNU General Public License
-    along with this program.  If not, see <http://www.gnu.org/licenses/>.
-*/
-
-#include "../include/zmq.h"
-
-#include "dummy_distributor.hpp"
-#include "pipe_writer.hpp"
-#include "err.hpp"
-#include "session.hpp"
-#include "msg.hpp"
-
-zmq::dummy_distributor_t::dummy_distributor_t () :
-    session (NULL)
-{
-}
-
-void zmq::dummy_distributor_t::set_session (session_t *session_)
-{
-    zmq_assert (!session);
-    session = session_;
-}
-
-void zmq::dummy_distributor_t::shutdown ()
-{
-    //  No need to deallocate pipe here. It'll be deallocated during the
-    //  shutdown of the dispatcher.
-    delete this;
-}
-
-void zmq::dummy_distributor_t::terminate ()
-{
-   if (pipe)
-       pipe->terminate ();
-
-   delete this;
-}
-
-zmq::dummy_distributor_t::~dummy_distributor_t ()
-{
-}
-
-void zmq::dummy_distributor_t::attach_pipe (pipe_writer_t *pipe_)
-{
-    zmq_assert (!pipe);
-    pipe = pipe_;
-}
-
-void zmq::dummy_distributor_t::detach_pipe (pipe_writer_t *pipe_)
-{
-    zmq_assert (pipe == pipe_);
-    pipe = NULL;
-}
-
-bool zmq::dummy_distributor_t::empty ()
-{
-    return pipe == NULL;
-}
-
-bool zmq::dummy_distributor_t::send (zmq_msg *msg_)
-{
-    return pipe && pipe->write (msg_);
-}
-
-void zmq::dummy_distributor_t::flush ()
-{
-    if (pipe)
-        pipe->flush ();
-}
-
diff --git a/src/dummy_distributor.hpp b/src/dummy_distributor.hpp
deleted file mode 100644
index a71cc49..0000000
--- a/src/dummy_distributor.hpp
+++ /dev/null
@@ -1,68 +0,0 @@
-/*
-    Copyright (c) 2007-2009 FastMQ Inc.
-
-    This file is part of 0MQ.
-
-    0MQ is free software; you can redistribute it and/or modify it under
-    the terms of the Lesser GNU General Public License as published by
-    the Free Software Foundation; either version 3 of the License, or
-    (at your option) any later version.
-
-    0MQ 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
-    Lesser GNU General Public License for more details.
-
-    You should have received a copy of the Lesser GNU General Public License
-    along with this program.  If not, see <http://www.gnu.org/licenses/>.
-*/
-
-#ifndef __ZMQ_DUMMY_DISTRIBUTOR_HPP_INCLUDED__
-#define __ZMQ_DUMMY_DISTRIBUTOR_HPP_INCLUDED__
-
-#include <vector>
-
-#include <i_demux.hpp>
-
-namespace zmq
-{
-
-    //  Fake message distributor. There can be only one pipe bound to it
-    //  so there no real distribution going on. However, it is more efficient
-    //  than a real distributor and should be used where business logic
-    //  ensures there'll be at most one pipe bound.
-
-    class dummy_distributor_t : public i_demux
-    {
-    public:
-
-        dummy_distributor_t ();
-
-        //  i_demux implementation.
-        void set_session (class session_t *session_);
-        void shutdown ();
-        void terminate ();
-        void attach_pipe (class pipe_writer_t *pipe_);
-        void detach_pipe (class pipe_writer_t *pipe_);
-        bool empty ();
-        bool send (struct zmq_msg *msg_);
-        void flush ();
-
-    private:
-
-        //  Clean-up.
-        ~dummy_distributor_t ();
-
-        //  Reference to the owner session object.
-        class session_t *session;
-
-        //  The bound pipe.
-        class pipe_writer_t *pipe;
-
-        dummy_distributor_t (const dummy_distributor_t&);
-        void operator = (const dummy_distributor_t&);
-    };
-
-}
-
-#endif
diff --git a/src/epoll.cpp b/src/epoll.cpp
index c4c8fdb..15278c6 100644
--- a/src/epoll.cpp
+++ b/src/epoll.cpp
@@ -41,8 +41,11 @@ zmq::epoll_t::epoll_t () :
 
 zmq::epoll_t::~epoll_t ()
 {
-    close (epoll_fd);
+    //  Make sure there are no fds registered on shutdown.
+    zmq_assert (load.get () == 0);
 
+    worker.stop ();
+    close (epoll_fd);
     for (retired_t::iterator it = retired.begin (); it != retired.end (); it ++)
         delete *it;
 }
@@ -144,11 +147,6 @@ void zmq::epoll_t::stop ()
     stopping = true;
 }
 
-void zmq::epoll_t::join ()
-{
-    worker.stop ();
-}
-
 void zmq::epoll_t::loop ()
 {
     epoll_event ev_buf [max_io_events];
diff --git a/src/epoll.hpp b/src/epoll.hpp
index aa363ee..619d4f3 100644
--- a/src/epoll.hpp
+++ b/src/epoll.hpp
@@ -44,7 +44,7 @@ namespace zmq
     public:
 
         epoll_t ();
-        virtual ~epoll_t ();
+        ~epoll_t ();
 
         //  i_poller implementation.
         handle_t add_fd (fd_t fd_, i_poll_events *events_);
@@ -58,7 +58,6 @@ namespace zmq
         int get_load ();
         void start ();
         void stop ();
-        void join ();
 
     private:
 
diff --git a/src/fair_aggregator.cpp b/src/fair_aggregator.cpp
deleted file mode 100644
index 1e6937f..0000000
--- a/src/fair_aggregator.cpp
+++ /dev/null
@@ -1,143 +0,0 @@
-/*
-    Copyright (c) 2007-2009 FastMQ Inc.
-
-    This file is part of 0MQ.
-
-    0MQ is free software; you can redistribute it and/or modify it under
-    the terms of the Lesser GNU General Public License as published by
-    the Free Software Foundation; either version 3 of the License, or
-    (at your option) any later version.
-
-    0MQ 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
-    Lesser GNU General Public License for more details.
-
-    You should have received a copy of the Lesser GNU General Public License
-    along with this program.  If not, see <http://www.gnu.org/licenses/>.
-*/
-
-#include "../include/zmq.h"
-
-#include "fair_aggregator.hpp"
-#include "err.hpp"
-#include "pipe_reader.hpp"
-#include "session.hpp"
-
-//  Swaps pipes at specified indices. 
-#define swap_pipes(i1_, i2_) \
-    std::swap (pipes [i1_], pipes [i2_]);\
-    pipes [i1_]->set_index (i1_);\
-    pipes [i2_]->set_index (i2_);
-
-zmq::fair_aggregator_t::fair_aggregator_t () :
-    session (NULL),
-    active (0),
-    current (0)
-{
-}
-
-void zmq::fair_aggregator_t::set_session (session_t *session_)
-{
-    zmq_assert (!session);
-    session = session_;
-}
-
-void zmq::fair_aggregator_t::shutdown ()
-{
-    //  No need to deallocate pipes here. They'll be deallocated during the
-    //  shutdown of the dispatcher.
-    delete this;
-}
-
-void zmq::fair_aggregator_t::terminate ()
-{
-    //  Pipe unregisters itself during the call to terminate, so the pipes
-    //  list shinks by one in each iteration.
-    while (!pipes.empty ())
-        pipes [0]->terminate ();
-
-   delete this;
-}
-
-zmq::fair_aggregator_t::~fair_aggregator_t ()
-{
-}
-
-void zmq::fair_aggregator_t::attach_pipe (pipe_reader_t *pipe_)
-{
-    //  Associate new pipe with the mux object.
-    pipe_->set_mux (this);
-    pipes.push_back (pipe_);
-    active++;
-    if (pipes.size () > active)
-        swap_pipes (pipes.size () - 1, active - 1);
-    if (active == 1)
-        session->revive ();
-}
-
-void zmq::fair_aggregator_t::detach_pipe (pipe_reader_t *pipe_)
-{
-    //  Move the pipe from the list of active pipes to the list of idle pipes.
-    deactivate (pipe_);
-            
-    //  Move the pipe to the end of the idle list and remove it.
-    swap_pipes (pipe_->get_index (), pipes.size () - 1);
-    pipes.pop_back ();
-}
-
-bool zmq::fair_aggregator_t::empty ()
-{
-    return pipes.empty ();
-}
-
-bool zmq::fair_aggregator_t::recv (zmq_msg *msg_)
-{
-    //  Deallocate old content of the message.
-    zmq_msg_close (msg_);
-
-    //  O(1) fair queueing. Round-robin over the active pipes to get
-    //  next message.
-    for (pipes_t::size_type i = active; i != 0; i--) {
-
-        //  Update current.
-        current = (current + 1) % active;
-        
-        //  Try to read from current.
-        if (pipes [current]->read (msg_))
-            return true;
-    }
-
-    //  No message is available. Initialise the output parameter
-    //  to be a 0-byte message.
-    zmq_msg_init (msg_);
-    return false;
-}
-
-void zmq::fair_aggregator_t::deactivate (pipe_reader_t *pipe_)
-{
-    int index = pipe_->get_index ();
-
-    //  Suspend an active pipe.
-    swap_pipes (index, active - 1);
-    active--;
-
-    //  If the deactiveted pipe is the current one, shift the current one pipe
-    //  backwards so that the pipe that replaced the deactiveted one will be
-    //  processed immediately rather than skipped.
-    if (index == (int) current) {
-        index--;
-        if (index == -1)
-            index = active - 1;
-        current = index;
-    }
-}
-
-void zmq::fair_aggregator_t::reactivate (pipe_reader_t *pipe_)
-{
-    //  Revive an idle pipe.
-    swap_pipes (pipe_->get_index (), active);
-    active++;
-    if (active == 1)
-        session->revive ();
-}
diff --git a/src/fair_aggregator.hpp b/src/fair_aggregator.hpp
deleted file mode 100644
index 6ae1fc5..0000000
--- a/src/fair_aggregator.hpp
+++ /dev/null
@@ -1,77 +0,0 @@
-/*
-    Copyright (c) 2007-2009 FastMQ Inc.
-
-    This file is part of 0MQ.
-
-    0MQ is free software; you can redistribute it and/or modify it under
-    the terms of the Lesser GNU General Public License as published by
-    the Free Software Foundation; either version 3 of the License, or
-    (at your option) any later version.
-
-    0MQ 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
-    Lesser GNU General Public License for more details.
-
-    You should have received a copy of the Lesser GNU General Public License
-    along with this program.  If not, see <http://www.gnu.org/licenses/>.
-*/
-
-#ifndef __ZMQ_FAIR_AGGREGATOR_HPP_INCLUDED__
-#define __ZMQ_FAIR_AGGREGATOR_HPP_INCLUDED__
-
-#include <vector>
-
-#include "i_mux.hpp"
-
-namespace zmq
-{
-
-    //  Object to aggregate messages from inbound pipes.
-
-    class fair_aggregator_t : public i_mux
-    {
-    public:
-
-        fair_aggregator_t ();
-
-        //  i_mux interface implementation.
-        void set_session (session_t *session_);
-        void shutdown ();
-        void terminate ();
-        void attach_pipe (class pipe_reader_t *pipe_);
-        void detach_pipe (class pipe_reader_t *pipe_);
-        bool empty ();
-        void deactivate (class pipe_reader_t *pipe_);
-        void reactivate (class pipe_reader_t *pipe_);
-        bool recv (struct zmq_msg *msg_);
-
-
-    private:
-
-        //  Clean-up.
-        ~fair_aggregator_t ();
-
-        //  Reference to the owner session object.
-        class session_t *session;
-
-        //  The list of inbound pipes. The active pipes are occupying indices
-        //  from 0 to active-1. Suspended pipes occupy indices from 'active'
-        //  to the end of the array.
-        typedef std::vector <class pipe_reader_t*> pipes_t;
-        pipes_t pipes;
-
-        //  The number of active pipes.
-        pipes_t::size_type active;
-
-        //  Pipe to retrieve next message from. The messages are retrieved
-        //  from the pipes in round-robin fashion (a.k.a. fair queueing).
-        pipes_t::size_type current;
-
-        fair_aggregator_t (const fair_aggregator_t&);
-        void operator = (const fair_aggregator_t&);
-    };
-
-}
-
-#endif
diff --git a/src/i_api.hpp b/src/i_api.hpp
index fc7275b..a87e41d 100644
--- a/src/i_api.hpp
+++ b/src/i_api.hpp
@@ -1,28 +1,28 @@
 /*
-    Copyright (c) 2007-2009 FastMQ Inc.
-
-    This file is part of 0MQ.
-
-    0MQ is free software; you can redistribute it and/or modify it under
-    the terms of the Lesser GNU General Public License as published by
-    the Free Software Foundation; either version 3 of the License, or
-    (at your option) any later version.
-
-    0MQ 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
-    Lesser GNU General Public License for more details.
-
-    You should have received a copy of the Lesser GNU General Public License
-    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+Copyright (c) 2007-2009 FastMQ Inc.
+ 
+This file is part of 0MQ.
+ 
+0MQ is free software; you can redistribute it and/or modify it under
+the terms of the Lesser GNU General Public License as published by
+the Free Software Foundation; either version 3 of the License, or
+(at your option) any later version.
+ 
+0MQ 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
+Lesser GNU General Public License for more details.
+ 
+You should have received a copy of the Lesser GNU General Public License
+along with this program. If not, see <http://www.gnu.org/licenses/>.
 */
-
+ 
 #ifndef __ZMQ_I_API_HPP_INCLUDED__
 #define __ZMQ_I_API_HPP_INCLUDED__
-
+ 
 namespace zmq
 {
-
+ 
     struct i_api
     {
         virtual int bind (const char *addr_, struct zmq_opts *opts_) = 0;
@@ -33,7 +33,7 @@ namespace zmq
         virtual int recv (struct zmq_msg *msg_, int flags_) = 0;
         virtual int close () = 0;
     };
-
+ 
 }
-
+ 
 #endif
diff --git a/src/i_demux.hpp b/src/i_demux.hpp
deleted file mode 100644
index c4755b5..0000000
--- a/src/i_demux.hpp
+++ /dev/null
@@ -1,57 +0,0 @@
-/*
-    Copyright (c) 2007-2009 FastMQ Inc.
-
-    This file is part of 0MQ.
-
-    0MQ is free software; you can redistribute it and/or modify it under
-    the terms of the Lesser GNU General Public License as published by
-    the Free Software Foundation; either version 3 of the License, or
-    (at your option) any later version.
-
-    0MQ 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
-    Lesser GNU General Public License for more details.
-
-    You should have received a copy of the Lesser GNU General Public License
-    along with this program.  If not, see <http://www.gnu.org/licenses/>.
-*/
-
-#ifndef __ZMQ_I_DEMUX_HPP_INCLUDED__
-#define __ZMQ_I_DEMUX_HPP_INCLUDED__
-
-namespace zmq
-{
-
-    struct i_demux
-    {
-        //  Attaches mux to a particular session.
-        virtual void set_session (class session_t *session_) = 0;
-
-        //  To be called when the whole infrastrucure
-        //  is being closed (zmq_term).
-        virtual void shutdown () = 0;
-
-        //  To be called when session is being closed.
-        virtual void terminate () = 0;
-
-        //  Adds new pipe to the demux to send messages to.
-        virtual void attach_pipe (class pipe_writer_t *pipe_) = 0;
-
-        //  Removes pipe from the demux.
-        virtual void detach_pipe (class pipe_writer_t *pipe_) = 0;
-
-        //  Returns true if there's no pipe attached.
-        virtual bool empty () = 0;
-
-        //  Sends the message. Returns false if the message cannot be sent
-        //  because the pipes are full.
-        virtual bool send (struct zmq_msg *msg_) = 0;
-
-        //  Flushes messages downstream.
-        virtual void flush () = 0;
-    };
-
-}
-
-#endif
diff --git a/src/i_engine.hpp b/src/i_engine.hpp
deleted file mode 100644
index 8ca2007..0000000
--- a/src/i_engine.hpp
+++ /dev/null
@@ -1,53 +0,0 @@
-/*
-    Copyright (c) 2007-2009 FastMQ Inc.
-
-    This file is part of 0MQ.
-
-    0MQ is free software; you can redistribute it and/or modify it under
-    the terms of the Lesser GNU General Public License as published by
-    the Free Software Foundation; either version 3 of the License, or
-    (at your option) any later version.
-
-    0MQ 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
-    Lesser GNU General Public License for more details.
-
-    You should have received a copy of the Lesser GNU General Public License
-    along with this program.  If not, see <http://www.gnu.org/licenses/>.
-*/
-
-#ifndef __ZMQ_I_ENGINE_HPP_INCLUDED__
-#define __ZMQ_I_ENGINE_HPP_INCLUDED__
-
-namespace zmq
-{
-
-    //  Generic interface to access engines from MD objects.
-
-    struct i_engine
-    {
-        //  Attach the engine with specified context.
-        virtual void attach (struct i_poller *poller_,
-            struct i_session *session_) = 0;
-
-        //  Detach the engine from the current context.
-        virtual void detach () = 0;
-
-        //  Notify the engine that new messages are available.
-        virtual void revive () = 0;
-
-        //  Called by session when it decides the engine
-        //  should terminate itself.
-        virtual void schedule_terminate () = 0;
-
-        //  Called by normal object termination process.
-        virtual void terminate () = 0;
-
-        //  To be called by MD when terminal shutdown (zmq_term) is in progress.
-        virtual void shutdown () = 0;
-    };
-
-}
-
-#endif
diff --git a/src/i_mux.hpp b/src/i_mux.hpp
deleted file mode 100644
index 22e0a26..0000000
--- a/src/i_mux.hpp
+++ /dev/null
@@ -1,60 +0,0 @@
-/*
-    Copyright (c) 2007-2009 FastMQ Inc.
-
-    This file is part of 0MQ.
-
-    0MQ is free software; you can redistribute it and/or modify it under
-    the terms of the Lesser GNU General Public License as published by
-    the Free Software Foundation; either version 3 of the License, or
-    (at your option) any later version.
-
-    0MQ 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
-    Lesser GNU General Public License for more details.
-
-    You should have received a copy of the Lesser GNU General Public License
-    along with this program.  If not, see <http://www.gnu.org/licenses/>.
-*/
-
-#ifndef __ZMQ_I_MUX_HPP_INCLUDED__
-#define __ZMQ_I_MUX_HPP_INCLUDED__
-
-namespace zmq
-{
-
-    struct i_mux
-    {
-        //  Attaches mux to a particular session.
-        virtual void set_session (class session_t *session_) = 0;
-
-        //  To be called when the whole infrastrucure
-        //  is being closed (zmq_term).
-        virtual void shutdown () = 0;
-
-        //  To be called when session is being closed.
-        virtual void terminate () = 0;
-
-        //  Adds new pipe to the mux to send messages to.
-        virtual void attach_pipe (class pipe_reader_t *pipe_) = 0;
-
-        //  Removes pipe from the mux.
-        virtual void detach_pipe (class pipe_reader_t *pipe_) = 0;
-
-        //  Returns true if there's no pipe attached.
-        virtual bool empty () = 0;
-
-        //  Shifts the pipe from active to passive state and vice versa.
-        //  TODO: Check whether state transitions cannot be done by
-        //  mux object itself without a need for external APIs.
-        virtual void deactivate (class pipe_reader_t *pipe_) = 0;
-        virtual void reactivate (class pipe_reader_t *pipe_) = 0;
-
-        //  Receives a message. Returns false when there is no message
-        //  to receive.
-        virtual bool recv (struct zmq_msg *msg_) = 0;
-    };
-
-}
-
-#endif
diff --git a/src/i_poller.hpp b/src/i_poller.hpp
index 52ca095..2665e82 100644
--- a/src/i_poller.hpp
+++ b/src/i_poller.hpp
@@ -75,13 +75,8 @@ namespace zmq
         //  This method is called from a foreign thread.
         virtual void start () = 0;
 
-        //  Ask underlying I/O thread to stop. This method is called from
-        //  underlying thread (callback from io_thread object).
+        //  Ask underlying I/O thread to stop.
         virtual void stop () = 0;
-
-        //  Wait for termination of undelying I/O thread.
-        //  This method is called from a foreign thread.
-        virtual void join () = 0;
     };
 
 }
diff --git a/src/i_session.hpp b/src/i_session.hpp
deleted file mode 100644
index 21cdc0d..0000000
--- a/src/i_session.hpp
+++ /dev/null
@@ -1,37 +0,0 @@
-/*
-    Copyright (c) 2007-2009 FastMQ Inc.
-
-    This file is part of 0MQ.
-
-    0MQ is free software; you can redistribute it and/or modify it under
-    the terms of the Lesser GNU General Public License as published by
-    the Free Software Foundation; either version 3 of the License, or
-    (at your option) any later version.
-
-    0MQ 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
-    Lesser GNU General Public License for more details.
-
-    You should have received a copy of the Lesser GNU General Public License
-    along with this program.  If not, see <http://www.gnu.org/licenses/>.
-*/
-
-#ifndef __ZMQ_I_SESSION_HPP_INCLUDED__
-#define __ZMQ_I_SESSION_HPP_INCLUDED__
-
-namespace zmq
-{
-
-    struct i_session
-    {
-        virtual void set_engine (struct i_engine *engine_) = 0;
-        virtual void shutdown () = 0;
-        virtual bool read (struct zmq_msg *msg_) = 0;
-        virtual bool write (struct zmq_msg *msg_) = 0;
-        virtual void flush () = 0;
-    };
-
-}
-
-#endif
diff --git a/src/req.cpp b/src/i_socket.hpp
index 01018f5..99ade8a 100644
--- a/src/req.cpp
+++ b/src/i_socket.hpp
@@ -17,13 +17,20 @@
     along with this program.  If not, see <http://www.gnu.org/licenses/>.
 */
 
-#include "../include/zmq.h"
+#ifndef __ZMQ_I_SOCKET_HPP_INCLUDED__
+#define __ZMQ_I_SOCKET_HPP_INCLUDED__
 
-#include "req.hpp"
-#include "app_thread.hpp"
-#include "session.hpp"
-
-zmq::req_t::req_t (app_thread_t *thread_, session_t *session_) :
-    socket_base_t (thread_, session_)
+namespace zmq
 {
+
+    struct i_socket
+    {
+        virtual ~i_socket () {};
+
+        //  Start shutting down the socket.
+        virtual void stop () = 0;
+    };
+
 }
+
+#endif
diff --git a/src/i_thread.hpp b/src/i_thread.hpp
deleted file mode 100644
index 9f31592..0000000
--- a/src/i_thread.hpp
+++ /dev/null
@@ -1,38 +0,0 @@
-/*
-    Copyright (c) 2007-2009 FastMQ Inc.
-
-    This file is part of 0MQ.
-
-    0MQ is free software; you can redistribute it and/or modify it under
-    the terms of the Lesser GNU General Public License as published by
-    the Free Software Foundation; either version 3 of the License, or
-    (at your option) any later version.
-
-    0MQ 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
-    Lesser GNU General Public License for more details.
-
-    You should have received a copy of the Lesser GNU General Public License
-    along with this program.  If not, see <http://www.gnu.org/licenses/>.
-*/
-
-#ifndef __ZMQ_I_THREAD_HPP_INCLUDED__
-#define __ZMQ_I_THREAD_HPP_INCLUDED__
-
-namespace zmq
-{
-
-    //  Interface used by session object to communicate with the thread
-    //  it belongs to.
-
-    struct i_thread
-    {
-        virtual void attach_session (class session_t *session_) = 0;
-        virtual void detach_session (class session_t *session_) = 0;
-        virtual struct i_poller *get_poller () = 0;
-    };
-
-}
-
-#endif
diff --git a/src/io_object.cpp b/src/io_object.cpp
deleted file mode 100644
index ad379cf..0000000
--- a/src/io_object.cpp
+++ /dev/null
@@ -1,37 +0,0 @@
-/*
-    Copyright (c) 2007-2009 FastMQ Inc.
-
-    This file is part of 0MQ.
-
-    0MQ is free software; you can redistribute it and/or modify it under
-    the terms of the Lesser GNU General Public License as published by
-    the Free Software Foundation; either version 3 of the License, or
-    (at your option) any later version.
-
-    0MQ 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
-    Lesser GNU General Public License for more details.
-
-    You should have received a copy of the Lesser GNU General Public License
-    along with this program.  If not, see <http://www.gnu.org/licenses/>.
-*/
-
-#include "io_object.hpp"
-#include "io_thread.hpp"
-#include "i_poller.hpp"
-
-zmq::io_object_t::io_object_t (io_thread_t *thread_) :
-    object_t (thread_),
-    thread (thread_)
-{
-}
-
-zmq::io_object_t::~io_object_t ()
-{
-}
-
-zmq::i_poller *zmq::io_object_t::get_poller ()
-{
-    return thread->get_poller ();
-}
diff --git a/src/io_object.hpp b/src/io_object.hpp
deleted file mode 100644
index d3fa809..0000000
--- a/src/io_object.hpp
+++ /dev/null
@@ -1,51 +0,0 @@
-/*
-    Copyright (c) 2007-2009 FastMQ Inc.
-
-    This file is part of 0MQ.
-
-    0MQ is free software; you can redistribute it and/or modify it under
-    the terms of the Lesser GNU General Public License as published by
-    the Free Software Foundation; either version 3 of the License, or
-    (at your option) any later version.
-
-    0MQ 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
-    Lesser GNU General Public License for more details.
-
-    You should have received a copy of the Lesser GNU General Public License
-    along with this program.  If not, see <http://www.gnu.org/licenses/>.
-*/
-
-#ifndef __ZMQ_IO_OBJECT_HPP_INCLUDED__
-#define __ZMQ_IO_OBJECT_HPP_INCLUDED__
-
-#include "object.hpp"
-
-namespace zmq
-{
-
-    //  All objects running within the context of an I/O thread should be
-    //  derived from this class to allow owning application threads to
-    //  destroy them.
-
-    class io_object_t : public object_t
-    {
-    public:
-
-        io_object_t (class io_thread_t *thread_);
-        ~io_object_t ();
-
-        virtual void terminate () = 0;
-        virtual void shutdown () = 0;
-
-        struct i_poller *get_poller ();
-
-    private:
- 
-        class io_thread_t *thread;
-    };
-
-}
-
-#endif
diff --git a/src/io_thread.cpp b/src/io_thread.cpp
index 162ed4c..f5261a6 100644
--- a/src/io_thread.cpp
+++ b/src/io_thread.cpp
@@ -30,9 +30,7 @@
 #include "devpoll.hpp"
 #include "kqueue.hpp"
 #include "context.hpp"
-#include "session.hpp"
 #include "simple_semaphore.hpp"
-#include "session.hpp"
 
 zmq::io_thread_t::io_thread_t (context_t *context_, int thread_slot_) :
     object_t (context_, thread_slot_)
@@ -76,15 +74,6 @@ zmq::io_thread_t::io_thread_t (context_t *context_, int thread_slot_) :
     poller->set_pollin (signaler_handle);
 }
 
-void zmq::io_thread_t::shutdown ()
-{
-    //  Deallocate all the sessions associated with the thread.
-    while (!sessions.empty ())
-        sessions [0]->shutdown ();
-
-    delete this;
-}
-
 zmq::io_thread_t::~io_thread_t ()
 {
     delete poller;
@@ -101,11 +90,6 @@ void zmq::io_thread_t::stop ()
     send_stop ();
 }
 
-void zmq::io_thread_t::join ()
-{
-    poller->join ();
-}
-
 zmq::i_signaler *zmq::io_thread_t::get_signaler ()
 {
     return &signaler;
@@ -149,21 +133,6 @@ void zmq::io_thread_t::timer_event ()
     zmq_assert (false);
 }
 
-void zmq::io_thread_t::attach_session (session_t *session_)
-{
-    session_->set_index (sessions.size ());
-    sessions.push_back (session_); 
-}
-
-void zmq::io_thread_t::detach_session (session_t *session_)
-{
-    //  O(1) removal of the session from the list.
-    sessions_t::size_type i = session_->get_index ();
-    sessions [i] = sessions [sessions.size () - 1];
-    sessions [i]->set_index (i);
-    sessions.pop_back ();
-}
-
 zmq::i_poller *zmq::io_thread_t::get_poller ()
 {
     zmq_assert (poller);
diff --git a/src/io_thread.hpp b/src/io_thread.hpp
index 585a28b..43ee19e 100644
--- a/src/io_thread.hpp
+++ b/src/io_thread.hpp
@@ -23,7 +23,6 @@
 #include <vector>
 
 #include "object.hpp"
-#include "i_thread.hpp"
 #include "i_poller.hpp"
 #include "i_poll_events.hpp"
 #include "fd_signaler.hpp"
@@ -34,26 +33,22 @@ namespace zmq
     //  Generic part of the I/O thread. Polling-mechanism-specific features
     //  are implemented in separate "polling objects".
 
-    class io_thread_t : public object_t, public i_poll_events, public i_thread
+    class io_thread_t : public object_t, public i_poll_events
     {
     public:
 
         io_thread_t (class context_t *context_, int thread_slot_);
 
+        //  Clean-up. If the thread was started, it's neccessary to call 'stop'
+        //  before invoking destructor. Otherwise the destructor would hang up.
+        ~io_thread_t ();
+
         //  Launch the physical thread.
         void start ();
 
         //  Ask underlying thread to stop.
         void stop ();
 
-        //  Wait till undelying thread terminates.
-        void join ();
-
-        //  To be called when the whole infrastrucure is being closed (zmq_term).
-        //  It's vital to call the individual commands in this sequence:
-        //  stop, join, shutdown.
-        void shutdown ();
-
         //  Returns signaler associated with this I/O thread.
         i_signaler *get_signaler ();
 
@@ -62,9 +57,7 @@ namespace zmq
         void out_event ();
         void timer_event ();
 
-        //  i_thread implementation.
-        void attach_session (class session_t *session_);
-        void detach_session (class session_t *session_);
+        //  ???
         struct i_poller *get_poller ();
 
         //  Command handlers.
@@ -75,9 +68,6 @@ namespace zmq
 
     private:
 
-        //  Clean-up.
-        ~io_thread_t ();
-
         //  Poll thread gets notifications about incoming commands using
         //  this signaler.
         fd_signaler_t signaler;
@@ -87,11 +77,6 @@ namespace zmq
 
         //  I/O multiplexing is performed using a poller object.
         i_poller *poller;
-
-        //  Vector of all sessions associated with this app thread.
-        typedef std::vector <class session_t*> sessions_t;
-        sessions_t sessions;
-
     };
 
 }
diff --git a/src/kqueue.cpp b/src/kqueue.cpp
index 28c15de..f4c58a3 100644
--- a/src/kqueue.cpp
+++ b/src/kqueue.cpp
@@ -42,6 +42,10 @@ zmq::kqueue_t::kqueue_t ()
 
 zmq::kqueue_t::~kqueue_t ()
 {
+    //  Make sure there are no fds registered on shutdown.
+    zmq_assert (load.get () == 0);
+
+    worker.stop ();
     close (kqueue_fd);
 }
 
@@ -144,11 +148,6 @@ void zmq::kqueue_t::stop ()
     stopping = true;
 }
 
-void zmq::kqueue_t::join ()
-{
-    worker.stop ();
-}
-
 void zmq::kqueue_t::loop ()
 {
     while (!stopping) {
diff --git a/src/kqueue.hpp b/src/kqueue.hpp
index 2fd6819..eeb6f09 100644
--- a/src/kqueue.hpp
+++ b/src/kqueue.hpp
@@ -42,7 +42,7 @@ namespace zmq
     public:
 
         kqueue_t ();
-        virtual ~kqueue_t ();
+        ~kqueue_t ();
 
         //  i_poller implementation.
         handle_t add_fd (fd_t fd_, i_poll_events *events_);
@@ -56,7 +56,6 @@ namespace zmq
         int get_load ();
         void start ();
         void stop ();
-        void join ();
 
     private:
 
diff --git a/src/listener.cpp b/src/listener.cpp
deleted file mode 100644
index 823b21b..0000000
--- a/src/listener.cpp
+++ /dev/null
@@ -1,170 +0,0 @@
-/*
-    Copyright (c) 2007-2009 FastMQ Inc.
-
-    This file is part of 0MQ.
-
-    0MQ is free software; you can redistribute it and/or modify it under
-    the terms of the Lesser GNU General Public License as published by
-    the Free Software Foundation; either version 3 of the License, or
-    (at your option) any later version.
-
-    0MQ 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
-    Lesser GNU General Public License for more details.
-
-    You should have received a copy of the Lesser GNU General Public License
-    along with this program.  If not, see <http://www.gnu.org/licenses/>.
-*/
-
-#include "listener.hpp"
-#include "simple_semaphore.hpp"
-#include "zmq_tcp_engine.hpp"
-#include "io_thread.hpp"
-#include "session_stub.hpp"
-#include "session.hpp"
-#include "err.hpp"
-#include "dummy_aggregator.hpp"
-#include "dummy_distributor.hpp"
-
-zmq::listener_t::listener_t (io_thread_t *thread_, const char *addr_,
-      session_t *peer_, bool has_in_, bool has_out_, uint64_t taskset_) :
-    io_object_t (thread_),
-    poller (NULL),
-    addr (addr_),
-    peer (peer_),
-    taskset (taskset_),
-    has_in (has_in_),
-    has_out (has_out_)
-{
-}
-
-void zmq::listener_t::terminate ()
-{
-    for (session_stubs_t::size_type i = 0; i != session_stubs.size (); i++)
-        session_stubs [i]->terminate ();
-    delete this;
-}
-
-void zmq::listener_t::shutdown ()
-{
-    for (session_stubs_t::size_type i = 0; i != session_stubs.size (); i++)
-        session_stubs [i]->shutdown ();
-    delete this;
-}
-
-zmq::listener_t::~listener_t ()
-{
-}
-
-void zmq::listener_t::got_identity (session_stub_t *session_stub_,
-    const char *identity_)
-{
-    //  Get the engine allready disconnected from the stub and poller.
-    i_engine *engine = session_stub_->detach_engine ();
-    zmq_assert (engine);
-
-    //  Find the corresponding session.
-    session_t *session;
-    sessions_t::iterator it = sessions.find (identity_);
-
-    //  Destroy the stub.
-    int i = session_stub_->get_index ();
-    session_stubs [i] = session_stubs [session_stubs.size () - 1];
-    session_stubs [i]->set_index (i);
-    session_stubs.pop_back ();
-    session_stub_->terminate ();
-
-    //  If there's no session with the specified identity, create one.
-    if (it != sessions.end ()) {
-        session = it->second;
-        session->inc_seqnum ();
-    }
-    else {
-
-        //  Choose an I/O thread with the least load to handle the new session.
-        io_thread_t *io_thread = choose_io_thread (taskset);
-
-        //  Create the session and bind it to the I/O thread and peer. Make
-        //  sure that the peer session won't get deallocated till it processes
-        //  the subsequent bind command.
-        i_mux *mux = new dummy_aggregator_t;
-        zmq_assert (mux);
-        i_demux *demux = new dummy_distributor_t;
-        zmq_assert (demux);
-        session = new session_t (io_thread, io_thread, mux, demux, false, true);
-        zmq_assert (session);
-        session->inc_seqnum ();
-        session->inc_seqnum ();
-        peer->inc_seqnum ();
-        send_reg_and_bind (session, peer, has_in, has_out);
-    }
-
-    //  Attach the engine to the session.
-    send_engine (session, engine);
-}
-
-void zmq::listener_t::process_reg (simple_semaphore_t *smph_)
-{
-    zmq_assert (!poller);
-    poller = get_poller ();
-
-    //  Open the listening socket.
-    int rc = tcp_listener.open (addr.c_str ());
-    zmq_assert (rc == 0);
-
-    //  Unlock the application thread that created the listener.
-    if (smph_)
-        smph_->post ();
-
-    //  Start polling for incoming connections.
-    handle = poller->add_fd (tcp_listener.get_fd (), this);
-    poller->set_pollin (handle);
-}
-
-void zmq::listener_t::process_unreg (simple_semaphore_t *smph_)
-{
-    //  Disassociate listener from the poller.
-    zmq_assert (poller);
-    poller->rm_fd (handle);
-    poller = NULL;
-
-    //  Unlock the application thread closing the listener.
-    if (smph_)
-        smph_->post ();
-}
-
-void zmq::listener_t::in_event ()
-{
-    fd_t fd = tcp_listener.accept ();
-
-    //  If connection was reset by the peer in the meantime, just ignore it.
-    //  TODO: Handle specific errors like ENFILE/EMFILE etc.
-    if (fd == retired_fd)
-        return;
-
-    //  Create an session stub for the engine to take care for it till its
-    //  identity is retreived.
-    session_stub_t *session_stub = new session_stub_t (this);
-    zmq_assert (session_stub);
-    session_stub->set_index (session_stubs.size ());
-    session_stubs.push_back (session_stub);
-
-    //  Create an engine to encaspulate the socket. Engine will register itself
-    //  with the stub so the stub will be able to free it in case of shutdown.
-    zmq_tcp_engine_t *engine = new zmq_tcp_engine_t (fd);
-    zmq_assert (engine);
-    engine->attach (poller, session_stub);
-}
-
-void zmq::listener_t::out_event ()
-{
-    zmq_assert (false);
-}
-
-void zmq::listener_t::timer_event ()
-{
-    zmq_assert (false);
-}
-
-
diff --git a/src/listener.hpp b/src/listener.hpp
deleted file mode 100644
index 2fe93db..0000000
--- a/src/listener.hpp
+++ /dev/null
@@ -1,110 +0,0 @@
-/*
-    Copyright (c) 2007-2009 FastMQ Inc.
-
-    This file is part of 0MQ.
-
-    0MQ is free software; you can redistribute it and/or modify it under
-    the terms of the Lesser GNU General Public License as published by
-    the Free Software Foundation; either version 3 of the License, or
-    (at your option) any later version.
-
-    0MQ 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
-    Lesser GNU General Public License for more details.
-
-    You should have received a copy of the Lesser GNU General Public License
-    along with this program.  If not, see <http://www.gnu.org/licenses/>.
-*/
-
-#ifndef __ZMQ_LISTENER_HPP_INCLUDED__
-#define __ZMQ_LISTENER_HPP_INCLUDED__
-
-#include <map>
-#include <vector>
-#include <string>
-
-#include "io_object.hpp"
-#include "tcp_listener.hpp"
-#include "i_poller.hpp"
-#include "i_poll_events.hpp"
-#include "stdint.hpp"
-
-namespace zmq
-{
-
-    class listener_t : public io_object_t, public i_poll_events
-    {
-    public:
-
-        listener_t (class io_thread_t *thread_, const char *addr_,
-            class session_t *peer_, bool has_in_, bool has_out_,
-            uint64_t taskset_);
-
-        void terminate ();
-        void shutdown ();
-
-        //  This function is called by session stub once the identity
-        //  is retrieved from the incoming connection.
-        void got_identity (class session_stub_t *session_stub_,
-            const char *identity_);
-
-        void process_reg (class simple_semaphore_t *smph_);
-        void process_unreg (class simple_semaphore_t *smph_);
-
-        //  i_poll_events implementation.
-        void in_event ();
-        void out_event ();
-        void timer_event ();
-
-    private:
-
-        ~listener_t ();
-
-        struct i_poller *poller;
-
-        //  Handle corresponding to the listening socket.
-        handle_t handle;
-
-        //  Actual listening socket.
-        tcp_listener_t tcp_listener;
-
-        //  Address to bind to.
-        std::string addr;
-
-        //  Peer session. All the newly created connections should bind to
-        //  this session.
-        session_t *peer;
-
-        //  Taskset specifies which I/O threads are to be use to handle
-        //  newly created connections (0 = all).
-        uint64_t taskset;
-
-        //  Sessions created by this listener are stored in this map. They are
-        //  indexed by peer identities so that the same peer connects to the
-        //  same session after reconnection.
-        //  NB: Sessions are destroyed from other place and possibly later on,
-        //  so no need to care about them during listener object termination.
-        typedef std::map <std::string, class session_t*> sessions_t;
-        sessions_t sessions;
-
-        //  List of engines (bound to temorary session stubs) that we haven't
-        //  retrieved the identity from so far.
-        typedef std::vector <class session_stub_t*> session_stubs_t;
-        session_stubs_t session_stubs;
-
-        //  If true, create inbound pipe when binding new connection
-        //  to the peer.
-        bool has_in;
-
-        //  If true, create outbound pipe when binding new connection
-        //  to the peer.
-        bool has_out;      
-
-        listener_t (const listener_t&);
-        void operator = (const listener_t&);
-    };
-
-}
-
-#endif
diff --git a/src/load_balancer.cpp b/src/load_balancer.cpp
deleted file mode 100644
index 0d382a1..0000000
--- a/src/load_balancer.cpp
+++ /dev/null
@@ -1,130 +0,0 @@
-/*
-    Copyright (c) 2007-2009 FastMQ Inc.
-
-    This file is part of 0MQ.
-
-    0MQ is free software; you can redistribute it and/or modify it under
-    the terms of the Lesser GNU General Public License as published by
-    the Free Software Foundation; either version 3 of the License, or
-    (at your option) any later version.
-
-    0MQ 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
-    Lesser GNU General Public License for more details.
-
-    You should have received a copy of the Lesser GNU General Public License
-    along with this program.  If not, see <http://www.gnu.org/licenses/>.
-*/
-
-#include "../include/zmq.h"
-
-#include "load_balancer.hpp"
-#include "pipe_writer.hpp"
-#include "err.hpp"
-#include "session.hpp"
-#include "msg.hpp"
-
-zmq::load_balancer_t::load_balancer_t () :
-    session (NULL),
-    current (0)
-{
-}
-
-void zmq::load_balancer_t::set_session (session_t *session_)
-{
-    zmq_assert (!session);
-    session = session_;
-}
-
-void zmq::load_balancer_t::shutdown ()
-{
-    //  No need to deallocate pipes here. They'll be deallocated during the
-    //  shutdown of the dispatcher.
-    delete this;
-}
-
-void zmq::load_balancer_t::terminate ()
-{
-    //  Pipe unregisters itself during the call to terminate, so the pipes
-    //  list shinks by one in each iteration.
-    while (!pipes.empty ())
-        pipes [0]->terminate ();
-
-   delete this;
-}
-
-zmq::load_balancer_t::~load_balancer_t ()
-{
-}
-
-void zmq::load_balancer_t::attach_pipe (pipe_writer_t *pipe_)
-{
-    //  Associate demux with a new pipe.
-    pipe_->set_demux (this);
-    pipe_->set_index (pipes.size ());
-    pipes.push_back (pipe_);
-}
-
-void zmq::load_balancer_t::detach_pipe (pipe_writer_t *pipe_)
-{
-    //  Release the reference to the pipe.
-    int index = pipe_->get_index ();
-    pipe_->set_index (-1);
-    pipes [index] = pipes.back ();
-    pipes [index]->set_index (index);
-    pipes.pop_back ();
-}
-
-bool zmq::load_balancer_t::empty ()
-{
-    return pipes.empty ();
-}
-
-bool zmq::load_balancer_t::send (zmq_msg *msg_)
-{
-    //  If there are no pipes, message cannot be sent.
-    if (pipes.size () == 0)
-        return false;
-
-    //  Find the first pipe that is ready to accept the message.
-    bool found = false;
-    for (pipes_t::size_type i = 0; !found && i < pipes.size (); i++) {
-//        if (pipes [current]->check_write (msg))
-            found = true;
-//        else
-//            current = (current + 1) % pipes.size ();
-    }
-
-    //  Oops, no pipe is ready to accept the message.
-    if (!found)
-        return false;
-
-    //  Send the message to the selected pipe.
-    write_to_pipe (pipes [current], msg_);
-    current = (current + 1) % pipes.size ();
-
-    //  Detach the original message from the data buffer.
-    zmq_msg_init (msg_);
-
-    return true;
-}
-
-void zmq::load_balancer_t::flush ()
-{
-    //  Flush all pipes. If there's large number of pipes, it can be pretty
-    //  inefficient (especially if there's new message only in a single pipe).
-    //  Can it be improved?
-    for (pipes_t::iterator it = pipes.begin (); it != pipes.end (); it ++)
-        (*it)->flush ();
-}
-
-void zmq::load_balancer_t::write_to_pipe (class pipe_writer_t *pipe_,
-    struct zmq_msg *msg_)
-{
-    if (!pipe_->write (msg_)) {
-        //  TODO: Push gap notification to the pipe.
-        zmq_assert (false);
-    }
-}
-
diff --git a/src/load_balancer.hpp b/src/load_balancer.hpp
deleted file mode 100644
index 953ed3b..0000000
--- a/src/load_balancer.hpp
+++ /dev/null
@@ -1,73 +0,0 @@
-/*
-    Copyright (c) 2007-2009 FastMQ Inc.
-
-    This file is part of 0MQ.
-
-    0MQ is free software; you can redistribute it and/or modify it under
-    the terms of the Lesser GNU General Public License as published by
-    the Free Software Foundation; either version 3 of the License, or
-    (at your option) any later version.
-
-    0MQ 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
-    Lesser GNU General Public License for more details.
-
-    You should have received a copy of the Lesser GNU General Public License
-    along with this program.  If not, see <http://www.gnu.org/licenses/>.
-*/
-
-#ifndef __ZMQ_LOAD_BALANCER_HPP_INCLUDED__
-#define __ZMQ_LOAD_BALANCER_HPP_INCLUDED__
-
-#include <vector>
-
-#include <i_demux.hpp>
-
-namespace zmq
-{
-
-    //  Object to distribute messages to outbound pipes.
-
-    class load_balancer_t : public i_demux
-    {
-    public:
-
-        load_balancer_t ();
-
-        //  i_demux implementation.
-        void set_session (class session_t *session_);
-        void shutdown ();
-        void terminate ();
-        void attach_pipe (class pipe_writer_t *pipe_);
-        void detach_pipe (class pipe_writer_t *pipe_);
-        bool empty ();
-        bool send (struct zmq_msg *msg_);
-        void flush ();
-
-    private:
-
-        //  Clean-up.
-        ~load_balancer_t ();
-
-        //  Reference to the owner session object.
-        class session_t *session;
-
-        //  Writes the message to the pipe if possible. If it isn't, writes
-        //  a gap notification to the pipe.
-        void write_to_pipe (class pipe_writer_t *pipe_, struct zmq_msg *msg_);
-
-        //  The list of outbound pipes.
-        typedef std::vector <class pipe_writer_t*> pipes_t;
-        pipes_t pipes;
-
-        //  Current pipe to write next message to.
-        pipes_t::size_type current;
-
-        load_balancer_t (const load_balancer_t&);
-        void operator = (const load_balancer_t&);
-    };
-
-}
-
-#endif
diff --git a/src/object.cpp b/src/object.cpp
index 7c85212..36f3937 100644
--- a/src/object.cpp
+++ b/src/object.cpp
@@ -20,12 +20,8 @@
 #include "object.hpp"
 #include "context.hpp"
 #include "err.hpp"
-#include "pipe_reader.hpp"
-#include "pipe_writer.hpp"
-#include "session.hpp"
 #include "io_thread.hpp"
 #include "simple_semaphore.hpp"
-#include "i_engine.hpp"
 
 zmq::object_t::object_t (context_t *context_, int thread_slot_) :
     context (context_),
@@ -103,35 +99,6 @@ void zmq::object_t::process_command (command_t &cmd_)
     }
 }
 
-void zmq::object_t::create_pipe (object_t *reader_parent_,
-    object_t *writer_parent_, uint64_t hwm_, uint64_t lwm_,
-    pipe_reader_t **reader_, pipe_writer_t **writer_)
-{
-    context->create_pipe (reader_parent_, writer_parent_, hwm_, lwm_,
-        reader_, writer_);
-}
-
-void zmq::object_t::destroy_pipe (pipe_t *pipe_)
-{
-    context->destroy_pipe (pipe_);
-}
-
-int zmq::object_t::register_inproc_endpoint (const char *endpoint_,
-    session_t *session_)
-{
-    return context->register_inproc_endpoint (endpoint_, session_);
-}
-
-zmq::object_t *zmq::object_t::get_inproc_endpoint (const char *endpoint_)
-{
-    return context->get_inproc_endpoint (endpoint_);
-}
-
-void zmq::object_t::unregister_inproc_endpoints (session_t *session_)
-{
-    context->unregister_inproc_endpoints (session_);
-}
-
 zmq::io_thread_t *zmq::object_t::choose_io_thread (uint64_t taskset_)
 {
     return context->choose_io_thread (taskset_);
diff --git a/src/object.hpp b/src/object.hpp
index 796e7fa..5851c68 100644
--- a/src/object.hpp
+++ b/src/object.hpp
@@ -44,14 +44,6 @@ namespace zmq
         //  Derived object can use following functions to interact with
         //  global repositories. See context.hpp for function details.
         int thread_slot_count ();
-        void create_pipe (class object_t *reader_parent_,
-            class object_t *writer_parent_, uint64_t hwm_, uint64_t lwm_,
-            class pipe_reader_t **reader_, class pipe_writer_t **writer_);
-        void destroy_pipe (class pipe_t *pipe_);
-        int register_inproc_endpoint (const char *endpoint_,
-            class session_t *session_);
-        class object_t *get_inproc_endpoint (const char *endpoint_);
-        void unregister_inproc_endpoints (class session_t *session_);
         class io_thread_t *choose_io_thread (uint64_t taskset_);
 
         //  Derived object can use these functions to send commands
diff --git a/src/p2p.cpp b/src/p2p.cpp
deleted file mode 100644
index c83d8b1..0000000
--- a/src/p2p.cpp
+++ /dev/null
@@ -1,29 +0,0 @@
-/*
-    Copyright (c) 2007-2009 FastMQ Inc.
-
-    This file is part of 0MQ.
-
-    0MQ is free software; you can redistribute it and/or modify it under
-    the terms of the Lesser GNU General Public License as published by
-    the Free Software Foundation; either version 3 of the License, or
-    (at your option) any later version.
-
-    0MQ 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
-    Lesser GNU General Public License for more details.
-
-    You should have received a copy of the Lesser GNU General Public License
-    along with this program.  If not, see <http://www.gnu.org/licenses/>.
-*/
-
-#include "../include/zmq.h"
-
-#include "p2p.hpp"
-#include "app_thread.hpp"
-#include "session.hpp"
-
-zmq::p2p_t::p2p_t (app_thread_t *thread_, session_t *session_) :
-    socket_base_t (thread_, session_)
-{
-}
diff --git a/src/p2p.hpp b/src/p2p.hpp
deleted file mode 100644
index d3d9dc3..0000000
--- a/src/p2p.hpp
+++ /dev/null
@@ -1,42 +0,0 @@
-/*
-    Copyright (c) 2007-2009 FastMQ Inc.
-
-    This file is part of 0MQ.
-
-    0MQ is free software; you can redistribute it and/or modify it under
-    the terms of the Lesser GNU General Public License as published by
-    the Free Software Foundation; either version 3 of the License, or
-    (at your option) any later version.
-
-    0MQ 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
-    Lesser GNU General Public License for more details.
-
-    You should have received a copy of the Lesser GNU General Public License
-    along with this program.  If not, see <http://www.gnu.org/licenses/>.
-*/
-
-#ifndef __ZMQ_P2P_HPP_INCLUDED__
-#define __ZMQ_P2P_HPP_INCLUDED__
-
-#include "socket_base.hpp"
-
-namespace zmq
-{
-
-    class p2p_t : public socket_base_t
-    {
-    public:
-
-        p2p_t (class app_thread_t *thread_, class session_t *session_);
-
-    private:
-
-        p2p_t (const p2p_t&);
-        void operator = (const p2p_t&);
-    };
-
-}
-
-#endif
diff --git a/src/pipe.cpp b/src/pipe.cpp
deleted file mode 100644
index bf761b4..0000000
--- a/src/pipe.cpp
+++ /dev/null
@@ -1,47 +0,0 @@
-/*
-    Copyright (c) 2007-2009 FastMQ Inc.
-
-    This file is part of 0MQ.
-
-    0MQ is free software; you can redistribute it and/or modify it under
-    the terms of the Lesser GNU General Public License as published by
-    the Free Software Foundation; either version 3 of the License, or
-    (at your option) any later version.
-
-    0MQ 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
-    Lesser GNU General Public License for more details.
-
-    You should have received a copy of the Lesser GNU General Public License
-    along with this program.  If not, see <http://www.gnu.org/licenses/>.
-*/
-
-#include "pipe.hpp"
-
-zmq::pipe_t::pipe_t () :
-    ypipe_t <zmq_msg, false, message_pipe_granularity> (false),
-    index (-1)
-{
-}
-
-zmq::pipe_t::~pipe_t ()
-{
-    //  Flush any outstanding messages to the pipe.
-    flush ();
-
-    //  Deallocate all the messages in the pipe.
-    zmq_msg msg;
-    while (read (&msg))
-        zmq_msg_close (&msg);
-}
-
-void zmq::pipe_t::set_index (int index_)
-{
-    index = index_;
-}
-
-int zmq::pipe_t::get_index ()
-{
-    return index;
-}
diff --git a/src/pipe.hpp b/src/pipe.hpp
index 8894a22..d771120 100644
--- a/src/pipe.hpp
+++ b/src/pipe.hpp
@@ -28,29 +28,10 @@
 namespace zmq
 {
 
-    //  Message pipe. A simple wrapper on top of ypipe.
-
+    //  Message pipe.
     class pipe_t : public ypipe_t <zmq_msg, false, message_pipe_granularity>
     {
-        //  Context is a friend so that it can create & destroy the pipes.
-        //  By making constructor & destructor private we are sure that nobody
-        //  except context messes with pipes.
-        friend class context_t;
-
-    private:
-
-        pipe_t ();
-        ~pipe_t ();
-
-        void set_index (int index_);
-        int get_index ();
-
-        //  Index of the pipe in context's array of pipes.
-        int index;
-
-        pipe_t (const pipe_t&);
-        void operator = (const pipe_t&);
-    }; 
+    };
 
 }
 
diff --git a/src/pipe_reader.cpp b/src/pipe_reader.cpp
deleted file mode 100644
index 79dfe2e..0000000
--- a/src/pipe_reader.cpp
+++ /dev/null
@@ -1,118 +0,0 @@
-/*
-    Copyright (c) 2007-2009 FastMQ Inc.
-
-    This file is part of 0MQ.
-
-    0MQ is free software; you can redistribute it and/or modify it under
-    the terms of the Lesser GNU General Public License as published by
-    the Free Software Foundation; either version 3 of the License, or
-    (at your option) any later version.
-
-    0MQ 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
-    Lesser GNU General Public License for more details.
-
-    You should have received a copy of the Lesser GNU General Public License
-    along with this program.  If not, see <http://www.gnu.org/licenses/>.
-*/
-
-#include "../include/zmq.h"
-
-#include "pipe_reader.hpp"
-#include "pipe.hpp"
-#include "err.hpp"
-#include "i_mux.hpp"
-
-zmq::pipe_reader_t::pipe_reader_t (object_t *parent_, pipe_t *pipe_,
-      uint64_t hwm_, uint64_t lwm_) :
-    object_t (parent_),
-    pipe (pipe_),
-    peer (NULL),
-    mux (NULL),
-    index (-1),
-    hwm (hwm_),
-    lwm (lwm_),
-    head (0),
-    tail (0),
-    last_sent_head (0)
-{
-}
-
-void zmq::pipe_reader_t::set_peer (object_t *peer_)
-{
-    peer = peer_;
-}
-
-zmq::pipe_reader_t::~pipe_reader_t ()
-{
-}
-
-void zmq::pipe_reader_t::set_mux (i_mux *mux_)
-{
-    mux = mux_;
-}
-
-void zmq::pipe_reader_t::set_index (int index_)
-{
-    index = index_;
-}
-
-int zmq::pipe_reader_t::get_index ()
-{
-    return index;
-}
-
-void zmq::pipe_reader_t::process_tail (uint64_t bytes_)
-{
-    tail = bytes_;
-    mux->reactivate (this);
-}
-
-bool zmq::pipe_reader_t::read (struct zmq_msg *msg_)
-{
-    //  Read a message.
-    if (!pipe->read (msg_)) {
-        mux->deactivate (this);
-        return false;
-    }
-
-    //  If successfull, adjust the head of the pipe.
-    head += zmq_msg_size (msg_);
-
-    //  If pipe writer wasn't notified about the head position for long enough,
-    //  notify it.
-    if (head - last_sent_head >= hwm - lwm) {
-        send_head (peer, head);
-        last_sent_head = head;
-    }
-
-    if (zmq_msg_type (msg_) == ZMQ_DELIMITER) {
-
-        //  Detach the pipe from the mux and send termination request to
-        //  the pipe writer.
-        mux->detach_pipe (this);
-        mux = NULL;
-        send_terminate (peer);
-        return false;
-    }
-
-    return true;
-}
-
-void zmq::pipe_reader_t::terminate ()
-{
-    //  Detach the pipe from the mux and send termination request to
-    //  the pipe writer.
-    if (mux) {
-        mux->detach_pipe (this);
-        mux = NULL;
-    }
-    send_terminate (peer);
-}
-
-void zmq::pipe_reader_t::process_terminate_ack ()
-{
-    //  Ask context to deallocate the pipe.
-    destroy_pipe (pipe);
-}
diff --git a/src/pipe_reader.hpp b/src/pipe_reader.hpp
deleted file mode 100644
index cf45bb4..0000000
--- a/src/pipe_reader.hpp
+++ /dev/null
@@ -1,89 +0,0 @@
-/*
-    Copyright (c) 2007-2009 FastMQ Inc.
-
-    This file is part of 0MQ.
-
-    0MQ is free software; you can redistribute it and/or modify it under
-    the terms of the Lesser GNU General Public License as published by
-    the Free Software Foundation; either version 3 of the License, or
-    (at your option) any later version.
-
-    0MQ 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
-    Lesser GNU General Public License for more details.
-
-    You should have received a copy of the Lesser GNU General Public License
-    along with this program.  If not, see <http://www.gnu.org/licenses/>.
-*/
-
-#ifndef __ZMQ_PIPE_READER_HPP_INCLUDED__
-#define __ZMQ_PIPE_READER_HPP_INCLUDED__
-
-#include "object.hpp"
-#include "stdint.hpp"
-
-namespace zmq
-{
-
-    class pipe_reader_t : public object_t
-    {
-        //  Context is a friend so that it can create & destroy the reader.
-        //  By making constructor & destructor private we are sure that nobody
-        //  except context messes with readers.
-        friend class context_t;
-
-    public:
-
-        //  Set & get index in the associated mux object.
-        void set_mux (struct i_mux *mux_);
-        void set_index (int index_);
-        int get_index ();
-
-        //  Reads a message to the underlying pipe.
-        bool read (struct zmq_msg *msg_);
-
-        //  Asks pipe to destroy itself.
-        void terminate ();
-
-    private:
-
-        pipe_reader_t (class object_t *parent_, class pipe_t *pipe_,
-            uint64_t hwm_, uint64_t lwm_);
-        ~pipe_reader_t ();
-
-        //  Second step of reader construction. The parameter cannot be passed
-        //  in constructor as peer object doesn't yet exist at the time.
-        void set_peer (class object_t *peer_);
-
-        void process_tail (uint64_t bytes_);
-        void process_terminate_ack ();
-
-        //  The underlying pipe.
-        class pipe_t *pipe;
-
-        //  Pipe writer associated with the other side of the pipe.
-        class object_t *peer;
-
-        //  Associated mux object.
-        struct i_mux *mux;
-
-        //  Index in the associated mux object.
-        int index;
-
-        //  High and low watermarks for in-memory storage (in bytes).
-        uint64_t hwm;
-        uint64_t lwm;
-
-        //  Positions of head and tail of the pipe (in bytes).
-        uint64_t head;
-        uint64_t tail;
-        uint64_t last_sent_head;
-
-        pipe_reader_t (const pipe_reader_t&);
-        void operator = (const pipe_reader_t&);
-    };
-
-}
-
-#endif
diff --git a/src/pipe_writer.cpp b/src/pipe_writer.cpp
deleted file mode 100644
index a54034b..0000000
--- a/src/pipe_writer.cpp
+++ /dev/null
@@ -1,120 +0,0 @@
-/*
-    Copyright (c) 2007-2009 FastMQ Inc.
-
-    This file is part of 0MQ.
-
-    0MQ is free software; you can redistribute it and/or modify it under
-    the terms of the Lesser GNU General Public License as published by
-    the Free Software Foundation; either version 3 of the License, or
-    (at your option) any later version.
-
-    0MQ 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
-    Lesser GNU General Public License for more details.
-
-    You should have received a copy of the Lesser GNU General Public License
-    along with this program.  If not, see <http://www.gnu.org/licenses/>.
-*/
-
-#include "../include/zmq.h"
-
-#include "pipe_writer.hpp"
-#include "pipe.hpp"
-#include "i_demux.hpp"
-
-zmq::pipe_writer_t::pipe_writer_t (object_t *parent_, pipe_t *pipe_,
-      object_t *peer_, uint64_t hwm_, uint64_t lwm_) :
-    object_t (parent_),
-    pipe (pipe_),
-    peer (peer_),
-    demux (NULL),
-    index (-1),
-    hwm (hwm_),
-    lwm (lwm_),
-    head (0),
-    tail (0)
-{
-}
-
-zmq::pipe_writer_t::~pipe_writer_t ()
-{
-}
-
-void zmq::pipe_writer_t::set_demux (i_demux *demux_)
-{
-    demux = demux_;
-}
-
-void zmq::pipe_writer_t::set_index (int index_)
-{
-    index = index_;
-}
-
-int zmq::pipe_writer_t::get_index ()
-{
-    return index;
-}
-
-bool zmq::pipe_writer_t::write (zmq_msg *msg_)
-{
-    size_t msg_size = zmq_msg_size (msg_);
-
-    //  If message won't fit into the in-memory pipe, there's no way
-    //  to pass it further.
-    //  TODO: It should be discarded and 'oversized' notification should be
-    //        placed into the pipe.
-    zmq_assert (!hwm || msg_size <= hwm);
-
-    //  If there's not enough space in the pipe at the moment, return false.
-    if (hwm && tail + msg_size - head > hwm)
-        return false;
-
-    //  Write the message to the pipe and adjust tail position.
-    pipe->write (*msg_);
-    flush ();
-    tail += msg_size;
-
-    return true;
-}
-
-void zmq::pipe_writer_t::flush ()
-{
-    if (!pipe->flush ())
-        send_tail (peer, tail);
-}
-
-void zmq::pipe_writer_t::process_head (uint64_t bytes_)
-{
-    head = bytes_;
-}
-
-void zmq::pipe_writer_t::terminate ()
-{
-    //  Disconnect from the associated demux.
-    if (demux) {
-        demux->detach_pipe (this);
-        demux = NULL;
-    }
-
-    //  Push the delimiter to the pipe. Delimiter is a notification for pipe
-    //  reader that there will be no more messages in the pipe.
-    zmq_msg delimiter;
-    delimiter.content = (zmq_msg_content*) ZMQ_DELIMITER;
-    delimiter.shared = false;
-    delimiter.vsm_size = 0;
-    pipe->write (delimiter);
-    flush ();
-}
-
-void zmq::pipe_writer_t::process_terminate ()
-{
-    //  Disconnect from the associated demux.
-    if (demux) {
-        demux->detach_pipe (this);
-        demux = NULL;
-    }
-
-    //  Send termination acknowledgement to the pipe reader.
-    send_terminate_ack (peer);
-}
diff --git a/src/pipe_writer.hpp b/src/pipe_writer.hpp
deleted file mode 100644
index a727b1f..0000000
--- a/src/pipe_writer.hpp
+++ /dev/null
@@ -1,88 +0,0 @@
-/*
-    Copyright (c) 2007-2009 FastMQ Inc.
-
-    This file is part of 0MQ.
-
-    0MQ is free software; you can redistribute it and/or modify it under
-    the terms of the Lesser GNU General Public License as published by
-    the Free Software Foundation; either version 3 of the License, or
-    (at your option) any later version.
-
-    0MQ 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
-    Lesser GNU General Public License for more details.
-
-    You should have received a copy of the Lesser GNU General Public License
-    along with this program.  If not, see <http://www.gnu.org/licenses/>.
-*/
-
-#ifndef __ZMQ_PIPE_WRITER_HPP_INCLUDED__
-#define __ZMQ_PIPE_WRITER_HPP_INCLUDED__
-
-#include "object.hpp"
-#include "stdint.hpp"
-
-namespace zmq
-{
-
-    class pipe_writer_t : public object_t
-    {
-        //  Context is a friend so that it can create & destroy the writer.
-        //  By making constructor & destructor private we are sure that nobody
-        //  except context messes with writers.
-        friend class context_t;
-
-    public:
-
-        //  Set & get index in the associated demux object.
-        void set_demux (struct i_demux *demux_);
-        void set_index (int index_);
-        int get_index ();
-
-        //  Writes a message to the underlying pipe. Returns false if the
-        //  message cannot be written to the pipe at the moment.
-        bool write (struct zmq_msg *msg_);
-
-        //  Flush the messages downsteam.
-        void flush ();
-
-        //  Asks pipe to destroy itself.
-        void terminate ();
-
-    private:
-
-        pipe_writer_t (class object_t *parent_, class pipe_t *pipe_,
-            class object_t *peer_, uint64_t hwm_, uint64_t lwm_);
-        ~pipe_writer_t ();
-
-        void process_head (uint64_t bytes_);
-        void process_terminate ();
-
-        //  The underlying pipe.
-        class pipe_t *pipe;
-
-        //  Pipe reader associated with the other side of the pipe.
-        class object_t *peer;
-
-        //  Associated demux object.
-        struct i_demux *demux;
-
-        //  Index in the associated demux object.
-        int index;
-
-        //  High and low watermarks for in-memory storage (in bytes).
-        uint64_t hwm;
-        uint64_t lwm;
-
-        //  Positions of head and tail of the pipe (in bytes).
-        uint64_t head;
-        uint64_t tail;
-
-        pipe_writer_t (const pipe_writer_t&);
-        void operator = (const pipe_writer_t&);
-    };
-
-}
-
-#endif
diff --git a/src/poll.cpp b/src/poll.cpp
index 864cfad..94e4fd4 100644
--- a/src/poll.cpp
+++ b/src/poll.cpp
@@ -50,6 +50,14 @@ zmq::poll_t::poll_t () :
         fd_table [i].index = retired_fd;
 }
 
+zmq::poll_t::~poll_t ()
+{
+    //  Make sure there are no fds registered on shutdown.
+    zmq_assert (load.get () == 0);
+
+    worker.stop ();
+}
+
 zmq::handle_t zmq::poll_t::add_fd (fd_t fd_, i_poll_events *events_)
 {
     pollfd pfd = {fd_, 0, 0};
@@ -132,11 +140,6 @@ void zmq::poll_t::stop ()
     stopping = true;
 }
 
-void zmq::poll_t::join ()
-{
-    worker.stop ();
-}
-
 void zmq::poll_t::loop ()
 {
     while (!stopping) {
diff --git a/src/poll.hpp b/src/poll.hpp
index dbfa776..9fe6067 100644
--- a/src/poll.hpp
+++ b/src/poll.hpp
@@ -47,7 +47,7 @@ namespace zmq
     public:
 
         poll_t ();
-        virtual ~poll_t () {}
+        ~poll_t ();
 
         //  i_poller implementation.
         handle_t add_fd (fd_t fd_, i_poll_events *events_);
@@ -61,7 +61,6 @@ namespace zmq
         int get_load ();
         void start ();
         void stop ();
-        void join ();
 
     private:
 
diff --git a/src/pub.cpp b/src/pub.cpp
deleted file mode 100644
index 5dca0b8..0000000
--- a/src/pub.cpp
+++ /dev/null
@@ -1,38 +0,0 @@
-/*
-    Copyright (c) 2007-2009 FastMQ Inc.
-
-    This file is part of 0MQ.
-
-    0MQ is free software; you can redistribute it and/or modify it under
-    the terms of the Lesser GNU General Public License as published by
-    the Free Software Foundation; either version 3 of the License, or
-    (at your option) any later version.
-
-    0MQ 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
-    Lesser GNU General Public License for more details.
-
-    You should have received a copy of the Lesser GNU General Public License
-    along with this program.  If not, see <http://www.gnu.org/licenses/>.
-*/
-
-#include "../include/zmq.h"
-
-#include "pub.hpp"
-#include "app_thread.hpp"
-#include "session.hpp"
-#include "err.hpp"
-
-zmq::pub_t::pub_t (app_thread_t *thread_, session_t *session_) :
-    socket_base_t (thread_, session_)
-{
-    disable_in ();
-}
-
-int zmq::pub_t::recv (struct zmq_msg *msg_, int flags_)
-{
-    //  Publisher socket has no recv function.
-    errno = ENOTSUP;
-    return -1;
-}
diff --git a/src/pub.hpp b/src/pub.hpp
deleted file mode 100644
index 909e731..0000000
--- a/src/pub.hpp
+++ /dev/null
@@ -1,45 +0,0 @@
-/*
-    Copyright (c) 2007-2009 FastMQ Inc.
-
-    This file is part of 0MQ.
-
-    0MQ is free software; you can redistribute it and/or modify it under
-    the terms of the Lesser GNU General Public License as published by
-    the Free Software Foundation; either version 3 of the License, or
-    (at your option) any later version.
-
-    0MQ 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
-    Lesser GNU General Public License for more details.
-
-    You should have received a copy of the Lesser GNU General Public License
-    along with this program.  If not, see <http://www.gnu.org/licenses/>.
-*/
-
-#ifndef __ZMQ_PUB_HPP_INCLUDED__
-#define __ZMQ_PUB_HPP_INCLUDED__
-
-#include "socket_base.hpp"
-
-namespace zmq
-{
-
-    class pub_t : public socket_base_t
-    {
-    public:
-
-        pub_t (class app_thread_t *thread_, class session_t *session_);
-
-        //  i_api overloads.
-        int recv (struct zmq_msg *msg_, int flags_);
-
-    private:
-
-        pub_t (const pub_t&);
-        void operator = (const pub_t&);
-    };
-
-}
-
-#endif
diff --git a/src/rep.cpp b/src/rep.cpp
deleted file mode 100644
index 60767e1..0000000
--- a/src/rep.cpp
+++ /dev/null
@@ -1,29 +0,0 @@
-/*
-    Copyright (c) 2007-2009 FastMQ Inc.
-
-    This file is part of 0MQ.
-
-    0MQ is free software; you can redistribute it and/or modify it under
-    the terms of the Lesser GNU General Public License as published by
-    the Free Software Foundation; either version 3 of the License, or
-    (at your option) any later version.
-
-    0MQ 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
-    Lesser GNU General Public License for more details.
-
-    You should have received a copy of the Lesser GNU General Public License
-    along with this program.  If not, see <http://www.gnu.org/licenses/>.
-*/
-
-#include "../include/zmq.h"
-
-#include "rep.hpp"
-#include "app_thread.hpp"
-#include "session.hpp"
-
-zmq::rep_t::rep_t (app_thread_t *thread_, session_t *session_) :
-    socket_base_t (thread_, session_)
-{
-}
diff --git a/src/rep.hpp b/src/rep.hpp
deleted file mode 100644
index 92d2758..0000000
--- a/src/rep.hpp
+++ /dev/null
@@ -1,42 +0,0 @@
-/*
-    Copyright (c) 2007-2009 FastMQ Inc.
-
-    This file is part of 0MQ.
-
-    0MQ is free software; you can redistribute it and/or modify it under
-    the terms of the Lesser GNU General Public License as published by
-    the Free Software Foundation; either version 3 of the License, or
-    (at your option) any later version.
-
-    0MQ 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
-    Lesser GNU General Public License for more details.
-
-    You should have received a copy of the Lesser GNU General Public License
-    along with this program.  If not, see <http://www.gnu.org/licenses/>.
-*/
-
-#ifndef __ZMQ_REP_HPP_INCLUDED__
-#define __ZMQ_REP_HPP_INCLUDED__
-
-#include "socket_base.hpp"
-
-namespace zmq
-{
-
-    class rep_t : public socket_base_t
-    {
-    public:
-
-        rep_t (class app_thread_t *thread_, class session_t *session_);
-
-    private:
-
-        rep_t (const rep_t&);
-        void operator = (const rep_t&);
-    };
-
-}
-
-#endif
diff --git a/src/req.hpp b/src/req.hpp
deleted file mode 100644
index c279f0e..0000000
--- a/src/req.hpp
+++ /dev/null
@@ -1,42 +0,0 @@
-/*
-    Copyright (c) 2007-2009 FastMQ Inc.
-
-    This file is part of 0MQ.
-
-    0MQ is free software; you can redistribute it and/or modify it under
-    the terms of the Lesser GNU General Public License as published by
-    the Free Software Foundation; either version 3 of the License, or
-    (at your option) any later version.
-
-    0MQ 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
-    Lesser GNU General Public License for more details.
-
-    You should have received a copy of the Lesser GNU General Public License
-    along with this program.  If not, see <http://www.gnu.org/licenses/>.
-*/
-
-#ifndef __ZMQ_REQ_HPP_INCLUDED__
-#define __ZMQ_REQ_HPP_INCLUDED__
-
-#include "socket_base.hpp"
-
-namespace zmq
-{
-
-    class req_t : public socket_base_t
-    {
-    public:
-
-        req_t (class app_thread_t *thread_, class session_t *session_);
-
-    private:
-
-        req_t (const req_t&);
-        void operator = (const req_t&);
-    };
-
-}
-
-#endif
diff --git a/src/safe_object.cpp b/src/safe_object.cpp
deleted file mode 100644
index d4a92d7..0000000
--- a/src/safe_object.cpp
+++ /dev/null
@@ -1,76 +0,0 @@
-/*
-    Copyright (c) 2007-2009 FastMQ Inc.
-
-    This file is part of 0MQ.
-
-    0MQ is free software; you can redistribute it and/or modify it under
-    the terms of the Lesser GNU General Public License as published by
-    the Free Software Foundation; either version 3 of the License, or
-    (at your option) any later version.
-
-    0MQ 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
-    Lesser GNU General Public License for more details.
-
-    You should have received a copy of the Lesser GNU General Public License
-    along with this program.  If not, see <http://www.gnu.org/licenses/>.
-*/
-
-#include "safe_object.hpp"
-
-zmq::safe_object_t::safe_object_t (class context_t *context_,
-        int thread_slot_) :
-    object_t (context_, thread_slot_),
-    processed_seqnum (0),
-    terminating (false)
-{
-}
-
-zmq::safe_object_t::safe_object_t (object_t *parent_) :
-    object_t (parent_),
-    processed_seqnum (0),
-    terminating (false)
-{
-}
-
-void zmq::safe_object_t::inc_seqnum ()
-{
-    //  This function is called from the sender thread to ensure that this
-    //  object will still exist when the command sent to it arrives in the
-    //  destination thread.
-    sent_seqnum.add (1);
-}
-
-void zmq::safe_object_t::process_command (struct command_t &cmd_)
-{
-    object_t::process_command (cmd_);
-
-    //  Adjust sequence number of the last processed command.
-    processed_seqnum++;
-
-    //  If we are already in the termination phase and all commands sent to
-    //  this object are processed, it's safe to deallocate it.
-    if (terminating && sent_seqnum.get () == processed_seqnum)
-        delete this;
-}
-
-void zmq::safe_object_t::terminate ()
-{
-    //  Wait till all commands sent to this session are processed.
-    terminating = true;
-
-    //  If there's no pending command we can deallocate the session
-    //  straight saway.
-    if (sent_seqnum.get () == processed_seqnum)
-        delete this;
-}
-
-bool zmq::safe_object_t::is_terminating ()
-{
-    return terminating;
-}
-
-zmq::safe_object_t::~safe_object_t ()
-{
-}
diff --git a/src/safe_object.hpp b/src/safe_object.hpp
deleted file mode 100644
index b47db48..0000000
--- a/src/safe_object.hpp
+++ /dev/null
@@ -1,68 +0,0 @@
-/*
-    Copyright (c) 2007-2009 FastMQ Inc.
-
-    This file is part of 0MQ.
-
-    0MQ is free software; you can redistribute it and/or modify it under
-    the terms of the Lesser GNU General Public License as published by
-    the Free Software Foundation; either version 3 of the License, or
-    (at your option) any later version.
-
-    0MQ 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
-    Lesser GNU General Public License for more details.
-
-    You should have received a copy of the Lesser GNU General Public License
-    along with this program.  If not, see <http://www.gnu.org/licenses/>.
-*/
-
-#ifndef __ZMQ_SAFE_OBJECT_HPP_INCLUDED__
-#define __ZMQ_SAFE_OBJECT_HPP_INCLUDED__
-
-#include "object.hpp"
-#include "atomic_counter.hpp"	
-
-namespace zmq
-{
-
-    //  Same as object_t with the exception of termination mechanism. While
-    //  object_t is destroyed immediately on terminate (assuming that the caller
-    //  have ensured that there are no more commands for the object on the
-    //  fly), safe_object_t switches into termination mode and waits for all
-    //  the on-the-fly commands to be delivered before it deallocates itself.
-
-    class safe_object_t : public object_t
-    {
-    public:
-
-        safe_object_t (class context_t *context_, int thread_slot_);
-        safe_object_t (object_t *parent_);
-
-        void inc_seqnum ();
-        void process_command (struct command_t &cmd_);
-
-    protected:
-
-        void terminate ();
-        bool is_terminating ();
-
-        virtual ~safe_object_t ();
-
-    private:
-
-        //  Sequence number of the last command sent to the object and last
-        //  command processed by the object. The former is an atomic counter
-        //  meaning that other threads can increment it safely.
-        atomic_counter_t sent_seqnum;
-        uint32_t processed_seqnum;
-
-        bool terminating;
-
-        safe_object_t (const safe_object_t&);
-        void operator = (const safe_object_t&);
-    };
-
-}
-
-#endif
diff --git a/src/select.cpp b/src/select.cpp
index 68ec9a0..f10acdc 100644
--- a/src/select.cpp
+++ b/src/select.cpp
@@ -51,6 +51,14 @@ zmq::select_t::select_t () :
     FD_ZERO (&source_set_err);
 }
 
+zmq::select_t::~select_t ()
+{
+    //  Make sure there are no fds registered on shutdown.
+    zmq_assert (load.get () == 0);
+
+    worker.stop ();
+}
+
 zmq::handle_t zmq::select_t::add_fd (fd_t fd_, i_poll_events *events_)
 {
     //  Store the file descriptor.
@@ -156,11 +164,6 @@ void zmq::select_t::stop ()
     stopping = true;
 }
 
-void zmq::select_t::join ()
-{
-    worker.stop ();
-}
-
 void zmq::select_t::loop ()
 {
     while (!stopping) {
diff --git a/src/select.hpp b/src/select.hpp
index c1e72a7..c442477 100644
--- a/src/select.hpp
+++ b/src/select.hpp
@@ -50,6 +50,7 @@ namespace zmq
     public:
 
         select_t ();
+        ~select_t ();
 
         //  i_poller implementation.
         handle_t add_fd (fd_t fd_, i_poll_events *events_);
@@ -63,7 +64,6 @@ namespace zmq
         int get_load ();
         void start ();
         void stop ();
-        void join ();
 
     private:
 
diff --git a/src/session.cpp b/src/session.cpp
deleted file mode 100644
index b9a450d..0000000
--- a/src/session.cpp
+++ /dev/null
@@ -1,273 +0,0 @@
-/*
-    Copyright (c) 2007-2009 FastMQ Inc.
-
-    This file is part of 0MQ.
-
-    0MQ is free software; you can redistribute it and/or modify it under
-    the terms of the Lesser GNU General Public License as published by
-    the Free Software Foundation; either version 3 of the License, or
-    (at your option) any later version.
-
-    0MQ 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
-    Lesser GNU General Public License for more details.
-
-    You should have received a copy of the Lesser GNU General Public License
-    along with this program.  If not, see <http://www.gnu.org/licenses/>.
-*/
-
-#include "../include/zmq.h"
-
-#include "session.hpp"
-#include "i_engine.hpp"
-#include "i_thread.hpp"
-#include "i_mux.hpp"
-#include "i_demux.hpp"
-#include "err.hpp"
-#include "pipe.hpp"
-#include "pipe_reader.hpp"
-#include "pipe_writer.hpp"
-#include "simple_semaphore.hpp"
-
-zmq::session_t::session_t (object_t *parent_, i_thread *thread_,
-      i_mux *mux_, i_demux *demux_,
-      bool terminate_on_disconnect_, bool terminate_on_no_pipes_) :
-    safe_object_t (parent_),
-    mux (mux_),
-    demux (demux_),
-    thread (thread_),
-    engine (NULL),
-    terminate_on_disconnect (terminate_on_disconnect_),
-    terminate_on_no_pipes (false),
-    terminate_on_no_pipes_delayed (terminate_on_no_pipes_),
-    index (-1)
-{
-    //  At least one way to terminate the session should be allowed. Otherwise
-    //  the session can be orphaned forever.
-    zmq_assert (terminate_on_disconnect || terminate_on_no_pipes_delayed);
-
-    //  Give the mux and the demux callback pointer to ourselves.
-    if (mux)
-        mux->set_session (this);
-    if (demux)
-        demux->set_session (this);
-}
-
-void zmq::session_t::shutdown ()
-{
-    //  Session may live even without an associated engine, thus we have
-    //  to check if for NULL value.
-    if (engine)
-        engine->shutdown ();
-
-    //  Propagate the shutdown signal to both inbound and outbound pipes.
-    if (mux)
-        mux->shutdown ();
-    if (demux)
-        demux->shutdown ();
-
-    delete this;
-}
-
-void zmq::session_t::disconnected ()
-{
-    //  It's engine who calls this function so there's no need to deallocate
-    //  the engine. Just drop the reference.
-    engine = NULL;
-
-    //  Some sessions won't shut down because of disconnect. New engine will
-    //  attached to the session later on.
-    if (!terminate_on_disconnect)
-        return;
-
-    terminate ();
-}
-
-void zmq::session_t::bind (object_t *peer_, bool in_, bool out_)
-{
-    //  Create the out pipe (if required).
-    pipe_reader_t *pipe_reader = NULL;
-    if (out_) {
-        pipe_writer_t *pipe_writer;
-        create_pipe (peer_, this, 0, 0, &pipe_reader, &pipe_writer);
-        demux->attach_pipe (pipe_writer);
-
-        //  There's at least one pipe attached. We can deallocate the object
-        //  when there are no pipes (if required).
-        terminate_on_no_pipes = terminate_on_no_pipes_delayed;
-    }
-
-    //  Ask peer to attach to the out pipe (if one exists). If required, ask
-    //  it to create a pipe in opposite direction. It's assumed that peer's
-    //  seqnum was already incremented, so we don't need to care whether it's
-    //  alive at the moment.
-    if (in_)
-        inc_seqnum ();
-    send_bind (peer_, pipe_reader, in_ ? this : NULL);
-}
-
-void zmq::session_t::revive ()
-{
-    if (engine)
-        engine->revive ();
-}
-
-void zmq::session_t::terminate ()
-{
-    //  Terminate is always called by engine, thus it'll terminate itself,
-    //  we just have to drop the pointer.
-    engine = NULL;
-
-    //  Propagate the terminate signal to both inbound and outbound pipes.
-    if (mux) {
-        mux->terminate ();
-        mux = NULL;
-    }
-    if (demux) {
-        demux->terminate ();
-        demux = NULL;
-    }
-
-    //  Session cannot be deallocated at this point. There can still be
-    //  pending commands to process. Unregister session from global
-    //  repository thus ensuring that no new commands will be sent.
-    unregister_inproc_endpoints (this);
-
-    //  Move to terminating state.
-    safe_object_t::terminate ();
-}
-
-zmq::session_t::~session_t ()
-{
-    //  When session is actually deallocated it unregisters from its thread.
-    //  Unregistration cannot be done earlier as it would result in memory
-    //  leak if global shutdown happens in the middle of session termination.
-    thread->detach_session (this);
-}
-
-void zmq::session_t::set_engine (i_engine *engine_)
-{
-    zmq_assert (!engine || !engine_);
-    engine = engine_;
-}
-
-void zmq::session_t::set_index (int index_)
-{
-    index = index_;
-}
-
-int zmq::session_t::get_index ()
-{
-    return index;
-}
-
-bool zmq::session_t::write (zmq_msg *msg_)
-{
-    return demux->send (msg_);
-}
-
-void zmq::session_t::flush ()
-{
-    demux->flush ();
-}
-
-bool zmq::session_t::read (zmq_msg *msg_)
-{
-    bool retrieved = mux->recv (msg_);
-    if (terminate_on_no_pipes && mux->empty () && demux->empty ()) {
-        zmq_assert (engine);
-        engine->schedule_terminate ();
-        terminate ();
-    }
-    return retrieved;
-}
-
-void zmq::session_t::process_bind (pipe_reader_t *reader_, session_t *peer_)
-{
-    if (is_terminating ()) {
-
-        //  If session is already in termination phase, we'll ask newly arrived
-        //  pipe reader & writer to terminate straight away.
-        if (reader_)
-            reader_->terminate ();
-
-        //  Peer session has already incremented its seqnum. We have to send
-        //  a dummy command to avoid a memory leak.
-        if (peer_)
-            send_bind (peer_, NULL, NULL);
-
-        return;
-    }
-
-    //  If inbound pipe is provided, bind it to the mux.
-    if (reader_) {
-        mux->attach_pipe (reader_);
-
-        //  There's at least one pipe attached. We can deallocate the object
-        //  when there are no pipes (if required).
-        terminate_on_no_pipes = terminate_on_no_pipes_delayed;
-    }
-
-    //  If peer wants to get messages from ourselves, we'll bind to it.
-    if (peer_) {
-        pipe_reader_t *pipe_reader;
-        pipe_writer_t *pipe_writer;
-        create_pipe (peer_, this, 0, 0, &pipe_reader, &pipe_writer);
-        demux->attach_pipe (pipe_writer);
-        send_bind (peer_, pipe_reader, NULL);
-
-        //  There's at least one pipe attached. We can deallocate the object
-        //  when there are no pipes (if required).
-        terminate_on_no_pipes = terminate_on_no_pipes_delayed;
-    }
-}
-
-void zmq::session_t::process_reg (simple_semaphore_t *smph_)
-{
-    zmq_assert (!is_terminating ());
-
-    //  Add the session to the list of sessions associated with this I/O thread.
-    //  This way the session will be deallocated on the terminal shutdown.
-    thread->attach_session (this);
-
-    //  Release calling thead (if required).
-    if (smph_)
-        smph_->post ();
-}
-
-void zmq::session_t::process_reg_and_bind (session_t *peer_,
-    bool flow_in_, bool flow_out_)
-{
-    zmq_assert (!is_terminating ());
-
-    //  Add the session to the list of sessions associated with this I/O thread.
-    //  This way the session will be deallocated on the terminal shutdown.
-    thread->attach_session (this);
-
-    //  Bind to the peer. Note that caller have already incremented command
-    //  sequence number of the peer so we are sure it still exists.
-    pipe_reader_t *pipe_reader = NULL;
-    if (flow_out_) {
-        pipe_writer_t *pipe_writer;
-        create_pipe (peer_, this, 0, 0, &pipe_reader, &pipe_writer);
-        demux->attach_pipe (pipe_writer);
-
-        //  There's at least one pipe attached. We can deallocate the object
-        //  when there are no pipes (if required).
-        terminate_on_no_pipes = terminate_on_no_pipes_delayed;
-    }
-    send_bind (peer_, pipe_reader, flow_in_ ? this : NULL);
-}
-
-void zmq::session_t::process_engine (i_engine *engine_)
-{
-    if (is_terminating ()) {
-
-        //  Kill the engine. It won't be needed anymore.
-        engine_->terminate ();
-        return;
-    }
-
-    engine_->attach (thread->get_poller (), this);
-}
diff --git a/src/session.hpp b/src/session.hpp
deleted file mode 100644
index 855dd1d..0000000
--- a/src/session.hpp
+++ /dev/null
@@ -1,107 +0,0 @@
-/*
-    Copyright (c) 2007-2009 FastMQ Inc.
-
-    This file is part of 0MQ.
-
-    0MQ is free software; you can redistribute it and/or modify it under
-    the terms of the Lesser GNU General Public License as published by
-    the Free Software Foundation; either version 3 of the License, or
-    (at your option) any later version.
-
-    0MQ 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
-    Lesser GNU General Public License for more details.
-
-    You should have received a copy of the Lesser GNU General Public License
-    along with this program.  If not, see <http://www.gnu.org/licenses/>.
-*/
-
-#ifndef __ZMQ_SESSION_HPP_INCLUDED__
-#define __ZMQ_SESSION_HPP_INCLUDED__
-
-#include "i_session.hpp"
-#include "safe_object.hpp"
-#include "stdint.hpp"
-#include "atomic_counter.hpp"
-
-namespace zmq
-{
-
-    //  Object that encapsulates both mux and demux.
-
-    class session_t : public safe_object_t, public i_session
-    {
-    public:
-
-        //  Creates the session object.
-        session_t (struct object_t *parent_, struct i_thread *thread_,
-            struct i_mux *mux_, struct i_demux *demux_,
-            bool terminate_on_disconnect_, bool terminate_on_no_pipes_);
-
-        //  i_session implementation
-        void set_engine (struct i_engine *engine_);
-        void shutdown ();
-        bool read (struct zmq_msg *msg_);
-        bool write (struct zmq_msg *msg_);
-        void flush ();
-
-        //  Called by the engine when it is being closed.
-        void disconnected ();
-
-        //  Creates a message flow between this session and the peer session.
-        //  If in_ is true, the messages can flow from the peer to ourselves.
-        //  If out_ is true, messages can flow from ourselves to the peer.
-        //  It's assumed that peer's seqnum was already incremented.
-        void bind (class object_t *peer_, bool in_, bool out_);
-
-        //  Called by mux if new messages are available.
-        void revive ();
-
-        //  Functions to set & retrieve index of this MD in thread's array
-        //  of session objects.
-        void set_index (int index_);
-        int get_index ();
-
-    private:
-
-        //  Clean-up.
-        ~session_t ();
-
-        //  Terminate is private here. It is called by either when disconnected
-        //  or no_pipes event occurs.
-        void terminate ();
-
-        void process_bind (class pipe_reader_t *reader_,
-            class session_t *peer_);
-        void process_reg (class simple_semaphore_t *smph_);
-        void process_reg_and_bind (class session_t *peer_,
-            bool flow_in_, bool flow_out_);
-        void process_engine (i_engine *engine_);
-
-        struct i_mux *mux;
-        struct i_demux *demux;
-
-        struct i_thread *thread;
-        struct i_engine *engine;
-
-        //  If true termination of the session can be triggered by engine
-        //  disconnect/close.
-        bool terminate_on_disconnect;
-
-        //  If true termination of the session can be triggered when the last
-        //  pipe detaches from it.
-        bool terminate_on_no_pipes;
-
-        //  If true, terminate_on_no_pipes should be set when at least one
-        //  pipe was bound.
-        bool terminate_on_no_pipes_delayed;
-
-        //  Index in thread's session array.
-        int index;
-    };
-
-}
-
-#endif
-
diff --git a/src/session_stub.cpp b/src/session_stub.cpp
deleted file mode 100644
index 152b9fb..0000000
--- a/src/session_stub.cpp
+++ /dev/null
@@ -1,110 +0,0 @@
-/*
-    Copyright (c) 2007-2009 FastMQ Inc.
-
-    This file is part of 0MQ.
-
-    0MQ is free software; you can redistribute it and/or modify it under
-    the terms of the Lesser GNU General Public License as published by
-    the Free Software Foundation; either version 3 of the License, or
-    (at your option) any later version.
-
-    0MQ 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
-    Lesser GNU General Public License for more details.
-
-    You should have received a copy of the Lesser GNU General Public License
-    along with this program.  If not, see <http://www.gnu.org/licenses/>.
-*/
-
-#include <string>
-
-#include "../include/zmq.h"
-
-#include "session_stub.hpp"
-#include "i_engine.hpp"
-#include "listener.hpp"
-#include "err.hpp"
-
-zmq::session_stub_t::session_stub_t (listener_t *listener_) :
-    state (reading_identity),
-    engine (NULL),
-    listener (listener_),
-    index (-1)
-{
-}
-
-void zmq::session_stub_t::terminate ()
-{
-    if (engine)
-        engine->terminate ();
-    delete this;
-}
-
-void zmq::session_stub_t::shutdown ()
-{
-    if (engine)
-        engine->shutdown ();
-    delete this;
-}
-
-zmq::session_stub_t::~session_stub_t ()
-{
-}
-
-void zmq::session_stub_t::set_engine (i_engine *engine_)
-{
-    zmq_assert (!engine_ || !engine);
-    engine = engine_;
-}
-
-bool zmq::session_stub_t::read (struct zmq_msg *msg_)
-{
-    //  No messages are sent to the connecting peer.
-    return false;
-}
-
-bool zmq::session_stub_t::write (struct zmq_msg *msg_)
-{
-    //  The first message arrived is the connection identity.
-    if (state == reading_identity) {
-        identity = std::string ((const char*) zmq_msg_data (msg_),
-            zmq_msg_size (msg_));
-        state = has_identity;
-        return true;
-    }
-
-    //  We are not interested in any subsequent messages.
-    return false;
-}
-
-void zmq::session_stub_t::flush ()
-{
-    //  We have the identity. At this point we can find the correct session and
-    //  attach it to the connection.
-    if (state == has_identity) {
-
-        //  At this point the stub will be deleted. Return immediately without
-        //  touching 'this' pointer.
-        listener->got_identity (this, identity.c_str ());
-        return;
-    }
-}
-
-zmq::i_engine *zmq::session_stub_t::detach_engine ()
-{
-    //  Ask engine to unregister from the poller.
-    i_engine *e = engine;
-    engine->detach ();
-    return e;
-}
-
-void zmq::session_stub_t::set_index (int index_)
-{
-    index = index_;
-}
-
-int zmq::session_stub_t::get_index ()
-{
-    return index;
-}
diff --git a/src/session_stub.hpp b/src/session_stub.hpp
deleted file mode 100644
index 4499e45..0000000
--- a/src/session_stub.hpp
+++ /dev/null
@@ -1,83 +0,0 @@
-/*
-    Copyright (c) 2007-2009 FastMQ Inc.
-
-    This file is part of 0MQ.
-
-    0MQ is free software; you can redistribute it and/or modify it under
-    the terms of the Lesser GNU General Public License as published by
-    the Free Software Foundation; either version 3 of the License, or
-    (at your option) any later version.
-
-    0MQ 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
-    Lesser GNU General Public License for more details.
-
-    You should have received a copy of the Lesser GNU General Public License
-    along with this program.  If not, see <http://www.gnu.org/licenses/>.
-*/
-
-#ifndef __ZMQ_SESSION_STUB_HPP_INCLUDED__
-#define __ZMQ_SESSION_STUB_HPP_INCLUDED__
-
-#include <string>
-
-#include "i_session.hpp"
-
-namespace zmq
-{
-
-    //  This class is used instead of regular session till the identity of
-    //  incomming connection is established and connection is attached
-    //  to corresponding session.
-
-    class session_stub_t : public i_session
-    {
-    public:
-
-        session_stub_t (class listener_t *listener_);
-
-        //  i_session implementation.
-        void set_engine (struct i_engine *engine_);
-        void terminate ();
-        void shutdown ();
-        bool read (struct zmq_msg *msg_);
-        bool write (struct zmq_msg *msg_);
-        void flush ();
-
-        //  Detaches engine from the stub. Returns it to the caller.
-        struct i_engine *detach_engine ();
-
-        //  Manipulate stubs's index in listener's array of stubs.
-        void set_index (int index_);
-        int get_index ();
-
-    private:
-
-        //  Clean-up.
-        virtual ~session_stub_t ();
-
-        enum {
-            reading_identity,
-            has_identity
-        } state;
-
-        //  Reference to the associated engine.
-        i_engine *engine;
-
-        //  Reference to the listener object that owns this stub.
-        class listener_t *listener;
-
-        //  Index of the stub in listener's array of stubs.
-        int index;
-
-        //  Identity of the connection.
-        std::string identity;
-
-        session_stub_t (const session_stub_t&);
-        void operator = (const session_stub_t&);
-    };
-
-}
-
-#endif
diff --git a/src/socket_base.cpp b/src/socket_base.cpp
deleted file mode 100644
index 6718244..0000000
--- a/src/socket_base.cpp
+++ /dev/null
@@ -1,267 +0,0 @@
-/*
-    Copyright (c) 2007-2009 FastMQ Inc.
-
-    This file is part of 0MQ.
-
-    0MQ is free software; you can redistribute it and/or modify it under
-    the terms of the Lesser GNU General Public License as published by
-    the Free Software Foundation; either version 3 of the License, or
-    (at your option) any later version.
-
-    0MQ 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
-    Lesser GNU General Public License for more details.
-
-    You should have received a copy of the Lesser GNU General Public License
-    along with this program.  If not, see <http://www.gnu.org/licenses/>.
-*/
-
-#include <string>
-
-#include "../include/zmq.h"
-
-#include "socket_base.hpp"
-#include "app_thread.hpp"
-#include "err.hpp"
-#include "listener.hpp"
-#include "connecter.hpp"
-#include "simple_semaphore.hpp"
-#include "io_thread.hpp"
-#include "io_object.hpp"
-#include "session.hpp"
-#include "dummy_aggregator.hpp"
-#include "dummy_distributor.hpp"
-
-zmq::socket_base_t::socket_base_t (app_thread_t *thread_, session_t *session_) :
-    object_t (thread_),
-    thread (thread_),
-    session (session_),
-    has_in (true),
-    has_out (true)
-{
-    session->set_engine (this);
-}
-
-void zmq::socket_base_t::shutdown ()
-{
-    //  Destroy all the I/O objects created from this socket.
-    for (io_objects_t::size_type i = 0; i != io_objects.size (); i++)
-        io_objects [i]->shutdown ();
-
-    delete this;
-}
-
-void zmq::socket_base_t::schedule_terminate ()
-{
-    //  Terminate is never scheduled on socket engines.
-    zmq_assert (false);
-}
-
-void zmq::socket_base_t::terminate ()
-{
-    //  Destroy all the I/O objects created from this socket.
-    //  First unregister the object from I/O thread, then terminate it in
-    //  this application thread.
-    simple_semaphore_t smph;
-    for (io_objects_t::size_type i = 0; i != io_objects.size (); i++) {
-        send_unreg (io_objects [i], &smph);
-        smph.wait ();
-        io_objects [i]->terminate ();
-    }
-
-    zmq_assert (session);
-    session->disconnected ();
-
-    delete this;
-}
-
-zmq::socket_base_t::~socket_base_t ()
-{
-}
-
-void zmq::socket_base_t::disable_in ()
-{
-    has_in = false;
-}
-
-void zmq::socket_base_t::disable_out ()
-{
-    has_out = false;
-}
-
-int zmq::socket_base_t::bind (const char *addr_, zmq_opts *opts_)
-{
-    thread->process_commands (false);
-
-    std::string addr (addr_);
-    std::string::size_type pos = addr.find ("://");
-    if (pos == std::string::npos || addr.substr (0, pos) == "zmq.tcp") {
-
-        //  Choose the I/O thread with the least load, create the listener.
-        //  Note that same taskset is used to choose the I/O thread to handle
-        //  the listening socket and newly created connections.
-        //  Note that has_in and has_out are twisted at this place - listener
-        //  is going to create peer objects, so the message flows are viewed
-        //  from the opposite direction.
-        io_thread_t *io_thread = choose_io_thread (opts_ ? opts_->taskset : 0);
-        listener_t *listener = new listener_t (io_thread, addr_, session,
-            has_out, has_in, opts_ ? opts_->taskset : 0);
-
-        //  Ask it to start interacting with the I/O thread.
-        simple_semaphore_t smph;
-        send_reg (listener, &smph);
-
-        //  Store the reference to the listener so that it can be terminated
-        //  when the socket is closed.
-        io_objects.push_back (listener);
-
-        //  Wait while listener is actually registered with the I/O thread.
-        smph.wait ();
-
-        return 0;
-    }
-    else if (addr.substr (0, pos) == "inproc") {
-
-        //  For inproc transport the only thing we have to do is to register
-        //  this socket as an inproc endpoint with the supplied name.
-        return register_inproc_endpoint (addr.substr (pos + 3).c_str (),
-            session);
-    }
-    else {
-
-        //  Unknown protocol requested.
-        errno = EINVAL;
-        return -1;
-    }
-}
-
-int zmq::socket_base_t::connect (const char *addr_, zmq_opts *opts_)
-{
-    thread->process_commands (false);
-
-    std::string addr (addr_);
-    std::string::size_type pos = addr.find ("://");
-    if (pos == std::string::npos || addr.substr (0, pos) == "zmq.tcp") {
-
-        //  Choose the I/O thread with the least load, create the connecter and
-        //  session.
-        io_thread_t *io_thread = choose_io_thread (opts_ ? opts_->taskset : 0);
-        i_mux *mux = new dummy_aggregator_t;
-        zmq_assert (mux);
-        i_demux *demux = new dummy_distributor_t;
-        zmq_assert (demux);
-        session_t *peer = new session_t (io_thread, io_thread, mux, demux,
-            false, true);
-        zmq_assert (peer);
-        connecter_t *connecter = new connecter_t (io_thread, addr_, peer);
-        zmq_assert (connecter);
-
-        //  Increment session's command sequence number so that it won't get
-        //  deallocated till the subsequent bind command arrives.
-        peer->inc_seqnum ();
-
-        //  Register the connecter (and session) with its I/O thread.
-        simple_semaphore_t smph;
-        send_reg (connecter, &smph);
-
-        //  Store the reference to the connecter so that it can be terminated
-        //  when the socket is closed.
-        io_objects.push_back (connecter);
-
-        //  Wait till registration succeeds.
-        smph.wait ();
-
-        //  Bind local session with the connecter's session so that messages
-        //  can flow in both directions.
-        session->bind (peer, has_in, has_out);
-
-        return 0;
-    }
-    else if (addr.substr (0, pos) == "inproc") {
-
-        //  Get the MD responsible for the address. In case of invalid address
-        //  return error.
-        object_t *peer = get_inproc_endpoint (addr.substr (pos + 3).c_str ());
-        if (!peer) {
-            errno = EADDRNOTAVAIL;
-            return -1;
-        }
-
-        //  Create bidirectional message pipes between this session and
-        //  the peer session.
-        session->bind (peer, has_in, has_out);
-
-        return 0;
-    }
-    else {
-
-        //  Unknown protocol requested.
-        errno = EINVAL;
-        return -1;
-    }
-}
-
-int zmq::socket_base_t::subscribe (const char *criteria_)
-{
-    //  No implementation at the moment...
-    errno = ENOTSUP;
-    return -1;
-}
-
-int zmq::socket_base_t::send (zmq_msg *msg_, int flags_)
-{
-    thread->process_commands (false);
-    while (true) {
-        if (session->write (msg_))
-            return 0;
-        if (flags_ & ZMQ_NOBLOCK) {
-            errno = EAGAIN;
-            return -1;
-        }
-        thread->process_commands (true);
-    }
-}
-
-int zmq::socket_base_t::flush ()
-{
-    thread->process_commands (false);
-    session->flush ();
-    return 0;
-}
-
-int zmq::socket_base_t::recv (zmq_msg *msg_, int flags_)
-{
-    thread->process_commands (false);
-    while (true) {
-        if (session->read (msg_))
-            return 0;
-        if (flags_ & ZMQ_NOBLOCK) {
-            errno = EAGAIN;
-            return -1;
-        }
-        thread->process_commands (true);
-    }
-}
-
-int zmq::socket_base_t::close ()
-{
-    terminate ();
-    return 0;
-}
-
-void zmq::socket_base_t::attach (struct i_poller *poller_, i_session *session_)
-{
-    zmq_assert (false);
-}
-
-void zmq::socket_base_t::detach ()
-{
-    zmq_assert (false);
-}
-
-void zmq::socket_base_t::revive ()
-{
-    //  We can ignore the event safely here.
-}
-
diff --git a/src/socket_base.hpp b/src/socket_base.hpp
deleted file mode 100644
index c1de8e6..0000000
--- a/src/socket_base.hpp
+++ /dev/null
@@ -1,96 +0,0 @@
-/*
-    Copyright (c) 2007-2009 FastMQ Inc.
-
-    This file is part of 0MQ.
-
-    0MQ is free software; you can redistribute it and/or modify it under
-    the terms of the Lesser GNU General Public License as published by
-    the Free Software Foundation; either version 3 of the License, or
-    (at your option) any later version.
-
-    0MQ 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
-    Lesser GNU General Public License for more details.
-
-    You should have received a copy of the Lesser GNU General Public License
-    along with this program.  If not, see <http://www.gnu.org/licenses/>.
-*/
-
-#ifndef __ZMQ_SOCKET_BASE_HPP_INCLUDED__
-#define __ZMQ_SOCKET_BASE_HPP_INCLUDED__
-
-#include <vector>
-
-#include "i_engine.hpp"
-#include "i_api.hpp"
-#include "object.hpp"
-
-namespace zmq
-{
-
-    class socket_base_t : public object_t, public i_engine, public i_api
-    {
-    public:
-
-        //  TODO: Possibly, session can be attached to the engine using
-        //  attach function.
-        socket_base_t (class app_thread_t *thread_, class session_t *session_);
-
-        //  i_engine interface implementation.
-        void attach (struct i_poller *poller_, struct i_session *session_);
-        void detach ();
-        void revive ();
-        void schedule_terminate ();
-        void terminate ();
-        void shutdown ();
-
-        // i_api interface implementation.
-        int bind (const char *addr_, struct zmq_opts *opts_);
-        int connect (const char *addr_, struct zmq_opts *opts_);
-        int subscribe (const char *criteria_);
-        int send (struct zmq_msg *msg_, int flags_);
-        int flush ();
-        int recv (struct zmq_msg *msg_, int flags_);
-        int close ();
-
-    protected:
-
-        //  Clean-up. The function has to be protected rather than private,
-        //  otherwise auto-generated destructor in derived classes
-        //  cannot be compiled. It has to be virtual so that socket_base_t's
-        //  terminate & shutdown functions deallocate correct type of object.
-        virtual ~socket_base_t ();
-
-        //  By default, socket is able to pass messages in both inward and
-        //  outward directions. By calling these functions, particular
-        //  socket type is able to eliminate one direction.
-        void disable_in ();
-        void disable_out ();
-
-    private:
-
-        //  Pointer to the application thread the socket belongs to.
-        class app_thread_t *thread;
-
-        //  Pointer to the associated session object.
-        class session_t *session;
-
-        //  List of I/O object created via this socket. These have to be shut
-        //  down when the socket is closed.
-        typedef std::vector <class io_object_t*> io_objects_t;
-        io_objects_t io_objects;
-
-        //  If true, socket creates inbound pipe when binding to an engine.
-        bool has_in;
-
-        //  If true, socket creates outbound pipe when binding to an engine.
-        bool has_out;
-
-        socket_base_t (const socket_base_t&);
-        void operator = (const socket_base_t&);
-    };
-
-}
-
-#endif
diff --git a/src/sub.cpp b/src/sub.cpp
deleted file mode 100644
index 3d1d578..0000000
--- a/src/sub.cpp
+++ /dev/null
@@ -1,45 +0,0 @@
-/*
-    Copyright (c) 2007-2009 FastMQ Inc.
-
-    This file is part of 0MQ.
-
-    0MQ is free software; you can redistribute it and/or modify it under
-    the terms of the Lesser GNU General Public License as published by
-    the Free Software Foundation; either version 3 of the License, or
-    (at your option) any later version.
-
-    0MQ 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
-    Lesser GNU General Public License for more details.
-
-    You should have received a copy of the Lesser GNU General Public License
-    along with this program.  If not, see <http://www.gnu.org/licenses/>.
-*/
-
-#include "../include/zmq.h"
-
-#include "sub.hpp"
-#include "app_thread.hpp"
-#include "session.hpp"
-#include "err.hpp"
-
-zmq::sub_t::sub_t (app_thread_t *thread_, session_t *session_) :
-    socket_base_t (thread_, session_)
-{
-    disable_out ();
-}
-
-int zmq::sub_t::send (struct zmq_msg *msg_, int flags_)
-{
-    //  Subscriber socket has no send function.
-    errno = ENOTSUP;
-    return -1;
-}
-
-int zmq::sub_t::flush ()
-{
-    //  Subscriber socket has no flush function.
-    errno = ENOTSUP;
-    return -1;
-}
diff --git a/src/sub.hpp b/src/sub.hpp
deleted file mode 100644
index f3e23c1..0000000
--- a/src/sub.hpp
+++ /dev/null
@@ -1,46 +0,0 @@
-/*
-    Copyright (c) 2007-2009 FastMQ Inc.
-
-    This file is part of 0MQ.
-
-    0MQ is free software; you can redistribute it and/or modify it under
-    the terms of the Lesser GNU General Public License as published by
-    the Free Software Foundation; either version 3 of the License, or
-    (at your option) any later version.
-
-    0MQ 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
-    Lesser GNU General Public License for more details.
-
-    You should have received a copy of the Lesser GNU General Public License
-    along with this program.  If not, see <http://www.gnu.org/licenses/>.
-*/
-
-#ifndef __ZMQ_SUB_HPP_INCLUDED__
-#define __ZMQ_SUB_HPP_INCLUDED__
-
-#include "socket_base.hpp"
-
-namespace zmq
-{
-
-    class sub_t : public socket_base_t
-    {
-    public:
-
-        sub_t (class app_thread_t *thread_, class session_t *session_);
-
-        //  i_api overloads.
-        int send (struct zmq_msg *msg_, int flags_);
-        int flush ();
-
-    private:
-
-        sub_t (const sub_t&);
-        void operator = (const sub_t&);
-    };
-
-}
-
-#endif
diff --git a/src/zmq.cpp b/src/zmq.cpp
index 0fb6fe1..d19b229 100644
--- a/src/zmq.cpp
+++ b/src/zmq.cpp
@@ -176,7 +176,7 @@ void *zmq_init (int app_threads_, int io_threads_)
 
 int zmq_term (void *context_)
 {
-    ((zmq::context_t*) context_)->shutdown ();
+    delete (zmq::context_t*) context_;
     return 0;
 }
 
diff --git a/src/zmq_decoder.cpp b/src/zmq_decoder.cpp
deleted file mode 100644
index 0c491ea..0000000
--- a/src/zmq_decoder.cpp
+++ /dev/null
@@ -1,79 +0,0 @@
-/*
-    Copyright (c) 2007-2009 FastMQ Inc.
-
-    This file is part of 0MQ.
-
-    0MQ is free software; you can redistribute it and/or modify it under
-    the terms of the Lesser GNU General Public License as published by
-    the Free Software Foundation; either version 3 of the License, or
-    (at your option) any later version.
-
-    0MQ 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
-    Lesser GNU General Public License for more details.
-
-    You should have received a copy of the Lesser GNU General Public License
-    along with this program.  If not, see <http://www.gnu.org/licenses/>.
-*/
-
-#include "zmq_decoder.hpp"
-#include "i_session.hpp"
-#include "wire.hpp"
-
-zmq::zmq_decoder_t::zmq_decoder_t () :
-    destination (NULL)
-{
-    zmq_msg_init (&in_progress);
-
-    //  At the beginning, read one byte and go to one_byte_size_ready state.
-    next_step (tmpbuf, 1, &zmq_decoder_t::one_byte_size_ready);
-}
-
-zmq::zmq_decoder_t::~zmq_decoder_t ()
-{
-    zmq_msg_close (&in_progress);
-}
-
-void zmq::zmq_decoder_t::set_session (i_session *destination_)
-{
-    destination = destination_;
-}
-
-bool zmq::zmq_decoder_t::one_byte_size_ready ()
-{
-    //  First byte of size is read. If it is 0xff read 8-byte size.
-    //  Otherwise allocate the buffer for message data and read the
-    //  message data into it.
-    if (*tmpbuf == 0xff)
-        next_step (tmpbuf, 8, &zmq_decoder_t::eight_byte_size_ready);
-    else {
-        zmq_msg_init_size (&in_progress, *tmpbuf);
-        next_step (zmq_msg_data (&in_progress), *tmpbuf,
-            &zmq_decoder_t::message_ready);
-    }
-    return true;
-}
-
-bool zmq::zmq_decoder_t::eight_byte_size_ready ()
-{
-    //  8-byte size is read. Allocate the buffer for message body and
-    //  read the message data into it.
-    size_t size = (size_t) get_uint64 (tmpbuf);
-    zmq_msg_init_size (&in_progress, size);
-    next_step (zmq_msg_data (&in_progress), size,
-        &zmq_decoder_t::message_ready);
-    return true;
-}
-
-bool zmq::zmq_decoder_t::message_ready ()
-{
-    //  Message is completely read. Push it further and start reading
-    //  new message.
-    if (!destination->write (&in_progress))
-        return false;
-
-    next_step (tmpbuf, 1, &zmq_decoder_t::one_byte_size_ready);
-    return true;
-}
-
diff --git a/src/zmq_decoder.hpp b/src/zmq_decoder.hpp
deleted file mode 100644
index f648819..0000000
--- a/src/zmq_decoder.hpp
+++ /dev/null
@@ -1,57 +0,0 @@
-/*
-    Copyright (c) 2007-2009 FastMQ Inc.
-
-    This file is part of 0MQ.
-
-    0MQ is free software; you can redistribute it and/or modify it under
-    the terms of the Lesser GNU General Public License as published by
-    the Free Software Foundation; either version 3 of the License, or
-    (at your option) any later version.
-
-    0MQ 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
-    Lesser GNU General Public License for more details.
-
-    You should have received a copy of the Lesser GNU General Public License
-    along with this program.  If not, see <http://www.gnu.org/licenses/>.
-*/
-
-
-#ifndef __ZMQ_ZMQ_DECODER_HPP_INCLUDED__
-#define __ZMQ_ZMQ_DECODER_HPP_INCLUDED__
-
-#include "../include/zmq.h"
-
-#include "decoder.hpp"
-
-namespace zmq
-{
-    //  Decoder for 0MQ backend protocol. Converts data batches into messages.
-
-    class zmq_decoder_t : public decoder_t <zmq_decoder_t>
-    {
-    public:
-
-        zmq_decoder_t ();
-        ~zmq_decoder_t ();
-
-        void set_session (struct i_session *destination_);
-
-    private:
-
-        bool one_byte_size_ready ();
-        bool eight_byte_size_ready ();
-        bool message_ready ();
-
-        struct i_session *destination;
-        unsigned char tmpbuf [8];
-        ::zmq_msg in_progress;
-
-        zmq_decoder_t (const zmq_decoder_t&);
-        void operator = (const zmq_decoder_t&);
-    };
-
-}
-
-#endif
diff --git a/src/zmq_encoder.cpp b/src/zmq_encoder.cpp
deleted file mode 100644
index 8a713cf..0000000
--- a/src/zmq_encoder.cpp
+++ /dev/null
@@ -1,75 +0,0 @@
-/*
-    Copyright (c) 2007-2009 FastMQ Inc.
-
-    This file is part of 0MQ.
-
-    0MQ is free software; you can redistribute it and/or modify it under
-    the terms of the Lesser GNU General Public License as published by
-    the Free Software Foundation; either version 3 of the License, or
-    (at your option) any later version.
-
-    0MQ 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
-    Lesser GNU General Public License for more details.
-
-    You should have received a copy of the Lesser GNU General Public License
-    along with this program.  If not, see <http://www.gnu.org/licenses/>.
-*/
-
-#include "zmq_encoder.hpp"
-#include "i_session.hpp"
-#include "wire.hpp"
-
-zmq::zmq_encoder_t::zmq_encoder_t () :
-    source (NULL)
-{
-    zmq_msg_init (&in_progress);
-
-    //  Write 0 bytes to the batch and go to message_ready state.
-    next_step (NULL, 0, &zmq_encoder_t::message_ready, true);
-}
-
-zmq::zmq_encoder_t::~zmq_encoder_t ()
-{
-    zmq_msg_close (&in_progress);
-}
-
-void zmq::zmq_encoder_t::set_session (i_session *source_)
-{
-    source = source_;
-}
-
-bool zmq::zmq_encoder_t::size_ready ()
-{
-    //  Write message body into the buffer.
-    next_step (zmq_msg_data (&in_progress), zmq_msg_size (&in_progress),
-        &zmq_encoder_t::message_ready, false);
-    return true;
-}
-
-bool zmq::zmq_encoder_t::message_ready ()
-{
-    //  Read new message from the dispatcher. If there is none, return false.
-    //  Note that new state is set only if write is successful. That way
-    //  unsuccessful write will cause retry on the next state machine
-    //  invocation.
-    if (!source->read (&in_progress)) {
-        return false;
-    }
-    size_t size = zmq_msg_size (&in_progress);
-
-    //  For messages less than 255 bytes long, write one byte of message size.
-    //  For longer messages write 0xff escape character followed by 8-byte
-    //  message size.
-    if (size < 255) {
-        tmpbuf [0] = (unsigned char) size;
-        next_step (tmpbuf, 1, &zmq_encoder_t::size_ready, true);
-    }
-    else {
-        tmpbuf [0] = 0xff;
-        put_uint64 (tmpbuf + 1, size);
-        next_step (tmpbuf, 9, &zmq_encoder_t::size_ready, true);
-    }
-    return true;
-}
diff --git a/src/zmq_encoder.hpp b/src/zmq_encoder.hpp
deleted file mode 100644
index 829fd4b..0000000
--- a/src/zmq_encoder.hpp
+++ /dev/null
@@ -1,54 +0,0 @@
-/*
-    Copyright (c) 2007-2009 FastMQ Inc.
-
-    This file is part of 0MQ.
-
-    0MQ is free software; you can redistribute it and/or modify it under
-    the terms of the Lesser GNU General Public License as published by
-    the Free Software Foundation; either version 3 of the License, or
-    (at your option) any later version.
-
-    0MQ 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
-    Lesser GNU General Public License for more details.
-
-    You should have received a copy of the Lesser GNU General Public License
-    along with this program.  If not, see <http://www.gnu.org/licenses/>.
-*/
-
-#ifndef __ZMQ_ZMQ_ENCODER_HPP_INCLUDED__
-#define __ZMQ_ZMQ_ENCODER_HPP_INCLUDED__
-
-#include "../include/zmq.h"
-
-#include "encoder.hpp"
-
-namespace zmq
-{
-    //  Encoder for 0MQ backend protocol. Converts messages into data batches.
-
-    class zmq_encoder_t : public encoder_t <zmq_encoder_t>
-    {
-    public:
-
-        zmq_encoder_t ();
-        ~zmq_encoder_t ();
-
-        void set_session (struct i_session *source_);
-
-    private:
-
-        bool size_ready ();
-        bool message_ready ();
-
-        struct i_session *source;
-        ::zmq_msg in_progress;
-        unsigned char tmpbuf [9];
-
-        zmq_encoder_t (const zmq_encoder_t&);
-        void operator = (const zmq_encoder_t&);
-    };
-}
-
-#endif
diff --git a/src/zmq_tcp_engine.cpp b/src/zmq_tcp_engine.cpp
deleted file mode 100644
index 6091d80..0000000
--- a/src/zmq_tcp_engine.cpp
+++ /dev/null
@@ -1,185 +0,0 @@
-/*
-    Copyright (c) 2007-2009 FastMQ Inc.
-
-    This file is part of 0MQ.
-
-    0MQ is free software; you can redistribute it and/or modify it under
-    the terms of the Lesser GNU General Public License as published by
-    the Free Software Foundation; either version 3 of the License, or
-    (at your option) any later version.
-
-    0MQ 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
-    Lesser GNU General Public License for more details.
-
-    You should have received a copy of the Lesser GNU General Public License
-    along with this program.  If not, see <http://www.gnu.org/licenses/>.
-*/
-
-#include "zmq_tcp_engine.hpp"
-#include "config.hpp"
-#include "i_session.hpp"
-#include "err.hpp"
-
-zmq::zmq_tcp_engine_t::zmq_tcp_engine_t (fd_t fd_) :
-    poller (NULL),
-    session (NULL),
-    terminating (false),
-    insize (0),
-    inpos (0),
-    outsize (0),
-    outpos (0)
-{
-    //  Allocate read & write buffer.
-    inbuf = new unsigned char [in_batch_size];
-    zmq_assert (inbuf);
-    outbuf = new unsigned char [out_batch_size];
-    zmq_assert (outbuf);
-
-    //  Attach the socket.
-    int rc = socket.open (fd_);
-    zmq_assert (rc == 0);
-}
-
-void zmq::zmq_tcp_engine_t::attach (i_poller *poller_, i_session *session_)
-{
-    zmq_assert (!poller);
-    poller = poller_;
-    zmq_assert (!session);
-    session = session_;
-    encoder.set_session (session);
-    decoder.set_session (session);
-
-    //  Let session know we are here.
-    session->set_engine (this);
-
-    //  Register the engine with the polling thread.
-    handle = poller->add_fd (socket.get_fd (), this);
-    poller->set_pollin (handle);
-    poller->set_pollout (handle);
-
-    //  Flush any pending inbound messages.
-    in_event ();
-}
-
-void zmq::zmq_tcp_engine_t::detach ()
-{
-    zmq_assert (poller);
-    poller->rm_fd (handle);
-    poller = NULL;
-    zmq_assert (session);
-    session->set_engine (NULL);
-    session = NULL;
-    encoder.set_session (NULL);
-    decoder.set_session (NULL);
-}
-
-void zmq::zmq_tcp_engine_t::revive ()
-{
-    zmq_assert (poller);
-    poller->set_pollout (handle);
-}
-
-void zmq::zmq_tcp_engine_t::schedule_terminate ()
-{
-    terminating = true;
-}
-
-void zmq::zmq_tcp_engine_t::terminate ()
-{
-    delete this;
-}
-
-void zmq::zmq_tcp_engine_t::shutdown ()
-{
-    delete this;
-}
-
-zmq::zmq_tcp_engine_t::~zmq_tcp_engine_t ()
-{
-    detach ();
-    delete [] outbuf;
-    delete [] inbuf;
-}
-
-void zmq::zmq_tcp_engine_t::in_event ()
-{
-    //  If there's no data to process in the buffer, read new data.
-    if (inpos == insize) {
-
-        //  Read as much data as possible to the read buffer.
-        insize = socket.read (inbuf, in_batch_size);
-        inpos = 0;
-
-        //  Check whether the peer has closed the connection.
-        if (insize == -1) {
-            insize = 0;
-            error ();
-            return;
-        }
-    }
-
-    //  Following code should be executed even if there's not a single byte in
-    //  the buffer. There still can be a decoded messages stored in the decoder.
-
-    //  Push the data to the decoder.
-    int nbytes = decoder.write (inbuf + inpos, insize - inpos);
-
-    //  Adjust read position. Stop polling for input if we got stuck.
-    inpos += nbytes;
-    if (inpos < insize)
-        poller->reset_pollin (handle);
-
-    //  If at least one byte was processed, flush all messages the decoder
-    //  may have produced. If engine is disconnected from session, no need
-    //  to flush the messages. It's important that flush is called at the
-    //  very end of in_event as it may invoke in_event itself.
-    if (nbytes > 0 && session)
-        session->flush ();
-}
-
-void zmq::zmq_tcp_engine_t::out_event ()
-{
-    //  If write buffer is empty, try to read new data from the encoder.
-    if (outpos == outsize) {
-
-        outsize = encoder.read (outbuf, out_batch_size);
-        outpos = 0;
-
-        //  If there are no more pipes, engine can be deallocated.
-        if (terminating) {
-            terminate ();
-            return;
-        }
-
-        //  If there is no data to send, stop polling for output.
-        if (outsize == 0)
-            poller->reset_pollout (handle);
-    }
-
-    //  If there are any data to write in write buffer, write as much as
-    //  possible to the socket.
-    if (outpos < outsize) {
-        int nbytes = socket.write (outbuf + outpos, outsize - outpos);
-
-        //  Handle problems with the connection.
-        if (nbytes == -1) {
-            error ();
-            return;
-        }
-
-        outpos += nbytes;
-    }
-}
-
-void zmq::zmq_tcp_engine_t::timer_event ()
-{
-    zmq_assert (false);
-}
-
-void zmq::zmq_tcp_engine_t::error ()
-{
-    zmq_assert (false);
-}
-
diff --git a/src/zmq_tcp_engine.hpp b/src/zmq_tcp_engine.hpp
deleted file mode 100644
index 6a83cec..0000000
--- a/src/zmq_tcp_engine.hpp
+++ /dev/null
@@ -1,92 +0,0 @@
-/*
-    Copyright (c) 2007-2009 FastMQ Inc.
-
-    This file is part of 0MQ.
-
-    0MQ is free software; you can redistribute it and/or modify it under
-    the terms of the Lesser GNU General Public License as published by
-    the Free Software Foundation; either version 3 of the License, or
-    (at your option) any later version.
-
-    0MQ 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
-    Lesser GNU General Public License for more details.
-
-    You should have received a copy of the Lesser GNU General Public License
-    along with this program.  If not, see <http://www.gnu.org/licenses/>.
-*/
-
-#ifndef __ZMQ_ZMQ_TCP_ENGINE_HPP_INCLUDED__
-#define __ZMQ_ZMQ_TCP_ENGINE_HPP_INCLUDED__
-
-#include "i_engine.hpp"
-#include "i_poller.hpp"
-#include "i_poll_events.hpp"
-#include "fd.hpp"
-#include "tcp_socket.hpp"
-#include "zmq_encoder.hpp"
-#include "zmq_decoder.hpp"
-
-namespace zmq
-{
-
-    class zmq_tcp_engine_t : public i_engine, public i_poll_events
-    {
-    public:
-
-        zmq_tcp_engine_t (fd_t fd_);
-
-        //  i_engine implementation.
-        void attach (struct i_poller *poller_, struct i_session *session_);
-        void detach ();
-        void revive ();
-        void schedule_terminate ();
-        void terminate ();
-        void shutdown ();
-
-        //  i_poll_events implementation.
-        void in_event ();
-        void out_event ();
-        void timer_event ();
-
-    private:
-
-        void error ();
-
-        //  Clean-up.
-        ~zmq_tcp_engine_t ();
-
-        //  The underlying TCP socket.
-        tcp_socket_t socket;
-
-        //  Handle associated with the socket.
-        handle_t handle;
-
-        //  I/O thread that the engine runs in.
-        i_poller *poller;
-
-        //  Reference to the associated session object.
-        i_session *session;
-
-        //  If true, engine should terminate itself as soon as possible.
-        bool terminating;
-
-        unsigned char *inbuf;
-        int insize;
-        int inpos;
-
-        unsigned char *outbuf;
-        int outsize;
-        int outpos;
-
-        zmq_encoder_t encoder;
-        zmq_decoder_t decoder;
-
-        zmq_tcp_engine_t (const zmq_tcp_engine_t&);
-        void operator = (const zmq_tcp_engine_t&);
-    };
-
-}
-
-#endif