Decouple encoder_t and decoder_t from session_base_t

This patch introduces i_msg_sink and i_msg_source interfaces. This
allows us to make message encoder and decoder more general.

src/decoder.cpp

@@ -29,14 +29,14 @@
 #endif
 
 #include "decoder.hpp"
-#include "session_base.hpp"
+#include "i_msg_sink.hpp"
 #include "likely.hpp"
 #include "wire.hpp"
 #include "err.hpp"
 
 zmq::decoder_t::decoder_t (size_t bufsize_, int64_t maxmsgsize_) :
     decoder_base_t <decoder_t> (bufsize_),
-    session (NULL),
+    msg_sink (NULL),
     maxmsgsize (maxmsgsize_)
 {
     int rc = in_progress.init ();
@@ -52,9 +52,9 @@ zmq::decoder_t::~decoder_t ()
     errno_assert (rc == 0);
 }
 
-void zmq::decoder_t::set_session (session_base_t *session_)
+void zmq::decoder_t::set_msg_sink (i_msg_sink *msg_sink_)
 {
-    session = session_;
+    msg_sink = msg_sink_;
 }
 
 bool zmq::decoder_t::stalled () const
@@ -157,9 +157,9 @@ bool zmq::decoder_t::message_ready ()
 {
     //  Message is completely read. Push it further and start reading
     //  new message. (in_progress is a 0-byte message after this point.)
-    if (unlikely (!session))
+    if (unlikely (!msg_sink))
         return false;
-    int rc = session->write (&in_progress);
+    int rc = msg_sink->push_msg (&in_progress);
     if (unlikely (rc != 0)) {
         if (errno != EAGAIN)
             decoding_error ();

src/decoder.hpp

@@ -34,7 +34,7 @@
 namespace zmq
 {
 
-    class session_base_t;
+    class i_msg_sink;
 
     //  Helper base class for decoders that know the amount of data to read
     //  in advance at any moment. Knowing the amount in advance is a property
@@ -195,7 +195,8 @@ namespace zmq
         decoder_t (size_t bufsize_, int64_t maxmsgsize_);
         ~decoder_t ();
 
-        void set_session (zmq::session_base_t *session_);
+        //  Set the receiver of decoded messages.
+        void set_msg_sink (i_msg_sink *msg_sink_);
 
         //  Returns true if there is a decoded message
         //  waiting to be delivered to the session.
@@ -208,7 +209,7 @@ namespace zmq
         bool flags_ready ();
         bool message_ready ();
 
-        zmq::session_base_t *session;
+        i_msg_sink *msg_sink;
         unsigned char tmpbuf [8];
         msg_t in_progress;
 

src/encoder.cpp

@@ -21,13 +21,13 @@
 */
 
 #include "encoder.hpp"
-#include "session_base.hpp"
+#include "i_msg_source.hpp"
 #include "likely.hpp"
 #include "wire.hpp"
 
 zmq::encoder_t::encoder_t (size_t bufsize_) :
     encoder_base_t <encoder_t> (bufsize_),
-    session (NULL)
+    msg_source (NULL)
 {
     int rc = in_progress.init ();
     errno_assert (rc == 0);
@@ -42,9 +42,9 @@ zmq::encoder_t::~encoder_t ()
     errno_assert (rc == 0);
 }
 
-void zmq::encoder_t::set_session (session_base_t *session_)
+void zmq::encoder_t::set_msg_source (i_msg_source *msg_source_)
 {
-    session = session_;
+    msg_source = msg_source_;
 }
 
 bool zmq::encoder_t::size_ready ()
@@ -65,12 +65,12 @@ bool zmq::encoder_t::message_ready ()
     //  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 (unlikely (!session)) {
+    if (unlikely (!msg_source)) {
         rc = in_progress.init ();
         errno_assert (rc == 0);
         return false;
     }
-    rc = session->read (&in_progress);
+    rc = msg_source->pull_msg (&in_progress);
     if (unlikely (rc != 0)) {
         errno_assert (errno == EAGAIN);
         rc = in_progress.init ();

src/encoder.hpp

@@ -33,7 +33,7 @@
 namespace zmq
 {
 
-    class session_base_t;
+    class i_msg_source;
 
     //  Helper base class for encoders. It implements the state machine that
     //  fills the outgoing buffer. Derived classes should implement individual
@@ -168,14 +168,14 @@ namespace zmq
         encoder_t (size_t bufsize_);
         ~encoder_t ();
 
-        void set_session (zmq::session_base_t *session_);
+        void set_msg_source (i_msg_source *msg_source_);
 
     private:
 
         bool size_ready ();
         bool message_ready ();
 
-        zmq::session_base_t *session;
+        i_msg_source *msg_source;
         msg_t in_progress;
         unsigned char tmpbuf [10];
 

src/i_msg_sink.hpp

@@ -0,0 +1,43 @@
+/*
+    Copyright (c) 2007-2012 iMatix Corporation
+    Copyright (c) 2007-2012 Other contributors as noted in the AUTHORS file
+
+    This file is part of 0MQ.
+
+    0MQ is free software; you can redistribute it and/or modify it under
+    the terms of the GNU Lesser General Public License as published by
+    the Free Software Foundation; either version 3 of the License, or
+    (at your option) any later version.
+
+    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
+    GNU Lesser General Public License for more details.
+
+    You should have received a copy of the GNU Lesser General Public License
+    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+*/
+
+#ifndef __ZMQ_I_MSG_SINK_HPP_INCLUDED__
+#define __ZMQ_I_MSG_SINK_HPP_INCLUDED__
+
+namespace zmq
+{
+
+    //  Forward declaration
+    class msg_t;
+
+    //  Interface to be implemented by message sink.
+
+    struct i_msg_sink
+    {
+        virtual ~i_msg_sink () {}
+
+        //  Delivers a message. Returns 0 if successful; -1 otherwise.
+        //  The function takes ownership of the passed message.
+        virtual int push_msg (msg_t *msg_) = 0;
+    };
+
+}
+
+#endif

src/i_msg_source.hpp

@@ -0,0 +1,44 @@
+/*
+    Copyright (c) 2007-2012 iMatix Corporation
+    Copyright (c) 2007-2012 Other contributors as noted in the AUTHORS file
+
+    This file is part of 0MQ.
+
+    0MQ is free software; you can redistribute it and/or modify it under
+    the terms of the GNU Lesser General Public License as published by
+    the Free Software Foundation; either version 3 of the License, or
+    (at your option) any later version.
+
+    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
+    GNU Lesser General Public License for more details.
+
+    You should have received a copy of the GNU Lesser General Public License
+    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+*/
+
+#ifndef __ZMQ_I_MSG_SOURCE_HPP_INCLUDED__
+#define __ZMQ_I_MSG_SOURCE_HPP_INCLUDED__
+
+namespace zmq
+{
+
+    //  Forward declaration
+    class msg_t;
+
+    //  Interface to be implemented by message source.
+
+    struct i_msg_source
+    {
+        virtual ~i_msg_source () {}
+
+        //  Fetch a message. Returns 0 if successful; -1 otherwise.
+        //  The caller is responsible for freeing the message when no
+        //  longer used.
+        virtual int pull_msg (msg_t *msg_) = 0;
+    };
+
+}
+
+#endif

src/pgm_receiver.cpp

@@ -233,7 +233,7 @@ void zmq::pgm_receiver_t::in_event ()
             it->second.decoder = new (std::nothrow) decoder_t (0,
                 options.maxmsgsize);
             alloc_assert (it->second.decoder);
-            it->second.decoder->set_session (session);
+            it->second.decoder->set_msg_sink (session);
         }
 
         mru_decoder = it->second.decoder;

src/pgm_sender.cpp

@@ -72,7 +72,7 @@ void zmq::pgm_sender_t::plug (io_thread_t *io_thread_, session_base_t *session_)
     fd_t rdata_notify_fd = retired_fd;
     fd_t pending_notify_fd = retired_fd;
 
-    encoder.set_session (session_);
+    encoder.set_msg_source (session_);
 
     //  Fill fds from PGM transport and add them to the poller.
     pgm_socket.get_sender_fds (&downlink_socket_fd, &uplink_socket_fd,

src/req.cpp

@@ -150,27 +150,27 @@ zmq::req_session_t::~req_session_t ()
     state = options.recv_identity ? identity : bottom;
 }
 
-int zmq::req_session_t::write (msg_t *msg_)
+int zmq::req_session_t::push_msg (msg_t *msg_)
 {
     switch (state) {
     case bottom:
         if (msg_->flags () == msg_t::more && msg_->size () == 0) {
             state = body;
-            return dealer_session_t::write (msg_);
+            return dealer_session_t::push_msg (msg_);
         }
         break;
     case body:
         if (msg_->flags () == msg_t::more)
-            return dealer_session_t::write (msg_);
+            return dealer_session_t::push_msg (msg_);
         if (msg_->flags () == 0) {
             state = bottom;
-            return dealer_session_t::write (msg_);
+            return dealer_session_t::push_msg (msg_);
         }
         break;
     case identity:
         if (msg_->flags () == 0) {
             state = bottom;
-            return dealer_session_t::write (msg_);
+            return dealer_session_t::push_msg (msg_);
         }
         break;
     }

src/req.hpp

@@ -71,7 +71,7 @@ namespace zmq
         ~req_session_t ();
 
         //  Overloads of the functions from session_base_t.
-        int write (msg_t *msg_);
+        int push_msg (msg_t *msg_);
         void reset ();
 
     private:

src/session_base.cpp

@@ -150,7 +150,7 @@ void zmq::session_base_t::attach_pipe (pipe_t *pipe_)
     pipe->set_event_sink (this);
 }
 
-int zmq::session_base_t::read (msg_t *msg_)
+int zmq::session_base_t::pull_msg (msg_t *msg_)
 {
     //  First message to send is identity
     if (!identity_sent) {
@@ -172,7 +172,7 @@ int zmq::session_base_t::read (msg_t *msg_)
     return 0;
 }
 
-int zmq::session_base_t::write (msg_t *msg_)
+int zmq::session_base_t::push_msg (msg_t *msg_)
 {
     //  First message to receive is identity
     if (!identity_received) {
@@ -224,7 +224,7 @@ void zmq::session_base_t::clean_pipes ()
             msg_t msg;
             int rc = msg.init ();
             errno_assert (rc == 0);
-            if (!read (&msg)) {
+            if (!pull_msg (&msg)) {
                 zmq_assert (!incomplete_in);
                 break;
             }

src/session_base.hpp

@@ -29,6 +29,8 @@
 #include "own.hpp"
 #include "io_object.hpp"
 #include "pipe.hpp"
+#include "i_msg_source.hpp"
+#include "i_msg_sink.hpp"
 
 namespace zmq
 {
@@ -42,7 +44,9 @@ namespace zmq
     class session_base_t :
         public own_t,
         public io_object_t,
-        public i_pipe_events
+        public i_pipe_events,
+        public i_msg_source,
+        public i_msg_sink
     {
     public:
 
@@ -54,9 +58,13 @@ namespace zmq
         //  To be used once only, when creating the session.
         void attach_pipe (zmq::pipe_t *pipe_);
 
+        //  i_msg_source interface implementation.
+        virtual int pull_msg (msg_t *msg_);
+
+        //  i_msg_sink interface implementation.
+        virtual int push_msg (msg_t *msg_);
+
         //  Following functions are the interface exposed towards the engine.
-        virtual int read (msg_t *msg_);
-        virtual int write (msg_t *msg_);
         virtual void reset ();
         void flush ();
         void detach ();

src/stream_engine.cpp

@@ -156,8 +156,8 @@ void zmq::stream_engine_t::unplug ()
     io_object_t::unplug ();
 
     //  Disconnect from session object.
-    encoder.set_session (NULL);
+    encoder.set_msg_source (NULL);
-    decoder.set_session (NULL);
+    decoder.set_msg_sink (NULL);
     session = NULL;
 }
 
@@ -379,8 +379,8 @@ bool zmq::stream_engine_t::handshake ()
     //  We have received either a header of identity message
     //  or the whole greeting.
 
-    encoder.set_session (session);
+    encoder.set_msg_source (session);
-    decoder.set_session (session);
+    decoder.set_msg_sink (session);
 
     zmq_assert (greeting [0] != 0xff || greeting_bytes_read >= 10);