消息伺服器支援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 伺服器--配置TeamTalk 服務端分析 二、配置[圖]](data:image/gif;base64,R0lGODlhAQABAIAAAP///wAAACwAAAAAAQABAAACAkQBADs=)