ZooKeeper is shipped with an official C client library, which can be used to develop distributed applications in C/C++. Many of the other language bindings such as the Python client binding shipped with ZooKeeper distribution are built using the C library. In this section, we will learn about the C client APIs and illustrate their usage by developing a znode data-watcher client.
In the Connecting to ZooKeeper with C-based shell section of Chapter 1, A Crash Course in Apache ZooKeeper, we learned how to build the ZooKeeper C library and the C-based shell using the make command. Once we build the C library, the following two shared libraries get built:
- The multi-threaded library:
libzookeeper_mt.so.2.0.0 - The single-thread library:
libzookeeper_st.so.2.0.0
The libraries are usually installed in /usr/local/lib. You need to have this path exported through your LD_LIBRARY_PATH environment variables for the applications to find the libraries.
The multi-threaded library is most commonly used and recommended. It provides both the synchronous as well as the asynchronous APIs. This library functions by creating an I/O thread and an event dispatch thread to handle connections and event callbacks. The single-thread library provides only the asynchronous APIs and callbacks. This library is used only on platforms where the pthread library is not available.
The C header file zookeeper.h in src/c/include in the ZooKeeper distribution outlines the main reference for the C client APIs. The ZooKeeper C command line shell's implementation is available in src/c/src as cli.c. Readers are advised to go through this code to understand how client applications are written using the ZooKeeper C API.
The ZooKeeper C API provides a way for the client to create a session with the ZooKeeper server through a handle. The handle represents a connection to the ZooKeeper service and is needed to invoke any ZooKeeper function. A handle is declared as type zhandle_t, and is obtained by calling the zookeeper_init function:
ZOOAPI zhandle_t* zookeeper_init ( const char *host, /* comma separated host:port pairs */ watcher_fn fn, /* watcher callback function */ int recv_timeout, /* session expiration time */ const clientid_t *clientid, /* session id for reconnection */ void *context, /* context of the handle object */ int flags /* for future use, should be set to zero*/ )
The
zookeeper_init function returns a handle, which is a pointer to the opaque zhandle structure and is used to communicate with the ZooKeeper service. Also, a zookeeper session gets established after this call, which corresponds to the handle. If the function fails to create a new zhandle structure, it returns NULL, and the error is set accordingly. The session establishment is asynchronous and hence the session should not be considered established until (and unless) an event of the state ZOO_CONNECTED_STATE is received by the client program. This event can be processed by implementing a watcher function with the following signature:
typedef void (* watcher_fn)( zhandle_t *zh, /* the zookeeper handle */ int type,/* event type, e.g. ZOO_SESSION_EVENT */ int state,/* connection state, e.g. ZOO_CONNECTED_STATE */ const char *path, /* znode path for which the watcher is triggered. NULL for session events */ void *watcherCtx/* watcher context object */ )
Having seen how to connect to a ZooKeeper service, let's now write a simple program to connect to the ZooKeeper service and list the children under the / root znode path.
The source code for our ZooKeeper client that uses the C library follows next.
Let's name this file hello_zookeeper.c:
#include <stdio.h>
#include "zookeeper.h"
static zhandle_t *zh;
typedef struct String_vector zoo_string;
/* An empty Watcher function */
void my_watcher_func(zhandle_t *zzh, int type, int state,
const char *path, void *watcherCtx) {}
/* Main Function */
int
main(int argc, char *argv[])
{
int i, retval;
char *host_port = "localhost:2181";
char *zoo_root = "/";
zoo_string *children_list = (zoo_string *) malloc(sizeof(zoo_string));
/* Connect to ZooKeeper server */
zh = zookeeper_init(host_port, my_watcher_func, 2000, 0, NULL, 0);
if (zh == NULL)
{
fprintf(stderr, "Error connecting to ZooKeeper server!
");
exit(EXIT_FAILURE);
}
/* Get the list of children synchronously */
retval = zoo_get_children(zh, zoo_root, 0, children_list);
if (retval != ZOK)
{
fprintf(stderr, "Error retrieving znode from path %s!
", zoo_root);
exit(EXIT_FAILURE);
}
fprintf(stderr, "
=== znode listing === [ %s ]", zoo_root);
for (i = 0; i < children_list->count; i++)
{
fprintf(stderr, "
(%d): %s", i+1, children_list->data[i]);
}
fprintf(stderr, "
=== done ===
");
/* Finally close the ZooKeeper handle */
zookeeper_close(zh);
return 0;
}Assuming the ZooKeeper C client-shared libraries are installed in /usr/local/lib, we can compile the preceding program as follows:
$ gcc -Wall hello_zookeeper.c -o hello_zookeeper -I${ZK_HOME}/src/c/include -I${ZK_HOME}/src/c/generated/ -L/usr/local/lib/ -lzookeeper_mt
Before executing the program, let's create a few znodes in the ZooKeeper / root znode path using the ZooKeeper shell:
Now, let's try to list the znodes using the C client program that we have written, as shown in the following screenshot:
In the preceding output, we can see that as the C client executes, the client library displays a lot of informative log messages that pertain to the client library version, the operating system, and the architecture of the machine. Then, it tries to connect to the ZooKeeper server, and once connected, it prints the listing of znodes, which is the same as what we get through the ZooKeeper shell. Finally, it closes the session with the ZooKeeper server.
In this section, we will implement a znode data-watcher client similar to what we did while reading about the ZooKeeper Java client APIs. To illustrate the znode data watcher, we will write a client zdata_watcher.c using the C APIs, which will continuously listen for ZOO_CHANGED_EVENT events from the ZooKeeper server in a znode path called /MyData. Another client program, zdata_updater.c, will periodically update the data field in /MyData, which will result in the generation of events, and upon receiving these events, zdata_watcher.c will print the changed data into the terminal.
Let's take a look at the source code of zdata_watcher.c:
#include <stdio.h>
#include <errno.h>
#include <string.h>
#include <stdlib.h>
#include <unistd.h>
#include <zookeeper.h>
/* ZooKeeper Znode Data Length (1MB, the max supported) */
#define ZDATALEN 1024 * 1024
static char *host_port;
static char *zoo_path = "/MyData";
static zhandle_t *zh;
static int is_connected;
static char *watcher_ctx = "ZooKeeper Data Watcher";
/**
* Watcher function for connection state change events
*/
void connection_watcher(zhandle_t *zzh, int type, int state, const char *path, void* context)
{
if (type == ZOO_SESSION_EVENT)
{
if (state == ZOO_CONNECTED_STATE)
{
is_connected = 1;
}
else
{
is_connected = 0;
}
}
}
/**
* Data Watcher function for /MyData node
*/
static void
data_watcher(zhandle_t *wzh, int type, int state, const char *zpath, void *watcher_ctx)
{
char *zoo_data = malloc(ZDATALEN * sizeof(char));
int zoo_data_len = ZDATALEN;
if (state == ZOO_CONNECTED_STATE)
{
if (type == ZOO_CHANGED_EVENT)
{
/* Get the updated data and reset the watch */
zoo_wget(wzh, zoo_path, data_watcher,
(void *)watcher_ctx, zoo_data, &zoo_data_len, NULL);
fprintf(stderr, "!!! Data Change Detected !!!
");
fprintf(stderr, "%s
", zoo_data);
}
}
}
int main(int argc, char *argv[])
{
int zdata_len;
char *zdata_buf = NULL;
if (argc != 2)
{
fprintf(stderr, "USAGE: %s host:port
", argv[0]);
exit(EXIT_FAILURE);
}
host_port = argv[1];
zh = zookeeper_init(host_port, connection_watcher, 2000, 0, 0, 0);
if (zh == NULL)
{
fprintf(stderr, "Error connecting to ZooKeeper server[%d]!
", errno);
exit(EXIT_FAILURE);
}
while (1)
{
if (is_connected)
{
zdata_buf = (char *)malloc(ZDATALEN * sizeof(char));
if (ZNONODE == zoo_exists(zh, zoo_path, 0, NULL))
{
if (ZOK == zoo_create( zh, zoo_path, NULL, -1, & ZOO_OPEN_ACL_UNSAFE, 0, NULL, 0))
{
fprintf(stderr, "%s created!
", zoo_path);
}
else
{
fprintf(stderr,
"Error Creating %s!
", zoo_path);
exit(EXIT_FAILURE);
}
}
if (ZOK != zoo_wget(zh, zoo_path, data_watcher, watcher_ctx, zdata_buf, &zdata_len, NULL))
{
fprintf(stderr, "Error setting watch at %s!
", zoo_path);
}
pause();
}
}
free(zdata_buf);
return 0;
}The zdata_updater connects to the ZooKeeper instance that is running in the localhost and updates the data field of the znode path /MyData with the current local date and time. It updates the znode path /MyData every 5 seconds, which makes the ZooKeeper server to trigger an event of type ZOO_CHANGED_EVENT. The zdata_watcher, which had set a watch for this znode path, receives the notification for the data change event. It then retrieves the current data, resets the watch, and prints the data in the console.
The code for the data updater is illustrated next:
#include <time.h>
#include <stdio.h>
#include <errno.h>
#include <string.h>
#include <stdlib.h>
#include <unistd.h>
#include <zookeeper.h>
/* ZooKeeper Znode Data Length (1MB, the max supported) */
#define ZDATALEN 1024 * 1024
static char *host_port;
static char *zoo_path = "/MyData";
static zhandle_t *zh;
static int is_connected;
/**
* Watcher function for connection state change events
*/
void connection_watcher(zhandle_t *zzh, int type, int state, const char *path, void* context)
{
if (type == ZOO_SESSION_EVENT)
{
if (state == ZOO_CONNECTED_STATE)
{
is_connected = 1;
}
else
{
is_connected = 0;
}
}
}
int main(int argc, char *argv[])
{
char zdata_buf[128];
struct tm *local;
time_t t;
if (argc != 2)
{
fprintf(stderr, "USAGE: %s host:port
", argv[0]);
exit(EXIT_FAILURE);
}
host_port = argv[1];
zh = zookeeper_init(host_port, connection_watcher, 2000, 0, 0, 0);
if (zh == NULL)
{
fprintf(stderr, "Error connecting to ZooKeeper server[%d]!
", errno);
exit(EXIT_FAILURE);
}
sleep(3); /* Sleep a little for connection to complete */
if (is_connected)
{
if (ZNONODE == zoo_exists(zh, zoo_path, 0, NULL))
{
fprintf(stderr, "%s doesn't exist! Please start zdata_watcher.
", zoo_path);
exit(EXIT_FAILURE);
}
while(1)
{
t = time(NULL);
local = localtime(&t);
memset(zdata_buf,' ',strlen(zdata_buf));
strcpy(zdata_buf,asctime(local));
if (ZOK != zoo_set(zh, zoo_path, zdata_buf, strlen(zdata_buf), -1))
{
fprintf(stderr, "Error in write at %s!
", zoo_path);
}
sleep(5);
}
}
return 0;
}
}Let's compile these two programs and see them in action:
$ gcc -Wall zdata_watcher.c -o zdata_watcher -I${ZK_HOME}/src/c/include -I${ZK_HOME}/src/c/generated/ -L/usr/local/lib/ -lzookeeper_mt $ gcc -Wall zdata_updater.c -o zdata_updater -I${ZK_HOME}/src/c/include -I${ZK_HOME}/src/c/generated/ -L/usr/local/lib/ -lzookeeper_mt
In one terminal window, when you execute zdata_watcher, it creates a znode in the path /MyData, and starts listening for events of the type ZOO_CHANGED_EVENT:
Once we execute the zdata_watcher, it initiates a connection to the ZooKeeper server, and after the session is established, it creates a znode called /MyData in the root path of the ZooKeeper tree.
Now, in another terminal, when you run the zdata_updater, the zdata_watcher starts receiving the ZooKeeper events.
$ ./zdata_updater localhost:2181
After starting the zdata_updater client, you will see output similar to the following in the terminal window where the zdata_watcher is running:
Our data-watcher client is working successfully. From these examples, you will have a firm grasp on the ZooKeeper C client APIs. In these examples, we have used the synchronous APIs provided by the C APIs. However, in real applications, the asynchronous APIs are most commonly used. Implementation of the previous examples using the asynchronous API calls is left as an exercise for the readers. Also, it's recommended that you use the multithreaded version of the API. For example, with the single-threaded library, a callback might block a thread while performing other operations such as disk I/O, which might cause session timeout. The multithreaded library uses separate threads to handle callbacks, and this issue will not occur.
In the next section, we will learn about the ZooKeeper APIs for the Python programming language.