From f9f6fd136a7337518b6516a8897d812c0344c351 Mon Sep 17 00:00:00 2001
From: xiongguangjie <xiong_panda@163.com>
Date: Sat, 27 Aug 2022 15:06:03 +0800
Subject: [PATCH] srt handle packet send repeate by timer

---
 srt/SrtTransport.cpp | 74 +++++++++++++++++++++-----------------------
 srt/SrtTransport.hpp |  4 +--
 2 files changed, 37 insertions(+), 41 deletions(-)

diff --git a/srt/SrtTransport.cpp b/srt/SrtTransport.cpp
index dd6cc5a1..398486da 100644
--- a/srt/SrtTransport.cpp
+++ b/srt/SrtTransport.cpp
@@ -80,6 +80,9 @@ void SrtTransport::inputSockData(uint8_t *buf, int len, struct sockaddr_storage
     if (DataPacket::isDataPacket(buf, len)) {
         uint32_t socketId = DataPacket::getSocketID(buf, len);
         if (socketId == _socket_id) {
+            if(_handleshake_timer){
+                _handleshake_timer.reset();
+            }
             _pkt_recv_rate_context->inputPacket(_now);
             _estimated_link_capacity_context->inputPacket(_now);
             _recv_rate_context->inputPacket(_now, len);
@@ -116,39 +119,27 @@ void SrtTransport::inputSockData(uint8_t *buf, int len, struct sockaddr_storage
         }
     }
 }
-bool SrtTransport::isSameCon(HandshakePacket &pkt) {
-    if (_handleshake_res) {
-        if (_handleshake_res->handshake_type == HandshakePacket::HS_TYPE_INDUCTION) {
-            if (pkt.srt_socket_id == _handleshake_res->dst_socket_id
-                && pkt.initial_packet_sequence_number == _init_seq_number) {
-                return true;
-            }
-            // TraceL << getIdentifier() << " new client from same udp connection";
-            return false;
-        } else if (_handleshake_res->handshake_type == HandshakePacket::HS_TYPE_CONCLUSION) {
-            if (pkt.srt_socket_id == _handleshake_res->dst_socket_id && _sync_cookie == pkt.syn_cookie) {
-                return true;
-            }
-            // TraceL << getIdentifier() << " new client from new same udp connection ";
-            return false;
-        } else {
-            WarnL << "not reach this";
-        }
-        return false;
-    }
-    return true;
-}
+
 void SrtTransport::handleHandshakeInduction(HandshakePacket &pkt, struct sockaddr_storage *addr) {
     // Induction Phase
     if (_handleshake_res) {
-        if(isSameCon(pkt)){
-            TraceL << getIdentifier() <<" Induction repeate "<<SockUtil::inet_ntoa((struct sockaddr *)addr) << ":" << SockUtil::inet_port((struct sockaddr *)addr);
-            for(int i=0;i<3;++i){
+        if(_handleshake_res->handshake_type == HandshakePacket::HS_TYPE_INDUCTION){
+            if(pkt.srt_socket_id == _handleshake_res->dst_socket_id){
+                TraceL << getIdentifier() <<" Induction repeate "<<SockUtil::inet_ntoa((struct sockaddr *)addr) << ":" << SockUtil::inet_port((struct sockaddr *)addr);
                 sendControlPacket(_handleshake_res, true);
+            }else{
+                TraceL << getIdentifier() <<" new connection fron client "<<SockUtil::inet_ntoa((struct sockaddr *)addr) << ":" << SockUtil::inet_port((struct sockaddr *)addr);
+                onShutdown(SockException(Err_other, "client new connection"));
             }
+            return;
+        }else if(_handleshake_res->handshake_type == HandshakePacket::HS_TYPE_CONCLUSION){
+            if(_handleshake_res->dst_socket_id != pkt.srt_socket_id){
+                TraceL << getIdentifier() <<" new connection fron client "<<SockUtil::inet_ntoa((struct sockaddr *)addr) << ":" << SockUtil::inet_port((struct sockaddr *)addr);
+                onShutdown(SockException(Err_other, "client new connection"));
+            }
+            return;
         }else{
-            TraceL << getIdentifier() <<" new connection fron client "<<SockUtil::inet_ntoa((struct sockaddr *)addr) << ":" << SockUtil::inet_port((struct sockaddr *)addr);
-            onShutdown(SockException(Err_other, "client new connection"));
+            WarnL<<"not reach this";
         }
         return;
     }else{
@@ -181,9 +172,12 @@ void SrtTransport::handleHandshakeInduction(HandshakePacket &pkt, struct sockadd
     res->storeToData();
 
     registerSelfHandshake();
-    for(int i=0;i<3;++i){
-        sendControlPacket(res, true);
-    }
+    sendControlPacket(res, true);
+
+    _handleshake_timer = std::make_shared<Timer>(0.02,[this]()->bool{
+        sendControlPacket(_handleshake_res, true);
+        return true;
+    },getPoller());
 }
 
 void SrtTransport::handleHandshakeConclusion(HandshakePacket &pkt, struct sockaddr_storage *addr) {
@@ -220,7 +214,7 @@ void SrtTransport::handleHandshakeConclusion(HandshakePacket &pkt, struct sockad
             srt_flag = req->srt_flag;
             delay = delay <= req->recv_tsbpd_delay ? req->recv_tsbpd_delay : delay;
         }
-        TraceL << getIdentifier() << " CONCLUSION Phase ";
+        TraceL << getIdentifier() << " CONCLUSION Phase from"<<SockUtil::inet_ntoa((struct sockaddr *)addr) << ":" << SockUtil::inet_port((struct sockaddr *)addr);;
         HandshakePacket::Ptr res = std::make_shared<HandshakePacket>();
         res->dst_socket_id = _peer_socket_id;
         res->timestamp = DurationCountMicroseconds(_now - _start_timestamp);
@@ -244,9 +238,7 @@ void SrtTransport::handleHandshakeConclusion(HandshakePacket &pkt, struct sockad
         _handleshake_res = res;
         unregisterSelfHandshake();
         registerSelf();
-        for(int i=0;i<3;++i){
-            sendControlPacket(res, true);
-        }
+        sendControlPacket(res, true);
         TraceL << " buf size = " << res->max_flow_window_size << " init seq =" << _init_seq_number
                << " latency=" << delay;
         _recv_buf = std::make_shared<PacketRecvQueue>(getPktBufSize(), _init_seq_number, delay * 1e3,srt_flag);
@@ -255,15 +247,19 @@ void SrtTransport::handleHandshakeConclusion(HandshakePacket &pkt, struct sockad
         _buf_delay = delay;
         onHandShakeFinished(_stream_id, addr);
     } else {
-        if(isSameCon(pkt)){
-            TraceL << getIdentifier() <<" CONCLUSION repeate "<<SockUtil::inet_ntoa((struct sockaddr *)addr) << ":" << SockUtil::inet_port((struct sockaddr *)addr);
-            for(int i=0;i<3;++i){
+        if(_handleshake_res->handshake_type == HandshakePacket::HS_TYPE_CONCLUSION){
+            if(_handleshake_res->dst_socket_id != pkt.srt_socket_id){
+                TraceL << getIdentifier() <<" new connection fron client "<<SockUtil::inet_ntoa((struct sockaddr *)addr) << ":" << SockUtil::inet_port((struct sockaddr *)addr);
+                onShutdown(SockException(Err_other, "client new connection"));
+            }else{
+                TraceL << getIdentifier() <<" CONCLUSION repeate "<<SockUtil::inet_ntoa((struct sockaddr *)addr) << ":" << SockUtil::inet_port((struct sockaddr *)addr);
                 sendControlPacket(_handleshake_res, true);
             }
+
         }else{
-            TraceL << getIdentifier() <<" new connection fron client "<<SockUtil::inet_ntoa((struct sockaddr *)addr) << ":" << SockUtil::inet_port((struct sockaddr *)addr);
-            onShutdown(SockException(Err_other, "client new connection"));
+            WarnL<<"not reach this";
         }
+        return;
         
     }
     _last_ack_pkt_seq_num = _init_seq_number;
diff --git a/srt/SrtTransport.hpp b/srt/SrtTransport.hpp
index 27219fab..9d0c186c 100644
--- a/srt/SrtTransport.hpp
+++ b/srt/SrtTransport.hpp
@@ -88,8 +88,6 @@ private:
 
     size_t getPayloadSize();
 
-    bool isSameCon(HandshakePacket &pkt);
-
 protected:
     void sendDataPacket(DataPacket::Ptr pkt, char *buf, int len, bool flush = false);
     void sendControlPacket(ControlPacket::Ptr pkt, bool flush = true);
@@ -143,6 +141,8 @@ private:
     // 保持发送的握手消息，防止丢失重发
     HandshakePacket::Ptr _handleshake_res;
 
+    Timer::Ptr _handleshake_timer;
+
     ResourcePool<BufferRaw> _packet_pool;
 };