Commit da49e5a4 authored by Martin Sustrik's avatar Martin Sustrik

devices exit in case of context termination

parent ca057c7d
...@@ -30,8 +30,19 @@ int zmq::forwarder (socket_base_t *insocket_, socket_base_t *outsocket_) ...@@ -30,8 +30,19 @@ int zmq::forwarder (socket_base_t *insocket_, socket_base_t *outsocket_)
errno_assert (rc == 0); errno_assert (rc == 0);
while (true) { while (true) {
insocket_->recv (&msg, 0); rc = insocket_->recv (&msg, 0);
outsocket_->send (&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; return 0;
......
...@@ -49,7 +49,11 @@ int zmq::queue (class socket_base_t *insocket_, ...@@ -49,7 +49,11 @@ 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);
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 // The algorithm below asumes ratio of request and replies processed
// under full load to be 1:1. Although processing requests replies // under full load to be 1:1. Although processing requests replies
...@@ -61,14 +65,26 @@ int zmq::queue (class socket_base_t *insocket_, ...@@ -61,14 +65,26 @@ int zmq::queue (class socket_base_t *insocket_,
while (true) { while (true) {
rc = insocket_->recv (&msg, 0); rc = insocket_->recv (&msg, 0);
errno_assert (rc == 0); if (rc < 0) {
if (errno == ETERM)
return -1;
errno_assert (false);
}
moresz = sizeof (more); moresz = sizeof (more);
rc = insocket_->getsockopt (ZMQ_RCVMORE, &more, &moresz); 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); 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) if (!more)
break; break;
...@@ -80,14 +96,26 @@ int zmq::queue (class socket_base_t *insocket_, ...@@ -80,14 +96,26 @@ int zmq::queue (class socket_base_t *insocket_,
while (true) { while (true) {
rc = outsocket_->recv (&msg, 0); rc = outsocket_->recv (&msg, 0);
errno_assert (rc == 0); if (rc < 0) {
if (errno == ETERM)
return -1;
errno_assert (false);
}
moresz = sizeof (more); moresz = sizeof (more);
rc = outsocket_->getsockopt (ZMQ_RCVMORE, &more, &moresz); 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); 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) if (!more)
break; break;
......
...@@ -30,8 +30,19 @@ int zmq::streamer (socket_base_t *insocket_, socket_base_t *outsocket_) ...@@ -30,8 +30,19 @@ int zmq::streamer (socket_base_t *insocket_, socket_base_t *outsocket_)
errno_assert (rc == 0); errno_assert (rc == 0);
while (true) { while (true) {
insocket_->recv (&msg, 0); rc = insocket_->recv (&msg, 0);
outsocket_->send (&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; return 0;
......
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