diff options
Diffstat (limited to 'src/object.hpp')
| -rw-r--r-- | src/object.hpp | 50 |
1 files changed, 28 insertions, 22 deletions
diff --git a/src/object.hpp b/src/object.hpp index a38b0a6..bc1b325 100644 --- a/src/object.hpp +++ b/src/object.hpp @@ -20,6 +20,8 @@ #ifndef __ZMQ_OBJECT_HPP_INCLUDED__ #define __ZMQ_OBJECT_HPP_INCLUDED__ +#include "../include/zmq.h" + #include "stdint.hpp" #include "blob.hpp" @@ -32,18 +34,14 @@ namespace zmq { public: - object_t (class ctx_t *ctx_, uint32_t thread_slot_); + object_t (class ctx_t *ctx_, uint32_t slot_); object_t (object_t *parent_); virtual ~object_t (); - uint32_t get_thread_slot (); + uint32_t get_slot (); ctx_t *get_ctx (); void process_command (struct command_t &cmd_); - // Allow pipe to access corresponding context functions. - void register_pipe (class pipe_t *pipe_); - void unregister_pipe (class pipe_t *pipe_); - protected: // Using following function, socket is able to access global @@ -52,45 +50,53 @@ namespace zmq void unregister_endpoints (class socket_base_t *socket_); class socket_base_t *find_endpoint (const char *addr_); + // Logs an message. + void log (zmq_msg_t *msg_); + // Chooses least loaded I/O thread. - class io_thread_t *choose_io_thread (uint64_t taskset_); + class io_thread_t *choose_io_thread (uint64_t affinity_); + + // Zombify particular socket. In other words, pass the ownership to + // the context. + void zombify_socket (class socket_base_t *socket_); // Derived object can use these functions to send commands // to other objects. void send_stop (); - void send_plug (class owned_t *destination_, bool inc_seqnum_ = true); - void send_own (class socket_base_t *destination_, - class owned_t *object_); + void send_plug (class own_t *destination_, + bool inc_seqnum_ = true); + void send_own (class own_t *destination_, + class own_t *object_); void send_attach (class session_t *destination_, struct i_engine *engine_, const blob_t &peer_identity_, bool inc_seqnum_ = true); - void send_bind (class socket_base_t *destination_, + void send_bind (class own_t *destination_, class reader_t *in_pipe_, class writer_t *out_pipe_, const blob_t &peer_identity_, bool inc_seqnum_ = true); - void send_revive (class object_t *destination_); - void send_reader_info (class writer_t *destination_, + void send_activate_reader (class reader_t *destination_); + void send_activate_writer (class writer_t *destination_, uint64_t msgs_read_); void send_pipe_term (class writer_t *destination_); void send_pipe_term_ack (class reader_t *destination_); - void send_term_req (class socket_base_t *destination_, - class owned_t *object_); - void send_term (class owned_t *destination_); - void send_term_ack (class socket_base_t *destination_); + void send_term_req (class own_t *destination_, + class own_t *object_); + void send_term (class own_t *destination_); + void send_term_ack (class own_t *destination_); // These handlers can be overloaded by the derived objects. They are // called when command arrives from another thread. virtual void process_stop (); virtual void process_plug (); - virtual void process_own (class owned_t *object_); + virtual void process_own (class own_t *object_); virtual void process_attach (struct i_engine *engine_, const blob_t &peer_identity_); virtual void process_bind (class reader_t *in_pipe_, class writer_t *out_pipe_, const blob_t &peer_identity_); - virtual void process_revive (); - virtual void process_reader_info (uint64_t msgs_read_); + virtual void process_activate_reader (); + virtual void process_activate_writer (uint64_t msgs_read_); virtual void process_pipe_term (); virtual void process_pipe_term_ack (); - virtual void process_term_req (class owned_t *object_); + virtual void process_term_req (class own_t *object_); virtual void process_term (); virtual void process_term_ack (); @@ -105,7 +111,7 @@ namespace zmq class ctx_t *ctx; // Slot ID of the thread the object belongs to. - uint32_t thread_slot; + uint32_t slot; void send_command (command_t &cmd_); |