0
0
mirror of https://github.com/zeromq/libzmq.git synced 2024-12-25 22:30:51 +08:00
libzmq/tests/test_stream.cpp
Luca Boccassi da31917f4f Relicense from LGPL3 + exceptions to Mozilla Public License version 2.0
Relicense permission collected from all relevant authors as tallied at:
https://github.com/rlenferink/libzmq-relicense/blob/master/checklist.md
The relicense grants are collected under RELICENSE/ and will be moved
to the above repository in a later commit.

Fixes https://github.com/zeromq/libzmq/issues/2376
2023-06-05 20:31:47 +01:00

277 lines
10 KiB
C++

/* SPDX-License-Identifier: MPL-2.0 */
#include "testutil.hpp"
#include "testutil_unity.hpp"
#include <string.h>
SETUP_TEARDOWN_TESTCONTEXT
// ZMTP protocol greeting structure
typedef uint8_t byte;
typedef struct
{
byte signature[10]; // 0xFF 8*0x00 0x7F
byte version[2]; // 0x03 0x01 for ZMTP/3.1
byte mechanism[20]; // "NULL"
byte as_server;
byte filler[31];
} zmtp_greeting_t;
#define ZMTP_DEALER 5 // Socket type constants
// This is a greeting matching what 0MQ will send us; note the
// 8-byte size is set to 1 for backwards compatibility
static zmtp_greeting_t greeting = {
{0xFF, 0, 0, 0, 0, 0, 0, 0, 1, 0x7F}, {3, 1}, {'N', 'U', 'L', 'L'}, 0, {0}};
static void test_stream_to_dealer ()
{
int rc;
char my_endpoint[MAX_SOCKET_STRING];
// We'll be using this socket in raw mode
void *stream = test_context_socket (ZMQ_STREAM);
int zero = 0;
TEST_ASSERT_SUCCESS_ERRNO (
zmq_setsockopt (stream, ZMQ_LINGER, &zero, sizeof (zero)));
int enabled = 1;
TEST_ASSERT_SUCCESS_ERRNO (
zmq_setsockopt (stream, ZMQ_STREAM_NOTIFY, &enabled, sizeof (enabled)));
bind_loopback_ipv4 (stream, my_endpoint, sizeof my_endpoint);
// We'll be using this socket as the other peer
void *dealer = test_context_socket (ZMQ_DEALER);
TEST_ASSERT_SUCCESS_ERRNO (
zmq_setsockopt (dealer, ZMQ_LINGER, &zero, sizeof (zero)));
TEST_ASSERT_SUCCESS_ERRNO (zmq_connect (dealer, my_endpoint));
// Send a message on the dealer socket
send_string_expect_success (dealer, "Hello", 0);
// Connecting sends a zero message
// First frame is routing id
zmq_msg_t routing_id;
TEST_ASSERT_SUCCESS_ERRNO (zmq_msg_init (&routing_id));
TEST_ASSERT_SUCCESS_ERRNO (zmq_msg_recv (&routing_id, stream, 0));
TEST_ASSERT_TRUE (zmq_msg_more (&routing_id));
// Verify the existence of Peer-Address metadata
char const *peer_address = zmq_msg_gets (&routing_id, "Peer-Address");
TEST_ASSERT_NOT_NULL (peer_address);
TEST_ASSERT_EQUAL_STRING ("127.0.0.1", peer_address);
// Second frame is zero
byte buffer[255];
TEST_ASSERT_EQUAL_INT (
0, TEST_ASSERT_SUCCESS_ERRNO (zmq_recv (stream, buffer, 255, 0)));
// Verify the existence of Peer-Address metadata
peer_address = zmq_msg_gets (&routing_id, "Peer-Address");
TEST_ASSERT_NOT_NULL (peer_address);
TEST_ASSERT_EQUAL_STRING ("127.0.0.1", peer_address);
// Real data follows
// First frame is routing id
TEST_ASSERT_SUCCESS_ERRNO (zmq_msg_recv (&routing_id, stream, 0));
TEST_ASSERT_TRUE (zmq_msg_more (&routing_id));
// Verify the existence of Peer-Address metadata
peer_address = zmq_msg_gets (&routing_id, "Peer-Address");
TEST_ASSERT_NOT_NULL (peer_address);
TEST_ASSERT_EQUAL_STRING ("127.0.0.1", peer_address);
// Second frame is greeting signature
recv_array_expect_success (stream, greeting.signature, 0);
// Send our own protocol greeting
TEST_ASSERT_SUCCESS_ERRNO (zmq_msg_send (&routing_id, stream, ZMQ_SNDMORE));
TEST_ASSERT_EQUAL_INT (
sizeof (greeting), TEST_ASSERT_SUCCESS_ERRNO (
zmq_send (stream, &greeting, sizeof (greeting), 0)));
// Now we expect the data from the DEALER socket
// We want the rest of greeting along with the Ready command
int bytes_read = 0;
while (bytes_read < 97) {
// First frame is the routing id of the connection (each time)
TEST_ASSERT_GREATER_THAN_INT (
0, TEST_ASSERT_SUCCESS_ERRNO (zmq_msg_recv (&routing_id, stream, 0)));
TEST_ASSERT_TRUE (zmq_msg_more (&routing_id));
// Second frame contains the next chunk of data
TEST_ASSERT_SUCCESS_ERRNO (
rc = zmq_recv (stream, buffer + bytes_read, 255 - bytes_read, 0));
bytes_read += rc;
}
// First two bytes are major and minor version numbers.
TEST_ASSERT_EQUAL_INT (3, buffer[0]); // ZMTP/3.1
TEST_ASSERT_EQUAL_INT (1, buffer[1]);
// Mechanism is "NULL"
TEST_ASSERT_EQUAL_INT8_ARRAY (buffer + 2,
"NULL\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0", 20);
TEST_ASSERT_EQUAL_INT8_ARRAY (buffer + 54, "\4\51\5READY", 8);
TEST_ASSERT_EQUAL_INT8_ARRAY (buffer + 62, "\13Socket-Type\0\0\0\6DEALER",
22);
TEST_ASSERT_EQUAL_INT8_ARRAY (buffer + 84, "\10Identity\0\0\0\0", 13);
// Announce we are ready
memcpy (buffer, "\4\51\5READY", 8);
memcpy (buffer + 8, "\13Socket-Type\0\0\0\6ROUTER", 22);
memcpy (buffer + 30, "\10Identity\0\0\0\0", 13);
// Send Ready command
TEST_ASSERT_GREATER_THAN_INT (0, TEST_ASSERT_SUCCESS_ERRNO (zmq_msg_send (
&routing_id, stream, ZMQ_SNDMORE)));
TEST_ASSERT_EQUAL_INT (
43, TEST_ASSERT_SUCCESS_ERRNO (zmq_send (stream, buffer, 43, 0)));
// Now we expect the data from the DEALER socket
// First frame is, again, the routing id of the connection
TEST_ASSERT_GREATER_THAN_INT (
0, TEST_ASSERT_SUCCESS_ERRNO (zmq_msg_recv (&routing_id, stream, 0)));
TEST_ASSERT_TRUE (zmq_msg_more (&routing_id));
// Third frame contains Hello message from DEALER
TEST_ASSERT_EQUAL_INT (7, TEST_ASSERT_SUCCESS_ERRNO (
zmq_recv (stream, buffer, sizeof buffer, 0)));
// Then we have a 5-byte message "Hello"
TEST_ASSERT_EQUAL_INT (0, buffer[0]); // Flags = 0
TEST_ASSERT_EQUAL_INT (5, buffer[1]); // Size = 5
TEST_ASSERT_EQUAL_INT8_ARRAY (buffer + 2, "Hello", 5);
// Send "World" back to DEALER
TEST_ASSERT_GREATER_THAN_INT (0, TEST_ASSERT_SUCCESS_ERRNO (zmq_msg_send (
&routing_id, stream, ZMQ_SNDMORE)));
byte world[] = {0, 5, 'W', 'o', 'r', 'l', 'd'};
TEST_ASSERT_EQUAL_INT (
sizeof (world),
TEST_ASSERT_SUCCESS_ERRNO (zmq_send (stream, world, sizeof (world), 0)));
// Expect response on DEALER socket
recv_string_expect_success (dealer, "World", 0);
// Test large messages over STREAM socket
#define size 64000
uint8_t msgout[size];
memset (msgout, 0xAB, size);
zmq_send (dealer, msgout, size, 0);
uint8_t msgin[9 + size];
memset (msgin, 0, 9 + size);
bytes_read = 0;
while (bytes_read < 9 + size) {
// Get routing id frame
TEST_ASSERT_GREATER_THAN_INT (
0, TEST_ASSERT_SUCCESS_ERRNO (zmq_recv (stream, buffer, 256, 0)));
// Get next chunk
TEST_ASSERT_GREATER_THAN_INT (
0,
TEST_ASSERT_SUCCESS_ERRNO (rc = zmq_recv (stream, msgin + bytes_read,
9 + size - bytes_read, 0)));
bytes_read += rc;
}
for (int byte_nbr = 0; byte_nbr < size; byte_nbr++) {
TEST_ASSERT_EQUAL_UINT8 (0xAB, msgin[9 + byte_nbr]);
}
test_context_socket_close (dealer);
test_context_socket_close (stream);
}
static void test_stream_to_stream ()
{
char my_endpoint[MAX_SOCKET_STRING];
// Set-up our context and sockets
void *server = test_context_socket (ZMQ_STREAM);
int enabled = 1;
TEST_ASSERT_SUCCESS_ERRNO (
zmq_setsockopt (server, ZMQ_STREAM_NOTIFY, &enabled, sizeof (enabled)));
bind_loopback_ipv4 (server, my_endpoint, sizeof my_endpoint);
void *client = test_context_socket (ZMQ_STREAM);
TEST_ASSERT_SUCCESS_ERRNO (
zmq_setsockopt (client, ZMQ_STREAM_NOTIFY, &enabled, sizeof (enabled)));
TEST_ASSERT_SUCCESS_ERRNO (zmq_connect (client, my_endpoint));
uint8_t id[256];
uint8_t buffer[256];
// Connecting sends a zero message
// Server: First frame is routing id, second frame is zero
TEST_ASSERT_GREATER_THAN_INT (
0, TEST_ASSERT_SUCCESS_ERRNO (zmq_recv (server, id, 256, 0)));
TEST_ASSERT_EQUAL_INT (
0, TEST_ASSERT_SUCCESS_ERRNO (zmq_recv (server, buffer, 256, 0)));
// Client: First frame is routing id, second frame is zero
TEST_ASSERT_GREATER_THAN_INT (
0, TEST_ASSERT_SUCCESS_ERRNO (zmq_recv (client, id, 256, 0)));
TEST_ASSERT_EQUAL_INT (
0, TEST_ASSERT_SUCCESS_ERRNO (zmq_recv (client, buffer, 256, 0)));
// Sent HTTP request on client socket
// Get server routing id
size_t id_size = sizeof id;
TEST_ASSERT_SUCCESS_ERRNO (
zmq_getsockopt (client, ZMQ_ROUTING_ID, id, &id_size));
// First frame is server routing id
TEST_ASSERT_EQUAL_INT ((int) id_size, TEST_ASSERT_SUCCESS_ERRNO (zmq_send (
client, id, id_size, ZMQ_SNDMORE)));
// Second frame is HTTP GET request
TEST_ASSERT_EQUAL_INT (
7, TEST_ASSERT_SUCCESS_ERRNO (zmq_send (client, "GET /\n\n", 7, 0)));
// Get HTTP request; ID frame and then request
TEST_ASSERT_GREATER_THAN_INT (
0, TEST_ASSERT_SUCCESS_ERRNO (zmq_recv (server, id, 256, 0)));
TEST_ASSERT_SUCCESS_ERRNO (zmq_recv (server, buffer, 256, 0));
TEST_ASSERT_EQUAL_INT8_ARRAY (buffer, "GET /\n\n", 7);
// Send reply back to client
char http_response[] = "HTTP/1.0 200 OK\r\n"
"Content-Type: text/plain\r\n"
"\r\n"
"Hello, World!";
TEST_ASSERT_SUCCESS_ERRNO (zmq_send (server, id, id_size, ZMQ_SNDMORE));
TEST_ASSERT_SUCCESS_ERRNO (
zmq_send (server, http_response, sizeof (http_response), ZMQ_SNDMORE));
// Send zero to close connection to client
TEST_ASSERT_SUCCESS_ERRNO (zmq_send (server, id, id_size, ZMQ_SNDMORE));
TEST_ASSERT_SUCCESS_ERRNO (zmq_send (server, NULL, 0, ZMQ_SNDMORE));
// Get reply at client and check that it's complete
TEST_ASSERT_GREATER_THAN_INT (
0, TEST_ASSERT_SUCCESS_ERRNO (zmq_recv (client, id, 256, 0)));
TEST_ASSERT_EQUAL_INT (
sizeof http_response,
TEST_ASSERT_SUCCESS_ERRNO (zmq_recv (client, buffer, 256, 0)));
TEST_ASSERT_EQUAL_INT8_ARRAY (buffer, http_response,
sizeof (http_response));
// // Get disconnection notification
// FIXME: why does this block? Bug in STREAM disconnect notification?
// id_size = zmq_recv (client, id, 256, 0);
// assert (id_size > 0);
// rc = zmq_recv (client, buffer, 256, 0);
// assert (rc == 0);
test_context_socket_close (server);
test_context_socket_close (client);
}
int main ()
{
setup_test_environment ();
UNITY_BEGIN ();
RUN_TEST (test_stream_to_dealer);
RUN_TEST (test_stream_to_stream);
return UNITY_END ();
}