0MQ/2.1 wire format compatibility implemented

- XS_PROTOCOL option added
- libxs ignores when unused flags are set to 1 (0MQ/2.1 bug)
- compatibility tests added

Signed-off-by: Martin Sustrik <sustrik@250bpm.com>

3 files changed, 184 insertions, 1 deletions

diff --git a/tests/Makefile.am b/tests/Makefile.am
index 817b0c1..8e2bbbf 100644
--- a/tests/Makefile.am
+++ b/tests/Makefile.am
@@ -21,7 +21,8 @@ noinst_PROGRAMS = pair_inproc \
                   max_sockets \
                   emptyctx \
                   polltimeo \
-                  wireformat
+                  wireformat \
+                  libzmq21
 
 pair_inproc_SOURCES = pair_inproc.cpp testutil.hpp
 pair_tcp_SOURCES = pair_tcp.cpp testutil.hpp
@@ -42,5 +43,6 @@ max_sockets_SOURCES = max_sockets.cpp
 emptyctx_SOURCES = emptyctx.cpp
 polltimeo_SOURCES = polltimeo.cpp testutil.hpp
 wireformat_SOURCES = wireformat.cpp
+libzmq21_SOURCES = libzmq21.cpp
 
 TESTS = $(noinst_PROGRAMS)
diff --git a/tests/libzmq21.cpp b/tests/libzmq21.cpp
new file mode 100644
index 0000000..e7affae
--- /dev/null
+++ b/tests/libzmq21.cpp
@@ -0,0 +1,175 @@
+/*
+    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 <http://www.gnu.org/licenses/>.
+*/
+
+#include "testutil.hpp"
+
+#if defined XS_HAVE_WINDOWS
+#include <winsock2.h>
+#else
+#include <fcntl.h>
+#include <sys/types.h>
+#include <sys/socket.h>
+#include <netinet/in.h>
+#include <netinet/tcp.h>
+#include <arpa/inet.h>
+#endif
+
+#if defined XS_HAVE_OPENVMS
+#include <ioctl.h>
+#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 ;
+}
diff --git a/tests/tests.cpp b/tests/tests.cpp
index d4756ab..f5c403a 100644
--- a/tests/tests.cpp
+++ b/tests/tests.cpp
@@ -103,6 +103,10 @@
 #include "wireformat.cpp"
 #undef XS_TEST_MAIN
 
+#define XS_TEST_MAIN libzmq21
+#include "libzmq21.cpp"
+#undef XS_TEST_MAIN
+
 int main ()
 {
     int rc;
@@ -143,6 +147,8 @@ int main ()
     assert (rc == 0);
     rc = wireformat ();
     assert (rc == 0);
+    rc = libzmq21 ();
+    assert (rc == 0);
 
     fprintf (stderr, "SUCCESS\n");
     sleep (1);