Adding ZMQ/Pub support for txpool_add and chain_main events

2 files changed, 22 insertions, 14 deletions

diff --git a/src/net/zmq.cpp b/src/net/zmq.cpp
index 1a0edb4b9..15560ca7e 100644
--- a/src/net/zmq.cpp
+++ b/src/net/zmq.cpp
@@ -158,20 +158,6 @@ namespace zmq
                 return unsigned(max_out) < added ? max_out : int(added);
             }
         };
-
-        template<typename F, typename... T>
-        expect<void> retry_op(F op, T&&... args) noexcept(noexcept(op(args...)))
-        {
-            for (;;)
-            {
-                if (0 <= op(args...))
-                    return success();
-
-                const int error = zmq_errno();
-                if (error != EINTR)
-                    return make_error_code(error);
-            }
-        }
     } // anonymous
 
     expect<std::string> receive(void* const socket, const int flags)
diff --git a/src/net/zmq.h b/src/net/zmq.h
index 8c587ed7c..fa4ef2fc9 100644
--- a/src/net/zmq.h
+++ b/src/net/zmq.h
@@ -26,6 +26,8 @@
 // STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF
 // THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 
+#pragma once
+
 #include <memory>
 #include <string>
 #include <system_error>
@@ -105,6 +107,26 @@ namespace zmq
     //! Unique ZMQ socket handle, calls `zmq_close` on destruction.
     using socket = std::unique_ptr<void, close>;
 
+  /*! Retry a ZMQ function on `EINTR` errors. `F` must return an int with
+      values less than 0 on error.
+
+      \param op The ZMQ function to execute + retry
+      \param args Forwarded to `op`. Must be resuable in case of retry.
+      \return All errors except for `EINTR`. */
+    template<typename F, typename... T>
+    expect<void> retry_op(F op, T&&... args) noexcept(noexcept(op(args...)))
+    {
+      for (;;)
+      {
+        if (0 <= op(args...))
+          return success();
+
+        const int error = zmq_errno();
+        if (error != EINTR)
+          return make_error_code(error);
+      }
+    }
+
     /*! Read all parts of the next message on `socket`. Blocks until the entire
         next message (all parts) are read, or until `zmq_term` is called on the
         `zmq_context` associated with `socket`. If the context is terminated,