c++ 使用zookeeper

---

zookeeper是一个分布式的，开源的分布式应用程序协调程序，可以为分布式应用提供一致性服务，包括：配置维护、域名服务、分布式同步、组服务，一般我们常用的功能包括使用zookeeper进行配置统一管理、master选举、分布式锁、服务器注册和发现等

zookeeper 官方只提供了 c client api, 没有c++的api，为了使用更加的方便，接口更加好用，笔者简单对c client进行了下封装，使c++用户用起来更方便

封装好的cpp接口见 https://github.com/yandaren/zk_cpp
下面简单介绍下，zk_cpp的一些函数接口

1. 一些静态的函数

这些函数主要是全局作用范围的

    /**
     * @brief get the errocode string
     */
    static const char*  error_string(int32_t rc);

    /** 
     * @brief state to string
     */
    static const char*  state_to_string(int32_t state);

    /** 
     * @brief set zookeeper client internal log level
     */
    static void         set_log_lvl(zoo_log_lvl lvl);

    /** 
     * @brief set the log stream
     */
    static void         set_log_stream(FILE* file);

2. 然后就是对节点的一些常规访问操作接口

• 创建节点

    zoo_rc      create_persistent_node(const char* path, const std::string& value, const std::vector<zoo_acl_t>& acl);
    zoo_rc      create_sequence_node(const char* path, const std::string& value, const std::vector<zoo_acl_t>& acl, std::string& returned_path_name);
    zoo_rc      create_ephemeral_node(const char* path, const std::string& value, const std::vector<zoo_acl_t>& acl);
    zoo_rc      create_sequance_ephemeral_node(const char* path, const std::string& value, const std::vector<zoo_acl_t>& acl, std::string& returned_path_name);

• 设置节点的值

    zoo_rc      set_node(const char* path, const std::string& value, int32_t version);

• 获取节点的值

    zoo_rc      get_node(const char* path, std::string& out_value, zoo_state_t* info, bool watch);

• 获取节点的所有子节点

    zoo_rc      get_children(const char* path, std::vector<std::string>& children, bool watch);

• 删除节点

    zoo_rc      delete_node(const char* path, int32_t version);

• 节点是否存在

    zoo_rc      exists_node(const char* path, zoo_state_t* info, bool watch);

• 设置节点的acl

    zoo_rc      set_acl(const char* path, const std::vector<zoo_acl_t>& acl, int32_t version);

• 获取节点的acl

    zoo_rc      get_acl(const char* path, std::vector<zoo_acl_t>& acl);

• 添加权限认证

    zoo_rc      add_auth(const std::string& user_name, const std::string& user_passwd);

3. 设置一些事件回调

• 设置节点的值变化的通知回调函数

    zoo_rc      watch_data_change(const char* path, const data_change_event_handler_t& handler, std::string* value);

• 设置节点的子节点变化(增/减)的通知回调函数

    zoo_rc      watch_children_event(const char* path, const child_event_handler_t& handler, std::vector<std::string>* out_children );

4. 具体的使用见zk_cpp_test.cpp

#include "zk_cpp/zk_cpp.h"
#include <stdio.h>
#include <iostream>
#include <string>

namespace utils {
    static std::string perms_to_string(int32_t perms) {
        if (perms == utility::zoo_perm_all) {
            return "all";
        }

        std::string ret;
        if (perms & utility::zoo_perm_create) {
            ret.append("c");
        }
        if (perms & utility::zoo_perm_read) {
            ret.append("r");
        }
        if (perms & utility::zoo_perm_delete) {
            ret.append("d");
        }
        if (perms & utility::zoo_perm_write) {
            ret.append("w");
        }
        if (perms & utility::zoo_perm_admin) {
            ret.append("a");
        }
        return ret;
    }

    static int32_t perms_string_to_int(const std::string& perm_str) {
        if (perm_str == "all") {
            return utility::zoo_perm_all;
        }

        int32_t perms = 0;
        for (auto c : perm_str) {
            if (c == 'c') {
                perms |= utility::zoo_perm_create;
            }
            else if (c == 'r') {
                perms |= utility::zoo_perm_read;
            }
            else if (c == 'd') {
                perms |= utility::zoo_perm_delete;
            }
            else if (c == 'w') {
                perms |= utility::zoo_perm_write;
            }
            else if (c == 'a') {
                perms |= utility::zoo_perm_admin;
            }
        }
        return perms;
    }

    static void        string_splits(const char* in_str, const char* sep_str, std::vector<std::string>& out_splits){
        std::string in(in_str);
        std::string sep(sep_str);
        std::size_t start_pos = 0;
        while (start_pos < in.size()){
            std::size_t pos = in.find(sep, start_pos);
            if (pos != std::string::npos){
                out_splits.push_back(std::move(in.substr(start_pos, pos - start_pos)));
                start_pos = pos + 1;
            }
            else{
                out_splits.push_back(std::move(in.substr(start_pos, in.size() - start_pos)));
                break;
            }
        }
    }
}

void print_zk_cpp_usage() {
    fprintf(stderr, "usage\n");
    fprintf(stderr, "    create <path> <value> <flag>\n"
                    "                          0 - persistence\n"
                    "                          1 - ephemeral \n"
                    "                          2 - sequence \n"
                    "                          3 - sequence and ephemeral\n");
    fprintf(stderr, "    delete <path>\n");
    fprintf(stderr, "    set <path> <data>\n");
    fprintf(stderr, "    get <path>\n");
    fprintf(stderr, "    ls <path>\n");
    fprintf(stderr, "    exists <path>\n");
    fprintf(stderr, "    setacl <path> scheme:id:perm\n");
    fprintf(stderr, "    getacl <path>\n");
    fprintf(stderr, "    addauth username passwd\n");
    fprintf(stderr, "    watch_data <path> \n");
    fprintf(stderr, "    watch_child <path> \n");
}

void data_change_event(const std::string& path, const std::string& new_value) {
    printf("data_change_event, path[%s] new_data[%s]\n", path.c_str(), new_value.c_str());
}

void child_change_events(const std::string& path, const std::vector<std::string>& children) {
    printf("child_change_events, path[%s] new_child_count[%d]\n", path.c_str(), (int32_t)children.size());

    for (int32_t i = 0; i < (int32_t)children.size(); ++i) {
        printf("%d, %s\n", i, children[i].c_str());
    }
}

int main() {
    printf("zk_cpp test\n");

    /** format is "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002" */
    std::string urls;

    printf("url formt is '127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002'\n");
    printf("input zk server urls:\n");
    std::cin >> urls;

    utility::zk_cpp zk;
    
    do {
        utility::zoo_rc ret = zk.connect(urls);
        if (ret != utility::z_ok) {
            printf("try connect zk server failed, code[%d][%s]\n", 
                ret, utility::zk_cpp::error_string(ret));
            break;
        }

        print_zk_cpp_usage();

        std::string cmd;
        while (std::cin >> cmd) {
            if (cmd == "create") {
                std::string path, value;
                int32_t flag;

                std::cin >> path >> value >> flag;

                std::string rpath = path;
                utility::zoo_rc ret = utility::z_ok;
                if (flag == 0) {
                    std::vector<utility::zoo_acl_t> acl;
                    acl.push_back(utility::zk_cpp::create_world_acl(utility::zoo_perm_all));
                    ret = zk.create_persistent_node(path.c_str(), value, acl);
                }
                else if (flag == 1) {
                    std::vector<utility::zoo_acl_t> acl;
                    acl.push_back(utility::zk_cpp::create_world_acl(utility::zoo_perm_all));
                    ret = zk.create_ephemeral_node(path.c_str(), value, acl);
                }
                else if (flag == 2) {
                    std::vector<utility::zoo_acl_t> acl;
                    acl.push_back(utility::zk_cpp::create_world_acl(utility::zoo_perm_all));
                    ret = zk.create_sequence_node(path.c_str(), value, acl, rpath);
                }
                else if (flag == 3) {
                    std::vector<utility::zoo_acl_t> acl;
                    acl.push_back(utility::zk_cpp::create_world_acl(utility::zoo_perm_all));
                    ret = zk.create_sequance_ephemeral_node(path.c_str(), value, acl, rpath);
                }
                else {
                    printf("invalid create path flag[%d]\n", flag);
                    continue;
                }

                printf("create path[%s] flag[%d] ret[%d][%s], rpath[%s]\n",
                    path.c_str(), flag, ret, utility::zk_cpp::error_string(ret), rpath.c_str());
            }
            else if (cmd == "get") {
                std::string path;
                std::string value;

                std::cin >> path;

                utility::zoo_rc ret = zk.get_node(path.c_str(), value, nullptr, true);
                printf("try get path[%s]'s value, value[%s] ret[%d][%s]\n",
                    path.c_str(), value.c_str(), ret, utility::zk_cpp::error_string(ret));
            }
            else if (cmd == "set") {
                std::string path;
                std::string value;

                std::cin >> path >> value;

                utility::zoo_rc ret = zk.set_node(path.c_str(), value, -1);
                printf("try set path[%s]'s value to [%s] ret[%d][%s]\n",
                    path.c_str(), value.c_str(), ret, utility::zk_cpp::error_string(ret));
            }
            else if (cmd == "exist") {
                std::string path;

                std::cin >> path;

                utility::zoo_rc ret = zk.exists_node(path.c_str(), nullptr, true);
                printf("try_check path[%s] exist[%d], ret[%d][%s]\n",
                    path.c_str(), ret == utility::z_ok, ret, utility::zk_cpp::error_string(ret));
            }
            else if (cmd == "ls") {
                std::string path;
                std::vector<std::string> children;

                std::cin >> path;

                utility::zoo_rc ret = zk.get_children(path.c_str(), children, true);
                printf("try get path[%s]'s children's, children count[%d], ret[%d][%s]\n",
                    path.c_str(), (int32_t)children.size(), ret, utility::zk_cpp::error_string(ret));

                std::string list;
                list.append("[");

                for (int32_t i = 0; i < (int32_t)children.size(); ++i) {
                    //printf("%s\n", children[i].c_str());
                    list.append(children[i]).append(", ");
                }

                list.append("]");
                printf("%s\n", list.c_str());
            }
            else if (cmd == "delete") {
                std::string path;

                std::cin >> path;

                utility::zoo_rc ret = zk.delete_node(path.c_str(), -1);
                printf("try delete path[%s], ret[%d][%s]\n",
                    path.c_str(), ret, utility::zk_cpp::error_string(ret));
            }
            else if (cmd == "setacl") {
                /* setacl <path> scheme:id:perm */
                std::string path;
                std::string acl_string;

                std::cin >> path >> acl_string;

                std::vector<std::string> splits;

                utils::string_splits(acl_string.c_str(), ":", splits);

                if (splits.size() < 3) {
                    printf("acl formt[%s] error\n", acl_string.c_str());
                    continue;
                }
                std::vector<utility::zoo_acl_t> acl;

                const std::string& scheme = splits[0];
                int32_t perms = 0; 
                if (scheme == "world") {
                    const std::string& id = splits[1];
                    if (id != "anyone") {
                        printf("acl world formt error, id[%s]\n", id.c_str());
                        continue;
                    }
                    const std::string& perm_str = splits[2];
                    perms = utils::perms_string_to_int(perm_str);

                    auto ac = utility::zk_cpp::create_world_acl(perms);
                    acl.push_back(ac);
                }
                else if (scheme == "auth") {
                    // "id" is empty
                    const std::string& perm_str = splits[2];
                    perms = utils::perms_string_to_int(perm_str);

                    auto ac = utility::zk_cpp::create_auth_acl(perms);
                    acl.push_back(ac);
                }
                else if (scheme == "digest") {
                    if (splits.size() < 4) {
                        printf("acl digest formt error, acl_str[%s]\n", acl_string.c_str());
                        continue;
                    }

                    const std::string& user_name = splits[1];
                    const std::string& passwd = splits[2];
                    const std::string& perm_str = splits[3];
                    perms = utils::perms_string_to_int(perm_str);

                    auto ac = utility::zk_cpp::create_digest_acl(perms, user_name, passwd);
                    acl.push_back(ac);
                }
                else if (scheme == "ip") {
                    const std::string& id = splits[1];
                    const std::string& perm_str = splits[2];
                    perms = utils::perms_string_to_int(perm_str);
                    auto ac = utility::zk_cpp::create_ip_acl(perms, id);
                    acl.push_back(ac);
                }
                else {
                    printf("unsupported scheme[%s]\n", scheme.c_str());
                    continue;
                }

                auto ret = zk.set_acl(path.c_str(), acl, -1);
                printf("set acl for path[%s], ret[%d][%s]\n", path.c_str(), ret, utility::zk_cpp::error_string(ret));
            }
            else if (cmd == "getacl") {
                std::string path;
                std::cin >> path;

                std::vector<utility::zoo_acl_t> acl;
                utility::zoo_rc rt = zk.get_acl(path.c_str(), acl);
                printf("get acl of path[%s], ret[%d][%s], acl_count[%d]\n",
                    path.c_str(), ret, utility::zk_cpp::error_string(ret), (int32_t)acl.size());

                for (auto& ac : acl) {
                    printf("%s:%s:%s\n", ac.scheme.c_str(), ac.id.c_str(), utils::perms_to_string(ac.perm).c_str());
                }
            }
            else if (cmd == "addauth") {
                std::string user_name;
                std::string user_passwd;
                std::cin >> user_name >> user_passwd;

                utility::zoo_rc rt = zk.add_auth(user_name, user_passwd);
                printf("add_auth, ret[%d][%s].\n", ret, utility::zk_cpp::error_string(ret));
            }
            else if (cmd == "watch_data") {
                std::string path;

                std::cin >> path;
                std::string value;

                utility::zoo_rc ret = zk.watch_data_change(path.c_str(), data_change_event, &value);

                printf("try watch_data change of path[%s], value[%s] ret[%d][%s]\n",
                    path.c_str(), value.c_str(), ret, utility::zk_cpp::error_string(ret));
            }
            else if (cmd == "watch_child") {
                std::string path;
                std::cin >> path;

                std::vector<std::string> children;

                utility::zoo_rc ret = zk.watch_children_event(path.c_str(), child_change_events, &children);

                printf("try watch_child change of path[%s], child_count[%d] ret[%d][%s]\n",
                    path.c_str(), (int32_t)children.size(), ret, utility::zk_cpp::error_string(ret));

                for (int32_t i = 0; i < (int32_t)children.size(); ++i) {
                    printf("%d, %s\n", i, children[i].c_str());
                }
            }
            else {
                printf("unsupported msg[%s]\n", cmd.c_str());
            }
        }

    } while (0);

#ifdef _WIN32
    system("pause");
#endif

    return 0;
}