Skip to content

L
libzmq
Commit 3cb84b5c authored by Jon Dyte's avatar Jon Dyte Committed by Martin Sustrik
Browse files
Options

forwarder and streamer devices handle multi-part messages correctly

parent 43f2c6ff
Show whitespace changes
Inline Side-by-side
Showing with 36 additions and 13 deletions
src/forwarder.cpp
View file @ 3cb84b5c
...@@ -21,6 +21,7 @@ ...@@ -21,6 +21,7 @@
#include "forwarder.hpp" #include "forwarder.hpp"
#include "socket_base.hpp" #include "socket_base.hpp"
#include "likely.hpp"
#include "err.hpp" #include "err.hpp"
int zmq::forwarder (socket_base_t *insocket_, socket_base_t *outsocket_) int zmq::forwarder (socket_base_t *insocket_, socket_base_t *outsocket_)
...@@ -29,16 +30,26 @@ int zmq::forwarder (socket_base_t *insocket_, socket_base_t *outsocket_) ...@@ -29,16 +30,26 @@ int zmq::forwarder (socket_base_t *insocket_, socket_base_t *outsocket_)
int rc = zmq_msg_init (&msg); int rc = zmq_msg_init (&msg);
errno_assert (rc == 0); errno_assert (rc == 0);
int64_t more;
size_t more_sz = sizeof (more);
while (true) { while (true) {
rc = insocket_->recv (&msg, 0); rc = insocket_->recv (&msg, 0);
if (rc < 0) { if (unlikely (rc < 0)) {
if (errno == ETERM)
return -1;
errno_assert (false);
}
rc = insocket_->getsockopt (ZMQ_RCVMORE, &more, &more_sz);
if (unlikely (rc < 0)) {
if (errno == ETERM) if (errno == ETERM)
return -1; return -1;
errno_assert (false); errno_assert (false);
} }
rc = outsocket_->send (&msg, 0); rc = outsocket_->send (&msg, more ? ZMQ_SNDMORE : 0);
if (rc < 0) { if (unlikely (rc < 0)) {
if (errno == ETERM) if (errno == ETERM)
return -1; return -1;
errno_assert (false); errno_assert (false);
......
src/queue.cpp
View file @ 3cb84b5c
...@@ -23,6 +23,7 @@ ...@@ -23,6 +23,7 @@
#include "queue.hpp" #include "queue.hpp"
#include "socket_base.hpp" #include "socket_base.hpp"
#include "likely.hpp"
#include "err.hpp" #include "err.hpp"
int zmq::queue (class socket_base_t *insocket_, int zmq::queue (class socket_base_t *insocket_,
...@@ -49,7 +50,7 @@ int zmq::queue (class socket_base_t *insocket_, ...@@ -49,7 +50,7 @@ int zmq::queue (class socket_base_t *insocket_,
// Wait while there are either requests or replies to process. // Wait while there are either requests or replies to process.
rc = zmq_poll (&items [0], 2, -1); rc = zmq_poll (&items [0], 2, -1);
if (rc < 0) { if (unlikely (rc < 0)) {
if (errno == ETERM) if (errno == ETERM)
return -1; return -1;
errno_assert (false); errno_assert (false);
...@@ -65,7 +66,7 @@ int zmq::queue (class socket_base_t *insocket_, ...@@ -65,7 +66,7 @@ int zmq::queue (class socket_base_t *insocket_,
while (true) { while (true) {
rc = insocket_->recv (&msg, 0); rc = insocket_->recv (&msg, 0);
if (rc < 0) { if (unlikely (rc < 0)) {
if (errno == ETERM) if (errno == ETERM)
return -1; return -1;
errno_assert (false); errno_assert (false);
...@@ -73,14 +74,14 @@ int zmq::queue (class socket_base_t *insocket_, ...@@ -73,14 +74,14 @@ int zmq::queue (class socket_base_t *insocket_,
moresz = sizeof (more); moresz = sizeof (more);
rc = insocket_->getsockopt (ZMQ_RCVMORE, &more, &moresz); rc = insocket_->getsockopt (ZMQ_RCVMORE, &more, &moresz);
if (rc < 0) { if (unlikely (rc < 0)) {
if (errno == ETERM) if (errno == ETERM)
return -1; return -1;
errno_assert (false); errno_assert (false);
} }
rc = outsocket_->send (&msg, more ? ZMQ_SNDMORE : 0); rc = outsocket_->send (&msg, more ? ZMQ_SNDMORE : 0);
if (rc < 0) { if (unlikely (rc < 0)) {
if (errno == ETERM) if (errno == ETERM)
return -1; return -1;
errno_assert (false); errno_assert (false);
...@@ -96,7 +97,7 @@ int zmq::queue (class socket_base_t *insocket_, ...@@ -96,7 +97,7 @@ int zmq::queue (class socket_base_t *insocket_,
while (true) { while (true) {
rc = outsocket_->recv (&msg, 0); rc = outsocket_->recv (&msg, 0);
if (rc < 0) { if (unlikely (rc < 0)) {
if (errno == ETERM) if (errno == ETERM)
return -1; return -1;
errno_assert (false); errno_assert (false);
...@@ -104,14 +105,14 @@ int zmq::queue (class socket_base_t *insocket_, ...@@ -104,14 +105,14 @@ int zmq::queue (class socket_base_t *insocket_,
moresz = sizeof (more); moresz = sizeof (more);
rc = outsocket_->getsockopt (ZMQ_RCVMORE, &more, &moresz); rc = outsocket_->getsockopt (ZMQ_RCVMORE, &more, &moresz);
if (rc < 0) { if (unlikely (rc < 0)) {
if (errno == ETERM) if (errno == ETERM)
return -1; return -1;
errno_assert (false); errno_assert (false);
} }
rc = insocket_->send (&msg, more ? ZMQ_SNDMORE : 0); rc = insocket_->send (&msg, more ? ZMQ_SNDMORE : 0);
if (rc < 0) { if (unlikely (rc < 0)) {
if (errno == ETERM) if (errno == ETERM)
return -1; return -1;
errno_assert (false); errno_assert (false);
......
src/streamer.cpp
View file @ 3cb84b5c
...@@ -21,6 +21,7 @@ ...@@ -21,6 +21,7 @@
#include "streamer.hpp" #include "streamer.hpp"
#include "socket_base.hpp" #include "socket_base.hpp"
#include "likely.hpp"
#include "err.hpp" #include "err.hpp"
int zmq::streamer (socket_base_t *insocket_, socket_base_t *outsocket_) int zmq::streamer (socket_base_t *insocket_, socket_base_t *outsocket_)
...@@ -29,16 +30,26 @@ int zmq::streamer (socket_base_t *insocket_, socket_base_t *outsocket_) ...@@ -29,16 +30,26 @@ int zmq::streamer (socket_base_t *insocket_, socket_base_t *outsocket_)
int rc = zmq_msg_init (&msg); int rc = zmq_msg_init (&msg);
errno_assert (rc == 0); errno_assert (rc == 0);
int64_t more;
size_t more_sz = sizeof (more);
while (true) { while (true) {
rc = insocket_->recv (&msg, 0); rc = insocket_->recv (&msg, 0);
if (rc < 0) { if (unlikely (rc < 0)) {
if (errno == ETERM)
return -1;
errno_assert (false);
}
rc = insocket_->getsockopt (ZMQ_RCVMORE, &more, &more_sz);
if (unlikely (rc < 0)) {
if (errno == ETERM) if (errno == ETERM)
return -1; return -1;
errno_assert (false); errno_assert (false);
} }
rc = outsocket_->send (&msg, 0); rc = outsocket_->send (&msg, more ? ZMQ_SNDMORE : 0);
if (rc < 0) { if (unlikely (rc < 0)) {
if (errno == ETERM) if (errno == ETERM)
return -1; return -1;
errno_assert (false); errno_assert (false);
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment