From 9d5a3793319769a1403ebc49e583afb5ba150aba Mon Sep 17 00:00:00 2001 From: Erik Hugne Date: Wed, 30 Oct 2013 16:03:27 +0100 Subject: [PATCH] zmq: add TIPC transport tests The tests are identical to the TCP ones, only the addressing is changed. Signed-off-by: Erik Hugne --- AUTHORS | 1 + tests/Makefile.am | 21 +++ tests/test_connect_delay_tipc.cpp | 238 +++++++++++++++++++++++++++ tests/test_pair_tipc.cpp | 54 ++++++ tests/test_reqrep_device_tipc.cpp | 143 ++++++++++++++++ tests/test_reqrep_tipc.cpp | 54 ++++++ tests/test_router_mandatory_tipc.cpp | 62 +++++++ tests/test_shutdown_stress_tipc.cpp | 93 +++++++++++ tests/test_sub_forward_tipc.cpp | 99 +++++++++++ tests/test_term_endpoint_tipc.cpp | 119 ++++++++++++++ 10 files changed, 884 insertions(+) create mode 100644 tests/test_connect_delay_tipc.cpp create mode 100644 tests/test_pair_tipc.cpp create mode 100644 tests/test_reqrep_device_tipc.cpp create mode 100644 tests/test_reqrep_tipc.cpp create mode 100644 tests/test_router_mandatory_tipc.cpp create mode 100644 tests/test_shutdown_stress_tipc.cpp create mode 100644 tests/test_sub_forward_tipc.cpp create mode 100644 tests/test_term_endpoint_tipc.cpp diff --git a/AUTHORS b/AUTHORS index 88e3ce25..75672929 100644 --- a/AUTHORS +++ b/AUTHORS @@ -37,6 +37,7 @@ Dhruva Krishnamurthy Dirk O. Kaar Douglas Creager Erich Heine +Erik Hugne Erik Rigtorp Fabien Ninoles Frank Denis diff --git a/tests/Makefile.am b/tests/Makefile.am index 5496b0f5..37dab0f1 100644 --- a/tests/Makefile.am +++ b/tests/Makefile.am @@ -51,6 +51,17 @@ noinst_PROGRAMS += test_shutdown_stress \ test_fork endif +if ON_LINUX +noinst_PROGRAMS += test_connect_delay_tipc \ + test_pair_tipc \ + test_reqrep_device_tipc \ + test_reqrep_tipc \ + test_router_mandatory_tipc \ + test_shutdown_stress_tipc \ + test_sub_forward_tipc \ + test_term_endpoint_tipc +endif + test_system_SOURCES = test_system.cpp test_pair_inproc_SOURCES = test_pair_inproc.cpp testutil.hpp test_pair_tcp_SOURCES = test_pair_tcp.cpp testutil.hpp @@ -97,6 +108,16 @@ test_reqrep_ipc_SOURCES = test_reqrep_ipc.cpp testutil.hpp test_timeo_SOURCES = test_timeo.cpp test_fork_SOURCES = test_fork.cpp endif +if ON_LINUX +test_connect_delay_tipc_SOURCES = test_connect_delay_tipc.cpp +test_pair_tipc_SOURCES = test_pair_tipc.cpp +test_reqrep_device_tipc_SOURCES = test_reqrep_device_tipc.cpp +test_reqrep_tipc_SOURCES = test_reqrep_tipc.cpp +test_router_mandatory_tipc_SOURCES = test_router_mandatory_tipc.cpp +test_shutdown_stress_tipc_SOURCES = test_shutdown_stress_tipc.cpp +test_sub_forward_tipc_SOURCES = test_sub_forward_tipc.cpp +test_term_endpoint_tipc_SOURCES = test_term_endpoint_tipc.cpp +endif # Run the test cases TESTS = $(noinst_PROGRAMS) diff --git a/tests/test_connect_delay_tipc.cpp b/tests/test_connect_delay_tipc.cpp new file mode 100644 index 00000000..940edaaa --- /dev/null +++ b/tests/test_connect_delay_tipc.cpp @@ -0,0 +1,238 @@ +/* + Copyright (c) 2007-2013 Contributors as noted in the AUTHORS file + + This file is part of 0MQ. + + 0MQ is free software; you can redistribute it and/or modify it under + the terms of the GNU Lesser General Public License as published by + the Free Software Foundation; either version 3 of the License, or + (at your option) any later version. + + 0MQ is distributed in the hope that it will be useful, + but WITHOUT ANY WARRANTY; without even the implied warranty of + MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + GNU Lesser General Public License for more details. + + You should have received a copy of the GNU Lesser General Public License + along with this program. If not, see . +*/ + +#include "../include/zmq.h" +#include +#include +#include +#include +#include + +#undef NDEBUG +#include + +int main (void) +{ + int val; + int rc; + char buffer[16]; + // 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 *context = zmq_ctx_new(); + assert (context); + void *to = zmq_socket(context, ZMQ_PULL); + assert (to); + + // Bind the one valid receiver + val = 0; + rc = zmq_setsockopt(to, ZMQ_LINGER, &val, sizeof(val)); + assert (rc == 0); + rc = zmq_bind (to, "tipc://{6555,0,0}"); + assert (rc == 0); + + // Create a socket pushing to two endpoints - only 1 message should arrive. + void *from = zmq_socket (context, ZMQ_PUSH); + assert(from); + + val = 0; + zmq_setsockopt (from, ZMQ_LINGER, &val, sizeof (val)); + // This pipe will not connect + rc = zmq_connect (from, "tipc://{5556,0}"); + assert (rc == 0); + // This pipe will + rc = zmq_connect (from, "tipc://{6555,0}"); + assert (rc == 0); + + // We send 10 messages, 5 should just get stuck in the queue + // for the not-yet-connected pipe + for (int i = 0; i < 10; ++i) { + rc = zmq_send (from, "Hello", 5, 0); + assert (rc == 5); + } + + // We now consume from the connected pipe + // - we should see just 5 + int timeout = 100; + rc = zmq_setsockopt (to, ZMQ_RCVTIMEO, &timeout, sizeof (int)); + assert (rc == 0); + + int seen = 0; + while (true) { + rc = zmq_recv (to, &buffer, sizeof (buffer), 0); + if (rc == -1) + break; // Break when we didn't get a message + seen++; + } + assert (seen == 5); + + rc = zmq_close (from); + assert (rc == 0); + + rc = zmq_close (to); + assert (rc == 0); + + rc = zmq_term (context); + assert (rc == 0); + + // 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. + context = zmq_ctx_new(); + + // Bind the valid socket + to = zmq_socket (context, ZMQ_PULL); + assert (to); + rc = zmq_bind (to, "tipc://{5560,0,0}"); + assert (rc == 0); + + val = 0; + rc = zmq_setsockopt (to, ZMQ_LINGER, &val, sizeof(val)); + assert (rc == 0); + + // Create a socket pushing to two endpoints - all messages should arrive. + from = zmq_socket (context, ZMQ_PUSH); + assert (from); + + val = 0; + rc = zmq_setsockopt (from, ZMQ_LINGER, &val, sizeof(val)); + assert (rc == 0); + + // Set the key flag + val = 1; + rc = zmq_setsockopt (from, ZMQ_DELAY_ATTACH_ON_CONNECT, &val, sizeof(val)); + assert (rc == 0); + + // Connect to the invalid socket + rc = zmq_connect (from, "tipc://{5561,0}"); + assert (rc == 0); + // Connect to the valid socket + rc = zmq_connect (from, "tipc://{5560,0}"); + assert (rc == 0); + + // Send 10 messages, all should be routed to the connected pipe + for (int i = 0; i < 10; ++i) { + rc = zmq_send (from, "Hello", 5, 0); + assert (rc == 5); + } + rc = zmq_setsockopt (to, ZMQ_RCVTIMEO, &timeout, sizeof (int)); + assert (rc == 0); + + seen = 0; + while (true) { + rc = zmq_recv (to, &buffer, sizeof (buffer), 0); + if (rc == -1) + break; // Break when we didn't get a message + seen++; + } + assert (seen == 10); + + rc = zmq_close (from); + assert (rc == 0); + + rc = zmq_close (to); + assert (rc == 0); + + rc = zmq_term (context); + assert (rc == 0); + + // 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. + context = zmq_ctx_new (); + + void *backend = zmq_socket (context, ZMQ_DEALER); + assert (backend); + void *frontend = zmq_socket (context, ZMQ_DEALER); + assert (frontend); + int zero = 0; + rc = zmq_setsockopt (backend, ZMQ_LINGER, &zero, sizeof (zero)); + assert (rc == 0); + rc = zmq_setsockopt (frontend, ZMQ_LINGER, &zero, sizeof (zero)); + assert (rc == 0); + + // Frontend connects to backend using DELAY_ATTACH_ON_CONNECT + int on = 1; + rc = zmq_setsockopt (frontend, ZMQ_DELAY_ATTACH_ON_CONNECT, &on, sizeof (on)); + assert (rc == 0); + rc = zmq_bind (backend, "tipc://{5560,0,0}"); + assert (rc == 0); + rc = zmq_connect (frontend, "tipc://{5560,0}"); + assert (rc == 0); + + // Ping backend to frontend so we know when the connection is up + rc = zmq_send (backend, "Hello", 5, 0); + assert (rc == 5); + rc = zmq_recv (frontend, buffer, 255, 0); + assert (rc == 5); + + // Send message from frontend to backend + rc = zmq_send (frontend, "Hello", 5, ZMQ_DONTWAIT); + assert (rc == 5); + + rc = zmq_close (backend); + assert (rc == 0); + + // Give time to process disconnect + // There's no way to do this except with a sleep + struct timespec t = { 0, 250 * 1000000 }; + nanosleep (&t, NULL); + + // Send a message, should fail + rc = zmq_send (frontend, "Hello", 5, ZMQ_DONTWAIT); + assert (rc == -1); + + // Recreate backend socket + backend = zmq_socket (context, ZMQ_DEALER); + assert (backend); + rc = zmq_setsockopt (backend, ZMQ_LINGER, &zero, sizeof (zero)); + assert (rc == 0); + rc = zmq_bind (backend, "tipc://{5560,0,0}"); + assert (rc == 0); + + // Ping backend to frontend so we know when the connection is up + rc = zmq_send (backend, "Hello", 5, 0); + assert (rc == 5); + rc = zmq_recv (frontend, buffer, 255, 0); + assert (rc == 5); + + // After the reconnect, should succeed + rc = zmq_send (frontend, "Hello", 5, ZMQ_DONTWAIT); + assert (rc == 5); + + rc = zmq_close (backend); + assert (rc == 0); + + rc = zmq_close (frontend); + assert (rc == 0); + + rc = zmq_term (context); + assert (rc == 0); +} + diff --git a/tests/test_pair_tipc.cpp b/tests/test_pair_tipc.cpp new file mode 100644 index 00000000..9cca6394 --- /dev/null +++ b/tests/test_pair_tipc.cpp @@ -0,0 +1,54 @@ +/* + Copyright (c) 2010-2011 250bpm s.r.o. + Copyright (c) 2011 iMatix Corporation + Copyright (c) 2010-2011 Other contributors as noted in the AUTHORS file + + This file is part of 0MQ. + + 0MQ is free software; you can redistribute it and/or modify it under + the terms of the GNU Lesser General Public License as published by + the Free Software Foundation; either version 3 of the License, or + (at your option) any later version. + + 0MQ is distributed in the hope that it will be useful, + but WITHOUT ANY WARRANTY; without even the implied warranty of + MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + GNU Lesser General Public License for more details. + + You should have received a copy of the GNU Lesser General Public License + along with this program. If not, see . +*/ + +#include +#include "testutil.hpp" + +int main (void) +{ + fprintf (stderr, "test_pair_tipc running...\n"); + + void *ctx = zmq_init (1); + assert (ctx); + + void *sb = zmq_socket (ctx, ZMQ_PAIR); + assert (sb); + int rc = zmq_bind (sb, "tipc://{5560,0,0}"); + assert (rc == 0); + + void *sc = zmq_socket (ctx, ZMQ_PAIR); + assert (sc); + rc = zmq_connect (sc, "tipc://{5560,0}"); + assert (rc == 0); + + bounce (sb, sc); + + rc = zmq_close (sc); + assert (rc == 0); + + rc = zmq_close (sb); + assert (rc == 0); + + rc = zmq_term (ctx); + assert (rc == 0); + + return 0 ; +} diff --git a/tests/test_reqrep_device_tipc.cpp b/tests/test_reqrep_device_tipc.cpp new file mode 100644 index 00000000..ffc6ff9e --- /dev/null +++ b/tests/test_reqrep_device_tipc.cpp @@ -0,0 +1,143 @@ +/* + Copyright (c) 2010-2011 250bpm s.r.o. + Copyright (c) 2011 VMware, Inc. + Copyright (c) 2010-2011 Other contributors as noted in the AUTHORS file + + This file is part of 0MQ. + + 0MQ is free software; you can redistribute it and/or modify it under + the terms of the GNU Lesser General Public License as published by + the Free Software Foundation; either version 3 of the License, or + (at your option) any later version. + + 0MQ is distributed in the hope that it will be useful, + but WITHOUT ANY WARRANTY; without even the implied warranty of + MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + GNU Lesser General Public License for more details. + + You should have received a copy of the GNU Lesser General Public License + along with this program. If not, see . +*/ + +#include "../include/zmq.h" +#include +#include + +#undef NDEBUG +#include + +int main (void) +{ + fprintf (stderr, "test_reqrep_device_tipc running...\n"); + + void *ctx = zmq_init (1); + assert (ctx); + + // Create a req/rep device. + void *dealer = zmq_socket (ctx, ZMQ_DEALER); + assert (dealer); + int rc = zmq_bind (dealer, "tipc://{5560,0,0}"); + assert (rc == 0); + void *router = zmq_socket (ctx, ZMQ_ROUTER); + assert (router); + rc = zmq_bind (router, "tipc://{5561,0,0}"); + assert (rc == 0); + + // Create a worker. + void *rep = zmq_socket (ctx, ZMQ_REP); + assert (rep); + rc = zmq_connect (rep, "tipc://{5560,0}"); + assert (rc == 0); + + // Create a client. + void *req = zmq_socket (ctx, ZMQ_REQ); + assert (req); + rc = zmq_connect (req, "tipc://{5561,0}"); + assert (rc == 0); + + // Send a request. + rc = zmq_send (req, "ABC", 3, ZMQ_SNDMORE); + assert (rc == 3); + rc = zmq_send (req, "DEF", 3, 0); + assert (rc == 3); + + // Pass the request through the device. + for (int i = 0; i != 4; i++) { + zmq_msg_t msg; + rc = zmq_msg_init (&msg); + assert (rc == 0); + rc = zmq_recvmsg (router, &msg, 0); + assert (rc >= 0); + int rcvmore; + size_t sz = sizeof (rcvmore); + rc = zmq_getsockopt (router, ZMQ_RCVMORE, &rcvmore, &sz); + assert (rc == 0); + rc = zmq_sendmsg (dealer, &msg, rcvmore ? ZMQ_SNDMORE : 0); + assert (rc >= 0); + } + + // Receive the request. + char buff [3]; + rc = zmq_recv (rep, buff, 3, 0); + assert (rc == 3); + assert (memcmp (buff, "ABC", 3) == 0); + int rcvmore; + size_t sz = sizeof (rcvmore); + rc = zmq_getsockopt (rep, ZMQ_RCVMORE, &rcvmore, &sz); + assert (rc == 0); + assert (rcvmore); + rc = zmq_recv (rep, buff, 3, 0); + assert (rc == 3); + assert (memcmp (buff, "DEF", 3) == 0); + rc = zmq_getsockopt (rep, ZMQ_RCVMORE, &rcvmore, &sz); + assert (rc == 0); + assert (!rcvmore); + + // Send the reply. + rc = zmq_send (rep, "GHI", 3, ZMQ_SNDMORE); + assert (rc == 3); + rc = zmq_send (rep, "JKL", 3, 0); + assert (rc == 3); + + // Pass the reply through the device. + for (int i = 0; i != 4; i++) { + zmq_msg_t msg; + rc = zmq_msg_init (&msg); + assert (rc == 0); + rc = zmq_recvmsg (dealer, &msg, 0); + assert (rc >= 0); + int rcvmore; + rc = zmq_getsockopt (dealer, ZMQ_RCVMORE, &rcvmore, &sz); + assert (rc == 0); + rc = zmq_sendmsg (router, &msg, rcvmore ? ZMQ_SNDMORE : 0); + assert (rc >= 0); + } + + // Receive the reply. + rc = zmq_recv (req, buff, 3, 0); + assert (rc == 3); + assert (memcmp (buff, "GHI", 3) == 0); + rc = zmq_getsockopt (req, ZMQ_RCVMORE, &rcvmore, &sz); + assert (rc == 0); + assert (rcvmore); + rc = zmq_recv (req, buff, 3, 0); + assert (rc == 3); + assert (memcmp (buff, "JKL", 3) == 0); + rc = zmq_getsockopt (req, ZMQ_RCVMORE, &rcvmore, &sz); + assert (rc == 0); + assert (!rcvmore); + + // Clean up. + rc = zmq_close (req); + assert (rc == 0); + rc = zmq_close (rep); + assert (rc == 0); + rc = zmq_close (router); + assert (rc == 0); + rc = zmq_close (dealer); + assert (rc == 0); + rc = zmq_term (ctx); + assert (rc == 0); + + return 0 ; +} diff --git a/tests/test_reqrep_tipc.cpp b/tests/test_reqrep_tipc.cpp new file mode 100644 index 00000000..fabe19f3 --- /dev/null +++ b/tests/test_reqrep_tipc.cpp @@ -0,0 +1,54 @@ +/* + Copyright (c) 2010-2011 250bpm s.r.o. + Copyright (c) 2011 iMatix Corporation + Copyright (c) 2010-2011 Other contributors as noted in the AUTHORS file + + This file is part of 0MQ. + + 0MQ is free software; you can redistribute it and/or modify it under + the terms of the GNU Lesser General Public License as published by + the Free Software Foundation; either version 3 of the License, or + (at your option) any later version. + + 0MQ is distributed in the hope that it will be useful, + but WITHOUT ANY WARRANTY; without even the implied warranty of + MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + GNU Lesser General Public License for more details. + + You should have received a copy of the GNU Lesser General Public License + along with this program. If not, see . +*/ + +#include +#include "testutil.hpp" + +int main (void) +{ + fprintf (stderr, "test_reqrep_tipc running...\n"); + + void *ctx = zmq_init (1); + assert (ctx); + + void *sb = zmq_socket (ctx, ZMQ_REP); + assert (sb); + int rc = zmq_bind (sb, "tipc://{5560,0,0}"); + assert (rc == 0); + + void *sc = zmq_socket (ctx, ZMQ_REQ); + assert (sc); + rc = zmq_connect (sc, "tipc://{5560,0}"); + assert (rc == 0); + + bounce (sb, sc); + + rc = zmq_close (sc); + assert (rc == 0); + + rc = zmq_close (sb); + assert (rc == 0); + + rc = zmq_term (ctx); + assert (rc == 0); + + return 0 ; +} diff --git a/tests/test_router_mandatory_tipc.cpp b/tests/test_router_mandatory_tipc.cpp new file mode 100644 index 00000000..0e972bbe --- /dev/null +++ b/tests/test_router_mandatory_tipc.cpp @@ -0,0 +1,62 @@ +/* + Copyright (c) 2010-2011 250bpm s.r.o. + Copyright (c) 2011 iMatix Corporation + Copyright (c) 2010-2011 Other contributors as noted in the AUTHORS file + + This file is part of 0MQ. + + 0MQ is free software; you can redistribute it and/or modify it under + the terms of the GNU Lesser General Public License as published by + the Free Software Foundation; either version 3 of the License, or + (at your option) any later version. + + 0MQ is distributed in the hope that it will be useful, + but WITHOUT ANY WARRANTY; without even the implied warranty of + MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + GNU Lesser General Public License for more details. + + You should have received a copy of the GNU Lesser General Public License + along with this program. If not, see . +*/ + +#include +#include "testutil.hpp" + +int main (void) +{ + fprintf (stderr, "test_router_mandatory_tipc running...\n"); + + void *ctx = zmq_init (1); + assert (ctx); + + // Creating the first socket. + void *sa = zmq_socket (ctx, ZMQ_ROUTER); + assert (sa); + + int rc = zmq_bind (sa, "tipc://{15560,0,0}"); + assert (rc == 0); + + // Sending a message to an unknown peer with the default setting + rc = zmq_send (sa, "UNKNOWN", 7, ZMQ_SNDMORE); + assert (rc == 7); + rc = zmq_send (sa, "DATA", 4, 0); + assert (rc == 4); + + int mandatory = 1; + + // Set mandatory routing on socket + rc = zmq_setsockopt (sa, ZMQ_ROUTER_MANDATORY, &mandatory, sizeof (mandatory)); + assert (rc == 0); + + // Send a message and check that it fails + rc = zmq_send (sa, "UNKNOWN", 7, ZMQ_SNDMORE | ZMQ_DONTWAIT); + assert (rc == -1 && errno == EHOSTUNREACH); + + rc = zmq_close (sa); + assert (rc == 0); + + rc = zmq_term (ctx); + assert (rc == 0); + + return 0 ; +} diff --git a/tests/test_shutdown_stress_tipc.cpp b/tests/test_shutdown_stress_tipc.cpp new file mode 100644 index 00000000..51a5b988 --- /dev/null +++ b/tests/test_shutdown_stress_tipc.cpp @@ -0,0 +1,93 @@ +/* + Copyright (c) 2010-2011 250bpm s.r.o. + Copyright (c) 2011 iMatix Corporation + Copyright (c) 2010-2011 Other contributors as noted in the AUTHORS file + + This file is part of 0MQ. + + 0MQ is free software; you can redistribute it and/or modify it under + the terms of the GNU Lesser General Public License as published by + the Free Software Foundation; either version 3 of the License, or + (at your option) any later version. + + 0MQ is distributed in the hope that it will be useful, + but WITHOUT ANY WARRANTY; without even the implied warranty of + MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + GNU Lesser General Public License for more details. + + You should have received a copy of the GNU Lesser General Public License + along with this program. If not, see . +*/ + +#include "../include/zmq.h" +#include +#include +#include + +#undef NDEBUG +#include + +#define THREAD_COUNT 100 + +extern "C" +{ + static void *worker (void *s) + { + int rc; + + rc = zmq_connect (s, "tipc://{5560,0}"); + assert (rc == 0); + + // Start closing the socket while the connecting process is underway. + rc = zmq_close (s); + assert (rc == 0); + + return NULL; + } +} + +int main (void) +{ + void *ctx; + void *s1; + void *s2; + int i; + int j; + int rc; + pthread_t threads [THREAD_COUNT]; + + fprintf (stderr, "test_shutdown_stress_tipc running...\n"); + + for (j = 0; j != 10; j++) { + + // Check the shutdown with many parallel I/O threads. + ctx = zmq_init (7); + assert (ctx); + + s1 = zmq_socket (ctx, ZMQ_PUB); + assert (s1); + + rc = zmq_bind (s1, "tipc://{5560,0,0}"); + assert (rc == 0); + + for (i = 0; i != THREAD_COUNT; i++) { + s2 = zmq_socket (ctx, ZMQ_SUB); + assert (s2); + rc = pthread_create (&threads [i], NULL, worker, s2); + assert (rc == 0); + } + + for (i = 0; i != THREAD_COUNT; i++) { + rc = pthread_join (threads [i], NULL); + assert (rc == 0); + } + + rc = zmq_close (s1); + assert (rc == 0); + + rc = zmq_term (ctx); + assert (rc == 0); + } + + return 0; +} diff --git a/tests/test_sub_forward_tipc.cpp b/tests/test_sub_forward_tipc.cpp new file mode 100644 index 00000000..1dda2f3a --- /dev/null +++ b/tests/test_sub_forward_tipc.cpp @@ -0,0 +1,99 @@ +/* + Copyright (c) 2010-2011 250bpm s.r.o. + Copyright (c) 2011 iMatix Corporation + Copyright (c) 2010-2011 Other contributors as noted in the AUTHORS file + + This file is part of 0MQ. + + 0MQ is free software; you can redistribute it and/or modify it under + the terms of the GNU Lesser General Public License as published by + the Free Software Foundation; either version 3 of the License, or + (at your option) any later version. + + 0MQ is distributed in the hope that it will be useful, + but WITHOUT ANY WARRANTY; without even the implied warranty of + MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + GNU Lesser General Public License for more details. + + You should have received a copy of the GNU Lesser General Public License + along with this program. If not, see . +*/ + +#include "../include/zmq.h" +#include "../include/zmq_utils.h" +#include + +#undef NDEBUG +#include + +int main (void) +{ + fprintf (stderr, "test_sub_forward running...\n"); + + void *ctx = zmq_init (1); + assert (ctx); + + // First, create an intermediate device. + void *xpub = zmq_socket (ctx, ZMQ_XPUB); + assert (xpub); + int rc = zmq_bind (xpub, "tipc://{5560,0,0}"); + assert (rc == 0); + void *xsub = zmq_socket (ctx, ZMQ_XSUB); + assert (xsub); + rc = zmq_bind (xsub, "tipc://{5561,0,0}"); + assert (rc == 0); + + // Create a publisher. + void *pub = zmq_socket (ctx, ZMQ_PUB); + assert (pub); + rc = zmq_connect (pub, "tipc://{5561,0}"); + assert (rc == 0); + + // Create a subscriber. + void *sub = zmq_socket (ctx, ZMQ_SUB); + assert (sub); + rc = zmq_connect (sub, "tipc://{5560,0}"); + assert (rc == 0); + + // Subscribe for all messages. + rc = zmq_setsockopt (sub, ZMQ_SUBSCRIBE, "", 0); + assert (rc == 0); + + // Pass the subscription upstream through the device. + char buff [32]; + rc = zmq_recv (xpub, buff, sizeof (buff), 0); + assert (rc >= 0); + rc = zmq_send (xsub, buff, rc, 0); + assert (rc >= 0); + + // Wait a bit till the subscription gets to the publisher. + zmq_sleep (1); + + // Send an empty message. + rc = zmq_send (pub, NULL, 0, 0); + assert (rc == 0); + + // Pass the message downstream through the device. + rc = zmq_recv (xsub, buff, sizeof (buff), 0); + assert (rc >= 0); + rc = zmq_send (xpub, buff, rc, 0); + assert (rc >= 0); + + // Receive the message in the subscriber. + rc = zmq_recv (sub, buff, sizeof (buff), 0); + assert (rc == 0); + + // Clean up. + rc = zmq_close (xpub); + assert (rc == 0); + rc = zmq_close (xsub); + assert (rc == 0); + rc = zmq_close (pub); + assert (rc == 0); + rc = zmq_close (sub); + assert (rc == 0); + rc = zmq_term (ctx); + assert (rc == 0); + + return 0 ; +} diff --git a/tests/test_term_endpoint_tipc.cpp b/tests/test_term_endpoint_tipc.cpp new file mode 100644 index 00000000..c338ade4 --- /dev/null +++ b/tests/test_term_endpoint_tipc.cpp @@ -0,0 +1,119 @@ +/* + Copyright (c) 2010-2011 250bpm s.r.o. + Copyright (c) 2011 iMatix Corporation + Copyright (c) 2010-2011 Other contributors as noted in the AUTHORS file + + This file is part of 0MQ. + + 0MQ is free software; you can redistribute it and/or modify it under + the terms of the GNU Lesser General Public License as published by + the Free Software Foundation; either version 3 of the License, or + (at your option) any later version. + + 0MQ is distributed in the hope that it will be useful, + but WITHOUT ANY WARRANTY; without even the implied warranty of + MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + GNU Lesser General Public License for more details. + + You should have received a copy of the GNU Lesser General Public License + along with this program. If not, see . +*/ + +#include "../include/zmq.h" +#include "../include/zmq_utils.h" +#include +#include + +#undef NDEBUG +#include + +int main (void) +{ + int rc; + char buf[32]; + const char *ep = "tipc://{5560,0,0}"; + const char *name = "tipc://{5560,0}"; + + fprintf (stderr, "unbind endpoint test running...\n"); + + // Create infrastructure. + void *ctx = zmq_init (1); + assert (ctx); + void *push = zmq_socket (ctx, ZMQ_PUSH); + assert (push); + rc = zmq_bind (push, ep); + assert (rc == 0); + void *pull = zmq_socket (ctx, ZMQ_PULL); + assert (pull); + rc = zmq_connect (pull, name); + assert (rc == 0); + + // Pass one message through to ensure the connection is established. + rc = zmq_send (push, "ABC", 3, 0); + assert (rc == 3); + rc = zmq_recv (pull, buf, sizeof (buf), 0); + assert (rc == 3); + + // Unbind the lisnening endpoint + rc = zmq_unbind (push, ep); + assert (rc == 0); + + // Let events some time + zmq_sleep (1); + + // Check that sending would block (there's no outbound connection). + rc = zmq_send (push, "ABC", 3, ZMQ_DONTWAIT); + assert (rc == -1 && zmq_errno () == EAGAIN); + + // Clean up. + rc = zmq_close (pull); + assert (rc == 0); + rc = zmq_close (push); + assert (rc == 0); + rc = zmq_term (ctx); + assert (rc == 0); + + + // Now the other way round. + fprintf (stderr, "disconnect endpoint test running...\n"); + + + // Create infrastructure. + ctx = zmq_init (1); + assert (ctx); + push = zmq_socket (ctx, ZMQ_PUSH); + assert (push); + rc = zmq_connect (push, name); + assert (rc == 0); + pull = zmq_socket (ctx, ZMQ_PULL); + assert (pull); + rc = zmq_bind (pull, ep); + assert (rc == 0); + + // Pass one message through to ensure the connection is established. + rc = zmq_send (push, "ABC", 3, 0); + assert (rc == 3); + rc = zmq_recv (pull, buf, sizeof (buf), 0); + assert (rc == 3); + + // Disconnect the bound endpoint + rc = zmq_disconnect (push, name); + assert (rc == 0); + + // Let events some time + zmq_sleep (1); + + // Check that sending would block (there's no inbound connections). + rc = zmq_send (push, "ABC", 3, ZMQ_DONTWAIT); + assert (rc == -1 && zmq_errno () == EAGAIN); + + // Clean up. + rc = zmq_close (pull); + assert (rc == 0); + rc = zmq_close (push); + assert (rc == 0); + rc = zmq_term (ctx); + assert (rc == 0); + + return 0; +}