myvtm/Module/mod_chromium/EntitySessionManager.cpp

#if (defined _WIN32 || defined _WIN64)
#include "stdafx.h"
#endif
#include "MessageType.h"
#include "EntitySessionManager.h"
#include "baseEx.h"
#include<ctime>
#include <string>
#include "guitask/format.hpp"
#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
	{
		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){
	m_session_ack_map.insert(std::pair<int, std::string>(transID, entityName));
}

std::vector<std::string> EntitySessionManager::queryUnLinkSession()
{
	std::vector<std::string> ret;
	for (auto it : m_session_map)
	{
		if (it.second == -1)
			ret.push_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()){
	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:
		DbgEx("EntitySessionManager::RequestProcess switch default!");
		ret = PROCESS_NOTHING;
	}

	StoreSessionReq(msg->getTransID(), entityName);
	if (Request == msg->getMessageType())
		msg->exchangeSessionIdAndTransId();

	return ret;
}

RequestProcessType EntitySessionManager::ProcessWithInfo(CMessage* msg, std::string entityName, unsigned int hdlID){
	DbgEx("EntitySessionManager::ProcessWithInfo");
	auto s = GetSessionIDByEntityName(entityName);
	if (false == s.first || -1 == s.second)
		return PROCESS_NOTHING;
	msg->setSessionID(s.second, false);
	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)
	{
		DbgEx("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){
	DbgEx("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){
	DbgEx("EntitySessionManager::ProcessWithEndSession");
	return PROCESS_NOTHING;
}

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

	return PROCESS_SEND;
}

RequestProcessType EntitySessionManager::ProcessWithRegister(CMessage* msg, std::string entityName, unsigned int hdlID){
	DbgEx("EntitySessionManager::ProcessWithRegister");
	auto it = m_session_map.find(entityName);
	if (it == m_session_map.end() || -1 == it->second)
	{
		DbgEx("no session found for entity : %s", UtfToGbk(entityName.c_str()).c_str());
		return PROCESS_STARTSESSION;
	}

	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){
	DbgEx("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){
	DbgEx("EntitySessionManager::ProcessWithSetVarReq");
	msg->setTransID(MakeNewTransID(msg, hdlID));
	return PROCESS_SEND;
}

RequestProcessType EntitySessionManager::ProcessWithGetVarReq(CMessage* msg, std::string entityName, 
	unsigned int hdlID){
	DbgEx("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("\"}"));
	DbgEx("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;
	for (auto it : m_session_map)
	{
		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,\"%s\":%s}", 16, transId, 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);
	bool needChange = true;
	if (m_session_map.end() != sessionIt)
	{

		if (sessionIt->second == session)
			needChange = false;
		m_session_map.erase(sessionIt);
	}
	if(needChange)
		DbgEx("update sessionId :<%s><%d>", entityname.c_str(), session);
	m_session_map.insert(std::make_pair(entityname, session));
	return *this;
}

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

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

	//RemoveAckListNode(globalID);
	if (hdlID == 0){
		return ACKPROCESS_NOTHING;
	}
	
	return ACKPROCESS_SEND;
}

std::pair<bool, std::string> EntitySessionManager::DoSessionRemove(unsigned int session)
{
	for (auto it : m_session_map)
	{
		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);
}



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.push_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:
		ret = ProcessWithAck(msg, hdlID);
		break;
	default:
		DbgEx("AckProcess default!");
		ret = ACKPROCESS_NOTHING;
	}
	return ret;
}

}