problem: no API to disconnect a specific peer by routing id

Background: JeroMQ added disconnectPeer(routingId) to allow dropping a single peer connection. In libzmq there was no C API to disconnect a single connection by its routing id; applications had to tear down entire endpoints or sockets. This also created divergence between JeroMQ and libzmq for the PEER/SERVER patterns.

solution: introduce a virtual xdisconnect_peer on socket_base_t and a public socket_base_t::disconnect_peer that delegates to it. Implement xdisconnect_peer in server_t to look up the outbound pipe by routing id and terminate it, which removes it from bookkeeping via xpipe_terminated. Expose a new DRAFT C API zmq_disconnect_peer(void*, uint32_t) that always calls socket_base_t::disconnect_peer (matching JeroMQ pattern of calling the socket regardless of type); unsupported types return ENOTSUP. This brings libzmq to feature parity with JeroMQ’s commit 57de9b8 (Peer support disconnect).

API: add draft function zmq_disconnect_peer(void *socket, uint32_t routing_id). For ZMQ_SERVER and ZMQ_PEER this disconnects that specific connection; subsequent sends to that routing id fail with EHOSTUNREACH until a new connection is formed. Other socket types return ENOTSUP. The method is thread-safe when using thread-safe sockets.

Tests: add tests/test_peer_disconnect.cpp which creates two ZMQ_PEER sockets, exchanges a message to learn the remote routing id, calls zmq_disconnect_peer on the bound side, and verifies sending back fails with EHOSTUNREACH. Hook the test into Makefile.am under ENABLE_DRAFTS.

Docs: add doc/zmq_disconnect_peer.adoc manpage; link it from doc/zmq_socket.adoc; include it in doc/Makefile.am MAN3 so it is built and installed. Note the draft status (requires --enable-drafts).

Implementation notes: the server_t override simply terminates the matched pipe; removal from maps happens in xpipe_terminated. The C shim intentionally does not gate by socket type and relies on the underlying implementation (returning ENOTSUP where appropriate), mirroring JeroMQ’s design. Reference JeroMQ change: 57de9b84 (kaplan-shaked: Peer support disconnect).

Author: somdoron

Makefile.am

@@ -1059,6 +1059,7 @@ test_apps += tests/test_poller \
 	tests/test_xpub_manual_last_value \
 	tests/test_router_notify \
 	tests/test_peer \
+	tests/test_peer_disconnect \
 	tests/test_reconnect_options \
 	tests/test_msg_init \
 	tests/test_hello_msg \
@@ -1113,6 +1114,10 @@ tests_test_peer_SOURCES = tests/test_peer.cpp
 tests_test_peer_LDADD = ${TESTUTIL_LIBS} src/libzmq.la
 tests_test_peer_CPPFLAGS = ${TESTUTIL_CPPFLAGS}
 
+tests_test_peer_disconnect_SOURCES = tests/test_peer_disconnect.cpp
+tests_test_peer_disconnect_LDADD = ${TESTUTIL_LIBS} src/libzmq.la
+tests_test_peer_disconnect_CPPFLAGS = ${TESTUTIL_CPPFLAGS}
+
 tests_test_reconnect_options_SOURCES = tests/test_reconnect_options.cpp
 tests_test_reconnect_options_LDADD = ${TESTUTIL_LIBS} src/libzmq.la
 tests_test_reconnect_options_CPPFLAGS = ${TESTUTIL_CPPFLAGS}

doc/Makefile.am

@@ -2,7 +2,7 @@
 # documentation
 #
 MAN3 = \
-    zmq_bind.3 zmq_unbind.3 zmq_connect.3 zmq_connect_peer.3 zmq_disconnect.3 zmq_close.3 \
+    zmq_bind.3 zmq_unbind.3 zmq_connect.3 zmq_connect_peer.3 zmq_disconnect_peer.3 zmq_disconnect.3 zmq_close.3 \
     zmq_ctx_new.3 zmq_ctx_term.3 zmq_ctx_get.3 zmq_ctx_set.3 zmq_ctx_shutdown.3 \
     zmq_msg_init.3 zmq_msg_init_data.3 zmq_msg_init_size.3 zmq_msg_init_buffer.3 \
     zmq_msg_move.3 zmq_msg_copy.3 zmq_msg_size.3 zmq_msg_data.3 zmq_msg_close.3 \

doc/zmq_disconnect_peer.adoc

@@ -0,0 +1,67 @@
+= zmq_disconnect_peer(3)
+
+
+== NAME
+zmq_disconnect_peer - disconnect a specific peer connection by routing id.
+
+
+== SYNOPSIS
+*int zmq_disconnect_peer (void '*socket', uint32_t 'routing_id');*
+
+
+== DESCRIPTION
+The _zmq_disconnect_peer()_ function disconnects a specific connection from a
+socket given the peer 'routing_id'. After a successful disconnection, attempts
+to send messages addressed with that 'routing_id' will fail with 'EHOSTUNREACH'
+until a new connection with a different 'routing_id' is established.
+
+This function is supported on socket types that manage per-peer routing ids:
+'ZMQ_SERVER' and 'ZMQ_PEER'. Calling it on other socket types will fail with
+'ENOTSUP'.
+
+
+== RETURN VALUE
+The _zmq_disconnect_peer()_ function returns `0` if successful. Otherwise it
+returns `-1` and sets 'errno' to one of the values defined below.
+
+
+== ERRORS
+*EHOSTUNREACH*::
+No connection exists for the given 'routing_id'.
+*ENOTSUP*::
+The socket type does not support disconnecting by 'routing_id'.
+*ENOTSOCK*::
+The provided 'socket' was invalid.
+*ETERM*::
+The 0MQ 'context' associated with the specified 'socket' was terminated.
+
+
+== EXAMPLE
+.Disconnect a peer by routing id
+----
+void *server = zmq_socket (context, ZMQ_SERVER);
+assert (server);
+assert (zmq_bind (server, "tcp://*:5555") == 0);
+/* ... receive a message and read its routing id ... */
+uint32_t routing_id = /* obtained from zmq_msg_routing_id(...) */;
+/* Disconnect that specific peer */
+int rc = zmq_disconnect_peer (server, routing_id);
+assert (rc == 0);
+/* Any attempt to send to that routing id now will fail with EHOSTUNREACH */
+----
+
+
+== SEE ALSO
+* xref:zmq_connect_peer.adoc[zmq_connect_peer]
+* xref:zmq_disconnect.adoc[zmq_disconnect]
+* xref:zmq_msg_routing_id.adoc[zmq_msg_routing_id]
+* xref:zmq_msg_set_routing_id.adoc[zmq_msg_set_routing_id]
+* xref:zmq_socket.adoc[zmq_socket]
+* xref:zmq.adoc[zmq]
+
+
+== AUTHORS
+This page was written by the 0MQ community. To make a change please
+read the 0MQ Contribution Policy at <https://zeromq.org/how-to-contribute/>.
+
+

doc/zmq_socket.adoc

@@ -447,6 +447,7 @@ ZMQ_PEER
 A 'ZMQ_PEER' socket talks to a set of 'ZMQ_PEER' sockets.
 
 To connect and fetch the 'routing_id' of the peer use xref:zmq_connect_peer.adoc[zmq_connect_peer]
+and to disconnect a specific peer by its 'routing_id' use xref:zmq_disconnect_peer.adoc[zmq_disconnect_peer].
 
 Each received message has a 'routing_id' that is a 32-bit unsigned integer.
 The application can fetch this with xref:zmq_msg_routing_id.adoc[zmq_msg_routing_id]

include/zmq.h

@@ -692,6 +692,7 @@ ZMQ_EXPORT int zmq_ctx_get_ext (void *context_,
 ZMQ_EXPORT int zmq_join (void *s, const char *group);
 ZMQ_EXPORT int zmq_leave (void *s, const char *group);
 ZMQ_EXPORT uint32_t zmq_connect_peer (void *s_, const char *addr_);
+ZMQ_EXPORT int zmq_disconnect_peer (void *s_, uint32_t routing_id_);
 
 /*  DRAFT Msg methods.                                                        */
 ZMQ_EXPORT int zmq_msg_set_routing_id (zmq_msg_t *msg, uint32_t routing_id);

src/server.cpp

@@ -155,3 +155,16 @@ bool zmq::server_t::xhas_out ()
     //  to be routed to.
     return true;
 }
+
+int zmq::server_t::xdisconnect_peer (uint32_t routing_id_)
+{
+    out_pipes_t::iterator it = _out_pipes.find (routing_id_);
+    if (it == _out_pipes.end ()) {
+        errno = EHOSTUNREACH;
+        return -1;
+    }
+
+    // Terminate the pipe; xpipe_terminated will erase it from maps.
+    it->second.pipe->terminate (false);
+    return 0;
+}

src/server.hpp

@@ -35,6 +35,7 @@ class server_t : public socket_base_t
     void xread_activated (zmq::pipe_t *pipe_);
     void xwrite_activated (zmq::pipe_t *pipe_);
     void xpipe_terminated (zmq::pipe_t *pipe_);
+    int xdisconnect_peer (uint32_t routing_id_);
 
   private:
     //  Fair queueing object for inbound pipes.

src/socket_base.cpp

@@ -493,6 +493,13 @@ int zmq::socket_base_t::leave (const char *group_)
     return xleave (group_);
 }
 
+int zmq::socket_base_t::disconnect_peer (uint32_t routing_id_)
+{
+    scoped_optional_lock_t sync_lock (_thread_safe ? &_sync : NULL);
+
+    return xdisconnect_peer (routing_id_);
+}
+
 void zmq::socket_base_t::add_signaler (signaler_t *s_)
 {
     zmq_assert (_thread_safe);
@@ -1637,6 +1644,12 @@ int zmq::socket_base_t::xleave (const char *group_)
     return -1;
 }
 
+int zmq::socket_base_t::xdisconnect_peer (uint32_t)
+{
+    errno = ENOTSUP;
+    return -1;
+}
+
 int zmq::socket_base_t::xrecv (msg_t *)
 {
     errno = ENOTSUP;

src/socket_base.hpp

@@ -140,6 +140,9 @@ class socket_base_t : public own_t,
 
     bool is_disconnected () const;
 
+    // Disconnect a specific peer given its routing id. Default ENOTSUP.
+    int disconnect_peer (uint32_t routing_id_);
+
   protected:
     socket_base_t (zmq::ctx_t *parent_,
                    uint32_t tid_,
@@ -182,6 +185,9 @@ class socket_base_t : public own_t,
     virtual int xjoin (const char *group_);
     virtual int xleave (const char *group_);
 
+    // Default implementation returns ENOTSUP. Specific sockets may override.
+    virtual int xdisconnect_peer (uint32_t routing_id_);
+
     //  Delay actual destruction of the socket.
     void process_destroy () ZMQ_FINAL;
 

src/zmq.cpp

@@ -333,6 +333,14 @@ uint32_t zmq_connect_peer (void *s_, const char *addr_)
     return s->connect_peer (addr_);
 }
 
+int zmq_disconnect_peer (void *s_, uint32_t routing_id_)
+{
+    zmq::socket_base_t *s = as_socket_base_t (s_);
+    if (!s)
+        return -1;
+    return s->disconnect_peer (routing_id_);
+}
+
 
 int zmq_unbind (void *s_, const char *addr_)
 {