Skip to content

Z
ZLMediaKit
Commit 49fc17d7 by ziyue
Options

hls m3u8文件直接通过内存读写

parent 642c9c07
正在显示 包含 96 行增加 和 69 行删除
src/Http/HttpFileManager.cpp
......@@ -399,7 +399,8 @@ static void accessFile(TcpSession &sender, const Parser &parser, const MediaInfo
return;
}
auto response_file = [file_exist, is_hls](const HttpServerCookie::Ptr &cookie, const HttpFileManager::invoker &cb, const string &strFile, const Parser &parser) {
auto response_file = [file_exist, is_hls](const HttpServerCookie::Ptr &cookie, const HttpFileManager::invoker &cb,
const string &strFile, const Parser &parser, bool is_path = true) {
StrCaseMap httpHeader;
if (cookie) {
httpHeader["Set-Cookie"] = cookie->getCookie(cookie->getAttach<HttpCookieAttachment>()._path);
......@@ -413,7 +414,7 @@ static void accessFile(TcpSession &sender, const Parser &parser, const MediaInfo
}
cb(code, HttpFileManager::getContentType(strFile.data()), headerOut, body);
};
invoker.responseFile(parser.getHeader(), httpHeader, strFile, !is_hls);
invoker.responseFile(parser.getHeader(), httpHeader, strFile, !is_hls, is_path);
};
if (!is_hls) {
......@@ -429,6 +430,13 @@ static void accessFile(TcpSession &sender, const Parser &parser, const MediaInfo
have_find_media_src = attach._have_find_media_source;
if (!have_find_media_src) {
const_cast<HttpCookieAttachment &>(attach)._have_find_media_source = true;
} else {
auto src = attach._hls_data->getMediaSource();
if (src) {
//直接从内存获取m3u8索引文件(而不是从文件系统)
response_file(cookie, cb, src->getIndexFile(), parser, false);
return;
}
}
}
if (have_find_media_src) {
......@@ -440,13 +448,11 @@ static void accessFile(TcpSession &sender, const Parser &parser, const MediaInfo
MediaSource::findAsync(mediaInfo, strongSession, [response_file, cookie, cb, strFile, parser](const MediaSource::Ptr &src) {
if (cookie) {
//尝试添加HlsMediaSource的观看人数(HLS是按需生成的，这样可以触发HLS文件的生成)
cookie->getAttach<HttpCookieAttachment>()._hls_data->addByteUsage(0);
}
if (src && File::is_file(strFile.data())) {
//流和m3u8文件都存在，那么直接返回文件
response_file(cookie, cb, strFile, parser);
return;
auto &attach = cookie->getAttach<HttpCookieAttachment>();
attach._hls_data->addByteUsage(0);
attach._hls_data->setMediaSource(dynamic_pointer_cast<HlsMediaSource>(src));
}
auto hls = dynamic_pointer_cast<HlsMediaSource>(src);
if (!hls) {
//流不存在，那么直接返回文件(相当于纯粹的HLS文件服务器,但是会挂起播放器15秒左右(用于等待HLS流的注册))
......@@ -454,9 +460,9 @@ static void accessFile(TcpSession &sender, const Parser &parser, const MediaInfo
return;
}
//流存在，但是m3u8文件不存在，那么等待生成m3u8文件(HLS源注册后，并不会立即生成HLS文件，有人观看才会按需生成HLS文件)
hls->waitForFile([response_file, cookie, cb, strFile, parser]() {
response_file(cookie, cb, strFile, parser);
//可能异步获取m3u8索引文件
hls->getIndexFile([response_file, cookie, cb, parser](const string &file) {
response_file(cookie, cb, file, parser, false);
});
});
});
......@@ -567,10 +573,18 @@ HttpResponseInvokerImp::HttpResponseInvokerImp(const HttpResponseInvokerImp::Htt
void HttpResponseInvokerImp::responseFile(const StrCaseMap &requestHeader,
const StrCaseMap &responseHeader,
const string &filePath,
bool use_mmap) const {
const string &file,
bool use_mmap,
bool is_path) const {
if (!is_path) {
//file是文件内容
(*this)(200, responseHeader, std::make_shared<HttpStringBody>(file));
return;
}
//file是文件路径
StrCaseMap &httpHeader = const_cast<StrCaseMap &>(responseHeader);
auto fileBody = std::make_shared<HttpFileBody>(filePath, use_mmap);
auto fileBody = std::make_shared<HttpFileBody>(file, use_mmap);
if (fileBody->remainSize() < 0) {
//打开文件失败
GET_CONFIG(string, notFound, Http::kNotFound);
......
src/Http/HttpFileManager.h
......@@ -35,7 +35,7 @@ public:
void operator()(int code, const StrCaseMap &headerOut, const HttpBody::Ptr &body) const;
void operator()(int code, const StrCaseMap &headerOut, const std::string &body) const;
void responseFile(const StrCaseMap &requestHeader,const StrCaseMap &responseHeader,const std::string &filePath, bool use_mmap = true) const;
void responseFile(const StrCaseMap &requestHeader,const StrCaseMap &responseHeader,const std::string &file, bool use_mmap = true, bool is_path = true) const;
operator bool();
private:
HttpResponseInvokerLambda0 _lambad;
......
src/Record/HlsMaker.cpp
......@@ -70,7 +70,7 @@ void HlsMaker::makeIndexFile(bool eof) {
snprintf(file_content, sizeof(file_content), "#EXT-X-ENDLIST\n");
m3u8.append(file_content);
}
onWriteHls(m3u8.data(), m3u8.size());
onWriteHls(m3u8);
}
......
src/Record/HlsMaker.h
......@@ -72,10 +72,8 @@ protected:
/**
* 写m3u8文件回调
* @param data
* @param len
*/
virtual void onWriteHls(const char *data, size_t len) = 0;
virtual void onWriteHls(const std::string &data) = 0;
/**
* 上一个 ts 切片写入完成, 可在这里进行通知处理
......
src/Record/HlsMakerImp.cpp
......@@ -111,13 +111,13 @@ void HlsMakerImp::onWriteSegment(const char *data, size_t len) {
}
}
void HlsMakerImp::onWriteHls(const char *data, size_t len) {
void HlsMakerImp::onWriteHls(const std::string &data) {
auto hls = makeFile(_path_hls);
if (hls) {
fwrite(data, len, 1, hls.get());
fwrite(data.data(), data.size(), 1, hls.get());
hls.reset();
if (_media_src) {
_media_src->registHls(true);
_media_src->registHls(data);
}
} else {
WarnL << "create hls file failed," << _path_hls << " " << get_uv_errmsg();
......
src/Record/HlsMakerImp.h
......@@ -53,7 +53,7 @@ protected:
std::string onOpenSegment(uint64_t index) override ;
void onDelSegment(uint64_t index) override;
void onWriteSegment(const char *data, size_t len) override;
void onWriteHls(const char *data, size_t len) override;
void onWriteHls(const std::string &data) override;
void onFlushLastSegment(uint32_t duration_ms) override;
private:
......
src/Record/HlsMediaSource.cpp
......@@ -12,7 +12,7 @@
using namespace toolkit;
namespace mediakit{
namespace mediakit {
HlsCookieData::HlsCookieData(const MediaInfo &info, const std::shared_ptr<SockInfo> &sock_info) {
_info = info;
......@@ -21,15 +21,15 @@ HlsCookieData::HlsCookieData(const MediaInfo &info, const std::shared_ptr<SockIn
addReaderCount();
}
void HlsCookieData::addReaderCount(){
if(!*_added){
auto src = std::dynamic_pointer_cast<HlsMediaSource>(MediaSource::find(HLS_SCHEMA,_info._vhost,_info._app,_info._streamid));
if(src){
void HlsCookieData::addReaderCount() {
if (!*_added) {
auto src = std::dynamic_pointer_cast<HlsMediaSource>(MediaSource::find(HLS_SCHEMA, _info._vhost, _info._app, _info._streamid));
if (src) {
*_added = true;
_ring_reader = src->getRing()->attach(EventPollerPool::Instance().getPoller());
auto added = _added;
_ring_reader->setDetachCB([added](){
//HlsMediaSource已经销毁
_ring_reader->setDetachCB([added]() {
// HlsMediaSource已经销毁
*added = false;
});
}
......@@ -39,14 +39,14 @@ void HlsCookieData::addReaderCount(){
HlsCookieData::~HlsCookieData() {
if (*_added) {
uint64_t duration = (_ticker.createdTime() - _ticker.elapsedTime()) / 1000;
WarnL << _sock_info->getIdentifier() << "(" << _sock_info->get_peer_ip() << ":" << _sock_info->get_peer_port() << ") "
<< "HLS播放器(" << _info._vhost << "/" << _info._app << "/" << _info._streamid
WarnL << _sock_info->getIdentifier() << "(" << _sock_info->get_peer_ip() << ":" << _sock_info->get_peer_port()
<< ") " << "HLS播放器(" << _info._vhost << "/" << _info._app << "/" << _info._streamid
<< ")断开,耗时(s):" << duration;
GET_CONFIG(uint32_t, iFlowThreshold, General::kFlowThreshold);
uint64_t bytes = _bytes.load();
if (bytes >= iFlowThreshold * 1024) {
NoticeCenter::Instance().emitEvent(Broadcast::kBroadcastFlowReport, _info, bytes, duration, true, static_cast<SockInfo&>(*_sock_info));
NoticeCenter::Instance().emitEvent(Broadcast::kBroadcastFlowReport, _info, bytes, duration, true, static_cast<SockInfo &>(*_sock_info));
}
}
}
......@@ -57,6 +57,12 @@ void HlsCookieData::addByteUsage(size_t bytes) {
_ticker.resetTime();
}
void HlsCookieData::setMediaSource(const HlsMediaSource::Ptr &src) {
_src = src;
}
}//namespace mediakit
HlsMediaSource::Ptr HlsCookieData::getMediaSource() const {
return _src.lock();
}
} // namespace mediakit
src/Record/HlsMediaSource.h
......@@ -11,11 +11,11 @@
#ifndef ZLMEDIAKIT_HLSMEDIASOURCE_H
#define ZLMEDIAKIT_HLSMEDIASOURCE_H
#include <atomic>
#include "Util/TimeTicker.h"
#include "Common/MediaSource.h"
#include "Util/TimeTicker.h"
#include <atomic>
namespace mediakit{
namespace mediakit {
class HlsMediaSource : public MediaSource {
public:
......@@ -24,28 +24,25 @@ public:
using RingType = toolkit::RingBuffer<std::string>;
using Ptr = std::shared_ptr<HlsMediaSource>;
HlsMediaSource(const std::string &vhost, const std::string &app, const std::string &stream_id) : MediaSource(HLS_SCHEMA, vhost, app, stream_id){}
HlsMediaSource(const std::string &vhost, const std::string &app, const std::string &stream_id)
: MediaSource(HLS_SCHEMA, vhost, app, stream_id) {}
~HlsMediaSource() override = default;
/**
* 获取媒体源的环形缓冲
*/
const RingType::Ptr &getRing() const {
return _ring;
}
const RingType::Ptr &getRing() const { return _ring; }
/**
* 获取播放器个数
*/
int readerCount() override {
return _ring ? _ring->readerCount() : 0;
}
int readerCount() override { return _ring ? _ring->readerCount() : 0; }
/**
* 生成m3u8文件时触发
* @param file_created 是否产生了hls文件
* @param index_file m3u8文件内容
*/
void registHls(bool file_created){
void registHls(std::string index_file) {
if (!_is_regist) {
_is_regist = true;
std::weak_ptr<HlsMediaSource> weakSelf = std::dynamic_pointer_cast<HlsMediaSource>(shared_from_this());
......@@ -61,56 +58,68 @@ public:
regist();
}
if (!file_created) {
//没产生文件
if (index_file.empty()) {
//没产生索引文件, 只是为了触发媒体注册
return;
}
//m3u8文件生成，发送给播放器
decltype(_list_cb) copy;
{
std::lock_guard<std::mutex> lck(_mtx_cb);
copy.swap(_list_cb);
}
copy.for_each([](const std::function<void()> &cb) {
cb();
});
//赋值m3u8索引文件内容
std::lock_guard<std::mutex> lck(_mtx_index);
_index_file = std::move(index_file);
_list_cb.for_each([&](const std::function<void(const std::string &str)> &cb) { cb(_index_file); });
_list_cb.clear();
}
void waitForFile(std::function<void()> cb) {
void getIndexFile(std::function<void(const std::string &str)> cb) {
std::lock_guard<std::mutex> lck(_mtx_index);
if (!_index_file.empty()) {
cb(_index_file);
return;
}
//等待生成m3u8文件
std::lock_guard<std::mutex> lck(_mtx_cb);
_list_cb.emplace_back(std::move(cb));
}
void onSegmentSize(size_t bytes) {
_speed[TrackVideo] += bytes;
std::string getIndexFile() const {
std::lock_guard<std::mutex> lck(_mtx_index);
return _index_file;
}
void onSegmentSize(size_t bytes) { _speed[TrackVideo] += bytes; }
private:
bool _is_regist = false;
RingType::Ptr _ring;
std::mutex _mtx_cb;
toolkit::List<std::function<void()> > _list_cb;
std::string _index_file;
mutable std::mutex _mtx_index;
toolkit::List<std::function<void(const std::string &)> > _list_cb;
};
class HlsCookieData{
class HlsCookieData {
public:
typedef std::shared_ptr<HlsCookieData> Ptr;
using Ptr = std::shared_ptr<HlsCookieData>;
HlsCookieData(const MediaInfo &info, const std::shared_ptr<toolkit::SockInfo> &sock_info);
~HlsCookieData();
void addByteUsage(size_t bytes);
void setMediaSource(const HlsMediaSource::Ptr &src);
HlsMediaSource::Ptr getMediaSource() const;
private:
void addReaderCount();
private:
std::atomic<uint64_t> _bytes {0};
std::atomic<uint64_t> _bytes { 0 };
MediaInfo _info;
std::shared_ptr<bool> _added;
toolkit::Ticker _ticker;
std::weak_ptr<HlsMediaSource> _src;
std::shared_ptr<toolkit::SockInfo> _sock_info;
HlsMediaSource::RingType::RingReader::Ptr _ring_reader;
};
}//namespace mediakit
#endif //ZLMEDIAKIT_HLSMEDIASOURCE_H
} // namespace mediakit
#endif // ZLMEDIAKIT_HLSMEDIASOURCE_H
src/Record/HlsRecorder.h
......@@ -39,7 +39,7 @@ public:
setDelegate(listener);
_hls->getMediaSource()->setListener(shared_from_this());
//先注册媒体流，后续可以按需生成
_hls->getMediaSource()->registHls(false);
_hls->getMediaSource()->registHls("");
}
int readerCount() {
......
Markdown 格式
0% 或
您添加了 0 人 到此讨论。请谨慎行事。
请先完成此评论的编辑！
请 注册 或者 后发表评论