amass/ZLMediaKit
1
0
Fork 0
mirror of https://github.com/ZLMediaKit/ZLMediaKit.git synced 2024-11-23 03:10:04 +08:00
Code Issues Actions 6 Packages Projects Releases Wiki Activity

解决单Track时，等待时间太长的问题

Browse Source
This commit is contained in:
xiongziliang 2020-11-29 09:38:04 +08:00
parent a5b4da7db5
commit a64ce68592
8 changed files with 112 additions and 74 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

117
src/Rtp/Decoder.cpp
View File

@ -64,6 +64,9 @@ DecoderImp::DecoderImp(const Decoder::Ptr &decoder, MediaSinkInterface *sink){
_decoder->setOnDecode([this](int stream, int codecid, int flags, int64_t pts, int64_t dts, const void *data, int bytes) { _decoder->setOnDecode([this](int stream, int codecid, int flags, int64_t pts, int64_t dts, const void *data, int bytes) {
onDecode(stream, codecid, flags, pts, dts, data, bytes); onDecode(stream, codecid, flags, pts, dts, data, bytes);
}); });
_decoder->setOnStream([this](int stream, int codecid, const void *extra, int bytes, int finish) {
onStream(stream, codecid, extra, bytes, finish);
});
} }
#if defined(ENABLE_RTPPROXY) || defined(ENABLE_HLS) #if defined(ENABLE_RTPPROXY) || defined(ENABLE_HLS)
@ -115,25 +118,65 @@ void FrameMerger::inputFrame(const Frame::Ptr &frame,const function<void(uint32_
_frameCached.emplace_back(Frame::getCacheAbleFrame(frame)); _frameCached.emplace_back(Frame::getCacheAbleFrame(frame));
} }
void DecoderImp::onStream(int stream, int codecid, const void *extra, int bytes, int finish){
switch (codecid) {
case PSI_STREAM_H264: {
InfoL << "got video track: H264";
auto track = std::make_shared<H264Track>();
onTrack(track);
break;
}
case PSI_STREAM_H265: {
InfoL << "got video track: H265";
auto track = std::make_shared<H265Track>();
onTrack(track);
break;
}
case PSI_STREAM_AAC: {
InfoL<< "got audio track: AAC";
auto track = std::make_shared<AACTrack>();
onTrack(track);
break;
}
case PSI_STREAM_AUDIO_G711A:
case PSI_STREAM_AUDIO_G711U: {
auto codec = codecid == PSI_STREAM_AUDIO_G711A ? CodecG711A : CodecG711U;
InfoL << "got audio track: G711";
//G711传统只支持 8000/1/16的规格，FFmpeg貌似做了扩展，但是这里不管它了
auto track = std::make_shared<G711Track>(codec, 8000, 1, 16);
onTrack(track);
break;
}
case PSI_STREAM_AUDIO_OPUS: {
InfoL << "got audio track: opus";
auto track = std::make_shared<OpusTrack>();
onTrack(track);
break;
}
default:
if(codecid != 0){
WarnL<< "unsupported codec type:" << getCodecName(codecid) << " " << (int)codecid;
}
break;
}
if (finish) {
_sink->addTrackCompleted();
InfoL << "add track finished";
}
}
void DecoderImp::onDecode(int stream,int codecid,int flags,int64_t pts,int64_t dts,const void *data,int bytes) { void DecoderImp::onDecode(int stream,int codecid,int flags,int64_t pts,int64_t dts,const void *data,int bytes) {
pts /= 90; pts /= 90;
dts /= 90; dts /= 90;
switch (codecid) { switch (codecid) {
case PSI_STREAM_H264: { case PSI_STREAM_H264: {
if (!_codecid_video) {
//获取到视频
_codecid_video = codecid;
InfoL<< "got video track: H264";
auto track = std::make_shared<H264Track>();
onTrack(track);
}
if (codecid != _codecid_video) {
WarnL<< "video track change to H264 from codecid:" << getCodecName(_codecid_video);
return;
}
auto frame = std::make_shared<H264FrameNoCacheAble>((char *) data, bytes, dts, pts,0); auto frame = std::make_shared<H264FrameNoCacheAble>((char *) data, bytes, dts, pts,0);
_merger.inputFrame(frame,[this](uint32_t dts, uint32_t pts, const Buffer::Ptr &buffer) { _merger.inputFrame(frame,[this](uint32_t dts, uint32_t pts, const Buffer::Ptr &buffer) {
onFrame(std::make_shared<FrameWrapper<H264FrameNoCacheAble> >(buffer, dts, pts, prefixSize(buffer->data(), buffer->size()), 0)); onFrame(std::make_shared<FrameWrapper<H264FrameNoCacheAble> >(buffer, dts, pts, prefixSize(buffer->data(), buffer->size()), 0));
@ -142,17 +185,6 @@ void DecoderImp::onDecode(int stream,int codecid,int flags,int64_t pts,int64_t d
} }
case PSI_STREAM_H265: { case PSI_STREAM_H265: {
if (!_codecid_video) {
//获取到视频
_codecid_video = codecid;
InfoL<< "got video track: H265";
auto track = std::make_shared<H265Track>();
onTrack(track);
}
if (codecid != _codecid_video) {
WarnL<< "video track change to H265 from codecid:" << getCodecName(_codecid_video);
return;
}
auto frame = std::make_shared<H265FrameNoCacheAble>((char *) data, bytes, dts, pts, 0); auto frame = std::make_shared<H265FrameNoCacheAble>((char *) data, bytes, dts, pts, 0);
_merger.inputFrame(frame,[this](uint32_t dts, uint32_t pts, const Buffer::Ptr &buffer) { _merger.inputFrame(frame,[this](uint32_t dts, uint32_t pts, const Buffer::Ptr &buffer) {
onFrame(std::make_shared<FrameWrapper<H265FrameNoCacheAble> >(buffer, dts, pts, prefixSize(buffer->data(), buffer->size()), 0)); onFrame(std::make_shared<FrameWrapper<H265FrameNoCacheAble> >(buffer, dts, pts, prefixSize(buffer->data(), buffer->size()), 0));
@ -166,18 +198,6 @@ void DecoderImp::onDecode(int stream,int codecid,int flags,int64_t pts,int64_t d
//这不是aac //这不是aac
break; break;
} }
if (!_codecid_audio) {
//获取到音频
_codecid_audio = codecid;
InfoL<< "got audio track: AAC";
auto track = std::make_shared<AACTrack>();
onTrack(track);
}
if (codecid != _codecid_audio) {
WarnL<< "audio track change to AAC from codecid:" << getCodecName(_codecid_audio);
return;
}
onFrame(std::make_shared<FrameFromPtr>(CodecAAC, (char *) data, bytes, dts, 0, ADTS_HEADER_LEN)); onFrame(std::make_shared<FrameFromPtr>(CodecAAC, (char *) data, bytes, dts, 0, ADTS_HEADER_LEN));
break; break;
} }
@ -185,36 +205,11 @@ void DecoderImp::onDecode(int stream,int codecid,int flags,int64_t pts,int64_t d
case PSI_STREAM_AUDIO_G711A: case PSI_STREAM_AUDIO_G711A:
case PSI_STREAM_AUDIO_G711U: { case PSI_STREAM_AUDIO_G711U: {
auto codec = codecid == PSI_STREAM_AUDIO_G711A ? CodecG711A : CodecG711U; auto codec = codecid == PSI_STREAM_AUDIO_G711A ? CodecG711A : CodecG711U;
if (!_codecid_audio) {
//获取到音频
_codecid_audio = codecid;
InfoL<< "got audio track: G711";
//G711传统只支持 8000/1/16的规格，FFmpeg貌似做了扩展，但是这里不管它了
auto track = std::make_shared<G711Track>(codec, 8000, 1, 16);
onTrack(track);
}
if (codecid != _codecid_audio) {
WarnL<< "audio track change to G711 from codecid:" << getCodecName(_codecid_audio);
return;
}
onFrame(std::make_shared<FrameFromPtr>(codec, (char *) data, bytes, dts)); onFrame(std::make_shared<FrameFromPtr>(codec, (char *) data, bytes, dts));
break; break;
} }
case PSI_STREAM_AUDIO_OPUS: { case PSI_STREAM_AUDIO_OPUS: {
if (!_codecid_audio) {
//获取到音频
_codecid_audio = codecid;
InfoL << "got audio track: opus";
auto track = std::make_shared<OpusTrack>();
onTrack(track);
}
if (codecid != _codecid_audio) {
WarnL << "audio track change to opus from codecid:" << getCodecName(_codecid_audio);
return;
}
onFrame(std::make_shared<FrameFromPtr>(CodecOpus, (char *) data, bytes, dts)); onFrame(std::make_shared<FrameFromPtr>(CodecOpus, (char *) data, bytes, dts));
break; break;
} }

9
src/Rtp/Decoder.h
View File

@ -23,8 +23,12 @@ class Decoder {
public: public:
typedef std::shared_ptr<Decoder> Ptr; typedef std::shared_ptr<Decoder> Ptr;
typedef std::function<void(int stream, int codecid, int flags, int64_t pts, int64_t dts, const void *data, int bytes)> onDecode; typedef std::function<void(int stream, int codecid, int flags, int64_t pts, int64_t dts, const void *data, int bytes)> onDecode;
typedef std::function<void(int stream, int codecid, const void *extra, int bytes, int finish)> onStream;
virtual int input(const uint8_t *data, int bytes) = 0; virtual int input(const uint8_t *data, int bytes) = 0;
virtual void setOnDecode(const onDecode &decode) = 0; virtual void setOnDecode(onDecode cb) = 0;
virtual void setOnStream(onStream cb) = 0;
protected: protected:
Decoder() = default; Decoder() = default;
virtual ~Decoder() = default; virtual ~Decoder() = default;
@ -62,13 +66,12 @@ protected:
private: private:
DecoderImp(const Decoder::Ptr &decoder, MediaSinkInterface *sink); DecoderImp(const Decoder::Ptr &decoder, MediaSinkInterface *sink);
void onDecode(int stream, int codecid, int flags, int64_t pts, int64_t dts, const void *data, int bytes); void onDecode(int stream, int codecid, int flags, int64_t pts, int64_t dts, const void *data, int bytes);
void onStream(int stream, int codecid, const void *extra, int bytes, int finish);
private: private:
Decoder::Ptr _decoder; Decoder::Ptr _decoder;
MediaSinkInterface *_sink; MediaSinkInterface *_sink;
FrameMerger _merger; FrameMerger _merger;
int _codecid_video = 0;
int _codecid_audio = 0;
}; };
}//namespace mediakit }//namespace mediakit

18
src/Rtp/PSDecoder.cpp
View File

@ -28,6 +28,16 @@ PSDecoder::PSDecoder() {
} }
return 0; return 0;
},this); },this);
ps_demuxer_notify_t notify = {
[](void *param, int stream, int codecid, const void *extra, int bytes, int finish) {
PSDecoder *thiz = (PSDecoder *) param;
if (thiz->_on_stream) {
thiz->_on_stream(stream, codecid, extra, bytes, finish);
}
}
};
ps_demuxer_set_notify((struct ps_demuxer_t *) _ps_demuxer, &notify, this);
} }
PSDecoder::~PSDecoder() { PSDecoder::~PSDecoder() {
@ -38,8 +48,12 @@ int PSDecoder::input(const uint8_t *data, int bytes) {
return ps_demuxer_input((struct ps_demuxer_t*)_ps_demuxer,data,bytes); return ps_demuxer_input((struct ps_demuxer_t*)_ps_demuxer,data,bytes);
} }
void PSDecoder::setOnDecode(const Decoder::onDecode &decode) { void PSDecoder::setOnDecode(Decoder::onDecode cb) {
_on_decode = decode; _on_decode = std::move(cb);
}
void PSDecoder::setOnStream(Decoder::onStream cb) {
_on_stream = std::move(cb);
} }
}//namespace mediakit }//namespace mediakit

5
src/Rtp/PSDecoder.h
View File

@ -22,10 +22,13 @@ public:
PSDecoder(); PSDecoder();
~PSDecoder(); ~PSDecoder();
int input(const uint8_t* data, int bytes) override; int input(const uint8_t* data, int bytes) override;
void setOnDecode(const onDecode &decode) override; void setOnDecode(onDecode cb) override;
void setOnStream(onStream cb) override;
private: private:
void *_ps_demuxer = nullptr; void *_ps_demuxer = nullptr;
onDecode _on_decode; onDecode _on_decode;
onStream _on_stream;
}; };
}//namespace mediakit }//namespace mediakit

4
src/Rtp/RtpProcess.cpp
View File

@ -129,6 +129,10 @@ void RtpProcess::addTrack(const Track::Ptr &track) {
_muxer->addTrack(track); _muxer->addTrack(track);
} }
void RtpProcess::addTrackCompleted() {
_muxer->addTrackCompleted();
}
bool RtpProcess::alive() { bool RtpProcess::alive() {
GET_CONFIG(int, timeoutSec, RtpProxy::kTimeoutSec) GET_CONFIG(int, timeoutSec, RtpProxy::kTimeoutSec)
if (_last_frame_time.elapsedTime() / 1000 < timeoutSec) { if (_last_frame_time.elapsedTime() / 1000 < timeoutSec) {

1
src/Rtp/RtpProcess.h
View File

@ -66,6 +66,7 @@ public:
protected: protected:
void inputFrame(const Frame::Ptr &frame) override; void inputFrame(const Frame::Ptr &frame) override;
void addTrack(const Track::Ptr & track) override; void addTrack(const Track::Ptr & track) override;
void addTrackCompleted() override;
void resetTracks() override {}; void resetTracks() override {};
//// MediaSourceEvent override //// //// MediaSourceEvent override ////

19
src/Rtp/TSDecoder.cpp
View File

@ -61,6 +61,16 @@ TSDecoder::TSDecoder() : _ts_segment() {
} }
return 0; return 0;
},this); },this);
ts_demuxer_notify_t notify = {
[](void *param, int stream, int codecid, const void *extra, int bytes, int finish) {
TSDecoder *thiz = (TSDecoder *) param;
if (thiz->_on_stream) {
thiz->_on_stream(stream, codecid, extra, bytes, finish);
}
}
};
ts_demuxer_set_notify((struct ts_demuxer_t *) _demuxer_ctx, &notify, this);
} }
TSDecoder::~TSDecoder() { TSDecoder::~TSDecoder() {
@ -75,9 +85,14 @@ int TSDecoder::input(const uint8_t *data, int bytes) {
return bytes; return bytes;
} }
void TSDecoder::setOnDecode(const Decoder::onDecode &decode) { void TSDecoder::setOnDecode(Decoder::onDecode cb) {
_on_decode = decode; _on_decode = std::move(cb);
} }
void TSDecoder::setOnStream(Decoder::onStream cb) {
_on_stream = std::move(cb);
}
#endif//defined(ENABLE_HLS) #endif//defined(ENABLE_HLS)
}//namespace mediakit }//namespace mediakit

5
src/Rtp/TSDecoder.h
View File

@ -44,11 +44,14 @@ public:
TSDecoder(); TSDecoder();
~TSDecoder(); ~TSDecoder();
int input(const uint8_t* data, int bytes) override ; int input(const uint8_t* data, int bytes) override ;
void setOnDecode(const onDecode &decode) override; void setOnDecode(onDecode cb) override;
void setOnStream(onStream cb) override;
private: private:
TSSegment _ts_segment; TSSegment _ts_segment;
struct ts_demuxer_t* _demuxer_ctx = nullptr; struct ts_demuxer_t* _demuxer_ctx = nullptr;
onDecode _on_decode; onDecode _on_decode;
onStream _on_stream;
}; };
#endif//defined(ENABLE_HLS) #endif//defined(ENABLE_HLS)