Skip to content

Z
ZLMediaKit
Commit be1e872f by xiongziliang
Options

完善按需转协议(包括hls)

parent 268a7fec
正在显示 包含 225 行增加 和 123 行删除
server/FFmpegSource.cpp
......@@ -249,8 +249,8 @@ void FFmpegSource::onGetMediaSource(const MediaSource::Ptr &src) {
auto listener = src->getListener();
if (listener.lock().get() != this) {
//防止多次进入onGetMediaSource函数导致无效递归调用的bug
src->setListener(shared_from_this());
_listener = listener;
src->setListener(shared_from_this());
} else {
WarnL << "多次触发onGetMediaSource事件:"
<< src->getSchema() << "/"
......
src/Common/MediaSource.cpp
......@@ -8,7 +8,6 @@
* may be found in the AUTHORS file in the root of the source tree.
*/
#include <math.h>
#include "MediaSource.h"
#include "Record/MP4Reader.h"
#include "Util/util.h"
......@@ -316,36 +315,6 @@ MediaSource::Ptr MediaSource::find(const string &vhost, const string &app, const
return MediaSource::find(HLS_SCHEMA, vhost, app, stream_id);
}
static string getTrackInfoStr(const TrackSource *track_src){
_StrPrinter codec_info;
auto tracks = track_src->getTracks(true);
for (auto &track : tracks) {
auto codec_type = track->getTrackType();
codec_info << track->getCodecName();
switch (codec_type) {
case TrackAudio : {
auto audio_track = dynamic_pointer_cast<AudioTrack>(track);
codec_info << "["
<< audio_track->getAudioSampleRate() << "/"
<< audio_track->getAudioChannel() << "/"
<< audio_track->getAudioSampleBit() << "] ";
break;
}
case TrackVideo : {
auto video_track = dynamic_pointer_cast<VideoTrack>(track);
codec_info << "["
<< video_track->getVideoWidth() << "/"
<< video_track->getVideoHeight() << "/"
<< round(video_track->getVideoFps()) << "] ";
break;
}
default:
break;
}
}
return codec_info;
}
void MediaSource::emitEvent(bool regist){
auto listener = _listener.lock();
if (listener) {
......@@ -354,7 +323,7 @@ void MediaSource::emitEvent(bool regist){
}
//触发广播
NoticeCenter::Instance().emitEvent(Broadcast::kBroadcastMediaChanged, regist, *this);
InfoL << (regist ? "媒体注册:" : "媒体注销:") << _schema << " " << _vhost << " " << _app << " " << _stream_id << " " << getTrackInfoStr(this);
InfoL << (regist ? "媒体注册:" : "媒体注销:") << _schema << " " << _vhost << " " << _app << " " << _stream_id;
}
void MediaSource::regist() {
......
src/Common/MultiMediaSourceMuxer.cpp
......@@ -8,6 +8,7 @@
* may be found in the AUTHORS file in the root of the source tree.
*/
#include <math.h>
#include "MultiMediaSourceMuxer.h"
namespace mediakit {
......@@ -16,6 +17,7 @@ namespace mediakit {
MultiMuxerPrivate::~MultiMuxerPrivate() {}
MultiMuxerPrivate::MultiMuxerPrivate(const string &vhost, const string &app, const string &stream, float dur_sec,
bool enable_rtsp, bool enable_rtmp, bool enable_hls, bool enable_mp4) {
_stream_url = vhost + " " + app + " " + stream;
if (enable_rtmp) {
_rtmp = std::make_shared<RtmpMediaSourceMuxer>(vhost, app, stream, std::make_shared<TitleMeta>(dur_sec));
}
......@@ -24,7 +26,7 @@ MultiMuxerPrivate::MultiMuxerPrivate(const string &vhost, const string &app, con
}
if (enable_hls) {
_hls = Recorder::createRecorder(Recorder::type_hls, vhost, app, stream);
_hls = dynamic_pointer_cast<HlsRecorder>(Recorder::createRecorder(Recorder::type_hls, vhost, app, stream));
}
if (enable_mp4) {
......@@ -53,24 +55,22 @@ void MultiMuxerPrivate::resetTracks() {
}
void MultiMuxerPrivate::setMediaListener(const std::weak_ptr<MediaSourceEvent> &listener) {
_listener = listener;
if (_rtmp) {
_rtmp->setListener(listener);
}
if (_rtsp) {
_rtsp->setListener(listener);
}
auto hls_src = getHlsMediaSource();
if (hls_src) {
hls_src->setListener(listener);
auto hls = _hls;
if (hls) {
hls->setListener(listener);
}
_listener = listener;
}
int MultiMuxerPrivate::totalReaderCount() const {
auto hls_src = getHlsMediaSource();
return (_rtsp ? _rtsp->readerCount() : 0) + (_rtmp ? _rtmp->readerCount() : 0) + (hls_src ? hls_src->readerCount() : 0);
auto hls = _hls;
return (_rtsp ? _rtsp->readerCount() : 0) + (_rtmp ? _rtmp->readerCount() : 0) + (hls ? hls->readerCount() : 0);
}
static std::shared_ptr<MediaSinkInterface> makeRecorder(const vector<Track::Ptr> &tracks, Recorder::type type, const string &custom_path, MediaSource &sender){
......@@ -87,12 +87,12 @@ bool MultiMuxerPrivate::setupRecord(MediaSource &sender, Recorder::type type, bo
case Recorder::type_hls : {
if (start && !_hls) {
//开始录制
_hls = makeRecorder(getTracks(true), type, custom_path, sender);
auto hls_src = getHlsMediaSource();
if (hls_src) {
auto hls = dynamic_pointer_cast<HlsRecorder>(makeRecorder(getTracks(true), type, custom_path, sender));
if (hls) {
//设置HlsMediaSource的事件监听器
hls_src->setListener(_listener);
hls->setListener(_listener);
}
_hls = hls;
} else if (!start && _hls) {
//停止录制
_hls = nullptr;
......@@ -158,7 +158,10 @@ void MultiMuxerPrivate::onTrackReady(const Track::Ptr &track) {
}
bool MultiMuxerPrivate::isEnabled(){
return (_rtmp ? _rtmp->isEnabled() : false) || (_rtsp ? _rtsp->isEnabled() : false) || _hls || _mp4;
auto hls = _hls;
return (_rtmp ? _rtmp->isEnabled() : false) ||
(_rtsp ? _rtsp->isEnabled() : false) ||
(hls ? hls->isEnabled() : false) || _mp4;
}
void MultiMuxerPrivate::onTrackFrame(const Frame::Ptr &frame) {
......@@ -180,6 +183,36 @@ void MultiMuxerPrivate::onTrackFrame(const Frame::Ptr &frame) {
}
}
static string getTrackInfoStr(const TrackSource *track_src){
_StrPrinter codec_info;
auto tracks = track_src->getTracks(true);
for (auto &track : tracks) {
auto codec_type = track->getTrackType();
codec_info << track->getCodecName();
switch (codec_type) {
case TrackAudio : {
auto audio_track = dynamic_pointer_cast<AudioTrack>(track);
codec_info << "["
<< audio_track->getAudioSampleRate() << "/"
<< audio_track->getAudioChannel() << "/"
<< audio_track->getAudioSampleBit() << "] ";
break;
}
case TrackVideo : {
auto video_track = dynamic_pointer_cast<VideoTrack>(track);
codec_info << "["
<< video_track->getVideoWidth() << "/"
<< video_track->getVideoHeight() << "/"
<< round(video_track->getVideoFps()) << "] ";
break;
}
default:
break;
}
}
return codec_info;
}
void MultiMuxerPrivate::onAllTrackReady() {
if (_rtmp) {
_rtmp->onAllTrackReady();
......@@ -187,18 +220,10 @@ void MultiMuxerPrivate::onAllTrackReady() {
if (_rtsp) {
_rtsp->onAllTrackReady();
}
if (_track_listener) {
_track_listener->onAllTrackReady();
}
}
MediaSource::Ptr MultiMuxerPrivate::getHlsMediaSource() const {
auto recorder = dynamic_pointer_cast<HlsRecorder>(_hls);
if (recorder) {
return recorder->getMediaSource();
}
return nullptr;
InfoL << "stream: " << _stream_url << " , codec info: " << getTrackInfoStr(this);
}
///////////////////////////////MultiMediaSourceMuxer//////////////////////////////////
......@@ -212,9 +237,9 @@ MultiMediaSourceMuxer::MultiMediaSourceMuxer(const string &vhost, const string &
}
void MultiMediaSourceMuxer::setMediaListener(const std::weak_ptr<MediaSourceEvent> &listener) {
_listener = listener;
//拦截事件
_muxer->setMediaListener(shared_from_this());
_listener = listener;
}
void MultiMediaSourceMuxer::setTrackListener(const std::weak_ptr<MultiMuxerPrivate::Listener> &listener) {
......@@ -242,7 +267,7 @@ int MultiMediaSourceMuxer::totalReaderCount(MediaSource &sender) {
}
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);
}
bool MultiMediaSourceMuxer::isRecording(MediaSource &sender, Recorder::type type) {
......@@ -373,8 +398,9 @@ void MultiMediaSourceMuxer::inputFrame(const Frame::Ptr &frame_in) {
bool MultiMediaSourceMuxer::isEnabled(){
#if defined(ENABLE_RTPPROXY)
return (_muxer->isEnabled() || _ps_rtp_sender);
#endif //ENABLE_RTPPROXY
#else
return _muxer->isEnabled();
#endif //ENABLE_RTPPROXY
}
......
src/Common/MultiMediaSourceMuxer.h
......@@ -46,13 +46,13 @@ private:
void onTrackReady(const Track::Ptr & track) override;
void onTrackFrame(const Frame::Ptr &frame) override;
void onAllTrackReady() override;
MediaSource::Ptr getHlsMediaSource() const;
private:
string _stream_url;
Listener *_track_listener = nullptr;
RtmpMediaSourceMuxer::Ptr _rtmp;
RtspMediaSourceMuxer::Ptr _rtsp;
MediaSinkInterface::Ptr _hls;
HlsRecorder::Ptr _hls;
MediaSinkInterface::Ptr _mp4;
std::weak_ptr<MediaSourceEvent> _listener;
};
......
src/Http/HttpFileManager.cpp
......@@ -557,8 +557,22 @@ static void accessFile(TcpSession &sender, const Parser &parser, const MediaInfo
}
//hls文件不存在，我们等待其生成并延后回复
MediaSource::findAsync(mediaInfo, strongSession, [response_file, cookie, cb, strFile, parser](const MediaSource::Ptr &src) {
//hls已经生成或者超时后仍未生成，那么不管怎么样都返回客户端
response_file(cookie, cb, strFile, parser);
if (src && File::is_file(strFile.data())) {
//流和m3u8文件都存在，那么直接返回文件
response_file(cookie, cb, strFile, parser);
return;
}
auto hls = dynamic_pointer_cast<HlsMediaSource>(src);
if (!hls) {
//流不存在，那么直接返回文件
response_file(cookie, cb, strFile, parser);
return;
}
//流存在，但是m3u8文件不存在，那么等待生成m3u8文件
hls->waitForHls([response_file, cookie, cb, strFile, parser]() {
response_file(cookie, cb, strFile, parser);
});
});
}
});
......
src/Record/HlsMaker.cpp
......@@ -124,4 +124,13 @@ bool HlsMaker::isLive() {
return _seg_number != 0;
}
void HlsMaker::clear(){
_seg_dur_list.clear();
_last_file_name.clear();
_ticker_last_data.resetTime();
_ticker.resetTime();
_file_index = 0;
}
}//namespace mediakit
\ No newline at end of file
src/Record/HlsMaker.h
......@@ -39,6 +39,17 @@ public:
* @param is_idr_fast_packet 是否为关键帧第一个包
*/
void inputData(void *data, uint32_t len, uint32_t timestamp, bool is_idr_fast_packet);
/**
* 是否为直播
*/
bool isLive();
/**
* 清空记录
*/
void clear();
protected:
/**
* 创建ts切片文件回调
......@@ -73,10 +84,6 @@ protected:
*/
void flushLastSegment(bool eof = false);
/**
* 是否为直播
*/
bool isLive();
private:
/**
* 生成m3u8文件
......@@ -94,6 +101,7 @@ private:
* @param timestamp
*/
void addNewSegment(uint32_t timestamp);
private:
uint32_t _seg_number = 0;
float _seg_duration = 0;
......
src/Record/HlsMakerImp.cpp
......@@ -11,6 +11,7 @@
#include "HlsMakerImp.h"
#include "Util/util.h"
#include "Util/uv_errno.h"
using namespace toolkit;
namespace mediakit {
......@@ -24,37 +25,44 @@ HlsMakerImp::HlsMakerImp(const string &m3u8_file,
_path_hls = m3u8_file;
_params = params;
_buf_size = bufSize;
_file_buf.reset(new char[bufSize],[](char *ptr){
_file_buf.reset(new char[bufSize], [](char *ptr) {
delete[] ptr;
});
}
HlsMakerImp::~HlsMakerImp() {
clearCache();
}
void HlsMakerImp::clearCache() {
//录制完了
flushLastSegment(true);
if(isLive()){
if (isLive()) {
//hls直播才删除文件
clear();
_file = nullptr;
_segment_file_paths.clear();
File::delete_file(_path_prefix.data());
}
}
string HlsMakerImp::onOpenSegment(int index) {
string segment_name , segment_path;
string segment_name, segment_path;
{
auto strDate = getTimeStr("%Y-%m-%d");
auto strHour = getTimeStr("%H");
auto strTime = getTimeStr("%M-%S");
segment_name = StrPrinter << strDate + "/" + strHour + "/" + strTime << "_" << index << ".ts";
segment_path = _path_prefix + "/" + segment_name;
if(isLive()){
_segment_file_paths.emplace(index,segment_path);
segment_path = _path_prefix + "/" + segment_name;
if (isLive()) {
_segment_file_paths.emplace(index, segment_path);
}
}
_file = makeFile(segment_path, true);
if(!_file){
WarnL << "create file falied," << segment_path << " " << get_uv_errmsg();
if (!_file) {
WarnL << "create file failed," << segment_path << " " << get_uv_errmsg();
}
if(_params.empty()){
if (_params.empty()) {
return std::move(segment_name);
}
return std::move(segment_name + "?" + _params);
......@@ -62,7 +70,7 @@ string HlsMakerImp::onOpenSegment(int index) {
void HlsMakerImp::onDelSegment(int index) {
auto it = _segment_file_paths.find(index);
if(it == _segment_file_paths.end()){
if (it == _segment_file_paths.end()) {
return;
}
File::delete_file(it->second.data());
......@@ -77,27 +85,27 @@ void HlsMakerImp::onWriteSegment(const char *data, int len) {
void HlsMakerImp::onWriteHls(const char *data, int len) {
auto hls = makeFile(_path_hls);
if(hls){
fwrite(data,len,1,hls.get());
if (hls) {
fwrite(data, len, 1, hls.get());
hls.reset();
if(_media_src){
if (_media_src) {
_media_src->registHls();
}
} else{
WarnL << "create hls file falied," << _path_hls << " " << get_uv_errmsg();
} else {
WarnL << "create hls file failed," << _path_hls << " " << get_uv_errmsg();
}
//DebugL << "\r\n" << string(data,len);
}
std::shared_ptr<FILE> HlsMakerImp::makeFile(const string &file,bool setbuf) {
std::shared_ptr<FILE> HlsMakerImp::makeFile(const string &file, bool setbuf) {
auto file_buf = _file_buf;
auto ret= shared_ptr<FILE>(File::create_file(file.data(), "wb"), [file_buf](FILE *fp) {
auto ret = shared_ptr<FILE>(File::create_file(file.data(), "wb"), [file_buf](FILE *fp) {
if (fp) {
fclose(fp);
}
});
if(ret && setbuf){
if (ret && setbuf) {
setvbuf(ret.get(), _file_buf.get(), _IOFBF, _buf_size);
}
return ret;
......@@ -107,7 +115,7 @@ void HlsMakerImp::setMediaSource(const string &vhost, const string &app, const s
_media_src = std::make_shared<HlsMediaSource>(vhost, app, stream_id);
}
MediaSource::Ptr HlsMakerImp::getMediaSource() const{
HlsMediaSource::Ptr HlsMakerImp::getMediaSource() const {
return _media_src;
}
......
src/Record/HlsMakerImp.h
......@@ -27,7 +27,8 @@ public:
uint32_t bufSize = 64 * 1024,
float seg_duration = 5,
uint32_t seg_number = 3);
virtual ~HlsMakerImp();
~HlsMakerImp() override;
/**
* 设置媒体信息
......@@ -41,23 +42,31 @@ public:
* 获取MediaSource
* @return
*/
MediaSource::Ptr getMediaSource() const;
HlsMediaSource::Ptr getMediaSource() const;
/**
* 清空缓存
*/
void clearCache();
protected:
string onOpenSegment(int index) override ;
void onDelSegment(int index) override;
void onWriteSegment(const char *data, int len) override;
void onWriteHls(const char *data, int len) override;
private:
std::shared_ptr<FILE> makeFile(const string &file,bool setbuf = false);
private:
HlsMediaSource::Ptr _media_src;
map<int /*index*/,string/*file_path*/> _segment_file_paths;
int _buf_size;
string _params;
string _path_hls;
string _path_prefix;
std::shared_ptr<FILE> _file;
std::shared_ptr<char> _file_buf;
string _path_prefix;
string _path_hls;
string _params;
int _buf_size;
HlsMediaSource::Ptr _media_src;
map<int /*index*/,string/*file_path*/> _segment_file_paths;
};
}//namespace mediakit
......
src/Record/HlsMediaSource.h
......@@ -22,7 +22,7 @@ public:
typedef RingBuffer<string> RingType;
typedef std::shared_ptr<HlsMediaSource> Ptr;
HlsMediaSource(const string &vhost, const string &app, const string &stream_id) : MediaSource(HLS_SCHEMA, vhost, app, stream_id){
_readerCount = 0;
_reader_count = 0;
_ring = std::make_shared<RingType>();
}
......@@ -40,18 +40,34 @@ public:
* @return
*/
int readerCount() override {
return _readerCount.load();
return _reader_count.load();
}
/**
* 注册hls
* 生成m3u8文件时触发
*/
void registHls(){
if (!_registed) {
_registed = true;
if (!_is_regist) {
_is_regist = true;
onReaderChanged(0);
regist();
}
//m3u8文件生成，发送给播放器
decltype(_list_cb) copy;
{
lock_guard<mutex> lck(_mtx_cb);
copy.swap(_list_cb);
}
copy.for_each([](const function<void()> &cb) {
cb();
});
}
void waitForHls(function<void()> cb){
//等待生成m3u8文件
lock_guard<mutex> lck(_mtx_cb);
_list_cb.emplace_back(std::move(cb));
}
private:
......@@ -61,16 +77,19 @@ private:
*/
void modifyReaderCount(bool add) {
if (add) {
++_readerCount;
++_reader_count;
} else {
--_readerCount;
--_reader_count;
}
onReaderChanged(_readerCount);
onReaderChanged(_reader_count);
}
private:
atomic_int _readerCount;
bool _registed = false;
bool _is_regist = false;
atomic_int _reader_count;
RingType::Ptr _ring;
mutex _mtx_cb;
List<function<void()> > _list_cb;
};
class HlsCookieData{
......
src/Record/HlsRecorder.h
......@@ -15,37 +15,76 @@
#include "TsMuxer.h"
namespace mediakit {
class HlsRecorder
class HlsRecorder : public MediaSourceEventInterceptor, public std::enable_shared_from_this<HlsRecorder>
#if defined(ENABLE_HLS)
: public TsMuxer
, public TsMuxer
#endif
{
public:
typedef std::shared_ptr<HlsRecorder> Ptr;
HlsRecorder(const string &m3u8_file, const string &params){
GET_CONFIG(uint32_t,hlsNum,Hls::kSegmentNum);
GET_CONFIG(uint32_t,hlsBufSize,Hls::kFileBufSize);
GET_CONFIG(uint32_t,hlsDuration,Hls::kSegmentDuration);
_hls = new HlsMakerImp(m3u8_file,params,hlsBufSize,hlsDuration,hlsNum);
GET_CONFIG(uint32_t, hlsNum, Hls::kSegmentNum);
GET_CONFIG(uint32_t, hlsBufSize, Hls::kFileBufSize);
GET_CONFIG(uint32_t, hlsDuration, Hls::kSegmentDuration);
_hls = std::make_shared<HlsMakerImp>(m3u8_file, params, hlsBufSize, hlsDuration, hlsNum);
//清空上次的残余文件
_hls->clearCache();
}
~HlsRecorder(){
delete _hls;
}
void setMediaSource(const string &vhost, const string &app, const string &stream_id){
~HlsRecorder(){}
void setMediaSource(const string &vhost, const string &app, const string &stream_id) {
_hls->setMediaSource(vhost, app, stream_id);
}
MediaSource::Ptr getMediaSource() const{
return _hls->getMediaSource();
void setListener(const std::weak_ptr<MediaSourceEvent> &listener) {
_listener = listener;
_hls->getMediaSource()->setListener(shared_from_this());
//先注册媒体流，后续可以按需生成
_hls->getMediaSource()->registHls();
}
int readerCount() {
return _hls->getMediaSource()->readerCount();
}
void onReaderChanged(MediaSource &sender, int size) override {
//hls保留切片个数为0时代表为hls录制(不删除切片)，那么不管有无观看者都一直生成hls
_enabled = _hls->isLive() ? size : true;
if (!size && _hls->isLive()) {
//hls直播时，如果无人观看就删除视频缓存，目的是为了防止视频跳跃
_clear_cache = true;
}
MediaSourceEventInterceptor::onReaderChanged(sender, size);
}
bool isEnabled() {
//缓存尚未清空时，还允许触发inputFrame函数，以便及时清空缓存
return _clear_cache ? true : _enabled;
}
#if defined(ENABLE_HLS)
protected:
void onTs(const void *packet, int bytes,uint32_t timestamp,bool is_idr_fast_packet) override {
_hls->inputData((char *)packet,bytes,timestamp, is_idr_fast_packet);
};
void inputFrame(const Frame::Ptr &frame) override{
if (_clear_cache) {
_clear_cache = false;
_hls->clearCache();
}
if (_enabled) {
TsMuxer::inputFrame(frame);
}
}
private:
void onTs(const void *packet, int bytes, uint32_t timestamp, bool is_idr_fast_packet) override {
_hls->inputData((char *) packet, bytes, timestamp, is_idr_fast_packet);
}
#endif
private:
HlsMakerImp *_hls;
//默认不生成hls文件，有播放器时再生成
bool _enabled = false;
bool _clear_cache = false;
std::shared_ptr<HlsMakerImp> _hls;
};
}//namespace mediakit
#endif //HLSRECORDER_H
src/Rtmp/RtmpMediaSourceMuxer.h
......@@ -33,6 +33,7 @@ public:
void setListener(const std::weak_ptr<MediaSourceEvent> &listener){
_listener = listener;
_media_src->setListener(shared_from_this());
}
void setTimeStamp(uint32_t stamp){
......@@ -45,7 +46,6 @@ public:
void onAllTrackReady(){
makeConfigPacket();
_media_src->setListener(shared_from_this());
_media_src->setMetaData(getMetadata());
}
......
src/Rtsp/RtspMediaSourceMuxer.h
......@@ -33,6 +33,7 @@ public:
void setListener(const std::weak_ptr<MediaSourceEvent> &listener){
_listener = listener;
_media_src->setListener(shared_from_this());
}
int readerCount() const{
......@@ -44,7 +45,6 @@ public:
}
void onAllTrackReady(){
_media_src->setListener(shared_from_this());
_media_src->setSdp(getSdp());
}
......
Markdown 格式
0% 或
您添加了 0 人 到此讨论。请谨慎行事。
请先完成此评论的编辑！
请 注册 或者 后发表评论