Mirror/ZLMediaKit
1
0
Fork 0
mirror of https://github.com/ZLMediaKit/ZLMediaKit.git synced 2026-06-22 00:42:22 +08:00
Code Issues Packages Projects Releases Wiki Activity

merge upstream

Browse Source
This commit is contained in:
monktan
2020-10-25 17:48:37 +08:00
parent e95ed256ea 5c7a08eb7c
commit a1fe5ee817
45 changed files with 743 additions and 404 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
3rdpart/media-server

Submodule 3rdpart/media-server updated: 66033ee475...698840bf5e

1
README.md
View File

@@ -85,6 +85,7 @@
- 支持UDP/TCP国标RTP(PS或TS)推流服务器，可以转换成RTSP/RTMP/HLS等协议
- 支持RTSP/RTMP/HLS转国标推流客户端，支持TCP/UDP模式，提供相应restful api
- 支持H264/H265/AAC/G711/OPUS编码
- 支持海康ehome推流
- MP4点播与录制
- 支持录制为FLV/HLS/MP4

14
server/FFmpegSource.cpp
View File

@@ -233,7 +233,7 @@ void FFmpegSource::setOnClose(const function<void()> &cb){
}
bool FFmpegSource::close(MediaSource &sender, bool force) {
auto listener = _listener.lock();
auto listener = getDelegate();
if(listener && !listener->close(sender,force)){
//关闭失败
return false;
@@ -258,17 +258,11 @@ std::shared_ptr<SockInfo> FFmpegSource::getOriginSock(MediaSource &sender) const
}
void FFmpegSource::onGetMediaSource(const MediaSource::Ptr &src) {
auto listener = src->getListener();
auto listener = src->getListener(true);
if (listener.lock().get() != this) {
//防止多次进入onGetMediaSource函数导致无效递归调用的bug
_listener = listener;
//防止多次进入onGetMediaSource函数导致无限递归调用的bug
setDelegate(listener);
src->setListener(shared_from_this());
} else {
WarnL << "多次触发onGetMediaSource事件:"
<< src->getSchema() << "/"
<< src->getVhost() << "/"
<< src->getApp() << "/"
<< src->getId();
}
}

9
server/WebApi.cpp
View File

@@ -504,12 +504,12 @@ void installWebApi() {
allArgs["vhost"],
allArgs["app"],
allArgs["stream"]);
if(src){
if (src) {
bool flag = src->close(allArgs["force"].as<bool>());
val["result"] = flag ? 0 : -1;
val["msg"] = flag ? "success" : "close failed";
val["code"] = API::OtherFailed;
}else{
val["code"] = flag ? API::Success : API::OtherFailed;
} else {
val["result"] = -2;
val["msg"] = "can not find the stream";
val["code"] = API::OtherFailed;
@@ -1309,6 +1309,8 @@ void installWebApi() {
}
void unInstallWebApi(){
RtpSelector::Instance().clear();
{
lock_guard<recursive_mutex> lck(s_proxyMapMtx);
s_proxyMap.clear();
@@ -1318,6 +1320,7 @@ void unInstallWebApi(){
lock_guard<recursive_mutex> lck(s_ffmpegMapMtx);
s_ffmpegMap.clear();
}
{
#if defined(ENABLE_RTPPROXY)
lock_guard<recursive_mutex> lck(s_rtpServerMapMtx);

54
src/Common/MediaSource.cpp
View File

@@ -20,16 +20,16 @@ recursive_mutex s_media_source_mtx;
MediaSource::SchemaVhostAppStreamMap s_media_source_map;
string getOriginTypeString(MediaOriginType type){
#define SWITCH_CASE(type) case type : return #type
#define SWITCH_CASE(type) case MediaOriginType::type : return #type
switch (type) {
SWITCH_CASE(MediaOriginType::unknown);
SWITCH_CASE(MediaOriginType::rtmp_push);
SWITCH_CASE(MediaOriginType::rtsp_push);
SWITCH_CASE(MediaOriginType::rtp_push);
SWITCH_CASE(MediaOriginType::pull);
SWITCH_CASE(MediaOriginType::ffmpeg_pull);
SWITCH_CASE(MediaOriginType::mp4_vod);
SWITCH_CASE(MediaOriginType::device_chn);
SWITCH_CASE(unknown);
SWITCH_CASE(rtmp_push);
SWITCH_CASE(rtsp_push);
SWITCH_CASE(rtp_push);
SWITCH_CASE(pull);
SWITCH_CASE(ffmpeg_pull);
SWITCH_CASE(mp4_vod);
SWITCH_CASE(device_chn);
}
}
@@ -92,8 +92,19 @@ void MediaSource::setListener(const std::weak_ptr<MediaSourceEvent> &listener){
_listener = listener;
}
const std::weak_ptr<MediaSourceEvent>& MediaSource::getListener() const{
return _listener;
std::weak_ptr<MediaSourceEvent> MediaSource::getListener(bool next) const{
if (!next) {
return _listener;
}
auto listener = dynamic_pointer_cast<MediaSourceEventInterceptor>(_listener.lock());
if (!listener) {
//不是MediaSourceEventInterceptor对象或者对象已经销毁
return _listener;
}
//获取被拦截的对象
auto next_obj = listener->getDelegate();
//有则返回之
return next_obj ? next_obj : _listener;
}
int MediaSource::totalReaderCount(){
@@ -168,7 +179,7 @@ bool MediaSource::isRecording(Recorder::type type){
return listener->isRecording(*this, type);
}
void MediaSource::startSendRtp(const string &dst_url, uint16_t dst_port, uint32_t ssrc, bool is_udp, const function<void(const SockException &ex)> &cb){
void MediaSource::startSendRtp(const string &dst_url, uint16_t dst_port, const string &ssrc, bool is_udp, const function<void(const SockException &ex)> &cb){
auto listener = _listener.lock();
if (!listener) {
cb(SockException(Err_other, "尚未设置事件监听器"));
@@ -627,7 +638,7 @@ vector<Track::Ptr> MediaSourceEventInterceptor::getTracks(MediaSource &sender, b
return listener->getTracks(sender, trackReady);
}
void MediaSourceEventInterceptor::startSendRtp(MediaSource &sender, const string &dst_url, uint16_t dst_port, uint32_t ssrc, bool is_udp, const function<void(const SockException &ex)> &cb){
void MediaSourceEventInterceptor::startSendRtp(MediaSource &sender, const string &dst_url, uint16_t dst_port, const string &ssrc, bool is_udp, const function<void(const SockException &ex)> &cb){
auto listener = _listener.lock();
if (listener) {
listener->startSendRtp(sender, dst_url, dst_port, ssrc, is_udp, cb);
@@ -644,9 +655,20 @@ bool MediaSourceEventInterceptor::stopSendRtp(MediaSource &sender){
return false;
}
void MediaSourceEventInterceptor::setDelegate(const std::weak_ptr<MediaSourceEvent> &listener) {
if (listener.lock().get() == this) {
throw std::invalid_argument("can not set self as a delegate");
}
_listener = listener;
}
std::shared_ptr<MediaSourceEvent> MediaSourceEventInterceptor::getDelegate() const{
return _listener.lock();
}
/////////////////////////////////////FlushPolicy//////////////////////////////////////
static bool isFlushAble_default(bool is_video, uint32_t last_stamp, uint32_t new_stamp, int cache_size) {
static bool isFlushAble_default(bool is_video, uint64_t last_stamp, uint64_t new_stamp, int cache_size) {
if (new_stamp + 500 < last_stamp) {
//时间戳回退比较大(可能seek中)，由于rtp中时间戳是pts，是可能存在一定程度的回退的
return true;
@@ -656,7 +678,7 @@ static bool isFlushAble_default(bool is_video, uint32_t last_stamp, uint32_t new
return last_stamp != new_stamp || cache_size >= 1024;
}
static bool isFlushAble_merge(bool is_video, uint32_t last_stamp, uint32_t new_stamp, int cache_size, int merge_ms) {
static bool isFlushAble_merge(bool is_video, uint64_t last_stamp, uint64_t new_stamp, int cache_size, int merge_ms) {
if (new_stamp + 500 < last_stamp) {
//时间戳回退比较大(可能seek中)，由于rtp中时间戳是pts，是可能存在一定程度的回退的
return true;
@@ -672,7 +694,7 @@ static bool isFlushAble_merge(bool is_video, uint32_t last_stamp, uint32_t new_s
return cache_size >= 1024;
}
bool FlushPolicy::isFlushAble(bool is_video, bool is_key, uint32_t new_stamp, int cache_size) {
bool FlushPolicy::isFlushAble(bool is_video, bool is_key, uint64_t new_stamp, int cache_size) {
bool flush_flag = false;
if (is_key && is_video) {
//遇到关键帧flush掉前面的数据，确保关键帧为该组数据的第一帧，确保GOP缓存有效

31
src/Common/MediaSource.h
View File

@@ -83,7 +83,7 @@ public:
// 获取所有track相关信息
virtual vector<Track::Ptr> getTracks(MediaSource &sender, bool trackReady = true) const { return vector<Track::Ptr>(); };
// 开始发送ps-rtp
virtual void startSendRtp(MediaSource &sender, const string &dst_url, uint16_t dst_port, uint32_t ssrc, bool is_udp, const function<void(const SockException &ex)> &cb) { cb(SockException(Err_other, "not implemented"));};
virtual void startSendRtp(MediaSource &sender, const string &dst_url, uint16_t dst_port, const string &ssrc, bool is_udp, const function<void(const SockException &ex)> &cb) { cb(SockException(Err_other, "not implemented"));};
// 停止发送ps-rtp
virtual bool stopSendRtp(MediaSource &sender) {return false; }
@@ -97,6 +97,9 @@ public:
MediaSourceEventInterceptor(){}
~MediaSourceEventInterceptor() override {}
void setDelegate(const std::weak_ptr<MediaSourceEvent> &listener);
std::shared_ptr<MediaSourceEvent> getDelegate() const;
MediaOriginType getOriginType(MediaSource &sender) const override;
string getOriginUrl(MediaSource &sender) const override;
std::shared_ptr<SockInfo> getOriginSock(MediaSource &sender) const override;
@@ -109,10 +112,10 @@ public:
bool setupRecord(MediaSource &sender, Recorder::type type, bool start, const string &custom_path) override;
bool isRecording(MediaSource &sender, Recorder::type type) override;
vector<Track::Ptr> getTracks(MediaSource &sender, bool trackReady = true) const override;
void startSendRtp(MediaSource &sender, const string &dst_url, uint16_t dst_port, uint32_t ssrc, bool is_udp, const function<void(const SockException &ex)> &cb) override;
void startSendRtp(MediaSource &sender, const string &dst_url, uint16_t dst_port, const string &ssrc, bool is_udp, const function<void(const SockException &ex)> &cb) override;
bool stopSendRtp(MediaSource &sender) override;
protected:
private:
std::weak_ptr<MediaSourceEvent> _listener;
};
@@ -226,9 +229,9 @@ public:
////////////////MediaSourceEvent相关接口实现////////////////
// 设置监听者
void setListener(const std::weak_ptr<MediaSourceEvent> &listener);
virtual void setListener(const std::weak_ptr<MediaSourceEvent> &listener);
// 获取监听者
const std::weak_ptr<MediaSourceEvent>& getListener() const;
std::weak_ptr<MediaSourceEvent> getListener(bool next = false) const;
// 本协议获取观看者个数，可能返回本协议的观看人数，也可能返回总人数
virtual int readerCount() = 0;
@@ -253,7 +256,7 @@ public:
// 获取录制状态
bool isRecording(Recorder::type type);
// 开始发送ps-rtp
void startSendRtp(const string &dst_url, uint16_t dst_port, uint32_t ssrc, bool is_udp, const function<void(const SockException &ex)> &cb);
void startSendRtp(const string &dst_url, uint16_t dst_port, const string &ssrc, bool is_udp, const function<void(const SockException &ex)> &cb);
// 停止发送ps-rtp
bool stopSendRtp();
@@ -301,18 +304,10 @@ public:
FlushPolicy() = default;
~FlushPolicy() = default;
uint32_t getStamp(const RtpPacket::Ptr &packet) {
return packet->timeStamp;
}
uint32_t getStamp(const RtmpPacket::Ptr &packet) {
return packet->time_stamp;
}
bool isFlushAble(bool is_video, bool is_key, uint32_t new_stamp, int cache_size);
bool isFlushAble(bool is_video, bool is_key, uint64_t new_stamp, int cache_size);
private:
uint32_t _last_stamp[2] = {0, 0};
uint64_t _last_stamp[2] = {0, 0};
};
/// 合并写缓存模板
@@ -328,8 +323,8 @@ public:
virtual ~PacketCache() = default;
void inputPacket(bool is_video, std::shared_ptr<packet> pkt, bool key_pos) {
if (_policy.isFlushAble(is_video, key_pos, _policy.getStamp(pkt), _cache->size())) {
void inputPacket(uint64_t stamp, bool is_video, std::shared_ptr<packet> pkt, bool key_pos) {
if (_policy.isFlushAble(is_video, key_pos, stamp, _cache->size())) {
flushAll();
}

28
src/Common/MultiMediaSourceMuxer.cpp
View File

@@ -323,7 +323,7 @@ MultiMediaSourceMuxer::MultiMediaSourceMuxer(const string &vhost, const string &
}
void MultiMediaSourceMuxer::setMediaListener(const std::weak_ptr<MediaSourceEvent> &listener) {
_listener = listener;
setDelegate(listener);
//拦截事件
_muxer->setMediaListener(shared_from_this());
}
@@ -345,7 +345,7 @@ vector<Track::Ptr> MultiMediaSourceMuxer::getTracks(MediaSource &sender, bool tr
}
int MultiMediaSourceMuxer::totalReaderCount(MediaSource &sender) {
auto listener = _listener.lock();
auto listener = getDelegate();
if (!listener) {
return totalReaderCount();
}
@@ -360,21 +360,21 @@ bool MultiMediaSourceMuxer::isRecording(MediaSource &sender, Recorder::type type
return _muxer->isRecording(sender,type);
}
void MultiMediaSourceMuxer::startSendRtp(MediaSource &sender, const string &dst_url, uint16_t dst_port, uint32_t ssrc, bool is_udp, const function<void(const SockException &ex)> &cb){
void MultiMediaSourceMuxer::startSendRtp(MediaSource &sender, const string &dst_url, uint16_t dst_port, const string &ssrc, bool is_udp, const function<void(const SockException &ex)> &cb){
#if defined(ENABLE_RTPPROXY)
auto ps_rtp_sender = std::make_shared<PSRtpSender>(ssrc);
RtpSender::Ptr rtp_sender = std::make_shared<RtpSender>(atoi(ssrc.data()));
weak_ptr<MultiMediaSourceMuxer> weak_self = shared_from_this();
ps_rtp_sender->startSend(dst_url, dst_port, is_udp, [weak_self, ps_rtp_sender, cb](const SockException &ex) {
rtp_sender->startSend(dst_url, dst_port, is_udp, [weak_self, rtp_sender, cb](const SockException &ex) {
cb(ex);
auto strong_self = weak_self.lock();
if (!strong_self || ex) {
return;
}
for (auto &track : strong_self->_muxer->getTracks(false)) {
ps_rtp_sender->addTrack(track);
rtp_sender->addTrack(track);
}
ps_rtp_sender->addTrackCompleted();
strong_self->_ps_rtp_sender = ps_rtp_sender;
rtp_sender->addTrackCompleted();
strong_self->_rtp_sender = rtp_sender;
});
#else
cb(SockException(Err_other, "该功能未启用，编译时请打开ENABLE_RTPPROXY宏"));
@@ -383,8 +383,8 @@ void MultiMediaSourceMuxer::startSendRtp(MediaSource &sender, const string &dst_
bool MultiMediaSourceMuxer::stopSendRtp(MediaSource &sender){
#if defined(ENABLE_RTPPROXY)
if (_ps_rtp_sender) {
_ps_rtp_sender = nullptr;
if (_rtp_sender) {
_rtp_sender = nullptr;
return true;
}
#endif//ENABLE_RTPPROXY
@@ -473,9 +473,9 @@ void MultiMediaSourceMuxer::inputFrame(const Frame::Ptr &frame_in) {
_muxer->inputFrame(frame);
#if defined(ENABLE_RTPPROXY)
auto ps_rtp_sender = _ps_rtp_sender;
if (ps_rtp_sender) {
ps_rtp_sender->inputFrame(frame);
auto rtp_sender = _rtp_sender;
if (rtp_sender) {
rtp_sender->inputFrame(frame);
}
#endif //ENABLE_RTPPROXY
@@ -483,7 +483,7 @@ void MultiMediaSourceMuxer::inputFrame(const Frame::Ptr &frame_in) {
bool MultiMediaSourceMuxer::isEnabled(){
#if defined(ENABLE_RTPPROXY)
return (_muxer->isEnabled() || _ps_rtp_sender);
return (_muxer->isEnabled() || _rtp_sender);
#else
return _muxer->isEnabled();
#endif //ENABLE_RTPPROXY

6
src/Common/MultiMediaSourceMuxer.h
View File

@@ -12,7 +12,7 @@
#define ZLMEDIAKIT_MULTIMEDIASOURCEMUXER_H
#include "Common/Stamp.h"
#include "Rtp/PSRtpSender.h"
#include "Rtp/RtpSender.h"
#include "Record/Recorder.h"
#include "Record/HlsRecorder.h"
#include "Record/HlsMediaSource.h"
@@ -143,7 +143,7 @@ public:
* @param is_udp 是否为udp
* @param cb 启动成功或失败回调
*/
void startSendRtp(MediaSource &sender, const string &dst_url, uint16_t dst_port, uint32_t ssrc, bool is_udp, const function<void(const SockException &ex)> &cb) override;
void startSendRtp(MediaSource &sender, const string &dst_url, uint16_t dst_port, const string &ssrc, bool is_udp, const function<void(const SockException &ex)> &cb) override;
/**
* 停止ps-rtp发送
@@ -188,7 +188,7 @@ private:
MultiMuxerPrivate::Ptr _muxer;
std::weak_ptr<MultiMuxerPrivate::Listener> _track_listener;
#if defined(ENABLE_RTPPROXY)
PSRtpSender::Ptr _ps_rtp_sender;
RtpSender::Ptr _rtp_sender;
#endif //ENABLE_RTPPROXY
};

19
src/Extension/H264Rtp.cpp
View File

@@ -195,13 +195,13 @@ H264RtpEncoder::H264RtpEncoder(uint32_t ui32Ssrc,
void H264RtpEncoder::inputFrame(const Frame::Ptr &frame) {
GET_CONFIG(uint32_t,cycleMS,Rtp::kCycleMS);
auto ptr = frame->data() + frame->prefixSize();
auto pts = frame->pts() % cycleMS;
auto len = frame->size() - frame->prefixSize();
auto pts = frame->pts() % cycleMS;
auto nal_type = H264_TYPE(ptr[0]);
auto max_rtp_size = _ui32MtuSize - 2;
auto payload_size = _ui32MtuSize - 2;
//超过MTU则按照FU-A模式打包
if (len > max_rtp_size) {
if (len > payload_size + 1) {
//最高位bit为forbidden_zero_bit,
//后面2bit为nal_ref_idc(帧重要程度),00:可以丢,11:不能丢
//末尾5bit为nalu type，固定为28(FU-A)
@@ -211,11 +211,10 @@ void H264RtpEncoder::inputFrame(const Frame::Ptr &frame) {
bool mark_bit = false;
int offset = 1;
while (!mark_bit) {
if (len <= offset + max_rtp_size) {
//已经拆分结束
max_rtp_size = len - offset;
mark_bit = true;
if (len <= offset + payload_size) {
//FU-A end
mark_bit = true;
payload_size = len - offset;
s_e_r_flags = (1 << 6) | nal_type;
} else if (fu_a_start) {
//FU-A start
@@ -227,7 +226,7 @@ void H264RtpEncoder::inputFrame(const Frame::Ptr &frame) {
{
//传入nullptr先不做payload的内存拷贝
auto rtp = makeRtp(getTrackType(), nullptr, max_rtp_size + 2, mark_bit, pts);
auto rtp = makeRtp(getTrackType(), nullptr, payload_size + 2, mark_bit, pts);
//rtp payload 负载部分
uint8_t *payload = (uint8_t*)rtp->data() + rtp->offset;
//FU-A 第1个字节
@@ -235,11 +234,11 @@ void H264RtpEncoder::inputFrame(const Frame::Ptr &frame) {
//FU-A 第2个字节
payload[1] = s_e_r_flags;
//H264 数据
memcpy(payload + 2, (unsigned char *) ptr + offset, max_rtp_size);
memcpy(payload + 2, (unsigned char *) ptr + offset, payload_size);
//输入到rtp环形缓存
RtpCodec::inputRtp(rtp, fu_a_start && nal_type == H264Frame::NAL_IDR);
}
offset += max_rtp_size;
offset += payload_size;
fu_a_start = false;
}
} else {

60
src/Extension/H265Rtp.cpp
View File

@@ -18,10 +18,10 @@ namespace mediakit{
//44 +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
//45 |F| Type | LayerId | TID |
//46 +-------------+-----------------+
//48 F = 0
//49 Type = 49 (fragmentation unit (FU))
//50 LayerId = 0
//51 TID = 1
//48 F = 0, 1bit
//49 Type = 49 (fragmentation unit (FU)), 6bit
//50 LayerId = 0, 6bit
//51 TID = 1, 3bit
//56 /*
//57 create the FU header
//58
@@ -29,7 +29,6 @@ namespace mediakit{
//60 +-+-+-+-+-+-+-+-+
//61 |S|E| FuType |
//62 +---------------+
//63
//64 S = variable
//65 E = variable
//66 FuType = NAL unit type
@@ -150,40 +149,39 @@ H265RtpEncoder::H265RtpEncoder(uint32_t ui32Ssrc,
}
void H265RtpEncoder::inputFrame(const Frame::Ptr &frame) {
GET_CONFIG(uint32_t,cycleMS,Rtp::kCycleMS);
uint8_t *pcData = (uint8_t*)frame->data() + frame->prefixSize();
auto uiStamp = frame->pts();
auto iLen = frame->size() - frame->prefixSize();
unsigned char naluType = H265_TYPE(pcData[0]); //获取NALU的5bit 帧类型
uiStamp %= cycleMS;
GET_CONFIG(uint32_t, cycleMS, Rtp::kCycleMS);
auto ptr = (uint8_t *) frame->data() + frame->prefixSize();
auto len = frame->size() - frame->prefixSize();
auto pts = frame->pts() % cycleMS;
auto nal_type = H265_TYPE(ptr[0]); //获取NALU的5bit 帧类型
auto payload_size = _ui32MtuSize - 3;
int maxSize = _ui32MtuSize - 3;
//超过MTU,按照FU方式打包
if (iLen > maxSize) {
if (len > payload_size + 2) {
//获取帧头数据，1byte
unsigned char s_e_flags;
bool bFirst = true;
bool mark = false;
int nOffset = 2;
while (!mark) {
if (iLen <= nOffset + maxSize) { //是否拆分结束
maxSize = iLen - nOffset;
mark = true;
bool fu_start = true;
bool mark_bit = false;
int offset = 2;
while (!mark_bit) {
if (len <= offset + payload_size) {
//FU end
s_e_flags = (1 << 6) | naluType;
} else if (bFirst) {
mark_bit = true;
payload_size = len - offset;
s_e_flags = (1 << 6) | nal_type;
} else if (fu_start) {
//FU start
s_e_flags = (1 << 7) | naluType;
s_e_flags = (1 << 7) | nal_type;
} else {
//FU mid
s_e_flags = naluType;
s_e_flags = nal_type;
}
{
//传入nullptr先不做payload的内存拷贝
auto rtp = makeRtp(getTrackType(), nullptr, maxSize + 3, mark, uiStamp);
auto rtp = makeRtp(getTrackType(), nullptr, payload_size + 3, mark_bit, pts);
//rtp payload 负载部分
uint8_t *payload = (uint8_t*)rtp->data() + rtp->offset;
uint8_t *payload = (uint8_t *) rtp->data() + rtp->offset;
//FU 第1个字节，表明为FU
payload[0] = 49 << 1;
//FU 第2个字节貌似固定为1
@@ -191,16 +189,16 @@ void H265RtpEncoder::inputFrame(const Frame::Ptr &frame) {
//FU 第3个字节
payload[2] = s_e_flags;
//H265 数据
memcpy(payload + 3,pcData + nOffset, maxSize);
memcpy(payload + 3, ptr + offset, payload_size);
//输入到rtp环形缓存
RtpCodec::inputRtp(rtp,bFirst && H265Frame::isKeyFrame(naluType));
RtpCodec::inputRtp(rtp, fu_start && H265Frame::isKeyFrame(nal_type));
}
nOffset += maxSize;
bFirst = false;
offset += payload_size;
fu_start = false;
}
} else {
makeH265Rtp(naluType,pcData, iLen, false, true, uiStamp);
makeH265Rtp(nal_type, ptr, len, false, true, pts);
}
}

18
src/FMP4/FMP4MediaSource.h
View File

@@ -30,19 +30,8 @@ public:
uint32_t time_stamp = 0;
};
//FMP4直播合并写策略类
class FMP4FlushPolicy : public FlushPolicy{
public:
FMP4FlushPolicy() = default;
~FMP4FlushPolicy() = default;
uint32_t getStamp(const FMP4Packet::Ptr &packet) {
return packet->time_stamp;
}
};
//FMP4直播源
class FMP4MediaSource : public MediaSource, public RingDelegate<FMP4Packet::Ptr>, public PacketCache<FMP4Packet, FMP4FlushPolicy>{
class FMP4MediaSource : public MediaSource, public RingDelegate<FMP4Packet::Ptr>, public PacketCache<FMP4Packet>{
public:
using Ptr = std::shared_ptr<FMP4MediaSource>;
using RingDataType = std::shared_ptr<List<FMP4Packet::Ptr> >;
@@ -100,14 +89,15 @@ public:
_have_video = true;
}
_speed += packet->size();
PacketCache<FMP4Packet, FMP4FlushPolicy>::inputPacket(true, std::move(packet), key);
auto stamp = packet->time_stamp;
PacketCache<FMP4Packet>::inputPacket(stamp, true, std::move(packet), key);
}
/**
* 情况GOP缓存
*/
void clearCache() override {
PacketCache<FMP4Packet, FMP4FlushPolicy>::clearCache();
PacketCache<FMP4Packet>::clearCache();
_ring->clearCache();
}

2
src/FMP4/FMP4MediaSourceMuxer.h
View File

@@ -32,7 +32,7 @@ public:
~FMP4MediaSourceMuxer() override = default;
void setListener(const std::weak_ptr<MediaSourceEvent> &listener){
_listener = listener;
setDelegate(listener);
_media_src->setListener(shared_from_this());
}

2
src/Player/MediaPlayer.cpp
View File

@@ -37,7 +37,7 @@ void MediaPlayer::play(const string &url) {
_delegate->setOnShutdown(_shutdownCB);
_delegate->setOnPlayResult(_playResultCB);
_delegate->setOnResume(_resumeCB);
_delegate->setMediaSouce(_pMediaSrc);
_delegate->setMediaSource(_pMediaSrc);
_delegate->mINI::operator=(*this);
_delegate->play(url);
}

6
src/Player/PlayerBase.h
View File

@@ -103,7 +103,7 @@ public:
* 设置一个MediaSource，直接生产rtsp/rtmp代理
* @param src
*/
virtual void setMediaSouce(const MediaSource::Ptr & src) {}
virtual void setMediaSource(const MediaSource::Ptr & src) {}
/**
* 获取丢包率，只支持rtsp
@@ -181,9 +181,9 @@ public:
return Parent::seekTo(fProgress);
}
void setMediaSouce(const MediaSource::Ptr & src) override {
void setMediaSource(const MediaSource::Ptr & src) override {
if (_delegate) {
_delegate->setMediaSouce(src);
_delegate->setMediaSource(src);
}
_pMediaSrc = src;
}

8
src/Player/PlayerProxy.cpp
View File

@@ -93,6 +93,10 @@ void PlayerProxy::play(const string &strUrlTmp) {
if(!strongSelf) {
return;
}
//注销直接拉流代理产生的流：#532
strongSelf->setMediaSource(nullptr);
if(strongSelf->_muxer) {
auto tracks = strongSelf->MediaPlayer::getTracks(false);
for (auto & track : tracks){
@@ -126,7 +130,7 @@ void PlayerProxy::play(const string &strUrlTmp) {
mediaSource = std::make_shared<RtmpMediaSource>(_vhost, _app, _stream_id);
}
if(mediaSource){
setMediaSouce(mediaSource);
setMediaSource(mediaSource);
mediaSource->setListener(shared_from_this());
}
}
@@ -163,7 +167,7 @@ bool PlayerProxy::close(MediaSource &sender,bool force) {
return;
}
strongSelf->_muxer.reset();
strongSelf->setMediaSouce(nullptr);
strongSelf->setMediaSource(nullptr);
strongSelf->teardown();
if (strongSelf->_on_close) {
strongSelf->_on_close();

2
src/Record/HlsRecorder.h
View File

@@ -34,7 +34,7 @@ public:
}
void setListener(const std::weak_ptr<MediaSourceEvent> &listener) {
_listener = listener;
setDelegate(listener);
_hls->getMediaSource()->setListener(shared_from_this());
//先注册媒体流，后续可以按需生成
_hls->getMediaSource()->registHls(false);

3
src/Rtmp/RtmpMediaSource.h
View File

@@ -154,7 +154,8 @@ public:
}
bool key = pkt->isVideoKeyFrame();
bool is_video = pkt->type_id == MSG_VIDEO;
PacketCache<RtmpPacket>::inputPacket(is_video, std::move(pkt), key);
auto stamp = pkt->time_stamp;
PacketCache<RtmpPacket>::inputPacket(stamp, is_video, std::move(pkt), key);
}
/**

16
src/Rtmp/RtmpMediaSourceImp.h
View File

@@ -86,7 +86,7 @@ public:
_muxer->setMediaListener(getListener());
_muxer->setTrackListener(static_pointer_cast<RtmpMediaSourceImp>(shared_from_this()));
//让_muxer对象拦截一部分事件(比如说录像相关事件)
setListener(_muxer);
MediaSource::setListener(_muxer);
for(auto &track : _demuxer->getTracks(false)){
_muxer->addTrack(track);
@@ -119,6 +119,20 @@ public:
}
}
/**
* 设置事件监听器
* @param listener 监听器
*/
void setListener(const std::weak_ptr<MediaSourceEvent> &listener) override{
if (_muxer) {
//_muxer对象不能处理的事件再给listener处理
_muxer->setMediaListener(listener);
} else {
//未创建_muxer对象，事件全部给listener处理
MediaSource::setListener(listener);
}
}
private:
bool _all_track_ready = false;
bool _recreate_metadata = false;

2
src/Rtmp/RtmpMediaSourceMuxer.h
View File

@@ -32,7 +32,7 @@ public:
~RtmpMediaSourceMuxer() override{}
void setListener(const std::weak_ptr<MediaSourceEvent> &listener){
_listener = listener;
setDelegate(listener);
_media_src->setListener(shared_from_this());
}

116
src/Rtp/GB28181Process.cpp Normal file
View File

@@ -0,0 +1,116 @@
/*
* Copyright (c) 2016 The ZLMediaKit project authors. All Rights Reserved.
*
* This file is part of ZLMediaKit(https://github.com/xiongziliang/ZLMediaKit).
*
* Use of this source code is governed by MIT license that can be found in the
* LICENSE file in the root of the source tree. All contributing project authors
* may be found in the AUTHORS file in the root of the source tree.
*/
#if defined(ENABLE_RTPPROXY)
#include "GB28181Process.h"
#include "Util/File.h"
#include "Http/HttpTSPlayer.h"
#include "Extension/CommonRtp.h"
namespace mediakit{
//判断是否为ts负载
static inline bool checkTS(const uint8_t *packet, int bytes){
return bytes % TS_PACKET_SIZE == 0 && packet[0] == TS_SYNC_BYTE;
}
GB28181Process::GB28181Process(const MediaInfo &media_info, MediaSinkInterface *interface) {
assert(interface);
_media_info = media_info;
_interface = interface;
}
GB28181Process::~GB28181Process() {}
bool GB28181Process::inputRtp(bool, const char *data, int data_len) {
return handleOneRtp(0, TrackVideo, 90000, (unsigned char *) data, data_len);
}
void GB28181Process::onRtpSorted(const RtpPacket::Ptr &rtp, int) {
if (!_rtp_decoder) {
switch (rtp->PT) {
case 33:
case 96: {
//ts或ps负载
_rtp_decoder = std::make_shared<CommonRtpDecoder>(CodecInvalid, 256 * 1024);
//设置dump目录
GET_CONFIG(string, dump_dir, RtpProxy::kDumpDir);
if (!dump_dir.empty()) {
auto save_path = File::absolutePath(_media_info._streamid + ".mp2", dump_dir);
_save_file_ps.reset(File::create_file(save_path.data(), "wb"), [](FILE *fp) {
if (fp) {
fclose(fp);
}
});
}
break;
}
default:
WarnL << "不支持的rtp负载类型:" << (int) rtp->PT;
return;
}
//设置frame回调
_rtp_decoder->addDelegate(std::make_shared<FrameWriterInterfaceHelper>([this](const Frame::Ptr &frame) {
onRtpDecode(frame);
}));
}
//解码rtp
_rtp_decoder->inputRtp(rtp, false);
}
const char *GB28181Process::onSearchPacketTail(const char *packet,int bytes){
try {
auto ret = _decoder->input((uint8_t *) packet, bytes);
if (ret > 0) {
return packet + ret;
}
return nullptr;
} catch (std::exception &ex) {
InfoL << "解析ps或ts异常: bytes=" << bytes
<< " ,exception=" << ex.what()
<< " ,hex=" << hexdump((uint8_t *) packet, bytes);
if (remainDataSize() > 256 * 1024) {
//缓存太多数据无法处理则上抛异常
throw;
}
return nullptr;
}
}
void GB28181Process::onRtpDecode(const Frame::Ptr &frame) {
//这是TS或PS
if (_save_file_ps) {
fwrite(frame->data(), frame->size(), 1, _save_file_ps.get());
}
if (!_decoder) {
//创建解码器
if (checkTS((uint8_t *) frame->data(), frame->size())) {
//猜测是ts负载
InfoL << _media_info._streamid << " judged to be TS";
_decoder = DecoderImp::createDecoder(DecoderImp::decoder_ts, _interface);
} else {
//猜测是ps负载
InfoL << _media_info._streamid << " judged to be PS";
_decoder = DecoderImp::createDecoder(DecoderImp::decoder_ps, _interface);
}
}
if (_decoder) {
HttpRequestSplitter::input(frame->data(), frame->size());
}
}
}//namespace mediakit
#endif//defined(ENABLE_RTPPROXY)

56
src/Rtp/GB28181Process.h Normal file
View File

@@ -0,0 +1,56 @@
/*
* Copyright (c) 2016 The ZLMediaKit project authors. All Rights Reserved.
*
* This file is part of ZLMediaKit(https://github.com/xiongziliang/ZLMediaKit).
*
* Use of this source code is governed by MIT license that can be found in the
* LICENSE file in the root of the source tree. All contributing project authors
* may be found in the AUTHORS file in the root of the source tree.
*/
#ifndef ZLMEDIAKIT_GB28181ROCESS_H
#define ZLMEDIAKIT_GB28181ROCESS_H
#if defined(ENABLE_RTPPROXY)
#include "Decoder.h"
#include "ProcessInterface.h"
#include "Rtsp/RtpCodec.h"
#include "Rtsp/RtpReceiver.h"
#include "Http/HttpRequestSplitter.h"
namespace mediakit{
class GB28181Process : public HttpRequestSplitter, public RtpReceiver, public ProcessInterface{
public:
typedef std::shared_ptr<GB28181Process> Ptr;
GB28181Process(const MediaInfo &media_info, MediaSinkInterface *interface);
~GB28181Process() override;
/**
* 输入rtp
* @param data rtp数据指针
* @param data_len rtp数据长度
* @return 是否解析成功
*/
bool inputRtp(bool, const char *data, int data_len) override;
protected:
void onRtpSorted(const RtpPacket::Ptr &rtp, int track_index) override ;
const char *onSearchPacketTail(const char *data,int len) override;
int64_t onRecvHeader(const char *data,uint64_t len) override { return 0; };
private:
void onRtpDecode(const Frame::Ptr &frame);
private:
MediaInfo _media_info;
DecoderImp::Ptr _decoder;
MediaSinkInterface *_interface;
std::shared_ptr<FILE> _save_file_ps;
std::shared_ptr<RtpCodec> _rtp_decoder;
};
}//namespace mediakit
#endif//defined(ENABLE_RTPPROXY)
#endif //ZLMEDIAKIT_GB28181ROCESS_H

38
src/Rtp/PSEncoder.cpp
View File

@@ -167,5 +167,43 @@ void PSEncoder::inputFrame(const Frame::Ptr &frame) {
}
}
////////////////////////////////////////////////////////////////////////////////////////////////
class RingDelegateHelper : public RingDelegate<RtpPacket::Ptr> {
public:
typedef function<void(RtpPacket::Ptr in, bool is_key)> onRtp;
~RingDelegateHelper() override{}
RingDelegateHelper(onRtp on_rtp){
_on_rtp = std::move(on_rtp);
}
void onWrite(RtpPacket::Ptr in, bool is_key) override{
_on_rtp(std::move(in), is_key);
}
private:
onRtp _on_rtp;
};
////////////////////////////////////////////////////////////////////////////////////////////////////////////////
PSEncoderImp::PSEncoderImp(uint32_t ssrc, uint8_t payload_type) {
GET_CONFIG(uint32_t,video_mtu,Rtp::kVideoMtuSize);
_rtp_encoder = std::make_shared<CommonRtpEncoder>(CodecInvalid, ssrc, video_mtu, 90000, payload_type, 0);
_rtp_encoder->setRtpRing(std::make_shared<RtpRing::RingType>());
_rtp_encoder->getRtpRing()->setDelegate(std::make_shared<RingDelegateHelper>([this](RtpPacket::Ptr rtp, bool is_key){
onRTP(std::move(rtp));
}));
InfoL << this << " " << printSSRC(_rtp_encoder->getSsrc());
}
PSEncoderImp::~PSEncoderImp() {
InfoL << this << " " << printSSRC(_rtp_encoder->getSsrc());
}
void PSEncoderImp::onPS(uint32_t stamp, void *packet, size_t bytes) {
_rtp_encoder->inputFrame(std::make_shared<FrameFromPtr>((char *) packet, bytes, stamp));
}
}//namespace mediakit
#endif//defined(ENABLE_RTPPROXY)

17
src/Rtp/PSEncoder.h
View File

@@ -14,6 +14,7 @@
#include "mpeg-ps.h"
#include "Common/MediaSink.h"
#include "Common/Stamp.h"
#include "Extension/CommonRtp.h"
namespace mediakit{
//该类实现mpeg-ps容器格式的打包
@@ -65,6 +66,22 @@ private:
unordered_map<int, track_info> _codec_to_trackid;
};
class PSEncoderImp : public PSEncoder{
public:
PSEncoderImp(uint32_t ssrc, uint8_t payload_type = 96);
~PSEncoderImp() override;
protected:
//rtp打包后回调
virtual void onRTP(Buffer::Ptr rtp) = 0;
protected:
void onPS(uint32_t stamp, void *packet, size_t bytes) override;
private:
std::shared_ptr<CommonRtpEncoder> _rtp_encoder;
};
}//namespace mediakit
#endif //ENABLE_RTPPROXY
#endif //ZLMEDIAKIT_PSENCODER_H

37
src/Rtp/ProcessInterface.h Normal file
View File

@@ -0,0 +1,37 @@
/*
* Copyright (c) 2016 The ZLMediaKit project authors. All Rights Reserved.
*
* This file is part of ZLMediaKit(https://github.com/xiongziliang/ZLMediaKit).
*
* Use of this source code is governed by MIT license that can be found in the
* LICENSE file in the root of the source tree. All contributing project authors
* may be found in the AUTHORS file in the root of the source tree.
*/
#ifndef ZLMEDIAKIT_PROCESSINTERFACE_H
#define ZLMEDIAKIT_PROCESSINTERFACE_H
#include <stdint.h>
#include <memory>
namespace mediakit {
class ProcessInterface {
public:
using Ptr = std::shared_ptr<ProcessInterface>;
ProcessInterface() = default;
virtual ~ProcessInterface() = default;
/**
* 输入rtp
* @param is_udp 是否为udp模式
* @param data rtp数据指针
* @param data_len rtp数据长度
* @return 是否解析成功
*/
virtual bool inputRtp(bool is_udp, const char *data, int data_len) = 0;
};
}//namespace mediakit
#endif //ZLMEDIAKIT_PROCESSINTERFACE_H

33
src/Rtp/RtpCache.cpp Normal file
View File

@@ -0,0 +1,33 @@
/*
* Copyright (c) 2016 The ZLMediaKit project authors. All Rights Reserved.
*
* This file is part of ZLMediaKit(https://github.com/xiongziliang/ZLMediaKit).
*
* Use of this source code is governed by MIT license that can be found in the
* LICENSE file in the root of the source tree. All contributing project authors
* may be found in the AUTHORS file in the root of the source tree.
*/
#include "RtpCache.h"
namespace mediakit{
RtpCache::RtpCache(onFlushed cb) {
_cb = std::move(cb);
}
void RtpCache::onFlush(std::shared_ptr<List<Buffer::Ptr> > rtp_list, bool) {
_cb(std::move(rtp_list));
}
void RtpCache::input(uint64_t stamp, Buffer::Ptr buffer) {
inputPacket(stamp, true, std::move(buffer), false);
}
void RtpCachePS::onRTP(Buffer::Ptr buffer) {
auto rtp = static_pointer_cast<RtpPacket>(buffer);
auto stamp = rtp->timeStamp;
input(stamp, std::move(buffer));
}
}//namespace mediakit

49
src/Rtp/RtpCache.h Normal file
View File

@@ -0,0 +1,49 @@
/*
* Copyright (c) 2016 The ZLMediaKit project authors. All Rights Reserved.
*
* This file is part of ZLMediaKit(https://github.com/xiongziliang/ZLMediaKit).
*
* Use of this source code is governed by MIT license that can be found in the
* LICENSE file in the root of the source tree. All contributing project authors
* may be found in the AUTHORS file in the root of the source tree.
*/
#ifndef ZLMEDIAKIT_RTPCACHE_H
#define ZLMEDIAKIT_RTPCACHE_H
#include "PSEncoder.h"
#include "Extension/CommonRtp.h"
namespace mediakit{
class RtpCache : public PacketCache<Buffer> {
public:
using onFlushed = function<void(std::shared_ptr<List<Buffer::Ptr> >)>;
RtpCache(onFlushed cb);
~RtpCache() override = default;
protected:
/**
* 输入rtp(目的是为了合并写)
* @param buffer rtp数据
*/
void input(uint64_t stamp, Buffer::Ptr buffer);
protected:
void onFlush(std::shared_ptr<List<Buffer::Ptr> > rtp_list, bool) override;
private:
onFlushed _cb;
};
class RtpCachePS : public RtpCache, public PSEncoderImp{
public:
RtpCachePS(onFlushed cb, uint32_t ssrc, uint8_t payload_type = 96) : RtpCache(std::move(cb)), PSEncoderImp(ssrc, payload_type) {};
~RtpCachePS() override = default;
protected:
void onRTP(Buffer::Ptr rtp) override;
};
}//namespace mediakit
#endif //ZLMEDIAKIT_RTPCACHE_H

166
src/Rtp/RtpProcess.cpp
View File

@@ -9,14 +9,17 @@
*/
#if defined(ENABLE_RTPPROXY)
#include "GB28181Process.h"
#include "RtpProcess.h"
#include "RtpSplitter.h"
#include "Util/File.h"
#include "Http/HttpTSPlayer.h"
#define RTP_APP_NAME "rtp"
namespace mediakit{
namespace mediakit {
static string printAddress(const struct sockaddr *addr){
static string printAddress(const struct sockaddr *addr) {
return StrPrinter << SockUtil::inet_ntoa(((struct sockaddr_in *) addr)->sin_addr) << ":" << ntohs(((struct sockaddr_in *) addr)->sin_port);
}
@@ -26,20 +29,11 @@ RtpProcess::RtpProcess(const string &stream_id) {
_media_info._app = RTP_APP_NAME;
_media_info._streamid = stream_id;
GET_CONFIG(string,dump_dir,RtpProxy::kDumpDir);
GET_CONFIG(string, dump_dir, RtpProxy::kDumpDir);
{
FILE *fp = !dump_dir.empty() ? File::create_file(File::absolutePath(_media_info._streamid + ".rtp", dump_dir).data(), "wb") : nullptr;
if(fp){
_save_file_rtp.reset(fp,[](FILE *fp){
fclose(fp);
});
}
}
{
FILE *fp = !dump_dir.empty() ? File::create_file(File::absolutePath(_media_info._streamid + ".mp2", dump_dir).data(), "wb") : nullptr;
if(fp){
_save_file_ps.reset(fp,[](FILE *fp){
if (fp) {
_save_file_rtp.reset(fp, [](FILE *fp) {
fclose(fp);
});
}
@@ -47,20 +41,16 @@ RtpProcess::RtpProcess(const string &stream_id) {
{
FILE *fp = !dump_dir.empty() ? File::create_file(File::absolutePath(_media_info._streamid + ".video", dump_dir).data(), "wb") : nullptr;
if(fp){
_save_file_video.reset(fp,[](FILE *fp){
if (fp) {
_save_file_video.reset(fp, [](FILE *fp) {
fclose(fp);
});
}
}
_rtp_decoder = std::make_shared<CommonRtpDecoder>(CodecInvalid, 256 * 1024);
_rtp_decoder->addDelegate(std::make_shared<FrameWriterInterfaceHelper>([this](const Frame::Ptr &frame){
onRtpDecode((uint8_t *) frame->data(), frame->size(), frame->dts());
}));
}
RtpProcess::~RtpProcess() {
uint64_t duration = (_last_rtp_time.createdTime() - _last_rtp_time.elapsedTime()) / 1000;
uint64_t duration = (_last_frame_time.createdTime() - _last_frame_time.elapsedTime()) / 1000;
WarnP(this) << "RTP推流器("
<< _media_info._vhost << "/"
<< _media_info._app << "/"
@@ -79,95 +69,47 @@ RtpProcess::~RtpProcess() {
}
}
bool RtpProcess::inputRtp(const Socket::Ptr &sock, const char *data, int data_len,const struct sockaddr *addr,uint32_t *dts_out) {
GET_CONFIG(bool,check_source,RtpProxy::kCheckSource);
bool RtpProcess::inputRtp(bool is_udp, const Socket::Ptr &sock, const char *data, int len, const struct sockaddr *addr, uint32_t *dts_out) {
GET_CONFIG(bool, check_source, RtpProxy::kCheckSource);
//检查源是否合法
if(!_addr){
if (!_addr) {
_addr = new struct sockaddr;
_sock = sock;
memcpy(_addr,addr, sizeof(struct sockaddr));
memcpy(_addr, addr, sizeof(struct sockaddr));
DebugP(this) << "bind to address:" << printAddress(_addr);
//推流鉴权
emitOnPublish();
}
if(!_muxer){
if (!_muxer) {
//无权限推流
return false;
}
if(check_source && memcmp(_addr,addr,sizeof(struct sockaddr)) != 0){
if (check_source && memcmp(_addr, addr, sizeof(struct sockaddr)) != 0) {
DebugP(this) << "address dismatch:" << printAddress(addr) << " != " << printAddress(_addr);
return false;
}
_total_bytes += data_len;
bool ret = handleOneRtp(0, TrackVideo, 90000, (unsigned char *) data, data_len);
if(dts_out){
_total_bytes += len;
if (_save_file_rtp) {
uint16_t size = len;
size = htons(size);
fwrite((uint8_t *) &size, 2, 1, _save_file_rtp.get());
fwrite((uint8_t *) data, len, 1, _save_file_rtp.get());
}
if (!_process) {
_process = std::make_shared<GB28181Process>(_media_info, this);
}
bool ret = _process ? _process->inputRtp(is_udp, data, len) : false;
if (dts_out) {
*dts_out = _dts;
}
return ret;
}
//判断是否为ts负载
static inline bool checkTS(const uint8_t *packet, int bytes){
return bytes % TS_PACKET_SIZE == 0 && packet[0] == TS_SYNC_BYTE;
}
void RtpProcess::onRtpSorted(const RtpPacket::Ptr &rtp, int) {
if(rtp->sequence != (uint16_t)(_sequence + 1) && _sequence != 0){
WarnP(this) << "rtp丢包:" << rtp->sequence << " != " << _sequence << "+1" << ",公网环境下请使用tcp方式推流";
}
_sequence = rtp->sequence;
if(_save_file_rtp){
uint16_t size = rtp->size() - 4;
size = htons(size);
fwrite((uint8_t *) &size, 2, 1, _save_file_rtp.get());
fwrite((uint8_t *) rtp->data() + 4, rtp->size() - 4, 1, _save_file_rtp.get());
}
_rtp_decoder->inputRtp(rtp);
}
const char *RtpProcess::onSearchPacketTail(const char *packet,int bytes){
try {
auto ret = _decoder->input((uint8_t *) packet, bytes);
if (ret > 0) {
return packet + ret;
}
return nullptr;
} catch (std::exception &ex) {
InfoP(this) << "解析ps或ts异常: bytes=" << bytes
<< " ,exception=" << ex.what();
//<< " ,hex=" << hexdump((uint8_t *) packet, bytes);
return nullptr;
}
}
void RtpProcess::onRtpDecode(const uint8_t *packet, int bytes, uint32_t timestamp) {
if(_save_file_ps){
fwrite((uint8_t *)packet,bytes, 1, _save_file_ps.get());
}
if (!_decoder) {
//创建解码器
if (checkTS(packet, bytes)) {
//猜测是ts负载
InfoP(this) << "judged to be TS";
_decoder = DecoderImp::createDecoder(DecoderImp::decoder_ts, this);
} else {
//猜测是ps负载
InfoP(this) << "judged to be PS";
_decoder = DecoderImp::createDecoder(DecoderImp::decoder_ps, this);
}
}
if (_decoder) {
HttpRequestSplitter::input((char *) packet, bytes);
}
}
void RtpProcess::inputFrame(const Frame::Ptr &frame){
_last_rtp_time.resetTime();
void RtpProcess::inputFrame(const Frame::Ptr &frame) {
_last_frame_time.resetTime();
_dts = frame->dts();
if (_save_file_video && frame->getTrackType() == TrackVideo) {
fwrite((uint8_t *) frame->data(), frame->size(), 1, _save_file_video.get());
@@ -175,7 +117,7 @@ void RtpProcess::inputFrame(const Frame::Ptr &frame){
_muxer->inputFrame(frame);
}
void RtpProcess::addTrack(const Track::Ptr & track){
void RtpProcess::addTrack(const Track::Ptr &track) {
_muxer->addTrack(track);
}
@@ -190,14 +132,14 @@ bool RtpProcess::alive() {
}
GET_CONFIG(int,timeoutSec,RtpProxy::kTimeoutSec)
if(_last_rtp_time.elapsedTime() / 1000 < timeoutSec){
if(_last_frame_time.elapsedTime() / 1000 < timeoutSec){
return true;
}
return false;
}
void RtpProcess::onDetach(){
if(_on_detach){
void RtpProcess::onDetach() {
if (_on_detach) {
_on_detach();
}
}
@@ -207,47 +149,43 @@ void RtpProcess::setOnDetach(const function<void()> &cb) {
}
string RtpProcess::get_peer_ip() {
if(_addr){
if (_addr) {
return SockUtil::inet_ntoa(((struct sockaddr_in *) _addr)->sin_addr);
}
return "0.0.0.0";
}
uint16_t RtpProcess::get_peer_port() {
if(!_addr){
if (!_addr) {
return 0;
}
return ntohs(((struct sockaddr_in *) _addr)->sin_port);
}
string RtpProcess::get_local_ip() {
if(_sock){
if (_sock) {
return _sock->get_local_ip();
}
return "0.0.0.0";
}
uint16_t RtpProcess::get_local_port() {
if(_sock){
return _sock->get_local_port();
if (_sock) {
return _sock->get_local_port();
}
return 0;
}
string RtpProcess::getIdentifier() const{
string RtpProcess::getIdentifier() const {
return _media_info._streamid;
}
int RtpProcess::totalReaderCount(){
int RtpProcess::totalReaderCount() {
return _muxer ? _muxer->totalReaderCount() : 0;
}
void RtpProcess::setListener(const std::weak_ptr<MediaSourceEvent> &listener){
if(_muxer){
_muxer->setMediaListener(listener);
}else{
_listener = listener;
}
void RtpProcess::setListener(const std::weak_ptr<MediaSourceEvent> &listener) {
setDelegate(listener);
}
void RtpProcess::setRtpPause(bool pause)
@@ -268,7 +206,7 @@ void RtpProcess::emitOnPublish() {
strongSelf->_media_info._app,
strongSelf->_media_info._streamid, 0,
true, true, enableHls, enableMP4);
strongSelf->_muxer->setMediaListener(strongSelf->_listener);
strongSelf->_muxer->setMediaListener(strongSelf);
InfoP(strongSelf) << "允许RTP推流";
} else {
WarnP(strongSelf) << "禁止RTP推流:" << err;
@@ -277,7 +215,7 @@ void RtpProcess::emitOnPublish() {
//触发推流鉴权事件
auto flag = NoticeCenter::Instance().emitEvent(Broadcast::kBroadcastMediaPublish, _media_info, invoker, static_cast<SockInfo &>(*this));
if(!flag){
if (!flag) {
//该事件无人监听,默认不鉴权
GET_CONFIG(bool, toHls, General::kPublishToHls);
GET_CONFIG(bool, toMP4, General::kPublishToMP4);
@@ -285,5 +223,17 @@ void RtpProcess::emitOnPublish() {
}
}
MediaOriginType RtpProcess::getOriginType(MediaSource &sender) const{
return MediaOriginType::rtp_push;
}
string RtpProcess::getOriginUrl(MediaSource &sender) const {
return _media_info._full_url;
}
std::shared_ptr<SockInfo> RtpProcess::getOriginSock(MediaSource &sender) const{
return const_cast<RtpProcess *>(this)->shared_from_this();
}
}//namespace mediakit
#endif//defined(ENABLE_RTPPROXY)

43
src/Rtp/RtpProcess.h
View File

@@ -12,18 +12,14 @@
#define ZLMEDIAKIT_RTPPROCESS_H
#if defined(ENABLE_RTPPROXY)
#include "ProcessInterface.h"
#include "Common/MultiMediaSourceMuxer.h"
#include "Rtsp/RtpReceiver.h"
#include "Decoder.h"
#include "Common/Device.h"
#include "Common/Stamp.h"
#include "Http/HttpRequestSplitter.h"
#include "Extension/CommonRtp.h"
using namespace mediakit;
namespace mediakit{
namespace mediakit {
class RtpProcess : public HttpRequestSplitter, public RtpReceiver, public SockInfo, public MediaSinkInterface, public std::enable_shared_from_this<RtpProcess>{
class RtpProcess : public SockInfo, public MediaSinkInterface, public MediaSourceEventInterceptor, public std::enable_shared_from_this<RtpProcess>{
public:
typedef std::shared_ptr<RtpProcess> Ptr;
friend class RtpProcessHelper;
@@ -32,14 +28,15 @@ public:
/**
* 输入rtp
* @param is_udp 是否为udp模式
* @param sock 本地监听的socket
* @param data rtp数据指针
* @param data_len rtp数据长度
* @param len rtp数据长度
* @param addr 数据源地址
* @param dts_out 解析出最新的dts
* @return 是否解析成功
*/
bool inputRtp(const Socket::Ptr &sock, const char *data,int data_len, const struct sockaddr *addr , uint32_t *dts_out = nullptr);
bool inputRtp(bool is_udp, const Socket::Ptr &sock, const char *data, int len, const struct sockaddr *addr , uint32_t *dts_out = nullptr);
/**
* 是否超时，用于超时移除对象
@@ -72,34 +69,30 @@ public:
void setRtpPause(bool pause);
protected:
void onRtpSorted(const RtpPacket::Ptr &rtp, int track_index) override ;
void inputFrame(const Frame::Ptr &frame) override;
void addTrack(const Track::Ptr & track) override;
void resetTracks() override {};
const char *onSearchPacketTail(const char *data,int len) override;
int64_t onRecvHeader(const char *data,uint64_t len) override { return 0; };
//// MediaSourceEvent override ////
MediaOriginType getOriginType(MediaSource &sender) const override;
string getOriginUrl(MediaSource &sender) const override;
std::shared_ptr<SockInfo> getOriginSock(MediaSource &sender) const override;
private:
void emitOnPublish();
void onRtpDecode(const uint8_t *packet, int bytes, uint32_t timestamp);
private:
std::shared_ptr<CommonRtpDecoder> _rtp_decoder;
std::shared_ptr<FILE> _save_file_rtp;
std::shared_ptr<FILE> _save_file_ps;
std::shared_ptr<FILE> _save_file_video;
struct sockaddr *_addr = nullptr;
uint16_t _sequence = 0;
MultiMediaSourceMuxer::Ptr _muxer;
Ticker _last_rtp_time;
uint32_t _dts = 0;
DecoderImp::Ptr _decoder;
std::weak_ptr<MediaSourceEvent> _listener;
MediaInfo _media_info;
uint64_t _total_bytes = 0;
struct sockaddr *_addr = nullptr;
Socket::Ptr _sock;
MediaInfo _media_info;
Ticker _last_frame_time;
function<void()> _on_detach;
std::shared_ptr<FILE> _save_file_rtp;
std::shared_ptr<FILE> _save_file_video;
ProcessInterface::Ptr _process;
MultiMediaSourceMuxer::Ptr _muxer;
bool _paused = false;
Ticker _pause_rtp_time;

29
src/Rtp/RtpSelector.cpp
View File

@@ -9,25 +9,34 @@
*/
#if defined(ENABLE_RTPPROXY)
#include <stddef.h>
#include "RtpSelector.h"
#include "RtpSplitter.h"
namespace mediakit{
INSTANCE_IMP(RtpSelector);
void RtpSelector::clear(){
lock_guard<decltype(_mtx_map)> lck(_mtx_map);
_map_rtp_process.clear();
}
bool RtpSelector::inputRtp(const Socket::Ptr &sock, const char *data, int data_len,
const struct sockaddr *addr,uint32_t *dts_out) {
//使用ssrc为流id
uint32_t ssrc = 0;
if (!getSSRC(data, data_len, ssrc)) {
WarnL << "get ssrc from rtp failed:" << data_len;
return false;
}
//假定指定了流id，那么通过流id来区分是否为一路流(哪怕可能同时收到多路流)
auto process = getProcess(printSSRC(ssrc), true);
if (process) {
return process->inputRtp(sock, data, data_len, addr, dts_out);
try {
return process->inputRtp(true, sock, data, data_len, addr, dts_out);
} catch (...) {
delProcess(printSSRC(ssrc), process.get());
throw;
}
}
return false;
}
@@ -145,18 +154,6 @@ int RtpProcessHelper::totalReaderCount(MediaSource &sender) {
return _process ? _process->totalReaderCount() : sender.totalReaderCount();
}
MediaOriginType RtpProcessHelper::getOriginType(MediaSource &sender) const{
return MediaOriginType::rtp_push;
}
string RtpProcessHelper::getOriginUrl(MediaSource &sender) const {
return _process ? _process->_media_info._full_url : "";
}
std::shared_ptr<SockInfo> RtpProcessHelper::getOriginSock(MediaSource &sender) const{
return _process;
}
RtpProcess::Ptr &RtpProcessHelper::getProcess() {
return _process;
}

11
src/Rtp/RtpSelector.h
View File

@@ -34,12 +34,6 @@ protected:
bool close(MediaSource &sender,bool force) override;
// 观看总人数
int totalReaderCount(MediaSource &sender) override;
// 获取媒体源类型
MediaOriginType getOriginType(MediaSource &sender) const override;
// 获取媒体源url或者文件路径
string getOriginUrl(MediaSource &sender) const override;
// 获取媒体源客户端相关信息
std::shared_ptr<SockInfo> getOriginSock(MediaSource &sender) const override;
private:
weak_ptr<RtpSelector > _parent;
@@ -55,6 +49,11 @@ public:
static bool getSSRC(const char *data,int data_len, uint32_t &ssrc);
static RtpSelector &Instance();
/**
* 清空所有对象
*/
void clear();
/**
* 输入多个rtp流，根据ssrc分流
* @param sock 本地socket

69
src/Rtp/PSRtpSender.cpp → src/Rtp/RtpSender.cpp
View File

@@ -9,33 +9,29 @@
*/
#if defined(ENABLE_RTPPROXY)
#include "PSRtpSender.h"
#include "RtpSender.h"
#include "Rtsp/RtspSession.h"
#include "Thread/WorkThreadPool.h"
#include "RtpCache.h"
namespace mediakit{
PSRtpSender::PSRtpSender(uint32_t ssrc, uint8_t payload_type) {
GET_CONFIG(uint32_t,video_mtu,Rtp::kVideoMtuSize);
_rtp_encoder = std::make_shared<CommonRtpEncoder>(CodecInvalid, ssrc, video_mtu, 90000, payload_type, 0);
_rtp_encoder->setRtpRing(std::make_shared<RtpRing::RingType>());
_rtp_encoder->getRtpRing()->setDelegate(std::make_shared<RingDelegateHelper>([this](const RtpPacket::Ptr &rtp, bool is_key){
onRtp(rtp, is_key);
}));
RtpSender::RtpSender(uint32_t ssrc, uint8_t payload_type) {
_poller = EventPollerPool::Instance().getPoller();
InfoL << this << " " << printSSRC(_rtp_encoder->getSsrc());
_interface = std::make_shared<RtpCachePS>([this](std::shared_ptr<List<Buffer::Ptr> > list) {
onFlushRtpList(std::move(list));
}, ssrc, payload_type);
}
PSRtpSender::~PSRtpSender() {
InfoL << this << " " << printSSRC(_rtp_encoder->getSsrc());
RtpSender::~RtpSender() {
}
void PSRtpSender::startSend(const string &dst_url, uint16_t dst_port, bool is_udp, const function<void(const SockException &ex)> &cb){
void RtpSender::startSend(const string &dst_url, uint16_t dst_port, bool is_udp, const function<void(const SockException &ex)> &cb){
_is_udp = is_udp;
_socket = Socket::createSocket(_poller, false);
_dst_url = dst_url;
_dst_port = dst_port;
weak_ptr<PSRtpSender> weak_self = shared_from_this();
weak_ptr<RtpSender> weak_self = shared_from_this();
if (is_udp) {
_socket->bindUdpSock(0);
auto poller = _poller;
@@ -73,7 +69,7 @@ void PSRtpSender::startSend(const string &dst_url, uint16_t dst_port, bool is_ud
}
}
void PSRtpSender::onConnect(){
void RtpSender::onConnect(){
_is_connect = true;
//加大发送缓存,防止udp丢包之类的问题
SockUtil::setSendBuf(_socket->rawFD(), 4 * 1024 * 1024);
@@ -83,37 +79,38 @@ void PSRtpSender::onConnect(){
_socket->setSendFlags(SOCKET_DEFAULE_FLAGS | FLAG_MORE);
}
//连接建立成功事件
weak_ptr<PSRtpSender> weak_self = shared_from_this();
weak_ptr<RtpSender> weak_self = shared_from_this();
_socket->setOnErr([weak_self](const SockException &err) {
auto strong_self = weak_self.lock();
if (strong_self) {
strong_self->onErr(err);
}
});
InfoL << "开始发送 ps rtp:" << _socket->get_peer_ip() << ":" << _socket->get_peer_port() << ", 是否为udp方式:" << _is_udp;
InfoL << "开始发送 rtp:" << _socket->get_peer_ip() << ":" << _socket->get_peer_port() << ", 是否为udp方式:" << _is_udp;
}
void RtpSender::addTrack(const Track::Ptr &track){
_interface->addTrack(track);
}
void RtpSender::addTrackCompleted(){
_interface->addTrackCompleted();
}
void RtpSender::resetTracks(){
_interface->resetTracks();
}
//此函数在其他线程执行
void PSRtpSender::inputFrame(const Frame::Ptr &frame) {
void RtpSender::inputFrame(const Frame::Ptr &frame) {
if (_is_connect) {
//连接成功后才做实质操作(节省cpu资源)
PSEncoder::inputFrame(frame);
_interface->inputFrame(frame);
}
}
//此函数在其他线程执行
void PSRtpSender::onPS(uint32_t stamp, void *packet, size_t bytes) {
_rtp_encoder->inputFrame(std::make_shared<FrameFromPtr>((char *) packet, bytes, stamp));
}
//此函数在其他线程执行
void PSRtpSender::onRtp(const RtpPacket::Ptr &rtp, bool) {
//开启合并写提高发送性能
PacketCache<RtpPacket>::inputPacket(true, rtp, false);
}
//此函数在其他线程执行
void PSRtpSender::onFlush(shared_ptr<List<RtpPacket::Ptr> > rtp_list, bool) {
void RtpSender::onFlushRtpList(shared_ptr<List<Buffer::Ptr> > rtp_list) {
if(!_is_connect){
//连接成功后才能发送数据
return;
@@ -124,29 +121,29 @@ void PSRtpSender::onFlush(shared_ptr<List<RtpPacket::Ptr> > rtp_list, bool) {
_poller->async([rtp_list, is_udp, socket]() {
int i = 0;
int size = rtp_list->size();
rtp_list->for_each([&](const RtpPacket::Ptr &packet) {
rtp_list->for_each([&](Buffer::Ptr &packet) {
if (is_udp) {
//udp模式，rtp over tcp前4个字节可以忽略
socket->send(std::make_shared<BufferRtp>(packet, 4), nullptr, 0, ++i == size);
socket->send(std::make_shared<BufferRtp>(std::move(packet), 4), nullptr, 0, ++i == size);
} else {
//tcp模式, rtp over tcp前2个字节可以忽略,只保留后续rtp长度的2个字节
socket->send(std::make_shared<BufferRtp>(packet, 2), nullptr, 0, ++i == size);
socket->send(std::make_shared<BufferRtp>(std::move(packet), 2), nullptr, 0, ++i == size);
}
});
});
}
void PSRtpSender::onErr(const SockException &ex, bool is_connect) {
void RtpSender::onErr(const SockException &ex, bool is_connect) {
_is_connect = false;
//监听socket断开事件，方便重连
if (is_connect) {
WarnL << "重连" << _dst_url << ":" << _dst_port << "失败, 原因为:" << ex.what();
} else {
WarnL << "停止发送 ps rtp:" << _dst_url << ":" << _dst_port << ", 原因为:" << ex.what();
WarnL << "停止发送 rtp:" << _dst_url << ":" << _dst_port << ", 原因为:" << ex.what();
}
weak_ptr<PSRtpSender> weak_self = shared_from_this();
weak_ptr<RtpSender> weak_self = shared_from_this();
_connect_timer = std::make_shared<Timer>(10.0, [weak_self]() {
auto strong_self = weak_self.lock();
if (!strong_self) {

59
src/Rtp/PSRtpSender.h → src/Rtp/RtpSender.h
View File

@@ -8,42 +8,27 @@
* may be found in the AUTHORS file in the root of the source tree.
*/
#ifndef ZLMEDIAKIT_PSRTPSENDER_H
#define ZLMEDIAKIT_PSRTPSENDER_H
#ifndef ZLMEDIAKIT_RTPSENDER_H
#define ZLMEDIAKIT_RTPSENDER_H
#if defined(ENABLE_RTPPROXY)
#include "PSEncoder.h"
#include "Extension/CommonRtp.h"
namespace mediakit{
class RingDelegateHelper : public RingDelegate<RtpPacket::Ptr> {
//rtp发送客户端，支持发送GB28181协议
class RtpSender : public MediaSinkInterface, public std::enable_shared_from_this<RtpSender>{
public:
typedef function<void(const RtpPacket::Ptr &in, bool is_key)> onRtp;
typedef std::shared_ptr<RtpSender> Ptr;
~RingDelegateHelper() override{}
RingDelegateHelper(onRtp on_rtp){
_on_rtp = std::move(on_rtp);
}
void onWrite(RtpPacket::Ptr in, bool is_key) override{
_on_rtp(in, is_key);
}
private:
onRtp _on_rtp;
};
//该类在PSEncoder的基础上，实现了mpeg-ps的rtp打包以及发送
class PSRtpSender : public PSEncoder, public std::enable_shared_from_this<PSRtpSender>, public PacketCache<RtpPacket>{
public:
typedef std::shared_ptr<PSRtpSender> Ptr;
~RtpSender() override;
/**
* 构造函数
* 构造函数，创建GB28181 RTP发送客户端
* @param ssrc rtp的ssrc
* @param payload_type 国标中ps-rtp的pt一般为96
*/
PSRtpSender(uint32_t ssrc, uint8_t payload_type = 96);
~PSRtpSender() override;
RtpSender(uint32_t ssrc, uint8_t payload_type = 96);
/**
* 开始发送ps-rtp包
@@ -59,20 +44,26 @@ public:
*/
void inputFrame(const Frame::Ptr &frame) override;
protected:
//mpeg-ps回调
void onPS(uint32_t stamp, void *packet, size_t bytes) override;
/**
* 添加track，内部会调用Track的clone方法
* 只会克隆sps pps这些信息 ，而不会克隆Delegate相关关系
* @param track
*/
virtual void addTrack(const Track::Ptr & track) override;
/**
* 批量flush rtp包时触发该函数
* @param rtp_list rtp包列表
* @param key_pos 是否包含关键帧
* 添加所有Track完毕
*/
void onFlush(std::shared_ptr<List<RtpPacket::Ptr> > rtp_list, bool key_pos) override;
virtual void addTrackCompleted() override;
/**
* 重置track
*/
virtual void resetTracks() override;
private:
//rtp打包后回调
void onRtp(const RtpPacket::Ptr &in, bool is_key);
//合并写输出
void onFlushRtpList(std::shared_ptr<List<Buffer::Ptr> > rtp_list);
//udp/tcp连接成功回调
void onConnect();
//异常断开socket事件
@@ -86,9 +77,9 @@ private:
Socket::Ptr _socket;
EventPoller::Ptr _poller;
Timer::Ptr _connect_timer;
std::shared_ptr<CommonRtpEncoder> _rtp_encoder;
MediaSinkInterface::Ptr _interface;
};
}//namespace mediakit
#endif// defined(ENABLE_RTPPROXY)
#endif //ZLMEDIAKIT_PSRTPSENDER_H
#endif //ZLMEDIAKIT_RTPSENDER_H

2
src/Rtp/RtpServer.cpp
View File

@@ -52,7 +52,7 @@ void RtpServer::start(uint16_t local_port, const string &stream_id, bool enable
//指定了流id，那么一个端口一个流(不管是否包含多个ssrc的多个流，绑定rtp源后，会筛选掉ip端口不匹配的流)
process = RtpSelector::Instance().getProcess(stream_id, true);
udp_server->setOnRead([udp_server, process](const Buffer::Ptr &buf, struct sockaddr *addr, int) {
process->inputRtp(udp_server, buf->data(), buf->size(), addr);
process->inputRtp(true, udp_server, buf->data(), buf->size(), addr);
});
} else {
//未指定流id，一个端口多个流，通过ssrc来分流

2
src/Rtp/RtpSession.cpp
View File

@@ -73,7 +73,7 @@ void RtpSession::onRtpPacket(const char *data, uint64_t len) {
_process = RtpSelector::Instance().getProcess(_stream_id, true);
_process->setListener(dynamic_pointer_cast<RtpSession>(shared_from_this()));
}
_process->inputRtp(getSock(), data, len, &addr);
_process->inputRtp(false, getSock(), data, len, &addr);
_ticker.resetTime();
}

7
src/Rtp/RtpSession.h
View File

@@ -35,13 +35,14 @@ protected:
bool close(MediaSource &sender,bool force) override;
// 观看总人数
int totalReaderCount(MediaSource &sender) override;
void onRtpPacket(const char *data,uint64_t len) override;
// 收到rtp回调
void onRtpPacket(const char *data, uint64_t len) override;
private:
RtpProcess::Ptr _process;
Ticker _ticker;
struct sockaddr addr;
string _stream_id;
struct sockaddr addr;
RtpProcess::Ptr _process;
};
}//namespace mediakit

46
src/Rtp/RtpSplitter.cpp
View File

@@ -9,18 +9,57 @@
*/
#if defined(ENABLE_RTPPROXY)
#include <netinet/in.h>
#include <string.h>
#include "RtpSplitter.h"
namespace mediakit{
RtpSplitter::RtpSplitter() {}
static const char kEHOME_MAGIC[] = "\x01\x00\x01\x00";
static const int kEHOME_OFFSET = 256;
RtpSplitter::RtpSplitter() {}
RtpSplitter::~RtpSplitter() {}
int64_t RtpSplitter::onRecvHeader(const char *data,uint64_t len){
//忽略偏移量
data += _offset;
len -= _offset;
if (_offset == kEHOME_OFFSET + 4 && len > 12 && data[12] == '\r') {
//这是ehome,移除第12个字节
memmove((char *) data + 1, data, 12);
data += 1;
len -= 1;
}
onRtpPacket(data, len);
return 0;
}
static bool isEhome(const char *data, int len){
if (len < 4) {
return false;
}
return memcmp(data, kEHOME_MAGIC, sizeof(kEHOME_MAGIC) - 1) == 0;
}
const char *RtpSplitter::onSearchPacketTail(const char *data, int len) {
if (len < 4) {
//数据不够
return nullptr;
}
if (isEhome(data, len)) {
//是ehome协议
if (len < kEHOME_OFFSET + 4) {
//数据不够
return nullptr;
}
//忽略ehome私有头后是rtsp样式的rtp，多4个字节，
_offset = kEHOME_OFFSET + 4;
//忽略ehome私有头
return onSearchPacketTail_l(data + kEHOME_OFFSET + 2, len - kEHOME_OFFSET - 2);
}
if (data[0] == '$') {
//可能是4个字节的rtp头
_offset = 4;
@@ -42,10 +81,5 @@ const char *RtpSplitter::onSearchPacketTail_l(const char *data, int len) {
return data + 2 + length;
}
int64_t RtpSplitter::onRecvHeader(const char *data, uint64_t len) {
onRtpPacket(data + _offset, len - _offset);
return 0;
}
}//namespace mediakit
#endif//defined(ENABLE_RTPPROXY)

12
src/Rtp/RtpSplitter.h
View File

@@ -19,18 +19,20 @@ namespace mediakit{
class RtpSplitter : public HttpRequestSplitter{
public:
RtpSplitter();
virtual ~RtpSplitter();
~RtpSplitter() override;
protected:
/**
* 收到rtp包回调
* @param data RTP包数据指针
* @param len RTP包数据长度
*/
virtual void onRtpPacket(const char *data,uint64_t len) = 0;
virtual void onRtpPacket(const char *data, uint64_t len) = 0;
protected:
const char *onSearchPacketTail(const char *data,int len) override ;
const char *onSearchPacketTail_l(const char *data,int len);
int64_t onRecvHeader(const char *data,uint64_t len) override;
int64_t onRecvHeader(const char *data, uint64_t len) override;
const char *onSearchPacketTail(const char *data, int len) override;
const char *onSearchPacketTail_l(const char *data, int len);
private:
int _offset = 0;

20
src/Rtsp/RtpReceiver.cpp
View File

@@ -36,7 +36,11 @@ bool RtpReceiver::handleOneRtp(int track_index, TrackType type, int samplerate,
return false;
}
uint32_t version = rtp_raw_ptr[0] >> 6;
uint8_t padding = 0;
uint8_t ext = rtp_raw_ptr[0] & 0x10;
uint8_t csrc = rtp_raw_ptr[0] & 0x0f;
if (rtp_raw_ptr[0] & 0x20) {
//获取padding大小
padding = rtp_raw_ptr[rtp_raw_len - 1];
@@ -46,6 +50,10 @@ bool RtpReceiver::handleOneRtp(int track_index, TrackType type, int samplerate,
rtp_raw_len -= padding;
}
if (version != 2) {
throw std::invalid_argument("非法的rtp，version != 2");
}
auto rtp_ptr = _rtp_pool.obtain();
auto &rtp = *rtp_ptr;
@@ -95,8 +103,6 @@ bool RtpReceiver::handleOneRtp(int track_index, TrackType type, int samplerate,
//获取rtp中媒体数据偏移量
rtp.offset = 12 + 4;
int csrc = rtp_raw_ptr[0] & 0x0f;
int ext = rtp_raw_ptr[0] & 0x10;
rtp.offset += 4 * csrc;
if (ext && rtp_raw_len >= rtp.offset) {
/* calculate the header extension length (stored as number of 32-bit words) */
@@ -124,13 +130,11 @@ bool RtpReceiver::handleOneRtp(int track_index, TrackType type, int samplerate,
payload_ptr[3] = (rtp_raw_len & 0x00FF);
//拷贝rtp负载
memcpy(payload_ptr + 4, rtp_raw_ptr, rtp_raw_len);
//排序rtp
sortRtp(std::move(rtp_ptr), track_index);
return true;
}
void RtpReceiver::sortRtp(const RtpPacket::Ptr &rtp,int track_index){
_rtp_sortor[track_index].sortPacket(rtp->sequence, rtp);
//排序rtp
auto seq = rtp_ptr->sequence;
_rtp_sortor[track_index].sortPacket(seq, std::move(rtp_ptr));
return true;
}
void RtpReceiver::clear() {

3
src/Rtsp/RtpReceiver.h
View File

@@ -187,9 +187,6 @@ protected:
int getJitterSize(int track_index);
int getCycleCount(int track_index);
private:
void sortRtp(const RtpPacket::Ptr &rtp , int track_index);
private:
uint32_t _ssrc[2] = {0, 0};
//ssrc不匹配计数

3
src/Rtsp/RtspMediaSource.h
View File

@@ -183,7 +183,8 @@ public:
}
}
bool is_video = rtp->type == TrackVideo;
PacketCache<RtpPacket>::inputPacket(is_video, std::move(rtp), keyPos);
auto stamp = rtp->timeStamp;
PacketCache<RtpPacket>::inputPacket(stamp, is_video, std::move(rtp), keyPos);
}
void clearCache() override{

16
src/Rtsp/RtspMediaSourceImp.h
View File

@@ -77,7 +77,7 @@ public:
_muxer->setMediaListener(getListener());
_muxer->setTrackListener(static_pointer_cast<RtspMediaSourceImp>(shared_from_this()));
//让_muxer对象拦截一部分事件(比如说录像相关事件)
setListener(_muxer);
MediaSource::setListener(_muxer);
for(auto &track : _demuxer->getTracks(false)){
_muxer->addTrack(track);
@@ -102,6 +102,20 @@ public:
_all_track_ready = true;
}
/**
* 设置事件监听器
* @param listener 监听器
*/
void setListener(const std::weak_ptr<MediaSourceEvent> &listener) override{
if (_muxer) {
//_muxer对象不能处理的事件再给listener处理
_muxer->setMediaListener(listener);
} else {
//未创建_muxer对象，事件全部给listener处理
MediaSource::setListener(listener);
}
}
private:
RtspDemuxer::Ptr _demuxer;
MultiMediaSourceMuxer::Ptr _muxer;

2
src/Rtsp/RtspMediaSourceMuxer.h
View File

@@ -32,7 +32,7 @@ public:
~RtspMediaSourceMuxer() override{}
void setListener(const std::weak_ptr<MediaSourceEvent> &listener){
_listener = listener;
setDelegate(listener);
_media_src->setListener(shared_from_this());
}

8
src/Rtsp/RtspSession.h
View File

@@ -37,17 +37,19 @@ class RtspSession;
class BufferRtp : public Buffer{
public:
typedef std::shared_ptr<BufferRtp> Ptr;
BufferRtp(const RtpPacket::Ptr & pkt,uint32_t offset = 0 ):_rtp(pkt),_offset(offset){}
virtual ~BufferRtp(){}
BufferRtp(Buffer::Ptr pkt, uint32_t offset = 0) : _rtp(std::move(pkt)), _offset(offset) {}
~BufferRtp() override{}
char *data() const override {
return (char *)_rtp->data() + _offset;
}
uint32_t size() const override {
return _rtp->size() - _offset;
}
private:
RtpPacket::Ptr _rtp;
Buffer::Ptr _rtp;
uint32_t _offset;
};

18
src/TS/TSMediaSource.h
View File

@@ -30,19 +30,8 @@ public:
uint32_t time_stamp = 0;
};
//TS直播合并写策略类
class TSFlushPolicy : public FlushPolicy{
public:
TSFlushPolicy() = default;
~TSFlushPolicy() = default;
uint32_t getStamp(const TSPacket::Ptr &packet) {
return packet->time_stamp;
}
};
//TS直播源
class TSMediaSource : public MediaSource, public RingDelegate<TSPacket::Ptr>, public PacketCache<TSPacket, TSFlushPolicy>{
class TSMediaSource : public MediaSource, public RingDelegate<TSPacket::Ptr>, public PacketCache<TSPacket>{
public:
using PoolType = ResourcePool<TSPacket>;
using Ptr = std::shared_ptr<TSMediaSource>;
@@ -83,14 +72,15 @@ public:
if (key) {
_have_video = true;
}
PacketCache<TSPacket, TSFlushPolicy>::inputPacket(true, std::move(packet), key);
auto stamp = packet->time_stamp;
PacketCache<TSPacket>::inputPacket(stamp, true, std::move(packet), key);
}
/**
* 情况GOP缓存
*/
void clearCache() override {
PacketCache<TSPacket, TSFlushPolicy>::clearCache();
PacketCache<TSPacket>::clearCache();
_ring->clearCache();
}

2
src/TS/TSMediaSourceMuxer.h
View File

@@ -31,7 +31,7 @@ public:
~TSMediaSourceMuxer() override = default;
void setListener(const std::weak_ptr<MediaSourceEvent> &listener){
_listener = listener;
setDelegate(listener);
_media_src->setListener(shared_from_this());
}