summaryrefslogtreecommitdiff
path: root/src/streamer.cpp
diff options
context:
space:
mode:
Diffstat (limited to 'src/streamer.cpp')
-rw-r--r--src/streamer.cpp26
1 files changed, 24 insertions, 2 deletions
diff --git a/src/streamer.cpp b/src/streamer.cpp
index 796771b..7c03365 100644
--- a/src/streamer.cpp
+++ b/src/streamer.cpp
@@ -21,6 +21,7 @@
#include "streamer.hpp"
#include "socket_base.hpp"
+#include "likely.hpp"
#include "err.hpp"
int zmq::streamer (socket_base_t *insocket_, socket_base_t *outsocket_)
@@ -29,9 +30,30 @@ int zmq::streamer (socket_base_t *insocket_, socket_base_t *outsocket_)
int rc = zmq_msg_init (&msg);
errno_assert (rc == 0);
+ int64_t more;
+ size_t more_sz = sizeof (more);
+
while (true) {
- insocket_->recv (&msg, 0);
- outsocket_->send (&msg, 0);
+ rc = insocket_->recv (&msg, 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)
+ return -1;
+ errno_assert (false);
+ }
+
+ rc = outsocket_->send (&msg, more ? ZMQ_SNDMORE : 0);
+ if (unlikely (rc < 0)) {
+ if (errno == ETERM)
+ return -1;
+ errno_assert (false);
+ }
}
return 0;