/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */

#include "baseconsumer.h"

#include <signal.h>
#include <unistd.h>

#include <sstream>

#include "client_service.h"
#include "const_config.h"
#include "const_rpc.h"
#include "logger.h"
#include "singleton.h"
#include "transport.h"
#include "tubemq/tubemq_config.h"
#include "tubemq/tubemq_errcode.h"
#include "tubemq_transport.h"
#include "utils.h"
#include "version.h"

namespace tubemq {

using std::lock_guard;
using std::stringstream;

BaseConsumer::BaseConsumer() : BaseClient(false) {
  status_.Set(0);
  unreport_times_ = 0;
  visit_token_.Set(tb_config::kInvalidValue);
  nextauth_2_master.Set(false);
  nextauth_2_broker.Set(false);
  masters_map_.clear();
  is_master_actived_.Set(false);
  master_reg_status_.Set(0);
  master_hb_status_.Set(0);
  last_master_hbtime_ = 0;
  master_sh_retry_cnt_ = 0;
}

BaseConsumer::~BaseConsumer() {
  //
}

bool BaseConsumer::Start(string& err_info, const ConsumerConfig& config) {
  ConsumerConfig tmp_config;
  if (!status_.CompareAndSet(0, 1)) {
    err_info = "Ok";
    return true;
  }
  // check configure
  if (config.GetGroupName().length() == 0 || config.GetMasterAddrInfo().length() == 0) {
    err_info = "Parameter error: not set master address info or group name!";
    return false;
  }
  //
  if (!TubeMQService::Instance()->IsRunning()) {
    err_info = "TubeMQ Service not startted!";
    return false;
  }
  if (!TubeMQService::Instance()->AddClientObj(err_info, shared_from_this())) {
    client_index_ = tb_config::kInvalidValue;
    status_.CompareAndSet(1, 0);
    return false;
  }
  config_ = config;
  if (!initMasterAddress(err_info, config.GetMasterAddrInfo())) {
    TubeMQService::Instance()->RmvClientObj(shared_from_this());
    return false;
  }
  client_uuid_ = buildUUID();
  sub_info_.SetConsumeTarget(config_);
  rmtdata_cache_.SetConsumerInfo(client_uuid_, config_.GetGroupName());
  // initial resource

  // register to master
  int32_t error_code;
  if (!register2Master(error_code, err_info, false)) {
    TubeMQService::Instance()->RmvClientObj(shared_from_this());
    status_.CompareAndSet(1, 0);
    return false;
  }
  status_.CompareAndSet(1, 2);
  heart_beat_timer_ = TubeMQService::Instance()->CreateTimer();
  heart_beat_timer_->expires_after(std::chrono::milliseconds(config_.GetHeartbeatPeriodMs() / 2));
  auto self = shared_from_this();
  heart_beat_timer_->async_wait([self, this](const std::error_code& ec) {
    if (ec) {
      return;
    }
    heartBeat2Master();
  });
  rebalance_thread_ptr_ =
      std::make_shared<std::thread>([self, this]() { processRebalanceEvent(); });
  LOG_INFO("[CONSUMER] start consumer success, client=%s", client_uuid_.c_str());
  err_info = "Ok";
  return true;
}

void BaseConsumer::ShutDown() {
  if (!status_.CompareAndSet(2, 0)) {
    return;
  }
  LOG_INFO("[CONSUMER] ShutDown consumer begin, client=%s", client_uuid_.c_str());
  // 1. exist rebalance thread
  ConsumerEvent empty_event;
  rmtdata_cache_.OfferEvent(empty_event);
  // 2. close to master
  close2Master();
  // 3. close all brokers
  closeAllBrokers();
  // 4. check master hb thread status
  int check_count = 0;
  while (master_hb_status_.Get() != 0 || master_reg_status_.Get() != 0) {
    std::this_thread::sleep_for(std::chrono::milliseconds(40));
    if (++check_count % 10 == 0) {
      if (check_count >= 1000) {
          LOG_INFO("[CONSUMER] Found hb[%d] or reg[%d] not zero, count=%d, exit, client=%s",
                   master_hb_status_.Get(), master_reg_status_.Get(), check_count, client_uuid_.c_str());
        break;
      } else {
        LOG_INFO("[CONSUMER] Found hb[%d] or reg[%d] not zero, count=%d, continue, client=%s",
                 master_hb_status_.Get(), master_reg_status_.Get(), check_count, client_uuid_.c_str());
      }
    }
  }
  // 5. join hb thread;
  rebalance_thread_ptr_->join();
  heart_beat_timer_ = nullptr;
  rebalance_thread_ptr_ = nullptr;
  // 6. remove client stub
  TubeMQService::Instance()->RmvClientObj(shared_from_this());
  client_index_ = tb_config::kInvalidValue;
  LOG_INFO("[CONSUMER] ShutDown consumer finished, client=%s", client_uuid_.c_str());
}

bool BaseConsumer::GetMessage(ConsumerResult& result) {
  int32_t error_code;
  string err_info;
  PartitionExt partition_ext;
  string confirm_context;

  if (!IsConsumeReady(result)) {
    return false;
  }
  if (!rmtdata_cache_.SelectPartition(error_code, err_info, partition_ext, confirm_context)) {
    result.SetFailureResult(error_code, err_info);
    return false;
  }
  int64_t curr_offset = tb_config::kInvalidValue;
  bool filter_consume = sub_info_.IsFilterConsume(partition_ext.GetTopic());
  PeerInfo peer_info(partition_ext.GetBrokerHost(), partition_ext.GetPartitionId(),
                     partition_ext.GetPartitionKey(), curr_offset);
  auto request = std::make_shared<RequestContext>();
  TubeMQCodec::ReqProtocolPtr req_protocol = TubeMQCodec::GetReqProtocol();
  // build getmessage request
  buidGetMessageC2B(partition_ext, req_protocol);
  request->codec_ = std::make_shared<TubeMQCodec>();
  request->ip_ = partition_ext.GetBrokerHost();
  request->port_ = partition_ext.GetBrokerPort();
  request->timeout_ = config_.GetRpcReadTimeoutMs();
  request->request_id_ = Singleton<UniqueSeqId>::Instance().Next();
  req_protocol->request_id_ = request->request_id_;
  req_protocol->rpc_read_timeout_ = config_.GetRpcReadTimeoutMs() - 500;
  // send message to target
  ResponseContext response_context;
  ErrorCode error = SyncRequest(response_context, request, req_protocol);
  if (!TubeMQService::Instance()->IsRunning()) {
    result.SetFailureResult(err_code::kErrMQServiceStop, "TubeMQ Service stopped!");
    return false;
  }
  if (!isClientRunning()) {
    result.SetFailureResult(err_code::kErrClientStop, "TubeMQ Client stopped!");
    return false;
  }
  if (error.Value() == err_code::kErrSuccess) {
    // process response
    auto rsp = any_cast<TubeMQCodec::RspProtocolPtr>(response_context.rsp_);
    processGetMessageRspB2C(result, peer_info, filter_consume, partition_ext, confirm_context, rsp);
    return result.IsSuccess();
  } else {
    rmtdata_cache_.RelPartition(err_info, filter_consume, confirm_context, false);
    result.SetFailureResult(error.Value(), error.Message(), partition_ext.GetTopic(), peer_info);
    return false;
  }
}

bool BaseConsumer::IsConsumeReady(ConsumerResult& result) {
  int32_t ret_code;
  int64_t start_time = Utils::GetCurrentTimeMillis();
  while (true) {
    if (!TubeMQService::Instance()->IsRunning()) {
      result.SetFailureResult(err_code::kErrMQServiceStop, "TubeMQ Service stopped!");
      return false;
    }
    if (!isClientRunning()) {
      result.SetFailureResult(err_code::kErrClientStop, "TubeMQ Client stopped!");
      return false;
    }
    ret_code = rmtdata_cache_.GetCurConsumeStatus();
    if (err_code::kErrSuccess == ret_code) {
      return true;
    }
    if ((config_.GetMaxPartCheckPeriodMs() >= 0) &&
        (Utils::GetCurrentTimeMillis() - start_time >= config_.GetMaxPartCheckPeriodMs())) {
      switch (ret_code) {
        case err_code::kErrNoPartAssigned: {
          result.SetFailureResult(ret_code,
                                  "No partition info in local cache, please retry later!");
        } break;

        case err_code::kErrAllPartInUse: {
          result.SetFailureResult(ret_code, "No idle partition to consume, please retry later!");
        } break;

        case err_code::kErrAllPartWaiting:
        default: {
          result.SetFailureResult(ret_code,
                                  "All partitions reach max position, please retry later!");
        } break;
      }
      return false;
    }
    std::this_thread::sleep_for(std::chrono::milliseconds(config_.GetPartCheckSliceMs()));
  }
  return true;
}

bool BaseConsumer::GetCurConsumedInfo(map<string, ConsumeOffsetInfo>& consume_info_map) {
  bool has_data = false;
  consume_info_map.clear();
  map<string, int64_t> part_offset_map;
  map<string, int64_t>::iterator it_part;
  rmtdata_cache_.GetCurPartitionOffsets(part_offset_map);
  for (it_part = part_offset_map.begin(); it_part != part_offset_map.end(); ++it_part) {
    ConsumeOffsetInfo tmp_info(it_part->first, it_part->second);
    consume_info_map[it_part->first] = tmp_info;
    has_data = true;
  }
  return has_data;
}

bool BaseConsumer::Confirm(const string& confirm_context, bool is_consumed,
                           ConsumerResult& result) {
  if (!TubeMQService::Instance()->IsRunning()) {
    result.SetFailureResult(err_code::kErrMQServiceStop, "TubeMQ Service stopped!");
    return false;
  }
  if (!isClientRunning()) {
    result.SetFailureResult(err_code::kErrClientStop, "TubeMQ Client stopped!");
    return false;
  }
  string token1 = delimiter::kDelimiterAt;
  string token2 = delimiter::kDelimiterColon;
  string::size_type pos1, pos2;
  pos1 = confirm_context.find(token1);
  if (string::npos == pos1) {
    result.SetFailureResult(
        err_code::kErrBadRequest,
        "Illegel confirm_context content: unregular confirm_context value format!");
    return false;
  }
  string part_key = Utils::Trim(confirm_context.substr(0, pos1));
  string booked_time_str =
      Utils::Trim(confirm_context.substr(pos1 + token1.size(), confirm_context.size()));
  int64_t booked_time = atol(booked_time_str.c_str());
  pos1 = part_key.find(token2);
  if (string::npos == pos1) {
    result.SetFailureResult(err_code::kErrBadRequest,
                            "Illegel confirm_context content: unregular index key value format!");
    return false;
  }
  pos1 = pos1 + token1.size();
  string topic_name = part_key.substr(pos1);
  pos2 = topic_name.rfind(token2);
  if (string::npos == pos2) {
    result.SetFailureResult(
        err_code::kErrBadRequest,
        "Illegel confirm_context content: unregular index's topic key value format!");
    return false;
  }
  topic_name = topic_name.substr(0, pos2);
  if (!rmtdata_cache_.IsPartitionInUse(part_key, booked_time)) {
    result.SetFailureResult(err_code::kErrConfirmTimeout, "The confirm_context's value invalid!");
    return false;
  }
  PartitionExt partition_ext;
  bool ret_result = rmtdata_cache_.GetPartitionExt(part_key, partition_ext);
  if (!ret_result) {
    result.SetFailureResult(err_code::kErrConfirmTimeout,
                            "Not found the partition by confirm_context!");
    return false;
  }
  int64_t curr_offset = tb_config::kInvalidValue;
  PeerInfo peer_info(partition_ext.GetBrokerHost(), partition_ext.GetPartitionId(),
                     partition_ext.GetPartitionKey(), curr_offset);
  auto request = std::make_shared<RequestContext>();
  TubeMQCodec::ReqProtocolPtr req_protocol = TubeMQCodec::GetReqProtocol();
  // build CommitC2B request
  buidCommitC2B(partition_ext, is_consumed, req_protocol);
  request->codec_ = std::make_shared<TubeMQCodec>();
  request->ip_ = partition_ext.GetBrokerHost();
  request->port_ = partition_ext.GetBrokerPort();
  request->timeout_ = config_.GetRpcReadTimeoutMs();
  request->request_id_ = Singleton<UniqueSeqId>::Instance().Next();
  req_protocol->request_id_ = request->request_id_;
  req_protocol->rpc_read_timeout_ = config_.GetRpcReadTimeoutMs() - 500;
  // send message to target
  ResponseContext response_context;
  ErrorCode error = SyncRequest(response_context, request, req_protocol);
  if (!TubeMQService::Instance()->IsRunning()) {
    result.SetFailureResult(err_code::kErrMQServiceStop, "TubeMQ Service stopped!");
    return false;
  }
  if (!isClientRunning()) {
    result.SetFailureResult(err_code::kErrClientStop, "TubeMQ Client stopped!");
    return false;
  }
  if (error.Value() == err_code::kErrSuccess) {
    // process response
    auto rsp = any_cast<TubeMQCodec::RspProtocolPtr>(response_context.rsp_);
    if (rsp->success_) {
      CommitOffsetResponseB2C rsp_b2c;
      ret_result = rsp_b2c.ParseFromArray(rsp->rsp_body_.data().c_str(),
                                          (int32_t)(rsp->rsp_body_.data().length()));
      if (ret_result) {
        if (rsp_b2c.success()) {
          curr_offset = rsp_b2c.curroffset();
          peer_info.SetCurrOffset(curr_offset);
          result.SetSuccessResult(err_code::kErrSuccess, topic_name, peer_info);
        } else {
          result.SetFailureResult(rsp_b2c.errcode(), rsp_b2c.errmsg(), topic_name, peer_info);
        }
      } else {
        result.SetFailureResult(err_code::kErrParseFailure,
                                "Parse CommitOffsetResponseB2C response failure!", topic_name,
                                peer_info);
      }
    } else {
      result.SetFailureResult(rsp->code_, rsp->error_msg_, topic_name, peer_info);
    }
  } else {
    result.SetFailureResult(error.Value(), error.Message(), topic_name, peer_info);
  }
  string err_info;
  rmtdata_cache_.BookedPartionInfo(part_key, curr_offset);
  rmtdata_cache_.RelPartition(err_info, sub_info_.IsFilterConsume(topic_name), confirm_context,
                              is_consumed);
  return result.IsSuccess();
}

bool BaseConsumer::register2Master(int32_t& error_code, string& err_info, bool need_change) {
  string target_ip;
  int target_port;

  // set regist process status to begin
  if (!master_reg_status_.CompareAndSet(0, 1)) {
    err_info = "register2Master process has began!";
    return false;
  }

  LOG_INFO("register2Master process begin: ");
  // check client status
  if (status_.Get() == 0) {
    master_reg_status_.CompareAndSet(1, 0);
    err_info = "Consumer not startted!";
    return false;
  }
  LOG_DEBUG("[CONSUMER], initial register2master request, clientId=%s", client_uuid_.c_str());
  // get master address and port
  if (need_change) {
    getNextMasterAddr(target_ip, target_port);
  } else {
    getCurrentMasterAddr(target_ip, target_port);
  }
  bool result = false;
  int retry_count = 0;
  int maxRetrycount = masters_map_.size();
  err_info = "Master register failure, no online master service!";
  while (retry_count < maxRetrycount) {
    if (!TubeMQService::Instance()->IsRunning()) {
      err_info = "TubeMQ Service is stopped!";
      master_reg_status_.CompareAndSet(1, 0);
      return false;
    }
    auto request = std::make_shared<RequestContext>();
    TubeMQCodec::ReqProtocolPtr req_protocol = TubeMQCodec::GetReqProtocol();
    // build register request
    buidRegisterRequestC2M(req_protocol);
    // set parameters
    request->codec_ = std::make_shared<TubeMQCodec>();
    request->ip_ = target_ip;
    request->port_ = target_port;
    request->timeout_ = config_.GetRpcReadTimeoutMs();
    request->request_id_ = Singleton<UniqueSeqId>::Instance().Next();
    req_protocol->request_id_ = request->request_id_;
    req_protocol->rpc_read_timeout_ = config_.GetRpcReadTimeoutMs() - 500;
    // send message to target
    ResponseContext response_context;
    ErrorCode error = SyncRequest(response_context, request, req_protocol);
    LOG_INFO("register2Master response come, error.value is %d", error.Value());
    if (error.Value() == err_code::kErrSuccess) {
      // process response
      auto rsp = any_cast<TubeMQCodec::RspProtocolPtr>(response_context.rsp_);
      result = processRegisterResponseM2C(error_code, err_info, rsp);
      if (result) {
        err_info = "Ok";
        is_master_actived_.Set(true);
        last_master_hbtime_ = Utils::GetCurrentTimeMillis();
        break;
      } else {
        is_master_actived_.Set(false);
      }
    } else {
      error_code = error.Value();
      err_info = error.Message();
    }
    if (error_code == err_code::kErrConsumeGroupForbidden ||
        error_code == err_code::kErrConsumeContentForbidden) {
      // set regist process status to existed
      master_reg_status_.CompareAndSet(1, 0);
      LOG_WARN("[CONSUMER] register2master(%s:%d) failure exist register, client=%s,reason:%s",
               target_ip.c_str(), target_port, client_uuid_.c_str(), err_info.c_str());
      return false;
    } else {
      LOG_WARN(
          "[CONSUMER] register2master(%s:%d) failure, client=%s, retrycount=(%d-%d), reason:%s",
          target_ip.c_str(), target_port, client_uuid_.c_str(), maxRetrycount, retry_count + 1,
          err_info.c_str());
    }
    retry_count++;
    getNextMasterAddr(target_ip, target_port);
  }
  // set regist process status to existed
  master_reg_status_.CompareAndSet(1, 0);
  LOG_INFO("[CONSUMER] register2Master finished, client=%s, result:%d, err_info:%s",
           client_uuid_.c_str(), result, err_info.c_str());
  return result;
}

void BaseConsumer::asyncRegister2Master(bool need_change) {
  auto self = shared_from_this();
  TubeMQService::Instance()->Post([self, this, need_change]() {
    int32_t error_code;
    string error_info;
    if (!is_master_actived_.Get()) {
      auto ret_result = register2Master(error_code, error_info, need_change);
      LOG_INFO("[CONSUMER] asyncRegister2Master ret_result:%d, master_sh_retry_cnt_:%d", ret_result,
               master_sh_retry_cnt_);
      if (ret_result) {
        is_master_actived_.Set(true);
        master_sh_retry_cnt_ = 0;
      } else {
        master_sh_retry_cnt_++;
      }
    }
    heart_beat_timer_->expires_after(std::chrono::milliseconds(nextHeartBeatPeriodms()));
    heart_beat_timer_->async_wait([self, this](const std::error_code& ec) {
      if (ec) {
        return;
      }
      heartBeat2Master();
    });
  });
}

void BaseConsumer::heartBeat2Master() {
  // timer task
  // 1. check if need re-register, if true, first call register
  // 2. call heartbeat to master
  // 3. process response
  // 4. call timer again
  string target_ip;
  int target_port;

  // set heartbeat process status to begin
  if (!master_hb_status_.CompareAndSet(0, 1)) {
    LOG_INFO("check hb process status, heartBeat2Master process has began!");
    return;
  }
  if (!TubeMQService::Instance()->IsRunning()) {
    master_hb_status_.CompareAndSet(1, 0);
    LOG_INFO("[CONSUMER] heartBeat2Master failure: TubeMQ Service is stopped! client=%s",
             client_uuid_.c_str());
    return;
  }

  if (!isClientRunning()) {
    master_hb_status_.CompareAndSet(1, 0);
    LOG_INFO("[CONSUMER] heartBeat2Master failure: TubeMQ Client stopped! client=%s",
             client_uuid_.c_str());
    return;
  }
  // check status in master
  // if not actived first register, or send heartbeat
  if (!is_master_actived_.Get()) {
    LOG_INFO("[CONSUMER] heartBeat2Master found master not active, re-register first! client=%s",
             client_uuid_.c_str());
    asyncRegister2Master(false);
    master_hb_status_.CompareAndSet(1, 0);
    return;
  }
  // check partition status
  if (Utils::GetCurrentTimeMillis() - last_master_hbtime_ > 30000) {
    rmtdata_cache_.handleExpiredPartitions(config_.GetMaxConfirmWaitPeriodMs());
  }
  // select current master
  getCurrentMasterAddr(target_ip, target_port);
  auto request = std::make_shared<RequestContext>();
  TubeMQCodec::ReqProtocolPtr req_protocol = TubeMQCodec::GetReqProtocol();
  // build heartbeat 2 master request
  buidHeartRequestC2M(req_protocol);
  request->codec_ = std::make_shared<TubeMQCodec>();
  request->ip_ = target_ip;
  request->port_ = target_port;
  request->timeout_ = config_.GetRpcReadTimeoutMs();
  request->request_id_ = Singleton<UniqueSeqId>::Instance().Next();
  req_protocol->request_id_ = request->request_id_;
  req_protocol->rpc_read_timeout_ = config_.GetRpcReadTimeoutMs() - 500;
  // send message to target
  AsyncRequest(request, req_protocol)
      .AddCallBack([=](ErrorCode error, const ResponseContext& response_context) {
        if (GetClientIndex() == tb_config::kInvalidValue ||
          !TubeMQService::Instance()->IsRunning() ||
          !isClientRunning()) {
          master_hb_status_.CompareAndSet(1, 0);
          return;
        }
        if (error.Value() != err_code::kErrSuccess) {
          master_sh_retry_cnt_++;
          LOG_WARN("[CONSUMER] heartBeat2Master failue to (%s:%d) : %s, client=%s",
                   target_ip.c_str(), target_port, error.Message().c_str(), client_uuid_.c_str());
          if (master_sh_retry_cnt_ >= tb_config::kMaxMasterHBRetryCount) {
            LOG_WARN("[CONSUMER] heartBeat2Master found over max-hb-retry(%d), client=%s",
                     master_sh_retry_cnt_, client_uuid_.c_str());
            master_sh_retry_cnt_ = 0;
            is_master_actived_.Set(false);
            asyncRegister2Master(true);
            master_hb_status_.CompareAndSet(1, 0);
            return;
          }
        } else {
          // process response
          auto rsp = any_cast<TubeMQCodec::RspProtocolPtr>(response_context.rsp_);
          int32_t error_code = 0;
          std::string error_info;
          auto ret_result = processHBResponseM2C(error_code, error_info, rsp);
          if (ret_result) {
            is_master_actived_.Set(true);
            master_sh_retry_cnt_ = 0;
          } else {
            master_sh_retry_cnt_++;
            if (error_code == err_code::kErrHbNoNode ||
                error_info.find("StandbyException") != string::npos) {
              is_master_actived_.Set(false);
              bool need_change = !(error_code == err_code::kErrHbNoNode);
              LOG_WARN("[CONSUMER] heartBeat2Master found no-node or standby, client=%s, change=%d",
                       client_uuid_.c_str(), need_change);
              asyncRegister2Master(need_change);
              master_hb_status_.CompareAndSet(1, 0);
              return;
            }
          }
        }
        if (GetClientIndex() == tb_config::kInvalidValue ||
          !TubeMQService::Instance()->IsRunning() ||
          !isClientRunning()) {
          master_hb_status_.CompareAndSet(1, 0);
          return;
        }
        heart_beat_timer_->expires_after(std::chrono::milliseconds(nextHeartBeatPeriodms()));
        auto self = shared_from_this();
        heart_beat_timer_->async_wait([self, this](const std::error_code& ec) {
          if (ec) {
            return;
          }
          heartBeat2Master();
        });
        master_hb_status_.CompareAndSet(1, 0);
      });
  return;
}

int32_t BaseConsumer::nextHeartBeatPeriodms() {
  int32_t next_hb_periodms = config_.GetHeartbeatPeriodMs();
  if (master_sh_retry_cnt_ >= config_.GetMaxHeartBeatRetryTimes()) {
    next_hb_periodms = config_.GetHeartbeatPeriodAftFailMs();
  }
  return next_hb_periodms;
}

void BaseConsumer::processRebalanceEvent() {
  // thread wait until event come
  LOG_INFO("[CONSUMER] rebalance event Handler startted!");
  while (true) {
    if (!TubeMQService::Instance()->IsRunning()) {
      LOG_INFO("[CONSUMER] Rebalance found Service stopped, existed, client=%s",
               client_uuid_.c_str());
      break;
    }
    if (!isClientRunning()) {
      LOG_INFO("[CONSUMER] Rebalance found Client stopped, existed, client=%s",
               client_uuid_.c_str());
      break;
    }
    ConsumerEvent event;
    rmtdata_cache_.TakeEvent(event);
    if (event.GetEventStatus() == tb_config::kInvalidValue &&
        event.GetRebalanceId() == tb_config::kInvalidValue) {
      LOG_INFO("[CONSUMER] Rebalance found Shutdown notify, existed, client=%s",
               client_uuid_.c_str());
      break;
    }
    rmtdata_cache_.ClearEvent();
    switch (event.GetEventType()) {
      case 2:
      case 20: {
        processDisConnect2Broker(event);
        rmtdata_cache_.OfferEventResult(event);
      } break;

      case 1:
      case 10: {
        processConnect2Broker(event);
        rmtdata_cache_.OfferEventResult(event);
      } break;

      default: {
        //
      } break;
    }
  }

  LOG_INFO("[CONSUMER] rebalance event Handler stopped!");
  return;
}

void BaseConsumer::close2Master() {
  string target_ip;
  int target_port;

  LOG_INFO("[CONSUMER] close2Master begin, clientid=%s", client_uuid_.c_str());
  getCurrentMasterAddr(target_ip, target_port);
  auto request = std::make_shared<RequestContext>();
  TubeMQCodec::ReqProtocolPtr req_protocol = TubeMQCodec::GetReqProtocol();
  // build close2master request
  buidCloseRequestC2M(req_protocol);
  request->codec_ = std::make_shared<TubeMQCodec>();
  request->ip_ = target_ip;
  request->port_ = target_port;
  request->timeout_ = config_.GetRpcReadTimeoutMs();
  request->request_id_ = Singleton<UniqueSeqId>::Instance().Next();
  req_protocol->request_id_ = request->request_id_;
  req_protocol->rpc_read_timeout_ = config_.GetRpcReadTimeoutMs() - 500;
  // send message to target
  AsyncRequest(request, req_protocol);
  LOG_INFO("[CONSUMER] close2Master finished, clientid=%s", client_uuid_.c_str());
  // not need wait response
  return;
}

void BaseConsumer::processConnect2Broker(ConsumerEvent& event) {
  bool ret_result;
  int32_t error_code;
  string error_info;
  list<PartitionExt> subscribed_partitions;
  list<PartitionExt> unsub_partitions;
  list<PartitionExt>::iterator it;
  LOG_TRACE("[processConnect2Broker] connect event begin, clientid=%s", client_uuid_.c_str());
  if (!isClientRunning()) {
    return;
  }
  list<SubscribeInfo> subscribe_info = event.GetSubscribeInfoList();
  if (!subscribe_info.empty()) {
    rmtdata_cache_.FilterPartitions(subscribe_info, subscribed_partitions, unsub_partitions);
    if (!unsub_partitions.empty()) {
      for (it = unsub_partitions.begin(); it != unsub_partitions.end(); it++) {
        if (!isClientRunning()) {
          LOG_TRACE("[processConnect2Broker] client stopped, break pos1, clientid=%s", client_uuid_.c_str());
          break;
        }
        LOG_TRACE("[processConnect2Broker] connect to %s, clientid=%s",
                  it->GetPartitionKey().c_str(), client_uuid_.c_str());
        auto request = std::make_shared<RequestContext>();
        TubeMQCodec::ReqProtocolPtr req_protocol = TubeMQCodec::GetReqProtocol();
        // build close2master request
        buidRegisterRequestC2B(*it, req_protocol);
        request->codec_ = std::make_shared<TubeMQCodec>();
        request->ip_ = it->GetBrokerHost();
        request->port_ = it->GetBrokerPort();
        request->timeout_ = config_.GetRpcReadTimeoutMs();
        request->request_id_ = Singleton<UniqueSeqId>::Instance().Next();
        req_protocol->request_id_ = request->request_id_;
        req_protocol->rpc_read_timeout_ = config_.GetRpcReadTimeoutMs() - 500;
        // send message to target
        ResponseContext response_context;
        ErrorCode error = SyncRequest(response_context, request, req_protocol);
        if (!isClientRunning()) {
          LOG_TRACE("[processConnect2Broker] client stopped, break pos2, clientid=%s", client_uuid_.c_str());
          break;
        }
        if (error.Value() != err_code::kErrSuccess) {
          LOG_WARN("[Connect2Broker] request network failure to (%s:%d) : %s",
                   it->GetBrokerHost().c_str(), it->GetBrokerPort(), error.Message().c_str());
        } else {
          // process response
          auto rsp = any_cast<TubeMQCodec::RspProtocolPtr>(response_context.rsp_);
          ret_result = processRegResponseB2C(error_code, error_info, rsp);
          if (ret_result) {
            rmtdata_cache_.AddNewPartition(*it);
            addBrokerHBTimer(it->GetBrokerInfo());
          }
        }
      }
    }
  }
  sub_info_.BookRegistered();
  event.SetEventStatus(2);
  LOG_TRACE("[processConnect2Broker] connect event finished, clientid=%s", client_uuid_.c_str());
}

void BaseConsumer::processDisConnect2Broker(ConsumerEvent& event) {
  LOG_TRACE("[processDisConnect2Broker] begin to process disConnect event, clientid=%s",
            client_uuid_.c_str());
  if (!isClientRunning()) {
    return;
  }
  list<SubscribeInfo> subscribe_info = event.GetSubscribeInfoList();
  if (!subscribe_info.empty()) {
    map<NodeInfo, list<PartitionExt> > rmv_partitions;
    rmtdata_cache_.RemoveAndGetPartition(subscribe_info, config_.IsRollbackIfConfirmTimeout(),
                                         rmv_partitions);
    if (!rmv_partitions.empty()) {
      unregister2Brokers(rmv_partitions, true);
    }
  }
  event.SetEventStatus(2);
  LOG_TRACE("[processDisConnect2Broker] out disConnect event process, clientid=%s",
            client_uuid_.c_str());
  return;
}

void BaseConsumer::closeAllBrokers() {
  map<NodeInfo, list<PartitionExt> > broker_parts;
  LOG_INFO("[CONSUMER] closeAllBrokers begin, clientid=%s", client_uuid_.c_str());
  rmtdata_cache_.GetAllClosedBrokerParts(broker_parts);
  if (!broker_parts.empty()) {
    unregister2Brokers(broker_parts, false);
  }
  LOG_INFO("[CONSUMER] closeAllBrokers end, clientid=%s", client_uuid_.c_str());
}

void BaseConsumer::processHeartBeat2Broker(NodeInfo broker_info) {
  if (!isClientRunning()) {
    return;
  }
  list<PartitionExt> partition_list;
  list<PartitionExt>::iterator it;
  rmtdata_cache_.GetPartitionByBroker(broker_info, partition_list);
  if (partition_list.empty()) {
    reSetBrokerHBTimer(broker_info);
    return;
  }
  set<string> req_part_keys;
  for (it = partition_list.begin(); it != partition_list.end(); ++it) {
    req_part_keys.insert(it->GetPartitionKey());
  }
  auto request = std::make_shared<RequestContext>();
  TubeMQCodec::ReqProtocolPtr req_protocol = TubeMQCodec::GetReqProtocol();
  // build heartbeat2broker request
  buidHeartBeatC2B(partition_list, req_protocol);
  request->codec_ = std::make_shared<TubeMQCodec>();
  request->ip_ = broker_info.GetHost();
  request->port_ = broker_info.GetPort();
  request->timeout_ = config_.GetRpcReadTimeoutMs();
  request->request_id_ = Singleton<UniqueSeqId>::Instance().Next();
  req_protocol->request_id_ = request->request_id_;
  req_protocol->rpc_read_timeout_ = config_.GetRpcReadTimeoutMs() - 500;
  // send message to target
  AsyncRequest(request, req_protocol)
      .AddCallBack([=](ErrorCode error, const ResponseContext& response_context) {
        if (GetClientIndex() == tb_config::kInvalidValue ||
          !TubeMQService::Instance()->IsRunning() ||
          !isClientRunning()) {
          return;
        }
        if (error.Value() != err_code::kErrSuccess) {
          LOG_WARN("[Heartbeat2Broker] request network  to failure (%s), ression is %s",
                   broker_info.GetAddrInfo().c_str(), error.Message().c_str());
        } else {
          // process response
          auto rsp = any_cast<TubeMQCodec::RspProtocolPtr>(response_context.rsp_);
          if (rsp->success_) {
            HeartBeatResponseB2C rsp_b2c;
            bool result = rsp_b2c.ParseFromArray(rsp->rsp_body_.data().c_str(),
                                                 (int32_t)(rsp->rsp_body_.data().length()));
            if (result) {
              set<string> partition_keys;
              if (rsp_b2c.success()) {
                if (rsp_b2c.has_haspartfailure() && rsp_b2c.haspartfailure()) {
                  for (int tmp_i = 0; tmp_i < rsp_b2c.failureinfo_size(); tmp_i++) {
                    string token_key = delimiter::kDelimiterColon;
                    string fullpart_str = rsp_b2c.failureinfo(tmp_i);
                    string::size_type pos1 = fullpart_str.find(token_key);
                    if (pos1 == string::npos) {
                      continue;
                    }
                    // int error_code = atoi(fullpart_str.substr(0, pos1).c_str());
                    string part_str =
                        fullpart_str.substr(pos1 + token_key.size(), fullpart_str.size());
                    Partition part(part_str);
                    partition_keys.insert(part.GetPartitionKey());
                    LOG_TRACE("[Heartbeat2Broker] found partiton(%s) hb failure!",
                              part.GetPartitionKey().c_str());
                  }
                }
                rmtdata_cache_.RemovePartition(partition_keys);
              } else {
                if (rsp_b2c.errcode() == err_code::kErrCertificateFailure) {
                  rmtdata_cache_.RemovePartition(req_part_keys);
                  LOG_WARN("[Heartbeat2Broker] request (%s) CertificateFailure",
                           broker_info.GetAddrInfo().c_str());
                }
              }
            }
          }
        }
        LOG_TRACE("[Heartbeat2Broker] out hb response process, add broker(%s) timer!",
                  broker_info.GetAddrInfo().c_str());
        reSetBrokerHBTimer(broker_info);
      });
}

void BaseConsumer::buidRegisterRequestC2M(TubeMQCodec::ReqProtocolPtr& req_protocol) {
  string reg_msg;
  RegisterRequestC2M c2m_request;
  list<string>::iterator it_topics;
  list<SubscribeInfo>::iterator it_sub;
  c2m_request.set_clientid(client_uuid_);
  c2m_request.set_hostname(TubeMQService::Instance()->GetLocalHost());
  c2m_request.set_requirebound(sub_info_.IsBoundConsume());
  c2m_request.set_groupname(config_.GetGroupName());
  c2m_request.set_sessiontime(sub_info_.GetSubscribedTime());
  // subscribed topic list
  list<string> sub_topics = sub_info_.GetSubTopics();
  for (it_topics = sub_topics.begin(); it_topics != sub_topics.end(); ++it_topics) {
    c2m_request.add_topiclist(*it_topics);
  }
  c2m_request.set_defflowcheckid(rmtdata_cache_.GetDefFlowCtrlId());
  c2m_request.set_groupflowcheckid(rmtdata_cache_.GetGroupFlowCtrlId());
  c2m_request.set_qrypriorityid(rmtdata_cache_.GetGroupQryPriorityId());
  // reported subscribed info
  list<SubscribeInfo> subscribe_lst;
  rmtdata_cache_.GetSubscribedInfo(subscribe_lst);
  for (it_sub = subscribe_lst.begin(); it_sub != subscribe_lst.end(); ++it_sub) {
    c2m_request.add_subscribeinfo(it_sub->ToString());
  }
  // get topic conditions
  list<string> topic_conds = sub_info_.GetTopicConds();
  for (it_topics = topic_conds.begin(); it_topics != topic_conds.end(); ++it_topics) {
    c2m_request.add_topiccondition(*it_topics);
  }
  // add bound consume info
  if (sub_info_.IsBoundConsume()) {
    c2m_request.set_sessionkey(sub_info_.GetSessionKey());
    c2m_request.set_selectbig(sub_info_.SelectBig());
    c2m_request.set_totalcount(sub_info_.GetSourceCnt());
    c2m_request.set_requiredpartition(sub_info_.GetBoundPartInfo());
    c2m_request.set_notallocated(sub_info_.IsNotAllocated());
  }
  // authenticate info
  if (needGenMasterCertificateInfo(true)) {
    MasterCertificateInfo* pmst_certinfo = c2m_request.mutable_authinfo();
    AuthenticateInfo* pauthinfo = pmst_certinfo->mutable_authinfo();
    genMasterAuthenticateToken(pauthinfo, config_.GetUsrName(), config_.GetUsrPassWord());
  }
  //
  c2m_request.SerializeToString(&reg_msg);
  req_protocol->method_id_ = rpc_config::kMasterMethoddConsumerRegister;
  req_protocol->prot_msg_ = reg_msg;
}

void BaseConsumer::buidHeartRequestC2M(TubeMQCodec::ReqProtocolPtr& req_protocol) {
  string hb_msg;
  HeartRequestC2M c2m_request;
  list<string>::iterator it_topics;
  list<SubscribeInfo>::iterator it_sub;
  c2m_request.set_clientid(client_uuid_);
  c2m_request.set_groupname(config_.GetGroupName());
  c2m_request.set_defflowcheckid(rmtdata_cache_.GetDefFlowCtrlId());
  c2m_request.set_groupflowcheckid(rmtdata_cache_.GetGroupFlowCtrlId());
  c2m_request.set_qrypriorityid(rmtdata_cache_.GetGroupQryPriorityId());
  c2m_request.set_reportsubscribeinfo(false);
  ConsumerEvent event;
  list<SubscribeInfo>::iterator it;
  list<SubscribeInfo> subscribe_info_lst;
  bool has_event = rmtdata_cache_.PollEventResult(event);
  // judge if report subscribe info
  if ((has_event) || (++unreport_times_ > config_.GetMaxSubinfoReportIntvl())) {
    unreport_times_ = 0;
    c2m_request.set_reportsubscribeinfo(true);
    rmtdata_cache_.GetSubscribedInfo(subscribe_info_lst);
    if (has_event) {
      EventProto* event_proto = c2m_request.mutable_event();
      event_proto->set_rebalanceid(event.GetRebalanceId());
      event_proto->set_optype(event.GetEventType());
      event_proto->set_status(event.GetEventStatus());
      list<SubscribeInfo> event_sub = event.GetSubscribeInfoList();
      for (it = event_sub.begin(); it != event_sub.end(); it++) {
        event_proto->add_subscribeinfo(it->ToString());
      }
    }
    if (!subscribe_info_lst.empty()) {
      for (it = subscribe_info_lst.begin(); it != subscribe_info_lst.end(); it++) {
        c2m_request.add_subscribeinfo(it->ToString());
      }
    }
  }
  if (needGenMasterCertificateInfo(true)) {
    MasterCertificateInfo* pmst_certinfo = c2m_request.mutable_authinfo();
    AuthenticateInfo* pauthinfo = pmst_certinfo->mutable_authinfo();
    genMasterAuthenticateToken(pauthinfo, config_.GetUsrName(), config_.GetUsrPassWord());
  }
  c2m_request.SerializeToString(&hb_msg);
  req_protocol->method_id_ = rpc_config::kMasterMethoddConsumerHeatbeat;
  req_protocol->prot_msg_ = hb_msg;
}

void BaseConsumer::buidCloseRequestC2M(TubeMQCodec::ReqProtocolPtr& req_protocol) {
  string close_msg;
  CloseRequestC2M c2m_request;
  c2m_request.set_clientid(client_uuid_);
  c2m_request.set_groupname(config_.GetGroupName());
  if (needGenMasterCertificateInfo(true)) {
    MasterCertificateInfo* pmst_certinfo = c2m_request.mutable_authinfo();
    AuthenticateInfo* pauthinfo = pmst_certinfo->mutable_authinfo();
    genMasterAuthenticateToken(pauthinfo, config_.GetUsrName(), config_.GetUsrPassWord());
  }
  c2m_request.SerializeToString(&close_msg);
  req_protocol->method_id_ = rpc_config::kMasterMethoddConsumerClose;
  req_protocol->prot_msg_ = close_msg;
}

void BaseConsumer::buidRegisterRequestC2B(const PartitionExt& partition,
                                          TubeMQCodec::ReqProtocolPtr& req_protocol) {
  string register_msg;
  set<string> filter_cond_set;
  map<string, set<string> > filter_map;
  RegisterRequestC2B c2b_request;
  c2b_request.set_clientid(client_uuid_);
  c2b_request.set_groupname(config_.GetGroupName());
  c2b_request.set_optype(rpc_config::kRegOpTypeRegister);
  c2b_request.set_topicname(partition.GetTopic());
  c2b_request.set_partitionid(partition.GetPartitionId());
  c2b_request.set_qrypriorityid(rmtdata_cache_.GetGroupQryPriorityId());
  bool is_first_reg = rmtdata_cache_.IsPartitionFirstReg(partition.GetPartitionKey());
  c2b_request.set_readstatus(getConsumeReadStatus(is_first_reg));
  if (sub_info_.IsFilterConsume(partition.GetTopic())) {
    filter_map = sub_info_.GetTopicFilterMap();
    if (filter_map.find(partition.GetTopic()) != filter_map.end()) {
      filter_cond_set = filter_map[partition.GetTopic()];
      for (set<string>::iterator it_cond = filter_cond_set.begin();
           it_cond != filter_cond_set.end(); it_cond++) {
        c2b_request.add_filtercondstr(*it_cond);
      }
    }
  }
  int64_t part_offset = tb_config::kInvalidValue;
  sub_info_.GetAssignedPartOffset(partition.GetPartitionKey(), part_offset);
  if (part_offset != tb_config::kInvalidValue) {
    c2b_request.set_curroffset(part_offset);
  }
  AuthorizedInfo* p_authInfo = c2b_request.mutable_authinfo();
  genBrokerAuthenticInfo(p_authInfo, true);
  c2b_request.SerializeToString(&register_msg);
  req_protocol->method_id_ = rpc_config::kBrokerMethoddConsumerRegister;
  req_protocol->prot_msg_ = register_msg;
}

void BaseConsumer::buidUnRegRequestC2B(const PartitionExt& partition,
                                       TubeMQCodec::ReqProtocolPtr& req_protocol) {
  string unreg_msg;
  RegisterRequestC2B c2b_request;
  c2b_request.set_clientid(client_uuid_);
  c2b_request.set_groupname(config_.GetGroupName());
  c2b_request.set_optype(rpc_config::kRegOpTypeUnReg);
  c2b_request.set_topicname(partition.GetTopic());
  c2b_request.set_partitionid(partition.GetPartitionId());
  c2b_request.set_readstatus(1);
  AuthorizedInfo* p_authInfo = c2b_request.mutable_authinfo();
  genBrokerAuthenticInfo(p_authInfo, true);
  c2b_request.SerializeToString(&unreg_msg);
  req_protocol->method_id_ = rpc_config::kBrokerMethoddConsumerRegister;
  req_protocol->prot_msg_ = unreg_msg;
}

void BaseConsumer::buidHeartBeatC2B(const list<PartitionExt>& partitions,
                                    TubeMQCodec::ReqProtocolPtr& req_protocol) {
  string hb_msg;
  HeartBeatRequestC2B c2b_request;
  list<PartitionExt>::const_iterator it_part;
  c2b_request.set_clientid(client_uuid_);
  c2b_request.set_groupname(config_.GetGroupName());
  c2b_request.set_readstatus(getConsumeReadStatus(false));
  c2b_request.set_qrypriorityid(rmtdata_cache_.GetGroupQryPriorityId());
  for (it_part = partitions.begin(); it_part != partitions.end(); ++it_part) {
    c2b_request.add_partitioninfo(it_part->ToString());
  }
  AuthorizedInfo* p_authInfo = c2b_request.mutable_authinfo();
  genBrokerAuthenticInfo(p_authInfo, true);
  c2b_request.SerializeToString(&hb_msg);
  req_protocol->method_id_ = rpc_config::kBrokerMethoddConsumerHeatbeat;
  req_protocol->prot_msg_ = hb_msg;
}

void BaseConsumer::buidGetMessageC2B(const PartitionExt& partition,
                                     TubeMQCodec::ReqProtocolPtr& req_protocol) {
  string get_msg;
  GetMessageRequestC2B c2b_request;
  c2b_request.set_clientid(client_uuid_);
  c2b_request.set_groupname(config_.GetGroupName());
  c2b_request.set_topicname(partition.GetTopic());
  c2b_request.set_escflowctrl(rmtdata_cache_.IsUnderGroupCtrl());
  c2b_request.set_partitionid(partition.GetPartitionId());
  c2b_request.set_lastpackconsumed(partition.IsLastConsumed());
  c2b_request.set_manualcommitoffset(false);
  c2b_request.SerializeToString(&get_msg);
  req_protocol->method_id_ = rpc_config::kBrokerMethoddConsumerGetMsg;
  req_protocol->prot_msg_ = get_msg;
}

void BaseConsumer::buidCommitC2B(const PartitionExt& partition, bool is_last_consumed,
                                 TubeMQCodec::ReqProtocolPtr& req_protocol) {
  string commit_msg;
  CommitOffsetRequestC2B c2b_request;
  c2b_request.set_clientid(client_uuid_);
  c2b_request.set_groupname(config_.GetGroupName());
  c2b_request.set_topicname(partition.GetTopic());
  c2b_request.set_partitionid(partition.GetPartitionId());
  c2b_request.set_lastpackconsumed(is_last_consumed);
  c2b_request.SerializeToString(&commit_msg);
  req_protocol->method_id_ = rpc_config::kBrokerMethoddConsumerCommit;
  req_protocol->prot_msg_ = commit_msg;
}

bool BaseConsumer::processRegisterResponseM2C(int32_t& error_code, string& err_info,
                                              const TubeMQCodec::RspProtocolPtr& rsp_protocol) {
  if (!rsp_protocol->success_) {
    error_code = rsp_protocol->code_;
    err_info = rsp_protocol->error_msg_;
    return false;
  }
  RegisterResponseM2C rsp_m2c;
  bool result = rsp_m2c.ParseFromArray(rsp_protocol->rsp_body_.data().c_str(),
                                       (int32_t)(rsp_protocol->rsp_body_.data().length()));
  if (!result) {
    error_code = err_code::kErrParseFailure;
    err_info = "Parse RegisterResponseM2C response failure!";
    return false;
  }
  if (!rsp_m2c.success()) {
    error_code = rsp_m2c.errcode();
    err_info = rsp_m2c.errmsg();
    return false;
  }
  // update policy
  if (rsp_m2c.has_notallocated() && !rsp_m2c.notallocated()) {
    sub_info_.CompAndSetNotAllocated(true, false);
  }
  if (rsp_m2c.has_defflowcheckid() || rsp_m2c.has_groupflowcheckid()) {
    if (rsp_m2c.has_defflowcheckid()) {
      rmtdata_cache_.UpdateDefFlowCtrlInfo(rsp_m2c.defflowcheckid(), rsp_m2c.defflowcontrolinfo());
    }
    int qryPriorityId = rsp_m2c.has_qrypriorityid() ? rsp_m2c.qrypriorityid()
                                                    : rmtdata_cache_.GetGroupQryPriorityId();
    rmtdata_cache_.UpdateGroupFlowCtrlInfo(qryPriorityId, rsp_m2c.groupflowcheckid(),
                                           rsp_m2c.groupflowcontrolinfo());
  }
  if (rsp_m2c.has_authorizedinfo()) {
    processAuthorizedToken(rsp_m2c.authorizedinfo());
  }
  error_code = err_code::kErrSuccess;
  err_info = "Ok";
  return true;
}

bool BaseConsumer::processHBResponseM2C(int32_t& error_code, string& err_info,
                                        const TubeMQCodec::RspProtocolPtr& rsp_protocol) {
  if (!rsp_protocol->success_) {
    error_code = rsp_protocol->code_;
    err_info = rsp_protocol->error_msg_;
    return false;
  }
  HeartResponseM2C rsp_m2c;
  bool result = rsp_m2c.ParseFromArray(rsp_protocol->rsp_body_.data().c_str(),
                                       (int32_t)(rsp_protocol->rsp_body_.data().length()));
  if (!result) {
    error_code = err_code::kErrParseFailure;
    err_info = "Parse HeartResponseM2C response failure!";
    return false;
  }
  if (!rsp_m2c.success()) {
    error_code = rsp_m2c.errcode();
    err_info = rsp_m2c.errmsg();
    return false;
  }
  // update policy
  if (rsp_m2c.has_notallocated() && !rsp_m2c.notallocated()) {
    sub_info_.CompAndSetNotAllocated(true, false);
  }
  if (rsp_m2c.has_defflowcheckid() || rsp_m2c.has_groupflowcheckid()) {
    if (rsp_m2c.has_defflowcheckid()) {
      rmtdata_cache_.UpdateDefFlowCtrlInfo(rsp_m2c.defflowcheckid(), rsp_m2c.defflowcontrolinfo());
    }
    int qryPriorityId = rsp_m2c.has_qrypriorityid() ? rsp_m2c.qrypriorityid()
                                                    : rmtdata_cache_.GetGroupQryPriorityId();
    rmtdata_cache_.UpdateGroupFlowCtrlInfo(qryPriorityId, rsp_m2c.groupflowcheckid(),
                                           rsp_m2c.groupflowcontrolinfo());
  }
  if (rsp_m2c.has_authorizedinfo()) {
    processAuthorizedToken(rsp_m2c.authorizedinfo());
  }
  if (rsp_m2c.has_requireauth()) {
    nextauth_2_master.Set(rsp_m2c.requireauth());
  }
  // Get the latest rebalance task
  if (rsp_m2c.has_event()) {
    EventProto eventProto = rsp_m2c.event();
    if (eventProto.rebalanceid() > 0) {
      list<SubscribeInfo> subcribe_infos;
      for (int i = 0; i < eventProto.subscribeinfo_size(); i++) {
        SubscribeInfo sub_info(eventProto.subscribeinfo(i));
        subcribe_infos.push_back(sub_info);
      }
      ConsumerEvent new_event(eventProto.rebalanceid(), eventProto.optype(), subcribe_infos, 0);
      rmtdata_cache_.OfferEvent(new_event);
    }
  }
  last_master_hbtime_ = Utils::GetCurrentTimeMillis();
  error_code = err_code::kErrSuccess;
  err_info = "Ok";
  return true;
}

void BaseConsumer::unregister2Brokers(map<NodeInfo, list<PartitionExt> >& unreg_partitions,
                                      bool wait_rsp) {
  string err_info;
  map<NodeInfo, list<PartitionExt> >::iterator it;
  list<PartitionExt>::iterator it_part;
  if (unreg_partitions.empty()) {
    return;
  }
  for (it = unreg_partitions.begin(); it != unreg_partitions.end(); ++it) {
    list<PartitionExt> part_list = it->second;
    for (it_part = part_list.begin(); it_part != part_list.end(); ++it_part) {
      LOG_TRACE("unregister2Brokers, partitionkey=%s", it_part->GetPartitionKey().c_str());
      auto request = std::make_shared<RequestContext>();
      TubeMQCodec::ReqProtocolPtr req_protocol = TubeMQCodec::GetReqProtocol();
      // build unregister 2 broker request
      buidUnRegRequestC2B(*it_part, req_protocol);
      request->codec_ = std::make_shared<TubeMQCodec>();
      request->ip_ = it_part->GetBrokerHost();
      request->port_ = it_part->GetBrokerPort();
      request->timeout_ = config_.GetRpcReadTimeoutMs();
      request->request_id_ = Singleton<UniqueSeqId>::Instance().Next();
      req_protocol->request_id_ = request->request_id_;
      req_protocol->rpc_read_timeout_ = config_.GetRpcReadTimeoutMs() - 500;
      // send message to target
      // not need process
      ResponseContext response_context;
      if (wait_rsp) {
        SyncRequest(response_context, request, req_protocol);
      } else {
        AsyncRequest(request, req_protocol);
      }
    }
  }
}

bool BaseConsumer::processRegResponseB2C(int32_t& error_code, string& err_info,
                                         const TubeMQCodec::RspProtocolPtr& rsp_protocol) {
  if (!rsp_protocol->success_) {
    error_code = rsp_protocol->code_;
    err_info = rsp_protocol->error_msg_;
    return false;
  }
  RegisterResponseB2C rsp_b2c;
  bool result = rsp_b2c.ParseFromArray(rsp_protocol->rsp_body_.data().c_str(),
                                       (int32_t)(rsp_protocol->rsp_body_.data().length()));
  if (!result) {
    error_code = err_code::kErrParseFailure;
    err_info = "Parse RegisterResponseB2C response failure!";
    return false;
  }
  if (!rsp_b2c.success()) {
    error_code = rsp_b2c.errcode();
    err_info = rsp_b2c.errmsg();
    return false;
  }
  error_code = err_code::kErrSuccess;
  err_info = "Ok";
  return true;
}

void BaseConsumer::convertMessages(int32_t& msg_size, list<Message>& message_list,
                                   bool filter_consume, const string& topic_name,
                                   GetMessageResponseB2C& rsp_b2c) {
  // #lizard forgives
  msg_size = 0;
  message_list.clear();
  if (rsp_b2c.messages_size() == 0) {
    return;
  }
  for (int i = 0; i < rsp_b2c.messages_size(); i++) {
    TransferedMessage tsfMsg = rsp_b2c.messages(i);
    int32_t flag = tsfMsg.flag();
    int64_t message_id = tsfMsg.messageid();
    int32_t in_check_sum = tsfMsg.checksum();
    int32_t payload_length = tsfMsg.payloaddata().length();
    int32_t calc_checksum = Utils::Crc32(tsfMsg.payloaddata());
    if (in_check_sum != calc_checksum) {
      continue;
    }
    int read_pos = 0;
    int data_len = payload_length;
    map<string, string> properties;
    std::unique_ptr<char[]> payload_data(new char[payload_length]);
    memcpy(&payload_data[0], tsfMsg.payloaddata().c_str(), payload_length);
    if ((flag & tb_config::kMsgFlagIncProperties) == 1) {
      if (payload_length < 4) {
        continue;
      }
      int32_t attr_len = ntohl(*reinterpret_cast<int*>(&payload_data[0]));
      read_pos += 4;
      data_len -= 4;
      if (attr_len > data_len) {
        continue;
      }
      string attribute(&payload_data[0] + read_pos, attr_len);
      read_pos += attr_len;
      data_len -= attr_len;
      Utils::Split(attribute, properties, delimiter::kDelimiterComma, delimiter::kDelimiterEqual);
      if (filter_consume) {
        map<string, set<string> > topic_filter_map = sub_info_.GetTopicFilterMap();
        map<string, set<string> >::const_iterator it = topic_filter_map.find(topic_name);
        if (properties.find(tb_config::kRsvPropKeyFilterItem) != properties.end()) {
          string msg_key = properties[tb_config::kRsvPropKeyFilterItem];
          if (it != topic_filter_map.end()) {
            set<string> filters = it->second;
            if (filters.find(msg_key) == filters.end()) {
              continue;
            }
          }
        }
      }
    }
    Message message(topic_name, flag, message_id, &payload_data[0] + read_pos, data_len,
                    properties);
    message_list.push_back(message);
    msg_size += data_len;
  }
  return;
}

bool BaseConsumer::processGetMessageRspB2C(ConsumerResult& result, PeerInfo& peer_info,
                                           bool filter_consume, const PartitionExt& partition_ext,
                                           const string& confirm_context,
                                           const TubeMQCodec::RspProtocolPtr& rsp) {
  // #lizard forgives
  string err_info;

  if (!rsp->success_) {
    rmtdata_cache_.RelPartition(err_info, filter_consume, confirm_context, false);
    result.SetFailureResult(rsp->code_, rsp->error_msg_, partition_ext.GetTopic(), peer_info);
    return false;
  }
  GetMessageResponseB2C rsp_b2c;
  bool ret_result = rsp_b2c.ParseFromArray(rsp->rsp_body_.data().c_str(),
                                           (int32_t)(rsp->rsp_body_.data().length()));
  if (!ret_result) {
    rmtdata_cache_.RelPartition(err_info, filter_consume, confirm_context, false);
    result.SetFailureResult(err_code::kErrServerError,
                            "Parse GetMessageResponseB2C response failure!",
                            partition_ext.GetTopic(), peer_info);
    return false;
  }

  switch (rsp_b2c.errcode()) {
    case err_code::kErrSuccess: {
      bool esc_limit = (rsp_b2c.has_escflowctrl() && rsp_b2c.escflowctrl());
      int64_t data_dltval =
          rsp_b2c.has_currdatadlt() ? rsp_b2c.currdatadlt() : tb_config::kInvalidValue;
      int64_t curr_offset =
          rsp_b2c.has_curroffset() ? rsp_b2c.curroffset() : tb_config::kInvalidValue;
      bool req_slow = rsp_b2c.has_requireslow() ? rsp_b2c.requireslow() : false;
      int msg_size = 0;
      list<Message> message_list;
      convertMessages(msg_size, message_list, filter_consume, partition_ext.GetTopic(), rsp_b2c);
      rmtdata_cache_.BookedPartionInfo(partition_ext.GetPartitionKey(), curr_offset,
                                       err_code::kErrSuccess, esc_limit, msg_size, 0, data_dltval,
                                       req_slow);
      peer_info.SetCurrOffset(curr_offset);
      result.SetSuccessResult(err_code::kErrSuccess, partition_ext.GetTopic(), peer_info,
                              confirm_context, message_list);
      LOG_TRACE("[CONSUMER] getMessage count=%ld, from %s, client=%s", message_list.size(),
                partition_ext.GetPartitionKey().c_str(), client_uuid_.c_str());
      return true;
    }

    case err_code::kErrHbNoNode:
    case err_code::kErrCertificateFailure:
    case err_code::kErrDuplicatePartition: {
      rmtdata_cache_.RemovePartition(err_info, confirm_context);
      result.SetFailureResult(rsp_b2c.errcode(), rsp_b2c.errmsg(), partition_ext.GetTopic(),
                              peer_info);
      return false;
    }

    case err_code::kErrConsumeSpeedLimit: {
      // Process with server side speed limit
      int64_t def_dlttime = rsp_b2c.has_minlimittime() ? rsp_b2c.minlimittime()
                                                       : config_.GetMsgNotFoundWaitPeriodMs();
      rmtdata_cache_.RelPartition(err_info, filter_consume, confirm_context, false,
                                  tb_config::kInvalidValue, rsp_b2c.errcode(), false, 0,
                                  def_dlttime, tb_config::kInvalidValue);
      result.SetFailureResult(rsp_b2c.errcode(), rsp_b2c.errmsg(), partition_ext.GetTopic(),
                              peer_info);
      return false;
    }

    case err_code::kErrNotFound:
    case err_code::kErrForbidden:
    case err_code::kErrMoved:
    case err_code::kErrServiceUnavilable:
    default: {
      // Slow down the request based on the limitation configuration when meet these errors
      int64_t limit_dlt = 300;
      switch (rsp_b2c.errcode()) {
        case err_code::kErrForbidden: {
          limit_dlt = 2000;
          break;
        }
        case err_code::kErrServiceUnavilable: {
          limit_dlt = 300;
          break;
        }
        case err_code::kErrMoved: {
          limit_dlt = 200;
          break;
        }
        case err_code::kErrNotFound: {
          limit_dlt = config_.GetMsgNotFoundWaitPeriodMs();
          break;
        }
        default: {
          //
        }
      }
      rmtdata_cache_.RelPartition(err_info, filter_consume, confirm_context, false,
                                  tb_config::kInvalidValue, rsp_b2c.errcode(), false, 0, limit_dlt,
                                  tb_config::kInvalidValue);
      result.SetFailureResult(rsp_b2c.errcode(), rsp_b2c.errmsg(), partition_ext.GetTopic(),
                              peer_info);
      return false;
    }
  }
  return true;
}

bool BaseConsumer::isClientRunning() { return (status_.Get() == 2); }

string BaseConsumer::buildUUID() {
  stringstream ss;
  ss << config_.GetGroupName();
  ss << "_";
  ss << TubeMQService::Instance()->GetLocalHost();
  ss << "-";
  ss << getpid();
  ss << "-";
  ss << Utils::GetCurrentTimeMillis();
  ss << "-";
  ss << GetClientIndex();
  ss << "-";
  ss << kTubeMQClientVersion;
  return ss.str();
}

int32_t BaseConsumer::getConsumeReadStatus(bool is_first_reg) {
  int32_t readStatus = rpc_config::kConsumeStatusNormal;
  if (is_first_reg) {
    if (config_.GetConsumePosition() == 0) {
      readStatus = rpc_config::kConsumeStatusFromMax;
      LOG_INFO("[Consumer From Max Offset], clientId=%s", client_uuid_.c_str());
    } else if (config_.GetConsumePosition() > 0) {
      readStatus = rpc_config::kConsumeStatusFromMaxAlways;
      LOG_INFO("[Consumer From Max Offset Always], clientId=%s", client_uuid_.c_str());
    }
  }
  return readStatus;
}

bool BaseConsumer::initMasterAddress(string& err_info, const string& master_info) {
  masters_map_.clear();
  Utils::Split(master_info, masters_map_, delimiter::kDelimiterComma, delimiter::kDelimiterColon);
  if (masters_map_.empty()) {
    err_info = "Illegal parameter: master_info is blank!";
    return false;
  }
  bool needXfs = false;
  map<string, int32_t>::iterator it;
  for (it = masters_map_.begin(); it != masters_map_.end(); it++) {
    if (Utils::NeedDnsXfs(it->first)) {
      needXfs = true;
      break;
    }
  }
  it = masters_map_.begin();
  curr_master_addr_ = it->first;
  if (needXfs) {
    TubeMQService::Instance()->AddMasterAddress(err_info, master_info);
  }
  err_info = "Ok";
  return true;
}

void BaseConsumer::getNextMasterAddr(string& ipaddr, int32_t& port) {
  map<string, int32_t>::iterator it;
  it = masters_map_.find(curr_master_addr_);
  if (it != masters_map_.end()) {
    it++;
    if (it == masters_map_.end()) {
      it = masters_map_.begin();
    }
  } else {
    it = masters_map_.begin();
  }
  ipaddr = it->first;
  port = it->second;
  curr_master_addr_ = it->first;
  if (Utils::NeedDnsXfs(ipaddr)) {
    TubeMQService::Instance()->GetXfsMasterAddress(curr_master_addr_, ipaddr);
  }
  LOG_TRACE("getNextMasterAddr address is %s:%d", ipaddr.c_str(), port);
}

void BaseConsumer::getCurrentMasterAddr(string& ipaddr, int32_t& port) {
  ipaddr = curr_master_addr_;
  port = masters_map_[curr_master_addr_];
  if (Utils::NeedDnsXfs(ipaddr)) {
    TubeMQService::Instance()->GetXfsMasterAddress(curr_master_addr_, ipaddr);
  }
  LOG_TRACE("getCurrentMasterAddr address is %s:%d", ipaddr.c_str(), port);
}

bool BaseConsumer::needGenMasterCertificateInfo(bool force) {
  bool needAdd = false;
  if (config_.IsAuthenticEnabled()) {
    if (force) {
      needAdd = true;
      nextauth_2_master.Set(false);
    } else if (nextauth_2_master.Get()) {
      if (nextauth_2_master.CompareAndSet(true, false)) {
        needAdd = true;
      }
    }
  }
  return needAdd;
}

void BaseConsumer::genBrokerAuthenticInfo(AuthorizedInfo* p_authInfo, bool force) {
  bool needAdd = false;
  p_authInfo->set_visitauthorizedtoken(visit_token_.Get());
  if (config_.IsAuthenticEnabled()) {
    if (force) {
      needAdd = true;
      nextauth_2_broker.Set(false);
    } else if (nextauth_2_broker.Get()) {
      if (nextauth_2_broker.CompareAndSet(true, false)) {
        needAdd = true;
      }
    }
    if (needAdd) {
      string auth_token =
          Utils::GenBrokerAuthenticateToken(config_.GetUsrName(), config_.GetUsrPassWord());
      p_authInfo->set_authauthorizedtoken(auth_token);
    }
  }
}

void BaseConsumer::genMasterAuthenticateToken(AuthenticateInfo* pauthinfo, const string& username,
                                              const string usrpassword) {
  //
}

void BaseConsumer::processAuthorizedToken(const MasterAuthorizedInfo& authorized_token_info) {
  visit_token_.Set(authorized_token_info.visitauthorizedtoken());
  if (authorized_token_info.has_authauthorizedtoken()) {
    lock_guard<mutex> lck(auth_lock_);

    if (authorized_info_ != authorized_token_info.authauthorizedtoken()) {
      authorized_info_ = authorized_token_info.authauthorizedtoken();
    }
  }
}

void BaseConsumer::addBrokerHBTimer(const NodeInfo broker) {
  SteadyTimerPtr timer;
  int32_t hb_periodms = config_.GetHeartbeatPeriodMs();
  lock_guard<mutex> lck(broker_timer_lock_);
  if (broker_timer_map_.find(broker) == broker_timer_map_.end()) {
    timer = TubeMQService::Instance()->CreateTimer();
    broker_timer_map_[broker] = timer;
    timer->expires_after(std::chrono::milliseconds(hb_periodms / 2));
    auto self = shared_from_this();
    timer->async_wait([self, this, broker](const std::error_code& ec) {
      if (ec) {
        return;
      }
      processHeartBeat2Broker(broker);
    });
  }
}

void BaseConsumer::reSetBrokerHBTimer(const NodeInfo broker) {
  SteadyTimerPtr timer;
  list<PartitionExt> partition_list;
  int32_t hb_periodms = config_.GetHeartbeatPeriodMs();
  lock_guard<mutex> lck(broker_timer_lock_);
  rmtdata_cache_.GetPartitionByBroker(broker, partition_list);
  if (partition_list.empty()) {
    broker_timer_map_.erase(broker);
  } else {
    if (broker_timer_map_.find(broker) != broker_timer_map_.end()) {
      timer = broker_timer_map_[broker];
      timer->expires_after(std::chrono::milliseconds(hb_periodms));
      auto self = shared_from_this();
      timer->async_wait([self, this, broker](const std::error_code& ec) {
        if (ec) {
          return;
        }
        processHeartBeat2Broker(broker);
      });
    }
  }
}
}  // namespace tubemq