From 65a9670065ed7c155f9226a8d0164598eb9e1cb2 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Pontus=20Sk=C3=B6ldstr=C3=B6m?= <pontus.skoldstrom@ri.se>
Date: Mon, 19 Mar 2018 17:46:29 +0100
Subject: [PATCH] Problem: ZMQ_DISH over UDP triggers errno_assert() after
 hitting watermark

---
 src/udp_engine.cpp | 16 ++++++++++++++--
 1 file changed, 14 insertions(+), 2 deletions(-)

diff --git a/src/udp_engine.cpp b/src/udp_engine.cpp
index 5317ad86..6dedd2d8 100644
--- a/src/udp_engine.cpp
+++ b/src/udp_engine.cpp
@@ -376,11 +376,11 @@ void zmq::udp_engine_t::in_event ()
         body_size = nbytes - 1 - group_size;
         body_offset = 1 + group_size;
     }
-
+    // Push group description to session
     rc = session->push_msg (&msg);
     errno_assert (rc == 0 || (rc == -1 && errno == EAGAIN));
 
-    //  Pipe is full
+    //  Group description message doesn't fit in the pipe, drop
     if (rc != 0) {
         rc = msg.close ();
         errno_assert (rc == 0);
@@ -394,7 +394,19 @@ void zmq::udp_engine_t::in_event ()
     rc = msg.init_size (body_size);
     errno_assert (rc == 0);
     memcpy (msg.data (), in_buffer + body_offset, body_size);
+
+    // Push message body to session
     rc = session->push_msg (&msg);
+    // Message body message doesn't fit in the pipe, drop and reset session state
+    if (rc != 0) {
+        rc = msg.close ();
+        errno_assert (rc == 0);
+
+        session->reset ();
+        reset_pollin (handle);
+        return;
+    }
+
     errno_assert (rc == 0);
     rc = msg.close ();
     errno_assert (rc == 0);