Google ProtoBuff + ZeroMq -- C++

In the last series of posts we demonstrated ZeroMq as a technology that supports 'sockets on steroids', supporting multiple platforms as well as multiple languages.  The examples to-date have been transmitting strings between senders and receivers.  While interesting, to effectively create a distributed heterogeneous system we need to be capable of transmitting meaningful messages, preferably complex data structures rather than just strings.  That's where Google's Protobuff comes into play: http://code.google.com/p/protobuf/

Building off our previously created Ubuntu 12.04 32-bit VM, let's start by installing the additional necessary packages;

$ sudo apt-get install libprotoc-dev

With the developer libraries installed, we can now extend our previous C++ example to transmit a ProtoBuff message.

We'll extend our Makefile to add the necessary libraries and a target (e.g. msgs) to generate the C++ files for the message.

$ cat Makefile 
CC=g++
SRCS=main.cpp Messages.pb.cc
OBJS=$(subst .cpp,.o,$(SRCS))
INCLUDES += -I.
LIBS += -lpthread -lrt -lzmq -lprotobuf
.cpp.o:
 $(CC) -c $<
main: msgs ${OBJS} 
 ${CC} ${CFLAGS} -o $@ ${OBJS} ${LIBS}
msgs:
 ${SH} protoc -I. --cpp_out=. Messages.proto
clean:
 ${RM} ${OBJS} main *.pb.*

Oh, we should take a look at our simple Protobuff message file:

$ cat Messages.proto 
message Person {
  required int32 id=1;
  required string name=2;
}

Finally, our extended main file:

$ cat main.cpp 
#include 
#include 
#include 
#include 
#include 
#include 
#include "Messages.pb.h"
void* ctx=zmq_init(1);
char* EndPoint="tcp://127.0.0.1:8000";
static const int N=100;
static const int BufferSize=128;
void* sender(void*)
{
  printf("(%s:%d) running\n",__FILE__,__LINE__);
  void* pub=zmq_socket(ctx, ZMQ_PUB);
  assert(pub);
  int rc=zmq_bind(pub,EndPoint);
  assert(rc==0);
  Person p;
  p.set_name("fatslowkid");
  p.set_id(01);
  for(int i=0; i
  {
    zmq_msg_t msg;
    std::string S=p.SerializeAsString();
    char* content=(char*)S.c_str();
    int rc=zmq_msg_init_size(&msg, BufferSize);
    assert(rc==0);
    rc=zmq_msg_init_data(&msg, content, strlen(content), 0,0);
    assert(rc==0);
    rc=zmq_send(pub, &msg, 0);
    assert(rc==0);
    ::usleep(100000);
  }
}
void* receiver(void*)
{
  printf("(%s:%d) running\n",__FILE__,__LINE__);
  void* sub=zmq_socket(ctx, ZMQ_SUB);
  assert(sub);
  int rc=zmq_connect(sub,EndPoint);
  assert(rc==0);
  char* filter="";
  rc=zmq_setsockopt(sub, ZMQ_SUBSCRIBE, filter, strlen(filter));
  assert(rc==0);
  for(int i=0; i
  {
    zmq_msg_t msg;
    zmq_msg_init_size(&msg, BufferSize);
    const int rc=zmq_recv (sub, &msg, 0);
    char* content=(char*)zmq_msg_data(&msg);
    Person p;
    p.ParseFromString(content);
    printf("(%s:%d) received: '%s'\n",__FILE__,__LINE__,p.name().c_str());
    zmq_msg_close(&msg);
  }
}
int main(int argc, char* argv[])
{
  printf("(%s:%d) main process initializing\n",__FILE__,__LINE__);
  int major, minor, patch;
  zmq_version (&major, &minor, &patch);
  printf("(%s:%d) zmq version: %d.%d.%d\n",__FILE__,__LINE__,major,minor,patch);
  pthread_t rId;
  pthread_create(&rId, 0, receiver, 0);
  pthread_t sId;
  pthread_create(&sId, 0, sender, 0);
  pthread_join(rId,0);
  pthread_join(sId,0);
  printf("(%s:%d) main process terminating\n",__FILE__,__LINE__);
}

Notice that we now transmit and receive Protobuf messages, serialized as strings.  The value of this is that the serialization mechanism is multi-platform & multi-language support.

Cheers.