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 "ResponseFuture.h"
  #include "Logging.h"
  #include "TcpRemotingClient.h"
  
  namespace rocketmq {
  //<!************************************************************************
  ResponseFuture::ResponseFuture(int requestCode, int opaque,
                                 TcpRemotingClient* powner, int64 timeout,
                                 bool bAsync /* = false */,
                                 AsyncCallbackWrap* pcall /* = NULL */) {
    m_bAsync.store(bAsync);
    m_requestCode = requestCode;
    m_opaque = opaque;
    m_timeout = timeout;
    m_pCallbackWrap = pcall;
    m_pResponseCommand = NULL;
    m_sendRequestOK = false;
    m_maxRetrySendTimes = 1;
    m_retrySendTimes = 1;
    m_brokerAddr = "";
    m_beginTimestamp = UtilAll::currentTimeMillis();
    m_asyncCallbackStatus = asyncCallBackStatus_init;
    if (getASyncFlag()) {
      m_asyncResponse.store(false);
      m_syncResponse.store(true);
    } else {
      m_asyncResponse.store(true);
      m_syncResponse.store(false);
    }
  }
  
  ResponseFuture::~ResponseFuture() {
    deleteAndZero(m_pCallbackWrap);
    /*
      do not set m_pResponseCommand to NULL when destruct, as m_pResponseCommand
      is used by MQClientAPIImpl concurrently, and will be released by producer or
      consumer;
      m_pResponseCommand = NULL;
    */
  }
  
  void ResponseFuture::releaseThreadCondition() { m_defaultEvent.notify_all(); }
  
  RemotingCommand* ResponseFuture::waitResponse(int timeoutMillis) {
    boost::unique_lock<boost::mutex> lk(m_defaultEventLock);
    if (!m_defaultEvent.timed_wait(
            lk, boost::posix_time::milliseconds(timeoutMillis))) {
      LOG_WARN("waitResponse of code:%d with opaque:%d timeout", m_requestCode,
               m_opaque);
      m_syncResponse.store(true);
    }
    return m_pResponseCommand;
  }
  
  void ResponseFuture::setResponse(RemotingCommand* pResponseCommand) {
    // LOG_DEBUG("setResponse of opaque:%d",m_opaque);
    m_pResponseCommand = pResponseCommand;
  
    if (!getASyncFlag()) {
      if (m_syncResponse.load() == false) {
        m_defaultEvent.notify_all();
        m_syncResponse.store(true);
      }
    }
  }
  
  const bool ResponseFuture::getSyncResponseFlag() {
    if (m_syncResponse.load() == true) {
      return true;
    }
    return false;
  }
  
  const bool ResponseFuture::getAsyncResponseFlag() {
    if (m_asyncResponse.load() == true) {
      // LOG_DEBUG("ASYNC flag is TRUE,opaque is:%d",getOpaque() );
      return true;
    }
  
    return false;
  }
  
  void ResponseFuture::setAsyncResponseFlag() { m_asyncResponse.store(true); }
  
  const bool ResponseFuture::getASyncFlag() {
    if (m_bAsync.load() == true) {
      // LOG_DEBUG("ASYNC flag is TRUE,opaque is:%d",getOpaque() );
      return true;
    }
    return false;
  }
  
  bool ResponseFuture::isSendRequestOK() { return m_sendRequestOK; }
  
  void ResponseFuture::setSendRequestOK(bool sendRequestOK) {
    m_sendRequestOK = sendRequestOK;
  }
  
  int ResponseFuture::getOpaque() const { return m_opaque; }
  
  int ResponseFuture::getRequestCode() const { return m_requestCode; }
  
  void ResponseFuture::setAsyncCallBackStatus(
      asyncCallBackStatus asyncCallbackStatus) {
    boost::lock_guard<boost::mutex> lock(m_asyncCallbackLock);
    if (m_asyncCallbackStatus == asyncCallBackStatus_init) {
      m_asyncCallbackStatus = asyncCallbackStatus;
    }
  }
  
  void ResponseFuture::executeInvokeCallback() {
    if (m_pCallbackWrap == NULL) {
      deleteAndZero(m_pResponseCommand);
      return;
    } else {
      if (m_asyncCallbackStatus == asyncCallBackStatus_response) {
        m_pCallbackWrap->operationComplete(this, true);
      } else {
        if (m_pResponseCommand)
          deleteAndZero(m_pResponseCommand);  // the responseCommand from
                                              // RemotingCommand::Decode(mem) will
                                              // only deleted by operationComplete
                                              // automatically
        LOG_WARN(
            "timeout and response incoming concurrently of opaque:%d, and "
            "executeInvokeCallbackException was called earlier",
            m_opaque);
      }
    }
  }
  
  void ResponseFuture::executeInvokeCallbackException() {
    if (m_pCallbackWrap == NULL) {
      LOG_ERROR("m_pCallbackWrap is NULL, critical error");
      return;
    } else {
      if (m_asyncCallbackStatus == asyncCallBackStatus_timeout) {
  
  	//here no need retrySendTimes process because of it have timeout
  	LOG_ERROR("send msg, callback timeout, opaque:%d, sendTimes:%d, maxRetryTimes:%d", getOpaque(), getRetrySendTimes(), getMaxRetrySendTimes());
  
        m_pCallbackWrap->onException();
      } else {
        LOG_WARN(
            "timeout and response incoming concurrently of opaque:%d, and "
            "executeInvokeCallback was called earlier",
            m_opaque);
      }
    }
  }
  
  bool ResponseFuture::isTimeOut() const {
    int64 diff = UtilAll::currentTimeMillis() - m_beginTimestamp;
    //<!only async;
    return m_bAsync.load() == 1 && diff > m_timeout;
  }
  
  int ResponseFuture::getMaxRetrySendTimes() const {
  	return m_maxRetrySendTimes;
  } 
  int ResponseFuture::getRetrySendTimes() const {
  	return m_retrySendTimes;
  }
  
  void ResponseFuture::setMaxRetrySendTimes(int maxRetryTimes) {
  	m_maxRetrySendTimes = maxRetryTimes;
  }
  void ResponseFuture::setRetrySendTimes(int retryTimes) {
  	m_retrySendTimes = retryTimes;
  }
  
  void ResponseFuture::setBrokerAddr(const std::string& brokerAddr) {
  	m_brokerAddr = brokerAddr;
  }
  void ResponseFuture::setRequestCommand(const RemotingCommand& requestCommand) {
  	m_requestCommand = requestCommand;
  }
  
  const RemotingCommand& ResponseFuture::getRequestCommand() {
  	return m_requestCommand;
  }
  std::string ResponseFuture::getBrokerAddr() const {
  	return m_brokerAddr;
  }
  
  int64 ResponseFuture::leftTime() const {
  	int64 diff = UtilAll::currentTimeMillis() - m_beginTimestamp;
  	return m_timeout - diff;
  }
  
  RemotingCommand* ResponseFuture::getCommand() const {
    return m_pResponseCommand;
  }
  
  AsyncCallbackWrap* ResponseFuture::getAsyncCallbackWrap() {
    return m_pCallbackWrap;
  }
  
  //<!************************************************************************
  }  //<!end namespace;
  