Problem: multipart send drops unsent frames

tests/multipart.cpp

@@ -1,10 +1,123 @@
 #include <catch2/catch_all.hpp>
 #include <zmq_addon.hpp>
+#include "testutil.hpp"
 
 #include <utility>
 
 #ifdef ZMQ_HAS_RVALUE_REFS
 
+TEST_CASE("multipart send retains parts on EAGAIN", "[multipart]")
+{
+    const int part_count = GENERATE(1, 3);
+    const std::string first = GENERATE(std::string(), std::string("first"));
+    zmq::context_t context;
+    zmq::socket_t output(context, zmq::socket_type::push);
+    output.set(zmq::sockopt::linger, 0);
+    output.bind("inproc://multipart.send.retry");
+    zmq::multipart_t message;
+    message.addstr(first);
+    if (part_count > 1) {
+        message.addstr(std::string(256, '\0'));
+        message.addstr("last");
+    }
+    const zmq::multipart_t expected = message.clone();
+
+    SECTION("nonblocking integer flags")
+    {
+        for (int attempt = 0; attempt < 2; ++attempt) {
+            REQUIRE_FALSE(message.send(output, ZMQ_DONTWAIT));
+            REQUIRE(message == expected);
+        }
+    }
+    SECTION("timed send_flags overload")
+    {
+        output.set(zmq::sockopt::sndtimeo, 1);
+        for (int attempt = 0; attempt < 2; ++attempt) {
+            REQUIRE_FALSE(message.send(output, zmq::send_flags::none));
+            REQUIRE(message == expected);
+        }
+    }
+
+    zmq::socket_t input(context, zmq::socket_type::pull);
+    input.set(zmq::sockopt::rcvtimeo, 1000);
+    input.connect("inproc://multipart.send.retry");
+    output.set(zmq::sockopt::sndtimeo, 1000);
+    REQUIRE(message.send(output));
+    CHECK(message.empty());
+    zmq::multipart_t received;
+    REQUIRE(received.recv(input));
+    CHECK(received == expected);
+}
+
+TEST_CASE("multipart send retains parts at high water mark", "[multipart]")
+{
+    zmq::context_t context;
+    zmq::socket_t output(context, zmq::socket_type::push);
+    zmq::socket_t input(context, zmq::socket_type::pull);
+    output.set(zmq::sockopt::linger, 0);
+    output.set(zmq::sockopt::sndhwm, 1);
+    output.set(zmq::sockopt::sndtimeo, 1000);
+    input.set(zmq::sockopt::rcvhwm, 1);
+    input.set(zmq::sockopt::rcvtimeo, 1000);
+    output.bind("inproc://multipart.send.hwm");
+    input.connect("inproc://multipart.send.hwm");
+
+    int queued = 0;
+    while (queued < 16
+           && output.send(zmq::str_buffer("queued"), zmq::send_flags::dontwait))
+        ++queued;
+    REQUIRE(queued > 0);
+    REQUIRE(queued < 16);
+
+    zmq::multipart_t message;
+    message.addstr("first");
+    message.addstr("last");
+    const zmq::multipart_t expected = message.clone();
+    REQUIRE_FALSE(message.send(output, zmq::send_flags::dontwait));
+    REQUIRE(message == expected);
+
+    for (int i = 0; i < queued; ++i) {
+        zmq::message_t part;
+        REQUIRE(input.recv(part));
+        CHECK(part.to_string() == "queued");
+    }
+    REQUIRE(message.send(output));
+    CHECK(message.empty());
+    zmq::multipart_t received;
+    REQUIRE(received.recv(input));
+    CHECK(received == expected);
+}
+
+TEST_CASE("multipart send retains parts on exception", "[multipart]")
+{
+    zmq::context_t context;
+    zmq::socket_t output(context, zmq::socket_type::rep);
+    zmq::socket_t input(context, zmq::socket_type::req);
+    output.set(zmq::sockopt::linger, 0);
+    output.set(zmq::sockopt::sndtimeo, 1000);
+    output.set(zmq::sockopt::rcvtimeo, 1000);
+    input.set(zmq::sockopt::linger, 0);
+    input.set(zmq::sockopt::sndtimeo, 1000);
+    input.set(zmq::sockopt::rcvtimeo, 1000);
+    output.bind("inproc://multipart.send.exception");
+    input.connect("inproc://multipart.send.exception");
+    zmq::multipart_t message;
+    message.addstr("first");
+    message.addstr("last");
+    const zmq::multipart_t expected = message.clone();
+    CHECK_THROWS_ZMQ_ERROR(EFSM, message.send(output));
+    REQUIRE(message == expected);
+
+    REQUIRE(input.send(zmq::str_buffer("request")));
+    zmq::message_t request;
+    REQUIRE(output.recv(request));
+    REQUIRE(message.send(output));
+    CHECK(message.empty());
+    zmq::multipart_t received;
+    REQUIRE(received.recv(input));
+    CHECK(received == expected);
+}
+
 #ifdef ZMQ_CPP17
 using multipart_send_int_t = bool (zmq::multipart_t::*)(zmq::socket_ref, int);
 using multipart_send_flags_t = bool (zmq::multipart_t::*)(zmq::socket_ref,

zmq_addon.hpp

@@ -476,8 +476,8 @@ class multipart_t
         flags &= ~(ZMQ_SNDMORE);
         bool more = size() > 0;
         while (more) {
-            message_t message = pop();
-            more = size() > 0;
+            message_t &message = m_parts.front();
+            more = size() > 1;
 #ifdef ZMQ_CPP11
             if (!socket.send(message, static_cast<send_flags>(
                                         (more ? ZMQ_SNDMORE : 0) | flags)))
@@ -486,6 +486,7 @@ class multipart_t
             if (!socket.send(message, (more ? ZMQ_SNDMORE : 0) | flags))
                 return false;
 #endif
+            m_parts.pop_front();
         }
         clear();
         return true;