amass/ZLMediaKit
1
0
Fork 0
mirror of https://github.com/ZLMediaKit/ZLMediaKit.git synced 2024-11-22 18:50:20 +08:00
Code Issues Actions 6 Packages Projects Releases Wiki Activity

格式化srt相关代码

Browse Source
This commit is contained in:
ziyue 2022-06-07 09:52:20 +08:00
parent 83d75c9a72
commit e415230e47
20 changed files with 821 additions and 789 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
conf/config.ini
View File

@ -291,8 +291,8 @@ timeoutSec=5
#该端口是多线程的，同时支持客户端网络切换导致的连接迁移 #该端口是多线程的，同时支持客户端网络切换导致的连接迁移
port=9000 port=9000
#srt 协议中延迟缓存的估算参数，在握手阶段估算rtt ,然后lantencyMul*rtt 为最大缓存时长，此参数越大，表示等待重传的时长就越大 #srt 协议中延迟缓存的估算参数，在握手阶段估算rtt ,然后latencyMul*rtt 为最大缓存时长，此参数越大，表示等待重传的时长就越大
lantencyMul=4 latencyMul=4
[rtsp] [rtsp]

36
srt/Ack.cpp
View File

@ -2,8 +2,9 @@
#include "Common.hpp" #include "Common.hpp"
namespace SRT { namespace SRT {
bool ACKPacket::loadFromData(uint8_t *buf, size_t len) { bool ACKPacket::loadFromData(uint8_t *buf, size_t len) {
if(len < ACK_CIF_SIZE + ControlPacket::HEADER_SIZE){ if (len < ACK_CIF_SIZE + ControlPacket::HEADER_SIZE) {
return false; return false;
} }
@ -11,7 +12,7 @@ bool ACKPacket::loadFromData(uint8_t *buf, size_t len) {
_data->assign((char *)(buf), len); _data->assign((char *)(buf), len);
ControlPacket::loadHeader(); ControlPacket::loadHeader();
ack_number = loadUint32(type_specific_info); ack_number = loadUint32(type_specific_info);
uint8_t* ptr = (uint8_t*)_data->data()+ControlPacket::HEADER_SIZE; uint8_t *ptr = (uint8_t *)_data->data() + ControlPacket::HEADER_SIZE;
last_ack_pkt_seq_number = loadUint32(ptr); last_ack_pkt_seq_number = loadUint32(ptr);
ptr += 4; ptr += 4;
@ -36,6 +37,7 @@ bool ACKPacket::loadFromData(uint8_t *buf, size_t len) {
return true; return true;
} }
bool ACKPacket::storeToData() { bool ACKPacket::storeToData() {
_data = BufferRaw::create(); _data = BufferRaw::create();
_data->setCapacity(HEADER_SIZE + ACK_CIF_SIZE); _data->setCapacity(HEADER_SIZE + ACK_CIF_SIZE);
@ -43,41 +45,41 @@ bool ACKPacket::storeToData() {
control_type = ControlPacket::ACK; control_type = ControlPacket::ACK;
sub_type = 0; sub_type = 0;
storeUint32(type_specific_info,ack_number); storeUint32(type_specific_info, ack_number);
storeToHeader(); storeToHeader();
uint8_t* ptr = (uint8_t*)_data->data()+ControlPacket::HEADER_SIZE; uint8_t *ptr = (uint8_t *)_data->data() + ControlPacket::HEADER_SIZE;
storeUint32(ptr,last_ack_pkt_seq_number); storeUint32(ptr, last_ack_pkt_seq_number);
ptr += 4; ptr += 4;
storeUint32(ptr,rtt); storeUint32(ptr, rtt);
ptr += 4; ptr += 4;
storeUint32(ptr,rtt_variance); storeUint32(ptr, rtt_variance);
ptr += 4; ptr += 4;
storeUint32(ptr,pkt_recv_rate); storeUint32(ptr, pkt_recv_rate);
ptr += 4; ptr += 4;
storeUint32(ptr,available_buf_size); storeUint32(ptr, available_buf_size);
ptr += 4; ptr += 4;
storeUint32(ptr,estimated_link_capacity); storeUint32(ptr, estimated_link_capacity);
ptr += 4; ptr += 4;
storeUint32(ptr,recv_rate); storeUint32(ptr, recv_rate);
ptr += 4; ptr += 4;
return true; return true;
} }
std::string ACKPacket::dump(){ std::string ACKPacket::dump() {
_StrPrinter printer; _StrPrinter printer;
printer << "last_ack_pkt_seq_number="<<last_ack_pkt_seq_number<<\ printer << "last_ack_pkt_seq_number=" << last_ack_pkt_seq_number << " rtt=" << rtt
" rtt="<<rtt<<" rtt_variance="<<rtt_variance<<\ << " rtt_variance=" << rtt_variance << " pkt_recv_rate=" << pkt_recv_rate
" pkt_recv_rate="<<pkt_recv_rate<<" available_buf_size="<<available_buf_size<<\ << " available_buf_size=" << available_buf_size << " estimated_link_capacity=" << estimated_link_capacity
" estimated_link_capacity="<<estimated_link_capacity<<" recv_rate="<<recv_rate; << " recv_rate=" << recv_rate;
return std::move(printer); return std::move(printer);
} }
} // namespace } // namespace SRT

26
srt/Ack.hpp
View File

@ -2,8 +2,7 @@
#define ZLMEDIAKIT_SRT_ACK_H #define ZLMEDIAKIT_SRT_ACK_H
#include "Packet.hpp" #include "Packet.hpp"
namespace SRT {
namespace SRT{
/* /*
0 1 2 3 0 1 2 3
0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
@ -33,16 +32,13 @@ namespace SRT{
Figure 13: ACK control packet Figure 13: ACK control packet
https://haivision.github.io/srt-rfc/draft-sharabayko-srt.html#name-ack-acknowledgment https://haivision.github.io/srt-rfc/draft-sharabayko-srt.html#name-ack-acknowledgment
*/ */
class ACKPacket : public ControlPacket class ACKPacket : public ControlPacket {
{
public: public:
using Ptr = std::shared_ptr<ACKPacket>; using Ptr = std::shared_ptr<ACKPacket>;
ACKPacket() = default; ACKPacket() = default;
~ACKPacket() = default; ~ACKPacket() = default;
enum{ enum { ACK_CIF_SIZE = 7 * 4 };
ACK_CIF_SIZE = 7*4
};
std::string dump(); std::string dump();
///////ControlPacket override/////// ///////ControlPacket override///////
bool loadFromData(uint8_t *buf, size_t len) override; bool loadFromData(uint8_t *buf, size_t len) override;
@ -59,15 +55,14 @@ public:
uint32_t recv_rate; uint32_t recv_rate;
}; };
class ACKACKPacket : public ControlPacket {
class ACKACKPacket : public ControlPacket{
public: public:
using Ptr = std::shared_ptr<ACKACKPacket>; using Ptr = std::shared_ptr<ACKACKPacket>;
ACKACKPacket() = default; ACKACKPacket() = default;
~ACKACKPacket() = default; ~ACKACKPacket() = default;
///////ControlPacket override/////// ///////ControlPacket override///////
bool loadFromData(uint8_t *buf, size_t len) override{ bool loadFromData(uint8_t *buf, size_t len) override {
if(len < ControlPacket::HEADER_SIZE){ if (len < ControlPacket::HEADER_SIZE) {
return false; return false;
} }
_data = BufferRaw::create(); _data = BufferRaw::create();
@ -76,21 +71,20 @@ public:
ack_number = loadUint32(type_specific_info); ack_number = loadUint32(type_specific_info);
return true; return true;
} }
bool storeToData() override{ bool storeToData() override {
_data = BufferRaw::create(); _data = BufferRaw::create();
_data->setCapacity(HEADER_SIZE); _data->setCapacity(HEADER_SIZE);
_data->setSize(HEADER_SIZE ); _data->setSize(HEADER_SIZE);
control_type = ControlPacket::ACKACK; control_type = ControlPacket::ACKACK;
sub_type = 0; sub_type = 0;
storeUint32(type_specific_info,ack_number); storeUint32(type_specific_info, ack_number);
storeToHeader(); storeToHeader();
return true; return true;
} }
uint32_t ack_number; uint32_t ack_number;
}; };
} //namespace SRT } // namespace SRT
#endif // ZLMEDIAKIT_SRT_ACK_H #endif // ZLMEDIAKIT_SRT_ACK_H

50
srt/Common.hpp
View File

@ -2,83 +2,73 @@
#define ZLMEDIAKIT_SRT_COMMON_H #define ZLMEDIAKIT_SRT_COMMON_H
#include <chrono> #include <chrono>
namespace SRT namespace SRT {
{
using SteadyClock = std::chrono::steady_clock; using SteadyClock = std::chrono::steady_clock;
using TimePoint = std::chrono::time_point<SteadyClock>; using TimePoint = std::chrono::time_point<SteadyClock>;
using Microseconds = std::chrono::microseconds; using Microseconds = std::chrono::microseconds;
using Milliseconds = std::chrono::milliseconds; using Milliseconds = std::chrono::milliseconds;
inline int64_t DurationCountMicroseconds( SteadyClock::duration dur){ static inline int64_t DurationCountMicroseconds(SteadyClock::duration dur) {
return std::chrono::duration_cast<std::chrono::microseconds>(dur).count(); return std::chrono::duration_cast<std::chrono::microseconds>(dur).count();
} }
inline uint32_t loadUint32(uint8_t *ptr) { static inline uint32_t loadUint32(uint8_t *ptr) {
return ptr[0] << 24 | ptr[1] << 16 | ptr[2] << 8 | ptr[3]; return ptr[0] << 24 | ptr[1] << 16 | ptr[2] << 8 | ptr[3];
} }
inline uint16_t loadUint16(uint8_t *ptr) {
static inline uint16_t loadUint16(uint8_t *ptr) {
return ptr[0] << 8 | ptr[1]; return ptr[0] << 8 | ptr[1];
} }
inline void storeUint32(uint8_t *buf, uint32_t val) { static inline void storeUint32(uint8_t *buf, uint32_t val) {
buf[0] = val >> 24; buf[0] = val >> 24;
buf[1] = (val >> 16) & 0xff; buf[1] = (val >> 16) & 0xff;
buf[2] = (val >> 8) & 0xff; buf[2] = (val >> 8) & 0xff;
buf[3] = val & 0xff; buf[3] = val & 0xff;
} }
inline void storeUint16(uint8_t *buf, uint16_t val) { static inline void storeUint16(uint8_t *buf, uint16_t val) {
buf[0] = (val >> 8) & 0xff; buf[0] = (val >> 8) & 0xff;
buf[1] = val & 0xff; buf[1] = val & 0xff;
} }
inline void storeUint32LE(uint8_t *buf, uint32_t val) { static inline void storeUint32LE(uint8_t *buf, uint32_t val) {
buf[0] = val & 0xff; buf[0] = val & 0xff;
buf[1] = (val >> 8) & 0xff; buf[1] = (val >> 8) & 0xff;
buf[2] = (val >> 16) & 0xff; buf[2] = (val >> 16) & 0xff;
buf[3] = (val >>24) & 0xff; buf[3] = (val >> 24) & 0xff;
} }
inline void storeUint16LE(uint8_t *buf, uint16_t val) { static inline void storeUint16LE(uint8_t *buf, uint16_t val) {
buf[0] = val & 0xff; buf[0] = val & 0xff;
buf[1] = (val>>8) & 0xff; buf[1] = (val >> 8) & 0xff;
} }
inline uint32_t srtVersion(int major, int minor, int patch) static inline uint32_t srtVersion(int major, int minor, int patch) {
{ return patch + minor * 0x100 + major * 0x10000;
return patch + minor*0x100 + major*0x10000;
} }
class UTicker { class UTicker {
public: public:
UTicker() { UTicker() { _created = _begin = SteadyClock::now(); }
_created = _begin = SteadyClock::now(); ~UTicker() = default;
}
~UTicker() {
}
/** /**
* 获取创建时间，单位微妙 * 获取创建时间，单位微妙
*/ */
int64_t elapsedTime(TimePoint now) const { int64_t elapsedTime(TimePoint now) const { return DurationCountMicroseconds(now - _begin); }
return DurationCountMicroseconds(now - _begin);
}
/** /**
* 获取上次resetTime后至今的时间，单位毫秒 * 获取上次resetTime后至今的时间，单位毫秒
*/ */
int64_t createdTime(TimePoint now) const { int64_t createdTime(TimePoint now) const { return DurationCountMicroseconds(now - _created); }
return DurationCountMicroseconds(now - _created);
}
/** /**
* 重置计时器 * 重置计时器
*/ */
void resetTime(TimePoint now) { void resetTime(TimePoint now) { _begin = now; }
_begin = now;
}
private: private:
TimePoint _begin; TimePoint _begin;
@ -87,4 +77,4 @@ private:
} // namespace SRT } // namespace SRT
#endif //ZLMEDIAKIT_SRT_COMMON_H #endif // ZLMEDIAKIT_SRT_COMMON_H

87
srt/HSExt.cpp
View File

@ -1,19 +1,20 @@
#include "HSExt.hpp" #include "HSExt.hpp"
namespace SRT { namespace SRT {
bool HSExtMessage::loadFromData(uint8_t *buf, size_t len) { bool HSExtMessage::loadFromData(uint8_t *buf, size_t len) {
if(buf == NULL || len != HSEXT_MSG_SIZE){ if (buf == NULL || len != HSEXT_MSG_SIZE) {
return false; return false;
} }
_data = BufferRaw::create(); _data = BufferRaw::create();
_data->assign((char*)buf,len); _data->assign((char *)buf, len);
extension_length = 3; extension_length = 3;
HSExt::loadHeader(); HSExt::loadHeader();
assert(extension_type == SRT_CMD_HSREQ || extension_type == SRT_CMD_HSRSP); assert(extension_type == SRT_CMD_HSREQ || extension_type == SRT_CMD_HSRSP);
uint8_t* ptr = (uint8_t*)_data->data()+4; uint8_t *ptr = (uint8_t *)_data->data() + 4;
srt_version = loadUint32(ptr); srt_version = loadUint32(ptr);
ptr += 4; ptr += 4;
@ -27,105 +28,107 @@ bool HSExtMessage::loadFromData(uint8_t *buf, size_t len) {
ptr += 2; ptr += 2;
return true; return true;
}
} std::string HSExtMessage::dump() {
std::string HSExtMessage::dump(){
_StrPrinter printer; _StrPrinter printer;
printer << "srt version : "<<std::hex<<srt_version<<" srt flag : "<<std::hex<<srt_flag<<\ printer << "srt version : " << std::hex << srt_version << " srt flag : " << std::hex << srt_flag
" recv_tsbpd_delay="<<recv_tsbpd_delay<<" send_tsbpd_delay = "<<send_tsbpd_delay; << " recv_tsbpd_delay=" << recv_tsbpd_delay << " send_tsbpd_delay = " << send_tsbpd_delay;
return std::move(printer); return std::move(printer);
} }
bool HSExtMessage::storeToData() { bool HSExtMessage::storeToData() {
_data = BufferRaw::create(); _data = BufferRaw::create();
_data->setCapacity(HSEXT_MSG_SIZE); _data->setCapacity(HSEXT_MSG_SIZE);
_data->setSize(HSEXT_MSG_SIZE); _data->setSize(HSEXT_MSG_SIZE);
extension_length = 3; extension_length = 3;
HSExt::storeHeader(); HSExt::storeHeader();
uint8_t* ptr = (uint8_t*)_data->data()+4; uint8_t *ptr = (uint8_t *)_data->data() + 4;
storeUint32(ptr,srt_version); storeUint32(ptr, srt_version);
ptr += 4; ptr += 4;
storeUint32(ptr,srt_flag); storeUint32(ptr, srt_flag);
ptr += 4; ptr += 4;
storeUint16(ptr,recv_tsbpd_delay); storeUint16(ptr, recv_tsbpd_delay);
ptr += 2; ptr += 2;
storeUint16(ptr,send_tsbpd_delay); storeUint16(ptr, send_tsbpd_delay);
ptr += 2; ptr += 2;
return true; return true;
} }
bool HSExtStreamID::loadFromData(uint8_t *buf, size_t len) { bool HSExtStreamID::loadFromData(uint8_t *buf, size_t len) {
if(buf == NULL || len < 4){ if (buf == NULL || len < 4) {
return false; return false;
} }
_data = BufferRaw::create(); _data = BufferRaw::create();
_data->assign((char*)buf,len); _data->assign((char *)buf, len);
HSExt::loadHeader(); HSExt::loadHeader();
size_t content_size = extension_length*4; size_t content_size = extension_length * 4;
if(len < content_size+4){ if (len < content_size + 4) {
return false; return false;
} }
streamid.clear(); streamid.clear();
char* ptr = _data->data()+4; char *ptr = _data->data() + 4;
for(size_t i = 0; i<extension_length; ++i){ for (size_t i = 0; i < extension_length; ++i) {
streamid.push_back(*(ptr+3)); streamid.push_back(*(ptr + 3));
streamid.push_back(*(ptr+2)); streamid.push_back(*(ptr + 2));
streamid.push_back(*(ptr+1)); streamid.push_back(*(ptr + 1));
streamid.push_back(*(ptr)); streamid.push_back(*(ptr));
ptr+=4; ptr += 4;
} }
char zero = 0x00; char zero = 0x00;
if(streamid.back() == zero){ if (streamid.back() == zero) {
streamid.erase(streamid.find_first_of(zero),streamid.size()); streamid.erase(streamid.find_first_of(zero), streamid.size());
} }
return true; return true;
}
}
bool HSExtStreamID::storeToData() { bool HSExtStreamID::storeToData() {
size_t content_size = ((streamid.length()+4)+3)/4*4; size_t content_size = ((streamid.length() + 4) + 3) / 4 * 4;
_data = BufferRaw::create(); _data = BufferRaw::create();
_data->setCapacity(content_size); _data->setCapacity(content_size);
_data->setSize(content_size); _data->setSize(content_size);
extension_length = (content_size-4)/4; extension_length = (content_size - 4) / 4;
extension_type = SRT_CMD_SID; extension_type = SRT_CMD_SID;
HSExt::storeHeader(); HSExt::storeHeader();
auto ptr = _data->data()+4; auto ptr = _data->data() + 4;
memset(ptr,0,extension_length*4); memset(ptr, 0, extension_length * 4);
const char* src = streamid.c_str(); const char *src = streamid.c_str();
for(size_t i = 0; i< streamid.length()/4;++i){ for (size_t i = 0; i < streamid.length() / 4; ++i) {
*ptr = *(src+3+i*4); *ptr = *(src + 3 + i * 4);
ptr++; ptr++;
*ptr = *(src+2+i*4); *ptr = *(src + 2 + i * 4);
ptr++; ptr++;
*ptr = *(src+1+i*4); *ptr = *(src + 1 + i * 4);
ptr++; ptr++;
*ptr = *(src+0+i*4); *ptr = *(src + 0 + i * 4);
ptr++; ptr++;
} }
ptr += 3; ptr += 3;
size_t offset = streamid.length()/4*4; size_t offset = streamid.length() / 4 * 4;
for(size_t i = 0; i<streamid.length()%4;++i){ for (size_t i = 0; i < streamid.length() % 4; ++i) {
*ptr = *(src+offset+i); *ptr = *(src + offset + i);
ptr -= 1; ptr -= 1;
} }
return true; return true;
} }
std::string HSExtStreamID::dump(){
std::string HSExtStreamID::dump() {
_StrPrinter printer; _StrPrinter printer;
printer << " streamid : "<< streamid; printer << " streamid : " << streamid;
return std::move(printer); return std::move(printer);
} }
} // namespace SRT } // namespace SRT

5
srt/HSExt.hpp
View File

@ -11,6 +11,7 @@ class HSExt : public Buffer {
public: public:
HSExt() = default; HSExt() = default;
virtual ~HSExt() = default; virtual ~HSExt() = default;
enum { enum {
SRT_CMD_REJECT = 0, SRT_CMD_REJECT = 0,
SRT_CMD_HSREQ = 1, SRT_CMD_HSREQ = 1,
@ -36,13 +37,13 @@ public:
return _data->data(); return _data->data();
} }
return nullptr; return nullptr;
}; }
size_t size() const override { size_t size() const override {
if (_data) { if (_data) {
return _data->size(); return _data->size();
} }
return 0; return 0;
}; }
protected: protected:
void loadHeader() { void loadHeader() {

222
srt/Packet.cpp
View File

@ -10,19 +10,13 @@
#include <netdb.h> #include <netdb.h>
#endif // defined(_WIN32) #endif // defined(_WIN32)
#include <atomic> #include <atomic>
#include "Util/logger.h" #include "Util/logger.h"
#include "Util/MD5.h" #include "Util/MD5.h"
#include "Packet.hpp" #include "Packet.hpp"
namespace SRT { namespace SRT {
const size_t DataPacket::HEADER_SIZE; const size_t DataPacket::HEADER_SIZE;
const size_t ControlPacket::HEADER_SIZE; const size_t ControlPacket::HEADER_SIZE;
const size_t HandshakePacket::HS_CONTENT_MIN_SIZE; const size_t HandshakePacket::HS_CONTENT_MIN_SIZE;
@ -38,7 +32,7 @@ bool DataPacket::isDataPacket(uint8_t *buf, size_t len) {
return false; return false;
} }
uint32_t DataPacket::getSocketID(uint8_t *buf, size_t len){ uint32_t DataPacket::getSocketID(uint8_t *buf, size_t len) {
uint8_t *ptr = buf; uint8_t *ptr = buf;
ptr += 12; ptr += 12;
return loadUint32(ptr); return loadUint32(ptr);
@ -51,7 +45,7 @@ bool DataPacket::loadFromData(uint8_t *buf, size_t len) {
} }
uint8_t *ptr = buf; uint8_t *ptr = buf;
f = ptr[0] >> 7; f = ptr[0] >> 7;
packet_seq_number = loadUint32(ptr)&0x7fffffff; packet_seq_number = loadUint32(ptr) & 0x7fffffff;
ptr += 4; ptr += 4;
PP = ptr[0] >> 6; PP = ptr[0] >> 6;
@ -71,7 +65,8 @@ bool DataPacket::loadFromData(uint8_t *buf, size_t len) {
_data->assign((char *)(buf), len); _data->assign((char *)(buf), len);
return true; return true;
} }
bool DataPacket::storeToHeader(){
bool DataPacket::storeToHeader() {
if (!_data || _data->size() < HEADER_SIZE) { if (!_data || _data->size() < HEADER_SIZE) {
WarnL << "data size less " << HEADER_SIZE; WarnL << "data size less " << HEADER_SIZE;
return false; return false;
@ -101,6 +96,7 @@ bool DataPacket::storeToHeader(){
ptr += 4; ptr += 4;
return true; return true;
} }
bool DataPacket::storeToData(uint8_t *buf, size_t len) { bool DataPacket::storeToData(uint8_t *buf, size_t len) {
_data = BufferRaw::create(); _data = BufferRaw::create();
_data->setCapacity(len + HEADER_SIZE); _data->setCapacity(len + HEADER_SIZE);
@ -139,6 +135,7 @@ char *DataPacket::data() const {
return nullptr; return nullptr;
return _data->data(); return _data->data();
} }
size_t DataPacket::size() const { size_t DataPacket::size() const {
if (!_data) { if (!_data) {
return 0; return 0;
@ -151,6 +148,7 @@ char *DataPacket::payloadData() {
return nullptr; return nullptr;
return _data->data() + HEADER_SIZE; return _data->data() + HEADER_SIZE;
} }
size_t DataPacket::payloadSize() { size_t DataPacket::payloadSize() {
if (!_data) { if (!_data) {
return 0; return 0;
@ -158,8 +156,6 @@ size_t DataPacket::payloadSize() {
return _data->size() - HEADER_SIZE; return _data->size() - HEADER_SIZE;
} }
bool ControlPacket::isControlPacket(uint8_t *buf, size_t len) { bool ControlPacket::isControlPacket(uint8_t *buf, size_t len) {
if (len < HEADER_SIZE) { if (len < HEADER_SIZE) {
WarnL << "data size" << len << " less " << HEADER_SIZE; WarnL << "data size" << len << " less " << HEADER_SIZE;
@ -199,6 +195,7 @@ bool ControlPacket::loadHeader() {
ptr += 4; ptr += 4;
return true; return true;
} }
bool ControlPacket::storeToHeader() { bool ControlPacket::storeToHeader() {
uint8_t *ptr = (uint8_t *)_data->data(); uint8_t *ptr = (uint8_t *)_data->data();
ptr[0] = 0x80; ptr[0] = 0x80;
@ -228,17 +225,20 @@ char *ControlPacket::data() const {
return nullptr; return nullptr;
return _data->data(); return _data->data();
} }
size_t ControlPacket::size() const { size_t ControlPacket::size() const {
if (!_data) { if (!_data) {
return 0; return 0;
} }
return _data->size(); return _data->size();
} }
uint32_t ControlPacket::getSocketID(uint8_t *buf, size_t len){
return loadUint32(buf+12); uint32_t ControlPacket::getSocketID(uint8_t *buf, size_t len) {
return loadUint32(buf + 12);
} }
bool HandshakePacket::loadFromData(uint8_t *buf, size_t len) { bool HandshakePacket::loadFromData(uint8_t *buf, size_t len) {
if(HEADER_SIZE+HS_CONTENT_MIN_SIZE > len){ if (HEADER_SIZE + HS_CONTENT_MIN_SIZE > len) {
ErrorL << "size too smalle " << encryption_field; ErrorL << "size too smalle " << encryption_field;
return false; return false;
} }
@ -282,79 +282,75 @@ bool HandshakePacket::loadFromData(uint8_t *buf, size_t len) {
ErrorL << "not support encryption " << encryption_field; ErrorL << "not support encryption " << encryption_field;
} }
if(extension_field == 0){ if (extension_field == 0) {
return true; return true;
} }
if(len == HEADER_SIZE+HS_CONTENT_MIN_SIZE){ if (len == HEADER_SIZE + HS_CONTENT_MIN_SIZE) {
//ErrorL << "extension filed not exist " << extension_field; // ErrorL << "extension filed not exist " << extension_field;
return true; return true;
} }
return loadExtMessage(ptr,len-HS_CONTENT_MIN_SIZE-HEADER_SIZE); return loadExtMessage(ptr, len - HS_CONTENT_MIN_SIZE - HEADER_SIZE);
} }
bool HandshakePacket::loadExtMessage(uint8_t *buf,size_t len){
uint8_t* ptr = buf; bool HandshakePacket::loadExtMessage(uint8_t *buf, size_t len) {
uint8_t *ptr = buf;
ext_list.clear(); ext_list.clear();
uint16_t type; uint16_t type;
uint16_t length; uint16_t length;
HSExt::Ptr ext; HSExt::Ptr ext;
while(ptr<buf+len){ while (ptr < buf + len) {
type = loadUint16(ptr); type = loadUint16(ptr);
length = loadUint16(ptr+2); length = loadUint16(ptr + 2);
switch (type) switch (type) {
{
case HSExt::SRT_CMD_HSREQ: case HSExt::SRT_CMD_HSREQ:
case HSExt::SRT_CMD_HSRSP: case HSExt::SRT_CMD_HSRSP: ext = std::make_shared<HSExtMessage>(); break;
ext = std::make_shared<HSExtMessage>(); case HSExt::SRT_CMD_SID: ext = std::make_shared<HSExtStreamID>(); break;
break;
case HSExt::SRT_CMD_SID:
ext = std::make_shared<HSExtStreamID>();
break;
default: default:
WarnL<<"not support ext "<<type; WarnL << "not support ext " << type;
break; break;
} }
if(ext){ if (ext) {
if(ext->loadFromData(ptr,length*4+4)){ if (ext->loadFromData(ptr, length * 4 + 4)) {
ext_list.push_back(std::move(ext)); ext_list.push_back(std::move(ext));
}else{ } else {
WarnL<<"parse HS EXT failed type="<<type<<" len="<<length; WarnL << "parse HS EXT failed type=" << type << " len=" << length;
} }
ext = nullptr; ext = nullptr;
} }
ptr += length*4+4; ptr += length * 4 + 4;
} }
return true; return true;
} }
bool HandshakePacket::storeExtMessage() bool HandshakePacket::storeExtMessage() {
{ uint8_t *buf = (uint8_t *)_data->data() + HEADER_SIZE + 48;
uint8_t* buf = (uint8_t*)_data->data()+HEADER_SIZE+48; size_t len = _data->size() - HEADER_SIZE - 48;
size_t len = _data->size()- HEADER_SIZE-48; for (auto ex : ext_list) {
for(auto ex : ext_list){ memcpy(buf, ex->data(), ex->size());
memcpy(buf,ex->data(),ex->size());
buf += ex->size(); buf += ex->size();
} }
return true; return true;
} }
size_t HandshakePacket::getExtSize(){ size_t HandshakePacket::getExtSize() {
size_t size = 0; size_t size = 0;
for(auto it : ext_list){ for (auto it : ext_list) {
size += it->size(); size += it->size();
} }
return size; return size;
} }
bool HandshakePacket::storeToData() { bool HandshakePacket::storeToData() {
_data = BufferRaw::create(); _data = BufferRaw::create();
for(auto ex : ext_list){ for (auto ex : ext_list) {
ex->storeToData(); ex->storeToData();
} }
auto ext_size = getExtSize(); auto ext_size = getExtSize();
_data->setCapacity(HEADER_SIZE + 48+ext_size); _data->setCapacity(HEADER_SIZE + 48 + ext_size);
_data->setSize(HEADER_SIZE + 48+ext_size); _data->setSize(HEADER_SIZE + 48 + ext_size);
control_type = ControlPacket::HANDSHAKE; control_type = ControlPacket::HANDSHAKE;
sub_type = 0; sub_type = 0;
@ -399,54 +395,53 @@ bool HandshakePacket::storeToData() {
assert(encryption_field == NO_ENCRYPTION); assert(encryption_field == NO_ENCRYPTION);
return storeExtMessage(); return storeExtMessage();
} }
bool HandshakePacket::isHandshakePacket(uint8_t *buf, size_t len){ bool HandshakePacket::isHandshakePacket(uint8_t *buf, size_t len) {
if(!ControlPacket::isControlPacket(buf,len)){ if (!ControlPacket::isControlPacket(buf, len)) {
return false; return false;
} }
if(len < HEADER_SIZE+48){ if (len < HEADER_SIZE + 48) {
return false; return false;
} }
return ControlPacket::getControlType(buf,len) == HANDSHAKE; return ControlPacket::getControlType(buf, len) == HANDSHAKE;
} }
uint32_t HandshakePacket::getHandshakeType(uint8_t *buf, size_t len){ uint32_t HandshakePacket::getHandshakeType(uint8_t *buf, size_t len) {
uint8_t *ptr = buf+HEADER_SIZE+5*4; uint8_t *ptr = buf + HEADER_SIZE + 5 * 4;
return loadUint32(ptr); return loadUint32(ptr);
} }
uint32_t HandshakePacket::getSynCookie(uint8_t *buf, size_t len){ uint32_t HandshakePacket::getSynCookie(uint8_t *buf, size_t len) {
uint8_t *ptr = buf+HEADER_SIZE+7*4; uint8_t *ptr = buf + HEADER_SIZE + 7 * 4;
return loadUint32(ptr); return loadUint32(ptr);
} }
void HandshakePacket::assignPeerIP(struct sockaddr_storage* addr){
memset(peer_ip_addr,0,sizeof(peer_ip_addr)*sizeof(peer_ip_addr[0])); void HandshakePacket::assignPeerIP(struct sockaddr_storage *addr) {
if(addr->ss_family == AF_INET){ memset(peer_ip_addr, 0, sizeof(peer_ip_addr) * sizeof(peer_ip_addr[0]));
struct sockaddr_in * ipv4 = (struct sockaddr_in *)addr; if (addr->ss_family == AF_INET) {
struct sockaddr_in *ipv4 = (struct sockaddr_in *)addr;
//抓包 奇怪好像是小头端？？？ //抓包 奇怪好像是小头端？？？
storeUint32LE(peer_ip_addr,ipv4->sin_addr.s_addr); storeUint32LE(peer_ip_addr, ipv4->sin_addr.s_addr);
}else if(addr->ss_family == AF_INET6){ } else if (addr->ss_family == AF_INET6) {
if (IN6_IS_ADDR_V4MAPPED(&((struct sockaddr_in6 *)addr)->sin6_addr)) { if (IN6_IS_ADDR_V4MAPPED(&((struct sockaddr_in6 *)addr)->sin6_addr)) {
struct in_addr addr4; struct in_addr addr4;
memcpy(&addr4, 12 + (char *)&(((struct sockaddr_in6 *)addr)->sin6_addr), 4); memcpy(&addr4, 12 + (char *)&(((struct sockaddr_in6 *)addr)->sin6_addr), 4);
storeUint32LE(peer_ip_addr,addr4.s_addr); storeUint32LE(peer_ip_addr, addr4.s_addr);
}else{ } else {
const sockaddr_in6* ipv6 = (struct sockaddr_in6 *)addr; const sockaddr_in6 *ipv6 = (struct sockaddr_in6 *)addr;
memcpy(peer_ip_addr,ipv6->sin6_addr.s6_addr,sizeof(peer_ip_addr)*sizeof(peer_ip_addr[0])); memcpy(peer_ip_addr, ipv6->sin6_addr.s6_addr, sizeof(peer_ip_addr) * sizeof(peer_ip_addr[0]));
} }
} }
} }
uint32_t HandshakePacket::generateSynCookie(struct sockaddr_storage* addr,TimePoint ts,uint32_t current_cookie, int correction ){
static std::atomic<uint32_t> distractor{0}; uint32_t HandshakePacket::generateSynCookie(
struct sockaddr_storage *addr, TimePoint ts, uint32_t current_cookie, int correction) {
static std::atomic<uint32_t> distractor { 0 };
uint32_t rollover = distractor.load() + 10; uint32_t rollover = distractor.load() + 10;
for (;;) while (true) {
{
// SYN cookie // SYN cookie
char clienthost[NI_MAXHOST]; char clienthost[NI_MAXHOST];
char clientport[NI_MAXSERV]; char clientport[NI_MAXSERV];
@ -466,31 +461,33 @@ uint32_t HandshakePacket::generateSynCookie(struct sockaddr_storage* addr,TimePo
uint32_t cookie_val; uint32_t cookie_val;
}; };
MD5 md5(cookiestr.str()); MD5 md5(cookiestr.str());
memcpy(cookie,md5.rawdigest().c_str(),16); memcpy(cookie, md5.rawdigest().c_str(), 16);
if (cookie_val != current_cookie) if (cookie_val != current_cookie) {
return cookie_val; return cookie_val;
}
++distractor; ++distractor;
// This is just to make the loop formally breakable, // This is just to make the loop formally breakable,
// but this is virtually impossible to happen. // but this is virtually impossible to happen.
if (distractor == rollover) if (distractor == rollover) {
return cookie_val; return cookie_val;
} }
}
} }
bool KeepLivePacket::loadFromData(uint8_t *buf, size_t len){ bool KeepLivePacket::loadFromData(uint8_t *buf, size_t len) {
if (len < HEADER_SIZE) { if (len < HEADER_SIZE) {
WarnL << "data size" << len << " less " << HEADER_SIZE; WarnL << "data size" << len << " less " << HEADER_SIZE;
return false; return false;
} }
_data = BufferRaw::create(); _data = BufferRaw::create();
_data->assign((char*)buf,len); _data->assign((char *)buf, len);
return loadHeader(); return loadHeader();
} }
bool KeepLivePacket::storeToData(){ bool KeepLivePacket::storeToData() {
control_type = ControlPacket::KEEPALIVE; control_type = ControlPacket::KEEPALIVE;
sub_type = 0; sub_type = 0;
@ -506,22 +503,21 @@ bool NAKPacket::loadFromData(uint8_t *buf, size_t len) {
return false; return false;
} }
_data = BufferRaw::create(); _data = BufferRaw::create();
_data->assign((char*)buf,len); _data->assign((char *)buf, len);
loadHeader(); loadHeader();
uint8_t* ptr = (uint8_t*)_data->data()+HEADER_SIZE; uint8_t *ptr = (uint8_t *)_data->data() + HEADER_SIZE;
uint8_t* end = (uint8_t*)_data->data()+_data->size(); uint8_t *end = (uint8_t *)_data->data() + _data->size();
LostPair lost; LostPair lost;
while (ptr<end) while (ptr < end) {
{ if ((*ptr) & 0x80) {
if((*ptr)&0x80){ lost.first = loadUint32(ptr) & 0x7fffffff;
lost.first = loadUint32(ptr)&0x7fffffff; lost.second = loadUint32(ptr + 4) & 0x7fffffff;
lost.second = loadUint32(ptr+4)&0x7fffffff;
lost.second += 1; lost.second += 1;
ptr += 8; ptr += 8;
}else{ } else {
lost.first = loadUint32(ptr); lost.first = loadUint32(ptr);
lost.second = lost.first +1; lost.second = lost.first + 1;
ptr += 4; ptr += 4;
} }
lost_list.push_back(lost); lost_list.push_back(lost);
@ -534,24 +530,24 @@ bool NAKPacket::storeToData() {
size_t cif_size = getCIFSize(); size_t cif_size = getCIFSize();
_data = BufferRaw::create(); _data = BufferRaw::create();
_data->setCapacity(HEADER_SIZE+cif_size); _data->setCapacity(HEADER_SIZE + cif_size);
_data->setSize(HEADER_SIZE+cif_size); _data->setSize(HEADER_SIZE + cif_size);
storeToHeader(); storeToHeader();
uint8_t* ptr = (uint8_t*)_data->data()+HEADER_SIZE; uint8_t *ptr = (uint8_t *)_data->data() + HEADER_SIZE;
for(auto it : lost_list){ for (auto it : lost_list) {
if(it.first+1 ==it.second){ if (it.first + 1 == it.second) {
storeUint32(ptr,it.first); storeUint32(ptr, it.first);
ptr[0] = ptr[0]&0x7f; ptr[0] = ptr[0] & 0x7f;
ptr += 4; ptr += 4;
}else{ } else {
storeUint32(ptr,it.first); storeUint32(ptr, it.first);
ptr[0] |= 0x80; ptr[0] |= 0x80;
storeUint32(ptr+4,it.second-1); storeUint32(ptr + 4, it.second - 1);
//ptr[4] = ptr[4]&0x7f; // ptr[4] = ptr[4]&0x7f;
ptr += 8; ptr += 8;
} }
@ -560,36 +556,36 @@ bool NAKPacket::storeToData() {
return true; return true;
} }
size_t NAKPacket::getCIFSize(){ size_t NAKPacket::getCIFSize() {
size_t size = 0; size_t size = 0;
for(auto it : lost_list){ for (auto it : lost_list) {
if(it.first+1 ==it.second){ if (it.first + 1 == it.second) {
size += 4; size += 4;
}else{ } else {
size += 8; size += 8;
} }
} }
return size; return size;
} }
std::string NAKPacket::dump(){ std::string NAKPacket::dump() {
_StrPrinter printer; _StrPrinter printer;
for (auto it : lost_list) { for (auto it : lost_list) {
printer<<"[ "<<it.first<<" , "<<it.second-1<<" ]"; printer << "[ " << it.first << " , " << it.second - 1 << " ]";
} }
return std::move(printer); return std::move(printer);
} }
bool MsgDropReqPacket::loadFromData(uint8_t *buf, size_t len) { bool MsgDropReqPacket::loadFromData(uint8_t *buf, size_t len) {
if (len < HEADER_SIZE+8) { if (len < HEADER_SIZE + 8) {
WarnL << "data size" << len << " less " << HEADER_SIZE; WarnL << "data size" << len << " less " << HEADER_SIZE;
return false; return false;
} }
_data = BufferRaw::create(); _data = BufferRaw::create();
_data->assign((char*)buf,len); _data->assign((char *)buf, len);
loadHeader(); loadHeader();
uint8_t* ptr = (uint8_t*)_data->data()+HEADER_SIZE; uint8_t *ptr = (uint8_t *)_data->data() + HEADER_SIZE;
first_pkt_seq_num = loadUint32(ptr); first_pkt_seq_num = loadUint32(ptr);
ptr += 4; ptr += 4;
@ -602,17 +598,17 @@ bool MsgDropReqPacket::storeToData() {
control_type = DROPREQ; control_type = DROPREQ;
sub_type = 0; sub_type = 0;
_data = BufferRaw::create(); _data = BufferRaw::create();
_data->setCapacity(HEADER_SIZE+8); _data->setCapacity(HEADER_SIZE + 8);
_data->setSize(HEADER_SIZE+8); _data->setSize(HEADER_SIZE + 8);
storeToHeader(); storeToHeader();
uint8_t* ptr = (uint8_t*)_data->data()+HEADER_SIZE; uint8_t *ptr = (uint8_t *)_data->data() + HEADER_SIZE;
storeUint32(ptr,first_pkt_seq_num); storeUint32(ptr, first_pkt_seq_num);
ptr += 4; ptr += 4;
storeUint32(ptr,last_pkt_seq_num); storeUint32(ptr, last_pkt_seq_num);
ptr += 4; ptr += 4;
return true; return true;
} }

34
srt/Packet.hpp
View File

@ -5,6 +5,7 @@
#include <vector> #include <vector>
#include "Network/Buffer.h" #include "Network/Buffer.h"
#include "Network/sockutil.h"
#include "Util/logger.h" #include "Util/logger.h"
#include "Common.hpp" #include "Common.hpp"
@ -182,17 +183,15 @@ public:
enum { HS_EXT_FILED_HSREQ = 0x00000001, HS_EXT_FILED_KMREQ = 0x00000002, HS_EXT_FILED_CONFIG = 0x00000004 }; enum { HS_EXT_FILED_HSREQ = 0x00000001, HS_EXT_FILED_KMREQ = 0x00000002, HS_EXT_FILED_CONFIG = 0x00000004 };
HandshakePacket() = default; HandshakePacket() = default;
~HandshakePacket() = default; ~HandshakePacket() = default;
static bool isHandshakePacket(uint8_t *buf, size_t len); static bool isHandshakePacket(uint8_t *buf, size_t len);
static uint32_t getHandshakeType(uint8_t *buf, size_t len); static uint32_t getHandshakeType(uint8_t *buf, size_t len);
static uint32_t getSynCookie(uint8_t *buf, size_t len); static uint32_t getSynCookie(uint8_t *buf, size_t len);
static uint32_t generateSynCookie(struct sockaddr_storage* addr,TimePoint ts,uint32_t current_cookie = 0, int correction = 0); static uint32_t generateSynCookie(struct sockaddr_storage *addr, TimePoint ts, uint32_t current_cookie = 0, int correction = 0);
void assignPeerIP(struct sockaddr_storage* addr); void assignPeerIP(struct sockaddr_storage *addr);
///////ControlPacket override/////// ///////ControlPacket override///////
bool loadFromData(uint8_t *buf, size_t len) override; bool loadFromData(uint8_t *buf, size_t len) override;
bool storeToData() override; bool storeToData() override;
@ -209,8 +208,9 @@ public:
uint8_t peer_ip_addr[16]; uint8_t peer_ip_addr[16];
std::vector<HSExt::Ptr> ext_list; std::vector<HSExt::Ptr> ext_list;
private: private:
bool loadExtMessage(uint8_t *buf,size_t len); bool loadExtMessage(uint8_t *buf, size_t len);
bool storeExtMessage(); bool storeExtMessage();
size_t getExtSize(); size_t getExtSize();
}; };
@ -229,8 +229,7 @@ private:
Figure 12: Keep-Alive control packet Figure 12: Keep-Alive control packet
https://haivision.github.io/srt-rfc/draft-sharabayko-srt.html#name-keep-alive https://haivision.github.io/srt-rfc/draft-sharabayko-srt.html#name-keep-alive
*/ */
class KeepLivePacket : public ControlPacket class KeepLivePacket : public ControlPacket {
{
public: public:
using Ptr = std::shared_ptr<KeepLivePacket>; using Ptr = std::shared_ptr<KeepLivePacket>;
KeepLivePacket() = default; KeepLivePacket() = default;
@ -265,11 +264,10 @@ An SRT NAK packet is formatted as follows:
Figure 14: NAK control packet Figure 14: NAK control packet
https://haivision.github.io/srt-rfc/draft-sharabayko-srt.html#name-nak-control-packet https://haivision.github.io/srt-rfc/draft-sharabayko-srt.html#name-nak-control-packet
*/ */
class NAKPacket : public ControlPacket class NAKPacket : public ControlPacket {
{
public: public:
using Ptr = std::shared_ptr<NAKPacket>; using Ptr = std::shared_ptr<NAKPacket>;
using LostPair = std::pair<uint32_t,uint32_t>; using LostPair = std::pair<uint32_t, uint32_t>;
NAKPacket() = default; NAKPacket() = default;
~NAKPacket() = default; ~NAKPacket() = default;
std::string dump(); std::string dump();
@ -278,11 +276,11 @@ public:
bool storeToData() override; bool storeToData() override;
std::list<LostPair> lost_list; std::list<LostPair> lost_list;
private: private:
size_t getCIFSize(); size_t getCIFSize();
}; };
/* /*
0 1 2 3 0 1 2 3
0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
@ -302,9 +300,8 @@ private:
Figure 18: Drop Request control packet Figure 18: Drop Request control packet
https://haivision.github.io/srt-rfc/draft-sharabayko-srt.html#name-message-drop-request https://haivision.github.io/srt-rfc/draft-sharabayko-srt.html#name-message-drop-request
*/ */
class MsgDropReqPacket : public ControlPacket class MsgDropReqPacket : public ControlPacket {
{ public:
public:
using Ptr = std::shared_ptr<MsgDropReqPacket>; using Ptr = std::shared_ptr<MsgDropReqPacket>;
MsgDropReqPacket() = default; MsgDropReqPacket() = default;
~MsgDropReqPacket() = default; ~MsgDropReqPacket() = default;
@ -332,12 +329,12 @@ class MsgDropReqPacket : public ControlPacket
https://haivision.github.io/srt-rfc/draft-sharabayko-srt.html#name-shutdown https://haivision.github.io/srt-rfc/draft-sharabayko-srt.html#name-shutdown
*/ */
class ShutDownPacket : public ControlPacket class ShutDownPacket : public ControlPacket {
{
public: public:
using Ptr = std::shared_ptr<ShutDownPacket>; using Ptr = std::shared_ptr<ShutDownPacket>;
ShutDownPacket() = default; ShutDownPacket() = default;
~ShutDownPacket() = default; ~ShutDownPacket() = default;
///////ControlPacket override/////// ///////ControlPacket override///////
bool loadFromData(uint8_t *buf, size_t len) override { bool loadFromData(uint8_t *buf, size_t len) override {
if (len < HEADER_SIZE) { if (len < HEADER_SIZE) {
@ -348,7 +345,7 @@ public:
_data->assign((char *)buf, len); _data->assign((char *)buf, len);
return loadHeader(); return loadHeader();
}; }
bool storeToData() override { bool storeToData() override {
control_type = ControlPacket::SHUTDOWN; control_type = ControlPacket::SHUTDOWN;
sub_type = 0; sub_type = 0;
@ -356,8 +353,9 @@ public:
_data->setCapacity(HEADER_SIZE); _data->setCapacity(HEADER_SIZE);
_data->setSize(HEADER_SIZE); _data->setSize(HEADER_SIZE);
return storeToHeader(); return storeToHeader();
}; }
}; };
} // namespace SRT } // namespace SRT
#endif //ZLMEDIAKIT_SRT_PACKET_H #endif //ZLMEDIAKIT_SRT_PACKET_H

147
srt/PacketQueue.cpp
View File

@ -4,78 +4,84 @@ namespace SRT {
#define MAX_SEQ 0x7fffffff #define MAX_SEQ 0x7fffffff
#define MAX_TS 0xffffffff #define MAX_TS 0xffffffff
inline uint32_t genExpectedSeq(uint32_t seq){
static inline uint32_t genExpectedSeq(uint32_t seq) {
return MAX_SEQ & seq; return MAX_SEQ & seq;
} }
inline bool isSeqEdge(uint32_t seq,uint32_t cap){
if(seq >(MAX_SEQ - cap)){ static inline bool isSeqEdge(uint32_t seq, uint32_t cap) {
if (seq > (MAX_SEQ - cap)) {
return true; return true;
} }
return false; return false;
} }
inline bool isTSCycle(uint32_t first,uint32_t second){ static inline bool isTSCycle(uint32_t first, uint32_t second) {
uint32_t diff; uint32_t diff;
if(first>second){ if (first > second) {
diff = first - second; diff = first - second;
}else{ } else {
diff = second - first; diff = second - first;
} }
if(diff > (MAX_TS>>1)){ if (diff > (MAX_TS >> 1)) {
return true; return true;
}else{ } else {
return false; return false;
} }
} }
PacketQueue::PacketQueue(uint32_t max_size, uint32_t init_seq, uint32_t lantency)
: _pkt_expected_seq(init_seq)
, _pkt_cap(max_size)
, _pkt_lantency(lantency) {
}
void PacketQueue::tryInsertPkt(DataPacket::Ptr pkt){
PacketQueue::PacketQueue(uint32_t max_size, uint32_t init_seq, uint32_t latency)
: _pkt_cap(max_size)
, _pkt_latency(latency)
, _pkt_expected_seq(init_seq) {}
void PacketQueue::tryInsertPkt(DataPacket::Ptr pkt) {
if (_pkt_expected_seq <= pkt->packet_seq_number) { if (_pkt_expected_seq <= pkt->packet_seq_number) {
auto diff = pkt->packet_seq_number - _pkt_expected_seq; auto diff = pkt->packet_seq_number - _pkt_expected_seq;
if(diff >= (MAX_SEQ>>1)){ if (diff >= (MAX_SEQ >> 1)) {
TraceL << "drop packet too later for cycle "<< "expected seq=" << _pkt_expected_seq << " pkt seq=" << pkt->packet_seq_number; TraceL << "drop packet too later for cycle "
<< "expected seq=" << _pkt_expected_seq << " pkt seq=" << pkt->packet_seq_number;
return; return;
}else{ } else {
_pkt_map.emplace(pkt->packet_seq_number, pkt); _pkt_map.emplace(pkt->packet_seq_number, pkt);
} }
} else { } else {
auto diff = _pkt_expected_seq - pkt->packet_seq_number; auto diff = _pkt_expected_seq - pkt->packet_seq_number;
if(diff >= (MAX_SEQ>>1)){ if (diff >= (MAX_SEQ >> 1)) {
_pkt_map.emplace(pkt->packet_seq_number, pkt); _pkt_map.emplace(pkt->packet_seq_number, pkt);
TraceL<<" cycle packet "<<"expected seq=" << _pkt_expected_seq << " pkt seq=" << pkt->packet_seq_number; TraceL << " cycle packet "
}else{ << "expected seq=" << _pkt_expected_seq << " pkt seq=" << pkt->packet_seq_number;
//TraceL << "drop packet too later "<< "expected seq=" << _pkt_expected_seq << " pkt seq=" << pkt->packet_seq_number; } else {
// TraceL << "drop packet too later "<< "expected seq=" << _pkt_expected_seq << " pkt seq=" <<
// pkt->packet_seq_number;
} }
} }
} }
bool PacketQueue::inputPacket(DataPacket::Ptr pkt,std::list<DataPacket::Ptr>& out) {
bool PacketQueue::inputPacket(DataPacket::Ptr pkt, std::list<DataPacket::Ptr> &out) {
tryInsertPkt(pkt); tryInsertPkt(pkt);
auto it = _pkt_map.find(_pkt_expected_seq); auto it = _pkt_map.find(_pkt_expected_seq);
while ( it != _pkt_map.end()) { while (it != _pkt_map.end()) {
out.push_back(it->second); out.push_back(it->second);
_pkt_map.erase(it); _pkt_map.erase(it);
_pkt_expected_seq = genExpectedSeq(_pkt_expected_seq+1); _pkt_expected_seq = genExpectedSeq(_pkt_expected_seq + 1);
it = _pkt_map.find(_pkt_expected_seq); it = _pkt_map.find(_pkt_expected_seq);
} }
while (_pkt_map.size() > _pkt_cap) { while (_pkt_map.size() > _pkt_cap) {
// 防止回环 // 防止回环
it = _pkt_map.find(_pkt_expected_seq); it = _pkt_map.find(_pkt_expected_seq);
if(it != _pkt_map.end()){ if (it != _pkt_map.end()) {
out.push_back(it->second); out.push_back(it->second);
_pkt_map.erase(it); _pkt_map.erase(it);
} }
_pkt_expected_seq = genExpectedSeq(_pkt_expected_seq + 1); _pkt_expected_seq = genExpectedSeq(_pkt_expected_seq + 1);
} }
while (timeLantency() > _pkt_lantency) { while (timeLatency() > _pkt_latency) {
it = _pkt_map.find(_pkt_expected_seq); it = _pkt_map.find(_pkt_expected_seq);
if(it != _pkt_map.end()){ if (it != _pkt_map.end()) {
out.push_back(it->second); out.push_back(it->second);
_pkt_map.erase(it); _pkt_map.erase(it);
} }
@ -85,22 +91,22 @@ bool PacketQueue::inputPacket(DataPacket::Ptr pkt,std::list<DataPacket::Ptr>& ou
return true; return true;
} }
bool PacketQueue::drop(uint32_t first, uint32_t last,std::list<DataPacket::Ptr>& out){ bool PacketQueue::drop(uint32_t first, uint32_t last, std::list<DataPacket::Ptr> &out) {
uint32_t end = genExpectedSeq(last+1); uint32_t end = genExpectedSeq(last + 1);
decltype(_pkt_map.end()) it; decltype(_pkt_map.end()) it;
for(uint32_t i =_pkt_expected_seq;i< end;){ for (uint32_t i = _pkt_expected_seq; i < end;) {
it = _pkt_map.find(i); it = _pkt_map.find(i);
if(it != _pkt_map.end()){ if (it != _pkt_map.end()) {
out.push_back(it->second); out.push_back(it->second);
_pkt_map.erase(it); _pkt_map.erase(it);
} }
i = genExpectedSeq(i+1); i = genExpectedSeq(i + 1);
} }
_pkt_expected_seq = end; _pkt_expected_seq = end;
return true; return true;
} }
uint32_t PacketQueue::timeLantency() { uint32_t PacketQueue::timeLatency() {
if (_pkt_map.empty()) { if (_pkt_map.empty()) {
return 0; return 0;
} }
@ -108,15 +114,15 @@ uint32_t PacketQueue::timeLantency() {
auto first = _pkt_map.begin()->second->timestamp; auto first = _pkt_map.begin()->second->timestamp;
auto last = _pkt_map.rbegin()->second->timestamp; auto last = _pkt_map.rbegin()->second->timestamp;
uint32_t dur; uint32_t dur;
if(last>first){ if (last > first) {
dur = last - first; dur = last - first;
}else{ } else {
dur = first - last; dur = first - last;
} }
if(dur > 0x80000000){ if (dur > 0x80000000) {
dur = MAX_TS - dur; dur = MAX_TS - dur;
WarnL<<"cycle dur "<<dur; WarnL << "cycle dur " << dur;
} }
return dur; return dur;
@ -124,16 +130,16 @@ uint32_t PacketQueue::timeLantency() {
std::list<PacketQueue::LostPair> PacketQueue::getLostSeq() { std::list<PacketQueue::LostPair> PacketQueue::getLostSeq() {
std::list<PacketQueue::LostPair> re; std::list<PacketQueue::LostPair> re;
if(_pkt_map.empty()){ if (_pkt_map.empty()) {
return re; return re;
} }
if(getExpectedSize() == getSize()){ if (getExpectedSize() == getSize()) {
return re; return re;
} }
uint32_t end = 0; uint32_t end = 0;
uint32_t first,last; uint32_t first, last;
first = _pkt_map.begin()->second->packet_seq_number; first = _pkt_map.begin()->second->packet_seq_number;
last = _pkt_map.rbegin()->second->packet_seq_number; last = _pkt_map.rbegin()->second->packet_seq_number;
@ -149,71 +155,76 @@ std::list<PacketQueue::LostPair> PacketQueue::getLostSeq() {
uint32_t i = _pkt_expected_seq; uint32_t i = _pkt_expected_seq;
bool finish = true; bool finish = true;
for(i = _pkt_expected_seq;i<=end;){ for (i = _pkt_expected_seq; i <= end;) {
if(_pkt_map.find(i) == _pkt_map.end()){ if (_pkt_map.find(i) == _pkt_map.end()) {
if(finish){ if (finish) {
finish = false; finish = false;
lost.first = i; lost.first = i;
lost.second = i+1; lost.second = i + 1;
}else{ } else {
lost.second = i+1; lost.second = i + 1;
} }
}else{ } else {
if(!finish){ if (!finish) {
finish = true; finish = true;
re.push_back(lost); re.push_back(lost);
} }
} }
i = genExpectedSeq(i+1); i = genExpectedSeq(i + 1);
} }
return re; return re;
} }
size_t PacketQueue::getSize(){ size_t PacketQueue::getSize() {
return _pkt_map.size(); return _pkt_map.size();
} }
size_t PacketQueue::getExpectedSize() { size_t PacketQueue::getExpectedSize() {
if(_pkt_map.empty()){ if (_pkt_map.empty()) {
return 0; return 0;
} }
uint32_t max = _pkt_map.rbegin()->first; uint32_t max = _pkt_map.rbegin()->first;
uint32_t min = _pkt_map.begin()->first; uint32_t min = _pkt_map.begin()->first;
if((max-min)>=(MAX_SEQ>>1)){ if ((max - min) >= (MAX_SEQ >> 1)) {
TraceL<<"cycle "<<"expected seq "<<_pkt_expected_seq<<" min "<<min<<" max "<<max<<" size "<<_pkt_map.size(); TraceL << "cycle "
return MAX_SEQ-_pkt_expected_seq+min+1; << "expected seq " << _pkt_expected_seq << " min " << min << " max " << max << " size "
}else{ << _pkt_map.size();
return max-_pkt_expected_seq+1; return MAX_SEQ - _pkt_expected_seq + min + 1;
} else {
return max - _pkt_expected_seq + 1;
} }
} }
size_t PacketQueue::getAvailableBufferSize(){ size_t PacketQueue::getAvailableBufferSize() {
auto size = getExpectedSize(); auto size = getExpectedSize();
if(_pkt_cap > size){ if (_pkt_cap > size) {
return _pkt_cap - size; return _pkt_cap - size;
} }
if(_pkt_cap > _pkt_map.size()){ if (_pkt_cap > _pkt_map.size()) {
return _pkt_cap - _pkt_map.size(); return _pkt_cap - _pkt_map.size();
} }
WarnL<<" cap "<<_pkt_cap<<" expected size "<<size<<" map size "<<_pkt_map.size(); WarnL << " cap " << _pkt_cap << " expected size " << size << " map size " << _pkt_map.size();
return _pkt_cap; return _pkt_cap;
} }
uint32_t PacketQueue::getExpectedSeq(){ uint32_t PacketQueue::getExpectedSeq() {
return _pkt_expected_seq; return _pkt_expected_seq;
} }
std::string PacketQueue::dump(){
std::string PacketQueue::dump() {
_StrPrinter printer; _StrPrinter printer;
if(_pkt_map.empty()){ if (_pkt_map.empty()) {
printer<<" expected seq :"<<_pkt_expected_seq; printer << " expected seq :" << _pkt_expected_seq;
}else{ } else {
printer<<" expected seq :"<<_pkt_expected_seq<<" size:"<<_pkt_map.size()<<" first:"<<_pkt_map.begin()->second->packet_seq_number; printer << " expected seq :" << _pkt_expected_seq << " size:" << _pkt_map.size()
printer<<" last:"<<_pkt_map.rbegin()->second->packet_seq_number; << " first:" << _pkt_map.begin()->second->packet_seq_number;
printer<<" latency:"<<timeLantency()/1e3; printer << " last:" << _pkt_map.rbegin()->second->packet_seq_number;
printer << " latency:" << timeLatency() / 1e3;
} }
return std::move(printer); return std::move(printer);
} }
} // namespace SRT } // namespace SRT

21
srt/PacketQueue.hpp
View File

@ -3,8 +3,8 @@
#include "Packet.hpp" #include "Packet.hpp"
#include <algorithm> #include <algorithm>
#include <list> #include <list>
#include <memory>
#include <map> #include <map>
#include <memory>
#include <tuple> #include <tuple>
#include <utility> #include <utility>
@ -16,11 +16,11 @@ public:
using Ptr = std::shared_ptr<PacketQueue>; using Ptr = std::shared_ptr<PacketQueue>;
using LostPair = std::pair<uint32_t, uint32_t>; using LostPair = std::pair<uint32_t, uint32_t>;
PacketQueue(uint32_t max_size, uint32_t init_seq, uint32_t lantency); PacketQueue(uint32_t max_size, uint32_t init_seq, uint32_t latency);
~PacketQueue() = default; ~PacketQueue() = default;
bool inputPacket(DataPacket::Ptr pkt,std::list<DataPacket::Ptr>& out); bool inputPacket(DataPacket::Ptr pkt, std::list<DataPacket::Ptr> &out);
uint32_t timeLantency(); uint32_t timeLatency();
std::list<LostPair> getLostSeq(); std::list<LostPair> getLostSeq();
size_t getSize(); size_t getSize();
@ -28,18 +28,17 @@ public:
size_t getAvailableBufferSize(); size_t getAvailableBufferSize();
uint32_t getExpectedSeq(); uint32_t getExpectedSeq();
bool drop(uint32_t first, uint32_t last,std::list<DataPacket::Ptr>& out);
std::string dump(); std::string dump();
bool drop(uint32_t first, uint32_t last, std::list<DataPacket::Ptr> &out);
private: private:
void tryInsertPkt(DataPacket::Ptr pkt); void tryInsertPkt(DataPacket::Ptr pkt);
private: private:
std::map<uint32_t, DataPacket::Ptr> _pkt_map;
uint32_t _pkt_expected_seq = 0;
uint32_t _pkt_cap; uint32_t _pkt_cap;
uint32_t _pkt_lantency; uint32_t _pkt_latency;
uint32_t _pkt_expected_seq = 0;
std::map<uint32_t, DataPacket::Ptr> _pkt_map;
}; };
} // namespace SRT } // namespace SRT

12
srt/PacketSendQueue.cpp
View File

@ -2,9 +2,10 @@
namespace SRT { namespace SRT {
PacketSendQueue::PacketSendQueue(uint32_t max_size, uint32_t lantency) PacketSendQueue::PacketSendQueue(uint32_t max_size, uint32_t latency)
: _pkt_cap(max_size) : _pkt_cap(max_size)
, _pkt_lantency(lantency) {} , _pkt_latency(latency) {}
bool PacketSendQueue::drop(uint32_t num) { bool PacketSendQueue::drop(uint32_t num) {
decltype(_pkt_cache.begin()) it; decltype(_pkt_cache.begin()) it;
for (it = _pkt_cache.begin(); it != _pkt_cache.end(); ++it) { for (it = _pkt_cache.begin(); it != _pkt_cache.end(); ++it) {
@ -17,12 +18,13 @@ bool PacketSendQueue::drop(uint32_t num) {
} }
return true; return true;
} }
bool PacketSendQueue::inputPacket(DataPacket::Ptr pkt) { bool PacketSendQueue::inputPacket(DataPacket::Ptr pkt) {
_pkt_cache.push_back(pkt); _pkt_cache.push_back(pkt);
while (_pkt_cache.size() > _pkt_cap) { while (_pkt_cache.size() > _pkt_cap) {
_pkt_cache.pop_front(); _pkt_cache.pop_front();
} }
while (timeLantency() > _pkt_lantency) { while (timeLatency() > _pkt_latency) {
_pkt_cache.pop_front(); _pkt_cache.pop_front();
} }
return true; return true;
@ -53,7 +55,7 @@ std::list<DataPacket::Ptr> PacketSendQueue::findPacketBySeq(uint32_t start, uint
return re; return re;
} }
uint32_t PacketSendQueue::timeLantency() { uint32_t PacketSendQueue::timeLatency() {
if (_pkt_cache.empty()) { if (_pkt_cache.empty()) {
return 0; return 0;
} }
@ -67,7 +69,7 @@ uint32_t PacketSendQueue::timeLantency() {
dur = first - last; dur = first - last;
} }
if (dur > (0x01 << 31)) { if (dur > (0x01 << 31)) {
TraceL << "cycle timeLantency " << dur; TraceL << "cycle timeLatency " << dur;
dur = 0xffffffff - dur; dur = 0xffffffff - dur;
} }

18
srt/PacketSendQueue.hpp
View File

@ -1,5 +1,6 @@
#ifndef ZLMEDIAKIT_SRT_PACKET_SEND_QUEUE_H #ifndef ZLMEDIAKIT_SRT_PACKET_SEND_QUEUE_H
#define ZLMEDIAKIT_SRT_PACKET_SEND_QUEUE_H #define ZLMEDIAKIT_SRT_PACKET_SEND_QUEUE_H
#include "Packet.hpp" #include "Packet.hpp"
#include <algorithm> #include <algorithm>
#include <list> #include <list>
@ -7,23 +8,30 @@
#include <set> #include <set>
#include <tuple> #include <tuple>
#include <utility> #include <utility>
namespace SRT { namespace SRT {
class PacketSendQueue { class PacketSendQueue {
public: public:
using Ptr = std::shared_ptr<PacketSendQueue>; using Ptr = std::shared_ptr<PacketSendQueue>;
using LostPair = std::pair<uint32_t, uint32_t>; using LostPair = std::pair<uint32_t, uint32_t>;
PacketSendQueue(uint32_t max_size, uint32_t lantency);
PacketSendQueue(uint32_t max_size, uint32_t latency);
~PacketSendQueue() = default; ~PacketSendQueue() = default;
bool drop(uint32_t num); bool drop(uint32_t num);
bool inputPacket(DataPacket::Ptr pkt); bool inputPacket(DataPacket::Ptr pkt);
std::list<DataPacket::Ptr> findPacketBySeq(uint32_t start,uint32_t end); std::list<DataPacket::Ptr> findPacketBySeq(uint32_t start, uint32_t end);
private: private:
uint32_t timeLantency(); uint32_t timeLatency();
private: private:
std::list<DataPacket::Ptr> _pkt_cache;
uint32_t _pkt_cap; uint32_t _pkt_cap;
uint32_t _pkt_lantency; uint32_t _pkt_latency;
std::list<DataPacket::Ptr> _pkt_cache;
}; };
} // namespace SRT } // namespace SRT
#endif // ZLMEDIAKIT_SRT_PACKET_SEND_QUEUE_H #endif // ZLMEDIAKIT_SRT_PACKET_SEND_QUEUE_H

66
srt/SrtSession.cpp
View File

@ -10,10 +10,10 @@ using namespace mediakit;
SrtSession::SrtSession(const Socket::Ptr &sock) SrtSession::SrtSession(const Socket::Ptr &sock)
: UdpSession(sock) { : UdpSession(sock) {
socklen_t addr_len = sizeof(_peer_addr); socklen_t addr_len = sizeof(_peer_addr);
memset(&_peer_addr,0,addr_len); memset(&_peer_addr, 0, addr_len);
//TraceL<<"before addr len "<<addr_len; // TraceL<<"before addr len "<<addr_len;
getpeername(sock->rawFD(), (struct sockaddr *)&_peer_addr, &addr_len); getpeername(sock->rawFD(), (struct sockaddr *)&_peer_addr, &addr_len);
//TraceL<<"after addr len "<<addr_len<<" family "<<_peer_addr.ss_family; // TraceL<<"after addr len "<<addr_len<<" family "<<_peer_addr.ss_family;
} }
SrtSession::~SrtSession() { SrtSession::~SrtSession() {
@ -21,40 +21,42 @@ SrtSession::~SrtSession() {
} }
EventPoller::Ptr SrtSession::queryPoller(const Buffer::Ptr &buffer) { EventPoller::Ptr SrtSession::queryPoller(const Buffer::Ptr &buffer) {
uint8_t* data = (uint8_t*)buffer->data(); uint8_t *data = (uint8_t *)buffer->data();
size_t size = buffer->size(); size_t size = buffer->size();
if(DataPacket::isDataPacket(data,size)){ if (DataPacket::isDataPacket(data, size)) {
uint32_t socket_id = DataPacket::getSocketID(data,size); uint32_t socket_id = DataPacket::getSocketID(data, size);
auto trans = SrtTransportManager::Instance().getItem(std::to_string(socket_id)); auto trans = SrtTransportManager::Instance().getItem(std::to_string(socket_id));
return trans ? trans->getPoller() : nullptr; return trans ? trans->getPoller() : nullptr;
} }
if(HandshakePacket::isHandshakePacket(data,size)){ if (HandshakePacket::isHandshakePacket(data, size)) {
auto type = HandshakePacket::getHandshakeType(data,size); auto type = HandshakePacket::getHandshakeType(data, size);
if(type == HandshakePacket::HS_TYPE_INDUCTION){ if (type == HandshakePacket::HS_TYPE_INDUCTION) {
// 握手第一阶段 // 握手第一阶段
return nullptr; return nullptr;
}else if(type == HandshakePacket::HS_TYPE_CONCLUSION){ } else if (type == HandshakePacket::HS_TYPE_CONCLUSION) {
// 握手第二阶段 // 握手第二阶段
uint32_t sync_cookie = HandshakePacket::getSynCookie(data,size); uint32_t sync_cookie = HandshakePacket::getSynCookie(data, size);
auto trans = SrtTransportManager::Instance().getHandshakeItem(std::to_string(sync_cookie)); auto trans = SrtTransportManager::Instance().getHandshakeItem(std::to_string(sync_cookie));
return trans ? trans->getPoller() : nullptr; return trans ? trans->getPoller() : nullptr;
}else{ } else {
WarnL<<" not reach there"; WarnL << " not reach there";
} }
}else{ } else {
uint32_t socket_id = ControlPacket::getSocketID(data,size); uint32_t socket_id = ControlPacket::getSocketID(data, size);
auto trans = SrtTransportManager::Instance().getItem(std::to_string(socket_id)); auto trans = SrtTransportManager::Instance().getItem(std::to_string(socket_id));
return trans ? trans->getPoller() : nullptr; return trans ? trans->getPoller() : nullptr;
} }
return nullptr; return nullptr;
} }
void SrtSession::attachServer(const toolkit::Server &server){
SockUtil::setRecvBuf(getSock()->rawFD(),1024 * 1024); void SrtSession::attachServer(const toolkit::Server &server) {
SockUtil::setRecvBuf(getSock()->rawFD(), 1024 * 1024);
} }
void SrtSession::onRecv(const Buffer::Ptr &buffer) { void SrtSession::onRecv(const Buffer::Ptr &buffer) {
uint8_t* data = (uint8_t*)buffer->data(); uint8_t *data = (uint8_t *)buffer->data();
size_t size = buffer->size(); size_t size = buffer->size();
if (_find_transport) { if (_find_transport) {
@ -64,10 +66,10 @@ void SrtSession::onRecv(const Buffer::Ptr &buffer) {
if (DataPacket::isDataPacket(data, size)) { if (DataPacket::isDataPacket(data, size)) {
uint32_t socket_id = DataPacket::getSocketID(data, size); uint32_t socket_id = DataPacket::getSocketID(data, size);
auto trans = SrtTransportManager::Instance().getItem(std::to_string(socket_id)); auto trans = SrtTransportManager::Instance().getItem(std::to_string(socket_id));
if(trans){ if (trans) {
_transport = std::move(trans); _transport = std::move(trans);
}else{ } else {
WarnL<<" data packet not find transport "; WarnL << " data packet not find transport ";
} }
} }
@ -92,24 +94,24 @@ void SrtSession::onRecv(const Buffer::Ptr &buffer) {
} else { } else {
uint32_t socket_id = ControlPacket::getSocketID(data, size); uint32_t socket_id = ControlPacket::getSocketID(data, size);
auto trans = SrtTransportManager::Instance().getItem(std::to_string(socket_id)); auto trans = SrtTransportManager::Instance().getItem(std::to_string(socket_id));
if(trans){ if (trans) {
_transport = std::move(trans); _transport = std::move(trans);
}else{ } else {
WarnL << " not find transport"; WarnL << " not find transport";
} }
} }
if(_transport){ if (_transport) {
_transport->setSession(shared_from_this()); _transport->setSession(shared_from_this());
} }
InfoP(this); InfoP(this);
} }
_ticker.resetTime(); _ticker.resetTime();
if(_transport){ if (_transport) {
_transport->inputSockData(data,size,&_peer_addr); _transport->inputSockData(data, size, &_peer_addr);
}else{ } else {
//WarnL<< "ingore data"; // WarnL<< "ingore data";
} }
} }
@ -125,15 +127,17 @@ void SrtSession::onError(const SockException &err) {
// 防止互相引用导致不释放 // 防止互相引用导致不释放
auto transport = std::move(_transport); auto transport = std::move(_transport);
getPoller()->async([transport,err] { getPoller()->async(
[transport, err] {
//延时减引用，防止使用transport对象时，销毁对象 //延时减引用，防止使用transport对象时，销毁对象
transport->onShutdown(err); transport->onShutdown(err);
}, false); },
false);
} }
void SrtSession::onManager() { void SrtSession::onManager() {
GET_CONFIG(float, timeoutSec, kTimeOutSec); GET_CONFIG(float, timeoutSec, kTimeOutSec);
if (_ticker.elapsedTime() > timeoutSec*1000) { if (_ticker.elapsedTime() > timeoutSec * 1000) {
shutdown(SockException(Err_timeout, "srt connection timeout")); shutdown(SockException(Err_timeout, "srt connection timeout"));
return; return;
} }

3
srt/SrtSession.hpp
View File

@ -24,8 +24,7 @@ private:
Ticker _ticker; Ticker _ticker;
struct sockaddr_storage _peer_addr; struct sockaddr_storage _peer_addr;
SrtTransport::Ptr _transport; SrtTransport::Ptr _transport;
}; };
} // namespace SRT } // namespace SRT
#endif //ZLMEDIAKIT_SRT_SESSION_H #endif // ZLMEDIAKIT_SRT_SESSION_H

351
srt/SrtTransport.cpp
View File

@ -1,32 +1,33 @@
#include <stdlib.h> #include <stdlib.h>
#include "Ack.hpp"
#include "Packet.hpp"
#include "SrtTransport.hpp"
#include "Util/onceToken.h" #include "Util/onceToken.h"
#include "SrtTransport.hpp"
#include "Packet.hpp"
#include "Ack.hpp"
namespace SRT { namespace SRT {
#define SRT_FIELD "srt." #define SRT_FIELD "srt."
//srt 超时时间 // srt 超时时间
const std::string kTimeOutSec = SRT_FIELD"timeoutSec"; const std::string kTimeOutSec = SRT_FIELD "timeoutSec";
//srt 单端口udp服务器 // srt 单端口udp服务器
const std::string kPort = SRT_FIELD"port"; const std::string kPort = SRT_FIELD "port";
const std::string kLatencyMul = SRT_FIELD "latencyMul";
const std::string kLantencyMul = SRT_FIELD"lantencyMul"; static std::atomic<uint32_t> s_srt_socket_id_generate { 125 };
static std::atomic<uint32_t> s_srt_socket_id_generate{125};
//////////// SrtTransport ////////////////////////// //////////// SrtTransport //////////////////////////
SrtTransport::SrtTransport(const EventPoller::Ptr &poller) SrtTransport::SrtTransport(const EventPoller::Ptr &poller)
: _poller(poller) { : _poller(poller) {
_start_timestamp = SteadyClock::now(); _start_timestamp = SteadyClock::now();
_socket_id = s_srt_socket_id_generate.fetch_add(1);\ _socket_id = s_srt_socket_id_generate.fetch_add(1);
_pkt_recv_rate_context = std::make_shared<PacketRecvRateContext>(_start_timestamp); _pkt_recv_rate_context = std::make_shared<PacketRecvRateContext>(_start_timestamp);
_recv_rate_context = std::make_shared<RecvRateContext>(_start_timestamp); _recv_rate_context = std::make_shared<RecvRateContext>(_start_timestamp);
_estimated_link_capacity_context = std::make_shared<EstimatedLinkCapacityContext>(_start_timestamp); _estimated_link_capacity_context = std::make_shared<EstimatedLinkCapacityContext>(_start_timestamp);
}
SrtTransport::~SrtTransport(){
TraceL<<" ";
} }
SrtTransport::~SrtTransport() {
TraceL << " ";
}
const EventPoller::Ptr &SrtTransport::getPoller() const { const EventPoller::Ptr &SrtTransport::getPoller() const {
return _poller; return _poller;
} }
@ -40,24 +41,25 @@ void SrtTransport::setSession(Session::Ptr session) {
} }
_selected_session = session; _selected_session = session;
} }
const Session::Ptr &SrtTransport::getSession() const { const Session::Ptr &SrtTransport::getSession() const {
return _selected_session; return _selected_session;
} }
void SrtTransport::switchToOtherTransport(uint8_t *buf, int len,uint32_t socketid, struct sockaddr_storage *addr){ void SrtTransport::switchToOtherTransport(uint8_t *buf, int len, uint32_t socketid, struct sockaddr_storage *addr) {
BufferRaw::Ptr tmp = BufferRaw::create(); BufferRaw::Ptr tmp = BufferRaw::create();
struct sockaddr_storage tmp_addr = *addr; struct sockaddr_storage tmp_addr = *addr;
tmp->assign((char*)buf,len); tmp->assign((char *)buf, len);
auto trans = SrtTransportManager::Instance().getItem(std::to_string(socketid)); auto trans = SrtTransportManager::Instance().getItem(std::to_string(socketid));
if(trans){ if (trans) {
trans->getPoller()->async([tmp,tmp_addr,trans]{ trans->getPoller()->async([tmp, tmp_addr, trans] {
trans->inputSockData((uint8_t*)tmp->data(),tmp->size(),(struct sockaddr_storage*)&tmp_addr); trans->inputSockData((uint8_t *)tmp->data(), tmp->size(), (struct sockaddr_storage *)&tmp_addr);
}); });
} }
} }
void SrtTransport::inputSockData(uint8_t *buf, int len, struct sockaddr_storage *addr) { void SrtTransport::inputSockData(uint8_t *buf, int len, struct sockaddr_storage *addr) {
using srt_control_handler = void (SrtTransport::*)(uint8_t* buf,int len,struct sockaddr_storage *addr); using srt_control_handler = void (SrtTransport::*)(uint8_t * buf, int len, struct sockaddr_storage *addr);
static std::unordered_map<uint16_t, srt_control_handler> s_control_functions; static std::unordered_map<uint16_t, srt_control_handler> s_control_functions;
static onceToken token([]() { static onceToken token([]() {
s_control_functions.emplace(ControlPacket::HANDSHAKE, &SrtTransport::handleHandshake); s_control_functions.emplace(ControlPacket::HANDSHAKE, &SrtTransport::handleHandshake);
@ -74,23 +76,23 @@ void SrtTransport::inputSockData(uint8_t *buf, int len, struct sockaddr_storage
_now = SteadyClock::now(); _now = SteadyClock::now();
// 处理srt数据 // 处理srt数据
if (DataPacket::isDataPacket(buf, len)) { if (DataPacket::isDataPacket(buf, len)) {
uint32_t socketId = DataPacket::getSocketID(buf,len); uint32_t socketId = DataPacket::getSocketID(buf, len);
if(socketId == _socket_id){ if (socketId == _socket_id) {
_pkt_recv_rate_context->inputPacket(_now); _pkt_recv_rate_context->inputPacket(_now);
_estimated_link_capacity_context->inputPacket(_now); _estimated_link_capacity_context->inputPacket(_now);
_recv_rate_context->inputPacket(_now, len); _recv_rate_context->inputPacket(_now, len);
handleDataPacket(buf, len, addr); handleDataPacket(buf, len, addr);
}else{ } else {
switchToOtherTransport(buf,len,socketId,addr); switchToOtherTransport(buf, len, socketId, addr);
} }
} else { } else {
if (ControlPacket::isControlPacket(buf, len)) { if (ControlPacket::isControlPacket(buf, len)) {
uint32_t socketId = ControlPacket::getSocketID(buf,len); uint32_t socketId = ControlPacket::getSocketID(buf, len);
uint16_t type = ControlPacket::getControlType(buf,len); uint16_t type = ControlPacket::getControlType(buf, len);
if(type != ControlPacket::HANDSHAKE && socketId != _socket_id && _socket_id != 0){ if (type != ControlPacket::HANDSHAKE && socketId != _socket_id && _socket_id != 0) {
// socket id not same // socket id not same
switchToOtherTransport(buf,len,socketId,addr); switchToOtherTransport(buf, len, socketId, addr);
return; return;
} }
_pkt_recv_rate_context->inputPacket(_now); _pkt_recv_rate_context->inputPacket(_now);
@ -99,10 +101,10 @@ void SrtTransport::inputSockData(uint8_t *buf, int len, struct sockaddr_storage
auto it = s_control_functions.find(type); auto it = s_control_functions.find(type);
if (it == s_control_functions.end()) { if (it == s_control_functions.end()) {
WarnL<<" not support type ignore" << ControlPacket::getControlType(buf,len); WarnL << " not support type ignore" << ControlPacket::getControlType(buf, len);
return; return;
}else{ } else {
(this->*(it->second))(buf,len,addr); (this->*(it->second))(buf, len, addr);
} }
} else { } else {
// not reach // not reach
@ -146,9 +148,10 @@ void SrtTransport::handleHandshakeInduction(HandshakePacket &pkt, struct sockadd
registerSelfHandshake(); registerSelfHandshake();
sendControlPacket(res, true); sendControlPacket(res, true);
} }
void SrtTransport::handleHandshakeConclusion(HandshakePacket &pkt, struct sockaddr_storage *addr) { void SrtTransport::handleHandshakeConclusion(HandshakePacket &pkt, struct sockaddr_storage *addr) {
if(!_handleshake_res){ if (!_handleshake_res) {
ErrorL<<"must Induction Phase for handleshake "; ErrorL << "must Induction Phase for handleshake ";
return; return;
} }
@ -157,21 +160,21 @@ void SrtTransport::handleHandshakeConclusion(HandshakePacket &pkt, struct sockad
HSExtMessage::Ptr req; HSExtMessage::Ptr req;
HSExtStreamID::Ptr sid; HSExtStreamID::Ptr sid;
uint32_t srt_flag = 0xbf; uint32_t srt_flag = 0xbf;
uint16_t delay = DurationCountMicroseconds(_now - _induction_ts)*getLantencyMul()/1000; uint16_t delay = DurationCountMicroseconds(_now - _induction_ts) * getLatencyMul() / 1000;
for (auto ext : pkt.ext_list) { for (auto ext : pkt.ext_list) {
//TraceL << getIdentifier() << " ext " << ext->dump(); // TraceL << getIdentifier() << " ext " << ext->dump();
if (!req) { if (!req) {
req = std::dynamic_pointer_cast<HSExtMessage>(ext); req = std::dynamic_pointer_cast<HSExtMessage>(ext);
} }
if(!sid){ if (!sid) {
sid = std::dynamic_pointer_cast<HSExtStreamID>(ext); sid = std::dynamic_pointer_cast<HSExtStreamID>(ext);
} }
} }
if(sid){ if (sid) {
_stream_id = sid->streamid; _stream_id = sid->streamid;
} }
if(req){ if (req) {
srt_flag = req->srt_flag; srt_flag = req->srt_flag;
delay = delay <= req->recv_tsbpd_delay ? req->recv_tsbpd_delay : delay; delay = delay <= req->recv_tsbpd_delay ? req->recv_tsbpd_delay : delay;
} }
@ -200,167 +203,174 @@ void SrtTransport::handleHandshakeConclusion(HandshakePacket &pkt, struct sockad
unregisterSelfHandshake(); unregisterSelfHandshake();
registerSelf(); registerSelf();
sendControlPacket(res, true); sendControlPacket(res, true);
TraceL<<" buf size = "<<res->max_flow_window_size<<" init seq ="<<_init_seq_number<<" lantency="<<delay; TraceL << " buf size = " << res->max_flow_window_size << " init seq =" << _init_seq_number << " latency=" << delay;
_recv_buf = std::make_shared<PacketQueue>(res->max_flow_window_size,_init_seq_number, delay*1e3); _recv_buf = std::make_shared<PacketQueue>(res->max_flow_window_size, _init_seq_number, delay * 1e3);
_send_buf = std::make_shared<PacketSendQueue>(res->max_flow_window_size, delay*1e3); _send_buf = std::make_shared<PacketSendQueue>(res->max_flow_window_size, delay * 1e3);
_send_packet_seq_number = _init_seq_number; _send_packet_seq_number = _init_seq_number;
_buf_delay = delay; _buf_delay = delay;
onHandShakeFinished(_stream_id,addr); onHandShakeFinished(_stream_id, addr);
} else { } else {
TraceL << getIdentifier() << " CONCLUSION handle repeate "; TraceL << getIdentifier() << " CONCLUSION handle repeate ";
sendControlPacket(_handleshake_res, true); sendControlPacket(_handleshake_res, true);
} }
_last_ack_pkt_seq_num = _init_seq_number; _last_ack_pkt_seq_num = _init_seq_number;
} }
void SrtTransport::handleHandshake(uint8_t *buf, int len, struct sockaddr_storage *addr){
HandshakePacket pkt;
assert(pkt.loadFromData(buf,len));
if(pkt.handshake_type == HandshakePacket::HS_TYPE_INDUCTION){ void SrtTransport::handleHandshake(uint8_t *buf, int len, struct sockaddr_storage *addr) {
handleHandshakeInduction(pkt,addr); HandshakePacket pkt;
}else if(pkt.handshake_type == HandshakePacket::HS_TYPE_CONCLUSION){ assert(pkt.loadFromData(buf, len));
handleHandshakeConclusion(pkt,addr);
}else{ if (pkt.handshake_type == HandshakePacket::HS_TYPE_INDUCTION) {
WarnL<<" not support handshake type = "<< pkt.handshake_type; handleHandshakeInduction(pkt, addr);
} else if (pkt.handshake_type == HandshakePacket::HS_TYPE_CONCLUSION) {
handleHandshakeConclusion(pkt, addr);
} else {
WarnL << " not support handshake type = " << pkt.handshake_type;
} }
_ack_ticker.resetTime(_now); _ack_ticker.resetTime(_now);
_nak_ticker.resetTime(_now); _nak_ticker.resetTime(_now);
} }
void SrtTransport::handleKeeplive(uint8_t *buf, int len, struct sockaddr_storage *addr){
//TraceL; void SrtTransport::handleKeeplive(uint8_t *buf, int len, struct sockaddr_storage *addr) {
// TraceL;
sendKeepLivePacket(); sendKeepLivePacket();
} }
void SrtTransport::sendKeepLivePacket(){ void SrtTransport::sendKeepLivePacket() {
KeepLivePacket::Ptr pkt = std::make_shared<KeepLivePacket>(); KeepLivePacket::Ptr pkt = std::make_shared<KeepLivePacket>();
pkt->dst_socket_id = _peer_socket_id; pkt->dst_socket_id = _peer_socket_id;
pkt->timestamp = DurationCountMicroseconds(_now -_start_timestamp); pkt->timestamp = DurationCountMicroseconds(_now - _start_timestamp);
pkt->storeToData(); pkt->storeToData();
sendControlPacket(pkt,true); sendControlPacket(pkt, true);
} }
void SrtTransport::handleACK(uint8_t *buf, int len, struct sockaddr_storage *addr){
//TraceL; void SrtTransport::handleACK(uint8_t *buf, int len, struct sockaddr_storage *addr) {
// TraceL;
ACKPacket ack; ACKPacket ack;
if(!ack.loadFromData(buf,len)){ if (!ack.loadFromData(buf, len)) {
return; return;
} }
ACKACKPacket::Ptr pkt = std::make_shared<ACKACKPacket>(); ACKACKPacket::Ptr pkt = std::make_shared<ACKACKPacket>();
pkt->dst_socket_id = _peer_socket_id; pkt->dst_socket_id = _peer_socket_id;
pkt->timestamp = DurationCountMicroseconds(_now -_start_timestamp); pkt->timestamp = DurationCountMicroseconds(_now - _start_timestamp);
pkt->ack_number = ack.ack_number; pkt->ack_number = ack.ack_number;
pkt->storeToData(); pkt->storeToData();
_send_buf->drop(ack.last_ack_pkt_seq_number); _send_buf->drop(ack.last_ack_pkt_seq_number);
sendControlPacket(pkt,true); sendControlPacket(pkt, true);
//TraceL<<"ack number "<<ack.ack_number; // TraceL<<"ack number "<<ack.ack_number;
} }
void SrtTransport::sendMsgDropReq(uint32_t first ,uint32_t last){
void SrtTransport::sendMsgDropReq(uint32_t first, uint32_t last) {
MsgDropReqPacket::Ptr pkt = std::make_shared<MsgDropReqPacket>(); MsgDropReqPacket::Ptr pkt = std::make_shared<MsgDropReqPacket>();
pkt->dst_socket_id = _peer_socket_id; pkt->dst_socket_id = _peer_socket_id;
pkt->timestamp = DurationCountMicroseconds(_now -_start_timestamp); pkt->timestamp = DurationCountMicroseconds(_now - _start_timestamp);
pkt->first_pkt_seq_num = first; pkt->first_pkt_seq_num = first;
pkt->last_pkt_seq_num = last; pkt->last_pkt_seq_num = last;
pkt->storeToData(); pkt->storeToData();
sendControlPacket(pkt,true); sendControlPacket(pkt, true);
} }
void SrtTransport::handleNAK(uint8_t *buf, int len, struct sockaddr_storage *addr){
//TraceL; void SrtTransport::handleNAK(uint8_t *buf, int len, struct sockaddr_storage *addr) {
// TraceL;
NAKPacket pkt; NAKPacket pkt;
pkt.loadFromData(buf,len); pkt.loadFromData(buf, len);
bool empty = false; bool empty = false;
bool flush = false; bool flush = false;
for(auto it : pkt.lost_list){ for (auto it : pkt.lost_list) {
if(pkt.lost_list.back() == it){ if (pkt.lost_list.back() == it) {
flush = true; flush = true;
} }
empty = true; empty = true;
auto re_list = _send_buf->findPacketBySeq(it.first,it.second-1); auto re_list = _send_buf->findPacketBySeq(it.first, it.second - 1);
for(auto pkt : re_list){ for (auto pkt : re_list) {
pkt->R = 1; pkt->R = 1;
pkt->storeToHeader(); pkt->storeToHeader();
sendPacket(pkt,flush); sendPacket(pkt, flush);
empty = false; empty = false;
} }
if(empty){ if (empty) {
sendMsgDropReq(it.first,it.second-1); sendMsgDropReq(it.first, it.second - 1);
} }
} }
} }
void SrtTransport::handleCongestionWarning(uint8_t *buf, int len, struct sockaddr_storage *addr){
void SrtTransport::handleCongestionWarning(uint8_t *buf, int len, struct sockaddr_storage *addr) {
TraceL; TraceL;
} }
void SrtTransport::handleShutDown(uint8_t *buf, int len, struct sockaddr_storage *addr){
void SrtTransport::handleShutDown(uint8_t *buf, int len, struct sockaddr_storage *addr) {
TraceL; TraceL;
onShutdown(SockException(Err_shutdown, "peer close connection")); onShutdown(SockException(Err_shutdown, "peer close connection"));
} }
void SrtTransport::handleDropReq(uint8_t *buf, int len, struct sockaddr_storage *addr){
void SrtTransport::handleDropReq(uint8_t *buf, int len, struct sockaddr_storage *addr) {
MsgDropReqPacket pkt; MsgDropReqPacket pkt;
pkt.loadFromData(buf,len); pkt.loadFromData(buf, len);
std::list<DataPacket::Ptr> list; std::list<DataPacket::Ptr> list;
//TraceL<<"drop "<<pkt.first_pkt_seq_num<<" last "<<pkt.last_pkt_seq_num; // TraceL<<"drop "<<pkt.first_pkt_seq_num<<" last "<<pkt.last_pkt_seq_num;
_recv_buf->drop(pkt.first_pkt_seq_num,pkt.last_pkt_seq_num,list); _recv_buf->drop(pkt.first_pkt_seq_num, pkt.last_pkt_seq_num, list);
if(list.empty()){ if (list.empty()) {
return; return;
} }
for(auto data : list){ for (auto data : list) {
onSRTData(std::move(data)); onSRTData(std::move(data));
} }
auto nak_interval = (_rtt+_rtt_variance*4)/2; auto nak_interval = (_rtt + _rtt_variance * 4) / 2;
if(nak_interval <= 20*1000){ if (nak_interval <= 20 * 1000) {
nak_interval = 20*1000; nak_interval = 20 * 1000;
} }
if(_nak_ticker.elapsedTime(_now)>nak_interval){ if (_nak_ticker.elapsedTime(_now) > nak_interval) {
auto lost = _recv_buf->getLostSeq(); auto lost = _recv_buf->getLostSeq();
if(!lost.empty()){ if (!lost.empty()) {
sendNAKPacket(lost); sendNAKPacket(lost);
} }
_nak_ticker.resetTime(_now); _nak_ticker.resetTime(_now);
} }
if(_ack_ticker.elapsedTime(_now)>10*1000){ if (_ack_ticker.elapsedTime(_now) > 10 * 1000) {
_light_ack_pkt_count = 0; _light_ack_pkt_count = 0;
_ack_ticker.resetTime(_now); _ack_ticker.resetTime(_now);
// send a ack per 10 ms for receiver // send a ack per 10 ms for receiver
sendACKPacket(); sendACKPacket();
}else{ } else {
if(_light_ack_pkt_count >= 64){ if (_light_ack_pkt_count >= 64) {
// for high bitrate stream send light ack // for high bitrate stream send light ack
// TODO // TODO
sendLightACKPacket(); sendLightACKPacket();
TraceL<<"send light ack"; TraceL << "send light ack";
} }
_light_ack_pkt_count = 0; _light_ack_pkt_count = 0;
} }
_light_ack_pkt_count++; _light_ack_pkt_count++;
} }
void SrtTransport::handleUserDefinedType(uint8_t *buf, int len, struct sockaddr_storage *addr){
void SrtTransport::handleUserDefinedType(uint8_t *buf, int len, struct sockaddr_storage *addr) {
TraceL; TraceL;
} }
void SrtTransport::handleACKACK(uint8_t *buf, int len, struct sockaddr_storage *addr){ void SrtTransport::handleACKACK(uint8_t *buf, int len, struct sockaddr_storage *addr) {
//TraceL; // TraceL;
ACKACKPacket::Ptr pkt = std::make_shared<ACKACKPacket>(); ACKACKPacket::Ptr pkt = std::make_shared<ACKACKPacket>();
pkt->loadFromData(buf,len); pkt->loadFromData(buf, len);
uint32_t rtt = DurationCountMicroseconds(_now - _ack_send_timestamp[pkt->ack_number]); uint32_t rtt = DurationCountMicroseconds(_now - _ack_send_timestamp[pkt->ack_number]);
_rtt_variance = (3*_rtt_variance+abs((long)_rtt - (long)rtt))/4; _rtt_variance = (3 * _rtt_variance + abs((long)_rtt - (long)rtt)) / 4;
_rtt = (7*rtt+_rtt)/8; _rtt = (7 * rtt + _rtt) / 8;
// TraceL<<" rtt:"<<_rtt<<" rtt variance:"<<_rtt_variance;
//TraceL<<" rtt:"<<_rtt<<" rtt variance:"<<_rtt_variance;
_ack_send_timestamp.erase(pkt->ack_number); _ack_send_timestamp.erase(pkt->ack_number);
} }
void SrtTransport::handlePeerError(uint8_t *buf, int len, struct sockaddr_storage *addr){ void SrtTransport::handlePeerError(uint8_t *buf, int len, struct sockaddr_storage *addr) {
TraceL; TraceL;
} }
void SrtTransport::sendACKPacket() { void SrtTransport::sendACKPacket() {
ACKPacket::Ptr pkt=std::make_shared<ACKPacket>(); ACKPacket::Ptr pkt = std::make_shared<ACKPacket>();
pkt->dst_socket_id = _peer_socket_id; pkt->dst_socket_id = _peer_socket_id;
pkt->timestamp = DurationCountMicroseconds(_now - _start_timestamp); pkt->timestamp = DurationCountMicroseconds(_now - _start_timestamp);
pkt->ack_number = ++_ack_number_count; pkt->ack_number = ++_ack_number_count;
@ -374,11 +384,12 @@ void SrtTransport::sendACKPacket() {
pkt->storeToData(); pkt->storeToData();
_ack_send_timestamp[pkt->ack_number] = _now; _ack_send_timestamp[pkt->ack_number] = _now;
_last_ack_pkt_seq_num = pkt->last_ack_pkt_seq_number; _last_ack_pkt_seq_num = pkt->last_ack_pkt_seq_number;
sendControlPacket(pkt,true); sendControlPacket(pkt, true);
//TraceL<<"send ack "<<pkt->dump(); // TraceL<<"send ack "<<pkt->dump();
} }
void SrtTransport::sendLightACKPacket() { void SrtTransport::sendLightACKPacket() {
ACKPacket::Ptr pkt=std::make_shared<ACKPacket>(); ACKPacket::Ptr pkt = std::make_shared<ACKPacket>();
pkt->dst_socket_id = _peer_socket_id; pkt->dst_socket_id = _peer_socket_id;
pkt->timestamp = DurationCountMicroseconds(_now - _start_timestamp); pkt->timestamp = DurationCountMicroseconds(_now - _start_timestamp);
@ -392,11 +403,11 @@ void SrtTransport::sendLightACKPacket() {
pkt->recv_rate = 0; pkt->recv_rate = 0;
pkt->storeToData(); pkt->storeToData();
_last_ack_pkt_seq_num = pkt->last_ack_pkt_seq_number; _last_ack_pkt_seq_num = pkt->last_ack_pkt_seq_number;
sendControlPacket(pkt,true); sendControlPacket(pkt, true);
TraceL<<"send ack "<<pkt->dump(); TraceL << "send ack " << pkt->dump();
} }
void SrtTransport::sendNAKPacket(std::list<PacketQueue::LostPair>& lost_list){ void SrtTransport::sendNAKPacket(std::list<PacketQueue::LostPair> &lost_list) {
NAKPacket::Ptr pkt = std::make_shared<NAKPacket>(); NAKPacket::Ptr pkt = std::make_shared<NAKPacket>();
pkt->dst_socket_id = _peer_socket_id; pkt->dst_socket_id = _peer_socket_id;
@ -405,112 +416,118 @@ void SrtTransport::sendNAKPacket(std::list<PacketQueue::LostPair>& lost_list){
pkt->storeToData(); pkt->storeToData();
//TraceL<<"send NAK "<<pkt->dump(); // TraceL<<"send NAK "<<pkt->dump();
sendControlPacket(pkt,true); sendControlPacket(pkt, true);
} }
void SrtTransport::sendShutDown(){ void SrtTransport::sendShutDown() {
ShutDownPacket::Ptr pkt = std::make_shared<ShutDownPacket>(); ShutDownPacket::Ptr pkt = std::make_shared<ShutDownPacket>();
pkt->dst_socket_id = _peer_socket_id; pkt->dst_socket_id = _peer_socket_id;
pkt->timestamp = DurationCountMicroseconds(_now - _start_timestamp); pkt->timestamp = DurationCountMicroseconds(_now - _start_timestamp);
pkt->storeToData(); pkt->storeToData();
sendControlPacket(pkt,true); sendControlPacket(pkt, true);
} }
void SrtTransport::handleDataPacket(uint8_t *buf, int len, struct sockaddr_storage *addr){
void SrtTransport::handleDataPacket(uint8_t *buf, int len, struct sockaddr_storage *addr) {
DataPacket::Ptr pkt = std::make_shared<DataPacket>(); DataPacket::Ptr pkt = std::make_shared<DataPacket>();
pkt->loadFromData(buf,len); pkt->loadFromData(buf, len);
pkt->get_ts = _now; pkt->get_ts = _now;
std::list<DataPacket::Ptr> list; std::list<DataPacket::Ptr> list;
//TraceL<<" seq="<< pkt->packet_seq_number<<" ts="<<pkt->timestamp<<" size="<<pkt->payloadSize()<<\ //TraceL<<" seq="<< pkt->packet_seq_number<<" ts="<<pkt->timestamp<<" size="<<pkt->payloadSize()<<\
//" PP="<<(int)pkt->PP<<" O="<<(int)pkt->O<<" kK="<<(int)pkt->KK<<" R="<<(int)pkt->R; //" PP="<<(int)pkt->PP<<" O="<<(int)pkt->O<<" kK="<<(int)pkt->KK<<" R="<<(int)pkt->R;
_recv_buf->inputPacket(pkt,list); _recv_buf->inputPacket(pkt, list);
for(auto data : list){ for (auto data : list) {
onSRTData(std::move(data)); onSRTData(std::move(data));
} }
auto nak_interval = (_rtt+_rtt_variance*4)/2; auto nak_interval = (_rtt + _rtt_variance * 4) / 2;
if(nak_interval <= 20*1000){ if (nak_interval <= 20 * 1000) {
nak_interval = 20*1000; nak_interval = 20 * 1000;
} }
if(list.empty()){ if (list.empty()) {
//TraceL<<_recv_buf->dump()<<" nake interval:"<<nak_interval/1000<<" ticker:"<<_nak_ticker.elapsedTime(_now)/1000; // TraceL<<_recv_buf->dump()<<" nake interval:"<<nak_interval/1000<<"
// ticker:"<<_nak_ticker.elapsedTime(_now)/1000;
} }
if(_nak_ticker.elapsedTime(_now)>nak_interval){ if (_nak_ticker.elapsedTime(_now) > nak_interval) {
auto lost = _recv_buf->getLostSeq(); auto lost = _recv_buf->getLostSeq();
if(!lost.empty()){ if (!lost.empty()) {
sendNAKPacket(lost); sendNAKPacket(lost);
//TraceL<<"send NAK"; // TraceL<<"send NAK";
}else{ } else {
//TraceL<<"lost is empty"; // TraceL<<"lost is empty";
} }
_nak_ticker.resetTime(_now); _nak_ticker.resetTime(_now);
} }
if(_ack_ticker.elapsedTime(_now)>10*1000){ if (_ack_ticker.elapsedTime(_now) > 10 * 1000) {
_light_ack_pkt_count = 0; _light_ack_pkt_count = 0;
_ack_ticker.resetTime(_now); _ack_ticker.resetTime(_now);
// send a ack per 10 ms for receiver // send a ack per 10 ms for receiver
sendACKPacket(); sendACKPacket();
}else{ } else {
if(_light_ack_pkt_count >= 64){ if (_light_ack_pkt_count >= 64) {
// for high bitrate stream send light ack // for high bitrate stream send light ack
// TODO // TODO
sendLightACKPacket(); sendLightACKPacket();
TraceL<<"send light ack"; TraceL << "send light ack";
} }
_light_ack_pkt_count = 0; _light_ack_pkt_count = 0;
} }
_light_ack_pkt_count++; _light_ack_pkt_count++;
//bufCheckInterval(); // bufCheckInterval();
} }
void SrtTransport::sendDataPacket(DataPacket::Ptr pkt,char* buf,int len, bool flush) { void SrtTransport::sendDataPacket(DataPacket::Ptr pkt, char *buf, int len, bool flush) {
pkt->storeToData((uint8_t*)buf,len); pkt->storeToData((uint8_t *)buf, len);
sendPacket(pkt,flush); sendPacket(pkt, flush);
_send_buf->inputPacket(pkt); _send_buf->inputPacket(pkt);
} }
void SrtTransport::sendControlPacket(ControlPacket::Ptr pkt, bool flush) { void SrtTransport::sendControlPacket(ControlPacket::Ptr pkt, bool flush) {
sendPacket(pkt,flush); sendPacket(pkt, flush);
} }
void SrtTransport::sendPacket(Buffer::Ptr pkt,bool flush){
if(_selected_session){ void SrtTransport::sendPacket(Buffer::Ptr pkt, bool flush) {
if (_selected_session) {
auto tmp = _packet_pool.obtain2(); auto tmp = _packet_pool.obtain2();
tmp->assign(pkt->data(),pkt->size()); tmp->assign(pkt->data(), pkt->size());
_selected_session->setSendFlushFlag(flush); _selected_session->setSendFlushFlag(flush);
_selected_session->send(std::move(tmp)); _selected_session->send(std::move(tmp));
}else{ } else {
WarnL<<"not reach this"; WarnL << "not reach this";
} }
} }
std::string SrtTransport::getIdentifier(){
std::string SrtTransport::getIdentifier() {
return _selected_session ? _selected_session->getIdentifier() : ""; return _selected_session ? _selected_session->getIdentifier() : "";
} }
void SrtTransport::registerSelfHandshake() { void SrtTransport::registerSelfHandshake() {
SrtTransportManager::Instance().addHandshakeItem(std::to_string(_sync_cookie),shared_from_this()); SrtTransportManager::Instance().addHandshakeItem(std::to_string(_sync_cookie), shared_from_this());
} }
void SrtTransport::unregisterSelfHandshake() { void SrtTransport::unregisterSelfHandshake() {
if(_sync_cookie == 0){ if (_sync_cookie == 0) {
return; return;
} }
SrtTransportManager::Instance().removeHandshakeItem(std::to_string(_sync_cookie)); SrtTransportManager::Instance().removeHandshakeItem(std::to_string(_sync_cookie));
} }
void SrtTransport::registerSelf() { void SrtTransport::registerSelf() {
if(_socket_id == 0){ if (_socket_id == 0) {
return; return;
} }
SrtTransportManager::Instance().addItem(std::to_string(_socket_id),shared_from_this()); SrtTransportManager::Instance().addItem(std::to_string(_socket_id), shared_from_this());
} }
void SrtTransport::unregisterSelf() { void SrtTransport::unregisterSelf() {
SrtTransportManager::Instance().removeItem(std::to_string(_socket_id)); SrtTransportManager::Instance().removeItem(std::to_string(_socket_id));
} }
void SrtTransport::onShutdown(const SockException &ex){ void SrtTransport::onShutdown(const SockException &ex) {
sendShutDown(); sendShutDown();
WarnL << ex.what(); WarnL << ex.what();
unregisterSelfHandshake(); unregisterSelfHandshake();
@ -522,23 +539,25 @@ void SrtTransport::onShutdown(const SockException &ex){
} }
} }
} }
size_t SrtTransport::getPayloadSize(){
size_t ret = (_mtu - 28 -16)/188*188; size_t SrtTransport::getPayloadSize() {
size_t ret = (_mtu - 28 - 16) / 188 * 188;
return ret; return ret;
} }
void SrtTransport::onSendTSData(const Buffer::Ptr &buffer, bool flush){
//TraceL; void SrtTransport::onSendTSData(const Buffer::Ptr &buffer, bool flush) {
// TraceL;
DataPacket::Ptr pkt; DataPacket::Ptr pkt;
size_t payloadSize = getPayloadSize(); size_t payloadSize = getPayloadSize();
size_t size = buffer->size(); size_t size = buffer->size();
char* ptr = buffer->data(); char *ptr = buffer->data();
char* end = buffer->data()+size; char *end = buffer->data() + size;
while(ptr < end && size >=payloadSize){ while (ptr < end && size >= payloadSize) {
pkt = std::make_shared<DataPacket>(); pkt = std::make_shared<DataPacket>();
pkt->f = 0; pkt->f = 0;
pkt->packet_seq_number = _send_packet_seq_number&0x7fffffff; pkt->packet_seq_number = _send_packet_seq_number & 0x7fffffff;
_send_packet_seq_number = (_send_packet_seq_number+1)&0x7fffffff; _send_packet_seq_number = (_send_packet_seq_number + 1) & 0x7fffffff;
pkt->PP = 3; pkt->PP = 3;
pkt->O = 0; pkt->O = 0;
pkt->KK = 0; pkt->KK = 0;
@ -546,16 +565,16 @@ void SrtTransport::onSendTSData(const Buffer::Ptr &buffer, bool flush){
pkt->msg_number = _send_msg_number++; pkt->msg_number = _send_msg_number++;
pkt->dst_socket_id = _peer_socket_id; pkt->dst_socket_id = _peer_socket_id;
pkt->timestamp = DurationCountMicroseconds(SteadyClock::now() - _start_timestamp); pkt->timestamp = DurationCountMicroseconds(SteadyClock::now() - _start_timestamp);
sendDataPacket(pkt,ptr,(int)payloadSize,flush); sendDataPacket(pkt, ptr, (int)payloadSize, flush);
ptr += payloadSize; ptr += payloadSize;
size -= payloadSize; size -= payloadSize;
} }
if(size >0 && ptr <end){ if (size > 0 && ptr < end) {
pkt = std::make_shared<DataPacket>(); pkt = std::make_shared<DataPacket>();
pkt->f = 0; pkt->f = 0;
pkt->packet_seq_number = _send_packet_seq_number&0x7fffffff; pkt->packet_seq_number = _send_packet_seq_number & 0x7fffffff;
_send_packet_seq_number = (_send_packet_seq_number+1)&0x7fffffff; _send_packet_seq_number = (_send_packet_seq_number + 1) & 0x7fffffff;
pkt->PP = 3; pkt->PP = 3;
pkt->O = 0; pkt->O = 0;
pkt->KK = 0; pkt->KK = 0;
@ -563,11 +582,12 @@ void SrtTransport::onSendTSData(const Buffer::Ptr &buffer, bool flush){
pkt->msg_number = _send_msg_number++; pkt->msg_number = _send_msg_number++;
pkt->dst_socket_id = _peer_socket_id; pkt->dst_socket_id = _peer_socket_id;
pkt->timestamp = DurationCountMicroseconds(SteadyClock::now() - _start_timestamp); pkt->timestamp = DurationCountMicroseconds(SteadyClock::now() - _start_timestamp);
sendDataPacket(pkt,ptr,(int)size,flush); sendDataPacket(pkt, ptr, (int)size, flush);
} }
} }
//////////// SrtTransportManager ////////////////////////// //////////// SrtTransportManager //////////////////////////
SrtTransportManager &SrtTransportManager::Instance() { SrtTransportManager &SrtTransportManager::Instance() {
static SrtTransportManager s_instance; static SrtTransportManager s_instance;
return s_instance; return s_instance;
@ -599,10 +619,12 @@ void SrtTransportManager::addHandshakeItem(const std::string &key, const SrtTran
std::lock_guard<std::mutex> lck(_handshake_mtx); std::lock_guard<std::mutex> lck(_handshake_mtx);
_handshake_map[key] = ptr; _handshake_map[key] = ptr;
} }
void SrtTransportManager::removeHandshakeItem(const std::string &key) { void SrtTransportManager::removeHandshakeItem(const std::string &key) {
std::lock_guard<std::mutex> lck(_handshake_mtx); std::lock_guard<std::mutex> lck(_handshake_mtx);
_handshake_map.erase(key); _handshake_map.erase(key);
} }
SrtTransport::Ptr SrtTransportManager::getHandshakeItem(const std::string &key) { SrtTransport::Ptr SrtTransportManager::getHandshakeItem(const std::string &key) {
if (key.empty()) { if (key.empty()) {
return nullptr; return nullptr;
@ -615,5 +637,4 @@ SrtTransport::Ptr SrtTransportManager::getHandshakeItem(const std::string &key)
return it->second.lock(); return it->second.lock();
} }
} // namespace SRT } // namespace SRT

54
srt/SrtTransport.hpp
View File

@ -1,10 +1,10 @@
#ifndef ZLMEDIAKIT_SRT_TRANSPORT_H #ifndef ZLMEDIAKIT_SRT_TRANSPORT_H
#define ZLMEDIAKIT_SRT_TRANSPORT_H #define ZLMEDIAKIT_SRT_TRANSPORT_H
#include <mutex> #include <atomic>
#include <chrono> #include <chrono>
#include <memory> #include <memory>
#include <atomic> #include <mutex>
#include "Network/Session.h" #include "Network/Session.h"
#include "Poller/EventPoller.h" #include "Poller/EventPoller.h"
@ -17,11 +17,12 @@
#include "Statistic.hpp" #include "Statistic.hpp"
namespace SRT { namespace SRT {
using namespace toolkit; using namespace toolkit;
extern const std::string kPort; extern const std::string kPort;
extern const std::string kTimeOutSec; extern const std::string kTimeOutSec;
extern const std::string kLantencyMul; extern const std::string kLatencyMul;
class SrtTransport : public std::enable_shared_from_this<SrtTransport> { class SrtTransport : public std::enable_shared_from_this<SrtTransport> {
public: public:
@ -33,6 +34,7 @@ public:
const EventPoller::Ptr &getPoller() const; const EventPoller::Ptr &getPoller() const;
void setSession(Session::Ptr session); void setSession(Session::Ptr session);
const Session::Ptr &getSession() const; const Session::Ptr &getSession() const;
/** /**
* socket收到udp数据 * socket收到udp数据
* @param buf 数据指针 * @param buf 数据指针
@ -43,26 +45,26 @@ public:
virtual void onSendTSData(const Buffer::Ptr &buffer, bool flush); virtual void onSendTSData(const Buffer::Ptr &buffer, bool flush);
std::string getIdentifier(); std::string getIdentifier();
void unregisterSelfHandshake();
void unregisterSelf(); void unregisterSelf();
void unregisterSelfHandshake();
protected: protected:
virtual void onHandShakeFinished(std::string& streamid,struct sockaddr_storage *addr){}; virtual bool isPusher() { return true; };
virtual void onSRTData(DataPacket::Ptr pkt){}; virtual void onSRTData(DataPacket::Ptr pkt) {};
virtual void onShutdown(const SockException &ex); virtual void onShutdown(const SockException &ex);
virtual bool isPusher(){ virtual void onHandShakeFinished(std::string &streamid, struct sockaddr_storage *addr) {};
return true; virtual void sendPacket(Buffer::Ptr pkt, bool flush = true);
}; virtual int getLatencyMul() { return 4; };
private: private:
void registerSelfHandshake();
void registerSelf(); void registerSelf();
void registerSelfHandshake();
void switchToOtherTransport(uint8_t *buf, int len,uint32_t socketid, struct sockaddr_storage *addr); void switchToOtherTransport(uint8_t *buf, int len, uint32_t socketid, struct sockaddr_storage *addr);
void handleHandshake(uint8_t *buf, int len, struct sockaddr_storage *addr); void handleHandshake(uint8_t *buf, int len, struct sockaddr_storage *addr);
void handleHandshakeInduction(HandshakePacket& pkt,struct sockaddr_storage *addr); void handleHandshakeInduction(HandshakePacket &pkt, struct sockaddr_storage *addr);
void handleHandshakeConclusion(HandshakePacket& pkt,struct sockaddr_storage *addr); void handleHandshakeConclusion(HandshakePacket &pkt, struct sockaddr_storage *addr);
void handleKeeplive(uint8_t *buf, int len, struct sockaddr_storage *addr); void handleKeeplive(uint8_t *buf, int len, struct sockaddr_storage *addr);
void handleACK(uint8_t *buf, int len, struct sockaddr_storage *addr); void handleACK(uint8_t *buf, int len, struct sockaddr_storage *addr);
@ -75,26 +77,24 @@ private:
void handlePeerError(uint8_t *buf, int len, struct sockaddr_storage *addr); void handlePeerError(uint8_t *buf, int len, struct sockaddr_storage *addr);
void handleDataPacket(uint8_t *buf, int len, struct sockaddr_storage *addr); void handleDataPacket(uint8_t *buf, int len, struct sockaddr_storage *addr);
void sendNAKPacket(std::list<PacketQueue::LostPair>& lost_list); void sendNAKPacket(std::list<PacketQueue::LostPair> &lost_list);
void sendACKPacket(); void sendACKPacket();
void sendLightACKPacket(); void sendLightACKPacket();
void sendKeepLivePacket(); void sendKeepLivePacket();
void sendShutDown(); void sendShutDown();
void sendMsgDropReq(uint32_t first ,uint32_t last); void sendMsgDropReq(uint32_t first, uint32_t last);
size_t getPayloadSize(); size_t getPayloadSize();
protected: protected:
void sendDataPacket(DataPacket::Ptr pkt,char* buf,int len,bool flush = false); void sendDataPacket(DataPacket::Ptr pkt, char *buf, int len, bool flush = false);
void sendControlPacket(ControlPacket::Ptr pkt,bool flush = true); void sendControlPacket(ControlPacket::Ptr pkt, bool flush = true);
virtual void sendPacket(Buffer::Ptr pkt,bool flush = true);
virtual int getLantencyMul(){
return 4;
};
private: private:
//当前选中的udp链接 //当前选中的udp链接
Session::Ptr _selected_session; Session::Ptr _selected_session;
//链接迁移前后使用过的udp链接 //链接迁移前后使用过的udp链接
std::unordered_map<Session *, std::weak_ptr<Session> > _history_sessions; std::unordered_map<Session *, std::weak_ptr<Session>> _history_sessions;
EventPoller::Ptr _poller; EventPoller::Ptr _poller;
@ -119,13 +119,13 @@ private:
PacketSendQueue::Ptr _send_buf; PacketSendQueue::Ptr _send_buf;
uint32_t _buf_delay = 120; uint32_t _buf_delay = 120;
PacketQueue::Ptr _recv_buf; PacketQueue::Ptr _recv_buf;
uint32_t _rtt = 100*1000; uint32_t _rtt = 100 * 1000;
uint32_t _rtt_variance =50*1000; uint32_t _rtt_variance = 50 * 1000;
uint32_t _light_ack_pkt_count = 0; uint32_t _light_ack_pkt_count = 0;
uint32_t _ack_number_count = 0; uint32_t _ack_number_count = 0;
uint32_t _last_ack_pkt_seq_num = 0; uint32_t _last_ack_pkt_seq_num = 0;
UTicker _ack_ticker; UTicker _ack_ticker;
std::map<uint32_t,TimePoint> _ack_send_timestamp; std::map<uint32_t, TimePoint> _ack_send_timestamp;
std::shared_ptr<PacketRecvRateContext> _pkt_recv_rate_context; std::shared_ptr<PacketRecvRateContext> _pkt_recv_rate_context;
std::shared_ptr<EstimatedLinkCapacityContext> _estimated_link_capacity_context; std::shared_ptr<EstimatedLinkCapacityContext> _estimated_link_capacity_context;
@ -137,7 +137,6 @@ private:
HandshakePacket::Ptr _handleshake_res; HandshakePacket::Ptr _handleshake_res;
ResourcePool<BufferRaw> _packet_pool; ResourcePool<BufferRaw> _packet_pool;
}; };
class SrtTransportManager { class SrtTransportManager {
@ -150,6 +149,7 @@ public:
void addHandshakeItem(const std::string &key, const SrtTransport::Ptr &ptr); void addHandshakeItem(const std::string &key, const SrtTransport::Ptr &ptr);
void removeHandshakeItem(const std::string &key); void removeHandshakeItem(const std::string &key);
SrtTransport::Ptr getHandshakeItem(const std::string &key); SrtTransport::Ptr getHandshakeItem(const std::string &key);
private: private:
SrtTransportManager() = default; SrtTransportManager() = default;

104
srt/SrtTransportImp.cpp
View File

@ -4,8 +4,7 @@
#include "SrtTransportImp.hpp" #include "SrtTransportImp.hpp"
namespace SRT { namespace SRT {
SrtTransportImp::SrtTransportImp(const EventPoller::Ptr &poller) SrtTransportImp::SrtTransportImp(const EventPoller::Ptr &poller) : SrtTransport(poller) {}
: SrtTransport(poller) {}
SrtTransportImp::~SrtTransportImp() { SrtTransportImp::~SrtTransportImp() {
InfoP(this); InfoP(this);
@ -23,52 +22,56 @@ SrtTransportImp::~SrtTransportImp() {
} }
} }
void SrtTransportImp::onHandShakeFinished(std::string &streamid,struct sockaddr_storage *addr) { void SrtTransportImp::onHandShakeFinished(std::string &streamid, struct sockaddr_storage *addr) {
// TODO parse stream id like this zlmediakit.com/live/test?token=1213444&type=push
// TODO parse streamid like this zlmediakit.com/live/test?token=1213444&type=push if (!_addr) {
if(!_addr){
_addr.reset(new sockaddr_storage(*((sockaddr_storage *)addr))); _addr.reset(new sockaddr_storage(*((sockaddr_storage *)addr)));
} }
_is_pusher = false; _is_pusher = false;
TraceL<<" stream id "<<streamid; TraceL << " stream id " << streamid;
if(streamid.empty()){ if (streamid.empty()) {
onShutdown(SockException(Err_shutdown, "streamid not empty")); onShutdown(SockException(Err_shutdown, "stream id not empty"));
return; return;
} }
_media_info.parse("srt://"+streamid); _media_info.parse("srt://" + streamid);
auto params = Parser::parseArgs(_media_info._param_strs); auto params = Parser::parseArgs(_media_info._param_strs);
if(params["type"] == "push"){ if (params["type"] == "push") {
_is_pusher = true; _is_pusher = true;
_decoder = DecoderImp::createDecoder(DecoderImp::decoder_ts, this); _decoder = DecoderImp::createDecoder(DecoderImp::decoder_ts, this);
emitOnPublish(); emitOnPublish();
}else{ } else {
_is_pusher = false; _is_pusher = false;
emitOnPlay(); emitOnPlay();
} }
} }
void SrtTransportImp::onSRTData(DataPacket::Ptr pkt) { void SrtTransportImp::onSRTData(DataPacket::Ptr pkt) {
if(!_is_pusher){ if (!_is_pusher) {
WarnP(this)<<"this is a player data ignore"; WarnP(this) << "this is a player data ignore";
return; return;
} }
if (_decoder) { if (_decoder) {
_decoder->input(reinterpret_cast<const uint8_t *>(pkt->payloadData()), pkt->payloadSize()); _decoder->input(reinterpret_cast<const uint8_t *>(pkt->payloadData()), pkt->payloadSize());
}else{ } else {
WarnP(this)<<" not reach this"; WarnP(this) << " not reach this";
} }
} }
void SrtTransportImp::onShutdown(const SockException &ex) { void SrtTransportImp::onShutdown(const SockException &ex) {
SrtTransport::onShutdown(ex); SrtTransport::onShutdown(ex);
} }
bool SrtTransportImp::close(mediakit::MediaSource &sender, bool force){ bool SrtTransportImp::close(mediakit::MediaSource &sender, bool force) {
if (!force && totalReaderCount(sender)) { if (!force && totalReaderCount(sender)) {
return false; return false;
} }
std::string err = StrPrinter << "close media:" << sender.getSchema() << "/" << sender.getVhost() << "/" std::string err = StrPrinter << "close media:" << sender.getSchema() << "/"
<< sender.getApp() << "/" << sender.getId() << " " << force; << sender.getVhost() << "/"
<< sender.getApp() << "/"
<< sender.getId() << " " << force;
weak_ptr<SrtTransportImp> weak_self = static_pointer_cast<SrtTransportImp>(shared_from_this()); weak_ptr<SrtTransportImp> weak_self = static_pointer_cast<SrtTransportImp>(shared_from_this());
getPoller()->async([weak_self, err]() { getPoller()->async([weak_self, err]() {
auto strong_self = weak_self.lock(); auto strong_self = weak_self.lock();
@ -80,20 +83,24 @@ bool SrtTransportImp::close(mediakit::MediaSource &sender, bool force){
}); });
return true; return true;
} }
// 播放总人数 // 播放总人数
int SrtTransportImp::totalReaderCount(mediakit::MediaSource &sender){ int SrtTransportImp::totalReaderCount(mediakit::MediaSource &sender) {
return _muxer ? _muxer->totalReaderCount() : sender.readerCount(); return _muxer ? _muxer->totalReaderCount() : sender.readerCount();
} }
// 获取媒体源类型 // 获取媒体源类型
mediakit::MediaOriginType SrtTransportImp::getOriginType(mediakit::MediaSource &sender) const{ mediakit::MediaOriginType SrtTransportImp::getOriginType(mediakit::MediaSource &sender) const {
return MediaOriginType::srt_push; return MediaOriginType::srt_push;
} }
// 获取媒体源url或者文件路径 // 获取媒体源url或者文件路径
std::string SrtTransportImp::getOriginUrl(mediakit::MediaSource &sender) const{ std::string SrtTransportImp::getOriginUrl(mediakit::MediaSource &sender) const {
return _media_info._full_url; return _media_info._full_url;
} }
// 获取媒体源客户端相关信息 // 获取媒体源客户端相关信息
std::shared_ptr<SockInfo> SrtTransportImp::getOriginSock(mediakit::MediaSource &sender) const{ std::shared_ptr<SockInfo> SrtTransportImp::getOriginSock(mediakit::MediaSource &sender) const {
return static_pointer_cast<SockInfo>(getSession()); return static_pointer_cast<SockInfo>(getSession());
} }
@ -114,7 +121,7 @@ void SrtTransportImp::emitOnPublish() {
InfoP(strong_self) << "允许 srt 推流"; InfoP(strong_self) << "允许 srt 推流";
} else { } else {
WarnP(strong_self) << "禁止 srt 推流:" << err; WarnP(strong_self) << "禁止 srt 推流:" << err;
strong_self->onShutdown(SockException(Err_refused,err)); strong_self->onShutdown(SockException(Err_refused, err));
} }
}; };
@ -126,47 +133,46 @@ void SrtTransportImp::emitOnPublish() {
} }
} }
void SrtTransportImp::emitOnPlay() {
void SrtTransportImp::emitOnPlay(){
std::weak_ptr<SrtTransportImp> weak_self = static_pointer_cast<SrtTransportImp>(shared_from_this()); std::weak_ptr<SrtTransportImp> weak_self = static_pointer_cast<SrtTransportImp>(shared_from_this());
Broadcast::AuthInvoker invoker = [weak_self](const string &err){ Broadcast::AuthInvoker invoker = [weak_self](const string &err) {
auto strong_self = weak_self.lock(); auto strong_self = weak_self.lock();
if (!strong_self) { if (!strong_self) {
return; return;
} }
strong_self->getPoller()->async([strong_self,err]{ strong_self->getPoller()->async([strong_self, err] {
if(err != ""){ if (err != "") {
strong_self->onShutdown(SockException(Err_refused,err)); strong_self->onShutdown(SockException(Err_refused, err));
}else{ } else {
strong_self->doPlay(); strong_self->doPlay();
} }
}); });
}; };
auto flag = NoticeCenter::Instance().emitEvent(Broadcast::kBroadcastMediaPlayed, _media_info, invoker, static_cast<SockInfo &>(*this)); auto flag = NoticeCenter::Instance().emitEvent(Broadcast::kBroadcastMediaPlayed, _media_info, invoker, static_cast<SockInfo &>(*this));
if(!flag){ if (!flag) {
doPlay(); doPlay();
} }
} }
void SrtTransportImp::doPlay(){
//异步查找直播流
std::weak_ptr<SrtTransportImp> weak_self = static_pointer_cast<SrtTransportImp>(shared_from_this());
void SrtTransportImp::doPlay() {
//异步查找直播流
MediaInfo info = _media_info; MediaInfo info = _media_info;
info._schema = TS_SCHEMA; info._schema = TS_SCHEMA;
std::weak_ptr<SrtTransportImp> weak_self = static_pointer_cast<SrtTransportImp>(shared_from_this());
MediaSource::findAsync(info, getSession(), [weak_self](const MediaSource::Ptr &src) { MediaSource::findAsync(info, getSession(), [weak_self](const MediaSource::Ptr &src) {
auto strong_self = weak_self.lock(); auto strong_self = weak_self.lock();
if (!strong_self) { if (!strong_self) {
//本对象已经销毁 //本对象已经销毁
TraceL<<"本对象已经销毁"; TraceL << "本对象已经销毁";
return; return;
} }
if (!src) { if (!src) {
//未找到该流 //未找到该流
TraceL<<"未找到该流"; TraceL << "未找到该流";
strong_self->onShutdown(SockException(Err_shutdown)); strong_self->onShutdown(SockException(Err_shutdown));
} else { } else {
TraceL<<"找到该流"; TraceL << "找到该流";
auto ts_src = dynamic_pointer_cast<TSMediaSource>(src); auto ts_src = dynamic_pointer_cast<TSMediaSource>(src);
assert(ts_src); assert(ts_src);
ts_src->pause(false); ts_src->pause(false);
@ -189,9 +195,10 @@ void SrtTransportImp::doPlay(){
auto size = ts_list->size(); auto size = ts_list->size();
ts_list->for_each([&](const TSPacket::Ptr &ts) { strong_self->onSendTSData(ts, ++i == size); }); ts_list->for_each([&](const TSPacket::Ptr &ts) { strong_self->onSendTSData(ts, ++i == size); });
}); });
}; }
}); });
} }
std::string SrtTransportImp::get_peer_ip() { std::string SrtTransportImp::get_peer_ip() {
if (!_addr) { if (!_addr) {
return "::"; return "::";
@ -236,9 +243,7 @@ bool SrtTransportImp::inputFrame(const Frame::Ptr &frame) {
} }
auto frame_cached = Frame::getCacheAbleFrame(frame); auto frame_cached = Frame::getCacheAbleFrame(frame);
lock_guard<recursive_mutex> lck(_func_mtx); lock_guard<recursive_mutex> lck(_func_mtx);
_cached_func.emplace_back([this, frame_cached]() { _cached_func.emplace_back([this, frame_cached]() { _muxer->inputFrame(frame_cached); });
_muxer->inputFrame(frame_cached);
});
return true; return true;
} }
@ -248,9 +253,7 @@ bool SrtTransportImp::addTrack(const Track::Ptr &track) {
} }
lock_guard<recursive_mutex> lck(_func_mtx); lock_guard<recursive_mutex> lck(_func_mtx);
_cached_func.emplace_back([this, track]() { _cached_func.emplace_back([this, track]() { _muxer->addTrack(track); });
_muxer->addTrack(track);
});
return true; return true;
} }
@ -259,9 +262,7 @@ void SrtTransportImp::addTrackCompleted() {
_muxer->addTrackCompleted(); _muxer->addTrackCompleted();
} else { } else {
lock_guard<recursive_mutex> lck(_func_mtx); lock_guard<recursive_mutex> lck(_func_mtx);
_cached_func.emplace_back([this]() { _cached_func.emplace_back([this]() { _muxer->addTrackCompleted(); });
_muxer->addTrackCompleted();
});
} }
} }
@ -273,10 +274,9 @@ void SrtTransportImp::doCachedFunc() {
_cached_func.clear(); _cached_func.clear();
} }
int SrtTransportImp::getLantencyMul(){ int SrtTransportImp::getLatencyMul() {
GET_CONFIG(int, lantencyMul, kLantencyMul); GET_CONFIG(int, latencyMul, kLatencyMul);
return lantencyMul; return latencyMul;
} }
} // namespace SRT } // namespace SRT

49
srt/SrtTransportImp.hpp
View File

@ -1,16 +1,18 @@
#ifndef ZLMEDIAKIT_SRT_TRANSPORT_IMP_H #ifndef ZLMEDIAKIT_SRT_TRANSPORT_IMP_H
#define ZLMEDIAKIT_SRT_TRANSPORT_IMP_H #define ZLMEDIAKIT_SRT_TRANSPORT_IMP_H
#include <mutex>
#include "Common/MultiMediaSourceMuxer.h"
#include "Rtp/Decoder.h"
#include "TS/TSMediaSource.h"
#include "SrtTransport.hpp"
#include <mutex>
#include "Rtp/Decoder.h"
#include "SrtTransport.hpp"
#include "TS/TSMediaSource.h"
#include "Common/MultiMediaSourceMuxer.h"
namespace SRT { namespace SRT {
using namespace toolkit;
using namespace mediakit; using namespace std;
using namespace std; using namespace toolkit;
using namespace mediakit;
class SrtTransportImp class SrtTransportImp
: public SrtTransport : public SrtTransport
, public toolkit::SockInfo , public toolkit::SockInfo
@ -19,13 +21,13 @@ class SrtTransportImp
public: public:
SrtTransportImp(const EventPoller::Ptr &poller); SrtTransportImp(const EventPoller::Ptr &poller);
~SrtTransportImp(); ~SrtTransportImp();
void inputSockData(uint8_t *buf, int len, struct sockaddr_storage *addr){
SrtTransport::inputSockData(buf,len,addr); void inputSockData(uint8_t *buf, int len, struct sockaddr_storage *addr) override {
SrtTransport::inputSockData(buf, len, addr);
_total_bytes += len; _total_bytes += len;
} }
void onSendTSData(const Buffer::Ptr &buffer, bool flush){ void onSendTSData(const Buffer::Ptr &buffer, bool flush) override { SrtTransport::onSendTSData(buffer, flush); }
SrtTransport::onSendTSData(buffer,flush);
}
/// SockInfo override /// SockInfo override
std::string get_local_ip() override; std::string get_local_ip() override;
uint16_t get_local_port() override; uint16_t get_local_port() override;
@ -35,20 +37,18 @@ public:
protected: protected:
///////SrtTransport override/////// ///////SrtTransport override///////
void onHandShakeFinished(std::string& streamid,struct sockaddr_storage *addr) override; int getLatencyMul() override;
void onSRTData(DataPacket::Ptr pkt) override; void onSRTData(DataPacket::Ptr pkt) override;
void onShutdown(const SockException &ex) override; void onShutdown(const SockException &ex) override;
int getLantencyMul() override; void onHandShakeFinished(std::string &streamid, struct sockaddr_storage *addr) override;
void sendPacket(Buffer::Ptr pkt,bool flush = true) override{ void sendPacket(Buffer::Ptr pkt, bool flush = true) override {
_total_bytes += pkt->size(); _total_bytes += pkt->size();
SrtTransport::sendPacket(pkt,flush); SrtTransport::sendPacket(pkt, flush);
};
bool isPusher() override{
return _is_pusher;
} }
bool isPusher() override { return _is_pusher; }
///////MediaSourceEvent override/////// ///////MediaSourceEvent override///////
// 关闭 // 关闭
bool close(mediakit::MediaSource &sender, bool force) override; bool close(mediakit::MediaSource &sender, bool force) override;
@ -61,10 +61,11 @@ protected:
// 获取媒体源客户端相关信息 // 获取媒体源客户端相关信息
std::shared_ptr<SockInfo> getOriginSock(mediakit::MediaSource &sender) const override; std::shared_ptr<SockInfo> getOriginSock(mediakit::MediaSource &sender) const override;
bool inputFrame(const Frame::Ptr &frame) override; ///////MediaSinkInterface override///////
bool addTrack(const Track::Ptr & track) override;
void addTrackCompleted() override;
void resetTracks() override {}; void resetTracks() override {};
void addTrackCompleted() override;
bool addTrack(const Track::Ptr &track) override;
bool inputFrame(const Frame::Ptr &frame) override;
private: private:
void emitOnPublish(); void emitOnPublish();

54
srt/Statistic.cpp
View File

@ -1,14 +1,17 @@
#include <algorithm> #include <algorithm>
#include "Statistic.hpp" #include "Statistic.hpp"
namespace SRT { namespace SRT {
void PacketRecvRateContext::inputPacket(TimePoint& ts) {
if(_pkt_map.size()>100){ void PacketRecvRateContext::inputPacket(TimePoint &ts) {
if (_pkt_map.size() > 100) {
_pkt_map.erase(_pkt_map.begin()); _pkt_map.erase(_pkt_map.begin());
} }
auto tmp = DurationCountMicroseconds(ts - _start); auto tmp = DurationCountMicroseconds(ts - _start);
_pkt_map.emplace(tmp,tmp); _pkt_map.emplace(tmp, tmp);
} }
uint32_t PacketRecvRateContext::getPacketRecvRate() { uint32_t PacketRecvRateContext::getPacketRecvRate() {
if (_pkt_map.size() < 2) { if (_pkt_map.size() < 2) {
return 50000; return 50000;
@ -17,79 +20,78 @@ uint32_t PacketRecvRateContext::getPacketRecvRate() {
for (auto it = _pkt_map.begin(); it != _pkt_map.end(); ++it) { for (auto it = _pkt_map.begin(); it != _pkt_map.end(); ++it) {
auto next = it; auto next = it;
++next; ++next;
if (next != _pkt_map.end()) { if (next == _pkt_map.end()) {
break;
}
if ((next->first - it->first) < dur) { if ((next->first - it->first) < dur) {
dur = next->first - it->first; dur = next->first - it->first;
} }
} else {
break;
}
} }
double rate = 1e6 / (double)dur; double rate = 1e6 / (double)dur;
if(rate <=1000){ if (rate <= 1000) {
return 50000; return 50000;
} }
return rate; return rate;
} }
void EstimatedLinkCapacityContext::inputPacket(TimePoint& ts) { void EstimatedLinkCapacityContext::inputPacket(TimePoint &ts) {
if (_pkt_map.size() > 16) { if (_pkt_map.size() > 16) {
_pkt_map.erase(_pkt_map.begin()); _pkt_map.erase(_pkt_map.begin());
} }
auto tmp = DurationCountMicroseconds(ts - _start); auto tmp = DurationCountMicroseconds(ts - _start);
_pkt_map.emplace(tmp, tmp); _pkt_map.emplace(tmp, tmp);
} }
uint32_t EstimatedLinkCapacityContext::getEstimatedLinkCapacity() { uint32_t EstimatedLinkCapacityContext::getEstimatedLinkCapacity() {
decltype(_pkt_map.begin()) next; decltype(_pkt_map.begin()) next;
std::vector<int64_t> tmp; std::vector<int64_t> tmp;
for(auto it = _pkt_map.begin();it != _pkt_map.end();++it){ for (auto it = _pkt_map.begin(); it != _pkt_map.end(); ++it) {
next = it; next = it;
++next; ++next;
if(next != _pkt_map.end()){ if (next != _pkt_map.end()) {
tmp.push_back(next->first -it->first); tmp.push_back(next->first - it->first);
}else{ } else {
break; break;
} }
} }
std::sort(tmp.begin(),tmp.end()); std::sort(tmp.begin(), tmp.end());
if(tmp.empty()){ if (tmp.empty()) {
return 1000; return 1000;
} }
if(tmp.size()<16){ if (tmp.size() < 16) {
return 1000; return 1000;
} }
double dur =tmp[0]/1e6; double dur = tmp[0] / 1e6;
return (uint32_t)(1.0 / dur);
return (uint32_t)(1.0/dur);
} }
void RecvRateContext::inputPacket(TimePoint& ts, size_t size ) { void RecvRateContext::inputPacket(TimePoint &ts, size_t size) {
if (_pkt_map.size() > 100) { if (_pkt_map.size() > 100) {
_pkt_map.erase(_pkt_map.begin()); _pkt_map.erase(_pkt_map.begin());
} }
auto tmp = DurationCountMicroseconds(ts - _start); auto tmp = DurationCountMicroseconds(ts - _start);
_pkt_map.emplace(tmp, tmp); _pkt_map.emplace(tmp, tmp);
} }
uint32_t RecvRateContext::getRecvRate() { uint32_t RecvRateContext::getRecvRate() {
if(_pkt_map.size()<2){ if (_pkt_map.size() < 2) {
return 0; return 0;
} }
auto first = _pkt_map.begin(); auto first = _pkt_map.begin();
auto last = _pkt_map.rbegin(); auto last = _pkt_map.rbegin();
double dur = (last->first - first->first)/1000000.0; double dur = (last->first - first->first) / 1000000.0;
size_t bytes = 0; size_t bytes = 0;
for(auto it : _pkt_map){ for (auto it : _pkt_map) {
bytes += it.second; bytes += it.second;
} }
double rate = (double)bytes/dur; double rate = (double)bytes / dur;
return (uint32_t)rate; return (uint32_t)rate;
} }

27
srt/Statistic.hpp
View File

@ -6,41 +6,42 @@
#include "Packet.hpp" #include "Packet.hpp"
namespace SRT { namespace SRT {
class PacketRecvRateContext { class PacketRecvRateContext {
public: public:
PacketRecvRateContext(TimePoint start):_start(start){}; PacketRecvRateContext(TimePoint start): _start(start) {};
~PacketRecvRateContext() = default; ~PacketRecvRateContext() = default;
void inputPacket(TimePoint& ts); void inputPacket(TimePoint &ts);
uint32_t getPacketRecvRate(); uint32_t getPacketRecvRate();
private:
std::map<int64_t,int64_t> _pkt_map;
TimePoint _start;
private:
TimePoint _start;
std::map<int64_t, int64_t> _pkt_map;
}; };
class EstimatedLinkCapacityContext { class EstimatedLinkCapacityContext {
public: public:
EstimatedLinkCapacityContext(TimePoint start):_start(start){}; EstimatedLinkCapacityContext(TimePoint start) : _start(start) {};
~EstimatedLinkCapacityContext() = default; ~EstimatedLinkCapacityContext() = default;
void inputPacket(TimePoint& ts); void inputPacket(TimePoint &ts);
uint32_t getEstimatedLinkCapacity(); uint32_t getEstimatedLinkCapacity();
private: private:
std::map<int64_t,int64_t> _pkt_map;
TimePoint _start; TimePoint _start;
std::map<int64_t, int64_t> _pkt_map;
}; };
class RecvRateContext { class RecvRateContext {
public: public:
RecvRateContext(TimePoint start):_start(start){}; RecvRateContext(TimePoint start): _start(start) {};
~RecvRateContext() = default; ~RecvRateContext() = default;
void inputPacket(TimePoint& ts,size_t size); void inputPacket(TimePoint &ts, size_t size);
uint32_t getRecvRate(); uint32_t getRecvRate();
private: private:
std::map<int64_t,size_t> _pkt_map;
TimePoint _start; TimePoint _start;
std::map<int64_t, size_t> _pkt_map;
}; };
} // namespace SRT } // namespace SRT
#endif // ZLMEDIAKIT_SRT_STATISTIC_H #endif // ZLMEDIAKIT_SRT_STATISTIC_H