Client reports when it gets a confused message

client.c

//
//  Clone client model 1
//
#include "zhelpers.h"

int main (void)
{
    s_catch_signals ();
    while (!s_interrupted) {
        //  Prepare our context and socket
        void *context = zmq_init (1);
        void *socket = zmq_socket (context, ZMQ_SUB);
        zmq_setsockopt (socket, ZMQ_SUBSCRIBE, "", 0);
        zmq_connect (socket, "tcp://localhost:5556");

        int sequence = 0;
        //  Error happens randomly on first message, so no
        //  point getting more than a handful...
#       define NBR_FRAMES 2
        while (!s_interrupted && sequence < 10) {
            zmq_msg_t msg [NBR_FRAMES];
            int64_t more;
            size_t more_size = sizeof (more);

            int frame_nbr;
            for (frame_nbr = 0; frame_nbr < NBR_FRAMES; frame_nbr++) {
                zmq_msg_init (&msg [frame_nbr]);
                int rc = zmq_recv (socket, &msg [frame_nbr], 0);
                assert (rc == 0);
                zmq_getsockopt (socket, ZMQ_RCVMORE, &more, &more_size);
                int expected = (frame_nbr < NBR_FRAMES - 1)? 1: 0;
                if (more != expected) {
                    printf ("BAD MULTIPART frame=%d more=%d expect=%d sequence=%d\n",
                            frame_nbr, (int) more, (int) expected, sequence);

                    //  Dump the actual frames we get
                    int f;
                    for (f = 0; f <= frame_nbr; f++) {
                        size_t size = zmq_msg_size (&msg [f]);
                        char *body = zmq_msg_data (&msg [f]);
                        fprintf (stderr, "[frame %d] [%03zd] ", f, size);
                        int char_nbr;
                        for (char_nbr = 0; char_nbr < size; char_nbr++)
                            fprintf (stderr, "%c", body [char_nbr]);
                        fprintf (stderr, "\n");
                    }
                    exit (1);
                }
            }
            for (frame_nbr = 0; frame_nbr < NBR_FRAMES; frame_nbr++)
                zmq_msg_close (&msg [frame_nbr]);
            sequence++;
        }
        printf ("."); fflush (stdout);
        zmq_close (socket);
        zmq_term (context);
    }
    return 0;
}

server.c

//
//  Pubsub issue - server
//  Lose frames at start of connection
//  To provoke it we send 10k messages in a new session and
//  loop over and over. Each message has 3 parts.
//
#include "zhelpers.h"

int main (void) 
{
    s_catch_signals ();
    while (!s_interrupted) {
        //  Prepare our context and socket
        void *context = zmq_init (1);
        void *socket = zmq_socket (context, ZMQ_PUB);
        zmq_bind (socket, "tcp://*:5556");
        s_sleep (200);

        int sequence = 0;
        while (!s_interrupted && sequence < 10000) {
            zmq_msg_t msg;

            zmq_msg_init_size (&msg, 8);
            sprintf (zmq_msg_data (&msg), "%07d", sequence);
            zmq_send (socket, &msg, ZMQ_SNDMORE);
            zmq_msg_close (&msg);

            zmq_msg_init_size (&msg, 8);
            memcpy (zmq_msg_data (&msg), "BODYDATA", 8);
            zmq_send (socket, &msg, 0);
            zmq_msg_close (&msg);
        }
        zmq_close (socket);
        zmq_term (context);
    }
    return 0;
}

zhelpers.h

/*  =========================================================================
    zhelpers.h

    Helper header file for example applications.

    -------------------------------------------------------------------------
    Copyright (c) 1991-2011 iMatix Corporation <www.imatix.com>
    Copyright other contributors as noted in the AUTHORS file.

    This file is part of the ZeroMQ Guide: http://zguide.zeromq.org

    This 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.

    This software is distributed in the hope that it will be useful, but
    WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABIL-
    ITY 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/>.
    =========================================================================
*/

#ifndef __ZHELPERS_H_INCLUDED__
#define __ZHELPERS_H_INCLUDED__

//  Include a bunch of headers that we will need in the examples

#include <zmq.h>

#include <pthread.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <stdarg.h>
#include <string.h>
#include <sys/time.h>
#include <time.h>
#include <unistd.h>
#include <assert.h>
#include <inttypes.h>
#include <signal.h>

//  Bring Windows MSVC up to C99 scratch
#if (defined (__WINDOWS__))
    typedef unsigned long ulong;
    typedef unsigned int  uint;
    typedef __int64 int64_t;
#elif (defined (__APPLE__))
    typedef unsigned long ulong;
    typedef unsigned int  uint;
#endif

//  Provide random number from 0..(num-1)
#define randof(num)   (int) ((float) (num) * random () / (RAND_MAX + 1.0))

//  Receive 0MQ string from socket and convert into C string
//  Caller must free returned string. Returns NULL if the context
//  is being terminated.
static char *
s_recv (void *socket) {
    zmq_msg_t message;
    zmq_msg_init (&message);
    if (zmq_recv (socket, &message, 0))
        return (NULL);
    int size = zmq_msg_size (&message);
    char *string = malloc (size + 1);
    memcpy (string, zmq_msg_data (&message), size);
    zmq_msg_close (&message);
    string [size] = 0;
    return (string);
}

//  Convert C string to 0MQ string and send to socket
static int
s_send (void *socket, char *string) {
    int rc;
    zmq_msg_t message;
    zmq_msg_init_size (&message, strlen (string));
    memcpy (zmq_msg_data (&message), string, strlen (string));
    rc = zmq_send (socket, &message, 0);
    zmq_msg_close (&message);
    return (rc);
}

//  Sends string as 0MQ string, as multipart non-terminal
static int
s_sendmore (void *socket, char *string) {
    int rc;
    zmq_msg_t message;
    zmq_msg_init_size (&message, strlen (string));
    memcpy (zmq_msg_data (&message), string, strlen (string));
    rc = zmq_send (socket, &message, ZMQ_SNDMORE);
    zmq_msg_close (&message);
    return (rc);
}

//  Receives all message parts from socket, prints neatly
//
static void
s_dump (void *socket)
{
    puts ("----------------------------------------");
    while (1) {
        //  Process all parts of the message
        zmq_msg_t message;
        zmq_msg_init (&message);
        zmq_recv (socket, &message, 0);

        //  Dump the message as text or binary
        char *data = zmq_msg_data (&message);
        int size = zmq_msg_size (&message);
        int is_text = 1;
        int char_nbr;
        for (char_nbr = 0; char_nbr < size; char_nbr++)
            if ((unsigned char) data [char_nbr] < 32
            ||  (unsigned char) data [char_nbr] > 127)
                is_text = 0;

        printf ("[%03d] ", size);
        for (char_nbr = 0; char_nbr < size; char_nbr++) {
            if (is_text)
                printf ("%c", data [char_nbr]);
            else
                printf ("%02X", (unsigned char) data [char_nbr]);
        }
        printf ("\n");

        int64_t more;           //  Multipart detection
        size_t more_size = sizeof (more);
        zmq_getsockopt (socket, ZMQ_RCVMORE, &more, &more_size);
        zmq_msg_close (&message);
        if (!more)
            break;      //  Last message part
    }
}

//  Set simple random printable identity on socket
//
static void
s_set_id (void *socket)
{
    char identity [10];
    sprintf (identity, "%04X-%04X", randof (0x10000), randof (0x10000));
    zmq_setsockopt (socket, ZMQ_IDENTITY, identity, strlen (identity));
}


//  Report 0MQ version number
//
static void
s_version (void)
{
    int major, minor, patch;
    zmq_version (&major, &minor, &patch);
    printf ("Current 0MQ version is %d.%d.%d\n", major, minor, patch);
}

//  Require at least some specified version
static void
s_version_assert (int want_major, int want_minor)
{
    int major, minor, patch;
    zmq_version (&major, &minor, &patch);
    if (major < want_major
    || (major == want_major && minor < want_minor)) {
        printf ("Current 0MQ version is %d.%d\n", major, minor);
        printf ("Application needs at least %d.%d - cannot continue\n",
            want_major, want_minor);
        exit (EXIT_FAILURE);
    }
}

//  Sleep for a number of milliseconds
static void
s_sleep (int msecs)
{
#if (defined (__WINDOWS__))
    Sleep (msecs);
#else
    struct timespec t;
    t.tv_sec  =  msecs / 1000;
    t.tv_nsec = (msecs % 1000) * 1000000;
    nanosleep (&t, NULL);
#endif
}

//  Return current system clock as milliseconds
static int64_t
s_clock (void)
{
#if (defined (__WINDOWS__))
    SYSTEMTIME st;
    GetSystemTime (&st);
    return (int64_t) st.wSecond * 1000 + st.wMilliseconds;
#else
    struct timeval tv;
    gettimeofday (&tv, NULL);
    return (int64_t) (tv.tv_sec * 1000 + tv.tv_usec / 1000);
#endif
}

//  Print formatted string to stdout, prefixed by date/time and
//  terminated with a newline.

static void
s_console (const char *format, ...)
{
    time_t curtime = time (NULL);
    struct tm *loctime = localtime (&curtime);
    char *formatted = malloc (20);
    strftime (formatted, 20, "%y-%m-%d %H:%M:%S ", loctime);
    printf ("%s", formatted);
    free (formatted);

    va_list argptr;
    va_start (argptr, format);
    vprintf (format, argptr);
    va_end (argptr);
    printf ("\n");
}


//  --------------------------------------------------------------------------
//  Signal handling
//
//  Call s_catch_signals() in your application at startup, and then exit your
//  main loop if s_interrupted is ever 1. Works especially well with zmq_poll.

static int s_interrupted = 0;
static void s_signal_handler (int signal_value)
{
    s_interrupted = 1;
}

static void s_catch_signals (void)
{
    struct sigaction action;
    action.sa_handler = s_signal_handler;
    action.sa_flags = 0;
    sigemptyset (&action.sa_mask);
    sigaction (SIGINT, &action, NULL);
    sigaction (SIGTERM, &action, NULL);
}


#endif  //  __ZHELPERS_H_INCLUDED__