Apache Zookeeper examples for distributed configuration service, distributed registry, distributed coordination etc

Apache Zookeeper is an excellent piece of software that is really helpful for achieving the below stuffs in distributed systems:

• coordination 
• synchronization
• lock service
• configuration service
• naming registry

etc. etc.
Zookeeper supports high availability by running multiple instances of the service. In case of one of the server that the clients are connecting to goes down, then it will switch over another server transparently. Great thing is irrespective of the server a client may connect to, it will see the updates in the same order. It is a high performance system and can be used for large distributed systems as well.
Zookeeper lets clients coordinate by a shared hierarchical namespace. It is a tree like structure as in normal file system. A client will be connected to a single zookeeper server (as long as the server is accessible).

Please go through Apache Zookeeper getting started guide for how to build, configure zookeeper first.
All these examples are available  in github.

Below is an example (zoo_create_node.c) for how to create Znodes (nodes in the zookeeper database).
In this example we demonstrate a simple way to create nodes in the zookeeper
server. Twenty nodes will be created with path /testpath0, /testpath1,     
/testpath2, /testpath3, ....., /testpath19.                                
All these nodes will be initialized to the same value "myvalue1".          
We will use zookeeper synchronus API to create the nodes. As soon as our   
client enters connected state, we start creating the nodes.                

All the examples used latest stable version of zookeeper that is version
3.3.6                                                                  
We may use the zookeeper client program to get the nodes and examine their
contents.                                                                
Suppose you have downloaded and extracted zookeeper to a directory       
/home/yourdir/packages/zookeeper-3.3.6 . Then after you build the C libraries
,they will be available at /home/yourdir/packages/zookeeper-3.3.6/src/c/.libs/
and the c command line tool will available at                                
/home/yourdir/packages/zookeeper-3.3.6/src/c. The command line tools are cli_st
and cli_mt and cli. They are convenient tool to examine zookeeper data.       

Compile the below code as shown below:
$gcc -o testzk1 zoo_create_node.c -I \
/home/yourdir/packages/zookeeper-3.3.6/src/c/include -I \
/home/yourdir/packages/zookeeper-3.3.6/src/c/generated -L \
/home/yourdir/packages/zookeeper-3.3.6/src/c/.libs/ -lzookeeper_mt
 *                                                               
Make sure that your LD_LIBRARY_PATH includes the zookeeper C libraries to run
the example. Before you run the example, you have to configure and run the  
zookeeper server. Please go through the zookeeper wiki how to do that.      
Now you run the example as shown below:                                     
./testzk1 127.0.0.1:22181  # Assuming zookeeper server is listening on port 
22181 and IP 127.0.0.1                                                      
Now use one of the cli tools to examine the znodes created and also their   
values.

#include <unistd.h>
#include <sys/select.h>
#include <sys/time.h>  
#include <time.h>      
#include <stdlib.h>    
#include <stdio.h>     
#include <string.h>    
#include <errno.h>     

#include <zookeeper.h>
static const char *hostPort;
static zhandle_t *zh;       
static clientid_t myid;     
static int connected;       

void watcher(zhandle_t *zzh, int type, int state, const char *path,
             void* context)                                        
{                                                                  
    if (type == ZOO_SESSION_EVENT) {                               
        if (state == ZOO_CONNECTED_STATE) {                        
            connected = 1;                                         
        } else if (state == ZOO_AUTH_FAILED_STATE) {               
            zookeeper_close(zzh);                                  
            exit(1);                                               
        } else if (state == ZOO_EXPIRED_SESSION_STATE) {           
            zookeeper_close(zzh);                                  
            exit(1);                                               
        }                                                          
    }                                                              
}                                                                  


int main(int argc, char *argv[])
{                               
    int rc;                     
    int fd;                     
    int interest;               
    int events;                 
    struct timeval tv;          
    fd_set rfds, wfds, efds;    

    if (argc != 2) {
        fprintf(stderr, "USAGE: %s host:port\n", argv[0]);
        exit(1);                                          
    }                                                     

    FD_ZERO(&rfds);
    FD_ZERO(&wfds);
    FD_ZERO(&efds);

    zoo_set_debug_level(ZOO_LOG_LEVEL_INFO);
    zoo_deterministic_conn_order(1);        
    hostPort = argv[1];                     
    int x = 0;                              
    zh = zookeeper_init(hostPort, watcher, 30000, &myid, 0, 0);
    if (!zh) {                                                 
        return errno;                                          
    }                                                          
    while (1) {                                                
        char mypath[255];                                      
        zookeeper_interest(zh, &fd, &interest, &tv);           
        usleep(10);                                            
        memset(mypath, 0, 255);                                
        if (connected) {                                       
            while (x < 20) {                                   
                sprintf(mypath, "/testpath%d", x);             
                usleep(10);                                    
                rc = zoo_create(zh, mypath, "myvalue1", 9, &ZOO_OPEN_ACL_UNSAFE, 0, 0, 0);
                if (rc){                                                                  
                    printf("Problems %s %d\n", mypath, rc);                               
                }                                                                         
                x++;                                                                      
            }                                                                             
            connected++;
        }
        if (fd != -1) {
            if (interest&ZOOKEEPER_READ) {
                FD_SET(fd, &rfds);
            } else {
                FD_CLR(fd, &rfds);
            }
            if (interest&ZOOKEEPER_WRITE) {
                FD_SET(fd, &wfds);
            } else {
                FD_CLR(fd, &wfds);
            }
        } else {
            fd = 0;
        }
        FD_SET(0, &rfds);
        rc = select(fd+1, &rfds, &wfds, &efds, &tv);
        events = 0;
        if (rc > 0) {
            if (FD_ISSET(fd, &rfds)) {
                    events |= ZOOKEEPER_READ;
            }
            if (FD_ISSET(fd, &wfds)) {
                events |= ZOOKEEPER_WRITE;
            }
        }
        zookeeper_process(zh, events);
        if (2 == connected ) {
            // We created the nodes, so we will exit now
            zookeeper_close(zh);
            break;
        }
    }
    return 0;
}

In the below example (zoo_get_node.c) we demonstrate a simple way to get nodes in the zookeeper
server. Twenty nodes with path /testpath0, /testpath1,                       
/testpath2, /testpath3, ....., /testpath19 will be examined and value        
associated with them will be printed.                                        
We will use zookeeper synchronus API to get the value of the nodes. As soon  
as our client enters connected state, we start getting the node values.      

Compile the code as shown below:
$gcc -o testzk1 zoo_get_node.c -I \
/home/yourdir/packages/zookeeper-3.3.6/src/c/include -I \
/home/yourdir/packages/zookeeper-3.3.6/src/c/generated -L \
/home/yourdir/packages/zookeeper-3.3.6/src/c/.libs/ -lzookeeper_mt
 *                                                               
Now you run the example as shown below:                          
./testzk1 127.0.0.1:22181  # Assuming zookeeper server is listening on port
22181 and IP 127.0.0.1                                                    

#include <unistd.h>
#include <sys/select.h>
#include <sys/time.h>  
#include <time.h>      
#include <stdlib.h>    
#include <stdio.h>     
#include <string.h>    
#include <errno.h>     

#include <zookeeper.h>
static const char *hostPort;
static zhandle_t *zh;       
static clientid_t myid;     
static int connected;       

void watcher(zhandle_t *zzh, int type, int state, const char *path,
             void* context)                                        
{                                                                  
    if (type == ZOO_SESSION_EVENT) {                               
        if (state == ZOO_CONNECTED_STATE) {                        
            connected = 1;                                         
        } else if (state == ZOO_AUTH_FAILED_STATE) {               
            zookeeper_close(zzh);                                  
            exit(1);                                               
        } else if (state == ZOO_EXPIRED_SESSION_STATE) {           
            zookeeper_close(zzh);                                  
            exit(1);                                               
        }                                                          
    }                                                              
}                                                                  

void watcherforwget(zhandle_t *zzh, int type, int state, const char *path,
             void* context)                                               
{                                                                         
    char *p = (char *)context;                                            
    if (type == ZOO_SESSION_EVENT) {                                      
        if (state == ZOO_CONNECTED_STATE) {                               
            connected = 1;                                                
        } else if (state == ZOO_AUTH_FAILED_STATE) {                      
            zookeeper_close(zzh);                                         
            exit(1);                                                      
        } else if (state == ZOO_EXPIRED_SESSION_STATE) {                  
            zookeeper_close(zzh);                                         
            exit(1);                                                      
        }                                                                 
    }                                                                     
    printf("Watcher context %s\n", p);                                    
}                                                                         

int main(int argc, char *argv[])
{                               
    int rc;                     
    int fd;                     
    int interest;               
    int events;                 
    struct timeval tv;          
    fd_set rfds, wfds, efds;    

    if (argc != 2) {
        fprintf(stderr, "USAGE: %s host:port\n", argv[0]);
        exit(1);                                          
    }                                                     

    FD_ZERO(&rfds);
    FD_ZERO(&wfds);
    FD_ZERO(&efds);

    zoo_set_debug_level(ZOO_LOG_LEVEL_INFO);
    zoo_deterministic_conn_order(1);        
    hostPort = argv[1];                     
    int x = 0;                              
    zh = zookeeper_init(hostPort, watcher, 30000, &myid, 0, 0);
    if (!zh) {                                                 
        return errno;                                          
    }                                                          
    while (1) {                                                
        char mypath[255];                                      
        char buffer[255];                                      
        struct Stat st;                                        
        zookeeper_interest(zh, &fd, &interest, &tv);           
        usleep(10);                                            
        memset(mypath, 0, 255);                                
        memset(buffer, 0, 255);                                
        if (connected) {                                       
            char mycontext[] = "This is context data for test";
            int len = 254;                                     
            while (x < 20) {                                   
                sprintf(mypath, "/testpath%d", x);             
                usleep(10);                                    
                rc = zoo_wget(zh, mypath, watcherforwget , mycontext, buffer, &len, &st);
                if (ZOK != rc){                                                          
                    printf("Problems %s %d\n", mypath, rc);                              
                } else if (len >= 0) {                                                   
                   buffer[len] = 0;                                                      
                   printf("Path: %s Data: %s\n", mypath, buffer);                        
                }                                                                        
                x++;                                                                     
                len = 254;                                                               
            }                                                                            
            connected++;
        }
        if (fd != -1) {
            if (interest&ZOOKEEPER_READ) {
                FD_SET(fd, &rfds);
            } else {
                FD_CLR(fd, &rfds);
            }
            if (interest&ZOOKEEPER_WRITE) {
                FD_SET(fd, &wfds);
            } else {
                FD_CLR(fd, &wfds);
            }
        } else {
            fd = 0;
        }
        FD_SET(0, &rfds);
        rc = select(fd+1, &rfds, &wfds, &efds, &tv);
        events = 0;
        if (rc > 0) {
            if (FD_ISSET(fd, &rfds)) {
                    events |= ZOOKEEPER_READ;
            }
            if (FD_ISSET(fd, &wfds)) {
                events |= ZOOKEEPER_WRITE;
            }
        }
        zookeeper_process(zh, events);
        if (2 == connected ) {
            // We created the nodes, so we will exit now
            zookeeper_close(zh);
            break;
        }
    }
    return 0;
}

In this example below (zoo_data_watches.c) we demonstrate a simple way to watch nodes in the zookeeper                                          
server. Twenty nodes with path /testpath0, /testpath1, /testpath2, /testpath3, .....,                                      
/testpath19 will be watches for any changes in their values.                                                               
We will use zookeeper synchronus API to watch the nodes. As soon                                                           
as our client enters connected state, we start putting watches for the nodes.                                              

Compile the below code as shown below:
$gcc -o testzk1 zoo_data_watches.c -I \
/home/yourdir/packages/zookeeper-3.3.6/src/c/include -I \
/home/yourdir/packages/zookeeper-3.3.6/src/c/generated -L \
/home/yourdir/packages/zookeeper-3.3.6/src/c/.libs/ -lzookeeper_mt
 *                                                               
Now you run the example as shown below:                          
./testzk1 127.0.0.1:22181  # Assuming zookeeper server is listening on port
22181 and IP 127.0.0.1                                    

#include <unistd.h>
#include <sys/select.h>
#include <sys/time.h>  
#include <time.h>      
#include <stdlib.h>    
#include <stdio.h>     
#include <string.h>    
#include <errno.h>     

#include <zookeeper.h>
static const char *hostPort;
static zhandle_t *zh;       
static clientid_t myid;     
static int connected;       
static char mycontext[] = "This is context data for test";

void watcher(zhandle_t *zzh, int type, int state, const char *path,
             void* context)                                        
{                                                                  
    if (type == ZOO_SESSION_EVENT) {                               
        if (state == ZOO_CONNECTED_STATE) {                        
            connected = 1;                                         
        } else if (state == ZOO_AUTH_FAILED_STATE) {               
            zookeeper_close(zzh);                                  
            exit(1);                                               
        } else if (state == ZOO_EXPIRED_SESSION_STATE) {           
            zookeeper_close(zzh);                                  
            exit(1);                                               
        }                                                          
    }                                                              
}                                                                  

void watcherforwget(zhandle_t *zzh, int type, int state, const char *path,
             void* context)                                               
{                                                                         
    char buffer[255];                                                     
    int len, rc;                                                          
    struct Stat st;                                                       
    char *p = (char *)context;                                            
    if (type == ZOO_SESSION_EVENT) {                                      
        if (state == ZOO_CONNECTED_STATE) {                               
            return;                                                       
        } else if (state == ZOO_AUTH_FAILED_STATE) {                      
            zookeeper_close(zzh);                                         
            exit(1);                                                      
        } else if (state == ZOO_EXPIRED_SESSION_STATE) {                  
            zookeeper_close(zzh);                                         
            exit(1);                                                      
        }                                                                 
    } else if (type == ZOO_CHANGED_EVENT) {                               
        printf("Data changed for %s \n", path);                           
        len = 254;                                                        
        //get the changed data and set an watch again                     
        rc = zoo_wget(zh, path, watcherforwget , mycontext, buffer, &len, &st);
        if (ZOK != rc){                                                        
            printf("Problems %s %d\n", path, rc);                              
        } else if (len >= 0) {                                                 
           buffer[len] = 0;                                                    
           printf("Path: %s changed data: %s\n", path, buffer);                
        }                                                                      
    }                                                                          

    printf("Watcher context %s\n", p);
}                                     

int main(int argc, char *argv[])
{                               
    int rc;                     
    int fd;                     
    int interest;               
    int events;                 
    struct timeval tv;          
    fd_set rfds, wfds, efds;    

    if (argc != 2) {
        fprintf(stderr, "USAGE: %s host:port\n", argv[0]);
        exit(1);                                          
    }                                                     

    FD_ZERO(&rfds);
    FD_ZERO(&wfds);
    FD_ZERO(&efds);

    zoo_set_debug_level(ZOO_LOG_LEVEL_INFO);
    zoo_deterministic_conn_order(1);        
    hostPort = argv[1];                     
    int x = 0;                              
    zh = zookeeper_init(hostPort, watcher, 30000, &myid, 0, 0);
    if (!zh) {                                                 
        return errno;                                          
    }                                                          
    while (1) {                                                
        char mypath[255];                                      
        char buffer[255];                                      
        struct Stat st;                                        
        zookeeper_interest(zh, &fd, &interest, &tv);           
        usleep(10);                                            
        memset(mypath, 0, 255);                                
        memset(buffer, 0, 255);                                
        if (connected) {                                       
            //Put the watches for the nodes                    
            int len = 254;                                     
            while (x < 20) {                                   
                sprintf(mypath, "/testpath%d", x);             
                usleep(10);                                    
                rc = zoo_wget(zh, mypath, watcherforwget , mycontext, buffer, &len, &st);
                if (ZOK != rc){                                                          
                    printf("Problems %s %d\n", mypath, rc);                              
                } else if (len >= 0) {                                                   
                   buffer[len] = 0;                                                      
                   printf("Path: %s Data: %s\n", mypath, buffer);
                }
                x++;
                len = 254;
            }
            connected++;
        }
        if (fd != -1) {
            if (interest&ZOOKEEPER_READ) {
                FD_SET(fd, &rfds);
            } else {
                FD_CLR(fd, &rfds);
            }
            if (interest&ZOOKEEPER_WRITE) {
                FD_SET(fd, &wfds);
            } else {
                FD_CLR(fd, &wfds);
            }
        } else {
            fd = 0;
        }
        FD_SET(0, &rfds);
        rc = select(fd+1, &rfds, &wfds, &efds, &tv);
        events = 0;
        if (rc > 0) {
            if (FD_ISSET(fd, &rfds)) {
                    events |= ZOOKEEPER_READ;
            }
            if (FD_ISSET(fd, &wfds)) {
                events |= ZOOKEEPER_WRITE;
            }
        }
        zookeeper_process(zh, events);
    }
    return 0;
}

In this example below (zoo_data_watches.c) we demonstrate a simple way to watch the appearance of a node
in the server. It will also watch if the node is deleted after it was created.

In this program we will watch for the appearance and deletion of a node
"/testforappearance". Once we create the node from another program, the watch
event will be sent to the client and the watcher routine woll be called.    
 *                                                                          

Compile the below code as shown below:
$gcc -o testzk1 zoo_exist_watch.c -I \
/home/yourdir/packages/zookeeper-3.3.6/src/c/include -I \
/home/yourdir/packages/zookeeper-3.3.6/src/c/generated -L \
/home/yourdir/packages/zookeeper-3.3.6/src/c/.libs/ -lzookeeper_mt
 *                                                               
./testzk1 127.0.0.1:22181  # Assuming zookeeper server is listening on port 22181 and
IP 127.0.0.1                                                                         

#include <unistd.h>
#include <sys/select.h>
#include <sys/time.h> 
#include <time.h>     
#include <stdlib.h>   
#include <stdio.h>    
#include <string.h>   
#include <errno.h>    

#include <zookeeper.h>
static const char *hostPort;
static zhandle_t *zh;      
static clientid_t myid;    
static int connected;      
static char mycontext[] = "This is context data for test";

void watcher(zhandle_t *zzh, int type, int state, const char *path,
             void* context)                                       
{                                                                 
    if (type == ZOO_SESSION_EVENT) {                              
        if (state == ZOO_CONNECTED_STATE) {                       
            connected = 1;                                        
        } else if (state == ZOO_AUTH_FAILED_STATE) {              
            zookeeper_close(zzh);                                 
            exit(1);                                              
        } else if (state == ZOO_EXPIRED_SESSION_STATE) {          
            zookeeper_close(zzh);                                 
            exit(1);                                              
        }                                                         
    }                                                             
}                                                                 

void watchexistence(zhandle_t *zzh, int type, int state, const char *path,
             void* context)                                              
{                                                                        
    static struct Stat st;                                               
    int rc;                                                              

    if (type == ZOO_SESSION_EVENT) {
        if (state == ZOO_CONNECTED_STATE) {
            return;                       
        } else if (state == ZOO_AUTH_FAILED_STATE) {
            zookeeper_close(zzh);                  
            exit(1);                               
        } else if (state == ZOO_EXPIRED_SESSION_STATE) {
            zookeeper_close(zzh);                      
            exit(1);                                   
        }                                              
    } else if (type == ZOO_CREATED_EVENT) {            
        printf("Node appeared %s, now Let us watch for its delete \n", path);
        rc = zoo_wexists(zh, path,                                          
                watchexistence , mycontext, &st);                           
        if (ZOK != rc){                                                     
            printf("Problems  %d\n", rc);                                   
        }                                                                   
    } else if (type == ZOO_DELETED_EVENT) {                                 
        printf("Node deleted %s, now Let us watch for its creation \n", path);
        rc = zoo_wexists(zh, path,                                           
                watchexistence , mycontext, &st);                            
        if (ZOK != rc){                                                      
            printf("Problems  %d\n", rc);                                    
        }                                                                    
    }                                                                        
}                                                                            

int main(int argc, char *argv[])
{                              
    int rc;                    
    int fd;                    
    int interest;              
    int events;                
    struct timeval tv;         
    fd_set rfds, wfds, efds;   

    if (argc != 2) {
        fprintf(stderr, "USAGE: %s host:port\n", argv[0]);
        exit(1);                                         
    }                                                    

    FD_ZERO(&rfds);
    FD_ZERO(&wfds);
    FD_ZERO(&efds);

    zoo_set_debug_level(ZOO_LOG_LEVEL_INFO);
    zoo_deterministic_conn_order(1);       
    hostPort = argv[1];                    

    zh = zookeeper_init(hostPort, watcher, 30000, &myid, 0, 0);
    if (!zh) {                                                
        return errno;                                         
    }                                                         

    while (1) {
        static struct Stat st;

        zookeeper_interest(zh, &fd, &interest, &tv);
        usleep(10);                                
        if (connected == 1) {                      
            // watch existence of the node         
            usleep(10);                            
            rc = zoo_wexists(zh, "/testforappearance",
                    watchexistence , mycontext, &st);
            if (ZOK != rc){
                printf("Problems  %d\n", rc);
            }
            connected++;
        }
        if (fd != -1) {
            if (interest & ZOOKEEPER_READ) {
                FD_SET(fd, &rfds);
            } else {
                FD_CLR(fd, &rfds);
            }
            if (interest & ZOOKEEPER_WRITE) {
                FD_SET(fd, &wfds);
            } else {
                FD_CLR(fd, &wfds);
            }
        } else {
            fd = 0;
        }
        FD_SET(0, &rfds);
        rc = select(fd+1, &rfds, &wfds, &efds, &tv);
        events = 0;
        if (rc > 0) {
            if (FD_ISSET(fd, &rfds)) {
                    events |= ZOOKEEPER_READ;
            }
            if (FD_ISSET(fd, &wfds)) {
                events |= ZOOKEEPER_WRITE;
            }
        }
        zookeeper_process(zh, events);
    }
    return 0;
}

In this example below(zoo_children_watch.c) we demonstrate a simple way to monitot the appearance of children znode in a path
in the server. We will check znode /testpath1 for its children and will examine if any children is
added or deleted under this path.

Compile the below code as shown below:
$gcc -o testzk1 zoo_childten_watch.c -I \
/home/yourdir/packages/zookeeper-3.3.6/src/c/include -I \
/home/yourdir/packages/zookeeper-3.3.6/src/c/generated -L \
/home/yourdir/packages/zookeeper-3.3.6/src/c/.libs/ -lzookeeper_mt
*
Make sure that your LD_LIBRARY_PATH includes the zookeeper C libraries to run the example. Before
you run the example, you have to configure and run the zookeeper server. Please go through the
zookeeper wiki how to do that. Now you run the example as shown below:
./testzk1 127.0.0.1:22181 # Assuming zookeeper server is listening on port 22181 and IP 127.0.0.1

#include <unistd.h>
#include <sys/select.h>
#include <sys/time.h>  
#include <time.h>      
#include <stdlib.h>    
#include <stdio.h>     
#include <string.h>    
#include <errno.h>     

#include <zookeeper.h>
static const char *hostPort;
static zhandle_t *zh;       
static clientid_t myid;     
static int connected;       
static char mycontext[] = "This is context data for test";

void watcher(zhandle_t *zzh, int type, int state, const char *path,
             void* context)                                        
{                                                                  
    if (type == ZOO_SESSION_EVENT) {                               
        if (state == ZOO_CONNECTED_STATE) {                        
            connected = 1;                                         
        } else if (state == ZOO_AUTH_FAILED_STATE) {               
            zookeeper_close(zzh);                                  
            exit(1);                                               
        } else if (state == ZOO_EXPIRED_SESSION_STATE) {           
            zookeeper_close(zzh);                                  
            exit(1);                                               
        }                                                          
    }                                                              
}                                                                  

void watchchildren(zhandle_t *zzh, int type, int state, const char *path,
             void* context)                                              
{                                                                        
    struct String_vector str;                                            
    int rc;                                                              

    printf("The event path %s, event type %d\n", path, type);
    if (type == ZOO_SESSION_EVENT) {                         
        if (state == ZOO_CONNECTED_STATE) {                  
            return;                                          
        } else if (state == ZOO_AUTH_FAILED_STATE) {         
            zookeeper_close(zzh);                            
            exit(1);                                         
        } else if (state == ZOO_EXPIRED_SESSION_STATE) {     
            zookeeper_close(zzh);                            
            exit(1);                                         
        }                                                    
    }                                                        
    // Put the watch again                                   
    rc = zoo_wget_children(zh, "/testpath1",                 
            watchchildren , mycontext, &str);                
    if (ZOK != rc){                                          
        printf("Problems  %d\n", rc);                        
    } else {                                                 
        int i = 0;                                           
        while (i < str.count) {                              
            printf("Children %s\n", str.data[i++]);          
        }                                                    
        if (str.count) {                                     
            deallocate_String_vector(&str);                  
        }                                                    
    }                                                        
}                                                            

int main(int argc, char *argv[])
{                               
    int rc;                     
    int fd;                     
    int interest;               
    int events;                 
    struct timeval tv;          
    fd_set rfds, wfds, efds;    

    if (argc != 2) {
        fprintf(stderr, "USAGE: %s host:port\n", argv[0]);
        exit(1);                                          
    }                                                     

    FD_ZERO(&rfds);
    FD_ZERO(&wfds);
    FD_ZERO(&efds);

    zoo_set_debug_level(ZOO_LOG_LEVEL_INFO);
    zoo_deterministic_conn_order(1);        
    hostPort = argv[1];                     

    zh = zookeeper_init(hostPort, watcher, 30000, &myid, 0, 0);
    if (!zh) {                                                 
        return errno;                                          
    }                                                          

    while (1) {
        zookeeper_interest(zh, &fd, &interest, &tv);
        usleep(10);                                 
        if (connected == 1) {                       
            struct String_vector str;               

            usleep(10);
            // watch existence of the node
            rc = zoo_wget_children(zh, "/testpath1", 
                    watchchildren , mycontext, &str);
            if (ZOK != rc){                          
                printf("Problems  %d\n", rc);        
            } else {                                 
                int i = 0;                           
                while (i < str.count) {              
                    printf("Children %s\n", str.data[i++]);
                }
                if (str.count) {
                    deallocate_String_vector(&str);
                }
            }
            connected++;
        }
        if (fd != -1) {
            if (interest & ZOOKEEPER_READ) {
                FD_SET(fd, &rfds);
            } else {
                FD_CLR(fd, &rfds);
            }
            if (interest & ZOOKEEPER_WRITE) {
                FD_SET(fd, &wfds);
            } else {
                FD_CLR(fd, &wfds);
            }
        } else {
            fd = 0;
        }
        FD_SET(0, &rfds);
        rc = select(fd+1, &rfds, &wfds, &efds, &tv);
        events = 0;
        if (rc > 0) {
            if (FD_ISSET(fd, &rfds)) {
                    events |= ZOOKEEPER_READ;
            }
            if (FD_ISSET(fd, &wfds)) {
                events |= ZOOKEEPER_WRITE;
            }
        }
        zookeeper_process(zh, events);
    }
    return 0;
}

Print all the permutations of the characters in a word

A interesting computer programming problem is to print the all permutations of a the characters in a word.

How can we do that? Below is a simple approach:

Suppose your word has 'n' chars,
then first letter of your word can have n options,
second n -1 options
third n -2 options
and so on....
last will have just one option.
Trick is to add the options successively to the prefix, and sending
the suffix (i.e.) the options for remaining positions to function
and call the function for printing permutations recursively.


Below is the code for printing permutations of a word. Here I am
not taking into account the case of repeated letters. Which results
in printing duplicates. More thoughts for avoiding printing duplicates
efficiently:). I don't want to use a set to store the words and printing them
at the end. I want an algorithmic way out to solve the issue. Also, in this example the maximum word size is taken as 26, but that can be easily changed; just replace 26 with a bigger number:)

// Below is the code for printing permutations of an word. Here I am
// not taking into account the case of repeated letters. Which results
// in printing duplicates. More thoughts for avoiding printing duplicates
// efficiently. I don't want to use a store the words and printing them
// at them at the end. I want an algorithmic way out to solve the issue
//
// How does it work ?
// Suppose your whord has 'n' chars,
// then first letter of your word can have n options,
// second n -1 options
// third n -2 options
// and so on....
// last will have just one option.
// Trick is to add the options successively to the prefix, and sending
// the suffix (i.e.) the options for remaining positions to function
// and call the function print_permutations_word repeatedly
//

#include <stdio.h>
#include <string.h>

int print_permutations_word(const char *prefix, const char *suffix, int suffixlen)
{
    char myprefix[26] = "";
    char mysuffix[26] = "";
    int i = 0, j = 0, k = 0;
    if (suffixlen == 0) {
        printf("word %s\n", prefix);
        return 0;
    }

    while (i < suffixlen) {
        memset(myprefix, 0, 26);
        memset(mysuffix, 0, 26);
        snprintf(myprefix,26,"%s%c", prefix,suffix[i]);
        j = 0;
        k = 0;
        while (j < suffixlen) {
           if (i != j){
                mysuffix[k++] = suffix[j];
           }
           j++;
        }
        i++;
        print_permutations_word(myprefix, mysuffix, suffixlen - 1);
    }
    return 0;

}

#if 0
//example run
int main()
{
print_permutations_word("", "abcde", 5);
return 0;
}
#endif

Next approach describes how we can print all permutations of the characters using iterative method based on the principle of getting the next permutation. The advantage of this approach is that we don't print duplicates and efficient for longer strings. We start with "smallest" word and successively print the next bigger word stopping at the "biggest" word. Heart of this approach is in getting the next permutation based on the current permutation.

# Given a word(current permutation), how do we find the next permutation?
# ------------------------------
# Start from last char and successively compare a char to the character in its
# left. If the left character is smaller than the current character (say the
# position of the left character is exchange position), exchange the left
# charecter with a charecter to its right which is bigger than it and
# nearest to it. When the exchange happens, sort the character array to the
# right of the exchange position.
# Repeat the above process until the iteration when no exchange of characters
# happens (i.e. it reaches the "biggest" word)
Below example describes the approach:

Current permutation “aerqpdcb”, it is represented in the array below:

a

e

r

q

p

d

c

b

Start with rightmost char (b) and proced left to find the first character which is smaller than its right character.

So, we reached 'e' (exchange position is 1).

On the right side of e, we find p is the character which is bigger than e and nearerst to it.

Exchange, p and e and the array now becomes:

a

p

r

q

e

d

c

b

Now, sort the array to the right of exchange position and the array becomes:

a

p

b

c

d

e

q

r

So, next permutation after “aerqpdcb” is “apbcdeqr”


Below is the Python program (also available at github )  that
implements the approach:

# Below function demonstrates how we can print the permutations of the letters
# in a word

###############################################################################
# HOW it works?
# it starts with the "smallest" possible word and successively prints the next
# bigger word. For example, the smallest possible word from the chars in 
# word 'axprq' is 'apqrx' and then the next bigger word is 'apqxr' and so on.
# It stops after printing the "biggest" word which is 'xrqpa'.
#
# The advantage of this algorithm is that we don't print any duplicate words.
#
#
# Given a word(current permutation), how do we find the next permutation?
# ------------------------------ 
# Start from last char and successively compare a char to the character in its
# left. If the left character is smaller than the current character (say the 
# position of the left character is exchange position), exchange the left 
# charecter with a charecter to its right which is bigger than it and
# nearest to it. When the exchange happens, sort the character array to the 
# right of the exchange position.
# Repeat the above process until the iteration when no exchange of characters
# happens (i.e. it reaches the "biggest" word)

def print_permute(word):
    a = sorted(word)
    l = len(word)
    exchanged = True
    while exchanged:
        print ''.join(a)
        i = l - 1
        exchanged = False
        while i != 0:
            if a[i] <= a[i - 1]:
                i -= 1
            else:
                exchanged = True
                j = i + 1
                while j < l:
                    if a[i - 1] < a[j]:
                        j += 1 
                    else:
                        break
                j -= 1
                a[i-1], a[j] = a[j], a[i-1]
                a[i:] = sorted(a[i:])
                break

#Example Run
print ('Printing permutations of xyz')
print_permute('xyz')
print('.............................')
print('.............................')
print ('Printing permutations of abcdd')
print_permute('abcdd')
print('.............................')
print('.............................')
print ('Printing permutations of axprq')
print_permute('axprq')

Apache Thrift File based transport

Today I will demonstrate how we can use Apache Thrift FileTransport. Thrift file based transport is useful when we want to store the messages in some local files if the remote server is down. We may replay the files when the remote server comes up again. If we want to implement a Thrift based messaging system, then also this feature will be useful for "store and forwarding message" approach.

exception BadOperation {
1: i32 what,
2: optional string reason
}
enum Operation {
CREATE,
DELETE,
FILECONTENT,
DIRCONTENT
}
struct Work {
1: Operation op
2: string filename
3: string data
4: string rootdir
}
service FileService {
i32 createFile(1:Work w) throws (1:BadOperation badop),
list<string> getFiles(1:Work w) throws (1:BadOperation badop)
}

Please refer my previous post or google for how to use the thrift compiler to generate code for handling thrift messages following the above definitions.My example will be in C++, but a PHP, Java or Python example would look very much similar.The generated code has a file FileService.cpp. Examining the code will show that createFileinterface calls two member functions of FileServiceClient class. They aresend_createFile and recv_createFile. send_createFile actually serialize our data,and send them over the socket in case of a typical client-server scenario.recv_createFile processes the response back from the server. In case we are using File based transport to "send" data, then there won't be any response. Hence, we don't need to call recv_createFile function. In stead we just call send_createFile to "send" or write the messages to the file. This is the trick.Same is the case with writing "getFiles" messages to the file as well.Actually we can make this much more efficient by serializing and batching the writes to the file ourselves. But here I am not showing that, as my purpose is to demonstrate the basic operations of File based transport.
Below I have pasted my code for "writing" messages to the file (file_writer.cpp)

#include <stdlib.h>
#include <time.h>  
#include <iostream>
#include <sstream> 
#include <protocol/TBinaryProtocol.h>
#include <transport/TSocket.h>       
#include <transport/TTransportUtils.h>

#include "FileService.h"

using std::cout;
using std::endl;
using std::stringstream;
using namespace apache::thrift;
using namespace apache::thrift::protocol;
using namespace apache::thrift::transport;

using namespace FileHandler;

using namespace boost;

int main(int argc, char** argv) {
    shared_ptr<TTransport> file(new TFileTransport("testfiletransport"));
    shared_ptr<TTransport> transport(new TBufferedTransport(file));      
    shared_ptr<TProtocol> protocol(new TBinaryProtocol(transport));      
    FileServiceClient client(protocol);                                  
    try {                                                                
        int i = 0;                                                       
        stringstream ss (stringstream::in | stringstream::out);          
        Work work;
        work.data = "mydata";
        work.rootdir = "/home/nipun/test";

        try {
            while (i++ < 100) {
                work.op = Operation::CREATE;
                ss << "filename" << i ;
                work.filename = ss.str();
                ss.clear();
                ss.str("");
                ss << "data" << rand() << "-" << rand() << time(0);
                work.data = ss.str();
                ss.clear();
                ss.str("");
                client.send_createFile(work);

                work.op = Operation::DIRCONTENT;
                ss << "filename" << i  << rand();
                work.filename = ss.str();
                ss.clear();
                ss.str("");
                ss << "data" << rand() << "-" << rand() << time(0) << rand();
                work.data = ss.str();
                ss.clear();
                ss.str("");
                client.send_getFiles(work);
            }
        } catch (BadOperation &op) {
            cout << "Exception "  <<  op.reason  << endl;
        }

    } catch (TException &tx) {
        cout <<  "ERROR: " << tx.what() << endl;
    }
}

Here we wrote "createFile" and "getFiles" messages 100 times each to the file "testfiletransport" in current directory. We used send_createFiles and send_getFiles routines for the same. Remember that we have to use exact similar type of transport while reading from the file also. TBinary protocol specifies how the data is serialized and TBuffered transport is used to buffer data before they are flushed to underlying transport which is the file "testfiletransport" here.

Now is the time to read the data from the file. Generally we write the data to a file and read it later when want to replay them later. In such cases the messages are read and send over another transport which may be over TSocket (socket based transport class defined in Thrift library). But in this example we will just read the messages and print them to standard out.
 

Code for file_reader.cpp

#include <stdlib.h>
#include <time.h>  
#include <iostream>
#include <sstream> 
#include <string>  
#include <protocol/TBinaryProtocol.h>
#include <transport/TSocket.h>       
#include <transport/TTransportUtils.h>

#include "FileService.h"

using std::cout;
using std::endl;
using std::stringstream;
using std::string;      
using namespace apache::thrift;
using namespace apache::thrift::protocol;
using namespace apache::thrift::transport;

using namespace FileHandler;

using namespace boost;

int main(int argc, char** argv) {
    shared_ptr<TTransport> file(new TFileTransport("testfiletransport", true));
    shared_ptr<TTransport> transport(new TBufferedTransport(file));
    shared_ptr<TProtocol> protocol(new TBinaryProtocol(transport));
    FileServiceClient client(protocol);
    string fname;
    TMessageType mtype;
    int32_t seqid;
    Work w;
    try {
        while (true) {
            protocol->readMessageBegin(fname, mtype, seqid);
            cout << fname << endl;
            if (fname == "createFile") {
                FileService_createFile_args args;
                args.read(protocol.get());
                protocol->readMessageEnd();
                w = args.w;
            } else if (fname == "getFiles") {
                FileService_getFiles_args args;
                args.read(protocol.get());
                protocol->readMessageEnd();
                w = args.w;
            }
            cout <<"\tData:\t" << w.data << endl;
            cout <<"\tFilename:\t" << w.filename << endl;
            cout <<"\tRootdir:\t" << w.rootdir << endl;
            cout <<"\tOperation:\t" << w.op << endl;

        }
    } catch (TTransportException& e) {
        if (e.getType() == TTransportException::END_OF_FILE) {
            cout << "\n\n\t\tRead All Data successfully\n";
        }
    } catch (TException &tx) {
        cout << "ERROR " <<  tx.what() << endl;
    }
}

In the file_reader.cpp example, we first read the message type using readMessageBegin interface of binary protocol.  Then we use "fname" (function name) field to decide if the message is a "createFile" or "getFiles" message. Then we use use appropriate FileService args class to read the actual message.
After that we just print the message on our terminal :)

Hope this will be useful for you and if so please don't forget to leave a comment here :)

In my next post, I will show how we may build a simple messaging system using Apache Thrift. Hopefully you will visit again.

Distributed scalable log aggreation with Scribe

Today I will explain how to configure scribe log server and how to aggregate log from different location to a central location.
Scribe uses boost library and Apache Thrift. Both scribe and Apache Thrift  were developed by Facebook and were open sourced. Latest release of scribe was 3 years ago and hence it may not build successfully with latest g++ compiler. I am using 4.4.4 version of g++ and I am building on Linux (CentOS 6.0 64 bit) platform and hence all my examples are for Linux platform only.
I am using boost library (Boost version 1.46.1). Problem is scribe won't build with this library also and hence we will need to change few cpp files in scribe to build it successfully.

We download and extract Boost archive and go to the directory where it was extracted. Then issue the below commands:

$ sh bootstrap.sh --prefix=/usr/local
$ ./bjam install

It will boost libraries and header files under /usr/local/include and /usr/local/lib.
Then download thrift and build it. Thrift will need libevent libarary.

$rpm -qa | grep libevent-devel

If the above command doesn't return anything, then we execute the command below:

$yum install libevent-devel

We have to do apt-get on debian/ubuntu.

Now extract the thrift archive and build it.
Cd to the directory where you extracted the thrift sources.
$ ./configure
$  make
$  make install

Generally, thrift will install in /usr/local directory. If you had installed maven , it will build and install JAVA jars as well in /usr/local/lib.

It is better to add the below headers in the /usr/local/incude/thrift/Thrift.h. Reason being, I was getting few compilation errors such as uint32_t/int32_t not being recognized; htons,ntohl etc. being reported as unknown functions etc. etc.

#include <stdint.h>
#include <inttypes.h>
#include <arpa/inet.h>

Now thrift is there for us. We built boost already.

Building scribe
Now, it's time to compile scribe. I downloaded scribe-2.2.tar.gz and  I explain building for this version only.
Extract the scribe archive and cd to the directory where you extracted the archive. Now issue the below commands:

$export LD_LIBRARY_PATH=

 In my case both of them were installed in /usr/local/lib. So, I issued the below command:

$export LD_LIBRARY_PATH=/usr/local/lib

$sh bootstrap.sh --with-boost=. In my case boost headers/libraries installed in /usr/local. So, I issued the below command:
$sh bootstrap.sh --with-boost=/usr/local
$make

I got compilation error for conflicting return type for virtual scribe::thrift::ResultCode scribeHandler::Log in scribe_server.h. I solved that by looking at the type declared in scribe.h and following the same for scribe_server.h , i.e., I changed return type of scribeHandler::Log to scribe::thrift::ResultCode::type.

Then I got the below error in another source file file.cpp.
file.cpp:203: error: ‘class boost::filesystem3::directory_entry’ has no member named ‘filename’
This was due to the higher version of boost library I am using. So, I made it compatible by replacing the below line in file.cpp 

 _return.push_back(dir_iter->filename());

with 

_return.push_back(dir_iter->path().filename().string()); 

Till there are more compilation errors :). This time the error is in conn_pool.cpp and it says TRY_LATER and OK are not declared (:
So, we replace occurrences of TRY_LATER with ResultCode::TRY_LATER and OK with ResultCode::OK. Also, we have to replace "ResulCode result" with "ResultCode::type result".

Now the compilation error scribe_server.cpp is resolved by replacing "scribe::thrift::ResultCode scribeHandler::Log" with scribe::thrift::ResultCode::type scribeHandler::Log. Fix for ResultCode related compilation issues are fixed in the same way we did for conn_pool.cpp.
Finally, everything compiled successfully and hence we can install scribe.

$make install

Now we can run scribe! Running scribe is simple because the required configuration file is not complex to understand and the scribe package already have some good examples.

In the directory "if" under the root scribe directory (where we extracted scribe.tar.gz) there is the thrift idl file scribe.thrift. This file describes the Log service and also the structure of the log messages that can be sent to scribe log server.
Below is the content of the scribe.thrift:

include "fb303/if/fb303.thrift"

namespace cpp scribe.thrift

enum ResultCode
{
  OK,
  TRY_LATER
}

struct LogEntry
{
  1:  string category,
  2:  string message
}

service scribe extends fb303.FacebookService
{
  ResultCode Log(1: list messages);
}

This shows that each messages are to be sent via the "Log" routine defined by scribe service and it takes a vector of LogEntry Messages as input. LogEntry message has two fields, category and message and both are strings.  By default scribe logger creates a separate directory for each category of messages (unless we explicitly configure scribe not to do so).  This file also include the file fb303/if/fb303.thrift. This was installed on /usr/local/share/fb303/if/fb303.thrift on my system. So, lets generate the cpp files by using thrift compiler.

$thrift -I /usr/local/share -gen cpp   scribe.thrift

This will create gen-cpp directory.

$ls gen-cpp
scribe_constants.cpp  scribe_constants.h  scribe.cpp  scribe.h  scribe_server.skeleton.cpp  scribe_types.cpp  scribe_types.h

We have all the files except the client executable which will call the Log routine and send the messages to the scribe server(s).
Below is the simple client code (save the code in client.cpp).

#include <stdio.h>            
#include <unistd.h>           
#include <sys/time.h>         

#include <iostream>
#include <sstream> 

#include <protocol/TBinaryProtocol.h>
#include <transport/TSocket.h>       
#include <transport/TTransportUtils.h>

#include "scribe.h"

using std::cout;
using std::endl;
using std::vector;
using boost::shared_ptr;
using std::stringstream;

using namespace apache::thrift;
using namespace apache::thrift::protocol;
using namespace apache::thrift::transport;
using namespace scribe;                   
using namespace scribe::thrift;           


int main(int argc, char** argv) {

    if (argc != 3) {
        cout << "Usage: " << argv[0] << "  host " <<  "port" << endl;
        exit(1);                                                                                   
    }                                                                                              

    shared_ptr<TTransport> socket(new TSocket(argv[1], atoi(argv[2])));
    shared_ptr<TTransport> transport(new TFramedTransport(socket));
    shared_ptr<TProtocol> protocol(new TBinaryProtocol(transport));

    // Cretae a scribe client
    scribeClient client(protocol);
    // Vector of Logentry
    vector<LogEntry> logdata;

    try {
        stringstream ss;
        try {
            transport->open();
            int i = 0;
            LogEntry a ;
            a.category = "MyCat";

            // We will send a 1000 messages to the scribe log server
            while ( i++ < 1000 ) {
                ss << "Hello " << i ;
                a.message = ss.str();
                ss.str("");
                ss.clear();
                logdata.push_back(a);
            }
            client.Log(logdata);
            logdata.clear();
        } catch (...) {
            cout << "Got some exception " << endl;
        }
        transport->close();
    } catch (TException &tex) {
        printf("Exception: %s\n", tex.what());
    }
}

Now we do the following:
1. Compile the client programs as shown below:

$g++ -g  -o scribe_client -I . -I /usr/local/include/thrift -I /usr/local/include/thrift/fb303 client.cpp scribe_types.cpp  scribe_constants.cpp  scribe.cpp     -L /usr/local/lib -lthriftnb -lthrift -levent  /usr/local/lib/libfb303.a
Note that we provided the include directory location, and library locations for thrift,boost and facebook interface library (fb303).
We will run all the central logger, client logger and the client program in the same machine for demonstration.
 
2. Scribe central logger example configuration is present in examples directory of scribe source (example2central.conf). I modified it to add a newline after every log messages by adding "add_newlines=1"  in the section of the file. Log files will be created under some sub-directory in /tmp/scribetest. For our example, the su-directory is /tmp/scribetest/MyCat. We run the scribe central logger as shown below:

$scribed -c example2central.conf

3.  example2client.conf is configured to send logs to the central logger. the client logger listens on 1464 port for log messages and it forwards the messages to the central logger. We run scribe client logger by issuing the below command:

$scribed -c example2client.conf

4. Now we run our client. It connects to the client logger and sends logs to the client logger who is listening on port 1464. We run the client program:
$./scribe_client 127.0.0.1 1464

Now we can see our log messages in files in /tmp/scribetest/MyCat directory.

How do we write a client in java? It is simple. Follow the steps given below:
$thrift -I /usr/local/share -gen java   scribe.thrift
This generates the below files under gen-java:

• LogEntry.java  
• ResultCode.java  
• scribe.java

We have to build libfb303-0.8.0.jar if it is not done already.  Cd to thrift-0.8.0/contrib/fb303/java and issue "ant "  build command. Basically we may need the below jars (which are part of thrift package or generated while building thrift).

• commons-codec-1.4.jar
• commons-lang-2.5.jar
• commons-logging-1.1.1.jar
• httpclient-4.1.2.jar
• httpcore-4.1.3.jar
• junit-4.4.jar
• libfb303-0.8.0.jar
• libthrift-0.8.0.jar
• log4j-1.2.14.jar
• servlet-api-2.5.jar
• slf4j-api-1.5.8.jar
• slf4j-log4j12-1.5.8.jar  

We need to write few lines of code to send messages to scribe logger though which will basically calls the "Log" routine defined in scribe.java.
Below is the code:

import org.apache.thrift.TException;
import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.protocol.TProtocol;      
import org.apache.thrift.transport.TSocket;       
import org.apache.thrift.transport.TTransport;    
import org.apache.thrift.transport.TFramedTransport;
import org.apache.thrift.transport.TTransportException;
import java.util.List;                                 
import java.util.ArrayList;                            

public class Main {

        public static void main(String[] args) {

                if (args.length != 2) {
                        System.out.println(" <Host> <Port> missing");
                        System.exit(1);
                }
                int port = -1;
                try {
                        port = Integer.parseInt(args[1]);
                } catch (NumberFormatException e) {
                        System.exit(1);
                }
                System.out.println(args[0]);
                System.out.println(args[1]);
                TTransport tr = new TFramedTransport(new TSocket(args[0], port));
                TProtocol proto = new TBinaryProtocol(tr);
                scribe.Client client = new scribe.Client(proto);

                int i = 0;
                List<LogEntry> list = new ArrayList<LogEntry>();
                LogEntry log = null;
                while (i < 100){
                        log = new LogEntry();
                        log.setCategory("javamessage");
                        log.setMessage("My Message " + i);
                        list.add(log);
                        i++;
                }
                try {
                        tr.open();
                        client.Log(list);
                } catch (org.apache.thrift.TException e){
                        e.printStackTrace();
                }
        }
}  

Save this in a file Main.java and compile this along with LogEntry.java, scribe.java,
ResultCode.java. While compiling and running, we have to put the jars listed above
in java classpath.

A simple REST framework on C/C++

REpresentational State Transfer (REST) is a software architecture pattern heavily used
on distributed systems, especially for web services.
It is really a client-server architecture. Client sends requests for resources and 
server responds with representation of the resources.In web-based REST system 
individual resources are exposed as URIs. The client and servers communicate using
HTTP protocol. All the HTTP verbs such as GET, DELTE,PUT,POST etc. can be used. The
input for the REST API may come from request headers or parameters. E.g. if an API
requires the client authentication, then it may look for the authentication token by
examining some header or parameter values.

So, for implementing a REST server we need the following bare minimum:
An HTTP server
A parser to parse the API parameters data
An executor which gets the representation of the resource
Generally you produce the REST response in XML or JSON format. But there is no 
restrictions regarding the formats.

So,We need a HTTP library, a JSON/XML parser and of course we have to design and
implement the back-end logic for our APIs. In this example I choose libmicrohttpd and 
Boost Property Tree to generate XML/JSON.

Here is a samll example for how we can create a REST server using available
http library in C/C++.We will create JSON or XML data as API responeses.
All the example code is available on GitHub (click here)           

REST allows all HTTP methods such as GET,POST,PUT,DELETE etc. But for
demonstration purpose we will only use GET method. Also, we will support only
http and not https. In next post, I will demonstrate how to use https for secure
data transfer for this example.                                                 

***PLEASE download and build libmicrohttpd, boost library as we need them to
compile and run this example.***                                            

The examples defines three simple APIs or resources.
sysinfo
diskinfo
procinfo

sysinfo returns few system related information of your Linux box. It optioanlly
takes three flags "cpus", "memory", "os" as the value of parameter "fields", so
that user can select a set of information. All the APIs take another parameter
"type" to select the response format and valid values are JSON or XML.

diskinfo returns few disks related information of your Linux box. It optioanlly
takes two flags "totalparts", "spaceinfo" as the value of parameter "fields", so
that user can select a set of information.

procinfo returns few processes related information about the processes runnining
on your Linux box. It also takes three flags "percentmemory", "percentcpu" as the
values for parameter "fields", so that user can select a set of information.

We need to compile the exmaples as shown below (assuming you built and installed
libmicrohttpd in /usr/local)g** httphandler.cpp  strutil.cpp  api.cpp executor.cpp -I \
g++ -o example_rest_server httphandler.cpp  strutil.cpp  api.cpp executor.cpp -I\
     /usr/local/lib -lmicrohttpd \
     libboost_regex.a

After we build the example, we run it as shown below
$ exectuable_name_for_our_example port_number_where_it_listens
E.g.
$ ./example_rest_server 1234

We access the API as shown below:
http://127.0.0.1:1234/procinfo?type=xml&fields=percentcpu,percentmemory
http://127.0.0.1:1234/procinfo?type=json&fields=percentcpu
http://127.0.0.1:1234/diskinfo
http://127.0.0.1:1234/sysinfo?fields=memory

Point your browser to any of the above locations and see the responses :)

 Below is the code for HTTP request handler (httphandler.cpp)

#include <signal.h>                 
#include <pthread.h>                
#include <platform.h>               
#include <microhttpd.h>             
#include <iostream>                 
#include <map>                      
#include <string>                   

#include <api.hpp>

using std::map;
using std::string;

#define PAGE "<html><head><title>Error</title></head><body>Bad data</body></html>"

static int shouldNotExit = 1;

static int send_bad_response( struct MHD_Connection *connection)
{                                                               
    static char *bad_response = (char *)PAGE;                   
    int bad_response_len = strlen(bad_response);                
    int ret;                                                    
    struct MHD_Response *response;                              

    response = MHD_create_response_from_buffer ( bad_response_len,
                bad_response,MHD_RESPMEM_PERSISTENT);             
    if (response == 0){                                           
        return MHD_NO;                                            
    }                                                             
    ret = MHD_queue_response (connection, MHD_HTTP_OK, response); 
    MHD_destroy_response (response);                              
    return ret;                                                   
}                                                                 


static int get_url_args(void *cls, MHD_ValueKind kind,
                    const char *key , const char* value)
{                                                       
    map<string, string> * url_args = static_cast<map<string, string> *>(cls);

    if (url_args->find(key) == url_args->end()) {
         if (!value)                                   
             (*url_args)[key] = "";                    
         else                                          
            (*url_args)[key] = value;                  
    }                                                  
    return MHD_YES;                                    

}
                
static int url_handler (void *cls,
    struct MHD_Connection *connection,
    const char *url,                  
    const char *method,               
    const char *version,              
    const char *upload_data, size_t *upload_data_size, void **ptr)
{                                                                 
    static int aptr;                                              
    const char *fmt = (const char *)cls;                          
    const char *val;                                              
    char *me;                                                     
    const char *typexml = "xml";                                  
    const char *typejson = "json";                                
    const char *type = typejson;                                  

    struct MHD_Response *response;
    int ret;                      
    map<string, string> url_args;
    map<string, string>:: iterator  it;
    ourapi::api callapi;                     
    string respdata;                         

    // Support only GET for demonstration
    if (0 != strcmp (method, "GET"))     
        return MHD_NO;                   


    if (&aptr != *ptr) {
        *ptr = &aptr;   
        return MHD_YES; 
    }                   


    if (MHD_get_connection_values (connection, MHD_GET_ARGUMENT_KIND, 
                           get_url_args, &url_args) < 0) {         
        return send_bad_response(connection);                         

    }

    callapi.executeAPI(url, url_args, respdata);

    *ptr = 0;                  /* reset when done */
    val = MHD_lookup_connection_value (connection, MHD_GET_ARGUMENT_KIND, "q");
    me = (char *)malloc (respdata.size() + 1);                                 
    if (me == 0)                                                               
        return MHD_NO;                                                         
    strncpy(me, respdata.c_str(), respdata.size() + 1);                        
    response = MHD_create_response_from_buffer (strlen (me), me,               
                                              MHD_RESPMEM_MUST_FREE);          
    if (response == 0){                                                        
        free (me);                                                             
        return MHD_NO;                                                         
    }                                                                          

    it = url_args.find("type");
    if (it != url_args.end() && strcasecmp(it->second.c_str(), "xml") == 0)
        type = typexml;                                                       

    MHD_add_response_header(response, "Content-Type", "text");
    MHD_add_response_header(response, "OurHeader", type);     

    ret = MHD_queue_response (connection, MHD_HTTP_OK, response);
    MHD_destroy_response (response);                             
    return ret;                                                  
}                                                                

void handle_term(int signo)
{                          
    shouldNotExit = 0;     
}                          

void* http(void *arg)
{
    int *port = (int *)arg;
    struct MHD_Daemon *d;

    d = MHD_start_daemon (MHD_USE_SELECT_INTERNALLY | MHD_USE_DEBUG | MHD_USE_POLL,
                        *port,
                        0, 0, &url_handler, (void *)PAGE, MHD_OPTION_END);
    if (d == 0){
        return 0;
    }
    while(shouldNotExit) {
        sleep(1);
    }
    MHD_stop_daemon (d);
    return 0;
}

int main (int argc, char *const *argv)
{

    if (argc != 2){
        printf ("%s PORT\n", argv[0]);
        exit(1);
    }
    daemon(0,0);
    signal(SIGTERM, handle_term);
    int port = atoi(argv[1]);
    pthread_t  thread;
    if ( 0 != pthread_create(&thread, 0 , http, &port)){
        exit(1);
    }
    pthread_join(thread, 0);
    return 0;
}

Below is the code for API parser (api.cpp). 
Here we implement the logic for parsing API parameters, and validating the request,
and calling the appropriate back-end executor routines. 

#include <string.h>         

#include <boost/foreach.hpp>

#include <api.hpp>
#include <strutil.hpp>

using namespace ourapi;

struct validate_data
{                   
    string api;     
    set <string>* params; 
};                              

api::api()
{         
    set<string> params;
    string sysinfoparams[] = {"cpus", "memory", "os"}; 
    string processinfoparams[] = {"percentmemory", "percentcpu" };
    string diskinfoparamas[] = {"totalparts", "spaceinfo" };      

    _apiparams["/sysinfo"] =  set<string>(sysinfoparams, sysinfoparams + 3);
    _apiparams["/procinfo"] = set<string>(processinfoparams, processinfoparams  + 2);
    _apiparams["/diskinfo"] = set<string>(diskinfoparamas, diskinfoparamas + 2);     
}                                                                                          

bool api::executeAPI(const string& url, const map<string, string>& argvals, string& response)
{                                                                                                  
    // Ignore all the args except the "fields" param                                               
    validate_data vdata ;                                                                          
    vdata.api = url;                                                                               
    Executor::outputType type = Executor::TYPE_JSON;                                               
    vector<string> params;                                                                   
    set<string> uniqueparams;                                                                
    map<string,string>::const_iterator it1 = argvals.find("fields");                         

    if (it1 != argvals.end()) {
        string prms = it1->second;
        StrUtil::eraseWhiteSpace(prms);
        StrUtil::splitString(prms, ",", params);   
    }                                              
    BOOST_FOREACH( string pr, params ) {           
        uniqueparams.insert(pr);                   
    }                                              
    vdata.params = &uniqueparams;                  

    if ( !_validate(&vdata)) {
        _getInvalidResponse(response);
        return false;                 
    }                                 

    it1 = argvals.find("type");
    if (it1 != argvals.end()){ 
        const string outputtype = it1->second;
        if (strcasecmp(outputtype.c_str(), "xml") == 0 ) {
            type = Executor::TYPE_XML;                    
        }                                                 
    }                                                     

    return _executeAPI(url, uniqueparams, type, response);
}                                                         

bool api::_executeAPI(const string& url, const set<string>& argvals, 
        Executor::outputType type, string& response)                       
{                                                                          
    bool ret = false;
    if (url == "/sysinfo")
        ret = _executor.sysinfo(argvals, type,  response);
    if (url == "/diskinfo")
        ret = _executor.diskinfo(argvals, type, response);
    if (url == "/procinfo")
        ret = _executor.procinfo(argvals, type, response);

    return ret;
}

bool api::_validate(const void *data)
{
    const validate_data *vdata = static_cast<const validate_data *>(data );
    map<string, set<string> > ::iterator it =  _apiparams.find(vdata->api);

    it = _apiparams.find(vdata->api);

    if ( it == _apiparams.end()){
        return false;
    }
    set<string>::iterator it2 = vdata->params->begin();
    while (it2 != vdata->params->end()) {
        if (it->second.find(*it2) == it->second.end())
            return false;
        ++it2;
    }

    return true;
}

void api::_getInvalidResponse(string& response)
{
    response = "Some error in your data ";
}

Below is the code for API back-end logic (executor.cpp) 
Here we generate the response, i.e., the representation of the resource. 

#include <stdio.h>               
#include <iostream>              
#include <vector>                
#include <sstream>               

#include <stdint.h>
#include <boost/regex.hpp>
#include <boost/format.hpp>
#include <boost/lexical_cast.hpp>
#include <boost/property_tree/ptree.hpp>
#include <boost/property_tree/json_parser.hpp>
#include <boost/property_tree/xml_parser.hpp> 

#include <executor.hpp>
#include <strutil.hpp> 

using namespace ourapi;
using std::vector;     
using boost::property_tree::ptree;
using std::make_pair;             
using boost::lexical_cast;        
using boost::bad_lexical_cast;    
using boost::format;              
using boost::regex_search;        
using boost::match_default;       
using boost::match_results;       
using boost::regex;               


Executor::Executor()
{                   
}                   

bool Executor::diskinfo(const set<string>& args, outputType type, 
        string& response)                                               
{                                                                       
    const char *command = "df | sed 's/ \\+/ /g'  | tail -n +2 ";       
    char line[255];                                                     
    vector<string> tokens;                                        
    int i = 0,j;                                                        
    bool spaceinfo = false;                                             
    bool totalparts = false;                                            
    uint64_t totalspace = 0;                                            
    uint64_t usedspace = 0;                                             
    int32_t partnum = 0;                                                

    FILE *fp = popen(command, "r");
    if (!fp){                      
        return false;              
    }                              
    while (fgets(line, 255, fp) != 0){
        response += string(line);     
    }                                 
    fclose(fp);                       

    if (args.find("spaceinfo") != args.end()) {
        spaceinfo = true;                      
    }                                          
    if (args.find("totalparts") != args.end()) {
        totalparts = true;                      
    }                                           


    StrUtil::splitString( response, " \t\n", tokens); 
                                                      
    j = tokens.size();                                
    ptree diskinforoot ;                              
    ptree diskinfo;                                   

    ptree::iterator  ptit = diskinforoot.push_back(make_pair("diskinfo", diskinfo ));
    ptree::iterator pit ;                                                            
    while (i < j) {                                                               
        {                                                                            
            ptree temp;                                                              
            pit = ptit->second.push_back(make_pair("FileSystem", temp));          
        }                                                                            
        pit->second.push_back(make_pair("Name", tokens[i++]));                    
        try {                                                                        
            if (spaceinfo) {                                                         
                totalspace += lexical_cast<uint64_t>(tokens[i]);               
            }                                                                        
            pit->second.push_back(make_pair("Size", tokens[i++]));                
            usedspace += lexical_cast<uint64_t>(tokens[i]);                    
            pit->second.push_back(make_pair("Used", tokens[i++]));                

        } catch ( bad_lexical_cast& e) {
        }                               
        pit->second.push_back(make_pair("Avail", tokens[i++]));
        pit->second.push_back(make_pair("PercentUse", tokens[i++]));
        pit->second.push_back(make_pair("MountedOn", tokens[i++])); 
        partnum++;                                                     
    }                                                                  

    if (spaceinfo) {
        ptree temp; 
        format fmter("%1%");
        pit = ptit->second.push_back(make_pair("SpaceInfo", temp));
        fmter % totalspace;                                           
        pit->second.push_back(make_pair("TotalSpace", fmter.str()));
        fmter.clear();                                                 
        fmter % usedspace;                                             
        pit->second.push_back(make_pair("UsedSpae", fmter.str()));  
        fmter.clear();                                                 

    }

    if (totalparts) {
        ptree temp;  
        format fmter("%1%");
        fmter % partnum;    
        ptit->second.push_back(make_pair("TotalParts", fmter.str()));
        fmter.clear();                                                  
    }                                                                   

    _generateOutput(&diskinforoot, type, response);
    std::cout << response << std::endl;
    return true;                                   
}                                                  

bool Executor::procinfo(const set<string>& args, outputType type, 
        string& response)                                               
{                                                                       
    const char *command = "ps auxef | tail -n +2 |awk ' { printf \"%s %s %s %s \", $1, $2, $3, $3 ; for (i = 11; i <= NF; i++) {printf \"%s \", $i }  print \"\" }  ' ";                                                                                                                                                                     
    char line[8096];                                                                                                                                                    
    FILE *fp = popen(command, "r");                                                                                                                                     

    if (!fp) {
        return false;
    }                

    string read_line;
    ptree prcinforoot ;
    ptree prcinfo;     
    string::const_iterator start, end;
    match_results<string::const_iterator > what;
    ptree::iterator  ptit = prcinforoot.push_back(make_pair("prcinfo", prcinfo ));
    ptree::iterator pit;                                                          
    regex expression("(.*?) (.*?) (.*?) (.*?) (.*)");                             
    ptree temp;                                                                   
    bool percentcpu = false;                                                      
    bool percentmemory = false;                                                   

    if (args.find("percentcpu") != args.end()) {
        percentcpu = true;                      
    }                                           
    if (args.find("percentmemory") != args.end()) {
        percentmemory = true;                      
    }                                              

    while (fgets(line, 8096, fp) != 0){
        read_line = line;              
        start = read_line.begin();     
        end = read_line.end();         
        if (!regex_search(start, end, what, expression, match_default)){
            continue;                                                   
        }                                                               
        if (what.size() != 6){                                          
            continue;                                                   
        }                                                               
        pit = ptit->second.push_back(make_pair("process", temp));    
        pit->second.push_back(make_pair("owner", string(what[1].first, what[1].second)));
        pit->second.push_back(make_pair("processid", string(what[2].first, what[2].second)));
        if (percentcpu)                                                                         
            pit->second.push_back(make_pair("percentcpu", string(what[3].first, what[3].second)));
        if (percentmemory)                                                                           
            pit->second.push_back(make_pair("percentmemory", string(what[4].first, what[4].second)));
        pit->second.push_back(make_pair("processcommand", string(what[5].first, what[5].second)));   
    }                                                                                                   
    fclose(fp);                                                                                         
    _generateOutput(&prcinforoot, type, response);                                                      
    std::cout << response << std::endl;                                                     
    return true;                                                                                        
}                                                                                                       

bool Executor::sysinfo(const set<string>& args, outputType type, 
        string& response)                                              
{                                                                      
    const char *commandcpu = "cat /proc/cpuinfo |  sed 's/\\s\\+: /:/g'";
    const char *commandmemory = "cat /proc/meminfo |  sed 's/:\\s\\+/:/g'";
    const char *commandos = "uname -a";                                    
    FILE *fp;                                                              
    char commandout[1048];                                                 
    string line;                                                           
    ptree sysinforoot ;                                                    
    ptree sysinfo;                                                         
    ptree::iterator  ptit = sysinforoot.push_back(make_pair("sysinfo", sysinfo ));

    while (args.empty() || args.find("cpus") != args.end()) {
        fp = popen(commandcpu, "r");                         
        if (!fp)                                             
            break;                                           
        ptree temp;                                          
        string field;                                        
        string value;                                        
        size_t index;                                        
        ptree::iterator pit;                                 
        while (fgets(commandout, 1048, fp) != 0){            
            line = commandout;                               
            StrUtil::eraseAllChars(line, ")( \r\n\t");       
            if (strncasecmp(line.c_str(),"processor:", 10) == 0) {
                pit = ptit->second.push_back(make_pair("cpus", temp));
            }                                                            
            index = line.find(":");                                      
            if (string::npos == index)                                   
                continue;                                                
            field = line.substr(0, index);                               
            value = line.substr(index + 1);                              
            pit->second.push_back(make_pair(field, value));           
        }                                                                
        fclose(fp);                                                      
        break;                                                           
    }                                                                    
                                                                         
    while (args.empty()  ||  args.find("memory") != args.end()) {        
        fp = popen(commandmemory, "r");                                  
        if (!fp)                                                         
            break;                                                       
        ptree temp;                                                      
        string field;                                                    
        string value;                                                    
        size_t index;                                                    
        ptree::iterator pit = ptit->second.push_back(make_pair("memory", temp));
        while (fgets(commandout, 1048, fp) != 0){                                  
            line = commandout;                                                     
            StrUtil::eraseAllChars(line, ")( \n\r\t");                             
            index = line.find(":");                                                
            if (string::npos == index)                                             
                continue;                                                          
            field = line.substr(0, index );                                        
            value = line.substr(index + 1);                                        
            pit->second.push_back(make_pair(field, value));                     
        }                                                                          
        fclose(fp);                                                                
        break;                                                                     
    }                                                                              
    while (args.empty() || args.find("os") != args.end()) {                        
        fp = popen(commandos, "r");
        if (!fp)
            break;
        if (fgets(commandout, 1048, fp) == 0) {
            fclose(fp);
            break;
        }
        line = commandout;
        ptree temp;
        string field;
        string value;
        size_t index;
        ptree::iterator pit = ptit->second.push_back(make_pair("os", temp));
        pit->second.push_back(make_pair("osdetails", line));
        fclose(fp);
        break;
    }

    _generateOutput(&sysinforoot, type, response);
    std::cout << response << std::endl;

    return true;
}

void Executor::_generateOutput(void *data, outputType type, string& output)
{
    std::ostringstream ostr;
    ptree *pt = (ptree *) data;
    if (TYPE_JSON == type)
        write_json(ostr, *pt);
    else if (TYPE_XML == type)
        write_xml(ostr, *pt);

    output = ostr.str();
}