Session class separated into socket-type-specific sessions

This is a preliminary patch allowing for socket-type-specific
functionality in the I/O thread. For example, message format
can be checked asynchronously and misbehaved connections dropped
straight away.

Signed-off-by: Martin Sustrik <sustrik@250bpm.com>

50 files changed, 494 insertions, 82 deletions

diff --git a/src/Makefile.am b/src/Makefile.am
index 7992ab8..3b7dec6 100644
--- a/src/Makefile.am
+++ b/src/Makefile.am
@@ -57,7 +57,7 @@ libzmq_la_SOURCES = \
     req.hpp \
     router.hpp \
     select.hpp \
-    session.hpp \
+    session_base.hpp \
     signaler.hpp \
     socket_base.hpp \
     stdint.hpp \
@@ -117,7 +117,7 @@ libzmq_la_SOURCES = \
     rep.cpp \
     req.cpp \
     select.cpp \
-    session.cpp \
+    session_base.cpp \
     signaler.cpp \
     socket_base.cpp \
     stream_engine.cpp \
diff --git a/src/decoder.cpp b/src/decoder.cpp
index 01ce0bb..9e93b73 100644
--- a/src/decoder.cpp
+++ b/src/decoder.cpp
@@ -22,7 +22,7 @@
 #include <string.h>
 
 #include "decoder.hpp"
-#include "session.hpp"
+#include "session_base.hpp"
 #include "wire.hpp"
 #include "err.hpp"
 
@@ -44,7 +44,7 @@ zmq::decoder_t::~decoder_t ()
     errno_assert (rc == 0);
 }
 
-void zmq::decoder_t::set_session (session_t *session_)
+void zmq::decoder_t::set_session (session_base_t *session_)
 {
     session = session_;
 }
diff --git a/src/decoder.hpp b/src/decoder.hpp
index 3ac4f7c..01021c4 100644
--- a/src/decoder.hpp
+++ b/src/decoder.hpp
@@ -184,7 +184,7 @@ namespace zmq
         decoder_t (size_t bufsize_, int64_t maxmsgsize_);
         ~decoder_t ();
 
-        void set_session (class session_t *session_);
+        void set_session (class session_base_t *session_);
 
     private:
 
@@ -193,7 +193,7 @@ namespace zmq
         bool flags_ready ();
         bool message_ready ();
 
-        class session_t *session;
+        class session_base_t *session;
         unsigned char tmpbuf [8];
         msg_t in_progress;
 
diff --git a/src/encoder.cpp b/src/encoder.cpp
index 087735d..6d09384 100644
--- a/src/encoder.cpp
+++ b/src/encoder.cpp
@@ -19,7 +19,7 @@
 */
 
 #include "encoder.hpp"
-#include "session.hpp"
+#include "session_base.hpp"
 #include "wire.hpp"
 
 zmq::encoder_t::encoder_t (size_t bufsize_) :
@@ -39,7 +39,7 @@ zmq::encoder_t::~encoder_t ()
     errno_assert (rc == 0);
 }
 
-void zmq::encoder_t::set_session (session_t *session_)
+void zmq::encoder_t::set_session (session_base_t *session_)
 {
     session = session_;
 }
diff --git a/src/encoder.hpp b/src/encoder.hpp
index b8784a3..f7e3cbc 100644
--- a/src/encoder.hpp
+++ b/src/encoder.hpp
@@ -163,14 +163,14 @@ namespace zmq
         encoder_t (size_t bufsize_);
         ~encoder_t ();
 
-        void set_session (class session_t *session_);
+        void set_session (class session_base_t *session_);
 
     private:
 
         bool size_ready ();
         bool message_ready ();
 
-        class session_t *session;
+        class session_base_t *session;
         msg_t in_progress;
         unsigned char tmpbuf [10];
 
diff --git a/src/i_engine.hpp b/src/i_engine.hpp
index c49a107..26e475b 100644
--- a/src/i_engine.hpp
+++ b/src/i_engine.hpp
@@ -32,7 +32,7 @@ namespace zmq
 
         //  Plug the engine to the session.
         virtual void plug (class io_thread_t *io_thread_,
-            class session_t *session_) = 0;
+            class session_base_t *session_) = 0;
 
         //  Unplug the engine from the session.
         virtual void unplug () = 0;
diff --git a/src/ipc_connecter.cpp b/src/ipc_connecter.cpp
index 9b8520d..a54e8fe 100644
--- a/src/ipc_connecter.cpp
+++ b/src/ipc_connecter.cpp
@@ -38,7 +38,7 @@
 #include <sys/un.h>
 
 zmq::ipc_connecter_t::ipc_connecter_t (class io_thread_t *io_thread_,
-      class session_t *session_, const options_t &options_,
+      class session_base_t *session_, const options_t &options_,
       const char *address_, bool wait_) :
     own_t (io_thread_, options_),
     io_object_t (io_thread_),
diff --git a/src/ipc_connecter.hpp b/src/ipc_connecter.hpp
index 0bb9d69..721bcf4 100644
--- a/src/ipc_connecter.hpp
+++ b/src/ipc_connecter.hpp
@@ -41,7 +41,7 @@ namespace zmq
         //  If 'delay' is true connecter first waits for a while, then starts
         //  connection process.
         ipc_connecter_t (class io_thread_t *io_thread_,
-            class session_t *session_, const options_t &options_,
+            class session_base_t *session_, const options_t &options_,
             const char *address_, bool delay_);
         ~ipc_connecter_t ();
 
@@ -101,7 +101,7 @@ namespace zmq
         bool wait;
 
         //  Reference to the session we belong to.
-        class session_t *session;
+        class session_base_t *session;
 
         //  Current reconnect ivl, updated for backoff strategy
         int current_reconnect_ivl;
diff --git a/src/ipc_listener.cpp b/src/ipc_listener.cpp
index cad58ba..5ba41be 100644
--- a/src/ipc_listener.cpp
+++ b/src/ipc_listener.cpp
@@ -29,7 +29,7 @@
 #include "stream_engine.hpp"
 #include "ipc_address.hpp"
 #include "io_thread.hpp"
-#include "session.hpp"
+#include "session_base.hpp"
 #include "config.hpp"
 #include "err.hpp"
 #include "ip.hpp"
@@ -87,9 +87,9 @@ void zmq::ipc_listener_t::in_event ()
     zmq_assert (io_thread);
 
     //  Create and launch a session object. 
-    session_t *session = new (std::nothrow)
-        session_t (io_thread, false, socket, options, NULL, NULL);
-    alloc_assert (session);
+    session_base_t *session = session_base_t::create (io_thread, false, socket,
+        options, NULL, NULL);
+    errno_assert (session);
     session->inc_seqnum ();
     launch_child (session);
     send_attach (session, engine, false);
diff --git a/src/object.cpp b/src/object.cpp
index 7f7d7f8..807fb04 100644
--- a/src/object.cpp
+++ b/src/object.cpp
@@ -26,7 +26,7 @@
 #include "err.hpp"
 #include "pipe.hpp"
 #include "io_thread.hpp"
-#include "session.hpp"
+#include "session_base.hpp"
 #include "socket_base.hpp"
 
 zmq::object_t::object_t (ctx_t *ctx_, uint32_t tid_) :
@@ -201,8 +201,8 @@ void zmq::object_t::send_own (own_t *destination_, own_t *object_)
     send_command (cmd);
 }
 
-void zmq::object_t::send_attach (session_t *destination_, i_engine *engine_,
-    bool inc_seqnum_)
+void zmq::object_t::send_attach (session_base_t *destination_,
+    i_engine *engine_, bool inc_seqnum_)
 {
     if (inc_seqnum_)
         destination_->inc_seqnum ();
diff --git a/src/object.hpp b/src/object.hpp
index e05b958..1a38b24 100644
--- a/src/object.hpp
+++ b/src/object.hpp
@@ -62,7 +62,7 @@ namespace zmq
             bool inc_seqnum_ = true);
         void send_own (class own_t *destination_,
             class own_t *object_);
-        void send_attach (class session_t *destination_,
+        void send_attach (class session_base_t *destination_,
              struct i_engine *engine_, bool inc_seqnum_ = true);
         void send_bind (class own_t *destination_, class pipe_t *pipe_,
              bool inc_seqnum_ = true);
diff --git a/src/pair.cpp b/src/pair.cpp
index 12a1881..2fa4eac 100644
--- a/src/pair.cpp
+++ b/src/pair.cpp
@@ -116,3 +116,15 @@ bool zmq::pair_t::xhas_out ()
     return result;
 }
 
+zmq::pair_session_t::pair_session_t (io_thread_t *io_thread_, bool connect_,
+      socket_base_t *socket_, const options_t &options_,
+      const char *protocol_, const char *address_) :
+    session_base_t (io_thread_, connect_, socket_, options_, protocol_,
+         address_)
+{
+}
+
+zmq::pair_session_t::~pair_session_t ()
+{
+}
+
diff --git a/src/pair.hpp b/src/pair.hpp
index 59300ae..e7390d6 100644
--- a/src/pair.hpp
+++ b/src/pair.hpp
@@ -22,6 +22,7 @@
 #define __ZMQ_PAIR_HPP_INCLUDED__
 
 #include "socket_base.hpp"
+#include "session_base.hpp"
 
 namespace zmq
 {
@@ -52,6 +53,21 @@ namespace zmq
         const pair_t &operator = (const pair_t&);
     };
 
+    class pair_session_t : public session_base_t
+    {
+    public:
+
+        pair_session_t (class io_thread_t *io_thread_, bool connect_,
+            class socket_base_t *socket_, const options_t &options_,
+            const char *protocol_, const char *address_);
+        ~pair_session_t ();
+
+    private:
+
+        pair_session_t (const pair_session_t&);
+        const pair_session_t &operator = (const pair_session_t&);
+    };
+
 }
 
 #endif
diff --git a/src/pgm_receiver.cpp b/src/pgm_receiver.cpp
index 5c1517d..6c292cd 100644
--- a/src/pgm_receiver.cpp
+++ b/src/pgm_receiver.cpp
@@ -29,7 +29,7 @@
 #endif
 
 #include "pgm_receiver.hpp"
-#include "session.hpp"
+#include "session_base.hpp"
 #include "stdint.hpp"
 #include "wire.hpp"
 #include "err.hpp"
@@ -57,7 +57,8 @@ int zmq::pgm_receiver_t::init (bool udp_encapsulation_, const char *network_)
     return pgm_socket.init (udp_encapsulation_, network_);
 }
 
-void zmq::pgm_receiver_t::plug (io_thread_t *io_thread_, session_t *session_)
+void zmq::pgm_receiver_t::plug (io_thread_t *io_thread_,
+    session_base_t *session_)
 {
     //  Retrieve PGM fds and start polling.
     fd_t socket_fd = retired_fd;
diff --git a/src/pgm_receiver.hpp b/src/pgm_receiver.hpp
index f66c592..b9e9a05 100644
--- a/src/pgm_receiver.hpp
+++ b/src/pgm_receiver.hpp
@@ -52,7 +52,8 @@ namespace zmq
         int init (bool udp_encapsulation_, const char *network_);
 
         //  i_engine interface implementation.
-        void plug (class io_thread_t *io_thread_, class session_t *session_);
+        void plug (class io_thread_t *io_thread_,
+            class session_base_t *session_);
         void unplug ();
         void terminate ();
         void activate_in ();
@@ -105,7 +106,7 @@ namespace zmq
         options_t options;
 
         //  Associated session.
-        class session_t *session;
+        class session_base_t *session;
 
         //  Most recently used decoder.
         decoder_t *mru_decoder;
diff --git a/src/pgm_sender.cpp b/src/pgm_sender.cpp
index e103d9a..733b1ec 100644
--- a/src/pgm_sender.cpp
+++ b/src/pgm_sender.cpp
@@ -30,7 +30,7 @@
 
 #include "io_thread.hpp"
 #include "pgm_sender.hpp"
-#include "session.hpp"
+#include "session_base.hpp"
 #include "err.hpp"
 #include "wire.hpp"
 #include "stdint.hpp"
@@ -62,7 +62,7 @@ int zmq::pgm_sender_t::init (bool udp_encapsulation_, const char *network_)
     return rc;
 }
 
-void zmq::pgm_sender_t::plug (io_thread_t *io_thread_, session_t *session_)
+void zmq::pgm_sender_t::plug (io_thread_t *io_thread_, session_base_t *session_)
 {
     //  Alocate 2 fds for PGM socket.
     fd_t downlink_socket_fd = retired_fd;
diff --git a/src/pgm_sender.hpp b/src/pgm_sender.hpp
index 366e385..d3d5924 100644
--- a/src/pgm_sender.hpp
+++ b/src/pgm_sender.hpp
@@ -50,7 +50,8 @@ namespace zmq
         int init (bool udp_encapsulation_, const char *network_);
 
         //  i_engine interface implementation.
-        void plug (class io_thread_t *io_thread_, class session_t *session_);
+        void plug (class io_thread_t *io_thread_,
+            class session_base_t *session_);
         void unplug ();
         void terminate ();
         void activate_in ();
diff --git a/src/pub.cpp b/src/pub.cpp
index 4787c32..15ec291 100644
--- a/src/pub.cpp
+++ b/src/pub.cpp
@@ -43,3 +43,15 @@ bool zmq::pub_t::xhas_in ()
     return false;
 }
 
+zmq::pub_session_t::pub_session_t (io_thread_t *io_thread_, bool connect_,
+      socket_base_t *socket_, const options_t &options_,
+      const char *protocol_, const char *address_) :
+    xpub_session_t (io_thread_, connect_, socket_, options_, protocol_,
+        address_)
+{
+}
+
+zmq::pub_session_t::~pub_session_t ()
+{
+}
+
diff --git a/src/pub.hpp b/src/pub.hpp
index c8db55f..4a4da0f 100644
--- a/src/pub.hpp
+++ b/src/pub.hpp
@@ -43,6 +43,21 @@ namespace zmq
         const pub_t &operator = (const pub_t&);
     };
 
+    class pub_session_t : public xpub_session_t
+    {
+    public:
+
+        pub_session_t (class io_thread_t *io_thread_, bool connect_,
+            class socket_base_t *socket_, const options_t &options_,
+            const char *protocol_, const char *address_);
+        ~pub_session_t ();
+
+    private:
+
+        pub_session_t (const pub_session_t&);
+        const pub_session_t &operator = (const pub_session_t&);
+    };
+
 }
 
 #endif
diff --git a/src/pull.cpp b/src/pull.cpp
index afde236..06575da 100644
--- a/src/pull.cpp
+++ b/src/pull.cpp
@@ -59,3 +59,15 @@ bool zmq::pull_t::xhas_in ()
     return fq.has_in ();
 }
 
+zmq::pull_session_t::pull_session_t (io_thread_t *io_thread_, bool connect_,
+      socket_base_t *socket_, const options_t &options_,
+      const char *protocol_, const char *address_) :
+    session_base_t (io_thread_, connect_, socket_, options_, protocol_,
+        address_)
+{
+}
+
+zmq::pull_session_t::~pull_session_t ()
+{
+}
+
diff --git a/src/pull.hpp b/src/pull.hpp
index be82af9..6a46ead 100644
--- a/src/pull.hpp
+++ b/src/pull.hpp
@@ -22,6 +22,7 @@
 #define __ZMQ_PULL_HPP_INCLUDED__
 
 #include "socket_base.hpp"
+#include "session_base.hpp"
 #include "fq.hpp"
 
 namespace zmq
@@ -54,6 +55,21 @@ namespace zmq
 
     };
 
+    class pull_session_t : public session_base_t
+    {
+    public:
+
+        pull_session_t (class io_thread_t *io_thread_, bool connect_,
+            class socket_base_t *socket_, const options_t &options_,
+            const char *protocol_, const char *address_);
+        ~pull_session_t ();
+
+    private:
+
+        pull_session_t (const pull_session_t&);
+        const pull_session_t &operator = (const pull_session_t&);
+    };
+
 }
 
 #endif
diff --git a/src/push.cpp b/src/push.cpp
index 77cc9d8..e91b789 100644
--- a/src/push.cpp
+++ b/src/push.cpp
@@ -59,3 +59,15 @@ bool zmq::push_t::xhas_out ()
     return lb.has_out ();
 }
 
+zmq::push_session_t::push_session_t (io_thread_t *io_thread_, bool connect_,
+      socket_base_t *socket_, const options_t &options_,
+      const char *protocol_, const char *address_) :
+    session_base_t (io_thread_, connect_, socket_, options_, protocol_,
+        address_)
+{
+}
+
+zmq::push_session_t::~push_session_t ()
+{
+}
+
diff --git a/src/push.hpp b/src/push.hpp
index 222a62d..1feb71d 100644
--- a/src/push.hpp
+++ b/src/push.hpp
@@ -22,6 +22,7 @@
 #define __ZMQ_PUSH_HPP_INCLUDED__
 
 #include "socket_base.hpp"
+#include "session_base.hpp"
 #include "lb.hpp"
 
 namespace zmq
@@ -53,6 +54,21 @@ namespace zmq
         const push_t &operator = (const push_t&);
     };
 
+    class push_session_t : public session_base_t
+    {
+    public:
+
+        push_session_t (class io_thread_t *io_thread_, bool connect_,
+            class socket_base_t *socket_, const options_t &options_,
+            const char *protocol_, const char *address_);
+        ~push_session_t ();
+
+    private:
+
+        push_session_t (const push_session_t&);
+        const push_session_t &operator = (const push_session_t&);
+    };
+
 }
 
 #endif
diff --git a/src/rep.cpp b/src/rep.cpp
index 2ad494d..564fa89 100644
--- a/src/rep.cpp
+++ b/src/rep.cpp
@@ -110,3 +110,15 @@ bool zmq::rep_t::xhas_out ()
     return xrep_t::xhas_out ();
 }
 
+zmq::rep_session_t::rep_session_t (io_thread_t *io_thread_, bool connect_,
+      socket_base_t *socket_, const options_t &options_,
+      const char *protocol_, const char *address_) :
+    xrep_session_t (io_thread_, connect_, socket_, options_, protocol_,
+        address_)
+{
+}
+
+zmq::rep_session_t::~rep_session_t ()
+{
+}
+
diff --git a/src/rep.hpp b/src/rep.hpp
index a13853d..55d57bd 100644
--- a/src/rep.hpp
+++ b/src/rep.hpp
@@ -54,6 +54,21 @@ namespace zmq
 
     };
 
+    class rep_session_t : public xrep_session_t
+    {
+    public:
+
+        rep_session_t (class io_thread_t *io_thread_, bool connect_,
+            class socket_base_t *socket_, const options_t &options_,
+            const char *protocol_, const char *address_);
+        ~rep_session_t ();
+
+    private:
+
+        rep_session_t (const rep_session_t&);
+        const rep_session_t &operator = (const rep_session_t&);
+    };
+
 }
 
 #endif
diff --git a/src/req.cpp b/src/req.cpp
index b3a9359..323e058 100644
--- a/src/req.cpp
+++ b/src/req.cpp
@@ -146,4 +146,15 @@ bool zmq::req_t::xhas_out ()
     return xreq_t::xhas_out ();
 }
 
+zmq::req_session_t::req_session_t (io_thread_t *io_thread_, bool connect_,
+      socket_base_t *socket_, const options_t &options_,
+      const char *protocol_, const char *address_) :
+    xreq_session_t (io_thread_, connect_, socket_, options_, protocol_,
+        address_)
+{
+}
+
+zmq::req_session_t::~req_session_t ()
+{
+}
 
diff --git a/src/req.hpp b/src/req.hpp
index 50dcb44..2c2cbc4 100644
--- a/src/req.hpp
+++ b/src/req.hpp
@@ -58,6 +58,21 @@ namespace zmq
         const req_t &operator = (const req_t&);
     };
 
+    class req_session_t : public xreq_session_t
+    {
+    public:
+
+        req_session_t (class io_thread_t *io_thread_, bool connect_,
+            class socket_base_t *socket_, const options_t &options_,
+            const char *protocol_, const char *address_);
+        ~req_session_t ();
+
+    private:
+
+        req_session_t (const req_session_t&);
+        const req_session_t &operator = (const req_session_t&);
+    };
+
 }
 
 #endif
diff --git a/src/router.cpp b/src/router.cpp
index 2c9ade9..c8cc278 100755
--- a/src/router.cpp
+++ b/src/router.cpp
@@ -270,5 +270,15 @@ bool zmq::router_t::xhas_out ()
     return true;
 }
 
+zmq::router_session_t::router_session_t (io_thread_t *io_thread_, bool connect_,
+      socket_base_t *socket_, const options_t &options_,
+      const char *protocol_, const char *address_) :
+    session_base_t (io_thread_, connect_, socket_, options_, protocol_,
+        address_)
+{
+}
 
+zmq::router_session_t::~router_session_t ()
+{
+}
 
diff --git a/src/router.hpp b/src/router.hpp
index aeac865..9a5c0f9 100755
--- a/src/router.hpp
+++ b/src/router.hpp
@@ -25,6 +25,7 @@
 #include <deque>
 
 #include "socket_base.hpp"
+#include "session_base.hpp"
 #include "stdint.hpp"
 #include "msg.hpp"
 #include "fq.hpp"
@@ -102,6 +103,21 @@ namespace zmq
         const router_t &operator = (const router_t&);
     };
 
+    class router_session_t : public session_base_t
+    {
+    public:
+
+        router_session_t (class io_thread_t *io_thread_, bool connect_,
+            class socket_base_t *socket_, const options_t &options_,
+            const char *protocol_, const char *address_);
+        ~router_session_t ();
+
+    private:
+
+        router_session_t (const router_session_t&);
+        const router_session_t &operator = (const router_session_t&);
+    };
+
 }
 
 #endif
diff --git a/src/session.cpp b/src/session_base.cpp
index 8001ba8..7d4c5ab 100644
--- a/src/session.cpp
+++ b/src/session_base.cpp
@@ -18,7 +18,7 @@
     along with this program.  If not, see <http://www.gnu.org/licenses/>.
 */
 
-#include "session.hpp"
+#include "session_base.hpp"
 #include "socket_base.hpp"
 #include "i_engine.hpp"
 #include "err.hpp"
@@ -30,8 +30,82 @@
 #include "pgm_sender.hpp"
 #include "pgm_receiver.hpp"
 
-zmq::session_t::session_t (class io_thread_t *io_thread_, bool connect_,
-      class socket_base_t *socket_, const options_t &options_,
+#include "req.hpp"
+#include "xreq.hpp"
+#include "rep.hpp"
+#include "xrep.hpp"
+#include "pub.hpp"
+#include "xpub.hpp"
+#include "sub.hpp"
+#include "xsub.hpp"
+#include "push.hpp"
+#include "pull.hpp"
+#include "router.hpp"
+#include "pair.hpp"
+
+zmq::session_base_t *zmq::session_base_t::create (class io_thread_t *io_thread_,
+    bool connect_, class socket_base_t *socket_, const options_t &options_,
+    const char *protocol_, const char *address_)
+{
+    session_base_t *s = NULL;
+    switch (options_.type) {
+    case ZMQ_REQ:
+        s = new (std::nothrow) req_session_t (io_thread_, connect_,
+            socket_, options_, protocol_, address_);
+        break;
+    case ZMQ_XREQ:
+        s = new (std::nothrow) xreq_session_t (io_thread_, connect_,
+            socket_, options_, protocol_, address_);
+    case ZMQ_REP:
+        s = new (std::nothrow) rep_session_t (io_thread_, connect_,
+            socket_, options_, protocol_, address_);
+        break;
+    case ZMQ_XREP:
+        s = new (std::nothrow) xrep_session_t (io_thread_, connect_,
+            socket_, options_, protocol_, address_);
+        break;
+    case ZMQ_PUB:
+        s = new (std::nothrow) pub_session_t (io_thread_, connect_,
+            socket_, options_, protocol_, address_);
+        break;
+    case ZMQ_XPUB:
+        s = new (std::nothrow) xpub_session_t (io_thread_, connect_,
+            socket_, options_, protocol_, address_);
+        break;
+    case ZMQ_SUB:
+        s = new (std::nothrow) sub_session_t (io_thread_, connect_,
+            socket_, options_, protocol_, address_);
+        break;
+    case ZMQ_XSUB:
+        s = new (std::nothrow) xsub_session_t (io_thread_, connect_,
+            socket_, options_, protocol_, address_);
+        break;
+    case ZMQ_PUSH:
+        s = new (std::nothrow) push_session_t (io_thread_, connect_,
+            socket_, options_, protocol_, address_);
+        break;
+    case ZMQ_PULL:
+        s = new (std::nothrow) pull_session_t (io_thread_, connect_,
+            socket_, options_, protocol_, address_);
+        break;
+    case ZMQ_ROUTER:
+        s = new (std::nothrow) router_session_t (io_thread_, connect_,
+            socket_, options_, protocol_, address_);
+        break;
+    case ZMQ_PAIR:
+        s = new (std::nothrow) pair_session_t (io_thread_, connect_,
+            socket_, options_, protocol_, address_);
+        break;
+    default:
+        errno = EINVAL;
+        return NULL;
+    }
+    alloc_assert (s);
+    return s;
+}
+
+zmq::session_base_t::session_base_t (class io_thread_t *io_thread_,
+      bool connect_, class socket_base_t *socket_, const options_t &options_,
       const char *protocol_, const char *address_) :
     own_t (io_thread_, options_),
     io_object_t (io_thread_),
@@ -50,7 +124,7 @@ zmq::session_t::session_t (class io_thread_t *io_thread_, bool connect_,
         address = address_;
 }
 
-zmq::session_t::~session_t ()
+zmq::session_base_t::~session_base_t ()
 {
     zmq_assert (!pipe);
 
@@ -65,7 +139,7 @@ zmq::session_t::~session_t ()
         engine->terminate ();
 }
 
-void zmq::session_t::attach_pipe (pipe_t *pipe_)
+void zmq::session_base_t::attach_pipe (pipe_t *pipe_)
 {
     zmq_assert (!is_terminating ());
     zmq_assert (!pipe);
@@ -74,7 +148,7 @@ void zmq::session_t::attach_pipe (pipe_t *pipe_)
     pipe->set_event_sink (this);
 }
 
-bool zmq::session_t::read (msg_t *msg_)
+bool zmq::session_base_t::read (msg_t *msg_)
 {
     if (!pipe)
         return false;
@@ -87,7 +161,7 @@ bool zmq::session_t::read (msg_t *msg_)
     return true;
 }
 
-bool zmq::session_t::write (msg_t *msg_)
+bool zmq::session_base_t::write (msg_t *msg_)
 {
     if (pipe && pipe->write (msg_)) {
         int rc = msg_->init ();
@@ -98,13 +172,13 @@ bool zmq::session_t::write (msg_t *msg_)
     return false;
 }
 
-void zmq::session_t::flush ()
+void zmq::session_base_t::flush ()
 {
     if (pipe)
         pipe->flush ();
 }
 
-void zmq::session_t::clean_pipes ()
+void zmq::session_base_t::clean_pipes ()
 {
     if (pipe) {
 
@@ -128,7 +202,7 @@ void zmq::session_t::clean_pipes ()
     }
 }
 
-void zmq::session_t::terminated (pipe_t *pipe_)
+void zmq::session_base_t::terminated (pipe_t *pipe_)
 {
     //  Drop the reference to the deallocated pipe.
     zmq_assert (pipe == pipe_);
@@ -141,7 +215,7 @@ void zmq::session_t::terminated (pipe_t *pipe_)
         proceed_with_term ();
 }
 
-void zmq::session_t::read_activated (pipe_t *pipe_)
+void zmq::session_base_t::read_activated (pipe_t *pipe_)
 {
     zmq_assert (pipe == pipe_);
 
@@ -151,7 +225,7 @@ void zmq::session_t::read_activated (pipe_t *pipe_)
         pipe->check_read ();
 }
 
-void zmq::session_t::write_activated (pipe_t *pipe_)
+void zmq::session_base_t::write_activated (pipe_t *pipe_)
 {
     zmq_assert (pipe == pipe_);
 
@@ -159,20 +233,20 @@ void zmq::session_t::write_activated (pipe_t *pipe_)
         engine->activate_in ();
 }
 
-void zmq::session_t::hiccuped (pipe_t *pipe_)
+void zmq::session_base_t::hiccuped (pipe_t *pipe_)
 {
     //  Hiccups are always sent from session to socket, not the other
     //  way round.
     zmq_assert (false);
 }
 
-void zmq::session_t::process_plug ()
+void zmq::session_base_t::process_plug ()
 {
     if (connect)
         start_connecting (false);
 }
 
-void zmq::session_t::process_attach (i_engine *engine_)
+void zmq::session_base_t::process_attach (i_engine *engine_)
 {
     //  If some other object (e.g. init) notifies us that the connection failed
     //  without creating an engine we need to start the reconnection process.
@@ -208,7 +282,7 @@ void zmq::session_t::process_attach (i_engine *engine_)
     engine->plug (io_thread, this);
 }
 
-void zmq::session_t::detach ()
+void zmq::session_base_t::detach ()
 {
     //  Engine is dead. Let's forget about it.
     engine = NULL;
@@ -224,7 +298,7 @@ void zmq::session_t::detach ()
         pipe->check_read ();
 }
 
-void zmq::session_t::process_term (int linger_)
+void zmq::session_base_t::process_term (int linger_)
 {
     zmq_assert (!pending);
 
@@ -257,7 +331,7 @@ void zmq::session_t::process_term (int linger_)
     pipe->check_read ();
 }
 
-void zmq::session_t::proceed_with_term ()
+void zmq::session_base_t::proceed_with_term ()
 {
     //  The pending phase have just ended.
     pending = false;
@@ -266,7 +340,7 @@ void zmq::session_t::proceed_with_term ()
     own_t::process_term (0);
 }
 
-void zmq::session_t::timer_event (int id_)
+void zmq::session_base_t::timer_event (int id_)
 {
     //  Linger period expired. We can proceed with termination even though
     //  there are still pending messages to be sent.
@@ -278,7 +352,7 @@ void zmq::session_t::timer_event (int id_)
     pipe->terminate (false);
 }
 
-void zmq::session_t::detached ()
+void zmq::session_base_t::detached ()
 {
     //  Transient session self-destructs after peer disconnects.
     if (!connect) {
@@ -295,7 +369,7 @@ void zmq::session_t::detached ()
         pipe->hiccup ();  
 }
 
-void zmq::session_t::start_connecting (bool wait_)
+void zmq::session_base_t::start_connecting (bool wait_)
 {
     zmq_assert (connect);
 
diff --git a/src/session.hpp b/src/session_base.hpp
index a155357..175a11d 100644
--- a/src/session.hpp
+++ b/src/session_base.hpp
@@ -18,8 +18,8 @@
     along with this program.  If not, see <http://www.gnu.org/licenses/>.
 */
 
-#ifndef __ZMQ_SESSION_HPP_INCLUDED__
-#define __ZMQ_SESSION_HPP_INCLUDED__
+#ifndef __ZMQ_SESSION_BASE_HPP_INCLUDED__
+#define __ZMQ_SESSION_BASE_HPP_INCLUDED__
 
 #include <string>
 
@@ -31,16 +31,18 @@
 namespace zmq
 {
 
-    class session_t :
+    class session_base_t :
         public own_t,
         public io_object_t,
         public i_pipe_events
     {
     public:
 
-        session_t (class io_thread_t *io_thread_, bool connect_,
-            class socket_base_t *socket_, const options_t &options_,
-            const char *protocol_, const char *address_);
+        //  Create a session of the particular type.
+        static session_base_t *create (class io_thread_t *io_thread_,
+            bool connect_, class socket_base_t *socket_,
+            const options_t &options_, const char *protocol_,
+            const char *address_);
 
         //  To be used once only, when creating the session.
         void attach_pipe (class pipe_t *pipe_);
@@ -57,9 +59,14 @@ namespace zmq
         void hiccuped (class pipe_t *pipe_);
         void terminated (class pipe_t *pipe_);
 
-    private:
+    protected:
 
-        ~session_t ();
+        session_base_t (class io_thread_t *io_thread_, bool connect_,
+            class socket_base_t *socket_, const options_t &options_,
+            const char *protocol_, const char *address_);
+        ~session_base_t ();
+
+    private:
 
         void start_connecting (bool wait_);
 
@@ -115,8 +122,8 @@ namespace zmq
         std::string protocol;
         std::string address;
 
-        session_t (const session_t&);
-        const session_t &operator = (const session_t&);
+        session_base_t (const session_base_t&);
+        const session_base_t &operator = (const session_base_t&);
     };
 
 }
diff --git a/src/socket_base.cpp b/src/socket_base.cpp
index 4209a69..a4d89db 100644
--- a/src/socket_base.cpp
+++ b/src/socket_base.cpp
@@ -39,7 +39,7 @@
 #include "vtcp_listener.hpp"
 #include "tcp_connecter.hpp"
 #include "io_thread.hpp"
-#include "session.hpp"
+#include "session_base.hpp"
 #include "config.hpp"
 #include "clock.hpp"
 #include "pipe.hpp"
@@ -480,9 +480,9 @@ int zmq::socket_base_t::connect (const char *addr_)
     }
 
     //  Create session.
-    session_t *session = new (std::nothrow) session_t (
-        io_thread, true, this, options, protocol.c_str (), address.c_str ());
-    alloc_assert (session);
+    session_base_t *session = session_base_t::create (io_thread, true, this,
+        options, protocol.c_str (), address.c_str ());
+    errno_assert (session);
 
     //  Create a bi-directional pipe.
     object_t *parents [2] = {this, session};
diff --git a/src/stream_engine.cpp b/src/stream_engine.cpp
index 15e7c21..2647795 100644
--- a/src/stream_engine.cpp
+++ b/src/stream_engine.cpp
@@ -36,7 +36,7 @@
 
 #include "stream_engine.hpp"
 #include "io_thread.hpp"
-#include "session.hpp"
+#include "session_base.hpp"
 #include "config.hpp"
 #include "err.hpp"
 #include "ip.hpp"
@@ -102,7 +102,8 @@ zmq::stream_engine_t::~stream_engine_t ()
     }
 }
 
-void zmq::stream_engine_t::plug (io_thread_t *io_thread_, session_t *session_)
+void zmq::stream_engine_t::plug (io_thread_t *io_thread_,
+    session_base_t *session_)
 {
     zmq_assert (!plugged);
     plugged = true;
diff --git a/src/stream_engine.hpp b/src/stream_engine.hpp
index ac9a5be..92fc55f 100644
--- a/src/stream_engine.hpp
+++ b/src/stream_engine.hpp
@@ -44,7 +44,8 @@ namespace zmq
         ~stream_engine_t ();
 
         //  i_engine interface implementation.
-        void plug (class io_thread_t *io_thread_, class session_t *session_);
+        void plug (class io_thread_t *io_thread_,
+           class session_base_t *session_);
         void unplug ();
         void terminate ();
         void activate_in ();
@@ -84,10 +85,10 @@ namespace zmq
         encoder_t encoder;
 
         //  The session this engine is attached to.
-        class session_t *session;
+        class session_base_t *session;
 
         //  Detached transient session.
-        class session_t *leftover_session;
+        class session_base_t *leftover_session;
 
         options_t options;
 
diff --git a/src/sub.cpp b/src/sub.cpp
index 81082a2..d9f2f2e 100644
--- a/src/sub.cpp
+++ b/src/sub.cpp
@@ -79,3 +79,15 @@ bool zmq::sub_t::xhas_out ()
     return false;
 }
 
+zmq::sub_session_t::sub_session_t (io_thread_t *io_thread_, bool connect_,
+      socket_base_t *socket_, const options_t &options_,
+      const char *protocol_, const char *address_) :
+    xsub_session_t (io_thread_, connect_, socket_, options_, protocol_,
+        address_)
+{
+}
+
+zmq::sub_session_t::~sub_session_t ()
+{
+}
+
diff --git a/src/sub.hpp b/src/sub.hpp
index b5980ba..7d3cf0b 100644
--- a/src/sub.hpp
+++ b/src/sub.hpp
@@ -45,6 +45,21 @@ namespace zmq
         const sub_t &operator = (const sub_t&);
     };
 
+    class sub_session_t : public xsub_session_t
+    {
+    public:
+
+        sub_session_t (class io_thread_t *io_thread_, bool connect_,
+            class socket_base_t *socket_, const options_t &options_,
+            const char *protocol_, const char *address_);
+        ~sub_session_t ();
+
+    private:
+
+        sub_session_t (const sub_session_t&);
+        const sub_session_t &operator = (const sub_session_t&);
+    };
+
 }
 
 #endif
diff --git a/src/tcp_connecter.cpp b/src/tcp_connecter.cpp
index 487f4f5..fe99252 100644
--- a/src/tcp_connecter.cpp
+++ b/src/tcp_connecter.cpp
@@ -46,7 +46,7 @@
 #endif
 
 zmq::tcp_connecter_t::tcp_connecter_t (class io_thread_t *io_thread_,
-      class session_t *session_, const options_t &options_,
+      class session_base_t *session_, const options_t &options_,
       const char *address_, bool wait_) :
     own_t (io_thread_, options_),
     io_object_t (io_thread_),
diff --git a/src/tcp_connecter.hpp b/src/tcp_connecter.hpp
index 17f8a70..d1a93cd 100644
--- a/src/tcp_connecter.hpp
+++ b/src/tcp_connecter.hpp
@@ -37,7 +37,7 @@ namespace zmq
         //  If 'delay' is true connecter first waits for a while, then starts
         //  connection process.
         tcp_connecter_t (class io_thread_t *io_thread_,
-            class session_t *session_, const options_t &options_,
+            class session_base_t *session_, const options_t &options_,
             const char *address_, bool delay_);
         ~tcp_connecter_t ();
 
@@ -97,7 +97,7 @@ namespace zmq
         bool wait;
 
         //  Reference to the session we belong to.
-        class session_t *session;
+        class session_base_t *session;
 
         //  Current reconnect ivl, updated for backoff strategy
         int current_reconnect_ivl;
diff --git a/src/tcp_listener.cpp b/src/tcp_listener.cpp
index da476a4..9b6068c 100644
--- a/src/tcp_listener.cpp
+++ b/src/tcp_listener.cpp
@@ -26,7 +26,7 @@
 #include "tcp_listener.hpp"
 #include "stream_engine.hpp"
 #include "io_thread.hpp"
-#include "session.hpp"
+#include "session_base.hpp"
 #include "config.hpp"
 #include "err.hpp"
 #include "ip.hpp"
@@ -97,9 +97,9 @@ void zmq::tcp_listener_t::in_event ()
     zmq_assert (io_thread);
 
     //  Create and launch a session object. 
-    session_t *session = new (std::nothrow)
-        session_t (io_thread, false, socket, options, NULL, NULL);
-    alloc_assert (session);
+    session_base_t *session = session_base_t::create (io_thread, false, socket,
+        options, NULL, NULL);
+    errno_assert (session);
     session->inc_seqnum ();
     launch_child (session);
     send_attach (session, engine, false);
diff --git a/src/vtcp_connecter.cpp b/src/vtcp_connecter.cpp
index 3d2900e..5dc147e 100644
--- a/src/vtcp_connecter.cpp
+++ b/src/vtcp_connecter.cpp
@@ -50,7 +50,7 @@
 #endif
 
 zmq::vtcp_connecter_t::vtcp_connecter_t (class io_thread_t *io_thread_,
-      class session_t *session_, const options_t &options_,
+      class session_base_t *session_, const options_t &options_,
       const char *address_, bool wait_) :
     own_t (io_thread_, options_),
     io_object_t (io_thread_),
diff --git a/src/vtcp_connecter.hpp b/src/vtcp_connecter.hpp
index f467b5f..fe5260e 100644
--- a/src/vtcp_connecter.hpp
+++ b/src/vtcp_connecter.hpp
@@ -43,7 +43,7 @@ namespace zmq
         //  If 'delay' is true connecter first waits for a while, then starts
         //  connection process.
         vtcp_connecter_t (class io_thread_t *io_thread_,
-            class session_t *session_, const options_t &options_,
+            class session_base_t *session_, const options_t &options_,
             const char *address_, bool delay_);
         ~vtcp_connecter_t ();
 
@@ -104,7 +104,7 @@ namespace zmq
         bool wait;
 
         //  Reference to the session we belong to.
-        class session_t *session;
+        class session_base_t *session;
 
         //  Current reconnect ivl, updated for backoff strategy
         int current_reconnect_ivl;
diff --git a/src/vtcp_listener.cpp b/src/vtcp_listener.cpp
index b394833..7e496e5 100644
--- a/src/vtcp_listener.cpp
+++ b/src/vtcp_listener.cpp
@@ -27,7 +27,7 @@
 #include <vtcp.h>
 
 #include "stream_engine.hpp"
-#include "session.hpp"
+#include "session_base.hpp"
 #include "stdint.hpp"
 #include "err.hpp"
 #include "ip.hpp"
@@ -113,8 +113,8 @@ void zmq::vtcp_listener_t::in_event ()
     zmq_assert (io_thread);
 
     //  Create and launch a session object. 
-    session_t *session = new (std::nothrow)
-        session_t (io_thread, false, socket, options, NULL, NULL);
+    session_base_t *session = session_base_t::create (io_thread, false, socket,
+        options, NULL, NULL);
     alloc_assert (session);
     session->inc_seqnum ();
     launch_child (session);
diff --git a/src/xpub.cpp b/src/xpub.cpp
index 8da9cf9..a245fea 100644
--- a/src/xpub.cpp
+++ b/src/xpub.cpp
@@ -169,3 +169,15 @@ void zmq::xpub_t::send_unsubscription (unsigned char *data_, size_t size_,
     }
 }
 
+zmq::xpub_session_t::xpub_session_t (io_thread_t *io_thread_, bool connect_,
+      socket_base_t *socket_, const options_t &options_,
+      const char *protocol_, const char *address_) :
+    session_base_t (io_thread_, connect_, socket_, options_, protocol_,
+        address_)
+{
+}
+
+zmq::xpub_session_t::~xpub_session_t ()
+{
+}
+
diff --git a/src/xpub.hpp b/src/xpub.hpp
index 001fa2d..b410e6c 100644
--- a/src/xpub.hpp
+++ b/src/xpub.hpp
@@ -25,6 +25,7 @@
 #include <string>
 
 #include "socket_base.hpp"
+#include "session_base.hpp"
 #include "mtrie.hpp"
 #include "array.hpp"
 #include "dist.hpp"
@@ -79,6 +80,21 @@ namespace zmq
         const xpub_t &operator = (const xpub_t&);
     };
 
+    class xpub_session_t : public session_base_t
+    {
+    public:
+
+        xpub_session_t (class io_thread_t *io_thread_, bool connect_,
+            class socket_base_t *socket_, const options_t &options_,
+            const char *protocol_, const char *address_);
+        ~xpub_session_t ();
+
+    private:
+
+        xpub_session_t (const xpub_session_t&);
+        const xpub_session_t &operator = (const xpub_session_t&);
+    };
+
 }
 
 #endif
diff --git a/src/xrep.cpp b/src/xrep.cpp
index a11b8c1..c304463 100644
--- a/src/xrep.cpp
+++ b/src/xrep.cpp
@@ -243,5 +243,15 @@ bool zmq::xrep_t::xhas_out ()
     return true;
 }
 
+zmq::xrep_session_t::xrep_session_t (io_thread_t *io_thread_, bool connect_,
+      socket_base_t *socket_, const options_t &options_,
+      const char *protocol_, const char *address_) :
+    session_base_t (io_thread_, connect_, socket_, options_, protocol_,
+        address_)
+{
+}
 
+zmq::xrep_session_t::~xrep_session_t ()
+{
+}
 
diff --git a/src/xrep.hpp b/src/xrep.hpp
index 07f10ba..562f87d 100644
--- a/src/xrep.hpp
+++ b/src/xrep.hpp
@@ -24,6 +24,7 @@
 #include <map>
 
 #include "socket_base.hpp"
+#include "session_base.hpp"
 #include "stdint.hpp"
 #include "msg.hpp"
 #include "fq.hpp"
@@ -93,6 +94,21 @@ namespace zmq
         const xrep_t &operator = (const xrep_t&);
     };
 
+    class xrep_session_t : public session_base_t
+    {
+    public:
+
+        xrep_session_t (class io_thread_t *io_thread_, bool connect_,
+            class socket_base_t *socket_, const options_t &options_,
+            const char *protocol_, const char *address_);
+        ~xrep_session_t ();
+
+    private:
+
+        xrep_session_t (const xrep_session_t&);
+        const xrep_session_t &operator = (const xrep_session_t&);
+    };
+
 }
 
 #endif
diff --git a/src/xreq.cpp b/src/xreq.cpp
index 7b66137..79b3b94 100644
--- a/src/xreq.cpp
+++ b/src/xreq.cpp
@@ -79,3 +79,15 @@ void zmq::xreq_t::xterminated (pipe_t *pipe_)
     lb.terminated (pipe_);
 }
 
+zmq::xreq_session_t::xreq_session_t (io_thread_t *io_thread_, bool connect_,
+      socket_base_t *socket_, const options_t &options_,
+      const char *protocol_, const char *address_) :
+    session_base_t (io_thread_, connect_, socket_, options_, protocol_,
+        address_)
+{
+}
+
+zmq::xreq_session_t::~xreq_session_t ()
+{
+}
+
diff --git a/src/xreq.hpp b/src/xreq.hpp
index a427ba9..d7e28c4 100644
--- a/src/xreq.hpp
+++ b/src/xreq.hpp
@@ -23,6 +23,7 @@
 #define __ZMQ_XREQ_HPP_INCLUDED__
 
 #include "socket_base.hpp"
+#include "session_base.hpp"
 #include "fq.hpp"
 #include "lb.hpp"
 
@@ -60,6 +61,21 @@ namespace zmq
         const xreq_t &operator = (const xreq_t&);
     };
 
+    class xreq_session_t : public session_base_t
+    {
+    public:
+
+        xreq_session_t (class io_thread_t *io_thread_, bool connect_,
+            class socket_base_t *socket_, const options_t &options_,
+            const char *protocol_, const char *address_);
+        ~xreq_session_t ();
+
+    private:
+
+        xreq_session_t (const xreq_session_t&);
+        const xreq_session_t &operator = (const xreq_session_t&);
+    };
+
 }
 
 #endif
diff --git a/src/xsub.cpp b/src/xsub.cpp
index 4122c67..b24f082 100644
--- a/src/xsub.cpp
+++ b/src/xsub.cpp
@@ -213,4 +213,15 @@ void zmq::xsub_t::send_subscription (unsigned char *data_, size_t size_,
     zmq_assert (sent);
 }
 
+zmq::xsub_session_t::xsub_session_t (io_thread_t *io_thread_, bool connect_,
+      socket_base_t *socket_, const options_t &options_,
+      const char *protocol_, const char *address_) :
+    session_base_t (io_thread_, connect_, socket_, options_, protocol_,
+        address_)
+{
+}
+
+zmq::xsub_session_t::~xsub_session_t ()
+{
+}
 
diff --git a/src/xsub.hpp b/src/xsub.hpp
index ea59cdb..310df6e 100644
--- a/src/xsub.hpp
+++ b/src/xsub.hpp
@@ -22,6 +22,7 @@
 #define __ZMQ_XSUB_HPP_INCLUDED__
 
 #include "socket_base.hpp"
+#include "session_base.hpp"
 #include "dist.hpp"
 #include "fq.hpp"
 #include "trie.hpp"
@@ -83,6 +84,21 @@ namespace zmq
         const xsub_t &operator = (const xsub_t&);
     };
 
+    class xsub_session_t : public session_base_t
+    {
+    public:
+
+        xsub_session_t (class io_thread_t *io_thread_, bool connect_,
+            class socket_base_t *socket_, const options_t &options_,
+            const char *protocol_, const char *address_);
+        ~xsub_session_t ();
+
+    private:
+
+        xsub_session_t (const xsub_session_t&);
+        const xsub_session_t &operator = (const xsub_session_t&);
+    };
+
 }
 
 #endif