zkwrapper.h

// Copyright 2017 Rice University, COMP 413 2017

#ifndef ZKWRAPPER_INCLUDE_ZKWRAPPER_H_
#define ZKWRAPPER_INCLUDE_ZKWRAPPER_H_

#include <zookeeper.h>
#include <easylogging++.h>
#include <string.h>
#include <string>
#include <vector>
#include <iostream>
#include <memory>
#include <cstring>
#include <map>
#include "LRUCache.h"


/*
 * These extend the set of error codes defined by ZooKeeper to provide more
 * error codes that can be returned by zkwrapper.
 */
enum ZK_ERRORS {
  ZKWRAPPERINSUFFICIENTBUFFER = -997,
  ZKWRAPPERUNINITIALIZED = -998,
  ZKWRAPPERDEFAULTERROR = -999
};

class ZkNnClient;

class ZooOp {
 public:
  ZooOp(const std::string &path_in,
      const std::vector<std::uint8_t> &data_in) {
    this->path = new char[path_in.size() + 1];
    snprintf(this->path, path_in.size() + 1, "%s", path_in.c_str());
    if (data_in.size() != 0) {  // Only save non-empty data
      this->num_bytes = data_in.size();
      this->data = new char[this->num_bytes];
      memcpy(this->data, data_in.data(), data_in.size());
    }
    op = new zoo_op_t();
  }

  ~ZooOp() {
    delete path;
    if (data) {
      delete data;
    }
    delete op;
  }

  zoo_op_t *op = nullptr;
  char *path = nullptr;
  char *data = nullptr;
  int num_bytes = 0;
};

class ZKWrapper {
 public:
  ZKWrapper(const std::string &host, int &error_code, const std::string &root);

  ~ZKWrapper();

  std::string prepend_zk_root(const std::string &path) const;

  std::string removeZKRoot(const std::string &path) const;

  std::string removeZKRootAndDir(const std::string &prefix,
                                 const std::string &path) const;

  static std::string translate_error(int error_code);

  static std::string translate_watch_event_type(int type);

  static std::string translate_watch_state(int state);

  static void watcher_znode_data(zhandle_t *zzh,
                                 int type,
                                 int state,
                                 const char *path,
                                 void *watcherCtx);

  bool create(const std::string &path,
              const std::vector<std::uint8_t> &data,
              int &error_code,
              bool ephemeral,
              bool sync = true) const;

  bool create_ephemeral(const std::string &path,
                        const std::vector<std::uint8_t> &data,
                        int &error_code,
                        bool sync = true) const;

  bool create_sequential(const std::string &path,
                         const std::vector<std::uint8_t> &data,
                         std::string &new_path,
                         bool ephemeral,
                         int &error_code,
                         bool sync = true) const;

  bool recursive_create(const std::string &path,
                        const std::vector<std::uint8_t> &data,
                        int &error_code,
                        bool sync = true) const;

  bool exists(const std::string &path, bool &exist, int &error_code) const;

  bool wexists(const std::string &path,
               bool &exist,
               watcher_fn watch,
               void *watcherCtx,
               int &error_code) const;

  bool delete_node(const std::string &path,
                   int &error_code,
                   bool sync = true) const;

  bool recursive_delete(const std::string &path, int &error_code) const;

  bool get_children(const std::string &path,
                    std::vector<std::string> &children,
                    int &error_code) const;

  bool wget_children(const std::string &path,
                     std::vector<std::string> &children,
                     watcher_fn watch,
                     void *watcherCtx,
                     int &error_code) const;

  bool get(const std::string &path,
           std::vector<std::uint8_t> &data,
           int &error_code,
           bool resize) const;

  bool get_info(const std::string &path,
                struct Stat &stat,
                int &error_code) const;

  bool wget(const std::string &path,
            std::vector<std::uint8_t> &data,
            watcher_fn watch,
            void *watcherCtx,
            int &error_code,
            bool resize) const;

  bool set(const std::string &path,
           const std::vector<std::uint8_t> &data,
           int &error_code,
           bool sync = true,
           int version = -1) const;

  // TODO(2016): Crexate a path buffer for returning sequential path names
  std::shared_ptr<ZooOp> build_create_op(const std::string &path,
                                         const std::vector<std::uint8_t> &data,
                                         const int flags = 0) const;

  std::shared_ptr<ZooOp> build_delete_op(const std::string &path,
                                         int version = -1) const;

  std::shared_ptr<ZooOp> build_set_op(const std::string &path,
                                      const std::vector<std::uint8_t> &data,
                                      int version = -1) const;

  bool execute_multi(const std::vector<std::shared_ptr<ZooOp>> operations,
                     std::vector<zoo_op_result> &results,
                     int &error_code,
                     bool sync = true) const;

  bool flush(const std::string &full_path, bool synchronous = false) const;

  void close();

  static std::vector<uint8_t> get_byte_vector(const std::string &string);

  static void print_error(int error) {
    LOG(ERROR) << "[zkwrapper] Got error: " << translate_error(error);
  }

  static const std::vector<std::uint8_t> EMPTY_VECTOR;

 private:
  friend void watcher(zhandle_t *zzh, int type, int state, const char *path,
                      void *watcherCtx);

  lru::Cache<std::string, std::shared_ptr<std::vector<unsigned char>>> *cache;
  zhandle_t *zh;
  std::string root = "";
  /*
   * Was a connection to ZooKeeper successfully established? Note that a
   * successful connection requires a successful call to zookeeper_init
   * and an invocation of the global watcher (passed into zookeeper_init)
   * with the desired state (ZOO_CONNECTED_STATE).
   */
  bool connected = false;
  /*
   * Use the "initializing" flag to explicitly allow certain methods to execute
   * in the constructor. Otherwise, the "initialized" flag prevents any useful
   * method from being invoked on an unsuccessfully initialized instance.
   */
  bool initializing = true;
  bool initialized = false;

  static const int MAX_PAYLOAD;
  static const int MAX_PATH_LEN;
  static const int NUM_SEQUENTIAL_DIGITS;
  static const int DEFAULT_ZK_RECV_TIMEOUT;
  static const int INITIAL_CONNECTION_TIMEOUT_MILLIS;
  static const int INITIAL_CONNECTION_RETRY_INTERVAL_MILLIS;
  static const int ROOT_CREATION_RETRY_LIMIT;
  static const int ROOT_CREATION_RETRY_INTERVAL_MILLIS;

  /*
   * TODO(2017): This field (as well fields with the same name in several other
   * classes) is not being used anywhere. We have no idea what the 2016 folks
   * intended to do with it.
   */
  static const std::string CLASS_NAME;
};

#endif  // ZKWRAPPER_INCLUDE_ZKWRAPPER_H_