zmq: add TIPC transport tests

The tests are identical to the TCP ones, only the
addressing is changed.
Signed-off-by: Erik Hugne <erik.hugne@ericsson.com>

AUTHORS

@@ -37,6 +37,7 @@ Dhruva Krishnamurthy <dhruva@ymail.com>
 Dirk O. Kaar <dok@dok-net.net>
 Douglas Creager <douglas.creager@redjack.com>
 Erich Heine <sophacles@gmail.com>
+Erik Hugne <erik.hugne@ericsson.com>
 Erik Rigtorp <erik@rigtorp.com>
 Fabien Ninoles <fabien@tzone.org>
 Frank Denis <zeromq@pureftpd.org>

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)

tests/test_connect_delay_tipc.cpp

+/*
+    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 <http://www.gnu.org/licenses/>.
+*/
+
+#include "../include/zmq.h"
+#include <errno.h>
+#include <stdlib.h>
+#include <string.h>
+#include <unistd.h>
+#include <string>
+
+#undef NDEBUG
+#include <assert.h>
+
+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);
+}
+

tests/test_pair_tipc.cpp

+/*
+    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 <http://www.gnu.org/licenses/>.
+*/
+
+#include <stdio.h>
+#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 ;
+}

tests/test_reqrep_device_tipc.cpp

+/*
+    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 <http://www.gnu.org/licenses/>.
+*/
+
+#include "../include/zmq.h"
+#include <stdio.h>
+#include <string.h>
+
+#undef NDEBUG
+#include <assert.h>
+
+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 ;
+}

tests/test_reqrep_tipc.cpp

+/*
+    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 <http://www.gnu.org/licenses/>.
+*/
+
+#include <stdio.h>
+#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 ;
+}

tests/test_router_mandatory_tipc.cpp

+/*
+    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 <http://www.gnu.org/licenses/>.
+*/
+
+#include <stdio.h>
+#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 ;
+}

tests/test_shutdown_stress_tipc.cpp

+/*
+    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 <http://www.gnu.org/licenses/>.
+*/
+
+#include "../include/zmq.h"
+#include <pthread.h>
+#include <stddef.h>
+#include <stdio.h>
+
+#undef NDEBUG
+#include <assert.h>
+
+#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;
+}

tests/test_sub_forward_tipc.cpp

+/*
+    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 <http://www.gnu.org/licenses/>.
+*/
+
+#include "../include/zmq.h"
+#include "../include/zmq_utils.h"
+#include <stdio.h>
+
+#undef NDEBUG
+#include <assert.h>
+
+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 ;
+}

tests/test_term_endpoint_tipc.cpp

+/*
+    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 <http://www.gnu.org/licenses/>.
+*/
+
+#include "../include/zmq.h"
+#include "../include/zmq_utils.h"
+#include <string.h>
+#include <unistd.h>
+
+#undef NDEBUG
+#include <assert.h>
+
+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;
+}