diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/options.cpp | 30 | ||||
| -rw-r--r-- | src/options.hpp | 5 | ||||
| -rw-r--r-- | src/session.cpp | 6 | ||||
| -rw-r--r-- | src/socket_base.cpp | 21 |
4 files changed, 44 insertions, 18 deletions
diff --git a/src/options.cpp b/src/options.cpp index 39f8984..556ffd8 100644 --- a/src/options.cpp +++ b/src/options.cpp @@ -26,7 +26,8 @@ #include "err.hpp" zmq::options_t::options_t () : - hwm (0), + sndhwm (0), + rcvhwm (0), affinity (0), rate (100), recovery_ivl (10000), @@ -49,12 +50,20 @@ int zmq::options_t::setsockopt (int option_, const void *optval_, { switch (option_) { - case ZMQ_HWM: + case ZMQ_SNDHWM: if (optvallen_ != sizeof (int) || *((int*) optval_) < 0) { errno = EINVAL; return -1; } - hwm = *((int*) optval_); + sndhwm = *((int*) optval_); + return 0; + + case ZMQ_RCVHWM: + if (optvallen_ != sizeof (int) || *((int*) optval_) < 0) { + errno = EINVAL; + return -1; + } + rcvhwm = *((int*) optval_); return 0; case ZMQ_AFFINITY: @@ -168,13 +177,22 @@ int zmq::options_t::getsockopt (int option_, void *optval_, size_t *optvallen_) { switch (option_) { - case ZMQ_HWM: + case ZMQ_SNDHWM: if (*optvallen_ < sizeof (int)) { errno = EINVAL; return -1; } - *((int*) optval_) = hwm; - *optvallen_ = sizeof (uint64_t); + *((int*) optval_) = sndhwm; + *optvallen_ = sizeof (int); + return 0; + + case ZMQ_RCVHWM: + if (*optvallen_ < sizeof (int)) { + errno = EINVAL; + return -1; + } + *((int*) optval_) = rcvhwm; + *optvallen_ = sizeof (int); return 0; case ZMQ_AFFINITY: diff --git a/src/options.hpp b/src/options.hpp index 971e643..9ba06e3 100644 --- a/src/options.hpp +++ b/src/options.hpp @@ -35,8 +35,9 @@ namespace zmq int setsockopt (int option_, const void *optval_, size_t optvallen_); int getsockopt (int option_, void *optval_, size_t *optvallen_); - // High-water mark for messages in pipe. - int hwm; + // High-water marks for message pipes. + int sndhwm; + int rcvhwm; uint64_t affinity; blob_t identity; diff --git a/src/session.cpp b/src/session.cpp index 176f0ef..5f970cc 100644 --- a/src/session.cpp +++ b/src/session.cpp @@ -247,11 +247,13 @@ void zmq::session_t::process_attach (i_engine *engine_, // Create the pipes, as required. if (options.requires_in) { - create_pipe (socket, this, options.hwm, &socket_reader, &out_pipe); + create_pipe (socket, this, options.rcvhwm, &socket_reader, + &out_pipe); out_pipe->set_event_sink (this); } if (options.requires_out) { - create_pipe (this, socket, options.hwm, &in_pipe, &socket_writer); + create_pipe (this, socket, options.sndhwm, &in_pipe, + &socket_writer); in_pipe->set_event_sink (this); } diff --git a/src/socket_base.cpp b/src/socket_base.cpp index 374e342..bf873b6 100644 --- a/src/socket_base.cpp +++ b/src/socket_base.cpp @@ -377,20 +377,25 @@ int zmq::socket_base_t::connect (const char *addr_) // The total HWM for an inproc connection should be the sum of // the binder's HWM and the connector's HWM. - int hwm; - if (options.hwm == 0 || peer.options.hwm == 0) - hwm = 0; + int sndhwm; + int rcvhwm; + if (options.sndhwm == 0 || peer.options.rcvhwm == 0) + sndhwm = 0; else - hwm = options.hwm + peer.options.hwm; + sndhwm = options.sndhwm + peer.options.rcvhwm; + if (options.rcvhwm == 0 || peer.options.sndhwm == 0) + rcvhwm = 0; + else + rcvhwm = options.rcvhwm + peer.options.sndhwm; // Create inbound pipe, if required. if (options.requires_in) - create_pipe (this, peer.socket, hwm, &inpipe_reader, + create_pipe (this, peer.socket, rcvhwm, &inpipe_reader, &inpipe_writer); // Create outbound pipe, if required. if (options.requires_out) - create_pipe (peer.socket, this, hwm, &outpipe_reader, + create_pipe (peer.socket, this, sndhwm, &outpipe_reader, &outpipe_writer); // Attach the pipes to this socket object. @@ -429,12 +434,12 @@ int zmq::socket_base_t::connect (const char *addr_) // Create inbound pipe, if required. if (options.requires_in) - create_pipe (this, session, options.hwm, + create_pipe (this, session, options.rcvhwm, &inpipe_reader, &inpipe_writer); // Create outbound pipe, if required. if (options.requires_out) - create_pipe (session, this, options.hwm, + create_pipe (session, this, options.sndhwm, &outpipe_reader, &outpipe_writer); // Attach the pipes to the socket object. |