diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/own.hpp | 12 | ||||
| -rw-r--r-- | src/socket_base.cpp | 52 | ||||
| -rw-r--r-- | src/socket_base.hpp | 9 | ||||
| -rw-r--r-- | src/xs.cpp | 11 | ||||
| -rw-r--r-- | src/xszmq.cpp | 6 |
5 files changed, 79 insertions, 11 deletions
diff --git a/src/own.hpp b/src/own.hpp index a2f0e9f..2d47a39 100644 --- a/src/own.hpp +++ b/src/own.hpp @@ -67,6 +67,12 @@ namespace xs protected: + // Handlers for incoming commands. + void process_own (own_t *object_); + void process_term_req (own_t *object_); + void process_term_ack (); + void process_seqnum (); + // Launch the supplied object and become its owner. void launch_child (own_t *object_); @@ -101,12 +107,6 @@ namespace xs // Set owner of the object void set_owner (own_t *owner_); - // Handlers for incoming commands. - void process_own (own_t *object_); - void process_term_req (own_t *object_); - void process_term_ack (); - void process_seqnum (); - // Check whether all the peding term acks were delivered. // If so, deallocate this object. void check_term_acks (); diff --git a/src/socket_base.cpp b/src/socket_base.cpp index 3e57c28..ec5c8bd 100644 --- a/src/socket_base.cpp +++ b/src/socket_base.cpp @@ -347,7 +347,12 @@ int xs::socket_base_t::bind (const char *addr_) if (protocol == "inproc") { endpoint_t endpoint = {this, options}; rc = register_endpoint (addr_, endpoint); - return rc; + if (rc != 0) + return -1; + + // Endpoint IDs for inproc transport are not implemented at the + // moment. Thus we return 0 to the user. + return 0; } if (protocol == "pgm" || protocol == "epgm") { @@ -373,7 +378,7 @@ int xs::socket_base_t::bind (const char *addr_) return -1; } launch_child (listener); - return 0; + return add_endpoint (listener); } #if !defined XS_HAVE_WINDOWS && !defined XS_HAVE_OPENVMS @@ -387,7 +392,7 @@ int xs::socket_base_t::bind (const char *addr_) return -1; } launch_child (listener); - return 0; + return add_endpoint (listener); } #endif @@ -478,6 +483,7 @@ int xs::socket_base_t::connect (const char *addr_) // increased here. send_bind (peer.socket, ppair [1], false); + // Inproc endpoints are not yet implemented thus we return 0. return 0; } @@ -512,7 +518,32 @@ int xs::socket_base_t::connect (const char *addr_) // Activate the session. Make it a child of this socket. launch_child (session); + return add_endpoint (session); +} + +int xs::socket_base_t::shutdown (int how_) +{ + // Check whether the library haven't been shut down yet. + if (unlikely (ctx_terminated)) { + errno = ETERM; + return -1; + } + // Endpoint ID means 'shutdown not implemented'. + if (how_ <= 0) { + errno = ENOTSUP; + return -1; + } + + // Find the endpoint corresponding to the ID. + endpoints_t::iterator it = endpoints.find (how_); + if (it == endpoints.end ()) { + errno = EINVAL; + return -1; + } + + process_term_req (it->second); + endpoints.erase (it); return 0; } @@ -937,3 +968,18 @@ uint64_t xs::socket_base_t::now_ms () { return clock.now_ms (); } + +int xs::socket_base_t::add_endpoint (own_t *endpoint_) +{ + // Get a unique endpoint ID. + int id = 1; + for (endpoints_t::iterator it = endpoints.begin (); it != endpoints.end (); + ++it, ++id) + if (it->first != id) + break; + + // Remember the endpoint. + endpoints.insert (std::make_pair (id, endpoint_)); + return id; +} + diff --git a/src/socket_base.hpp b/src/socket_base.hpp index 54a5f0a..f73c413 100644 --- a/src/socket_base.hpp +++ b/src/socket_base.hpp @@ -23,6 +23,7 @@ #ifndef __XS_SOCKET_BASE_HPP_INCLUDED__ #define __XS_SOCKET_BASE_HPP_INCLUDED__ +#include <map> #include <string> #include "own.hpp" @@ -71,6 +72,7 @@ namespace xs int getsockopt (int option_, void *optval_, size_t *optvallen_); int bind (const char *addr_); int connect (const char *addr_); + int shutdown (int how_); int send (xs::msg_t *msg_, int flags_); int recv (xs::msg_t *msg_, int flags_); int close (); @@ -151,6 +153,9 @@ namespace xs // to be later retrieved by getsockopt. void extract_flags (msg_t *msg_); + // Creates new endpoint ID and adds the endpoint to the map. + int add_endpoint (own_t *endpoint_); + // Used to check whether the object is a socket. uint32_t tag; @@ -209,6 +214,10 @@ namespace xs // Improves efficiency of time measurement. clock_t clock; + // Map of open endpoints. + typedef std::map <int, own_t*> endpoints_t; + endpoints_t endpoints; + socket_base_t (const socket_base_t&); const socket_base_t &operator = (const socket_base_t&); }; @@ -212,6 +212,17 @@ int xs_connect (void *s_, const char *addr_) return rc; } +int xs_shutdown (void *s_, int how_) +{ + xs::socket_base_t *s = (xs::socket_base_t*) s_; + if (!s || !s->check_tag ()) { + errno = ENOTSOCK; + return -1; + } + int rc = s->shutdown (how_); + return rc; +} + int xs_send (void *s_, const void *buf_, size_t len_, int flags_) { xs_msg_t msg; diff --git a/src/xszmq.cpp b/src/xszmq.cpp index d7199f9..f6e6fa1 100644 --- a/src/xszmq.cpp +++ b/src/xszmq.cpp @@ -355,12 +355,14 @@ int zmq_getsockopt (void *s, int option, void *optval, int zmq_bind (void *s, const char *addr) { - return xs_bind (s, addr); + int rc = xs_bind (s, addr); + return rc < 0 ? -1 : 0; } int zmq_connect (void *s, const char *addr) { - return xs_connect (s, addr); + int rc = xs_connect (s, addr); + return rc < 0 ? -1 : 0; } int zmq_send (void *s, zmq_msg_t *msg, int flags) |