diff options
Diffstat (limited to 'src/socket_base.hpp')
| -rw-r--r-- | src/socket_base.hpp | 68 |
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&); |