Merge branch 'feature/srt' of github.com:ZLMediaKit/ZLMediaKit

This commit is contained in:
xiongziliang 2022-06-11 15:07:58 +08:00
commit fd11c53a78
30 changed files with 3663 additions and 2 deletions

View File

@ -106,6 +106,7 @@ option(ENABLE_FFMPEG "Enable FFmpeg" true)
option(ENABLE_MSVC_MT "Enable MSVC Mt/Mtd lib" true)
option(ENABLE_API_STATIC_LIB "Enable mk_api static lib" false)
option(USE_SOLUTION_FOLDERS "Enable solution dir supported" ON)
option(ENABLE_SRT "Enable SRT" true)
# ----------------------------------------------------------------------------
# Solution folders:
# ----------------------------------------------------------------------------
@ -486,6 +487,15 @@ if (ENABLE_WEBRTC)
endif ()
endif ()
if (ENABLE_SRT)
add_definitions(-DENABLE_SRT)
include_directories(./srt)
file(GLOB SRC_SRT_LIST ./srt/*.cpp ./srt/*.h ./srt/*.hpp)
add_library(srt ${SRC_SRT_LIST})
list(APPEND LINK_LIB_LIST srt)
message(STATUS "srt 功能已开启")
endif()
#c
if (ENABLE_API)
add_subdirectory(api)

View File

@ -107,7 +107,7 @@
- 支持rtp扩展解析
- 支持GOP缓冲webrtc播放秒开
- 支持datachannel
- [SRT支持](./srt/srt.md)
- 其他
- 支持丰富的restful api以及web hook事件
- 支持简单的telnet调试

View File

@ -61,7 +61,7 @@
- Support simulcast
- Support rtx/nack
- Support transport-cc rtcp/rtp ext
- [SRT support](./srt/srt_en.md)
- Others
- Support stream proxy by ffmpeg.
- RESTful http api and http hook event api.

View File

@ -284,6 +284,17 @@ preferredCodecA=PCMU,PCMA,opus,mpeg4-generic
#以下范例为所有支持的视频codec
preferredCodecV=H264,H265,AV1X,VP9,VP8
[srt]
#srt播放推流、播放超时时间,单位秒
timeoutSec=5
#srt udp服务器监听端口号所有srt客户端将通过该端口传输srt数据
#该端口是多线程的,同时支持客户端网络切换导致的连接迁移
port=9000
#srt 协议中延迟缓存的估算参数在握手阶段估算rtt ,然后latencyMul*rtt 为最大缓存时长,此参数越大,表示等待重传的时长就越大
latencyMul=4
[rtsp]
#rtsp专有鉴权方式是采用base64还是md5方式
authBasic=0

View File

@ -32,6 +32,11 @@
#include "../webrtc/WebRtcSession.h"
#endif
#if defined(ENABLE_SRT)
#include "../srt/SrtSession.hpp"
#include "../srt/SrtTransport.hpp"
#endif
#if defined(ENABLE_VERSION)
#include "version.h"
#endif
@ -284,6 +289,24 @@ int start_main(int argc,char *argv[]) {
uint16_t rtcPort = mINI::Instance()[RTC::kPort];
#endif//defined(ENABLE_WEBRTC)
#if defined(ENABLE_SRT)
auto srtSrv = std::make_shared<UdpServer>();
srtSrv->setOnCreateSocket([](const EventPoller::Ptr &poller, const Buffer::Ptr &buf, struct sockaddr *, int) {
if (!buf) {
return Socket::createSocket(poller, false);
}
auto new_poller = SRT::SrtSession::queryPoller(buf);
if (!new_poller) {
//握手第一阶段
return Socket::createSocket(poller, false);
}
return Socket::createSocket(new_poller, false);
});
uint16_t srtPort = mINI::Instance()[SRT::kPort];
#endif //defined(ENABLE_SRT)
try {
//rtsp服务器端口默认554
if (rtspPort) { rtspSrv->start<RtspSession>(rtspPort); }
@ -313,6 +336,14 @@ int start_main(int argc,char *argv[]) {
if (rtcPort) { rtcSrv->start<WebRtcSession>(rtcPort); }
#endif//defined(ENABLE_WEBRTC)
#if defined(ENABLE_SRT)
// srt udp服务器
if(srtPort){
srtSrv->start<SRT::SrtSession>(srtPort);
}
#endif//defined(ENABLE_SRT)
} catch (std::exception &ex) {
WarnL << "端口占用或无权限:" << ex.what() << endl;
ErrorL << "程序启动失败,请修改配置文件中端口号后重试!" << endl;

View File

@ -38,6 +38,7 @@ string getOriginTypeString(MediaOriginType type){
SWITCH_CASE(mp4_vod);
SWITCH_CASE(device_chn);
SWITCH_CASE(rtc_push);
SWITCH_CASE(srt_push);
default : return "unknown";
}
}

View File

@ -45,6 +45,7 @@ enum class MediaOriginType : uint8_t {
mp4_vod,
device_chn,
rtc_push,
srt_push
};
std::string getOriginTypeString(MediaOriginType type);

85
srt/Ack.cpp Normal file
View File

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

90
srt/Ack.hpp Normal file
View File

@ -0,0 +1,90 @@
#ifndef ZLMEDIAKIT_SRT_ACK_H
#define ZLMEDIAKIT_SRT_ACK_H
#include "Packet.hpp"
namespace SRT {
/*
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
+-+-+-+-+-+-+-+-+-+-+-+-+- SRT Header +-+-+-+-+-+-+-+-+-+-+-+-+-+
|1| Control Type | Reserved |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| Acknowledgement Number |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| Timestamp |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| Destination Socket ID |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+- CIF -+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| Last Acknowledged Packet Sequence Number |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| RTT |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| RTT Variance |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| Available Buffer Size |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| Packets Receiving Rate |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| Estimated Link Capacity |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| Receiving Rate |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
Figure 13: ACK control packet
https://haivision.github.io/srt-rfc/draft-sharabayko-srt.html#name-ack-acknowledgment
*/
class ACKPacket : public ControlPacket {
public:
using Ptr = std::shared_ptr<ACKPacket>;
ACKPacket() = default;
~ACKPacket() = default;
enum { ACK_CIF_SIZE = 7 * 4 };
std::string dump();
///////ControlPacket override///////
bool loadFromData(uint8_t *buf, size_t len) override;
bool storeToData() override;
uint32_t ack_number;
uint32_t last_ack_pkt_seq_number;
uint32_t rtt;
uint32_t rtt_variance;
uint32_t available_buf_size;
uint32_t pkt_recv_rate;
uint32_t estimated_link_capacity;
uint32_t recv_rate;
};
class ACKACKPacket : public ControlPacket {
public:
using Ptr = std::shared_ptr<ACKACKPacket>;
ACKACKPacket() = default;
~ACKACKPacket() = default;
///////ControlPacket override///////
bool loadFromData(uint8_t *buf, size_t len) override {
if (len < ControlPacket::HEADER_SIZE) {
return false;
}
_data = BufferRaw::create();
_data->assign((char *)(buf), len);
ControlPacket::loadHeader();
ack_number = loadUint32(type_specific_info);
return true;
}
bool storeToData() override {
_data = BufferRaw::create();
_data->setCapacity(HEADER_SIZE);
_data->setSize(HEADER_SIZE);
control_type = ControlPacket::ACKACK;
sub_type = 0;
storeUint32(type_specific_info, ack_number);
storeToHeader();
return true;
}
uint32_t ack_number;
};
} // namespace SRT
#endif // ZLMEDIAKIT_SRT_ACK_H

96
srt/Common.hpp Normal file
View File

@ -0,0 +1,96 @@
#ifndef ZLMEDIAKIT_SRT_COMMON_H
#define ZLMEDIAKIT_SRT_COMMON_H
#if defined(_WIN32)
#include <winsock2.h>
#include <ws2tcpip.h>
#include <Iphlpapi.h>
#pragma comment(lib, "Ws2_32.lib")
#pragma comment(lib, "Iphlpapi.lib")
#else
#include <netdb.h>
#include <sys/socket.h>
#endif // defined(_WIN32)
#include <chrono>
#define MAX_SEQ 0x7fffffff
#define MAX_TS 0xffffffff
namespace SRT {
using SteadyClock = std::chrono::steady_clock;
using TimePoint = std::chrono::time_point<SteadyClock>;
using Microseconds = std::chrono::microseconds;
using Milliseconds = std::chrono::milliseconds;
static inline int64_t DurationCountMicroseconds(SteadyClock::duration dur) {
return std::chrono::duration_cast<std::chrono::microseconds>(dur).count();
}
static inline uint32_t loadUint32(uint8_t *ptr) {
return ptr[0] << 24 | ptr[1] << 16 | ptr[2] << 8 | ptr[3];
}
static inline uint16_t loadUint16(uint8_t *ptr) {
return ptr[0] << 8 | ptr[1];
}
static inline void storeUint32(uint8_t *buf, uint32_t val) {
buf[0] = val >> 24;
buf[1] = (val >> 16) & 0xff;
buf[2] = (val >> 8) & 0xff;
buf[3] = val & 0xff;
}
static inline void storeUint16(uint8_t *buf, uint16_t val) {
buf[0] = (val >> 8) & 0xff;
buf[1] = val & 0xff;
}
static inline void storeUint32LE(uint8_t *buf, uint32_t val) {
buf[0] = val & 0xff;
buf[1] = (val >> 8) & 0xff;
buf[2] = (val >> 16) & 0xff;
buf[3] = (val >> 24) & 0xff;
}
static inline void storeUint16LE(uint8_t *buf, uint16_t val) {
buf[0] = val & 0xff;
buf[1] = (val >> 8) & 0xff;
}
static inline uint32_t srtVersion(int major, int minor, int patch) {
return patch + minor * 0x100 + major * 0x10000;
}
static inline uint32_t genExpectedSeq(uint32_t seq) {
return MAX_SEQ & seq;
}
class UTicker {
public:
UTicker() { _created = _begin = SteadyClock::now(); }
~UTicker() = default;
/**
*
*/
int64_t elapsedTime(TimePoint now) const { return DurationCountMicroseconds(now - _begin); }
/**
* resetTime后至今的时间
*/
int64_t createdTime(TimePoint now) const { return DurationCountMicroseconds(now - _created); }
/**
*
*/
void resetTime(TimePoint now) { _begin = now; }
private:
TimePoint _begin;
TimePoint _created;
};
} // namespace SRT
#endif // ZLMEDIAKIT_SRT_COMMON_H

134
srt/HSExt.cpp Normal file
View File

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

129
srt/HSExt.hpp Normal file
View File

@ -0,0 +1,129 @@
#ifndef ZLMEDIAKIT_SRT_HS_EXT_H
#define ZLMEDIAKIT_SRT_HS_EXT_H
#include "Network/Buffer.h"
#include "Common.hpp"
namespace SRT {
using namespace toolkit;
class HSExt : public Buffer {
public:
HSExt() = default;
virtual ~HSExt() = default;
enum {
SRT_CMD_REJECT = 0,
SRT_CMD_HSREQ = 1,
SRT_CMD_HSRSP = 2,
SRT_CMD_KMREQ = 3,
SRT_CMD_KMRSP = 4,
SRT_CMD_SID = 5,
SRT_CMD_CONGESTION = 6,
SRT_CMD_FILTER = 7,
SRT_CMD_GROUP = 8,
SRT_CMD_NONE = -1
};
using Ptr = std::shared_ptr<HSExt>;
uint16_t extension_type;
uint16_t extension_length;
virtual bool loadFromData(uint8_t *buf, size_t len) = 0;
virtual bool storeToData() = 0;
virtual std::string dump() = 0;
///////Buffer override///////
char *data() const override {
if (_data) {
return _data->data();
}
return nullptr;
}
size_t size() const override {
if (_data) {
return _data->size();
}
return 0;
}
protected:
void loadHeader() {
uint8_t *ptr = (uint8_t *)_data->data();
extension_type = loadUint16(ptr);
ptr += 2;
extension_length = loadUint16(ptr);
ptr += 2;
}
void storeHeader() {
uint8_t *ptr = (uint8_t *)_data->data();
SRT::storeUint16(ptr, extension_type);
ptr += 2;
storeUint16(ptr, extension_length);
}
protected:
BufferRaw::Ptr _data;
};
/*
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
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| SRT Version |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| SRT Flags |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| Receiver TSBPD Delay | Sender TSBPD Delay |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
Figure 6: Handshake Extension Message structure
https://haivision.github.io/srt-rfc/draft-sharabayko-srt.html#name-handshake-extension-message
*/
class HSExtMessage : public HSExt {
public:
using Ptr = std::shared_ptr<HSExtMessage>;
enum {
HS_EXT_MSG_TSBPDSND = 0x00000001,
HS_EXT_MSG_TSBPDRCV = 0x00000002,
HS_EXT_MSG_CRYPT = 0x00000004,
HS_EXT_MSG_TLPKTDROP = 0x00000008,
HS_EXT_MSG_PERIODICNAK = 0x00000010,
HS_EXT_MSG_REXMITFLG = 0x00000020,
HS_EXT_MSG_STREAM = 0x00000040,
HS_EXT_MSG_PACKET_FILTER = 0x00000080
};
enum { HSEXT_MSG_SIZE = 16 };
HSExtMessage() = default;
~HSExtMessage() = default;
bool loadFromData(uint8_t *buf, size_t len) override;
bool storeToData() override;
std::string dump() override;
uint32_t srt_version;
uint32_t srt_flag;
uint16_t recv_tsbpd_delay;
uint16_t send_tsbpd_delay;
};
/*
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
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| |
| Stream ID |
...
| |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
Figure 7: Stream ID Extension Message
https://haivision.github.io/srt-rfc/draft-sharabayko-srt.html#name-stream-id-extension-message
*/
class HSExtStreamID : public HSExt {
public:
using Ptr = std::shared_ptr<HSExtStreamID>;
HSExtStreamID() = default;
~HSExtStreamID() = default;
bool loadFromData(uint8_t *buf, size_t len) override;
bool storeToData() override;
std::string dump() override;
std::string streamid;
};
} // namespace SRT
#endif // ZLMEDIAKIT_SRT_HS_EXT_H

108
srt/NackContext.cpp Normal file
View File

@ -0,0 +1,108 @@
#include "NackContext.hpp"
namespace SRT {
void NackContext::update(TimePoint now, std::list<PacketQueue::LostPair> &lostlist) {
for (auto item : lostlist) {
mergeItem(now, item);
}
}
void NackContext::getLostList(
TimePoint now, uint32_t rtt, uint32_t rtt_variance, std::list<PacketQueue::LostPair> &lostlist) {
lostlist.clear();
std::list<uint32_t> tmp_list;
for (auto it = _nack_map.begin(); it != _nack_map.end(); ++it) {
if (!it->second._is_nack) {
tmp_list.push_back(it->first);
it->second._ts = now;
it->second._is_nack = true;
} else {
if (DurationCountMicroseconds(now - it->second._ts) > rtt) {
tmp_list.push_back(it->first);
it->second._ts = now;
}
}
}
tmp_list.sort();
if (tmp_list.empty()) {
return;
}
uint32_t min = *tmp_list.begin();
uint32_t max = *tmp_list.rbegin();
if ((max - min) >= (MAX_SEQ >> 1)) {
while ((max - tmp_list.front()) > (MAX_SEQ >> 1)) {
tmp_list.push_back(tmp_list.front());
tmp_list.pop_front();
}
}
PacketQueue::LostPair lost;
bool finish = true;
for (auto cur = tmp_list.begin(); cur != tmp_list.end(); ++cur) {
if (finish) {
lost.first = *cur;
lost.second = genExpectedSeq(*cur + 1);
finish = false;
} else {
if (lost.second == *cur) {
lost.second = genExpectedSeq(*cur + 1);
} else {
finish = true;
lostlist.push_back(lost);
}
}
}
}
void NackContext::drop(uint32_t seq) {
if (_nack_map.empty())
return;
uint32_t min = _nack_map.begin()->first;
uint32_t max = _nack_map.rbegin()->first;
bool is_cycle = false;
if ((max - min) >= (MAX_SEQ >> 1)) {
is_cycle = true;
}
for (auto it = _nack_map.begin(); it != _nack_map.end();) {
if (!is_cycle) {
// 不回环
if (it->first <= seq) {
it = _nack_map.erase(it);
} else {
it++;
}
} else {
if (it->first <= seq) {
if ((seq - it->first) >= (MAX_SEQ >> 1)) {
WarnL << "cycle seq " << seq << " " << it->first;
it++;
} else {
it = _nack_map.erase(it);
}
} else {
if ((it->first - seq) >= (MAX_SEQ >> 1)) {
it = _nack_map.erase(it);
WarnL << "cycle seq " << seq << " " << it->first;
} else {
it++;
}
}
}
}
}
void NackContext::mergeItem(TimePoint now, PacketQueue::LostPair &item) {
for (uint32_t i = item.first; i < item.second; ++i) {
auto it = _nack_map.find(i);
if (it != _nack_map.end()) {
} else {
NackItem tmp;
tmp._is_nack = false;
_nack_map.emplace(i, tmp);
}
}
}
} // namespace SRT

30
srt/NackContext.hpp Normal file
View File

@ -0,0 +1,30 @@
#ifndef ZLMEDIAKIT_SRT_NACK_CONTEXT_H
#define ZLMEDIAKIT_SRT_NACK_CONTEXT_H
#include "Common.hpp"
#include "PacketQueue.hpp"
#include <list>
namespace SRT {
class NackContext {
public:
NackContext() = default;
~NackContext() = default;
void update(TimePoint now, std::list<PacketQueue::LostPair> &lostlist);
void getLostList(TimePoint now, uint32_t rtt, uint32_t rtt_variance, std::list<PacketQueue::LostPair> &lostlist);
void drop(uint32_t seq);
private:
void mergeItem(TimePoint now, PacketQueue::LostPair &item);
private:
class NackItem {
public:
bool _is_nack = false;
TimePoint _ts; // send nak time
};
std::map<uint32_t, NackItem> _nack_map;
};
} // namespace SRT
#endif // ZLMEDIAKIT_SRT_NACK_CONTEXT_H

603
srt/Packet.cpp Normal file
View File

@ -0,0 +1,603 @@
#include "Util/MD5.h"
#include "Util/logger.h"
#include <atomic>
#include "Packet.hpp"
namespace SRT {
const size_t DataPacket::HEADER_SIZE;
const size_t ControlPacket::HEADER_SIZE;
const size_t HandshakePacket::HS_CONTENT_MIN_SIZE;
bool DataPacket::isDataPacket(uint8_t *buf, size_t len) {
if (len < HEADER_SIZE) {
WarnL << "data size" << len << " less " << HEADER_SIZE;
return false;
}
if (!(buf[0] & 0x80)) {
return true;
}
return false;
}
uint32_t DataPacket::getSocketID(uint8_t *buf, size_t len) {
uint8_t *ptr = buf;
ptr += 12;
return loadUint32(ptr);
}
bool DataPacket::loadFromData(uint8_t *buf, size_t len) {
if (len < HEADER_SIZE) {
WarnL << "data size" << len << " less " << HEADER_SIZE;
return false;
}
uint8_t *ptr = buf;
f = ptr[0] >> 7;
packet_seq_number = loadUint32(ptr) & 0x7fffffff;
ptr += 4;
PP = ptr[0] >> 6;
O = (ptr[0] & 0x20) >> 5;
KK = (ptr[0] & 0x18) >> 3;
R = (ptr[0] & 0x04) >> 2;
msg_number = (ptr[0] & 0x03) << 24 | ptr[1] << 12 | ptr[2] << 8 | ptr[3];
ptr += 4;
timestamp = loadUint32(ptr);
ptr += 4;
dst_socket_id = loadUint32(ptr);
ptr += 4;
_data = BufferRaw::create();
_data->assign((char *)(buf), len);
return true;
}
bool DataPacket::storeToHeader() {
if (!_data || _data->size() < HEADER_SIZE) {
WarnL << "data size less " << HEADER_SIZE;
return false;
}
uint8_t *ptr = (uint8_t *)_data->data();
ptr[0] = packet_seq_number >> 24;
ptr[1] = (packet_seq_number >> 16) & 0xff;
ptr[2] = (packet_seq_number >> 8) & 0xff;
ptr[3] = packet_seq_number & 0xff;
ptr += 4;
ptr[0] = PP << 6;
ptr[0] |= O << 5;
ptr[0] |= KK << 3;
ptr[0] |= R << 2;
ptr[0] |= (msg_number & 0xff000000) >> 24;
ptr[1] = (msg_number & 0xff0000) >> 16;
ptr[2] = (msg_number & 0xff00) >> 8;
ptr[3] = msg_number & 0xff;
ptr += 4;
storeUint32(ptr, timestamp);
ptr += 4;
storeUint32(ptr, dst_socket_id);
ptr += 4;
return true;
}
bool DataPacket::storeToData(uint8_t *buf, size_t len) {
_data = BufferRaw::create();
_data->setCapacity(len + HEADER_SIZE);
_data->setSize(len + HEADER_SIZE);
uint8_t *ptr = (uint8_t *)_data->data();
ptr[0] = packet_seq_number >> 24;
ptr[1] = (packet_seq_number >> 16) & 0xff;
ptr[2] = (packet_seq_number >> 8) & 0xff;
ptr[3] = packet_seq_number & 0xff;
ptr += 4;
ptr[0] = PP << 6;
ptr[0] |= O << 5;
ptr[0] |= KK << 3;
ptr[0] |= R << 2;
ptr[0] |= (msg_number & 0xff000000) >> 24;
ptr[1] = (msg_number & 0xff0000) >> 16;
ptr[2] = (msg_number & 0xff00) >> 8;
ptr[3] = msg_number & 0xff;
ptr += 4;
storeUint32(ptr, timestamp);
ptr += 4;
storeUint32(ptr, dst_socket_id);
ptr += 4;
memcpy(ptr, buf, len);
return true;
}
char *DataPacket::data() const {
if (!_data)
return nullptr;
return _data->data();
}
size_t DataPacket::size() const {
if (!_data) {
return 0;
}
return _data->size();
}
char *DataPacket::payloadData() {
if (!_data)
return nullptr;
return _data->data() + HEADER_SIZE;
}
size_t DataPacket::payloadSize() {
if (!_data) {
return 0;
}
return _data->size() - HEADER_SIZE;
}
bool ControlPacket::isControlPacket(uint8_t *buf, size_t len) {
if (len < HEADER_SIZE) {
WarnL << "data size" << len << " less " << HEADER_SIZE;
return false;
}
if (buf[0] & 0x80) {
return true;
}
return false;
}
uint16_t ControlPacket::getControlType(uint8_t *buf, size_t len) {
uint8_t *ptr = buf;
uint16_t control_type = (ptr[0] & 0x7f) << 8 | ptr[1];
return control_type;
}
bool ControlPacket::loadHeader() {
uint8_t *ptr = (uint8_t *)_data->data();
f = ptr[0] >> 7;
control_type = (ptr[0] & 0x7f) << 8 | ptr[1];
ptr += 2;
sub_type = loadUint16(ptr);
ptr += 2;
type_specific_info[0] = ptr[0];
type_specific_info[1] = ptr[1];
type_specific_info[2] = ptr[2];
type_specific_info[3] = ptr[3];
ptr += 4;
timestamp = loadUint32(ptr);
ptr += 4;
dst_socket_id = loadUint32(ptr);
ptr += 4;
return true;
}
bool ControlPacket::storeToHeader() {
uint8_t *ptr = (uint8_t *)_data->data();
ptr[0] = 0x80;
ptr[0] |= control_type >> 8;
ptr[1] = control_type & 0xff;
ptr += 2;
storeUint16(ptr, sub_type);
ptr += 2;
ptr[0] = type_specific_info[0];
ptr[1] = type_specific_info[1];
ptr[2] = type_specific_info[2];
ptr[3] = type_specific_info[3];
ptr += 4;
storeUint32(ptr, timestamp);
ptr += 4;
storeUint32(ptr, dst_socket_id);
ptr += 4;
return true;
}
char *ControlPacket::data() const {
if (!_data)
return nullptr;
return _data->data();
}
size_t ControlPacket::size() const {
if (!_data) {
return 0;
}
return _data->size();
}
uint32_t ControlPacket::getSocketID(uint8_t *buf, size_t len) {
return loadUint32(buf + 12);
}
bool HandshakePacket::loadFromData(uint8_t *buf, size_t len) {
if (HEADER_SIZE + HS_CONTENT_MIN_SIZE > len) {
ErrorL << "size too smalle " << encryption_field;
return false;
}
_data = BufferRaw::create();
_data->assign((char *)(buf), len);
ControlPacket::loadHeader();
uint8_t *ptr = (uint8_t *)_data->data() + HEADER_SIZE;
// parse CIF
version = loadUint32(ptr);
ptr += 4;
encryption_field = loadUint16(ptr);
ptr += 2;
extension_field = loadUint16(ptr);
ptr += 2;
initial_packet_sequence_number = loadUint32(ptr);
ptr += 4;
mtu = loadUint32(ptr);
ptr += 4;
max_flow_window_size = loadUint32(ptr);
ptr += 4;
handshake_type = loadUint32(ptr);
ptr += 4;
srt_socket_id = loadUint32(ptr);
ptr += 4;
syn_cookie = loadUint32(ptr);
ptr += 4;
memcpy(peer_ip_addr, ptr, sizeof(peer_ip_addr) * sizeof(peer_ip_addr[0]));
ptr += sizeof(peer_ip_addr) * sizeof(peer_ip_addr[0]);
if (encryption_field != NO_ENCRYPTION) {
ErrorL << "not support encryption " << encryption_field;
}
if (extension_field == 0) {
return true;
}
if (len == HEADER_SIZE + HS_CONTENT_MIN_SIZE) {
// ErrorL << "extension filed not exist " << extension_field;
return true;
}
return loadExtMessage(ptr, len - HS_CONTENT_MIN_SIZE - HEADER_SIZE);
}
bool HandshakePacket::loadExtMessage(uint8_t *buf, size_t len) {
uint8_t *ptr = buf;
ext_list.clear();
uint16_t type;
uint16_t length;
HSExt::Ptr ext;
while (ptr < buf + len) {
type = loadUint16(ptr);
length = loadUint16(ptr + 2);
switch (type) {
case HSExt::SRT_CMD_HSREQ:
case HSExt::SRT_CMD_HSRSP:
ext = std::make_shared<HSExtMessage>();
break;
case HSExt::SRT_CMD_SID:
ext = std::make_shared<HSExtStreamID>();
break;
default:
WarnL << "not support ext " << type;
break;
}
if (ext) {
if (ext->loadFromData(ptr, length * 4 + 4)) {
ext_list.push_back(std::move(ext));
} else {
WarnL << "parse HS EXT failed type=" << type << " len=" << length;
}
ext = nullptr;
}
ptr += length * 4 + 4;
}
return true;
}
bool HandshakePacket::storeExtMessage() {
uint8_t *buf = (uint8_t *)_data->data() + HEADER_SIZE + 48;
size_t len = _data->size() - HEADER_SIZE - 48;
for (auto ex : ext_list) {
memcpy(buf, ex->data(), ex->size());
buf += ex->size();
}
return true;
}
size_t HandshakePacket::getExtSize() {
size_t size = 0;
for (auto it : ext_list) {
size += it->size();
}
return size;
}
bool HandshakePacket::storeToData() {
_data = BufferRaw::create();
for (auto ex : ext_list) {
ex->storeToData();
}
auto ext_size = getExtSize();
_data->setCapacity(HEADER_SIZE + 48 + ext_size);
_data->setSize(HEADER_SIZE + 48 + ext_size);
control_type = ControlPacket::HANDSHAKE;
sub_type = 0;
ControlPacket::storeToHeader();
uint8_t *ptr = (uint8_t *)_data->data() + HEADER_SIZE;
storeUint32(ptr, version);
ptr += 4;
storeUint16(ptr, encryption_field);
ptr += 2;
storeUint16(ptr, extension_field);
ptr += 2;
storeUint32(ptr, initial_packet_sequence_number);
ptr += 4;
storeUint32(ptr, mtu);
ptr += 4;
storeUint32(ptr, max_flow_window_size);
ptr += 4;
storeUint32(ptr, handshake_type);
ptr += 4;
storeUint32(ptr, srt_socket_id);
ptr += 4;
storeUint32(ptr, syn_cookie);
ptr += 4;
memcpy(ptr, peer_ip_addr, sizeof(peer_ip_addr) * sizeof(peer_ip_addr[0]));
ptr += sizeof(peer_ip_addr) * sizeof(peer_ip_addr[0]);
if (encryption_field != NO_ENCRYPTION) {
ErrorL << "not support encryption " << encryption_field;
}
assert(encryption_field == NO_ENCRYPTION);
return storeExtMessage();
}
bool HandshakePacket::isHandshakePacket(uint8_t *buf, size_t len) {
if (!ControlPacket::isControlPacket(buf, len)) {
return false;
}
if (len < HEADER_SIZE + 48) {
return false;
}
return ControlPacket::getControlType(buf, len) == HANDSHAKE;
}
uint32_t HandshakePacket::getHandshakeType(uint8_t *buf, size_t len) {
uint8_t *ptr = buf + HEADER_SIZE + 5 * 4;
return loadUint32(ptr);
}
uint32_t HandshakePacket::getSynCookie(uint8_t *buf, size_t len) {
uint8_t *ptr = buf + HEADER_SIZE + 7 * 4;
return loadUint32(ptr);
}
void HandshakePacket::assignPeerIP(struct sockaddr_storage *addr) {
memset(peer_ip_addr, 0, sizeof(peer_ip_addr) * sizeof(peer_ip_addr[0]));
if (addr->ss_family == AF_INET) {
struct sockaddr_in *ipv4 = (struct sockaddr_in *)addr;
// 抓包 奇怪好像是小头端???
storeUint32LE(peer_ip_addr, ipv4->sin_addr.s_addr);
} else if (addr->ss_family == AF_INET6) {
if (IN6_IS_ADDR_V4MAPPED(&((struct sockaddr_in6 *)addr)->sin6_addr)) {
struct in_addr addr4;
memcpy(&addr4, 12 + (char *)&(((struct sockaddr_in6 *)addr)->sin6_addr), 4);
storeUint32LE(peer_ip_addr, addr4.s_addr);
} else {
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]));
}
}
}
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;
while (true) {
// SYN cookie
char clienthost[NI_MAXHOST];
char clientport[NI_MAXSERV];
getnameinfo(
(struct sockaddr *)addr, sizeof(struct sockaddr_storage), clienthost, sizeof(clienthost), clientport,
sizeof(clientport), NI_NUMERICHOST | NI_NUMERICSERV);
int64_t timestamp = (DurationCountMicroseconds(SteadyClock::now() - ts) / 60000000) + distractor.load()
+ correction; // secret changes every one minute
std::stringstream cookiestr;
cookiestr << clienthost << ":" << clientport << ":" << timestamp;
union {
unsigned char cookie[16];
uint32_t cookie_val;
};
MD5 md5(cookiestr.str());
memcpy(cookie, md5.rawdigest().c_str(), 16);
if (cookie_val != current_cookie) {
return cookie_val;
}
++distractor;
// This is just to make the loop formally breakable,
// but this is virtually impossible to happen.
if (distractor == rollover) {
return cookie_val;
}
}
}
bool KeepLivePacket::loadFromData(uint8_t *buf, size_t len) {
if (len < HEADER_SIZE) {
WarnL << "data size" << len << " less " << HEADER_SIZE;
return false;
}
_data = BufferRaw::create();
_data->assign((char *)buf, len);
return loadHeader();
}
bool KeepLivePacket::storeToData() {
control_type = ControlPacket::KEEPALIVE;
sub_type = 0;
_data = BufferRaw::create();
_data->setCapacity(HEADER_SIZE);
_data->setSize(HEADER_SIZE);
return storeToHeader();
}
bool NAKPacket::loadFromData(uint8_t *buf, size_t len) {
if (len < HEADER_SIZE) {
WarnL << "data size" << len << " less " << HEADER_SIZE;
return false;
}
_data = BufferRaw::create();
_data->assign((char *)buf, len);
loadHeader();
uint8_t *ptr = (uint8_t *)_data->data() + HEADER_SIZE;
uint8_t *end = (uint8_t *)_data->data() + _data->size();
LostPair lost;
while (ptr < end) {
if ((*ptr) & 0x80) {
lost.first = loadUint32(ptr) & 0x7fffffff;
lost.second = loadUint32(ptr + 4) & 0x7fffffff;
lost.second += 1;
ptr += 8;
} else {
lost.first = loadUint32(ptr);
lost.second = lost.first + 1;
ptr += 4;
}
lost_list.push_back(lost);
}
return true;
}
bool NAKPacket::storeToData() {
control_type = NAK;
sub_type = 0;
size_t cif_size = getCIFSize();
_data = BufferRaw::create();
_data->setCapacity(HEADER_SIZE + cif_size);
_data->setSize(HEADER_SIZE + cif_size);
storeToHeader();
uint8_t *ptr = (uint8_t *)_data->data() + HEADER_SIZE;
for (auto it : lost_list) {
if (it.first + 1 == it.second) {
storeUint32(ptr, it.first);
ptr[0] = ptr[0] & 0x7f;
ptr += 4;
} else {
storeUint32(ptr, it.first);
ptr[0] |= 0x80;
storeUint32(ptr + 4, it.second - 1);
// ptr[4] = ptr[4]&0x7f;
ptr += 8;
}
}
return true;
}
size_t NAKPacket::getCIFSize() {
size_t size = 0;
for (auto it : lost_list) {
if (it.first + 1 == it.second) {
size += 4;
} else {
size += 8;
}
}
return size;
}
std::string NAKPacket::dump() {
_StrPrinter printer;
for (auto it : lost_list) {
printer << "[ " << it.first << " , " << it.second - 1 << " ]";
}
return std::move(printer);
}
bool MsgDropReqPacket::loadFromData(uint8_t *buf, size_t len) {
if (len < HEADER_SIZE + 8) {
WarnL << "data size" << len << " less " << HEADER_SIZE;
return false;
}
_data = BufferRaw::create();
_data->assign((char *)buf, len);
loadHeader();
uint8_t *ptr = (uint8_t *)_data->data() + HEADER_SIZE;
first_pkt_seq_num = loadUint32(ptr);
ptr += 4;
last_pkt_seq_num = loadUint32(ptr);
ptr += 4;
return true;
}
bool MsgDropReqPacket::storeToData() {
control_type = DROPREQ;
sub_type = 0;
_data = BufferRaw::create();
_data->setCapacity(HEADER_SIZE + 8);
_data->setSize(HEADER_SIZE + 8);
storeToHeader();
uint8_t *ptr = (uint8_t *)_data->data() + HEADER_SIZE;
storeUint32(ptr, first_pkt_seq_num);
ptr += 4;
storeUint32(ptr, last_pkt_seq_num);
ptr += 4;
return true;
}
} // namespace SRT

360
srt/Packet.hpp Normal file
View File

@ -0,0 +1,360 @@
#ifndef ZLMEDIAKIT_SRT_PACKET_H
#define ZLMEDIAKIT_SRT_PACKET_H
#include <stdint.h>
#include <vector>
#include "Network/Buffer.h"
#include "Network/sockutil.h"
#include "Util/logger.h"
#include "Common.hpp"
#include "HSExt.hpp"
namespace SRT {
using namespace toolkit;
/*
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
+-+-+-+-+-+-+-+-+-+-+-+-+- SRT Header +-+-+-+-+-+-+-+-+-+-+-+-+-+
|0| Packet Sequence Number |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
|P P|O|K K|R| Message Number |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| Timestamp |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| Destination Socket ID |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| |
+ Data +
| |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
Figure 3: Data packet structure
reference https://haivision.github.io/srt-rfc/draft-sharabayko-srt.html#name-packet-structure
*/
class DataPacket : public Buffer {
public:
using Ptr = std::shared_ptr<DataPacket>;
DataPacket() = default;
~DataPacket() = default;
static const size_t HEADER_SIZE = 16;
static bool isDataPacket(uint8_t *buf, size_t len);
static uint32_t getSocketID(uint8_t *buf, size_t len);
bool loadFromData(uint8_t *buf, size_t len);
bool storeToData(uint8_t *buf, size_t len);
bool storeToHeader();
///////Buffer override///////
char *data() const override;
size_t size() const override;
char *payloadData();
size_t payloadSize();
uint8_t f;
uint32_t packet_seq_number;
uint8_t PP;
uint8_t O;
uint8_t KK;
uint8_t R;
uint32_t msg_number;
uint32_t timestamp;
uint32_t dst_socket_id;
private:
BufferRaw::Ptr _data;
};
/*
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
+-+-+-+-+-+-+-+-+-+-+-+-+- SRT Header +-+-+-+-+-+-+-+-+-+-+-+-+-+
|1| Control Type | Subtype |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| Type-specific Information |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| Timestamp |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| Destination Socket ID |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+- CIF -+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| |
+ Control Information Field +
| |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
Figure 4: Control packet structure
reference https://haivision.github.io/srt-rfc/draft-sharabayko-srt.html#name-control-packets
*/
class ControlPacket : public Buffer {
public:
using Ptr = std::shared_ptr<ControlPacket>;
static const size_t HEADER_SIZE = 16;
static bool isControlPacket(uint8_t *buf, size_t len);
static uint16_t getControlType(uint8_t *buf, size_t len);
static uint32_t getSocketID(uint8_t *buf, size_t len);
ControlPacket() = default;
virtual ~ControlPacket() = default;
virtual bool loadFromData(uint8_t *buf, size_t len) = 0;
virtual bool storeToData() = 0;
bool loadHeader();
bool storeToHeader();
///////Buffer override///////
char *data() const override;
size_t size() const override;
enum {
HANDSHAKE = 0x0000,
KEEPALIVE = 0x0001,
ACK = 0x0002,
NAK = 0x0003,
CONGESTIONWARNING = 0x0004,
SHUTDOWN = 0x0005,
ACKACK = 0x0006,
DROPREQ = 0x0007,
PEERERROR = 0x0008,
USERDEFINEDTYPE = 0x7FFF
};
uint32_t sub_type : 16;
uint32_t control_type : 15;
uint32_t f : 1;
uint8_t type_specific_info[4];
uint32_t timestamp;
uint32_t dst_socket_id;
protected:
BufferRaw::Ptr _data;
};
/**
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
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| Version |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| Encryption Field | Extension Field |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| Initial Packet Sequence Number |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| Maximum Transmission Unit Size |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| Maximum Flow Window Size |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| Handshake Type |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| SRT Socket ID |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| SYN Cookie |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| |
+ +
| |
+ Peer IP Address +
| |
+ +
| |
+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+
| Extension Type | Extension Length |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| |
+ Extension Contents +
| |
+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+
Figure 5: Handshake packet structure
https://haivision.github.io/srt-rfc/draft-sharabayko-srt.html#name-handshake
*/
class HandshakePacket : public ControlPacket {
public:
using Ptr = std::shared_ptr<HandshakePacket>;
enum { NO_ENCRYPTION = 0, AES_128 = 1, AES_196 = 2, AES_256 = 3 };
static const size_t HS_CONTENT_MIN_SIZE = 48;
enum {
HS_TYPE_DONE = 0xFFFFFFFD,
HS_TYPE_AGREEMENT = 0xFFFFFFFE,
HS_TYPE_CONCLUSION = 0xFFFFFFFF,
HS_TYPE_WAVEHAND = 0x00000000,
HS_TYPE_INDUCTION = 0x00000001
};
enum { HS_EXT_FILED_HSREQ = 0x00000001, HS_EXT_FILED_KMREQ = 0x00000002, HS_EXT_FILED_CONFIG = 0x00000004 };
HandshakePacket() = default;
~HandshakePacket() = default;
static bool isHandshakePacket(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
generateSynCookie(struct sockaddr_storage *addr, TimePoint ts, uint32_t current_cookie = 0, int correction = 0);
void assignPeerIP(struct sockaddr_storage *addr);
///////ControlPacket override///////
bool loadFromData(uint8_t *buf, size_t len) override;
bool storeToData() override;
uint32_t version;
uint16_t encryption_field;
uint16_t extension_field;
uint32_t initial_packet_sequence_number;
uint32_t mtu;
uint32_t max_flow_window_size;
uint32_t handshake_type;
uint32_t srt_socket_id;
uint32_t syn_cookie;
uint8_t peer_ip_addr[16];
std::vector<HSExt::Ptr> ext_list;
private:
bool loadExtMessage(uint8_t *buf, size_t len);
bool storeExtMessage();
size_t getExtSize();
};
/*
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
+-+-+-+-+-+-+-+-+-+-+-+-+- SRT Header +-+-+-+-+-+-+-+-+-+-+-+-+-+
|1| Control Type | Reserved |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| Type-specific Information |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| Timestamp |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| Destination Socket ID |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
Figure 12: Keep-Alive control packet
https://haivision.github.io/srt-rfc/draft-sharabayko-srt.html#name-keep-alive
*/
class KeepLivePacket : public ControlPacket {
public:
using Ptr = std::shared_ptr<KeepLivePacket>;
KeepLivePacket() = default;
~KeepLivePacket() = default;
///////ControlPacket override///////
bool loadFromData(uint8_t *buf, size_t len) override;
bool storeToData() override;
};
/*
An SRT NAK packet is formatted as follows:
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
+-+-+-+-+-+-+-+-+-+-+-+-+- SRT Header +-+-+-+-+-+-+-+-+-+-+-+-+-+
|1| Control Type | Reserved |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| Type-specific Information |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| Timestamp |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| Destination Socket ID |
+-+-+-+-+-+-+-+-+-+-+-+- CIF (Loss List) -+-+-+-+-+-+-+-+-+-+-+-+
|0| Lost packet sequence number |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
|1| Range of lost packets from sequence number |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
|0| Up to sequence number |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
|0| Lost packet sequence number |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
Figure 14: NAK control packet
https://haivision.github.io/srt-rfc/draft-sharabayko-srt.html#name-nak-control-packet
*/
class NAKPacket : public ControlPacket {
public:
using Ptr = std::shared_ptr<NAKPacket>;
using LostPair = std::pair<uint32_t, uint32_t>;
NAKPacket() = default;
~NAKPacket() = default;
std::string dump();
///////ControlPacket override///////
bool loadFromData(uint8_t *buf, size_t len) override;
bool storeToData() override;
std::list<LostPair> lost_list;
private:
size_t getCIFSize();
};
/*
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
+-+-+-+-+-+-+-+-+-+-+-+-+- SRT Header +-+-+-+-+-+-+-+-+-+-+-+-+-+
|1| Control Type = 7 | Reserved = 0 |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| Message Number |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| Timestamp |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| Destination Socket ID |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| First Packet Sequence Number |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| Last Packet Sequence Number |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
Figure 18: Drop Request control packet
https://haivision.github.io/srt-rfc/draft-sharabayko-srt.html#name-message-drop-request
*/
class MsgDropReqPacket : public ControlPacket {
public:
using Ptr = std::shared_ptr<MsgDropReqPacket>;
MsgDropReqPacket() = default;
~MsgDropReqPacket() = default;
///////ControlPacket override///////
bool loadFromData(uint8_t *buf, size_t len) override;
bool storeToData() override;
uint32_t first_pkt_seq_num;
uint32_t last_pkt_seq_num;
};
/*
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
+-+-+-+-+-+-+-+-+-+-+-+-+- SRT Header +-+-+-+-+-+-+-+-+-+-+-+-+-+
|1| Control Type | Reserved |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| Type-specific Information |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| Timestamp |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| Destination Socket ID |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
Figure 16: Shutdown control packet
https://haivision.github.io/srt-rfc/draft-sharabayko-srt.html#name-shutdown
*/
class ShutDownPacket : public ControlPacket {
public:
using Ptr = std::shared_ptr<ShutDownPacket>;
ShutDownPacket() = default;
~ShutDownPacket() = default;
///////ControlPacket override///////
bool loadFromData(uint8_t *buf, size_t len) override {
if (len < HEADER_SIZE) {
WarnL << "data size" << len << " less " << HEADER_SIZE;
return false;
}
_data = BufferRaw::create();
_data->assign((char *)buf, len);
return loadHeader();
}
bool storeToData() override {
control_type = ControlPacket::SHUTDOWN;
sub_type = 0;
_data = BufferRaw::create();
_data->setCapacity(HEADER_SIZE);
_data->setSize(HEADER_SIZE);
return storeToHeader();
}
};
} // namespace SRT
#endif // ZLMEDIAKIT_SRT_PACKET_H

223
srt/PacketQueue.cpp Normal file
View File

@ -0,0 +1,223 @@
#include "PacketQueue.hpp"
namespace SRT {
static inline bool isSeqEdge(uint32_t seq, uint32_t cap) {
if (seq > (MAX_SEQ - cap)) {
return true;
}
return false;
}
static inline bool isTSCycle(uint32_t first, uint32_t second) {
uint32_t diff;
if (first > second) {
diff = first - second;
} else {
diff = second - first;
}
if (diff > (MAX_TS >> 1)) {
return true;
} else {
return false;
}
}
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) {
auto diff = pkt->packet_seq_number - _pkt_expected_seq;
if (diff >= (MAX_SEQ >> 1)) {
TraceL << "drop packet too later for cycle "
<< "expected seq=" << _pkt_expected_seq << " pkt seq=" << pkt->packet_seq_number;
return;
} else {
_pkt_map.emplace(pkt->packet_seq_number, pkt);
}
} else {
auto diff = _pkt_expected_seq - pkt->packet_seq_number;
if (diff >= (MAX_SEQ >> 1)) {
_pkt_map.emplace(pkt->packet_seq_number, pkt);
TraceL << " cycle packet "
<< "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) {
tryInsertPkt(pkt);
auto it = _pkt_map.find(_pkt_expected_seq);
while (it != _pkt_map.end()) {
out.push_back(it->second);
_pkt_map.erase(it);
_pkt_expected_seq = genExpectedSeq(_pkt_expected_seq + 1);
it = _pkt_map.find(_pkt_expected_seq);
}
while (_pkt_map.size() > _pkt_cap) {
// 防止回环
it = _pkt_map.find(_pkt_expected_seq);
if (it != _pkt_map.end()) {
out.push_back(it->second);
_pkt_map.erase(it);
}
_pkt_expected_seq = genExpectedSeq(_pkt_expected_seq + 1);
}
while (timeLatency() > _pkt_latency) {
it = _pkt_map.find(_pkt_expected_seq);
if (it != _pkt_map.end()) {
out.push_back(it->second);
_pkt_map.erase(it);
}
_pkt_expected_seq = genExpectedSeq(_pkt_expected_seq + 1);
}
return true;
}
bool PacketQueue::drop(uint32_t first, uint32_t last, std::list<DataPacket::Ptr> &out) {
uint32_t end = genExpectedSeq(last + 1);
decltype(_pkt_map.end()) it;
for (uint32_t i = _pkt_expected_seq; i < end;) {
it = _pkt_map.find(i);
if (it != _pkt_map.end()) {
out.push_back(it->second);
_pkt_map.erase(it);
}
i = genExpectedSeq(i + 1);
}
_pkt_expected_seq = end;
return true;
}
uint32_t PacketQueue::timeLatency() {
if (_pkt_map.empty()) {
return 0;
}
auto first = _pkt_map.begin()->second->timestamp;
auto last = _pkt_map.rbegin()->second->timestamp;
uint32_t dur;
if (last > first) {
dur = last - first;
} else {
dur = first - last;
}
if (dur > 0x80000000) {
dur = MAX_TS - dur;
WarnL << "cycle dur " << dur;
}
return dur;
}
std::list<PacketQueue::LostPair> PacketQueue::getLostSeq() {
std::list<PacketQueue::LostPair> re;
if (_pkt_map.empty()) {
return re;
}
if (getExpectedSize() == getSize()) {
return re;
}
uint32_t end = 0;
uint32_t first, last;
first = _pkt_map.begin()->second->packet_seq_number;
last = _pkt_map.rbegin()->second->packet_seq_number;
if ((last - first) > (MAX_SEQ >> 1)) {
TraceL << " cycle seq first " << first << " last " << last << " size " << _pkt_map.size();
end = first;
} else {
end = last;
}
PacketQueue::LostPair lost;
lost.first = 0;
lost.second = 0;
uint32_t i = _pkt_expected_seq;
bool finish = true;
for (i = _pkt_expected_seq; i <= end;) {
if (_pkt_map.find(i) == _pkt_map.end()) {
if (finish) {
finish = false;
lost.first = i;
lost.second = genExpectedSeq(i + 1);
} else {
lost.second = genExpectedSeq(i + 1);
}
} else {
if (!finish) {
finish = true;
re.push_back(lost);
}
}
i = genExpectedSeq(i + 1);
}
return re;
}
size_t PacketQueue::getSize() {
return _pkt_map.size();
}
size_t PacketQueue::getExpectedSize() {
if (_pkt_map.empty()) {
return 0;
}
uint32_t max = _pkt_map.rbegin()->first;
uint32_t min = _pkt_map.begin()->first;
if ((max - min) >= (MAX_SEQ >> 1)) {
TraceL << "cycle "
<< "expected seq " << _pkt_expected_seq << " min " << min << " max " << max << " size "
<< _pkt_map.size();
return MAX_SEQ - _pkt_expected_seq + min + 1;
} else {
return max - _pkt_expected_seq + 1;
}
}
size_t PacketQueue::getAvailableBufferSize() {
auto size = getExpectedSize();
if (_pkt_cap > size) {
return _pkt_cap - size;
}
if (_pkt_cap > _pkt_map.size()) {
return _pkt_cap - _pkt_map.size();
}
WarnL << " cap " << _pkt_cap << " expected size " << size << " map size " << _pkt_map.size();
return _pkt_cap;
}
uint32_t PacketQueue::getExpectedSeq() {
return _pkt_expected_seq;
}
std::string PacketQueue::dump() {
_StrPrinter printer;
if (_pkt_map.empty()) {
printer << " expected seq :" << _pkt_expected_seq;
} else {
printer << " expected seq :" << _pkt_expected_seq << " size:" << _pkt_map.size()
<< " first:" << _pkt_map.begin()->second->packet_seq_number;
printer << " last:" << _pkt_map.rbegin()->second->packet_seq_number;
printer << " latency:" << timeLatency() / 1e3;
}
return std::move(printer);
}
} // namespace SRT

46
srt/PacketQueue.hpp Normal file
View File

@ -0,0 +1,46 @@
#ifndef ZLMEDIAKIT_SRT_PACKET_QUEUE_H
#define ZLMEDIAKIT_SRT_PACKET_QUEUE_H
#include "Packet.hpp"
#include <algorithm>
#include <list>
#include <map>
#include <memory>
#include <tuple>
#include <utility>
namespace SRT {
// for recv
class PacketQueue {
public:
using Ptr = std::shared_ptr<PacketQueue>;
using LostPair = std::pair<uint32_t, uint32_t>;
PacketQueue(uint32_t max_size, uint32_t init_seq, uint32_t latency);
~PacketQueue() = default;
bool inputPacket(DataPacket::Ptr pkt, std::list<DataPacket::Ptr> &out);
uint32_t timeLatency();
std::list<LostPair> getLostSeq();
size_t getSize();
size_t getExpectedSize();
size_t getAvailableBufferSize();
uint32_t getExpectedSeq();
std::string dump();
bool drop(uint32_t first, uint32_t last, std::list<DataPacket::Ptr> &out);
private:
void tryInsertPkt(DataPacket::Ptr pkt);
private:
uint32_t _pkt_cap;
uint32_t _pkt_latency;
uint32_t _pkt_expected_seq = 0;
std::map<uint32_t, DataPacket::Ptr> _pkt_map;
};
} // namespace SRT
#endif // ZLMEDIAKIT_SRT_PACKET_QUEUE_H

79
srt/PacketSendQueue.cpp Normal file
View File

@ -0,0 +1,79 @@
#include "PacketSendQueue.hpp"
namespace SRT {
PacketSendQueue::PacketSendQueue(uint32_t max_size, uint32_t latency)
: _pkt_cap(max_size)
, _pkt_latency(latency) {}
bool PacketSendQueue::drop(uint32_t num) {
decltype(_pkt_cache.begin()) it;
for (it = _pkt_cache.begin(); it != _pkt_cache.end(); ++it) {
if ((*it)->packet_seq_number == num) {
break;
}
}
if (it != _pkt_cache.end()) {
_pkt_cache.erase(_pkt_cache.begin(), it);
}
return true;
}
bool PacketSendQueue::inputPacket(DataPacket::Ptr pkt) {
_pkt_cache.push_back(pkt);
while (_pkt_cache.size() > _pkt_cap) {
_pkt_cache.pop_front();
}
while (timeLatency() > _pkt_latency) {
_pkt_cache.pop_front();
}
return true;
}
std::list<DataPacket::Ptr> PacketSendQueue::findPacketBySeq(uint32_t start, uint32_t end) {
std::list<DataPacket::Ptr> re;
decltype(_pkt_cache.begin()) it;
for (it = _pkt_cache.begin(); it != _pkt_cache.end(); ++it) {
if ((*it)->packet_seq_number == start) {
break;
}
}
if (start == end) {
if (it != _pkt_cache.end()) {
re.push_back(*it);
}
return re;
}
for (; it != _pkt_cache.end(); ++it) {
re.push_back(*it);
if ((*it)->packet_seq_number == end) {
break;
}
}
return re;
}
uint32_t PacketSendQueue::timeLatency() {
if (_pkt_cache.empty()) {
return 0;
}
auto first = _pkt_cache.front()->timestamp;
auto last = _pkt_cache.back()->timestamp;
uint32_t dur;
if (last > first) {
dur = last - first;
} else {
dur = first - last;
}
if (dur > ((uint32_t)0x01 << 31)) {
TraceL << "cycle timeLatency " << dur;
dur = 0xffffffff - dur;
}
return dur;
}
} // namespace SRT

36
srt/PacketSendQueue.hpp Normal file
View File

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

146
srt/SrtSession.cpp Normal file
View File

@ -0,0 +1,146 @@
#include "SrtSession.hpp"
#include "Packet.hpp"
#include "SrtTransportImp.hpp"
#include "Common/config.h"
namespace SRT {
using namespace mediakit;
SrtSession::SrtSession(const Socket::Ptr &sock)
: UdpSession(sock) {
socklen_t addr_len = sizeof(_peer_addr);
memset(&_peer_addr, 0, addr_len);
// TraceL<<"before addr len "<<addr_len;
getpeername(sock->rawFD(), (struct sockaddr *)&_peer_addr, &addr_len);
// TraceL<<"after addr len "<<addr_len<<" family "<<_peer_addr.ss_family;
}
SrtSession::~SrtSession() {
InfoP(this);
}
EventPoller::Ptr SrtSession::queryPoller(const Buffer::Ptr &buffer) {
uint8_t *data = (uint8_t *)buffer->data();
size_t size = buffer->size();
if (DataPacket::isDataPacket(data, size)) {
uint32_t socket_id = DataPacket::getSocketID(data, size);
auto trans = SrtTransportManager::Instance().getItem(std::to_string(socket_id));
return trans ? trans->getPoller() : nullptr;
}
if (HandshakePacket::isHandshakePacket(data, size)) {
auto type = HandshakePacket::getHandshakeType(data, size);
if (type == HandshakePacket::HS_TYPE_INDUCTION) {
// 握手第一阶段
return nullptr;
} else if (type == HandshakePacket::HS_TYPE_CONCLUSION) {
// 握手第二阶段
uint32_t sync_cookie = HandshakePacket::getSynCookie(data, size);
auto trans = SrtTransportManager::Instance().getHandshakeItem(std::to_string(sync_cookie));
return trans ? trans->getPoller() : nullptr;
} else {
WarnL << " not reach there";
}
} else {
uint32_t socket_id = ControlPacket::getSocketID(data, size);
auto trans = SrtTransportManager::Instance().getItem(std::to_string(socket_id));
return trans ? trans->getPoller() : nullptr;
}
return nullptr;
}
void SrtSession::attachServer(const toolkit::Server &server) {
SockUtil::setRecvBuf(getSock()->rawFD(), 1024 * 1024);
}
void SrtSession::onRecv(const Buffer::Ptr &buffer) {
uint8_t *data = (uint8_t *)buffer->data();
size_t size = buffer->size();
if (_find_transport) {
//只允许寻找一次transport
_find_transport = false;
if (DataPacket::isDataPacket(data, size)) {
uint32_t socket_id = DataPacket::getSocketID(data, size);
auto trans = SrtTransportManager::Instance().getItem(std::to_string(socket_id));
if (trans) {
_transport = std::move(trans);
} else {
WarnL << " data packet not find transport ";
}
}
if (HandshakePacket::isHandshakePacket(data, size)) {
auto type = HandshakePacket::getHandshakeType(data, size);
if (type == HandshakePacket::HS_TYPE_INDUCTION) {
// 握手第一阶段
_transport = std::make_shared<SrtTransportImp>(getPoller());
} else if (type == HandshakePacket::HS_TYPE_CONCLUSION) {
// 握手第二阶段
uint32_t sync_cookie = HandshakePacket::getSynCookie(data, size);
auto trans = SrtTransportManager::Instance().getHandshakeItem(std::to_string(sync_cookie));
if (trans) {
_transport = std::move(trans);
} else {
WarnL << " hanshake packet not find transport ";
}
} else {
WarnL << " not reach there";
}
} else {
uint32_t socket_id = ControlPacket::getSocketID(data, size);
auto trans = SrtTransportManager::Instance().getItem(std::to_string(socket_id));
if (trans) {
_transport = std::move(trans);
} else {
WarnL << " not find transport";
}
}
if (_transport) {
_transport->setSession(shared_from_this());
}
InfoP(this);
}
_ticker.resetTime();
if (_transport) {
_transport->inputSockData(data, size, &_peer_addr);
} else {
// WarnL<< "ingore data";
}
}
void SrtSession::onError(const SockException &err) {
// udp链接超时但是srt链接不一定超时因为可能存在udp链接迁移的情况
//在udp链接迁移时新的SrtSession对象将接管SrtSession对象的生命周期
//本SrtSession对象将在超时后自动销毁
WarnP(this) << err.what();
if (!_transport) {
return;
}
// 防止互相引用导致不释放
auto transport = std::move(_transport);
getPoller()->async(
[transport, err] {
//延时减引用防止使用transport对象时销毁对象
transport->onShutdown(err);
},
false);
}
void SrtSession::onManager() {
GET_CONFIG(float, timeoutSec, kTimeOutSec);
if (_ticker.elapsedTime() > timeoutSec * 1000) {
shutdown(SockException(Err_timeout, "srt connection timeout"));
return;
}
}
} // namespace SRT

30
srt/SrtSession.hpp Normal file
View File

@ -0,0 +1,30 @@
#ifndef ZLMEDIAKIT_SRT_SESSION_H
#define ZLMEDIAKIT_SRT_SESSION_H
#include "Network/Session.h"
#include "SrtTransport.hpp"
namespace SRT {
using namespace toolkit;
class SrtSession : public UdpSession {
public:
SrtSession(const Socket::Ptr &sock);
~SrtSession() override;
void onRecv(const Buffer::Ptr &) override;
void onError(const SockException &err) override;
void onManager() override;
void attachServer(const toolkit::Server &server) override;
static EventPoller::Ptr queryPoller(const Buffer::Ptr &buffer);
private:
bool _find_transport = true;
Ticker _ticker;
struct sockaddr_storage _peer_addr;
SrtTransport::Ptr _transport;
};
} // namespace SRT
#endif // ZLMEDIAKIT_SRT_SESSION_H

673
srt/SrtTransport.cpp Normal file
View File

@ -0,0 +1,673 @@
#include "Util/onceToken.h"
#include <stdlib.h>
#include "Ack.hpp"
#include "Packet.hpp"
#include "SrtTransport.hpp"
namespace SRT {
#define SRT_FIELD "srt."
// srt 超时时间
const std::string kTimeOutSec = SRT_FIELD "timeoutSec";
// srt 单端口udp服务器
const std::string kPort = SRT_FIELD "port";
const std::string kLatencyMul = SRT_FIELD "latencyMul";
static std::atomic<uint32_t> s_srt_socket_id_generate { 125 };
//////////// SrtTransport //////////////////////////
SrtTransport::SrtTransport(const EventPoller::Ptr &poller)
: _poller(poller) {
_start_timestamp = SteadyClock::now();
_socket_id = s_srt_socket_id_generate.fetch_add(1);
_pkt_recv_rate_context = std::make_shared<PacketRecvRateContext>(_start_timestamp);
_recv_rate_context = std::make_shared<RecvRateContext>(_start_timestamp);
_estimated_link_capacity_context = std::make_shared<EstimatedLinkCapacityContext>(_start_timestamp);
}
SrtTransport::~SrtTransport() {
TraceL << " ";
}
const EventPoller::Ptr &SrtTransport::getPoller() const {
return _poller;
}
void SrtTransport::setSession(Session::Ptr session) {
_history_sessions.emplace(session.get(), session);
if (_selected_session) {
InfoL << "srt network changed: " << _selected_session->get_peer_ip() << ":"
<< _selected_session->get_peer_port() << " -> " << session->get_peer_ip() << ":"
<< session->get_peer_port() << ", id:" << _selected_session->getIdentifier();
}
_selected_session = session;
}
const Session::Ptr &SrtTransport::getSession() const {
return _selected_session;
}
void SrtTransport::switchToOtherTransport(uint8_t *buf, int len, uint32_t socketid, struct sockaddr_storage *addr) {
BufferRaw::Ptr tmp = BufferRaw::create();
struct sockaddr_storage tmp_addr = *addr;
tmp->assign((char *)buf, len);
auto trans = SrtTransportManager::Instance().getItem(std::to_string(socketid));
if (trans) {
trans->getPoller()->async([tmp, tmp_addr, trans] {
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) {
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 onceToken token([]() {
s_control_functions.emplace(ControlPacket::HANDSHAKE, &SrtTransport::handleHandshake);
s_control_functions.emplace(ControlPacket::KEEPALIVE, &SrtTransport::handleKeeplive);
s_control_functions.emplace(ControlPacket::ACK, &SrtTransport::handleACK);
s_control_functions.emplace(ControlPacket::NAK, &SrtTransport::handleNAK);
s_control_functions.emplace(ControlPacket::CONGESTIONWARNING, &SrtTransport::handleCongestionWarning);
s_control_functions.emplace(ControlPacket::SHUTDOWN, &SrtTransport::handleShutDown);
s_control_functions.emplace(ControlPacket::ACKACK, &SrtTransport::handleACKACK);
s_control_functions.emplace(ControlPacket::DROPREQ, &SrtTransport::handleDropReq);
s_control_functions.emplace(ControlPacket::PEERERROR, &SrtTransport::handlePeerError);
s_control_functions.emplace(ControlPacket::USERDEFINEDTYPE, &SrtTransport::handleUserDefinedType);
});
_now = SteadyClock::now();
// 处理srt数据
if (DataPacket::isDataPacket(buf, len)) {
uint32_t socketId = DataPacket::getSocketID(buf, len);
if (socketId == _socket_id) {
_pkt_recv_rate_context->inputPacket(_now);
_estimated_link_capacity_context->inputPacket(_now);
_recv_rate_context->inputPacket(_now, len);
handleDataPacket(buf, len, addr);
} else {
switchToOtherTransport(buf, len, socketId, addr);
}
} else {
if (ControlPacket::isControlPacket(buf, len)) {
uint32_t socketId = ControlPacket::getSocketID(buf, len);
uint16_t type = ControlPacket::getControlType(buf, len);
if (type != ControlPacket::HANDSHAKE && socketId != _socket_id && _socket_id != 0) {
// socket id not same
switchToOtherTransport(buf, len, socketId, addr);
return;
}
_pkt_recv_rate_context->inputPacket(_now);
_estimated_link_capacity_context->inputPacket(_now);
_recv_rate_context->inputPacket(_now, len);
auto it = s_control_functions.find(type);
if (it == s_control_functions.end()) {
WarnL << " not support type ignore" << ControlPacket::getControlType(buf, len);
return;
} else {
(this->*(it->second))(buf, len, addr);
}
} else {
// not reach
WarnL << "not reach this";
}
}
}
void SrtTransport::handleHandshakeInduction(HandshakePacket &pkt, struct sockaddr_storage *addr) {
// Induction Phase
TraceL << getIdentifier() << " Induction Phase ";
if (_handleshake_res) {
TraceL << getIdentifier() << " Induction handle repeate ";
sendControlPacket(_handleshake_res, true);
return;
}
_induction_ts = _now;
_start_timestamp = _now;
_init_seq_number = pkt.initial_packet_sequence_number;
_max_window_size = pkt.max_flow_window_size;
_mtu = pkt.mtu;
_last_pkt_seq = _init_seq_number - 1;
_peer_socket_id = pkt.srt_socket_id;
HandshakePacket::Ptr res = std::make_shared<HandshakePacket>();
res->dst_socket_id = _peer_socket_id;
res->timestamp = DurationCountMicroseconds(_start_timestamp.time_since_epoch());
res->mtu = _mtu;
res->max_flow_window_size = _max_window_size;
res->initial_packet_sequence_number = _init_seq_number;
res->version = 5;
res->encryption_field = HandshakePacket::NO_ENCRYPTION;
res->extension_field = 0x4A17;
res->handshake_type = HandshakePacket::HS_TYPE_INDUCTION;
res->srt_socket_id = _peer_socket_id;
res->syn_cookie = HandshakePacket::generateSynCookie(addr, _start_timestamp);
_sync_cookie = res->syn_cookie;
memcpy(res->peer_ip_addr, pkt.peer_ip_addr, sizeof(pkt.peer_ip_addr) * sizeof(pkt.peer_ip_addr[0]));
_handleshake_res = res;
res->storeToData();
registerSelfHandshake();
sendControlPacket(res, true);
}
void SrtTransport::handleHandshakeConclusion(HandshakePacket &pkt, struct sockaddr_storage *addr) {
if (!_handleshake_res) {
ErrorL << "must Induction Phase for handleshake ";
return;
}
if (_handleshake_res->handshake_type == HandshakePacket::HS_TYPE_INDUCTION) {
// first
HSExtMessage::Ptr req;
HSExtStreamID::Ptr sid;
uint32_t srt_flag = 0xbf;
uint16_t delay = DurationCountMicroseconds(_now - _induction_ts) * getLatencyMul() / 1000;
for (auto ext : pkt.ext_list) {
// TraceL << getIdentifier() << " ext " << ext->dump();
if (!req) {
req = std::dynamic_pointer_cast<HSExtMessage>(ext);
}
if (!sid) {
sid = std::dynamic_pointer_cast<HSExtStreamID>(ext);
}
}
if (sid) {
_stream_id = sid->streamid;
}
if (req) {
srt_flag = req->srt_flag;
delay = delay <= req->recv_tsbpd_delay ? req->recv_tsbpd_delay : delay;
}
TraceL << getIdentifier() << " CONCLUSION Phase ";
HandshakePacket::Ptr res = std::make_shared<HandshakePacket>();
res->dst_socket_id = _peer_socket_id;
res->timestamp = DurationCountMicroseconds(_now - _start_timestamp);
res->mtu = _mtu;
res->max_flow_window_size = _max_window_size;
res->initial_packet_sequence_number = _init_seq_number;
res->version = 5;
res->encryption_field = HandshakePacket::NO_ENCRYPTION;
res->extension_field = HandshakePacket::HS_EXT_FILED_HSREQ;
res->handshake_type = HandshakePacket::HS_TYPE_CONCLUSION;
res->srt_socket_id = _socket_id;
res->syn_cookie = 0;
res->assignPeerIP(addr);
HSExtMessage::Ptr ext = std::make_shared<HSExtMessage>();
ext->extension_type = HSExt::SRT_CMD_HSRSP;
ext->srt_version = srtVersion(1, 5, 0);
ext->srt_flag = srt_flag;
ext->recv_tsbpd_delay = ext->send_tsbpd_delay = delay;
res->ext_list.push_back(std::move(ext));
res->storeToData();
_handleshake_res = res;
unregisterSelfHandshake();
registerSelf();
sendControlPacket(res, true);
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);
_send_buf = std::make_shared<PacketSendQueue>(res->max_flow_window_size, delay * 1e3);
_send_packet_seq_number = _init_seq_number;
_buf_delay = delay;
onHandShakeFinished(_stream_id, addr);
} else {
TraceL << getIdentifier() << " CONCLUSION handle repeate ";
sendControlPacket(_handleshake_res, true);
}
_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) {
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);
_nak_ticker.resetTime(_now);
}
void SrtTransport::handleKeeplive(uint8_t *buf, int len, struct sockaddr_storage *addr) {
// TraceL;
sendKeepLivePacket();
}
void SrtTransport::sendKeepLivePacket() {
KeepLivePacket::Ptr pkt = std::make_shared<KeepLivePacket>();
pkt->dst_socket_id = _peer_socket_id;
pkt->timestamp = DurationCountMicroseconds(_now - _start_timestamp);
pkt->storeToData();
sendControlPacket(pkt, true);
}
void SrtTransport::handleACK(uint8_t *buf, int len, struct sockaddr_storage *addr) {
// TraceL;
ACKPacket ack;
if (!ack.loadFromData(buf, len)) {
return;
}
ACKACKPacket::Ptr pkt = std::make_shared<ACKACKPacket>();
pkt->dst_socket_id = _peer_socket_id;
pkt->timestamp = DurationCountMicroseconds(_now - _start_timestamp);
pkt->ack_number = ack.ack_number;
pkt->storeToData();
_send_buf->drop(ack.last_ack_pkt_seq_number);
sendControlPacket(pkt, true);
// TraceL<<"ack number "<<ack.ack_number;
}
void SrtTransport::sendMsgDropReq(uint32_t first, uint32_t last) {
MsgDropReqPacket::Ptr pkt = std::make_shared<MsgDropReqPacket>();
pkt->dst_socket_id = _peer_socket_id;
pkt->timestamp = DurationCountMicroseconds(_now - _start_timestamp);
pkt->first_pkt_seq_num = first;
pkt->last_pkt_seq_num = last;
pkt->storeToData();
sendControlPacket(pkt, true);
}
void SrtTransport::handleNAK(uint8_t *buf, int len, struct sockaddr_storage *addr) {
// TraceL;
NAKPacket pkt;
pkt.loadFromData(buf, len);
bool empty = false;
bool flush = false;
for (auto it : pkt.lost_list) {
if (pkt.lost_list.back() == it) {
flush = true;
}
empty = true;
auto re_list = _send_buf->findPacketBySeq(it.first, it.second - 1);
for (auto pkt : re_list) {
pkt->R = 1;
pkt->storeToHeader();
sendPacket(pkt, flush);
empty = false;
}
if (empty) {
sendMsgDropReq(it.first, it.second - 1);
}
}
}
void SrtTransport::handleCongestionWarning(uint8_t *buf, int len, struct sockaddr_storage *addr) {
TraceL;
}
void SrtTransport::handleShutDown(uint8_t *buf, int len, struct sockaddr_storage *addr) {
TraceL;
onShutdown(SockException(Err_shutdown, "peer close connection"));
}
void SrtTransport::handleDropReq(uint8_t *buf, int len, struct sockaddr_storage *addr) {
MsgDropReqPacket pkt;
pkt.loadFromData(buf, len);
std::list<DataPacket::Ptr> list;
// 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);
if (list.empty()) {
return;
}
uint32_t max_seq = 0;
for (auto data : list) {
max_seq = data->packet_seq_number;
if (_last_pkt_seq + 1 != data->packet_seq_number) {
TraceL << "pkt lost " << _last_pkt_seq + 1 << "->" << data->packet_seq_number;
}
_last_pkt_seq = data->packet_seq_number;
onSRTData(std::move(data));
}
_recv_nack.drop(max_seq);
auto lost = _recv_buf->getLostSeq();
_recv_nack.update(_now, lost);
lost.clear();
_recv_nack.getLostList(_now, _rtt, _rtt_variance, lost);
if (!lost.empty()) {
sendNAKPacket(lost);
// TraceL << "check lost send nack";
}
auto nak_interval = (_rtt + _rtt_variance * 4) / 2;
if (nak_interval <= 20 * 1000) {
nak_interval = 20 * 1000;
}
if (_nak_ticker.elapsedTime(_now) > nak_interval) {
auto lost = _recv_buf->getLostSeq();
if (!lost.empty()) {
sendNAKPacket(lost);
}
_nak_ticker.resetTime(_now);
}
if (_ack_ticker.elapsedTime(_now) > 10 * 1000) {
_light_ack_pkt_count = 0;
_ack_ticker.resetTime(_now);
// send a ack per 10 ms for receiver
sendACKPacket();
} else {
if (_light_ack_pkt_count >= 64) {
// for high bitrate stream send light ack
// TODO
sendLightACKPacket();
TraceL << "send light ack";
}
_light_ack_pkt_count = 0;
}
_light_ack_pkt_count++;
}
void SrtTransport::handleUserDefinedType(uint8_t *buf, int len, struct sockaddr_storage *addr) {
TraceL;
}
void SrtTransport::handleACKACK(uint8_t *buf, int len, struct sockaddr_storage *addr) {
// TraceL;
ACKACKPacket::Ptr pkt = std::make_shared<ACKACKPacket>();
pkt->loadFromData(buf, len);
uint32_t rtt = DurationCountMicroseconds(_now - _ack_send_timestamp[pkt->ack_number]);
_rtt_variance = (3 * _rtt_variance + abs((long)_rtt - (long)rtt)) / 4;
_rtt = (7 * rtt + _rtt) / 8;
// TraceL<<" rtt:"<<_rtt<<" rtt variance:"<<_rtt_variance;
_ack_send_timestamp.erase(pkt->ack_number);
}
void SrtTransport::handlePeerError(uint8_t *buf, int len, struct sockaddr_storage *addr) {
TraceL;
}
void SrtTransport::sendACKPacket() {
ACKPacket::Ptr pkt = std::make_shared<ACKPacket>();
pkt->dst_socket_id = _peer_socket_id;
pkt->timestamp = DurationCountMicroseconds(_now - _start_timestamp);
pkt->ack_number = ++_ack_number_count;
pkt->last_ack_pkt_seq_number = _recv_buf->getExpectedSeq();
pkt->rtt = _rtt;
pkt->rtt_variance = _rtt_variance;
pkt->available_buf_size = _recv_buf->getAvailableBufferSize();
pkt->pkt_recv_rate = _pkt_recv_rate_context->getPacketRecvRate();
pkt->estimated_link_capacity = _estimated_link_capacity_context->getEstimatedLinkCapacity();
pkt->recv_rate = _recv_rate_context->getRecvRate();
pkt->storeToData();
_ack_send_timestamp[pkt->ack_number] = _now;
_last_ack_pkt_seq_num = pkt->last_ack_pkt_seq_number;
sendControlPacket(pkt, true);
// TraceL<<"send ack "<<pkt->dump();
}
void SrtTransport::sendLightACKPacket() {
ACKPacket::Ptr pkt = std::make_shared<ACKPacket>();
pkt->dst_socket_id = _peer_socket_id;
pkt->timestamp = DurationCountMicroseconds(_now - _start_timestamp);
pkt->ack_number = 0;
pkt->last_ack_pkt_seq_number = _recv_buf->getExpectedSeq();
pkt->rtt = 0;
pkt->rtt_variance = 0;
pkt->available_buf_size = 0;
pkt->pkt_recv_rate = 0;
pkt->estimated_link_capacity = 0;
pkt->recv_rate = 0;
pkt->storeToData();
_last_ack_pkt_seq_num = pkt->last_ack_pkt_seq_number;
sendControlPacket(pkt, true);
TraceL << "send ack " << pkt->dump();
}
void SrtTransport::sendNAKPacket(std::list<PacketQueue::LostPair> &lost_list) {
NAKPacket::Ptr pkt = std::make_shared<NAKPacket>();
pkt->dst_socket_id = _peer_socket_id;
pkt->timestamp = DurationCountMicroseconds(_now - _start_timestamp);
pkt->lost_list = lost_list;
pkt->storeToData();
// TraceL<<"send NAK "<<pkt->dump();
sendControlPacket(pkt, true);
}
void SrtTransport::sendShutDown() {
ShutDownPacket::Ptr pkt = std::make_shared<ShutDownPacket>();
pkt->dst_socket_id = _peer_socket_id;
pkt->timestamp = DurationCountMicroseconds(_now - _start_timestamp);
pkt->storeToData();
sendControlPacket(pkt, true);
}
void SrtTransport::handleDataPacket(uint8_t *buf, int len, struct sockaddr_storage *addr) {
DataPacket::Ptr pkt = std::make_shared<DataPacket>();
pkt->loadFromData(buf, len);
std::list<DataPacket::Ptr> list;
//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;
_recv_buf->inputPacket(pkt, list);
if (list.empty()) {
// when no data ok send nack to sender immediately
} else {
uint32_t last_seq;
for (auto data : list) {
last_seq = data->packet_seq_number;
if (_last_pkt_seq + 1 != data->packet_seq_number) {
TraceL << "pkt lost " << _last_pkt_seq + 1 << "->" << data->packet_seq_number;
}
_last_pkt_seq = data->packet_seq_number;
onSRTData(std::move(data));
}
_recv_nack.drop(last_seq);
}
auto lost = _recv_buf->getLostSeq();
_recv_nack.update(_now, lost);
lost.clear();
_recv_nack.getLostList(_now, _rtt, _rtt_variance, lost);
if (!lost.empty()) {
// TraceL << "check lost send nack immediately";
sendNAKPacket(lost);
}
auto nak_interval = (_rtt + _rtt_variance * 4) / 2;
if (nak_interval <= 20 * 1000) {
nak_interval = 20 * 1000;
}
if (_nak_ticker.elapsedTime(_now) > nak_interval) {
// Periodic NAK reports
auto lost = _recv_buf->getLostSeq();
if (!lost.empty()) {
sendNAKPacket(lost);
// TraceL<<"send NAK";
} else {
// TraceL<<"lost is empty";
}
_nak_ticker.resetTime(_now);
}
if (_ack_ticker.elapsedTime(_now) > 10 * 1000) {
_light_ack_pkt_count = 0;
_ack_ticker.resetTime(_now);
// send a ack per 10 ms for receiver
sendACKPacket();
} else {
if (_light_ack_pkt_count >= 64) {
// for high bitrate stream send light ack
// TODO
sendLightACKPacket();
TraceL << "send light ack";
}
_light_ack_pkt_count = 0;
}
_light_ack_pkt_count++;
// bufCheckInterval();
}
void SrtTransport::sendDataPacket(DataPacket::Ptr pkt, char *buf, int len, bool flush) {
pkt->storeToData((uint8_t *)buf, len);
sendPacket(pkt, flush);
_send_buf->inputPacket(pkt);
}
void SrtTransport::sendControlPacket(ControlPacket::Ptr pkt, bool flush) {
sendPacket(pkt, flush);
}
void SrtTransport::sendPacket(Buffer::Ptr pkt, bool flush) {
if (_selected_session) {
auto tmp = _packet_pool.obtain2();
tmp->assign(pkt->data(), pkt->size());
_selected_session->setSendFlushFlag(flush);
_selected_session->send(std::move(tmp));
} else {
WarnL << "not reach this";
}
}
std::string SrtTransport::getIdentifier() {
return _selected_session ? _selected_session->getIdentifier() : "";
}
void SrtTransport::registerSelfHandshake() {
SrtTransportManager::Instance().addHandshakeItem(std::to_string(_sync_cookie), shared_from_this());
}
void SrtTransport::unregisterSelfHandshake() {
if (_sync_cookie == 0) {
return;
}
SrtTransportManager::Instance().removeHandshakeItem(std::to_string(_sync_cookie));
}
void SrtTransport::registerSelf() {
if (_socket_id == 0) {
return;
}
SrtTransportManager::Instance().addItem(std::to_string(_socket_id), shared_from_this());
}
void SrtTransport::unregisterSelf() {
SrtTransportManager::Instance().removeItem(std::to_string(_socket_id));
}
void SrtTransport::onShutdown(const SockException &ex) {
sendShutDown();
WarnL << ex.what();
unregisterSelfHandshake();
unregisterSelf();
for (auto &pr : _history_sessions) {
auto session = pr.second.lock();
if (session) {
session->shutdown(ex);
}
}
}
size_t SrtTransport::getPayloadSize() {
size_t ret = (_mtu - 28 - 16) / 188 * 188;
return ret;
}
void SrtTransport::onSendTSData(const Buffer::Ptr &buffer, bool flush) {
// TraceL;
DataPacket::Ptr pkt;
size_t payloadSize = getPayloadSize();
size_t size = buffer->size();
char *ptr = buffer->data();
char *end = buffer->data() + size;
while (ptr < end && size >= payloadSize) {
pkt = std::make_shared<DataPacket>();
pkt->f = 0;
pkt->packet_seq_number = _send_packet_seq_number & 0x7fffffff;
_send_packet_seq_number = (_send_packet_seq_number + 1) & 0x7fffffff;
pkt->PP = 3;
pkt->O = 0;
pkt->KK = 0;
pkt->R = 0;
pkt->msg_number = _send_msg_number++;
pkt->dst_socket_id = _peer_socket_id;
pkt->timestamp = DurationCountMicroseconds(SteadyClock::now() - _start_timestamp);
sendDataPacket(pkt, ptr, (int)payloadSize, flush);
ptr += payloadSize;
size -= payloadSize;
}
if (size > 0 && ptr < end) {
pkt = std::make_shared<DataPacket>();
pkt->f = 0;
pkt->packet_seq_number = _send_packet_seq_number & 0x7fffffff;
_send_packet_seq_number = (_send_packet_seq_number + 1) & 0x7fffffff;
pkt->PP = 3;
pkt->O = 0;
pkt->KK = 0;
pkt->R = 0;
pkt->msg_number = _send_msg_number++;
pkt->dst_socket_id = _peer_socket_id;
pkt->timestamp = DurationCountMicroseconds(SteadyClock::now() - _start_timestamp);
sendDataPacket(pkt, ptr, (int)size, flush);
}
}
//////////// SrtTransportManager //////////////////////////
SrtTransportManager &SrtTransportManager::Instance() {
static SrtTransportManager s_instance;
return s_instance;
}
void SrtTransportManager::addItem(const std::string &key, const SrtTransport::Ptr &ptr) {
std::lock_guard<std::mutex> lck(_mtx);
_map[key] = ptr;
}
SrtTransport::Ptr SrtTransportManager::getItem(const std::string &key) {
if (key.empty()) {
return nullptr;
}
std::lock_guard<std::mutex> lck(_mtx);
auto it = _map.find(key);
if (it == _map.end()) {
return nullptr;
}
return it->second.lock();
}
void SrtTransportManager::removeItem(const std::string &key) {
std::lock_guard<std::mutex> lck(_mtx);
_map.erase(key);
}
void SrtTransportManager::addHandshakeItem(const std::string &key, const SrtTransport::Ptr &ptr) {
std::lock_guard<std::mutex> lck(_handshake_mtx);
_handshake_map[key] = ptr;
}
void SrtTransportManager::removeHandshakeItem(const std::string &key) {
std::lock_guard<std::mutex> lck(_handshake_mtx);
_handshake_map.erase(key);
}
SrtTransport::Ptr SrtTransportManager::getHandshakeItem(const std::string &key) {
if (key.empty()) {
return nullptr;
}
std::lock_guard<std::mutex> lck(_handshake_mtx);
auto it = _handshake_map.find(key);
if (it == _handshake_map.end()) {
return nullptr;
}
return it->second.lock();
}
} // namespace SRT

169
srt/SrtTransport.hpp Normal file
View File

@ -0,0 +1,169 @@
#ifndef ZLMEDIAKIT_SRT_TRANSPORT_H
#define ZLMEDIAKIT_SRT_TRANSPORT_H
#include <atomic>
#include <chrono>
#include <memory>
#include <mutex>
#include "Network/Session.h"
#include "Poller/EventPoller.h"
#include "Poller/Timer.h"
#include "Common.hpp"
#include "NackContext.hpp"
#include "Packet.hpp"
#include "PacketQueue.hpp"
#include "PacketSendQueue.hpp"
#include "Statistic.hpp"
namespace SRT {
using namespace toolkit;
extern const std::string kPort;
extern const std::string kTimeOutSec;
extern const std::string kLatencyMul;
class SrtTransport : public std::enable_shared_from_this<SrtTransport> {
public:
friend class SrtSession;
using Ptr = std::shared_ptr<SrtTransport>;
SrtTransport(const EventPoller::Ptr &poller);
virtual ~SrtTransport();
const EventPoller::Ptr &getPoller() const;
void setSession(Session::Ptr session);
const Session::Ptr &getSession() const;
/**
* socket收到udp数据
* @param buf
* @param len
* @param addr
*/
virtual void inputSockData(uint8_t *buf, int len, struct sockaddr_storage *addr);
virtual void onSendTSData(const Buffer::Ptr &buffer, bool flush);
std::string getIdentifier();
void unregisterSelf();
void unregisterSelfHandshake();
protected:
virtual bool isPusher() { return true; };
virtual void onSRTData(DataPacket::Ptr pkt) {};
virtual void onShutdown(const SockException &ex);
virtual void onHandShakeFinished(std::string &streamid, struct sockaddr_storage *addr) {};
virtual void sendPacket(Buffer::Ptr pkt, bool flush = true);
virtual int getLatencyMul() { return 4; };
private:
void registerSelf();
void registerSelfHandshake();
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 handleHandshakeInduction(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 handleACK(uint8_t *buf, int len, struct sockaddr_storage *addr);
void handleACKACK(uint8_t *buf, int len, struct sockaddr_storage *addr);
void handleNAK(uint8_t *buf, int len, struct sockaddr_storage *addr);
void handleCongestionWarning(uint8_t *buf, int len, struct sockaddr_storage *addr);
void handleShutDown(uint8_t *buf, int len, struct sockaddr_storage *addr);
void handleDropReq(uint8_t *buf, int len, struct sockaddr_storage *addr);
void handleUserDefinedType(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 sendNAKPacket(std::list<PacketQueue::LostPair> &lost_list);
void sendACKPacket();
void sendLightACKPacket();
void sendKeepLivePacket();
void sendShutDown();
void sendMsgDropReq(uint32_t first, uint32_t last);
size_t getPayloadSize();
protected:
void sendDataPacket(DataPacket::Ptr pkt, char *buf, int len, bool flush = false);
void sendControlPacket(ControlPacket::Ptr pkt, bool flush = true);
private:
// 当前选中的udp链接
Session::Ptr _selected_session;
// 链接迁移前后使用过的udp链接
std::unordered_map<Session *, std::weak_ptr<Session>> _history_sessions;
EventPoller::Ptr _poller;
uint32_t _peer_socket_id;
uint32_t _socket_id = 0;
TimePoint _now;
TimePoint _start_timestamp;
// for calculate rtt for delay
TimePoint _induction_ts;
uint32_t _mtu = 1500;
uint32_t _max_window_size = 8192;
uint32_t _init_seq_number = 0;
std::string _stream_id;
uint32_t _sync_cookie = 0;
uint32_t _send_packet_seq_number = 0;
uint32_t _send_msg_number = 1;
PacketSendQueue::Ptr _send_buf;
uint32_t _buf_delay = 120;
PacketQueue::Ptr _recv_buf;
NackContext _recv_nack;
uint32_t _rtt = 100 * 1000;
uint32_t _rtt_variance = 50 * 1000;
uint32_t _light_ack_pkt_count = 0;
uint32_t _ack_number_count = 0;
uint32_t _last_ack_pkt_seq_num = 0;
uint32_t _last_pkt_seq = 0;
UTicker _ack_ticker;
std::map<uint32_t, TimePoint> _ack_send_timestamp;
std::shared_ptr<PacketRecvRateContext> _pkt_recv_rate_context;
std::shared_ptr<EstimatedLinkCapacityContext> _estimated_link_capacity_context;
std::shared_ptr<RecvRateContext> _recv_rate_context;
UTicker _nak_ticker;
// 保持发送的握手消息,防止丢失重发
HandshakePacket::Ptr _handleshake_res;
ResourcePool<BufferRaw> _packet_pool;
};
class SrtTransportManager {
public:
static SrtTransportManager &Instance();
SrtTransport::Ptr getItem(const std::string &key);
void addItem(const std::string &key, const SrtTransport::Ptr &ptr);
void removeItem(const std::string &key);
void addHandshakeItem(const std::string &key, const SrtTransport::Ptr &ptr);
void removeHandshakeItem(const std::string &key);
SrtTransport::Ptr getHandshakeItem(const std::string &key);
private:
SrtTransportManager() = default;
private:
std::mutex _mtx;
std::unordered_map<std::string, std::weak_ptr<SrtTransport>> _map;
std::mutex _handshake_mtx;
std::unordered_map<std::string, std::weak_ptr<SrtTransport>> _handshake_map;
};
} // namespace SRT
#endif // ZLMEDIAKIT_SRT_TRANSPORT_H

282
srt/SrtTransportImp.cpp Normal file
View File

@ -0,0 +1,282 @@
#include <memory>
#include "Util/util.h"
#include "SrtTransportImp.hpp"
namespace SRT {
SrtTransportImp::SrtTransportImp(const EventPoller::Ptr &poller) : SrtTransport(poller) {}
SrtTransportImp::~SrtTransportImp() {
InfoP(this);
uint64_t duration = _alive_ticker.createdTime() / 1000;
WarnP(this) << (_is_pusher ? "srt 推流器(" : "srt 播放器(") << _media_info._vhost << "/" << _media_info._app << "/"
<< _media_info._streamid << ")断开,耗时(s):" << duration;
//流量统计事件广播
GET_CONFIG(uint32_t, iFlowThreshold, General::kFlowThreshold);
if (_total_bytes >= iFlowThreshold * 1024) {
NoticeCenter::Instance().emitEvent(
Broadcast::kBroadcastFlowReport, _media_info, _total_bytes, duration, false,
static_cast<SockInfo &>(*this));
}
}
void SrtTransportImp::onHandShakeFinished(std::string &streamid, struct sockaddr_storage *addr) {
// TODO parse stream id like this zlmediakit.com/live/test?token=1213444&type=push
if (!_addr) {
_addr.reset(new sockaddr_storage(*((sockaddr_storage *)addr)));
}
_is_pusher = false;
TraceL << " stream id " << streamid;
if (streamid.empty()) {
onShutdown(SockException(Err_shutdown, "stream id not empty"));
return;
}
_media_info.parse("srt://" + streamid);
auto params = Parser::parseArgs(_media_info._param_strs);
if (params["type"] == "push") {
_is_pusher = true;
_decoder = DecoderImp::createDecoder(DecoderImp::decoder_ts, this);
emitOnPublish();
} else {
_is_pusher = false;
emitOnPlay();
}
}
void SrtTransportImp::onSRTData(DataPacket::Ptr pkt) {
if (!_is_pusher) {
WarnP(this) << "this is a player data ignore";
return;
}
if (_decoder) {
_decoder->input(reinterpret_cast<const uint8_t *>(pkt->payloadData()), pkt->payloadSize());
} else {
WarnP(this) << " not reach this";
}
}
void SrtTransportImp::onShutdown(const SockException &ex) {
SrtTransport::onShutdown(ex);
}
bool SrtTransportImp::close(mediakit::MediaSource &sender, bool force) {
if (!force && totalReaderCount(sender)) {
return false;
}
std::string err = StrPrinter << "close media:" << sender.getSchema() << "/"
<< sender.getVhost() << "/"
<< sender.getApp() << "/"
<< sender.getId() << " " << force;
weak_ptr<SrtTransportImp> weak_self = static_pointer_cast<SrtTransportImp>(shared_from_this());
getPoller()->async([weak_self, err]() {
auto strong_self = weak_self.lock();
if (strong_self) {
strong_self->onShutdown(SockException(Err_shutdown, err));
//主动关闭推流,那么不延时注销
strong_self->_muxer = nullptr;
}
});
return true;
}
// 播放总人数
int SrtTransportImp::totalReaderCount(mediakit::MediaSource &sender) {
return _muxer ? _muxer->totalReaderCount() : sender.readerCount();
}
// 获取媒体源类型
mediakit::MediaOriginType SrtTransportImp::getOriginType(mediakit::MediaSource &sender) const {
return MediaOriginType::srt_push;
}
// 获取媒体源url或者文件路径
std::string SrtTransportImp::getOriginUrl(mediakit::MediaSource &sender) const {
return _media_info._full_url;
}
// 获取媒体源客户端相关信息
std::shared_ptr<SockInfo> SrtTransportImp::getOriginSock(mediakit::MediaSource &sender) const {
return static_pointer_cast<SockInfo>(getSession());
}
void SrtTransportImp::emitOnPublish() {
std::weak_ptr<SrtTransportImp> weak_self = static_pointer_cast<SrtTransportImp>(shared_from_this());
Broadcast::PublishAuthInvoker invoker = [weak_self](const std::string &err, const ProtocolOption &option) {
auto strong_self = weak_self.lock();
if (!strong_self) {
return;
}
if (err.empty()) {
strong_self->_muxer = std::make_shared<MultiMediaSourceMuxer>(
strong_self->_media_info._vhost, strong_self->_media_info._app, strong_self->_media_info._streamid,
0.0f, option);
strong_self->_muxer->setMediaListener(strong_self);
strong_self->doCachedFunc();
InfoP(strong_self) << "允许 srt 推流";
} else {
WarnP(strong_self) << "禁止 srt 推流:" << err;
strong_self->onShutdown(SockException(Err_refused, err));
}
};
//触发推流鉴权事件
auto flag = NoticeCenter::Instance().emitEvent(
Broadcast::kBroadcastMediaPublish, MediaOriginType::srt_push, _media_info, invoker,
static_cast<SockInfo &>(*this));
if (!flag) {
//该事件无人监听,默认不鉴权
invoker("", ProtocolOption());
}
}
void SrtTransportImp::emitOnPlay() {
std::weak_ptr<SrtTransportImp> weak_self = static_pointer_cast<SrtTransportImp>(shared_from_this());
Broadcast::AuthInvoker invoker = [weak_self](const string &err) {
auto strong_self = weak_self.lock();
if (!strong_self) {
return;
}
strong_self->getPoller()->async([strong_self, err] {
if (err != "") {
strong_self->onShutdown(SockException(Err_refused, err));
} else {
strong_self->doPlay();
}
});
};
auto flag = NoticeCenter::Instance().emitEvent(
Broadcast::kBroadcastMediaPlayed, _media_info, invoker, static_cast<SockInfo &>(*this));
if (!flag) {
doPlay();
}
}
void SrtTransportImp::doPlay() {
//异步查找直播流
MediaInfo info = _media_info;
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) {
auto strong_self = weak_self.lock();
if (!strong_self) {
//本对象已经销毁
TraceL << "本对象已经销毁";
return;
}
if (!src) {
//未找到该流
TraceL << "未找到该流";
strong_self->onShutdown(SockException(Err_shutdown));
} else {
TraceL << "找到该流";
auto ts_src = dynamic_pointer_cast<TSMediaSource>(src);
assert(ts_src);
ts_src->pause(false);
strong_self->_ts_reader = ts_src->getRing()->attach(strong_self->getPoller());
strong_self->_ts_reader->setDetachCB([weak_self]() {
auto strong_self = weak_self.lock();
if (!strong_self) {
//本对象已经销毁
return;
}
strong_self->onShutdown(SockException(Err_shutdown));
});
strong_self->_ts_reader->setReadCB([weak_self](const TSMediaSource::RingDataType &ts_list) {
auto strong_self = weak_self.lock();
if (!strong_self) {
//本对象已经销毁
return;
}
size_t i = 0;
auto size = ts_list->size();
ts_list->for_each([&](const TSPacket::Ptr &ts) { strong_self->onSendTSData(ts, ++i == size); });
});
}
});
}
std::string SrtTransportImp::get_peer_ip() {
if (!_addr) {
return "::";
}
return SockUtil::inet_ntoa((sockaddr *)_addr.get());
}
uint16_t SrtTransportImp::get_peer_port() {
if (!_addr) {
return 0;
}
return SockUtil::inet_port((sockaddr *)_addr.get());
}
std::string SrtTransportImp::get_local_ip() {
auto s = getSession();
if (s) {
return s->get_local_ip();
}
return "::";
}
uint16_t SrtTransportImp::get_local_port() {
auto s = getSession();
if (s) {
return s->get_local_port();
}
return 0;
}
std::string SrtTransportImp::getIdentifier() const {
return _media_info._streamid;
}
bool SrtTransportImp::inputFrame(const Frame::Ptr &frame) {
if (_muxer) {
return _muxer->inputFrame(frame);
}
if (_cached_func.size() > 200) {
WarnL << "cached frame of track(" << frame->getCodecName() << ") is too much, now dropped";
return false;
}
auto frame_cached = Frame::getCacheAbleFrame(frame);
lock_guard<recursive_mutex> lck(_func_mtx);
_cached_func.emplace_back([this, frame_cached]() { _muxer->inputFrame(frame_cached); });
return true;
}
bool SrtTransportImp::addTrack(const Track::Ptr &track) {
if (_muxer) {
return _muxer->addTrack(track);
}
lock_guard<recursive_mutex> lck(_func_mtx);
_cached_func.emplace_back([this, track]() { _muxer->addTrack(track); });
return true;
}
void SrtTransportImp::addTrackCompleted() {
if (_muxer) {
_muxer->addTrackCompleted();
} else {
lock_guard<recursive_mutex> lck(_func_mtx);
_cached_func.emplace_back([this]() { _muxer->addTrackCompleted(); });
}
}
void SrtTransportImp::doCachedFunc() {
lock_guard<recursive_mutex> lck(_func_mtx);
for (auto &func : _cached_func) {
func();
}
_cached_func.clear();
}
int SrtTransportImp::getLatencyMul() {
GET_CONFIG(int, latencyMul, kLatencyMul);
return latencyMul;
}
} // namespace SRT

92
srt/SrtTransportImp.hpp Normal file
View File

@ -0,0 +1,92 @@
#ifndef ZLMEDIAKIT_SRT_TRANSPORT_IMP_H
#define ZLMEDIAKIT_SRT_TRANSPORT_IMP_H
#include <mutex>
#include "Rtp/Decoder.h"
#include "SrtTransport.hpp"
#include "TS/TSMediaSource.h"
#include "Common/MultiMediaSourceMuxer.h"
namespace SRT {
using namespace std;
using namespace toolkit;
using namespace mediakit;
class SrtTransportImp
: public SrtTransport
, public toolkit::SockInfo
, public MediaSinkInterface
, public mediakit::MediaSourceEvent {
public:
SrtTransportImp(const EventPoller::Ptr &poller);
~SrtTransportImp();
void inputSockData(uint8_t *buf, int len, struct sockaddr_storage *addr) override {
SrtTransport::inputSockData(buf, len, addr);
_total_bytes += len;
}
void onSendTSData(const Buffer::Ptr &buffer, bool flush) override { SrtTransport::onSendTSData(buffer, flush); }
/// SockInfo override
std::string get_local_ip() override;
uint16_t get_local_port() override;
std::string get_peer_ip() override;
uint16_t get_peer_port() override;
std::string getIdentifier() const override;
protected:
///////SrtTransport override///////
int getLatencyMul() override;
void onSRTData(DataPacket::Ptr pkt) override;
void onShutdown(const SockException &ex) override;
void onHandShakeFinished(std::string &streamid, struct sockaddr_storage *addr) override;
void sendPacket(Buffer::Ptr pkt, bool flush = true) override {
_total_bytes += pkt->size();
SrtTransport::sendPacket(pkt, flush);
}
bool isPusher() override { return _is_pusher; }
///////MediaSourceEvent override///////
// 关闭
bool close(mediakit::MediaSource &sender, bool force) override;
// 播放总人数
int totalReaderCount(mediakit::MediaSource &sender) override;
// 获取媒体源类型
mediakit::MediaOriginType getOriginType(mediakit::MediaSource &sender) const override;
// 获取媒体源url或者文件路径
std::string getOriginUrl(mediakit::MediaSource &sender) const override;
// 获取媒体源客户端相关信息
std::shared_ptr<SockInfo> getOriginSock(mediakit::MediaSource &sender) const override;
///////MediaSinkInterface override///////
void resetTracks() override {};
void addTrackCompleted() override;
bool addTrack(const Track::Ptr &track) override;
bool inputFrame(const Frame::Ptr &frame) override;
private:
void emitOnPublish();
void emitOnPlay();
void doPlay();
void doCachedFunc();
private:
bool _is_pusher = true;
MediaInfo _media_info;
uint64_t _total_bytes = 0;
Ticker _alive_ticker;
std::unique_ptr<sockaddr_storage> _addr;
// for player
TSMediaSource::RingType::RingReader::Ptr _ts_reader;
// for pusher
MultiMediaSourceMuxer::Ptr _muxer;
DecoderImp::Ptr _decoder;
std::recursive_mutex _func_mtx;
std::deque<std::function<void()>> _cached_func;
};
} // namespace SRT
#endif // ZLMEDIAKIT_SRT_TRANSPORT_IMP_H

98
srt/Statistic.cpp Normal file
View File

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

49
srt/Statistic.hpp Normal file
View File

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

25
srt/srt.md Normal file
View File

@ -0,0 +1,25 @@
## 特性
- NACK(重传)
- listener 支持
- 推流只支持ts推流
- 拉流只支持ts拉流
- 协议实现 [参考](https://haivision.github.io/srt-rfc/draft-sharabayko-srt.html)
- 版本支持(>=1.3.0)
- fec与加密没有实现
## 使用
zlm中的srt更加streamid 来确定是推流还是拉流来确定vhost,app,streamid(ZLM中的)
srt中的streamid 为 `<vhost>/<app>/<streamid>?type=<push|play>& <other>=<other>`
- OBS 推流地址
`srt://192.168.1.105:9000?streamid=__defaultVhost__/live/test?type=push`
- ffmpeg 推流
`ffmpeg -re -stream_loop -1 -i test.ts -c:v copy -c:a copy -f mpegts srt://192.168.1.105:9000?streamid="__defaultVhost__/live/test?type=push"`
- ffplay 拉流
`ffplay -i srt://192.168.1.105:9000?streamid=__defaultVhost__/live/test`
- vlc 不支持,因为无法指定streamid[参考](https://github.com/Haivision/srt/issues/1015)

24
srt/srt_en.md Normal file
View File

@ -0,0 +1,24 @@
## feature
- NACK support
- listener support
- push stream payload must ts
- pull stream payload is ts
- protocol impliment [reference](https://haivision.github.io/srt-rfc/draft-sharabayko-srt.html)
- version support (>=1.3.0)
- fec and encriyped not support
## usage
zlm get vhost,app,streamid and push or play by streamid of srt like this `<vhost>/<app>/<streamid>?type=<push|play>& <other>=<other>`
- OBS push stream url
`srt://192.168.1.105:9000?streamid=__defaultVhost__/live/test?type=push`
- ffmpeg push
`ffmpeg -re -stream_loop -1 -i test.ts -c:v copy -c:a copy -f mpegts srt://192.168.1.105:9000?streamid="__defaultVhost__/live/test?type=push"`
- ffplay pull
`ffplay -i srt://192.168.1.105:9000?streamid=__defaultVhost__/live/test`
- vlc not support ,because can't set stream id [reference](https://github.com/Haivision/srt/issues/1015)