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

#include <cassert>
#include <cstring>

#include <typeindex>

#include "ClientRemotingProcessor.h"
#include "MQClientInstance.h"
#include "MessageBatch.h"
#include "MessageClientIDSetter.h"
#include "PullCallbackWrap.h"
#include "PullResultExt.hpp"
#include "SendCallbackWrap.h"
#include "TcpRemotingClient.h"
#include "protocol/body/LockBatchResponseBody.hpp"

namespace rocketmq {

MQClientAPIImpl::MQClientAPIImpl(ClientRemotingProcessor* clientRemotingProcessor,
                                 RPCHookPtr rpcHook,
                                 const MQClientConfig& clientConfig)
    : remoting_client_(new TcpRemotingClient(clientConfig.tcp_transport_worker_thread_nums(),
                                             clientConfig.tcp_transport_connect_timeout(),
                                             clientConfig.tcp_transport_try_lock_timeout())) {
  remoting_client_->registerRPCHook(rpcHook);
  remoting_client_->registerProcessor(CHECK_TRANSACTION_STATE, clientRemotingProcessor);
  remoting_client_->registerProcessor(NOTIFY_CONSUMER_IDS_CHANGED, clientRemotingProcessor);
  remoting_client_->registerProcessor(RESET_CONSUMER_CLIENT_OFFSET, clientRemotingProcessor);
  remoting_client_->registerProcessor(GET_CONSUMER_STATUS_FROM_CLIENT, clientRemotingProcessor);
  remoting_client_->registerProcessor(GET_CONSUMER_RUNNING_INFO, clientRemotingProcessor);
  remoting_client_->registerProcessor(CONSUME_MESSAGE_DIRECTLY, clientRemotingProcessor);
  remoting_client_->registerProcessor(PUSH_REPLY_MESSAGE_TO_CLIENT, clientRemotingProcessor);
}

MQClientAPIImpl::~MQClientAPIImpl() = default;

void MQClientAPIImpl::start() {
  remoting_client_->start();
}

void MQClientAPIImpl::shutdown() {
  remoting_client_->shutdown();
}

void MQClientAPIImpl::updateNameServerAddressList(const std::string& addrs) {
  // TODO: split addrs
  remoting_client_->updateNameServerAddressList(addrs);
}

void MQClientAPIImpl::createTopic(const std::string& addr, const std::string& defaultTopic, TopicConfig topicConfig) {
  auto* requestHeader = new CreateTopicRequestHeader();
  requestHeader->topic = topicConfig.topic_name();
  requestHeader->defaultTopic = defaultTopic;
  requestHeader->readQueueNums = topicConfig.read_queue_nums();
  requestHeader->writeQueueNums = topicConfig.write_queue_nums();
  requestHeader->perm = topicConfig.perm();
  requestHeader->topicFilterType = topicConfig.topic_filter_type();

  RemotingCommand request(UPDATE_AND_CREATE_TOPIC, requestHeader);

  std::unique_ptr<RemotingCommand> response(remoting_client_->invokeSync(addr, request));
  assert(response != nullptr);
  switch (response->code()) {
    case SUCCESS: {
      return;
    }
    default:
      break;
  }

  THROW_MQEXCEPTION(MQBrokerException, response->remark(), response->code());
}

std::unique_ptr<SendResult> MQClientAPIImpl::sendMessage(const std::string& addr,
                                                         const std::string& brokerName,
                                                         const MessagePtr msg,
                                                         std::unique_ptr<SendMessageRequestHeader> requestHeader,
                                                         int timeoutMillis,
                                                         CommunicationMode communicationMode,
                                                         DefaultMQProducerImplPtr producer) {
  return sendMessage(addr, brokerName, msg, std::move(requestHeader), timeoutMillis, communicationMode, nullptr,
                     nullptr, nullptr, 0, producer);
}

std::unique_ptr<SendResult> MQClientAPIImpl::sendMessage(const std::string& addr,
                                                         const std::string& brokerName,
                                                         const MessagePtr msg,
                                                         std::unique_ptr<SendMessageRequestHeader> requestHeader,
                                                         int timeoutMillis,
                                                         CommunicationMode communicationMode,
                                                         SendCallback* sendCallback,
                                                         TopicPublishInfoPtr topicPublishInfo,
                                                         MQClientInstancePtr instance,
                                                         int retryTimesWhenSendFailed,
                                                         DefaultMQProducerImplPtr producer) {
  int code = SEND_MESSAGE;
  std::unique_ptr<CommandCustomHeader> header;

  const auto& msgType = msg->getProperty(MQMessageConst::PROPERTY_MESSAGE_TYPE);
  bool isReply = msgType == REPLY_MESSAGE_FLAG;
  if (isReply) {
    code = SEND_REPLY_MESSAGE_V2;
  } else if (msg->isBatch()) {
    code = SEND_BATCH_MESSAGE;
  } else {
    code = SEND_MESSAGE_V2;
  }

  if (code != SEND_MESSAGE && code != SEND_REPLY_MESSAGE) {
    header = SendMessageRequestHeaderV2::createSendMessageRequestHeaderV2(requestHeader.get());
  } else {
    header = std::move(requestHeader);
  }

  RemotingCommand request(code, header.release());
  request.set_body(msg->body());

  switch (communicationMode) {
    case CommunicationMode::ONEWAY:
      remoting_client_->invokeOneway(addr, request);
      return nullptr;
    case CommunicationMode::ASYNC:
      sendMessageAsync(addr, brokerName, msg, std::move(request), sendCallback, topicPublishInfo, instance,
                       timeoutMillis, retryTimesWhenSendFailed, producer);
      return nullptr;
    case CommunicationMode::SYNC:
      return sendMessageSync(addr, brokerName, msg, request, timeoutMillis);
    default:
      assert(false);
      break;
  }

  return nullptr;
}

void MQClientAPIImpl::sendMessageAsync(const std::string& addr,
                                       const std::string& brokerName,
                                       const MessagePtr msg,
                                       RemotingCommand&& request,
                                       SendCallback* sendCallback,
                                       TopicPublishInfoPtr topicPublishInfo,
                                       MQClientInstancePtr instance,
                                       int64_t timeoutMillis,
                                       int retryTimesWhenSendFailed,
                                       DefaultMQProducerImplPtr producer) {
  std::unique_ptr<InvokeCallback> cbw(
      new SendCallbackWrap(addr, brokerName, msg, std::forward<RemotingCommand>(request), sendCallback,
                           topicPublishInfo, instance, retryTimesWhenSendFailed, 0, producer));
  sendMessageAsyncImpl(cbw, timeoutMillis);
}

void MQClientAPIImpl::sendMessageAsyncImpl(std::unique_ptr<InvokeCallback>& cbw, int64_t timeoutMillis) {
  auto* scbw = static_cast<SendCallbackWrap*>(cbw.get());
  const auto& addr = scbw->getAddr();
  auto& request = scbw->getRemotingCommand();
  remoting_client_->invokeAsync(addr, request, cbw, timeoutMillis);
}

std::unique_ptr<SendResult> MQClientAPIImpl::sendMessageSync(const std::string& addr,
                                                             const std::string& brokerName,
                                                             const MessagePtr msg,
                                                             RemotingCommand& request,
                                                             int timeoutMillis) {
  // block until response
  std::unique_ptr<RemotingCommand> response(remoting_client_->invokeSync(addr, request, timeoutMillis));
  assert(response != nullptr);
  return processSendResponse(brokerName, msg, response.get());
}

std::unique_ptr<SendResult> MQClientAPIImpl::processSendResponse(const std::string& brokerName,
                                                                 const MessagePtr msg,
                                                                 RemotingCommand* response) {
  SendStatus sendStatus = SEND_OK;
  switch (response->code()) {
    case FLUSH_DISK_TIMEOUT:
      sendStatus = SEND_FLUSH_DISK_TIMEOUT;
      break;
    case FLUSH_SLAVE_TIMEOUT:
      sendStatus = SEND_FLUSH_SLAVE_TIMEOUT;
      break;
    case SLAVE_NOT_AVAILABLE:
      sendStatus = SEND_SLAVE_NOT_AVAILABLE;
      break;
    case SUCCESS:
      sendStatus = SEND_OK;
      break;
    default:
      THROW_MQEXCEPTION(MQBrokerException, response->remark(), response->code());
      return nullptr;
  }

  auto* responseHeader = response->decodeCommandCustomHeader<SendMessageResponseHeader>();
  assert(responseHeader != nullptr);

  MQMessageQueue messageQueue(msg->topic(), brokerName, responseHeader->queueId);

  std::string uniqMsgId = MessageClientIDSetter::getUniqID(*msg);

  // MessageBatch
  if (msg->isBatch()) {
    const auto& messages = dynamic_cast<MessageBatch*>(msg.get())->messages();
    uniqMsgId.clear();
    uniqMsgId.reserve(33 * messages.size() + 1);
    for (const auto& message : messages) {
      uniqMsgId.append(MessageClientIDSetter::getUniqID(message));
      uniqMsgId.append(",");
    }
    if (!uniqMsgId.empty()) {
      uniqMsgId.resize(uniqMsgId.length() - 1);
    }
  }

  std::unique_ptr<SendResult> sendResult(
      new SendResult(sendStatus, uniqMsgId, responseHeader->msgId, messageQueue, responseHeader->queueOffset));
  sendResult->set_transaction_id(responseHeader->transactionId);

  return sendResult;
}

std::unique_ptr<PullResult> MQClientAPIImpl::pullMessage(const std::string& addr,
                                                         PullMessageRequestHeader* requestHeader,
                                                         int timeoutMillis,
                                                         CommunicationMode communicationMode,
                                                         PullCallback* pullCallback) {
  RemotingCommand request(PULL_MESSAGE, requestHeader);

  switch (communicationMode) {
    case CommunicationMode::ASYNC:
      pullMessageAsync(addr, request, timeoutMillis, pullCallback);
      return nullptr;
    case CommunicationMode::SYNC:
      return pullMessageSync(addr, request, timeoutMillis);
    default:
      assert(false);
      return nullptr;
  }
}

void MQClientAPIImpl::pullMessageAsync(const std::string& addr,
                                       RemotingCommand& request,
                                       int timeoutMillis,
                                       PullCallback* pullCallback) {
  std::unique_ptr<InvokeCallback> cbw(new PullCallbackWrap(pullCallback, this));
  remoting_client_->invokeAsync(addr, request, cbw, timeoutMillis);
}

std::unique_ptr<PullResult> MQClientAPIImpl::pullMessageSync(const std::string& addr,
                                                             RemotingCommand& request,
                                                             int timeoutMillis) {
  std::unique_ptr<RemotingCommand> response(remoting_client_->invokeSync(addr, request, timeoutMillis));
  assert(response != nullptr);
  return processPullResponse(response.get());
}

std::unique_ptr<PullResult> MQClientAPIImpl::processPullResponse(RemotingCommand* response) {
  PullStatus pullStatus = NO_NEW_MSG;
  switch (response->code()) {
    case SUCCESS:
      pullStatus = FOUND;
      break;
    case PULL_NOT_FOUND:
      pullStatus = NO_NEW_MSG;
      break;
    case PULL_RETRY_IMMEDIATELY:
      if ("OFFSET_OVERFLOW_BADLY" == response->remark()) {
        pullStatus = NO_LATEST_MSG;
      } else {
        pullStatus = NO_MATCHED_MSG;
      }
      break;
    case PULL_OFFSET_MOVED:
      pullStatus = OFFSET_ILLEGAL;
      break;
    default:
      THROW_MQEXCEPTION(MQBrokerException, response->remark(), response->code());
  }

  // return of decodeCommandCustomHeader is non-null
  auto* responseHeader = response->decodeCommandCustomHeader<PullMessageResponseHeader>();
  assert(responseHeader != nullptr);

  return std::unique_ptr<PullResult>(new PullResultExt(pullStatus, responseHeader->nextBeginOffset,
                                                       responseHeader->minOffset, responseHeader->maxOffset,
                                                       (int)responseHeader->suggestWhichBrokerId, response->body()));
}

MQMessageExt MQClientAPIImpl::viewMessage(const std::string& addr, int64_t phyoffset, int timeoutMillis) {
  auto* requestHeader = new ViewMessageRequestHeader();
  requestHeader->offset = phyoffset;

  RemotingCommand request(VIEW_MESSAGE_BY_ID, requestHeader);

  std::unique_ptr<RemotingCommand> response(remoting_client_->invokeSync(addr, request, timeoutMillis));
  assert(response != nullptr);
  switch (response->code()) {
    case SUCCESS: {
      // TODO: ...
    }
    default:
      break;
  }

  THROW_MQEXCEPTION(MQBrokerException, response->remark(), response->code());
}

int64_t MQClientAPIImpl::searchOffset(const std::string& addr,
                                      const std::string& topic,
                                      int queueId,
                                      int64_t timestamp,
                                      int timeoutMillis) {
  auto* requestHeader = new SearchOffsetRequestHeader();
  requestHeader->topic = topic;
  requestHeader->queueId = queueId;
  requestHeader->timestamp = timestamp;

  RemotingCommand request(SEARCH_OFFSET_BY_TIMESTAMP, requestHeader);

  std::unique_ptr<RemotingCommand> response(remoting_client_->invokeSync(addr, request, timeoutMillis));
  assert(response != nullptr);
  switch (response->code()) {
    case SUCCESS: {
      auto* responseHeader = response->decodeCommandCustomHeader<SearchOffsetResponseHeader>();
      assert(responseHeader != nullptr);
      return responseHeader->offset;
    }
    default:
      break;
  }

  THROW_MQEXCEPTION(MQBrokerException, response->remark(), response->code());
}

int64_t MQClientAPIImpl::getMaxOffset(const std::string& addr,
                                      const std::string& topic,
                                      int queueId,
                                      int timeoutMillis) {
  auto* requestHeader = new GetMaxOffsetRequestHeader();
  requestHeader->topic = topic;
  requestHeader->queueId = queueId;

  RemotingCommand request(GET_MAX_OFFSET, requestHeader);

  std::unique_ptr<RemotingCommand> response(remoting_client_->invokeSync(addr, request, timeoutMillis));
  assert(response != nullptr);
  switch (response->code()) {
    case SUCCESS: {
      auto* responseHeader = response->decodeCommandCustomHeader<GetMaxOffsetResponseHeader>();
      return responseHeader->offset;
    }
    default:
      break;
  }

  THROW_MQEXCEPTION(MQBrokerException, response->remark(), response->code());
}

int64_t MQClientAPIImpl::getMinOffset(const std::string& addr,
                                      const std::string& topic,
                                      int queueId,
                                      int timeoutMillis) {
  auto* requestHeader = new GetMinOffsetRequestHeader();
  requestHeader->topic = topic;
  requestHeader->queueId = queueId;

  RemotingCommand request(GET_MIN_OFFSET, requestHeader);

  std::unique_ptr<RemotingCommand> response(remoting_client_->invokeSync(addr, request, timeoutMillis));
  assert(response != nullptr);
  switch (response->code()) {
    case SUCCESS: {
      auto* responseHeader = response->decodeCommandCustomHeader<GetMinOffsetResponseHeader>();
      assert(responseHeader != nullptr);
      return responseHeader->offset;
    }
    default:
      break;
  }

  THROW_MQEXCEPTION(MQBrokerException, response->remark(), response->code());
}

int64_t MQClientAPIImpl::getEarliestMsgStoretime(const std::string& addr,
                                                 const std::string& topic,
                                                 int queueId,
                                                 int timeoutMillis) {
  auto* requestHeader = new GetEarliestMsgStoretimeRequestHeader();
  requestHeader->topic = topic;
  requestHeader->queueId = queueId;

  RemotingCommand request(GET_EARLIEST_MSG_STORETIME, requestHeader);

  std::unique_ptr<RemotingCommand> response(remoting_client_->invokeSync(addr, request, timeoutMillis));
  assert(response != nullptr);
  switch (response->code()) {
    case SUCCESS: {
      auto* responseHeader = response->decodeCommandCustomHeader<GetEarliestMsgStoretimeResponseHeader>();
      assert(responseHeader != nullptr);
      return responseHeader->timestamp;
    }
    default:
      break;
  }

  THROW_MQEXCEPTION(MQBrokerException, response->remark(), response->code());
}

void MQClientAPIImpl::getConsumerIdListByGroup(const std::string& addr,
                                               const std::string& consumerGroup,
                                               std::vector<std::string>& cids,
                                               int timeoutMillis) {
  auto* requestHeader = new GetConsumerListByGroupRequestHeader();
  requestHeader->consumerGroup = consumerGroup;

  RemotingCommand request(GET_CONSUMER_LIST_BY_GROUP, requestHeader);

  std::unique_ptr<RemotingCommand> response(remoting_client_->invokeSync(addr, request, timeoutMillis));
  assert(response != nullptr);
  switch (response->code()) {
    case SUCCESS: {
      auto responseBody = response->body();
      if (responseBody != nullptr && responseBody->size() > 0) {
        std::unique_ptr<GetConsumerListByGroupResponseBody> body(
            GetConsumerListByGroupResponseBody::Decode(*responseBody));
        cids = std::move(body->consumerIdList);
        return;
      }
    }
    case SYSTEM_ERROR:
    // no consumer for this group
    default:
      break;
  }

  THROW_MQEXCEPTION(MQBrokerException, response->remark(), response->code());
}

int64_t MQClientAPIImpl::queryConsumerOffset(const std::string& addr,
                                             QueryConsumerOffsetRequestHeader* requestHeader,
                                             int timeoutMillis) {
  RemotingCommand request(QUERY_CONSUMER_OFFSET, requestHeader);

  std::unique_ptr<RemotingCommand> response(remoting_client_->invokeSync(addr, request, timeoutMillis));
  assert(response != nullptr);
  switch (response->code()) {
    case SUCCESS: {
      auto* responseHeader = response->decodeCommandCustomHeader<QueryConsumerOffsetResponseHeader>();
      assert(responseHeader != nullptr);
      return responseHeader->offset;
    }
    default:
      break;
  }

  THROW_MQEXCEPTION(MQBrokerException, response->remark(), response->code());
}

void MQClientAPIImpl::updateConsumerOffset(const std::string& addr,
                                           UpdateConsumerOffsetRequestHeader* requestHeader,
                                           int timeoutMillis) {
  RemotingCommand request(UPDATE_CONSUMER_OFFSET, requestHeader);

  std::unique_ptr<RemotingCommand> response(remoting_client_->invokeSync(addr, request, timeoutMillis));
  assert(response != nullptr);
  switch (response->code()) {
    case SUCCESS: {
      return;
    }
    default:
      break;
  }

  THROW_MQEXCEPTION(MQBrokerException, response->remark(), response->code());
}

void MQClientAPIImpl::updateConsumerOffsetOneway(const std::string& addr,
                                                 UpdateConsumerOffsetRequestHeader* requestHeader,
                                                 int timeoutMillis) {
  RemotingCommand request(UPDATE_CONSUMER_OFFSET, requestHeader);

  remoting_client_->invokeOneway(addr, request);
}

void MQClientAPIImpl::sendHearbeat(const std::string& addr, HeartbeatData* heartbeatData, long timeoutMillis) {
  RemotingCommand request(HEART_BEAT, nullptr);
  request.set_body(heartbeatData->encode());

  std::unique_ptr<RemotingCommand> response(remoting_client_->invokeSync(addr, request, timeoutMillis));
  assert(response != nullptr);
  switch (response->code()) {
    case SUCCESS: {
      LOG_DEBUG_NEW("sendHeartbeat to broker:{} success", addr);
      return;
    }
    default:
      break;
  }

  THROW_MQEXCEPTION(MQBrokerException, response->remark(), response->code());
}

void MQClientAPIImpl::unregisterClient(const std::string& addr,
                                       const std::string& clientID,
                                       const std::string& producerGroup,
                                       const std::string& consumerGroup) {
  LOG_INFO("unregisterClient to broker:%s", addr.c_str());
  RemotingCommand request(UNREGISTER_CLIENT, new UnregisterClientRequestHeader(clientID, producerGroup, consumerGroup));

  std::unique_ptr<RemotingCommand> response(remoting_client_->invokeSync(addr, request));
  assert(response != nullptr);
  switch (response->code()) {
    case SUCCESS:
      LOG_INFO("unregisterClient to:%s success", addr.c_str());
      return;
    default:
      break;
  }

  LOG_WARN("unregisterClient fail:%s, %d", response->remark().c_str(), response->code());
  THROW_MQEXCEPTION(MQBrokerException, response->remark(), response->code());
}

void MQClientAPIImpl::endTransactionOneway(const std::string& addr,
                                           EndTransactionRequestHeader* requestHeader,
                                           const std::string& remark) {
  RemotingCommand request(END_TRANSACTION, requestHeader);
  request.set_remark(remark);

  remoting_client_->invokeOneway(addr, request);
}

void MQClientAPIImpl::consumerSendMessageBack(const std::string& addr,
                                              MessageExtPtr msg,
                                              const std::string& consumerGroup,
                                              int delayLevel,
                                              int timeoutMillis,
                                              int maxConsumeRetryTimes) {
  auto* requestHeader = new ConsumerSendMsgBackRequestHeader();
  requestHeader->group = consumerGroup;
  requestHeader->originTopic = msg->topic();
  requestHeader->offset = msg->commit_log_offset();
  requestHeader->delayLevel = delayLevel;
  requestHeader->originMsgId = msg->msg_id();
  requestHeader->maxReconsumeTimes = maxConsumeRetryTimes;

  RemotingCommand request(CONSUMER_SEND_MSG_BACK, requestHeader);

  std::unique_ptr<RemotingCommand> response(remoting_client_->invokeSync(addr, request, timeoutMillis));
  assert(response != nullptr);
  switch (response->code()) {
    case SUCCESS: {
      return;
    }
    default:
      break;
  }

  THROW_MQEXCEPTION(MQBrokerException, response->remark(), response->code());
}

void MQClientAPIImpl::lockBatchMQ(const std::string& addr,
                                  LockBatchRequestBody* requestBody,
                                  std::vector<MQMessageQueue>& mqs,
                                  int timeoutMillis) {
  RemotingCommand request(LOCK_BATCH_MQ, nullptr);
  request.set_body(requestBody->encode());

  std::unique_ptr<RemotingCommand> response(remoting_client_->invokeSync(addr, request, timeoutMillis));
  assert(response != nullptr);
  switch (response->code()) {
    case SUCCESS: {
      auto requestBody = response->body();
      if (requestBody != nullptr && requestBody->size() > 0) {
        std::unique_ptr<LockBatchResponseBody> body(LockBatchResponseBody::Decode(*requestBody));
        mqs = std::move(body->lock_ok_mq_set());
      } else {
        mqs.clear();
      }
      return;
    } break;
    default:
      break;
  }

  THROW_MQEXCEPTION(MQBrokerException, response->remark(), response->code());
}

void MQClientAPIImpl::unlockBatchMQ(const std::string& addr,
                                    UnlockBatchRequestBody* requestBody,
                                    int timeoutMillis,
                                    bool oneway) {
  RemotingCommand request(UNLOCK_BATCH_MQ, nullptr);
  request.set_body(requestBody->encode());

  if (oneway) {
    remoting_client_->invokeOneway(addr, request);
  } else {
    std::unique_ptr<RemotingCommand> response(remoting_client_->invokeSync(addr, request, timeoutMillis));
    assert(response != nullptr);
    switch (response->code()) {
      case SUCCESS: {
        return;
      } break;
      default:
        break;
    }

    THROW_MQEXCEPTION(MQBrokerException, response->remark(), response->code());
  }
}

std::unique_ptr<TopicRouteData> MQClientAPIImpl::getTopicRouteInfoFromNameServer(const std::string& topic,
                                                                                 int timeoutMillis) {
  RemotingCommand request(GET_ROUTEINFO_BY_TOPIC, new GetRouteInfoRequestHeader(topic));

  std::unique_ptr<RemotingCommand> response(remoting_client_->invokeSync(null, request, timeoutMillis));
  assert(response != nullptr);
  switch (response->code()) {
    case SUCCESS: {
      auto responseBody = response->body();
      if (responseBody != nullptr && responseBody->size() > 0) {
        return TopicRouteData::Decode(*responseBody);
      }
    }
    case TOPIC_NOT_EXIST:
    default:
      break;
  }

  THROW_MQEXCEPTION(MQClientException, response->remark(), response->code());
}

std::unique_ptr<TopicList> MQClientAPIImpl::getTopicListFromNameServer() {
  RemotingCommand request(GET_ALL_TOPIC_LIST_FROM_NAMESERVER, nullptr);

  std::unique_ptr<RemotingCommand> response(remoting_client_->invokeSync(null, request));
  assert(response != nullptr);
  switch (response->code()) {
    case SUCCESS: {
      auto responseBody = response->body();
      if (responseBody != nullptr && responseBody->size() > 0) {
        return TopicList::Decode(*responseBody);
      }
    }
    default:
      break;
  }

  THROW_MQEXCEPTION(MQClientException, response->remark(), response->code());
}

int MQClientAPIImpl::wipeWritePermOfBroker(const std::string& namesrvAddr,
                                           const std::string& brokerName,
                                           int timeoutMillis) {
  return 0;
}

void MQClientAPIImpl::deleteTopicInBroker(const std::string& addr, const std::string& topic, int timeoutMillis) {}

void MQClientAPIImpl::deleteTopicInNameServer(const std::string& addr, const std::string& topic, int timeoutMillis) {}

void MQClientAPIImpl::deleteSubscriptionGroup(const std::string& addr,
                                              const std::string& groupName,
                                              int timeoutMillis) {}

std::string MQClientAPIImpl::getKVConfigByValue(const std::string& projectNamespace,
                                                const std::string& projectGroup,
                                                int timeoutMillis) {
  return "";
}

void MQClientAPIImpl::deleteKVConfigByValue(const std::string& projectNamespace,
                                            const std::string& projectGroup,
                                            int timeoutMillis) {}

KVTable MQClientAPIImpl::getKVListByNamespace(const std::string& projectNamespace, int timeoutMillis) {
  return KVTable();
}

}  // namespace rocketmq