TeamTalk源码分析之file_server

2019年12月14日 阅读数:51
这篇文章主要向大家介绍TeamTalk源码分析之file_server,主要内容包括基础应用、实用技巧、原理机制等方面,希望对大家有所帮助。

1、链接情况介绍
windows

fileserver开始并非和客户端链接的,客户端是按需链接file_server的。可是file_server与msg_server倒是长链接。先启动file_server,再启动msg_server。msg_server初始化的时候,会去尝试链接file_server的8601端口。链接成功之后,会给file_server发送一个发包询问file_server侦听客户端链接的ip和端口号信息:服务器

[cpp]  view plain  copy
  1. void CFileServConn::OnConfirm()  
  2. {  
  3.     log("connect to file server success ");  
  4.     m_bOpen = true;  
  5.     m_connect_time = get_tick_count();  
  6.     g_file_server_list[m_serv_idx].reconnect_cnt = MIN_RECONNECT_CNT / 2;  
  7.       
  8.     //连上file_server之后,给file_server发送获取ip地址的数据包  
  9.     IM::Server::IMFileServerIPReq msg;  
  10.     CImPdu pdu;  
  11.     pdu.SetPBMsg(&msg);  
  12.     pdu.SetServiceId(SID_OTHER);  
  13.     pdu.SetCommandId(CID_OTHER_FILE_SERVER_IP_REQ);  
  14.     SendPdu(&pdu);  
  15. }  

file_server收到该数据包后,将本身的侦听客户端链接的ip地址和端口号发包告诉msg_server:

[cpp]  view plain  copy
  1. void FileMsgServerConn::_HandleGetServerAddressReq(CImPdu* pPdu) {  
  2.     IM::Server::IMFileServerIPRsp msg;  
  3.       
  4.     const std::list<IM::BaseDefine::IpAddr>& addrs = ConfigUtil::GetInstance()->GetAddressList();  
  5.       
  6.     for (std::list<IM::BaseDefine::IpAddr>::const_iterator it=addrs.begin(); it!=addrs.end(); ++it) {  
  7.         IM::BaseDefine::IpAddr* addr = msg.add_ip_addr_list();  
  8.         *addr = *it;  
  9.         log("Upload file_client_conn addr info, ip=%s, port=%d", addr->ip().c_str(), addr->port());  
  10.     }  
  11.       
  12.     SendMessageLite(this, SID_OTHER, CID_OTHER_FILE_SERVER_IP_RSP, pPdu->GetSeqNum(), &msg);  
  13. }  

获得的信息是file_server侦听的ip地址和端口号,默认配置的端口号是8600。也就是说file_server的8600用于客户端链接,8601端口用于msg_server链接。这样,客户端须要传输文件(注意:不是聊天图片,聊天图片使用另一个服务msfs进行传输),会先告诉msg_server它须要进行文件传输,msg_server收到消息后告诉客户端,你连file_server来传输文件吧,并把file_server的地址和端口号告诉客户端。客户端这个时候链接file_server进行文件传输。咱们来具体看一看这个流程的细节信息:

1. 客户端发包给msg_server说要进行文件发送并发


而后选择一个文件:dom




pc客户端发送文件逻辑:ide

[cpp]  view plain  copy
  1. //pc客户端代码(Modules工程SessionLayout.cpp)  
  2. void SessionLayout::Notify(TNotifyUI& msg)  
  3. {  
  4.     ...  
  5.     //省略无关代码  
  6.     else if (msg.pSender == m_pBtnsendfile) //文件传输  
  7.     {  
  8.         module::UserInfoEntity userInfo;  
  9.         if (!module::getUserListModule()->getUserInfoBySId(m_sId, userInfo))  
  10.         {  
  11.             LOG__(ERR, _T("SendFile can't find the sid"));  
  12.             return;  
  13.         }  
  14.   
  15.         CFileDialog fileDlg(TRUE, NULL, NULL, OFN_HIDEREADONLY | OFN_FILEMUSTEXIST  
  16.             , _T("文件|*.*||"), AfxGetMainWnd());  
  17.         fileDlg.m_ofn.Flags |= OFN_NOCHANGEDIR;  
  18.         fileDlg.DoModal();  
  19.   
  20.         CString sPathName;  
  21.         POSITION nPos = fileDlg.GetStartPosition();  
  22.         if (nPos != NULL)  
  23.         {  
  24.             sPathName = fileDlg.GetNextPathName(nPos);  
  25.         }  
  26.         if (sPathName.IsEmpty())  
  27.             return;  
  28.   
  29.         module::getFileTransferModule()->sendFile(sPathName, m_sId, userInfo.isOnlne());  
  30.     }  
  31.     ...  
  32.     //省略无关代码  
  33. }  


sPathName是文件的全饰路径;m_sId是收取文件的用户id,userInfo.isOnlne()判断m_sId表明的用户是否在线,以此来决定此次文件传输是在线文件仍是离线文件模式。
 
[cpp]  view plain  copy
  1. BOOL FileTransferModule_Impl::sendFile(IN const CString& sFilePath, IN const std::string& sSendToSID,IN BOOL bOnlineMode)  
  2. {  
  3.     if (TransferFileEntityManager::getInstance()->checkIfIsSending(sFilePath))  
  4.     {  
  5.         return FALSE;  
  6.     }  
  7.     TransferFileEntity fileEntity;  
  8.       
  9.     //获取文件大小  
  10.     struct __stat64 buffer;  
  11.     _wstat64(sFilePath, &buffer);  
  12.     fileEntity.nFileSize = (UInt32)buffer.st_size;  
  13.     if (0 != fileEntity.nFileSize)  
  14.     {  
  15.         CString strFileName = sFilePath;  
  16.         strFileName.Replace(_T("\\"), _T("/"));//mac上对于路径字符“\”须要作特殊处理,windows上则能够识别  
  17.         fileEntity.sFileName = util::cStringToString(strFileName);  
  18.         fileEntity.sFromID = module::getSysConfigModule()->userID();  
  19.         fileEntity.sToID = sSendToSID;  
  20.         uint32_t transMode = 0;  
  21.         transMode = bOnlineMode ? IM::BaseDefine::TransferFileType::FILE_TYPE_ONLINE : IM::BaseDefine::TransferFileType::FILE_TYPE_OFFLINE;  
  22.           
  23.         LOG__(DEBG,_T("FileTransferSevice_Impl::sendFile sTaskID = %s"), util::stringToCString(fileEntity.sTaskID));  
  24.   
  25.         imcore::IMLibCoreStartOperationWithLambda(  
  26.             [=]()  
  27.         {  
  28.             IM::File::IMFileReq imFileReq;  
  29.             LOG__(APP, _T("imFileReq,name=%s,size=%d,toId=%s"),util::stringToCString(fileEntity.sFileName)  
  30.                 ,fileEntity.nFileSize,util::stringToCString(fileEntity.sToID));  
  31.             imFileReq.set_from_user_id(util::stringToInt32(fileEntity.sFromID));  
  32.             imFileReq.set_to_user_id(util::stringToInt32(fileEntity.sToID));  
  33.             imFileReq.set_file_name(fileEntity.sFileName);  
  34.             imFileReq.set_file_size(fileEntity.nFileSize);  
  35.             imFileReq.set_trans_mode(static_cast<IM::BaseDefine::TransferFileType>(transMode));  
  36.   
  37.             module::getTcpClientModule()->sendPacket(IM::BaseDefine::ServiceID::SID_FILE  
  38.                 , IM::BaseDefine::FileCmdID::CID_FILE_REQUEST  
  39.                 , &imFileReq);  
  40.         });  
  41.           
  42.         return TRUE;  
  43.     }  
  44.     LOG__(ERR, _T("fileEntity FileSize error,size = %d"), fileEntity.nFileSize);  
  45.     return FALSE;  
  46. }  

上面代码中组装的包信息中含有要传输的文件路径、文件大小、发送人id、接收方id、文件传输模式(在线仍是离线),包的命令号是IM::BaseDefine::FileCmdID::CID_FILE_REQUEST。这个包发给msg_server之后,msg_server应答:

[cpp]  view plain  copy
  1. void CMsgConn::HandlePdu(CImPdu* pPdu)  
  2. {  
  3. ...  
  4. //省略无关代码  
  5. case CID_FILE_REQUEST:  
  6.             s_file_handler->HandleClientFileRequest(this, pPdu);  
  7.             break;  
  8. ...  
  9. //省略无关代码  
  10. }  

[cpp]  view plain  copy
  1. void CFileHandler::HandleClientFileRequest(CMsgConn* pMsgConn, CImPdu* pPdu)  
  2. {  
  3.     IM::File::IMFileReq msg;  
  4.     CHECK_PB_PARSE_MSG(msg.ParseFromArray(pPdu->GetBodyData(), pPdu->GetBodyLength()));  
  5.       
  6.     uint32_t from_id = pMsgConn->GetUserId();  
  7.     uint32_t to_id = msg.to_user_id();  
  8.     string file_name = msg.file_name();  
  9.     uint32_t file_size = msg.file_size();  
  10.     uint32_t trans_mode = msg.trans_mode();  
  11.     log("HandleClientFileRequest, %u->%u, fileName: %s, trans_mode: %u.", from_id, to_id, file_name.c_str(), trans_mode);  
  12.       
  13.     CDbAttachData attach(ATTACH_TYPE_HANDLE, pMsgConn->GetHandle());  
  14.     CFileServConn* pFileConn = get_random_file_serv_conn();  
  15.     if (pFileConn)  
  16.     {  
  17.         IM::Server::IMFileTransferReq msg2;  
  18.         msg2.set_from_user_id(from_id);  
  19.         msg2.set_to_user_id(to_id);  
  20.         msg2.set_file_name(file_name);  
  21.         msg2.set_file_size(file_size);  
  22.         msg2.set_trans_mode((IM::BaseDefine::TransferFileType)trans_mode);  
  23.         msg2.set_attach_data(attach.GetBuffer(), attach.GetLength());  
  24.         CImPdu pdu;  
  25.         pdu.SetPBMsg(&msg2);  
  26.         pdu.SetServiceId(SID_OTHER);  
  27.         pdu.SetCommandId(CID_OTHER_FILE_TRANSFER_REQ);  
  28.         pdu.SetSeqNum(pPdu->GetSeqNum());  
  29.           
  30.         if (IM::BaseDefine::FILE_TYPE_OFFLINE == trans_mode)  
  31.         {  
  32.             pFileConn->SendPdu(&pdu);  
  33.         }  
  34.         else //IM::BaseDefine::FILE_TYPE_ONLINE  
  35.         {  
  36.             CImUser* pUser = CImUserManager::GetInstance()->GetImUserById(to_id);  
  37.             if (pUser && pUser->GetPCLoginStatus())//已有对应的帐号pc登陆状态  
  38.             {  
  39.                 pFileConn->SendPdu(&pdu);  
  40.             }  
  41.             else//无对应用户的pc登陆状态,向route_server查询状态  
  42.             {  
  43.                 //no pc_client in this msg_server, check it from route_server  
  44.                 CPduAttachData attach_data(ATTACH_TYPE_HANDLE_AND_PDU_FOR_FILE, pMsgConn->GetHandle(), pdu.GetBodyLength(), pdu.GetBodyData());  
  45.                 IM::Buddy::IMUsersStatReq msg3;  
  46.                 msg3.set_user_id(from_id);  
  47.                 msg3.add_user_id_list(to_id);  
  48.                 msg3.set_attach_data(attach_data.GetBuffer(), attach_data.GetLength());  
  49.                 CImPdu pdu2;  
  50.                 pdu2.SetPBMsg(&msg3);  
  51.                 pdu2.SetServiceId(SID_BUDDY_LIST);  
  52.                 pdu2.SetCommandId(CID_BUDDY_LIST_USERS_STATUS_REQUEST);  
  53.                 pdu2.SetSeqNum(pPdu->GetSeqNum());  
  54.                 CRouteServConn* route_conn = get_route_serv_conn();  
  55.                 if (route_conn)  
  56.                 {  
  57.                     route_conn->SendPdu(&pdu2);  
  58.                 }  
  59.             }  
  60.         }  
  61.     }  
  62.     else  
  63.     {  
  64.         log("HandleClientFileRequest, no file server.   ");  
  65.         IM::File::IMFileRsp msg2;  
  66.         msg2.set_result_code(1);  
  67.         msg2.set_from_user_id(from_id);  
  68.         msg2.set_to_user_id(to_id);  
  69.         msg2.set_file_name(file_name);  
  70.         msg2.set_task_id("");  
  71.         msg2.set_trans_mode((IM::BaseDefine::TransferFileType)trans_mode);  
  72.         CImPdu pdu;  
  73.         pdu.SetPBMsg(&msg2);  
  74.         pdu.SetServiceId(SID_FILE);  
  75.         pdu.SetCommandId(CID_FILE_RESPONSE);  
  76.         pdu.SetSeqNum(pPdu->GetSeqNum());  
  77.         pMsgConn->SendPdu(&pdu);  
  78.     }  
  79. }  

这段代码,颇有讲究,msg_server会检测file_server是否已经启动,若是没有启动,则直接发包告诉客户端,file_server不存在。另外,若是该文件传输模式是在线文件,会判断接收文件的用户是否和发送用户在同一台msg_server上。不在的话,则给route_server发送消息,查找该用户所在的msg_server(这个不具体介绍了,后面分析route_server会专门介绍的)。不然,会将文件发送请求转发给file_server,包的命令号是CID_OTHER_FILE_TRANSFER_REQ。file_server收到该请求后,处理以下:

[cpp]  view plain  copy
  1. void FileMsgServerConn::HandlePdu(CImPdu* pdu) {  
  2.       
  3.      ...  
  4.     //省略无关代码        
  5.     case CID_OTHER_FILE_TRANSFER_REQ:  
  6.         _HandleMsgFileTransferReq(pdu);  
  7.         break ;  
  8.     ...  
  9.     //省略无关代码          
  10. }  

[cpp]  view plain  copy
  1. void FileMsgServerConn::_HandleMsgFileTransferReq(CImPdu* pdu) {  
  2.     IM::Server::IMFileTransferReq transfer_req;  
  3.     CHECK_PB_PARSE_MSG(transfer_req.ParseFromArray(pdu->GetBodyData(), pdu->GetBodyLength()));  
  4.       
  5.       
  6.     uint32_t from_id = transfer_req.from_user_id();  
  7.     uint32_t to_id = transfer_req.to_user_id();  
  8.       
  9.     IM::Server::IMFileTransferRsp transfer_rsp;  
  10.     transfer_rsp.set_result_code(1);  
  11.     transfer_rsp.set_from_user_id(from_id);  
  12.     transfer_rsp.set_to_user_id(to_id);  
  13.     transfer_rsp.set_file_name(transfer_req.file_name());  
  14.     transfer_rsp.set_file_size(transfer_req.file_size());  
  15.     transfer_rsp.set_task_id("");  
  16.     transfer_rsp.set_trans_mode(transfer_req.trans_mode());  
  17.     transfer_rsp.set_attach_data(transfer_req.attach_data());  
  18.   
  19.       
  20.     bool rv = false;  
  21.     do {  
  22.         std::string task_id = GenerateUUID();  
  23.         if (task_id.empty()) {  
  24.             log("Create task id failed");  
  25.             break;  
  26.         }  
  27.         log("trams_mode=%d, task_id=%s, from_id=%d, to_id=%d, file_name=%s, file_size=%d", transfer_req.trans_mode(), task_id.c_str(), from_id, to_id, transfer_req.file_name().c_str(), transfer_req.file_size());  
  28.           
  29.         BaseTransferTask* transfer_task = TransferTaskManager::GetInstance()->NewTransferTask(  
  30.                                                                                               transfer_req.trans_mode(),  
  31.                                                                                               task_id,  
  32.                                                                                               from_id,  
  33.                                                                                               to_id,  
  34.                                                                                               transfer_req.file_name(),  
  35.                                                                                               transfer_req.file_size());  
  36.           
  37.         if (transfer_task == NULL) {  
  38.             // 建立未成功  
  39.             // close connection with msg svr  
  40.             // need_close = true;  
  41.             log("Create task failed");  
  42.             break;  
  43.         }  
  44.           
  45.         transfer_rsp.set_result_code(0);  
  46.         transfer_rsp.set_task_id(task_id);  
  47.         rv = true;  
  48.         // need_seq_no = false;  
  49.           
  50.         log("Create task succeed, task id %s, task type %d, from user %d, to user %d", task_id.c_str(), transfer_req.trans_mode(), from_id, to_id);  
  51.     } while (0);  
  52.       
  53.     ::SendMessageLite(this, SID_OTHER, CID_OTHER_FILE_TRANSFER_RSP, pdu->GetSeqNum(), &transfer_rsp);  
  54.       
  55.     if (!rv) {  
  56.         // 未建立成功,关闭链接  
  57.         Close();  
  58.     }  
  59. }  

上述代码会为本次传输任务建立一个惟一的标识uuid做为taskid,而后根据离线文件仍是在线文件建立离线文件传输任务OfflineTransferTask或者在线文件传输任务OnlineTransferTask,并加入一个一个成员变量transfer_tasks_中进行管理:

[cpp]  view plain  copy
  1. BaseTransferTask* TransferTaskManager::NewTransferTask(uint32_t trans_mode, const std::string& task_id, uint32_t from_user_id, uint32_t to_user_id, const std::string& file_name, uint32_t file_size) {  
  2.     BaseTransferTask* transfer_task = NULL;  
  3.       
  4.     TransferTaskMap::iterator it = transfer_tasks_.find(task_id);  
  5.     if (it==transfer_tasks_.end()) {  
  6.         if (trans_mode == IM::BaseDefine::FILE_TYPE_ONLINE) {  
  7.             transfer_task = new OnlineTransferTask(task_id, from_user_id, to_user_id, file_name, file_size);  
  8.         } else if (trans_mode == IM::BaseDefine::FILE_TYPE_OFFLINE) {  
  9.             transfer_task = new OfflineTransferTask(task_id, from_user_id, to_user_id, file_name, file_size);  
  10.         } else {  
  11.             log("Invalid trans_mode = %d", trans_mode);  
  12.         }  
  13.           
  14.         if (transfer_task) {  
  15.             transfer_tasks_.insert(std::make_pair(task_id, transfer_task));  
  16.         }  
  17.     } else {  
  18.         log("Task existed by task_id=%s, why?????", task_id.c_str());  
  19.     }  
  20.       
  21.     return transfer_task;  
  22. }  

这个map transfer_tasks_是在定时器里面进行按期处理的,处理的依据是当前任务的状态,好比已经完成的任务就能够从map中移除了:

[cpp]  view plain  copy
  1. void TransferTaskManager::OnTimer(uint64_t tick) {  
  2.     for (TransferTaskMap::iterator it = transfer_tasks_.begin(); it != transfer_tasks_.end();) {  
  3.         BaseTransferTask* task = it->second;  
  4.         if (task == NULL) {  
  5.             transfer_tasks_.erase(it++);  
  6.             continue;  
  7.         }  
  8.           
  9.         if (task->state() != kTransferTaskStateWaitingUpload &&  
  10.             task->state() == kTransferTaskStateTransferDone) {  
  11.             long esp = time(NULL) - task->create_time();  
  12.             if (esp > ConfigUtil::GetInstance()->GetTaskTimeout()) {  
  13.                 if (task->GetFromConn()) {  
  14.                     FileClientConn* conn = reinterpret_cast<FileClientConn*>(task->GetFromConn());  
  15.                     conn->ClearTransferTask();  
  16.                 }  
  17.                 if (task->GetToConn()) {  
  18.                     FileClientConn* conn = reinterpret_cast<FileClientConn*>(task->GetToConn());  
  19.                     conn->ClearTransferTask();  
  20.                 }  
  21.                 delete task;  
  22.                 transfer_tasks_.erase(it++);  
  23.                 continue;  
  24.             }  
  25.         }  
  26.           
  27.         ++it;  
  28.     }  
  29. }  

完成这些工做之后,组装的应答包命令号是CID_OTHER_FILE_TRANSFER_RSP,回复给msg_server。msg_server收到该应答包后处理:

[cpp]  view plain  copy
  1. void CFileServConn::HandlePdu(CImPdu* pPdu)  
  2. {  
  3.     switch (pPdu->GetCommandId()) {  
  4.         ...  
  5.     //省略无关代码  
  6.         case CID_OTHER_FILE_TRANSFER_RSP:  
  7.             _HandleFileMsgTransRsp(pPdu);  
  8.             break;  
  9.         ...  
  10.     //省略无关代码   
  11.     }  
  12. }  

[cpp]  view plain  copy
  1. void CFileServConn::_HandleFileMsgTransRsp(CImPdu* pPdu)  
  2. {  
  3.     IM::Server::IMFileTransferRsp msg;  
  4.     CHECK_PB_PARSE_MSG(msg.ParseFromArray(pPdu->GetBodyData(), pPdu->GetBodyLength()));  
  5.   
  6.     uint32_t result = msg.result_code();  
  7.     uint32_t from_id = msg.from_user_id();  
  8.     uint32_t to_id = msg.to_user_id();  
  9.     string file_name = msg.file_name();  
  10.     uint32_t file_size = msg.file_size();  
  11.     string task_id = msg.task_id();  
  12.     uint32_t trans_mode = msg.trans_mode();  
  13.     CDbAttachData attach((uchar_t*)msg.attach_data().c_str(), msg.attach_data().length());  
  14.     log("HandleFileMsgTransRsp, result: %u, from_user_id: %u, to_user_id: %u, file_name: %s, \  
  15.         task_id: %s, trans_mode: %u. ", result, from_id, to_id,  
  16.         file_name.c_str(), task_id.c_str(), trans_mode);  
  17.   
  18.     const list<IM::BaseDefine::IpAddr>* ip_addr_list = GetFileServerIPList();  
  19.   
  20.     IM::File::IMFileRsp msg2;  
  21.     msg2.set_result_code(result);  
  22.     msg2.set_from_user_id(from_id);  
  23.     msg2.set_to_user_id(to_id);  
  24.     msg2.set_file_name(file_name);  
  25.     msg2.set_task_id(task_id);  
  26.     msg2.set_trans_mode((IM::BaseDefine::TransferFileType)trans_mode);  
  27.     for (list<IM::BaseDefine::IpAddr>::const_iterator it = ip_addr_list->begin(); it != ip_addr_list->end(); it++)  
  28.     {  
  29.         IM::BaseDefine::IpAddr ip_addr_tmp = *it;  
  30.         IM::BaseDefine::IpAddr* ip_addr = msg2.add_ip_addr_list();  
  31.         ip_addr->set_ip(ip_addr_tmp.ip());  
  32.         ip_addr->set_port(ip_addr_tmp.port());  
  33.     }  
  34.     CImPdu pdu;  
  35.     pdu.SetPBMsg(&msg2);  
  36.     pdu.SetServiceId(SID_FILE);  
  37.     pdu.SetCommandId(CID_FILE_RESPONSE);  
  38.     pdu.SetSeqNum(pPdu->GetSeqNum());  
  39.     uint32_t handle = attach.GetHandle();  
  40.       
  41.     CMsgConn* pFromConn = CImUserManager::GetInstance()->GetMsgConnByHandle(from_id, handle);  
  42.     if (pFromConn)  
  43.     {  
  44.         pFromConn->SendPdu(&pdu);  
  45.     }  
  46.       
  47.     if (result == 0)  
  48.     {  
  49.         IM::File::IMFileNotify msg3;  
  50.         msg3.set_from_user_id(from_id);  
  51.         msg3.set_to_user_id(to_id);  
  52.         msg3.set_file_name(file_name);  
  53.         msg3.set_file_size(file_size);  
  54.         msg3.set_task_id(task_id);  
  55.         msg3.set_trans_mode((IM::BaseDefine::TransferFileType)trans_mode);  
  56.         msg3.set_offline_ready(0);  
  57.         for (list<IM::BaseDefine::IpAddr>::const_iterator it = ip_addr_list->begin(); it != ip_addr_list->end(); it++)  
  58.         {  
  59.             IM::BaseDefine::IpAddr ip_addr_tmp = *it;  
  60.             IM::BaseDefine::IpAddr* ip_addr = msg3.add_ip_addr_list();  
  61.             ip_addr->set_ip(ip_addr_tmp.ip());  
  62.             ip_addr->set_port(ip_addr_tmp.port());  
  63.         }  
  64.         CImPdu pdu2;  
  65.         pdu2.SetPBMsg(&msg3);  
  66.         pdu2.SetServiceId(SID_FILE);  
  67.         pdu2.SetCommandId(CID_FILE_NOTIFY);  
  68.           
  69.         //send notify to target user  
  70.         CImUser* pToUser = CImUserManager::GetInstance()->GetImUserById(to_id);  
  71.         if (pToUser)  
  72.         {  
  73.             pToUser->BroadcastPduWithOutMobile(&pdu2);  
  74.         }  
  75.           
  76.         //send to route server  
  77.         CRouteServConn* pRouteConn = get_route_serv_conn();  
  78.         if (pRouteConn) {  
  79.             pRouteConn->SendPdu(&pdu2);  
  80.         }  
  81.     }  
  82. }  

msg_server收到包后,首先装包数据,并把file_server的ip地址和端口信息带上,发给请求发文件的客户端,命令号是CID_FILE_RESPONSE;接着查询通知接收方有人给其发文件(通知方式也是同样,若是接收方在该msg_server上,直接发给该用户;不在的话,发给路由服务route_server)。固然接收到文件发送的端只有pc端,移动端会被过滤掉的,也就是说移动端不会收到发送文件的请求。

咱们先看发送方pc客户端收到应答的逻辑(命令号是CID_FILE_RESPONSE):函数

[cpp]  view plain  copy
  1. void FileTransferModule_Impl::onPacket(imcore::TTPBHeader& header, std::string& pbBody)  
  2. {  
  3.     switch (header.getCommandId())  
  4.     {  
  5.     case IM::BaseDefine::FileCmdID::CID_FILE_RESPONSE://发送“文件请求/离线文件”-返回  
  6.         _sendfileResponse(pbBody);  
  7.         break;    
  8.     }  
  9. }  

[cpp]  view plain  copy
  1. void FileTransferModule_Impl::_sendfileResponse(IN std::string& body)  
  2. {  
  3.     IM::File::IMFileRsp imFileRsp;  
  4.     if (!imFileRsp.ParseFromString(body))  
  5.     {  
  6.         LOG__(ERR, _T("parse failed,body:%s"), util::stringToCString(body));  
  7.         return;  
  8.     }  
  9.   
  10.     UInt32 nResult = imFileRsp.result_code();  
  11.     if (nResult != 0)  
  12.     {  
  13.         LOG__(ERR, _T("_sendfileResponse result != 0"));  
  14.         module::getFileTransferModule()->asynNotifyObserver(module::KEY_FILESEVER_UPLOAD_FAILED);  
  15.     }  
  16.   
  17.     TransferFileEntity fileEntity;  
  18.     fileEntity.sTaskID = imFileRsp.task_id();  
  19.     assert(!fileEntity.sTaskID.empty());  
  20.     fileEntity.sFromID = util::uint32ToString(imFileRsp.from_user_id());  
  21.     fileEntity.sToID = util::uint32ToString(imFileRsp.to_user_id());  
  22.     fileEntity.sFileName = imFileRsp.file_name();  
  23.     fileEntity.setSaveFilePath(util::stringToCString(fileEntity.sFileName));//发送方文件地址,就是保存地址  
  24.     fileEntity.time = static_cast<UInt32>(time(0));  
  25.     uint32_t transMode = imFileRsp.trans_mode();  
  26.     if (IM::BaseDefine::TransferFileType::FILE_TYPE_ONLINE == transMode)  
  27.     {  
  28.         fileEntity.nClientMode = IM::BaseDefine::ClientFileRole::CLIENT_REALTIME_SENDER;  
  29.     }  
  30.     else if (IM::BaseDefine::TransferFileType::FILE_TYPE_OFFLINE == transMode)  
  31.     {  
  32.         fileEntity.nClientMode = IM::BaseDefine::ClientFileRole::CLIENT_OFFLINE_UPLOAD;  
  33.     }  
  34.     fileEntity.pFileObject = new TransferFile(util::stringToCString(fileEntity.sFileName),FALSE);  
  35.     if (fileEntity.pFileObject)  
  36.     {  
  37.         fileEntity.nFileSize = fileEntity.pFileObject->length();  
  38.     }  
  39.       
  40.     UINT32 nIPCount = imFileRsp.ip_addr_list_size();  
  41.     if (nIPCount <= 0)  
  42.     {  
  43.         return;  
  44.     }  
  45.     const IM::BaseDefine::IpAddr& ipAdd = imFileRsp.ip_addr_list(0);  
  46.     fileEntity.sIP = ipAdd.ip();  
  47.     fileEntity.nPort = ipAdd.port();  
  48.   
  49.     if (!TransferFileEntityManager::getInstance()->pushTransferFileEntity(fileEntity))  
  50.         TransferFileEntityManager::getInstance()->updateFileInfoBysTaskID(fileEntity);  
  51.   
  52.     LOG__(DEBG, _T("FileTransferSevice_Impl::准备链接文件服务器 sTaskId = %s"), util::stringToCString(fileEntity.sTaskID));  
  53.     TransferFileEntityManager::getInstance()->openFileSocketByTaskId(fileEntity.sTaskID);  
  54. }  

客户端在TransferFileEntityManager::getInstance()->openFileSocketByTaskId(fileEntity.sTaskID);里面实际去链接file_server并尝试发文件:

[cpp]  view plain  copy
  1. void TransferFileEntityManager::openFileSocketByTaskId(std::string& taskId)  
  2. {  
  3.     m_fileUIThread->openFileSocketByTaskId(taskId);  
  4. }  

[cpp]  view plain  copy
  1. void FileTransferUIThread::openFileSocketByTaskId(std::string& taskId)  
  2. {  
  3.     FileTransferSocket* pFileSocket = _findFileSocketByTaskId(taskId);  
  4.     if (!pFileSocket)  
  5.     {  
  6.         pFileSocket = new FileTransferSocket(taskId);  
  7.         m_lstFileTransSockets.push_back(pFileSocket);  
  8.         assert(m_hWnd);  
  9.         ::PostMessage(m_hWnd, WM_FILE_TRANSFER, 0, (LPARAM)pFileSocket);  
  10.     }  
  11. }  

[cpp]  view plain  copy
  1. LRESULT _stdcall FileTransferUIThread::_WndProc(HWND hWnd, UINT message, WPARAM wparam, LPARAM lparam)  
  2. {  
  3.     if (WM_FILE_TRANSFER == message)  
  4.     {  
  5.         FileTransferSocket* pFileSocket = (FileTransferSocket*)lparam;  
  6.         pFileSocket->startFileTransLink();  
  7.     }  
  8.   
  9.     return ::DefWindowProc(hWnd, message, wparam, lparam);  
  10. }  

[cpp]  view plain  copy
  1. BOOL FileTransferSocket::startFileTransLink()  
  2. {  
  3.     TransferFileEntity FileInfo;  
  4.     if (TransferFileEntityManager::getInstance()->getFileInfoByTaskId(m_sTaskId, FileInfo))  
  5.     {  
  6.         //大佛:使用msg server 传过来的IP和端口  
  7.         LOG__(APP, _T("connect IP=%s,Port=%d"), util::stringToCString(FileInfo.sIP), FileInfo.nPort);  
  8.         connect(util::stringToCString(FileInfo.sIP), FileInfo.nPort);  
  9.         //connect(util::stringToCString(module::FILETRANSFER_IP), module::FILETRANSFER_PORT);  
  10.         return TRUE;  
  11.     }  
  12.     LOG__(ERR, _T("can't find the TaskId"));  
  13.     return FALSE;  
  14. }  

注意,这里只是去链接file_server服务器,链接成功的状况下,会尝试登陆文件服务器,登陆file_server的命令号是CID_FILE_LOGIN_REQ:

[cpp]  view plain  copy
  1. void FileTransferSocket::onConnectDone()  
  2. {  
  3.     LOG__(APP, _T("FileTransferSocket::onConnected()"));  
  4.     startHeartbeat();  
  5.   
  6.     TransferFileEntity info;  
  7.     if (!TransferFileEntityManager::getInstance()->getFileInfoByTaskId(m_sTaskId, info))  
  8.     {  
  9.         LOG__(APP, _T("Can't get the file info,task id:%s"),util::stringToCString(m_sTaskId));  
  10.         return;  
  11.     }  
  12.           
  13.     //拉模式文件传输,传输taskid、token、client_mode  
  14.     IM::File::IMFileLoginReq imFileLoginReq;  
  15.     imFileLoginReq.set_user_id(module::getSysConfigModule()->userId());  
  16.     imFileLoginReq.set_task_id(info.sTaskID);  
  17.     imFileLoginReq.set_file_role(static_cast<IM::BaseDefine::ClientFileRole>(info.nClientMode));  
  18.   
  19.     LOG__(APP, _T("IMFileLoginReq,sTaskID:%s,nClientMode:%d"), util::stringToCString(info.sTaskID), info.nClientMode);  
  20.     //send packet  
  21.     LOG__(APP, _T("IMFileLoginReq,taskId:%s"), util::stringToCString(info.sTaskID));  
  22.     sendPacket(IM::BaseDefine::ServiceID::SID_FILE, IM::BaseDefine::FileCmdID::CID_FILE_LOGIN_REQ, &imFileLoginReq);  
  23.   
  24.     //CImPduClientFileLoginReq pduFileLoginReq(module::getSysConfigModule()->userID().c_str()  
  25.     //  , "", info.sTaskID.c_str(), );  
  26.     //sendPacket(&pduFileLoginReq);  
  27. }  

file_server收到该数据包处理以下:

[cpp]  view plain  copy
  1. void FileClientConn::HandlePdu(CImPdu* pdu) {  
  2.    
  3.     ...  
  4.     //省略无关代码  
  5.         case CID_FILE_LOGIN_REQ:  
  6.             _HandleClientFileLoginReq(pdu);  
  7.             break;  
  8. }  

[cpp]  view plain  copy
  1. void FileClientConn::_HandleClientFileLoginReq(CImPdu* pdu) {  
  2.     IM::File::IMFileLoginReq login_req;  
  3.     CHECK_PB_PARSE_MSG(login_req.ParseFromArray(pdu->GetBodyData(), pdu->GetBodyLength()));  
  4.       
  5.     uint32_t user_id = login_req.user_id();  
  6.     string task_id = login_req.task_id();  
  7.     IM::BaseDefine::ClientFileRole mode = login_req.file_role();  
  8.       
  9.     log("Client login, user_id=%d, task_id=%s, file_role=%d", user_id, task_id.c_str(), mode);  
  10.       
  11.     BaseTransferTask* transfer_task = NULL;  
  12.       
  13.     bool rv = false;  
  14.     do {  
  15.         // 查找任务是否存在  
  16.         transfer_task = TransferTaskManager::GetInstance()->FindByTaskID(task_id);  
  17.           
  18.         if (transfer_task == NULL) {  
  19.             if (mode == CLIENT_OFFLINE_DOWNLOAD) {  
  20.                 // 文件不存在,检查是不是离线下载,有多是文件服务器重启  
  21.                 // 尝试从磁盘加载  
  22.                 transfer_task = TransferTaskManager::GetInstance()->NewTransferTask(task_id, user_id);  
  23.                 // 须要再次判断是否加载成功  
  24.                 if (transfer_task == NULL) {  
  25.                     log("Find task id failed, user_id=%u, taks_id=%s, mode=%d", user_id, task_id.c_str(), mode);  
  26.                     break;  
  27.                 }  
  28.             } else {  
  29.                 log("Can't find task_id, user_id=%u, taks_id=%s, mode=%d", user_id, task_id.c_str(), mode);  
  30.                 break;  
  31.             }  
  32.         }  
  33.           
  34.         // 状态转换  
  35.         rv = transfer_task->ChangePullState(user_id, mode);  
  36.         if (!rv) {  
  37.             // log();  
  38.             break;  
  39.             //  
  40.         }  
  41.           
  42.         // Ok  
  43.         auth_ = true;  
  44.         transfer_task_ = transfer_task;  
  45.         user_id_ = user_id;  
  46.         // 设置conn  
  47.         transfer_task->SetConnByUserID(user_id, this);  
  48.         rv = true;  
  49.           
  50.     } while (0);  
  51.   
  52.     IM::File::IMFileLoginRsp login_rsp;  
  53.     login_rsp.set_result_code(rv?0:1);  
  54.     login_rsp.set_task_id(task_id);  
  55.   
  56.     ::SendMessageLite(this, SID_FILE, CID_FILE_LOGIN_RES, pdu->GetSeqNum(), &login_rsp);  
  57.   
  58.     if (rv) {  
  59.         if (transfer_task->GetTransMode() == FILE_TYPE_ONLINE) {  
  60.             if (transfer_task->state() == kTransferTaskStateWaitingTransfer) {  
  61.                 CImConn* conn = transfer_task_->GetToConn();  
  62.                 if (conn) {  
  63.                     _StatesNotify(CLIENT_FILE_PEER_READY, task_id, transfer_task_->from_user_id(), conn);  
  64.                 } else {  
  65.                     log("to_conn is close, close me!!!");  
  66.                     Close();  
  67.                 }  
  68.                 // _StatesNotify(CLIENT_FILE_PEER_READY, task_id, user_id, this);  
  69.                 // transfer_task->StatesNotify(CLIENT_FILE_PEER_READY, task_id, user_id_);  
  70.             }  
  71.         } else {  
  72.             if (transfer_task->state() == kTransferTaskStateWaitingUpload) {  
  73.                   
  74.                 OfflineTransferTask* offline = reinterpret_cast<OfflineTransferTask*>(transfer_task);  
  75.                   
  76.                 IM::File::IMFilePullDataReq pull_data_req;  
  77.                 pull_data_req.set_task_id(task_id);  
  78.                 pull_data_req.set_user_id(user_id);  
  79.                 pull_data_req.set_trans_mode(FILE_TYPE_OFFLINE);  
  80.                 pull_data_req.set_offset(0);  
  81.                 pull_data_req.set_data_size(offline->GetNextSegmentBlockSize());  
  82.                   
  83.                 ::SendMessageLite(this, SID_FILE, CID_FILE_PULL_DATA_REQ, &pull_data_req);  
  84.   
  85.                 log("Pull Data Req");  
  86.             }  
  87.         }  
  88.     } else {  
  89.         Close();  
  90.     }  
  91. }  


file_server应答客户端的命令号是CID_FILE_LOGIN_RES,客户端收到该包后处理以下:ui

[cpp]  view plain  copy
  1. void FileTransferSocket::onReceiveData(const char* data, int32_t size)  
  2. {  
  3.     std::string pbBody;  
  4.     imcore::TTPBHeader pbHeader;   
  5.     try  
  6.     {  
  7.         pbHeader.unSerialize((byte*)data, imcore::HEADER_LENGTH);  
  8.         pbBody.assign(data + imcore::HEADER_LENGTH, size - imcore::HEADER_LENGTH);  
  9.   
  10.         if (IM::BaseDefine::OtherCmdID::CID_OTHER_HEARTBEAT == pbHeader.getCommandId()  
  11.             && IM::BaseDefine::ServiceID::SID_OTHER == pbHeader.getModuleId())  
  12.             return;  
  13.     }  
  14.     catch (CPduException e)  
  15.     {  
  16.         LOG__(ERR, _T("onPacket CPduException serviceId:%d,commandId:%d,errCode:%d")  
  17.             , e.GetModuleId(), e.GetCommandId(), e.GetErrorCode());  
  18.         return;  
  19.     }  
  20.     catch (...)  
  21.     {  
  22.         LOG__(ERR, _T("FileTransferSocket onPacket unknown exception"));  
  23.         return;  
  24.     }  
  25.     UInt16 ncmdid = pbHeader.getCommandId();  
  26.     switch (ncmdid)  
  27.     {      
  28.     case IM::BaseDefine::FileCmdID::CID_FILE_LOGIN_RES:  
  29.         _fileLoginResponse(pbBody);  
  30.         break;  
  31.     //无关代码省略  
  32.     }  
  33. }  
[cpp]  view plain  copy
  1. void FileTransferSocket::_fileLoginResponse(IN std::string& body)  
  2. {  
  3.     IM::File::IMFileLoginRsp imFileLoginRsp;  
  4.     if (!imFileLoginRsp.ParseFromString(body))  
  5.     {  
  6.         LOG__(ERR, _T("parse failed,body:%s"), util::stringToCString(body));  
  7.         return;  
  8.     }  
  9.     if (imFileLoginRsp.result_code() != 0)  
  10.     {  
  11.         LOG__(ERR, _T("file server login failed! "));  
  12.         return;  
  13.     }  
  14.     //打开文件  
  15.     std::string taskId = imFileLoginRsp.task_id();  
  16.     TransferFileEntity fileEntity;  
  17.     if (!TransferFileEntityManager::getInstance()->getFileInfoByTaskId(taskId, fileEntity))  
  18.     {  
  19.         LOG__(ERR, _T("file server login:can't find the fileInfo "));  
  20.         return;  
  21.     }  
  22.   
  23.     LOG__(APP, _T("IMFileLoginRsp, file server login succeed"));  
  24.     //提示界面,界面上插入该项  
  25.     if (IM::BaseDefine::ClientFileRole::CLIENT_REALTIME_SENDER == fileEntity.nClientMode  
  26.         || IM::BaseDefine::ClientFileRole::CLIENT_OFFLINE_UPLOAD == fileEntity.nClientMode)  
  27.     {  
  28.         module::getFileTransferModule()->asynNotifyObserver(module::KEY_FILETRANSFER_SENDFILE, fileEntity.sTaskID);  
  29.     }  
  30.     else if (IM::BaseDefine::ClientFileRole::CLIENT_REALTIME_RECVER == fileEntity.nClientMode  
  31.         || IM::BaseDefine::ClientFileRole::CLIENT_OFFLINE_DOWNLOAD == fileEntity.nClientMode)  
  32.     {  
  33.         module::getFileTransferModule()->asynNotifyObserver(module::KEY_FILETRANSFER_REQUEST, fileEntity.sTaskID);  
  34.     }  
  35. }  

至此,不论是离线文件仍是在线文件发送,pc客户端会显示一个文件进度的对话框:




对于在线文件,须要对端赞成接收文件的传输,客户端才会读取文件,这个进度条才会发生变化。而对于离线文件,应该会立马读取文件上传文件数据到服务器。但是哪里会触发客户端读取文件并发送的逻辑呢?门道在于file_server在收到登陆请求CID_FILE_LOGIN_REQ后,不只会给客户端发送登陆应答数据包CID_FILE_LOGIN_RES。还会根据文件的传输模式,若是是离线文件则会给客户端发送拉取文件的数据包CID_FILE_PULL_DATA_REQ,代码咱们已经在上面的FileClientConn::_HandleClientFileLoginReq(CImPdu* pdu)中贴过了,咱们再贴一次:this

[cpp]  view plain  copy
  1. void FileClientConn::_HandleClientFileLoginReq(CImPdu* pdu) {  
  2.     IM::File::IMFileLoginReq login_req;  
  3.     CHECK_PB_PARSE_MSG(login_req.ParseFromArray(pdu->GetBodyData(), pdu->GetBodyLength()));  
  4.       
  5.     uint32_t user_id = login_req.user_id();  
  6.     string task_id = login_req.task_id();  
  7.     IM::BaseDefine::ClientFileRole mode = login_req.file_role();  
  8.       
  9.     log("Client login, user_id=%d, task_id=%s, file_role=%d", user_id, task_id.c_str(), mode);  
  10.       
  11.     BaseTransferTask* transfer_task = NULL;  
  12.       
  13.     bool rv = false;  
  14.     do {  
  15.         // 查找任务是否存在  
  16.         transfer_task = TransferTaskManager::GetInstance()->FindByTaskID(task_id);  
  17.           
  18.         if (transfer_task == NULL) {  
  19.             if (mode == CLIENT_OFFLINE_DOWNLOAD) {  
  20.                 // 文件不存在,检查是不是离线下载,有多是文件服务器重启  
  21.                 // 尝试从磁盘加载  
  22.                 transfer_task = TransferTaskManager::GetInstance()->NewTransferTask(task_id, user_id);  
  23.                 // 须要再次判断是否加载成功  
  24.                 if (transfer_task == NULL) {  
  25.                     log("Find task id failed, user_id=%u, taks_id=%s, mode=%d", user_id, task_id.c_str(), mode);  
  26.                     break;  
  27.                 }  
  28.             } else {  
  29.                 log("Can't find task_id, user_id=%u, taks_id=%s, mode=%d", user_id, task_id.c_str(), mode);  
  30.                 break;  
  31.             }  
  32.         }  
  33.           
  34.         // 状态转换  
  35.         rv = transfer_task->ChangePullState(user_id, mode);  
  36.         if (!rv) {  
  37.             // log();  
  38.             break;  
  39.             //  
  40.         }  
  41.           
  42.         // Ok  
  43.         auth_ = true;  
  44.         transfer_task_ = transfer_task;  
  45.         user_id_ = user_id;  
  46.         // 设置conn  
  47.         transfer_task->SetConnByUserID(user_id, this);  
  48.         rv = true;  
  49.           
  50.     } while (0);  
  51.   
  52.     IM::File::IMFileLoginRsp login_rsp;  
  53.     login_rsp.set_result_code(rv?0:1);  
  54.     login_rsp.set_task_id(task_id);  
  55.   
  56.     ::SendMessageLite(this, SID_FILE, CID_FILE_LOGIN_RES, pdu->GetSeqNum(), &login_rsp);  
  57.   
  58.     if (rv) {  
  59.         if (transfer_task->GetTransMode() == FILE_TYPE_ONLINE) {  
  60.             if (transfer_task->state() == kTransferTaskStateWaitingTransfer) {  
  61.                 CImConn* conn = transfer_task_->GetToConn();  
  62.                 if (conn) {  
  63.                     _StatesNotify(CLIENT_FILE_PEER_READY, task_id, transfer_task_->from_user_id(), conn);  
  64.                 } else {  
  65.                     log("to_conn is close, close me!!!");  
  66.                     Close();  
  67.                 }  
  68.                 // _StatesNotify(CLIENT_FILE_PEER_READY, task_id, user_id, this);  
  69.                 // transfer_task->StatesNotify(CLIENT_FILE_PEER_READY, task_id, user_id_);  
  70.             }  
  71.         } else {  
  72.             if (transfer_task->state() == kTransferTaskStateWaitingUpload) {  
  73.                   
  74.                 OfflineTransferTask* offline = reinterpret_cast<OfflineTransferTask*>(transfer_task);  
  75.                   
  76.                 IM::File::IMFilePullDataReq pull_data_req;  
  77.                 pull_data_req.set_task_id(task_id);  
  78.                 pull_data_req.set_user_id(user_id);  
  79.                 pull_data_req.set_trans_mode(FILE_TYPE_OFFLINE);  
  80.                 pull_data_req.set_offset(0);  
  81.                 pull_data_req.set_data_size(offline->GetNextSegmentBlockSize());  
  82.                   
  83.                 ::SendMessageLite(this, SID_FILE, CID_FILE_PULL_DATA_REQ, &pull_data_req);  
  84.   
  85.                 log("Pull Data Req");  
  86.             }  
  87.         }  
  88.     } else {  
  89.         Close();  
  90.     }  
  91. }  

pc端收到CID_FILE_PULL_DATA_REQ后,表示这是一个离线文件,就能够直接上传文件数据了:

[cpp]  view plain  copy
  1. case IM::BaseDefine::FileCmdID::CID_FILE_PULL_DATA_REQ://发文件  
  2.         _filePullDataReqResponse(pbBody);  

[cpp]  view plain  copy
  1. void FileTransferSocket::_filePullDataReqResponse(IN std::string& body)//发  
  2. {  
  3.     IM::File::IMFilePullDataReq imFilePullDataReq;  
  4.     if (!imFilePullDataReq.ParseFromString(body))  
  5.     {  
  6.         LOG__(ERR, _T("parse failed,body:%s"), util::stringToCString(body));  
  7.         return;  
  8.     }  
  9.     UInt32 fileSize = imFilePullDataReq.data_size();  
  10.     UInt32 fileOffset = imFilePullDataReq.offset();  
  11.     std::string taskId = imFilePullDataReq.task_id();  
  12.       
  13.     TransferFileEntity fileEntity;  
  14.     if (!TransferFileEntityManager::getInstance()->getFileInfoByTaskId(taskId, fileEntity))  
  15.     {  
  16.         LOG__(ERR, _T("PullDataReqResponse: can't find the fileInfo"));  
  17.         return;  
  18.     }  
  19.     LOG__(DEBG, _T("send:taskId=%s,filesize=%d,name=%s,BolckSize=%d")  
  20.         ,util::stringToCString(fileEntity.sTaskID)  
  21.         ,fileEntity.nFileSize  
  22.         ,fileEntity.getRealFileName()  
  23.         ,fileSize);  
  24.     std::string buff;  
  25.     if (nullptr == fileEntity.pFileObject)  
  26.     {  
  27.         LOG__(ERR, _T("PullDataReqResponse: file boject Destoryed!"));  
  28.         return;  
  29.     }  
  30.     fileEntity.pFileObject->readBlock(fileOffset, fileSize, buff);//读取本地文件的数据块  
  31.     IM::File::IMFilePullDataRsp imFilePullDataRsp;//todo check  
  32.     imFilePullDataRsp.set_result_code(0);  
  33.     imFilePullDataRsp.set_task_id(taskId);  
  34.     imFilePullDataRsp.set_user_id(util::stringToInt32(fileEntity.sFromID));  
  35.     imFilePullDataRsp.set_offset(fileOffset);  
  36.     imFilePullDataRsp.set_file_data((void*)buff.data(), fileSize);  
  37.   
  38.     //send packet  
  39.     sendPacket(IM::BaseDefine::ServiceID::SID_FILE, IM::BaseDefine::FileCmdID::CID_FILE_PULL_DATA_RSP  
  40.         , &imFilePullDataRsp);  
  41.   
  42.     fileEntity.nProgress = fileOffset + fileSize;  
  43.     if (fileEntity.nProgress < fileEntity.nFileSize)  
  44.     {  
  45.         //更新进度条  
  46.         TransferFileEntityManager::getInstance()->updateFileInfoBysTaskID(fileEntity);//保存当前进度  
  47.         module::getFileTransferModule()->asynNotifyObserver(module::KEY_FILESEVER_UPDATA_PROGRESSBAR  
  48.             , fileEntity.sTaskID);  
  49.     }  
  50.     else//传输完成  
  51.     {  
  52.         if (fileEntity.pFileObject)  
  53.         {  
  54.             delete fileEntity.pFileObject;  
  55.             fileEntity.pFileObject = nullptr;  
  56.         }  
  57.         module::getFileTransferModule()->asynNotifyObserver(module::KEY_FILESEVER_PROGRESSBAR_FINISHED  
  58.             , fileEntity.sTaskID);  
  59.     }  
  60.     TransferFileEntityManager::getInstance()->updateFileInfoBysTaskID(fileEntity);  
  61. }  

固然,若是文件比较大,一次发不完也不要紧,在CID_FILE_PULL_DATA_REQ中有当前文件的偏移量,客户端在读取文件和应答服务器时也带上这个偏移量fileOffset,应答给服务器的包是CID_FILE_PULL_DATA_RSP。file_server收到应答后处理:

[cpp]  view plain  copy
  1. void FileClientConn::HandlePdu(CImPdu* pdu) {  
  2.       
  3.             ...  
  4.     //省略无关代码  
  5.         case CID_FILE_PULL_DATA_RSP:  
  6.             _HandleClientFilePullFileRsp( pdu);  
  7.             break ;  
  8.                 ...  
  9.     //省略无关代码  
  10.   
  11. }  


[cpp]  view plain  copy
  1. void FileClientConn::_HandleClientFilePullFileRsp(CImPdu *pdu) {  
  2.     if (!auth_ || !transfer_task_) {  
  3.         log("auth is false");  
  4.         return;  
  5.     }  
  6.   
  7.     // 只有rsp  
  8.     IM::File::IMFilePullDataRsp pull_data_rsp;  
  9.     CHECK_PB_PARSE_MSG(pull_data_rsp.ParseFromArray(pdu->GetBodyData(), pdu->GetBodyLength()));  
  10.       
  11.     uint32_t user_id = pull_data_rsp.user_id();  
  12.     string task_id = pull_data_rsp.task_id();  
  13.     uint32_t offset = pull_data_rsp.offset();  
  14.     uint32_t data_size = static_cast<uint32_t>(pull_data_rsp.file_data().length());  
  15.     const char* data = pull_data_rsp.file_data().data();  
  16.   
  17.     // log("Recv FilePullFileRsp, user_id=%d, task_id=%s, file_role=%d, offset=%d, datasize=%d", user_id, task_id.c_str(), mode, offset, datasize);  
  18.     log("Recv FilePullFileRsp, task_id=%s, user_id=%u, offset=%u, data_size=%d", task_id.c_str(), user_id, offset, data_size);  
  19.   
  20.     int rv = -1;  
  21.     do {  
  22.         //  
  23.         // 检查user_id  
  24.         if (user_id != user_id_) {  
  25.             log("Received user_id valid, recv_user_id = %d, transfer_task.user_id = %d, user_id_ = %d", user_id, transfer_task_->from_user_id(), user_id_);  
  26.             break;  
  27.         }  
  28.           
  29.         // 检查task_id  
  30.         if (transfer_task_->task_id() != task_id) {  
  31.             log("Received task_id valid, recv_task_id = %s, this_task_id = %s", task_id.c_str(), transfer_task_->task_id().c_str());  
  32.             // Close();  
  33.             break;  
  34.         }  
  35.           
  36.         rv = transfer_task_->DoRecvData(user_id, offset, data, data_size);  
  37.         if (rv == -1) {  
  38.             break;  
  39.         }  
  40.           
  41.         if (transfer_task_->GetTransMode() == FILE_TYPE_ONLINE) {  
  42.             // 对于在线,直接转发  
  43.             OnlineTransferTask* online = reinterpret_cast<OnlineTransferTask*>(transfer_task_);  
  44.             pdu->SetSeqNum(online->GetSeqNum());  
  45.             // online->SetSeqNum(pdu->GetSeqNum());  
  46.   
  47.             CImConn* conn = transfer_task_->GetToConn();  
  48.             if (conn) {  
  49.                 conn->SendPdu(pdu);  
  50.             }  
  51.         } else {  
  52.             // 离线  
  53.             // all packages recved  
  54.             if (rv == 1) {  
  55.                 _StatesNotify(CLIENT_FILE_DONE, task_id, user_id, this);  
  56.                 // Close();  
  57.             } else {  
  58.                 OfflineTransferTask* offline = reinterpret_cast<OfflineTransferTask*>(transfer_task_);  
  59.                   
  60.                 IM::File::IMFilePullDataReq pull_data_req;  
  61.                 pull_data_req.set_task_id(task_id);  
  62.                 pull_data_req.set_user_id(user_id);  
  63.                 pull_data_req.set_trans_mode(static_cast<IM::BaseDefine::TransferFileType>(offline->GetTransMode()));  
  64.                 pull_data_req.set_offset(offline->GetNextOffset());  
  65.                 pull_data_req.set_data_size(offline->GetNextSegmentBlockSize());  
  66.                   
  67.                 ::SendMessageLite(this, SID_FILE, CID_FILE_PULL_DATA_REQ, &pull_data_req);  
  68.                 // log("size not match");  
  69.             }  
  70.         }  
  71.           
  72.     } while (0);  
  73.       
  74.     if (rv!=0) {  
  75.         // -1,出错关闭  
  76.         //  1, 离线上传完成  
  77.         Close();  
  78.     }  
  79. }  

若是是在线文件,就直接转发含有文件数据的包;若是是离线文件,则存入文件服务上,即写入文件:

[cpp]  view plain  copy
  1. int OfflineTransferTask::DoRecvData(uint32_t user_id, uint32_t offset, const char* data, uint32_t data_size) {  
  2.     // 离线文件上传  
  3.       
  4.     int rv = -1;  
  5.       
  6.     do {  
  7.         // 检查是否发送者  
  8.         if (!CheckFromUserID(user_id)) {  
  9.             log("rsp user_id=%d, but sender_id is %d", user_id, from_user_id_);  
  10.             break;  
  11.         }  
  12.           
  13.         // 检查状态  
  14.         if (state_ != kTransferTaskStateWaitingUpload && state_ != kTransferTaskStateUploading) {  
  15.             log("state=%d error, need kTransferTaskStateWaitingUpload or kTransferTaskStateUploading", state_);  
  16.             break;  
  17.         }  
  18.           
  19.         // 检查offset是否有效  
  20.         if (offset != transfered_idx_*SEGMENT_SIZE) {  
  21.             break;  
  22.         }  
  23.           
  24.         //if (data_size != GetNextSegmentBlockSize()) {  
  25.         //    break;  
  26.         //}  
  27.         // todo  
  28.         // 检查文件大小  
  29.           
  30.         data_size = GetNextSegmentBlockSize();  
  31.         log("Ready recv data, offset=%d, data_size=%d, segment_size=%d", offset, data_size, sengment_size_);  
  32.           
  33.         if (state_ == kTransferTaskStateWaitingUpload) {  
  34.             if (fp_ == NULL) {  
  35.                 fp_ = OpenByWrite(task_id_, to_user_id_);  
  36.                 if (fp_ == NULL) {  
  37.                     break;  
  38.                 }  
  39.             }  
  40.   
  41.             // 写文件头  
  42.             OfflineFileHeader file_header;  
  43.             memset(&file_header, 0, sizeof(file_header));  
  44.             file_header.set_create_time(time(NULL));  
  45.             file_header.set_task_id(task_id_);  
  46.             file_header.set_from_user_id(from_user_id_);  
  47.             file_header.set_to_user_id(to_user_id_);  
  48.             file_header.set_file_name("");  
  49.             file_header.set_file_size(file_size_);  
  50.             fwrite(&file_header, 1, sizeof(file_header), fp_);  
  51.             fflush(fp_);  
  52.   
  53.             state_ = kTransferTaskStateUploading;  
  54.         }  
  55.           
  56.         // 存储  
  57.         if (fp_ == NULL) {  
  58.             //  
  59.             break;  
  60.         }  
  61.           
  62.         fwrite(data, 1, data_size, fp_);  
  63.         fflush(fp_);  
  64.   
  65.         ++transfered_idx_;  
  66.         SetLastUpdateTime();  
  67.   
  68.         if (transfered_idx_ == sengment_size_) {  
  69.             state_ = kTransferTaskStateUploadEnd;  
  70.             fclose(fp_);  
  71.             fp_ = NULL;  
  72.             rv = 1;  
  73.         } else {  
  74.             rv = 0;  
  75.         }  
  76.     } while (0);  
  77.       
  78.     return rv;  
  79. }  

如此循环,直至文件传输完成。固然文件上传完成后file_server也会断开与客户端的链接。

到这里咱们介绍了发送文件方的逻辑,下面咱们看看接收方的逻辑,上文中介绍了接收方会收到接收文件的通知CID_FILE_NOTIFY,客户端处理这个命令号:spa

[cpp]  view plain  copy
  1. case IM::BaseDefine::FileCmdID::CID_FILE_NOTIFY://收到“发送文件请求”  
  2. <span style="white-space:pre">  </span>_fileNotify(pbBody);  
[cpp]  view plain  copy
  1. void FileTransferModule_Impl::_fileNotify(IN std::string& body)  
  2. {  
  3.     IM::File::IMFileNotify imFileNotify;  
  4.     if (!imFileNotify.ParseFromString(body))  
  5.     {  
  6.         LOG__(ERR, _T("parse failed,body:%s"), util::stringToCString(body));  
  7.         return;  
  8.     }  
  9.     TransferFileEntity file;  
  10.     file.sFileName = imFileNotify.file_name();  
  11.     file.sFromID = util::uint32ToString(imFileNotify.from_user_id());  
  12.     file.sToID = util::uint32ToString(imFileNotify.to_user_id());  
  13.     file.sTaskID = imFileNotify.task_id();  
  14.     file.nFileSize = imFileNotify.file_size();  
  15.   
  16.     UINT32 nIPCount = imFileNotify.ip_addr_list_size();  
  17.     if (nIPCount <= 0)  
  18.     {  
  19.         return;  
  20.     }  
  21.     const IM::BaseDefine::IpAddr& ipAdd = imFileNotify.ip_addr_list(0);  
  22.     file.sIP = ipAdd.ip();  
  23.     file.nPort = ipAdd.port();  
  24.   
  25.     uint32_t transMode = imFileNotify.trans_mode();  
  26.     if (IM::BaseDefine::TransferFileType::FILE_TYPE_ONLINE == transMode)  
  27.     {  
  28.         file.nClientMode = IM::BaseDefine::ClientFileRole::CLIENT_REALTIME_RECVER;  
  29.     }  
  30.     else if (IM::BaseDefine::TransferFileType::FILE_TYPE_OFFLINE == transMode)  
  31.     {  
  32.         file.nClientMode = IM::BaseDefine::ClientFileRole::CLIENT_OFFLINE_DOWNLOAD;  
  33.     }  
  34.     file.time = static_cast<UInt32>(time(0));  
  35.     TransferFileEntityManager::getInstance()->pushTransferFileEntity(file);  
  36.     LOG__(DEBG, _T("FileTransferSevice_Impl::给你发文件 sFileID = %s"), util::stringToCString(file.sTaskID));  
  37.   
  38.     if (1 == imFileNotify.offline_ready())  
  39.     {  
  40.         //TODO离线文件传输结束  
  41.     }  
  42.   
  43.     //链接服务器  
  44.     TransferFileEntityManager::getInstance()->openFileSocketByTaskId(file.sTaskID);  
  45. }  

其实也就是接收方会去链接文件服务器。链接成功之后,在对应的回调函数里面触发显示接收文件对话框。可是此时实际上还不能接收文件,由于发送方可能还没准备好。发送方要准备啥呢?前面咱们已经介绍了,咱们梳理一下上述流程:

1. 发送方先向msg_server请求发送文件,msg_server转发给file_server;.net

2. file_server应答msg_server并告诉msg_server本身的地址和端口号;

3. msg_server收到file_server的应答后,先回复发送方,再转发给接收方;

4. 发送方接着发送登陆请求给file_server,file_server收到请求决定是否给发送方发送拉取文件的数据包。若是是离线文件,则会马上给发送方发送拉取文件的数据包;若是是在线文件,则须要等待接收方赞成接收。


因此,必须过了步骤4,一直到file_server应答了发送方的登陆文件服务器请求后,发送方才算准备好。此时,file_server知道发送方已经准备好了,给接收方发送数据包CID_FILE_STATE。接收方收到这个命令号后:

[cpp]  view plain  copy
  1. case IM::BaseDefine::FileCmdID::CID_FILE_STATE://  
  2.         _fileState(pbBody);  

[cpp]  view plain  copy
  1. void FileTransferSocket::_fileState(IN std::string& body)  
  2. {  
  3.     IM::File::IMFileState imFileState;  
  4.     if (!imFileState.ParseFromString(body))  
  5.     {  
  6.         LOG__(ERR, _T("parse failed,body:%s"), util::stringToCString(body));  
  7.         return;  
  8.     }  
  9.     UINT32 nfileState = imFileState.state();  
  10.   
  11.     std::string taskId = imFileState.task_id();  
  12.     TransferFileEntity fileEntity;  
  13.     if (!TransferFileEntityManager::getInstance()->getFileInfoByTaskId(taskId, fileEntity))  
  14.     {  
  15.         LOG__(ERR, _T("fileState:can't find the fileInfo "));  
  16.         return;  
  17.     }  
  18.   
  19.     switch (nfileState)  
  20.     {  
  21.     case IM::BaseDefine::ClientFileState::CLIENT_FILE_PEER_READY:  
  22.         LOG__(APP, _T("fileState--CLIENT_FILE_PEER_READY "));  
  23.         break;  
  24.     case IM::BaseDefine::ClientFileState::CLIENT_FILE_CANCEL ://取消的了文件传输  
  25.         LOG__(APP, _T("fileState--CLIENT_FILE_CANCEL "));  
  26.         {  
  27.             if (fileEntity.pFileObject)  
  28.             {  
  29.                 delete fileEntity.pFileObject;  
  30.                 fileEntity.pFileObject = nullptr;  
  31.             }  
  32.             TransferFileEntityManager::getInstance()->updateFileInfoBysTaskID(fileEntity);  
  33.             module::getFileTransferModule()->asynNotifyObserver(module::KEY_FILESEVER_UPLOAD_CANCEL, fileEntity.sTaskID);  
  34.         }  
  35.         break;  
  36.     case IM::BaseDefine::ClientFileState::CLIENT_FILE_REFUSE://拒绝了文件  
  37.         LOG__(APP, _T("fileState--CLIENT_FILE_REFUSE "));  
  38.         {  
  39.             if (fileEntity.pFileObject)  
  40.             {  
  41.                 delete fileEntity.pFileObject;  
  42.                 fileEntity.pFileObject = nullptr;  
  43.             }  
  44.             TransferFileEntityManager::getInstance()->updateFileInfoBysTaskID(fileEntity);  
  45.             module::getFileTransferModule()->asynNotifyObserver(module::KEY_FILESEVER_UPLOAD_REJECT, fileEntity.sTaskID);  
  46.         }  
  47.         break;  
  48.     case IM::BaseDefine::ClientFileState::CLIENT_FILE_DONE:  
  49.         LOG__(APP, _T("fileState--CLIENT_FILE_DONE "));  
  50.         if (fileEntity.pFileObject)  
  51.         {  
  52.             delete fileEntity.pFileObject;  
  53.             fileEntity.pFileObject = nullptr;  
  54.         }  
  55.         TransferFileEntityManager::getInstance()->updateFileInfoBysTaskID(fileEntity);  
  56.         module::getFileTransferModule()->asynNotifyObserver(module::KEY_FILESEVER_PROGRESSBAR_FINISHED, fileEntity.sTaskID);  
  57.         break;  
  58.     default:  
  59.         break;  
  60.     }  
  61. }  

同理,对于接收方,选择接收仍是拒绝文件的逻辑也是在这里一块儿处理的,与此相似,这里就再也不重复叙述了。

接收方下载文件的逻辑和发送方上传文件的逻辑相似。这里也不在描述了。


最后说一点个人建议,teamtalk的file_server逻辑、以及与客户端还有msg_server的逻辑流程加上各类细节写的比较的细腻,代码实现上也比较好。强烈建议好好地阅读这部分的代码。毕竟不少人在本身实现一个文件服务器时,仍是存在很多问题的。


zhangyl 2017.05.20 

文中若是存在说的不对的地方,欢迎批评指正。

1、链接情况介绍

fileserver开始并非和客户端链接的,客户端是按需链接file_server的。可是file_server与msg_server倒是长链接。先启动file_server,再启动msg_server。msg_server初始化的时候,会去尝试链接file_server的8601端口。链接成功之后,会给file_server发送一个发包询问file_server侦听客户端链接的ip和端口号信息:

[cpp]  view plain  copy
  1. void CFileServConn::OnConfirm()  
  2. {  
  3.     log("connect to file server success ");  
  4.     m_bOpen = true;  
  5.     m_connect_time = get_tick_count();  
  6.     g_file_server_list[m_serv_idx].reconnect_cnt = MIN_RECONNECT_CNT / 2;  
  7.       
  8.     //连上file_server之后,给file_server发送获取ip地址的数据包  
  9.     IM::Server::IMFileServerIPReq msg;  
  10.     CImPdu pdu;  
  11.     pdu.SetPBMsg(&msg);  
  12.     pdu.SetServiceId(SID_OTHER);  
  13.     pdu.SetCommandId(CID_OTHER_FILE_SERVER_IP_REQ);  
  14.     SendPdu(&pdu);  
  15. }  

file_server收到该数据包后,将本身的侦听客户端链接的ip地址和端口号发包告诉msg_server:

[cpp]  view plain  copy
  1. void FileMsgServerConn::_HandleGetServerAddressReq(CImPdu* pPdu) {  
  2.     IM::Server::IMFileServerIPRsp msg;  
  3.       
  4.     const std::list<IM::BaseDefine::IpAddr>& addrs = ConfigUtil::GetInstance()->GetAddressList();  
  5.       
  6.     for (std::list<IM::BaseDefine::IpAddr>::const_iterator it=addrs.begin(); it!=addrs.end(); ++it) {  
  7.         IM::BaseDefine::IpAddr* addr = msg.add_ip_addr_list();  
  8.         *addr = *it;  
  9.         log("Upload file_client_conn addr info, ip=%s, port=%d", addr->ip().c_str(), addr->port());  
  10.     }  
  11.       
  12.     SendMessageLite(this, SID_OTHER, CID_OTHER_FILE_SERVER_IP_RSP, pPdu->GetSeqNum(), &msg);  
  13. }  

获得的信息是file_server侦听的ip地址和端口号,默认配置的端口号是8600。也就是说file_server的8600用于客户端链接,8601端口用于msg_server链接。这样,客户端须要传输文件(注意:不是聊天图片,聊天图片使用另一个服务msfs进行传输),会先告诉msg_server它须要进行文件传输,msg_server收到消息后告诉客户端,你连file_server来传输文件吧,并把file_server的地址和端口号告诉客户端。客户端这个时候链接file_server进行文件传输。咱们来具体看一看这个流程的细节信息:

1. 客户端发包给msg_server说要进行文件发送


而后选择一个文件:




pc客户端发送文件逻辑:

[cpp]  view plain  copy
  1. //pc客户端代码(Modules工程SessionLayout.cpp)  
  2. void SessionLayout::Notify(TNotifyUI& msg)  
  3. {  
  4.     ...  
  5.     //省略无关代码  
  6.     else if (msg.pSender == m_pBtnsendfile) //文件传输  
  7.     {  
  8.         module::UserInfoEntity userInfo;  
  9.         if (!module::getUserListModule()->getUserInfoBySId(m_sId, userInfo))  
  10.         {  
  11.             LOG__(ERR, _T("SendFile can't find the sid"));  
  12.             return;  
  13.         }  
  14.   
  15.         CFileDialog fileDlg(TRUE, NULL, NULL, OFN_HIDEREADONLY | OFN_FILEMUSTEXIST  
  16.             , _T("文件|*.*||"), AfxGetMainWnd());  
  17.         fileDlg.m_ofn.Flags |= OFN_NOCHANGEDIR;  
  18.         fileDlg.DoModal();  
  19.   
  20.         CString sPathName;  
  21.         POSITION nPos = fileDlg.GetStartPosition();  
  22.         if (nPos != NULL)  
  23.         {  
  24.             sPathName = fileDlg.GetNextPathName(nPos);  
  25.         }  
  26.         if (sPathName.IsEmpty())  
  27.             return;  
  28.   
  29.         module::getFileTransferModule()->sendFile(sPathName, m_sId, userInfo.isOnlne());  
  30.     }  
  31.     ...  
  32.     //省略无关代码  
  33. }  


sPathName是文件的全饰路径;m_sId是收取文件的用户id,userInfo.isOnlne()判断m_sId表明的用户是否在线,以此来决定此次文件传输是在线文件仍是离线文件模式。
 
[cpp]  view plain  copy
  1. BOOL FileTransferModule_Impl::sendFile(IN const CString& sFilePath, IN const std::string& sSendToSID,IN BOOL bOnlineMode)  
  2. {  
  3.     if (TransferFileEntityManager::getInstance()->checkIfIsSending(sFilePath))  
  4.     {  
  5.         return FALSE;  
  6.     }  
  7.     TransferFileEntity fileEntity;  
  8.       
  9.     //获取文件大小  
  10.     struct __stat64 buffer;  
  11.     _wstat64(sFilePath, &buffer);  
  12.     fileEntity.nFileSize = (UInt32)buffer.st_size;  
  13.     if (0 != fileEntity.nFileSize)  
  14.     {  
  15.         CString strFileName = sFilePath;  
  16.         strFileName.Replace(_T("\\"), _T("/"));//mac上对于路径字符“\”须要作特殊处理,windows上则能够识别  
  17.         fileEntity.sFileName = util::cStringToString(strFileName);  
  18.         fileEntity.sFromID = module::getSysConfigModule()->userID();  
  19.         fileEntity.sToID = sSendToSID;  
  20.         uint32_t transMode = 0;  
  21.         transMode = bOnlineMode ? IM::BaseDefine::TransferFileType::FILE_TYPE_ONLINE : IM::BaseDefine::TransferFileType::FILE_TYPE_OFFLINE;  
  22.           
  23.         LOG__(DEBG,_T("FileTransferSevice_Impl::sendFile sTaskID = %s"), util::stringToCString(fileEntity.sTaskID));  
  24.   
  25.         imcore::IMLibCoreStartOperationWithLambda(  
  26.             [=]()  
  27.         {  
  28.             IM::File::IMFileReq imFileReq;  
  29.             LOG__(APP, _T("imFileReq,name=%s,size=%d,toId=%s"),util::stringToCString(fileEntity.sFileName)  
  30.                 ,fileEntity.nFileSize,util::stringToCString(fileEntity.sToID));  
  31.             imFileReq.set_from_user_id(util::stringToInt32(fileEntity.sFromID));  
  32.             imFileReq.set_to_user_id(util::stringToInt32(fileEntity.sToID));  
  33.             imFileReq.set_file_name(fileEntity.sFileName);  
  34.             imFileReq.set_file_size(fileEntity.nFileSize);  
  35.             imFileReq.set_trans_mode(static_cast<IM::BaseDefine::TransferFileType>(transMode));  
  36.   
  37.             module::getTcpClientModule()->sendPacket(IM::BaseDefine::ServiceID::SID_FILE  
  38.                 , IM::BaseDefine::FileCmdID::CID_FILE_REQUEST  
  39.                 , &imFileReq);  
  40.         });  
  41.           
  42.         return TRUE;  
  43.     }  
  44.     LOG__(ERR, _T("fileEntity FileSize error,size = %d"), fileEntity.nFileSize);  
  45.     return FALSE;  
  46. }  

上面代码中组装的包信息中含有要传输的文件路径、文件大小、发送人id、接收方id、文件传输模式(在线仍是离线),包的命令号是IM::BaseDefine::FileCmdID::CID_FILE_REQUEST。这个包发给msg_server之后,msg_server应答:

[cpp]  view plain  copy
  1. void CMsgConn::HandlePdu(CImPdu* pPdu)  
  2. {  
  3. ...  
  4. //省略无关代码  
  5. case CID_FILE_REQUEST:  
  6.             s_file_handler->HandleClientFileRequest(this, pPdu);  
  7.             break;  
  8. ...  
  9. //省略无关代码  
  10. }  

[cpp]  view plain  copy
  1. void CFileHandler::HandleClientFileRequest(CMsgConn* pMsgConn, CImPdu* pPdu)  
  2. {  
  3.     IM::File::IMFileReq msg;  
  4.     CHECK_PB_PARSE_MSG(msg.ParseFromArray(pPdu->GetBodyData(), pPdu->GetBodyLength()));  
  5.       
  6.     uint32_t from_id = pMsgConn->GetUserId();  
  7.     uint32_t to_id = msg.to_user_id();  
  8.     string file_name = msg.file_name();  
  9.     uint32_t file_size = msg.file_size();  
  10.     uint32_t trans_mode = msg.trans_mode();  
  11.     log("HandleClientFileRequest, %u->%u, fileName: %s, trans_mode: %u.", from_id, to_id, file_name.c_str(), trans_mode);  
  12.       
  13.     CDbAttachData attach(ATTACH_TYPE_HANDLE, pMsgConn->GetHandle());  
  14.     CFileServConn* pFileConn = get_random_file_serv_conn();  
  15.     if (pFileConn)  
  16.     {  
  17.         IM::Server::IMFileTransferReq msg2;  
  18.         msg2.set_from_user_id(from_id);  
  19.         msg2.set_to_user_id(to_id);  
  20.         msg2.set_file_name(file_name);  
  21.         msg2.set_file_size(file_size);  
  22.         msg2.set_trans_mode((IM::BaseDefine::TransferFileType)trans_mode);  
  23.         msg2.set_attach_data(attach.GetBuffer(), attach.GetLength());  
  24.         CImPdu pdu;  
  25.         pdu.SetPBMsg(&msg2);  
  26.         pdu.SetServiceId(SID_OTHER);  
  27.         pdu.SetCommandId(CID_OTHER_FILE_TRANSFER_REQ);  
  28.         pdu.SetSeqNum(pPdu->GetSeqNum());  
  29.           
  30.         if (IM::BaseDefine::FILE_TYPE_OFFLINE == trans_mode)  
  31.         {  
  32.             pFileConn->SendPdu(&pdu);  
  33.         }  
  34.         else //IM::BaseDefine::FILE_TYPE_ONLINE  
  35.         {  
  36.             CImUser* pUser = CImUserManager::GetInstance()->GetImUserById(to_id);  
  37.             if (pUser && pUser->GetPCLoginStatus())//已有对应的帐号pc登陆状态  
  38.             {  
  39.                 pFileConn->SendPdu(&pdu);  
  40.             }  
  41.             else//无对应用户的pc登陆状态,向route_server查询状态  
  42.             {  
  43.                 //no pc_client in this msg_server, check it from route_server  
  44.                 CPduAttachData attach_data(ATTACH_TYPE_HANDLE_AND_PDU_FOR_FILE, pMsgConn->GetHandle(), pdu.GetBodyLength(), pdu.GetBodyData());  
  45.                 IM::Buddy::IMUsersStatReq msg3;  
  46.                 msg3.set_user_id(from_id);  
  47.                 msg3.add_user_id_list(to_id);  
  48.                 msg3.set_attach_data(attach_data.GetBuffer(), attach_data.GetLength());  
  49.                 CImPdu pdu2;  
  50.                 pdu2.SetPBMsg(&msg3);  
  51.                 pdu2.SetServiceId(SID_BUDDY_LIST);  
  52.                 pdu2.SetCommandId(CID_BUDDY_LIST_USERS_STATUS_REQUEST);  
  53.                 pdu2.SetSeqNum(pPdu->GetSeqNum());  
  54.                 CRouteServConn* route_conn = get_route_serv_conn();  
  55.                 if (route_conn)  
  56.                 {  
  57.                     route_conn->SendPdu(&pdu2);  
  58.                 }  
  59.             }  
  60.         }  
  61.     }  
  62.     else  
  63.     {  
  64.         log("HandleClientFileRequest, no file server.   ");  
  65.         IM::File::IMFileRsp msg2;  
  66.         msg2.set_result_code(1);  
  67.         msg2.set_from_user_id(from_id);  
  68.         msg2.set_to_user_id(to_id);  
  69.         msg2.set_file_name(file_name);  
  70.         msg2.set_task_id("");  
  71.         msg2.set_trans_mode((IM::BaseDefine::TransferFileType)trans_mode);  
  72.         CImPdu pdu;  
  73.         pdu.SetPBMsg(&msg2);  
  74.         pdu.SetServiceId(SID_FILE);  
  75.         pdu.SetCommandId(CID_FILE_RESPONSE);  
  76.         pdu.SetSeqNum(pPdu->GetSeqNum());  
  77.         pMsgConn->SendPdu(&pdu);  
  78.     }  
  79. }  

这段代码,颇有讲究,msg_server会检测file_server是否已经启动,若是没有启动,则直接发包告诉客户端,file_server不存在。另外,若是该文件传输模式是在线文件,会判断接收文件的用户是否和发送用户在同一台msg_server上。不在的话,则给route_server发送消息,查找该用户所在的msg_server(这个不具体介绍了,后面分析route_server会专门介绍的)。不然,会将文件发送请求转发给file_server,包的命令号是CID_OTHER_FILE_TRANSFER_REQ。file_server收到该请求后,处理以下:

[cpp]  view plain  copy
  1. void FileMsgServerConn::HandlePdu(CImPdu* pdu) {  
  2.       
  3.      ...  
  4.     //省略无关代码        
  5.     case CID_OTHER_FILE_TRANSFER_REQ:  
  6.         _HandleMsgFileTransferReq(pdu);  
  7.         break ;  
  8.     ...  
  9.     //省略无关代码          
  10. }  

[cpp]  view plain  copy
  1. void FileMsgServerConn::_HandleMsgFileTransferReq(CImPdu* pdu) {  
  2.     IM::Server::IMFileTransferReq transfer_req;  
  3.     CHECK_PB_PARSE_MSG(transfer_req.ParseFromArray(pdu->GetBodyData(), pdu->GetBodyLength()));  
  4.       
  5.       
  6.     uint32_t from_id = transfer_req.from_user_id();  
  7.     uint32_t to_id = transfer_req.to_user_id();  
  8.       
  9.     IM::Server::IMFileTransferRsp transfer_rsp;  
  10.     transfer_rsp.set_result_code(1);  
  11.     transfer_rsp.set_from_user_id(from_id);  
  12.     transfer_rsp.set_to_user_id(to_id);  
  13.     transfer_rsp.set_file_name(transfer_req.file_name());  
  14.     transfer_rsp.set_file_size(transfer_req.file_size());  
  15.     transfer_rsp.set_task_id("");  
  16.     transfer_rsp.set_trans_mode(transfer_req.trans_mode());  
  17.     transfer_rsp.set_attach_data(transfer_req.attach_data());  
  18.   
  19.       
  20.     bool rv = false;  
  21.     do {  
  22.         std::string task_id = GenerateUUID();  
  23.         if (task_id.empty()) {  
  24.             log("Create task id failed");  
  25.             break;  
  26.         }  
  27.         log("trams_mode=%d, task_id=%s, from_id=%d, to_id=%d, file_name=%s, file_size=%d", transfer_req.trans_mode(), task_id.c_str(), from_id, to_id, transfer_req.file_name().c_str(), transfer_req.file_size());  
  28.           
  29.         BaseTransferTask* transfer_task = TransferTaskManager::GetInstance()->NewTransferTask(  
  30.                                                                                               transfer_req.trans_mode(),  
  31.                                                                                               task_id,  
  32.                                                                                               from_id,  
  33.                                                                                               to_id,  
  34.                                                                                               transfer_req.file_name(),  
  35.                                                                                               transfer_req.file_size());  
  36.           
  37.         if (transfer_task == NULL) {  
  38.             // 建立未成功  
  39.             // close connection with msg svr  
  40.             // need_close = true;  
  41.             log("Create task failed");  
  42.             break;  
  43.         }  
  44.           
  45.         transfer_rsp.set_result_code(0);  
  46.         transfer_rsp.set_task_id(task_id);  
  47.         rv = true;  
  48.         // need_seq_no = false;  
  49.           
  50.         log("Create task succeed, task id %s, task type %d, from user %d, to user %d", task_id.c_str(), transfer_req.trans_mode(), from_id, to_id);  
  51.     } while (0);  
  52.       
  53.     ::SendMessageLite(this, SID_OTHER, CID_OTHER_FILE_TRANSFER_RSP, pdu->GetSeqNum(), &transfer_rsp);  
  54.       
  55.     if (!rv) {  
  56.         // 未建立成功,关闭链接  
  57.         Close();  
  58.     }  
  59. }  

上述代码会为本次传输任务建立一个惟一的标识uuid做为taskid,而后根据离线文件仍是在线文件建立离线文件传输任务OfflineTransferTask或者在线文件传输任务OnlineTransferTask,并加入一个一个成员变量transfer_tasks_中进行管理:

[cpp]  view plain  copy
  1. BaseTransferTask* TransferTaskManager::NewTransferTask(uint32_t trans_mode, const std::string& task_id, uint32_t from_user_id, uint32_t to_user_id, const std::string& file_name, uint32_t file_size) {  
  2.     BaseTransferTask* transfer_task = NULL;  
  3.       
  4.     TransferTaskMap::iterator it = transfer_tasks_.find(task_id);  
  5.     if (it==transfer_tasks_.end()) {  
  6.         if (trans_mode == IM::BaseDefine::FILE_TYPE_ONLINE) {  
  7.             transfer_task = new OnlineTransferTask(task_id, from_user_id, to_user_id, file_name, file_size);  
  8.         } else if (trans_mode == IM::BaseDefine::FILE_TYPE_OFFLINE) {  
  9.             transfer_task = new OfflineTransferTask(task_id, from_user_id, to_user_id, file_name, file_size);  
  10.         } else {  
  11.             log("Invalid trans_mode = %d", trans_mode);  
  12.         }  
  13.           
  14.         if (transfer_task) {  
  15.             transfer_tasks_.insert(std::make_pair(task_id, transfer_task));  
  16.         }  
  17.     } else {  
  18.         log("Task existed by task_id=%s, why?????", task_id.c_str());  
  19.     }  
  20.       
  21.     return transfer_task;  
  22. }  

这个map transfer_tasks_是在定时器里面进行按期处理的,处理的依据是当前任务的状态,好比已经完成的任务就能够从map中移除了:

[cpp]  view plain  copy
  1. void TransferTaskManager::OnTimer(uint64_t tick) {  
  2.     for (TransferTaskMap::iterator it = transfer_tasks_.begin(); it != transfer_tasks_.end();) {  
  3.         BaseTransferTask* task = it->second;  
  4.         if (task == NULL) {  
  5.             transfer_tasks_.erase(it++);  
  6.             continue;  
  7.         }  
  8.           
  9.         if (task->state() != kTransferTaskStateWaitingUpload &&  
  10.             task->state() == kTransferTaskStateTransferDone) {  
  11.             long esp = time(NULL) - task->create_time();  
  12.             if (esp > ConfigUtil::GetInstance()->GetTaskTimeout()) {  
  13.                 if (task->GetFromConn()) {  
  14.                     FileClientConn* conn = reinterpret_cast<FileClientConn*>(task->GetFromConn());  
  15.                     conn->ClearTransferTask();  
  16.                 }  
  17.                 if (task->GetToConn()) {  
  18.                     FileClientConn* conn = reinterpret_cast<FileClientConn*>(task->GetToConn());  
  19.                     conn->ClearTransferTask();  
  20.                 }  
  21.                 delete task;  
  22.                 transfer_tasks_.erase(it++);  
  23.                 continue;  
  24.             }  
  25.         }  
  26.           
  27.         ++it;  
  28.     }  
  29. }  

完成这些工做之后,组装的应答包命令号是CID_OTHER_FILE_TRANSFER_RSP,回复给msg_server。msg_server收到该应答包后处理:

[cpp]  view plain  copy
  1. void CFileServConn::HandlePdu(CImPdu* pPdu)  
  2. {  
  3.     switch (pPdu->GetCommandId()) {  
  4.         ...  
  5.     //省略无关代码  
  6.         case CID_OTHER_FILE_TRANSFER_RSP:  
  7.             _HandleFileMsgTransRsp(pPdu);  
  8.             break;  
  9.         ...  
  10.     //省略无关代码   
  11.     }  
  12. }  

[cpp]  view plain  copy
  1. void CFileServConn::_HandleFileMsgTransRsp(CImPdu* pPdu)  
  2. {  
  3.     IM::Server::IMFileTransferRsp msg;  
  4.     CHECK_PB_PARSE_MSG(msg.ParseFromArray(pPdu->GetBodyData(), pPdu->GetBodyLength()));  
  5.   
  6.     uint32_t result = msg.result_code();  
  7.     uint32_t from_id = msg.from_user_id();  
  8.     uint32_t to_id = msg.to_user_id();  
  9.     string file_name = msg.file_name();  
  10.     uint32_t file_size = msg.file_size();  
  11.     string task_id = msg.task_id();  
  12.     uint32_t trans_mode = msg.trans_mode();  
  13.     CDbAttachData attach((uchar_t*)msg.attach_data().c_str(), msg.attach_data().length());  
  14.     log("HandleFileMsgTransRsp, result: %u, from_user_id: %u, to_user_id: %u, file_name: %s, \  
  15.         task_id: %s, trans_mode: %u. ", result, from_id, to_id,  
  16.         file_name.c_str(), task_id.c_str(), trans_mode);  
  17.   
  18.     const list<IM::BaseDefine::IpAddr>* ip_addr_list = GetFileServerIPList();  
  19.   
  20.     IM::File::IMFileRsp msg2;  
  21.     msg2.set_result_code(result);  
  22.     msg2.set_from_user_id(from_id);  
  23.     msg2.set_to_user_id(to_id);  
  24.     msg2.set_file_name(file_name);  
  25.     msg2.set_task_id(task_id);  
  26.     msg2.set_trans_mode((IM::BaseDefine::TransferFileType)trans_mode);  
  27.     for (list<IM::BaseDefine::IpAddr>::const_iterator it = ip_addr_list->begin(); it != ip_addr_list->end(); it++)  
  28.     {  
  29.         IM::BaseDefine::IpAddr ip_addr_tmp = *it;  
  30.         IM::BaseDefine::IpAddr* ip_addr = msg2.add_ip_addr_list();  
  31.         ip_addr->set_ip(ip_addr_tmp.ip());  
  32.         ip_addr->set_port(ip_addr_tmp.port());  
  33.     }  
  34.     CImPdu pdu;  
  35.     pdu.SetPBMsg(&msg2);  
  36.     pdu.SetServiceId(SID_FILE);  
  37.     pdu.SetCommandId(CID_FILE_RESPONSE);  
  38.     pdu.SetSeqNum(pPdu->GetSeqNum());  
  39.     uint32_t handle = attach.GetHandle();  
  40.       
  41.     CMsgConn* pFromConn = CImUserManager::GetInstance()->GetMsgConnByHandle(from_id, handle);  
  42.     if (pFromConn)  
  43.     {  
  44.         pFromConn->SendPdu(&pdu);  
  45.     }  
  46.       
  47.     if (result == 0)  
  48.     {  
  49.         IM::File::IMFileNotify msg3;  
  50.         msg3.set_from_user_id(from_id);  
  51.         msg3.set_to_user_id(to_id);  
  52.         msg3.set_file_name(file_name);  
  53.         msg3.set_file_size(file_size);  
  54.         msg3.set_task_id(task_id);  
  55.         msg3.set_trans_mode((IM::BaseDefine::TransferFileType)trans_mode);  
  56.         msg3.set_offline_ready(0);  
  57.         for (list<IM::BaseDefine::IpAddr>::const_iterator it = ip_addr_list->begin(); it != ip_addr_list->end(); it++)  
  58.         {  
  59.             IM::BaseDefine::IpAddr ip_addr_tmp = *it;  
  60.             IM::BaseDefine::IpAddr* ip_addr = msg3.add_ip_addr_list();  
  61.             ip_addr->set_ip(ip_addr_tmp.ip());  
  62.             ip_addr->set_port(ip_addr_tmp.port());  
  63.         }  
  64.         CImPdu pdu2;  
  65.         pdu2.SetPBMsg(&msg3);  
  66.         pdu2.SetServiceId(SID_FILE);  
  67.         pdu2.SetCommandId(CID_FILE_NOTIFY);  
  68.           
  69.         //send notify to target user  
  70.         CImUser* pToUser = CImUserManager::GetInstance()->GetImUserById(to_id);  
  71.         if (pToUser)  
  72.         {  
  73.             pToUser->BroadcastPduWithOutMobile(&pdu2);  
  74.         }  
  75.           
  76.         //send to route server  
  77.         CRouteServConn* pRouteConn = get_route_serv_conn();  
  78.         if (pRouteConn) {  
  79.             pRouteConn->SendPdu(&pdu2);  
  80.         }  
  81.     }  
  82. }  

msg_server收到包后,首先装包数据,并把file_server的ip地址和端口信息带上,发给请求发文件的客户端,命令号是CID_FILE_RESPONSE;接着查询通知接收方有人给其发文件(通知方式也是同样,若是接收方在该msg_server上,直接发给该用户;不在的话,发给路由服务route_server)。固然接收到文件发送的端只有pc端,移动端会被过滤掉的,也就是说移动端不会收到发送文件的请求。

咱们先看发送方pc客户端收到应答的逻辑(命令号是CID_FILE_RESPONSE):

[cpp]  view plain  copy
  1. void FileTransferModule_Impl::onPacket(imcore::TTPBHeader& header, std::string& pbBody)  
  2. {  
  3.     switch (header.getCommandId())  
  4.     {  
  5.     case IM::BaseDefine::FileCmdID::CID_FILE_RESPONSE://发送“文件请求/离线文件”-返回  
  6.         _sendfileResponse(pbBody);  
  7.         break;    
  8.     }  
  9. }  

[cpp]  view plain  copy
  1. void FileTransferModule_Impl::_sendfileResponse(IN std::string& body)  
  2. {  
  3.     IM::File::IMFileRsp imFileRsp;  
  4.     if (!imFileRsp.ParseFromString(body))  
  5.     {  
  6.         LOG__(ERR, _T("parse failed,body:%s"), util::stringToCString(body));  
  7.         return;  
  8.     }  
  9.   
  10.     UInt32 nResult = imFileRsp.result_code();  
  11.     if (nResult != 0)  
  12.     {  
  13.         LOG__(ERR, _T("_sendfileResponse result != 0"));  
  14.         module::getFileTransferModule()->asynNotifyObserver(module::KEY_FILESEVER_UPLOAD_FAILED);  
  15.     }  
  16.   
  17.     TransferFileEntity fileEntity;  
  18.     fileEntity.sTaskID = imFileRsp.task_id();  
  19.     assert(!fileEntity.sTaskID.empty());  
  20.     fileEntity.sFromID = util::uint32ToString(imFileRsp.from_user_id());  
  21.     fileEntity.sToID = util::uint32ToString(imFileRsp.to_user_id());  
  22.     fileEntity.sFileName = imFileRsp.file_name();  
  23.     fileEntity.setSaveFilePath(util::stringToCString(fileEntity.sFileName));//发送方文件地址,就是保存地址  
  24.     fileEntity.time = static_cast<UInt32>(time(0));  
  25.     uint32_t transMode = imFileRsp.trans_mode();  
  26.     if (IM::BaseDefine::TransferFileType::FILE_TYPE_ONLINE == transMode)  
  27.     {  
  28.         fileEntity.nClientMode = IM::BaseDefine::ClientFileRole::CLIENT_REALTIME_SENDER;  
  29.     }  
  30.     else if (IM::BaseDefine::TransferFileType::FILE_TYPE_OFFLINE == transMode)  
  31.     {  
  32.         fileEntity.nClientMode = IM::BaseDefine::ClientFileRole::CLIENT_OFFLINE_UPLOAD;  
  33.     }  
  34.     fileEntity.pFileObject = new TransferFile(util::stringToCString(fileEntity.sFileName),FALSE);  
  35.     if (fileEntity.pFileObject)  
  36.     {  
  37.         fileEntity.nFileSize = fileEntity.pFileObject->length();  
  38.     }  
  39.       
  40.     UINT32 nIPCount = imFileRsp.ip_addr_list_size();  
  41.     if (nIPCount <= 0)  
  42.     {  
  43.         return;  
  44.     }  
  45.     const IM::BaseDefine::IpAddr& ipAdd = imFileRsp.ip_addr_list(0);  
  46.     fileEntity.sIP = ipAdd.ip();  
  47.     fileEntity.nPort = ipAdd.port();  
  48.   
  49.     if (!TransferFileEntityManager::getInstance()->pushTransferFileEntity(fileEntity))  
  50.         TransferFileEntityManager::getInstance()->updateFileInfoBysTaskID(fileEntity);  
  51.   
  52.     LOG__(DEBG, _T("FileTransferSevice_Impl::准备链接文件服务器 sTaskId = %s"), util::stringToCString(fileEntity.sTaskID));  
  53.     TransferFileEntityManager::getInstance()->openFileSocketByTaskId(fileEntity.sTaskID);  
  54. }  

客户端在TransferFileEntityManager::getInstance()->openFileSocketByTaskId(fileEntity.sTaskID);里面实际去链接file_server并尝试发文件:

[cpp]  view plain  copy
  1. void TransferFileEntityManager::openFileSocketByTaskId(std::string& taskId)  
  2. {  
  3.     m_fileUIThread->openFileSocketByTaskId(taskId);  
  4. }  

[cpp]  view plain  copy
  1. void FileTransferUIThread::openFileSocketByTaskId(std::string& taskId)  
  2. {  
  3.     FileTransferSocket* pFileSocket = _findFileSocketByTaskId(taskId);