添加tcp_p支持；

客户端断连后，不再reinvite

添加tcp_p支持；
客户端断连后，不再reinvite
Hu Chunming
1 parent 8735837d
Showing 4 changed files with 87 additions and 19 deletions
sip/SipServer.cpp
sip/SipServer.h
sip/WebSocketServer.cpp
sip/WebSocketServer.h
@@ -110,6 +110,7 @@ int SipServer::sip_event_handle(eXosip_event_t *evtp) {
         case EXOSIP_CALL_CLOSED://21
             LOG_INFO("EXOSIP_CALL_CLOSED type={}",evtp->type);
+            // response_message_answer(evtp,200);
             // this->dump_request(evtp);
             // this->dump_response(evtp);
             break;
@@ -168,12 +169,14 @@ int SipServer::sip_event_handle(eXosip_event_t *evtp) {
             break;
         case EXOSIP_CALL_SERVERFAILURE:
             LOG_INFO("EXOSIP_CALL_SERVERFAILURE cid={}", evtp->cid);
+            response_message_answer(evtp,200);
             break;
         case EXOSIP_IN_SUBSCRIPTION_NEW:
             LOG_INFO("EXOSIP_IN_SUBSCRIPTION_NEW cid={}", evtp->cid);
             break;
         case EXOSIP_CALL_MESSAGE_ANSWERED:
             LOG_INFO("EXOSIP_CALL_MESSAGE_ANSWERED cid={}", evtp->cid);
+            response_message_answer(evtp,200);
             break;
         default:
             LOG_INFO("type={} unknown", evtp->type);
@@ -296,16 +299,16 @@ void SipServer::Close() {
 void SipServer::response_message_answer(eXosip_event_t *evtp,int code){
     int returnCode = 0 ;
-    osip_message_t * pRegister = nullptr;
-    returnCode = eXosip_message_build_answer (mSipCtx,evtp->tid,code,&pRegister);
+    osip_message_t * pMsg = nullptr;
+    returnCode = eXosip_message_build_answer (mSipCtx,evtp->tid,code,&pMsg);
     bool bRegister = false;
-    if(pRegister){
+    if(pMsg){
         bRegister = true;
     }
     if (returnCode == 0 && bRegister)
     {
         eXosip_lock(mSipCtx);
-        eXosip_message_send_answer (mSipCtx,evtp->tid,code,pRegister);
+        eXosip_message_send_answer (mSipCtx,evtp->tid,code,pMsg);
         eXosip_unlock(mSipCtx);
     }
     else{
@@ -705,9 +708,9 @@ int SipServer::reInvite(int cid) {
             int ret = -1;
             string cmd = it_cmd->second;
             if (cmd == "udp") {
-                ret = RequestInvite_UDP(strChannelId.c_str(), strIp.c_str(), iPort);
-            } else if (cmd == "tcp_a") {
-                ret = RequestInvite_TCP_a(strChannelId.c_str(), strIp.c_str(), iPort);
+                ret = invite_UDP_nolock(strChannelId.c_str(), strIp.c_str(), iPort);
+            } else if (cmd == "tcp_p") {
+                ret = invite_TCP_p_nolock(strChannelId.c_str(), strIp.c_str(), iPort);
             }
             if (ret <= 0) {
@@ -721,6 +724,14 @@ int SipServer::reInvite(int cid) {
 }
 int SipServer::RequestInvite_UDP(const char* dst_channel, const char* rtpIp, int rtpPort) {
+    std::lock_guard<std::mutex> l_cmd(m_invite_cmd_map_mtx);
+    return invite_UDP_nolock(dst_channel, rtpIp, rtpPort);
+}
+
+int SipServer::invite_UDP_nolock(const char* dst_channel, const char* rtpIp, int rtpPort) {
+
+    // rtpIp = "192.168.70.1";
+    // rtpPort = 10000;
     // 检查设备是否在线
     if (!check_device_status(dst_channel)) {
@@ -755,7 +766,6 @@ int SipServer::RequestInvite_UDP(const char* dst_channel, const char* rtpIp, int
         "a=rtpmap:96 PS/90000\r\n"
         "a=rtpmap:98 H264/90000\r\n"
         "a=rtpmap:97 MPEG4/90000\r\n"
-        "a=setup:passive\r\n"
         "a=connection:new\r\n"
         "y=0100000001\r\n"
         "f=\r\n", mInfo.getSipId().c_str(), rtpIp, rtpIp, rtpPort);
@@ -785,18 +795,30 @@ int SipServer::RequestInvite_UDP(const char* dst_channel, const char* rtpIp, int
     return call_id;
 }
-int SipServer::RequestInvite_TCP_a(const char* dst_channel, const char* rtpIp, int rtpPort) {
+int SipServer::RequestInvite_TCP_p(const char* dst_channel, const char* rtpIp, int rtpPort) {
+    std::lock_guard<std::mutex> l_cmd(m_invite_cmd_map_mtx);
+    return invite_TCP_p_nolock(dst_channel, rtpIp, rtpPort);
+}
+
+int SipServer::invite_TCP_p_nolock(const char* dst_channel, const char* rtpIp, int rtpPort) {
+
+    //测试
+    // rtpIp = "192.168.70.1";
+    // rtpPort = 10000;
+
     // 检查设备是否在线
     if (!check_device_status(dst_channel)) {
+        LOG_ERROR("{} is not online!", dst_channel);
         return -2;
     }
     Client* client = get_parent_by_id(dst_channel);
     if (client == nullptr) {
+        LOG_ERROR("do not get parent device:{}", dst_channel);
         return -1;
     }
-    LOG_INFO("INVITE TCP active");
+    LOG_INFO("INVITE TCP Passive: {}  {}:{}", dst_channel, rtpIp, rtpPort);
     char session_exp[1024] = { 0 };
     osip_message_t* msg = nullptr;
@@ -804,8 +826,6 @@ int SipServer::RequestInvite_TCP_a(const char* dst_channel, const char* rtpIp, i
     char to[1024] = { 0 };
     char sdp[2048] = { 0 };
-    // const char* dst_channel = "34020000001320000001";
-
     sprintf(from, "sip:%s@%s:%d", mInfo.getSipId().c_str(), mInfo.getIp().c_str(), mInfo.getPort());
     sprintf(to, "sip:%s@%s:%d", dst_channel, client->getIp().c_str(), client->getPort());
     snprintf(sdp, 2048,
@@ -819,10 +839,10 @@ int SipServer::RequestInvite_TCP_a(const char* dst_channel, const char* rtpIp, i
         "a=rtpmap:96 PS/90000\r\n"
         "a=rtpmap:98 H264/90000\r\n"
         "a=rtpmap:97 MPEG4/90000\r\n"
-        "a=setup:active\r\n"
+        "a=setup:passive\r\n"
         "a=connection:new\r\n"
-        "y=0100000001\r\n"
-        "f=\r\n", mInfo.getSipId().c_str(), mInfo.getIp().c_str(), mInfo.getIp().c_str(), rtpPort);
+        "y=0200002494\r\n"
+        "f=\r\n", mInfo.getSipId().c_str(), rtpIp, rtpIp, rtpPort);
     int ret = eXosip_call_build_initial_invite(mSipCtx, &msg, to, from, nullptr, nullptr);
     if (ret) {
@@ -840,14 +860,20 @@ int SipServer::RequestInvite_TCP_a(const char* dst_channel, const char* rtpIp, i
     if (call_id > 0) {
         LOG_INFO("eXosip_call_send_initial_invite success: call_id={}", call_id);
         auto key = std::make_tuple(string(dst_channel), string(rtpIp), rtpPort);
-        m_invite_cmd_map[key] = "tcp_a";
+        m_invite_cmd_map[key] = "tcp_p";
     }
     else {
         LOG_ERROR("eXosip_call_send_initial_invite error: call_id={}", call_id);
     }
+
     return call_id;
 }
+void SipServer::RemoveInviteTask(tuple<std::string, std::string, int> key) {
+    std::lock_guard<std::mutex> l_cmd(m_invite_cmd_map_mtx);
+    m_invite_cmd_map.erase(key);
+}
+
 void SipServer::RequestCatalog(Client* client) {
     eXosip_lock(mSipCtx);
@@ -115,14 +115,23 @@ public:
     bool Init(ServerInfo* pInfo, WebSocketServer* pServer);
+    /*
+    GB28181流传输几种模式
+        UDP:服务端监听UDP端口，通过INVITE信令告知设备端口，设备主动向服务端发起流传输
+        TCP被动:服务端监听TCP端口，通过INVITE信令告知设备端口，设备向服务端发起流传输
+        TCP主动:设备端告知服务端监听的TCP端口情况，服务端主动向设备拉流，此种场景较少，且设备所在网络可以被服务所在网络访问（如下级设备与上级GB28181服务在同一个局域网，或者都在公网上能相互访问）。
+     */
+
     int RequestInvite_UDP(const char* dst_channel, const char* rtpIp, int rtpPort);
-    int RequestInvite_TCP_a(const char* dst_channel, const char* rtpIp, int rtpPort);
+    int RequestInvite_TCP_p(const char* dst_channel, const char* rtpIp, int rtpPort);
     int ByeInvite(std::string channel_id, string ip, int rtpPort);
     void Close();
+    void RemoveInviteTask(tuple<std::string, std::string, int>);
+
 public:
     void event_loop();
@@ -159,6 +168,10 @@ private:
     int check_device_type(string sip_id);
+    int invite_UDP_nolock(const char* dst_channel, const char* rtpIp, int rtpPort);
+
+    int invite_TCP_p_nolock(const char* dst_channel, const char* rtpIp, int rtpPort);
+
 private:
     bool mQuit{ false };
     eXosip_t *mSipCtx;
@@ -31,7 +31,7 @@ int WebSocketServer::parse_invite(vector&lt;string&gt;&amp; vec_msg, std::string ip) {
     if ("udp" == net_type) {
         ret = sip_server.RequestInvite_UDP(vec_msg[2].c_str(), ip.c_str(), atoi(vec_msg[3].c_str()));
     } else if ("tcp" == net_type)  {
-        ret = sip_server.RequestInvite_TCP_a(vec_msg[2].c_str(), ip.c_str(), atoi(vec_msg[3].c_str()));
+        ret = sip_server.RequestInvite_TCP_p(vec_msg[2].c_str(), ip.c_str(), atoi(vec_msg[3].c_str()));
     }
     return ret;
@@ -54,6 +54,8 @@ int WebSocketServer::msg_parser(websocketpp::connection_hdl hdl, string msg) {
     string ip = get_ip_from_hdl(hdl);
+    auto task_key = std::make_tuple(vec_msg[2], ip, atoi(vec_msg[3].c_str()));
+
     string response_msg = "";
     int ret = -100;
     if ( StringTools::to_lower(vec_msg[0]) == "invite") {
@@ -62,6 +64,10 @@ int WebSocketServer::msg_parser(websocketpp::connection_hdl hdl, string msg) {
         if (ret > 0) {
             std::lock_guard<std::mutex> l_hdl(m_channelport_hdl_map_mtx);
             m_channelport_hdl_map[strKey] = hdl;
+
+            std::lock_guard<std::mutex> l_task(m_hdl_task_mtx);
+            auto conn = ws_server.get_con_from_hdl(hdl);
+            m_hdl_task.emplace(conn, task_key);
         }
         response_msg = strKey +"|" + vec_msg[0] + "|" + vec_msg[1] + "|" + to_string(ret);
@@ -78,6 +84,15 @@ int WebSocketServer::msg_parser(websocketpp::connection_hdl hdl, string msg) {
             }
             ret = sip_server.ByeInvite(vec_msg[2], ip, atoi(vec_msg[3].c_str()));
+            if (ret == 0) {
+                std::lock_guard<std::mutex> l_task(m_hdl_task_mtx);
+                for(auto it_task = m_hdl_task.begin(); it_task != m_hdl_task.end(); it_task++) {
+                    if(it_task->second == task_key) {
+                        m_hdl_task.erase(it_task);
+                        break;
+                    }
+                }
+            }
         }
     } else {
         LOG_ERROR("on_message called with hdl: {}  and message:{}", hdl.lock().get(), msg);
@@ -136,16 +151,27 @@ void WebSocketServer::on_close(websocketpp::connection_hdl h){
     std::lock_guard<std::mutex> l_hdl(m_channelport_hdl_map_mtx);
+    server::connection_ptr con_std = ws_server.get_con_from_hdl(h);
     for (auto it = m_channelport_hdl_map.begin(); it != m_channelport_hdl_map.end(); )
     {
         server::connection_ptr con = ws_server.get_con_from_hdl(it->second);
-        server::connection_ptr con_std = ws_server.get_con_from_hdl(h);
+        
         if (con == con_std) {
             it = m_channelport_hdl_map.erase(it);
         } else {
             it++;
         }
     }
+
+    std::lock_guard<std::mutex> l_task(m_hdl_task_mtx);
+    for(auto it_task = m_hdl_task.begin(); it_task != m_hdl_task.end(); ) {
+        if(it_task->first == con_std) {
+            sip_server.RemoveInviteTask(it_task->second);
+            it_task = m_hdl_task.erase(it_task);
+        } else {
+            it_task++ ;
+        }
+    }
 }
@@ -43,6 +43,9 @@ private:
     std::map<string, websocketpp::connection_hdl> m_channelport_hdl_map;
     std::mutex m_channelport_hdl_map_mtx;
+
+    std::multimap<server::connection_ptr, tuple<std::string, std::string, int> > m_hdl_task;
+    std::mutex m_hdl_task_mtx;
 };
 #endif      // __WEB_SOCKET_SERVER__
 \ No newline at end of file