1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
|
/* SPDX-License-Identifier: MPL-2.0 */
#include "testutil.hpp"
#include "testutil_unity.hpp"
#include "testutil_security.hpp"
SETUP_TEARDOWN_TESTCONTEXT
void test_send_one_connected_one_unconnected ()
{
int val;
// TEST 1.
// First we're going to attempt to send messages to two
// pipes, one connected, the other not. We should see
// the PUSH load balancing to both pipes, and hence half
// of the messages getting queued, as connect() creates a
// pipe immediately.
void *to = test_context_socket (ZMQ_PULL);
int timeout = 5000;
TEST_ASSERT_SUCCESS_ERRNO (
zmq_setsockopt (to, ZMQ_LINGER, &timeout, sizeof (timeout)));
// Bind the one valid receiver
val = 0;
TEST_ASSERT_SUCCESS_ERRNO (
zmq_setsockopt (to, ZMQ_LINGER, &val, sizeof (val)));
TEST_ASSERT_SUCCESS_ERRNO (zmq_bind (to, "tipc://{6555,0,0}"));
// Create a socket pushing to two endpoints - only 1 message should arrive.
void *from = test_context_socket (ZMQ_PUSH);
val = 0;
TEST_ASSERT_SUCCESS_ERRNO (
zmq_setsockopt (from, ZMQ_LINGER, &val, sizeof (val)));
TEST_ASSERT_SUCCESS_ERRNO (
zmq_setsockopt (from, ZMQ_LINGER, &timeout, sizeof (timeout)));
// This pipe will not connect
TEST_ASSERT_SUCCESS_ERRNO (zmq_connect (from, "tipc://{5556,0}@0.0.0"));
// This pipe will
TEST_ASSERT_SUCCESS_ERRNO (zmq_connect (from, "tipc://{6555,0}@0.0.0"));
// We send 10 messages, 5 should just get stuck in the queue
// for the not-yet-connected pipe
const int send_count = 10;
for (int i = 0; i < send_count; ++i) {
send_string_expect_success (from, "Hello", 0);
}
// We now consume from the connected pipe
// - we should see just 5
timeout = SETTLE_TIME;
TEST_ASSERT_SUCCESS_ERRNO (
zmq_setsockopt (to, ZMQ_RCVTIMEO, &timeout, sizeof (int)));
int seen = 0;
while (true) {
char buffer[16];
int rc = zmq_recv (to, &buffer, sizeof (buffer), 0);
if (rc == -1) {
TEST_ASSERT_EQUAL_INT (EAGAIN, zmq_errno ());
break; // Break when we didn't get a message
}
seen++;
}
TEST_ASSERT_EQUAL_INT (send_count / 2, seen);
test_context_socket_close (from);
test_context_socket_close (to);
}
void test_send_one_connected_one_unconnected_with_delay ()
{
int val;
// TEST 2
// This time we will do the same thing, connect two pipes,
// one of which will succeed in connecting to a bound
// receiver, the other of which will fail. However, we will
// also set the delay attach on connect flag, which should
// cause the pipe attachment to be delayed until the connection
// succeeds.
// Bind the valid socket
void *to = test_context_socket (ZMQ_PULL);
TEST_ASSERT_SUCCESS_ERRNO (zmq_bind (to, "tipc://{5560,0,0}"));
int timeout = 5000;
TEST_ASSERT_SUCCESS_ERRNO (
zmq_setsockopt (to, ZMQ_LINGER, &timeout, sizeof (timeout)));
val = 0;
TEST_ASSERT_SUCCESS_ERRNO (
zmq_setsockopt (to, ZMQ_LINGER, &val, sizeof (val)));
// Create a socket pushing to two endpoints - all messages should arrive.
void *from = test_context_socket (ZMQ_PUSH);
val = 0;
TEST_ASSERT_SUCCESS_ERRNO (
zmq_setsockopt (from, ZMQ_LINGER, &val, sizeof (val)));
TEST_ASSERT_SUCCESS_ERRNO (
zmq_setsockopt (from, ZMQ_LINGER, &timeout, sizeof (timeout)));
// Set the key flag
val = 1;
TEST_ASSERT_SUCCESS_ERRNO (
zmq_setsockopt (from, ZMQ_DELAY_ATTACH_ON_CONNECT, &val, sizeof (val)));
// Connect to the invalid socket
TEST_ASSERT_SUCCESS_ERRNO (zmq_connect (from, "tipc://{5561,0}@0.0.0"));
// Connect to the valid socket
TEST_ASSERT_SUCCESS_ERRNO (zmq_connect (from, "tipc://{5560,0}@0.0.0"));
// Send 10 messages, all should be routed to the connected pipe
const int send_count = 10;
for (int i = 0; i < send_count; ++i) {
send_string_expect_success (from, "Hello", 0);
}
timeout = SETTLE_TIME;
TEST_ASSERT_SUCCESS_ERRNO (
zmq_setsockopt (to, ZMQ_RCVTIMEO, &timeout, sizeof (int)));
int seen = 0;
while (true) {
char buffer[16];
int rc = zmq_recv (to, &buffer, sizeof (buffer), 0);
if (rc == -1) {
TEST_ASSERT_EQUAL_INT (EAGAIN, zmq_errno ());
break; // Break when we didn't get a message
}
seen++;
}
TEST_ASSERT_EQUAL_INT (send_count, seen);
test_context_socket_close (from);
test_context_socket_close (to);
}
void test_send_disconnected_with_delay ()
{
// TEST 3
// This time we want to validate that the same blocking behaviour
// occurs with an existing connection that is broken. We will send
// messages to a connected pipe, disconnect and verify the messages
// block. Then we reconnect and verify messages flow again.
void *backend = test_context_socket (ZMQ_DEALER);
void *frontend = test_context_socket (ZMQ_DEALER);
void *monitor = test_context_socket (ZMQ_PAIR);
int rc;
int zero = 0;
TEST_ASSERT_SUCCESS_ERRNO (
zmq_setsockopt (backend, ZMQ_LINGER, &zero, sizeof (zero)));
TEST_ASSERT_SUCCESS_ERRNO (
zmq_setsockopt (frontend, ZMQ_LINGER, &zero, sizeof (zero)));
TEST_ASSERT_SUCCESS_ERRNO (zmq_socket_monitor (frontend, "inproc://monitor",
ZMQ_EVENT_DISCONNECTED));
int timeout = 5000;
TEST_ASSERT_SUCCESS_ERRNO (
zmq_setsockopt (backend, ZMQ_LINGER, &timeout, sizeof (timeout)));
TEST_ASSERT_SUCCESS_ERRNO (
zmq_setsockopt (frontend, ZMQ_LINGER, &timeout, sizeof (timeout)));
// Frontend connects to backend using DELAY_ATTACH_ON_CONNECT
int on = 1;
TEST_ASSERT_SUCCESS_ERRNO (
zmq_setsockopt (frontend, ZMQ_DELAY_ATTACH_ON_CONNECT, &on, sizeof (on)));
TEST_ASSERT_SUCCESS_ERRNO (zmq_bind (backend, "tipc://{5560,0,0}"));
TEST_ASSERT_SUCCESS_ERRNO (zmq_connect (monitor, "inproc://monitor"));
TEST_ASSERT_SUCCESS_ERRNO (zmq_connect (frontend, "tipc://{5560,0}@0.0.0"));
// Ping backend to frontend so we know when the connection is up
send_string_expect_success (backend, "Hello", 0);
recv_string_expect_success (frontend, "Hello", 0);
// Send message from frontend to backend
send_string_expect_success (frontend, "Hello", ZMQ_DONTWAIT);
test_context_socket_close (backend);
// Wait for disconnect to happen
expect_monitor_event (monitor, ZMQ_EVENT_DISCONNECTED);
// Send a message, might succeed depending on scheduling of the I/O thread
do {
rc = zmq_send (frontend, "Hello", 5, ZMQ_DONTWAIT);
TEST_ASSERT_TRUE (rc == 5 || (rc == -1 && zmq_errno () == EAGAIN));
} while (rc == 5);
// Recreate backend socket
backend = test_context_socket (ZMQ_DEALER);
TEST_ASSERT_SUCCESS_ERRNO (
zmq_setsockopt (backend, ZMQ_LINGER, &zero, sizeof (zero)));
TEST_ASSERT_SUCCESS_ERRNO (zmq_bind (backend, "tipc://{5560,0,0}"));
// Ping backend to frontend so we know when the connection is up
send_string_expect_success (backend, "Hello", 0);
recv_string_expect_success (frontend, "Hello", 0);
// After the reconnect, should succeed
send_string_expect_success (frontend, "Hello", ZMQ_DONTWAIT);
test_context_socket_close (monitor);
test_context_socket_close (backend);
test_context_socket_close (frontend);
}
int main (void)
{
if (!is_tipc_available ()) {
printf ("TIPC environment unavailable, skipping test\n");
return 77;
}
UNITY_BEGIN ();
RUN_TEST (test_send_one_connected_one_unconnected);
RUN_TEST (test_send_one_connected_one_unconnected_with_delay);
RUN_TEST (test_send_disconnected_with_delay);
return UNITY_END ();
}
|
