text:
#include "MQClientFactory.h"

#include <math.h>
#include <set>
#include <string>
#include <vector>
#include <algorithm>

#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"

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_pRemoteClientConfig, m_pClientRemotingProcessor);

	if (!m_clientConfig.getNamesrvAddr().empty())
	{
		m_pMQClientAPIImpl->updateNameServerAddressList(m_clientConfig.getNamesrvAddr());
	}

	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 = GetCurrentTimeMillis();

	m_pFetchNameServerAddr = new ScheduledTask(this,&MQClientFactory::fetchNameServerAddr);
	m_pUpdateTopicRouteInfoFromNameServerTask =
		new ScheduledTask(this,&MQClientFactory::updateTopicRouteInfoFromNameServerTask);
	m_pCleanBroker = 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;
}

MQClientFactory::~MQClientFactory()
{
	delete m_pRemoteClientConfig;
	delete m_pClientRemotingProcessor;
	delete m_pMQClientAPIImpl;
	delete m_pMQAdminImpl;
	delete m_pPullMessageService;
	delete m_pRebalanceService;
	delete m_pDefaultMQProducer;
	delete m_pFetchNameServerAddr;
	delete m_pUpdateTopicRouteInfoFromNameServerTask;
	delete m_pCleanBroker;
	delete m_pPersistAllConsumerOffsetTask;
	delete m_pRecordSnapshotPeriodicallyTask;
	delete m_pLogStatsPeriodicallyTask;
}

void MQClientFactory::start()
{
	Logger::get_logger()->info("Starting MQClientFactory");
	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);
		m_serviceState = RUNNING;
		Logger::get_logger()->info("MQClientFactory started");
		break;
	case RUNNING:
		Logger::get_logger()->warn("MQClientFactory is already running.");
		break;
	case SHUTDOWN_ALREADY:
		Logger::get_logger()->error("MQClientFactory should have already been shutted down");
		break;
	case START_FAILED:
		Logger::get_logger()->error("MQClientFactory started failed.");
		THROW_MQEXCEPTION(MQClientException,"The Factory object start failed",-1);
	default:
		break;
	}
}

void MQClientFactory::sendHeartbeatToAllBrokerWithLock()
{
	if (m_lockHeartbeat.TryLock())
	{
		try
		{
			sendHeartbeatToAllBroker();
		}
		catch (...)
		{

		}
		m_lockHeartbeat.Unlock();
	}
	else
	{

	}
}

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

	// Consumer¶ÔÏó
	{
		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
	{
		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)
{
	try
	{
		if (m_lockNamesrv.TryLock())
		{
			TopicRouteData* topicRouteData = NULL;
			try
			{

				if (isDefault && pDefaultMQProducer != NULL)
				{
					topicRouteData =
						m_pMQClientAPIImpl->getDefaultTopicRouteInfoFromNameServer(
							pDefaultMQProducer->getCreateTopicKey(), 1000 * 3);
					if (topicRouteData != 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 != NULL)
				{
					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);
						}
						else
						{
							//TODO log?
						}
					}
					else
					{
						changed=true;
					}

					if (changed)
					{
						// ºóÃæÅÅÐò»áÓ°ÏìÏ´εÄequalÂß¼Åжϣ¬ËùÒÔÏÈcloneÒ»·Ý
						TopicRouteData cloneTopicRouteData = *topicRouteData;

						// ¸üÐÂBrokerµØÖ·ÐÅÏ¢
						std::list<BrokerData> dataList = topicRouteData->getBrokerDatas();

						{
							kpr::ScopedLock<kpr::Mutex> lock(m_brokerAddrTableLock);
							std::list<BrokerData>::iterator it= dataList.begin();

							for(; it!=dataList.end(); it++)
							{
								m_brokerAddrTable[(*it).brokerName]=(*it).brokerAddrs;
							}
						}

						// ¸üз¢²¼¶ÓÁÐÐÅÏ¢
						{
							TopicPublishInfo* publishInfo =
								topicRouteData2TopicPublishInfo(topic, *topicRouteData);
							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);
							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);
								}
							}
						}

						m_topicRouteTable[topic]= cloneTopicRouteData;
						m_lockNamesrv.Unlock();

						delete topicRouteData;
						return true;
					}
				}
				else
				{
					//TODO log?
				}
			}
			catch (...)
			{
				//TODO log?
				if (topicRouteData != NULL)
				{
					delete topicRouteData;
				}
			}

			m_lockNamesrv.Unlock();
		}
		else
		{
			//TODO log?
		}
	}
	catch (...)
	{
		//TODO log?
	}

	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 = new MessageQueue(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))
			{
				// ÕâÀïÐèÒªÅжÏBrokerName¶ÔÓ¦µÄMasterÊÇ·ñ´æÔÚ£¬ÒòΪֻÄÜÏòMaster·¢ËÍÏûÏ¢
				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 = new MessageQueue(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++)
			{
				mqList->insert(MessageQueue(topic, qd.brokerName, i));
			}
		}
	}

	return mqList;
}

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

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

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

	{
		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.Close();

			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;
		}
	}
}

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

	if (m_consumerTable.find(group)!=m_consumerTable.end())
	{
		return false;
	}

	m_consumerTable[group] = pConsumer;

	return true;
}

void MQClientFactory::unregisterConsumer(const std::string& group)
{
	m_consumerTable.erase(group);
	unregisterClientWithLock("", group);
}

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

	if (m_producerTable.find(group)!=m_producerTable.end())
	{
		return false;
	}

	m_producerTable[group] = pProducer;

	return true;
}

void MQClientFactory::unregisterProducer(const std::string& group)
{
	m_producerTable.erase(group);
	unregisterClientWithLock(group, "");
}

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

	if (m_adminExtTable.find(group)!=m_adminExtTable.end())
	{
		return false;
	}

	m_adminExtTable[group] = pAdmin;

	return true;
}

void MQClientFactory::unregisterAdminExt(const std::string& group)
{
	m_adminExtTable.erase(group);
}

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

void MQClientFactory::doRebalance()
{
	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 (...)
			{

			}
		}
	}
}

MQProducerInner* MQClientFactory::selectProducer(const std::string& group)
{
	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)
{
	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)
{
	FindBrokerResult result;
	std::string brokerAddr;
	bool slave = false;
	bool found = false;
	kpr::ScopedLock<kpr::Mutex> lock(m_brokerAddrTableLock);
	std::map<std::string, std::map<int, std::string> >::iterator it = m_brokerAddrTable.find(brokerName);

	if (it!=m_brokerAddrTable.end())
	{
		// TODO Èç¹ûÓжà¸öSlave£¬¿ÉÄÜ»áÿ´Î¶¼Ñ¡ÖÐÏàͬµÄSlave£¬ÕâÀïÐèÒªÓÅ»¯
		std::map<int, std::string>::iterator it1 = it->second.begin();
		for (;it1!=it->second.end();it1++)
		{
			int id = it1->first;
			brokerAddr = it1->second;
			if (!brokerAddr.empty())
			{
				found = true;
				if (MixAll::MASTER_ID == id)
				{
					slave = false;
				}
				else
				{
					slave = true;
				}

				break;
			}
		}
	}

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

	return result;
}

std::string MQClientFactory::findBrokerAddressInPublish(const std::string& brokerName)
{
	kpr::ScopedLock<kpr::Mutex> 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::ScopedLock<kpr::Mutex> 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 (...)
		{

		}
	}

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

	return ids;
}

std::string MQClientFactory::findBrokerAddrByTopic(const std::string& topic)
{
	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)
{
	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()
{
	HeartbeatData* heartbeatData = prepareHeartbeatData();
	bool producerEmpty = heartbeatData->getProducerDataSet().empty();
	bool consumerEmpty = heartbeatData->getConsumerDataSet().empty();
	if (producerEmpty && consumerEmpty)
	{
		return;
	}

	std::vector<std::map<int, std::string> > addrlist;
	{
		kpr::ScopedLock<kpr::Mutex> lock(m_brokerAddrTableLock);
		std::map<std::string, std::map<int, std::string> >::iterator it = m_brokerAddrTable.begin();

		for (; it!=m_brokerAddrTable.end(); it++)
		{
			addrlist.push_back(it->second);
		}
	}

	std::vector<std::map<int, std::string> >::iterator it = addrlist.begin();

	for (; it!=addrlist.end(); it++)
	{
		std::map<int, std::string> addrs = *it;
		std::map<int, std::string>::iterator it1 = addrs.begin();
		for (; it1!=addrs.end(); it1++)
		{
			std::string& addr = it1->second;
			if (!addr.empty())
			{
				// ˵Ã÷Ö»ÓÐProducer£¬Ôò²»ÏòSlave·¢ÐÄÌø
				if (consumerEmpty)
				{
					if (it1->first != MixAll::MASTER_ID)
						continue;
				}

				try
				{
					m_pMQClientAPIImpl->sendHearbeat(addr, heartbeatData, 3000);
				}
				catch (...)
				{
				}
			}
		}
	}
}

HeartbeatData* MQClientFactory::prepareHeartbeatData()
{
	HeartbeatData* heartbeatData = new HeartbeatData();

	// clientID
	heartbeatData->setClientID(m_clientId);

	// Consumer
	{
		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
	{
		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 heartbeatData;
}

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

void MQClientFactory::fetchNameServerAddr()
{
	if (m_clientConfig.getNamesrvAddr().empty())
	{
		//1000 * 10, 1000 * 60 * 2
		try
		{
			m_pMQClientAPIImpl->fetchNameServerAddr();
		}
		catch (...)
		{
		}
	}
}

void MQClientFactory::updateTopicRouteInfoFromNameServerTask()
{
	//10, m_clientConfig.getPollNameServerInterval()
	try
	{
		updateTopicRouteInfoFromNameServer();
	}
	catch (...)
	{

	}
}

void MQClientFactory::cleanBroker()
{
	//1000, m_clientConfig.getHeartbeatBrokerInterval()
	try
	{
		cleanOfflineBroker();
		sendHeartbeatToAllBrokerWithLock();
	}
	catch (...)
	{

	}
}

void MQClientFactory::persistAllConsumerOffsetTask()
{
	//1000 * 10, m_clientConfig.getPersistConsumerOffsetInterval()
	try {
		Logger::get_logger()->info("Begin to persist consumer offsets.");
		persistAllConsumerOffset();
		Logger::get_logger()->info("All consumer offsets persisted.");
	} catch (...) {
		Logger::get_logger()->error("Failed to persist consumer offsets");
	}
}

void MQClientFactory::recordSnapshotPeriodicallyTask()
{
	// 1000 * 10, 1000,
	try
	{
		recordSnapshotPeriodically();
	}
	catch (...)
	{

	}
}

void MQClientFactory::logStatsPeriodicallyTask()
{
	//  1000 * 10, 1000 * 60
	try
	{
		logStatsPeriodically();
	}
	catch (...)
	{

	}
}

void MQClientFactory::startScheduledTask()
{
	// ¶¨Ê±»ñÈ¡Name ServerµØÖ·
	m_scheduledTaskIds[0] = m_timerTaskManager.RegisterTimer(1000 * 10, 1000 * 60 * 2,m_pFetchNameServerAddr);

	// ¶¨Ê±´ÓName Server»ñÈ¡Topic·ÓÉÐÅÏ¢
	m_scheduledTaskIds[1] = m_timerTaskManager.RegisterTimer(10, m_clientConfig.getPollNameServerInterval(), m_pUpdateTopicRouteInfoFromNameServerTask);

	// ¶¨Ê±ÇåÀíÏÂÏßµÄBroker
	// ÏòËùÓÐBroker·¢ËÍÐÄÌøÐÅÏ¢£¨°üº¬¶©ÔĹØϵµÈ£©
	m_scheduledTaskIds[2] = m_timerTaskManager.RegisterTimer(1000, m_clientConfig.getHeartbeatBrokerInterval(),m_pCleanBroker);

	// ¶¨Ê±³Ö¾Ã»¯ConsumerÏû·Ñ½ø¶È£¨¹ã²¥´æ´¢µ½±¾µØ£¬¼¯Èº´æ´¢µ½Broker£©
	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()
{
	std::map<std::string, std::map<int, std::string> > updatedTable;
	std::map<std::string, std::map<int, std::string> > tmpTable;

	{
		kpr::ScopedLock<kpr::Mutex> lock(m_brokerAddrTableLock);
		tmpTable = m_brokerAddrTable;
	}

	std::map<std::string, std::map<int, std::string> >::iterator it = tmpTable.begin();

	for (; it!=tmpTable.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::ScopedLock<kpr::Mutex> lock(m_brokerAddrTableLock);
		m_brokerAddrTable.clear();
		m_brokerAddrTable = updatedTable;
	}
}

bool MQClientFactory::isBrokerAddrExistInTopicRouteTable(const std::string& addr)
{
	std::map<std::string, TopicRouteData>::iterator it = m_topicRouteTable.begin();
	bool changed = false;

	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()
{
	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()
{
	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()
{
	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;
	// ²é¿´·¢²¼¶ÓÁÐÊÇ·ñÐèÒª¸üÐÂ
	{
		std::map<std::string, MQProducerInner*>::iterator it = m_producerTable.begin();

		for (; it!=m_producerTable.end(); it++)
		{
			MQProducerInner* inner = it->second;
			if (inner)
			{
				inner->isPublishTopicNeedUpdate(topic);
			}
		}
	}

	// ²é¿´¶©ÔĶÓÁÐÊÇ·ñÐèÒª¸üÐÂ
	{
		std::map<std::string, MQConsumerInner*>::iterator it = m_consumerTable.begin();

		for (; it!=m_consumerTable.end(); it++)
		{
			MQConsumerInner* inner = it->second;
			if (inner)
			{
				inner->isSubscribeTopicNeedUpdate(topic);
			}
		}
	}

	return result;
}

void MQClientFactory::unregisterClientWithLock(const std::string& producerGroup, const std::string& consumerGroup)
{
	try
	{
		if (m_lockHeartbeat.TryLock())
		{
			try
			{
				unregisterClient(producerGroup, consumerGroup);
				m_lockHeartbeat.Unlock();
			}
			catch (...)
			{
				m_lockHeartbeat.Unlock();
			}
		}
		else
		{
		}
	}
	catch (...)
	{

	}
}

void MQClientFactory::unregisterClient(const std::string& producerGroup, const std::string& consumerGroup)
{
	std::vector<std::map<int, std::string> > addrlist;
	{
		kpr::ScopedLock<kpr::Mutex> lock(m_brokerAddrTableLock);
		std::map<std::string, std::map<int, std::string> >::iterator it = m_brokerAddrTable.begin();

		for (; it!=m_brokerAddrTable.end(); it++)
		{
			addrlist.push_back(it->second);
		}
	}

	std::vector<std::map<int, std::string> >::iterator it = addrlist.begin();

	for (; it!=addrlist.end(); it++)
	{
		std::map<int, std::string>::iterator it1 = (*it).begin();

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

			if (!addr.empty())
			{
				try
				{
					m_pMQClientAPIImpl->unregisterClient(addr, m_clientId, producerGroup,
														 consumerGroup, 3000);
				}
				catch (...)
				{

				}
			}
		}
	}
}


text:
// NO INCLUDE GUARDS, THE HEADER IS INTENDED FOR MULTIPLE INCLUSION

// Copyright Aleksey Gurtovoy 2000-2004
//
// Distributed under the Boost Software License, Version 1.0.
// (See accompanying file LICENSE_1_0.txt or copy at
// http://www.boost.org/LICENSE_1_0.txt)
//
// See http://www.boost.org/libs/mpl for documentation.

// $Source: /cvsroot/boost/boost/boost/mpl/aux_/shift_op.hpp,v $
// $Date: 2004/09/07 08:51:32 $
// $Revision: 1.4 $

#if !defined(BOOST_MPL_PREPROCESSING_MODE)
#   include <boost/mpl/integral_c.hpp>
#   include <boost/mpl/aux_/value_wknd.hpp>
#endif

#if !defined(AUX778076_OP_PREFIX)
#   define AUX778076_OP_PREFIX AUX778076_OP_NAME
#endif

#define AUX778076_OP_ARITY 2

#include <boost/mpl/aux_/numeric_op.hpp>
#include <boost/mpl/aux_/config/static_constant.hpp>
#include <boost/mpl/aux_/config/use_preprocessed.hpp>

#if !defined(BOOST_MPL_CFG_NO_PREPROCESSED_HEADERS) \
    && !defined(BOOST_MPL_PREPROCESSING_MODE)

#   define BOOST_MPL_PREPROCESSED_HEADER AUX778076_OP_PREFIX.hpp
#   include <boost/mpl/aux_/include_preprocessed.hpp>

#else

#   include <boost/mpl/aux_/config/integral.hpp>
#   include <boost/preprocessor/cat.hpp>

namespace boost { namespace mpl {

#if defined(BOOST_MPL_CFG_NO_NESTED_VALUE_ARITHMETIC)
namespace aux {
template< typename T, typename Shift, T n, Shift s >
struct BOOST_PP_CAT(AUX778076_OP_PREFIX,_wknd)
{
    BOOST_STATIC_CONSTANT(T, value = (n AUX778076_OP_TOKEN s));
    typedef integral_c<T,value> type;
};
}
#endif

template<>
struct AUX778076_OP_IMPL_NAME<integral_c_tag,integral_c_tag>
{
    template< typename N, typename S > struct apply
#if !defined(BOOST_MPL_CFG_NO_NESTED_VALUE_ARITHMETIC)
        : integral_c<
              typename N::value_type
            , ( BOOST_MPL_AUX_VALUE_WKND(N)::value
                  AUX778076_OP_TOKEN BOOST_MPL_AUX_VALUE_WKND(S)::value
                )
            >
#else
        : aux::BOOST_PP_CAT(AUX778076_OP_PREFIX,_wknd)<
              typename N::value_type
            , typename S::value_type
            , N::value
            , S::value
            >::type
#endif
    {
    };
};

}}

#endif // BOOST_MPL_CFG_NO_PREPROCESSED_HEADERS

#undef AUX778076_OP_TAG_NAME
#undef AUX778076_OP_IMPL_NAME
#undef AUX778076_OP_ARITY
#undef AUX778076_OP_PREFIX
#undef AUX778076_OP_NAME
#undef AUX778076_OP_TOKEN

redpajama_set_name: RedPajamaGithub


text:
Since I don't have a direct contact at [Etsy](http://codeascraft.etsy.com/)'s security team (apart from security-reports@etsy.com), here is the question I would like to ask them (which hopefully will reach the right person).

- - -

_Dear Etsy security team, _

**_How can OWASP help?_**

_By Owasp, I mean OWASP Community (it's projects, chapters, people, ideas, activities, energy)._

_From the information posted on your website and presented at conferences, you really take security seriously. _

_You have been able to create a productive environment where secure code 'happens', and more importantly, there is a productive and pragmatic relationship between you (the security team), your developers and your management._

_So, assuming that you still have a couple things you would like to do better, is there a way (or place, or activity) where OWASP's community can help?_

  * _Maybe it is in creating better documentation or education materials for your developers/testers?_
  * _Maybe its an improved schema for AppSensor that would allow your multiple teams to create even better data (or metadata) for your amazing graphs?_
  * _Maybe it is a an special Summit on an topic that you care about? (see the amazing talent that we were able to gather in our last one)_
  * _Maybe is better SAST or DAST rules for your tools?_
  * _Maybe is better technical (and security focused) information on how Frameworks work and its security implications? (which will help with code reviews and code standards)_
  * _Maybe its a working group on CSP (Content Security Policies) to share best-practices and ideas on how to implement them? (with the key players from the browser vendors participating)_
  * _Maybe creating a series of events (or even a tour) around OWASP chapters and conferences where you can present your latest ideas and challenges? (the format is up to your imagination and availability) _
  * _Maybe its better connectors, parsers or data-transformations for the data you collect using StastD?_
  * _....fell free to propose your own (these are just ideas to kickstart the dialogue)_

_The idea is to start a collaboration with you._

_There is a lot that OWASP can learn from what you are doing, and the more we are able to capture it, the more we can help others who also want to protect their customers, business and applications._

_Thanks for your time_

_Dinis Cruz_

_Owasp Contributor_

- - -

_**Related Etsy posts:**

  * [Etsy.com - A case study on how to do security right?](http://diniscruz.blogspot.co.uk/2012/10/etsycom-case-study-on-how-to-do.html)
  * [Amazing presentation on integrating security into the SDL](http://diniscruz.blogspot.co.uk/2012/10/amazing-presentation-on-integrating.html)
  * [Measure Anything, Measure Everything, AppSensor and Simple Graphite Hosting](http://diniscruz.blogspot.co.uk/2012/10/measure-anything-measure-everything.html)


text:
<?php

namespace Nours\RestAdminBundle\Loader;

use Nours\RestAdminBundle\Domain\ResourceCollection;
use RecursiveDirectoryIterator;
use RecursiveIteratorIterator;
use Symfony\Component\Config\FileLocatorInterface;
use Symfony\Component\Config\Loader\Loader;
use Symfony\Component\Config\Resource\DirectoryResource;

/**
 * Load resources from controller directories
 *
 * @author David Coudrier <david.coudrier@gmail.com>
 */
class AnnotationDirectoryLoader extends Loader
{
    /**
     * @var AnnotationFileLoader
     */
    private $loader;
    /**
     * @var FileLocatorInterface
     */
    private $locator;

    public function __construct(
        FileLocatorInterface $locator,
        AnnotationFileLoader $loader
    ) {
        parent::__construct();

        $this->locator = $locator;
        $this->loader = $loader;
    }

    /**
     * {@inheritdoc}
     */
    public function load($resource, $type = null): ResourceCollection
    {
        $path = $this->locator->locate($resource);

        $resources = new ResourceCollection();
        $resources->addConfigResource(new DirectoryResource($path));

        $iterator = new RecursiveIteratorIterator(new RecursiveDirectoryIterator($path));
//        $iterator = new \DirectoryIterator($path);
        foreach ($iterator as $file) {
            if (!$file->isFile() || $file->getExtension() != 'php' || $file->isDir()) {
                continue;
            }

            $import = $this->loader->load($file->getRealPath());
            $resources->merge($import);
        }

        return $resources;
    }

    /**
     * {@inheritdoc}
     */
    public function supports($resource, string $type = null): bool
    {
        try {
            $path = $this->locator->locate($resource);
        } catch (\Exception $e) {
            return false;
        }

        return is_string($resource) && is_dir($path) && (!$type || 'annotation' === $type);
    }
}