Skip to content

Z
ZLMediaKit
Commit 198f223d by xiongziliang
Options

进一步抽象ts/ps解析代码

parent cf599167
正在显示 包含 283 行增加 和 210 行删除
src/Rtp/Decoder.cpp
...@@ -8,18 +8,203 @@ ...@@ -8,18 +8,203 @@
* may be found in the AUTHORS file in the root of the source tree. * may be found in the AUTHORS file in the root of the source tree.
*/ */
#if defined(ENABLE_RTPPROXY)
#include "Decoder.h" #include "Decoder.h"
#include "PSDecoder.h" #include "PSDecoder.h"
#include "TSDecoder.h" #include "TSDecoder.h"
#include "mpeg-ts-proto.h"
#include "Extension/H264.h"
#include "Extension/H265.h"
#include "Extension/AAC.h"
#include "Extension/G711.h"
namespace mediakit { namespace mediakit {
Decoder::Ptr Decoder::createDecoder(Decoder::Type type) { static Decoder::Ptr createDecoder_l(DecoderImp::Type type) {
switch (type){ switch (type){
case decoder_ps : return std::make_shared<PSDecoder>(); case DecoderImp::decoder_ps:
case decoder_ts : return std::make_shared<TSDecoder>(); #ifdef ENABLE_RTPPROXY
default : return nullptr; return std::make_shared<PSDecoder>();
#else
WarnL << "创建ps解复用器失败，请打开ENABLE_RTPPROXY然后重新编译";
return nullptr;
#endif//ENABLE_RTPPROXY
case DecoderImp::decoder_ts:
#ifdef ENABLE_HLS
return std::make_shared<TSDecoder>();
#else
WarnL << "创建mpegts解复用器失败，请打开ENABLE_HLS然后重新编译";
return nullptr;
#endif//ENABLE_HLS
default: return nullptr;
}
}
/////////////////////////////////////////////////////////////
DecoderImp::Ptr DecoderImp::createDecoder(Type type, const MediaSinkInterface::Ptr &sink){
auto decoder = createDecoder_l(type);
if(!decoder){
return nullptr;
}
return DecoderImp::Ptr(new DecoderImp(decoder, sink));
}
int DecoderImp::input(const uint8_t *data, int bytes){
return _decoder->input(data, bytes);
}
DecoderImp::DecoderImp(const Decoder::Ptr &decoder,const MediaSinkInterface::Ptr &sink){
_decoder = decoder;
_sink = sink;
_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);
});
}
#define SWITCH_CASE(codec_id) case codec_id : return #codec_id
static const char *getCodecName(int codec_id) {
switch (codec_id) {
SWITCH_CASE(PSI_STREAM_MPEG1);
SWITCH_CASE(PSI_STREAM_MPEG2);
SWITCH_CASE(PSI_STREAM_AUDIO_MPEG1);
SWITCH_CASE(PSI_STREAM_MP3);
SWITCH_CASE(PSI_STREAM_AAC);
SWITCH_CASE(PSI_STREAM_MPEG4);
SWITCH_CASE(PSI_STREAM_MPEG4_AAC_LATM);
SWITCH_CASE(PSI_STREAM_H264);
SWITCH_CASE(PSI_STREAM_MPEG4_AAC);
SWITCH_CASE(PSI_STREAM_H265);
SWITCH_CASE(PSI_STREAM_AUDIO_AC3);
SWITCH_CASE(PSI_STREAM_AUDIO_EAC3);
SWITCH_CASE(PSI_STREAM_AUDIO_DTS);
SWITCH_CASE(PSI_STREAM_VIDEO_DIRAC);
SWITCH_CASE(PSI_STREAM_VIDEO_VC1);
SWITCH_CASE(PSI_STREAM_VIDEO_SVAC);
SWITCH_CASE(PSI_STREAM_AUDIO_SVAC);
SWITCH_CASE(PSI_STREAM_AUDIO_G711A);
SWITCH_CASE(PSI_STREAM_AUDIO_G711U);
SWITCH_CASE(PSI_STREAM_AUDIO_G722);
SWITCH_CASE(PSI_STREAM_AUDIO_G723);
SWITCH_CASE(PSI_STREAM_AUDIO_G729);
default : return "unknown codec";
}
}
void FrameMerger::inputFrame(const Frame::Ptr &frame,const function<void(uint32_t dts,uint32_t pts,const Buffer::Ptr &buffer)> &cb){
if (!_frameCached.empty() && _frameCached.back()->dts() != frame->dts()) {
Frame::Ptr back = _frameCached.back();
Buffer::Ptr merged_frame = back;
if(_frameCached.size() != 1){
string merged;
_frameCached.for_each([&](const Frame::Ptr &frame){
merged.append(frame->data(),frame->size());
});
merged_frame = std::make_shared<BufferString>(std::move(merged));
}
cb(back->dts(),back->pts(),merged_frame);
_frameCached.clear();
}
_frameCached.emplace_back(Frame::getCacheAbleFrame(frame));
}
void DecoderImp::onDecode(int stream,int codecid,int flags,int64_t pts,int64_t dts,const void *data,int bytes) {
pts /= 90;
dts /= 90;
switch (codecid) {
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);
_merger.inputFrame(frame,[this](uint32_t dts, uint32_t pts, const Buffer::Ptr &buffer) {
onFrame(std::make_shared<H264FrameNoCacheAble>(buffer->data(), buffer->size(), dts, pts,4));
});
break;
}
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);
_merger.inputFrame(frame,[this](uint32_t dts, uint32_t pts, const Buffer::Ptr &buffer) {
onFrame(std::make_shared<H265FrameNoCacheAble>(buffer->data(), buffer->size(), dts, pts, 4));
});
break;
}
case PSI_STREAM_AAC: {
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<AACFrameNoCacheAble>((char *) data, bytes, dts, 0, 7));
break;
}
case PSI_STREAM_AUDIO_G711A:
case PSI_STREAM_AUDIO_G711U: {
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;
}
auto frame = std::make_shared<G711FrameNoCacheAble>((char *) data, bytes, dts);
frame->setCodec(codec);
onFrame(frame);
break;
}
default:
if(codecid != 0){
WarnL<< "unsupported codec type:" << getCodecName(codecid) << " " << (int)codecid;
}
break;
} }
} }
void DecoderImp::onTrack(const Track::Ptr &track) {
_sink->addTrack(track);
}
void DecoderImp::onFrame(const Frame::Ptr &frame) {
_sink->inputFrame(frame);
}
}//namespace mediakit }//namespace mediakit
#endif//defined(ENABLE_RTPPROXY)
src/Rtp/Decoder.h
...@@ -11,31 +11,66 @@ ...@@ -11,31 +11,66 @@
#ifndef ZLMEDIAKIT_DECODER_H #ifndef ZLMEDIAKIT_DECODER_H
#define ZLMEDIAKIT_DECODER_H #define ZLMEDIAKIT_DECODER_H
#if defined(ENABLE_RTPPROXY)
#include <stdint.h> #include <stdint.h>
#include <memory> #include <memory>
#include <functional> #include <functional>
#include "Decoder.h" #include "Decoder.h"
#include "Common/MediaSink.h"
using namespace std; using namespace std;
namespace mediakit { namespace mediakit {
class Decoder { class Decoder {
public: public:
typedef std::shared_ptr<Decoder> Ptr; typedef std::shared_ptr<Decoder> Ptr;
typedef enum {
decoder_ts = 0,
decoder_ps
}Type;
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;
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(const onDecode &decode) = 0;
static Ptr createDecoder(Type type);
protected: protected:
Decoder() = default; Decoder() = default;
virtual ~Decoder() = default; virtual ~Decoder() = default;
}; };
/**
* 合并一些时间戳相同的frame
*/
class FrameMerger {
public:
FrameMerger() = default;
~FrameMerger() = default;
void inputFrame(const Frame::Ptr &frame,const function<void(uint32_t dts,uint32_t pts,const Buffer::Ptr &buffer)> &cb);
private:
List<Frame::Ptr> _frameCached;
};
class DecoderImp{
public:
typedef enum {
decoder_ts = 0,
decoder_ps
}Type;
typedef std::shared_ptr<DecoderImp> Ptr;
~DecoderImp() = default;
static Ptr createDecoder(Type type, const MediaSinkInterface::Ptr &sink);
int input(const uint8_t *data, int bytes);
protected:
void onTrack(const Track::Ptr &track);
void onFrame(const Frame::Ptr &frame);
private:
DecoderImp(const Decoder::Ptr &decoder, const MediaSinkInterface::Ptr &sink);
void onDecode(int stream,int codecid,int flags,int64_t pts,int64_t dts,const void *data,int bytes);
private:
Decoder::Ptr _decoder;
MediaSinkInterface::Ptr _sink;
FrameMerger _merger;
int _codecid_video = 0;
int _codecid_audio = 0;
};
}//namespace mediakit }//namespace mediakit
#endif//defined(ENABLE_RTPPROXY)
#endif //ZLMEDIAKIT_DECODER_H #endif //ZLMEDIAKIT_DECODER_H
src/Rtp/RtpProcess.cpp
...@@ -9,44 +9,13 @@ ...@@ -9,44 +9,13 @@
*/ */
#if defined(ENABLE_RTPPROXY) #if defined(ENABLE_RTPPROXY)
#include "mpeg-ts-proto.h"
#include "RtpProcess.h" #include "RtpProcess.h"
#include "Util/File.h" #include "Util/File.h"
#include "Extension/H265.h" #include "Http/HttpTSPlayer.h"
#include "Extension/AAC.h"
#include "Extension/G711.h"
#define RTP_APP_NAME "rtp" #define RTP_APP_NAME "rtp"
namespace mediakit{ namespace mediakit{
/**
* 合并一些时间戳相同的frame
*/
class FrameMerger {
public:
FrameMerger() = default;
virtual ~FrameMerger() = default;
void inputFrame(const Frame::Ptr &frame,const function<void(uint32_t dts,uint32_t pts,const Buffer::Ptr &buffer)> &cb){
if (!_frameCached.empty() && _frameCached.back()->dts() != frame->dts()) {
Frame::Ptr back = _frameCached.back();
Buffer::Ptr merged_frame = back;
if(_frameCached.size() != 1){
string merged;
_frameCached.for_each([&](const Frame::Ptr &frame){
merged.append(frame->data(),frame->size());
});
merged_frame = std::make_shared<BufferString>(std::move(merged));
}
cb(back->dts(),back->pts(),merged_frame);
_frameCached.clear();
}
_frameCached.emplace_back(Frame::getCacheAbleFrame(frame));
}
private:
List<Frame::Ptr> _frameCached;
};
string printSSRC(uint32_t ui32Ssrc) { string printSSRC(uint32_t ui32Ssrc) {
char tmp[9] = { 0 }; char tmp[9] = { 0 };
ui32Ssrc = htonl(ui32Ssrc); ui32Ssrc = htonl(ui32Ssrc);
...@@ -101,7 +70,6 @@ RtpProcess::RtpProcess(uint32_t ssrc) { ...@@ -101,7 +70,6 @@ RtpProcess::RtpProcess(uint32_t ssrc) {
}); });
} }
} }
_merger = std::make_shared<FrameMerger>();
} }
RtpProcess::~RtpProcess() { RtpProcess::~RtpProcess() {
...@@ -157,7 +125,7 @@ bool RtpProcess::inputRtp(const Socket::Ptr &sock, const char *data, int data_le ...@@ -157,7 +125,7 @@ bool RtpProcess::inputRtp(const Socket::Ptr &sock, const char *data, int data_le
//判断是否为ts负载 //判断是否为ts负载
static inline bool checkTS(const uint8_t *packet, int bytes){ static inline bool checkTS(const uint8_t *packet, int bytes){
return bytes % 188 == 0 && packet[0] == 0x47; return bytes % TS_PACKET_SIZE == 0 && packet[0] == TS_SYNC_BYTE;
} }
void RtpProcess::onRtpSorted(const RtpPacket::Ptr &rtp, int) { void RtpProcess::onRtpSorted(const RtpPacket::Ptr &rtp, int) {
...@@ -179,153 +147,37 @@ void RtpProcess::onRtpDecode(const uint8_t *packet, int bytes, uint32_t timestam ...@@ -179,153 +147,37 @@ void RtpProcess::onRtpDecode(const uint8_t *packet, int bytes, uint32_t timestam
fwrite((uint8_t *)packet,bytes, 1, _save_file_ps.get()); fwrite((uint8_t *)packet,bytes, 1, _save_file_ps.get());
} }
if(!_decoder){ if (!_decoder) {
//创建解码器 //创建解码器
if(checkTS(packet, bytes)){ if (checkTS(packet, bytes)) {
//猜测是ts负载 //猜测是ts负载
InfoP(this) << "judged to be TS"; InfoP(this) << "judged to be TS";
_decoder = Decoder::createDecoder(Decoder::decoder_ts); _decoder = DecoderImp::createDecoder(DecoderImp::decoder_ts, shared_from_this());
}else{ } else {
//猜测是ps负载 //猜测是ps负载
InfoP(this) << "judged to be PS"; InfoP(this) << "judged to be PS";
_decoder = Decoder::createDecoder(Decoder::decoder_ps); _decoder = DecoderImp::createDecoder(DecoderImp::decoder_ps, shared_from_this());
} }
_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);
});
} }
auto ret = _decoder->input((uint8_t *)packet,bytes); if (_decoder) {
if(ret != bytes){ auto ret = _decoder->input((uint8_t *) packet, bytes);
WarnP(this) << ret << " != " << bytes << " " << flags; if (ret != bytes) {
WarnP(this) << ret << " != " << bytes << " " << flags;
}
} }
} }
#define SWITCH_CASE(codec_id) case codec_id : return #codec_id void RtpProcess::inputFrame(const Frame::Ptr &frame){
static const char *getCodecName(int codec_id) { _dts = frame->dts();
switch (codec_id) { if (_save_file_video && frame->getTrackType() == TrackVideo) {
SWITCH_CASE(PSI_STREAM_MPEG1); fwrite((uint8_t *) frame->data(), frame->size(), 1, _save_file_video.get());
SWITCH_CASE(PSI_STREAM_MPEG2);
SWITCH_CASE(PSI_STREAM_AUDIO_MPEG1);
SWITCH_CASE(PSI_STREAM_MP3);
SWITCH_CASE(PSI_STREAM_AAC);
SWITCH_CASE(PSI_STREAM_MPEG4);
SWITCH_CASE(PSI_STREAM_MPEG4_AAC_LATM);
SWITCH_CASE(PSI_STREAM_H264);
SWITCH_CASE(PSI_STREAM_MPEG4_AAC);
SWITCH_CASE(PSI_STREAM_H265);
SWITCH_CASE(PSI_STREAM_AUDIO_AC3);
SWITCH_CASE(PSI_STREAM_AUDIO_EAC3);
SWITCH_CASE(PSI_STREAM_AUDIO_DTS);
SWITCH_CASE(PSI_STREAM_VIDEO_DIRAC);
SWITCH_CASE(PSI_STREAM_VIDEO_VC1);
SWITCH_CASE(PSI_STREAM_VIDEO_SVAC);
SWITCH_CASE(PSI_STREAM_AUDIO_SVAC);
SWITCH_CASE(PSI_STREAM_AUDIO_G711A);
SWITCH_CASE(PSI_STREAM_AUDIO_G711U);
SWITCH_CASE(PSI_STREAM_AUDIO_G722);
SWITCH_CASE(PSI_STREAM_AUDIO_G723);
SWITCH_CASE(PSI_STREAM_AUDIO_G729);
default : return "unknown codec";
} }
_muxer->inputFrame(frame);
} }
void RtpProcess::onDecode(int stream,int codecid,int flags,int64_t pts,int64_t dts,const void *data,int bytes) { void RtpProcess::addTrack(const Track::Ptr & track){
pts /= 90; _muxer->addTrack(track);
dts /= 90;
_dts = dts;
switch (codecid) {
case PSI_STREAM_H264: {
if (!_codecid_video) {
//获取到视频
_codecid_video = codecid;
InfoP(this) << "got video track: H264";
auto track = std::make_shared<H264Track>();
_muxer->addTrack(track);
}
if (codecid != _codecid_video) {
WarnP(this) << "video track change to H264 from codecid:" << getCodecName(_codecid_video);
return;
}
if(_save_file_video){
fwrite((uint8_t *)data,bytes, 1, _save_file_video.get());
}
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) {
_muxer->inputFrame(std::make_shared<H264FrameNoCacheAble>(buffer->data(), buffer->size(), dts, pts,4));
});
break;
}
case PSI_STREAM_H265: {
if (!_codecid_video) {
//获取到视频
_codecid_video = codecid;
InfoP(this) << "got video track: H265";
auto track = std::make_shared<H265Track>();
_muxer->addTrack(track);
}
if (codecid != _codecid_video) {
WarnP(this) << "video track change to H265 from codecid:" << getCodecName(_codecid_video);
return;
}
if(_save_file_video){
fwrite((uint8_t *)data,bytes, 1, _save_file_video.get());
}
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) {
_muxer->inputFrame(std::make_shared<H265FrameNoCacheAble>(buffer->data(), buffer->size(), dts, pts, 4));
});
break;
}
case PSI_STREAM_AAC: {
if (!_codecid_audio) {
//获取到音频
_codecid_audio = codecid;
InfoP(this) << "got audio track: AAC";
auto track = std::make_shared<AACTrack>();
_muxer->addTrack(track);
}
if (codecid != _codecid_audio) {
WarnP(this) << "audio track change to AAC from codecid:" << getCodecName(_codecid_audio);
return;
}
_muxer->inputFrame(std::make_shared<AACFrameNoCacheAble>((char *) data, bytes, dts, 0, 7));
break;
}
case PSI_STREAM_AUDIO_G711A:
case PSI_STREAM_AUDIO_G711U: {
auto codec = codecid == PSI_STREAM_AUDIO_G711A ? CodecG711A : CodecG711U;
if (!_codecid_audio) {
//获取到音频
_codecid_audio = codecid;
InfoP(this) << "got audio track: G711";
//G711传统只支持 8000/1/16的规格，FFmpeg貌似做了扩展，但是这里不管它了
auto track = std::make_shared<G711Track>(codec, 8000, 1, 16);
_muxer->addTrack(track);
}
if (codecid != _codecid_audio) {
WarnP(this) << "audio track change to G711 from codecid:" << getCodecName(_codecid_audio);
return;
}
auto frame = std::make_shared<G711FrameNoCacheAble>((char *) data, bytes, dts);
frame->setCodec(codec);
_muxer->inputFrame(frame);
break;
}
default:
if(codecid != 0){
WarnP(this) << "unsupported codec type:" << getCodecName(codecid) << " " << (int)codecid;
}
return;
}
} }
bool RtpProcess::alive() { bool RtpProcess::alive() {
...@@ -410,6 +262,5 @@ void RtpProcess::emitOnPublish() { ...@@ -410,6 +262,5 @@ void RtpProcess::emitOnPublish() {
} }
} }
}//namespace mediakit }//namespace mediakit
#endif//defined(ENABLE_RTPPROXY) #endif//defined(ENABLE_RTPPROXY)
\ No newline at end of file
src/Rtp/RtpProcess.h
...@@ -23,8 +23,7 @@ using namespace mediakit; ...@@ -23,8 +23,7 @@ using namespace mediakit;
namespace mediakit{ namespace mediakit{
string printSSRC(uint32_t ui32Ssrc); string printSSRC(uint32_t ui32Ssrc);
class FrameMerger; class RtpProcess : public RtpReceiver , public RtpDecoder, public SockInfo, public MediaSinkInterface, public std::enable_shared_from_this<RtpProcess>{
class RtpProcess : public RtpReceiver , public RtpDecoder, public SockInfo, public std::enable_shared_from_this<RtpProcess>{
public: public:
typedef std::shared_ptr<RtpProcess> Ptr; typedef std::shared_ptr<RtpProcess> Ptr;
RtpProcess(uint32_t ssrc); RtpProcess(uint32_t ssrc);
...@@ -44,7 +43,9 @@ public: ...@@ -44,7 +43,9 @@ public:
protected: protected:
void onRtpSorted(const RtpPacket::Ptr &rtp, int track_index) override ; void onRtpSorted(const RtpPacket::Ptr &rtp, int track_index) override ;
void onRtpDecode(const uint8_t *packet, int bytes, uint32_t timestamp, int flags) override; void onRtpDecode(const uint8_t *packet, int bytes, uint32_t timestamp, int flags) override;
void onDecode(int stream,int codecid,int flags,int64_t pts,int64_t dts, const void *data,int bytes); void inputFrame(const Frame::Ptr &frame) override;
void addTrack(const Track::Ptr & track) override;
void resetTracks() override {};
private: private:
void emitOnPublish(); void emitOnPublish();
...@@ -57,13 +58,10 @@ private: ...@@ -57,13 +58,10 @@ private:
SdpTrack::Ptr _track; SdpTrack::Ptr _track;
struct sockaddr *_addr = nullptr; struct sockaddr *_addr = nullptr;
uint16_t _sequence = 0; uint16_t _sequence = 0;
int _codecid_video = 0;
int _codecid_audio = 0;
MultiMediaSourceMuxer::Ptr _muxer; MultiMediaSourceMuxer::Ptr _muxer;
std::shared_ptr<FrameMerger> _merger;
Ticker _last_rtp_time; Ticker _last_rtp_time;
uint32_t _dts = 0; uint32_t _dts = 0;
Decoder::Ptr _decoder; DecoderImp::Ptr _decoder;
std::weak_ptr<MediaSourceEvent> _listener; std::weak_ptr<MediaSourceEvent> _listener;
MediaInfo _media_info; MediaInfo _media_info;
uint64_t _total_bytes = 0; uint64_t _total_bytes = 0;
......
src/Rtp/TSDecoder.cpp
...@@ -8,34 +8,38 @@ ...@@ -8,34 +8,38 @@
* may be found in the AUTHORS file in the root of the source tree. * may be found in the AUTHORS file in the root of the source tree.
*/ */
#if defined(ENABLE_RTPPROXY)
#include "mpeg-ts.h"
#include "TSDecoder.h" #include "TSDecoder.h"
#define TS_PACKET_SIZE 188
namespace mediakit { namespace mediakit {
bool TSSegment::isTSPacket(const char *data, int len){
return len == TS_PACKET_SIZE && ((uint8_t*)data)[0] == TS_SYNC_BYTE;
}
void TSSegment::setOnSegment(const TSSegment::onSegment &cb) { void TSSegment::setOnSegment(const TSSegment::onSegment &cb) {
_onSegment = cb; _onSegment = cb;
} }
int64_t TSSegment::onRecvHeader(const char *data, uint64_t len) { int64_t TSSegment::onRecvHeader(const char *data, uint64_t len) {
if (!isTSPacket(data, len)) {
WarnL << "不是ts包:" << (int) (data[0]) << " " << len;
return 0;
}
_onSegment(data, len); _onSegment(data, len);
return 0; return 0;
} }
const char *TSSegment::onSearchPacketTail(const char *data, int len) { const char *TSSegment::onSearchPacketTail(const char *data, int len) {
if (len < _size + 1) { if (len < _size + 1) {
if (len == _size && ((uint8_t *) data)[0] == 0x47) { if (len == _size && ((uint8_t *) data)[0] == TS_SYNC_BYTE) {
return data + _size; return data + _size;
} }
return nullptr; return nullptr;
} }
//下一个包头 //下一个包头
if (((uint8_t *) data)[_size] == 0x47) { if (((uint8_t *) data)[_size] == TS_SYNC_BYTE) {
return data + _size; return data + _size;
} }
auto pos = memchr(data + _size, TS_SYNC_BYTE, len - _size);
auto pos = memchr(data + _size, 0x47, len - _size);
if (pos) { if (pos) {
return (char *) pos; return (char *) pos;
} }
...@@ -44,12 +48,10 @@ const char *TSSegment::onSearchPacketTail(const char *data, int len) { ...@@ -44,12 +48,10 @@ const char *TSSegment::onSearchPacketTail(const char *data, int len) {
//////////////////////////////////////////////////////////////// ////////////////////////////////////////////////////////////////
TSDecoder::TSDecoder() : _ts_segment(TS_PACKET_SIZE) { #if defined(ENABLE_HLS)
#include "mpeg-ts.h"
TSDecoder::TSDecoder() : _ts_segment() {
_ts_segment.setOnSegment([this](const char *data,uint64_t len){ _ts_segment.setOnSegment([this](const char *data,uint64_t len){
if(((uint8_t*)data)[0] != 0x47 || len != TS_PACKET_SIZE ){
WarnL << "不是ts包:" << (int)(data[0]) << " " << len;
return;
}
ts_demuxer_input(_demuxer_ctx,(uint8_t*)data,len); ts_demuxer_input(_demuxer_ctx,(uint8_t*)data,len);
}); });
_demuxer_ctx = ts_demuxer_create([](void* param, int program, int stream, int codecid, int flags, int64_t pts, int64_t dts, const void* data, size_t bytes){ _demuxer_ctx = ts_demuxer_create([](void* param, int program, int stream, int codecid, int flags, int64_t pts, int64_t dts, const void* data, size_t bytes){
...@@ -66,8 +68,8 @@ TSDecoder::~TSDecoder() { ...@@ -66,8 +68,8 @@ TSDecoder::~TSDecoder() {
} }
int TSDecoder::input(const uint8_t *data, int bytes) { int TSDecoder::input(const uint8_t *data, int bytes) {
if(bytes == TS_PACKET_SIZE && ((uint8_t*)data)[0] == 0x47){ if (TSSegment::isTSPacket((char *)data, bytes)) {
return ts_demuxer_input(_demuxer_ctx,(uint8_t*)data,bytes); return ts_demuxer_input(_demuxer_ctx, (uint8_t *) data, bytes);
} }
_ts_segment.input((char*)data,bytes); _ts_segment.input((char*)data,bytes);
return bytes; return bytes;
...@@ -76,6 +78,6 @@ int TSDecoder::input(const uint8_t *data, int bytes) { ...@@ -76,6 +78,6 @@ int TSDecoder::input(const uint8_t *data, int bytes) {
void TSDecoder::setOnDecode(const Decoder::onDecode &decode) { void TSDecoder::setOnDecode(const Decoder::onDecode &decode) {
_on_decode = decode; _on_decode = decode;
} }
#endif//defined(ENABLE_HLS)
}//namespace mediakit }//namespace mediakit
#endif//defined(ENABLE_RTPPROXY)
\ No newline at end of file
src/Rtp/TSDecoder.h
...@@ -11,7 +11,6 @@ ...@@ -11,7 +11,6 @@
#ifndef ZLMEDIAKIT_TSDECODER_H #ifndef ZLMEDIAKIT_TSDECODER_H
#define ZLMEDIAKIT_TSDECODER_H #define ZLMEDIAKIT_TSDECODER_H
#if defined(ENABLE_RTPPROXY)
#include "Util/logger.h" #include "Util/logger.h"
#include "Http/HttpRequestSplitter.h" #include "Http/HttpRequestSplitter.h"
#include "Decoder.h" #include "Decoder.h"
...@@ -19,13 +18,17 @@ ...@@ -19,13 +18,17 @@
using namespace toolkit; using namespace toolkit;
namespace mediakit { namespace mediakit {
//ts包拆分器 #define TS_PACKET_SIZE 188
#define TS_SYNC_BYTE 0x47
//TS包分割器，用于split一个一个的ts包
class TSSegment : public HttpRequestSplitter { class TSSegment : public HttpRequestSplitter {
public: public:
typedef std::function<void(const char *data,uint64_t len)> onSegment; typedef std::function<void(const char *data,uint64_t len)> onSegment;
TSSegment(int size = 188) : _size(size){} TSSegment(int size = TS_PACKET_SIZE) : _size(size){}
~TSSegment(){} ~TSSegment(){}
void setOnSegment(const onSegment &cb); void setOnSegment(const onSegment &cb);
static bool isTSPacket(const char *data, int len);
protected: protected:
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(const char *data, int len) override ;
...@@ -34,6 +37,7 @@ private: ...@@ -34,6 +37,7 @@ private:
onSegment _onSegment; onSegment _onSegment;
}; };
#if defined(ENABLE_HLS)
//ts解析器 //ts解析器
class TSDecoder : public Decoder { class TSDecoder : public Decoder {
public: public:
...@@ -46,7 +50,7 @@ private: ...@@ -46,7 +50,7 @@ private:
struct ts_demuxer_t* _demuxer_ctx = nullptr; struct ts_demuxer_t* _demuxer_ctx = nullptr;
onDecode _on_decode; onDecode _on_decode;
}; };
#endif//defined(ENABLE_HLS)
}//namespace mediakit }//namespace mediakit
#endif//defined(ENABLE_RTPPROXY)
#endif //ZLMEDIAKIT_TSDECODER_H #endif //ZLMEDIAKIT_TSDECODER_H
Markdown 格式
0% 或
您添加了 0 人 到此讨论。请谨慎行事。
请先完成此评论的编辑！
请 注册 或者 后发表评论