Merge branch 'master' of https://github.com/zeromq/libzmq

configure.in

@@ -246,6 +246,9 @@ stdlib.h string.h sys/socket.h sys/time.h time.h unistd.h limits.h)
 # Check if we have ifaddrs.h header file.
 AC_CHECK_HEADERS(ifaddrs.h, [AC_DEFINE(ZMQ_HAVE_IFADDRS, 1, [Have ifaddrs.h header.])])
 
+# Check if we have sys/uio.h header file.
+AC_CHECK_HEADERS(sys/uio.h, [AC_DEFINE(ZMQ_HAVE_UIO, 1, [Have uio.h header.])])
+
 # Force not to use eventfd
 AC_ARG_ENABLE([eventfd], [AS_HELP_STRING([--disable-eventfd], [disable eventfd [default=no]])],
     [zmq_disable_eventfd=yes], [zmq_disable_eventfd=no])

include/zmq.h

@@ -219,6 +219,9 @@ ZMQ_EXPORT int zmq_recv (void *s, void *buf, size_t len, int flags);
 ZMQ_EXPORT int zmq_sendmsg (void *s, zmq_msg_t *msg, int flags);
 ZMQ_EXPORT int zmq_recvmsg (void *s, zmq_msg_t *msg, int flags);
 
+ZMQ_EXPORT int zmq_sendv (void *s, struct iovec *iov, size_t count, int flags);
+ZMQ_EXPORT int zmq_recvmmsg (void *s, struct iovec *iov, size_t *count, int flags);
+
 /******************************************************************************/
 /*  I/O multiplexing.                                                         */
 /******************************************************************************/

src/blob.hpp

@@ -96,10 +96,11 @@ namespace std
 
       static char_type
       to_char_type(const int_type& __c)
-      { return char_type(); }
+      { ((void)__c); return char_type(); }
 
       static int_type
-      to_int_type(const char_type& __c) { return int_type(); }
+      to_int_type(const char_type& __c)
+      { ((void)__c); return int_type(); }
 
       static bool
       eq_int_type(const int_type& __c1, const int_type& __c2)

src/ctx.cpp

@@ -315,6 +315,7 @@ zmq::endpoint_t zmq::ctx_t::find_endpoint (const char *addr_)
 
 void zmq::ctx_t::log (const char *format_, va_list args_)
 {
+    ((void)args_);
     //  Create the log message.
     msg_t msg;
     int rc = msg.init_size (strlen (format_) + 1);

src/dist.cpp

@@ -128,6 +128,7 @@ int zmq::dist_t::send_to_matching (msg_t *msg_, int flags_)
 
 void zmq::dist_t::distribute (msg_t *msg_, int flags_)
 {
+    ((void)flags_);
     //  If there are no matching pipes available, simply drop the message.
     if (matching == 0) {
         int rc = msg_->close ();

src/err.cpp

@@ -65,6 +65,7 @@ const char *zmq::errno_to_string (int errno_)
 
 void zmq::zmq_abort(const char *errmsg_)
 {
+    ((void)errmsg_);
 #if defined ZMQ_HAVE_WINDOWS
 
     //  Raise STATUS_FATAL_APP_EXIT.

src/fq.cpp

@@ -70,6 +70,7 @@ int zmq::fq_t::recv (msg_t *msg_, int flags_)
 
 int zmq::fq_t::recvpipe (msg_t *msg_, int flags_, pipe_t **pipe_)
 {
+    ((void)flags_);
     //  Deallocate old content of the message.
     int rc = msg_->close ();
     errno_assert (rc == 0);

src/io_object.cpp

@@ -89,6 +89,7 @@ void zmq::io_object_t::add_timer (int timeout_, int id_)
 
 void zmq::io_object_t::cancel_timer (int id_)
 {
+    ((void)id_);
     poller->cancel_timer (this, id_);
 }
 
@@ -104,5 +105,6 @@ void zmq::io_object_t::out_event ()
 
 void zmq::io_object_t::timer_event (int id_)
 {
+    ((void)id_);
     zmq_assert (false);
 }

src/io_thread.cpp

@@ -91,6 +91,7 @@ void zmq::io_thread_t::out_event ()
 
 void zmq::io_thread_t::timer_event (int id_)
 {
+    ((void)id_);
     //  No timers here. This function is never called.
     zmq_assert (false);
 }

src/lb.cpp

@@ -73,6 +73,7 @@ void zmq::lb_t::activated (pipe_t *pipe_)
 
 int zmq::lb_t::send (msg_t *msg_, int flags_)
 {
+    ((void)flags_);
     //  Drop the message if required. If we are at the end of the message
     //  switch back to non-dropping mode.
     if (dropping) {

src/object.cpp

@@ -374,16 +374,19 @@ void zmq::object_t::process_plug ()
 
 void zmq::object_t::process_own (own_t *object_)
 {
+    ((void)object_);
     zmq_assert (false);
 }
 
 void zmq::object_t::process_attach (i_engine *engine_)
 {
+    ((void)engine_);
     zmq_assert (false);
 }
 
 void zmq::object_t::process_bind (pipe_t *pipe_)
 {
+    ((void)pipe_);
     zmq_assert (false);
 }
 
@@ -394,11 +397,13 @@ void zmq::object_t::process_activate_read ()
 
 void zmq::object_t::process_activate_write (uint64_t msgs_read_)
 {
+    ((void)msgs_read_);
     zmq_assert (false);
 }
 
 void zmq::object_t::process_hiccup (void *pipe_)
 {
+    ((void)pipe_);
     zmq_assert (false);
 }
 
@@ -414,11 +419,13 @@ void zmq::object_t::process_pipe_term_ack ()
 
 void zmq::object_t::process_term_req (own_t *object_)
 {
+    ((void)object_);
     zmq_assert (false);
 }
 
 void zmq::object_t::process_term (int linger_)
 {
+    ((void)linger_);
     zmq_assert (false);
 }
 
@@ -429,6 +436,7 @@ void zmq::object_t::process_term_ack ()
 
 void zmq::object_t::process_reap (class socket_base_t *socket_)
 {
+    ((void)socket_);
     zmq_assert (false);
 }
 

src/pair.cpp

@@ -38,6 +38,7 @@ zmq::pair_t::~pair_t ()
 
 void zmq::pair_t::xattach_pipe (pipe_t *pipe_, bool icanhasall_)
 {
+    ((void)icanhasall_);
     zmq_assert (!pipe);
     pipe = pipe_;
 }
@@ -50,12 +51,14 @@ void zmq::pair_t::xterminated (pipe_t *pipe_)
 
 void zmq::pair_t::xread_activated (pipe_t *pipe_)
 {
+    ((void)pipe_);
     //  There's just one pipe. No lists of active and inactive pipes.
     //  There's nothing to do here.
 }
 
 void zmq::pair_t::xwrite_activated (pipe_t *pipe_)
 {
+    ((void)pipe_);
     //  There's just one pipe. No lists of active and inactive pipes.
     //  There's nothing to do here.
 }
@@ -79,6 +82,7 @@ int zmq::pair_t::xsend (msg_t *msg_, int flags_)
 
 int zmq::pair_t::xrecv (msg_t *msg_, int flags_)
 {
+    ((void)flags_);
     //  Deallocate old content of the message.
     int rc = msg_->close ();
     errno_assert (rc == 0);

src/pipe.cpp

@@ -148,6 +148,7 @@ bool zmq::pipe_t::read (msg_t *msg_)
 
 bool zmq::pipe_t::check_write (msg_t *msg_)
 {
+    ((void)msg_);
     if (unlikely (!out_active || state != active))
         return false;
 

src/pub.cpp

@@ -34,6 +34,8 @@ zmq::pub_t::~pub_t ()
 
 int zmq::pub_t::xrecv (class msg_t *msg_, int flags_)
 {
+    ((void)msg_);
+    ((void)flags_);
     //  Messages cannot be received from PUB socket.
     errno = ENOTSUP;
     return -1;

src/pull.cpp

@@ -36,6 +36,7 @@ zmq::pull_t::~pull_t ()
 
 void zmq::pull_t::xattach_pipe (pipe_t *pipe_, bool icanhasall_)
 {
+    ((void)icanhasall_);
     zmq_assert (pipe_);
     fq.attach (pipe_);
 }

src/push.cpp

@@ -36,6 +36,7 @@ zmq::push_t::~push_t ()
 
 void zmq::push_t::xattach_pipe (pipe_t *pipe_, bool icanhasall_)
 {
+    ((void)icanhasall_);
     zmq_assert (pipe_);
     lb.attach (pipe_);
 }

src/reaper.cpp

@@ -80,6 +80,7 @@ void zmq::reaper_t::out_event ()
 
 void zmq::reaper_t::timer_event (int id_)
 {
+    ((void)id_);
     zmq_assert (false);
 }
 

src/session_base.cpp

@@ -248,6 +248,7 @@ void zmq::session_base_t::write_activated (pipe_t *pipe_)
 
 void zmq::session_base_t::hiccuped (pipe_t *pipe_)
 {
+    ((void)pipe_);
     //  Hiccups are always sent from session to socket, not the other
     //  way round.
     zmq_assert (false);

src/socket_base.cpp

@@ -762,6 +762,9 @@ void zmq::socket_base_t::process_destroy ()
 int zmq::socket_base_t::xsetsockopt (int option_, const void *optval_,
     size_t optvallen_)
 {
+    ((void)option_);
+    ((void)optval_);
+    ((void)optvallen_);
     errno = EINVAL;
     return -1;
 }
@@ -773,6 +776,8 @@ bool zmq::socket_base_t::xhas_out ()
 
 int zmq::socket_base_t::xsend (msg_t *msg_, int flags_)
 {
+    ((void)msg_);
+    ((void)flags_);
     errno = ENOTSUP;
     return -1;
 }
@@ -784,21 +789,26 @@ bool zmq::socket_base_t::xhas_in ()
 
 int zmq::socket_base_t::xrecv (msg_t *msg_, int flags_)
 {
+    ((void)msg_);
+    ((void)flags_);
     errno = ENOTSUP;
     return -1;
 }
 
 void zmq::socket_base_t::xread_activated (pipe_t *pipe_)
 {
+    ((void)pipe_);
     zmq_assert (false);
 }
 void zmq::socket_base_t::xwrite_activated (pipe_t *pipe_)
 {
+    ((void)pipe_);
     zmq_assert (false);
 }
 
 void zmq::socket_base_t::xhiccuped (pipe_t *pipe_)
 {
+    ((void)pipe_);
     zmq_assert (false);
 }
 
@@ -819,6 +829,7 @@ void zmq::socket_base_t::out_event ()
 
 void zmq::socket_base_t::timer_event (int id_)
 {
+    ((void)id_);
     zmq_assert (false);
 }
 

src/sub.cpp

@@ -69,6 +69,8 @@ int zmq::sub_t::xsetsockopt (int option_, const void *optval_,
 
 int zmq::sub_t::xsend (msg_t *msg_, int flags_)
 {
+    ((void)msg_);
+    ((void)flags_);
     //  Overload the XSUB's send.
     errno = ENOTSUP;
     return -1;

src/xpub.cpp

@@ -136,6 +136,7 @@ bool zmq::xpub_t::xhas_out ()
 
 int zmq::xpub_t::xrecv (msg_t *msg_, int flags_)
 {
+    ((void)flags_);
     //  If there is at least one 
     if (pending.empty ()) {
         errno = EAGAIN;

src/xrep.cpp

@@ -57,6 +57,7 @@ zmq::xrep_t::~xrep_t ()
 
 void zmq::xrep_t::xattach_pipe (pipe_t *pipe_, bool icanhasall_)
 {
+    ((void)icanhasall_);
     zmq_assert (pipe_);
 
     //  Generate a new unique peer identity.
@@ -113,6 +114,7 @@ void zmq::xrep_t::xwrite_activated (pipe_t *pipe_)
 
 int zmq::xrep_t::xsend (msg_t *msg_, int flags_)
 {
+    ((void)flags_);
     //  If this is the first part of the message it's the ID of the
     //  peer to send the message to.
     if (!more_out) {

src/xreq.cpp

@@ -48,6 +48,7 @@ zmq::xreq_t::~xreq_t ()
 
 void zmq::xreq_t::xattach_pipe (pipe_t *pipe_, bool icanhasall_)
 {
+    ((void)icanhasall_);
     zmq_assert (pipe_);
     fq.attach (pipe_);
     lb.attach (pipe_);

src/xsub.cpp

@@ -47,6 +47,7 @@ zmq::xsub_t::~xsub_t ()
 
 void zmq::xsub_t::xattach_pipe (pipe_t *pipe_, bool icanhasall_)
 {
+    ((void)icanhasall_);
     zmq_assert (pipe_);
     fq.attach (pipe_);
     dist.attach (pipe_);

src/zmq.cpp

@@ -50,6 +50,19 @@
 #include <unistd.h>
 #endif
 
+
+// XSI vector I/O
+#if ZMQ_HAVE_UIO
+#include <sys/uio.h>
+#else
+struct iovec 
+{
+    void *iov_base;
+    size_t iov_len;
+};
+#endif
+
+
 #include <string.h>
 #include <errno.h>
 #include <stdlib.h>
@@ -294,6 +307,10 @@ int zmq_sendmsg (void *s_, zmq_msg_t *msg_, int flags_)
 
 int zmq_send (void *s_, const void *buf_, size_t len_, int flags_)
 {
+    if (!s_ || !((zmq::socket_base_t*) s_)->check_tag ()) {
+        errno = ENOTSOCK;
+        return -1;
+    }
     zmq_msg_t msg;
     int rc = zmq_msg_init_size (&msg, len_);
     if (rc != 0)
@@ -317,6 +334,47 @@ int zmq_send (void *s_, const void *buf_, size_t len_, int flags_)
     return rc;
 }
 
+// Send multiple messages.
+//
+// If flag bit ZMQ_SNDMORE is set the vector is treated as
+// a single multi-part message, i.e. the last message has
+// ZMQ_SNDMORE bit switched off.
+//
+int zmq_sendv (void *s_, iovec *a_, size_t count_, int flags_)
+{
+    if (!s_ || !((zmq::socket_base_t*) s_)->check_tag ()) {
+        errno = ENOTSOCK;
+        return -1;
+    }
+    int rc = 0;
+    zmq_msg_t msg;
+    zmq::socket_base_t *s = (zmq::socket_base_t *) s_;
+    if(s->thread_safe()) s->lock();
+    for(size_t i = 0; i < count_; ++i)
+    {
+        rc = zmq_msg_init_size (&msg, a_[i].iov_len);
+        if (rc != 0)
+        {
+            rc = -1;
+            break;
+        }
+        memcpy (zmq_msg_data (&msg), a_[i].iov_base, a_[i].iov_len);
+
+        if (i == count_ - 1) flags_ = flags_ & ~ZMQ_SNDMORE;
+        rc = inner_sendmsg (s, &msg, flags_);
+        if (unlikely (rc < 0)) {
+           int err = errno;
+           int rc2 = zmq_msg_close (&msg);
+           errno_assert (rc2 == 0);
+           errno = err;
+           rc = -1;
+           break;
+        }
+    }
+    if(s->thread_safe()) s->unlock();
+    return rc; 
+}
+
 // Receiving functions.
 
 static int inner_recvmsg (zmq::socket_base_t *s_, zmq_msg_t *msg_, int flags_)
@@ -374,6 +432,70 @@ int zmq_recv (void *s_, void *buf_, size_t len_, int flags_)
     return nbytes;    
 }
 
+// Receive a multi-part message
+// 
+// Receives up to *count_ parts of a multi-part message.
+// Sets *count_ to the actual number of parts read.
+// ZMQ_RCVMORE is set to indicate if a complete multi-part message was read.
+// Returns number of message parts read, or -1 on error.
+//
+// Note: even if -1 is returned, some parts of the message
+// may have been read. Therefore the client must consult
+// *count_ to retrieve message parts successfully read,
+// even if -1 is returned.
+//
+// The iov_base* buffers of each iovec *a_ filled in by this 
+// function may be freed using free().
+//
+// Implementation note: We assume zmq::msg_t buffer allocated
+// by zmq::recvmsg can be freed by free().
+// We assume it is safe to steal these buffers by simply
+// not closing the zmq::msg_t.
+//
+int zmq_recvmmsg (void *s_, iovec *a_, size_t *count_, int flags_)
+{
+    if (!s_ || !((zmq::socket_base_t*) s_)->check_tag ()) {
+        errno = ENOTSOCK;
+        return -1;
+    }
+    zmq::socket_base_t *s = (zmq::socket_base_t *) s_;
+    if(s->thread_safe()) s->lock();
+
+    size_t count = (int)*count_;
+    int nread = 0;
+    bool recvmore = true;
+
+    for(size_t i = 0; recvmore && i < count; ++i)
+    {
+        // Cheat! We never close any msg
+        // because we want to steal the buffer.
+        zmq_msg_t msg;
+        int rc = zmq_msg_init (&msg);
+        errno_assert (rc == 0);
+
+        int nbytes = inner_recvmsg (s, &msg, flags_);
+        if (unlikely (nbytes < 0)) {
+            int err = errno;
+            rc = zmq_msg_close (&msg);
+            errno_assert (rc == 0);
+            errno = err;
+            nread = -1;
+            break;
+        }
+        ++*count_;
+        ++nread;
+
+        // Cheat: acquire zmq_msg buffer.
+        a_[i].iov_base = zmq_msg_data (&msg);
+        a_[i].iov_len = zmq_msg_size (&msg);
+
+        // Assume zmq_socket ZMQ_RVCMORE is properly set.
+        recvmore =((zmq::msg_t*) (void*) &msg)->flags () & zmq::msg_t::more;
+    }
+    if(s->thread_safe()) s->unlock();
+    return nread;    
+}
+
 // Message manipulators.
 
 int zmq_msg_init (zmq_msg_t *msg_)