1 files changed, 109 insertions, 2 deletions

diff --git a/src/sub.cpp b/src/sub.cpp
index 2d6ade6..11c4532 100644
--- a/src/sub.cpp
+++ b/src/sub.cpp
@@ -22,19 +22,39 @@
 #include "msg.hpp"
 
 zmq::sub_t::sub_t (class ctx_t *parent_, uint32_t tid_) :
-    xsub_t (parent_, tid_)
+    xsub_t (parent_, tid_),
+    has_message (false),
+    more (false)
 {
     options.type = ZMQ_SUB;
+    int rc = message.init ();
+    errno_assert (rc == 0);
 }
 
 zmq::sub_t::~sub_t ()
 {
+    int rc = message.close ();
+    errno_assert (rc == 0);
 }
 
 int zmq::sub_t::xsetsockopt (int option_, const void *optval_,
     size_t optvallen_)
 {
-    if (option_ != ZMQ_SUBSCRIBE && option_ != ZMQ_UNSUBSCRIBE) {
+    //  Process a subscription.
+    if (option_ == ZMQ_SUBSCRIBE)
+        subscriptions.add ((unsigned char*) optval_, optvallen_);
+
+    //  Process an unsubscription. Return error if there is no corresponding
+    //  subscription.
+    else if (option_ == ZMQ_UNSUBSCRIBE) {
+        if (!subscriptions.rm ((unsigned char*) optval_, optvallen_)) {
+            errno = EINVAL;
+            return -1;
+        }
+    }
+
+    //  Unknow option.
+    else {
         errno = EINVAL;
         return -1;
     }
@@ -74,3 +94,90 @@ bool zmq::sub_t::xhas_out ()
     //  Overload the XSUB's send.
     return false;
 }
+
+int zmq::sub_t::xrecv (msg_t *msg_, int flags_)
+{
+    //  If there's already a message prepared by a previous call to zmq_poll,
+    //  return it straight ahead.
+    if (has_message) {
+        int rc = msg_->move (message);
+        errno_assert (rc == 0);
+        has_message = false;
+        more = msg_->flags () & msg_t::more;
+        return 0;
+    }
+
+    //  TODO: This can result in infinite loop in the case of continuous
+    //  stream of non-matching messages which breaks the non-blocking recv
+    //  semantics.
+    while (true) {
+
+        //  Get a message using fair queueing algorithm.
+        int rc = xsub_t::xrecv (msg_, flags_);
+
+        //  If there's no message available, return immediately.
+        //  The same when error occurs.
+        if (rc != 0)
+            return -1;
+
+        //  Check whether the message matches at least one subscription.
+        //  Non-initial parts of the message are passed 
+        if (more || match (msg_)) {
+            more = msg_->flags () & msg_t::more;
+            return 0;
+        }
+
+        //  Message doesn't match. Pop any remaining parts of the message
+        //  from the pipe.
+        while (msg_->flags () & msg_t::more) {
+            rc = xsub_t::xrecv (msg_, ZMQ_DONTWAIT);
+            zmq_assert (rc == 0);
+        }
+    }
+}
+
+bool zmq::sub_t::xhas_in ()
+{
+    //  There are subsequent parts of the partly-read message available.
+    if (more)
+        return true;
+
+    //  If there's already a message prepared by a previous call to zmq_poll,
+    //  return straight ahead.
+    if (has_message)
+        return true;
+
+    //  TODO: This can result in infinite loop in the case of continuous
+    //  stream of non-matching messages.
+    while (true) {
+
+        //  Get a message using fair queueing algorithm.
+        int rc = xsub_t::xrecv (&message, ZMQ_DONTWAIT);
+
+        //  If there's no message available, return immediately.
+        //  The same when error occurs.
+        if (rc != 0) {
+            zmq_assert (errno == EAGAIN);
+            return false;
+        }
+
+        //  Check whether the message matches at least one subscription.
+        if (match (&message)) {
+            has_message = true;
+            return true;
+        }
+
+        //  Message doesn't match. Pop any remaining parts of the message
+        //  from the pipe.
+        while (message.flags () & msg_t::more) {
+            rc = xsub_t::xrecv (&message, ZMQ_DONTWAIT);
+            zmq_assert (rc == 0);
+        }
+    }
+}
+
+bool zmq::sub_t::match (msg_t *msg_)
+{
+    return subscriptions.check ((unsigned char*) msg_->data (), msg_->size ());
+}
+