client.c

//
//  Hello World client
//  Connects REQ socket to tcp://localhost:5555
//  Sends "Hello" to server, expects "World" back
//

#define ROLE "Client"
#define LOOPS 10000UL
#include "common.h"

#define ZMQ_URL "tcp://localhost:5555"

int32_t count (void) {

  static int i = 0;

  i += 10;

  return i;

}


int main (void)
{
  void *context = zmq_init (1);

  void *requester = zmq_socket (context, ZMQ_REQ);

  zmq_connect (requester, ZMQ_URL);

  zmq_msg_t request, reply;

  msgpack_sbuffer *sbuf;
  sbuf = msgpack_sbuffer_new ();

  msgpack_packer *mp;
  mp = msgpack_packer_new (sbuf, msgpack_sbuffer_write);


  msgpack_zone mempool;
  msgpack_zone_init(&mempool, 2048);

  msgpack_object deserialized;


  int i;

  for (i = 0; i != LOOPS; i++) {

    SLEEP();

    msgpack_sbuffer_clear (sbuf);

    msgpack_pack_array (mp, 2);
    msgpack_pack_int (mp, count());
    msgpack_pack_str (mp, "/i/am/a/clinet/");

    zmq_msg_init_data(&request, sbuf->data, sbuf->size, NULL, sbuf);

#if VERBOSE != 0
    msgpack_unpack (sbuf->data, sbuf->size,
        NULL, &mempool, &deserialized);
#endif

    msg_info (deserialized, "Sending\t\t");

    zmq_send (requester, &request, 0);
    zmq_msg_close (&request);

    zmq_msg_init (&reply);
    zmq_recv (requester, &reply, 0);

#if VERBOSE != 0
    msgpack_unpack (zmq_msg_data(&reply), zmq_msg_size(&reply),
        NULL, &mempool, &deserialized);
#endif

    msg_info (deserialized, "Received Reply\t\t");

    zmq_msg_close (&reply);

  }

  msgpack_zone_destroy(&mempool);
  msgpack_sbuffer_destroy(sbuf);

  zmq_close (requester);
  zmq_term (context);
  return 0;
}

server.c

//
//  Hello World server
//  Binds REP socket to tcp://*:5555
//  Expects "Hello" from client, replies with "World"
//
#define ROLE "Server"
#include "common.h"

#define ZMQ_URL "tcp://*:5555"

int32_t count (void) {

  static int i = -100;

  i += 10;

  return i;

}

int main (void)
{
  void *context = zmq_init (1);

  void *responder = zmq_socket (context, ZMQ_REP);

  zmq_bind (responder, ZMQ_URL);

  zmq_msg_t request, reply;

  msgpack_sbuffer *sbuf;
  sbuf = msgpack_sbuffer_new ();

  msgpack_zone mempool;
  msgpack_zone_init(&mempool, 2048);

  msgpack_packer *mp;
  mp = msgpack_packer_new (sbuf, msgpack_sbuffer_write);

  msgpack_object deserialized;

  while (1) {

    zmq_msg_init (&request);
    zmq_recv (responder, &request, 0);

#if VERBOSE != 0
    msgpack_unpack (zmq_msg_data(&request), zmq_msg_size(&request),
        NULL, &mempool, &deserialized);
#endif

    msg_info (deserialized, "Recieved\t\t");

    zmq_msg_close (&request);

    SLEEP();

    msgpack_pack_array (mp, 2);
    msgpack_pack_int (mp, count());
    msgpack_pack_str (mp, "/i/am/a/server/");

    zmq_msg_init_data(&reply, sbuf->data, sbuf->size, NULL, sbuf);

#if VERBOSE != 0
    msgpack_unpack (sbuf->data, sbuf->size,
        NULL, &mempool, &deserialized);
#endif

    msg_info (deserialized, "Sending Reply\t\t");

    zmq_send (responder, &reply, 0);
    zmq_msg_close (&reply);

  }

  msgpack_zone_destroy(&mempool);
  msgpack_sbuffer_destroy(sbuf);

  zmq_close (responder);
  zmq_term (context);
  return 0;
}