From c503749328b70c7b9ef6d4508ac6caf4467c9bfa Mon Sep 17 00:00:00 2001 From: xiongziliang <771730766@qq.com> Date: Thu, 6 Jun 2019 18:28:33 +0800 Subject: [PATCH] =?UTF-8?q?=E6=B7=BB=E5=8A=A0FFmpeg=E6=8B=89=E6=B5=81?= =?UTF-8?q?=E5=8A=9F=E8=83=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- server/FFmpegSource.cpp | 211 ++++++++++++++++++++++++++++++++++++++++ server/FFmpegSource.h | 47 +++++++++ server/WebApi.cpp | 113 +++++++++++++++++---- server/WebApi.h | 21 ++++ server/main.cpp | 1 + 5 files changed, 373 insertions(+), 20 deletions(-) create mode 100644 server/FFmpegSource.cpp create mode 100644 server/FFmpegSource.h create mode 100644 server/WebApi.h diff --git a/server/FFmpegSource.cpp b/server/FFmpegSource.cpp new file mode 100644 index 00000000..1f0faca3 --- /dev/null +++ b/server/FFmpegSource.cpp @@ -0,0 +1,211 @@ +// +// Created by xzl on 2018/5/24. +// + +#include "FFmpegSource.h" +#include "Common/config.h" +#include "Common/MediaSource.h" +#include "Util/File.h" +#include "System.h" + +namespace FFmpeg { +#define FFmpeg_FIELD "ffmpeg." +const char kBin[] = FFmpeg_FIELD"bin"; +const char kCmd[] = FFmpeg_FIELD"cmd"; +const char kLog[] = FFmpeg_FIELD"log"; + +onceToken token([]() { + mINI::Instance()[kBin] = trim(System::execute("which ffmpeg")); + mINI::Instance()[kCmd] = "%s -i %s -c:a aac -strict -2 -ar 44100 -ab 48k -c:v libx264 -f flv %s"; + mINI::Instance()[kLog] = exeDir() + "ffmpeg/ffmpeg.log"; +}); +} + +FFmpegSource::FFmpegSource() { + _poller = EventPollerPool::Instance().getPoller(); +} + +FFmpegSource::~FFmpegSource() { + NoticeCenter::Instance().delListener(this, Broadcast::kBroadcastStreamNoneReader); + DebugL; +} + + +void FFmpegSource::play(const string &src_url,const string &dst_url,int timeout_ms,const onPlay &cb) { + GET_CONFIG_AND_REGISTER(string,ffmpeg_bin,FFmpeg::kBin); + GET_CONFIG_AND_REGISTER(string,ffmpeg_cmd,FFmpeg::kCmd); + GET_CONFIG_AND_REGISTER(string,ffmpeg_log,FFmpeg::kLog); + + _src_url = src_url; + _dst_url = dst_url; + _media_info.parse(dst_url); + + char cmd[1024] = {0}; + snprintf(cmd, sizeof(cmd),ffmpeg_cmd.data(),ffmpeg_bin.data(),src_url.data(),dst_url.data()); + _process.run(cmd,ffmpeg_log); + InfoL << cmd; + + if(_media_info._host == "127.0.0.1"){ + //推流给自己的，通过判断流是否注册上来判断是否正常 + if(_media_info._schema != RTSP_SCHEMA && _media_info._schema != RTMP_SCHEMA){ + cb(SockException(Err_other,"本服务只支持rtmp/rtsp推流")); + return; + } + weak_ptr weakSelf = shared_from_this(); + findAsync(timeout_ms,[cb,weakSelf,timeout_ms](const MediaSource::Ptr &src){ + auto strongSelf = weakSelf.lock(); + if(!strongSelf){ + //自己已经销毁 + return; + } + if(src){ + //推流给自己成功 + cb(SockException()); + strongSelf->startTimer(timeout_ms); + return; + } + //推流失败 + if(!strongSelf->_process.wait(false)){ + //ffmpeg进程已经退出 + cb(SockException(Err_other,StrPrinter << "ffmpeg已经退出,exit code = " << strongSelf->_process.exit_code())); + return; + } + //ffmpeg进程还在线，但是等待推流超时 + cb(SockException(Err_other,"等待超时")); + }); + } else{ + //推流给其他服务器的，通过判断FFmpeg进程是否在线判断是否成功 + weak_ptr weakSelf = shared_from_this(); + _timer = std::make_shared(timeout_ms / 1000,[weakSelf,cb,timeout_ms](){ + auto strongSelf = weakSelf.lock(); + if(!strongSelf){ + //自身已经销毁 + return false; + } + //FFmpeg还在线，那么我们认为推流成功 + if(strongSelf->_process.wait(false)){ + cb(SockException()); + strongSelf->startTimer(timeout_ms); + return false; + } + //ffmpeg进程已经退出 + cb(SockException(Err_other,StrPrinter << "ffmpeg已经退出,exit code = " << strongSelf->_process.exit_code())); + return false; + },_poller); + } +} + +void FFmpegSource::findAsync(int maxWaitMS, const function &cb) { + auto src = MediaSource::find(_media_info._schema, + _media_info._vhost, + _media_info._app, + _media_info._streamid, + false); + if(src || !maxWaitMS){ + cb(src); + return; + } + + void *listener_tag = this; + //若干秒后执行等待媒体注册超时回调 + auto onRegistTimeout = _poller->doDelayTask(maxWaitMS,[cb,listener_tag](){ + //取消监听该事件 + NoticeCenter::Instance().delListener(listener_tag,Broadcast::kBroadcastMediaChanged); + cb(nullptr); + return 0; + }); + + weak_ptr weakSelf = shared_from_this(); + auto onRegist = [listener_tag,weakSelf,cb,onRegistTimeout](BroadcastMediaChangedArgs) { + auto strongSelf = weakSelf.lock(); + if(!strongSelf) { + //本身已经销毁，取消延时任务 + onRegistTimeout->cancel(); + NoticeCenter::Instance().delListener(listener_tag,Broadcast::kBroadcastMediaChanged); + return; + } + + if(!bRegist || + schema != strongSelf->_media_info._schema || + vhost != strongSelf->_media_info._vhost || + app != strongSelf->_media_info._app || + stream != strongSelf->_media_info._streamid){ + //不是自己感兴趣的事件，忽略之 + return; + } + + //查找的流终于注册上了；取消延时任务，防止多次回调 + onRegistTimeout->cancel(); + //取消事件监听 + NoticeCenter::Instance().delListener(listener_tag,Broadcast::kBroadcastMediaChanged); + + //切换到自己的线程再回复 + strongSelf->_poller->async([listener_tag,weakSelf,cb](){ + auto strongSelf = weakSelf.lock(); + if(!strongSelf) { + return; + } + //再找一遍媒体源，一般能找到 + strongSelf->findAsync(0,cb); + }, false); + }; + //监听媒体注册事件 + NoticeCenter::Instance().addListener(listener_tag, Broadcast::kBroadcastMediaChanged, onRegist); +} + +/** + * 定时检查媒体是否在线 + */ +void FFmpegSource::startTimer(int timeout_ms) { + weak_ptr weakSelf = shared_from_this(); + _timer = std::make_shared(1, [weakSelf, timeout_ms]() { + auto strongSelf = weakSelf.lock(); + if (!strongSelf) { + //自身已经销毁 + return false; + } + if (strongSelf->_media_info._host == "127.0.0.1") { + //推流给自己的，我们通过检查是否已经注册来判断FFmpeg是否工作正常 + strongSelf->findAsync(0, [&](const MediaSource::Ptr &src) { + //同步查找流 + if (!src) { + //流不在线，重新拉流 + strongSelf->play(strongSelf->_src_url, strongSelf->_dst_url, timeout_ms, + [](const SockException &) {}); + } + }); + } else { + //推流给其他服务器的，我们通过判断FFmpeg进程是否在线，如果FFmpeg推流中断，那么它应该会自动退出 + if (!strongSelf->_process.wait(false)) { + //ffmpeg不在线，重新拉流 + strongSelf->play(strongSelf->_src_url, strongSelf->_dst_url, timeout_ms, [](const SockException &) {}); + } + } + return true; + }, _poller); + + NoticeCenter::Instance().delListener(this, Broadcast::kBroadcastStreamNoneReader); + NoticeCenter::Instance().addListener(this, Broadcast::kBroadcastStreamNoneReader,[weakSelf](BroadcastStreamNoneReaderArgs) { + auto strongSelf = weakSelf.lock(); + if (!strongSelf) { + //自身已经销毁 + return; + } + + if(sender.getVhost() != strongSelf->_media_info._vhost || + sender.getApp() != strongSelf->_media_info._app || + sender.getId() != strongSelf->_media_info._streamid){ + //不是自己感兴趣的事件，忽略之 + return; + } + + //该流无人观看，我们停止吧 + if(strongSelf->_onClose){ + strongSelf->_onClose(); + } + }); +} + +void FFmpegSource::setOnClose(const function &cb){ + _onClose = cb; +} \ No newline at end of file diff --git a/server/FFmpegSource.h b/server/FFmpegSource.h new file mode 100644 index 00000000..cd589e70 --- /dev/null +++ b/server/FFmpegSource.h @@ -0,0 +1,47 @@ +// +// Created by xzl on 2018/5/24. +// + +#ifndef FFMPEG_SOURCE_H +#define FFMPEG_SOURCE_H + +#include +#include +#include +#include "Process.h" +#include "Util/TimeTicker.h" +#include "Network/Socket.h" +#include "Common/MediaSource.h" + +using namespace std; +using namespace toolkit; +using namespace mediakit; + +class FFmpegSource : public std::enable_shared_from_this{ +public: + typedef shared_ptr Ptr; + typedef function onPlay; + + FFmpegSource(); + virtual ~FFmpegSource(); + /** + * 设置主动关闭回调 + * @param cb + */ + void setOnClose(const function &cb); + void play(const string &src_url,const string &dst_url,int timeout_ms,const onPlay &cb); +private: + void findAsync(int maxWaitMS ,const function &cb); + void startTimer(int timeout_ms); +private: + Process _process; + Timer::Ptr _timer; + EventPoller::Ptr _poller; + MediaInfo _media_info; + string _src_url; + string _dst_url; + function _onClose; +}; + + +#endif //FFMPEG_SOURCE_H diff --git a/server/WebApi.cpp b/server/WebApi.cpp index c026f609..ccbaa58f 100644 --- a/server/WebApi.cpp +++ b/server/WebApi.cpp @@ -16,6 +16,9 @@ #include "Http/HttpSession.h" #include "Network/TcpServer.h" #include "Player/PlayerProxy.h" +#include "FFmpegSource.h" +#include "Util/MD5.h" +#include "WebApi.h" using namespace Json; using namespace toolkit; @@ -216,6 +219,9 @@ static inline string getProxyKey(const string &vhost,const string &app,const str return vhost + "/" + app + "/" + stream; } +static unordered_map s_ffmpegMap; +static recursive_mutex s_ffmpegMapMtx; + /** * 安装api接口 * 所有api都支持GET和POST两种方式 @@ -427,12 +433,11 @@ void installWebApi() { //指定RTP over TCP(播放rtsp时有效) (*player)[kRtpType] = rtp_type; //开始播放，如果播放失败或者播放中止，将会自动重试若干次，默认一直重试 - player->setPlayCallbackOnce([cb,player,key](const SockException &ex){ + player->setPlayCallbackOnce([cb,key](const SockException &ex){ if(ex){ lock_guard lck(s_proxyMapMtx); s_proxyMap.erase(key); } - const_cast(player).reset(); cb(ex,key); }); @@ -476,6 +481,62 @@ void installWebApi() { val["data"]["flag"] = s_proxyMap.erase(allArgs["key"]) == 1; }); + static auto addFFmepgSource = [](const string &src_url, + const string &dst_url, + int timeout_ms, + const function &cb){ + auto key = MD5(dst_url).hexdigest(); + lock_guard lck(s_ffmpegMapMtx); + if(s_ffmpegMap.find(key) != s_ffmpegMap.end()){ + //已经在拉流了 + cb(SockException(Err_success),key); + return; + } + + FFmpegSource::Ptr ffmpeg = std::make_shared(); + s_ffmpegMap[key] = ffmpeg; + + ffmpeg->setOnClose([key](){ + lock_guard lck(s_ffmpegMapMtx); + s_ffmpegMap.erase(key); + }); + ffmpeg->play(src_url, dst_url,timeout_ms,[cb , key](const SockException &ex){ + if(ex){ + lock_guard lck(s_ffmpegMapMtx); + s_ffmpegMap.erase(key); + } + cb(ex,key); + }); + }; + + //动态添加rtsp/rtmp拉流代理 + //测试url http://127.0.0.1/index/api/addFFmpegSource?src_url=http://live.hkstv.hk.lxdns.com/live/hks2/playlist.m3u8&dst_url=rtmp://127.0.0.1/live/hks2&timeout_ms=10000 + API_REGIST_INVOKER(api,addFFmpegSource,{ + CHECK_SECRET(); + CHECK_ARGS("src_url","dst_url","timeout_ms"); + auto src_url = allArgs["src_url"]; + auto dst_url = allArgs["dst_url"]; + int timeout_ms = allArgs["timeout_ms"]; + + addFFmepgSource(src_url,dst_url,timeout_ms,[invoker,val,headerOut](const SockException &ex,const string &key){ + if(ex){ + const_cast(val)["code"] = API::OtherFailed; + const_cast(val)["msg"] = ex.what(); + }else{ + const_cast(val)["data"]["key"] = key; + } + invoker("200 OK", headerOut, val.toStyledString()); + }); + }); + + //关闭拉流代理 + //测试url http://127.0.0.1/index/api/delFFmepgSource?key=key + API_REGIST(api,delFFmepgSource,{ + CHECK_SECRET(); + CHECK_ARGS("key"); + lock_guard lck(s_ffmpegMapMtx); + val["data"]["flag"] = s_ffmpegMap.erase(allArgs["key"]) == 1; + }); ////////////以下是注册的Hook API//////////// API_REGIST(hook,on_publish,{ @@ -517,22 +578,27 @@ void installWebApi() { //媒体未找到事件,我们都及时拉流hks作为替代品，目的是为了测试按需拉流 CHECK_SECRET(); CHECK_ARGS("vhost","app","stream"); - addStreamProxy(allArgs["vhost"], - allArgs["app"], - allArgs["stream"], - "rtmp://live.hkstv.hk.lxdns.com/live/hks2", - false, - false, - 0, - [invoker,val,headerOut](const SockException &ex,const string &key){ - if(ex){ - const_cast(val)["code"] = API::OtherFailed; - const_cast(val)["msg"] = ex.what(); - }else{ - const_cast(val)["data"]["key"] = key; - } - invoker("200 OK", headerOut, val.toStyledString()); - }); + GET_CONFIG(int,rtmp_port,Rtmp::kPort); + + string dst_url = StrPrinter + << "rtmp://127.0.0.1:" + << rtmp_port << "/" + << allArgs["app"] << "/" + << allArgs["stream"] << "?vhost=" + << allArgs["vhost"]; + + addFFmepgSource("http://live.hkstv.hk.lxdns.com/live/hks2/playlist.m3u8", + dst_url, + 10000, + [invoker,val,headerOut](const SockException &ex,const string &key){ + if(ex){ + const_cast(val)["code"] = API::OtherFailed; + const_cast(val)["msg"] = ex.what(); + }else{ + const_cast(val)["data"]["key"] = key; + } + invoker("200 OK", headerOut, val.toStyledString()); + }); }); @@ -555,6 +621,13 @@ void installWebApi() { } void unInstallWebApi(){ - lock_guard lck(s_proxyMapMtx); - s_proxyMap.clear(); + { + lock_guard lck(s_proxyMapMtx); + s_proxyMap.clear(); + } + + { + lock_guard lck(s_ffmpegMapMtx); + s_ffmpegMap.clear(); + } } \ No newline at end of file diff --git a/server/WebApi.h b/server/WebApi.h new file mode 100644 index 00000000..805e5ad4 --- /dev/null +++ b/server/WebApi.h @@ -0,0 +1,21 @@ +// +// Created by xzl on 2019-06-06. +// + +#ifndef ZLMEDIAKIT_WEBAPI_H +#define ZLMEDIAKIT_WEBAPI_H + +namespace mediakit { +////////////RTSP服务器配置/////////// +namespace Rtsp { + extern const char kPort[]; +} //namespace Rtsp + +////////////RTMP服务器配置/////////// +namespace Rtmp { + extern const char kPort[]; +} //namespace RTMP +} // namespace mediakit + + +#endif //ZLMEDIAKIT_WEBAPI_H diff --git a/server/main.cpp b/server/main.cpp index 24665416..11b7905a 100644 --- a/server/main.cpp +++ b/server/main.cpp @@ -44,6 +44,7 @@ #include "Player/PlayerProxy.h" #include "Http/WebSocketSession.h" #include "System.h" +#include "WebApi.h" using namespace std; using namespace toolkit;