obcdc 开发说明

rpm -ivh oceanbase-ce-libs-****.rpm

/**
 * Copyright (c) 2021 OceanBase
 * OceanBase CE is licensed under Mulan PubL v2.
 * You can use this software according to the terms and conditions of the Mulan PubL v2.
 * You may obtain a copy of Mulan PubL v2 at:
 *          http://license.coscl.org.cn/MulanPubL-2.0
 * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
 * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
 * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
 * See the Mulan PubL v2 for more details.
 *
 * obcdc_demo
 */

#include <iostream>
#include "include/libobcdc/libobcdc.h"
#include "include/libobcdc/ob_errno.h"

using namespace std;
using namespace oceanbase::libobcdc;
using namespace oceanbase::common;

typedef IBinlogRecord Record;

#define LOG(msg) \
    do { \
      std::cout << msg << std::endl; \
    } while (0)

int create_obcdc_instance(ObCDCFactory &cdc_factory, IObCDCInstance *&obcdc_instance)
{
  int ret = OB_SUCCESS;

  if (NULL == (obcdc_instance = cdc_factory.construct_obcdc())) {
    ret = OB_NOT_INIT;
    LOG("[ERROR] construct_obcdc failed");
  }

  return ret;
}

void destroy_obcdc_instance(ObCDCFactory &cdc_factory, IObCDCInstance *obcdc_instance)
{
  obcdc_instance->stop();
  cdc_factory.deconstruct(obcdc_instance);
}

int init_obcdc_instance(IObCDCInstance &obcdc_instance)
{
  int ret = OB_SUCCESS;
  const char *config_path = "conf/libobcdc.conf";

  if (OB_SUCCESS != (ret = obcdc_instance.init(config_path, 0))) {
    LOG("obcdc_instance init failed");
  } else if (OB_SUCCESS != (ret = obcdc_instance.launch())) {
    LOG("obcdc_instance launch failed");
  }

  return ret;
}

int fetch_next_cdc_record(IObCDCInstance &obcdc_instance, Record *record)
{
  int ret = OB_SUCCESS;
  const int64_t timeout = 10000; // usec

  if (OB_SUCCESS != (ret = obcdc_instance.next_record(&record, timeout))) {
    if (OB_TIMEOUT != ret) {
      LOG("[WARN] next_record failed");
    }
  } else if (NULL == record) {
    ret = OB_ERR_UNEXPECTED;
    LOG("invalid record");
  }

  return ret;
}

int release_cdc_record(IObCDCInstance &obcdc_instance, Record *record)
{
  int ret = OB_SUCCESS;

  obcdc_instance.release_record(record);

  return ret;
}
int handle_cdc_record(Record *record)
{
  int ret = OB_SUCCESS;
  return ret;
}

int main(int argc, char **argv)
{
  int ret = OB_SUCCESS;
  ObCDCFactory cdc_factory;
  IObCDCInstance *obcdc_instance = NULL;

  if (OB_SUCCESS != create_obcdc_instance(cdc_factory, obcdc_instance)) {
    LOG("[ERROR] construct_obcdc_instance failed");
  } else if (NULL == obcdc_instance) {
    ret = OB_ERR_UNEXPECTED;
    LOG("[ERROR] obcdc_instance should not be null!");
  } else {
    if (OB_SUCCESS != init_obcdc_instance(*obcdc_instance)) {
      LOG("[ERROR] obcdc_instance init failed");
    } else {
      while(OB_SUCCESS == ret) {
        Record *record = NULL;
        if (OB_SUCCESS != (ret = fetch_next_cdc_record(*obcdc_instance, record))) {
          if (OB_TIMEOUT == ret) {
            ret = OB_SUCCESS;
          } else {
            LOG("[ERROR] fetch_next_cdc_record failed");
          }
        } else if (OB_SUCCESS != (ret = handle_cdc_record(record))) {
          LOG("[ERROR] handle_cdc_record failed");
        } else if (OB_SUCCESS != (ret = release_cdc_record(*obcdc_instance, record))) {
          LOG("[ERROR] release_cdc_record failed");
        }
      }
    }

    destroy_obcdc_instance(cdc_factory, obcdc_instance);
  }

  return 0;
}

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

class ObCDCFactory
{
public:
  ObCDCFactory();
  ~ObCDCFactory();
public:
  IObCDCInstance *construct_obcdc();
  void deconstruct(IObCDCInstance *log);
};

}
}

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

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

class IObCDCInstance
{
public:
  virtual ~IObCDCInstance() {};
public:
  /*
   * init libobcdc
   * @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_sec,
      ERROR_CALLBACK err_cb = NULL) = 0;

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

  /*
   * init libobcdc
   * @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(ICDCRecord **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 ICDCRecord
   * @param [out] tenant_id     tenant id of ICDCRecord
   *
   * @param OB_SUCCESS          success
   * @param OB_TIMEOUT          timeout
   * @param other error code    fail
   */

  virtual int next_record(ICDCRecord **record,
      int32_t &major_version,
      uint64_t &tenant_id,
      const int64_t timeout_us) = 0;

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

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

  /*
   * Stop libobcdc
   */

  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;
};

}
}