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

新增媒体流flush机制：#1996

Browse Source
This commit is contained in:
ziyue
2022-10-16 19:49:56 +08:00
parent 80eef693c6
commit ac1abb34da
44 changed files with 287 additions and 124 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
src/Rtp/Decoder.cpp
View File

@@ -65,6 +65,10 @@ DecoderImp::Ptr DecoderImp::createDecoder(Type type, MediaSinkInterface *sink){
return DecoderImp::Ptr(new DecoderImp(decoder, sink));
}
void DecoderImp::flush() {
_merger.flush();
}
ssize_t DecoderImp::input(const uint8_t *data, size_t bytes){
return _decoder->input(data, bytes);
}
@@ -219,10 +223,7 @@ void DecoderImp::onDecode(int stream,int codecid,int flags,int64_t pts,int64_t d
default:
// 海康的 PS 流中会有 codecid 为 0xBD 的包
if (codecid != 0 && codecid != 0xBD) {
if (_last_unsported_print.elapsedTime() / 1000 > 5) {
_last_unsported_print.resetTime();
WarnL << "unsupported codec type:" << getCodecName(codecid) << " " << (int) codecid;
}
WarnL << "unsupported codec type:" << getCodecName(codecid) << " " << (int) codecid;
}
break;
}

7
src/Rtp/Decoder.h
View File

@@ -39,16 +39,14 @@ protected:
class DecoderImp{
public:
typedef enum {
decoder_ts = 0,
decoder_ps
}Type;
typedef enum { decoder_ts = 0, decoder_ps } Type;
typedef std::shared_ptr<DecoderImp> Ptr;
~DecoderImp() = default;
static Ptr createDecoder(Type type, MediaSinkInterface *sink);
ssize_t input(const uint8_t *data, size_t bytes);
void flush();
protected:
void onTrack(const Track::Ptr &track);
@@ -63,7 +61,6 @@ private:
Decoder::Ptr _decoder;
MediaSinkInterface *_sink;
FrameMerger _merger{FrameMerger::none};
toolkit::Ticker _last_unsported_print;
Track::Ptr _tracks[TrackMax];
};

8
src/Rtp/GB28181Process.cpp
View File

@@ -59,12 +59,16 @@ GB28181Process::GB28181Process(const MediaInfo &media_info, MediaSinkInterface *
_interface = sink;
}
GB28181Process::~GB28181Process() {}
void GB28181Process::onRtpSorted(RtpPacket::Ptr rtp) {
_rtp_decoder[rtp->getHeader()->pt]->inputRtp(rtp, false);
}
void GB28181Process::flush() {
if (_decoder) {
_decoder->flush();
}
}
bool GB28181Process::inputRtp(bool, const char *data, size_t data_len) {
GET_CONFIG(uint32_t, h264_pt, RtpProxy::kH264PT);
GET_CONFIG(uint32_t, h265_pt, RtpProxy::kH265PT);

7
src/Rtp/GB28181Process.h
View File

@@ -26,7 +26,7 @@ class GB28181Process : public ProcessInterface {
public:
typedef std::shared_ptr<GB28181Process> Ptr;
GB28181Process(const MediaInfo &media_info, MediaSinkInterface *sink);
~GB28181Process() override;
~GB28181Process() override = default;
/**
* 输入rtp
@@ -36,6 +36,11 @@ public:
*/
bool inputRtp(bool, const char *data, size_t data_len) override;
/**
* 刷新输出所有缓存
*/
void flush() override;
protected:
void onRtpSorted(RtpPacket::Ptr rtp);

5
src/Rtp/ProcessInterface.h
View File

@@ -31,6 +31,11 @@ public:
* @return 是否解析成功
*/
virtual bool inputRtp(bool is_udp, const char *data, size_t data_len) = 0;
/**
* 刷新输出所有缓存
*/
virtual void flush() {}
};
}//namespace mediakit

38
src/Rtp/RtpCache.cpp
View File

@@ -19,37 +19,49 @@ namespace mediakit{
RtpCache::RtpCache(onFlushed cb) {
_cb = std::move(cb);
}
bool RtpCache::firstKeyReady(bool in) {
if(_first_key){
if (_first_key) {
return _first_key;
}
_first_key = in;
return _first_key;
}
void RtpCache::onFlush(std::shared_ptr<List<Buffer::Ptr> > rtp_list, bool) {
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,bool is_key ) {
void RtpCache::input(uint64_t stamp, Buffer::Ptr buffer, bool is_key) {
inputPacket(stamp, true, std::move(buffer), is_key);
}
void RtpCachePS::onRTP(Buffer::Ptr buffer,bool is_key) {
if(!firstKeyReady(is_key)){
return;
}
auto rtp = std::static_pointer_cast<RtpPacket>(buffer);
auto stamp = rtp->getStampMS();
input(stamp, std::move(buffer),is_key);
void RtpCachePS::flush() {
PSEncoderImp::flush();
RtpCache::flush();
}
void RtpCacheRaw::onRTP(Buffer::Ptr buffer,bool is_key) {
if(!firstKeyReady(is_key)){
void RtpCachePS::onRTP(Buffer::Ptr buffer, bool is_key) {
if (!firstKeyReady(is_key)) {
return;
}
auto rtp = std::static_pointer_cast<RtpPacket>(buffer);
auto stamp = rtp->getStampMS();
input(stamp, std::move(buffer),is_key);
input(stamp, std::move(buffer), is_key);
}
void RtpCacheRaw::flush() {
RawEncoderImp::flush();
RtpCache::flush();
}
void RtpCacheRaw::onRTP(Buffer::Ptr buffer, bool is_key) {
if (!firstKeyReady(is_key)) {
return;
}
auto rtp = std::static_pointer_cast<RtpPacket>(buffer);
auto stamp = rtp->getStampMS();
input(stamp, std::move(buffer), is_key);
}
}//namespace mediakit

17
src/Rtp/RtpCache.h
View File

@@ -19,7 +19,7 @@
namespace mediakit{
class RtpCache : private PacketCache<toolkit::Buffer> {
class RtpCache : protected PacketCache<toolkit::Buffer> {
public:
using onFlushed = std::function<void(std::shared_ptr<toolkit::List<toolkit::Buffer::Ptr> >)>;
RtpCache(onFlushed cb);
@@ -33,30 +33,33 @@ protected:
void input(uint64_t stamp, toolkit::Buffer::Ptr buffer,bool is_key = false);
bool firstKeyReady(bool in);
protected:
void onFlush(std::shared_ptr<toolkit::List<toolkit::Buffer::Ptr> > rtp_list, bool) override;
private:
onFlushed _cb;
bool _first_key = false;
onFlushed _cb;
};
class RtpCachePS : public RtpCache, public PSEncoderImp{
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;
void flush() override;
protected:
void onRTP(toolkit::Buffer::Ptr rtp,bool is_key = false) override;
void onRTP(toolkit::Buffer::Ptr rtp, bool is_key = false) override;
};
class RtpCacheRaw : public RtpCache, public RawEncoderImp{
class RtpCacheRaw : public RtpCache, public RawEncoderImp {
public:
RtpCacheRaw(onFlushed cb, uint32_t ssrc, uint8_t payload_type = 96, bool sendAudio = true) : RtpCache(std::move(cb)), RawEncoderImp(ssrc, payload_type,sendAudio) {};
RtpCacheRaw(onFlushed cb, uint32_t ssrc, uint8_t payload_type = 96, bool send_audio = true) : RtpCache(std::move(cb)), RawEncoderImp(ssrc, payload_type, send_audio) {};
~RtpCacheRaw() override = default;
void flush() override;
protected:
void onRTP(toolkit::Buffer::Ptr rtp,bool is_key = false) override;
void onRTP(toolkit::Buffer::Ptr rtp, bool is_key = false) override;
};
}//namespace mediakit

3
src/Rtp/RtpProcess.cpp
View File

@@ -50,6 +50,9 @@ RtpProcess::RtpProcess(const string &stream_id) {
}
RtpProcess::~RtpProcess() {
if (_process) {
_process->flush();
}
uint64_t duration = (_last_frame_time.createdTime() - _last_frame_time.elapsedTime()) / 1000;
WarnP(this) << "RTP推流器("
<< _media_info.shortUrl()

10
src/Rtp/RtpSender.cpp
View File

@@ -25,6 +25,10 @@ RtpSender::RtpSender(EventPoller::Ptr poller) {
_socket_rtp = Socket::createSocket(_poller, false);
}
RtpSender::~RtpSender() {
flush();
}
void RtpSender::startSend(const MediaSourceEvent::SendRtpArgs &args, const function<void(uint16_t local_port, const SockException &ex)> &cb){
_args = args;
if (!_interface) {
@@ -231,6 +235,12 @@ void RtpSender::resetTracks(){
_interface->resetTracks();
}
void RtpSender::flush() {
if (_interface) {
_interface->flush();
}
}
//此函数在其他线程执行
bool RtpSender::inputFrame(const Frame::Ptr &frame) {
//连接成功后才做实质操作(节省cpu资源)

9
src/Rtp/RtpSender.h
View File

@@ -18,12 +18,12 @@
namespace mediakit{
//rtp发送客户端，支持发送GB28181协议
class RtpSender : public MediaSinkInterface, public std::enable_shared_from_this<RtpSender>{
class RtpSender final : public MediaSinkInterface, public std::enable_shared_from_this<RtpSender>{
public:
typedef std::shared_ptr<RtpSender> Ptr;
RtpSender(toolkit::EventPoller::Ptr poller = nullptr);
~RtpSender() override = default;
~RtpSender() override;
/**
* 开始发送ps-rtp包
@@ -37,6 +37,11 @@ public:
*/
bool inputFrame(const Frame::Ptr &frame) override;
/**
* 刷新输出frame缓存
*/
void flush() override;
/**
* 添加track，内部会调用Track的clone方法
* 只会克隆sps pps这些信息 ，而不会克隆Delegate相关关系