TeamTalk源码分析之msg_server

2024-04-11 03:32

本文主要是介绍TeamTalk源码分析之msg_server,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

消息服务器支持TCP长连接和HTTP长轮询两种接入方式,本节主要讲SOCKET的轮训,先看如下几个函数。

void CMsgConn::OnConnect(net_handle_thandle)

{

     m_handle = handle;

     m_login_time = get_tick_count();

     g_msg_conn_map.insert(make_pair(handle,this));

     netlib_option(handle,NETLIB_OPT_SET_CALLBACK, (void*)imconn_callback);

     netlib_option(handle,NETLIB_OPT_SET_CALLBACK_DATA, (void*)&g_msg_conn_map);

     netlib_option(handle,NETLIB_OPT_GET_REMOTE_IP, (void*)&m_peer_ip);

     netlib_option(handle,NETLIB_OPT_GET_REMOTE_PORT, (void*)&m_peer_port);

     //根据handle,为socket类添加如上4个参数。

}

// for client connect in

void msg_serv_callback(void* callback_data, uint8_t msg, uint32_t handle, void* pParam)

{

     if (msg ==NETLIB_MSG_CONNECT)

     {

         CMsgConn* pConn = new CMsgConn();

         pConn->OnConnect(handle);

         //有客户端connect后,将accept后的SOCKET跟新建的CMsgConn相关联,并将SOCKET加入轮训。

     }

     else

     {

         log("!!!error msg: %d\n", msg);

     }

}

int main(intargc,char*argv[])

{

   …  //读配置,初始化SOCKET等操作。

     CStrExplode listen_ip_list(listen_ip, ';');

     for (uint32_ti = 0;i <listen_ip_list.GetItemCnt();i++) {

         ret = netlib_listen(listen_ip_list.GetItem(i),listen_port,msg_serv_callback,NULL);

         if (ret ==NETLIB_ERROR)

              return ret;

     }

   ……

netlib_eventloop();  //开始轮训,处理定时回调,SOCKET。

}

netlib_eventloop执行后,进开始进行循环,处理定时回调,SOCKET轮训,以下以读为例。

void CBaseSocket::OnRead()

{

     if (m_state ==SOCKET_STATE_LISTENING)

     {

         _AcceptNewSocket();

     }

     else

     {

         u_long avail = 0;

         if ( (ioctlsocket(m_socket,FIONREAD, &avail) ==SOCKET_ERROR) || (avail == 0) )

         {

              m_callback(m_callback_data,NETLIB_MSG_CLOSE, (net_handle_t)m_socket,NULL);

         }

         else

         {

              m_callback(m_callback_data,NETLIB_MSG_READ, (net_handle_t)m_socket,NULL);

         }

     }

}

SOCKET可读后,执行如上CBaseSocket::OnRead,紧接着会执行m_callback(m_callback_data,NETLIB_MSG_READ, (net_handle_t)m_socket,NULL);,此时会转到如下回调函数中:

void imconn_callback(void* callback_data, uint8_t msg, uint32_t handle, void* pParam)

{

     NOTUSED_ARG(handle);

     NOTUSED_ARG(pParam);

     if (!callback_data)

         return;

     ConnMap_t* conn_map = (ConnMap_t*)callback_data;

     CImConn* pConn = FindImConn(conn_map,handle);

     if (!pConn)

         return;

     //log("msg=%d, handle=%d\n", msg, handle);

     switch (msg)

     {

     case NETLIB_MSG_CONFIRM:

         pConn->OnConfirm();

         break;

     case NETLIB_MSG_READ:

         pConn->OnRead();

         break;

     case NETLIB_MSG_WRITE:

         pConn->OnWrite();

         break;

     case NETLIB_MSG_CLOSE:

         pConn->OnClose();

         break;

     default:

         log("!!!imconn_callback error msg: %d\n", msg);

         break;

     }

     pConn->ReleaseRef();

}

可以看到,会执行如下函数,先从SOCKET中读出数据,在后面会执行CImPdu::ReadPdu。

void CImConn::OnRead()

{

     for (;;)

     {

         uint32_t free_buf_len = m_in_buf.GetAllocSize() -m_in_buf.GetWriteOffset();

         if (free_buf_len <READ_BUF_SIZE)

              m_in_buf.Extend(READ_BUF_SIZE);

 

         int ret = netlib_recv(m_handle,m_in_buf.GetBuffer() +m_in_buf.GetWriteOffset(),READ_BUF_SIZE);

         if (ret <= 0)

              break;

         m_recv_bytes += ret;

         m_in_buf.IncWriteOffset(ret);

         m_last_recv_tick = get_tick_count();

     }

     if (m_policy_conn) {

         return;

     }

     // no received data is read by ReadPdu(), check if this is a flash security policy request

     if (m_recv_bytes ==m_in_buf.GetWriteOffset()) {

         if ( (m_in_buf.GetBuffer()[0] =='<') && (g_policy_content !=NULL) ) {

              log("policy request, handle=%d\n", m_handle);

              m_policy_conn = true;

              Send(g_policy_content,g_policy_len);

              return;

         }

     }

     try {

         CImPdu* pPdu = NULL;

         while ( ( pPdu = CImPdu::ReadPdu(m_in_buf.GetBuffer(),m_in_buf.GetWriteOffset()) ) )

         {

              uint32_t pdu_len = pPdu->GetLength();

              HandlePdu(pPdu);

              m_in_buf.Read(NULL,pdu_len);

              delete pPdu;

              ++g_recv_pkt_cnt;

          }

     } catch (CPduException&ex) {

         log("!!!catch exception, sid=%u, cid=%u, err_code=%u, err_msg=%s, close the connection\n",

                   ex.GetModuleId(),ex.GetCommandId(),ex.GetErrorCode(),ex.GetErrorMsg());

         OnClose();

     }

}

CImPdu* CImPdu::ReadPdu(uchar_t *buf,uint32_tlen)

{

     uint32_t pdu_len = 0;

     if (!_IsPduAvailable(buf,len,pdu_len))

         return NULL;

     uint16_t service_id = CByteStream::ReadUint16(buf + 4);

     uint16_t command_id = CByteStream::ReadUint16(buf + 6);

     CImPdu* pPdu = NULL;

     switch (service_id)

     {

     case SID_LOGIN:

         pPdu = ReadPduLogin(command_id, buf, pdu_len);

         break;

     case SID_BUDDY_LIST:

         pPdu = ReadPduBuddyList(command_id, buf, pdu_len);

         break;

     case SID_GROUP:

         pPdu = ReadPduGroup(command_id, buf, pdu_len);

         break;

     case SID_MSG:

         pPdu = ReadPduMsg(command_id, buf, pdu_len);

         break;

     case SID_OTHER:

         pPdu = ReadPduOther(command_id, buf, pdu_len);

         break;

     case SID_SWITCH_SERVICE:

         pPdu = ReadPduSwitchService(command_id, buf, pdu_len);

         break;

     case SID_FILE:

         pPdu = ReadPduFile(command_id, buf, pdu_len);

         break;

     default:

         throw CPduException(service_id, command_id, ERROR_CODE_WRONG_SERVICE_ID,"wrong service id");

     }

     pPdu->_SetIncomingLen(pdu_len);

     pPdu->_SetIncomingBuf(buf);

     return pPdu;

}

现在以登陆为例子,进入CImPdu*CImPdu::ReadPduLogin。

CImPdu* CImPdu::ReadPduLogin(uint16_tcommand_id,uchar_t*pdu_buf,uint32_tpdu_len)

{

     CImPdu* pPdu = NULL;

     switch (command_id) {

     case CID_LOGIN_REQ_MSGSERVER:

         pPdu = new CImPduMsgServRequest(pdu_buf,pdu_len);

         break;

     case CID_LOGIN_RES_MSGSERVER:

         pPdu = new CImPduMsgServResponse(pdu_buf,pdu_len);

         break;

     case CID_LOGIN_REQ_USERLOGIN:

         pPdu = new CImPduLoginRequest(pdu_buf,pdu_len);

         break;

     case CID_LOGIN_RES_USERLOGIN:

         pPdu = new CImPduLoginResponse(pdu_buf,pdu_len);

         break;

     case CID_LOGIN_KICK_USER:

         pPdu = new CImPduKickUser(pdu_buf,pdu_len);

         break;

     default:

         throw CPduException(SID_LOGIN, command_id, ERROR_CODE_WRONG_COMMAND_ID, "wrong command id");

     }

     return pPdu;

}

可以看到是在打包回复的内容,返回pPdu,然后回到CImConn::OnRead,执行HandlePdu(pPdu)-> _HandleLoginRequest-> SendPdu(&pdu)-> CBaseSocket::Send。

这篇关于TeamTalk源码分析之msg_server的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/892966

相关文章

SQL Server配置管理器无法打开的四种解决方法

《SQLServer配置管理器无法打开的四种解决方法》本文总结了SQLServer配置管理器无法打开的四种解决方法,文中通过图文示例介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的... 目录方法一:桌面图标进入方法二:运行窗口进入检查版本号对照表php方法三:查找文件路径方法四:检查 S

怎样通过分析GC日志来定位Java进程的内存问题

《怎样通过分析GC日志来定位Java进程的内存问题》:本文主要介绍怎样通过分析GC日志来定位Java进程的内存问题,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录一、GC 日志基础配置1. 启用详细 GC 日志2. 不同收集器的日志格式二、关键指标与分析维度1.

MySQL中的表连接原理分析

《MySQL中的表连接原理分析》:本文主要介绍MySQL中的表连接原理分析,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录1、背景2、环境3、表连接原理【1】驱动表和被驱动表【2】内连接【3】外连接【4编程】嵌套循环连接【5】join buffer4、总结1、背景

python中Hash使用场景分析

《python中Hash使用场景分析》Python的hash()函数用于获取对象哈希值,常用于字典和集合,不可变类型可哈希,可变类型不可,常见算法包括除法、乘法、平方取中和随机数哈希,各有优缺点,需根... 目录python中的 Hash除法哈希算法乘法哈希算法平方取中法随机数哈希算法小结在Python中,

Java Stream的distinct去重原理分析

《JavaStream的distinct去重原理分析》Javastream中的distinct方法用于去除流中的重复元素,它返回一个包含过滤后唯一元素的新流,该方法会根据元素的hashcode和eq... 目录一、distinct 的基础用法与核心特性二、distinct 的底层实现原理1. 顺序流中的去重

关于MyISAM和InnoDB对比分析

《关于MyISAM和InnoDB对比分析》:本文主要介绍关于MyISAM和InnoDB对比分析,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录开篇:从交通规则看存储引擎选择理解存储引擎的基本概念技术原理对比1. 事务支持:ACID的守护者2. 锁机制:并发控制的艺

SQL Server修改数据库名及物理数据文件名操作步骤

《SQLServer修改数据库名及物理数据文件名操作步骤》在SQLServer中重命名数据库是一个常见的操作,但需要确保用户具有足够的权限来执行此操作,:本文主要介绍SQLServer修改数据... 目录一、背景介绍二、操作步骤2.1 设置为单用户模式(断开连接)2.2 修改数据库名称2.3 查找逻辑文件名

SQL Server数据库死锁处理超详细攻略

《SQLServer数据库死锁处理超详细攻略》SQLServer作为主流数据库管理系统,在高并发场景下可能面临死锁问题,影响系统性能和稳定性,这篇文章主要给大家介绍了关于SQLServer数据库死... 目录一、引言二、查询 Sqlserver 中造成死锁的 SPID三、用内置函数查询执行信息1. sp_w

MyBatis Plus 中 update_time 字段自动填充失效的原因分析及解决方案(最新整理)

《MyBatisPlus中update_time字段自动填充失效的原因分析及解决方案(最新整理)》在使用MyBatisPlus时,通常我们会在数据库表中设置create_time和update... 目录前言一、问题现象二、原因分析三、总结:常见原因与解决方法对照表四、推荐写法前言在使用 MyBATis

Python主动抛出异常的各种用法和场景分析

《Python主动抛出异常的各种用法和场景分析》在Python中,我们不仅可以捕获和处理异常,还可以主动抛出异常,也就是以类的方式自定义错误的类型和提示信息,这在编程中非常有用,下面我将详细解释主动抛... 目录一、为什么要主动抛出异常?二、基本语法:raise关键字基本示例三、raise的多种用法1. 抛