I/O object hierarchy implemented

1 files changed, 20 insertions, 48 deletions

diff --git a/src/socket_base.hpp b/src/socket_base.hpp
index 386fdbb..f76dc4c 100644
--- a/src/socket_base.hpp
+++ b/src/socket_base.hpp
@@ -20,13 +20,12 @@
 #ifndef __ZMQ_SOCKET_BASE_HPP_INCLUDED__
 #define __ZMQ_SOCKET_BASE_HPP_INCLUDED__
 
-#include <set>
 #include <map>
 #include <vector>
 
 #include "../include/zmq.h"
 
-#include "object.hpp"
+#include "own.hpp"
 #include "yarray_item.hpp"
 #include "mutex.hpp"
 #include "options.hpp"
@@ -35,12 +34,13 @@
 #include "signaler.hpp"
 #include "stdint.hpp"
 #include "blob.hpp"
+#include "own.hpp"
 
 namespace zmq
 {
 
     class socket_base_t :
-        public object_t,
+        public own_t,
         public yarray_item_t
     {
     public:
@@ -65,30 +65,16 @@ namespace zmq
         int recv (zmq_msg_t *msg_, int flags_);
         int close ();
 
-        //  When another owned object wants to send command to this object
-        //  it calls this function to let it know it should not shut down
-        //  before the command is delivered.
-        void inc_seqnum ();
-
         //  These functions are used by the polling mechanism to determine
         //  which events are to be reported from this socket.
         bool has_in ();
         bool has_out ();
 
-        //  The list of sessions cannot be accessed via inter-thread
-        //  commands as it is unacceptable to wait for the completion of the
-        //  action till user application yields control of the application
-        //  thread to 0MQ. Locking is used instead.
-        //  There are two distinct types of sessions: those identified by name
-        //  and those identified by ordinal number. Thus two sets of session
-        //  management functions.
+        //  Registry of named sessions.
         bool register_session (const blob_t &peer_identity_,
             class session_t *session_);
         void unregister_session (const blob_t &peer_identity_);
         class session_t *find_session (const blob_t &peer_identity_);
-        uint64_t register_session (class session_t *session_);
-        void unregister_session (uint64_t ordinal_);
-        class session_t *find_session (uint64_t ordinal_);
 
         //  i_reader_events interface implementation.
         void activated (class reader_t *pipe_);
@@ -99,7 +85,7 @@ namespace zmq
         void terminated (class writer_t *pipe_);
 
         //  This function should be called only on zombie sockets. It tries
-        //  to deallocate the zombie and returns true is successful.
+        //  to deallocate the zombie. Returns true if zombie is finally dead.
         bool dezombify ();
 
     protected:
@@ -109,11 +95,8 @@ namespace zmq
 
         //  Concrete algorithms for the x- methods are to be defined by
         //  individual socket types.
-
         virtual void xattach_pipes (class reader_t *inpipe_,
             class writer_t *outpipe_, const blob_t &peer_identity_) = 0;
-        virtual void xterm_pipes () = 0;
-        virtual bool xhas_pipes () = 0;
 
         //  The default implementation assumes there are no specific socket
         //  options for the particular socket type. If not so, overload this
@@ -132,12 +115,22 @@ namespace zmq
         //  Socket options.
         options_t options;
 
+        //  We are declaring termination handler as protected so that
+        //  individual socket types can hook into the termination process
+        //  by overloading it.
+        void process_term ();
+
+    private:
+
+//  TODO: Check whether we still need this flag...
         //  If true, socket was already closed but not yet deallocated
         //  because either shutdown is in process or there are still pipes
         //  attached to the socket.
         bool zombie;
 
-    private:
+        //  Check whether transport protocol, as specified in connect or
+        //  bind, is available and compatible with the socket type.
+        int check_protocol (const std::string &protocol_);
 
         //  If no identity set generate one and call xattach_pipes ().
         void attach_pipes (class reader_t *inpipe_, class writer_t *outpipe_,
@@ -151,12 +144,9 @@ namespace zmq
 
         //  Handlers for incoming commands.
         void process_stop ();
-        void process_own (class owned_t *object_);
         void process_bind (class reader_t *in_pipe_, class writer_t *out_pipe_,
             const blob_t &peer_identity_);
-        void process_term_req (class owned_t *object_);
-        void process_term_ack ();
-        void process_seqnum ();
+        void process_unplug ();
 
         //  App thread's signaler object.
         signaler_t signaler;
@@ -164,36 +154,18 @@ namespace zmq
         //  Timestamp of when commands were processed the last time.
         uint64_t last_processing_time;
 
-        //  List of all I/O objects owned by this socket. The socket is
-        //  responsible for deallocating them before it quits.
-        typedef std::set <class owned_t*> io_objects_t;
-        io_objects_t io_objects;
-
-        //  Number of I/O objects that were already asked to terminate
-        //  but haven't acknowledged it yet.
-        int pending_term_acks;
-
         //  Number of messages received since last command processing.
         int ticks;
 
         //  If true there's a half-read message in the socket.
         bool rcvmore;
 
-        //  Sequence number of the last command sent to this object.
-        atomic_counter_t sent_seqnum;
-
-        //  Sequence number of the last command processed by this object.
-        uint64_t processed_seqnum;
-
         //  Lists of existing sessions. This lists are never referenced from
-        //  within the socket, instead they are used by I/O objects owned by
+        //  within the socket, instead they are used by objects owned by
         //  the socket. As those objects can live in different threads,
         //  the access is synchronised by mutex.
-        typedef std::map <blob_t, session_t*> named_sessions_t;
-        named_sessions_t named_sessions;
-        typedef std::map <uint64_t, session_t*> unnamed_sessions_t;
-        unnamed_sessions_t unnamed_sessions;
-        uint64_t next_ordinal;
+        typedef std::map <blob_t, session_t*> sessions_t;
+        sessions_t sessions;
         mutex_t sessions_sync;
 
         socket_base_t (const socket_base_t&);