Merge branch 'master' of github.com:Dw9/ZLMediaKit

server/main.cpp

@@ -277,9 +277,10 @@ int start_main(int argc,char *argv[]) {
 #endif//defined(ENABLE_RTPPROXY)
 
 #if defined(ENABLE_WEBRTC)
+        auto rtcSrv_tcp = std::make_shared<TcpServer>();
         //webrtc udp服务器
-        auto rtcSrv = std::make_shared<UdpServer>();
-        rtcSrv->setOnCreateSocket([](const EventPoller::Ptr &poller, const Buffer::Ptr &buf, struct sockaddr *, int) {
+        auto rtcSrv_udp = std::make_shared<UdpServer>();
+        rtcSrv_udp->setOnCreateSocket([](const EventPoller::Ptr &poller, const Buffer::Ptr &buf, struct sockaddr *, int) {
             if (!buf) {
                 return Socket::createSocket(poller, false);
             }
@@ -337,7 +338,7 @@ int start_main(int argc,char *argv[]) {
 
 #if defined(ENABLE_WEBRTC)
             //webrtc udp服务器
-            if (rtcPort) { rtcSrv->start<WebRtcSession>(rtcPort); }
+            if (rtcPort) { rtcSrv_udp->start<WebRtcSession>(rtcPort); rtcSrv_tcp->start<WebRtcSession>(rtcPort); }
 #endif//defined(ENABLE_WEBRTC)
 
 #if defined(ENABLE_SRT)

webrtc/WebRtcSession.cpp

@@ -15,9 +15,7 @@ using namespace std;
 
 namespace mediakit {
 
-static string getUserName(const Buffer::Ptr &buffer) {
-    auto buf = buffer->data();
-    auto len = buffer->size();
+static string getUserName(const char *buf, size_t len) {
     if (!RTC::StunPacket::IsStun((const uint8_t *) buf, len)) {
         return "";
     }
@@ -35,7 +33,7 @@ static string getUserName(const Buffer::Ptr &buffer) {
 }
 
 EventPoller::Ptr WebRtcSession::queryPoller(const Buffer::Ptr &buffer) {
-    auto user_name = getUserName(buffer);
+    auto user_name = getUserName(buffer->data(), buffer->size());
     if (user_name.empty()) {
         return nullptr;
     }
@@ -45,20 +43,21 @@ EventPoller::Ptr WebRtcSession::queryPoller(const Buffer::Ptr &buffer) {
 
 ////////////////////////////////////////////////////////////////////////////////
 
-WebRtcSession::WebRtcSession(const Socket::Ptr &sock) : UdpSession(sock) {
+WebRtcSession::WebRtcSession(const Socket::Ptr &sock) : Session(sock) {
     socklen_t addr_len = sizeof(_peer_addr);
     getpeername(sock->rawFD(), (struct sockaddr *)&_peer_addr, &addr_len);
+    _over_tcp = sock->sockType() == SockNum::Sock_TCP;
 }
 
 WebRtcSession::~WebRtcSession() {
     InfoP(this);
 }
 
-void WebRtcSession::onRecv(const Buffer::Ptr &buffer) {
+void WebRtcSession::onRecv_l(const char *data, size_t len) {
     if (_find_transport) {
-        //只允许寻找一次transport
+        // 只允许寻找一次transport
         _find_transport = false;
-        auto user_name = getUserName(buffer);
+        auto user_name = getUserName(data, len);
         auto transport = WebRtcTransportManager::Instance().getItem(user_name);
         CHECK(transport && transport->getPoller()->isCurrentThread());
         transport->setSession(shared_from_this());
@@ -67,11 +66,19 @@ void WebRtcSession::onRecv(const Buffer::Ptr &buffer) {
     }
     _ticker.resetTime();
     CHECK(_transport);
-    _transport->inputSockData(buffer->data(), buffer->size(), (struct sockaddr *)&_peer_addr);
+    _transport->inputSockData((char *)data, len, (struct sockaddr *)&_peer_addr);
+}
+
+void WebRtcSession::onRecv(const Buffer::Ptr &buffer) {
+    if (_over_tcp) {
+        input(buffer->data(), buffer->size());
+    } else {
+        onRecv_l(buffer->data(), buffer->size());
+    }
 }
 
 void WebRtcSession::onError(const SockException &err) {
-    //udp链接超时，但是rtc链接不一定超时，因为可能存在udp链接迁移的情况
+    //udp链接超时，但是rtc链接不一定超时，因为可能存在链接迁移的情况
     //在udp链接迁移时，新的WebRtcSession对象将接管WebRtcTransport对象的生命周期
     //本WebRtcSession对象将在超时后自动销毁
     WarnP(this) << err.what();
@@ -97,6 +104,25 @@ void WebRtcSession::onManager() {
     }
 }
 
+ssize_t WebRtcSession::onRecvHeader(const char *data, size_t len) {
+    onRecv_l(data + 2, len - 2);
+    return 0;
+}
+
+const char *WebRtcSession::onSearchPacketTail(const char *data, size_t len) {
+    if (len < 2) {
+        // 数据不够
+        return nullptr;
+    }
+    uint16_t length = (((uint8_t *)data)[0] << 8) | ((uint8_t *)data)[1];
+    if (len < (size_t)(length + 2)) {
+        // 数据不够
+        return nullptr;
+    }
+    // 返回rtp包末尾
+    return data + 2 + length;
+}
+
 }// namespace mediakit
 
 

webrtc/WebRtcSession.h

@@ -15,10 +15,11 @@
 #include "Network/Session.h"
 #include "IceServer.hpp"
 #include "WebRtcTransport.h"
+#include "Http/HttpRequestSplitter.h"
 
 namespace mediakit {
 
-class WebRtcSession : public UdpSession {
+class WebRtcSession : public Session, public HttpRequestSplitter {
 public:
     WebRtcSession(const Socket::Ptr &sock);
     ~WebRtcSession() override;
@@ -26,11 +27,17 @@ public:
     void onRecv(const Buffer::Ptr &) override;
     void onError(const SockException &err) override;
     void onManager() override;
-    //std::string getIdentifier() const override;
-
     static EventPoller::Ptr queryPoller(const Buffer::Ptr &buffer);
 
 private:
+    //// HttpRequestSplitter override ////
+    ssize_t onRecvHeader(const char *data, size_t len) override;
+    const char *onSearchPacketTail(const char *data, size_t len) override;
+
+    void onRecv_l(const char *data, size_t len);
+
+private:
+    bool _over_tcp = false;
     bool _find_transport = true;
     Ticker _ticker;
     struct sockaddr_storage _peer_addr;

webrtc/WebRtcTransport.cpp

@@ -418,9 +418,21 @@ void WebRtcTransportImp::onSendSockData(Buffer::Ptr buf, bool flush, RTC::Transp
         WarnL << "send data failed:" << buf->size();
         return;
     }
+
     // 一次性发送一帧的rtp数据，提高网络io性能
-    _selected_session->setSendFlushFlag(flush);
+    if (_selected_session->getSock()->sockType() == SockNum::Sock_TCP) {
+        // 增加tcp两字节头
+        auto len = buf->size();
+        char tcp_len[2] = { 0 };
+        tcp_len[0] = (len >> 8) & 0xff;
+        tcp_len[1] = len & 0xff;
+        _selected_session->SockSender::send(tcp_len, 2);
+    }
     _selected_session->send(std::move(buf));
+
+    if (flush) {
+        _selected_session->flushAll();
+    }
 }
 
 ///////////////////////////////////////////////////////////////////
@@ -609,11 +621,13 @@ void WebRtcTransportImp::onRtcConfigure(RtcConfigure &configure) const {
     if (extern_ips.empty()) {
         std::string localIp = SockUtil::get_local_ip();
         configure.addCandidate(*makeIceCandidate(localIp, local_port, 120, "udp"));
+        configure.addCandidate(*makeIceCandidate(localIp, local_port, 110, "tcp"));
     } else {
         const uint32_t delta = 10;
         uint32_t priority = 100 + delta * extern_ips.size();
         for (auto ip : extern_ips) {
-            configure.addCandidate(*makeIceCandidate(ip, local_port, priority, "udp"));
+            configure.addCandidate(*makeIceCandidate(ip, local_port, priority + 5, "udp"));
+            configure.addCandidate(*makeIceCandidate(ip, local_port, priority, "tcp"));
             priority -= delta;
         }
     }
@@ -1042,6 +1056,7 @@ void WebRtcTransportImp::setSession(Session::Ptr session) {
               << session->get_peer_port() << ", id:" << getIdentifier();
     }
     _selected_session = std::move(session);
+    _selected_session->setSendFlushFlag(false);
     unrefSelf();
 }
 

webrtc/WebRtcTransport.h

@@ -179,7 +179,7 @@ private:
     std::shared_ptr<RTC::SrtpSession> _srtp_session_send;
     std::shared_ptr<RTC::SrtpSession> _srtp_session_recv;
     Ticker _ticker;
-    //循环池
+    // 循环池
     ResourcePool<BufferRaw> _packet_pool;
 
 #ifdef ENABLE_SCTP