1+ /* SPDX-License-Identifier: MPL-2.0 */
2+
3+ #define ZMQ_BUILD_DRAFT_API
4+
5+ #include " testutil.hpp"
6+ #include " testutil_unity.hpp"
7+
8+ #include < string.h>
9+
10+ #if defined ZMQ_HAVE_WINDOWS
11+ #include < winsock2.h>
12+ #else
13+ #include < arpa/inet.h>
14+ #endif
15+
16+ SETUP_TEARDOWN_TESTCONTEXT
17+
18+ // Helper to extract numeric host ID from the 5-byte ZMQ_STREAM frame [0x00][uint32]
19+ static uint32_t extract_id (zmq_msg_t *msg_)
20+ {
21+ TEST_ASSERT_EQUAL_INT (5 , zmq_msg_size (msg_));
22+ const unsigned char *id_ptr = (const unsigned char *) zmq_msg_data (msg_);
23+ uint32_t net_id;
24+ memcpy (&net_id, id_ptr + 1 , 4 );
25+ return ntohl (net_id);
26+ }
27+
28+ static void test_stream_disconnect_peer ()
29+ {
30+ char my_endpoint[MAX_SOCKET_STRING];
31+
32+ // We'll be using this socket to test the surgical disconnect API
33+ void *stream = test_context_socket (ZMQ_STREAM);
34+
35+ // Set timeouts to prevent the test from hanging indefinitely on failure
36+ int timeout = 500 ;
37+ TEST_ASSERT_SUCCESS_ERRNO (
38+ zmq_setsockopt (stream, ZMQ_SNDTIMEO, &timeout, sizeof (timeout)));
39+ TEST_ASSERT_SUCCESS_ERRNO (
40+ zmq_setsockopt (stream, ZMQ_RCVTIMEO, &timeout, sizeof (timeout)));
41+
42+ bind_loopback_ipv4 (stream, my_endpoint, sizeof (my_endpoint));
43+
44+ // Connect two distinct clients to test isolation and state reset
45+ fd_t fd_a = connect_socket (my_endpoint);
46+ fd_t fd_b = connect_socket (my_endpoint);
47+
48+ zmq_msg_t msg;
49+ TEST_ASSERT_SUCCESS_ERRNO (zmq_msg_init (&msg));
50+
51+ // Peer A Setup: Receive connection notification
52+ TEST_ASSERT_SUCCESS_ERRNO (zmq_msg_recv (&msg, stream, 0 ));
53+ uint32_t id_a_numeric = extract_id (&msg);
54+ unsigned char id_a_raw[5 ];
55+ memcpy (id_a_raw, zmq_msg_data (&msg), 5 );
56+ TEST_ASSERT_EQUAL_INT (
57+ 0 , TEST_ASSERT_SUCCESS_ERRNO (zmq_msg_recv (&msg, stream, 0 )));
58+
59+ // Peer B Setup: Receive connection notification
60+ TEST_ASSERT_SUCCESS_ERRNO (zmq_msg_recv (&msg, stream, 0 ));
61+ uint32_t id_b_numeric = extract_id (&msg);
62+ unsigned char id_b_raw[5 ];
63+ memcpy (id_b_raw, zmq_msg_data (&msg), 5 );
64+ TEST_ASSERT_EQUAL_INT (
65+ 0 , TEST_ASSERT_SUCCESS_ERRNO (zmq_msg_recv (&msg, stream, 0 )));
66+
67+ // Verify Peer IDs are unique
68+ TEST_ASSERT_NOT_EQUAL (id_a_numeric, id_b_numeric);
69+
70+ // --- CASE 1: THE DIRTY RESET ---
71+ // Start a multi-part message to Peer A.
72+ // This locks the socket state machine (_more_out = true, _current_out = Pipe A).
73+ TEST_ASSERT_EQUAL_INT (5 , zmq_send (stream, id_a_raw, 5 , ZMQ_SNDMORE));
74+
75+ // Use the new API to surgically disconnect Peer A.
76+ // This must force-reset the internal 'more' state and NULL the current pipe.
77+ TEST_ASSERT_SUCCESS_ERRNO (zmq_disconnect_peer (stream, id_a_numeric));
78+ msleep (SETTLE_TIME);
79+
80+ // Attempt to talk to Peer B immediately.
81+ // If the reset failed, this would misroute the ID frame as data for Peer A.
82+ TEST_ASSERT_EQUAL_INT (5 , zmq_send (stream, id_b_raw, 5 , ZMQ_SNDMORE));
83+ TEST_ASSERT_EQUAL_INT (5 , zmq_send (stream, " HELLO" 5 , 0 ));
84+
85+ // Verify Peer B actually received the data via raw TCP
86+ char recv_buf[5 ];
87+ int bytes = recv (fd_b, recv_buf, 5 , 0 );
88+ TEST_ASSERT_EQUAL_INT (5 , bytes);
89+ TEST_ASSERT_EQUAL_STRING_LEN (" HELLO" 5 );
90+
91+ // --- CASE 2: SURGICAL ISOLATION ---
92+ // Verify Peer A is gone from the routing table; sending to it should fail.
93+ int rc = zmq_send (stream, id_a_raw, 5 , ZMQ_SNDMORE);
94+ TEST_ASSERT_EQUAL_INT (-1 , rc);
95+ TEST_ASSERT_EQUAL_INT (EHOSTUNREACH, errno);
96+
97+ // --- CASE 3: INBOUND INTEGRITY ---
98+ // Ensure Peer B can still send data to the server (FQ remains intact).
99+ const char *ping = " PING"
100+ send (fd_b, ping, 4 , 0 );
101+ msleep (SETTLE_TIME);
102+
103+ TEST_ASSERT_SUCCESS_ERRNO (zmq_msg_recv (&msg, stream, 0 ));
104+ TEST_ASSERT_EQUAL_INT (id_b_numeric, extract_id (&msg));
105+ TEST_ASSERT_SUCCESS_ERRNO (zmq_msg_recv (&msg, stream, 0 ));
106+ TEST_ASSERT_EQUAL_STRING_LEN (ping, (char *) zmq_msg_data (&msg), 4 );
107+
108+ // --- CASE 4: ERROR HANDLING ---
109+ // Attempt to disconnect a non-existent ID
110+ rc = zmq_disconnect_peer (stream, 0x12345678 );
111+ TEST_ASSERT_EQUAL_INT (-1 , rc);
112+ TEST_ASSERT_EQUAL_INT (EHOSTUNREACH, errno);
113+
114+ // Cleanup
115+ TEST_ASSERT_SUCCESS_ERRNO (zmq_msg_close (&msg));
116+ close (fd_a);
117+ close (fd_b);
118+ test_context_socket_close (stream);
119+ }
120+
121+ int main (void )
122+ {
123+ setup_test_environment ();
124+
125+ UNITY_BEGIN ();
126+ RUN_TEST (test_stream_disconnect_peer);
127+ return UNITY_END ();
128+ }
0 commit comments