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 "OffsetStore.h"
  #include "Logging.h"
  #include "MQClientFactory.h"
  #include "MessageQueue.h"
  
  #include <fstream>
  #include <sstream>
  
  #include <boost/archive/binary_iarchive.hpp>
  #include <boost/archive/binary_oarchive.hpp>
  #include <boost/archive/text_iarchive.hpp>
  #include <boost/archive/text_oarchive.hpp>
  #include <boost/filesystem.hpp>
  #include <boost/serialization/map.hpp>
  namespace rocketmq {
  
  //<!***************************************************************************
  OffsetStore::OffsetStore(const string& groupName, MQClientFactory* pfactory)
      : m_groupName(groupName), m_pClientFactory(pfactory) {}
  
  OffsetStore::~OffsetStore() {
    m_pClientFactory = NULL;
    m_offsetTable.clear();
  }
  
  //<!***************************************************************************
  LocalFileOffsetStore::LocalFileOffsetStore(const string& groupName, MQClientFactory* pfactory)
      : OffsetStore(groupName, pfactory) {
    MQConsumer* pConsumer = pfactory->selectConsumer(groupName);
    if (pConsumer) {
      // 广播消费模式下，每个消费者把自己的消费进度存储在本地文件
      LOG_INFO("new LocalFileOffsetStore");
      string directoryName = UtilAll::getLocalAddress() + "@" + pConsumer->getInstanceName();
      m_storePath = ".rocketmq_offsets/" + directoryName + "/" + groupName;
      string homeDir(UtilAll::getHomeDirectory());
      m_storeFile = homeDir + "/" + m_storePath + "/offsets.Json";
  
      string storePath(homeDir);
      storePath.append("/").append(m_storePath);
      boost::filesystem::path dir(storePath);
      boost::system::error_code ec;
      if (!boost::filesystem::exists(dir, ec)) {
        if (!boost::filesystem::create_directories(dir, ec)) {
          LOG_ERROR("create offset store dir:%s error", storePath.c_str());
          string errorMsg("create offset store dir fail: ");
          errorMsg.append(storePath);
          THROW_MQEXCEPTION(MQClientException, errorMsg, -1);
        }
      }
    }
  }
  
  LocalFileOffsetStore::~LocalFileOffsetStore() {}
  
  void LocalFileOffsetStore::load() {
    std::ifstream ifs(m_storeFile.c_str(), std::ios::in);
    if (ifs.good()) {
      if (ifs.is_open()) {
        if (ifs.peek() != std::ifstream::traits_type::eof()) {
          map<string, int64> m_offsetTable_tmp;
          boost::system::error_code e;
          try {
            boost::archive::text_iarchive ia(ifs);
            ia >> m_offsetTable_tmp;
          } catch (...) {
            LOG_ERROR(
                "load offset store file failed, please check whether file: %s is "
                "cleared by operator, if so, delete this offsets.Json file and "
                "then restart consumer",
                m_storeFile.c_str());
            ifs.close();
            string errorMsg("load offset store file: ");
            errorMsg.append(m_storeFile)
                .append(
                    " failed, please check whether offsets.Json is cleared by "
                    "operator, if so, delete this offsets.Json file and then "
                    "restart consumer");
            THROW_MQEXCEPTION(MQClientException, errorMsg, -1);
          }
          ifs.close();
  
          for (map<string, int64>::iterator it = m_offsetTable_tmp.begin(); it != m_offsetTable_tmp.end(); ++it) {
            // LOG_INFO("it->first:%s, it->second:%lld", it->first.c_str(),
            // it->second);
            Json::Reader reader;
            Json::Value object;
            reader.parse(it->first.c_str(), object);
            MQMessageQueue mq(object["topic"].asString(), object["brokerName"].asString(), object["queueId"].asInt());
            // 某个 topic 在某个 broker 的某个 queueid 的进度
            updateOffset(mq, it->second);
          }
          m_offsetTable_tmp.clear();
        } else {
          LOG_ERROR(
              "open offset store file failed, please check whether file: %s is "
              "cleared by operator, if so, delete this offsets.Json file and "
              "then restart consumer",
              m_storeFile.c_str());
          THROW_MQEXCEPTION(MQClientException,
                            "open offset store file failed, please check whether "
                            "offsets.Json is cleared by operator, if so, delete "
                            "this offsets.Json file and then restart consumer",
                            -1);
        }
      } else {
        LOG_ERROR(
            "open offset store file failed, please check whether file:%s is "
            "deleted by operator and then restart consumer",
            m_storeFile.c_str());
        THROW_MQEXCEPTION(MQClientException,
                          "open offset store file failed, please check "
                          "directory:%s is deleted by operator or offset.Json "
                          "file is cleared by operator, and then restart "
                          "consumer",
                          -1);
      }
    } else {
      LOG_WARN(
          "offsets.Json file not exist, maybe this is the first time "
          "consumation");
    }
  }
  
  void LocalFileOffsetStore::updateOffset(const MQMessageQueue& mq, int64 offset) {
    boost::lock_guard<boost::mutex> lock(m_lock);
    m_offsetTable[mq] = offset;
  }
  
  int64 LocalFileOffsetStore::readOffset(const MQMessageQueue& mq,
                                         ReadOffsetType type,
                                         const SessionCredentials& session_credentials) {
    switch (type) {
      case MEMORY_FIRST_THEN_STORE:
      case READ_FROM_MEMORY: {
        boost::lock_guard<boost::mutex> lock(m_lock);
        MQ2OFFSET::iterator it = m_offsetTable.find(mq);
        if (it != m_offsetTable.end()) {
          return it->second;
        } else if (READ_FROM_MEMORY == type) {
          return -1;
        }
      }
      case READ_FROM_STORE: {
        try {
          load();
        } catch (MQException& e) {
          LOG_ERROR("catch exception when load local file");
          return -1;
        }
        boost::lock_guard<boost::mutex> lock(m_lock);
        MQ2OFFSET::iterator it = m_offsetTable.find(mq);
        if (it != m_offsetTable.end()) {
          return it->second;
        }
      }
      default:
        break;
    }
    LOG_ERROR("can not readOffset from offsetStore.json, maybe first time consumation");
    return -1;
  }
  
  void LocalFileOffsetStore::persist(const MQMessageQueue& mq, const SessionCredentials& session_credentials) {}
  
  void LocalFileOffsetStore::persistAll(const std::vector<MQMessageQueue>& mqs) {
    boost::lock_guard<boost::mutex> lock(m_lock);
  
    map<string, int64> m_offsetTable_tmp;
    vector<MQMessageQueue>::const_iterator it = mqs.begin();
    for (; it != mqs.end(); ++it) {
      MessageQueue mq_tmp((*it).getTopic(), (*it).getBrokerName(), (*it).getQueueId());
      string mqKey = mq_tmp.toJson().toStyledString();
      m_offsetTable_tmp[mqKey] = m_offsetTable[*it];
    }
  
    std::ofstream s;
    string storefile_bak(m_storeFile);
    storefile_bak.append(".bak");
    s.open(storefile_bak.c_str(), std::ios::out);
    if (s.is_open()) {
      try {
        boost::archive::text_oarchive oa(s);
        // Boost is nervous that archiving non-const class instances which might
        // cause a problem with object tracking if different tracked objects use
        // the same address.
        oa << const_cast<const map<string, int64>&>(m_offsetTable_tmp);
      } catch (...) {
        LOG_ERROR("persist offset store file:%s failed", m_storeFile.c_str());
        s.close();
        THROW_MQEXCEPTION(MQClientException, "persistAll:open offset store file failed", -1);
      }
      s.close();
      if (!UtilAll::ReplaceFile(storefile_bak, m_storeFile))
        LOG_ERROR("could not rename bak file:%s", strerror(errno));
      m_offsetTable_tmp.clear();
    } else {
      LOG_ERROR("open offset store file:%s failed", m_storeFile.c_str());
      m_offsetTable_tmp.clear();
      THROW_MQEXCEPTION(MQClientException, "persistAll:open offset store file failed", -1);
    }
  }
  
  void LocalFileOffsetStore::removeOffset(const MQMessageQueue& mq) {}
  
  //<!***************************************************************************
  RemoteBrokerOffsetStore::RemoteBrokerOffsetStore(const string& groupName, MQClientFactory* pfactory)
      : OffsetStore(groupName, pfactory) {}
  
  RemoteBrokerOffsetStore::~RemoteBrokerOffsetStore() {}
  
  void RemoteBrokerOffsetStore::load() {}
  
  void RemoteBrokerOffsetStore::updateOffset(const MQMessageQueue& mq, int64 offset) {
    boost::lock_guard<boost::mutex> lock(m_lock);
    m_offsetTable[mq] = offset;
  }
  
  int64 RemoteBrokerOffsetStore::readOffset(const MQMessageQueue& mq,
                                            ReadOffsetType type,
                                            const SessionCredentials& session_credentials) {
    switch (type) {
      case MEMORY_FIRST_THEN_STORE:
      case READ_FROM_MEMORY: {
        boost::lock_guard<boost::mutex> lock(m_lock);
  
        MQ2OFFSET::iterator it = m_offsetTable.find(mq);
        if (it != m_offsetTable.end()) {
          return it->second;
        } else if (READ_FROM_MEMORY == type) {
          return -1;
        }
      }
      case READ_FROM_STORE: {
        try {
          int64 brokerOffset = fetchConsumeOffsetFromBroker(mq, session_credentials);
          //<!update;
          updateOffset(mq, brokerOffset);
          return brokerOffset;
        } catch (MQBrokerException& e) {
          LOG_ERROR(e.what());
          return -1;
        } catch (MQException& e) {
          LOG_ERROR(e.what());
          return -2;
        }
      }
      default:
        break;
    }
    return -1;
  }
  
  void RemoteBrokerOffsetStore::persist(const MQMessageQueue& mq, const SessionCredentials& session_credentials) {
    MQ2OFFSET offsetTable;
    {
      boost::lock_guard<boost::mutex> lock(m_lock);
      offsetTable = m_offsetTable;
    }
  
    MQ2OFFSET::iterator it = offsetTable.find(mq);
    if (it != offsetTable.end()) {
      try {
        updateConsumeOffsetToBroker(mq, it->second, session_credentials);
      } catch (MQException& e) {
        LOG_ERROR("updateConsumeOffsetToBroker error");
      }
    }
  }
  
  void RemoteBrokerOffsetStore::persistAll(const std::vector<MQMessageQueue>& mq) {}
  
  void RemoteBrokerOffsetStore::removeOffset(const MQMessageQueue& mq) {
    boost::lock_guard<boost::mutex> lock(m_lock);
    if (m_offsetTable.find(mq) != m_offsetTable.end())
      m_offsetTable.erase(mq);
  }
  
  void RemoteBrokerOffsetStore::updateConsumeOffsetToBroker(const MQMessageQueue& mq,
                                                            int64 offset,
                                                            const SessionCredentials& session_credentials) {
    unique_ptr<FindBrokerResult> pFindBrokerResult(m_pClientFactory->findBrokerAddressInAdmin(mq.getBrokerName()));
  
    if (pFindBrokerResult == NULL) {
      m_pClientFactory->updateTopicRouteInfoFromNameServer(mq.getTopic(), session_credentials);
      pFindBrokerResult.reset(m_pClientFactory->findBrokerAddressInAdmin(mq.getBrokerName()));
    }
  
    if (pFindBrokerResult != NULL) {
      UpdateConsumerOffsetRequestHeader* pRequestHeader = new UpdateConsumerOffsetRequestHeader();
      pRequestHeader->topic = mq.getTopic();
      pRequestHeader->consumerGroup = m_groupName;
      pRequestHeader->queueId = mq.getQueueId();
      pRequestHeader->commitOffset = offset;
  
      try {
        LOG_INFO("oneway updateConsumeOffsetToBroker of mq:%s, its offset is:%lld", mq.toString().c_str(), offset);
        return m_pClientFactory->getMQClientAPIImpl()->updateConsumerOffsetOneway(
            pFindBrokerResult->brokerAddr, pRequestHeader, 1000 * 5, session_credentials);
      } catch (MQException& e) {
        LOG_ERROR(e.what());
      }
    }
    LOG_WARN("The broker not exist");
  }
  
  int64 RemoteBrokerOffsetStore::fetchConsumeOffsetFromBroker(const MQMessageQueue& mq,
                                                              const SessionCredentials& session_credentials) {
    unique_ptr<FindBrokerResult> pFindBrokerResult(m_pClientFactory->findBrokerAddressInAdmin(mq.getBrokerName()));
  
    if (pFindBrokerResult == NULL) {
      m_pClientFactory->updateTopicRouteInfoFromNameServer(mq.getTopic(), session_credentials);
      pFindBrokerResult.reset(m_pClientFactory->findBrokerAddressInAdmin(mq.getBrokerName()));
    }
  
    if (pFindBrokerResult != NULL) {
      QueryConsumerOffsetRequestHeader* pRequestHeader = new QueryConsumerOffsetRequestHeader();
      pRequestHeader->topic = mq.getTopic();
      pRequestHeader->consumerGroup = m_groupName;
      pRequestHeader->queueId = mq.getQueueId();
  
      return m_pClientFactory->getMQClientAPIImpl()->queryConsumerOffset(pFindBrokerResult->brokerAddr, pRequestHeader,
                                                                         1000 * 5, session_credentials);
    } else {
      LOG_ERROR("The broker not exist when fetchConsumeOffsetFromBroker");
      THROW_MQEXCEPTION(MQClientException, "The broker not exist", -1);
    }
  }
  //<!***************************************************************************
  }  //<!end namespace;
  