diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/Makefile.am | 6 | ||||
| -rw-r--r-- | src/config.hpp | 4 | ||||
| -rw-r--r-- | src/options.cpp | 4 | ||||
| -rw-r--r-- | src/options.hpp | 6 | ||||
| -rw-r--r-- | src/socket_base.cpp | 74 |
5 files changed, 79 insertions, 15 deletions
diff --git a/src/Makefile.am b/src/Makefile.am index 68b34fa..f4f338e 100644 --- a/src/Makefile.am +++ b/src/Makefile.am @@ -62,6 +62,8 @@ libzmq_la_SOURCES = $(pgm_sources) \ object.hpp \ options.hpp \ owned.hpp \ + pgm_sender.hpp \ + pgm_socket.hpp \ pipe.hpp \ platform.hpp \ poll.hpp \ @@ -101,6 +103,8 @@ libzmq_la_SOURCES = $(pgm_sources) \ object.cpp \ options.cpp \ owned.cpp \ + pgm_sender.cpp \ + pgm_socket.cpp \ pipe.cpp \ poll.cpp \ select.cpp \ @@ -122,7 +126,7 @@ libzmq_la_SOURCES = $(pgm_sources) \ zmq_listener.cpp \ zmq_listener_init.cpp -libzmq_la_LDFLAGS = -version-info @LTVER@ +libzmq_la_LDFLAGS = -version-info @LTVER@ @LIBZMQ_EXTRA_LDFLAFS@ if BUILD_PGM libzmq_la_CXXFLAGS = -I$(top_srcdir)/foreign/openpgm/@pgm_basename@/openpgm/pgm/include/ -Wall @LIBZMQ_EXTRA_CXXFLAGS@ diff --git a/src/config.hpp b/src/config.hpp index 17e67b9..43a4513 100644 --- a/src/config.hpp +++ b/src/config.hpp @@ -70,8 +70,10 @@ namespace zmq // Maximal number of non-accepted connections that can be held by // TCP listener object. - tcp_connection_backlog = 10 + tcp_connection_backlog = 10, + // Maximum transport data unit size for PGM (TPDU). + pgm_max_tpdu = 1500 }; } diff --git a/src/options.cpp b/src/options.cpp index cd07c44..804cb4f 100644 --- a/src/options.cpp +++ b/src/options.cpp @@ -24,6 +24,8 @@ zmq::options_t::options_t () : lwm (0), swap (0), mask (0), - affinity (0) + affinity (0), + rate (0), + recovery_ivl (0) { } diff --git a/src/options.hpp b/src/options.hpp index faf21b8..9f4a264 100644 --- a/src/options.hpp +++ b/src/options.hpp @@ -37,6 +37,12 @@ namespace zmq uint64_t mask; uint64_t affinity; std::string identity; + + // Maximum tranfer rate [kb/s]. + uint32_t rate; + + // Reliability time interval [s]. + uint32_t recovery_ivl; }; } diff --git a/src/socket_base.cpp b/src/socket_base.cpp index a26c280..2384d80 100644 --- a/src/socket_base.cpp +++ b/src/socket_base.cpp @@ -17,6 +17,8 @@ along with this program. If not, see <http://www.gnu.org/licenses/>. */ +#include <iostream> + #include <string> #include <algorithm> @@ -35,6 +37,7 @@ #include "uuid.hpp" #include "pipe.hpp" #include "err.hpp" +#include "platform.hpp" zmq::socket_base_t::socket_base_t (app_thread_t *parent_) : object_t (parent_), @@ -145,6 +148,22 @@ int zmq::socket_base_t::setsockopt (int option_, const void *optval_, errno = ENOTSUP; return -1; + case ZMQ_RATE: + if (optvallen_ != sizeof (uint32_t)) { + errno = EINVAL; + return -1; + } + options.rate = *((int32_t*) optval_); + return 0; + + case ZMQ_RECOVERY_IVL: + if (optvallen_ != sizeof (uint32_t)) { + errno = EINVAL; + return -1; + } + options.recovery_ivl = *((int32_t*) optval_); + return 0; + default: errno = EINVAL; return -1; @@ -170,6 +189,21 @@ int zmq::socket_base_t::connect (const char *addr_) std::string session_name ("#"); session_name += uuid_t ().to_string (); + // Parse addr_ string. + std::string addr_type; + std::string addr_args; + + std::string addr (addr_); + std::string::size_type pos = addr.find ("://"); + + if (pos == std::string::npos) { + errno = EINVAL; + return -1; + } + + addr_type = addr.substr (0, pos); + addr_args = addr.substr (pos + 3); + // Create the session. io_thread_t *io_thread = choose_io_thread (options.affinity); session_t *session = new session_t (io_thread, this, session_name.c_str (), @@ -198,20 +232,36 @@ int zmq::socket_base_t::connect (const char *addr_) send_plug (session); send_own (this, session); - // Create the connecter object. Supply it with the session name so that - // it can bind the new connection to the session once it is established. - zmq_connecter_t *connecter = new zmq_connecter_t ( - choose_io_thread (options.affinity), this, options, - session_name.c_str ()); - int rc = connecter->set_address (addr_); - if (rc != 0) { - delete connecter; - return -1; + if (addr_type == "tcp") { + + // Create the connecter object. Supply it with the session name so that + // it can bind the new connection to the session once it is established. + zmq_connecter_t *connecter = new zmq_connecter_t ( + choose_io_thread (options.affinity), this, options, + session_name.c_str ()); + int rc = connecter->set_address (addr_args.c_str ()); + if (rc != 0) { + delete connecter; + return -1; + } + send_plug (connecter); + send_own (this, connecter); + + return 0; } - send_plug (connecter); - send_own (this, connecter); - return 0; +#if defined ZMQ_HAVE_OPENPGM + if (addr_type == "pgm") { + + zmq_assert (false); + + return 0; + } +#endif + + // Unknown address type. + errno = ENOTSUP; + return -1; } int zmq::socket_base_t::send (::zmq_msg_t *msg_, int flags_) |