/* Copyright (c) 2012 250bpm s.r.o. Copyright (c) 2012 Paul Colomiets Copyright (c) 2012 Other contributors as noted in the AUTHORS file This file is part of Crossroads I/O project. Crossroads I/O 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. Crossroads 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 "testutil.hpp" #if defined XS_HAVE_WINDOWS #include #else #include #include #include #include #include #include #endif #if defined XS_HAVE_OPENVMS #include #endif int XS_TEST_MAIN () { fprintf (stderr, "libzmq21 test running...\n"); #if defined XS_HAVE_WINDOWS WSADATA info; int wsarc = WSAStartup (MAKEWORD(1,1), &info); assert (wsarc == 0); #endif // First, test up-to-date publisher with 0MQ/2.1-style subscriber. // Create the basic infrastructure. void *ctx = xs_init (); assert (ctx); void *pub = xs_socket (ctx, XS_PUB); assert (pub); int protocol = 1; int rc = xs_setsockopt (pub, XS_PROTOCOL, &protocol, sizeof (protocol)); assert (rc == 0); rc = xs_bind (pub, "tcp://127.0.0.1:5560"); assert (rc == 0); int oldsub = socket (AF_INET, SOCK_STREAM, IPPROTO_TCP); struct sockaddr_in address; address.sin_family = AF_INET; address.sin_addr.s_addr = inet_addr ("127.0.0.1"); address.sin_port = htons (5560); rc = connect (oldsub, (struct sockaddr*) &address, sizeof (address)); assert (rc == 0); // Wait a while to allow connection to be accepted on the publisher side. sleep (1); // Send a message and check whether it arrives although there was no // subscription sent. rc = xs_send (pub, "ABC", 3, 0); assert (rc == 3); char buf [5]; rc = recv (oldsub, buf, sizeof (buf), 0); assert (rc == 5); assert (!memcmp (buf, "\x04\0ABC", 5)); // Tear down the infrastructure. rc = xs_close (pub); assert (rc == 0); #if defined XS_HAVE_WINDOWS rc = closesocket (oldsub); assert (rc != SOCKET_ERROR); #else rc = close (oldsub); assert (rc == 0); #endif rc = xs_term (ctx); assert (rc == 0); // Second, test the 0MQ/2.1-style publisher with up-to-date subscriber. // Create the basic infrastructure. ctx = xs_init (); assert (ctx); void *sub = xs_socket (ctx, XS_SUB); assert (sub); protocol = 1; rc = xs_setsockopt (sub, XS_PROTOCOL, &protocol, sizeof (protocol)); assert (rc == 0); rc = xs_setsockopt (sub, XS_SUBSCRIBE, "", 0); assert (rc == 0); rc = xs_bind (sub, "tcp://127.0.0.1:5560"); assert (rc == 0); int oldpub = socket (AF_INET, SOCK_STREAM, IPPROTO_TCP); address.sin_family = AF_INET; address.sin_addr.s_addr = inet_addr ("127.0.0.1"); address.sin_port = htons (5560); rc = connect (oldpub, (struct sockaddr*) &address, sizeof (address)); assert (rc == 0); // Wait a while to allow connection to be accepted on the subscriber side. sleep (1); // Set the socket to the non-blocking mode. #ifdef XS_HAVE_WINDOWS u_long nonblock = 1; rc = ioctlsocket (oldpub, FIONBIO, &nonblock); assert (rc != SOCKET_ERROR); #elif XS_HAVE_OPENVMS int nonblock = 1; rc = ioctl (oldpub, FIONBIO, &nonblock); assert (rc != -1); #else int flags = fcntl (oldpub, F_GETFL, 0); if (flags == -1) flags = 0; rc = fcntl (oldpub, F_SETFL, flags | O_NONBLOCK); assert (rc != -1); #endif // Check that subscription haven't arrived at the publisher side. rc = recv (oldpub, buf, sizeof (buf), 0); #if defined XS_HAVE_WINDOWS assert (rc == SOCKET_ERROR && WSAGetLastError () == WSAEWOULDBLOCK); #else assert (rc == -1 && (errno == EAGAIN || errno == EWOULDBLOCK)); #endif // Pass one message through. rc = send (oldpub, "\x04\0ABC", 5, 0); assert (rc == 5); rc = xs_recv (sub, buf, sizeof (buf), 0); assert (rc == 3); // Pass one message with usused flags set (0MQ/2.1 bug). rc = send (oldpub, "\x04\xfe" "ABC", 5, 0); assert (rc == 5); rc = xs_recv (sub, buf, sizeof (buf), 0); assert (rc == 3); // Tear down the infrastructure. rc = xs_close (sub); assert (rc == 0); #if defined XS_HAVE_WINDOWS rc = closesocket (oldpub); assert (rc != SOCKET_ERROR); #else rc = close (oldpub); assert (rc == 0); #endif rc = xs_term (ctx); assert (rc == 0); #if defined XS_HAVE_WINDOWS WSACleanup (); #endif return 0 ; }