text: /*
   * 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 "RemotingCommand.h"
  #include "Logging.h"
  #include "MQProtos.h"
  #include "MQVersion.h"
  #include "SessionCredentials.h"
  
  namespace rocketmq {
  boost::atomic<int> RemotingCommand::s_seqNumber;
  boost::mutex RemotingCommand::m_clock;
  //<!************************************************************************
  RemotingCommand::RemotingCommand(int code,
                                   CommandHeader* pExtHeader /* = NULL */)
      : m_code(code),
        m_language("CPP"),
        m_version(MQVersion::s_CurrentVersion),
        m_flag(0),
        m_remark(""),
        m_pExtHeader(pExtHeader) {
    boost::lock_guard<boost::mutex> lock(m_clock);
    m_opaque = (s_seqNumber.load(boost::memory_order_acquire)) %
               (numeric_limits<int>::max());
    s_seqNumber.store(m_opaque, boost::memory_order_release);
    ++s_seqNumber;
  }
  
  RemotingCommand::RemotingCommand(int code, string language, int version,
                                   int opaque, int flag, string remark,
                                   CommandHeader* pExtHeader)
      : m_code(code),
        m_language(language),
        m_version(version),
        m_opaque(opaque),
        m_flag(flag),
        m_remark(remark),
        m_pExtHeader(pExtHeader) {}
  
  RemotingCommand::RemotingCommand(const RemotingCommand& command) {
      Assign(command);
  }
  
  RemotingCommand& RemotingCommand::operator=(const RemotingCommand& command) {
    if (this != &command) {
      Assign(command);
    }
    return *this;
  }
  
  RemotingCommand::~RemotingCommand() { m_pExtHeader = NULL; }
  
  void RemotingCommand::Assign(const RemotingCommand& command)
  {
      m_code = command.m_code;
      m_language = command.m_language;
      m_version = command.m_version;
      m_opaque = command.m_opaque;
      m_flag = command.m_flag;
      m_remark = command.m_remark;
      m_msgBody = command.m_msgBody;
      
      for (auto& it : command.m_extFields) {
        m_extFields[it.first] = it.second;
      }
      m_head = command.m_head;
      m_body = command.m_body;
      s_seqNumber.store(command.s_seqNumber.load());
      m_parsedJson = command.m_parsedJson;
      //m_pExtHeader = command.m_pExtHeader; //ignore this filed at this moment, if need please add it
  }
  
  void RemotingCommand::Encode() {
    Json::Value root;
    root["code"] = m_code;
    root["language"] = "CPP";
    root["version"] = m_version;
    root["opaque"] = m_opaque;
    root["flag"] = m_flag;
    root["remark"] = m_remark;
  
    if (m_pExtHeader) {
      Json::Value extJson;
      m_pExtHeader->Encode(extJson);
  
      extJson[SessionCredentials::Signature] =
          m_extFields[SessionCredentials::Signature];
      extJson[SessionCredentials::AccessKey] =
          m_extFields[SessionCredentials::AccessKey];
      extJson[SessionCredentials::ONSChannelKey] =
          m_extFields[SessionCredentials::ONSChannelKey];
  
      root["extFields"] = extJson;
    } else {  // for heartbeat
      Json::Value extJson;
      extJson[SessionCredentials::Signature] =
          m_extFields[SessionCredentials::Signature];
      extJson[SessionCredentials::AccessKey] =
          m_extFields[SessionCredentials::AccessKey];
      extJson[SessionCredentials::ONSChannelKey] =
          m_extFields[SessionCredentials::ONSChannelKey];
      root["extFields"] = extJson;
    }
  
    Json::FastWriter fastwrite;
    string data = fastwrite.write(root);
  
    uint32 headLen = data.size();
    uint32 totalLen = 4 + headLen + m_body.getSize();
  
    uint32 messageHeader[2];
    messageHeader[0] = htonl(totalLen);
    messageHeader[1] = htonl(headLen);
  
    //<!include self 4 bytes, see : doc/protocol.txt;
    m_head.setSize(4 + 4 + headLen);
    m_head.copyFrom(messageHeader, 0, sizeof(messageHeader));
    m_head.copyFrom(data.c_str(), sizeof(messageHeader), headLen);
  }
  
  const MemoryBlock* RemotingCommand::GetHead() const { return &m_head; }
  
  const MemoryBlock* RemotingCommand::GetBody() const { return &m_body; }
  
  void RemotingCommand::SetBody(const char* pData, int len) {
    m_body.reset();
    m_body.setSize(len);
    m_body.copyFrom(pData, 0, len);
  }
  
  RemotingCommand* RemotingCommand::Decode(const MemoryBlock& mem) {
    //<!decode 1 bytes,4+head+body
    uint32 messageHeader[1];
    mem.copyTo(messageHeader, 0, sizeof(messageHeader));
    int totalLen = mem.getSize();
    int headLen = ntohl(messageHeader[0]);
    int bodyLen = totalLen - 4 - headLen;
  
    //<!decode header;
    const char* const pData = static_cast<const char*>(mem.getData());
    Json::Reader reader;
    Json::Value object;
    const char* begin = pData + 4;
    const char* end = pData + 4 + headLen;
  
    if (!reader.parse(begin, end, object)) {
      THROW_MQEXCEPTION(MQClientException, "conn't parse json", -1);
    }
  
    int code = object["code"].asInt();
  
    string language = object["language"].asString();
    int version = object["version"].asInt();
    int opaque = object["opaque"].asInt();
    int flag = object["flag"].asInt();
    Json::Value v = object["remark"];
    string remark = "";
    if (!v.isNull()) {
      remark = object["remark"].asString();
    }
    LOG_DEBUG(
        "code:%d, remark:%s, version:%d, opaque:%d, flag:%d, remark:%s, "
        "headLen:%d, bodyLen:%d ",
        code, language.c_str(), version, opaque, flag, remark.c_str(), headLen,
        bodyLen);
    RemotingCommand* cmd =
        new RemotingCommand(code, language, version, opaque, flag, remark, NULL);
    cmd->setParsedJson(object);
    if (bodyLen > 0) {
      cmd->SetBody(pData + 4 + headLen, bodyLen);
    }
    return cmd;
  }
  
  void RemotingCommand::markResponseType() {
    int bits = 1 << RPC_TYPE;
    m_flag |= bits;
  }
  
  bool RemotingCommand::isResponseType() {
    int bits = 1 << RPC_TYPE;
    return (m_flag & bits) == bits;
  }
  
  void RemotingCommand::markOnewayRPC() {
    int bits = 1 << RPC_ONEWAY;
    m_flag |= bits;
  }
  
  bool RemotingCommand::isOnewayRPC() {
    int bits = 1 << RPC_ONEWAY;
    return (m_flag & bits) == bits;
  }
  
  void RemotingCommand::setOpaque(const int opa) { m_opaque = opa; }
  
  void RemotingCommand::SetExtHeader(int code) {
    try {
      Json::Value ext = m_parsedJson["extFields"];
      if (!ext.isNull()) {
        m_pExtHeader = NULL;
        switch (code) {
          case SEND_MESSAGE:
            m_pExtHeader.reset(SendMessageResponseHeader::Decode(ext));
            break;
          case PULL_MESSAGE:
            m_pExtHeader.reset(PullMessageResponseHeader::Decode(ext));
            break;
          case GET_MIN_OFFSET:
            m_pExtHeader.reset(GetMinOffsetResponseHeader::Decode(ext));
            break;
          case GET_MAX_OFFSET:
            m_pExtHeader.reset(GetMaxOffsetResponseHeader::Decode(ext));
            break;
          case SEARCH_OFFSET_BY_TIMESTAMP:
            m_pExtHeader.reset(SearchOffsetResponseHeader::Decode(ext));
            break;
          case GET_EARLIEST_MSG_STORETIME:
            m_pExtHeader.reset(
                GetEarliestMsgStoretimeResponseHeader::Decode(ext));
            break;
          case QUERY_CONSUMER_OFFSET:
            m_pExtHeader.reset(QueryConsumerOffsetResponseHeader::Decode(ext));
            break;
          case RESET_CONSUMER_CLIENT_OFFSET:
            m_pExtHeader.reset(ResetOffsetRequestHeader::Decode(ext));
            break;
          case GET_CONSUMER_RUNNING_INFO:
            m_pExtHeader.reset(GetConsumerRunningInfoRequestHeader::Decode(ext));
            break;
          case NOTIFY_CONSUMER_IDS_CHANGED:
            m_pExtHeader.reset(
                NotifyConsumerIdsChangedRequestHeader::Decode(ext));
          default:
            break;
        }
      }
    } catch (MQException& e) {
      LOG_ERROR("set response head error");
    }
  }
  
  void RemotingCommand::setCode(int code) { m_code = code; }
  
  int RemotingCommand::getCode() const { return m_code; }
  
  int RemotingCommand::getOpaque() const { return m_opaque; }
  
  string RemotingCommand::getRemark() const { return m_remark; }
  
  void RemotingCommand::setRemark(string mark) { m_remark = mark; }
  
  CommandHeader* RemotingCommand::getCommandHeader() const {
    return m_pExtHeader.get();
  }
  
  void RemotingCommand::setParsedJson(Json::Value json) { m_parsedJson = json; }
  
  const int RemotingCommand::getFlag() const { return m_flag; }
  
  const int RemotingCommand::getVersion() const { return m_version; }
  
  void RemotingCommand::setMsgBody(const string& body) { m_msgBody = body; }
  
  string RemotingCommand::getMsgBody() const { return m_msgBody; }
  
  void RemotingCommand::addExtField(const string& key, const string& value) {
    m_extFields[key] = value;
  }
  
  std::string RemotingCommand::ToString() const {
  	 std::stringstream ss;
  	 ss << "code:" << m_code
  	  <<",opaque:"<< m_opaque
  	  <<",flag:"<< m_flag
  	  <<",seqNumber:" << s_seqNumber
  	  <<",body.size:" << m_body.getSize()
  	  <<",header.size:" << m_head.getSize();
  	 return ss.str();
  }
  
  }  //<!end namespace;
  