amass/ZLMediaKit
1
0
Fork 0
mirror of https://github.com/ZLMediaKit/ZLMediaKit.git synced 2024-11-26 12:37:09 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

优化与拦截代码

Browse Source
This commit is contained in:
xiongziliang 2020-09-06 17:52:07 +08:00
parent 74d6689a65
commit eca9236955
7 changed files with 315 additions and 323 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

24
server/FFmpegSource.cpp
View File

@ -245,30 +245,6 @@ bool FFmpegSource::close(MediaSource &sender, bool force) {
return true; return true;
} }
int FFmpegSource::totalReaderCount(MediaSource &sender) {
auto listener = _listener.lock();
if(listener){
return listener->totalReaderCount(sender);
}
return sender.readerCount();
}
void FFmpegSource::onNoneReader(MediaSource &sender){
auto listener = _listener.lock();
if(listener){
listener->onNoneReader(sender);
return;
}
MediaSourceEvent::onNoneReader(sender);
}
void FFmpegSource::onRegist(MediaSource &sender, bool regist){
auto listener = _listener.lock();
if(listener){
listener->onRegist(sender, regist);
}
}
void FFmpegSource::onGetMediaSource(const MediaSource::Ptr &src) { void FFmpegSource::onGetMediaSource(const MediaSource::Ptr &src) {
auto listener = src->getListener(); auto listener = src->getListener();
if (listener.lock().get() != this) { if (listener.lock().get() != this) {

6
server/FFmpegSource.h
View File

@ -40,7 +40,7 @@ private:
~FFmpegSnap() = delete; ~FFmpegSnap() = delete;
}; };
class FFmpegSource : public std::enable_shared_from_this<FFmpegSource> , public MediaSourceEvent{ class FFmpegSource : public std::enable_shared_from_this<FFmpegSource> , public MediaSourceEventInterceptor{
public: public:
typedef shared_ptr<FFmpegSource> Ptr; typedef shared_ptr<FFmpegSource> Ptr;
typedef function<void(const SockException &ex)> onPlay; typedef function<void(const SockException &ex)> onPlay;
@ -60,9 +60,6 @@ private:
//MediaSourceEvent override //MediaSourceEvent override
bool close(MediaSource &sender,bool force) override; bool close(MediaSource &sender,bool force) override;
int totalReaderCount(MediaSource &sender) override;
void onNoneReader(MediaSource &sender) override;
void onRegist(MediaSource &sender, bool regist) override;
private: private:
Process _process; Process _process;
@ -72,7 +69,6 @@ private:
string _src_url; string _src_url;
string _dst_url; string _dst_url;
function<void()> _onClose; function<void()> _onClose;
std::weak_ptr<MediaSourceEvent> _listener;
Ticker _replay_ticker; Ticker _replay_ticker;
}; };

7
src/Common/MediaSink.h
View File

@ -36,6 +36,11 @@ public:
*/ */
virtual void addTrack(const Track::Ptr & track) = 0; virtual void addTrack(const Track::Ptr & track) = 0;
/**
* 添加所有Track完毕
*/
virtual void addTrackCompleted() {}
/** /**
* 重置track * 重置track
*/ */
@ -70,7 +75,7 @@ public:
* 这样会增加生成流的延时，如果添加了音视频双Track，那么可以不调用此方法 * 这样会增加生成流的延时，如果添加了音视频双Track，那么可以不调用此方法
* 否则为了降低流注册延时，请手动调用此方法 * 否则为了降低流注册延时，请手动调用此方法
*/ */
void addTrackCompleted(); void addTrackCompleted() override;
/** /**
* 重置track * 重置track

314
src/Common/MediaSource.cpp
View File

@ -17,16 +17,19 @@
using namespace toolkit; using namespace toolkit;
namespace mediakit { namespace mediakit {
recursive_mutex MediaSource::g_mtxMediaSrc; recursive_mutex s_media_source_mtx;
MediaSource::SchemaVhostAppStreamMap MediaSource::g_mapMediaSrc; MediaSource::SchemaVhostAppStreamMap s_media_source_map;
MediaSource::MediaSource(const string &strSchema, const string &strVhost, const string &strApp, const string &strId) : MediaSource::MediaSource(const string &schema, const string &vhost, const string &app, const string &stream_id){
_strSchema(strSchema), _strApp(strApp), _strId(strId) { GET_CONFIG(bool, enableVhost, General::kEnableVhost);
if (strVhost.empty()) { if (!enableVhost) {
_strVhost = DEFAULT_VHOST; _vhost = DEFAULT_VHOST;
} else { } else {
_strVhost = strVhost; _vhost = vhost.empty() ? DEFAULT_VHOST : vhost;
} }
_schema = schema;
_app = app;
_stream_id = stream_id;
} }
MediaSource::~MediaSource() { MediaSource::~MediaSource() {
@ -34,26 +37,26 @@ MediaSource::~MediaSource() {
} }
const string& MediaSource::getSchema() const { const string& MediaSource::getSchema() const {
return _strSchema; return _schema;
} }
const string& MediaSource::getVhost() const { const string& MediaSource::getVhost() const {
return _strVhost; return _vhost;
} }
const string& MediaSource::getApp() const { const string& MediaSource::getApp() const {
//获取该源的id //获取该源的id
return _strApp; return _app;
} }
const string& MediaSource::getId() const { const string& MediaSource::getId() const {
return _strId; return _stream_id;
} }
vector<Track::Ptr> MediaSource::getTracks(bool trackReady) const { vector<Track::Ptr> MediaSource::getTracks(bool ready) const {
auto strongPtr = _track_source.lock(); auto strongPtr = _track_source.lock();
if(strongPtr){ if(strongPtr){
return strongPtr->getTracks(trackReady); return strongPtr->getTracks(ready);
} }
return vector<Track::Ptr>(); return vector<Track::Ptr>();
} }
@ -77,12 +80,13 @@ int MediaSource::totalReaderCount(){
} }
return listener->totalReaderCount(*this); return listener->totalReaderCount(*this);
} }
bool MediaSource::seekTo(uint32_t ui32Stamp) {
bool MediaSource::seekTo(uint32_t stamp) {
auto listener = _listener.lock(); auto listener = _listener.lock();
if(!listener){ if(!listener){
return false; return false;
} }
return listener->seekTo(*this,ui32Stamp); return listener->seekTo(*this, stamp);
} }
bool MediaSource::close(bool force) { bool MediaSource::close(bool force) {
@ -121,12 +125,12 @@ bool MediaSource::isRecording(Recorder::type type){
} }
void MediaSource::for_each_media(const function<void(const MediaSource::Ptr &src)> &cb) { void MediaSource::for_each_media(const function<void(const MediaSource::Ptr &src)> &cb) {
decltype(g_mapMediaSrc) copy; decltype(s_media_source_map) copy;
{ {
//拷贝g_mapMediaSrc后再遍历，考虑到是高频使用的全局单例锁，并且在上锁时会执行回调代码 //拷贝s_media_source_map后再遍历，考虑到是高频使用的全局单例锁，并且在上锁时会执行回调代码
//很容易导致多个锁交叉死锁的情况，而且该函数使用频率不高，拷贝开销相对来说是可以接受的 //很容易导致多个锁交叉死锁的情况，而且该函数使用频率不高，拷贝开销相对来说是可以接受的
lock_guard<recursive_mutex> lock(g_mtxMediaSrc); lock_guard<recursive_mutex> lock(s_media_source_mtx);
copy = g_mapMediaSrc; copy = s_media_source_map;
} }
for (auto &pr0 : copy) { for (auto &pr0 : copy) {
@ -181,42 +185,75 @@ 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){ static MediaSource::Ptr find_l(const string &schema, const string &vhost_in, const string &app, const string &id, bool create_new) {
auto src = MediaSource::find_l(info._schema, info._vhost, info._app, info._streamid, true); string vhost = vhost_in;
GET_CONFIG(bool,enableVhost,General::kEnableVhost);
if(vhost.empty() || !enableVhost){
vhost = DEFAULT_VHOST;
}
MediaSource::Ptr ret;
{
lock_guard<recursive_mutex> lock(s_media_source_mtx);
//查找某一媒体源，找到后返回
searchMedia(s_media_source_map, schema, vhost, app, id,
[&](MediaSource::SchemaVhostAppStreamMap::iterator &it0, MediaSource::VhostAppStreamMap::iterator &it1,
MediaSource::AppStreamMap::iterator &it2, MediaSource::StreamMap::iterator &it3) {
ret = it3->second.lock();
if (!ret) {
//该对象已经销毁
it2->second.erase(it3);
eraseIfEmpty(s_media_source_map, it0, it1, it2);
return false;
}
return true;
});
}
if(!ret && create_new){
//未查找媒体源，则创建一个
ret = MediaSource::createFromMP4(schema, vhost, app, id);
}
return ret;
}
static void findAsync_l(const MediaInfo &info, const std::shared_ptr<TcpSession> &session, bool retry,
const function<void(const MediaSource::Ptr &src)> &cb){
auto src = find_l(info._schema, info._vhost, info._app, info._streamid, true);
if (src || !retry) { if (src || !retry) {
cb(src); cb(src);
return; return;
} }
void *listener_tag = session.get(); void *listener_tag = session.get();
weak_ptr<TcpSession> weakSession = session; weak_ptr<TcpSession> weak_session = session;
GET_CONFIG(int, maxWaitMS, General::kMaxStreamWaitTimeMS); GET_CONFIG(int, maxWaitMS, General::kMaxStreamWaitTimeMS);
auto onTimeout = session->getPoller()->doDelayTask(maxWaitMS, [cb, listener_tag]() { auto on_timeout = session->getPoller()->doDelayTask(maxWaitMS, [cb, listener_tag]() {
//最多等待一定时间，如果这个时间内，流未注册上，那么返回未找到流 //最多等待一定时间，如果这个时间内，流未注册上，那么返回未找到流
NoticeCenter::Instance().delListener(listener_tag, Broadcast::kBroadcastMediaChanged); NoticeCenter::Instance().delListener(listener_tag, Broadcast::kBroadcastMediaChanged);
cb(nullptr); cb(nullptr);
return 0; return 0;
}); });
auto cancelAll = [onTimeout, listener_tag]() { auto cancel_all = [on_timeout, listener_tag]() {
//取消延时任务，防止多次回调 //取消延时任务，防止多次回调
onTimeout->cancel(); on_timeout->cancel();
//取消媒体注册事件监听 //取消媒体注册事件监听
NoticeCenter::Instance().delListener(listener_tag, Broadcast::kBroadcastMediaChanged); NoticeCenter::Instance().delListener(listener_tag, Broadcast::kBroadcastMediaChanged);
}; };
function<void()> closePlayer = [cb, cancelAll]() { function<void()> close_player = [cb, cancel_all]() {
cancelAll(); cancel_all();
//告诉播放器，流不存在，这样会立即断开播放器 //告诉播放器，流不存在，这样会立即断开播放器
cb(nullptr); cb(nullptr);
}; };
auto onRegist = [weakSession, info, cb, cancelAll](BroadcastMediaChangedArgs) { auto on_regist = [weak_session, info, cb, cancel_all](BroadcastMediaChangedArgs) {
auto strongSession = weakSession.lock(); auto strong_session = weak_session.lock();
if (!strongSession) { if (!strong_session) {
//自己已经销毁 //自己已经销毁
cancelAll(); cancel_all();
return; return;
} }
@ -229,11 +266,11 @@ void MediaSource::findAsync_l(const MediaInfo &info, const std::shared_ptr<TcpSe
return; return;
} }
cancelAll(); cancel_all();
//播发器请求的流终于注册上了，切换到自己的线程再回复 //播发器请求的流终于注册上了，切换到自己的线程再回复
strongSession->async([weakSession, info, cb]() { strong_session->async([weak_session, info, cb]() {
auto strongSession = weakSession.lock(); auto strongSession = weak_session.lock();
if (!strongSession) { if (!strongSession) {
return; return;
} }
@ -244,9 +281,9 @@ void MediaSource::findAsync_l(const MediaInfo &info, const std::shared_ptr<TcpSe
}; };
//监听媒体注册事件 //监听媒体注册事件
NoticeCenter::Instance().addListener(listener_tag, Broadcast::kBroadcastMediaChanged, onRegist); NoticeCenter::Instance().addListener(listener_tag, Broadcast::kBroadcastMediaChanged, on_regist);
//广播未找到流,此时可以立即去拉流，这样还来得及 //广播未找到流,此时可以立即去拉流，这样还来得及
NoticeCenter::Instance().emitEvent(Broadcast::kBroadcastNotFoundStream, info, static_cast<SockInfo &>(*session), closePlayer); NoticeCenter::Instance().emitEvent(Broadcast::kBroadcastNotFoundStream, info, static_cast<SockInfo &>(*session), close_player);
} }
void MediaSource::findAsync(const MediaInfo &info, const std::shared_ptr<TcpSession> &session,const function<void(const Ptr &src)> &cb){ void MediaSource::findAsync(const MediaInfo &info, const std::shared_ptr<TcpSession> &session,const function<void(const Ptr &src)> &cb){
@ -257,54 +294,9 @@ MediaSource::Ptr MediaSource::find(const string &schema, const string &vhost, co
return find_l(schema, vhost, app, id, false); return find_l(schema, vhost, app, id, false);
} }
MediaSource::Ptr MediaSource::find_l(const string &schema, const string &vhost_tmp, const string &app, const string &id, bool bMake) { static string getTrackInfoStr(const TrackSource *track_src){
string vhost = vhost_tmp;
if(vhost.empty()){
vhost = DEFAULT_VHOST;
}
GET_CONFIG(bool,enableVhost,General::kEnableVhost);
if(!enableVhost){
vhost = DEFAULT_VHOST;
}
MediaSource::Ptr ret;
{
lock_guard<recursive_mutex> lock(g_mtxMediaSrc);
//查找某一媒体源，找到后返回
searchMedia(g_mapMediaSrc, schema, vhost, app, id, [&](SchemaVhostAppStreamMap::iterator &it0,
VhostAppStreamMap::iterator &it1,
AppStreamMap::iterator &it2,
StreamMap::iterator &it3) {
ret = it3->second.lock();
if (!ret) {
//该对象已经销毁
it2->second.erase(it3);
eraseIfEmpty(g_mapMediaSrc, it0, it1, it2);
return false;
}
return true;
});
}
if(!ret && bMake){
//未查找媒体源，则创建一个
ret = createFromMP4(schema, vhost, app, id);
}
return ret;
}
void MediaSource::regist() {
GET_CONFIG(bool,enableVhost,General::kEnableVhost);
if(!enableVhost){
_strVhost = DEFAULT_VHOST;
}
//注册该源，注册后服务器才能找到该源
{
lock_guard<recursive_mutex> lock(g_mtxMediaSrc);
g_mapMediaSrc[_strSchema][_strVhost][_strApp][_strId] = shared_from_this();
}
_StrPrinter codec_info; _StrPrinter codec_info;
auto tracks = getTracks(true); auto tracks = track_src->getTracks(true);
for (auto &track : tracks) { for (auto &track : tracks) {
auto codec_type = track->getTrackType(); auto codec_type = track->getTrackType();
codec_info << track->getCodecName(); codec_info << track->getCodecName();
@ -329,45 +321,51 @@ void MediaSource::regist() {
break; break;
} }
} }
return codec_info;
}
InfoL << _strSchema << " " << _strVhost << " " << _strApp << " " << _strId << " " << codec_info; void MediaSource::emitEvent(bool regist){
NoticeCenter::Instance().emitEvent(Broadcast::kBroadcastMediaChanged, true, *this);
auto listener = _listener.lock(); auto listener = _listener.lock();
if (listener) { if (listener) {
listener->onRegist(*this, true); //触发回调
listener->onRegist(*this, regist);
} }
//触发广播
NoticeCenter::Instance().emitEvent(Broadcast::kBroadcastMediaChanged, regist, *this);
InfoL << (regist ? "媒体注册:" : "媒体注销:") << _schema << " " << _vhost << " " << _app << " " << _stream_id << " " << getTrackInfoStr(this);
}
void MediaSource::regist() {
{
//减小互斥锁临界区
lock_guard<recursive_mutex> lock(s_media_source_mtx);
s_media_source_map[_schema][_vhost][_app][_stream_id] = shared_from_this();
}
emitEvent(true);
} }
//反注册该源 //反注册该源
bool MediaSource::unregist() { bool MediaSource::unregist() {
bool ret; bool ret;
{ {
lock_guard<recursive_mutex> lock(g_mtxMediaSrc); //减小互斥锁临界区
ret = searchMedia(g_mapMediaSrc, _strSchema, _strVhost, _strApp, _strId, lock_guard<recursive_mutex> lock(s_media_source_mtx);
[&](SchemaVhostAppStreamMap::iterator &it0, ret = searchMedia(s_media_source_map, _schema, _vhost, _app, _stream_id,
VhostAppStreamMap::iterator &it1, [&](SchemaVhostAppStreamMap::iterator &it0, VhostAppStreamMap::iterator &it1,
AppStreamMap::iterator &it2, AppStreamMap::iterator &it2, StreamMap::iterator &it3) {
StreamMap::iterator &it3) { auto strong_self = it3->second.lock();
auto strongMedia = it3->second.lock(); if (strong_self && this != strong_self.get()) {
if (strongMedia && this != strongMedia.get()) {
//不是自己,不允许反注册 //不是自己,不允许反注册
return false; return false;
} }
it2->second.erase(it3); it2->second.erase(it3);
eraseIfEmpty(g_mapMediaSrc, it0, it1, it2); eraseIfEmpty(s_media_source_map, it0, it1, it2);
return true; return true;
}); });
} }
if (ret) { if (ret) {
InfoL << _strSchema << " " << _strVhost << " " << _strApp << " " << _strId; emitEvent(false);
NoticeCenter::Instance().emitEvent(Broadcast::kBroadcastMediaChanged, false, *this);
auto listener = _listener.lock();
if (listener) {
listener->onRegist(*this, false);
}
} }
return ret; return ret;
} }
@ -403,23 +401,23 @@ void MediaInfo::parse(const string &url){
_app = split_vec[1]; _app = split_vec[1];
} }
if (split_vec.size() > 2) { if (split_vec.size() > 2) {
string steamid; string stream_id;
for (int i = 2; i < split_vec.size(); ++i) { for (int i = 2; i < split_vec.size(); ++i) {
steamid.append(split_vec[i] + "/"); stream_id.append(split_vec[i] + "/");
} }
if(steamid.back() == '/'){ if (stream_id.back() == '/') {
steamid.pop_back(); stream_id.pop_back();
} }
auto pos = steamid.find("?"); auto pos = stream_id.find("?");
if (pos != string::npos) { if (pos != string::npos) {
_streamid = steamid.substr(0,pos); _streamid = stream_id.substr(0, pos);
_param_strs = steamid.substr(pos + 1); _param_strs = stream_id.substr(pos + 1);
auto params = Parser::parseArgs(_param_strs); auto params = Parser::parseArgs(_param_strs);
if (params.find(VHOST_KEY) != params.end()) { if (params.find(VHOST_KEY) != params.end()) {
_vhost = params[VHOST_KEY]; _vhost = params[VHOST_KEY];
} }
} else { } else {
_streamid = steamid; _streamid = stream_id;
} }
} }
@ -430,14 +428,34 @@ void MediaInfo::parse(const string &url){
} }
} }
MediaSource::Ptr MediaSource::createFromMP4(const string &schema, const string &vhost, const string &app, const string &stream, const string &file_path , bool check_app){
GET_CONFIG(string, appName, Record::kAppName);
if (check_app && app != appName) {
return nullptr;
}
#ifdef ENABLE_MP4
try {
MP4Reader::Ptr pReader(new MP4Reader(vhost, app, stream, file_path));
pReader->startReadMP4();
return MediaSource::find(schema, vhost, app, stream);
} catch (std::exception &ex) {
WarnL << ex.what();
return nullptr;
}
#else
WarnL << "创建MP4点播失败，请编译时打开\"ENABLE_MP4\"选项";
return nullptr;
#endif //ENABLE_MP4
}
/////////////////////////////////////MediaSourceEvent////////////////////////////////////// /////////////////////////////////////MediaSourceEvent//////////////////////////////////////
void MediaSourceEvent::onNoneReader(MediaSource &sender){ void MediaSourceEvent::onNoneReader(MediaSource &sender){
GET_CONFIG(string, recordApp, Record::kAppName); GET_CONFIG(string, record_app, Record::kAppName);
GET_CONFIG(int, stream_none_reader_delay, General::kStreamNoneReaderDelayMS); GET_CONFIG(int, stream_none_reader_delay, General::kStreamNoneReaderDelayMS);
//如果mp4点播, 无人观看时我们强制关闭点播 //如果mp4点播, 无人观看时我们强制关闭点播
bool is_mp4_vod = sender.getApp() == recordApp; bool is_mp4_vod = sender.getApp() == record_app;
//没有任何人观看该视频源，表明该源可以关闭了 //没有任何人观看该视频源，表明该源可以关闭了
weak_ptr<MediaSource> weakSender = sender.shared_from_this(); weak_ptr<MediaSource> weakSender = sender.shared_from_this();
@ -475,25 +493,63 @@ void MediaSourceEvent::onNoneReader(MediaSource &sender){
}, nullptr); }, nullptr);
} }
MediaSource::Ptr MediaSource::createFromMP4(const string &schema, const string &vhost, const string &app, const string &stream, const string &filePath , bool checkApp){ bool MediaSourceEventInterceptor::seekTo(MediaSource &sender, uint32_t stamp) {
GET_CONFIG(string, appName, Record::kAppName); auto listener = _listener.lock();
if (checkApp && app != appName) { if (!listener) {
return nullptr; return false;
} }
#ifdef ENABLE_MP4 return listener->seekTo(sender, stamp);
try {
MP4Reader::Ptr pReader(new MP4Reader(vhost, app, stream, filePath));
pReader->startReadMP4();
return MediaSource::find(schema, vhost, app, stream);
} catch (std::exception &ex) {
WarnL << ex.what();
return nullptr;
} }
#else
WarnL << "创建MP4点播失败，请编译时打开\"ENABLE_MP4\"选项"; bool MediaSourceEventInterceptor::close(MediaSource &sender, bool force) {
return nullptr; auto listener = _listener.lock();
#endif //ENABLE_MP4 if (!listener) {
return false;
} }
return listener->close(sender, force);
}
int MediaSourceEventInterceptor::totalReaderCount(MediaSource &sender) {
auto listener = _listener.lock();
if (!listener) {
return sender.readerCount();
}
return listener->totalReaderCount(sender);
}
void MediaSourceEventInterceptor::onNoneReader(MediaSource &sender) {
auto listener = _listener.lock();
if (!listener) {
MediaSourceEvent::onNoneReader(sender);
return;
}
listener->onNoneReader(sender);
}
void MediaSourceEventInterceptor::onRegist(MediaSource &sender, bool regist) {
auto listener = _listener.lock();
if (listener) {
listener->onRegist(sender, regist);
}
}
bool MediaSourceEventInterceptor::setupRecord(MediaSource &sender, Recorder::type type, bool start, const string &custom_path) {
auto listener = _listener.lock();
if (!listener) {
return false;
}
return listener->setupRecord(sender, type, start, custom_path);
}
bool MediaSourceEventInterceptor::isRecording(MediaSource &sender, Recorder::type type) {
auto listener = _listener.lock();
if (!listener) {
return false;
}
return listener->isRecording(sender, type);
}
/////////////////////////////////////FlushPolicy//////////////////////////////////////
static bool isFlushAble_default(bool is_video, uint32_t last_stamp, uint32_t new_stamp, int cache_size) { static bool isFlushAble_default(bool is_video, uint32_t last_stamp, uint32_t new_stamp, int cache_size) {
if (new_stamp + 500 < last_stamp) { if (new_stamp + 500 < last_stamp) {

73
src/Common/MediaSource.h
View File

@ -44,33 +44,54 @@ public:
virtual ~MediaSourceEvent(){}; virtual ~MediaSourceEvent(){};
// 通知拖动进度条 // 通知拖动进度条
virtual bool seekTo(MediaSource &sender,uint32_t ui32Stamp){ return false; } virtual bool seekTo(MediaSource &sender, uint32_t stamp) { return false; }
// 通知其停止推流 // 通知其停止产生流
virtual bool close(MediaSource &sender, bool force) { return false; } virtual bool close(MediaSource &sender, bool force) { return false; }
// 观看总人数 // 获取观看总人数
virtual int totalReaderCount(MediaSource &sender) = 0; virtual int totalReaderCount(MediaSource &sender) = 0;
// 开启或关闭录制
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; };
// 通知无人观看 // 通知无人观看
virtual void onNoneReader(MediaSource &sender); virtual void onNoneReader(MediaSource &sender);
//流注册或注销事件 //流注册或注销事件
virtual void onRegist(MediaSource &sender, bool regist) {}; virtual void onRegist(MediaSource &sender, bool regist) {};
////////////////////////仅供MultiMediaSourceMuxer对象继承////////////////////////
// 开启或关闭录制
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: private:
Timer::Ptr _async_close_timer; Timer::Ptr _async_close_timer;
}; };
//该对象用于拦截感兴趣的MediaSourceEvent事件
class MediaSourceEventInterceptor : public MediaSourceEvent{
public:
MediaSourceEventInterceptor(){}
~MediaSourceEventInterceptor() override {}
bool seekTo(MediaSource &sender, uint32_t stamp) override;
bool close(MediaSource &sender, bool force) override;
int totalReaderCount(MediaSource &sender) override;
void onNoneReader(MediaSource &sender) override;
void onRegist(MediaSource &sender, bool regist) override;;
bool setupRecord(MediaSource &sender, Recorder::type type, bool start, const string &custom_path) override;
bool isRecording(MediaSource &sender, Recorder::type type) override;
protected:
std::weak_ptr<MediaSourceEvent> _listener;
};
/** /**
* 解析url获取媒体相关信息 * 解析url获取媒体相关信息
*/ */
class MediaInfo{ class MediaInfo{
public: public:
MediaInfo(){}
~MediaInfo() {} ~MediaInfo() {}
MediaInfo() {}
MediaInfo(const string &url) { parse(url); } MediaInfo(const string &url) { parse(url); }
void parse(const string &url); void parse(const string &url);
public: public:
string _schema; string _schema;
string _host; string _host;
@ -92,7 +113,7 @@ public:
typedef unordered_map<string, AppStreamMap > VhostAppStreamMap; typedef unordered_map<string, AppStreamMap > VhostAppStreamMap;
typedef unordered_map<string, VhostAppStreamMap > SchemaVhostAppStreamMap; typedef unordered_map<string, VhostAppStreamMap > SchemaVhostAppStreamMap;
MediaSource(const string &strSchema, const string &strVhost, const string &strApp, const string &strId) ; MediaSource(const string &schema, const string &vhost, const string &app, const string &stream_id) ;
virtual ~MediaSource() ; virtual ~MediaSource() ;
// 获取协议类型 // 获取协议类型
@ -107,7 +128,7 @@ public:
// 设置TrackSource // 设置TrackSource
void setTrackSource(const std::weak_ptr<TrackSource> &track_src); void setTrackSource(const std::weak_ptr<TrackSource> &track_src);
// 获取所有Track // 获取所有Track
vector<Track::Ptr> getTracks(bool trackReady = true) const override; vector<Track::Ptr> getTracks(bool ready = true) const override;
// 设置监听者 // 设置监听者
virtual void setListener(const std::weak_ptr<MediaSourceEvent> &listener); virtual void setListener(const std::weak_ptr<MediaSourceEvent> &listener);
@ -120,12 +141,12 @@ public:
virtual int totalReaderCount(); virtual int totalReaderCount();
// 获取流当前时间戳 // 获取流当前时间戳
virtual uint32_t getTimeStamp(TrackType trackType) { return 0; }; virtual uint32_t getTimeStamp(TrackType type) { return 0; };
// 设置时间戳 // 设置时间戳
virtual void setTimeStamp(uint32_t uiStamp) {}; virtual void setTimeStamp(uint32_t stamp) {};
// 拖动进度条 // 拖动进度条
bool seekTo(uint32_t ui32Stamp); bool seekTo(uint32_t stamp);
// 关闭该流 // 关闭该流
bool close(bool force); bool close(bool force);
// 该流无人观看 // 该流无人观看
@ -141,26 +162,26 @@ public:
static void findAsync(const MediaInfo &info, const std::shared_ptr<TcpSession> &session, const function<void(const Ptr &src)> &cb); static void findAsync(const MediaInfo &info, const std::shared_ptr<TcpSession> &session, const function<void(const Ptr &src)> &cb);
// 遍历所有流 // 遍历所有流
static void for_each_media(const function<void(const Ptr &src)> &cb); static void for_each_media(const function<void(const Ptr &src)> &cb);
// 从mp4文件生成MediaSource // 从mp4文件生成MediaSource
static MediaSource::Ptr createFromMP4(const string &schema, const string &vhost, const string &app, const string &stream, const string &filePath = "", bool checkApp = true); static MediaSource::Ptr createFromMP4(const string &schema, const string &vhost, const string &app, const string &stream, const string &file_path = "", bool check_app = true);
protected: protected:
//媒体注册
void regist(); void regist();
bool unregist();
private: private:
static Ptr find_l(const string &schema, const string &vhost, const string &app, const string &id, bool bMake); //媒体注销
static void findAsync_l(const MediaInfo &info, const std::shared_ptr<TcpSession> &session, bool retry, const function<void(const MediaSource::Ptr &src)> &cb); bool unregist();
//触发媒体事件
void emitEvent(bool regist);
private: private:
string _strSchema; string _schema;
string _strVhost; string _vhost;
string _strApp; string _app;
string _strId; string _stream_id;
std::weak_ptr<MediaSourceEvent> _listener;
weak_ptr<TrackSource> _track_source; weak_ptr<TrackSource> _track_source;
static SchemaVhostAppStreamMap g_mapMediaSrc; std::weak_ptr<MediaSourceEvent> _listener;
static recursive_mutex g_mtxMediaSrc;
}; };
///缓存刷新策略类 ///缓存刷新策略类
@ -221,9 +242,9 @@ private:
} }
private: private:
bool _key_pos = false;
policy _policy; policy _policy;
std::shared_ptr<packet_list> _cache; std::shared_ptr<packet_list> _cache;
bool _key_pos = false;
}; };
} /* namespace mediakit */ } /* namespace mediakit */

78
src/Common/MultiMediaSourceMuxer.cpp
View File

@ -11,15 +11,11 @@
#include "MultiMediaSourceMuxer.h" #include "MultiMediaSourceMuxer.h"
namespace mediakit { namespace mediakit {
///////////////////////////////MultiMuxerPrivate//////////////////////////////////
MultiMuxerPrivate::~MultiMuxerPrivate() {} MultiMuxerPrivate::~MultiMuxerPrivate() {}
MultiMuxerPrivate::MultiMuxerPrivate(const string &vhost, MultiMuxerPrivate::MultiMuxerPrivate(const string &vhost, const string &app, const string &stream, float dur_sec,
const string &app, bool enable_rtsp, bool enable_rtmp, bool enable_hls, bool enable_mp4) {
const string &stream,
float dur_sec,
bool enable_rtsp,
bool enable_rtmp,
bool enable_hls,
bool enable_mp4) {
if (enable_rtmp) { if (enable_rtmp) {
_rtmp = std::make_shared<RtmpMediaSourceMuxer>(vhost, app, stream, std::make_shared<TitleMeta>(dur_sec)); _rtmp = std::make_shared<RtmpMediaSourceMuxer>(vhost, app, stream, std::make_shared<TitleMeta>(dur_sec));
_enable_rtxp = true; _enable_rtxp = true;
@ -73,7 +69,7 @@ void MultiMuxerPrivate::setMediaListener(const std::weak_ptr<MediaSourceEvent> &
if (hls_src) { if (hls_src) {
hls_src->setListener(listener); hls_src->setListener(listener);
} }
_meida_listener = listener; _listener = listener;
} }
int MultiMuxerPrivate::totalReaderCount() const { int MultiMuxerPrivate::totalReaderCount() const {
@ -99,7 +95,7 @@ bool MultiMuxerPrivate::setupRecord(MediaSource &sender, Recorder::type type, bo
auto hls_src = getHlsMediaSource(); auto hls_src = getHlsMediaSource();
if (hls_src) { if (hls_src) {
//设置HlsMediaSource的事件监听器 //设置HlsMediaSource的事件监听器
hls_src->setListener(_meida_listener); hls_src->setListener(_listener);
hls_src->setTrackSource(shared_from_this()); hls_src->setTrackSource(shared_from_this());
} }
} else if (!start && _hls) { } else if (!start && _hls) {
@ -146,7 +142,7 @@ void MultiMuxerPrivate::setTimeStamp(uint32_t stamp) {
} }
void MultiMuxerPrivate::setTrackListener(Listener *listener) { void MultiMuxerPrivate::setTrackListener(Listener *listener) {
_listener = listener; _track_listener = listener;
} }
void MultiMuxerPrivate::onTrackReady(const Track::Ptr &track) { void MultiMuxerPrivate::onTrackReady(const Track::Ptr &track) {
@ -206,8 +202,8 @@ void MultiMuxerPrivate::onAllTrackReady() {
hls_src->setTrackSource(shared_from_this()); hls_src->setTrackSource(shared_from_this());
} }
if (_listener) { if (_track_listener) {
_listener->onAllTrackReady(); _track_listener->onAllTrackReady();
} }
} }
@ -219,22 +215,18 @@ MediaSource::Ptr MultiMuxerPrivate::getHlsMediaSource() const {
return nullptr; return nullptr;
} }
///////////////////////////////////////////////////////////////// ///////////////////////////////MultiMediaSourceMuxer//////////////////////////////////
MultiMediaSourceMuxer::~MultiMediaSourceMuxer() {} MultiMediaSourceMuxer::~MultiMediaSourceMuxer() {}
MultiMediaSourceMuxer::MultiMediaSourceMuxer(const string &vhost,
const string &app, MultiMediaSourceMuxer::MultiMediaSourceMuxer(const string &vhost, const string &app, const string &stream, float dur_sec,
const string &stream, bool enable_rtsp, bool enable_rtmp, bool enable_hls, bool enable_mp4) {
float dur_sec,
bool enable_rtsp,
bool enable_rtmp,
bool enable_hls,
bool enable_mp4) {
_muxer.reset(new MultiMuxerPrivate(vhost, app, stream, dur_sec, enable_rtsp, enable_rtmp, enable_hls, enable_mp4)); _muxer.reset(new MultiMuxerPrivate(vhost, app, stream, dur_sec, enable_rtsp, enable_rtmp, enable_hls, enable_mp4));
_muxer->setTrackListener(this); _muxer->setTrackListener(this);
} }
void MultiMediaSourceMuxer::setMediaListener(const std::weak_ptr<MediaSourceEvent> &listener) { void MultiMediaSourceMuxer::setMediaListener(const std::weak_ptr<MediaSourceEvent> &listener) {
//拦截事件
_muxer->setMediaListener(shared_from_this()); _muxer->setMediaListener(shared_from_this());
_listener = listener; _listener = listener;
} }
@ -255,46 +247,14 @@ vector<Track::Ptr> MultiMediaSourceMuxer::getTracks(bool trackReady) const {
return _muxer->getTracks(trackReady); return _muxer->getTracks(trackReady);
} }
bool MultiMediaSourceMuxer::seekTo(MediaSource &sender, uint32_t ui32Stamp) {
auto listener = _listener.lock();
if (!listener) {
return false;
}
return listener->seekTo(sender, ui32Stamp);
}
bool MultiMediaSourceMuxer::close(MediaSource &sender, bool force) {
auto listener = _listener.lock();
if (!listener) {
return false;
}
return listener->close(sender, force);
}
int MultiMediaSourceMuxer::totalReaderCount(MediaSource &sender) { int MultiMediaSourceMuxer::totalReaderCount(MediaSource &sender) {
auto listener = _listener.lock(); auto listener = _listener.lock();
if (!listener) { if (!listener) {
return _muxer->totalReaderCount(); return totalReaderCount();
} }
return listener->totalReaderCount(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) { bool MultiMediaSourceMuxer::setupRecord(MediaSource &sender, Recorder::type type, bool start, const string &custom_path) {
return _muxer->setupRecord(sender,type,start,custom_path); return _muxer->setupRecord(sender,type,start,custom_path);
} }
@ -313,9 +273,9 @@ void MultiMediaSourceMuxer::addTrackCompleted() {
void MultiMediaSourceMuxer::onAllTrackReady(){ void MultiMediaSourceMuxer::onAllTrackReady(){
_muxer->setMediaListener(shared_from_this()); _muxer->setMediaListener(shared_from_this());
auto track_listener = _track_listener.lock(); auto listener = _track_listener.lock();
if(track_listener){ if(listener){
track_listener->onAllTrackReady(); listener->onAllTrackReady();
} }
} }
@ -370,9 +330,9 @@ public:
return _frame->getCodecId(); return _frame->getCodecId();
} }
private: private:
Frame::Ptr _frame;
int64_t _dts; int64_t _dts;
int64_t _pts; int64_t _pts;
Frame::Ptr _frame;
}; };
void MultiMediaSourceMuxer::inputFrame(const Frame::Ptr &frame) { void MultiMediaSourceMuxer::inputFrame(const Frame::Ptr &frame) {

66
src/Common/MultiMediaSourceMuxer.h
View File

@ -47,17 +47,17 @@ private:
MediaSource::Ptr getHlsMediaSource() const; MediaSource::Ptr getHlsMediaSource() const;
private: private:
bool _enable_rtxp = false;
bool _enable_record = false;
Listener *_track_listener = nullptr;
RtmpMediaSourceMuxer::Ptr _rtmp; RtmpMediaSourceMuxer::Ptr _rtmp;
RtspMediaSourceMuxer::Ptr _rtsp; RtspMediaSourceMuxer::Ptr _rtsp;
MediaSinkInterface::Ptr _hls; MediaSinkInterface::Ptr _hls;
MediaSinkInterface::Ptr _mp4; MediaSinkInterface::Ptr _mp4;
Listener *_listener = nullptr; std::weak_ptr<MediaSourceEvent> _listener;
std::weak_ptr<MediaSourceEvent> _meida_listener;
bool _enable_rtxp = false;
bool _enable_record = false;
}; };
class MultiMediaSourceMuxer : public MediaSourceEvent, public MediaSinkInterface, public TrackSource, public MultiMuxerPrivate::Listener, public std::enable_shared_from_this<MultiMediaSourceMuxer>{ class MultiMediaSourceMuxer : public MediaSourceEventInterceptor, public MediaSinkInterface, public TrackSource, public MultiMuxerPrivate::Listener, public std::enable_shared_from_this<MultiMediaSourceMuxer>{
public: public:
typedef MultiMuxerPrivate::Listener Listener; typedef MultiMuxerPrivate::Listener Listener;
typedef std::shared_ptr<MultiMediaSourceMuxer> Ptr; typedef std::shared_ptr<MultiMediaSourceMuxer> Ptr;
@ -83,12 +83,19 @@ public:
*/ */
int totalReaderCount() const; int totalReaderCount() const;
/**
* 判断是否生效(是否正在转其他协议)
*/
bool isEnabled();
/** /**
* 设置MediaSource时间戳 * 设置MediaSource时间戳
* @param stamp 时间戳 * @param stamp 时间戳
*/ */
void setTimeStamp(uint32_t stamp); void setTimeStamp(uint32_t stamp);
/////////////////////////////////TrackSource override/////////////////////////////////
/** /**
* 获取所有Track * 获取所有Track
* @param trackReady 是否筛选过滤未就绪的track * @param trackReady 是否筛选过滤未就绪的track
@ -96,21 +103,7 @@ public:
*/ */
vector<Track::Ptr> getTracks(bool trackReady = true) const override; vector<Track::Ptr> getTracks(bool trackReady = true) const override;
/** /////////////////////////////////MediaSourceEvent override/////////////////////////////////
* 通知拖动进度条
* @param sender 事件发送者
* @param ui32Stamp 目标时间戳
* @return 是否成功
*/
bool seekTo(MediaSource &sender,uint32_t ui32Stamp) override;
/**
* 通知停止流生成
* @param sender 事件发送者
* @param force 是否强制关闭
* @return 成功与否
*/
bool close(MediaSource &sender,bool force) override;
/** /**
* 观看总人数 * 观看总人数
@ -119,19 +112,6 @@ public:
*/ */
int totalReaderCount(MediaSource &sender) override; 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 录制类型 * @param type 录制类型
@ -148,6 +128,8 @@ public:
*/ */
bool isRecording(MediaSource &sender, Recorder::type type) override; bool isRecording(MediaSource &sender, Recorder::type type) override;
/////////////////////////////////MediaSinkInterface override/////////////////////////////////
/** /**
* 添加track，内部会调用Track的clone方法 * 添加track，内部会调用Track的clone方法
* 只会克隆sps pps这些信息 ，而不会克隆Delegate相关关系 * 只会克隆sps pps这些信息 ，而不会克隆Delegate相关关系
@ -158,12 +140,7 @@ public:
/** /**
* 添加track完毕 * 添加track完毕
*/ */
void addTrackCompleted(); void addTrackCompleted() override;
/**
* 所有track全部就绪
*/
void onAllTrackReady() override;
/** /**
* 重置track * 重置track
@ -176,16 +153,17 @@ public:
*/ */
void inputFrame(const Frame::Ptr &frame) override; void inputFrame(const Frame::Ptr &frame) override;
/////////////////////////////////MultiMuxerPrivate::Listener override/////////////////////////////////
/** /**
* 判断是否生效(是否正在转其他协议) * 所有track全部就绪
*/ */
bool isEnabled(); void onAllTrackReady() override;
private: private:
MultiMuxerPrivate::Ptr _muxer;
std::weak_ptr<MediaSourceEvent> _listener;
std::weak_ptr<MultiMuxerPrivate::Listener> _track_listener;
Stamp _stamp[2]; Stamp _stamp[2];
MultiMuxerPrivate::Ptr _muxer;
std::weak_ptr<MultiMuxerPrivate::Listener> _track_listener;
}; };
}//namespace mediakit }//namespace mediakit