/**
* Copyright (C) 2013 kangliqiang ,kangliq@163.com
*
* Licensed 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 <math.h>
#include <set>
#include <string>
#include <iostream>
#include <vector>

#include "MQClientFactory.h"
#include "RemoteClientConfig.h"
#include "ClientRemotingProcessor.h"
#include "MQClientAPIImpl.h"
#include "MQAdminImpl.h"
#include "DefaultMQProducer.h"
#include "PullMessageService.h"
#include "RebalanceService.h"
#include "ScopedLock.h"
#include "KPRUtil.h"
#include "DefaultMQProducerImpl.h"
#include "DefaultMQPushConsumerImpl.h"
#include "MQClientException.h"
#include "MQConsumerInner.h"
#include "MQProducerInner.h"
#include "UtilAll.h"
#include "PermName.h"
#include "MQClientManager.h"
#include "ConsumerStatManage.h"
#include "TopicPublishInfo.h"
#include "MQVersion.h"

namespace rmq
{


long MQClientFactory::LockTimeoutMillis = 3000;

MQClientFactory::MQClientFactory(ClientConfig& clientConfig, int factoryIndex, const std::string& clientId)
{
    m_clientConfig = clientConfig;
    m_factoryIndex = factoryIndex;
    m_pRemoteClientConfig = new RemoteClientConfig();
    m_pRemoteClientConfig->clientCallbackExecutorThreads = clientConfig.getClientCallbackExecutorThreads();
    m_pClientRemotingProcessor = new ClientRemotingProcessor(this);
    m_pMQClientAPIImpl = new MQClientAPIImpl(m_clientConfig, *m_pRemoteClientConfig, m_pClientRemotingProcessor);

    if (!m_clientConfig.getNamesrvAddr().empty())
    {
        m_pMQClientAPIImpl->updateNameServerAddressList(m_clientConfig.getNamesrvAddr());
        RMQ_INFO("user specified name server address: {%s}", m_clientConfig.getNamesrvAddr().c_str());
    }

    m_clientId = clientId;

    m_pMQAdminImpl = new MQAdminImpl(this);
    m_pPullMessageService = new PullMessageService(this);
    m_pRebalanceService = new RebalanceService(this);
    m_pDefaultMQProducer = new DefaultMQProducer(MixAll::CLIENT_INNER_PRODUCER_GROUP);
    m_pDefaultMQProducer->resetClientConfig(clientConfig);
    m_bootTimestamp = KPRUtil::GetCurrentTimeMillis();

    m_pFetchNameServerAddrTask = new ScheduledTask(this, &MQClientFactory::fetchNameServerAddr);
    m_pUpdateTopicRouteInfoFromNameServerTask = new ScheduledTask(this, &MQClientFactory::updateTopicRouteInfoFromNameServerTask);
    m_pCleanBrokerTask = new ScheduledTask(this, &MQClientFactory::cleanBroker);
    m_pPersistAllConsumerOffsetTask = new ScheduledTask(this, &MQClientFactory::persistAllConsumerOffsetTask);
    m_pRecordSnapshotPeriodicallyTask = new ScheduledTask(this, &MQClientFactory::recordSnapshotPeriodicallyTask);
    m_pLogStatsPeriodicallyTask = new ScheduledTask(this, &MQClientFactory::logStatsPeriodicallyTask);

    m_serviceState = CREATE_JUST;

    RMQ_INFO("created a new client Instance, FactoryIndex: {%d} ClinetID: {%s} Config: {%s} Version: {%s}",
                m_factoryIndex,
                m_clientId.c_str(),
                m_clientConfig.toString().c_str(),
                MQVersion::getVersionDesc(MQVersion::s_CurrentVersion));
}

MQClientFactory::~MQClientFactory()
{
    delete m_pRemoteClientConfig;
    delete m_pClientRemotingProcessor;
    delete m_pMQClientAPIImpl;
    delete m_pMQAdminImpl;
    delete m_pPullMessageService;
    delete m_pRebalanceService;
    delete m_pDefaultMQProducer;
}

void MQClientFactory::start()
{
    RMQ_DEBUG("MQClientFactory::start()");
    kpr::ScopedLock<kpr::Mutex> lock(m_mutex);
    switch (m_serviceState)
    {
        case CREATE_JUST:
            makesureInstanceNameIsOnly(m_clientConfig.getInstanceName());

            m_serviceState = START_FAILED;
            if (m_clientConfig.getNamesrvAddr().empty())
            {
                m_clientConfig.setNamesrvAddr(m_pMQClientAPIImpl->fetchNameServerAddr());
            }

            m_pMQClientAPIImpl->start();
            m_timerTaskManager.Init(5, 1000);
            startScheduledTask();
            m_pPullMessageService->Start();
            m_pRebalanceService->Start();
            m_pDefaultMQProducer->getDefaultMQProducerImpl()->start(false);

			RMQ_INFO("the client factory [%s] start OK", m_clientId.c_str());
            m_serviceState = RUNNING;
            break;
        case RUNNING:
            RMQ_WARN("MQClientFactory is already running.");
            break;
        case SHUTDOWN_ALREADY:
            RMQ_ERROR("MQClientFactory should have already been shutted down");
            break;
        case START_FAILED:
            RMQ_ERROR("MQClientFactory started failed.");
            THROW_MQEXCEPTION(MQClientException, "The Factory object start failed", -1);
        default:
            break;
    }
}


void MQClientFactory::shutdown()
{
    RMQ_DEBUG("MQClientFactory::shutdown()");
    // Consumer
    if (!m_consumerTable.empty())
    {
        return;
    }

    // AdminExt
    if (!m_adminExtTable.empty())
    {
        return;
    }

    // Producer
    if (m_producerTable.size() > 1)
    {
        return;
    }

    RMQ_DEBUG("MQClientFactory::shutdown_begin");
    {
        kpr::ScopedLock<kpr::Mutex> lock(m_mutex);
        switch (m_serviceState)
        {
            case CREATE_JUST:
                break;
            case RUNNING:
                m_pDefaultMQProducer->getDefaultMQProducerImpl()->shutdown(false);

                for (int i = 0; i < 6; i++)
                {
                    m_timerTaskManager.UnRegisterTimer(m_scheduledTaskIds[i]);
                }

                m_timerTaskManager.Stop();

                m_pPullMessageService->stop();
                m_pPullMessageService->Join();

                m_pMQClientAPIImpl->shutdown();
                m_pRebalanceService->stop();
                m_pRebalanceService->Join();

                //closesocket(m_datagramSocket);

                MQClientManager::getInstance()->removeClientFactory(m_clientId);
                m_serviceState = SHUTDOWN_ALREADY;
                break;
            case SHUTDOWN_ALREADY:
                break;
            default:
                break;
        }
    }
}


void MQClientFactory::sendHeartbeatToAllBrokerWithLock()
{
	RMQ_DEBUG("TryLock m_lockHeartbeat: %p", &m_lockHeartbeat);
    if (m_lockHeartbeat.TryLock())
    {
        try
        {
            RMQ_DEBUG("TryLock m_lockHeartbeat ok");
            sendHeartbeatToAllBroker();
        }
        catch (...)
        {
            RMQ_ERROR("sendHeartbeatToAllBroker exception");
        }
        m_lockHeartbeat.Unlock();
    }
    else
    {
        RMQ_WARN("TryLock heartBeat fail");
    }
}

void MQClientFactory::updateTopicRouteInfoFromNameServer()
{
    std::set<std::string> topicList;

    // Consumer
    {
    	kpr::ScopedRLock<kpr::RWMutex> lock(m_consumerTableLock);
        std::map<std::string, MQConsumerInner*>::iterator it = m_consumerTable.begin();
        for (; it != m_consumerTable.end(); it++)
        {
            MQConsumerInner* inner = it->second;
            std::set<SubscriptionData> subList = inner->subscriptions();
            std::set<SubscriptionData>::iterator it1 = subList.begin();
            for (; it1 != subList.end(); it1++)
            {
                topicList.insert((*it1).getTopic());
            }
        }
    }

    // Producer
    {
    	kpr::ScopedRLock<kpr::RWMutex> lock(m_producerTableLock);
        std::map<std::string, MQProducerInner*>::iterator it = m_producerTable.begin();
        for (; it != m_producerTable.end(); it++)
        {
            MQProducerInner* inner = it->second;
            std::set<std::string> pubList = inner->getPublishTopicList();
            topicList.insert(pubList.begin(), pubList.end());
        }
    }

    std::set<std::string>::iterator it2 = topicList.begin();
    for (; it2 != topicList.end(); it2++)
    {
        updateTopicRouteInfoFromNameServer(*it2);
    }
}

bool MQClientFactory::updateTopicRouteInfoFromNameServer(const std::string& topic)
{
    return updateTopicRouteInfoFromNameServer(topic, false, NULL);
}

bool MQClientFactory::updateTopicRouteInfoFromNameServer(const std::string& topic,
        bool isDefault,
        DefaultMQProducer* pDefaultMQProducer)
{
    RMQ_DEBUG("TryLock m_lockNamesrv: 0x%p, topic: [%s]", &m_lockNamesrv, topic.c_str());
    if (m_lockNamesrv.TryLock(MQClientFactory::LockTimeoutMillis))
    {
        RMQ_DEBUG("TryLock m_lockNamesrv ok");
        TopicRouteDataPtr topicRouteData = NULL;
        try
        {
            if (isDefault && pDefaultMQProducer != NULL)
            {
                topicRouteData =
                    m_pMQClientAPIImpl->getDefaultTopicRouteInfoFromNameServer(
                        pDefaultMQProducer->getCreateTopicKey(), 1000 * 3);
                if (topicRouteData.ptr() != NULL)
                {
                    std::list<QueueData> dataList = topicRouteData->getQueueDatas();

                    std::list<QueueData>::iterator it = dataList.begin();
                    for (; it != dataList.end(); it++)
                    {
                        QueueData data = *it;

                        int queueNums =
                            std::min<int>(pDefaultMQProducer->getDefaultTopicQueueNums(),
                                          data.readQueueNums);
                        data.readQueueNums = (queueNums);
                        data.writeQueueNums = (queueNums);
                    }
                }
            }
            else
            {
                topicRouteData =
                    m_pMQClientAPIImpl->getTopicRouteInfoFromNameServer(topic, 1000 * 3);
            }

            if (topicRouteData.ptr() != NULL)
            {
                kpr::ScopedWLock<kpr::RWMutex> lock(m_topicRouteTableLock);
                std::map<std::string, TopicRouteData>::iterator it = m_topicRouteTable.find(topic);
                bool changed = false;

                if (it != m_topicRouteTable.end())
                {
                    changed = topicRouteDataIsChange(it->second, *topicRouteData);
                    if (!changed)
                    {
                        changed = isNeedUpdateTopicRouteInfo(topic);
						if (changed)
						{
                        	RMQ_INFO("the topic[{%s}] route info changed, old[{%s}] ,new[{%s}]",
                                 topic.c_str(), it->second.toString().c_str(),
                                 topicRouteData->toString().c_str());
                        }
                    }
                }
                else
                {
                    changed = true;
                }

                if (changed)
                {
                    TopicRouteData cloneTopicRouteData = *topicRouteData;

                    std::list<BrokerData> dataList = topicRouteData->getBrokerDatas();

                    std::list<BrokerData>::iterator it = dataList.begin();
                    for (; it != dataList.end(); it++)
                    {
                        kpr::ScopedWLock<kpr::RWMutex> lock(m_brokerAddrTableLock);
                        m_brokerAddrTable[(*it).brokerName] = (*it).brokerAddrs;
                    }

                    {
                        TopicPublishInfoPtr publishInfo =
                            topicRouteData2TopicPublishInfo(topic, *topicRouteData);
                        publishInfo->setHaveTopicRouterInfo(true);

                        kpr::ScopedRLock<kpr::RWMutex> lock(m_producerTableLock);
                        std::map<std::string, MQProducerInner*>::iterator it = m_producerTable.begin();
                        for (; it != m_producerTable.end(); it++)
                        {
                            MQProducerInner* impl = it->second;
                            if (impl)
                            {
                                impl->updateTopicPublishInfo(topic, *publishInfo);
                            }
                        }
                    }

                    {
                        std::set<MessageQueue>* subscribeInfo =
                            topicRouteData2TopicSubscribeInfo(topic, *topicRouteData);

                        kpr::ScopedRLock<kpr::RWMutex> lock(m_consumerTableLock);
                        std::map<std::string, MQConsumerInner*>::iterator it = m_consumerTable.begin();
                        for (; it != m_consumerTable.end(); it++)
                        {
                            MQConsumerInner* impl = it->second;
                            if (impl)
                            {
                                impl->updateTopicSubscribeInfo(topic, *subscribeInfo);
                            }
                        }
                        delete subscribeInfo;
                    }

                    m_topicRouteTable[topic] = cloneTopicRouteData;
                    m_lockNamesrv.Unlock();
                    RMQ_DEBUG("UnLock m_lockNamesrv ok");

                    RMQ_INFO("topicRouteTable.put[%s] = TopicRouteData[%s]",
                    	topic.c_str(), cloneTopicRouteData.toString().c_str());
                    return true;
                }
            }
            else
            {
                //TODO log?
                RMQ_WARN("updateTopicRouteInfoFromNameServer, getTopicRouteInfoFromNameServer return null, Topic: {%s}",
                         topic.c_str());
            }
        }
        catch (const std::exception& e)
        {
        	if (!(topic.find(MixAll::RETRY_GROUP_TOPIC_PREFIX) == 0) && topic != MixAll::DEFAULT_TOPIC)
        	{
            	RMQ_WARN("updateTopicRouteInfoFromNameServer Exception: %s", e.what());
            }
        }
        catch (...)
        {
            RMQ_WARN("updateTopicRouteInfoFromNameServer unknow Exception");
        }

        m_lockNamesrv.Unlock();
        RMQ_DEBUG("UnLock m_lockNamesrv ok");
    }
    else
    {
        RMQ_WARN("TryLock m_lockNamesrv timeout %ldms", MQClientFactory::LockTimeoutMillis);
    }

    return false;
}

TopicPublishInfo*  MQClientFactory::topicRouteData2TopicPublishInfo(const std::string& topic,
        TopicRouteData& route)
{
    TopicPublishInfo* info = new TopicPublishInfo();
    if (!route.getOrderTopicConf().empty())
    {
        std::vector<std::string> brokers;
        UtilAll::Split(brokers, route.getOrderTopicConf(), ";");
        for (size_t i = 0; i < brokers.size(); i++)
        {
            std::vector<std::string> item;
            UtilAll::Split(item, brokers[i], ":");
            int nums = atoi(item[1].c_str());
            for (int i = 0; i < nums; i++)
            {
                MessageQueue mq(topic, item[0], i);
                info->getMessageQueueList().push_back(mq);
            }
        }

        info->setOrderTopic(true);
    }
    else
    {
        std::list<QueueData> qds = route.getQueueDatas();
        qds.sort();
        std::list<QueueData>::iterator it = qds.begin();
        for (; it != qds.end(); it++)
        {
            QueueData& qd = (*it);
            if (PermName::isWriteable(qd.perm))
            {
                bool find = false;
                BrokerData brokerData;
                std::list<BrokerData> bds = route.getBrokerDatas();
                std::list<BrokerData>::iterator it1 = bds.begin();

                for (; it1 != bds.end(); it1++)
                {
                    BrokerData& bd = (*it1);
                    if (bd.brokerName == qd.brokerName)
                    {
                        brokerData = bd;
                        find = true;
                        break;
                    }
                }

                if (!find)
                {
                    continue;
                }

                if (brokerData.brokerAddrs.find(MixAll::MASTER_ID) == brokerData.brokerAddrs.end())
                {
                    continue;
                }

                for (int i = 0; i < qd.writeQueueNums; i++)
                {
                    MessageQueue mq(topic, qd.brokerName, i);
                    info->getMessageQueueList().push_back(mq);
                }
            }
        }

        info->setOrderTopic(false);
    }

    return info;
}

std::set<MessageQueue>* MQClientFactory::topicRouteData2TopicSubscribeInfo(const std::string& topic,
        TopicRouteData& route)
{
    std::set<MessageQueue>* mqList = new std::set<MessageQueue>();
    std::list<QueueData> qds = route.getQueueDatas();
    std::list<QueueData>::iterator it = qds.begin();
    for (; it != qds.end(); it++)
    {
        QueueData& qd = (*it);
        if (PermName::isReadable(qd.perm))
        {
            for (int i = 0; i < qd.readQueueNums; i++)
            {
                MessageQueue mq(topic, qd.brokerName, i);
                mqList->insert(mq);
            }
        }
    }

    return mqList;
}

bool MQClientFactory::registerConsumer(const std::string& group, MQConsumerInner* pConsumer)
{
    if (group.empty() || pConsumer == NULL)
    {
        return false;
    }

	kpr::ScopedWLock<kpr::RWMutex> lock(m_consumerTableLock);
    if (m_consumerTable.find(group) != m_consumerTable.end())
    {
        return false;
    }
    m_consumerTable[group] = pConsumer;

    return true;
}

void MQClientFactory::unregisterConsumer(const std::string& group)
{
	{
		kpr::ScopedWLock<kpr::RWMutex> lock(m_consumerTableLock);
	    m_consumerTable.erase(group);
    }
    unregisterClientWithLock("", group);
}

bool MQClientFactory::registerProducer(const std::string& group, DefaultMQProducerImpl* pProducer)
{
    if (group.empty() || pProducer == NULL)
    {
        return false;
    }

	kpr::ScopedWLock<kpr::RWMutex> lock(m_producerTableLock);
    if (m_producerTable.find(group) != m_producerTable.end())
    {
        return false;
    }
    m_producerTable[group] = pProducer;

    return true;
}

void MQClientFactory::unregisterProducer(const std::string& group)
{
	{
		kpr::ScopedWLock<kpr::RWMutex> lock(m_producerTableLock);
	    m_producerTable.erase(group);
    }
    unregisterClientWithLock(group, "");
}

bool MQClientFactory::registerAdminExt(const std::string& group, MQAdminExtInner* pAdmin)
{
    if (group.empty() || pAdmin == NULL)
    {
        return false;
    }

	kpr::ScopedWLock<kpr::RWMutex> lock(m_adminExtTableLock);
    if (m_adminExtTable.find(group) != m_adminExtTable.end())
    {
        return false;
    }
    m_adminExtTable[group] = pAdmin;

    return true;
}

void MQClientFactory::unregisterAdminExt(const std::string& group)
{
	kpr::ScopedWLock<kpr::RWMutex> lock(m_adminExtTableLock);
    m_adminExtTable.erase(group);
}

void MQClientFactory::rebalanceImmediately()
{
    m_pRebalanceService->wakeup();
}

void MQClientFactory::doRebalance()
{
	kpr::ScopedRLock<kpr::RWMutex> lock(m_consumerTableLock);
    std::map<std::string, MQConsumerInner*>::iterator it = m_consumerTable.begin();
    for (; it != m_consumerTable.end(); it++)
    {
        MQConsumerInner* impl = it->second;
        if (impl != NULL)
        {
            try
            {
                impl->doRebalance();
            }
            catch (std::exception& e)
            {
                RMQ_ERROR("doRebalance exception, %s", e.what());
            }
            catch (...)
            {
                RMQ_ERROR("doRebalance unknow exception");
            }
        }
    }
}

MQProducerInner* MQClientFactory::selectProducer(const std::string& group)
{
	kpr::ScopedRLock<kpr::RWMutex> lock(m_producerTableLock);
    std::map<std::string, MQProducerInner*>::iterator it = m_producerTable.find(group);
    if (it != m_producerTable.end())
    {
        return it->second;
    }

    return NULL;
}

MQConsumerInner* MQClientFactory::selectConsumer(const std::string& group)
{
	kpr::ScopedRLock<kpr::RWMutex> lock(m_consumerTableLock);
    std::map<std::string, MQConsumerInner*>::iterator it = m_consumerTable.find(group);
    if (it != m_consumerTable.end())
    {
        return it->second;
    }

    return NULL;
}

FindBrokerResult MQClientFactory::findBrokerAddressInAdmin(const std::string& brokerName)
{
    //TODO
    FindBrokerResult result;
    std::string brokerAddr;
    bool slave = false;
    bool found = false;

    kpr::ScopedRLock<kpr::RWMutex> lock(m_brokerAddrTableLock);
    typeof(m_brokerAddrTable.begin()) it = m_brokerAddrTable.find(brokerName);
    if (it != m_brokerAddrTable.end())
    {
        // TODO more slave
        typeof(it->second.begin()) it1 = it->second.begin();
        for (; it1 != it->second.end(); it1++)
        {
            int brockerId = it1->first;
            brokerAddr = it1->second;
            if (!brokerAddr.empty())
            {
                found = true;
                if (MixAll::MASTER_ID == brockerId)
                {
                    slave = false;
                }
                else
                {
                    slave = true;
                }
                break;
            }
        }
    }

    if (found)
    {
        result.brokerAddr = brokerAddr;
        result.slave = slave;
    }

    return result;
}

std::string MQClientFactory::findBrokerAddressInPublish(const std::string& brokerName)
{
    kpr::ScopedRLock<kpr::RWMutex> lock(m_brokerAddrTableLock);
    std::map<std::string, std::map<int, std::string> >::iterator it = m_brokerAddrTable.find(brokerName);
    if (it != m_brokerAddrTable.end())
    {
        std::map<int, std::string>::iterator it1 = it->second.find(MixAll::MASTER_ID);
        if (it1 != it->second.end())
        {
            return it1->second;
        }
    }

    return "";
}

FindBrokerResult MQClientFactory::findBrokerAddressInSubscribe(const std::string& brokerName,
        long brokerId,
        bool onlyThisBroker)
{
    std::string brokerAddr = "";
    bool slave = false;
    bool found = false;

    kpr::ScopedRLock<kpr::RWMutex> lock(m_brokerAddrTableLock);
    std::map<std::string, std::map<int, std::string> >::iterator it = m_brokerAddrTable.find(brokerName);
    if (it != m_brokerAddrTable.end())
    {
        std::map<int, std::string>::iterator it1 = it->second.find(brokerId);
        if (it1 != it->second.end())
        {
            brokerAddr = it1->second;
            slave = (brokerId != MixAll::MASTER_ID);
            found = true;
        }
        else
        {
            it1 = it->second.begin();
            brokerAddr = it1->second;
            slave = (brokerId != MixAll::MASTER_ID);
            found = true;
        }
    }

    FindBrokerResult result;
    result.brokerAddr = brokerAddr;
    result.slave = slave;

    return result;
}

std::list<std::string> MQClientFactory::findConsumerIdList(const std::string& topic, const std::string& group)
{
    std::string brokerAddr = findBrokerAddrByTopic(topic);

    if (brokerAddr.empty())
    {
        updateTopicRouteInfoFromNameServer(topic);
        brokerAddr = findBrokerAddrByTopic(topic);
    }

    if (!brokerAddr.empty())
    {
        try
        {
            return m_pMQClientAPIImpl->getConsumerIdListByGroup(brokerAddr, group, 3000);
        }
        catch (...)
        {
			RMQ_WARN("getConsumerIdListByGroup exception, %s, %s", brokerAddr.c_str(), group.c_str());
        }
    }

    std::list<std::string> ids;

    return ids;
}

std::string MQClientFactory::findBrokerAddrByTopic(const std::string& topic)
{
    kpr::ScopedRLock<kpr::RWMutex> lock(m_topicRouteTableLock);

    std::map<std::string, TopicRouteData>::iterator it = m_topicRouteTable.find(topic);
    if (it != m_topicRouteTable.end())
    {
        const std::list<BrokerData>& brokers = it->second.getBrokerDatas();

        if (!brokers.empty())
        {
            BrokerData bd = brokers.front();
            return TopicRouteData::selectBrokerAddr(bd);
        }
    }

    return "";
}

TopicRouteData MQClientFactory::getAnExistTopicRouteData(const std::string& topic)
{
    kpr::ScopedRLock<kpr::RWMutex> lock(m_topicRouteTableLock);

    std::map<std::string, TopicRouteData>::iterator it = m_topicRouteTable.find(topic);
    if (it != m_topicRouteTable.end())
    {
        return it->second;
    }

    TopicRouteData data;
    return data;
}

MQClientAPIImpl* MQClientFactory::getMQClientAPIImpl()
{
    return m_pMQClientAPIImpl;
}

MQAdminImpl* MQClientFactory::getMQAdminImpl()
{
    return m_pMQAdminImpl;
}

std::string MQClientFactory::getClientId()
{
    return m_clientId;
}

long long MQClientFactory::getBootTimestamp()
{
    return m_bootTimestamp;
}

PullMessageService* MQClientFactory::getPullMessageService()
{
    return m_pPullMessageService;
}


DefaultMQProducer* MQClientFactory::getDefaultMQProducer()
{
    return m_pDefaultMQProducer;
}

void MQClientFactory::sendHeartbeatToAllBroker()
{
    RMQ_DEBUG("sendHeartbeatToAllBroker begin");

    HeartbeatData heartbeatData;
    this->prepareHeartbeatData(heartbeatData);

    bool producerEmpty = heartbeatData.getProducerDataSet().empty();
    bool consumerEmpty = heartbeatData.getConsumerDataSet().empty();
    if (producerEmpty && consumerEmpty)
    {
        RMQ_ERROR("sending hearbeat, but no consumer and no producer");
        return;
    }

    RMQ_DEBUG("clientId=%s, m_brokerAddrTable=%u", heartbeatData.getClientID().c_str(), (unsigned)m_brokerAddrTable.size());

    kpr::ScopedRLock<kpr::RWMutex> lock(m_brokerAddrTableLock);
    std::map<std::string, std::map<int, std::string> >::iterator it = m_brokerAddrTable.begin();
    for (; it != m_brokerAddrTable.end(); it++)
    {
        std::map<int, std::string>::iterator it1 = it->second.begin();
        for (; it1 != it->second.end(); it1++)
        {
            std::string& addr = it1->second;
            if (!addr.empty())
            {
                if (consumerEmpty)
                {
                    if (it1->first != MixAll::MASTER_ID)
                    {
                        continue;
                    }
                }

                try
                {
                    m_pMQClientAPIImpl->sendHearbeat(addr, &heartbeatData, 3000);
                    RMQ_INFO("send heartbeat to broker[{%s} {%d} {%s}] success",
                    	it->first.c_str(), it1->first, addr.c_str());
                    RMQ_INFO("HeartbeatData %s", heartbeatData.toString().c_str());
                }
                catch (...)
                {
                    RMQ_ERROR("send heart beat to broker exception");
                }
            }
        }
    }

    RMQ_DEBUG("sendHeartbeatToAllBroker end");
}

void MQClientFactory::prepareHeartbeatData(HeartbeatData& heartbeatData)
{
    // clientID
    heartbeatData.setClientID(m_clientId);

    // Consumer
    {
    	kpr::ScopedRLock<kpr::RWMutex> lock(m_consumerTableLock);
        std::map<std::string, MQConsumerInner*>::iterator it = m_consumerTable.begin();
        for (; it != m_consumerTable.end(); it++)
        {
            MQConsumerInner* inner = it->second;
            if (inner)
            {
                ConsumerData consumerData;
                consumerData.groupName = inner->groupName();
                consumerData.consumeType = inner->consumeType();
                consumerData.messageModel = inner->messageModel();
                consumerData.consumeFromWhere = inner->consumeFromWhere();
                consumerData.subscriptionDataSet = inner->subscriptions();

                heartbeatData.getConsumerDataSet().insert(consumerData);
            }
        }
    }

    // Producer
    {
    	kpr::ScopedRLock<kpr::RWMutex> lock(m_producerTableLock);
        std::map<std::string, MQProducerInner*>::iterator it = m_producerTable.begin();
        for (; it != m_producerTable.end(); it++)
        {
            MQProducerInner* inner = it->second;
            if (inner)
            {
                ProducerData producerData;
                producerData.groupName = (it->first);

                heartbeatData.getProducerDataSet().insert(producerData);
            }
        }
    }

    return;
}

void MQClientFactory::makesureInstanceNameIsOnly(const std::string& instanceName)
{
    //TODO
}


void MQClientFactory::fetchNameServerAddr()
{
    //1000 * 10, 1000 * 60 * 2
    try
    {
        RMQ_DEBUG("Task: fetchNameServerAddr");
        m_pMQClientAPIImpl->fetchNameServerAddr();
    }
    catch (...)
    {
    	RMQ_ERROR("Task: fetchNameServerAddr exception");
    }
}

void MQClientFactory::updateTopicRouteInfoFromNameServerTask()
{
    //10, 1000 * 30, m_clientConfig.getPollNameServerInteval()
    try
    {
        RMQ_DEBUG("Task: updateTopicRouteInfoFromNameServerTask");
        updateTopicRouteInfoFromNameServer();
    }
    catch (...)
    {
		RMQ_ERROR("Task: fetchNameServerAddr exception");
    }
}

void MQClientFactory::cleanBroker()
{
    //1000, 1000 * 30, m_clientConfig.getHeartbeatBrokerInterval()
    try
    {
        RMQ_DEBUG("Task: cleanBroker");
        cleanOfflineBroker();
        sendHeartbeatToAllBrokerWithLock();
    }
    catch (...)
    {
		RMQ_ERROR("Task: cleanBroker exception");
    }
}

void MQClientFactory::persistAllConsumerOffsetTask()
{
    //1000 * 10, 1000 * 5, m_clientConfig.getPersistConsumerOffsetInterval()
    try
    {
        RMQ_DEBUG("Task: persistAllConsumerOffsetTask");
        persistAllConsumerOffset();
    }
    catch (...)
    {
		RMQ_ERROR("Task: persistAllConsumerOffsetTask exception");
    }
}

void MQClientFactory::recordSnapshotPeriodicallyTask()
{
    // 1000 * 10, 1000,
    try
    {
        //RMQ_DEBUG("Task: recordSnapshotPeriodicallyTask");
        recordSnapshotPeriodically();
    }
    catch (...)
    {
		RMQ_ERROR("Task: recordSnapshotPeriodically exception");
    }
}

void MQClientFactory::logStatsPeriodicallyTask()
{
    //  1000 * 10, 1000 * 60
    try
    {
        RMQ_DEBUG("Task: logStatsPeriodicallyTask");
        logStatsPeriodically();
    }
    catch (...)
    {
		RMQ_ERROR("Task: logStatsPeriodicallyTask exception");
    }
}

void MQClientFactory::startScheduledTask()
{
    m_scheduledTaskIds[0] = m_timerTaskManager.RegisterTimer(1000 * 10, 1000 * 60 * 2, m_pFetchNameServerAddrTask);

    m_scheduledTaskIds[1] = m_timerTaskManager.RegisterTimer(10, m_clientConfig.getPollNameServerInterval(), m_pUpdateTopicRouteInfoFromNameServerTask);

    m_scheduledTaskIds[2] = m_timerTaskManager.RegisterTimer(1000, m_clientConfig.getHeartbeatBrokerInterval(), m_pCleanBrokerTask);

    m_scheduledTaskIds[3] = m_timerTaskManager.RegisterTimer(1000 * 10, m_clientConfig.getPersistConsumerOffsetInterval(), m_pPersistAllConsumerOffsetTask);

    m_scheduledTaskIds[4] = m_timerTaskManager.RegisterTimer(1000 * 10, 1000, m_pRecordSnapshotPeriodicallyTask);
    m_scheduledTaskIds[5] = m_timerTaskManager.RegisterTimer(1000 * 10, 1000 * 60, m_pLogStatsPeriodicallyTask);
}

void MQClientFactory::cleanOfflineBroker()
{
    RMQ_DEBUG("TryLock m_lockNamesrv: 0x%p", &m_lockNamesrv);
    if (m_lockNamesrv.TryLock(MQClientFactory::LockTimeoutMillis))
    {
        RMQ_DEBUG("TryLock m_lockNamesrv ok");
        std::map<std::string, std::map<int, std::string> > updatedTable;
        {
            kpr::ScopedRLock<kpr::RWMutex> lock(m_brokerAddrTableLock);
            std::map<std::string, std::map<int, std::string> >::iterator it = m_brokerAddrTable.begin();

            for (; it != m_brokerAddrTable.end(); it++)
            {
                std::map<int, std::string> cloneTable = it->second;

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

                for (; it1 != cloneTable.end();)
                {
                    std::string& addr = it1->second;
                    if (!isBrokerAddrExistInTopicRouteTable(addr))
                    {
                        std::map<int, std::string>::iterator itTmp = it1;
                        it1++;
                        cloneTable.erase(itTmp);
                        continue;
                    }

                    it1++;
                }

                if (!cloneTable.empty())
                {
                    updatedTable[it->first] = cloneTable;
                }
            }
        }

        {
            kpr::ScopedWLock<kpr::RWMutex> lock(m_brokerAddrTableLock);
            m_brokerAddrTable.clear();
            m_brokerAddrTable = updatedTable;
        }

        m_lockNamesrv.Unlock();
        RMQ_DEBUG("UnLock m_lockNamesrv ok");
    }
    else
    {
        RMQ_DEBUG("TryLock m_lockNamesrv fail");
    }
}

bool MQClientFactory::isBrokerAddrExistInTopicRouteTable(const std::string& addr)
{
    kpr::ScopedRLock<kpr::RWMutex> lock(m_topicRouteTableLock);

    std::map<std::string, TopicRouteData>::iterator it = m_topicRouteTable.begin();
    for (; it != m_topicRouteTable.end(); it++)
    {
        const std::list<BrokerData>& brokers = it->second.getBrokerDatas();
        std::list<BrokerData>::const_iterator it1 = brokers.begin();

        for (; it1 != brokers.end(); it1++)
        {
            std::map<int, std::string>::const_iterator it2 = (*it1).brokerAddrs.begin();
            for (; it2 != (*it1).brokerAddrs.end(); it2++)
            {
                if (it2->second.find(addr) != std::string::npos)
                {
                    return true;
                }
            }
        }
    }

    return false;
}

void MQClientFactory::recordSnapshotPeriodically()
{
	kpr::ScopedRLock<kpr::RWMutex> lock(m_consumerTableLock);
    std::map<std::string, MQConsumerInner*>::iterator it = m_consumerTable.begin();
    for (; it != m_consumerTable.end(); it++)
    {
        MQConsumerInner* inner = it->second;
        if (inner)
        {
            DefaultMQPushConsumerImpl* consumer = dynamic_cast<DefaultMQPushConsumerImpl*>(inner);
            if (consumer)
            {
                consumer->getConsumerStatManager()->recordSnapshotPeriodically();
            }
        }
    }
}

void MQClientFactory::logStatsPeriodically()
{
	kpr::ScopedRLock<kpr::RWMutex> lock(m_consumerTableLock);
    std::map<std::string, MQConsumerInner*>::iterator it = m_consumerTable.begin();
    for (; it != m_consumerTable.end(); it++)
    {
        MQConsumerInner* inner = it->second;
        if (inner)
        {
            DefaultMQPushConsumerImpl* consumer = dynamic_cast<DefaultMQPushConsumerImpl*>(inner);
            if (consumer)
            {
                std::string group = it->first;
                consumer->getConsumerStatManager()->logStatsPeriodically(group, m_clientId);
            }
        }
    }
}

void MQClientFactory::persistAllConsumerOffset()
{
    kpr::ScopedRLock<kpr::RWMutex> lock(m_consumerTableLock);
    RMQ_DEBUG("persistAllConsumerOffset, m_consumerTable.size=%u", (unsigned)m_consumerTable.size());
    std::map<std::string, MQConsumerInner*>::iterator it = m_consumerTable.begin();
    for (; it != m_consumerTable.end(); it++)
    {
        MQConsumerInner* inner = it->second;
        if (inner)
        {
            inner->persistConsumerOffset();
        }
    }
}

bool MQClientFactory::topicRouteDataIsChange(TopicRouteData& olddata, TopicRouteData& nowdata)
{
    TopicRouteData old = olddata;
    TopicRouteData now = nowdata;

    old.getQueueDatas().sort();
    old.getBrokerDatas().sort();
    now.getQueueDatas().sort();
    now.getBrokerDatas().sort();

    return !(old == now);

}

bool MQClientFactory::isNeedUpdateTopicRouteInfo(const std::string& topic)
{
    bool result = false;
    {
    	kpr::ScopedRLock<kpr::RWMutex> lock(m_producerTableLock);
        std::map<std::string, MQProducerInner*>::iterator it = m_producerTable.begin();
        for (; it != m_producerTable.end(); it++)
        {
            MQProducerInner* inner = it->second;
            if (inner)
            {
                result = inner->isPublishTopicNeedUpdate(topic);
            }
        }
    }

    {
    	kpr::ScopedRLock<kpr::RWMutex> lock(m_consumerTableLock);
        std::map<std::string, MQConsumerInner*>::iterator it = m_consumerTable.begin();
        for (; it != m_consumerTable.end(); it++)
        {
            MQConsumerInner* inner = it->second;
            if (inner)
            {
                result = inner->isSubscribeTopicNeedUpdate(topic);
            }
        }
    }

    return result;
}

void MQClientFactory::unregisterClientWithLock(const std::string& producerGroup, const std::string& consumerGroup)
{
    RMQ_DEBUG("TryLock m_lockHeartbeat: 0x%p", &m_lockHeartbeat);
    if (m_lockHeartbeat.TryLock(MQClientFactory::LockTimeoutMillis))
    {
        try
        {
            RMQ_DEBUG("TryLock m_lockHeartbeat ok");
            unregisterClient(producerGroup, consumerGroup);
        }
        catch (...)
        {
            RMQ_ERROR("unregisterClientWithLock exception, %s %s",
                      producerGroup.c_str(), consumerGroup.c_str());
        }
        m_lockHeartbeat.Unlock();
        RMQ_DEBUG("Unlock m_lockHeartbeat ok");
    }
    else
    {
        RMQ_WARN("TryLock m_lockHeartbeat fail");
    }
}

void MQClientFactory::unregisterClient(const std::string& producerGroup, const std::string& consumerGroup)
{
    kpr::ScopedRLock<kpr::RWMutex> lock(m_brokerAddrTableLock);
    std::map<std::string, std::map<int, std::string> >::iterator it = m_brokerAddrTable.begin();
    for (; it != m_brokerAddrTable.end(); it++)
    {
        std::map<int, std::string>::iterator it1 = it->second.begin();

        for (; it1 != it->second.end(); it1++)
        {
            std::string& addr = it1->second;

            if (!addr.empty())
            {
                try
                {
                    m_pMQClientAPIImpl->unregisterClient(addr, m_clientId, producerGroup,
                                                         consumerGroup, 3000);
                }
                catch (...)
                {
                    RMQ_ERROR("unregister client exception from broker: %s", addr.c_str());
                }
            }
        }
    }
}

}