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

Merge pull request #18 from xiongziliang/master

Browse Source 
update
This commit is contained in:
baiyfcu
2020-07-20 11:37:23 +08:00
committed by GitHub
parent a46218965a 81a384049a
commit 90bbdf95a9
66 changed files with 2222 additions and 722 deletions
Download Patch File Download Diff File Expand all files Collapse all files

65
src/Common/MediaSource.cpp
View File

@@ -178,39 +178,44 @@ static void eraseIfEmpty(MAP &map, IT0 it0, IT1 it1, IT2 it2) {
}
}
}
};
}
void MediaSource::findAsync_l(const MediaInfo &info, const std::shared_ptr<TcpSession> &session, bool retry, const function<void(const MediaSource::Ptr &src)> &cb){
auto src = MediaSource::find_l(info._schema, info._vhost, info._app, info._streamid, true);
if(src || !retry){
if (src || !retry) {
cb(src);
return;
}
void *listener_tag = session.get();
weak_ptr<TcpSession> weakSession = session;
//广播未找到流,此时可以立即去拉流，这样还来得及
NoticeCenter::Instance().emitEvent(Broadcast::kBroadcastNotFoundStream,info, static_cast<SockInfo &>(*session));
//最多等待一定时间，如果这个时间内，流未注册上，那么返回未找到流
GET_CONFIG(int,maxWaitMS,General::kMaxStreamWaitTimeMS);
//若干秒后执行等待媒体注册超时回调
auto onRegistTimeout = session->getPoller()->doDelayTask(maxWaitMS,[cb,listener_tag](){
//取消监听该事件
NoticeCenter::Instance().delListener(listener_tag,Broadcast::kBroadcastMediaChanged);
GET_CONFIG(int, maxWaitMS, General::kMaxStreamWaitTimeMS);
auto onTimeout = session->getPoller()->doDelayTask(maxWaitMS, [cb, listener_tag]() {
//最多等待一定时间，如果这个时间内，流未注册上，那么返回未找到流
NoticeCenter::Instance().delListener(listener_tag, Broadcast::kBroadcastMediaChanged);
cb(nullptr);
return 0;
});
auto onRegist = [listener_tag,weakSession,info,cb,onRegistTimeout](BroadcastMediaChangedArgs) {
auto cancelAll = [onTimeout, listener_tag]() {
//取消延时任务，防止多次回调
onTimeout->cancel();
//取消媒体注册事件监听
NoticeCenter::Instance().delListener(listener_tag, Broadcast::kBroadcastMediaChanged);
};
function<void()> closePlayer = [cb, cancelAll]() {
cancelAll();
//告诉播放器，流不存在，这样会立即断开播放器
cb(nullptr);
};
auto onRegist = [weakSession, info, cb, cancelAll](BroadcastMediaChangedArgs) {
auto strongSession = weakSession.lock();
if(!strongSession) {
if (!strongSession) {
//自己已经销毁
//取消延时任务，防止多次回调
onRegistTimeout->cancel();
//取消事件监听
NoticeCenter::Instance().delListener(listener_tag,Broadcast::kBroadcastMediaChanged);
cancelAll();
return;
}
@@ -223,24 +228,24 @@ void MediaSource::findAsync_l(const MediaInfo &info, const std::shared_ptr<TcpSe
return;
}
//取消延时任务，防止多次回调
onRegistTimeout->cancel();
//取消事件监听
NoticeCenter::Instance().delListener(listener_tag,Broadcast::kBroadcastMediaChanged);
cancelAll();
//播发器请求的流终于注册上了，切换到自己的线程再回复
strongSession->async([weakSession,info,cb](){
strongSession->async([weakSession, info, cb]() {
auto strongSession = weakSession.lock();
if(!strongSession) {
if (!strongSession) {
return;
}
DebugL << "收到媒体注册事件,回复播放器:" << info._schema << "/" << info._vhost << "/" << info._app << "/" << info._streamid;
//再找一遍媒体源，一般能找到
findAsync_l(info,strongSession,false,cb);
findAsync_l(info, strongSession, false, cb);
}, false);
};
//监听媒体注册事件
NoticeCenter::Instance().addListener(listener_tag, Broadcast::kBroadcastMediaChanged, onRegist);
//广播未找到流,此时可以立即去拉流，这样还来得及
NoticeCenter::Instance().emitEvent(Broadcast::kBroadcastNotFoundStream, info, static_cast<SockInfo &>(*session), closePlayer);
}
void MediaSource::findAsync(const MediaInfo &info, const std::shared_ptr<TcpSession> &session,const function<void(const Ptr &src)> &cb){
@@ -295,7 +300,7 @@ void MediaSource::regist() {
//注册该源，注册后服务器才能找到该源
{
lock_guard<recursive_mutex> lock(g_mtxMediaSrc);
g_mapMediaSrc[_strSchema][_strVhost][_strApp][_strId] = shared_from_this();
g_mapMediaSrc[_strSchema][_strVhost][_strApp][_strId] = shared_from_this();
}
_StrPrinter codec_info;
auto tracks = getTracks(true);
@@ -326,6 +331,11 @@ void MediaSource::regist() {
InfoL << _strSchema << " " << _strVhost << " " << _strApp << " " << _strId << " " << codec_info;
NoticeCenter::Instance().emitEvent(Broadcast::kBroadcastMediaChanged, true, *this);
auto listener = _listener.lock();
if (listener) {
listener->onRegist(*this, true);
}
}
//反注册该源
@@ -352,6 +362,11 @@ bool MediaSource::unregist() {
if(ret){
InfoL << _strSchema << " " << _strVhost << " " << _strApp << " " << _strId;
NoticeCenter::Instance().emitEvent(Broadcast::kBroadcastMediaChanged, false, *this);
auto listener = _listener.lock();
if (listener) {
listener->onRegist(*this, false);
}
}
return ret;
}

6
src/Common/MediaSource.h
View File

@@ -53,9 +53,11 @@ public:
virtual bool setupRecord(MediaSource &sender, Recorder::type type, bool start, const string &custom_path) { return false; };
// 获取录制状态
virtual bool isRecording(MediaSource &sender, Recorder::type type) { return false; };
private:
// 通知无人观看
void onNoneReader(MediaSource &sender);
virtual void onNoneReader(MediaSource &sender);
//流注册或注销事件
virtual void onRegist(MediaSource &sender, bool regist) {};
private:
Timer::Ptr _async_close_timer;
};

16
src/Common/MultiMediaSourceMuxer.cpp
View File

@@ -278,6 +278,22 @@ int MultiMediaSourceMuxer::totalReaderCount(MediaSource &sender) {
return listener->totalReaderCount(sender);
}
void MultiMediaSourceMuxer::onNoneReader(MediaSource &sender){
auto listener = _listener.lock();
if (!listener) {
MediaSourceEvent::onNoneReader(sender);
return;
}
listener->onNoneReader(sender);
}
void MultiMediaSourceMuxer::onRegist(MediaSource &sender, bool regist){
auto listener = _listener.lock();
if (listener) {
listener->onRegist(sender, regist);
}
}
bool MultiMediaSourceMuxer::setupRecord(MediaSource &sender, Recorder::type type, bool start, const string &custom_path) {
return _muxer->setupRecord(sender,type,start,custom_path);
}

13
src/Common/MultiMediaSourceMuxer.h
View File

@@ -131,6 +131,19 @@ public:
*/
int totalReaderCount(MediaSource &sender) override;
/**
* 触发无人观看事件
* @param sender 触发者
*/
void onNoneReader(MediaSource &sender) override;
/**
* 媒体注册注销事件
* @param sender 触发者
* @param regist 是否为注册事件
*/
void onRegist(MediaSource &sender, bool regist) override;
/**
* 设置录制状态
* @param type 录制类型

2
src/Common/Stamp.cpp
View File

@@ -176,6 +176,8 @@ bool DtsGenerator::getDts_l(uint32_t pts, uint32_t &dts){
_sorter_max_size = _frames_since_last_max_pts;
//我们记录P帧间时间间隔(也就是多个B帧时间戳增量累计)
_dts_pts_offset = (pts - _last_max_pts);
//除以2，防止dts大于pts
_dts_pts_offset /= 2;
}
//遇到P帧或关键帧，连续B帧计数清零
_frames_since_last_max_pts = 0;

4
src/Common/config.cpp
View File

@@ -100,11 +100,15 @@ const string kCharSet = HTTP_FIELD"charSet";
const string kRootPath = HTTP_FIELD"rootPath";
//http 404错误提示内容
const string kNotFound = HTTP_FIELD"notFound";
//是否显示文件夹菜单
const string kDirMenu = HTTP_FIELD"dirMenu";
onceToken token([](){
mINI::Instance()[kSendBufSize] = 64 * 1024;
mINI::Instance()[kMaxReqSize] = 4*1024;
mINI::Instance()[kKeepAliveSecond] = 15;
mINI::Instance()[kDirMenu] = true;
#if defined(_WIN32)
mINI::Instance()[kCharSet] = "gb2312";
#else

4
src/Common/config.h
View File

@@ -111,7 +111,7 @@ extern const string kBroadcastFlowReport;
//未找到流后会广播该事件，请在监听该事件后去拉流或其他方式产生流，这样就能按需拉流了
extern const string kBroadcastNotFoundStream;
#define BroadcastNotFoundStreamArgs const MediaInfo &args,SockInfo &sender
#define BroadcastNotFoundStreamArgs const MediaInfo &args,SockInfo &sender, const function<void()> &closePlayer
//某个流无人消费时触发，目的为了实现无人观看时主动断开拉流等业务逻辑
extern const string kBroadcastStreamNoneReader;
@@ -193,6 +193,8 @@ extern const string kCharSet;
extern const string kRootPath;
//http 404错误提示内容
extern const string kNotFound;
//是否显示文件夹菜单
extern const string kDirMenu;
}//namespace Http
////////////SHELL配置///////////

8
src/Extension/AAC.h
View File

@@ -138,14 +138,18 @@ public:
if (_cfg.empty()) {
//未获取到aac_cfg信息
if (frame->prefixSize()) {
//7个字节的adts头
//根据7个字节的adts头生成aac config
_cfg = makeAacConfig((uint8_t *) (frame->data()), frame->prefixSize());
onReady();
} else {
WarnL << "无法获取adts头!";
}
}
AudioTrack::inputFrame(frame);
if (frame->size() > frame->prefixSize()) {
//除adts头外，有实际负载
AudioTrack::inputFrame(frame);
}
}
private:
/**

17
src/Extension/AACRtmp.cpp
View File

@@ -32,8 +32,10 @@ static string getAacCfg(const RtmpPacket &thiz) {
bool AACRtmpDecoder::inputRtmp(const RtmpPacket::Ptr &pkt, bool) {
if (pkt->isCfgFrame()) {
_aac_cfg = getAacCfg(*pkt);
onGetAAC(nullptr, 0, 0);
return false;
}
if (!_aac_cfg.empty()) {
onGetAAC(pkt->strBuf.data() + 2, pkt->strBuf.size() - 2, pkt->timeStamp);
}
@@ -42,7 +44,6 @@ bool AACRtmpDecoder::inputRtmp(const RtmpPacket::Ptr &pkt, bool) {
void AACRtmpDecoder::onGetAAC(const char* data, int len, uint32_t stamp) {
auto frame = ResourcePoolHelper<AACFrame>::obtainObj();
//生成adts头
char adts_header[32] = {0};
auto size = dumpAacConfig(_aac_cfg, len, (uint8_t *) adts_header, sizeof(adts_header));
@@ -54,12 +55,16 @@ void AACRtmpDecoder::onGetAAC(const char* data, int len, uint32_t stamp) {
frame->_prefix_size = 0;
}
//追加负载数据
frame->_buffer.append(data, len);
frame->_dts = stamp;
if(len > 0){
//追加负载数据
frame->_buffer.append(data, len);
frame->_dts = stamp;
}
//写入环形缓存
RtmpCodec::inputFrame(frame);
if(size > 0 || len > 0){
//有adts头或者实际aac负载
RtmpCodec::inputFrame(frame);
}
}
/////////////////////////////////////////////////////////////////////////////////////

9
src/Extension/Factory.cpp
View File

@@ -118,7 +118,7 @@ RtpCodec::Ptr Factory::getRtpEncoderBySdp(const Sdp::Ptr &sdp) {
case CodecH265 : return std::make_shared<H265RtpEncoder>(ssrc,mtu,sample_rate,pt,interleaved);
case CodecAAC : return std::make_shared<AACRtpEncoder>(ssrc,mtu,sample_rate,pt,interleaved);
case CodecG711A :
case CodecG711U : return std::make_shared<G711RtpEncoder>(ssrc, mtu, sample_rate, pt, interleaved);
case CodecG711U : return std::make_shared<G711RtpEncoder>(codec_id, ssrc, mtu, sample_rate, pt, interleaved);
default : WarnL << "暂不支持该CodecId:" << codec_id; return nullptr;
}
}
@@ -129,7 +129,7 @@ RtpCodec::Ptr Factory::getRtpDecoderByTrack(const Track::Ptr &track) {
case CodecH265 : return std::make_shared<H265RtpDecoder>();
case CodecAAC : return std::make_shared<AACRtpDecoder>(track->clone());
case CodecG711A :
case CodecG711U : return std::make_shared<G711RtpDecoder>(track->clone());
case CodecG711U : return std::make_shared<G711RtpDecoder>(track->getCodecId());
default : WarnL << "暂不支持该CodecId:" << track->getCodecName(); return nullptr;
}
}
@@ -243,9 +243,8 @@ RtmpCodec::Ptr Factory::getRtmpCodecByTrack(const Track::Ptr &track, bool is_enc
AMFValue Factory::getAmfByCodecId(CodecId codecId) {
switch (codecId){
//此处用string标明rtmp编码类型目的是为了兼容某些android系统
case CodecAAC: return AMFValue("mp4a");
case CodecH264: return AMFValue("avc1");
case CodecAAC: return AMFValue(FLV_CODEC_AAC);
case CodecH264: return AMFValue(FLV_CODEC_H264);
case CodecH265: return AMFValue(FLV_CODEC_H265);
case CodecG711A: return AMFValue(FLV_CODEC_G711A);
case CodecG711U: return AMFValue(FLV_CODEC_G711U);

18
src/Extension/G711Rtp.cpp
View File

@@ -12,8 +12,8 @@
namespace mediakit{
G711RtpDecoder::G711RtpDecoder(const Track::Ptr &track){
_codecid = track->getCodecId();
G711RtpDecoder::G711RtpDecoder(CodecId codecid){
_codecid = codecid;
_frame = obtainFrame();
}
@@ -59,16 +59,10 @@ void G711RtpDecoder::onGetG711(const G711Frame::Ptr &frame) {
/////////////////////////////////////////////////////////////////////////////////////
G711RtpEncoder::G711RtpEncoder(uint32_t ui32Ssrc,
uint32_t ui32MtuSize,
uint32_t ui32SampleRate,
uint8_t ui8PayloadType,
uint8_t ui8Interleaved) :
RtpInfo(ui32Ssrc,
ui32MtuSize,
ui32SampleRate,
ui8PayloadType,
ui8Interleaved) {
G711RtpEncoder::G711RtpEncoder(CodecId codecid, uint32_t ui32Ssrc, uint32_t ui32MtuSize,
uint32_t ui32SampleRate, uint8_t ui8PayloadType, uint8_t ui8Interleaved) :
G711RtpDecoder(codecid),
RtpInfo(ui32Ssrc, ui32MtuSize, ui32SampleRate, ui8PayloadType, ui8Interleaved) {
}
void G711RtpEncoder::inputFrame(const Frame::Ptr &frame) {

8
src/Extension/G711Rtp.h
View File

@@ -21,7 +21,7 @@ class G711RtpDecoder : public RtpCodec , public ResourcePoolHelper<G711Frame> {
public:
typedef std::shared_ptr<G711RtpDecoder> Ptr;
G711RtpDecoder(const Track::Ptr &track);
G711RtpDecoder(CodecId codecid);
~G711RtpDecoder() {}
/**
@@ -35,9 +35,6 @@ public:
return _codecid;
}
protected:
G711RtpDecoder() {}
private:
void onGetG711(const G711Frame::Ptr &frame);
G711Frame::Ptr obtainFrame();
@@ -61,7 +58,8 @@ public:
* @param ui8PayloadType pt类型
* @param ui8Interleaved rtsp interleaved 值
*/
G711RtpEncoder(uint32_t ui32Ssrc,
G711RtpEncoder(CodecId codecid,
uint32_t ui32Ssrc,
uint32_t ui32MtuSize,
uint32_t ui32SampleRate,
uint8_t ui8PayloadType = 0,

5
src/Extension/H264.h
View File

@@ -228,11 +228,6 @@ private:
}
break;
case H264Frame::NAL_SEI:{
//忽略SEI
break;
}
default:
VideoTrack::inputFrame(frame);
break;

7
src/Extension/H264Rtmp.cpp
View File

@@ -166,6 +166,9 @@ void H264RtmpEncoder::inputFrame(const Frame::Ptr &frame) {
auto pcData = frame->data() + frame->prefixSize();
auto iLen = frame->size() - frame->prefixSize();
auto type = H264_TYPE(((uint8_t*)pcData)[0]);
if(type == H264Frame::NAL_SEI){
return;
}
if (!_gotSpsPps) {
//尝试从frame中获取sps pps
@@ -187,10 +190,6 @@ void H264RtmpEncoder::inputFrame(const Frame::Ptr &frame) {
}
}
if(type == H264Frame::NAL_SEI){
return;
}
if(_lastPacket && _lastPacket->timeStamp != frame->dts()) {
RtmpCodec::inputRtmp(_lastPacket, _lastPacket->isVideoKeyFrame());
_lastPacket = nullptr;

4
src/Extension/H264Rtp.cpp
View File

@@ -234,7 +234,7 @@ void H264RtpEncoder::inputFrame(const Frame::Ptr &frame) {
bool mark = false;
int nOffset = 1;
while (!mark) {
if (iLen < nOffset + iSize) {
if (iLen <= nOffset + iSize) {
//已经拆分结束
iSize = iLen - nOffset;
mark = true;
@@ -274,4 +274,4 @@ void H264RtpEncoder::makeH264Rtp(int nal_type,const void* data, unsigned int len
RtpCodec::inputRtp(makeRtp(getTrackType(),data,len,mark,uiStamp),first_packet && nal_type == H264Frame::NAL_IDR);
}
}//namespace mediakit
}//namespace mediakit

4
src/Extension/H265Rtp.cpp
View File

@@ -166,7 +166,7 @@ void H265RtpEncoder::inputFrame(const Frame::Ptr &frame) {
bool mark = false;
int nOffset = 2;
while (!mark) {
if (iLen < nOffset + maxSize) { //是否拆分结束
if (iLen <= nOffset + maxSize) { //是否拆分结束
maxSize = iLen - nOffset;
mark = true;
//FU end
@@ -208,4 +208,4 @@ void H265RtpEncoder::makeH265Rtp(int nal_type,const void* data, unsigned int len
RtpCodec::inputRtp(makeRtp(getTrackType(),data,len,mark,uiStamp),first_packet && H265Frame::isKeyFrame(nal_type));
}
}//namespace mediakit
}//namespace mediakit

5
src/Http/HttpFileManager.cpp
View File

@@ -195,6 +195,11 @@ static string searchIndexFile(const string &dir){
}
static bool makeFolderMenu(const string &httpPath, const string &strFullPath, string &strRet) {
GET_CONFIG(bool, dirMenu, Http::kDirMenu);
if(!dirMenu){
//不允许浏览文件夹
return false;
}
string strPathPrefix(strFullPath);
string last_dir_name;
if(strPathPrefix.back() == '/'){

1
src/Record/MP4.cpp
View File

@@ -93,6 +93,7 @@ void MP4File::openFile(const char *file,const char *mode) {
//创建智能指针
_file.reset(fp,[file_buf](FILE *fp) {
fflush(fp);
fclose(fp);
});
}

8
src/Record/MP4Muxer.cpp
View File

@@ -11,6 +11,7 @@
#ifdef ENABLE_MP4
#include "MP4Muxer.h"
#include "Util/File.h"
#include "Extension/H264.h"
namespace mediakit{
MP4Muxer::MP4Muxer(const char *file) {
@@ -65,7 +66,12 @@ void MP4Muxer::inputFrame(const Frame::Ptr &frame) {
int64_t dts_out, pts_out;
switch (frame->getCodecId()) {
case CodecH264:
case CodecH264: {
int type = H264_TYPE(*((uint8_t *)frame->data() + frame->prefixSize()));
if(type == H264Frame::NAL_SEI){
break;
}
}
case CodecH265: {
//这里的代码逻辑是让SPS、PPS、IDR这些时间戳相同的帧打包到一起当做一个帧处理，
if (!_frameCached.empty() && _frameCached.back()->dts() != frame->dts()) {

8
src/Record/MP4Muxer.h
View File

@@ -25,6 +25,8 @@ namespace mediakit{
class MP4Muxer : public MediaSinkInterface, public MP4File{
public:
typedef std::shared_ptr<MP4Muxer> Ptr;
MP4Muxer(const char *file);
~MP4Muxer() override;
@@ -42,9 +44,13 @@ public:
*/
void resetTracks() override ;
/**
* 手动关闭文件(对象析构时会自动关闭)
*/
void closeMP4();
private:
void openMP4();
void closeMP4();
void stampSync();
private:

2
src/Record/MP4Recorder.cpp
View File

@@ -75,7 +75,7 @@ void MP4Recorder::asyncClose() {
//获取文件录制时间，放在关闭mp4之前是为了忽略关闭mp4执行时间
const_cast<MP4Info&>(info).ui64TimeLen = ::time(NULL) - info.ui64StartedTime;
//关闭mp4非常耗时，所以要放在后台线程执行
const_cast<MP4Muxer::Ptr &>(muxer).reset();
muxer->closeMP4();
//临时文件名改成正式文件名，防止mp4未完成时被访问
rename(strFileTmp.data(),strFile.data());
//获取文件大小

8
src/Record/TsMuxer.cpp
View File

@@ -12,6 +12,7 @@
#if defined(ENABLE_HLS)
#include "mpeg-ts-proto.h"
#include "mpeg-ts.h"
#include "Extension/H264.h"
namespace mediakit {
@@ -89,8 +90,13 @@ void TsMuxer::inputFrame(const Frame::Ptr &frame) {
int64_t dts_out, pts_out;
_is_idr_fast_packet = !_have_video;
switch (frame->getCodecId()){
case CodecH265:
case CodecH264: {
int type = H264_TYPE(*((uint8_t *)frame->data() + frame->prefixSize()));
if(type == H264Frame::NAL_SEI){
break;
}
}
case CodecH265: {
//这里的代码逻辑是让SPS、PPS、IDR这些时间戳相同的帧打包到一起当做一个帧处理，
if (!_frameCached.empty() && _frameCached.back()->dts() != frame->dts()) {
Frame::Ptr back = _frameCached.back();

2
src/Rtmp/RtmpDemuxer.cpp
View File

@@ -106,6 +106,7 @@ void RtmpDemuxer::makeVideoTrack(const AMFValue &videoCodec) {
//设置rtmp解码器代理，生成的frame写入该Track
_videoRtmpDecoder->addDelegate(_videoTrack);
onAddTrack(_videoTrack);
_tryedGetVideoTrack = true;
} else {
//找不到相应的rtmp解码器，该track无效
_videoTrack.reset();
@@ -123,6 +124,7 @@ void RtmpDemuxer::makeAudioTrack(const AMFValue &audioCodec,int sample_rate, int
//设置rtmp解码器代理，生成的frame写入该Track
_audioRtmpDecoder->addDelegate(_audioTrack);
onAddTrack(_audioTrack);
_tryedGetAudioTrack = true;
} else {
//找不到相应的rtmp解码器，该track无效
_audioTrack.reset();

11
src/Rtmp/RtmpSession.cpp
View File

@@ -257,8 +257,8 @@ void RtmpSession::sendPlayResponse(const string &err,const RtmpMediaSource::Ptr
invoke.clear();
invoke << "onMetaData" << metadata;
sendResponse(MSG_DATA, invoke.data());
auto duration = metadata["duration"].as_number();
if(duration > 0){
auto duration = metadata["duration"];
if(duration && duration.as_number() > 0){
//这是点播，使用绝对时间戳
_stamp[0].setPlayBack();
_stamp[1].setPlayBack();
@@ -380,7 +380,12 @@ string RtmpSession::getStreamId(const string &str){
//vlc和ffplay在播放 rtmp://127.0.0.1/record/0.mp4时，
//传过来的url会是rtmp://127.0.0.1/record/mp4:0,
//我们在这里还原成0.mp4
stream_id = stream_id.substr(pos + 1) + "." + stream_id.substr(0,pos);
//实际使用时发现vlc，mpv等会传过来rtmp://127.0.0.1/record/mp4:0.mp4,这里做个判断
auto ext = stream_id.substr(0,pos);
stream_id = stream_id.substr(pos + 1);
if(stream_id.find(ext) == string::npos){
stream_id = stream_id + "." + ext;
}
}
if(params.empty()){

5
src/Rtp/Decoder.cpp
View File

@@ -158,6 +158,11 @@ void DecoderImp::onDecode(int stream,int codecid,int flags,int64_t pts,int64_t d
}
case PSI_STREAM_AAC: {
uint8_t *ptr = (uint8_t *)data;
if(!(bytes > 7 && ptr[0] == 0xFF && (ptr[1] & 0xF0) == 0xF0)){
//这不是aac
break;
}
if (!_codecid_audio) {
//获取到音频
_codecid_audio = codecid;

1
src/Rtp/Decoder.h
View File

@@ -14,7 +14,6 @@
#include <stdint.h>
#include <memory>
#include <functional>
#include "Decoder.h"
#include "Common/MediaSink.h"
using namespace std;

45
src/Rtp/RtpProcess.cpp
View File

@@ -16,32 +16,15 @@
namespace mediakit{
string printSSRC(uint32_t ui32Ssrc) {
char tmp[9] = { 0 };
ui32Ssrc = htonl(ui32Ssrc);
uint8_t *pSsrc = (uint8_t *) &ui32Ssrc;
for (int i = 0; i < 4; i++) {
sprintf(tmp + 2 * i, "%02X", pSsrc[i]);
}
return tmp;
}
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);
}
RtpProcess::RtpProcess(uint32_t ssrc) {
_ssrc = ssrc;
_track = std::make_shared<SdpTrack>();
_track->_interleaved = 0;
_track->_samplerate = 90000;
_track->_type = TrackVideo;
_track->_ssrc = _ssrc;
RtpProcess::RtpProcess(const string &stream_id) {
_media_info._schema = RTP_APP_NAME;
_media_info._vhost = DEFAULT_VHOST;
_media_info._app = RTP_APP_NAME;
_media_info._streamid = printSSRC(_ssrc);
_media_info._streamid = stream_id;
GET_CONFIG(string,dump_dir,RtpProxy::kDumpDir);
{
@@ -73,11 +56,6 @@ RtpProcess::RtpProcess(uint32_t ssrc) {
}
RtpProcess::~RtpProcess() {
DebugP(this);
if (_addr) {
delete _addr;
}
uint64_t duration = (_last_rtp_time.createdTime() - _last_rtp_time.elapsedTime()) / 1000;
WarnP(this) << "RTP推流器("
<< _media_info._vhost << "/"
@@ -90,6 +68,11 @@ RtpProcess::~RtpProcess() {
if (_total_bytes > iFlowThreshold * 1024) {
NoticeCenter::Instance().emitEvent(Broadcast::kBroadcastFlowReport, _media_info, _total_bytes, duration, false, static_cast<SockInfo &>(*this));
}
if (_addr) {
delete _addr;
_addr = nullptr;
}
}
bool RtpProcess::inputRtp(const Socket::Ptr &sock, const char *data, int data_len,const struct sockaddr *addr,uint32_t *dts_out) {
@@ -115,7 +98,7 @@ bool RtpProcess::inputRtp(const Socket::Ptr &sock, const char *data, int data_le
}
_total_bytes += data_len;
bool ret = handleOneRtp(0,_track,(unsigned char *)data,data_len);
bool ret = handleOneRtp(0, TrackVideo, 90000, (unsigned char *) data, data_len);
if(dts_out){
*dts_out = _dts;
}
@@ -128,7 +111,7 @@ static inline bool checkTS(const uint8_t *packet, int bytes){
}
void RtpProcess::onRtpSorted(const RtpPacket::Ptr &rtp, int) {
if(rtp->sequence != _sequence + 1 && _sequence != 0){
if(rtp->sequence != _sequence + (uint16_t)1 && _sequence != 0){
WarnP(this) << "rtp丢包:" << rtp->sequence << " != " << _sequence << "+1" << ",公网环境下请使用tcp方式推流";
}
_sequence = rtp->sequence;
@@ -188,6 +171,16 @@ bool RtpProcess::alive() {
return false;
}
void RtpProcess::onDetach(){
if(_on_detach){
_on_detach();
}
}
void RtpProcess::setOnDetach(const function<void()> &cb) {
_on_detach = cb;
}
string RtpProcess::get_peer_ip() {
if(_addr){
return SockUtil::inet_ntoa(((struct sockaddr_in *) _addr)->sin_addr);

31
src/Rtp/RtpProcess.h
View File

@@ -22,15 +22,39 @@ using namespace mediakit;
namespace mediakit{
string printSSRC(uint32_t ui32Ssrc);
class RtpProcess : public RtpReceiver , public RtpDecoder, public SockInfo, public MediaSinkInterface, public std::enable_shared_from_this<RtpProcess>{
public:
typedef std::shared_ptr<RtpProcess> Ptr;
RtpProcess(uint32_t ssrc);
RtpProcess(const string &stream_id);
~RtpProcess();
/**
* 输入rtp
* @param sock 本地监听的socket
* @param data rtp数据指针
* @param data_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 alive();
/**
* 超时时被RtpSelector移除时触发
*/
void onDetach();
/**
* 设置onDetach事件回调
*/
void setOnDetach(const function<void()> &cb);
/// SockInfo override
string get_local_ip() override;
uint16_t get_local_port() override;
string get_peer_ip() override;
@@ -54,8 +78,6 @@ private:
std::shared_ptr<FILE> _save_file_rtp;
std::shared_ptr<FILE> _save_file_ps;
std::shared_ptr<FILE> _save_file_video;
uint32_t _ssrc;
SdpTrack::Ptr _track;
struct sockaddr *_addr = nullptr;
uint16_t _sequence = 0;
MultiMediaSourceMuxer::Ptr _muxer;
@@ -66,6 +88,7 @@ private:
MediaInfo _media_info;
uint64_t _total_bytes = 0;
Socket::Ptr _sock;
function<void()> _on_detach;
};
}//namespace mediakit

58
src/Rtp/RtpSelector.cpp
View File

@@ -15,37 +15,41 @@ namespace mediakit{
INSTANCE_IMP(RtpSelector);
bool RtpSelector::inputRtp(const Socket::Ptr &sock, const char *data, int data_len,const struct sockaddr *addr,uint32_t *dts_out) {
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)){
if (!getSSRC(data, data_len, ssrc)) {
WarnL << "get ssrc from rtp failed:" << data_len;
return false;
}
auto process = getProcess(ssrc, true);
if(process){
return process->inputRtp(sock, data,data_len, addr,dts_out);
//假定指定了流id，那么通过流id来区分是否为一路流(哪怕可能同时收到多路流)
auto process = getProcess(printSSRC(ssrc), true);
if (process) {
return process->inputRtp(sock, data, data_len, addr, dts_out);
}
return false;
}
bool RtpSelector::getSSRC(const char *data,int data_len, uint32_t &ssrc){
if(data_len < 12){
if (data_len < 12) {
return false;
}
uint32_t *ssrc_ptr = (uint32_t *)(data + 8);
uint32_t *ssrc_ptr = (uint32_t *) (data + 8);
ssrc = ntohl(*ssrc_ptr);
return true;
}
RtpProcess::Ptr RtpSelector::getProcess(uint32_t ssrc,bool makeNew) {
RtpProcess::Ptr RtpSelector::getProcess(const string &stream_id,bool makeNew) {
lock_guard<decltype(_mtx_map)> lck(_mtx_map);
auto it = _map_rtp_process.find(ssrc);
if(it == _map_rtp_process.end() && !makeNew){
auto it = _map_rtp_process.find(stream_id);
if (it == _map_rtp_process.end() && !makeNew) {
return nullptr;
}
RtpProcessHelper::Ptr &ref = _map_rtp_process[ssrc];
if(!ref){
ref = std::make_shared<RtpProcessHelper>(ssrc,shared_from_this());
RtpProcessHelper::Ptr &ref = _map_rtp_process[stream_id];
if (!ref) {
ref = std::make_shared<RtpProcessHelper>(stream_id, shared_from_this());
ref->attachEvent();
createTimer();
}
@@ -67,18 +71,18 @@ void RtpSelector::createTimer() {
}
}
void RtpSelector::delProcess(uint32_t ssrc,const RtpProcess *ptr) {
void RtpSelector::delProcess(const string &stream_id,const RtpProcess *ptr) {
lock_guard<decltype(_mtx_map)> lck(_mtx_map);
auto it = _map_rtp_process.find(ssrc);
if(it == _map_rtp_process.end()){
auto it = _map_rtp_process.find(stream_id);
if (it == _map_rtp_process.end()) {
return;
}
if(it->second->getProcess().get() != ptr){
if (it->second->getProcess().get() != ptr) {
return;
}
auto process = it->second->getProcess();
_map_rtp_process.erase(it);
process->onDetach();
}
void RtpSelector::onManager() {
@@ -88,8 +92,10 @@ void RtpSelector::onManager() {
++it;
continue;
}
WarnL << "RtpProcess timeout:" << printSSRC(it->first);
WarnL << "RtpProcess timeout:" << it->first;
auto process = it->second->getProcess();
it = _map_rtp_process.erase(it);
process->onDetach();
}
}
@@ -99,10 +105,10 @@ RtpSelector::RtpSelector() {
RtpSelector::~RtpSelector() {
}
RtpProcessHelper::RtpProcessHelper(uint32_t ssrc, const weak_ptr<RtpSelector> &parent) {
_ssrc = ssrc;
RtpProcessHelper::RtpProcessHelper(const string &stream_id, const weak_ptr<RtpSelector> &parent) {
_stream_id = stream_id;
_parent = parent;
_process = std::make_shared<RtpProcess>(_ssrc);
_process = std::make_shared<RtpProcess>(stream_id);
}
RtpProcessHelper::~RtpProcessHelper() {
@@ -114,14 +120,14 @@ void RtpProcessHelper::attachEvent() {
bool RtpProcessHelper::close(MediaSource &sender, bool force) {
//此回调在其他线程触发
if(!_process || (!force && _process->totalReaderCount())){
if (!_process || (!force && _process->totalReaderCount())) {
return false;
}
auto parent = _parent.lock();
if(!parent){
if (!parent) {
return false;
}
parent->delProcess(_ssrc,_process.get());
parent->delProcess(_stream_id, _process.get());
WarnL << "close media:" << sender.getSchema() << "/" << sender.getVhost() << "/" << sender.getApp() << "/" << sender.getId() << " " << force;
return true;
}

42
src/Rtp/RtpSelector.h
View File

@@ -24,19 +24,21 @@ class RtpSelector;
class RtpProcessHelper : public MediaSourceEvent , public std::enable_shared_from_this<RtpProcessHelper> {
public:
typedef std::shared_ptr<RtpProcessHelper> Ptr;
RtpProcessHelper(uint32_t ssrc,const weak_ptr<RtpSelector > &parent);
RtpProcessHelper(const string &stream_id, const weak_ptr<RtpSelector > &parent);
~RtpProcessHelper();
void attachEvent();
RtpProcess::Ptr & getProcess();
protected:
// 通知其停止推流
bool close(MediaSource &sender,bool force) override;
// 观看总人数
int totalReaderCount(MediaSource &sender) override;
private:
weak_ptr<RtpSelector > _parent;
RtpProcess::Ptr _process;
uint32_t _ssrc = 0;
string _stream_id;
};
class RtpSelector : public std::enable_shared_from_this<RtpSelector>{
@@ -44,16 +46,42 @@ public:
RtpSelector();
~RtpSelector();
static RtpSelector &Instance();
bool inputRtp(const Socket::Ptr &sock, const char *data,int data_len,const struct sockaddr *addr ,uint32_t *dts_out = nullptr );
static bool getSSRC(const char *data,int data_len, uint32_t &ssrc);
RtpProcess::Ptr getProcess(uint32_t ssrc,bool makeNew);
void delProcess(uint32_t ssrc,const RtpProcess *ptr);
static RtpSelector &Instance();
/**
* 输入多个rtp流，根据ssrc分流
* @param sock 本地socket
* @param data 收到的数据
* @param data_len 收到的数据长度
* @param addr rtp流源地址
* @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);
/**
* 获取一个rtp处理器
* @param stream_id 流id
* @param makeNew 不存在时是否新建
* @return rtp处理器
*/
RtpProcess::Ptr getProcess(const string &stream_id, bool makeNew);
/**
* 删除rtp处理器
* @param stream_id 流id
* @param ptr rtp处理器指针
*/
void delProcess(const string &stream_id, const RtpProcess *ptr);
private:
void onManager();
void createTimer();
private:
unordered_map<uint32_t,RtpProcessHelper::Ptr> _map_rtp_process;
unordered_map<string,RtpProcessHelper::Ptr> _map_rtp_process;
recursive_mutex _mtx_map;
Timer::Ptr _timer;
};

90
src/Rtp/RtpServer.cpp Normal file
View File

@@ -0,0 +1,90 @@
/*
* 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 "RtpServer.h"
#include "RtpSelector.h"
namespace mediakit{
RtpServer::RtpServer() {
}
RtpServer::~RtpServer() {
if(_on_clearup){
_on_clearup();
}
}
void RtpServer::start(uint16_t local_port, const string &stream_id, bool enable_tcp, const char *local_ip) {
//创建udp服务器
Socket::Ptr udp_server = std::make_shared<Socket>(nullptr, false);
if (!udp_server->bindUdpSock(local_port, local_ip)) {
throw std::runtime_error(StrPrinter << "bindUdpSock on " << local_ip << ":" << local_port << " failed:" << get_uv_errmsg(true));
}
//设置udp socket读缓存
SockUtil::setRecvBuf(udp_server->rawFD(), 4 * 1024 * 1024);
TcpServer::Ptr tcp_server;
if (enable_tcp) {
try {
//创建tcp服务器
tcp_server = std::make_shared<TcpServer>(udp_server->getPoller());
(*tcp_server)[RtpSession::kStreamID] = stream_id;
tcp_server->start<RtpSession>(udp_server->get_local_port(), local_ip);
} catch (...) {
throw;
}
}
RtpProcess::Ptr process;
if (!stream_id.empty()) {
//指定了流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);
});
} else {
//未指定流id，一个端口多个流，通过ssrc来分流
auto &ref = RtpSelector::Instance();
udp_server->setOnRead([&ref, udp_server](const Buffer::Ptr &buf, struct sockaddr *addr, int) {
ref.inputRtp(udp_server, buf->data(), buf->size(), addr);
});
}
_on_clearup = [udp_server, process, stream_id]() {
//去除循环引用
udp_server->setOnRead(nullptr);
if (process) {
//删除rtp处理器
RtpSelector::Instance().delProcess(stream_id, process.get());
}
};
_tcp_server = tcp_server;
_udp_server = udp_server;
_rtp_process = process;
}
void RtpServer::setOnDetach(const function<void()> &cb){
if(_rtp_process){
_rtp_process->setOnDetach(cb);
}
}
EventPoller::Ptr RtpServer::getPoller() {
return _udp_server->getPoller();
}
uint16_t RtpServer::getPort() {
return _udp_server ? _udp_server->get_local_port() : 0;
}
}//namespace mediakit
#endif//defined(ENABLE_RTPPROXY)

69
src/Rtp/RtpServer.h Normal file
View File

@@ -0,0 +1,69 @@
/*
* 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_RTPSERVER_H
#define ZLMEDIAKIT_RTPSERVER_H
#if defined(ENABLE_RTPPROXY)
#include <memory>
#include "Network/Socket.h"
#include "Network/TcpServer.h"
#include "RtpSession.h"
using namespace std;
using namespace toolkit;
namespace mediakit{
/**
* RTP服务器，支持UDP/TCP
*/
class RtpServer {
public:
typedef std::shared_ptr<RtpServer> Ptr;
typedef function<void(const Buffer::Ptr &buf)> onRecv;
RtpServer();
~RtpServer();
/**
* 开启服务器，可能抛异常
* @param local_port 本地端口，0时为随机端口
* @param stream_id 流id，置空则使用ssrc
* @param enable_tcp 是否启用tcp服务器
* @param local_ip 绑定的本地网卡ip
*/
void start(uint16_t local_port, const string &stream_id = "", bool enable_tcp = true, const char *local_ip = "0.0.0.0");
/**
* 获取绑定的本地端口
*/
uint16_t getPort();
/**
* 获取绑定的线程
*/
EventPoller::Ptr getPoller();
/**
* 设置RtpProcess onDetach事件回调
*/
void setOnDetach(const function<void()> &cb);
protected:
Socket::Ptr _udp_server;
TcpServer::Ptr _tcp_server;
RtpProcess::Ptr _rtp_process;
function<void()> _on_clearup;
};
}//namespace mediakit
#endif//defined(ENABLE_RTPPROXY)
#endif //ZLMEDIAKIT_RTPSERVER_H

23
src/Rtp/RtpSession.cpp
View File

@@ -11,8 +11,15 @@
#if defined(ENABLE_RTPPROXY)
#include "RtpSession.h"
#include "RtpSelector.h"
#include "Network/TcpServer.h"
namespace mediakit{
const string RtpSession::kStreamID = "stream_id";
void RtpSession::attachServer(const TcpServer &server) {
_stream_id = const_cast<TcpServer &>(server)[kStreamID];
}
RtpSession::RtpSession(const Socket::Ptr &sock) : TcpSession(sock) {
DebugP(this);
socklen_t addr_len = sizeof(addr);
@@ -21,7 +28,7 @@ RtpSession::RtpSession(const Socket::Ptr &sock) : TcpSession(sock) {
RtpSession::~RtpSession() {
DebugP(this);
if(_process){
RtpSelector::Instance().delProcess(_ssrc,_process.get());
RtpSelector::Instance().delProcess(_stream_id,_process.get());
}
}
@@ -36,7 +43,7 @@ void RtpSession::onRecv(const Buffer::Ptr &data) {
}
void RtpSession::onError(const SockException &err) {
WarnL << _ssrc << " " << err.what();
WarnL << _stream_id << " " << err.what();
}
void RtpSession::onManager() {
@@ -51,13 +58,19 @@ void RtpSession::onManager() {
void RtpSession::onRtpPacket(const char *data, uint64_t len) {
if (!_process) {
if (!RtpSelector::getSSRC(data + 2, len - 2, _ssrc)) {
uint32_t ssrc;
if (!RtpSelector::getSSRC(data + 2, len - 2, ssrc)) {
return;
}
_process = RtpSelector::Instance().getProcess(_ssrc, true);
if (_stream_id.empty()) {
//未指定流id就使用ssrc为流id
_stream_id = printSSRC(ssrc);
}
//tcp情况下，一个tcp链接只可能是一路流，不需要通过多个ssrc来区分，所以不需要频繁getProcess
_process = RtpSelector::Instance().getProcess(_stream_id, true);
_process->setListener(dynamic_pointer_cast<RtpSession>(shared_from_this()));
}
_process->inputRtp(_sock,data + 2, len - 2, &addr);
_process->inputRtp(_sock, data + 2, len - 2, &addr);
_ticker.resetTime();
}

6
src/Rtp/RtpSession.h
View File

@@ -22,22 +22,26 @@ namespace mediakit{
class RtpSession : public TcpSession , public RtpSplitter , public MediaSourceEvent{
public:
static const string kStreamID;
RtpSession(const Socket::Ptr &sock);
~RtpSession() override;
void onRecv(const Buffer::Ptr &) override;
void onError(const SockException &err) override;
void onManager() override;
void attachServer(const TcpServer &server) override;
protected:
// 通知其停止推流
bool close(MediaSource &sender,bool force) override;
// 观看总人数
int totalReaderCount(MediaSource &sender) override;
void onRtpPacket(const char *data,uint64_t len) override;
private:
uint32_t _ssrc = 0;
RtpProcess::Ptr _process;
Ticker _ticker;
struct sockaddr addr;
string _stream_id;
};
}//namespace mediakit

44
src/Rtp/UdpRecver.cpp
View File

@@ -1,44 +0,0 @@
/*
* 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 "UdpRecver.h"
#include "RtpSelector.h"
namespace mediakit{
UdpRecver::UdpRecver() {
}
UdpRecver::~UdpRecver() {
if(_sock){
_sock->setOnRead(nullptr);
}
}
bool UdpRecver::initSock(uint16_t local_port,const char *local_ip) {
_sock.reset(new Socket(nullptr, false));
onceToken token(nullptr,[&](){
SockUtil::setRecvBuf(_sock->rawFD(),4 * 1024 * 1024);
});
auto &ref = RtpSelector::Instance();
auto sock = _sock;
_sock->setOnRead([&ref, sock](const Buffer::Ptr &buf, struct sockaddr *addr, int ){
ref.inputRtp(sock, buf->data(),buf->size(),addr);
});
return _sock->bindUdpSock(local_port,local_ip);
}
EventPoller::Ptr UdpRecver::getPoller() {
return _sock->getPoller();
}
}//namespace mediakit
#endif//defined(ENABLE_RTPPROXY)

40
src/Rtp/UdpRecver.h
View File

@@ -1,40 +0,0 @@
/*
* 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_UDPRECVER_H
#define ZLMEDIAKIT_UDPRECVER_H
#if defined(ENABLE_RTPPROXY)
#include <memory>
#include "Network/Socket.h"
using namespace std;
using namespace toolkit;
namespace mediakit{
/**
* 组播接收器
*/
class UdpRecver {
public:
typedef std::shared_ptr<UdpRecver> Ptr;
typedef function<void(const Buffer::Ptr &buf)> onRecv;
UdpRecver();
virtual ~UdpRecver();
bool initSock(uint16_t local_port,const char *local_ip = "0.0.0.0");
EventPoller::Ptr getPoller();
protected:
Socket::Ptr _sock;
};
}//namespace mediakit
#endif//defined(ENABLE_RTPPROXY)
#endif //ZLMEDIAKIT_UDPRECVER_H

33
src/Rtsp/RtpReceiver.cpp
View File

@@ -27,7 +27,7 @@ namespace mediakit {
RtpReceiver::RtpReceiver() {}
RtpReceiver::~RtpReceiver() {}
bool RtpReceiver::handleOneRtp(int track_index,SdpTrack::Ptr &track, unsigned char *rtp_raw_ptr, unsigned int rtp_raw_len) {
bool RtpReceiver::handleOneRtp(int track_index, TrackType type, int samplerate, unsigned char *rtp_raw_ptr, unsigned int rtp_raw_len) {
if(rtp_raw_len < 12){
WarnL << "rtp包太小:" << rtp_raw_len;
return false;
@@ -46,8 +46,8 @@ bool RtpReceiver::handleOneRtp(int track_index,SdpTrack::Ptr &track, unsigned ch
auto rtp_ptr = _rtp_pool.obtain();
auto &rtp = *rtp_ptr;
rtp.type = track->_type;
rtp.interleaved = 2 * track->_type;
rtp.type = type;
rtp.interleaved = 2 * type;
rtp.mark = rtp_raw_ptr[1] >> 7;
rtp.PT = rtp_raw_ptr[1] & 0x7F;
@@ -59,29 +59,29 @@ bool RtpReceiver::handleOneRtp(int track_index,SdpTrack::Ptr &track, unsigned ch
memcpy(&rtp.timeStamp, rtp_raw_ptr + 4, 4);
rtp.timeStamp = ntohl(rtp.timeStamp);
if(!track->_samplerate){
if(!samplerate){
//无法把时间戳转换成毫秒
return false;
}
//时间戳转换成毫秒
rtp.timeStamp = rtp.timeStamp * 1000LL / track->_samplerate;
rtp.timeStamp = rtp.timeStamp * 1000LL / samplerate;
//ssrc,内存对齐
memcpy(&rtp.ssrc, rtp_raw_ptr + 8, 4);
rtp.ssrc = ntohl(rtp.ssrc);
if (track->_ssrc != rtp.ssrc) {
if (track->_ssrc == 0) {
if (_ssrc[track_index] != rtp.ssrc) {
if (_ssrc[track_index] == 0) {
//保存SSRC至track对象
track->_ssrc = rtp.ssrc;
_ssrc[track_index] = rtp.ssrc;
}else{
//ssrc错误
WarnL << "ssrc错误:" << rtp.ssrc << " != " << track->_ssrc;
WarnL << "ssrc错误:" << rtp.ssrc << " != " << _ssrc[track_index];
if (_ssrc_err_count[track_index]++ > 10) {
//ssrc切换后清除老数据
WarnL << "ssrc更换:" << track->_ssrc << " -> " << rtp.ssrc;
WarnL << "ssrc更换:" << _ssrc[track_index] << " -> " << rtp.ssrc;
_rtp_sort_cache_map[track_index].clear();
track->_ssrc = rtp.ssrc;
_ssrc[track_index] = rtp.ssrc;
}
return false;
}
@@ -168,11 +168,12 @@ void RtpReceiver::sortRtp(const RtpPacket::Ptr &rtp,int track_index){
}
void RtpReceiver::clear() {
CLEAR_ARR(_last_seq)
CLEAR_ARR(_ssrc_err_count)
CLEAR_ARR(_seq_ok_count)
CLEAR_ARR(_sort_started)
CLEAR_ARR(_seq_cycle_count)
CLEAR_ARR(_last_seq);
CLEAR_ARR(_ssrc);
CLEAR_ARR(_ssrc_err_count);
CLEAR_ARR(_seq_ok_count);
CLEAR_ARR(_sort_started);
CLEAR_ARR(_seq_cycle_count);
_rtp_sort_cache_map[0].clear();
_rtp_sort_cache_map[1].clear();

8
src/Rtsp/RtpReceiver.h
View File

@@ -32,12 +32,13 @@ protected:
/**
* 输入数据指针生成并排序rtp包
* @param track_index track下标索引
* @param track sdp track相关信息
* @param type track类型
* @param samplerate rtp时间戳基准时钟，视频为90000，音频为采样率
* @param rtp_raw_ptr rtp数据指针
* @param rtp_raw_len rtp数据指针长度
* @return 解析成功返回true
*/
bool handleOneRtp(int track_index,SdpTrack::Ptr &track, unsigned char *rtp_raw_ptr, unsigned int rtp_raw_len);
bool handleOneRtp(int track_index, TrackType type, int samplerate, unsigned char *rtp_raw_ptr, unsigned int rtp_raw_len);
/**
* rtp数据包排序后输出
@@ -49,9 +50,12 @@ protected:
void setPoolSize(int size);
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不匹配计数
uint32_t _ssrc_err_count[2] = { 0, 0 };
//上次seq

10
src/Rtsp/Rtsp.cpp
View File

@@ -404,4 +404,14 @@ std::pair<Socket::Ptr, Socket::Ptr> makeSockPair(const EventPoller::Ptr &poller,
}
}
string printSSRC(uint32_t ui32Ssrc) {
char tmp[9] = { 0 };
ui32Ssrc = htonl(ui32Ssrc);
uint8_t *pSsrc = (uint8_t *) &ui32Ssrc;
for (int i = 0; i < 4; i++) {
sprintf(tmp + 2 * i, "%02X", pSsrc[i]);
}
return tmp;
}
}//namespace mediakit

1
src/Rtsp/Rtsp.h
View File

@@ -272,6 +272,7 @@ private:
};
std::pair<Socket::Ptr, Socket::Ptr> makeSockPair(const EventPoller::Ptr &poller, const string &local_ip);
string printSSRC(uint32_t ui32Ssrc);
} //namespace mediakit
#endif //RTSP_RTSP_H_

20
src/Rtsp/RtspPlayer.cpp
View File

@@ -319,7 +319,8 @@ void RtspPlayer::handleResSETUP(const Parser &parser, unsigned int uiTrackIndex)
WarnL << "收到其他地址的rtp数据:" << SockUtil::inet_ntoa(((struct sockaddr_in *) addr)->sin_addr);
return;
}
strongSelf->handleOneRtp(uiTrackIndex, strongSelf->_sdp_track[uiTrackIndex], (unsigned char *) buf->data(), buf->size());
strongSelf->handleOneRtp(uiTrackIndex, strongSelf->_sdp_track[uiTrackIndex]->_type,
strongSelf->_sdp_track[uiTrackIndex]->_samplerate, (unsigned char *) buf->data(), buf->size());
});
if(pRtcpSockRef) {
@@ -464,14 +465,10 @@ void RtspPlayer::onRtpPacket(const char *data, uint64_t len) {
uint8_t interleaved = data[1];
if(interleaved %2 == 0){
trackIdx = getTrackIndexByInterleaved(interleaved);
if (trackIdx != -1) {
handleOneRtp(trackIdx, _sdp_track[trackIdx], (unsigned char *)data + 4, len - 4);
}
handleOneRtp(trackIdx, _sdp_track[trackIdx]->_type, _sdp_track[trackIdx]->_samplerate, (unsigned char *)data + 4, len - 4);
}else{
trackIdx = getTrackIndexByInterleaved(interleaved - 1);
if (trackIdx != -1) {
onRtcpPacket(trackIdx, _sdp_track[trackIdx], (unsigned char *) data + 4, len - 4);
}
onRtcpPacket(trackIdx, _sdp_track[trackIdx], (unsigned char *) data + 4, len - 4);
}
}
@@ -712,10 +709,7 @@ void RtspPlayer::onRecvRTP_l(const RtpPacket::Ptr &pkt, const SdpTrack::Ptr &tra
_rtp_recv_ticker.resetTime();
onRecvRTP(pkt, track);
int iTrackIndex = getTrackIndexByInterleaved(pkt->interleaved);
if (iTrackIndex == -1) {
return;
}
int iTrackIndex = getTrackIndexByTrackType(pkt->type);
RtcpCounter &counter = _rtcp_counter[iTrackIndex];
counter.pktCnt = pkt->sequence;
auto &ticker = _rtcp_send_ticker[iTrackIndex];
@@ -788,7 +782,7 @@ int RtspPlayer::getTrackIndexByInterleaved(int interleaved) const {
if (_sdp_track.size() == 1) {
return 0;
}
return -1;
throw SockException(Err_shutdown, StrPrinter << "no such track with interleaved:" << interleaved);
}
int RtspPlayer::getTrackIndexByTrackType(TrackType trackType) const {
@@ -800,7 +794,7 @@ int RtspPlayer::getTrackIndexByTrackType(TrackType trackType) const {
if (_sdp_track.size() == 1) {
return 0;
}
return -1;
throw SockException(Err_shutdown, StrPrinter << "no such track with type:" << (int) trackType);
}
} /* namespace mediakit */

2
src/Rtsp/RtspPusher.cpp
View File

@@ -347,7 +347,7 @@ inline int RtspPusher::getTrackIndexByTrackType(TrackType type) {
if(_aTrackInfo.size() == 1){
return 0;
}
return -1;
throw SockException(Err_shutdown, StrPrinter << "no such track with type:" << (int) type);
}
void RtspPusher::sendRecord() {

560
src/Rtsp/RtspSession.cpp
View File

File diff suppressed because it is too large Load Diff

169
src/Rtsp/RtspSession.h
View File

@@ -59,51 +59,31 @@ public:
//在请求明文密码时如果提供md5密码者则会导致认证失败
typedef std::function<void(bool encrypted,const string &pwd_or_md5)> onAuth;
RtspSession(const Socket::Ptr &pSock);
RtspSession(const Socket::Ptr &sock);
virtual ~RtspSession();
////TcpSession override////
void onRecv(const Buffer::Ptr &pBuf) override;
void onRecv(const Buffer::Ptr &buf) override;
void onError(const SockException &err) override;
void onManager() override;
protected:
//RtspSplitter override
/**
* 收到完整的rtsp包回调，包括sdp等content数据
* @param parser rtsp包
*/
/////RtspSplitter override/////
//收到完整的rtsp包回调，包括sdp等content数据
void onWholeRtspPacket(Parser &parser) override;
/**
* 收到rtp包回调
* @param data
* @param len
*/
void onRtpPacket(const char *data,uint64_t len) override;
/**
* 从rtsp头中获取Content长度
* @param parser
* @return
*/
//收到rtp包回调
void onRtpPacket(const char *data, uint64_t len) override;
//从rtsp头中获取Content长度
int64_t getContentLength(Parser &parser) override;
//RtpReceiver override
void onRtpSorted(const RtpPacket::Ptr &rtppt, int trackidx) override;
//MediaSourceEvent override
bool close(MediaSource &sender,bool force) override ;
////RtpReceiver override////
void onRtpSorted(const RtpPacket::Ptr &rtp, int track_idx) override;
////MediaSourceEvent override////
bool close(MediaSource &sender, bool force) override ;
int totalReaderCount(MediaSource &sender) override;
//TcpSession override
/////TcpSession override////
int send(const Buffer::Ptr &pkt) override;
//收到RTCP包回调
virtual void onRtcpPacket(int track_idx, SdpTrack::Ptr &track, unsigned char *data, unsigned int len);
/**
* 收到RTCP包回调
* @param iTrackidx
* @param track
* @param pucData
* @param uiLen
*/
virtual void onRtcpPacket(int iTrackidx, SdpTrack::Ptr &track, unsigned char *pucData, unsigned int uiLen);
private:
//处理options方法,获取服务器能力
void handleReq_Options(const Parser &parser);
@@ -127,103 +107,102 @@ private:
void handleReq_Post(const Parser &parser);
//处理SET_PARAMETER、GET_PARAMETER方法，一般用于心跳
void handleReq_SET_PARAMETER(const Parser &parser);
//rtsp资源未找到
void inline send_StreamNotFound();
void send_StreamNotFound();
//不支持的传输模式
void inline send_UnsupportedTransport();
void send_UnsupportedTransport();
//会话id错误
void inline send_SessionNotFound();
void send_SessionNotFound();
//一般rtsp服务器打开端口失败时触发
void inline send_NotAcceptable();
//ssrc转字符串
inline string printSSRC(uint32_t ui32Ssrc);
void send_NotAcceptable();
//获取track下标
inline int getTrackIndexByTrackType(TrackType type);
inline int getTrackIndexByControlSuffix(const string &controlSuffix);
inline int getTrackIndexByInterleaved(int interleaved);
int getTrackIndexByTrackType(TrackType type);
int getTrackIndexByControlSuffix(const string &control_suffix);
int getTrackIndexByInterleaved(int interleaved);
//一般用于接收udp打洞包，也用于rtsp推流
inline void onRcvPeerUdpData(int intervaled, const Buffer::Ptr &pBuf, const struct sockaddr &addr);
void onRcvPeerUdpData(int interleaved, const Buffer::Ptr &buf, const struct sockaddr &addr);
//配合onRcvPeerUdpData使用
inline void startListenPeerUdpData(int iTrackIdx);
void startListenPeerUdpData(int track_idx);
////rtsp专有认证相关////
//认证成功
void onAuthSuccess();
//认证失败
void onAuthFailed(const string &realm,const string &why,bool close = true);
void onAuthFailed(const string &realm, const string &why, bool close = true);
//开始走rtsp专有认证流程
void onAuthUser(const string &realm,const string &authorization);
void onAuthUser(const string &realm, const string &authorization);
//校验base64方式的认证加密
void onAuthBasic(const string &realm,const string &strBase64);
void onAuthBasic(const string &realm, const string &auth_base64);
//校验md5方式的认证加密
void onAuthDigest(const string &realm,const string &strMd5);
void onAuthDigest(const string &realm, const string &auth_md5);
//触发url鉴权事件
void emitOnPlay();
//发送rtp给客户端
void sendRtpPacket(const RtspMediaSource::RingDataType &pkt);
//触发rtcp发送
void onSendRtpPacket(const RtpPacket::Ptr &rtp);
//回复客户端
bool sendRtspResponse(const string &res_code,const std::initializer_list<string> &header, const string &sdp = "" , const char *protocol = "RTSP/1.0");
bool sendRtspResponse(const string &res_code,const StrCaseMap &header = StrCaseMap(), const string &sdp = "",const char *protocol = "RTSP/1.0");
bool sendRtspResponse(const string &res_code, const std::initializer_list<string> &header, const string &sdp = "", const char *protocol = "RTSP/1.0");
bool sendRtspResponse(const string &res_code, const StrCaseMap &header = StrCaseMap(), const string &sdp = "", const char *protocol = "RTSP/1.0");
//服务器发送rtcp
void sendSenderReport(bool overTcp,int iTrackIndex);
void sendSenderReport(bool over_tcp, int track_idx);
//设置socket标志
void setSocketFlags();
private:
//用于判断客户端是否超时
Ticker _ticker;
//收到的seq，回复时一致
int _iCseq = 0;
//ContentBase
string _strContentBase;
//Session号
string _strSession;
//记录是否需要rtsp专属鉴权，防止重复触发事件
string _rtsp_realm;
//是否已经触发on_play事件
bool _emit_on_play = false;
//url解析后保存的相关信息
MediaInfo _mediaInfo;
//rtsp播放器绑定的直播源
std::weak_ptr<RtspMediaSource> _pMediaSrc;
//直播源读取器
RtspMediaSource::RingType::RingReader::Ptr _pRtpReader;
//是否开始发送rtp
bool _enable_send_rtp;
//推流或拉流客户端采用的rtp传输方式
Rtsp::eRtpType _rtpType = Rtsp::RTP_Invalid;
Rtsp::eRtpType _rtp_type = Rtsp::RTP_Invalid;
//收到的seq，回复时一致
int _cseq = 0;
//rtsp推流起始时间戳，目的是为了同步
int64_t _start_stamp[2] = {-1, -1};
//消耗的总流量
uint64_t _bytes_usage = 0;
//ContentBase
string _content_base;
//Session号
string _sessionid;
//记录是否需要rtsp专属鉴权，防止重复触发事件
string _rtsp_realm;
//登录认证
string _auth_nonce;
//用于判断客户端是否超时
Ticker _alive_ticker;
//url解析后保存的相关信息
MediaInfo _media_info;
//rtsp推流相关绑定的源
RtspMediaSourceImp::Ptr _push_src;
//rtsp播放器绑定的直播源
std::weak_ptr<RtspMediaSource> _play_src;
//直播源读取器
RtspMediaSource::RingType::RingReader::Ptr _play_reader;
//sdp里面有效的track,包含音频或视频
vector<SdpTrack::Ptr> _aTrackInfo;
vector<SdpTrack::Ptr> _sdp_track;
//rtcp统计,trackid idx 为数组下标
RtcpCounter _rtcp_counter[2];
//rtcp发送时间,trackid idx 为数组下标
Ticker _rtcp_send_tickers[2];
////////RTP over udp////////
//RTP端口,trackid idx 为数组下标
Socket::Ptr _apRtpSock[2];
Socket::Ptr _rtp_socks[2];
//RTCP端口,trackid idx 为数组下标
Socket::Ptr _apRtcpSock[2];
Socket::Ptr _rtcp_socks[2];
//标记是否收到播放的udp打洞包,收到播放的udp打洞包后才能知道其外网udp端口号
unordered_set<int> _udpSockConnected;
unordered_set<int> _udp_connected_flags;
////////RTP over udp_multicast////////
//共享的rtp组播对象
RtpMultiCaster::Ptr _multicaster;
//登录认证
string _strNonce;
//消耗的总流量
uint64_t _ui64TotalBytes = 0;
//RTSP over HTTP
////////RTSP over HTTP ////////
//quicktime 请求rtsp会产生两次tcp连接，
//一次发送 get 一次发送post，需要通过x-sessioncookie关联起来
string _http_x_sessioncookie;
function<void(const Buffer::Ptr &pBuf)> _onRecv;
//是否开始发送rtp
bool _enableSendRtp;
//rtsp推流相关
RtspMediaSourceImp::Ptr _pushSrc;
//rtcp统计,trackid idx 为数组下标
RtcpCounter _aRtcpCnt[2];
//rtcp发送时间,trackid idx 为数组下标
Ticker _aRtcpTicker[2];
function<void(const Buffer::Ptr &)> _on_recv;
};
/**