ZLMediaKit/src/Rtmp/RtmpSession.cpp

575 lines
21 KiB
C++
Raw Normal View History

2017-10-09 22:11:01 +08:00
/*
2020-04-04 20:30:09 +08:00
* Copyright (c) 2016 The ZLMediaKit project authors. All Rights Reserved.
2017-09-27 16:20:30 +08:00
*
* This file is part of ZLMediaKit(https://github.com/xia-chu/ZLMediaKit).
2017-09-27 16:20:30 +08:00
*
2020-04-04 20:30:09 +08:00
* Use of this source code is governed by MIT license that can be found in the
* LICENSE file in the root of the source tree. All contributing project authors
* may be found in the AUTHORS file in the root of the source tree.
2017-04-01 16:35:56 +08:00
*/
#include "RtmpSession.h"
#include "Common/config.h"
2017-04-01 16:35:56 +08:00
#include "Util/onceToken.h"
2018-10-24 17:17:55 +08:00
namespace mediakit {
2017-04-01 16:35:56 +08:00
2020-08-30 10:48:34 +08:00
RtmpSession::RtmpSession(const Socket::Ptr &sock) : TcpSession(sock) {
2020-03-20 11:51:24 +08:00
DebugP(this);
2019-05-30 10:41:25 +08:00
GET_CONFIG(uint32_t,keep_alive_sec,Rtmp::kKeepAliveSecond);
2020-08-30 10:48:34 +08:00
sock->setSendTimeOutSecond(keep_alive_sec);
2017-04-01 16:35:56 +08:00
}
RtmpSession::~RtmpSession() {
2019-05-28 18:46:52 +08:00
DebugP(this);
2017-04-01 16:35:56 +08:00
}
void RtmpSession::onError(const SockException& err) {
2020-08-30 10:48:34 +08:00
bool isPlayer = !_publisher_src;
2020-02-13 12:10:08 +08:00
uint64_t duration = _ticker.createdTime()/1000;
2019-12-29 15:38:29 +08:00
WarnP(this) << (isPlayer ? "RTMP播放器(" : "RTMP推流器(")
2020-08-30 10:48:34 +08:00
<< _media_info._vhost << "/"
<< _media_info._app << "/"
<< _media_info._streamid
2020-02-13 12:10:08 +08:00
<< ")断开:" << err.what()
<< ",耗时(s):" << duration;
//流量统计事件广播
2019-05-28 17:14:36 +08:00
GET_CONFIG(uint32_t,iFlowThreshold,General::kFlowThreshold);
2018-02-09 11:42:55 +08:00
if(_total_bytes >= iFlowThreshold * 1024){
2020-08-30 10:48:34 +08:00
NoticeCenter::Instance().emitEvent(Broadcast::kBroadcastFlowReport, _media_info, _total_bytes, duration, isPlayer, static_cast<SockInfo &>(*this));
}
2017-04-01 16:35:56 +08:00
}
void RtmpSession::onManager() {
2019-08-28 18:20:40 +08:00
GET_CONFIG(uint32_t,handshake_sec,Rtmp::kHandshakeSecond);
2019-05-29 18:24:35 +08:00
GET_CONFIG(uint32_t,keep_alive_sec,Rtmp::kKeepAliveSecond);
2020-03-20 11:51:24 +08:00
if (_ticker.createdTime() > handshake_sec * 1000) {
2020-08-30 10:48:34 +08:00
if (!_ring_reader && !_publisher_src) {
2020-03-20 11:51:24 +08:00
shutdown(SockException(Err_timeout,"illegal connection"));
}
}
2020-08-30 10:48:34 +08:00
if (_publisher_src) {
2020-03-20 11:51:24 +08:00
//publisher
if (_ticker.elapsedTime() > keep_alive_sec * 1000) {
shutdown(SockException(Err_timeout,"recv data from rtmp pusher timeout"));
}
}
2017-04-01 16:35:56 +08:00
}
2020-08-30 10:48:34 +08:00
void RtmpSession::onRecv(const Buffer::Ptr &buf) {
2020-03-20 11:51:24 +08:00
_ticker.resetTime();
try {
2020-08-30 10:48:34 +08:00
_total_bytes += buf->size();
onParseRtmp(buf->data(), buf->size());
} catch (exception &ex) {
shutdown(SockException(Err_shutdown, ex.what()));
2020-03-20 11:51:24 +08:00
}
2017-04-01 16:35:56 +08:00
}
void RtmpSession::onCmd_connect(AMFDecoder &dec) {
2020-03-20 11:51:24 +08:00
auto params = dec.load<AMFValue>();
///////////set chunk size////////////////
sendChunkSize(60000);
////////////window Acknowledgement size/////
sendAcknowledgementSize(5000000);
///////////set peerBandwidth////////////////
sendPeerBandwidth(5000000);
2017-05-13 17:25:31 +08:00
2020-08-30 10:48:34 +08:00
_media_info._app = params["app"].as_string();
_tc_url = params["tcUrl"].as_string();
if(_tc_url.empty()){
//defaultVhost:默认vhost
2020-08-30 10:48:34 +08:00
_tc_url = string(RTMP_SCHEMA) + "://" + DEFAULT_VHOST + "/" + _media_info._app;
2021-01-23 10:06:38 +08:00
} else {
auto pos = _tc_url.rfind('?');
if (pos != string::npos) {
//tc_url 中可能包含?以及参数参见issue: #692
_tc_url = _tc_url.substr(0, pos);
}
}
2020-03-20 11:51:24 +08:00
bool ok = true; //(app == APP_NAME);
AMFValue version(AMF_OBJECT);
version.set("fmsVer", "FMS/3,0,1,123");
version.set("capabilities", 31.0);
AMFValue status(AMF_OBJECT);
status.set("level", ok ? "status" : "error");
status.set("code", ok ? "NetConnection.Connect.Success" : "NetConnection.Connect.InvalidApp");
status.set("description", ok ? "Connection succeeded." : "InvalidApp.");
2021-03-07 10:04:27 +08:00
status.set("objectEncoding", params["objectEncoding"]);
2020-03-20 11:51:24 +08:00
sendReply(ok ? "_result" : "_error", version, status);
if (!ok) {
2020-08-30 10:48:34 +08:00
throw std::runtime_error("Unsupported application: " + _media_info._app);
2020-03-20 11:51:24 +08:00
}
AMFEncoder invoke;
invoke << "onBWDone" << 0.0 << nullptr;
sendResponse(MSG_CMD, invoke.data());
2017-04-01 16:35:56 +08:00
}
void RtmpSession::onCmd_createStream(AMFDecoder &dec) {
2020-03-20 11:51:24 +08:00
sendReply("_result", nullptr, double(STREAM_MEDIA));
2017-04-01 16:35:56 +08:00
}
void RtmpSession::onCmd_publish(AMFDecoder &dec) {
2020-08-30 10:48:34 +08:00
std::shared_ptr<Ticker> ticker(new Ticker);
weak_ptr<RtmpSession> weak_self = dynamic_pointer_cast<RtmpSession>(shared_from_this());
std::shared_ptr<onceToken> pToken(new onceToken(nullptr,[ticker,weak_self](){
auto strong_self = weak_self.lock();
if(strong_self){
DebugP(strong_self.get()) << "publish 回复时间:" << ticker->elapsedTime() << "ms";
2019-05-28 18:46:52 +08:00
}
}));
2020-03-20 11:51:24 +08:00
dec.load<AMFValue>();/* NULL */
2020-08-30 10:48:34 +08:00
_media_info.parse(_tc_url + "/" + getStreamId(dec.load<std::string>()));
_media_info._schema = RTMP_SCHEMA;
2020-09-12 19:09:56 +08:00
auto on_res = [this,pToken](const string &err, bool enableHls, bool enableMP4){
auto src = dynamic_pointer_cast<RtmpMediaSource>(MediaSource::find(RTMP_SCHEMA,
2020-08-30 10:48:34 +08:00
_media_info._vhost,
_media_info._app,
_media_info._streamid));
bool auth_success = err.empty();
bool ok = (!src && !_publisher_src && auth_success);
AMFValue status(AMF_OBJECT);
status.set("level", ok ? "status" : "error");
2020-08-30 10:48:34 +08:00
status.set("code", ok ? "NetStream.Publish.Start" : (auth_success ? "NetStream.Publish.BadName" : "NetStream.Publish.BadAuth"));
status.set("description", ok ? "Started publishing stream." : (auth_success ? "Already publishing." : err.data()));
status.set("clientid", "0");
sendReply("onStatus", nullptr, status);
if (!ok) {
2020-08-30 10:48:34 +08:00
string errMsg = StrPrinter << (auth_success ? "already publishing:" : err.data()) << " "
<< _media_info._vhost << " "
<< _media_info._app << " "
<< _media_info._streamid;
2019-05-29 18:08:50 +08:00
shutdown(SockException(Err_shutdown,errMsg));
return;
}
2020-08-30 10:48:34 +08:00
_publisher_src.reset(new RtmpMediaSourceImp(_media_info._vhost, _media_info._app, _media_info._streamid));
_publisher_src->setListener(dynamic_pointer_cast<MediaSourceEvent>(shared_from_this()));
//设置转协议
2020-09-12 19:09:56 +08:00
_publisher_src->setProtocolTranslation(enableHls, enableMP4);
setSocketFlags();
};
2020-08-30 10:48:34 +08:00
if(_media_info._app.empty() || _media_info._streamid.empty()){
2020-05-25 16:40:41 +08:00
//不允许莫名其妙的推流url
2020-09-12 19:09:56 +08:00
on_res("rtmp推流url非法", false, false);
2020-05-25 16:40:41 +08:00
return;
}
2020-09-12 19:09:56 +08:00
Broadcast::PublishAuthInvoker invoker = [weak_self, on_res, pToken](const string &err, bool enableHls, bool enableMP4) {
2020-08-30 10:48:34 +08:00
auto strongSelf = weak_self.lock();
2020-09-12 19:09:56 +08:00
if (!strongSelf) {
return;
}
2020-09-12 19:09:56 +08:00
strongSelf->async([weak_self, on_res, err, pToken, enableHls, enableMP4]() {
2020-08-30 10:48:34 +08:00
auto strongSelf = weak_self.lock();
2020-09-12 19:09:56 +08:00
if (!strongSelf) {
return;
}
2020-09-12 19:09:56 +08:00
on_res(err, enableHls, enableMP4);
});
};
2020-08-30 10:48:34 +08:00
auto flag = NoticeCenter::Instance().emitEvent(Broadcast::kBroadcastMediaPublish, _media_info, invoker, static_cast<SockInfo &>(*this));
if(!flag){
//该事件无人监听,默认鉴权成功
2020-08-30 10:48:34 +08:00
GET_CONFIG(bool,to_hls,General::kPublishToHls);
GET_CONFIG(bool,to_mp4,General::kPublishToMP4);
2020-09-12 19:09:56 +08:00
on_res("", to_hls, to_mp4);
}
2017-04-01 16:35:56 +08:00
}
void RtmpSession::onCmd_deleteStream(AMFDecoder &dec) {
2020-03-20 11:51:24 +08:00
AMFValue status(AMF_OBJECT);
status.set("level", "status");
status.set("code", "NetStream.Unpublish.Success");
status.set("description", "Stop publishing.");
sendReply("onStatus", nullptr, status);
throw std::runtime_error(StrPrinter << "Stop publishing" << endl);
2017-04-01 16:35:56 +08:00
}
2018-10-31 12:11:14 +08:00
void RtmpSession::sendPlayResponse(const string &err,const RtmpMediaSource::Ptr &src){
2020-08-30 10:48:34 +08:00
bool auth_success = err.empty();
bool ok = (src.operator bool() && auth_success);
2018-03-14 22:35:54 +08:00
if (ok) {
//stream begin
sendUserControl(CONTROL_STREAM_BEGIN, STREAM_MEDIA);
}
// onStatus(NetStream.Play.Reset)
AMFValue status(AMF_OBJECT);
status.set("level", ok ? "status" : "error");
2020-08-30 10:48:34 +08:00
status.set("code", ok ? "NetStream.Play.Reset" : (auth_success ? "NetStream.Play.StreamNotFound" : "NetStream.Play.BadAuth"));
status.set("description", ok ? "Resetting and playing." : (auth_success ? "No such stream." : err.data()));
status.set("details", _media_info._streamid);
2018-03-14 22:35:54 +08:00
status.set("clientid", "0");
sendReply("onStatus", nullptr, status);
if (!ok) {
2020-08-30 10:48:34 +08:00
string err_msg = StrPrinter << (auth_success ? "no such stream:" : err.data()) << " "
<< _media_info._vhost << " "
<< _media_info._app << " "
<< _media_info._streamid;
shutdown(SockException(Err_shutdown, err_msg));
2018-03-14 22:35:54 +08:00
return;
}
2017-04-01 16:35:56 +08:00
2018-03-14 22:35:54 +08:00
// onStatus(NetStream.Play.Start)
status.clear();
status.set("level", "status");
status.set("code", "NetStream.Play.Start");
status.set("description", "Started playing.");
2020-08-30 10:48:34 +08:00
status.set("details", _media_info._streamid);
2018-03-14 22:35:54 +08:00
status.set("clientid", "0");
sendReply("onStatus", nullptr, status);
// |RtmpSampleAccess(true, true)
AMFEncoder invoke;
invoke << "|RtmpSampleAccess" << true << true;
sendResponse(MSG_DATA, invoke.data());
//onStatus(NetStream.Data.Start)
invoke.clear();
AMFValue obj(AMF_OBJECT);
obj.set("code", "NetStream.Data.Start");
invoke << "onStatus" << obj;
sendResponse(MSG_DATA, invoke.data());
//onStatus(NetStream.Play.PublishNotify)
status.clear();
status.set("level", "status");
status.set("code", "NetStream.Play.PublishNotify");
status.set("description", "Now published.");
2020-08-30 10:48:34 +08:00
status.set("details", _media_info._streamid);
2018-03-14 22:35:54 +08:00
status.set("clientid", "0");
sendReply("onStatus", nullptr, status);
2019-09-21 19:27:34 +08:00
auto &metadata = src->getMetaData();
if(metadata){
//在有metadata的情况下才发送metadata
//其实metadata没什么用有些推流器不产生metadata
// onMetaData
invoke.clear();
invoke << "onMetaData" << metadata;
sendResponse(MSG_DATA, invoke.data());
auto duration = metadata["duration"];
if(duration && duration.as_number() > 0){
2020-04-04 15:37:37 +08:00
//这是点播,使用绝对时间戳
_stamp[0].setPlayBack();
_stamp[1].setPlayBack();
}
2019-09-21 19:27:34 +08:00
}
2018-03-14 22:35:54 +08:00
src->getConfigFrame([&](const RtmpPacket::Ptr &pkt) {
2019-05-28 18:46:52 +08:00
//DebugP(this)<<"send initial frame";
2018-03-14 22:35:54 +08:00
onSendMedia(pkt);
});
//音频同步于视频
2020-05-15 18:08:54 +08:00
_stamp[0].syncTo(_stamp[1]);
2020-08-30 10:48:34 +08:00
_ring_reader = src->getRing()->attach(getPoller());
2018-03-14 22:35:54 +08:00
weak_ptr<RtmpSession> weakSelf = dynamic_pointer_cast<RtmpSession>(shared_from_this());
2020-08-30 10:48:34 +08:00
_ring_reader->setReadCB([weakSelf](const RtmpMediaSource::RingDataType &pkt) {
2018-03-14 22:35:54 +08:00
auto strongSelf = weakSelf.lock();
if (!strongSelf) {
return;
}
2020-04-09 16:19:03 +08:00
if(strongSelf->_paused){
return;
}
size_t i = 0;
auto size = pkt->size();
2020-04-09 16:19:03 +08:00
strongSelf->setSendFlushFlag(false);
pkt->for_each([&](const RtmpPacket::Ptr &rtmp){
if(++i == size){
strongSelf->setSendFlushFlag(true);
}
strongSelf->onSendMedia(rtmp);
});
2018-03-14 22:35:54 +08:00
});
2020-08-30 10:48:34 +08:00
_ring_reader->setDetachCB([weakSelf]() {
2018-03-14 22:35:54 +08:00
auto strongSelf = weakSelf.lock();
if (!strongSelf) {
return;
}
2019-05-29 18:08:50 +08:00
strongSelf->shutdown(SockException(Err_shutdown,"rtmp ring buffer detached"));
2018-03-14 22:35:54 +08:00
});
2020-08-30 10:48:34 +08:00
_player_src = src;
2019-12-28 16:57:35 +08:00
if (src->totalReaderCount() == 1) {
2018-03-14 22:35:54 +08:00
src->seekTo(0);
}
//提高服务器发送性能
setSocketFlags();
2018-03-14 22:35:54 +08:00
}
2018-10-31 12:11:14 +08:00
void RtmpSession::doPlayResponse(const string &err,const std::function<void(bool)> &cb){
if(!err.empty()){
//鉴权失败,直接返回播放失败
sendPlayResponse(err, nullptr);
cb(false);
return;
}
//鉴权成功,查找媒体源并回复
2020-08-30 10:48:34 +08:00
weak_ptr<RtmpSession> weak_self = dynamic_pointer_cast<RtmpSession>(shared_from_this());
MediaSource::findAsync(_media_info, weak_self.lock(), [weak_self,cb](const MediaSource::Ptr &src){
auto rtmp_src = dynamic_pointer_cast<RtmpMediaSource>(src);
2020-08-30 10:48:34 +08:00
auto strong_self = weak_self.lock();
if(strong_self){
strong_self->sendPlayResponse("", rtmp_src);
2018-10-31 12:11:14 +08:00
}
cb(rtmp_src.operator bool());
2018-10-31 12:11:14 +08:00
});
}
2018-03-14 22:35:54 +08:00
void RtmpSession::doPlay(AMFDecoder &dec){
2020-08-30 10:48:34 +08:00
std::shared_ptr<Ticker> ticker(new Ticker);
weak_ptr<RtmpSession> weak_self = dynamic_pointer_cast<RtmpSession>(shared_from_this());
std::shared_ptr<onceToken> token(new onceToken(nullptr, [ticker,weak_self](){
auto strongSelf = weak_self.lock();
if (strongSelf) {
DebugP(strongSelf.get()) << "play 回复时间:" << ticker->elapsedTime() << "ms";
2019-05-28 18:46:52 +08:00
}
}));
2020-08-30 10:48:34 +08:00
Broadcast::AuthInvoker invoker = [weak_self,token](const string &err){
auto strong_self = weak_self.lock();
if (!strong_self) {
return;
}
2020-08-30 10:48:34 +08:00
strong_self->async([weak_self, err, token]() {
auto strong_self = weak_self.lock();
if (!strong_self) {
return;
}
2020-08-30 10:48:34 +08:00
strong_self->doPlayResponse(err, [token](bool) {});
});
};
2020-08-30 10:48:34 +08:00
auto flag = NoticeCenter::Instance().emitEvent(Broadcast::kBroadcastMediaPlayed, _media_info, invoker, static_cast<SockInfo &>(*this));
if(!flag){
//该事件无人监听,默认不鉴权
2020-08-30 10:48:34 +08:00
doPlayResponse("",[token](bool){});
}
2017-05-18 16:24:36 +08:00
}
2020-08-30 10:48:34 +08:00
2017-05-18 16:24:36 +08:00
void RtmpSession::onCmd_play2(AMFDecoder &dec) {
2020-03-20 11:51:24 +08:00
doPlay(dec);
2017-05-18 16:24:36 +08:00
}
2019-09-06 18:30:42 +08:00
string RtmpSession::getStreamId(const string &str){
string stream_id;
string params;
auto pos = str.find('?');
2020-08-30 10:48:34 +08:00
if (pos != string::npos) {
2019-09-06 18:30:42 +08:00
//有url参数
2020-08-30 10:48:34 +08:00
stream_id = str.substr(0, pos);
2019-09-06 18:30:42 +08:00
//获取url参数
params = str.substr(pos + 1);
2020-08-30 10:48:34 +08:00
} else {
2019-09-06 18:30:42 +08:00
//没有url参数
stream_id = str;
}
pos = stream_id.find(":");
2020-08-30 10:48:34 +08:00
if (pos != string::npos) {
2019-09-06 18:30:42 +08:00
//vlc和ffplay在播放 rtmp://127.0.0.1/record/0.mp4时
//传过来的url会是rtmp://127.0.0.1/record/mp4:0,
//我们在这里还原成0.mp4
//实际使用时发现vlcmpv等会传过来rtmp://127.0.0.1/record/mp4:0.mp4,这里做个判断
2020-08-30 10:48:34 +08:00
auto ext = stream_id.substr(0, pos);
stream_id = stream_id.substr(pos + 1);
2020-08-30 10:48:34 +08:00
if (stream_id.find(ext) == string::npos) {
stream_id = stream_id + "." + ext;
}
2019-09-06 18:30:42 +08:00
}
2020-08-30 10:48:34 +08:00
if (params.empty()) {
2019-09-06 18:30:42 +08:00
//没有url参数
return stream_id;
}
//有url参数
return stream_id + '?' + params;
}
2017-05-18 16:24:36 +08:00
void RtmpSession::onCmd_play(AMFDecoder &dec) {
2020-03-20 11:51:24 +08:00
dec.load<AMFValue>();/* NULL */
2020-08-30 10:48:34 +08:00
_media_info.parse(_tc_url + "/" + getStreamId(dec.load<std::string>()));
_media_info._schema = RTMP_SCHEMA;
2020-03-20 11:51:24 +08:00
doPlay(dec);
2017-04-01 16:35:56 +08:00
}
void RtmpSession::onCmd_pause(AMFDecoder &dec) {
2020-03-20 11:51:24 +08:00
dec.load<AMFValue>();/* NULL */
bool paused = dec.load<bool>();
TraceP(this) << paused;
AMFValue status(AMF_OBJECT);
status.set("level", "status");
status.set("code", paused ? "NetStream.Pause.Notify" : "NetStream.Unpause.Notify");
status.set("description", paused ? "Paused stream." : "Unpaused stream.");
sendReply("onStatus", nullptr, status);
2020-04-09 16:19:03 +08:00
//streamBegin
sendUserControl(paused ? CONTROL_STREAM_EOF : CONTROL_STREAM_BEGIN, STREAM_MEDIA);
_paused = paused;
2017-04-01 16:35:56 +08:00
}
void RtmpSession::setMetaData(AMFDecoder &dec) {
2020-03-20 11:51:24 +08:00
std::string type = dec.load<std::string>();
if (type != "onMetaData") {
throw std::runtime_error("can only set metadata");
}
2021-04-01 21:56:21 +08:00
_publisher_metadata = dec.load<AMFValue>();
2017-04-01 16:35:56 +08:00
}
void RtmpSession::onProcessCmd(AMFDecoder &dec) {
2020-08-30 10:48:34 +08:00
typedef void (RtmpSession::*cmd_function)(AMFDecoder &dec);
static unordered_map<string, cmd_function> s_cmd_functions;
static onceToken token([]() {
2020-08-30 10:48:34 +08:00
s_cmd_functions.emplace("connect", &RtmpSession::onCmd_connect);
s_cmd_functions.emplace("createStream", &RtmpSession::onCmd_createStream);
s_cmd_functions.emplace("publish", &RtmpSession::onCmd_publish);
s_cmd_functions.emplace("deleteStream", &RtmpSession::onCmd_deleteStream);
s_cmd_functions.emplace("play", &RtmpSession::onCmd_play);
s_cmd_functions.emplace("play2", &RtmpSession::onCmd_play2);
s_cmd_functions.emplace("seek", &RtmpSession::onCmd_seek);
s_cmd_functions.emplace("pause", &RtmpSession::onCmd_pause);
});
std::string method = dec.load<std::string>();
2020-03-20 11:51:24 +08:00
auto it = s_cmd_functions.find(method);
if (it == s_cmd_functions.end()) {
2020-01-08 14:00:53 +08:00
// TraceP(this) << "can not support cmd:" << method;
2020-03-20 11:51:24 +08:00
return;
}
2020-08-30 10:48:34 +08:00
_recv_req_id = dec.load<double>();
2020-03-20 11:51:24 +08:00
auto fun = it->second;
(this->*fun)(dec);
2017-04-01 16:35:56 +08:00
}
2021-02-04 17:58:51 +08:00
void RtmpSession::onRtmpChunk(RtmpPacket::Ptr packet) {
auto &chunk_data = *packet;
2020-08-30 10:48:34 +08:00
switch (chunk_data.type_id) {
2020-03-20 11:51:24 +08:00
case MSG_CMD:
case MSG_CMD3: {
2021-04-18 21:02:01 +08:00
AMFDecoder dec(chunk_data.buffer, chunk_data.type_id == MSG_CMD3 ? 3 : 0);
2020-03-20 11:51:24 +08:00
onProcessCmd(dec);
break;
2020-08-30 10:48:34 +08:00
}
2020-03-20 11:51:24 +08:00
case MSG_DATA:
case MSG_DATA3: {
2021-04-18 21:02:01 +08:00
AMFDecoder dec(chunk_data.buffer, chunk_data.type_id == MSG_DATA3 ? 3 : 0);
2020-03-20 11:51:24 +08:00
std::string type = dec.load<std::string>();
if (type == "@setDataFrame") {
setMetaData(dec);
} else if(type == "onMetaData"){
//setMetaData(dec);
_publisher_metadata = dec.load<AMFValue>();
}else {
2020-01-08 14:03:56 +08:00
TraceP(this) << "unknown notify:" << type;
}
2020-03-20 11:51:24 +08:00
break;
2020-08-30 10:48:34 +08:00
}
2020-03-20 11:51:24 +08:00
case MSG_AUDIO:
case MSG_VIDEO: {
2020-08-30 10:48:34 +08:00
if (!_publisher_src) {
2021-04-01 21:56:21 +08:00
WarnL << "Not a rtmp publisher!";
return;
2020-03-20 11:51:24 +08:00
}
2020-08-30 10:48:34 +08:00
GET_CONFIG(bool, rtmp_modify_stamp, Rtmp::kModifyStamp);
if (rtmp_modify_stamp) {
2019-08-22 14:56:58 +08:00
int64_t dts_out;
2020-08-30 10:48:34 +08:00
_stamp[chunk_data.type_id % 2].revise(chunk_data.time_stamp, chunk_data.time_stamp, dts_out, dts_out, true);
chunk_data.time_stamp = (uint32_t)dts_out;
2019-08-22 14:56:58 +08:00
}
2020-01-13 15:48:55 +08:00
2021-04-01 21:56:21 +08:00
if (!_set_meta_data) {
2020-01-13 15:48:55 +08:00
_set_meta_data = true;
2021-04-01 21:56:21 +08:00
_publisher_src->setMetaData(_publisher_metadata ? _publisher_metadata : TitleMeta().getMetadata());
2020-01-13 15:48:55 +08:00
}
2021-02-04 17:58:51 +08:00
_publisher_src->onWrite(std::move(packet));
2020-03-20 11:51:24 +08:00
break;
2020-08-30 10:48:34 +08:00
}
2020-03-20 11:51:24 +08:00
default:
2020-08-30 10:48:34 +08:00
WarnP(this) << "unhandled message:" << (int) chunk_data.type_id << hexdump(chunk_data.buffer.data(), chunk_data.buffer.size());
2020-03-20 11:51:24 +08:00
break;
}
2017-04-01 16:35:56 +08:00
}
void RtmpSession::onCmd_seek(AMFDecoder &dec) {
dec.load<AMFValue>();/* NULL */
2020-03-20 11:51:24 +08:00
AMFValue status(AMF_OBJECT);
AMFEncoder invoke;
status.set("level", "status");
status.set("code", "NetStream.Seek.Notify");
status.set("description", "Seeking.");
sendReply("onStatus", nullptr, status);
2020-04-04 15:37:37 +08:00
auto milliSeconds = (uint32_t)(dec.load<AMFValue>().as_number());
2020-04-04 15:37:37 +08:00
InfoP(this) << "rtmp seekTo(ms):" << milliSeconds;
2020-08-30 10:48:34 +08:00
auto strong_src = _player_src.lock();
if (strong_src) {
strong_src->seekTo(milliSeconds);
2020-04-04 15:37:37 +08:00
}
2017-04-01 16:35:56 +08:00
}
2017-12-04 23:55:09 +08:00
void RtmpSession::onSendMedia(const RtmpPacket::Ptr &pkt) {
2019-08-22 14:56:58 +08:00
//rtmp播放器时间戳从零开始
int64_t dts_out;
2020-08-30 10:48:34 +08:00
_stamp[pkt->type_id % 2].revise(pkt->time_stamp, 0, dts_out, dts_out);
sendRtmp(pkt->type_id, pkt->stream_index, pkt, (uint32_t)dts_out, pkt->chunk_id);
2017-04-01 16:35:56 +08:00
}
2018-03-14 22:35:54 +08:00
2019-05-31 15:40:55 +08:00
bool RtmpSession::close(MediaSource &sender,bool force) {
//此回调在其他线程触发
2020-08-30 10:48:34 +08:00
if(!_publisher_src || (!force && _publisher_src->totalReaderCount())){
2019-05-31 15:40:55 +08:00
return false;
}
string err = StrPrinter << "close media:" << sender.getSchema() << "/" << sender.getVhost() << "/" << sender.getApp() << "/" << sender.getId() << " " << force;
safeShutdown(SockException(Err_shutdown,err));
return true;
}
2019-12-28 16:48:11 +08:00
int RtmpSession::totalReaderCount(MediaSource &sender) {
2020-08-30 10:48:34 +08:00
return _publisher_src ? _publisher_src->totalReaderCount() : sender.readerCount();
2019-12-28 16:48:11 +08:00
}
2020-09-27 11:32:49 +08:00
MediaOriginType RtmpSession::getOriginType(MediaSource &sender) const{
return MediaOriginType::rtmp_push;
}
string RtmpSession::getOriginUrl(MediaSource &sender) const {
return _media_info._full_url;
}
std::shared_ptr<SockInfo> RtmpSession::getOriginSock(MediaSource &sender) const {
return const_cast<RtmpSession *>(this)->shared_from_this();
}
void RtmpSession::setSocketFlags(){
2020-08-30 10:48:34 +08:00
GET_CONFIG(int, merge_write_ms, General::kMergeWriteMS);
if (merge_write_ms > 0) {
//推流模式下关闭TCP_NODELAY会增加推流端的延时但是服务器性能将提高
SockUtil::setNoDelay(getSock()->rawFD(), false);
//播放模式下开启MSG_MORE会增加延时但是能提高发送性能
2020-04-23 17:50:12 +08:00
setSendFlags(SOCKET_DEFAULE_FLAGS | FLAG_MORE);
}
}
2019-09-21 19:27:34 +08:00
void RtmpSession::dumpMetadata(const AMFValue &metadata) {
2020-08-30 10:48:34 +08:00
if (metadata.type() != AMF_OBJECT && metadata.type() != AMF_ECMA_ARRAY) {
WarnL << "invalid metadata type:" << metadata.type();
2020-08-30 10:48:34 +08:00
return;
2019-09-21 19:27:34 +08:00
}
_StrPrinter printer;
2020-08-30 10:48:34 +08:00
metadata.object_for_each([&](const string &key, const AMFValue &val) {
printer << "\r\n" << key << "\t:" << val.to_string();
2019-09-21 19:27:34 +08:00
});
2020-08-30 10:48:34 +08:00
InfoL << _media_info._vhost << " " << _media_info._app << " " << _media_info._streamid << (string) printer;
2019-09-21 19:27:34 +08:00
}
2018-10-24 17:17:55 +08:00
} /* namespace mediakit */