Skip to content

Z
ZLMediaKit
Commit d432888a by 夏楚 Committed by GitHub
Options

Merge pull request #1419 from ZLMediaKit/feature/mmap 

通过共享mmap方式提高http文件/hls服务器性能
parents 72caa43c a2b63448
正在显示 包含 258 行增加 和 165 行删除
src/Http/HlsPlayer.cpp
......@@ -76,15 +76,17 @@ void HlsPlayer::fetchSegment() {
}
return Socket::createSocket(poller, true);
});
_http_ts_player->setOnPacket([weak_self](const char *data, size_t len) {
auto strong_self = weak_self.lock();
if (!strong_self) {
return;
}
//收到ts包
strong_self->onPacket_l(data, len);
});
auto benchmark_mode = (*this)[Client::kBenchmarkMode].as<int>();
if (!benchmark_mode) {
_http_ts_player->setOnPacket([weak_self](const char *data, size_t len) {
auto strong_self = weak_self.lock();
if (!strong_self) {
return;
}
//收到ts包
strong_self->onPacket_l(data, len);
});
}
if (!(*this)[Client::kNetAdapter].empty()) {
_http_ts_player->setNetAdapter((*this)[Client::kNetAdapter]);
......@@ -349,7 +351,8 @@ void HlsPlayerImp::addTrackCompleted() {
}
void HlsPlayerImp::onPlayResult(const SockException &ex) {
if (ex) {
auto benchmark_mode = (*this)[Client::kBenchmarkMode].as<int>();
if (ex || benchmark_mode) {
PlayerImp<HlsPlayer, PlayerBase>::onPlayResult(ex);
} else {
auto demuxer = std::make_shared<HlsDemuxer>();
......@@ -364,6 +367,9 @@ void HlsPlayerImp::onShutdown(const SockException &ex) {
}
vector<Track::Ptr> HlsPlayerImp::getTracks(bool ready) const {
if (!_demuxer) {
return vector<Track::Ptr>();
}
return static_pointer_cast<HlsDemuxer>(_demuxer)->getTracks(ready);
}
......
src/Http/HttpBody.h
......@@ -37,7 +37,7 @@ public:
/**
* 剩余数据大小，如果返回-1, 那么就不设置content-length
*/
virtual ssize_t remainSize() { return 0;};
virtual int64_t remainSize() { return 0;};
/**
* 读取一定字节数，返回大小可能小于size
......@@ -77,7 +77,7 @@ public:
HttpStringBody(std::string str);
~HttpStringBody() override = default;
ssize_t remainSize() override;
int64_t remainSize() override;
toolkit::Buffer::Ptr readData(size_t size) override ;
private:
......@@ -94,7 +94,7 @@ public:
HttpBufferBody(toolkit::Buffer::Ptr buffer);
~HttpBufferBody() override = default;
ssize_t remainSize() override;
int64_t remainSize() override;
toolkit::Buffer::Ptr readData(size_t size) override;
private:
......@@ -104,32 +104,32 @@ private:
/**
* 文件类型的content
*/
class HttpFileBody : public HttpBody{
class HttpFileBody : public HttpBody {
public:
typedef std::shared_ptr<HttpFileBody> Ptr;
/**
* 构造函数
* @param fp 文件句柄，文件的偏移量必须为0
* @param offset 相对文件头的偏移量
* @param max_size 最大读取字节数，未判断是否大于文件真实大小
* @param file_path 文件路径
* @param use_mmap 是否使用mmap方式访问文件
*/
HttpFileBody(const std::shared_ptr<FILE> &fp, size_t offset, size_t max_size, bool use_mmap = true);
HttpFileBody(const std::string &file_path, bool use_mmap = true);
~HttpFileBody() override = default;
ssize_t remainSize() override ;
/**
* 设置读取范围
* @param offset 相对文件头的偏移量
* @param max_size 最大读取字节数
*/
void setRange(uint64_t offset, uint64_t max_size);
int64_t remainSize() override;
toolkit::Buffer::Ptr readData(size_t size) override;
int sendFile(int fd) override;
private:
void init(const std::shared_ptr<FILE> &fp,size_t offset,size_t max_size, bool use_mmap);
private:
size_t _max_size;
size_t _offset = 0;
size_t _file_offset = 0;
int64_t _read_to = 0;
uint64_t _file_offset = 0;
std::shared_ptr<FILE> _fp;
std::shared_ptr<char> _map_addr;
toolkit::ResourcePool<toolkit::BufferRaw> _pool;
......@@ -152,7 +152,7 @@ public:
*/
HttpMultiFormBody(const HttpArgs &args,const std::string &filePath,const std::string &boundary = "0xKhTmLbOuNdArY");
virtual ~HttpMultiFormBody(){}
ssize_t remainSize() override ;
int64_t remainSize() override ;
toolkit::Buffer::Ptr readData(size_t size) override;
public:
......@@ -161,8 +161,8 @@ public:
static std::string multiFormContentType(const std::string &boundary);
private:
size_t _offset = 0;
size_t _totalSize;
uint64_t _offset = 0;
int64_t _totalSize;
std::string _bodyPrefix;
std::string _bodySuffix;
HttpFileBody::Ptr _fileBody;
......
src/Http/HttpCookie.cpp
......@@ -10,7 +10,7 @@
#include "HttpCookie.h"
#include "Util/util.h"
#include "Util/logger.h"
#include "Util/onceToken.h"
#if defined(_WIN32)
#include "Util/strptime_win.h"
......@@ -21,28 +21,75 @@ using namespace std;
namespace mediakit {
void HttpCookie::setPath(const string &path){
void HttpCookie::setPath(const string &path) {
_path = path;
}
void HttpCookie::setHost(const string &host){
void HttpCookie::setHost(const string &host) {
_host = host;
}
static time_t timeStrToInt(const string &date){
static long s_gmtoff = 0; //时间差
static onceToken s_token([]() {
#ifdef _WIN32
TIME_ZONE_INFORMATION tzinfo;
DWORD dwStandardDaylight;
long bias;
dwStandardDaylight = GetTimeZoneInformation(&tzinfo);
bias = tzinfo.Bias;
if (dwStandardDaylight == TIME_ZONE_ID_STANDARD) {
bias += tzinfo.StandardBias;
}
if (dwStandardDaylight == TIME_ZONE_ID_DAYLIGHT) {
bias += tzinfo.DaylightBias;
}
s_gmtoff = -bias * 60; //时间差(分钟)
#else
s_gmtoff = getLocalTime(time(nullptr)).tm_gmtoff;
#endif // _WIN32
});
// from https://gmbabar.wordpress.com/2010/12/01/mktime-slow-use-custom-function/#comment-58
static time_t time_to_epoch(const struct tm *ltm, int utcdiff) {
const int mon_days[] = { 31, 28, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31 };
long tyears, tdays, leaps, utc_hrs;
int i;
tyears = ltm->tm_year - 70; // tm->tm_year is from 1900.
leaps = (tyears + 2) / 4; // no of next two lines until year 2100.
// i = (ltm->tm_year – 100) / 100;
// leaps -= ( (i/4)*3 + i%4 );
tdays = 0;
for (i = 0; i < ltm->tm_mon; i++)
tdays += mon_days[i];
tdays += ltm->tm_mday - 1; // days of month passed.
tdays = tdays + (tyears * 365) + leaps;
utc_hrs = ltm->tm_hour + utcdiff; // for your time zone.
return (tdays * 86400) + (utc_hrs * 3600) + (ltm->tm_min * 60) + ltm->tm_sec;
}
static time_t timeStrToInt(const string &date) {
struct tm tt;
strptime(date.data(),"%a, %b %d %Y %H:%M:%S %Z",&tt);
return mktime(&tt);
strptime(date.data(), "%a, %b %d %Y %H:%M:%S %Z", &tt);
// mktime内部有使用互斥锁，非常影响性能
return time_to_epoch(&tt, s_gmtoff / 3600); // mktime(&tt);
}
void HttpCookie::setExpires(const string &expires,const string &server_date){
void HttpCookie::setExpires(const string &expires, const string &server_date) {
_expire = timeStrToInt(expires);
if(!server_date.empty()){
_expire = time(NULL) + (_expire - timeStrToInt(server_date));
if (!server_date.empty()) {
_expire = time(NULL) + (_expire - timeStrToInt(server_date));
}
}
void HttpCookie::setKeyVal(const string &key,const string &val){
void HttpCookie::setKeyVal(const string &key, const string &val) {
_key = key;
_val = val;
}
HttpCookie::operator bool (){
HttpCookie::operator bool() {
return !_host.empty() && !_key.empty() && !_val.empty() && (_expire > time(NULL));
}
......@@ -50,19 +97,18 @@ const string &HttpCookie::getVal() const {
return _val;
}
const string &HttpCookie::getKey() const{
const string &HttpCookie::getKey() const {
return _key;
}
HttpCookieStorage &HttpCookieStorage::Instance(){
HttpCookieStorage &HttpCookieStorage::Instance() {
static HttpCookieStorage instance;
return instance;
}
void HttpCookieStorage::set(const HttpCookie::Ptr &cookie) {
lock_guard<mutex> lck(_mtx_cookie);
if(!cookie || !(*cookie)){
if (!cookie || !(*cookie)) {
return;
}
_all_cookie[cookie->_host][cookie->_path][cookie->_key] = cookie;
......@@ -71,20 +117,20 @@ void HttpCookieStorage::set(const HttpCookie::Ptr &cookie) {
vector<HttpCookie::Ptr> HttpCookieStorage::get(const string &host, const string &path) {
vector<HttpCookie::Ptr> ret(0);
lock_guard<mutex> lck(_mtx_cookie);
auto it = _all_cookie.find(host);
if(it == _all_cookie.end()){
auto it = _all_cookie.find(host);
if (it == _all_cookie.end()) {
//未找到该host相关记录
return ret;
}
//遍历该host下所有path
for(auto &pr : it->second){
if(path.find(pr.first) != 0){
for (auto &pr : it->second) {
if (path.find(pr.first) != 0) {
//这个path不匹配
continue;
}
//遍历该path下的各个cookie
for(auto it_cookie = pr.second.begin() ; it_cookie != pr.second.end() ; ){
if(!*(it_cookie->second)){
for (auto it_cookie = pr.second.begin(); it_cookie != pr.second.end();) {
if (!*(it_cookie->second)) {
//该cookie已经过期，移除之
it_cookie = pr.second.erase(it_cookie);
continue;
......@@ -97,5 +143,4 @@ vector<HttpCookie::Ptr> HttpCookieStorage::get(const string &host, const string
return ret;
}
} /* namespace mediakit */
src/Http/HttpCookieManager.h
......@@ -11,13 +11,13 @@
#ifndef SRC_HTTP_COOKIEMANAGER_H
#define SRC_HTTP_COOKIEMANAGER_H
#include <memory>
#include <unordered_map>
#include "Common/Parser.h"
#include "Network/Socket.h"
#include "Util/TimeTicker.h"
#include "Util/mini.h"
#include "Util/util.h"
#include "Util/TimeTicker.h"
#include "Network/Socket.h"
#include "Common/Parser.h"
#include <memory>
#include <unordered_map>
#define COOKIE_DEFAULT_LIFE (7 * 24 * 60 * 60)
......@@ -28,9 +28,9 @@ class HttpCookieManager;
/**
* cookie对象，用于保存cookie的一些相关属性
*/
class HttpServerCookie : public toolkit::AnyStorage , public toolkit::noncopyable{
class HttpServerCookie : public toolkit::noncopyable {
public:
typedef std::shared_ptr<HttpServerCookie> Ptr;
using Ptr = std::shared_ptr<HttpServerCookie>;
/**
* 构建cookie
* @param manager cookie管理者对象
......@@ -40,12 +40,10 @@ public:
* @param max_elapsed 最大过期时间，单位秒
*/
HttpServerCookie(const std::shared_ptr<HttpCookieManager> &manager,
const std::string &cookie_name,
const std::string &uid,
const std::string &cookie,
uint64_t max_elapsed);
~HttpServerCookie() ;
HttpServerCookie(
const std::shared_ptr<HttpCookieManager> &manager, const std::string &cookie_name, const std::string &uid,
const std::string &cookie, uint64_t max_elapsed);
~HttpServerCookie();
/**
* 获取uid
......@@ -65,13 +63,13 @@ public:
* 获取cookie随机字符串
* @return cookie随机字符串
*/
const std::string& getCookie() const;
const std::string &getCookie() const;
/**
* 获取该cookie名
* @return
*/
const std::string& getCookieName() const;
const std::string &getCookieName() const;
/**
* 更新该cookie的过期时间，可以让此cookie不失效
......@@ -85,26 +83,35 @@ public:
bool isExpired();
/**
* 获取区域锁
* @return
* 设置附加数据
*/
std::shared_ptr<std::lock_guard<std::recursive_mutex> > getLock();
void setAttach(std::shared_ptr<void> attach);
/*
* 获取附加数据
*/
template <class T>
const T& getAttach() const {
return *static_cast<const T *>(_attach.get());
}
private:
std::string cookieExpireTime() const ;
std::string cookieExpireTime() const;
private:
std::string _uid;
std::string _cookie_name;
std::string _cookie_uuid;
uint64_t _max_elapsed;
toolkit::Ticker _ticker;
std::recursive_mutex _mtx;
std::shared_ptr<void> _attach;
std::weak_ptr<HttpCookieManager> _manager;
};
/**
* cookie随机字符串生成器
*/
class RandStrGeneator{
class RandStrGeneator {
public:
RandStrGeneator() = default;
~RandStrGeneator() = default;
......@@ -120,8 +127,10 @@ public:
* @param str 随机字符串
*/
void release(const std::string &str);
private:
std::string obtain_l();
private:
//碰撞库
std::unordered_set<std::string> _obtained;
......@@ -135,8 +144,8 @@ private:
*/
class HttpCookieManager : public std::enable_shared_from_this<HttpCookieManager> {
public:
typedef std::shared_ptr<HttpCookieManager> Ptr;
friend class HttpServerCookie;
using Ptr = std::shared_ptr<HttpCookieManager>;
~HttpCookieManager();
/**
......@@ -152,7 +161,10 @@ public:
* @param max_elapsed 该cookie过期时间，单位秒
* @return cookie对象
*/
HttpServerCookie::Ptr addCookie(const std::string &cookie_name,const std::string &uid, uint64_t max_elapsed = COOKIE_DEFAULT_LIFE,int max_client = 1);
HttpServerCookie::Ptr addCookie(
const std::string &cookie_name, const std::string &uid, uint64_t max_elapsed = COOKIE_DEFAULT_LIFE,
std::shared_ptr<void> attach = nullptr,
int max_client = 1);
/**
* 根据cookie随机字符串查找cookie对象
......@@ -160,7 +172,7 @@ public:
* @param cookie cookie随机字符串
* @return cookie对象，可以为nullptr
*/
HttpServerCookie::Ptr getCookie(const std::string &cookie_name,const std::string &cookie);
HttpServerCookie::Ptr getCookie(const std::string &cookie_name, const std::string &cookie);
/**
* 从http头中获取cookie对象
......@@ -168,7 +180,7 @@ public:
* @param http_header http头
* @return cookie对象
*/
HttpServerCookie::Ptr getCookie(const std::string &cookie_name,const StrCaseMap &http_header);
HttpServerCookie::Ptr getCookie(const std::string &cookie_name, const StrCaseMap &http_header);
/**
* 根据uid获取cookie
......@@ -176,7 +188,7 @@ public:
* @param uid 用户id
* @return cookie对象
*/
HttpServerCookie::Ptr getCookieByUid(const std::string &cookie_name,const std::string &uid);
HttpServerCookie::Ptr getCookieByUid(const std::string &cookie_name, const std::string &uid);
/**
* 删除cookie，用户登出时使用
......@@ -184,8 +196,10 @@ public:
* @return
*/
bool delCookie(const HttpServerCookie::Ptr &cookie);
private:
HttpCookieManager();
void onManager();
/**
* 构造cookie对象时触发，目的是记录某账号下多个cookie
......@@ -193,7 +207,7 @@ private:
* @param uid 用户id
* @param cookie cookie随机字符串
*/
void onAddCookie(const std::string &cookie_name,const std::string &uid,const std::string &cookie);
void onAddCookie(const std::string &cookie_name, const std::string &uid, const std::string &cookie);
/**
* 析构cookie对象时触发
......@@ -201,7 +215,7 @@ private:
* @param uid 用户id
* @param cookie cookie随机字符串
*/
void onDelCookie(const std::string &cookie_name,const std::string &uid,const std::string &cookie);
void onDelCookie(const std::string &cookie_name, const std::string &uid, const std::string &cookie);
/**
* 获取某用户名下最先登录时的cookie，目的是实现某用户下最多登录若干个设备
......@@ -210,7 +224,7 @@ private:
* @param max_client 最多登录的设备个数
* @return 最早的cookie随机字符串
*/
std::string getOldestCookie(const std::string &cookie_name,const std::string &uid, int max_client = 1);
std::string getOldestCookie(const std::string &cookie_name, const std::string &uid, int max_client = 1);
/**
* 删除cookie
......@@ -218,16 +232,21 @@ private:
* @param cookie cookie随机字符串
* @return 成功true
*/
bool delCookie(const std::string &cookie_name,const std::string &cookie);
bool delCookie(const std::string &cookie_name, const std::string &cookie);
private:
std::unordered_map<std::string/*cookie_name*/,std::unordered_map<std::string/*cookie*/,HttpServerCookie::Ptr/*cookie_data*/> >_map_cookie;
std::unordered_map<std::string/*cookie_name*/,std::unordered_map<std::string/*uid*/,std::map<uint64_t/*cookie time stamp*/,std::string/*cookie*/> > >_map_uid_to_cookie;
std::unordered_map<
std::string /*cookie_name*/, std::unordered_map<std::string /*cookie*/, HttpServerCookie::Ptr /*cookie_data*/>>
_map_cookie;
std::unordered_map<
std::string /*cookie_name*/,
std::unordered_map<std::string /*uid*/, std::map<uint64_t /*cookie time stamp*/, std::string /*cookie*/>>>
_map_uid_to_cookie;
std::recursive_mutex _mtx_cookie;
toolkit::Timer::Ptr _timer;
RandStrGeneator _geneator;
};
}//namespace mediakit
} // namespace mediakit
#endif //SRC_HTTP_COOKIEMANAGER_H
#endif // SRC_HTTP_COOKIEMANAGER_H
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/Http/HttpSession.cpp
......@@ -519,16 +519,16 @@ void HttpSession::sendResponse(int code,
GET_CONFIG(uint32_t,keepAliveSec,Http::kKeepAliveSecond);
//body默认为空
ssize_t size = 0;
int64_t size = 0;
if (body && body->remainSize()) {
//有body，获取body大小
size = body->remainSize();
}
if(no_content_length){
//http-flv直播是Keep-Alive类型
if (no_content_length) {
// http-flv直播是Keep-Alive类型
bClose = false;
}else if((size_t) size >= SIZE_MAX || size < 0 ){
} else if ((size_t)size >= SIZE_MAX || size < 0) {
//不固定长度的body，那么发送完body后应该关闭socket，以便浏览器做下载完毕的判断
bClose = true;
}
......@@ -537,47 +537,47 @@ void HttpSession::sendResponse(int code,
headerOut.emplace(kDate, dateStr());
headerOut.emplace(kServer, kServerName);
headerOut.emplace(kConnection, bClose ? "close" : "keep-alive");
if(!bClose){
if (!bClose) {
string keepAliveString = "timeout=";
keepAliveString += to_string(keepAliveSec);
keepAliveString += ", max=100";
headerOut.emplace(kKeepAlive,std::move(keepAliveString));
headerOut.emplace(kKeepAlive, std::move(keepAliveString));
}
if(!_origin.empty()){
if (!_origin.empty()) {
//设置跨域
headerOut.emplace(kAccessControlAllowOrigin,_origin);
headerOut.emplace(kAccessControlAllowOrigin, _origin);
headerOut.emplace(kAccessControlAllowCredentials, "true");
}
if(!no_content_length && size >= 0 && (size_t)size < SIZE_MAX){
if (!no_content_length && size >= 0 && (size_t)size < SIZE_MAX) {
//文件长度为固定值,且不是http-flv强制设置Content-Length
headerOut[kContentLength] = to_string(size);
}
if(size && !pcContentType){
if (size && !pcContentType) {
//有body时，设置缺省类型
pcContentType = "text/plain";
}
if((size || no_content_length) && pcContentType){
if ((size || no_content_length) && pcContentType) {
//有body时，设置文件类型
string strContentType = pcContentType;
strContentType += "; charset=";
strContentType += charSet;
headerOut.emplace(kContentType,std::move(strContentType));
headerOut.emplace(kContentType, std::move(strContentType));
}
//发送http头
string str;
str.reserve(256);
str += "HTTP/1.1 " ;
str += "HTTP/1.1 ";
str += to_string(code);
str += ' ';
str += getHttpStatusMessage(code) ;
str += getHttpStatusMessage(code);
str += "\r\n";
for (auto &pr : header) {
str += pr.first ;
str += pr.first;
str += ": ";
str += pr.second;
str += "\r\n";
......@@ -586,18 +586,21 @@ void HttpSession::sendResponse(int code,
SockSender::send(std::move(str));
_ticker.resetTime();
if(!size){
if (!size) {
//没有body
if(bClose){
if (bClose) {
shutdown(SockException(Err_shutdown,StrPrinter << "close connection after send http header completed with status code:" << code));
}
return;
}
#if 0
//sendfile跟共享mmap相比并没有性能上的优势，相反，sendfile还有功能上的缺陷，先屏蔽
if (typeid(*this) == typeid(HttpSession) && !body->sendFile(getSock()->rawFD())) {
//http支持sendfile优化
// http支持sendfile优化
return;
}
#endif
GET_CONFIG(uint32_t, sendBufSize, Http::kSendBufSize);
if (body->remainSize() > sendBufSize) {
......@@ -607,9 +610,7 @@ void HttpSession::sendResponse(int code,
//发送http body
AsyncSenderData::Ptr data = std::make_shared<AsyncSenderData>(shared_from_this(), body, bClose);
getSock()->setOnFlush([data]() {
return AsyncSender::onSocketFlushed(data);
});
getSock()->setOnFlush([data]() { return AsyncSender::onSocketFlushed(data); });
AsyncSender::onSocketFlushed(data);
}
......
src/Http/TsPlayer.cpp
......@@ -20,6 +20,7 @@ TsPlayer::TsPlayer(const EventPoller::Ptr &poller) : HttpTSPlayer(poller, true)
void TsPlayer::play(const string &url) {
TraceL << "play http-ts: " << url;
_play_result = false;
_benchmark_mode = (*this)[Client::kBenchmarkMode].as<int>();
setHeaderTimeout((*this)[Client::kTimeoutMS].as<int>());
setBodyTimeout((*this)[Client::kMediaTimeoutMS].as<int>());
setMethod("GET");
......@@ -45,7 +46,9 @@ void TsPlayer::onResponseBody(const char *buf, size_t size) {
_play_result = true;
onPlayResult(SockException(Err_success, "play http-ts success"));
}
HttpTSPlayer::onResponseBody(buf, size);
if (!_benchmark_mode) {
HttpTSPlayer::onResponseBody(buf, size);
}
}
} // namespace mediakit
\ No newline at end of file
src/Http/TsPlayer.h
......@@ -16,7 +16,7 @@
namespace mediakit {
class TsPlayer : public HttpTSPlayer , public PlayerBase {
class TsPlayer : public HttpTSPlayer, public PlayerBase {
public:
TsPlayer(const toolkit::EventPoller::Ptr &poller);
~TsPlayer() override = default;
......@@ -37,6 +37,7 @@ protected:
private:
bool _play_result = true;
bool _benchmark_mode = false;
};
} // namespace mediakit
......
src/Http/TsplayerImp.cpp
......@@ -33,7 +33,8 @@ void TsPlayerImp::addTrackCompleted() {
}
void TsPlayerImp::onPlayResult(const SockException &ex) {
if (ex) {
auto benchmark_mode = (*this)[Client::kBenchmarkMode].as<int>();
if (ex || benchmark_mode) {
PlayerImp<TsPlayer, PlayerBase>::onPlayResult(ex);
} else {
auto demuxer = std::make_shared<HlsDemuxer>();
......@@ -47,7 +48,10 @@ void TsPlayerImp::onShutdown(const SockException &ex) {
_demuxer = nullptr;
}
vector <Track::Ptr> TsPlayerImp::getTracks(bool ready) const {
vector<Track::Ptr> TsPlayerImp::getTracks(bool ready) const {
if (!_demuxer) {
return vector<Track::Ptr>();
}
return static_pointer_cast<HlsDemuxer>(_demuxer)->getTracks(ready);
}
......
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 人 到此讨论。请谨慎行事。
请先完成此评论的编辑！
请 注册 或者 后发表评论