/*

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

#include "ConsumerRunningInfo.h"

#include "Logging.h"

#include "MQClientManager.h"

#include "MQVersion.h"

#include "PullRequest.h"

#include "Rebalance.h"

#include "TopicPublishInfo.h"

#include "TransactionMQProducerImpl.h"

#define MAX_BUFF_SIZE 8192

#define SAFE_BUFF_SIZE 7936 // 8192 - 256 = 7936

#define PROCESS_NAME_BUF_SIZE 256

namespace rocketmq {

//<!***************************************************************************

MQClientFactory::MQClientFactory(const string& clientID, bool enableSsl, const std::string& sslPropertyFile)

: m_bFetchNSService(true) {

m_clientId = clientID;

}

MQClientFactory::MQClientFactory(const string& clientID,

int pullThreadNum,

uint64_t tcpConnectTimeout,

uint64_t tcpTransportTryLockTimeout,

string unitName,

bool enableSsl,

const std::string& sslPropertyFile)

: m_bFetchNSService(true) {

m_clientId = clientID;

// default Topic register;

boost::shared_ptr<TopicPublishInfo> pDefaultTopicInfo(new TopicPublishInfo());

m_topicPublishInfoTable[DEFAULT_TOPIC] = pDefaultTopicInfo;

m_pClientRemotingProcessor.reset(new ClientRemotingProcessor(this));

m_pClientAPIImpl.reset(new MQClientAPIImpl(m_clientId, m_pClientRemotingProcessor.get(), pullThreadNum,

tcpConnectTimeout, tcpTransportTryLockTimeout, unitName, enableSsl,

sslPropertyFile));

m_serviceState = CREATE_JUST;

LOG_DEBUG("MQClientFactory construct");

}

MQClientFactory::~MQClientFactory() {

LOG_INFO("MQClientFactory:%s destruct", m_clientId.c_str());

for (TRDMAP::iterator itp = m_topicRouteTable.begin(); itp != m_topicRouteTable.end(); ++itp) {

delete itp->second;

}

m_producerTable.clear();

m_consumerTable.clear();

m_topicRouteTable.clear();

m_brokerAddrTable.clear();

m_topicPublishInfoTable.clear();

}

void MQClientFactory::start() {

switch (m_serviceState) {

case CREATE_JUST:

LOG_INFO("MQClientFactory:%s start", m_clientId.c_str());

m_serviceState = START_FAILED;

//<!start time task;

m_async_service_thread.reset(

new boost::thread(boost::bind(&MQClientFactory::startScheduledTask, this, m_bFetchNSService)));

m_serviceState = RUNNING;

break;

case RUNNING:

LOG_INFO("The Factory object:%s start before with now state:%d", m_clientId.c_str(), m_serviceState);

break;

case SHUTDOWN_ALREADY:

case START_FAILED:

LOG_INFO("The Factory object:%s start failed with fault state:%d", m_clientId.c_str(), m_serviceState);

break;

default:

break;

}

}

void MQClientFactory::updateTopicRouteInfo(boost::system::error_code& ec,

boost::shared_ptr<boost::asio::deadline_timer> t) {

if ((getConsumerTableSize() == 0) && (getProducerTableSize() == 0)) {

return;

}

set<string> topicList;

//<!Consumer;

getTopicListFromConsumerSubscription(topicList);

//<!Producer;

getTopicListFromTopicPublishInfo(topicList);

//<! update;

{

SessionCredentials session_credentials;

getSessionCredentialsFromOneOfProducerOrConsumer(session_credentials);

set<string>::iterator it = topicList.begin();

for (; it != topicList.end(); ++it) {

updateTopicRouteInfoFromNameServer(*it, session_credentials);

}

}

boost::system::error_code e;

t->expires_from_now(t->expires_from_now() + boost::posix_time::seconds(30), e);

t->async_wait(boost::bind(&MQClientFactory::updateTopicRouteInfo, this, ec, t));

}

TopicRouteData* MQClientFactory::getTopicRouteData(const string& topic) {

boost::lock_guard<boost::mutex> lock(m_topicRouteTableMutex);

if (m_topicRouteTable.find(topic) != m_topicRouteTable.end()) {

return m_topicRouteTable[topic];

}

return NULL;

}

void MQClientFactory::addTopicRouteData(const string& topic, TopicRouteData* pTopicRouteData) {

boost::lock_guard<boost::mutex> lock(m_topicRouteTableMutex);

if (m_topicRouteTable.find(topic) != m_topicRouteTable.end()) {

delete m_topicRouteTable[topic];

m_topicRouteTable.erase(topic);

}

m_topicRouteTable[topic] = pTopicRouteData;

}

boost::shared_ptr<TopicPublishInfo> MQClientFactory::tryToFindTopicPublishInfo(

const string& topic,

const SessionCredentials& session_credentials) {

boost::lock_guard<boost::mutex> lock(m_topicPublishInfoLock); // add topicPublishInfoLock to avoid con-current

// excuting updateTopicRouteInfoFromNameServer

// when producer send msg before topicRouteInfo

// was got;

if (!isTopicInfoValidInTable(topic)) {

updateTopicRouteInfoFromNameServer(topic, session_credentials);

}

//<!if not exist ,update default topic;

if (!isTopicInfoValidInTable(topic)) {

LOG_INFO("updateTopicRouteInfoFromNameServer with default");

updateTopicRouteInfoFromNameServer(topic, session_credentials, true);

}

if (!isTopicInfoValidInTable(topic)) {

LOG_WARN("tryToFindTopicPublishInfo null:%s", topic.c_str());

boost::shared_ptr<TopicPublishInfo> pTopicPublishInfo;

return pTopicPublishInfo;

}

return getTopicPublishInfoFromTable(topic);

}

bool MQClientFactory::updateTopicRouteInfoFromNameServer(const string& topic,

const SessionCredentials& session_credentials,

bool isDefault /* = false */) {

boost::lock_guard<boost::mutex> lock(m_factoryLock);

unique_ptr<TopicRouteData> pTopicRouteData;

LOG_DEBUG("updateTopicRouteInfoFromNameServer start. Topic:%s", topic.c_str());

if (isDefault) {

pTopicRouteData.reset(

m_pClientAPIImpl->getTopicRouteInfoFromNameServer(DEFAULT_TOPIC, 1000 * 5, session_credentials));

if (pTopicRouteData != NULL) {

vector<QueueData>& queueDatas = pTopicRouteData->getQueueDatas();

vector<QueueData>::iterator it = queueDatas.begin();

for (; it != queueDatas.end(); ++it) {

int queueNums = std::min(4, it->readQueueNums);

it->readQueueNums = queueNums;

it->writeQueueNums = queueNums;

}

}

LOG_DEBUG("getTopicRouteInfoFromNameServer is null for topic :%s", topic.c_str());

} else {

pTopicRouteData.reset(m_pClientAPIImpl->getTopicRouteInfoFromNameServer(topic, 1000 * 5, session_credentials));

}

if (pTopicRouteData != NULL) {

LOG_DEBUG("updateTopicRouteInfoFromNameServer has data");

TopicRouteData* pTemp = getTopicRouteData(topic);

bool changed = true;

if (pTemp != NULL) {

changed = !(*pTemp == *pTopicRouteData);

}

if (getConsumerTableSize() > 0) {

vector<MQMessageQueue> mqs;

topicRouteData2TopicSubscribeInfo(topic, pTopicRouteData.get(), mqs);

updateConsumerSubscribeTopicInfo(topic, mqs);

}

if (changed) {

//<!update Broker addr

LOG_INFO("updateTopicRouteInfoFromNameServer changed:%s", topic.c_str());

vector<BrokerData> brokerList = pTopicRouteData->getBrokerDatas();

vector<BrokerData>::iterator it = brokerList.begin();

for (; it != brokerList.end(); ++it) {

LOG_INFO("updateTopicRouteInfoFromNameServer changed with broker name:%s", (*it).brokerName.c_str());

addBrokerToAddrMap((*it).brokerName, (*it).brokerAddrs);

}

//<! update publish info;

{

boost::shared_ptr<TopicPublishInfo> publishInfo(topicRouteData2TopicPublishInfo(topic, pTopicRouteData.get()));

addTopicInfoToTable(topic, publishInfo); // erase first, then add

}

//<! update subscribe info

addTopicRouteData(topic, pTopicRouteData.release());

}

LOG_DEBUG("updateTopicRouteInfoFromNameServer end:%s", topic.c_str());

return true;

}

LOG_DEBUG("updateTopicRouteInfoFromNameServer end:%s", topic.c_str());

return false;

}

boost::shared_ptr<TopicPublishInfo> MQClientFactory::topicRouteData2TopicPublishInfo(const string& topic,

TopicRouteData* pRoute) {

boost::shared_ptr<TopicPublishInfo> info(new TopicPublishInfo());

string OrderTopicConf = pRoute->getOrderTopicConf();

//<! order msg

if (!OrderTopicConf.empty()) {

// "broker-a:8";"broker-b:8"

vector<string> brokers;

UtilAll::Split(brokers, OrderTopicConf, ';');

for (size_t i = 0; i < brokers.size(); i++) {

vector<string> item;

UtilAll::Split(item, brokers[i], ':');

size_t nums = (item.size() > 1) ? atoi(item[1].c_str()) : 0;

for (size_t i = 0; i < nums; i++) {

MQMessageQueue mq(topic, item[0], i);

info->updateMessageQueueList(mq);

}

}

}

//<!no order msg

else {

vector<QueueData>& queueDatas = pRoute->getQueueDatas();

vector<QueueData>::iterator it = queueDatas.begin();

for (; it != queueDatas.end(); ++it) {

QueueData& qd = (*it);

if (PermName::isWriteable(qd.perm)) {

string addr = findBrokerAddressInPublish(qd.brokerName);

if (addr.empty()) {

continue;

}

for (int i = 0; i < qd.writeQueueNums; i++) {

MQMessageQueue mq(topic, qd.brokerName, i);

info->updateMessageQueueList(mq);

}

}

}

}

return info;

}

void MQClientFactory::topicRouteData2TopicSubscribeInfo(const string& topic,

TopicRouteData* pRoute,

vector<MQMessageQueue>& mqs) {

mqs.clear();

vector<QueueData>& queueDatas = pRoute->getQueueDatas();

vector<QueueData>::iterator it = queueDatas.begin();

for (; it != queueDatas.end(); ++it) {

QueueData& qd = (*it);

if (PermName::isReadable(qd.perm)) {

for (int i = 0; i < qd.readQueueNums; i++) {

MQMessageQueue mq(topic, qd.brokerName, i);

mqs.push_back(mq);

}

}

}

}

void MQClientFactory::shutdown() {

if (getConsumerTableSize() != 0)

return;

if (getProducerTableSize() != 0)

return;

switch (m_serviceState) {

case CREATE_JUST:

case RUNNING: {

if (m_consumer_async_service_thread) {

m_consumer_async_ioService.stop();

m_consumer_async_service_thread->interrupt();

m_consumer_async_service_thread->join();

m_consumer_async_service_thread.reset();

}

if (m_async_service_thread) {

m_async_ioService.stop();

m_async_service_thread->interrupt();

m_async_service_thread->join();

m_async_service_thread.reset();

}

if (m_pClientAPIImpl) {

m_pClientAPIImpl->stopAllTcpTransportThread(); // Note: stop all

// TcpTransport Threads

// and release all

// responseFuture

// conditions

m_pClientAPIImpl.reset();

}

m_serviceState = SHUTDOWN_ALREADY;

LOG_INFO("MQClientFactory:%s shutdown", m_clientId.c_str());

break;

}

case SHUTDOWN_ALREADY:

break;

default:

break;

}

MQClientManager::getInstance()->removeClientFactory(m_clientId);

}

bool MQClientFactory::registerProducer(MQProducer* pProducer) {

string groupName = pProducer->getGroupName();

string namesrvaddr = pProducer->getNamesrvAddr();

if (groupName.empty()) {

return false;

}

if (!addProducerToTable(groupName, pProducer)) {

return false;

}

LOG_DEBUG("registerProducer success:%s", groupName.c_str());

//<!set nameserver;

if (namesrvaddr.empty()) {

string nameSrvDomain(pProducer->getNamesrvDomain());

if (!nameSrvDomain.empty())

m_nameSrvDomain = nameSrvDomain;

pProducer->setNamesrvAddr(m_pClientAPIImpl->fetchNameServerAddr(m_nameSrvDomain));

} else {

m_bFetchNSService = false;

m_pClientAPIImpl->updateNameServerAddr(namesrvaddr);

LOG_INFO("user specfied name server address: %s", namesrvaddr.c_str());

}

return true;

}

void MQClientFactory::unregisterProducer(MQProducer* pProducer) {

string groupName = pProducer->getGroupName();

unregisterClient(groupName, "", pProducer->getSessionCredentials());

eraseProducerFromTable(groupName);

}

bool MQClientFactory::registerConsumer(MQConsumer* pConsumer) {

string groupName = pConsumer->getGroupName();

string namesrvaddr = pConsumer->getNamesrvAddr();

if (groupName.empty()) {

return false;

}

if (!addConsumerToTable(groupName, pConsumer)) {

return false;

}

LOG_DEBUG("registerConsumer success:%s", groupName.c_str());

//<!set nameserver;

if (namesrvaddr.empty()) {

string nameSrvDomain(pConsumer->getNamesrvDomain());

if (!nameSrvDomain.empty())

m_nameSrvDomain = nameSrvDomain;

pConsumer->setNamesrvAddr(m_pClientAPIImpl->fetchNameServerAddr(m_nameSrvDomain));

} else {

m_bFetchNSService = false;

m_pClientAPIImpl->updateNameServerAddr(namesrvaddr);

LOG_INFO("user specfied name server address: %s", namesrvaddr.c_str());

}

return true;

}

void MQClientFactory::unregisterConsumer(MQConsumer* pConsumer) {

string groupName = pConsumer->getGroupName();

unregisterClient("", groupName, pConsumer->getSessionCredentials());

eraseConsumerFromTable(groupName);

}

MQProducer* MQClientFactory::selectProducer(const string& producerName) {

boost::lock_guard<boost::mutex> lock(m_producerTableMutex);

if (m_producerTable.find(producerName) != m_producerTable.end()) {

return m_producerTable[producerName];

}

return NULL;

}

bool MQClientFactory::getSessionCredentialFromProducerTable(SessionCredentials& sessionCredentials) {

boost::lock_guard<boost::mutex> lock(m_producerTableMutex);

for (MQPMAP::iterator it = m_producerTable.begin(); it != m_producerTable.end(); ++it) {

if (it->second)

sessionCredentials = it->second->getSessionCredentials();

}

if (sessionCredentials.isValid())

return true;

return false;

}

bool MQClientFactory::addProducerToTable(const string& producerName, MQProducer* pMQProducer) {

boost::lock_guard<boost::mutex> lock(m_producerTableMutex);

if (m_producerTable.find(producerName) != m_producerTable.end())

return false;

m_producerTable[producerName] = pMQProducer;

return true;

}

void MQClientFactory::eraseProducerFromTable(const string& producerName) {

boost::lock_guard<boost::mutex> lock(m_producerTableMutex);

if (m_producerTable.find(producerName) != m_producerTable.end())

m_producerTable.erase(producerName);

}

int MQClientFactory::getProducerTableSize() {

boost::lock_guard<boost::mutex> lock(m_producerTableMutex);

return m_producerTable.size();

}

void MQClientFactory::insertProducerInfoToHeartBeatData(HeartbeatData* pHeartbeatData) {

boost::lock_guard<boost::mutex> lock(m_producerTableMutex);

for (MQPMAP::iterator it = m_producerTable.begin(); it != m_producerTable.end(); ++it) {

ProducerData producerData;

producerData.groupName = it->first;

pHeartbeatData->insertDataToProducerDataSet(producerData);

}

}

MQConsumer* MQClientFactory::selectConsumer(const string& group) {

boost::lock_guard<boost::recursive_mutex> lock(m_consumerTableMutex);

if (m_consumerTable.find(group) != m_consumerTable.end()) {

return m_consumerTable[group];

}

return NULL;

}

bool MQClientFactory::getSessionCredentialFromConsumerTable(SessionCredentials& sessionCredentials) {

boost::lock_guard<boost::recursive_mutex> lock(m_consumerTableMutex);

for (MQCMAP::iterator it = m_consumerTable.begin(); it != m_consumerTable.end(); ++it) {

if (it->second)

sessionCredentials = it->second->getSessionCredentials();

}

if (sessionCredentials.isValid())

return true;

return false;

}

bool MQClientFactory::getSessionCredentialFromConsumer(const string& consumerGroup,

SessionCredentials& sessionCredentials) {

boost::lock_guard<boost::recursive_mutex> lock(m_consumerTableMutex);

if (m_consumerTable.find(consumerGroup) != m_consumerTable.end()) {

sessionCredentials = m_consumerTable[consumerGroup]->getSessionCredentials();

}

if (sessionCredentials.isValid())

return true;

return false;

}

bool MQClientFactory::addConsumerToTable(const string& consumerName, MQConsumer* pMQConsumer) {

boost::lock_guard<boost::recursive_mutex> lock(m_consumerTableMutex);

if (m_consumerTable.find(consumerName) != m_consumerTable.end())

return false;

m_consumerTable[consumerName] = pMQConsumer;

return true;

}

void MQClientFactory::eraseConsumerFromTable(const string& consumerName) {

boost::lock_guard<boost::recursive_mutex> lock(m_consumerTableMutex);

if (m_consumerTable.find(consumerName) != m_consumerTable.end())

m_consumerTable.erase(consumerName); // do not need freee pConsumer, as it

// was allocated by user

else

LOG_WARN("could not find consumer:%s from table", consumerName.c_str());

}

int MQClientFactory::getConsumerTableSize() {

boost::lock_guard<boost::recursive_mutex> lock(m_consumerTableMutex);

return m_consumerTable.size();

}

void MQClientFactory::getTopicListFromConsumerSubscription(set<string>& topicList) {

boost::lock_guard<boost::recursive_mutex> lock(m_consumerTableMutex);

for (MQCMAP::iterator it = m_consumerTable.begin(); it != m_consumerTable.end(); ++it) {

vector<SubscriptionData> result;

it->second->getSubscriptions(result);

vector<SubscriptionData>::iterator iter = result.begin();

for (; iter != result.end(); ++iter) {

topicList.insert((*iter).getTopic());

}

}

}

void MQClientFactory::updateConsumerSubscribeTopicInfo(const string& topic, vector<MQMessageQueue> mqs) {

boost::lock_guard<boost::recursive_mutex> lock(m_consumerTableMutex);

for (MQCMAP::iterator it = m_consumerTable.begin(); it != m_consumerTable.end(); ++it) {

it->second->updateTopicSubscribeInfo(topic, mqs);

}

}

void MQClientFactory::insertConsumerInfoToHeartBeatData(HeartbeatData* pHeartbeatData) {

boost::lock_guard<boost::recursive_mutex> lock(m_consumerTableMutex);

for (MQCMAP::iterator it = m_consumerTable.begin(); it != m_consumerTable.end(); ++it) {

MQConsumer* pConsumer = it->second;

ConsumerData consumerData;

consumerData.groupName = pConsumer->getGroupName();

consumerData.consumeType = pConsumer->getConsumeType();

consumerData.messageModel = pConsumer->getMessageModel();

consumerData.consumeFromWhere = pConsumer->getConsumeFromWhere();

//<!fill data;

vector<SubscriptionData> result;

pConsumer->getSubscriptions(result);

consumerData.subscriptionDataSet.swap(result);

pHeartbeatData->insertDataToConsumerDataSet(consumerData);

}

}

void MQClientFactory::addTopicInfoToTable(const string& topic, boost::shared_ptr<TopicPublishInfo> pTopicPublishInfo) {

boost::lock_guard<boost::mutex> lock(m_topicPublishInfoTableMutex);

if (m_topicPublishInfoTable.find(topic) != m_topicPublishInfoTable.end()) {

m_topicPublishInfoTable.erase(topic);

}

m_topicPublishInfoTable[topic] = pTopicPublishInfo;

}

void MQClientFactory::eraseTopicInfoFromTable(const string& topic) {

boost::lock_guard<boost::mutex> lock(m_topicPublishInfoTableMutex);

if (m_topicPublishInfoTable.find(topic) != m_topicPublishInfoTable.end()) {

m_topicPublishInfoTable.erase(topic);

}

}

bool MQClientFactory::isTopicInfoValidInTable(const string& topic) {

boost::lock_guard<boost::mutex> lock(m_topicPublishInfoTableMutex);

if (m_topicPublishInfoTable.find(topic) != m_topicPublishInfoTable.end()) {

if (m_topicPublishInfoTable[topic]->ok())

return true;

}

return false;

}

boost::shared_ptr<TopicPublishInfo> MQClientFactory::getTopicPublishInfoFromTable(const string& topic) {

boost::lock_guard<boost::mutex> lock(m_topicPublishInfoTableMutex);

if (m_topicPublishInfoTable.find(topic) != m_topicPublishInfoTable.end()) {

return m_topicPublishInfoTable[topic];

}

boost::shared_ptr<TopicPublishInfo> pTopicPublishInfo;

return pTopicPublishInfo;

}

void MQClientFactory::getTopicListFromTopicPublishInfo(set<string>& topicList) {

boost::lock_guard<boost::mutex> lock(m_topicPublishInfoTableMutex);

for (TPMap::iterator itp = m_topicPublishInfoTable.begin(); itp != m_topicPublishInfoTable.end(); ++itp) {

topicList.insert(itp->first);

}

}

void MQClientFactory::clearBrokerAddrMap() {

boost::lock_guard<boost::mutex> lock(m_brokerAddrlock);

m_brokerAddrTable.clear();

}

bool MQClientFactory::isBrokerAddressInUse(const std::string& address) {

if (m_topicRouteTableMutex.try_lock()) {

boost::lock_guard<boost::mutex> lk(m_topicRouteTableMutex, boost::adopt_lock_t());

for (TRDMAP::iterator it = m_topicRouteTable.begin(); it != m_topicRouteTable.end(); it++) {

TopicRouteData* topicRouteData = it->second;

vector<BrokerData>& brokerData = topicRouteData->getBrokerDatas();

for (vector<BrokerData>::iterator next = brokerData.begin(); next != brokerData.end(); next++) {

map<int, string>& brokerAddresses = next->brokerAddrs;

for (map<int, string>::iterator entry = brokerAddresses.begin(); entry != brokerAddresses.end(); entry++) {

if (address == entry->second) {

return true;

}

}

}

}

return false;

} else {

LOG_WARN("Cannot lock m_topicRouteTableMutex. Assume %s is still in use", address.c_str());

return true;

}

}

void MQClientFactory::addBrokerToAddrMap(const string& brokerName, map<int, string>& brokerAddrs) {

boost::lock_guard<boost::mutex> lock(m_brokerAddrlock);

if (m_brokerAddrTable.find(brokerName) != m_brokerAddrTable.end()) {

m_brokerAddrTable.erase(brokerName);

}

m_brokerAddrTable[brokerName] = brokerAddrs;

}

MQClientFactory::BrokerAddrMAP MQClientFactory::getBrokerAddrMap() {

boost::lock_guard<boost::mutex> lock(m_brokerAddrlock);

return m_brokerAddrTable;

}

string MQClientFactory::findBrokerAddressInPublish(const string& brokerName) {

/*reslove the concurrent access m_brokerAddrTable by

findBrokerAddressInPublish(called by sendKernlImpl) And

sendHeartbeatToAllBroker, which leads hign RT of sendMsg

1. change m_brokerAddrTable from hashMap to map;

2. do not add m_factoryLock here, but copy m_brokerAddrTable,

this is used to avoid con-current access m_factoryLock by

findBrokerAddressInPublish(called by sendKernlImpl) And

updateTopicRouteInfoFromNameServer

Note: after copying m_brokerAddrTable, updateTopicRouteInfoFromNameServer

modify m_brokerAddrTable imediatly,

after 1st send fail, producer will get topicPushlibshInfo again

before next try, so 2nd try will get correct broker to send ms;

*/

BrokerAddrMAP brokerTable(getBrokerAddrMap());

string brokerAddr;

bool found = false;

if (brokerTable.find(brokerName) != brokerTable.end()) {

map<int, string> brokerMap(brokerTable[brokerName]);

map<int, string>::iterator it1 = brokerMap.find(MASTER_ID);

if (it1 != brokerMap.end()) {

brokerAddr = it1->second;

found = true;

}

}

brokerTable.clear();

if (found)

return brokerAddr;

return "";

}

FindBrokerResult* MQClientFactory::findBrokerAddressInSubscribe(const string& brokerName,

int brokerId,

bool onlyThisBroker) {

string brokerAddr;

bool slave = false;

bool found = false;

BrokerAddrMAP brokerTable(getBrokerAddrMap());

if (brokerTable.find(brokerName) != brokerTable.end()) {

map<int, string> brokerMap(brokerTable[brokerName]);

if (!brokerMap.empty()) {

auto iter = brokerMap.find(brokerId);

if (iter != brokerMap.end()) {

brokerAddr = iter->second;

slave = (brokerId != MASTER_ID);

found = true;

} else if (!onlyThisBroker) { // not only from master

iter = brokerMap.begin();

brokerAddr = iter->second;

slave = iter->first != MASTER_ID;

found = true;

}

}

}

brokerTable.clear();

if (found) {

return new FindBrokerResult(brokerAddr, slave);

}

return nullptr;

}

FindBrokerResult* MQClientFactory::findBrokerAddressInAdmin(const string& brokerName) {

BrokerAddrMAP brokerTable(getBrokerAddrMap());

bool found = false;

bool slave = false;

string brokerAddr;

if (brokerTable.find(brokerName) != brokerTable.end()) {

map<int, string> brokerMap(brokerTable[brokerName]);

map<int, string>::iterator it1 = brokerMap.begin();

if (it1 != brokerMap.end()) {

slave = (it1->first != MASTER_ID);

found = true;

brokerAddr = it1->second;

}

}

brokerTable.clear();

if (found)

return new FindBrokerResult(brokerAddr, slave);

return NULL;

}

void MQClientFactory::checkTransactionState(const std::string& addr,

const MQMessageExt& messageExt,

const CheckTransactionStateRequestHeader& checkRequestHeader) {

string group = messageExt.getProperty(MQMessage::PROPERTY_PRODUCER_GROUP);

if (!group.empty()) {

MQProducer* producer = selectProducer(group);

if (producer != nullptr) {

TransactionMQProducerImpl* transProducer = dynamic_cast<TransactionMQProducerImpl*>(producer);

if (transProducer != nullptr) {

transProducer->checkTransactionState(addr, messageExt, checkRequestHeader.m_tranStateTableOffset,

checkRequestHeader.m_commitLogOffset, checkRequestHeader.m_msgId,

checkRequestHeader.m_transactionId, checkRequestHeader.m_offsetMsgId);

} else {

LOG_ERROR("checkTransactionState, producer not TransactionMQProducer failed, msg:%s",

messageExt.toString().data());

}

} else {

LOG_ERROR("checkTransactionState, pick producer by group[%s] failed, msg:%s", group.data(),

messageExt.toString().data());

}

} else {

LOG_ERROR("checkTransactionState, pick producer group failed, msg:%s", messageExt.toString().data());

}

}

MQClientAPIImpl* MQClientFactory::getMQClientAPIImpl() {

return m_pClientAPIImpl.get();

}

void MQClientFactory::cleanOfflineBrokers() {

LOG_DEBUG("Begin to clean offline brokers");

boost::lock_guard<boost::mutex> lock(m_brokerAddrlock);

for (BrokerAddrMAP::iterator it = m_brokerAddrTable.begin(); it != m_brokerAddrTable.end();) {

std::string brokerName = it->first;

map<int, std::string> brokerIdAddressMap = it->second;

for (map<int, std::string>::iterator next = brokerIdAddressMap.begin(); next != brokerIdAddressMap.end();) {

if (!isBrokerAddressInUse(next->second)) {

LOG_INFO("Remove broker address: %s", (next->second).c_str());

brokerIdAddressMap.erase(next++);

} else {

next++;

}

}

if (brokerIdAddressMap.empty()) {

m_brokerAddrTable.erase(it++);

LOG_INFO("Broker name: %s is purged from client", brokerName.c_str());

} else {

LOG_DEBUG("Broker: %s is alive", brokerName.c_str());

it++;

}

}

LOG_DEBUG("Exit of cleaning offline brokers");

}

void MQClientFactory::sendHeartbeatToAllBroker() {

BrokerAddrMAP brokerTable(getBrokerAddrMap());

if (brokerTable.size() == 0) {

LOG_WARN("sendheartbeat brokeradd is empty");

return;

}

unique_ptr<HeartbeatData> heartbeatData(prepareHeartbeatData());

bool producerEmpty = heartbeatData->isProducerDataSetEmpty();

bool consumerEmpty = heartbeatData->isConsumerDataSetEmpty();

if (producerEmpty && consumerEmpty) {

LOG_WARN("sendheartbeat heartbeatData empty");

brokerTable.clear();

return;

}

SessionCredentials session_credentials;

getSessionCredentialsFromOneOfProducerOrConsumer(session_credentials);

for (BrokerAddrMAP::iterator it = brokerTable.begin(); it != brokerTable.end(); ++it) {

map<int, string> brokerMap(it->second);

map<int, string>::iterator it1 = brokerMap.begin();

for (; it1 != brokerMap.end(); ++it1) {

string& addr = it1->second;

if (consumerEmpty && it1->first != MASTER_ID)

continue;

try {

m_pClientAPIImpl->sendHeartbeat(addr, heartbeatData.get(), session_credentials);

} catch (MQException& e) {

LOG_ERROR("%s", e.what());

}

}

}

brokerTable.clear();

}

void MQClientFactory::persistAllConsumerOffset(boost::system::error_code& ec,

boost::shared_ptr<boost::asio::deadline_timer> t) {

{

boost::lock_guard<boost::recursive_mutex> lock(m_consumerTableMutex);

if (m_consumerTable.size() > 0) {

for (MQCMAP::iterator it = m_consumerTable.begin(); it != m_consumerTable.end(); ++it) {

LOG_DEBUG("Client factory start persistAllConsumerOffset");

it->second->persistConsumerOffset();

}

}

}

boost::system::error_code e;

t->expires_from_now(t->expires_from_now() + boost::posix_time::seconds(5), e);

t->async_wait(boost::bind(&MQClientFactory::persistAllConsumerOffset, this, ec, t));

}

HeartbeatData* MQClientFactory::prepareHeartbeatData() {

HeartbeatData* pHeartbeatData = new HeartbeatData();

// clientID

pHeartbeatData->setClientID(m_clientId);

// Consumer

insertConsumerInfoToHeartBeatData(pHeartbeatData);

// Producer

insertProducerInfoToHeartBeatData(pHeartbeatData);

return pHeartbeatData;

}

void MQClientFactory::timerCB_sendHeartbeatToAllBroker(boost::system::error_code& ec,

boost::shared_ptr<boost::asio::deadline_timer> t) {

sendHeartbeatToAllBroker();

boost::system::error_code e;

t->expires_from_now(t->expires_from_now() + boost::posix_time::seconds(30), e);

t->async_wait(boost::bind(&MQClientFactory::timerCB_sendHeartbeatToAllBroker, this, ec, t));

}

void MQClientFactory::timerCB_cleanOfflineBrokers(boost::system::error_code& ec,

boost::shared_ptr<boost::asio::deadline_timer> t) {

cleanOfflineBrokers();

boost::system::error_code e;

t->expires_from_now(t->expires_from_now() + boost::posix_time::seconds(30), e);

t->async_wait(boost::bind(&MQClientFactory::timerCB_cleanOfflineBrokers, this, ec, t));

}

void MQClientFactory::fetchNameServerAddr(boost::system::error_code& ec,

boost::shared_ptr<boost::asio::deadline_timer> t) {

m_pClientAPIImpl->fetchNameServerAddr(m_nameSrvDomain);

boost::system::error_code e;

t->expires_from_now(t->expires_from_now() + boost::posix_time::seconds(60 * 2), e);

t->async_wait(boost::bind(&MQClientFactory::fetchNameServerAddr, this, ec, t));

}

void MQClientFactory::startScheduledTask(bool startFetchNSService) {

boost::asio::io_service::work work(m_async_ioService); // avoid async io

// service stops after

// first timer timeout

// callback

boost::system::error_code ec1;

boost::shared_ptr<boost::asio::deadline_timer> t1 =

boost::make_shared<boost::asio::deadline_timer>(m_async_ioService, boost::posix_time::seconds(3));

t1->async_wait(boost::bind(&MQClientFactory::updateTopicRouteInfo, this, ec1, t1));

boost::system::error_code ec2;

boost::shared_ptr<boost::asio::deadline_timer> t2 =

boost::make_shared<boost::asio::deadline_timer>(m_async_ioService, boost::posix_time::milliseconds(10));

t2->async_wait(boost::bind(&MQClientFactory::timerCB_sendHeartbeatToAllBroker, this, ec2, t2));

boost::system::error_code ec3;

boost::shared_ptr<boost::asio::deadline_timer> t3 =

boost::make_shared<boost::asio::deadline_timer>(m_async_ioService, boost::posix_time::seconds(3));

t3->async_wait(boost::bind(&MQClientFactory::timerCB_cleanOfflineBrokers, this, ec3, t3));

if (startFetchNSService) {

boost::system::error_code ec5;

boost::shared_ptr<boost::asio::deadline_timer> t5 =

boost::make_shared<boost::asio::deadline_timer>(m_async_ioService, boost::posix_time::seconds(60 * 2));

t5->async_wait(boost::bind(&MQClientFactory::fetchNameServerAddr, this, ec5, t5));

}

LOG_INFO("start scheduled task:%s", m_clientId.c_str());

boost::system::error_code ec;

m_async_ioService.run(ec);

}

void MQClientFactory::rebalanceImmediately() {

// m_consumer_async_service_thread will be only started once for all consumer

if (m_consumer_async_service_thread == NULL) {

doRebalance();

m_consumer_async_service_thread.reset(

new boost::thread(boost::bind(&MQClientFactory::consumer_timerOperation, this)));

}

}

void MQClientFactory::consumer_timerOperation() {

LOG_INFO("clientFactory:%s start consumer_timerOperation", m_clientId.c_str());

boost::asio::io_service::work work(m_consumer_async_ioService); // avoid async io

// service stops after

// first timer timeout

// callback

boost::system::error_code ec1;

boost::shared_ptr<boost::asio::deadline_timer> t1 =

boost::make_shared<boost::asio::deadline_timer>(m_consumer_async_ioService, boost::posix_time::seconds(10));

t1->async_wait(boost::bind(&MQClientFactory::timerCB_doRebalance, this, ec1, t1));

boost::system::error_code ec2;

boost::shared_ptr<boost::asio::deadline_timer> t2 =

boost::make_shared<boost::asio::deadline_timer>(m_consumer_async_ioService, boost::posix_time::seconds(5));

t2->async_wait(boost::bind(&MQClientFactory::persistAllConsumerOffset, this, ec2, t2));

boost::system::error_code ec;

m_consumer_async_ioService.run(ec);

LOG_INFO("clientFactory:%s stop consumer_timerOperation", m_clientId.c_str());

}

void MQClientFactory::timerCB_doRebalance(boost::system::error_code& ec,

boost::shared_ptr<boost::asio::deadline_timer> t) {

doRebalance();

boost::system::error_code e;

t->expires_from_now(t->expires_from_now() + boost::posix_time::seconds(10), e);

t->async_wait(boost::bind(&MQClientFactory::timerCB_doRebalance, this, ec, t));

}

void MQClientFactory::doRebalance() {

LOG_DEBUG("Client factory:%s start doRebalance", m_clientId.c_str());

if (getConsumerTableSize() > 0) {

boost::lock_guard<boost::recursive_mutex> lock(m_consumerTableMutex);

for (MQCMAP::iterator it = m_consumerTable.begin(); it != m_consumerTable.end(); ++it) {

it->second->doRebalance();

}

}

LOG_DEBUG("Client factory:%s finish doRebalance", m_clientId.c_str());

}

void MQClientFactory::doRebalanceByConsumerGroup(const string& consumerGroup) {

boost::lock_guard<boost::recursive_mutex> lock(m_consumerTableMutex);

if (m_consumerTable.find(consumerGroup) != m_consumerTable.end()) {

LOG_INFO("Client factory:%s start dorebalance for consumer:%s", m_clientId.c_str(), consumerGroup.c_str());

MQConsumer* pMQConsumer = m_consumerTable[consumerGroup];

pMQConsumer->doRebalance();

}

}

void MQClientFactory::endTransactionOneway(const MQMessageQueue& mq,

EndTransactionRequestHeader* requestHeader,

const SessionCredentials& sessionCredentials) {

string brokerAddr = findBrokerAddressInPublish(mq.getBrokerName());

string remark = "";

if (!brokerAddr.empty()) {

try {

getMQClientAPIImpl()->endTransactionOneway(brokerAddr, requestHeader, remark, sessionCredentials);

} catch (MQException& e) {

LOG_ERROR("endTransactionOneway exception:%s", e.what());

throw e;

}

} else {

THROW_MQEXCEPTION(MQClientException, "The broker[" + mq.getBrokerName() + "] not exist", -1);

}

}

void MQClientFactory::unregisterClient(const string& producerGroup,

const string& consumerGroup,

const SessionCredentials& sessionCredentials) {

BrokerAddrMAP brokerTable(getBrokerAddrMap());

for (BrokerAddrMAP::iterator it = brokerTable.begin(); it != brokerTable.end(); ++it) {

map<int, string> brokerMap(it->second);

map<int, string>::iterator it1 = brokerMap.begin();

for (; it1 != brokerMap.end(); ++it1) {

string& addr = it1->second;

m_pClientAPIImpl->unregisterClient(addr, m_clientId, producerGroup, consumerGroup, sessionCredentials);

}

}

}

//<!************************************************************************

void MQClientFactory::fetchSubscribeMessageQueues(const string& topic,

vector<MQMessageQueue>& mqs,

const SessionCredentials& sessionCredentials) {

TopicRouteData* pTopicRouteData = getTopicRouteData(topic);

if (pTopicRouteData == NULL) {

updateTopicRouteInfoFromNameServer(topic, sessionCredentials);

pTopicRouteData = getTopicRouteData(topic);

}

if (pTopicRouteData != NULL) {

topicRouteData2TopicSubscribeInfo(topic, pTopicRouteData, mqs);

if (mqs.empty()) {

THROW_MQEXCEPTION(MQClientException, "Can not find Message Queue", -1);

}