/*
 * 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 "TcpRemotingClient.h"
#include <stddef.h>
#if !defined(WIN32) && !defined(__APPLE__)
#include <sys/prctl.h>
#endif
#include "Logging.h"
#include "MemoryOutputStream.h"
#include "TopAddressing.h"
#include "UtilAll.h"

namespace rocketmq {

//<!************************************************************************
TcpRemotingClient::TcpRemotingClient(int pullThreadNum, uint64_t tcpConnectTimeout, uint64_t tcpTransportTryLockTimeout)
    : m_pullThreadNum(pullThreadNum),
      m_tcpConnectTimeout(tcpConnectTimeout),
      m_tcpTransportTryLockTimeout(tcpTransportTryLockTimeout),
      m_namesrvIndex(0),
      m_ioServiceWork(m_ioService) {
#if !defined(WIN32) && !defined(__APPLE__)
  string taskName = UtilAll::getProcessName();
  prctl(PR_SET_NAME, "networkTP", 0, 0, 0);
#endif
  for (int i = 0; i != pullThreadNum; ++i) {
    m_threadpool.create_thread(boost::bind(&boost::asio::io_service::run, &m_ioService));
  }
#if !defined(WIN32) && !defined(__APPLE__)
  prctl(PR_SET_NAME, taskName.c_str(), 0, 0, 0);
#endif
  LOG_INFO(
      "m_tcpConnectTimeout:%ju, m_tcpTransportTryLockTimeout:%ju, "
      "m_pullThreadNum:%d",
      m_tcpConnectTimeout, m_tcpTransportTryLockTimeout, m_pullThreadNum);
  m_async_service_thread.reset(new boost::thread(boost::bind(&TcpRemotingClient::boost_asio_work, this)));
}

void TcpRemotingClient::boost_asio_work() {
  LOG_INFO("TcpRemotingClient::boost asio async service runing");
  boost::asio::io_service::work work(m_async_ioService);  // avoid async io
                                                          // service stops after
                                                          // first timer timeout
                                                          // callback
  m_async_ioService.run();
}

TcpRemotingClient::~TcpRemotingClient() {
  m_tcpTable.clear();
  m_futureTable.clear();
  m_asyncFutureTable.clear();
  m_namesrvAddrList.clear();
  removeAllTimerCallback();
}

void TcpRemotingClient::stopAllTcpTransportThread() {
  LOG_DEBUG("TcpRemotingClient::stopAllTcpTransportThread Begin");
  m_async_ioService.stop();
  m_async_service_thread->interrupt();
  m_async_service_thread->join();
  removeAllTimerCallback();

  {
    TcpMap::iterator it = m_tcpTable.begin();
    for (; it != m_tcpTable.end(); ++it) {
      it->second->disconnect(it->first);
    }
    m_tcpTable.clear();
  }

  m_ioService.stop();
  m_threadpool.join_all();

  {
    boost::lock_guard<boost::mutex> lock(m_futureTableMutex);
    for (ResMap::iterator it = m_futureTable.begin(); it != m_futureTable.end(); ++it) {
      if (it->second)
        it->second->releaseThreadCondition();
    }
  }
  LOG_DEBUG("TcpRemotingClient::stopAllTcpTransportThread End");
}

void TcpRemotingClient::updateNameServerAddressList(const string& addrs) {
  LOG_INFO("updateNameServerAddressList: [%s]", addrs.c_str());
  if (!addrs.empty()) {
    boost::unique_lock<boost::timed_mutex> lock(m_namesrvlock, boost::try_to_lock);
    if (!lock.owns_lock()) {
      if (!lock.timed_lock(boost::get_system_time() + boost::posix_time::seconds(10))) {
        LOG_ERROR("updateNameServerAddressList get timed_mutex timeout");
        return;
      }
    }
    // clear first;
    m_namesrvAddrList.clear();

    vector<string> out;
    UtilAll::Split(out, addrs, ";");
    for (size_t i = 0; i < out.size(); i++) {
      string addr = out[i];
      UtilAll::Trim(addr);

      string hostName;
      short portNumber;
      if (UtilAll::SplitURL(addr, hostName, portNumber)) {
        LOG_INFO("update Namesrv:%s", addr.c_str());
        m_namesrvAddrList.push_back(addr);
      } else {
        LOG_INFO("This may be invalid namer server: [%s]", addr.c_str());
      }
    }
    out.clear();
  }
}

bool TcpRemotingClient::invokeHeartBeat(const string& addr, RemotingCommand& request) {
  boost::shared_ptr<TcpTransport> pTcp = GetTransport(addr, true);
  if (pTcp != NULL) {
    int code = request.getCode();
    int opaque = request.getOpaque();
    boost::shared_ptr<ResponseFuture> responseFuture(new ResponseFuture(code, opaque, this, 3000, false, NULL));
    addResponseFuture(opaque, responseFuture);
    // LOG_INFO("invokeHeartbeat success, addr:%s, code:%d, opaque:%d,
    // timeoutms:%d", addr.c_str(), code, opaque, 3000);

    if (SendCommand(pTcp, request)) {
      responseFuture->setSendRequestOK(true);
      unique_ptr<RemotingCommand> pRsp(responseFuture->waitResponse(3000));
      if (pRsp == NULL) {
        LOG_ERROR("wait response timeout of heartbeat, so closeTransport of addr:%s", addr.c_str());
        findAndDeleteResponseFuture(opaque);
        CloseTransport(addr, pTcp);
        return false;
      } else if (pRsp->getCode() == SUCCESS_VALUE) {
        return true;
      } else {
        LOG_WARN("get error response:%d of heartbeat to addr:%s", pRsp->getCode(), addr.c_str());
        findAndDeleteResponseFuture(opaque);
        return false;
      }
    } else {
      CloseTransport(addr, pTcp);
    }
  }
  return false;
}

RemotingCommand* TcpRemotingClient::invokeSync(const string& addr,
                                               RemotingCommand& request,
                                               int timeoutMillis /* = 3000 */) {
  LOG_DEBUG("InvokeSync:", addr.c_str());
  boost::shared_ptr<TcpTransport> pTcp = GetTransport(addr, true);
  if (pTcp != NULL) {
    int code = request.getCode();
    int opaque = request.getOpaque();
    boost::shared_ptr<ResponseFuture> responseFuture(
        new ResponseFuture(code, opaque, this, timeoutMillis, false, NULL));
    addResponseFuture(opaque, responseFuture);

    if (SendCommand(pTcp, request)) {
      // LOG_INFO("invokeSync success, addr:%s, code:%d, opaque:%d,
      // timeoutms:%d", addr.c_str(), code, opaque, timeoutMillis);
      responseFuture->setSendRequestOK(true);
      RemotingCommand* pRsp = responseFuture->waitResponse(timeoutMillis);
      if (pRsp == NULL) {
        if (code != GET_CONSUMER_LIST_BY_GROUP) {
          LOG_WARN(
              "wait response timeout or get NULL response of code:%d, so "
              "closeTransport of addr:%s",
              code, addr.c_str());
          CloseTransport(addr, pTcp);
        }
        // avoid responseFuture leak;
        findAndDeleteResponseFuture(opaque);
        return NULL;
      } else {
        return pRsp;
      }
    } else {
      // avoid responseFuture leak;
      findAndDeleteResponseFuture(opaque);
      CloseTransport(addr, pTcp);
    }
  }
  LOG_DEBUG("InvokeSync [%s] Failed: Cannot Get Transport.", addr.c_str());
  return NULL;
}

bool TcpRemotingClient::invokeAsync(const string& addr,
                                    RemotingCommand& request,
                                    AsyncCallbackWrap* cbw,
                                    int64 timeoutMilliseconds,
                                    int maxRetrySendTimes,
                                    int retrySendTimes) {
  boost::shared_ptr<TcpTransport> pTcp = GetTransport(addr, true);
  if (pTcp != NULL) {
    //<!not delete, for callback to delete;
    int code = request.getCode();
    int opaque = request.getOpaque();
    boost::shared_ptr<ResponseFuture> responseFuture(
        new ResponseFuture(code, opaque, this, timeoutMilliseconds, true, cbw));
    responseFuture->setMaxRetrySendTimes(maxRetrySendTimes);
    responseFuture->setRetrySendTimes(retrySendTimes);
    responseFuture->setBrokerAddr(addr);
    responseFuture->setRequestCommand(request);
    addAsyncResponseFuture(opaque, responseFuture);
    if (cbw) {
      boost::asio::deadline_timer* t =
          new boost::asio::deadline_timer(m_async_ioService, boost::posix_time::milliseconds(timeoutMilliseconds));
      addTimerCallback(t, opaque);
      boost::system::error_code e;
      t->async_wait(boost::bind(&TcpRemotingClient::handleAsyncPullForResponseTimeout, this, e, opaque));
    }

    if (SendCommand(pTcp, request))  // Even if send failed, asyncTimerThread
                                     // will trigger next pull request or report
                                     // send msg failed
    {
      LOG_DEBUG("invokeAsync success, addr:%s, code:%d, opaque:%d", addr.c_str(), code, opaque);
      responseFuture->setSendRequestOK(true);
    }
    return true;
  }
  LOG_ERROR("invokeAsync failed of addr:%s", addr.c_str());
  return false;
}

void TcpRemotingClient::invokeOneway(const string& addr, RemotingCommand& request) {
  //<!not need callback;
  boost::shared_ptr<TcpTransport> pTcp = GetTransport(addr, true);
  if (pTcp != NULL) {
    request.markOnewayRPC();
    LOG_DEBUG("invokeOneway success, addr:%s, code:%d", addr.c_str(), request.getCode());
    SendCommand(pTcp, request);
  }
}

boost::shared_ptr<TcpTransport> TcpRemotingClient::GetTransport(const string& addr, bool needRespons) {
  if (addr.empty()) {
    LOG_DEBUG("GetTransport of NameServer");
    return CreateNameserverTransport(needRespons);
  }
  return CreateTransport(addr, needRespons);
}

boost::shared_ptr<TcpTransport> TcpRemotingClient::CreateTransport(const string& addr, bool needRespons) {
  boost::shared_ptr<TcpTransport> tts;
  {
    // try get m_tcpLock util m_tcpTransportTryLockTimeout to avoid blocking
    // long
    // time, if could not get m_tcpLock, return NULL
    bool bGetMutex = false;
    boost::unique_lock<boost::timed_mutex> lock(m_tcpLock, boost::try_to_lock);
    if (!lock.owns_lock()) {
      if (!lock.timed_lock(boost::get_system_time() + boost::posix_time::seconds(m_tcpTransportTryLockTimeout))) {
        LOG_ERROR("GetTransport of:%s get timed_mutex timeout", addr.c_str());
        boost::shared_ptr<TcpTransport> pTcp;
        return pTcp;
      } else {
        bGetMutex = true;
      }
    } else {
      bGetMutex = true;
    }
    if (bGetMutex) {
      if (m_tcpTable.find(addr) != m_tcpTable.end()) {
        boost::weak_ptr<TcpTransport> weakPtcp(m_tcpTable[addr]);
        boost::shared_ptr<TcpTransport> tcp = weakPtcp.lock();
        if (tcp) {
          tcpConnectStatus connectStatus = tcp->getTcpConnectStatus();
          if (connectStatus == e_connectWaitResponse) {
            boost::shared_ptr<TcpTransport> pTcp;
            return pTcp;
          } else if (connectStatus == e_connectFail) {
            LOG_ERROR("tcpTransport with server disconnected, erase server:%s", addr.c_str());
            tcp->disconnect(addr);  // avoid coredump when connection with broker was broken
            m_tcpTable.erase(addr);
          } else if (connectStatus == e_connectSuccess) {
            return tcp;
          } else {
            LOG_ERROR(
                "go to fault state, erase:%s from tcpMap, and reconnect "
                "it",
                addr.c_str());
            m_tcpTable.erase(addr);
          }
        }
      }

      //<!callback;
      READ_CALLBACK callback = needRespons ? &TcpRemotingClient::static_messageReceived : NULL;

      tts.reset(new TcpTransport(this, callback));
      tcpConnectStatus connectStatus = tts->connect(addr, m_tcpConnectTimeout);
      if (connectStatus != e_connectWaitResponse) {
        LOG_WARN("can not connect to :%s", addr.c_str());
        tts->disconnect(addr);
        boost::shared_ptr<TcpTransport> pTcp;
        return pTcp;
      } else {
        m_tcpTable[addr] = tts;  // even if connecting failed finally, this
                                 // server transport will be erased by next
                                 // CreateTransport
      }
    } else {
      LOG_WARN("get tcpTransport mutex failed :%s", addr.c_str());
      boost::shared_ptr<TcpTransport> pTcp;
      return pTcp;
    }
  }

  tcpConnectStatus connectStatus = tts->waitTcpConnectEvent(m_tcpConnectTimeout);
  if (connectStatus != e_connectSuccess) {
    LOG_WARN("can not connect to server:%s", addr.c_str());
    tts->disconnect(addr);
    boost::shared_ptr<TcpTransport> pTcp;
    return pTcp;
  } else {
    LOG_INFO("connect server with addr:%s success", addr.c_str());
    return tts;
  }
}

boost::shared_ptr<TcpTransport> TcpRemotingClient::CreateNameserverTransport(bool needRespons) {
  // m_namesrvLock was added to avoid operation of nameServer was blocked by
  // m_tcpLock, it was used by single Thread mostly, so no performance impact
  // try get m_tcpLock util m_tcpTransportTryLockTimeout to avoid blocking long
  // time, if could not get m_namesrvlock, return NULL
  LOG_DEBUG("--CreateNameserverTransport--");
  bool bGetMutex = false;
  boost::unique_lock<boost::timed_mutex> lock(m_namesrvlock, boost::try_to_lock);
  if (!lock.owns_lock()) {
    if (!lock.timed_lock(boost::get_system_time() + boost::posix_time::seconds(m_tcpTransportTryLockTimeout))) {
      LOG_ERROR("CreateNameserverTransport get timed_mutex timeout");
      boost::shared_ptr<TcpTransport> pTcp;
      return pTcp;
    } else {
      bGetMutex = true;
    }
  } else {
    bGetMutex = true;
  }

  if (bGetMutex) {
    if (!m_namesrvAddrChoosed.empty()) {
      boost::shared_ptr<TcpTransport> pTcp = GetTransport(m_namesrvAddrChoosed, true);
      if (pTcp)
        return pTcp;
      else
        m_namesrvAddrChoosed.clear();
    }

    vector<string>::iterator itp = m_namesrvAddrList.begin();
    for (; itp != m_namesrvAddrList.end(); ++itp) {
      unsigned int index = m_namesrvIndex % m_namesrvAddrList.size();
      if (m_namesrvIndex == numeric_limits<unsigned int>::max())
        m_namesrvIndex = 0;
      m_namesrvIndex++;
      LOG_INFO("namesrvIndex is:%d, index:%d, namesrvaddrlist size:" SIZET_FMT "", m_namesrvIndex, index,
               m_namesrvAddrList.size());
      boost::shared_ptr<TcpTransport> pTcp = GetTransport(m_namesrvAddrList[index], true);
      if (pTcp) {
        m_namesrvAddrChoosed = m_namesrvAddrList[index];
        return pTcp;
      }
    }
    boost::shared_ptr<TcpTransport> pTcp;
    return pTcp;
  } else {
    LOG_WARN("get nameServer tcpTransport mutex failed");
    boost::shared_ptr<TcpTransport> pTcp;
    return pTcp;
  }
}

void TcpRemotingClient::CloseTransport(const string& addr, boost::shared_ptr<TcpTransport> pTcp) {
  if (addr.empty()) {
    return CloseNameServerTransport(pTcp);
  }

  bool bGetMutex = false;
  boost::unique_lock<boost::timed_mutex> lock(m_tcpLock, boost::try_to_lock);
  if (!lock.owns_lock()) {
    if (!lock.timed_lock(boost::get_system_time() + boost::posix_time::seconds(m_tcpTransportTryLockTimeout))) {
      LOG_ERROR("CloseTransport of:%s get timed_mutex timeout", addr.c_str());
      return;
    } else {
      bGetMutex = true;
    }
  } else {
    bGetMutex = true;
  }
  LOG_ERROR("CloseTransport of:%s", addr.c_str());
  if (bGetMutex) {
    bool removeItemFromTable = true;
    if (m_tcpTable.find(addr) != m_tcpTable.end()) {
      if (m_tcpTable[addr]->getStartTime() != pTcp->getStartTime()) {
        LOG_INFO(
            "tcpTransport with addr:%s has been closed before, and has been "
            "created again, nothing to do",
            addr.c_str());
        removeItemFromTable = false;
      }
    } else {
      LOG_INFO("tcpTransport with addr:%s had been removed from tcpTable before", addr.c_str());
      removeItemFromTable = false;
    }

    if (removeItemFromTable == true) {
      LOG_WARN("closeTransport: disconnect broker:%s with state:%d", addr.c_str(),
               m_tcpTable[addr]->getTcpConnectStatus());
      if (m_tcpTable[addr]->getTcpConnectStatus() == e_connectSuccess)
        m_tcpTable[addr]->disconnect(addr);  // avoid coredump when connection with server was broken
      LOG_WARN("closeTransport: erase broker: %s", addr.c_str());
      m_tcpTable.erase(addr);
    }
  } else {
    LOG_WARN("CloseTransport::get tcpTransport mutex failed:%s", addr.c_str());
    return;
  }
  LOG_ERROR("CloseTransport of:%s end", addr.c_str());
}

void TcpRemotingClient::CloseNameServerTransport(boost::shared_ptr<TcpTransport> pTcp) {
  bool bGetMutex = false;
  boost::unique_lock<boost::timed_mutex> lock(m_namesrvlock, boost::try_to_lock);
  if (!lock.owns_lock()) {
    if (!lock.timed_lock(boost::get_system_time() + boost::posix_time::seconds(m_tcpTransportTryLockTimeout))) {
      LOG_ERROR("CreateNameserverTransport get timed_mutex timeout");
      return;
    } else {
      bGetMutex = true;
    }
  } else {
    bGetMutex = true;
  }
  if (bGetMutex) {
    string addr = m_namesrvAddrChoosed;
    bool removeItemFromTable = true;
    if (m_tcpTable.find(addr) != m_tcpTable.end()) {
      if (m_tcpTable[addr]->getStartTime() != pTcp->getStartTime()) {
        LOG_INFO(
            "tcpTransport with addr:%s has been closed before, and has been "
            "created again, nothing to do",
            addr.c_str());
        removeItemFromTable = false;
      }
    } else {
      LOG_INFO("tcpTransport with addr:%s had been removed from tcpTable before", addr.c_str());
      removeItemFromTable = false;
    }

    if (removeItemFromTable == true) {
      m_tcpTable[addr]->disconnect(addr);  // avoid coredump when connection with server was broken
      LOG_WARN("closeTransport: erase broker: %s", addr.c_str());
      m_tcpTable.erase(addr);
      m_namesrvAddrChoosed.clear();
    }
  } else {
    LOG_WARN("CloseNameServerTransport::get tcpTransport mutex failed:%s", m_namesrvAddrChoosed.c_str());
    return;
  }
}

bool TcpRemotingClient::SendCommand(boost::shared_ptr<TcpTransport> pTts, RemotingCommand& msg) {
  const MemoryBlock* phead = msg.GetHead();
  const MemoryBlock* pbody = msg.GetBody();

  unique_ptr<MemoryOutputStream> result(new MemoryOutputStream(1024));
  if (phead->getData()) {
    result->write(phead->getData(), phead->getSize());
  }
  if (pbody->getData()) {
    result->write(pbody->getData(), pbody->getSize());
  }
  const char* pData = static_cast<const char*>(result->getData());
  int len = result->getDataSize();
  return pTts->sendMessage(pData, len);
}

void TcpRemotingClient::static_messageReceived(void* context, const MemoryBlock& mem, const string& addr) {
  TcpRemotingClient* pTcpRemotingClient = (TcpRemotingClient*)context;
  if (pTcpRemotingClient)
    pTcpRemotingClient->messageReceived(mem, addr);
}

void TcpRemotingClient::messageReceived(const MemoryBlock& mem, const string& addr) {
  m_ioService.post(boost::bind(&TcpRemotingClient::ProcessData, this, mem, addr));
}

void TcpRemotingClient::ProcessData(const MemoryBlock& mem, const string& addr) {
  RemotingCommand* pRespondCmd = NULL;
  try {
    pRespondCmd = RemotingCommand::Decode(mem);
  } catch (...) {
    LOG_ERROR("processData_error");
    return;
  }

  int opaque = pRespondCmd->getOpaque();

  //<!process self;
  if (pRespondCmd->isResponseType()) {
    boost::shared_ptr<ResponseFuture> pFuture(findAndDeleteAsyncResponseFuture(opaque));
    if (!pFuture) {
      pFuture = findAndDeleteResponseFuture(opaque);
      if (pFuture) {
        if (pFuture->getSyncResponseFlag()) {
          LOG_WARN("waitResponse already timeout of opaque:%d", opaque);
          deleteAndZero(pRespondCmd);
          return;
        }
        LOG_DEBUG("find_response opaque:%d", opaque);
      } else {
        LOG_DEBUG("responseFuture was deleted by timeout of opaque:%d", opaque);
        deleteAndZero(pRespondCmd);
        return;
      }
    }
    processResponseCommand(pRespondCmd, pFuture);
  } else {
    processRequestCommand(pRespondCmd, addr);
  }
}

void TcpRemotingClient::processResponseCommand(RemotingCommand* pCmd, boost::shared_ptr<ResponseFuture> pfuture) {
  int code = pfuture->getRequestCode();
  int opaque = pCmd->getOpaque();
  LOG_DEBUG("processResponseCommand, code:%d,opaque:%d, maxRetryTimes:%d, retrySendTimes:%d", code, opaque,
            pfuture->getMaxRetrySendTimes(), pfuture->getRetrySendTimes());
  pCmd->SetExtHeader(code);  // set head , for response use

  pfuture->setResponse(pCmd);

  if (pfuture->getASyncFlag()) {
    if (!pfuture->getAsyncResponseFlag()) {
      pfuture->setAsyncResponseFlag();
      pfuture->setAsyncCallBackStatus(asyncCallBackStatus_response);
      cancelTimerCallback(opaque);
      pfuture->executeInvokeCallback();
    }
  }
}

void TcpRemotingClient::processRequestCommand(RemotingCommand* pCmd, const string& addr) {
  unique_ptr<RemotingCommand> pRequestCommand(pCmd);
  int requestCode = pRequestCommand->getCode();
  if (m_requestTable.find(requestCode) == m_requestTable.end()) {
    LOG_ERROR("can_not_find request:%d processor", requestCode);
  } else {
    unique_ptr<RemotingCommand> pResponse(m_requestTable[requestCode]->processRequest(addr, pRequestCommand.get()));
    if (!pRequestCommand->isOnewayRPC()) {
      if (pResponse) {
        pResponse->setOpaque(pRequestCommand->getOpaque());
        pResponse->markResponseType();
        pResponse->Encode();

        invokeOneway(addr, *pResponse);
      }
    }
  }
}

void TcpRemotingClient::addResponseFuture(int opaque, boost::shared_ptr<ResponseFuture> pfuture) {
  boost::lock_guard<boost::mutex> lock(m_futureTableMutex);
  m_futureTable[opaque] = pfuture;
}

// Note: after call this function, shared_ptr of m_syncFutureTable[opaque] will
// be erased, so caller must ensure the life cycle of returned shared_ptr;
boost::shared_ptr<ResponseFuture> TcpRemotingClient::findAndDeleteResponseFuture(int opaque) {
  boost::lock_guard<boost::mutex> lock(m_futureTableMutex);
  boost::shared_ptr<ResponseFuture> pResponseFuture;
  if (m_futureTable.find(opaque) != m_futureTable.end()) {
    pResponseFuture = m_futureTable[opaque];
    m_futureTable.erase(opaque);
  }
  return pResponseFuture;
}

void TcpRemotingClient::handleAsyncPullForResponseTimeout(const boost::system::error_code& e, int opaque) {
  if (e == boost::asio::error::operation_aborted) {
    LOG_INFO("handleAsyncPullForResponseTimeout aborted opaque:%d, e_code:%d, msg:%s", opaque, e.value(),
             e.message().data());
    return;
  }

  LOG_DEBUG("handleAsyncPullForResponseTimeout opaque:%d, e_code:%d, msg:%s", opaque, e.value(), e.message().data());
  boost::shared_ptr<ResponseFuture> pFuture(findAndDeleteAsyncResponseFuture(opaque));
  if (pFuture && pFuture->getASyncFlag() && (pFuture->getAsyncCallbackWrap())) {
    if ((pFuture->getAsyncResponseFlag() != true))  // if no response received, then check timeout or not
    {
      LOG_ERROR("no response got for opaque:%d", opaque);
      pFuture->setAsyncCallBackStatus(asyncCallBackStatus_timeout);
      pFuture->executeInvokeCallbackException();
    }
  }

  eraseTimerCallback(opaque);
}

void TcpRemotingClient::addAsyncResponseFuture(int opaque, boost::shared_ptr<ResponseFuture> pfuture) {
  boost::lock_guard<boost::mutex> lock(m_asyncFutureLock);
  m_asyncFutureTable[opaque] = pfuture;
}

// Note: after call this function, shared_ptr of m_asyncFutureTable[opaque] will
// be erased, so caller must ensure the life cycle of returned shared_ptr;
boost::shared_ptr<ResponseFuture> TcpRemotingClient::findAndDeleteAsyncResponseFuture(int opaque) {
  boost::lock_guard<boost::mutex> lock(m_asyncFutureLock);
  boost::shared_ptr<ResponseFuture> pResponseFuture;
  if (m_asyncFutureTable.find(opaque) != m_asyncFutureTable.end()) {
    pResponseFuture = m_asyncFutureTable[opaque];
    m_asyncFutureTable.erase(opaque);
  }

  return pResponseFuture;
}

void TcpRemotingClient::registerProcessor(MQRequestCode requestCode, ClientRemotingProcessor* clientRemotingProcessor) {
  if (m_requestTable.find(requestCode) != m_requestTable.end())
    m_requestTable.erase(requestCode);
  m_requestTable[requestCode] = clientRemotingProcessor;
}

void TcpRemotingClient::addTimerCallback(boost::asio::deadline_timer* t, int opaque) {
  boost::lock_guard<boost::mutex> lock(m_timerMapMutex);
  if (m_async_timer_map.find(opaque) != m_async_timer_map.end()) {
    LOG_DEBUG("addTimerCallback:erase timerCallback opaque:%lld", opaque);
    boost::asio::deadline_timer* old_t = m_async_timer_map[opaque];
    old_t->cancel();
    delete old_t;
    old_t = NULL;
    m_async_timer_map.erase(opaque);
  }
  m_async_timer_map[opaque] = t;
}

void TcpRemotingClient::eraseTimerCallback(int opaque) {
  boost::lock_guard<boost::mutex> lock(m_timerMapMutex);
  if (m_async_timer_map.find(opaque) != m_async_timer_map.end()) {
    LOG_DEBUG("eraseTimerCallback: opaque:%lld", opaque);
    boost::asio::deadline_timer* t = m_async_timer_map[opaque];
    delete t;
    t = NULL;
    m_async_timer_map.erase(opaque);
  }
}

void TcpRemotingClient::cancelTimerCallback(int opaque) {
  boost::lock_guard<boost::mutex> lock(m_timerMapMutex);
  if (m_async_timer_map.find(opaque) != m_async_timer_map.end()) {
    LOG_DEBUG("cancelTimerCallback: opaque:%lld", opaque);
    boost::asio::deadline_timer* t = m_async_timer_map[opaque];
    t->cancel();
    delete t;
    t = NULL;
    m_async_timer_map.erase(opaque);
  }
}

void TcpRemotingClient::removeAllTimerCallback() {
  boost::lock_guard<boost::mutex> lock(m_timerMapMutex);
  for (asyncTimerMap::iterator it = m_async_timer_map.begin(); it != m_async_timer_map.end(); ++it) {
    boost::asio::deadline_timer* t = it->second;
    t->cancel();
    delete t;
    t = NULL;
  }
  m_async_timer_map.clear();
}

void TcpRemotingClient::deleteOpaqueForDropPullRequest(const MQMessageQueue& mq, int opaque) {
  // delete the map record of opaque<->ResponseFuture, so the answer for the pull request will discard when receive it
  // later
  boost::shared_ptr<ResponseFuture> pFuture(findAndDeleteAsyncResponseFuture(opaque));
  if (!pFuture) {
    pFuture = findAndDeleteResponseFuture(opaque);
    if (pFuture) {
      LOG_DEBUG("succ deleted the sync pullrequest for opaque:%d, mq:%s", opaque, mq.toString().data());
    }
  } else {
    LOG_DEBUG("succ deleted the async pullrequest for opaque:%d, mq:%s", opaque, mq.toString().data());
  }
  // delete the timeout timer for opaque for pullrequest
  cancelTimerCallback(opaque);
}

//<!************************************************************************
}  // namespace rocketmq