用libevent和共享内存实现的一个游戏server

// // Created by DGuco on 17-7-13. // #include <dbmessage.pb.h> #include <client_comm_engine.h> #include <my_assert.h> #include "dbctrl.h" #include "config.h" #include "connector.h" #include "server_comm_engine.h" #include "message_factory.h" #include "database_mysql.h" template<> shared_ptr<CDBCtrl> CSingleton<CDBCtrl>::spSingleton = NULL; int CDBCtrl::m_iProxyId = 0; CByteBuff CDBCtrl::m_acRecvBuff = CByteBuff(); CDBCtrl::CDBCtrl() { m_iRunFlag = 0; m_tLastSendKeepAlive = 0; m_tLastRecvKeepAlive = 0; m_pNetWork = CNetWork::GetSingletonPtr(); m_pServerConfig = CServerConfig::GetSingletonPtr(); m_pMsgFactory = std::make_shared<CMessageFactory>(); m_pDatabase = std::make_shared<DatabaseMysql>(); } CDBCtrl::~CDBCtrl() { } void CDBCtrl::SetRunFlag(int iFlag) { m_iRunFlag = iFlag; LOG_INFO("default", "Set DealMsg Flag {}, All Flag Is {}", iFlag, m_iRunFlag); } void CDBCtrl::ClearRunFlag(int iFlag) { m_iRunFlag = 0; LOG_INFO("default", "Clear DealMsg Flag {}", iFlag); } bool CDBCtrl::IsRunFlagSet(int iFlag) { return iFlag == m_iRunFlag; } int CDBCtrl::SendMessageTo(CProxyMessage *pMsg) { if (pMsg == NULL) { LOG_ERROR("default", "in CDBHandle::SendMessageTo, pmsg is null"); return -1; } CProxyHead *stProxyHead = pMsg->mutable_msghead(); CByteBuff tmBuff; unsigned short nCodeLength = 0; ServerInfo *dbInfo = m_pServerConfig->GetServerInfo(enServerType::FE_DBSERVER); ServerInfo *proxyInfo = m_pServerConfig->GetServerInfo(enServerType::FE_PROXYSERVER); pbmsg_setproxy(stProxyHead, enServerType::FE_PROXYSERVER, proxyInfo->m_iServerId, enServerType::FE_DBSERVER, dbInfo->m_iServerId, GetMSTime(), enMessageCmd::MESS_REGIST); int iRet = CServerCommEngine::ConvertMsgToStream(pMsg, &tmBuff, nCodeLength); if (iRet != 0) { LOG_ERROR("default", "CDBCtrl::RegisterToProxyServer ConvertMsgToStream failed, iRet = {}.", iRet); return 0; } //int nSendReturn = m_pProxySvrdConns[ m_current_proxy_index ].GetSocket()->SendOneCode( nCodeLength, abyCodeBuf ); IBufferEvent *tmpConnector = m_pNetWork->FindConnector(CDBCtrl::m_iProxyId); if (tmpConnector == NULL) { LOG_ERROR("default", "Connection to proxyserver has gone"); return -1; } int nSendReturn = tmpConnector->Send(tmBuff.GetData(), nCodeLength); if (nSendReturn < 0) { LOG_ERROR("default", "Send Code(len:{}) To Proxy faild(error={})", nCodeLength, nSendReturn); return -1; } CGooMess *pUnknownMessagePara = (CGooMess *) pMsg->msgpara(); MY_ASSERT(pUnknownMessagePara != NULL, return 0); const ::google::protobuf::Descriptor *pDescriptor = pUnknownMessagePara->GetDescriptor(); LOG_DEBUG("default", "SendMessageTo: MsgName[{}] ProxyHead[{}] MsgHead[{}] MsgPara[{}]", pDescriptor->name().c_str(), stProxyHead->ShortDebugString().c_str(), pMsg->ShortDebugString().c_str(), ((CGooMess *) pMsg->msgpara())->ShortDebugString().c_str()); return 0; } // 执行相应的指令 int CDBCtrl::Event(CProxyMessage *pMsg) { if (pMsg == NULL) { LOG_ERROR("default", "In CDBHandle::Event, input msg null"); return -1; } CGooMess *pUnknownMessagePara = (CGooMess *) pMsg->msgpara(); MY_ASSERT(pUnknownMessagePara != NULL, return 0); const ::google::protobuf::Descriptor *pDescriptor = pUnknownMessagePara->GetDescriptor(); LOG_DEBUG("default", "ReceveMsg: MsgName[{}] MsgHead[{}] MsgPara[{}]", pDescriptor->name().c_str(), pMsg->ShortDebugString().c_str(), ((CGooMess *) pMsg->msgpara())->ShortDebugString().c_str()); switch (pMsg->msghead().messageid()) { case CMsgExecuteSqlRequest::MsgID: // 服务器执行SQL请求 { ProcessExecuteSqlRequest(pMsg); break; } default: { break; } } return 0; } // 执行相应的指令 int CDBCtrl::ProcessExecuteSqlRequest(CProxyMessage *pMsg) { if (pMsg == NULL) { LOG_ERROR("default", "Error: [{}][{}][{}], invalid input.\n", __MY_FILE__, __LINE__, __FUNCTION__); return -1; } CMsgExecuteSqlRequest *pReqMsg = (CMsgExecuteSqlRequest * )(pMsg->msgpara()); if (pReqMsg == NULL) { LOG_ERROR("default", "Error: [{}][{}][{}], msgpara null.\n", __MY_FILE__, __LINE__, __FUNCTION__); return -1; } if (pReqMsg->sql().length() <= 0) { LOG_ERROR("default", "Error: [{}][{}][{}], sql len({}) invalid.\n", __MY_FILE__, __LINE__, __FUNCTION__, pReqMsg->sql().length()); return -1; } LOG_ERROR("default", "{} \n", pReqMsg->ShortDebugString().c_str()); string sqlStr = ""; if (pReqMsg->hasblob() == HASBLOB) { // 有blob字段的处理 sqlStr += pReqMsg->sql(); if (pReqMsg->bufsize() > 0) { if (pReqMsg->bufsize() >= MAX_PACKAGE_LEN) { // blob字段超出长度 LOG_ERROR("default", "Error: [{}][{}][{}], exec sql {} faild. sql_len:{} > MAX_PACKAGE_LEN\n", __MY_FILE__, __LINE__, __FUNCTION__, pReqMsg->sql().c_str(), pReqMsg->bufsize()); return -1; } char sqlBuff[MAX_PACKAGE_LEN * 2 + 1] = {0}; int len = m_pDatabase->escape_string(sqlBuff, pReqMsg->buffer().c_str(), pReqMsg->bufsize()); if (len <= 0) { LOG_ERROR("default", "Error: [{}][{}][{}], escape_string error!!!!\n", __MY_FILE__, __LINE__, __FUNCTION__); return -1; } sqlStr += sqlBuff; } sqlStr += pReqMsg->sqlwhere(); } else { // 没有blob字段的处理 sqlStr += pReqMsg->sql(); } // 需要回执 if (MUSTCALLBACK == pReqMsg->callback()) { CProxyMessage tMsg; // 该消息为回执 CMsgExecuteSqlResponse tSqlResMsg; // tSqlResMsg 为消息体 tMsg.Clear(); tSqlResMsg.Clear(); tSqlResMsg.set_logictype(pReqMsg->logictype()); tSqlResMsg.set_sessionid(pReqMsg->sessionid()); tSqlResMsg.set_timestamp(pReqMsg->timestamp()); tSqlResMsg.set_teamid(pReqMsg->teamid()); LOG_ERROR("default", "Execute SQL: {}", sqlStr.c_str()); // 用于在日志中查看SQL查询语句 CProxyHead *pTmpHead = tMsg.mutable_msghead(); pTmpHead->set_messageid(CMsgExecuteSqlResponse::MsgID); if (pMsg->has_msghead()) { if (pMsg->mutable_msghead()->has_msghead()) { CClientCommEngine::CopyMesHead(pMsg->mutable_msghead()->mutable_msghead(), pTmpHead->mutable_msghead()); } } tMsg.set_msgpara((uint64) &tSqlResMsg); if (pReqMsg->sqltype() == SELECT || pReqMsg->sqltype() == CALL) { QueryResult *res = NULL; if (pReqMsg->sqltype() == SELECT) { res = m_pDatabase->Query(sqlStr.c_str(), sqlStr.length()); } else { res = m_pDatabase->QueryForprocedure(sqlStr.c_str(), sqlStr.length(), pReqMsg->outnumber()); } if (res == NULL) // 无返回结果 { LOG_ERROR("default", "sql {} exec, but no resultset returned", sqlStr.c_str()); tSqlResMsg.set_resultcode(0); } else // 有返回结果 { tSqlResMsg.set_resultcode(1); tSqlResMsg.set_rowcount(res->GetRowCount()); // 行 tSqlResMsg.set_colcount(res->GetFieldCount()); // 列 //TODO: 列名 + 列类型 暂时不赋值 if (res->GetRowCount() > 0) { // 列值 和 值长度 do { Field *pField = res->Fetch(); if (pField == NULL) { LOG_ERROR("default", "ERROR: do sql {} success, row[{}], col[{}], but some row is null\n", sqlStr.c_str(), res->GetRowCount(), res->GetFieldCount()); //TODO: 出错设为0 tSqlResMsg.set_rowcount(0); break; } for (int j = 0; j < (int) res->GetFieldCount(); j++) { tSqlResMsg.add_fieldvalue(pField[j].GetString(), pField[j].GetValueLen()); tSqlResMsg.add_fieldvaluelen(pField[j].GetValueLen()); } } while (res->NextRow()); } } // 安全释放结果集 ReleaseResult(res); } else // 非select操作 { bool bExecResult = m_pDatabase->RealDirectExecute(sqlStr.c_str(), sqlStr.length()); // 执行操作 tSqlResMsg.set_resultcode(bExecResult ? 1 : 0); } // 回复客户端 SendMessageTo(&tMsg); } else { // 同步执行 if (m_pDatabase->RealDirectExecute(sqlStr.