/*
 * 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 "DefaultLitePullConsumerImpl.h"

#ifndef WIN32
#include <signal.h>
#endif

#include "AssignedMessageQueue.hpp"
#include "FilterAPI.hpp"
#include "LocalFileOffsetStore.h"
#include "MQAdminImpl.h"
#include "MQClientAPIImpl.h"
#include "MQClientInstance.h"
#include "NamespaceUtil.h"
#include "PullAPIWrapper.h"
#include "PullSysFlag.h"
#include "RebalanceLitePullImpl.h"
#include "RemoteBrokerOffsetStore.h"
#include "UtilAll.h"
#include "Validators.h"

static const long PULL_TIME_DELAY_MILLS_WHEN_PAUSE = 1000;
static const long PULL_TIME_DELAY_MILLS_WHEN_FLOW_CONTROL = 50;

namespace rocketmq {

class DefaultLitePullConsumerImpl::MessageQueueListenerImpl : public MessageQueueListener {
 public:
  MessageQueueListenerImpl(DefaultLitePullConsumerImplPtr pull_consumer) : default_lite_pull_consumer_(pull_consumer) {}

  ~MessageQueueListenerImpl() = default;

  void messageQueueChanged(const std::string& topic,
                           std::vector<MQMessageQueue>& mq_all,
                           std::vector<MQMessageQueue>& mq_divided) override {
    auto consumer = default_lite_pull_consumer_.lock();
    if (nullptr == consumer) {
      return;
    }
    switch (consumer->messageModel()) {
      case BROADCASTING:
        consumer->updateAssignedMessageQueue(topic, mq_all);
        consumer->updatePullTask(topic, mq_all);
        break;
      case CLUSTERING:
        consumer->updateAssignedMessageQueue(topic, mq_divided);
        consumer->updatePullTask(topic, mq_divided);
        break;
      default:
        break;
    }
  }

 private:
  std::weak_ptr<DefaultLitePullConsumerImpl> default_lite_pull_consumer_;
};

class DefaultLitePullConsumerImpl::ConsumeRequest {
 public:
  ConsumeRequest(std::vector<MessageExtPtr>&& message_exts,
                 const MQMessageQueue& message_queue,
                 ProcessQueuePtr process_queue)
      : message_exts_(std::move(message_exts)), message_queue_(message_queue), process_queue_(process_queue) {}

 public:
  std::vector<MessageExtPtr>& message_exts() { return message_exts_; }

  MQMessageQueue& message_queue() { return message_queue_; }

  ProcessQueuePtr process_queue() { return process_queue_; }

 private:
  std::vector<MessageExtPtr> message_exts_;
  MQMessageQueue message_queue_;
  ProcessQueuePtr process_queue_;
};

class DefaultLitePullConsumerImpl::PullTaskImpl : public std::enable_shared_from_this<PullTaskImpl> {
 public:
  PullTaskImpl(DefaultLitePullConsumerImplPtr pull_consumer, const MQMessageQueue& message_queue)
      : default_lite_pull_consumer_(pull_consumer), message_queue_(message_queue), cancelled_(false) {}

  void run() {
    auto consumer = default_lite_pull_consumer_.lock();
    if (nullptr == consumer) {
      LOG_WARN_NEW("PullTaskImpl::run: DefaultLitePullConsumerImpl is released.");
      return;
    }

    if (cancelled_) {
      return;
    }

    if (consumer->assigned_message_queue_->isPaused(message_queue_)) {
      consumer->scheduled_thread_pool_executor_.schedule(
          std::bind(&DefaultLitePullConsumerImpl::PullTaskImpl::run, shared_from_this()),
          PULL_TIME_DELAY_MILLS_WHEN_PAUSE, time_unit::milliseconds);
      LOG_DEBUG_NEW("Message Queue: {} has been paused!", message_queue_.toString());
      return;
    }

    auto process_queue = consumer->assigned_message_queue_->getProcessQueue(message_queue_);
    if (nullptr == process_queue || process_queue->dropped()) {
      LOG_INFO_NEW("The message queue not be able to poll, because it's dropped. group={}, messageQueue={}",
                   consumer->groupName(), message_queue_.toString());
      return;
    }

    auto config = consumer->getDefaultLitePullConsumerConfig();

    if (consumer->consume_request_cache_.size() * config->pull_batch_size() > config->pull_threshold_for_all()) {
      consumer->scheduled_thread_pool_executor_.schedule(
          std::bind(&DefaultLitePullConsumerImpl::PullTaskImpl::run, shared_from_this()),
          PULL_TIME_DELAY_MILLS_WHEN_FLOW_CONTROL, time_unit::milliseconds);
      if ((consumer->consume_request_flow_control_times_++ % 1000) == 0)
        LOG_WARN_NEW(
            "The consume request count exceeds threshold {}, so do flow control, consume request count={}, "
            "flowControlTimes={}",
            config->pull_threshold_for_all(), consumer->consume_request_cache_.size(),
            consumer->consume_request_flow_control_times_);
      return;
    }

    auto cached_message_count = process_queue->getCacheMsgCount();
    if (cached_message_count > config->pull_threshold_for_queue()) {
      consumer->scheduled_thread_pool_executor_.schedule(
          std::bind(&DefaultLitePullConsumerImpl::PullTaskImpl::run, shared_from_this()),
          PULL_TIME_DELAY_MILLS_WHEN_FLOW_CONTROL, time_unit::milliseconds);
      if ((consumer->queue_flow_control_times_++ % 1000) == 0) {
        LOG_WARN_NEW(
            "The cached message count exceeds the threshold {}, so do flow control, minOffset={}, maxOffset={}, "
            "count={}, size={} MiB, flowControlTimes={}",
            config->pull_threshold_for_queue(), process_queue->getCacheMinOffset(), process_queue->getCacheMaxOffset(),
            cached_message_count, "unknown", consumer->queue_flow_control_times_);
      }
      return;
    }

    // long cachedMessageSizeInMiB = processQueue->getMsgSize() / (1024 * 1024);
    // if (cachedMessageSizeInMiB > consumer.getPullThresholdSizeForQueue()) {
    //   scheduledThreadPoolExecutor.schedule(this, PULL_TIME_DELAY_MILLS_WHEN_FLOW_CONTROL, TimeUnit.MILLISECONDS);
    //   if ((queueFlowControlTimes++ % 1000) == 0) {
    //     log.warn(
    //         "The cached message size exceeds the threshold {} MiB, so do flow control, minOffset={}, maxOffset={},
    //         "
    //         "count={}, size={} MiB, flowControlTimes={}",
    //         consumer.getPullThresholdSizeForQueue(), processQueue.getMsgTreeMap().firstKey(),
    //         processQueue.getMsgTreeMap().lastKey(), cachedMessageCount, cachedMessageSizeInMiB,
    //         queueFlowControlTimes);
    //   }
    //   return;
    // }

    // if (processQueue.getMaxSpan() > consumer.getConsumeMaxSpan()) {
    //   scheduledThreadPoolExecutor.schedule(this, PULL_TIME_DELAY_MILLS_WHEN_FLOW_CONTROL, TimeUnit.MILLISECONDS);
    //   if ((queueMaxSpanFlowControlTimes++ % 1000) == 0) {
    //     log.warn(
    //         "The queue's messages, span too long, so do flow control, minOffset={}, maxOffset={}, maxSpan={}, "
    //         "flowControlTimes={}",
    //         processQueue.getMsgTreeMap().firstKey(), processQueue.getMsgTreeMap().lastKey(),
    //         processQueue.getMaxSpan(),
    //         queueMaxSpanFlowControlTimes);
    //   }
    //   return;
    // }

    auto offset = consumer->nextPullOffset(message_queue_);
    long pull_delay_time_millis = 0;
    SubscriptionData* subscription_data = nullptr;
    try {
      if (consumer->subscription_type_ == SubscriptionType::SUBSCRIBE) {
        subscription_data = consumer->rebalance_impl_->getSubscriptionData(message_queue_.topic());
      } else {
        subscription_data = FilterAPI::buildSubscriptionData(message_queue_.topic(), SUB_ALL).release();
      }

      std::unique_ptr<PullResult> pull_result(
          consumer->pull(message_queue_, subscription_data, offset, config->pull_batch_size()));

      switch (pull_result->pull_status()) {
        case PullStatus::FOUND: {
          auto objLock = consumer->message_queue_lock_.fetchLockObject(message_queue_);
          std::lock_guard<std::mutex> lock(*objLock);
          if (!pull_result->msg_found_list().empty() &&
              consumer->assigned_message_queue_->getSeekOffset(message_queue_) == -1) {
            process_queue->putMessage(pull_result->msg_found_list());
            consumer->submitConsumeRequest(
                new ConsumeRequest(std::move(pull_result->msg_found_list()), message_queue_, process_queue));
          }
        } break;
        case PullStatus::OFFSET_ILLEGAL:
          LOG_WARN_NEW("The pull request offset illegal, {}", pull_result->toString());
          break;
        case PullStatus::NO_NEW_MSG:
        case PullStatus::NO_MATCHED_MSG:
          pull_delay_time_millis = 1000;
          break;
        case PullStatus::NO_LATEST_MSG:
          pull_delay_time_millis = config->pull_time_delay_millis_when_exception();
          break;
        default:
          break;
      }

      consumer->updatePullOffset(message_queue_, pull_result->next_begin_offset());
    } catch (std::exception& e) {
      pull_delay_time_millis = config->pull_time_delay_millis_when_exception();
      LOG_ERROR_NEW("An error occurred in pull message process. {}", e.what());
    }

    if (consumer->subscription_type_ != SubscriptionType::SUBSCRIBE) {
      delete subscription_data;
    }

    if (!cancelled_) {
      consumer->scheduled_thread_pool_executor_.schedule(
          std::bind(&DefaultLitePullConsumerImpl::PullTaskImpl::run, shared_from_this()), pull_delay_time_millis,
          time_unit::milliseconds);
    } else {
      LOG_WARN_NEW("The Pull Task is cancelled after doPullTask, {}", message_queue_.toString());
    }
  }

 public:
  inline const MQMessageQueue& message_queue() { return message_queue_; }

  inline bool is_cancelled() const { return cancelled_; }
  inline void set_cancelled(bool cancelled) { cancelled_ = cancelled; }

 private:
  std::weak_ptr<DefaultLitePullConsumerImpl> default_lite_pull_consumer_;
  MQMessageQueue message_queue_;
  volatile bool cancelled_;
};

DefaultLitePullConsumerImpl::DefaultLitePullConsumerImpl(DefaultLitePullConsumerConfigPtr config)
    : DefaultLitePullConsumerImpl(config, nullptr) {}

DefaultLitePullConsumerImpl::DefaultLitePullConsumerImpl(DefaultLitePullConsumerConfigPtr config, RPCHookPtr rpcHook)
    : MQClientImpl(config, rpcHook),
      start_time_(UtilAll::currentTimeMillis()),
      subscription_type_(SubscriptionType::NONE),
      consume_request_flow_control_times_(0),
      queue_flow_control_times_(0),
      next_auto_commit_deadline_(-1LL),
      auto_commit_(true),
      message_queue_listener_(nullptr),
      assigned_message_queue_(new AssignedMessageQueue()),
      scheduled_thread_pool_executor_("PullMsgThread", config->pull_thread_nums(), false),
      scheduled_executor_service_("MonitorMessageQueueChangeThread", false),
      rebalance_impl_(new RebalanceLitePullImpl(this)),
      pull_api_wrapper_(nullptr),
      offset_store_(nullptr) {}

DefaultLitePullConsumerImpl::~DefaultLitePullConsumerImpl() = default;

void DefaultLitePullConsumerImpl::start() {
#ifndef WIN32
  /* Ignore the SIGPIPE */
  struct sigaction sa;
  memset(&sa, 0, sizeof(struct sigaction));
  sa.sa_handler = SIG_IGN;
  sa.sa_flags = 0;
  ::sigaction(SIGPIPE, &sa, 0);
#endif

  switch (service_state_) {
    case CREATE_JUST: {
      // wrap namespace
      client_config_->set_group_name(
          NamespaceUtil::wrapNamespace(client_config_->name_space(), client_config_->group_name()));

      LOG_INFO_NEW("the consumer [{}] start beginning.", client_config_->group_name());

      service_state_ = START_FAILED;

      checkConfig();

      if (messageModel() == MessageModel::CLUSTERING) {
        client_config_->changeInstanceNameToPID();
      }

      // init client_instance_
      MQClientImpl::start();

      // init rebalance_impl_
      rebalance_impl_->set_consumer_group(client_config_->group_name());
      rebalance_impl_->set_message_model(getDefaultLitePullConsumerConfig()->message_model());
      rebalance_impl_->set_allocate_mq_strategy(getDefaultLitePullConsumerConfig()->allocate_mq_strategy());
      rebalance_impl_->set_client_instance(client_instance_.get());

      // init pull_api_wrapper_
      pull_api_wrapper_.reset(new PullAPIWrapper(client_instance_.get(), client_config_->group_name()));
      // TODO: registerFilterMessageHook

      // init offset_store_
      switch (getDefaultLitePullConsumerConfig()->message_model()) {
        case MessageModel::BROADCASTING:
          offset_store_.reset(new LocalFileOffsetStore(client_instance_.get(), client_config_->group_name()));
          break;
        case MessageModel::CLUSTERING:
          offset_store_.reset(new RemoteBrokerOffsetStore(client_instance_.get(), client_config_->group_name()));
          break;
      }
      offset_store_->load();

      scheduled_thread_pool_executor_.set_thread_nums(getDefaultLitePullConsumerConfig()->pull_thread_nums());
      scheduled_thread_pool_executor_.startup();
      scheduled_executor_service_.startup();

      // register consumer
      bool registerOK = client_instance_->registerConsumer(client_config_->group_name(), this);
      if (!registerOK) {
        service_state_ = CREATE_JUST;
        THROW_MQEXCEPTION(MQClientException, "The cousumer group[" + client_config_->group_name() +
                                                 "] has been created before, specify another name please.",
                          -1);
      }

      client_instance_->start();

      startScheduleTask();

      LOG_INFO_NEW("the consumer [{}] start OK", client_config_->group_name());
      service_state_ = RUNNING;

      operateAfterRunning();
      break;
    }
    case RUNNING:
    case START_FAILED:
    case SHUTDOWN_ALREADY:
      THROW_MQEXCEPTION(MQClientException, "The PullConsumer service state not OK, maybe started once", -1);
      break;
    default:
      break;
  };
}

void DefaultLitePullConsumerImpl::checkConfig() {
  const auto& groupname = client_config_->group_name();

  // check consumerGroup
  Validators::checkGroup(groupname);

  // consumerGroup
  if (DEFAULT_CONSUMER_GROUP == groupname) {
    THROW_MQEXCEPTION(MQClientException,
                      "consumerGroup can not equal " + DEFAULT_CONSUMER_GROUP + ", please specify another one.", -1);
  }

  // messageModel
  if (getDefaultLitePullConsumerConfig()->message_model() != BROADCASTING &&
      getDefaultLitePullConsumerConfig()->message_model() != CLUSTERING) {
    THROW_MQEXCEPTION(MQClientException, "messageModel is valid", -1);
  }

  // allocateMessageQueueStrategy
  if (nullptr == getDefaultLitePullConsumerConfig()->allocate_mq_strategy()) {
    THROW_MQEXCEPTION(MQClientException, "allocateMessageQueueStrategy is null", -1);
  }

  // if (getDefaultLitePullConsumerConfig()->getConsumerTimeoutMillisWhenSuspend() <
  //     getDefaultLitePullConsumerConfig()->getBrokerSuspendMaxTimeMillis()) {
  //   THROW_MQEXCEPTION(
  //       MQClientException,
  //       "Long polling mode, the consumer consumerTimeoutMillisWhenSuspend must greater than
  //       brokerSuspendMaxTimeMillis",
  //       -1);
  // }
}

void DefaultLitePullConsumerImpl::startScheduleTask() {
  scheduled_executor_service_.schedule(
      std::bind(&DefaultLitePullConsumerImpl::fetchTopicMessageQueuesAndComparePeriodically, this), 1000 * 10,
      time_unit::milliseconds);
}

void DefaultLitePullConsumerImpl::fetchTopicMessageQueuesAndComparePeriodically() {
  try {
    fetchTopicMessageQueuesAndCompare();
  } catch (std::exception& e) {
    LOG_ERROR_NEW("ScheduledTask fetchMessageQueuesAndCompare exception: {}", e.what());
  }

  // next round
  scheduled_executor_service_.schedule(
      std::bind(&DefaultLitePullConsumerImpl::fetchTopicMessageQueuesAndComparePeriodically, this),
      getDefaultLitePullConsumerConfig()->topic_metadata_check_interval_millis(), time_unit::milliseconds);
}

void DefaultLitePullConsumerImpl::fetchTopicMessageQueuesAndCompare() {
  std::lock_guard<std::mutex> lock(mutex_);  // synchronized
  for (const auto& it : topic_message_queue_change_listener_map_) {
    const auto& topic = it.first;
    auto* topic_message_queue_change_listener = it.second;
    std::vector<MQMessageQueue> old_message_queues = message_queues_for_topic_[topic];
    std::vector<MQMessageQueue> new_message_queues = fetchMessageQueues(topic);
    bool isChanged = !isSetEqual(new_message_queues, old_message_queues);
    if (isChanged) {
      message_queues_for_topic_[topic] = new_message_queues;
      if (topic_message_queue_change_listener != nullptr) {
        topic_message_queue_change_listener->onChanged(topic, new_message_queues);
      }
    }
  }
}

bool DefaultLitePullConsumerImpl::isSetEqual(std::vector<MQMessageQueue>& new_message_queues,
                                             std::vector<MQMessageQueue>& old_message_queues) {
  if (new_message_queues.size() != old_message_queues.size()) {
    return false;
  }
  std::sort(new_message_queues.begin(), new_message_queues.end());
  std::sort(old_message_queues.begin(), old_message_queues.end());
  return new_message_queues == old_message_queues;
}

void DefaultLitePullConsumerImpl::operateAfterRunning() {
  // If subscribe function invoke before start function, then update topic subscribe info after initialization.
  if (subscription_type_ == SubscriptionType::SUBSCRIBE) {
    updateTopicSubscribeInfoWhenSubscriptionChanged();
  }
  // If assign function invoke before start function, then update pull task after initialization.
  else if (subscription_type_ == SubscriptionType::ASSIGN) {
    auto message_queues = assigned_message_queue_->messageQueues();
    updateAssignPullTask(message_queues);
  }

  for (const auto& it : topic_message_queue_change_listener_map_) {
    const auto& topic = it.first;
    auto messageQueues = fetchMessageQueues(topic);
    message_queues_for_topic_[topic] = std::move(messageQueues);
  }
  // client_instance_->checkClientInBroker();
}

void DefaultLitePullConsumerImpl::updateTopicSubscribeInfoWhenSubscriptionChanged() {
  auto& subTable = rebalance_impl_->getSubscriptionInner();
  for (const auto& it : subTable) {
    const auto& topic = it.first;
    bool ret = client_instance_->updateTopicRouteInfoFromNameServer(topic);
    if (!ret) {
      LOG_WARN_NEW("The topic:[{}] not exist", topic);
    }
  }
}

void DefaultLitePullConsumerImpl::updateAssignPullTask(std::vector<MQMessageQueue>& mq_new_set) {
  std::sort(mq_new_set.begin(), mq_new_set.end());
  std::lock_guard<std::mutex> lock(task_table_mutex_);
  for (auto it = task_table_.begin(); it != task_table_.end();) {
    auto& mq = it->first;
    if (!std::binary_search(mq_new_set.begin(), mq_new_set.end(), mq)) {
      it->second->set_cancelled(true);
      it = task_table_.erase(it);
      continue;
    }
    it++;
  }
  startPullTask(mq_new_set);
}

void DefaultLitePullConsumerImpl::shutdown() {
  switch (service_state_) {
    case CREATE_JUST:
      break;
    case RUNNING:
      persistConsumerOffset();
      client_instance_->unregisterConsumer(client_config_->group_name());
      scheduled_thread_pool_executor_.shutdown();
      scheduled_executor_service_.shutdown();
      client_instance_->shutdown();
      rebalance_impl_->destroy();
      service_state_ = ServiceState::SHUTDOWN_ALREADY;
      LOG_INFO_NEW("the consumer [{}] shutdown OK", client_config_->group_name());
      break;
    default:
      break;
  }
}

void DefaultLitePullConsumerImpl::subscribe(const std::string& topic, const std::string& subExpression) {
  std::lock_guard<std::mutex> lock(mutex_);  // synchronized
  try {
    if (topic.empty()) {
      THROW_MQEXCEPTION(MQClientException, "Topic can not be null or empty.", -1);
    }
    set_subscription_type(SubscriptionType::SUBSCRIBE);
    rebalance_impl_->setSubscriptionData(topic, FilterAPI::buildSubscriptionData(topic, subExpression));

    message_queue_listener_.reset(new MessageQueueListenerImpl(shared_from_this()));
    assigned_message_queue_->set_rebalance_impl(rebalance_impl_.get());

    if (service_state_ == ServiceState::RUNNING) {
      client_instance_->sendHeartbeatToAllBrokerWithLock();
      updateTopicSubscribeInfoWhenSubscriptionChanged();
    }
  } catch (std::exception& e) {
    THROW_MQEXCEPTION2(MQClientException, "subscribe exception", -1, std::make_exception_ptr(e));
  }
}

void DefaultLitePullConsumerImpl::subscribe(const std::string& topic, const MessageSelector& selector) {
  // TODO:
}

void DefaultLitePullConsumerImpl::unsubscribe(const std::string& topic) {
  // TODO:
}

std::vector<SubscriptionData> DefaultLitePullConsumerImpl::subscriptions() const {
  std::vector<SubscriptionData> result;
  auto& subTable = rebalance_impl_->getSubscriptionInner();
  for (const auto& it : subTable) {
    result.push_back(*(it.second));
  }
  return result;
}

void DefaultLitePullConsumerImpl::updateTopicSubscribeInfo(const std::string& topic,
                                                           std::vector<MQMessageQueue>& info) {
  rebalance_impl_->setTopicSubscribeInfo(topic, info);
}

void DefaultLitePullConsumerImpl::doRebalance() {
  if (rebalance_impl_ != nullptr) {
    rebalance_impl_->doRebalance(false);
  }
}

void DefaultLitePullConsumerImpl::updateAssignedMessageQueue(const std::string& topic,
                                                             std::vector<MQMessageQueue>& assigned_message_queue) {
  assigned_message_queue_->updateAssignedMessageQueue(topic, assigned_message_queue);
}

void DefaultLitePullConsumerImpl::updatePullTask(const std::string& topic, std::vector<MQMessageQueue>& mq_new_set) {
  std::sort(mq_new_set.begin(), mq_new_set.end());
  std::lock_guard<std::mutex> lock(task_table_mutex_);
  for (auto it = task_table_.begin(); it != task_table_.end();) {
    auto& mq = it->first;
    if (mq.topic() == topic) {
      // remove unnecessary PullTask
      if (!std::binary_search(mq_new_set.begin(), mq_new_set.end(), mq)) {
        it->second->set_cancelled(true);
        it = task_table_.erase(it);
        continue;
      }
    }
    it++;
  }
  startPullTask(mq_new_set);
}

void DefaultLitePullConsumerImpl::startPullTask(std::vector<MQMessageQueue>& mq_set) {
  for (const auto& mq : mq_set) {
    // add new PullTask
    if (task_table_.find(mq) == task_table_.end()) {
      auto pull_task = std::make_shared<PullTaskImpl>(shared_from_this(), mq);
      task_table_.emplace(mq, pull_task);
      scheduled_thread_pool_executor_.submit(std::bind(&PullTaskImpl::run, pull_task));
    }
  }
}

int64_t DefaultLitePullConsumerImpl::nextPullOffset(const MQMessageQueue& message_queue) {
  int64_t offset = -1;
  int64_t seek_offset = assigned_message_queue_->getSeekOffset(message_queue);
  if (seek_offset != -1) {
    offset = seek_offset;
    assigned_message_queue_->updateConsumeOffset(message_queue, offset);
    assigned_message_queue_->setSeekOffset(message_queue, -1);
  } else {
    offset = assigned_message_queue_->getPullOffset(message_queue);
    if (offset == -1) {
      offset = fetchConsumeOffset(message_queue);
    }
  }
  return offset;
}

int64_t DefaultLitePullConsumerImpl::fetchConsumeOffset(const MQMessageQueue& messageQueue) {
  // checkServiceState();
  return rebalance_impl_->computePullFromWhere(messageQueue);
}

std::unique_ptr<PullResult> DefaultLitePullConsumerImpl::pull(const MQMessageQueue& mq,
                                                              SubscriptionData* subscription_data,
                                                              int64_t offset,
                                                              int max_nums) {
  return pull(mq, subscription_data, offset, max_nums,
              getDefaultLitePullConsumerConfig()->consumer_pull_timeout_millis());
}

std::unique_ptr<PullResult> DefaultLitePullConsumerImpl::pull(const MQMessageQueue& mq,
                                                              SubscriptionData* subscription_data,
                                                              int64_t offset,
                                                              int max_nums,
                                                              long timeout) {
  return pullSyncImpl(mq, subscription_data, offset, max_nums,
                      getDefaultLitePullConsumerConfig()->long_polling_enable(), timeout);
}

std::unique_ptr<PullResult> DefaultLitePullConsumerImpl::pullSyncImpl(const MQMessageQueue& mq,
                                                                      SubscriptionData* subscription_data,
                                                                      int64_t offset,
                                                                      int max_nums,
                                                                      bool block,
                                                                      long timeout) {
  if (offset < 0) {
    THROW_MQEXCEPTION(MQClientException, "offset < 0", -1);
  }

  if (max_nums <= 0) {
    THROW_MQEXCEPTION(MQClientException, "maxNums <= 0", -1);
  }

  int sysFlag = PullSysFlag::buildSysFlag(false, block, true, false, true);

  long timeoutMillis = block ? getDefaultLitePullConsumerConfig()->consumer_timeout_millis_when_suspend() : timeout;

  bool isTagType = ExpressionType::isTagType(subscription_data->expression_type());

  std::unique_ptr<PullResult> pull_result(pull_api_wrapper_->pullKernelImpl(
      mq,                                                                    // mq
      subscription_data->sub_string(),                                       // subExpression
      subscription_data->expression_type(),                                  // expressionType
      isTagType ? 0L : subscription_data->sub_version(),                     // subVersion
      offset,                                                                // offset
      max_nums,                                                              // maxNums
      sysFlag,                                                               // sysFlag
      0,                                                                     // commitOffset
      getDefaultLitePullConsumerConfig()->broker_suspend_max_time_millis(),  // brokerSuspendMaxTimeMillis
      timeoutMillis,                                                         // timeoutMillis
      CommunicationMode::SYNC,                                               // communicationMode
      nullptr));                                                             // pullCallback

  return pull_api_wrapper_->processPullResult(mq, std::move(pull_result), subscription_data);
}

void DefaultLitePullConsumerImpl::submitConsumeRequest(ConsumeRequest* consume_request) {
  consume_request_cache_.push_back(consume_request);
}

void DefaultLitePullConsumerImpl::updatePullOffset(const MQMessageQueue& message_queue, int64_t next_pull_offset) {
  if (assigned_message_queue_->getSeekOffset(message_queue) == -1) {
    assigned_message_queue_->updatePullOffset(message_queue, next_pull_offset);
  }
}

std::vector<MQMessageExt> DefaultLitePullConsumerImpl::poll() {
  return poll(getDefaultLitePullConsumerConfig()->poll_timeout_millis());
}

std::vector<MQMessageExt> DefaultLitePullConsumerImpl::poll(long timeout) {
  // checkServiceState();
  if (auto_commit_) {
    maybeAutoCommit();
  }

  int64_t endTime = UtilAll::currentTimeMillis() + timeout;

  auto consume_request = consume_request_cache_.pop_front(timeout, time_unit::milliseconds);
  if (endTime - UtilAll::currentTimeMillis() > 0) {
    while (consume_request != nullptr && consume_request->process_queue()->dropped()) {
      consume_request = consume_request_cache_.pop_front();
      if (endTime - UtilAll::currentTimeMillis() <= 0) {
        break;
      }
    }
  }

  if (consume_request != nullptr && !consume_request->process_queue()->dropped()) {
    auto& messages = consume_request->message_exts();
    long offset = consume_request->process_queue()->removeMessage(messages);
    assigned_message_queue_->updateConsumeOffset(consume_request->message_queue(), offset);
    // If namespace not null , reset Topic without namespace.
    resetTopic(messages);
    return MQMessageExt::from_list(messages);
  }

  return std::vector<MQMessageExt>();
}

void DefaultLitePullConsumerImpl::maybeAutoCommit() {
  auto now = UtilAll::currentTimeMillis();
  if (now >= next_auto_commit_deadline_) {
    commitAll();
    next_auto_commit_deadline_ = now + getDefaultLitePullConsumerConfig()->auto_commit_interval_millis();
  }
}

void DefaultLitePullConsumerImpl::resetTopic(std::vector<MessageExtPtr>& msg_list) {
  if (msg_list.empty()) {
    return;
  }

  // If namespace not null , reset Topic without namespace.
  const auto& name_space = getDefaultLitePullConsumerConfig()->name_space();
  if (!name_space.empty()) {
    for (auto& message_ext : msg_list) {
      message_ext->set_topic(NamespaceUtil::withoutNamespace(message_ext->topic(), name_space));
    }
  }
}

void DefaultLitePullConsumerImpl::commitAll() {
  try {
    std::vector<MQMessageQueue> message_queues = assigned_message_queue_->messageQueues();
    for (const auto& message_queue : message_queues) {
      long consumer_offset = assigned_message_queue_->getConsumerOffset(message_queue);
      if (consumer_offset != -1) {
        auto process_queue = assigned_message_queue_->getProcessQueue(message_queue);
        if (process_queue != nullptr && !process_queue->dropped()) {
          updateConsumeOffset(message_queue, consumer_offset);
        }
      }
    }
    if (getDefaultLitePullConsumerConfig()->message_model() == MessageModel::BROADCASTING) {
      offset_store_->persistAll(message_queues);
    }
  } catch (std::exception& e) {
    LOG_ERROR_NEW("An error occurred when update consume offset Automatically.");
  }
}

void DefaultLitePullConsumerImpl::updateConsumeOffset(const MQMessageQueue& mq, int64_t offset) {
  // checkServiceState();
  offset_store_->updateOffset(mq, offset, false);
}

void DefaultLitePullConsumerImpl::persistConsumerOffset() {
  if (isServiceStateOk()) {
    std::vector<MQMessageQueue> allocated_mqs;
    if (subscription_type_ == SubscriptionType::SUBSCRIBE) {
      allocated_mqs = rebalance_impl_->getAllocatedMQ();
    } else if (subscription_type_ == SubscriptionType::ASSIGN) {
      allocated_mqs = assigned_message_queue_->messageQueues();
    }
    offset_store_->persistAll(allocated_mqs);
  }
}

std::vector<MQMessageQueue> DefaultLitePullConsumerImpl::fetchMessageQueues(const std::string& topic) {
  std::vector<MQMessageQueue> result;
  if (isServiceStateOk()) {
    client_instance_->getMQAdminImpl()->fetchSubscribeMessageQueues(topic, result);
    parseMessageQueues(result);
  }
  return result;
}

void DefaultLitePullConsumerImpl::parseMessageQueues(std::vector<MQMessageQueue>& queueSet) {
  const auto& name_space = client_config_->name_space();
  if (name_space.empty()) {
    return;
  }
  for (auto& messageQueue : queueSet) {
    auto user_topic = NamespaceUtil::withoutNamespace(messageQueue.topic(), name_space);
    messageQueue.set_topic(user_topic);
  }
}

void DefaultLitePullConsumerImpl::assign(const std::vector<MQMessageQueue>& messageQueues) {
  // TODO:
}

void DefaultLitePullConsumerImpl::seek(const MQMessageQueue& messageQueue, int64_t offset) {
  // TODO:
}

void DefaultLitePullConsumerImpl::seekToBegin(const MQMessageQueue& message_queue) {
  auto begin = minOffset(message_queue);
  seek(message_queue, begin);
}

void DefaultLitePullConsumerImpl::seekToEnd(const MQMessageQueue& message_queue) {
  auto end = maxOffset(message_queue);
  seek(message_queue, end);
}

int64_t DefaultLitePullConsumerImpl::offsetForTimestamp(const MQMessageQueue& message_queue, int64_t timestamp) {
  return searchOffset(message_queue, timestamp);
}

void DefaultLitePullConsumerImpl::pause(const std::vector<MQMessageQueue>& message_queues) {
  assigned_message_queue_->pause(message_queues);
}

void DefaultLitePullConsumerImpl::resume(const std::vector<MQMessageQueue>& message_queues) {
  assigned_message_queue_->resume(message_queues);
}

void DefaultLitePullConsumerImpl::commitSync() {
  commitAll();
}

int64_t DefaultLitePullConsumerImpl::committed(const MQMessageQueue& message_queue) {
  // checkServiceState();
  auto offset = offset_store_->readOffset(message_queue, ReadOffsetType::MEMORY_FIRST_THEN_STORE);
  if (offset == -2) {
    THROW_MQEXCEPTION(MQClientException, "Fetch consume offset from broker exception", -1);
  }
  return offset;
}

void DefaultLitePullConsumerImpl::registerTopicMessageQueueChangeListener(
    const std::string& topic,
    TopicMessageQueueChangeListener* topicMessageQueueChangeListener) {
  std::lock_guard<std::mutex> lock(mutex_);  // synchronized
  if (topic.empty() || nullptr == topicMessageQueueChangeListener) {
    THROW_MQEXCEPTION(MQClientException, "Topic or listener is null", -1);
  }
  if (topic_message_queue_change_listener_map_.find(topic) != topic_message_queue_change_listener_map_.end()) {
    LOG_WARN_NEW("Topic {} had been registered, new listener will overwrite the old one", topic);
  }

  topic_message_queue_change_listener_map_[topic] = topicMessageQueueChangeListener;
  if (service_state_ == ServiceState::RUNNING) {
    auto messageQueues = fetchMessageQueues(topic);
    message_queues_for_topic_[topic] = std::move(messageQueues);
  }
}

std::unique_ptr<ConsumerRunningInfo> DefaultLitePullConsumerImpl::consumerRunningInfo() {
  std::unique_ptr<ConsumerRunningInfo> info(new ConsumerRunningInfo());

  info->setProperty(ConsumerRunningInfo::PROP_CONSUMER_START_TIMESTAMP, UtilAll::to_string(start_time_));

  info->setSubscriptionSet(subscriptions());

  auto processQueueTable = rebalance_impl_->getProcessQueueTable();
  for (const auto& it : processQueueTable) {
    const auto& mq = it.first;
    const auto& pq = it.second;

    ProcessQueueInfo pq_info;
    pq_info.setCommitOffset(offset_store_->readOffset(mq, MEMORY_FIRST_THEN_STORE));
    pq->fillProcessQueueInfo(pq_info);
    info->setMqTable(mq, pq_info);
  }

  return info;
}

bool DefaultLitePullConsumerImpl::isAutoCommit() const {
  return auto_commit_;
}

void DefaultLitePullConsumerImpl::setAutoCommit(bool auto_commit) {
  auto_commit_ = auto_commit;
}

const std::string& DefaultLitePullConsumerImpl::groupName() const {
  return client_config_->group_name();
}

MessageModel DefaultLitePullConsumerImpl::messageModel() const {
  return getDefaultLitePullConsumerConfig()->message_model();
};

ConsumeType DefaultLitePullConsumerImpl::consumeType() const {
  return CONSUME_ACTIVELY;
}

ConsumeFromWhere DefaultLitePullConsumerImpl::consumeFromWhere() const {
  return getDefaultLitePullConsumerConfig()->consume_from_where();
}

void DefaultLitePullConsumerImpl::set_subscription_type(SubscriptionType subscription_type) {
  if (subscription_type_ == SubscriptionType::NONE) {
    subscription_type_ = subscription_type;
  } else if (subscription_type_ != subscription_type) {
    THROW_MQEXCEPTION(MQClientException, "Subscribe and assign are mutually exclusive.", -1);
  }
}

}  // namespace rocketmq