obcdc 开发说明

#include "libobcdc.h"
#include "ob_errno.h"

using namespace oceanbase::common;
using namespace oceanbase::liboblog;

namespace oceanbase
{
namespace liboblog
{

class ObLogDemo
{
public:
  static ObLogDemo& get_instance();
  static const int64_t NEXT_RECORD_TIMEOUT = 1000000;
public:
  ObLogDemo();
  ~ObLogDemo();
public:
  int init(const char* config_file_path, const uint64_t start_tstamp);
  void destroy();
  int start();
  void stop();
  void run();
private:
  int handle_log_record(ILogRecord *log_record, volatile bool stop_flag);
private:
  ObLogFactory oblog_factory_;
  IObLog* oblog_;
  bool inited_;
  volatile bool stop_flag_;
};

}
}

#include "oblog_demo.h"
#include "ob_log_module.h" // ONLY FOR OB LOG MODULE, DEVELOPER CAN REPLACE IT AND LOG_XXX BELOW WITY YOUR OWN LOG MODULE

#define USING_LOG_PREFIX OBLOG_TAILF // ONLY FOR LOG

namespace oceanbase
{
namespace liboblog
{

ObLogDemo& ObLogDemo::get_instance()
{
  static ObLogDemo oblog_instance;
  return oblog_instance;
}

ObLogDemo::ObLogDemo() :
    oblog_factory_(),
    oblog_(NULL),
    inited_(false),
    stop_flag_(true)
{}

ObLogDemo::~ObLogDemo()
{
    destroy();
}

int ObLogDemo::init(const char* config_file_path, const uint64_t start_tstamp)
{
  int ret = OB_SUCCESS;

  if (NULL == (oblog_ = oblog_factory_.construct_oblog())) {
    LOG_ERROR("failed to construct liboblog", KR(ret));
  } else if (OB_SUCCESS != oblog_->init(config_file_path, start_tstamp)) {
    LOG_ERROR("failed to init liboblog instance", KR(ret), K(config_file_path, K(start_tstamp)));
  } else {
    inited_ = true;
  }

  return ret;
}

void ObLogDemo::destroy()
{
  stop();
  inited_ = false;
  if (NULL != oblog_) {
    oblog_->destroy(); // will call ObLog::stop for another time
    oblog_factory_.deconstruct(oblog_);
    oblog_ = NULL;
  }
}

int ObLogDemo::start()
{
  int ret = OB_SUCCESS;

  if (!inited_) {
    ret = OB_NOT_INIT;
    LOG_ERROR("ObLogDemo not init, please call init function first!", KR(ret));
  } else if (!stop_flag_){
    ret = OB_ERR_UNEXPECTED;
    LOG_ERROR("ObLogDemo already start!", KR(ret), K_(inited), K_(stop_flag));
  } else {
    if (NULL == oblog_) {
      ret = OB_ERR_UNEXPECTED;
      LOG_ERROR("oblog instance should not be null while start ObLogDemo", KR(ret), K_(inited), K_(stop_flag));
    } else if (OB_SUCCESS != oblog_->launch()) {
      LOG_ERROR("liboblog instance launch failed", KR(ret));
    } else {
      LOG_INFO("liboblog instance launch success");
      stop_flag_ = false;
    }
  }

  return ret;
}

void ObLogDemo::stop()
{
  stop_flag_ = true;
  if (NULL != oblog_) {
      oblog_->stop();
  }
}

void ObLogDemo::run()
{
  int ret = OB_SUCCESS;
  if (inited_ && NULL != oblog_) {
    while (OB_SUCCESS == ret && ! stop_flag_) {
      ILogRecord* record = NULL;
      if (OB_SUCCESS == oblog_->next_record(&record, NEXT_RECORD_TIMEOUT)) {
        if (OB_SUCCESS != handle_log_record(record, stop_flag_)) {
          LOG_ERROR("failed to handle LogRecord", KR(ret), K(record));
        }
        // must call release_record
        oblog_->release_record(record);
        record = NULL;
      } else if (OB_TIMEOUT == ret) {
        LOG_WARN("fetch record timeout, may don't have any record", KR(ret));
        // fetch record timeout may cause by no data produced by OB
        // process this situation with your own logic
        // e.g. retry or exit progress, etc.
      } else if (OB_IN_STOP_STATE == ret) {
        stop_flag_ = true;
        ret = OB_SUCCESS;
      } else {
        LOG_ERROR("next_record fail", KR(ret));
        stop_flag_ = true;
      }
    }
  } else {
    LOG_ERROR("should not run ObLogDemo while not inid or oblog instance is NULL, will exit");
  }
}

int ObLogDemo::handle_log_record(ILogRecord *log_record, volatile bool stop_flag)
{
  int ret = OB_SUCCESS;

  // do your process for log_record
  // u can get  stop signal by volatile variable stop_flag
  if (!stop_flag_) {
    LOG_DEBUG("handle log record", K(log_record));
  }
  return ret;
}

}
}

// libobcdc.h: ObLogFactory
namespace oceanbase{
namespace liboblog{


class ObLogFactory
{
public:
  ObLogFactory();
  ~ObLogFactory();
public:
  // 构建 liboblog 实例
  IObLog *construct_oblog();
  // 销毁 liboblog 实例
  void deconstruct(IObLog *log);
};

}
}

// libobcdc.h: IObLog
namespace oceanbase{
namespace liboblog{

// IObLog 是 liboblog 对外提供的接口
// 注意：这里忽略接口中需要传入的参数

class IObLog
{
public:
  virtual ~IObLog() {};
public:
  /*
   * init liboblog
   * @param config_file       config file name
   * @param start_timestamp   start timestamp (by second)
   * @param err_cb            error callback function pointer
   */
  virtual int init(const char *config_file,
      const uint64_t start_timestamp,
      ERROR_CALLBACK err_cb = NULL) = 0;

  /*
   * init liboblog
   * @param configs         config by map
   * @param start_timestamp start timestamp (by secon)
   * @param err_cb          error callback function pointer
   */
  virtual int init(const std::map<std::string, std::string>& configs,
      const uint64_t start_timestamp,
      ERROR_CALLBACK err_cb = NULL) = 0;

  /*
   * init liboblog
   * @param configs         config by map
   * @param start_timestamp start timestamp by microsecond
   * @param err_cb          error callback function pointer
   */
  virtual int init_with_start_tstamp_usec(const std::map<std::string, std::string>& configs,
      const uint64_t start_timestamp_usec,
      ERROR_CALLBACK err_cb = NULL) = 0;

  virtual void destroy() = 0;

  /*
   * fetch next binlog record from OB cluster
   * @param record           binlog record, memory allocated by oblog, support release_record(corresponding times) after mutli next_record
   * @param OB_SUCCESS       success
   * @param OB_TIMEOUT       timeout
   * @param other errorcode  fail
   */
  virtual int next_record(ILogRecord **record, const int64_t timeout_us) = 0;

  /*
   * fetch next binlog record from OB cluster
   * @param [out] record        binlog record, memory allocated by oblog, support release_record(corresponding tiems) after mutli next_record
   * @param [out] major_version major version of ILogRecord
   * @param [out] tenant_id     tenant id of ILogRecord
   *
   * @param OB_SUCCESS          success
   * @param OB_TIMEOUT          timeout
   * @param other error code    fail
   */
  virtual int next_record(ILogRecord **record,
      int32_t &major_version,
      uint64_t &tenant_id,
      const int64_t timeout_us) = 0;

  /*
   * release recorcd for EACH ILogRecord
   * @param record
   */
  virtual void release_record(ILogRecord *record) = 0;

  /*
   * Launch liboblog
   * @retval OB_SUCCESS on success
   * @retval ! OB_SUCCESS on fail
   */
  virtual int launch() = 0;

  /*
   * Stop liboblog
   */
  virtual void stop() = 0;

  /// get all serving tenant id list after oblog inited
  ///
  /// @param [out]            tenant_ids tenant ids that oblog serving
  ///
  /// @retval OB_SUCCESS      success
  /// @retval other value     fail
  virtual int get_tenant_ids(std::vector<uint64_t> &tenant_ids) = 0;
};

}
}