listener/connecter/init/session added

25 files changed, 1069 insertions, 454 deletions

diff --git a/src/Makefile.am b/src/Makefile.am
index 8524067..9d6127c 100644
--- a/src/Makefile.am
+++ b/src/Makefile.am
@@ -15,6 +15,7 @@ libzmq_la_SOURCES = \
     err.hpp \
     fd.hpp \
     fd_signaler.hpp \
+    i_inout.hpp \
     io_object.hpp \
     io_thread.hpp \
     ip.hpp \
@@ -25,10 +26,13 @@ libzmq_la_SOURCES = \
     msg.hpp \
     mutex.hpp \
     object.hpp \
+    options.hpp \
+    owner.hpp \
     pipe.hpp \
     platform.hpp \
     poll.hpp \
     select.hpp \
+    session.hpp \
     simple_semaphore.hpp \
     socket_base.hpp \
     stdint.hpp \
@@ -43,7 +47,10 @@ libzmq_la_SOURCES = \
     ypollset.hpp \
     yqueue.hpp \
     zmq_connecter.hpp \
+    zmq_decoder.hpp \
+    zmq_encoder.hpp \
     zmq_engine.hpp \
+    zmq_init.hpp \
     zmq_listener.hpp \
     app_thread.cpp \
     devpoll.cpp \
@@ -56,8 +63,11 @@ libzmq_la_SOURCES = \
     ip.cpp \
     kqueue.cpp \
     object.cpp \
+    options.cpp \
+    owned.cpp \
     poll.cpp \
     select.cpp \
+    session.cpp \
     socket_base.cpp \
     tcp_connecter.cpp \
     tcp_listener.cpp \
@@ -67,7 +77,10 @@ libzmq_la_SOURCES = \
     ypollset.cpp \
     zmq.cpp \
     zmq_connecter.cpp \
+    zmq_decoder.cpp \
+    zmq_encoder.cpp \
     zmq_engine.cpp \
+    zmq_init.cpp \
     zmq_listener.cpp
 
 libzmq_la_LDFLAGS = -version-info 0:0:0
diff --git a/src/atomic.hpp b/src/atomic.hpp
deleted file mode 100644
index e581593..0000000
--- a/src/atomic.hpp
+++ /dev/null
@@ -1,310 +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_ATOMIC_HPP_INCLUDED__
-#define __ZMQ_ATOMIC_HPP_INCLUDED__
-
-#include "stdint.hpp"
-
-#if defined ZMQ_FORCE_MUTEXES
-#define ZMQ_ATOMIC_MUTEX
-#elif (defined __i386__ || defined __x86_64__) && defined __GNUC__
-#define ZMQ_ATOMIC_X86
-#elif defined ZMQ_HAVE_WINDOWS
-#define ZMQ_ATOMIC_WINDOWS
-#elif defined ZMQ_HAVE_SOLARIS
-#define ZMQ_ATOMIC_SOLARIS
-#else
-#define ZMQ_ATOMIC_MUTEX
-#endif
-
-namespace zmq
-{
-
-    //  Atomic assignement.
-    inline void atomic_uint32_set (volatile uint32_t *p_, uint32_t value_)
-    {
-        *p_ = value_;
-        //  StoreLoad memory barrier should go here on platforms with
-        //  memory models that require it.
-    }
-
-    //  Atomic retrieval of an integer.
-    inline uint32_t atomic_uint32_get (volatile uint32_t *p_)
-    {
-        //  StoreLoad memory barrier should go here on platforms with
-        //  memory models that require it.
-        return *p_;
-    }
-
-    //  Atomic addition. Returns the old value.
-    inline uint32_t atomic_uint32_add (volatile uint32_t *p_, uint32_t delta_)
-    {
-#if defined ZMQ_ATOMIC_WINDOWS
-        return InterlockedExchangeAdd ((LONG*) &value, increment_);
-#elif defined ZMQ_ATOMIC_SOLARIS
-        return atomic_add_32_nv (&value, increment_) - delta_;
-#elif defined ZMQ_ATOMIC_X86
-        uint32_t old;
-        __asm__ volatile (
-            "lock; xadd %0, %1\n\t"
-            : "=r" (old), "=m" (*p_)
-            : "0" (delta_), "m" (*p_)
-            : "cc", "memory");
-        return old;
-#else
-#error // TODO: 
-        sync.lock ();
-        uint32_t old = *p_;
-        *p_ += delta_;
-        sync.unlock ();
-#endif
-    }
-
-    //  Atomic subtraction. Returns the old value.
-    inline uint32_t atomic_uint32_sub (volatile uint32_t *p_, uint32_t delta_)
-    {
-#if defined ZMQ_ATOMIC_WINDOWS
-        LONG delta = - ((LONG) delta_);
-        return InterlockedExchangeAdd ((LONG*) &value, delta);
-#elif defined ZMQ_ATOMIC_SOLARIS
-        int32_t delta = - ((int32_t) delta_);
-        return atomic_add_32_nv (&value, delta) + delta_;
-#elif defined ZMQ_ATOMIC_X86
-        uint32_t old = -delta_;
-        __asm__ volatile ("lock; xaddl %0,%1"
-            : "=r" (old), "=m" (*p_)
-            : "0" (old), "m" (*p_)
-            : "cc");
-        return old;
-#else
-#error // TODO: 
-        sync.lock ();
-        uint32_t old = *p_;
-        *p_ -= delta_;
-        sync.unlock ();
-        return old;
-#endif
-    }
-
-    //  Atomic assignement.
-    template <typename T>
-    inline void atomic_ptr_set (volatile T **p_, T *value_)
-    {
-        *p_ = value_;
-        //  StoreLoad memory barrier should go here on platforms with
-        //  memory models that require it.
-    }
-
-    //  Perform atomic 'exchange pointers' operation. Old value is returned.
-    template <typename T>
-    inline void *atomic_ptr_xchg (volatile T **p_, T *value_)
-    {
-#if defined ZMQ_ATOMIC_WINDOWS
-        return InterlockedExchangePointer (p_, value_);
-#elif defined ZMQ_ATOMIC_SOLARIS
-        return atomic_swap_ptr (p_, value_);
-#elif defined ZMQ_ATOMIC_X86
-        void *old;
-        __asm__ volatile (
-            "lock; xchg %0, %2"
-            : "=r" (old), "=m" (*p_)
-            : "m" (*p_), "0" (value_));
-        return old;
-#else
-#error //  TODO:
-        sync.lock ();
-        void *old = *p_;
-        *p_ = value_;
-        sync.unlock ();
-        return old;
-#endif
-    }
-
-    //  Perform atomic 'compare and swap' operation on the pointer.
-    //  The pointer is compared to 'cmp' argument and if they are
-    //  equal, its value is set to 'value'. Old value of the pointer
-    //  is returned.
-    template <typename T>
-    inline void *atomic_ptr_cas (volatile T **p_, T *cmp_, T *value_)
-    {
-#if defined ZMQ_ATOMIC_WINDOWS
-        return InterlockedCompareExchangePointer (p_, value_, cmp_);
-#elif defined ZMQ_ATOMIC_SOLARIS
-        return atomic_cas_ptr (p_, cmp_, value_);
-#elif defined ZMQ_ATOMIC_X86
-        void *old;
-        __asm__ volatile (
-            "lock; cmpxchg %2, %3"
-            : "=a" (old), "=m" (*p_)
-            : "r" (value_), "m" (*p_), "0" (cmp_)
-            : "cc");
-        return old;
-#else
-#error //  TODO:
-        sync.lock ();
-        void *old = *p_;
-        if (old == cmp_)
-            *p_ = value_;
-        sync.unlock ();
-        return old;
-#endif
-    }
-
-#if defined ZMQ_ATOMIC_X86 && defined __x86_64__
-    typedef uint64_t atomic_bitmap_t;
-#else
-    typedef uint32_t atomic_bitmap_t;
-#endif
-
-    //  Atomic assignement.
-    inline void atomic_bitmap_set (volatile atomic_bitmap_t *p_,
-        atomic_bitmap_t value_)
-    {
-        *p_ = value_;
-        //  StoreLoad memory barrier should go here on platforms with
-        //  memory models that require it.
-    }
-
-    //  Bit-test-set-and-reset. Sets one bit of the value and resets
-    //  another one. Returns the original value of the reset bit.
-    inline bool atomic_bitmap_btsr (volatile atomic_bitmap_t *p_,
-        int set_index_, int reset_index_)
-    {
-#if defined ZMQ_ATOMIC_WINDOWS
-        while (true) {
-            atomic_bitmap_t oldval = *p_;
-            atomic_bitmap_t newval = (oldval | (atomic_bitmap_t (1) <<
-                set_index_)) & ~(integer_t (1) << reset_index_);
-            if (InterlockedCompareExchange ((volatile LONG*) p_, newval,
-                  oldval) == (LONG) oldval)
-                return (oldval & (atomic_bitmap_t (1) << reset_index_)) ?
-                    true : false; 
-        }
-#elif defined ZMQ_ATOMIC_SOLARIS
-        while (true) {
-            atomic_bitmap_t oldval = *p_;
-            atomic_bitmap_t newval = (oldval | (atomic_bitmap_t (1) <<
-                set_index_)) & ~(integer_t (1) << reset_index_);
-            if (atomic_cas_32 (p_, oldval, newval) == oldval)
-                return (oldval & (atomic_bitmap_t (1) << reset_index_)) ?
-                    true : false; 
-        }
-#elif defined ZMQ_ATOMIC_X86
-        atomic_bitmap_t oldval, dummy;
-        __asm__ volatile (
-            "mov %0, %1\n\t"
-            "1:"
-            "mov %1, %2\n\t"
-            "bts %3, %2\n\t"
-            "btr %4, %2\n\t"
-            "lock cmpxchg %2, %0\n\t"
-            "jnz 1b\n\t"
-            : "+m" (*p_), "=&a" (oldval), "=&r" (dummy)
-            : "r" (atomic_bitmap_t (set_index_)),
-                "r" (atomic_bitmap_t (reset_index_))
-            : "cc");
-        return (bool) (oldval & (atomic_bitmap_t (1) << reset_index_)); 
-#else
-#error // TODO:
-        sync.lock ();
-        atomic_bitmap_t oldval = *p_;
-        *p_ = (oldval | (atomic_bitmap_t (1) << set_index_)) &
-            ~(atomic_bitmap_t (1) << reset_index_);
-        sync.unlock ();
-        return (oldval & (atomic_bitmap_t (1) << reset_index_)) ? true : false;
-#endif
-    }
-
-    //  Sets value to newval. Returns the original value.
-    inline atomic_bitmap_t atomic_bitmap_xchg (volatile atomic_bitmap_t *p_,
-        atomic_bitmap_t newval_)
-    {
-#if defined ZMQ_ATOMIC_WINDOWS
-        return InterlockedExchange ((volatile LONG*) p_, newval_);
-#elif defined ZMQ_ATOMIC_SOLARIS
-        return atomic_swap_32 (p_, newval_);
-#elif defined ZMQ_ATOMIC_X86
-        atomic_bitmap_t oldval = newval_;
-        __asm__ volatile (
-            "lock; xchg %0, %1"
-            : "=r" (oldval)
-            : "m" (*p_), "0" (oldval)
-            : "memory");
-        return oldval; 
-#else
-#error //  TODO:
-        sync.lock ();
-        atomic_bitmap_t oldval = *p_;
-        *p_ = newval_;
-        sync.unlock ();
-#endif
-    }
-
-    //  izte is "if-zero-then-else" atomic operation - if the value is zero
-    //  it substitutes it by 'thenval' else it rewrites it by 'elseval'.
-    //  Original value of the integer is returned from this function.
-    inline atomic_bitmap_t atomic_bitmap_izte (volatile atomic_bitmap_t *p_,
-            atomic_bitmap_t thenval_,  atomic_bitmap_t elseval_)
-    {
-#if defined ZMQ_ATOMIC_WINDOWS
-        while (true) {
-            atomic_bitmap_t oldval = *p_;
-            atomic_bitmap_t newval = (oldval ? elseval_ : thenval_); 
-            if (InterlockedCompareExchange ((volatile LONG*) p_, newval,
-                  oldval) == (LONG) oldval)
-                return oldval;
-        }
-#elif defined ZMQ_ATOMIC_SOLARIS
-        while (true) {
-            atomic_bitmap_t oldval = *p_;
-            atomic_bitmap_t newval = (oldval ? elseval_ : thenval_); 
-            if (atomic_cas_32 (p_, oldval, newval) == oldval)
-                return oldval;
-        }
-#elif defined ZMQ_ATOMIC_X86
-        atomic_bitmap_t oldval;
-        atomic_bitmap_t dummy;
-        __asm__ volatile (
-            "mov %0, %1\n\t"
-            "1:"
-            "mov %3, %2\n\t"
-            "test %1, %1\n\t"
-            "jz 2f\n\t"
-            "mov %4, %2\n\t"
-            "2:"
-            "lock cmpxchg %2, %0\n\t"
-            "jnz 1b\n\t"
-            : "+m" (*p_), "=&a" (oldval), "=&r" (dummy)
-            : "r" (thenval_), "r" (elseval_)
-            : "cc");
-        return oldval;
-#else
-#error //  TODO:
-        sync.lock ();
-        atomic_bitmap_t oldval = *p_;
-        *p_ = oldval ? elseval_ : thenval_;
-        sync.unlock ();
-        return oldval;
-#endif
-    }
-
-}
-
-#endif
diff --git a/src/i_inout.hpp b/src/i_inout.hpp
new file mode 100644
index 0000000..be2e007
--- /dev/null
+++ b/src/i_inout.hpp
@@ -0,0 +1,37 @@
+/*
+    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_INOUT_HPP_INCLUDED__
+#define __ZMQ_I_INOUT_HPP_INCLUDED__
+
+#include "../include/zmq.h"
+
+namespace zmq
+{
+
+    struct i_inout
+    {
+        virtual bool read (::zmq_msg *msg_) = 0;
+        virtual bool write (::zmq_msg *msg_) = 0;
+        virtual void flush () = 0;
+    };
+
+}
+
+#endif
diff --git a/src/io_object.cpp b/src/io_object.cpp
index d8cc1c0..f61e5f0 100644
--- a/src/io_object.cpp
+++ b/src/io_object.cpp
@@ -21,35 +21,19 @@
 #include "io_thread.hpp"
 #include "err.hpp"
 
-zmq::io_object_t::io_object_t (io_thread_t *parent_, object_t *owner_) :
-    object_t (parent_),
-    owner (owner_),
-    plugged_in (false),
-    terminated (false)
+zmq::io_object_t::io_object_t (io_thread_t *io_thread_)
 {
     //  Retrieve the poller from the thread we are running in.
-    poller = parent_->get_poller ();
+    poller = io_thread_->get_poller ();
 }
 
 zmq::io_object_t::~io_object_t ()
 {
 }
 
-void zmq::io_object_t::process_plug ()
+void zmq::io_object_t::set_io_thread (io_thread_t *io_thread_)
 {
-    zmq_assert (!plugged_in);
-
-    //  If termination of the object was already requested, destroy it and
-    //  send the termination acknowledgement.
-    if (terminated) {
-        send_term_ack (owner);
-        delete this;
-        return;
-    }
-
-    //  Notify the generic termination mechanism (io_object_t) that the object
-    //  is already plugged in.
-    plugged_in = true;
+    poller = io_thread_->get_poller ();
 }
 
 zmq::handle_t zmq::io_object_t::add_fd (fd_t fd_)
@@ -106,26 +90,3 @@ void zmq::io_object_t::timer_event ()
 {
     zmq_assert (false);
 }
-
-void zmq::io_object_t::term ()
-{
-    send_term_req (owner, this);
-}
-
-void zmq::io_object_t::process_term ()
-{
-    zmq_assert (!terminated);
-
-    //  If termination request has occured even before the object was plugged in
-    //  wait till plugging in happens, then acknowledge the termination.
-    if (!plugged_in) {
-        terminated = true;
-        return;
-    }
-
-    //  Otherwise, destroy the object and acknowledge the termination
-    //  straight away.
-    send_term_ack (owner);
-    process_unplug ();
-    delete this;
-}
diff --git a/src/io_object.hpp b/src/io_object.hpp
index 4f323ad..e5582db 100644
--- a/src/io_object.hpp
+++ b/src/io_object.hpp
@@ -20,41 +20,31 @@
 #ifndef __ZMQ_IO_OBJECT_HPP_INCLUDED__
 #define __ZMQ_IO_OBJECT_HPP_INCLUDED__
 
-#include "object.hpp"
+#include <stddef.h>
+
 #include "i_poller.hpp"
 #include "i_poll_events.hpp"
 
 namespace zmq
 {
 
-    class io_object_t : public object_t, public i_poll_events
+    //  Simple base class for objects that live in I/O threads.
+    //  It makes communication with the poller object easier and
+    //  makes defining unneeded event handlers unnecessary.
+
+    class io_object_t : public i_poll_events
     {
     public:
 
-        //  I/O object will live in the thread inherited from the parent.
-        //  However, it's lifetime is managed by the owner.
-        io_object_t (class io_thread_t *parent_, object_t *owner_);
+        io_object_t (class io_thread_t *io_thread_ = NULL);
+        ~io_object_t ();
 
     protected:
 
-        //  Ask owner socket to terminate this I/O object. This may not happen
-        void term ();
-
-        //  I/O object destroys itself. No point in allowing others to invoke
-        //  the destructor. At the same time, it has to be virtual so that
-        //  generic io_object deallocation mechanism destroys specific type
-        //  of I/O object correctly.
-        virtual ~io_object_t ();
-
-        //  Handlers for incoming commands. It vital that every I/O object
-        //  invokes io_object_t::process_plug at the end of it's own plug
-        //  handler.
-        void process_plug ();
-
-        //  io_object_t defines a new handler used to disconnect the object
-        //  from the poller object. Implement the handlen in the derived
-        //  classes to ensure sane cleanup.
-        virtual void process_unplug () = 0;
+        //  Derived class can init/swap the underlying I/O thread.
+        //  Caution: Remove all the file descriptors from the old I/O thread
+        //  before swapping to the new one!
+        void set_io_thread (class io_thread_t *io_thread_);
 
         //  Methods to access underlying poller object.
         handle_t add_fd (fd_t fd_);
@@ -71,24 +61,10 @@ namespace zmq
         void out_event ();
         void timer_event ();
 
-        //  Socket owning this I/O object. It is responsible for destroying
-        //  it when it's being closed.
-        object_t *owner;
-
     private:
 
-        //  Set to true when object is plugged in.
-        bool plugged_in;
-
-        //  Set to true when object was terminated before it was plugged in.
-        //  In such case destruction is delayed till 'plug' command arrives.
-        bool terminated;
-
         struct i_poller *poller;
 
-        //  Handlers for incoming commands.
-        void process_term ();
-
         io_object_t (const io_object_t&);
         void operator = (const io_object_t&);
     };
diff --git a/src/options.cpp b/src/options.cpp
new file mode 100644
index 0000000..cd07c44
--- /dev/null
+++ b/src/options.cpp
@@ -0,0 +1,29 @@
+/*
+    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 "options.hpp"
+
+zmq::options_t::options_t () :
+    hwm (0),
+    lwm (0),
+    swap (0),
+    mask (0),
+    affinity (0)
+{
+}
diff --git a/src/options.hpp b/src/options.hpp
new file mode 100644
index 0000000..7d78da2
--- /dev/null
+++ b/src/options.hpp
@@ -0,0 +1,42 @@
+/*
+    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_OPTIONS_HPP_INCLUDED__
+#define __ZMQ_OPTIONS_HPP_INCLUDED__
+
+#include <string>
+
+namespace zmq
+{
+
+    struct options_t
+    {
+        options_t ();
+
+        int64_t hwm;
+        int64_t lwm;
+        int64_t swap;
+        uint64_t mask;
+        uint64_t affinity;
+        std::string identity;
+    };
+
+}
+
+#endif
diff --git a/src/owned.cpp b/src/owned.cpp
new file mode 100644
index 0000000..22e257f
--- /dev/null
+++ b/src/owned.cpp
@@ -0,0 +1,74 @@
+/*
+    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 "owned.hpp"
+#include "err.hpp"
+
+zmq::owned_t::owned_t (object_t *parent_, object_t *owner_) :
+    object_t (parent_),
+    owner (owner_),
+    plugged_in (false),
+    terminated (false)
+{
+}
+
+zmq::owned_t::~owned_t ()
+{
+}
+
+void zmq::owned_t::process_plug ()
+{
+    zmq_assert (!plugged_in);
+
+    //  If termination of the object was already requested, destroy it and
+    //  send the termination acknowledgement.
+    if (terminated) {
+        send_term_ack (owner);
+        delete this;
+        return;
+    }
+
+    //  Notify the generic termination mechanism (io_object_t) that the object
+    //  is already plugged in.
+    plugged_in = true;
+}
+
+void zmq::owned_t::term ()
+{
+    send_term_req (owner, this);
+}
+
+void zmq::owned_t::process_term ()
+{
+    zmq_assert (!terminated);
+
+    //  If termination request has occured even before the object was plugged in
+    //  wait till plugging in happens, then acknowledge the termination.
+    if (!plugged_in) {
+        terminated = true;
+        return;
+    }
+
+    //  Otherwise, destroy the object and acknowledge the termination
+    //  straight away.
+    send_term_ack (owner);
+    process_unplug ();
+    delete this;
+}
+
diff --git a/src/owned.hpp b/src/owned.hpp
new file mode 100644
index 0000000..164622e
--- /dev/null
+++ b/src/owned.hpp
@@ -0,0 +1,82 @@
+/*
+    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_OWNED_HPP_INCLUDED__
+#define __ZMQ_OWNED_HPP_INCLUDED__
+
+#include "object.hpp"
+
+namespace zmq
+{
+
+    //  Base class for objects owned by individual sockets. Handles
+    //  initialisation and destruction of such objects.
+
+    class owned_t : public object_t
+    {
+    public:
+
+        //  The object will live in parent's thread, however, its lifetime
+        //  will be managed by its owner socket.
+        owned_t (object_t *parent_, object_t *owner_);
+
+    protected:
+
+        //  Ask owner socket to terminate this object.
+        void term ();
+
+        //  Derived object destroys owned_t. No point in allowing others to
+        //  invoke the destructor. At the same time, it has to be virtual so
+        //  that generic owned_t deallocation mechanism destroys specific type
+        //  of the owned object correctly.
+        virtual ~owned_t ();
+
+        //  Handlers for incoming commands. It's vital that every I/O object
+        //  invokes io_object_t::process_plug at the end of it's own plug
+        //  handler.
+        void process_plug ();
+
+        //  io_object_t defines a new handler used to disconnect the object
+        //  from the poller object. Implement the handlen in the derived
+        //  classes to ensure sane cleanup.
+        virtual void process_unplug () = 0;
+
+        //  Socket owning this object. It is responsible for destroying
+        //  it when it's being closed.
+        object_t *owner;
+
+    private:
+
+        //  Handlers for incoming commands.
+        void process_term ();
+
+        //  Set to true when object is plugged in.
+        bool plugged_in;
+
+        //  Set to true when object was terminated before it was plugged in.
+        //  In such case destruction is delayed till 'plug' command arrives.
+        bool terminated;
+
+        owned_t (const owned_t&);
+        void operator = (const owned_t&);
+    };
+
+}
+
+#endif
diff --git a/src/session.cpp b/src/session.cpp
new file mode 100644
index 0000000..fa29dd3
--- /dev/null
+++ b/src/session.cpp
@@ -0,0 +1,58 @@
+/*
+    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 "session.hpp"
+#include "zmq_engine.hpp"
+#include "err.hpp"
+
+zmq::session_t::session_t (object_t *parent_, object_t *owner_,
+      zmq_engine_t *engine_) :
+    owned_t (parent_, owner_),
+    engine (engine_)
+{
+}
+
+zmq::session_t::~session_t ()
+{
+}
+
+bool zmq::session_t::read (::zmq_msg *msg_)
+{
+    return false;
+}
+
+bool zmq::session_t::write (::zmq_msg *msg_)
+{
+    return false;
+}
+
+void zmq::session_t::flush ()
+{
+}
+
+void zmq::session_t::process_plug ()
+{
+    engine->plug (this);
+    owned_t::process_plug ();
+}
+
+void zmq::session_t::process_unplug ()
+{
+    engine->unplug ();
+}
diff --git a/src/session.hpp b/src/session.hpp
new file mode 100644
index 0000000..4228fd9
--- /dev/null
+++ b/src/session.hpp
@@ -0,0 +1,57 @@
+/*
+    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_inout.hpp"
+#include "owned.hpp"
+
+namespace zmq
+{
+
+    class session_t : public owned_t, public i_inout
+    {
+    public:
+
+        session_t (object_t *parent_, object_t *owner_,
+            class zmq_engine_t *engine_);
+
+    private:
+
+        ~session_t ();
+
+        //  i_inout interface implementation.
+        bool read (::zmq_msg *msg_);
+        bool write (::zmq_msg *msg_);
+        void flush ();
+
+        //  Handlers for incoming commands.
+        void process_plug ();
+        void process_unplug ();
+
+        class zmq_engine_t *engine;
+
+        session_t (const session_t&);
+        void operator = (const session_t&);
+    };
+
+}
+
+#endif
diff --git a/src/socket_base.cpp b/src/socket_base.cpp
index c179a93..4fadad2 100644
--- a/src/socket_base.cpp
+++ b/src/socket_base.cpp
@@ -32,12 +32,7 @@
 zmq::socket_base_t::socket_base_t (app_thread_t *parent_) :
     object_t (parent_),
     pending_term_acks (0),
-    app_thread (parent_),
-    hwm (0),
-    lwm (0),
-    swap (0),
-    mask (0),
-    affinity (0)
+    app_thread (parent_)
 {    
 }
 
@@ -77,7 +72,7 @@ int zmq::socket_base_t::setsockopt (int option_, void *optval_,
             errno = EINVAL;
             return -1;
         }
-        hwm = *((int64_t*) optval_);
+        options.hwm = *((int64_t*) optval_);
         return 0;
 
     case ZMQ_LWM:
@@ -85,7 +80,7 @@ int zmq::socket_base_t::setsockopt (int option_, void *optval_,
             errno = EINVAL;
             return -1;
         }
-        lwm = *((int64_t*) optval_);
+        options.lwm = *((int64_t*) optval_);
         return 0;
 
     case ZMQ_SWAP:
@@ -93,7 +88,7 @@ int zmq::socket_base_t::setsockopt (int option_, void *optval_,
             errno = EINVAL;
             return -1;
         }
-        swap = *((int64_t*) optval_);
+        options.swap = *((int64_t*) optval_);
         return 0;
 
     case ZMQ_MASK:
@@ -101,7 +96,7 @@ int zmq::socket_base_t::setsockopt (int option_, void *optval_,
             errno = EINVAL;
             return -1;
         }
-        mask = (uint64_t) *((int64_t*) optval_);
+        options.mask = (uint64_t) *((int64_t*) optval_);
         return 0;
 
     case ZMQ_AFFINITY:
@@ -109,15 +104,15 @@ int zmq::socket_base_t::setsockopt (int option_, void *optval_,
             errno = EINVAL;
             return -1;
         }
-        affinity = (uint64_t) *((int64_t*) optval_);
+        options.affinity = (uint64_t) *((int64_t*) optval_);
         return 0;
 
-    case ZMQ_SESSIONID:
+    case ZMQ_IDENTITY:
         if (optvallen_ != sizeof (const char*)) {
             errno = EINVAL;
             return -1;
         }
-        session_id = (const char*) optval_;
+        options.identity = (const char*) optval_;
         return 0;
 
     default:
@@ -128,8 +123,8 @@ int zmq::socket_base_t::setsockopt (int option_, void *optval_,
 
 int zmq::socket_base_t::bind (const char *addr_)
 {
-    zmq_listener_t *listener =
-        new zmq_listener_t (choose_io_thread (affinity), this);
+    zmq_listener_t *listener = new zmq_listener_t (
+        choose_io_thread (options.affinity), this, options);
     int rc = listener->set_address (addr_);
     if (rc != 0)
         return -1;
@@ -141,8 +136,8 @@ int zmq::socket_base_t::bind (const char *addr_)
 
 int zmq::socket_base_t::connect (const char *addr_)
 {
-    zmq_connecter_t *connecter =
-        new zmq_connecter_t (choose_io_thread (affinity), this);
+    zmq_connecter_t *connecter = new zmq_connecter_t (
+        choose_io_thread (options.affinity), this, options);
     int rc = connecter->set_address (addr_);
     if (rc != 0)
         return -1;
diff --git a/src/socket_base.hpp b/src/socket_base.hpp
index 2257fbe..4ccac48 100644
--- a/src/socket_base.hpp
+++ b/src/socket_base.hpp
@@ -24,6 +24,7 @@
 #include <string>
 
 #include "object.hpp"
+#include "options.hpp"
 #include "stdint.hpp"
 
 namespace zmq
@@ -66,12 +67,7 @@ namespace zmq
         class app_thread_t *app_thread;
 
         //  Socket options.
-        int64_t hwm;
-        int64_t lwm;
-        int64_t swap;
-        uint64_t mask;
-        uint64_t affinity;
-        std::string session_id;
+        options_t options;
 
         socket_base_t (const socket_base_t&);
         void operator = (const socket_base_t&);
diff --git a/src/zmq_connecter.cpp b/src/zmq_connecter.cpp
index 513508d..4416a70 100644
--- a/src/zmq_connecter.cpp
+++ b/src/zmq_connecter.cpp
@@ -18,11 +18,16 @@
 */
 
 #include "zmq_connecter.hpp"
+#include "zmq_init.hpp"
+#include "io_thread.hpp"
 #include "err.hpp"
 
-zmq::zmq_connecter_t::zmq_connecter_t (io_thread_t *parent_, object_t *owner_) :
-    io_object_t (parent_, owner_),
-    waiting (false)
+zmq::zmq_connecter_t::zmq_connecter_t (io_thread_t *parent_, object_t *owner_,
+      const options_t &options_) :
+    owned_t (parent_, owner_),
+    io_object_t (parent_),
+    handle_valid (false),
+    options (options_)
 {
 }
 
@@ -38,12 +43,12 @@ int zmq::zmq_connecter_t::set_address (const char *addr_)
 void zmq::zmq_connecter_t::process_plug ()
 {
     start_connecting ();
-    io_object_t::process_plug ();
+    owned_t::process_plug ();
 }
 
 void zmq::zmq_connecter_t::process_unplug ()
 {
-    if (!waiting)
+    if (handle_valid)
         rm_fd (handle);
 }
 
@@ -58,30 +63,31 @@ void zmq::zmq_connecter_t::in_event ()
 void zmq::zmq_connecter_t::out_event ()
 {
     fd_t fd = tcp_connecter.connect ();
+    rm_fd (handle);
+    handle_valid = false;
 
     //  If there was error during the connecting, close the socket and wait
     //  for a while before trying to reconnect.
     if (fd == retired_fd) {
-        rm_fd (handle);
         tcp_connecter.close ();
-        waiting = true;
         add_timer ();
         return;
     }
 
-    zmq_assert (false);
+    //  Create an init object. 
+    io_thread_t *io_thread = choose_io_thread (options.affinity);
+    zmq_init_t *init = new zmq_init_t (io_thread, owner, fd, true, options);
+    zmq_assert (init);
+    send_plug (init);
+    send_own (owner, init);
 
-/*
-    object_t *engine = new zmq_engine_t (choose_io_thread (0), owner);
-    send_plug (engine);
-    send_own (owner, engine);
-*/
+    //  Ask owner socket to shut the connecter down.
+    term ();
 }
 
 void zmq::zmq_connecter_t::timer_event ()
 {
     //  Reconnect period have elapsed.
-    waiting = false;
     start_connecting ();
 }
 
@@ -99,12 +105,12 @@ void zmq::zmq_connecter_t::start_connecting ()
     //  Connection establishment may be dealyed. Poll for its completion.
     else if (rc == -1 && errno == EINPROGRESS) {
         handle = add_fd (tcp_connecter.get_fd ());
+        handle_valid = true;
         set_pollout (handle);
         return;
     }
 
     //  If none of the above is true, synchronous error occured.
     //  Wait for a while and retry.
-    waiting = true;
     add_timer ();
 }
diff --git a/src/zmq_connecter.hpp b/src/zmq_connecter.hpp
index d346396..93497cb 100644
--- a/src/zmq_connecter.hpp
+++ b/src/zmq_connecter.hpp
@@ -20,17 +20,21 @@
 #ifndef __ZMQ_ZMQ_CONNECTER_HPP_INCLUDED__
 #define __ZMQ_ZMQ_CONNECTER_HPP_INCLUDED__
 
+#include "owned.hpp"
 #include "io_object.hpp"
 #include "tcp_connecter.hpp"
+#include "options.hpp"
+#include "stdint.hpp"
 
 namespace zmq
 {
 
-    class zmq_connecter_t : public io_object_t
+    class zmq_connecter_t : public owned_t, public io_object_t
     {
     public:
 
-        zmq_connecter_t (class io_thread_t *parent_, object_t *owner_);
+        zmq_connecter_t (class io_thread_t *parent_, object_t *owner_,
+            const options_t &options_);
 
         //  Set IP address to connect to.
         int set_address (const char *addr_);
@@ -57,9 +61,12 @@ namespace zmq
         //  Handle corresponding to the listening socket.
         handle_t handle;
 
-        //  True, if we are waiting for a period of time before trying to
-        //  reconnect.
-        bool waiting;
+        //  If true file descriptor is registered with the poller and 'handle'
+        //  contains valid value.
+        bool handle_valid;
+
+        //  Associated socket options.
+        options_t options;
 
         zmq_connecter_t (const zmq_connecter_t&);
         void operator = (const zmq_connecter_t&);
diff --git a/src/zmq_decoder.cpp b/src/zmq_decoder.cpp
new file mode 100644
index 0000000..b461234
--- /dev/null
+++ b/src/zmq_decoder.cpp
@@ -0,0 +1,78 @@
+/*
+    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_inout.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_inout (i_inout *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
new file mode 100644
index 0000000..17c28f8
--- /dev/null
+++ b/src/zmq_decoder.hpp
@@ -0,0 +1,57 @@
+/*
+    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_inout (struct i_inout *destination_);
+
+    private:
+
+        bool one_byte_size_ready ();
+        bool eight_byte_size_ready ();
+        bool message_ready ();
+
+        struct i_inout *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
new file mode 100644
index 0000000..124d77b
--- /dev/null
+++ b/src/zmq_encoder.cpp
@@ -0,0 +1,76 @@
+/*
+    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_inout.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_inout (i_inout *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
new file mode 100644
index 0000000..89af265
--- /dev/null
+++ b/src/zmq_encoder.hpp
@@ -0,0 +1,55 @@
+/*
+    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_inout (struct i_inout *source_);
+
+    private:
+
+        bool size_ready ();
+        bool message_ready ();
+
+        struct i_inout *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_engine.cpp b/src/zmq_engine.cpp
index 3708c9a..3620d30 100644
--- a/src/zmq_engine.cpp
+++ b/src/zmq_engine.cpp
@@ -19,17 +19,118 @@
 
 #include "zmq_engine.hpp"
 #include "io_thread.hpp"
+#include "i_inout.hpp"
+#include "config.hpp"
+#include "err.hpp"
 
-zmq::zmq_engine_t::zmq_engine_t (io_thread_t *parent_, object_t *owner_) :
-    io_object_t (parent_, owner_)
+zmq::zmq_engine_t::zmq_engine_t (io_thread_t *parent_, fd_t fd_) :
+    io_object_t (parent_),
+    insize (0),
+    inpos (0),
+    outsize (0),
+    outpos (0),
+    inout (NULL)
 {
+    //  Allocate read & write buffer.
+    inbuf = (unsigned char*) malloc (in_batch_size);
+    zmq_assert (inbuf);
+    outbuf = (unsigned char*) malloc (out_batch_size);
+    zmq_assert (outbuf);
+
+    //  Initialise the underlying socket.
+    int rc = tcp_socket.open (fd_);
+    zmq_assert (rc == 0);
 }
 
 zmq::zmq_engine_t::~zmq_engine_t ()
 {
+    free (outbuf);
+    free (inbuf);
+}
+
+void zmq::zmq_engine_t::plug (i_inout *inout_)
+{
+    encoder.set_inout (inout_);
+    decoder.set_inout (inout_);
+
+    handle = add_fd (tcp_socket.get_fd ());
+    set_pollin (handle);
+    set_pollout (handle);
+
+    inout = inout_;
+}
+
+void zmq::zmq_engine_t::unplug ()
+{
+    rm_fd (handle);
+    inout = NULL;
+}
+
+void zmq::zmq_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 = tcp_socket.read (inbuf, in_batch_size);
+printf ("%d bytes read\n", (int) insize);
+        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)
+        reset_pollin (handle);
+
+    //  If at least one byte was processed, flush all messages the decoder
+    //  may have produced.
+    if (nbytes > 0)
+        inout->flush ();
+
 }
 
-void zmq::zmq_engine_t::process_plug ()
+void zmq::zmq_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 is no data to send, stop polling for output.
+        if (outsize == 0)
+            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 = tcp_socket.write (outbuf + outpos, outsize - outpos);
+
+        //  Handle problems with the connection.
+        if (nbytes == -1) {
+            error ();
+            return;
+        }
+
+        outpos += nbytes;
+    }
 }
 
+void zmq::zmq_engine_t::error ()
+{
+    zmq_assert (false);
+}
diff --git a/src/zmq_engine.hpp b/src/zmq_engine.hpp
index ad739c7..38a390d 100644
--- a/src/zmq_engine.hpp
+++ b/src/zmq_engine.hpp
@@ -21,6 +21,9 @@
 #define __ZMQ_ZMQ_ENGINE_HPP_INCLUDED__
 
 #include "io_object.hpp"
+#include "tcp_socket.hpp"
+#include "zmq_encoder.hpp"
+#include "zmq_decoder.hpp"
 
 namespace zmq
 {
@@ -29,14 +32,36 @@ namespace zmq
     {
     public:
 
-        zmq_engine_t (class io_thread_t *parent_, object_t *owner_);
+        zmq_engine_t (class io_thread_t *parent_, fd_t fd_);
+        ~zmq_engine_t ();
+
+        void plug (struct i_inout *inout_);
+        void unplug ();
+
+        //  i_poll_events interface implementation.
+        void in_event ();
+        void out_event ();
 
     private:
 
-        ~zmq_engine_t ();
+        //  Function to handle network disconnections.
+        void error ();
+
+        tcp_socket_t tcp_socket;
+        handle_t handle;
+
+        unsigned char *inbuf;
+        int insize;
+        int inpos;
+
+        unsigned char *outbuf;
+        int outsize;
+        int outpos;
+
+        i_inout *inout;
 
-        //  Handlers for incoming commands.
-        void process_plug ();
+        zmq_encoder_t encoder;
+        zmq_decoder_t decoder;
 
         zmq_engine_t (const zmq_engine_t&);
         void operator = (const zmq_engine_t&);
diff --git a/src/zmq_init.cpp b/src/zmq_init.cpp
new file mode 100644
index 0000000..fea1452
--- /dev/null
+++ b/src/zmq_init.cpp
@@ -0,0 +1,110 @@
+/*
+    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_init.hpp"
+#include "io_thread.hpp"
+#include "session.hpp"
+#include "err.hpp"
+
+zmq::zmq_init_t::zmq_init_t (io_thread_t *parent_, object_t *owner_, fd_t fd_,
+      bool connected_, const options_t &options_) :
+    owned_t (parent_, owner_),
+    connected (connected_),
+    options (options_)
+{
+    //  Create associated engine object.
+    engine = new zmq_engine_t (parent_, fd_);
+    zmq_assert (engine);
+}
+
+zmq::zmq_init_t::~zmq_init_t ()
+{
+    if (engine)
+        delete engine;
+}
+
+bool zmq::zmq_init_t::read (::zmq_msg *msg_)
+{
+    //  On the listening side, no initialisation data are sent to the peer.
+    if (!connected)
+        return false;
+
+    //  Send identity.
+    int rc = zmq_msg_init_size (msg_, options.identity.size ());
+    zmq_assert (rc == 0);
+    memcpy (zmq_msg_data (msg_), options.identity.c_str (),
+        options.identity.size ());
+
+    //  Initialisation is done.
+    create_session ();
+
+    return true;
+}
+
+bool zmq::zmq_init_t::write (::zmq_msg *msg_)
+{
+    //  On the connecting side no initialisation data are expected.
+    if (connected)
+        return false;
+
+    //  Retreieve the identity.
+    options.identity = std::string ((const char*) zmq_msg_data (msg_),
+        zmq_msg_size (msg_));
+
+    //  Initialisation is done.
+    create_session ();
+
+    return true;
+}
+
+void zmq::zmq_init_t::flush ()
+{
+    //  No need to do anything. zmq_init_t does no batching of messages.
+    //  Each message is processed immediately on write.
+}
+
+void zmq::zmq_init_t::process_plug ()
+{
+    engine->plug (this);
+    owned_t::process_plug ();
+}
+
+void zmq::zmq_init_t::process_unplug ()
+{
+    engine->unplug ();
+}
+
+void zmq::zmq_init_t::create_session ()
+{
+    //  Disconnect engine from the init object.
+    engine->unplug ();
+
+    //  Create the session instance.
+    io_thread_t *io_thread = choose_io_thread (options.affinity);
+    session_t *session = new session_t (io_thread, owner, engine);
+    zmq_assert (session);
+    engine = NULL;
+
+    //  Pass session/engine pair to a chosen I/O thread.
+    send_plug (session);
+    send_own (owner, session);
+
+    //  Destroy the init object.
+    term ();
+}
diff --git a/src/zmq_init.hpp b/src/zmq_init.hpp
new file mode 100644
index 0000000..2e0910a
--- /dev/null
+++ b/src/zmq_init.hpp
@@ -0,0 +1,82 @@
+/*
+    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_INIT_HPP_INCLUDED__
+#define __ZMQ_ZMQ_INIT_HPP_INCLUDED__
+
+#include <string>
+
+#include "i_inout.hpp"
+#include "owned.hpp"
+#include "zmq_engine.hpp"
+#include "stdint.hpp"
+#include "fd.hpp"
+#include "options.hpp"
+
+namespace zmq
+{
+
+    //  The class handles initialisation phase of native 0MQ wire-level
+    //  protocol. Currently it can be used to handle both sides of the
+    //  connection. If it grows to complex, we can separate the two into
+    //  distinct classes.
+
+    class zmq_init_t : public owned_t, public i_inout
+    {
+    public:
+
+        //  Set 'connected' to true if the connection was created by 'connect'
+        //  function. If it was accepted from a listening socket, set it to
+        //  false.
+        zmq_init_t (class io_thread_t *parent_, object_t *owner_, fd_t fd_,
+            bool connected_, const options_t &options);
+        ~zmq_init_t ();
+
+    private:
+
+        //  i_inout interface implementation.
+        bool read (::zmq_msg *msg_);
+        bool write (::zmq_msg *msg_);
+        void flush ();
+
+        //  Handlers for incoming commands.
+        void process_plug ();
+        void process_unplug ();
+
+        void create_session ();
+
+        //  Engine is created by zmq_init_t object. Once the initialisation
+        //  phase is over it is passed to a session object, possibly running
+        //  in a different I/O thread.
+        zmq_engine_t *engine;
+
+        //  If true, we are on the connecting side. If false, we are on the
+        //  listening side.
+        bool connected;
+
+        //  Associated socket options.
+        options_t options;
+
+        zmq_init_t (const zmq_init_t&);
+        void operator = (const zmq_init_t&);
+    };
+
+}
+
+#endif
diff --git a/src/zmq_listener.cpp b/src/zmq_listener.cpp
index 463a1dd..c990468 100644
--- a/src/zmq_listener.cpp
+++ b/src/zmq_listener.cpp
@@ -18,12 +18,15 @@
 */
 
 #include "zmq_listener.hpp"
-#include "zmq_engine.hpp"
+#include "zmq_init.hpp"
 #include "io_thread.hpp"
 #include "err.hpp"
 
-zmq::zmq_listener_t::zmq_listener_t (io_thread_t *parent_, object_t *owner_) :
-    io_object_t (parent_, owner_)
+zmq::zmq_listener_t::zmq_listener_t (io_thread_t *parent_, object_t *owner_,
+      const options_t &options_) :
+    owned_t (parent_, owner_),
+    io_object_t (parent_),
+    options (options_)
 {
 }
 
@@ -46,7 +49,7 @@ void zmq::zmq_listener_t::process_plug ()
     handle = add_fd (tcp_listener.get_fd ());
     set_pollin (handle);
 
-    io_object_t::process_plug ();
+    owned_t::process_plug ();
 }
 
 void zmq::zmq_listener_t::process_unplug ()
@@ -63,14 +66,12 @@ void zmq::zmq_listener_t::in_event ()
     if (fd == retired_fd)
         return;
 
-    //  TODO
-    zmq_assert (false);
-
-/*
-    object_t *engine = new zmq_engine_t (choose_io_thread (0), owner);
-    send_plug (engine);
-    send_own (owner, engine);
-*/
+    //  Create an init object. 
+    io_thread_t *io_thread = choose_io_thread (options.affinity);
+    zmq_init_t *init = new zmq_init_t (io_thread, owner, fd, false, options);
+    zmq_assert (init);
+    send_plug (init);
+    send_own (owner, init);
 }
 
 
diff --git a/src/zmq_listener.hpp b/src/zmq_listener.hpp
index ef252b3..f85ad5a 100644
--- a/src/zmq_listener.hpp
+++ b/src/zmq_listener.hpp
@@ -20,17 +20,21 @@
 #ifndef __ZMQ_ZMQ_LISTENER_HPP_INCLUDED__
 #define __ZMQ_ZMQ_LISTENER_HPP_INCLUDED__
 
+#include "owned.hpp"
 #include "io_object.hpp"
 #include "tcp_listener.hpp"
+#include "options.hpp"
+#include "stdint.hpp"
 
 namespace zmq
 {
 
-    class zmq_listener_t : public io_object_t
+    class zmq_listener_t : public owned_t, public io_object_t
     {
     public:
 
-        zmq_listener_t (class io_thread_t *parent_, object_t *owner_);
+        zmq_listener_t (class io_thread_t *parent_, object_t *owner_,
+            const options_t &options_);
 
         //  Set IP address to listen on.
         int set_address (const char *addr_);
@@ -52,6 +56,9 @@ namespace zmq
         //  Handle corresponding to the listening socket.
         handle_t handle;
 
+        //  Associated socket options.
+        options_t options;
+
         zmq_listener_t (const zmq_listener_t&);
         void operator = (const zmq_listener_t&);
     };