myvtm/Module/mod_chromium/EntitySessionManager.cpp

#include "stdafx.h"
#include "MessageType.h"
#include "EntitySessionManager.h"
#include "baseEx.h"
#include<ctime>
#include <string>
#include "../../Other/libpublicFun/publicFunExport.h"
#include "boost/lexical_cast.hpp"

namespace Chromium {

	static unsigned int m_global_transid = 0x0F000001;	//用于替换来自web的transId,用来和框架进行交互
	static unsigned int m_createSession_transId = 0x00F00001;	//用于建立本地session,为避免创建的sessionId重复

	auto createMapFun = []()->std::map<std::string, int> {
		std::map<std::string, int> nullMap;
		return nullMap;
	};

	std::map<std::string, int> EntitySessionManager::m_session_map = createMapFun();

	EntitySessionManager::EntitySessionManager() {
		m_global_transid = 0x0F000001;
		m_createSession_transId = 0x00F00001;
	}

	EntitySessionManager::~EntitySessionManager() {}

	std::pair<bool, int> EntitySessionManager::GetSessionIDByEntityName(std::string entityname) {
		auto it = m_session_map.find(entityname);
		if (it == m_session_map.end())
			return std::make_pair(false, -1);	// 发出 StartSession request
		return std::make_pair(true, it->second);
	}


	int EntitySessionManager::MakeNewTransID(CMessage* msg, unsigned int hdlID) {
		WSClientReqInfo info;
		info.transID = msg->getTransID();
		info.hdlID = hdlID;
		info.next = NULL;
		m_trans_map.insert(std::pair<int, WSClientReqInfo>(m_global_transid, info));
		return m_global_transid++;
	}

	std::pair<bool, int> EntitySessionManager::getSrcTransID(int newTransID)
	{
		auto tmp = m_trans_map.find(newTransID);
		if (tmp != m_trans_map.end())
			return std::make_pair(true, tmp->second.transID);
		else
		{
			if ((newTransID & 0xF00000) == false)
				DbgWithLink(LOG_LEVEL_INFO, LOG_TYPE_SYSTEM).setAPI(__FUNCTION__)("can not find transId:%d, may be fault", newTransID);
			//DbgEx("can not find transId:%d, may be true:%s", newTransID, newTransID & 0xF00000 ? "true" : "false");
			return std::make_pair(false, 0);
		}
	}

	void EntitySessionManager::StoreSessionReq(int transID, std::string entityName) {
		if(entityName.length() > 0)
			m_session_ack_map.insert(std::pair<int, std::string>(transID, entityName));
	}

	std::vector<std::string> EntitySessionManager::queryUnLinkSession()
	{
		std::vector<std::string> ret;
#if (defined _WIN32 || defined _WIN64)
		for each (auto it in m_session_map)
#else
		for (auto it : m_session_map)
#endif
		{
			if (it.second == -1)
				ret.emplace_back(it.first);
		}
		return ret;
	}

	std::map<std::string, int> EntitySessionManager::queryAllSessionInfo()
	{
		return m_session_map;
	}

	WSClientReqInfo* EntitySessionManager::ReduceOriginTransID(CMessage* msg) {//获取transId对应的连接
		int tansId = msg->getTransID();
		std::map<int, WSClientReqInfo>::iterator it = m_trans_map.find(tansId);
		if (it == m_trans_map.end())
			return NULL;
		return &(it->second);
	}

	void EntitySessionManager::RemoveAckListNode(int transid) {
		std::map<int, WSClientReqInfo>::iterator it = m_trans_map.find(transid);
		if (it == m_trans_map.end())
			return;
		m_trans_map.erase(it);
		return;
	}

	RequestProcessType EntitySessionManager::RequestProcess(CMessage* msg, std::string entityName, unsigned int hdlID) {
		if (msg == NULL)
			return PROCESS_NOTHING;
		RequestProcessType ret = PROCESS_NOTHING;

		switch (msg->getMessageType() & 0x0000FFFF) {
		case Info:
			ret = ProcessWithInfo(msg, entityName, hdlID);
			break;
		case BeginSession:
			ret = ProcessWithBeginSession(msg, entityName, hdlID);
			break;
		case EndSession:
			ret = ProcessWithEndSession(msg, entityName, hdlID);
			break;
		case Request:
			ret = ProcessWithRequest(msg, entityName, hdlID);
			break;
		case Register:
			ret = ProcessWithRegister(msg, entityName, hdlID);
			break;
		case Unregister:
			ret = ProcessWithUnregister(msg, entityName, hdlID);
			break;
		case Event:
			ret = PROCESS_SEND;
			break;
		case LogEventMsgType:
			ret = PROCESS_SEND;
			break;
		case LogWarnMsgType:
			return PROCESS_SEND;
		case SetVarReq:
			ret = ProcessWithSetVarReq(msg, entityName, hdlID);
			break;
		case GetVarReq:
			ret = ProcessWithGetVarReq(msg, entityName, hdlID);
			break;
		case Broadcast:
			return PROCESS_NOTHING;
		default:
			DbgWithLink(LOG_LEVEL_INFO, LOG_TYPE_SYSTEM).setAPI(__FUNCTION__)("EntitySessionManager::RequestProcess switch default!");
			ret = PROCESS_NOTHING;
		}
		int srcTransId = msg->getTransID();
		StoreSessionReq(srcTransId, entityName);
		if (Request == (msg->getMessageType() & 0x0000FFFF))
			msg->exchangeSessionIdAndTransId();

		return ret;
	}

	RequestProcessType EntitySessionManager::ProcessWithInfo(CMessage* msg, std::string entityName, unsigned int hdlID) {
		auto it = m_session_map.find(entityName);
		if (it == m_session_map.end() || -1 == it->second)
		{
			DbgWithLink(LOG_LEVEL_INFO, LOG_TYPE_SYSTEM).setAPI(__FUNCTION__)("no session found for entity : %s", UtfToGbk(entityName.c_str()).c_str());
			return PROCESS_STARTSESSION;
		}
		msg->setSessionID(it->second, false);
		DbgWithLink(LOG_LEVEL_DEBUG, LOG_TYPE_SYSTEM).setAPI(__FUNCTION__)("EntitySessionManager::ProcessWithInfo, replace sessionID:%d", it->second);
		return PROCESS_SEND;
	}

	void EntitySessionManager::updateBeginSessionTime(std::string entityName, bool doCleanSessionTime)
	{
		time_t now_time = doCleanSessionTime ? 0 : time(NULL);	//doCleanSessionTime 会重置记录时间为0,5s内重复发起的beginSession能被接受
		auto curSession = t_BeginSessionTime.find(entityName);
		if (curSession == t_BeginSessionTime.end())
			t_BeginSessionTime.insert(std::make_pair(entityName, now_time));
		else
			curSession->second = now_time;
		//DbgEx("save BeginSession Time:%s, %d", entityName.c_str(), now_time);
	}

	bool EntitySessionManager::checkBeginSession(std::string entityName)
	{
		time_t now_time = time(NULL);
		auto curSession = t_BeginSessionTime.find(entityName);
		if (curSession == t_BeginSessionTime.end() || curSession->second < now_time - 5)
		{
			DbgWithLink(LOG_LEVEL_DEBUG, LOG_TYPE_SYSTEM).setAPI(__FUNCTION__)("checkBeginSession record begin session msg, oldTime:%lld, newTime:%lld", curSession != t_BeginSessionTime.end() ? curSession->second : -1, now_time);
			return true;//5s内不重复发起session
		}
		else
			return false;
	}

	RequestProcessType EntitySessionManager::ProcessWithBeginSession(CMessage* msg, std::string entityName, unsigned int hdlID) {
		DbgWithLink(LOG_LEVEL_DEBUG, LOG_TYPE_SYSTEM).setAPI(__FUNCTION__)("EntitySessionManager::ProcessWithBeginSession");
		auto s = GetSessionIDByEntityName(entityName);
		if (s.first && -1 != s.second)
			return PROCESS_FINDSESSION;

		if (checkBeginSession(entityName))
		{
			// 换transID 发送beginsession
			int transID = MakeNewTransID(msg, hdlID);
			msg->setTransID(transID);
			updateBeginSessionTime(entityName);
			return PROCESS_SEND;
		}
		else
			return PROCESS_RECORDMSG;//5s内发起过session
	}

	RequestProcessType EntitySessionManager::ProcessWithEndSession(CMessage* msg, std::string entityName,
		unsigned int hdlID) {
		DbgWithLink(LOG_LEVEL_DEBUG, LOG_TYPE_SYSTEM).setAPI(__FUNCTION__)("EntitySessionManager::ProcessWithEndSession");
		return PROCESS_NOTHING;
	}

	RequestProcessType EntitySessionManager::ProcessWithRequest(CMessage* msg, std::string entityName,
		unsigned int hdlID) {
		
		auto it = m_session_map.find(entityName);
		if (it == m_session_map.end() || -1 == it->second)
		{
			DbgWithLink(LOG_LEVEL_INFO, LOG_TYPE_SYSTEM).setAPI(__FUNCTION__)("no session found for entity : %s", UtfToGbk(entityName.c_str()).c_str());
			return PROCESS_STARTSESSION;
		}
		msg->setSessionID(it->second, true);
		auto transId = MakeNewTransID(msg, hdlID);
		msg->setTransID(transId);
		DbgWithLink(LOG_LEVEL_DEBUG, LOG_TYPE_SYSTEM).setAPI(__FUNCTION__)("EntitySessionManager::ProcessWithRequest, replace sessionID:%d, replace transId:%d", it->second, transId);
		return PROCESS_SEND;
	}

	RequestProcessType EntitySessionManager::ProcessWithRegister(CMessage* msg, std::string entityName, unsigned int hdlID) {
		DbgWithLink(LOG_LEVEL_DEBUG, LOG_TYPE_SYSTEM).setAPI(__FUNCTION__)("EntitySessionManager::ProcessWithRegister");
		auto _broadcast_it = m_broadcast_map.find(entityName);
		if (_broadcast_it == m_broadcast_map.end())
		{
			// 尚未有客户端订阅过
			int transid = MakeNewTransID(msg, hdlID);
			msg->setTransID(transid);
			m_broadcast_map.insert(std::pair<std::string, int>(entityName, transid));
			return PROCESS_SEND;
		}

		std::map<int, WSClientReqInfo>::iterator _trans_it = m_trans_map.find(_broadcast_it->second);
		if (_trans_it == m_trans_map.end())
		{
			// 尚未有客户端订阅过
			int transid = MakeNewTransID(msg, hdlID);
			msg->setTransID(transid);
			m_broadcast_map.insert(std::pair<std::string, int>(entityName, transid));
			return PROCESS_SEND;
		}
		// 已经有过订阅 在链表后加
		WSClientReqInfo* pNode = &_trans_it->second;
		if (pNode->hdlID == hdlID)
		{
			// 该客户端订阅过，避免重复订阅
			return PROCESS_NOTHING;
		}
		while (pNode->next != NULL) {
			if (pNode->hdlID == hdlID)
			{
				// 该客户端订阅过，避免重复订阅
				return PROCESS_NOTHING;
			}
			pNode = pNode->next;
		}
		// 该客户端从未订阅过                                                             
		pNode->next = new WSClientReqInfo();
		pNode->next->next = NULL;
		pNode->next->hdlID = hdlID;
		pNode->next->transID = msg->getTransID();
		return PROCESS_NOTHING;
	}

	RequestProcessType EntitySessionManager::ProcessWithUnregister(CMessage* msg, std::string entityName,
		unsigned int hdlID) {
		DbgWithLink(LOG_LEVEL_DEBUG, LOG_TYPE_SYSTEM).setAPI(__FUNCTION__)("EntitySessionManager::ProcessWithUnregister");
		std::map<std::string, int>::iterator _broadcast_it = m_broadcast_map.find(entityName);
		if (_broadcast_it == m_broadcast_map.end())
		{
			// 没订阅过 不用反注册
			return PROCESS_NOTHING;
		}
		// 订阅过 找到对应的客户端节点 删掉
		std::map<int, WSClientReqInfo>::iterator _trans_it = m_trans_map.find(_broadcast_it->second);
		if (_trans_it == m_trans_map.end())
		{
			// 没找到对应transid 理论上不可能
			m_broadcast_map.erase(_broadcast_it);
			return PROCESS_NOTHING;
		}
		// 查找节点删掉
		WSClientReqInfo* pNode = &_trans_it->second;
		if (pNode->hdlID == hdlID)
		{
			// 找到了就删掉
			if (pNode->next == NULL)
			{
				// 只有一个节点 就全都删掉
				m_trans_map.erase(_trans_it);
				m_broadcast_map.erase(_broadcast_it);
				return PROCESS_SEND;
			}
			// 后面有其他节点
			WSClientReqInfo* tp = pNode->next;
			pNode->transID = tp->transID;
			pNode->hdlID = tp->hdlID;
			pNode->next = tp->next;
			free(tp);
			return PROCESS_NOTHING;
		}
		// 首个节点不匹配时
		WSClientReqInfo* pHead = pNode;
		pNode = pNode->next;
		while (pNode != NULL) {
			if (pNode->hdlID == hdlID)
			{
				pHead->next = pNode->next;
				free(pNode);
				return PROCESS_NOTHING;
			}
			pHead = pNode;
			pNode = pNode->next;
		}
		return PROCESS_NOTHING;
	}

	RequestProcessType EntitySessionManager::ProcessWithSetVarReq(CMessage* msg, std::string entityName,
		unsigned int hdlID) {
		DbgWithLink(LOG_LEVEL_INFO, LOG_TYPE_SYSTEM).setAPI(__FUNCTION__)("EntitySessionManager::ProcessWithSetVarReq");
		msg->setTransID(MakeNewTransID(msg, hdlID));
		return PROCESS_SEND;
	}

	RequestProcessType EntitySessionManager::ProcessWithGetVarReq(CMessage* msg, std::string entityName,
		unsigned int hdlID) {
		DbgWithLink(LOG_LEVEL_DEBUG, LOG_TYPE_SYSTEM).setAPI(__FUNCTION__)("EntitySessionManager::ProcessWithGetVarReq");
		msg->setTransID(MakeNewTransID(msg, hdlID));
		return PROCESS_SEND;
	}


	std::pair<int, std::string>  EntitySessionManager::GetStartSessionRequest(std::string entityname, std::string className) {
		auto curSession = m_createSession_transId++;
		std::string js = "{\"messageType\":1,\"transID\":" + std::to_string((LONGLONG)curSession) + ",\"functionName\":\"\",\"entity\":\"";
		js.append(entityname.append("\",\"class\":\""));
		js.append(className.append("\"}"));
		DbgWithLink(LOG_LEVEL_DEBUG, LOG_TYPE_SYSTEM).setAPI(__FUNCTION__)("start sessionRequest: %s", UtfToGbk(js.c_str()).c_str());
		return std::make_pair(curSession, js);
	}

	std::pair<int, std::string> EntitySessionManager::GetAllSessionRequest(int transId)
	{
		std::map<std::string, std::string> sessionInfo;
#if (defined _WIN32 || defined _WIN64)
		for each (auto it in m_session_map)
#else
		for (auto it : m_session_map)
#endif
		{
			if (it.second > 0)
				sessionInfo.insert(std::make_pair(it.first, boost::lexical_cast<std::string>(it.second)));
		}
		auto sessionJson = generateJsonStr(sessionInfo);
		if (sessionJson.first)
		{
			char sessionJs[10240] = "";
			sprintf(sessionJs, "{\"messageType\":%d,\"transID\":%d,\"InitCfg\":%d,\"%s\":%s}", 16, transId, g_hasInitCfg, PARAMLIST_HEAD, sessionJson.second.c_str());
			return std::make_pair(true, sessionJs);
		}
		return std::make_pair(false, "error generateJsonStr");
	}

	std::string EntitySessionManager::GetStartSessionAck(CMessage* msg, std::string entityname) {
		auto s = GetSessionIDByEntityName(entityname);
		std::string sessionId = std::to_string((LONGLONG)s.second);
		std::string transid = std::to_string((LONGLONG)msg->getTransID());
		std::string js = "{\"messageType\":5,\"errorCode\":\"\",\"errorMsg\":\"\",\"sessionID\":";
		js.append(sessionId.c_str());
		js.append(",\"transID\":");
		js.append(transid.c_str());
		js.append("}");
		return js;
	}

	std::string EntitySessionManager::UpdateSessionMap(int transid, int sessionId)
	{
		auto it = m_session_ack_map.find(transid);
		if (it == m_session_ack_map.end())
			return "";
		DoSessionUpdate(it->second, sessionId);
		return it->second;
	}

	EntitySessionManager& EntitySessionManager::DoSessionUpdate(std::string entityname, int session)
	{
		auto sessionIt = m_session_map.find(entityname);

		if(m_session_map.end() != sessionIt)
		{
			if (sessionIt->second == session)
				return *this;//相等不处理
			else if(sessionIt->second == -1 || session == -1)
			{
				DbgWithLink(LOG_LEVEL_INFO, LOG_TYPE_SYSTEM).setAPI(__FUNCTION__)("update sessionId(regenerate):<%s><%d>", entityname.c_str(), session);
				m_session_map.erase(sessionIt);
				m_session_map.insert(std::make_pair(entityname, session));
			}
			else
				DbgWithLink(LOG_LEVEL_INFO, LOG_TYPE_SYSTEM).setAPI(__FUNCTION__)("session Error(src:%d), do not update sessionId:<%s><%d>", sessionIt->second, entityname.c_str(), session);
		}
		else
			m_session_map.insert(std::make_pair(entityname, session));//not exist
		return *this;
	}

	AckProcessType EntitySessionManager::ProcessWithAck(CMessage* msg, unsigned int& hdlID) {
		int globalID = msg->getTransID();
		WSClientReqInfoStruct* p = ReduceOriginTransID(msg);
		if (NULL == p)
			return ACKPROCESS_NOTHING;// 异常 应该不会有这种情况出现

		hdlID = p->hdlID;
		UpdateSessionMap(globalID, msg->getSessionID());

		if (hdlID == 0) {
			return ACKPROCESS_NOTHING;
		}

		return ACKPROCESS_SEND;
	}

	std::pair<bool, std::string> EntitySessionManager::DoSessionRemove(unsigned int session)
	{
#if (defined _WIN32 || defined _WIN64)
		for each (auto it in m_session_map)
#else
		for (auto it : m_session_map)
#endif
		{
			if (it.second == session)
			{
				DoSessionUpdate(it.first, -1);
				return std::make_pair(true, it.first);
			}
		}
		return std::make_pair(false, "");
	}

	std::pair<AckProcessType, std::string> EntitySessionManager::AskProcessSession(CMessage* msg, unsigned int& hdlID) {
		int globalID = msg->getTransID();
		/*
		替换真实的transId
		*/

		if (globalID & 0xF00000)
			hdlID = 0;	////本地transId，无记录的链接
		else
		{
			WSClientReqInfoStruct* p = ReduceOriginTransID(msg);
			if (NULL == p)
				return std::make_pair(ACKPROCESS_NOTHING, "can not find WSClient!");// 异常 应该不会有这种情况出现
			msg->setTransID(p->transID);
			hdlID = p->hdlID;
		}
		auto sessionId = msg->getSessionID();
		auto entityName = UpdateSessionMap(globalID, sessionId);
		if ("" == entityName)
			return std::make_pair(ACKPROCESS_NOTHING, "Update SessionMap error!");

		updateBeginSessionTime(entityName, -1 == sessionId);	//clean the begin Sessiontime
		return std::make_pair(ACKPROCESS_SEND, entityName);
	}

	void EntitySessionManager::doWithErrorCode(int errorCode, int transId)
	{
		switch (errorCode)
		{
		case Error_Break:
		case Error_MethodNotFound:
		case Error_MethodSignatureFailed:
			UpdateSessionMap(transId, -1);
			break;
		default:
			break;
		}
	}

	AckProcessType EntitySessionManager::ProcessWithRecvEvent(CMessage* msg, unsigned int& hdlID) {
		WSClientReqInfoStruct* p = ReduceOriginTransID(msg);
		if (NULL == p)
			return ACKPROCESS_NOTHING;// 异常 应该不会有这种情况出现

		msg->setTransID(p->transID);
		hdlID = p->hdlID;
		if (hdlID == 0) {
			return ACKPROCESS_NOTHING;
		}
		return ACKPROCESS_SEND;
	}

	AckProcessType EntitySessionManager::AskProcessEvent(CMessage* msg, std::vector<std::pair<int, int>>& transIdandHdlIDArr)
	{
		WSClientReqInfoStruct* p = ReduceOriginTransID(msg);
		if (NULL == p)
			return ACKPROCESS_NOTHING;
		transIdandHdlIDArr.clear();
		while (NULL != p)
		{
			if (0 != p->transID)
				transIdandHdlIDArr.emplace_back(std::make_pair(p->transID, p->hdlID));
			p = p->next;
		}
		if (transIdandHdlIDArr.size() > 0)
			return ACKPROCESS_SEND;
		else
			return ACKPROCESS_NOTHING;
	}

	AckProcessType EntitySessionManager::AckProcess(CMessage* msg, unsigned int& hdlID) {
		AckProcessType ret = ACKPROCESS_NOTHING;
		switch (msg->getMessageType()) {
		case Event:
			ProcessWithRecvEvent(msg, hdlID);
			break;
		case SessionAck:
		case RequestAck:
		case GetVarAck:
		case SetVarAck:
			ProcessWithAck(msg, hdlID);
			break;
		default:
			DbgWithLink(LOG_LEVEL_DEBUG, LOG_TYPE_SYSTEM).setAPI(__FUNCTION__)("AckProcess default!");
			ret = ACKPROCESS_NOTHING;
		}
		return ret;
	}

}