diff options
| -rw-r--r-- | src/forwarder.cpp | 15 | ||||
| -rw-r--r-- | src/queue.cpp | 42 | ||||
| -rw-r--r-- | src/streamer.cpp | 15 |
3 files changed, 61 insertions, 11 deletions
diff --git a/src/forwarder.cpp b/src/forwarder.cpp index 5aab8f2..503868b 100644 --- a/src/forwarder.cpp +++ b/src/forwarder.cpp @@ -30,8 +30,19 @@ int zmq::forwarder (socket_base_t *insocket_, socket_base_t *outsocket_) errno_assert (rc == 0); while (true) { - insocket_->recv (&msg, 0); - outsocket_->send (&msg, 0); + rc = insocket_->recv (&msg, 0); + if (rc < 0) { + if (errno == ETERM) + return -1; + errno_assert (false); + } + + rc = outsocket_->send (&msg, 0); + if (rc < 0) { + if (errno == ETERM) + return -1; + errno_assert (false); + } } return 0; diff --git a/src/queue.cpp b/src/queue.cpp index 470ea67..311a8c1 100644 --- a/src/queue.cpp +++ b/src/queue.cpp @@ -49,7 +49,11 @@ int zmq::queue (class socket_base_t *insocket_, // Wait while there are either requests or replies to process. rc = zmq_poll (&items [0], 2, -1); - errno_assert (rc > 0); + if (rc < 0) { + if (errno == ETERM) + return -1; + errno_assert (false); + } // The algorithm below asumes ratio of request and replies processed // under full load to be 1:1. Although processing requests replies @@ -61,14 +65,26 @@ int zmq::queue (class socket_base_t *insocket_, while (true) { rc = insocket_->recv (&msg, 0); - errno_assert (rc == 0); + if (rc < 0) { + if (errno == ETERM) + return -1; + errno_assert (false); + } moresz = sizeof (more); rc = insocket_->getsockopt (ZMQ_RCVMORE, &more, &moresz); - errno_assert (rc == 0); + if (rc < 0) { + if (errno == ETERM) + return -1; + errno_assert (false); + } rc = outsocket_->send (&msg, more ? ZMQ_SNDMORE : 0); - errno_assert (rc == 0); + if (rc < 0) { + if (errno == ETERM) + return -1; + errno_assert (false); + } if (!more) break; @@ -80,14 +96,26 @@ int zmq::queue (class socket_base_t *insocket_, while (true) { rc = outsocket_->recv (&msg, 0); - errno_assert (rc == 0); + if (rc < 0) { + if (errno == ETERM) + return -1; + errno_assert (false); + } moresz = sizeof (more); rc = outsocket_->getsockopt (ZMQ_RCVMORE, &more, &moresz); - errno_assert (rc == 0); + if (rc < 0) { + if (errno == ETERM) + return -1; + errno_assert (false); + } rc = insocket_->send (&msg, more ? ZMQ_SNDMORE : 0); - errno_assert (rc == 0); + if (rc < 0) { + if (errno == ETERM) + return -1; + errno_assert (false); + } if (!more) break; diff --git a/src/streamer.cpp b/src/streamer.cpp index 796771b..9799007 100644 --- a/src/streamer.cpp +++ b/src/streamer.cpp @@ -30,8 +30,19 @@ int zmq::streamer (socket_base_t *insocket_, socket_base_t *outsocket_) errno_assert (rc == 0); while (true) { - insocket_->recv (&msg, 0); - outsocket_->send (&msg, 0); + rc = insocket_->recv (&msg, 0); + if (rc < 0) { + if (errno == ETERM) + return -1; + errno_assert (false); + } + + rc = outsocket_->send (&msg, 0); + if (rc < 0) { + if (errno == ETERM) + return -1; + errno_assert (false); + } } return 0; |