summaryrefslogtreecommitdiff
path: root/src/socket_base.hpp
diff options
context:
space:
mode:
Diffstat (limited to 'src/socket_base.hpp')
-rw-r--r--src/socket_base.hpp68
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&);