tab统一替换为4个空格键:#242

This commit is contained in:
xiongziliang
2020-03-20 11:51:24 +08:00
parent 2a0d78fd12
commit 1168174c2b
84 changed files with 6541 additions and 6520 deletions

View File

@@ -41,27 +41,27 @@ namespace mediakit {
class RtmpDemuxer : public Demuxer{
public:
typedef std::shared_ptr<RtmpDemuxer> Ptr;
typedef std::shared_ptr<RtmpDemuxer> Ptr;
RtmpDemuxer() = default;
virtual ~RtmpDemuxer() = default;
RtmpDemuxer() = default;
virtual ~RtmpDemuxer() = default;
void loadMetaData(const AMFValue &metadata);
void loadMetaData(const AMFValue &metadata);
/**
* 开始解复用
* @param pkt rtmp包
* @return true 代表是i帧
*/
bool inputRtmp(const RtmpPacket::Ptr &pkt);
/**
* 开始解复用
* @param pkt rtmp包
* @return true 代表是i帧
*/
bool inputRtmp(const RtmpPacket::Ptr &pkt);
private:
void makeVideoTrack(const AMFValue &val);
void makeAudioTrack(const AMFValue &val);
void makeVideoTrack(const AMFValue &val);
void makeAudioTrack(const AMFValue &val);
private:
bool _tryedGetVideoTrack = false;
bool _tryedGetAudioTrack = false;
RtmpCodec::Ptr _audioRtmpDecoder;
RtmpCodec::Ptr _videoRtmpDecoder;
bool _tryedGetVideoTrack = false;
bool _tryedGetAudioTrack = false;
RtmpCodec::Ptr _audioRtmpDecoder;
RtmpCodec::Ptr _videoRtmpDecoder;
};
} /* namespace mediakit */

View File

@@ -58,163 +58,163 @@ namespace mediakit {
*/
class RtmpMediaSource : public MediaSource, public RingDelegate<RtmpPacket::Ptr> {
public:
typedef std::shared_ptr<RtmpMediaSource> Ptr;
typedef RingBuffer<RtmpPacket::Ptr> RingType;
typedef std::shared_ptr<RtmpMediaSource> Ptr;
typedef RingBuffer<RtmpPacket::Ptr> RingType;
/**
* 构造函数
* @param vhost 虚拟主机名
* @param app 应用名
* @param stream_id 流id
* @param ring_size 可以设置固定的环形缓冲大小0则自适应
*/
RtmpMediaSource(const string &vhost,
const string &app,
const string &stream_id,
int ring_size = RTMP_GOP_SIZE) :
MediaSource(RTMP_SCHEMA, vhost, app, stream_id), _ring_size(ring_size) {
}
/**
* 构造函数
* @param vhost 虚拟主机名
* @param app 应用名
* @param stream_id 流id
* @param ring_size 可以设置固定的环形缓冲大小0则自适应
*/
RtmpMediaSource(const string &vhost,
const string &app,
const string &stream_id,
int ring_size = RTMP_GOP_SIZE) :
MediaSource(RTMP_SCHEMA, vhost, app, stream_id), _ring_size(ring_size) {
}
virtual ~RtmpMediaSource() {}
virtual ~RtmpMediaSource() {}
/**
* 获取媒体源的环形缓冲
*/
const RingType::Ptr &getRing() const {
return _ring;
}
/**
* 获取媒体源的环形缓冲
*/
const RingType::Ptr &getRing() const {
return _ring;
}
/**
* 获取播放器个数
* @return
*/
int readerCount() override {
return _ring ? _ring->readerCount() : 0;
}
/**
* 获取播放器个数
* @return
*/
int readerCount() override {
return _ring ? _ring->readerCount() : 0;
}
/**
* 获取metadata
*/
const AMFValue &getMetaData() const {
lock_guard<recursive_mutex> lock(_mtx);
return _metadata;
}
/**
* 获取metadata
*/
const AMFValue &getMetaData() const {
lock_guard<recursive_mutex> lock(_mtx);
return _metadata;
}
/**
* 获取所有的config帧
*/
template<typename FUNC>
void getConfigFrame(const FUNC &f) {
lock_guard<recursive_mutex> lock(_mtx);
for (auto &pr : _config_frame_map) {
f(pr.second);
}
}
/**
* 获取所有的config帧
*/
template<typename FUNC>
void getConfigFrame(const FUNC &f) {
lock_guard<recursive_mutex> lock(_mtx);
for (auto &pr : _config_frame_map) {
f(pr.second);
}
}
/**
* 设置metadata
*/
virtual void setMetaData(const AMFValue &metadata) {
lock_guard<recursive_mutex> lock(_mtx);
_metadata = metadata;
if(_ring){
regist();
}
}
/**
* 设置metadata
*/
virtual void setMetaData(const AMFValue &metadata) {
lock_guard<recursive_mutex> lock(_mtx);
_metadata = metadata;
if(_ring){
regist();
}
}
/**
* 输入rtmp包
* @param pkt rtmp包
* @param key 是否为关键帧
*/
void onWrite(const RtmpPacket::Ptr &pkt, bool key = true) override {
lock_guard<recursive_mutex> lock(_mtx);
if(pkt->typeId == MSG_VIDEO){
//有视频那么启用GOP缓存
/**
* 输入rtmp包
* @param pkt rtmp包
* @param key 是否为关键帧
*/
void onWrite(const RtmpPacket::Ptr &pkt, bool key = true) override {
lock_guard<recursive_mutex> lock(_mtx);
if(pkt->typeId == MSG_VIDEO){
//有视频那么启用GOP缓存
_have_video = true;
}
if (pkt->isCfgFrame()) {
_config_frame_map[pkt->typeId] = pkt;
return;
}
}
if (pkt->isCfgFrame()) {
_config_frame_map[pkt->typeId] = pkt;
return;
}
if (!_ring) {
weak_ptr<RtmpMediaSource> weakSelf = dynamic_pointer_cast<RtmpMediaSource>(shared_from_this());
auto lam = [weakSelf](const EventPoller::Ptr &, int size, bool) {
auto strongSelf = weakSelf.lock();
if (!strongSelf) {
return;
}
strongSelf->onReaderChanged(size);
};
if (!_ring) {
weak_ptr<RtmpMediaSource> weakSelf = dynamic_pointer_cast<RtmpMediaSource>(shared_from_this());
auto lam = [weakSelf](const EventPoller::Ptr &, int size, bool) {
auto strongSelf = weakSelf.lock();
if (!strongSelf) {
return;
}
strongSelf->onReaderChanged(size);
};
//rtmp包缓存最大允许512个如果是纯视频(25fps)大概为20秒数据
//但是这个是GOP缓存的上限值真实的GOP缓存大小等于两个I帧之间的包数的两倍
//而且每次遇到I帧则会清空GOP缓存所以真实的GOP缓存远小于此值
_ring = std::make_shared<RingType>(_ring_size,std::move(lam));
onReaderChanged(0);
//rtmp包缓存最大允许512个如果是纯视频(25fps)大概为20秒数据
//但是这个是GOP缓存的上限值真实的GOP缓存大小等于两个I帧之间的包数的两倍
//而且每次遇到I帧则会清空GOP缓存所以真实的GOP缓存远小于此值
_ring = std::make_shared<RingType>(_ring_size,std::move(lam));
onReaderChanged(0);
if(_metadata){
regist();
}
}
_track_stamps_map[pkt->typeId] = pkt->timeStamp;
//不存在视频为了减少缓存延时那么关闭GOP缓存
_ring->write(pkt, _have_video ? pkt->isVideoKeyFrame() : true);
checkNoneReader();
}
if(_metadata){
regist();
}
}
_track_stamps_map[pkt->typeId] = pkt->timeStamp;
//不存在视频为了减少缓存延时那么关闭GOP缓存
_ring->write(pkt, _have_video ? pkt->isVideoKeyFrame() : true);
checkNoneReader();
}
/**
* 获取当前时间戳
*/
uint32_t getTimeStamp(TrackType trackType) override {
lock_guard<recursive_mutex> lock(_mtx);
switch (trackType) {
case TrackVideo:
return _track_stamps_map[MSG_VIDEO];
case TrackAudio:
return _track_stamps_map[MSG_AUDIO];
default:
return MAX(_track_stamps_map[MSG_VIDEO], _track_stamps_map[MSG_AUDIO]);
}
}
/**
* 获取当前时间戳
*/
uint32_t getTimeStamp(TrackType trackType) override {
lock_guard<recursive_mutex> lock(_mtx);
switch (trackType) {
case TrackVideo:
return _track_stamps_map[MSG_VIDEO];
case TrackAudio:
return _track_stamps_map[MSG_AUDIO];
default:
return MAX(_track_stamps_map[MSG_VIDEO], _track_stamps_map[MSG_AUDIO]);
}
}
private:
/**
* 每次增减消费者都会触发该函数
*/
void onReaderChanged(int size) {
//我们记录最后一次活动时间
_reader_changed_ticker.resetTime();
if (size != 0 || totalReaderCount() != 0) {
//还有消费者正在观看该流
_async_emit_none_reader = false;
return;
}
_async_emit_none_reader = true;
}
/**
* 每次增减消费者都会触发该函数
*/
void onReaderChanged(int size) {
//我们记录最后一次活动时间
_reader_changed_ticker.resetTime();
if (size != 0 || totalReaderCount() != 0) {
//还有消费者正在观看该流
_async_emit_none_reader = false;
return;
}
_async_emit_none_reader = true;
}
/**
* 检查是否无人消费该流,
* 如果无人消费且超过一定时间会触发onNoneReader事件
*/
void checkNoneReader() {
GET_CONFIG(int, stream_none_reader_delay, General::kStreamNoneReaderDelayMS);
if (_async_emit_none_reader && _reader_changed_ticker.elapsedTime() > stream_none_reader_delay) {
_async_emit_none_reader = false;
onNoneReader();
}
}
/**
* 检查是否无人消费该流,
* 如果无人消费且超过一定时间会触发onNoneReader事件
*/
void checkNoneReader() {
GET_CONFIG(int, stream_none_reader_delay, General::kStreamNoneReaderDelayMS);
if (_async_emit_none_reader && _reader_changed_ticker.elapsedTime() > stream_none_reader_delay) {
_async_emit_none_reader = false;
onNoneReader();
}
}
protected:
int _ring_size;
bool _async_emit_none_reader = false;
bool _have_video = false;
mutable recursive_mutex _mtx;
Ticker _reader_changed_ticker;
AMFValue _metadata;
RingBuffer<RtmpPacket::Ptr>::Ptr _ring;
unordered_map<int, uint32_t> _track_stamps_map;
unordered_map<int, RtmpPacket::Ptr> _config_frame_map;
int _ring_size;
bool _async_emit_none_reader = false;
bool _have_video = false;
mutable recursive_mutex _mtx;
Ticker _reader_changed_ticker;
AMFValue _metadata;
RingBuffer<RtmpPacket::Ptr>::Ptr _ring;
unordered_map<int, uint32_t> _track_stamps_map;
unordered_map<int, RtmpPacket::Ptr> _config_frame_map;
};
} /* namespace mediakit */

View File

@@ -45,46 +45,46 @@ using namespace toolkit;
namespace mediakit {
class RtmpMediaSourceImp: public RtmpMediaSource, public Demuxer::Listener , public MultiMediaSourceMuxer::Listener {
public:
typedef std::shared_ptr<RtmpMediaSourceImp> Ptr;
typedef std::shared_ptr<RtmpMediaSourceImp> Ptr;
/**
* 构造函数
* @param vhost 虚拟主机
* @param app 应用名
* @param id 流id
* @param ringSize 环形缓存大小
*/
RtmpMediaSourceImp(const string &vhost, const string &app, const string &id, int ringSize = RTMP_GOP_SIZE) : RtmpMediaSource(vhost, app, id, ringSize) {
_demuxer = std::make_shared<RtmpDemuxer>();
_demuxer->setTrackListener(this);
}
/**
* 构造函数
* @param vhost 虚拟主机
* @param app 应用名
* @param id 流id
* @param ringSize 环形缓存大小
*/
RtmpMediaSourceImp(const string &vhost, const string &app, const string &id, int ringSize = RTMP_GOP_SIZE) : RtmpMediaSource(vhost, app, id, ringSize) {
_demuxer = std::make_shared<RtmpDemuxer>();
_demuxer->setTrackListener(this);
}
~RtmpMediaSourceImp() = default;
~RtmpMediaSourceImp() = default;
/**
* 设置metadata
*/
void setMetaData(const AMFValue &metadata) override{
_demuxer->loadMetaData(metadata);
RtmpMediaSource::setMetaData(metadata);
}
/**
* 设置metadata
*/
void setMetaData(const AMFValue &metadata) override{
_demuxer->loadMetaData(metadata);
RtmpMediaSource::setMetaData(metadata);
}
/**
* 输入rtmp并解析
*/
void onWrite(const RtmpPacket::Ptr &pkt,bool key_pos = true) override {
key_pos = _demuxer->inputRtmp(pkt);
RtmpMediaSource::onWrite(pkt,key_pos);
}
/**
* 输入rtmp并解析
*/
void onWrite(const RtmpPacket::Ptr &pkt,bool key_pos = true) override {
key_pos = _demuxer->inputRtmp(pkt);
RtmpMediaSource::onWrite(pkt,key_pos);
}
/**
* 设置监听器
* @param listener
*/
void setListener(const std::weak_ptr<MediaSourceEvent> &listener) override {
/**
* 设置监听器
* @param listener
*/
void setListener(const std::weak_ptr<MediaSourceEvent> &listener) override {
RtmpMediaSource::setListener(listener);
if(_muxer){
_muxer->setListener(listener);
_muxer->setListener(listener);
}
}
@@ -95,42 +95,42 @@ public:
return readerCount() + (_muxer ? _muxer->totalReaderCount() : 0);
}
/**
* 设置协议转换
* @param enableRtsp 是否转换成rtsp
* @param enableHls 是否转换成hls
* @param enableMP4 是否mp4录制
*/
void setProtocolTranslation(bool enableRtsp, bool enableHls, bool enableMP4) {
//不重复生成rtmp
_muxer = std::make_shared<MultiMediaSourceMuxer>(getVhost(), getApp(), getId(), _demuxer->getDuration(), enableRtsp, false, enableHls, enableMP4);
_muxer->setListener(getListener());
_muxer->setTrackListener(this);
/**
* 设置协议转换
* @param enableRtsp 是否转换成rtsp
* @param enableHls 是否转换成hls
* @param enableMP4 是否mp4录制
*/
void setProtocolTranslation(bool enableRtsp, bool enableHls, bool enableMP4) {
//不重复生成rtmp
_muxer = std::make_shared<MultiMediaSourceMuxer>(getVhost(), getApp(), getId(), _demuxer->getDuration(), enableRtsp, false, enableHls, enableMP4);
_muxer->setListener(getListener());
_muxer->setTrackListener(this);
for(auto &track : _demuxer->getTracks(false)){
_muxer->addTrack(track);
track->addDelegate(_muxer);
}
}
}
/**
* _demuxer触发的添加Track事件
*/
void onAddTrack(const Track::Ptr &track) override {
if(_muxer){
_muxer->addTrack(track);
track->addDelegate(_muxer);
}
}
/**
* _demuxer触发的添加Track事件
*/
void onAddTrack(const Track::Ptr &track) override {
if(_muxer){
_muxer->addTrack(track);
track->addDelegate(_muxer);
}
}
/**
* _muxer触发的所有Track就绪的事件
*/
void onAllTrackReady() override{
setTrackSource(_muxer);
}
/**
* _muxer触发的所有Track就绪的事件
*/
void onAllTrackReady() override{
setTrackSource(_muxer);
}
private:
RtmpDemuxer::Ptr _demuxer;
MultiMediaSourceMuxer::Ptr _muxer;
RtmpDemuxer::Ptr _demuxer;
MultiMediaSourceMuxer::Ptr _muxer;
};
} /* namespace mediakit */

View File

@@ -38,35 +38,35 @@ RtmpPlayer::RtmpPlayer(const EventPoller::Ptr &poller) : TcpClient(poller) {
}
RtmpPlayer::~RtmpPlayer() {
DebugL << endl;
DebugL << endl;
}
void RtmpPlayer::teardown() {
if (alive()) {
shutdown(SockException(Err_shutdown,"teardown"));
}
_strApp.clear();
_strStream.clear();
_strTcUrl.clear();
_pBeatTimer.reset();
_pPlayTimer.reset();
_pMediaTimer.reset();
_iSeekTo = 0;
RtmpProtocol::reset();
if (alive()) {
shutdown(SockException(Err_shutdown,"teardown"));
}
_strApp.clear();
_strStream.clear();
_strTcUrl.clear();
_pBeatTimer.reset();
_pPlayTimer.reset();
_pMediaTimer.reset();
_iSeekTo = 0;
RtmpProtocol::reset();
CLEAR_ARR(_aiFistStamp);
CLEAR_ARR(_aiNowStamp);
CLEAR_ARR(_aiFistStamp);
CLEAR_ARR(_aiNowStamp);
lock_guard<recursive_mutex> lck(_mtxOnResultCB);
_mapOnResultCB.clear();
lock_guard<recursive_mutex> lck2(_mtxOnStatusCB);
_dqOnStatusCB.clear();
lock_guard<recursive_mutex> lck(_mtxOnResultCB);
_mapOnResultCB.clear();
lock_guard<recursive_mutex> lck2(_mtxOnStatusCB);
_dqOnStatusCB.clear();
}
void RtmpPlayer::play(const string &strUrl) {
teardown();
string strHost = FindField(strUrl.data(), "://", "/");
_strApp = FindField(strUrl.data(), (strHost + "/").data(), "/");
teardown();
string strHost = FindField(strUrl.data(), "://", "/");
_strApp = FindField(strUrl.data(), (strHost + "/").data(), "/");
_strStream = FindField(strUrl.data(), (strHost + "/" + _strApp + "/").data(), NULL);
_strTcUrl = string("rtmp://") + strHost + "/" + _strApp;
@@ -74,48 +74,48 @@ void RtmpPlayer::play(const string &strUrl) {
onPlayResult_l(SockException(Err_other,"rtmp url非法"),false);
return;
}
DebugL << strHost << " " << _strApp << " " << _strStream;
DebugL << strHost << " " << _strApp << " " << _strStream;
auto iPort = atoi(FindField(strHost.data(), ":", NULL).data());
if (iPort <= 0) {
auto iPort = atoi(FindField(strHost.data(), ":", NULL).data());
if (iPort <= 0) {
//rtmp 默认端口1935
iPort = 1935;
} else {
iPort = 1935;
} else {
//服务器域名
strHost = FindField(strHost.data(), NULL, ":");
}
if(!(*this)[kNetAdapter].empty()){
setNetAdapter((*this)[kNetAdapter]);
}
strHost = FindField(strHost.data(), NULL, ":");
}
if(!(*this)[kNetAdapter].empty()){
setNetAdapter((*this)[kNetAdapter]);
}
weak_ptr<RtmpPlayer> weakSelf= dynamic_pointer_cast<RtmpPlayer>(shared_from_this());
float playTimeOutSec = (*this)[kTimeoutMS].as<int>() / 1000.0;
_pPlayTimer.reset( new Timer(playTimeOutSec, [weakSelf]() {
auto strongSelf=weakSelf.lock();
if(!strongSelf) {
return false;
}
strongSelf->onPlayResult_l(SockException(Err_timeout,"play rtmp timeout"),false);
return false;
},getPoller()));
weak_ptr<RtmpPlayer> weakSelf= dynamic_pointer_cast<RtmpPlayer>(shared_from_this());
float playTimeOutSec = (*this)[kTimeoutMS].as<int>() / 1000.0;
_pPlayTimer.reset( new Timer(playTimeOutSec, [weakSelf]() {
auto strongSelf=weakSelf.lock();
if(!strongSelf) {
return false;
}
strongSelf->onPlayResult_l(SockException(Err_timeout,"play rtmp timeout"),false);
return false;
},getPoller()));
_metadata_got = false;
startConnect(strHost, iPort , playTimeOutSec);
_metadata_got = false;
startConnect(strHost, iPort , playTimeOutSec);
}
void RtmpPlayer::onErr(const SockException &ex){
//定时器_pPlayTimer为空后表明握手结束了
onPlayResult_l(ex, !_pPlayTimer);
onPlayResult_l(ex, !_pPlayTimer);
}
void RtmpPlayer::onPlayResult_l(const SockException &ex , bool handshakeCompleted) {
WarnL << ex.getErrCode() << " " << ex.what();
WarnL << ex.getErrCode() << " " << ex.what();
if(!ex){
//播放成功恢复rtmp接收超时定时器
_mediaTicker.resetTime();
weak_ptr<RtmpPlayer> weakSelf = dynamic_pointer_cast<RtmpPlayer>(shared_from_this());
int timeoutMS = (*this)[kMediaTimeoutMS].as<int>();
//创建rtmp数据接收超时检测定时器
//创建rtmp数据接收超时检测定时器
_pMediaTimer.reset( new Timer(timeoutMS / 2000.0, [weakSelf,timeoutMS]() {
auto strongSelf=weakSelf.lock();
if(!strongSelf) {
@@ -130,110 +130,110 @@ void RtmpPlayer::onPlayResult_l(const SockException &ex , bool handshakeComplete
},getPoller()));
}
if (!handshakeCompleted) {
//开始播放阶段
_pPlayTimer.reset();
onPlayResult(ex);
} else if (ex) {
//播放成功后异常断开回调
onShutdown(ex);
} else {
//恢复播放
onResume();
}
if (!handshakeCompleted) {
//开始播放阶段
_pPlayTimer.reset();
onPlayResult(ex);
} else if (ex) {
//播放成功后异常断开回调
onShutdown(ex);
} else {
//恢复播放
onResume();
}
if(ex){
teardown();
}
if(ex){
teardown();
}
}
void RtmpPlayer::onConnect(const SockException &err){
if(err.getErrCode() != Err_success) {
onPlayResult_l(err, false);
return;
}
weak_ptr<RtmpPlayer> weakSelf= dynamic_pointer_cast<RtmpPlayer>(shared_from_this());
startClientSession([weakSelf](){
if(err.getErrCode() != Err_success) {
onPlayResult_l(err, false);
return;
}
weak_ptr<RtmpPlayer> weakSelf= dynamic_pointer_cast<RtmpPlayer>(shared_from_this());
startClientSession([weakSelf](){
auto strongSelf=weakSelf.lock();
if(!strongSelf) {
if(!strongSelf) {
return;
}
strongSelf->send_connect();
});
strongSelf->send_connect();
});
}
void RtmpPlayer::onRecv(const Buffer::Ptr &pBuf){
try {
onParseRtmp(pBuf->data(), pBuf->size());
} catch (exception &e) {
SockException ex(Err_other, e.what());
try {
onParseRtmp(pBuf->data(), pBuf->size());
} catch (exception &e) {
SockException ex(Err_other, e.what());
//定时器_pPlayTimer为空后表明握手结束了
onPlayResult_l(ex, !_pPlayTimer);
}
onPlayResult_l(ex, !_pPlayTimer);
}
}
void RtmpPlayer::pause(bool bPause) {
send_pause(bPause);
send_pause(bPause);
}
inline void RtmpPlayer::send_connect() {
AMFValue obj(AMF_OBJECT);
obj.set("app", _strApp);
obj.set("tcUrl", _strTcUrl);
//未使用代理
obj.set("fpad", false);
//参考librtmp,什么作用?
obj.set("capabilities", 15);
//SUPPORT_VID_CLIENT_SEEK 支持seek
obj.set("videoFunction", 1);
AMFValue obj(AMF_OBJECT);
obj.set("app", _strApp);
obj.set("tcUrl", _strTcUrl);
//未使用代理
obj.set("fpad", false);
//参考librtmp,什么作用?
obj.set("capabilities", 15);
//SUPPORT_VID_CLIENT_SEEK 支持seek
obj.set("videoFunction", 1);
//只支持aac
obj.set("audioCodecs", (double)(0x0400));
//只支持H264
obj.set("videoCodecs", (double)(0x0080));
sendInvoke("connect", obj);
addOnResultCB([this](AMFDecoder &dec){
//TraceL << "connect result";
dec.load<AMFValue>();
auto val = dec.load<AMFValue>();
auto level = val["level"].as_string();
auto code = val["code"].as_string();
if(level != "status"){
throw std::runtime_error(StrPrinter <<"connect 失败:" << level << " " << code << endl);
}
send_createStream();
});
sendInvoke("connect", obj);
addOnResultCB([this](AMFDecoder &dec){
//TraceL << "connect result";
dec.load<AMFValue>();
auto val = dec.load<AMFValue>();
auto level = val["level"].as_string();
auto code = val["code"].as_string();
if(level != "status"){
throw std::runtime_error(StrPrinter <<"connect 失败:" << level << " " << code << endl);
}
send_createStream();
});
}
inline void RtmpPlayer::send_createStream() {
AMFValue obj(AMF_NULL);
sendInvoke("createStream", obj);
addOnResultCB([this](AMFDecoder &dec){
//TraceL << "createStream result";
dec.load<AMFValue>();
_ui32StreamId = dec.load<int>();
send_play();
});
AMFValue obj(AMF_NULL);
sendInvoke("createStream", obj);
addOnResultCB([this](AMFDecoder &dec){
//TraceL << "createStream result";
dec.load<AMFValue>();
_ui32StreamId = dec.load<int>();
send_play();
});
}
inline void RtmpPlayer::send_play() {
AMFEncoder enc;
enc << "play" << ++_iReqID << nullptr << _strStream << (double)_ui32StreamId;
sendRequest(MSG_CMD, enc.data());
auto fun = [this](AMFValue &val){
//TraceL << "play onStatus";
auto level = val["level"].as_string();
auto code = val["code"].as_string();
if(level != "status"){
throw std::runtime_error(StrPrinter <<"play 失败:" << level << " " << code << endl);
}
};
addOnStatusCB(fun);
addOnStatusCB(fun);
AMFEncoder enc;
enc << "play" << ++_iReqID << nullptr << _strStream << (double)_ui32StreamId;
sendRequest(MSG_CMD, enc.data());
auto fun = [this](AMFValue &val){
//TraceL << "play onStatus";
auto level = val["level"].as_string();
auto code = val["code"].as_string();
if(level != "status"){
throw std::runtime_error(StrPrinter <<"play 失败:" << level << " " << code << endl);
}
};
addOnStatusCB(fun);
addOnStatusCB(fun);
}
inline void RtmpPlayer::send_pause(bool bPause) {
AMFEncoder enc;
enc << "pause" << ++_iReqID << nullptr << bPause;
sendRequest(MSG_CMD, enc.data());
auto fun = [this,bPause](AMFValue &val){
AMFEncoder enc;
enc << "pause" << ++_iReqID << nullptr << bPause;
sendRequest(MSG_CMD, enc.data());
auto fun = [this,bPause](AMFValue &val){
//TraceL << "pause onStatus";
auto level = val["level"].as_string();
auto code = val["code"].as_string();
@@ -250,147 +250,147 @@ inline void RtmpPlayer::send_pause(bool bPause) {
_pMediaTimer.reset();
}
}
};
addOnStatusCB(fun);
};
addOnStatusCB(fun);
_pBeatTimer.reset();
if(bPause){
weak_ptr<RtmpPlayer> weakSelf = dynamic_pointer_cast<RtmpPlayer>(shared_from_this());
_pBeatTimer.reset(new Timer((*this)[kBeatIntervalMS].as<int>() / 1000.0,[weakSelf](){
auto strongSelf = weakSelf.lock();
if (!strongSelf){
return false;
}
uint32_t timeStamp = ::time(NULL);
strongSelf->sendUserControl(CONTROL_PING_REQUEST, timeStamp);
return true;
},getPoller()));
}
_pBeatTimer.reset();
if(bPause){
weak_ptr<RtmpPlayer> weakSelf = dynamic_pointer_cast<RtmpPlayer>(shared_from_this());
_pBeatTimer.reset(new Timer((*this)[kBeatIntervalMS].as<int>() / 1000.0,[weakSelf](){
auto strongSelf = weakSelf.lock();
if (!strongSelf){
return false;
}
uint32_t timeStamp = ::time(NULL);
strongSelf->sendUserControl(CONTROL_PING_REQUEST, timeStamp);
return true;
},getPoller()));
}
}
void RtmpPlayer::onCmd_result(AMFDecoder &dec){
auto iReqId = dec.load<int>();
lock_guard<recursive_mutex> lck(_mtxOnResultCB);
auto it = _mapOnResultCB.find(iReqId);
if(it != _mapOnResultCB.end()){
it->second(dec);
_mapOnResultCB.erase(it);
}else{
WarnL << "unhandled _result";
}
auto iReqId = dec.load<int>();
lock_guard<recursive_mutex> lck(_mtxOnResultCB);
auto it = _mapOnResultCB.find(iReqId);
if(it != _mapOnResultCB.end()){
it->second(dec);
_mapOnResultCB.erase(it);
}else{
WarnL << "unhandled _result";
}
}
void RtmpPlayer::onCmd_onStatus(AMFDecoder &dec) {
AMFValue val;
while(true){
val = dec.load<AMFValue>();
if(val.type() == AMF_OBJECT){
break;
}
}
if(val.type() != AMF_OBJECT){
throw std::runtime_error("onStatus:the result object was not found");
}
AMFValue val;
while(true){
val = dec.load<AMFValue>();
if(val.type() == AMF_OBJECT){
break;
}
}
if(val.type() != AMF_OBJECT){
throw std::runtime_error("onStatus:the result object was not found");
}
lock_guard<recursive_mutex> lck(_mtxOnStatusCB);
if(_dqOnStatusCB.size()){
_dqOnStatusCB.front()(val);
_dqOnStatusCB.pop_front();
}else{
auto level = val["level"];
auto code = val["code"].as_string();
if(level.type() == AMF_STRING){
if(level.as_string() != "status"){
throw std::runtime_error(StrPrinter <<"onStatus 失败:" << level.as_string() << " " << code << endl);
}
}
//WarnL << "unhandled onStatus:" << code;
if(_dqOnStatusCB.size()){
_dqOnStatusCB.front()(val);
_dqOnStatusCB.pop_front();
}else{
auto level = val["level"];
auto code = val["code"].as_string();
if(level.type() == AMF_STRING){
if(level.as_string() != "status"){
throw std::runtime_error(StrPrinter <<"onStatus 失败:" << level.as_string() << " " << code << endl);
}
}
//WarnL << "unhandled onStatus:" << code;
}
}
void RtmpPlayer::onCmd_onMetaData(AMFDecoder &dec) {
//TraceL;
auto val = dec.load<AMFValue>();
if(!onCheckMeta(val)){
throw std::runtime_error("onCheckMeta failed");
}
_metadata_got = true;
//TraceL;
auto val = dec.load<AMFValue>();
if(!onCheckMeta(val)){
throw std::runtime_error("onCheckMeta failed");
}
_metadata_got = true;
}
void RtmpPlayer::onStreamDry(uint32_t ui32StreamId) {
//TraceL << ui32StreamId;
onPlayResult_l(SockException(Err_other,"rtmp stream dry"), true);
//TraceL << ui32StreamId;
onPlayResult_l(SockException(Err_other,"rtmp stream dry"), true);
}
void RtmpPlayer::onMediaData_l(const RtmpPacket::Ptr &packet) {
_mediaTicker.resetTime();
if(!_pPlayTimer){
//已经触发了onPlayResult事件直接触发onMediaData事件
onMediaData(packet);
return;
}
_mediaTicker.resetTime();
if(!_pPlayTimer){
//已经触发了onPlayResult事件直接触发onMediaData事件
onMediaData(packet);
return;
}
if(packet->isCfgFrame()){
//输入配置帧以便初始化完成各个track
onMediaData(packet);
}else{
//先触发onPlayResult事件这个时候解码器才能初始化完毕
onPlayResult_l(SockException(Err_success,"play rtmp success"), false);
//触发onPlayResult事件后再把帧数据输入到解码器
onMediaData(packet);
}
if(packet->isCfgFrame()){
//输入配置帧以便初始化完成各个track
onMediaData(packet);
}else{
//先触发onPlayResult事件这个时候解码器才能初始化完毕
onPlayResult_l(SockException(Err_success,"play rtmp success"), false);
//触发onPlayResult事件后再把帧数据输入到解码器
onMediaData(packet);
}
}
void RtmpPlayer::onRtmpChunk(RtmpPacket &chunkData) {
typedef void (RtmpPlayer::*rtmp_func_ptr)(AMFDecoder &dec);
static unordered_map<string, rtmp_func_ptr> s_func_map;
static onceToken token([]() {
s_func_map.emplace("_error",&RtmpPlayer::onCmd_result);
s_func_map.emplace("_result",&RtmpPlayer::onCmd_result);
s_func_map.emplace("onStatus",&RtmpPlayer::onCmd_onStatus);
s_func_map.emplace("onMetaData",&RtmpPlayer::onCmd_onMetaData);
}, []() {});
typedef void (RtmpPlayer::*rtmp_func_ptr)(AMFDecoder &dec);
static unordered_map<string, rtmp_func_ptr> s_func_map;
static onceToken token([]() {
s_func_map.emplace("_error",&RtmpPlayer::onCmd_result);
s_func_map.emplace("_result",&RtmpPlayer::onCmd_result);
s_func_map.emplace("onStatus",&RtmpPlayer::onCmd_onStatus);
s_func_map.emplace("onMetaData",&RtmpPlayer::onCmd_onMetaData);
}, []() {});
switch (chunkData.typeId) {
case MSG_CMD:
case MSG_CMD3:
case MSG_DATA:
case MSG_DATA3: {
AMFDecoder dec(chunkData.strBuf, 0);
std::string type = dec.load<std::string>();
auto it = s_func_map.find(type);
if(it != s_func_map.end()){
auto fun = it->second;
(this->*fun)(dec);
}else{
WarnL << "can not support cmd:" << type;
}
}
break;
case MSG_AUDIO:
case MSG_VIDEO: {
switch (chunkData.typeId) {
case MSG_CMD:
case MSG_CMD3:
case MSG_DATA:
case MSG_DATA3: {
AMFDecoder dec(chunkData.strBuf, 0);
std::string type = dec.load<std::string>();
auto it = s_func_map.find(type);
if(it != s_func_map.end()){
auto fun = it->second;
(this->*fun)(dec);
}else{
WarnL << "can not support cmd:" << type;
}
}
break;
case MSG_AUDIO:
case MSG_VIDEO: {
auto idx = chunkData.typeId%2;
if (_aNowStampTicker[idx].elapsedTime() > 500) {
//计算播放进度时间轴用
//计算播放进度时间轴用
_aiNowStamp[idx] = chunkData.timeStamp;
}
if(!_metadata_got){
if(!onCheckMeta(TitleMeta().getMetadata())){
throw std::runtime_error("onCheckMeta failed");
}
_metadata_got = true;
}
onMediaData_l(std::make_shared<RtmpPacket>(std::move(chunkData)));
}
break;
default:
//WarnL << "unhandled message:" << (int) chunkData.typeId << hexdump(chunkData.strBuf.data(), chunkData.strBuf.size());
break;
}
if(!_metadata_got){
if(!onCheckMeta(TitleMeta().getMetadata())){
throw std::runtime_error("onCheckMeta failed");
}
_metadata_got = true;
}
onMediaData_l(std::make_shared<RtmpPacket>(std::move(chunkData)));
}
break;
default:
//WarnL << "unhandled message:" << (int) chunkData.typeId << hexdump(chunkData.strBuf.data(), chunkData.strBuf.size());
break;
}
}
uint32_t RtmpPlayer::getProgressMilliSecond() const{
uint32_t iTime[2] = {0,0};
uint32_t iTime[2] = {0,0};
for(auto i = 0 ;i < 2 ;i++){
iTime[i] = _aiNowStamp[i] - _aiFistStamp[i];
}
@@ -407,7 +407,7 @@ void RtmpPlayer::seekToMilliSecond(uint32_t seekMS){
//TraceL << "seek result";
_aNowStampTicker[0].resetTime();
_aNowStampTicker[1].resetTime();
int iTimeInc = seekMS - getProgressMilliSecond();
int iTimeInc = seekMS - getProgressMilliSecond();
for(auto i = 0 ;i < 2 ;i++){
_aiFistStamp[i] = _aiNowStamp[i] + iTimeInc;
_aiNowStamp[i] = _aiFistStamp[i];

View File

@@ -47,77 +47,77 @@ namespace mediakit {
//实现了rtmp播放器协议部分的功能及数据接收功能
class RtmpPlayer:public PlayerBase, public TcpClient, public RtmpProtocol{
public:
typedef std::shared_ptr<RtmpPlayer> Ptr;
RtmpPlayer(const EventPoller::Ptr &poller);
virtual ~RtmpPlayer();
typedef std::shared_ptr<RtmpPlayer> Ptr;
RtmpPlayer(const EventPoller::Ptr &poller);
virtual ~RtmpPlayer();
void play(const string &strUrl) override;
void pause(bool bPause) override;
void teardown() override;
void play(const string &strUrl) override;
void pause(bool bPause) override;
void teardown() override;
protected:
virtual bool onCheckMeta(const AMFValue &val) =0;
virtual void onMediaData(const RtmpPacket::Ptr &chunkData) =0;
uint32_t getProgressMilliSecond() const;
void seekToMilliSecond(uint32_t ms);
virtual bool onCheckMeta(const AMFValue &val) =0;
virtual void onMediaData(const RtmpPacket::Ptr &chunkData) =0;
uint32_t getProgressMilliSecond() const;
void seekToMilliSecond(uint32_t ms);
protected:
void onMediaData_l(const RtmpPacket::Ptr &chunkData);
//在获取config帧后才触发onPlayResult_l(而不是收到play命令回复)所以此时所有track都初始化完毕了
void onPlayResult_l(const SockException &ex, bool handshakeCompleted);
void onMediaData_l(const RtmpPacket::Ptr &chunkData);
//在获取config帧后才触发onPlayResult_l(而不是收到play命令回复)所以此时所有track都初始化完毕了
void onPlayResult_l(const SockException &ex, bool handshakeCompleted);
//form Tcpclient
void onRecv(const Buffer::Ptr &pBuf) override;
void onConnect(const SockException &err) override;
void onErr(const SockException &ex) override;
//from RtmpProtocol
void onRtmpChunk(RtmpPacket &chunkData) override;
void onStreamDry(uint32_t ui32StreamId) override;
//form Tcpclient
void onRecv(const Buffer::Ptr &pBuf) override;
void onConnect(const SockException &err) override;
void onErr(const SockException &ex) override;
//from RtmpProtocol
void onRtmpChunk(RtmpPacket &chunkData) override;
void onStreamDry(uint32_t ui32StreamId) override;
void onSendRawData(const Buffer::Ptr &buffer) override{
send(buffer);
}
template<typename FUN>
inline void addOnResultCB(const FUN &fun) {
lock_guard<recursive_mutex> lck(_mtxOnResultCB);
_mapOnResultCB.emplace(_iReqID, fun);
}
template<typename FUN>
inline void addOnStatusCB(const FUN &fun) {
lock_guard<recursive_mutex> lck(_mtxOnStatusCB);
_dqOnStatusCB.emplace_back(fun);
}
template<typename FUN>
inline void addOnResultCB(const FUN &fun) {
lock_guard<recursive_mutex> lck(_mtxOnResultCB);
_mapOnResultCB.emplace(_iReqID, fun);
}
template<typename FUN>
inline void addOnStatusCB(const FUN &fun) {
lock_guard<recursive_mutex> lck(_mtxOnStatusCB);
_dqOnStatusCB.emplace_back(fun);
}
void onCmd_result(AMFDecoder &dec);
void onCmd_onStatus(AMFDecoder &dec);
void onCmd_onMetaData(AMFDecoder &dec);
void onCmd_result(AMFDecoder &dec);
void onCmd_onStatus(AMFDecoder &dec);
void onCmd_onMetaData(AMFDecoder &dec);
inline void send_connect();
inline void send_createStream();
inline void send_play();
inline void send_pause(bool bPause);
inline void send_connect();
inline void send_createStream();
inline void send_play();
inline void send_pause(bool bPause);
private:
string _strApp;
string _strStream;
string _strTcUrl;
bool _bPaused = false;
string _strApp;
string _strStream;
string _strTcUrl;
bool _bPaused = false;
unordered_map<int, function<void(AMFDecoder &dec)> > _mapOnResultCB;
recursive_mutex _mtxOnResultCB;
deque<function<void(AMFValue &dec)> > _dqOnStatusCB;
recursive_mutex _mtxOnStatusCB;
unordered_map<int, function<void(AMFDecoder &dec)> > _mapOnResultCB;
recursive_mutex _mtxOnResultCB;
deque<function<void(AMFValue &dec)> > _dqOnStatusCB;
recursive_mutex _mtxOnStatusCB;
//超时功能实现
Ticker _mediaTicker;
std::shared_ptr<Timer> _pMediaTimer;
std::shared_ptr<Timer> _pPlayTimer;
//心跳定时器
std::shared_ptr<Timer> _pBeatTimer;
//超时功能实现
Ticker _mediaTicker;
std::shared_ptr<Timer> _pMediaTimer;
std::shared_ptr<Timer> _pPlayTimer;
//心跳定时器
std::shared_ptr<Timer> _pBeatTimer;
//播放进度控制
uint32_t _iSeekTo = 0;
uint32_t _aiFistStamp[2] = { 0, 0 };
uint32_t _aiNowStamp[2] = { 0, 0 };
Ticker _aNowStampTicker[2];
bool _metadata_got = false;
//播放进度控制
uint32_t _iSeekTo = 0;
uint32_t _aiFistStamp[2] = { 0, 0 };
uint32_t _aiNowStamp[2] = { 0, 0 };
Ticker _aNowStampTicker[2];
bool _metadata_got = false;
};
} /* namespace mediakit */

View File

@@ -73,7 +73,7 @@ private:
return true;
}
void onMediaData(const RtmpPacket::Ptr &chunkData) override {
if(_pRtmpMediaSrc){
if(_pRtmpMediaSrc){
if(!_set_meta_data && !chunkData->isCfgFrame()){
_set_meta_data = true;
_pRtmpMediaSrc->setMetaData(TitleMeta().getMetadata());
@@ -81,7 +81,7 @@ private:
_pRtmpMediaSrc->onWrite(chunkData);
}
if(!_delegate){
//这个流没有metadata
//这个流没有metadata
_delegate.reset(new RtmpDemuxer());
}
_delegate->inputRtmp(chunkData);

File diff suppressed because it is too large Load Diff

View File

@@ -45,77 +45,77 @@ namespace mediakit {
class RtmpProtocol {
public:
RtmpProtocol();
virtual ~RtmpProtocol();
//作为客户端发送c0c1等待s0s1s2并且回调
void startClientSession(const function<void()> &cb);
void onParseRtmp(const char *pcRawData,int iSize);
void reset();
RtmpProtocol();
virtual ~RtmpProtocol();
//作为客户端发送c0c1等待s0s1s2并且回调
void startClientSession(const function<void()> &cb);
void onParseRtmp(const char *pcRawData,int iSize);
void reset();
protected:
virtual void onSendRawData(const Buffer::Ptr &buffer) = 0;
virtual void onRtmpChunk(RtmpPacket &chunkData) = 0;
virtual void onStreamBegin(uint32_t ui32StreamId){
_ui32StreamId = ui32StreamId;
}
virtual void onStreamEof(uint32_t ui32StreamId){};
virtual void onStreamDry(uint32_t ui32StreamId){};
virtual void onSendRawData(const Buffer::Ptr &buffer) = 0;
virtual void onRtmpChunk(RtmpPacket &chunkData) = 0;
virtual void onStreamBegin(uint32_t ui32StreamId){
_ui32StreamId = ui32StreamId;
}
virtual void onStreamEof(uint32_t ui32StreamId){};
virtual void onStreamDry(uint32_t ui32StreamId){};
protected:
void sendAcknowledgement(uint32_t ui32Size);
void sendAcknowledgementSize(uint32_t ui32Size);
void sendPeerBandwidth(uint32_t ui32Size);
void sendChunkSize(uint32_t ui32Size);
void sendPingRequest(uint32_t ui32TimeStamp = ::time(NULL));
void sendPingResponse(uint32_t ui32TimeStamp = ::time(NULL));
void sendSetBufferLength(uint32_t ui32StreamId, uint32_t ui32Length);
void sendUserControl(uint16_t ui16EventType, uint32_t ui32EventData);
void sendUserControl(uint16_t ui16EventType, const string &strEventData);
void sendAcknowledgement(uint32_t ui32Size);
void sendAcknowledgementSize(uint32_t ui32Size);
void sendPeerBandwidth(uint32_t ui32Size);
void sendChunkSize(uint32_t ui32Size);
void sendPingRequest(uint32_t ui32TimeStamp = ::time(NULL));
void sendPingResponse(uint32_t ui32TimeStamp = ::time(NULL));
void sendSetBufferLength(uint32_t ui32StreamId, uint32_t ui32Length);
void sendUserControl(uint16_t ui16EventType, uint32_t ui32EventData);
void sendUserControl(uint16_t ui16EventType, const string &strEventData);
void sendInvoke(const string &strCmd, const AMFValue &val);
void sendRequest(int iCmd, const string &str);
void sendResponse(int iType, const string &str);
void sendRtmp(uint8_t ui8Type, uint32_t ui32StreamId, const std::string &strBuf, uint32_t ui32TimeStamp, int iChunkID);
void sendRtmp(uint8_t ui8Type, uint32_t ui32StreamId, const Buffer::Ptr &buffer, uint32_t ui32TimeStamp, int iChunkID);
void sendInvoke(const string &strCmd, const AMFValue &val);
void sendRequest(int iCmd, const string &str);
void sendResponse(int iType, const string &str);
void sendRtmp(uint8_t ui8Type, uint32_t ui32StreamId, const std::string &strBuf, uint32_t ui32TimeStamp, int iChunkID);
void sendRtmp(uint8_t ui8Type, uint32_t ui32StreamId, const Buffer::Ptr &buffer, uint32_t ui32TimeStamp, int iChunkID);
protected:
int _iReqID = 0;
uint32_t _ui32StreamId = STREAM_CONTROL;
int _iNowStreamID = 0;
int _iNowChunkID = 0;
bool _bDataStarted = false;
inline BufferRaw::Ptr obtainBuffer();
inline BufferRaw::Ptr obtainBuffer(const void *data, int len);
//ResourcePool<BufferRaw,MAX_SEND_PKT> _bufferPool;
int _iReqID = 0;
uint32_t _ui32StreamId = STREAM_CONTROL;
int _iNowStreamID = 0;
int _iNowChunkID = 0;
bool _bDataStarted = false;
inline BufferRaw::Ptr obtainBuffer();
inline BufferRaw::Ptr obtainBuffer(const void *data, int len);
//ResourcePool<BufferRaw,MAX_SEND_PKT> _bufferPool;
private:
void handle_S0S1S2(const function<void()> &cb);
void handle_C0C1();
void handle_C1_simple();
void handle_S0S1S2(const function<void()> &cb);
void handle_C0C1();
void handle_C1_simple();
#ifdef ENABLE_OPENSSL
void handle_C1_complex();
string get_C1_digest(const uint8_t *ptr,char **digestPos);
string get_C1_key(const uint8_t *ptr);
void check_C1_Digest(const string &digest,const string &data);
void send_complex_S0S1S2(int schemeType,const string &digest);
void handle_C1_complex();
string get_C1_digest(const uint8_t *ptr,char **digestPos);
string get_C1_key(const uint8_t *ptr);
void check_C1_Digest(const string &digest,const string &data);
void send_complex_S0S1S2(int schemeType,const string &digest);
#endif //ENABLE_OPENSSL
void handle_C2();
void handle_rtmp();
void handle_rtmpChunk(RtmpPacket &chunkData);
void handle_C2();
void handle_rtmp();
void handle_rtmpChunk(RtmpPacket &chunkData);
private:
////////////ChunkSize////////////
size_t _iChunkLenIn = DEFAULT_CHUNK_LEN;
size_t _iChunkLenOut = DEFAULT_CHUNK_LEN;
////////////Acknowledgement////////////
uint32_t _ui32ByteSent = 0;
uint32_t _ui32LastSent = 0;
uint32_t _ui32WinSize = 0;
///////////PeerBandwidth///////////
uint32_t _ui32Bandwidth = 2500000;
uint8_t _ui8LimitType = 2;
////////////Chunk////////////
unordered_map<int, RtmpPacket> _mapChunkData;
//////////Rtmp parser//////////
string _strRcvBuf;
function<void()> _nextHandle;
////////////ChunkSize////////////
size_t _iChunkLenIn = DEFAULT_CHUNK_LEN;
size_t _iChunkLenOut = DEFAULT_CHUNK_LEN;
////////////Acknowledgement////////////
uint32_t _ui32ByteSent = 0;
uint32_t _ui32LastSent = 0;
uint32_t _ui32WinSize = 0;
///////////PeerBandwidth///////////
uint32_t _ui32Bandwidth = 2500000;
uint8_t _ui8LimitType = 2;
////////////Chunk////////////
unordered_map<int, RtmpPacket> _mapChunkData;
//////////Rtmp parser//////////
string _strRcvBuf;
function<void()> _nextHandle;
};
} /* namespace mediakit */

View File

@@ -34,55 +34,55 @@ using namespace mediakit::Client;
namespace mediakit {
RtmpPusher::RtmpPusher(const EventPoller::Ptr &poller,const RtmpMediaSource::Ptr &src) : TcpClient(poller){
_pMediaSrc=src;
_pMediaSrc=src;
}
RtmpPusher::~RtmpPusher() {
teardown();
DebugL << endl;
teardown();
DebugL << endl;
}
void RtmpPusher::teardown() {
if (alive()) {
_strApp.clear();
_strStream.clear();
_strTcUrl.clear();
{
lock_guard<recursive_mutex> lck(_mtxOnResultCB);
_mapOnResultCB.clear();
}
if (alive()) {
_strApp.clear();
_strStream.clear();
_strTcUrl.clear();
{
lock_guard<recursive_mutex> lck(_mtxOnResultCB);
_mapOnResultCB.clear();
}
{
lock_guard<recursive_mutex> lck(_mtxOnStatusCB);
_dqOnStatusCB.clear();
}
_pPublishTimer.reset();
_pPublishTimer.reset();
reset();
shutdown(SockException(Err_shutdown,"teardown"));
}
}
}
void RtmpPusher::onPublishResult(const SockException &ex,bool handshakeCompleted) {
if(!handshakeCompleted){
//播放结果回调
_pPublishTimer.reset();
if(_onPublished){
_onPublished(ex);
}
} else {
//播放成功后异常断开回调
if(_onShutdown){
_onShutdown(ex);
}
}
if(!handshakeCompleted){
//播放结果回调
_pPublishTimer.reset();
if(_onPublished){
_onPublished(ex);
}
} else {
//播放成功后异常断开回调
if(_onShutdown){
_onShutdown(ex);
}
}
if(ex){
teardown();
}
if(ex){
teardown();
}
}
void RtmpPusher::publish(const string &strUrl) {
teardown();
string strHost = FindField(strUrl.data(), "://", "/");
_strApp = FindField(strUrl.data(), (strHost + "/").data(), "/");
teardown();
string strHost = FindField(strUrl.data(), "://", "/");
_strApp = FindField(strUrl.data(), (strHost + "/").data(), "/");
_strStream = FindField(strUrl.data(), (strHost + "/" + _strApp + "/").data(), NULL);
_strTcUrl = string("rtmp://") + strHost + "/" + _strApp;
@@ -90,16 +90,16 @@ void RtmpPusher::publish(const string &strUrl) {
onPublishResult(SockException(Err_other,"rtmp url非法"),false);
return;
}
DebugL << strHost << " " << _strApp << " " << _strStream;
DebugL << strHost << " " << _strApp << " " << _strStream;
auto iPort = atoi(FindField(strHost.data(), ":", NULL).data());
if (iPort <= 0) {
auto iPort = atoi(FindField(strHost.data(), ":", NULL).data());
if (iPort <= 0) {
//rtmp 默认端口1935
iPort = 1935;
} else {
iPort = 1935;
} else {
//服务器域名
strHost = FindField(strHost.data(), NULL, ":");
}
strHost = FindField(strHost.data(), NULL, ":");
}
weak_ptr<RtmpPusher> weakSelf = dynamic_pointer_cast<RtmpPusher>(shared_from_this());
float publishTimeOutSec = (*this)[kTimeoutMS].as<int>() / 1000.0;
@@ -116,23 +116,23 @@ void RtmpPusher::publish(const string &strUrl) {
setNetAdapter((*this)[kNetAdapter]);
}
startConnect(strHost, iPort);
startConnect(strHost, iPort);
}
void RtmpPusher::onErr(const SockException &ex){
//定时器_pPublishTimer为空后表明握手结束了
onPublishResult(ex,!_pPublishTimer);
//定时器_pPublishTimer为空后表明握手结束了
onPublishResult(ex,!_pPublishTimer);
}
void RtmpPusher::onConnect(const SockException &err){
if(err) {
onPublishResult(err,false);
return;
}
//推流器不需要多大的接收缓存,节省内存占用
_sock->setReadBuffer(std::make_shared<BufferRaw>(1 * 1024));
if(err) {
onPublishResult(err,false);
return;
}
//推流器不需要多大的接收缓存,节省内存占用
_sock->setReadBuffer(std::make_shared<BufferRaw>(1 * 1024));
weak_ptr<RtmpPusher> weakSelf = dynamic_pointer_cast<RtmpPusher>(shared_from_this());
startClientSession([weakSelf](){
weak_ptr<RtmpPusher> weakSelf = dynamic_pointer_cast<RtmpPusher>(shared_from_this());
startClientSession([weakSelf](){
auto strongSelf=weakSelf.lock();
if(!strongSelf) {
return;
@@ -140,63 +140,63 @@ void RtmpPusher::onConnect(const SockException &err){
strongSelf->sendChunkSize(60000);
strongSelf->send_connect();
});
});
}
void RtmpPusher::onRecv(const Buffer::Ptr &pBuf){
try {
onParseRtmp(pBuf->data(), pBuf->size());
} catch (exception &e) {
SockException ex(Err_other, e.what());
//定时器_pPublishTimer为空后表明握手结束了
onPublishResult(ex,!_pPublishTimer);
}
try {
onParseRtmp(pBuf->data(), pBuf->size());
} catch (exception &e) {
SockException ex(Err_other, e.what());
//定时器_pPublishTimer为空后表明握手结束了
onPublishResult(ex,!_pPublishTimer);
}
}
inline void RtmpPusher::send_connect() {
AMFValue obj(AMF_OBJECT);
obj.set("app", _strApp);
obj.set("type", "nonprivate");
obj.set("tcUrl", _strTcUrl);
obj.set("swfUrl", _strTcUrl);
sendInvoke("connect", obj);
addOnResultCB([this](AMFDecoder &dec){
//TraceL << "connect result";
dec.load<AMFValue>();
auto val = dec.load<AMFValue>();
auto level = val["level"].as_string();
auto code = val["code"].as_string();
if(level != "status"){
throw std::runtime_error(StrPrinter <<"connect 失败:" << level << " " << code << endl);
}
send_createStream();
});
AMFValue obj(AMF_OBJECT);
obj.set("app", _strApp);
obj.set("type", "nonprivate");
obj.set("tcUrl", _strTcUrl);
obj.set("swfUrl", _strTcUrl);
sendInvoke("connect", obj);
addOnResultCB([this](AMFDecoder &dec){
//TraceL << "connect result";
dec.load<AMFValue>();
auto val = dec.load<AMFValue>();
auto level = val["level"].as_string();
auto code = val["code"].as_string();
if(level != "status"){
throw std::runtime_error(StrPrinter <<"connect 失败:" << level << " " << code << endl);
}
send_createStream();
});
}
inline void RtmpPusher::send_createStream() {
AMFValue obj(AMF_NULL);
sendInvoke("createStream", obj);
addOnResultCB([this](AMFDecoder &dec){
//TraceL << "createStream result";
dec.load<AMFValue>();
_ui32StreamId = dec.load<int>();
send_publish();
});
AMFValue obj(AMF_NULL);
sendInvoke("createStream", obj);
addOnResultCB([this](AMFDecoder &dec){
//TraceL << "createStream result";
dec.load<AMFValue>();
_ui32StreamId = dec.load<int>();
send_publish();
});
}
inline void RtmpPusher::send_publish() {
AMFEncoder enc;
enc << "publish" << ++_iReqID << nullptr << _strStream << _strApp ;
sendRequest(MSG_CMD, enc.data());
AMFEncoder enc;
enc << "publish" << ++_iReqID << nullptr << _strStream << _strApp ;
sendRequest(MSG_CMD, enc.data());
addOnStatusCB([this](AMFValue &val) {
auto level = val["level"].as_string();
auto code = val["code"].as_string();
if(level != "status") {
throw std::runtime_error(StrPrinter <<"publish 失败:" << level << " " << code << endl);
}
//start send media
send_metaData();
});
addOnStatusCB([this](AMFValue &val) {
auto level = val["level"].as_string();
auto code = val["code"].as_string();
if(level != "status") {
throw std::runtime_error(StrPrinter <<"publish 失败:" << level << " " << code << endl);
}
//start send media
send_metaData();
});
}
inline void RtmpPusher::send_metaData(){
@@ -216,11 +216,11 @@ inline void RtmpPusher::send_metaData(){
_pRtmpReader = src->getRing()->attach(getPoller());
weak_ptr<RtmpPusher> weakSelf = dynamic_pointer_cast<RtmpPusher>(shared_from_this());
_pRtmpReader->setReadCB([weakSelf](const RtmpPacket::Ptr &pkt){
auto strongSelf = weakSelf.lock();
if(!strongSelf) {
return;
}
strongSelf->sendRtmp(pkt->typeId, strongSelf->_ui32StreamId, pkt, pkt->timeStamp, pkt->chunkId);
auto strongSelf = weakSelf.lock();
if(!strongSelf) {
return;
}
strongSelf->sendRtmp(pkt->typeId, strongSelf->_ui32StreamId, pkt, pkt->timeStamp, pkt->chunkId);
});
_pRtmpReader->setDetachCB([weakSelf](){
auto strongSelf = weakSelf.lock();
@@ -229,84 +229,84 @@ inline void RtmpPusher::send_metaData(){
}
});
onPublishResult(SockException(Err_success,"success"), false);
//提升发送性能
setSocketFlags();
//提升发送性能
setSocketFlags();
}
void RtmpPusher::setSocketFlags(){
GET_CONFIG(bool,ultraLowDelay,General::kUltraLowDelay);
if(!ultraLowDelay) {
GET_CONFIG(bool,ultraLowDelay,General::kUltraLowDelay);
if(!ultraLowDelay) {
//提高发送性能
(*this) << SocketFlags(SOCKET_DEFAULE_FLAGS | FLAG_MORE);
SockUtil::setNoDelay(_sock->rawFD(), false);
}
}
}
void RtmpPusher::onCmd_result(AMFDecoder &dec){
auto iReqId = dec.load<int>();
lock_guard<recursive_mutex> lck(_mtxOnResultCB);
auto it = _mapOnResultCB.find(iReqId);
if(it != _mapOnResultCB.end()){
it->second(dec);
_mapOnResultCB.erase(it);
}else{
WarnL << "unhandled _result";
}
auto iReqId = dec.load<int>();
lock_guard<recursive_mutex> lck(_mtxOnResultCB);
auto it = _mapOnResultCB.find(iReqId);
if(it != _mapOnResultCB.end()){
it->second(dec);
_mapOnResultCB.erase(it);
}else{
WarnL << "unhandled _result";
}
}
void RtmpPusher::onCmd_onStatus(AMFDecoder &dec) {
AMFValue val;
while(true){
val = dec.load<AMFValue>();
if(val.type() == AMF_OBJECT){
break;
}
}
if(val.type() != AMF_OBJECT){
throw std::runtime_error("onStatus:the result object was not found");
}
AMFValue val;
while(true){
val = dec.load<AMFValue>();
if(val.type() == AMF_OBJECT){
break;
}
}
if(val.type() != AMF_OBJECT){
throw std::runtime_error("onStatus:the result object was not found");
}
lock_guard<recursive_mutex> lck(_mtxOnStatusCB);
if(_dqOnStatusCB.size()){
_dqOnStatusCB.front()(val);
_dqOnStatusCB.pop_front();
}else{
auto level = val["level"];
auto code = val["code"].as_string();
if(level.type() == AMF_STRING){
if(level.as_string() != "status"){
throw std::runtime_error(StrPrinter <<"onStatus 失败:" << level.as_string() << " " << code << endl);
}
}
if(_dqOnStatusCB.size()){
_dqOnStatusCB.front()(val);
_dqOnStatusCB.pop_front();
}else{
auto level = val["level"];
auto code = val["code"].as_string();
if(level.type() == AMF_STRING){
if(level.as_string() != "status"){
throw std::runtime_error(StrPrinter <<"onStatus 失败:" << level.as_string() << " " << code << endl);
}
}
}
}
void RtmpPusher::onRtmpChunk(RtmpPacket &chunkData) {
switch (chunkData.typeId) {
case MSG_CMD:
case MSG_CMD3: {
typedef void (RtmpPusher::*rtmpCMDHandle)(AMFDecoder &dec);
static unordered_map<string, rtmpCMDHandle> g_mapCmd;
static onceToken token([]() {
g_mapCmd.emplace("_error",&RtmpPusher::onCmd_result);
g_mapCmd.emplace("_result",&RtmpPusher::onCmd_result);
g_mapCmd.emplace("onStatus",&RtmpPusher::onCmd_onStatus);
}, []() {});
switch (chunkData.typeId) {
case MSG_CMD:
case MSG_CMD3: {
typedef void (RtmpPusher::*rtmpCMDHandle)(AMFDecoder &dec);
static unordered_map<string, rtmpCMDHandle> g_mapCmd;
static onceToken token([]() {
g_mapCmd.emplace("_error",&RtmpPusher::onCmd_result);
g_mapCmd.emplace("_result",&RtmpPusher::onCmd_result);
g_mapCmd.emplace("onStatus",&RtmpPusher::onCmd_onStatus);
}, []() {});
AMFDecoder dec(chunkData.strBuf, 0);
std::string type = dec.load<std::string>();
auto it = g_mapCmd.find(type);
if(it != g_mapCmd.end()){
auto fun = it->second;
(this->*fun)(dec);
}else{
WarnL << "can not support cmd:" << type;
}
}
break;
default:
//WarnL << "unhandled message:" << (int) chunkData.typeId << hexdump(chunkData.strBuf.data(), chunkData.strBuf.size());
break;
}
AMFDecoder dec(chunkData.strBuf, 0);
std::string type = dec.load<std::string>();
auto it = g_mapCmd.find(type);
if(it != g_mapCmd.end()){
auto fun = it->second;
(this->*fun)(dec);
}else{
WarnL << "can not support cmd:" << type;
}
}
break;
default:
//WarnL << "unhandled message:" << (int) chunkData.typeId << hexdump(chunkData.strBuf.data(), chunkData.strBuf.size());
break;
}
}

View File

@@ -36,66 +36,66 @@ namespace mediakit {
class RtmpPusher: public RtmpProtocol , public TcpClient , public PusherBase{
public:
typedef std::shared_ptr<RtmpPusher> Ptr;
RtmpPusher(const EventPoller::Ptr &poller,const RtmpMediaSource::Ptr &src);
virtual ~RtmpPusher();
typedef std::shared_ptr<RtmpPusher> Ptr;
RtmpPusher(const EventPoller::Ptr &poller,const RtmpMediaSource::Ptr &src);
virtual ~RtmpPusher();
void publish(const string &strUrl) override ;
void publish(const string &strUrl) override ;
void teardown() override;
void teardown() override;
void setOnPublished(const Event &cb) override {
_onPublished = cb;
}
void setOnPublished(const Event &cb) override {
_onPublished = cb;
}
void setOnShutdown(const Event &cb) override{
_onShutdown = cb;
}
void setOnShutdown(const Event &cb) override{
_onShutdown = cb;
}
protected:
//for Tcpclient override
void onRecv(const Buffer::Ptr &pBuf) override;
void onConnect(const SockException &err) override;
void onErr(const SockException &ex) override;
//for Tcpclient override
void onRecv(const Buffer::Ptr &pBuf) override;
void onConnect(const SockException &err) override;
void onErr(const SockException &ex) override;
//for RtmpProtocol override
void onRtmpChunk(RtmpPacket &chunkData) override;
void onSendRawData(const Buffer::Ptr &buffer) override{
send(buffer);
}
//for RtmpProtocol override
void onRtmpChunk(RtmpPacket &chunkData) override;
void onSendRawData(const Buffer::Ptr &buffer) override{
send(buffer);
}
private:
void onPublishResult(const SockException &ex,bool handshakeCompleted);
void onPublishResult(const SockException &ex,bool handshakeCompleted);
template<typename FUN>
inline void addOnResultCB(const FUN &fun) {
lock_guard<recursive_mutex> lck(_mtxOnResultCB);
_mapOnResultCB.emplace(_iReqID, fun);
}
template<typename FUN>
inline void addOnStatusCB(const FUN &fun) {
lock_guard<recursive_mutex> lck(_mtxOnStatusCB);
_dqOnStatusCB.emplace_back(fun);
}
template<typename FUN>
inline void addOnResultCB(const FUN &fun) {
lock_guard<recursive_mutex> lck(_mtxOnResultCB);
_mapOnResultCB.emplace(_iReqID, fun);
}
template<typename FUN>
inline void addOnStatusCB(const FUN &fun) {
lock_guard<recursive_mutex> lck(_mtxOnStatusCB);
_dqOnStatusCB.emplace_back(fun);
}
void onCmd_result(AMFDecoder &dec);
void onCmd_onStatus(AMFDecoder &dec);
void onCmd_onMetaData(AMFDecoder &dec);
void onCmd_result(AMFDecoder &dec);
void onCmd_onStatus(AMFDecoder &dec);
void onCmd_onMetaData(AMFDecoder &dec);
inline void send_connect();
inline void send_createStream();
inline void send_publish();
inline void send_metaData();
void setSocketFlags();
inline void send_connect();
inline void send_createStream();
inline void send_publish();
inline void send_metaData();
void setSocketFlags();
private:
string _strApp;
string _strStream;
string _strTcUrl;
string _strApp;
string _strStream;
string _strTcUrl;
unordered_map<int, function<void(AMFDecoder &dec)> > _mapOnResultCB;
recursive_mutex _mtxOnResultCB;
deque<function<void(AMFValue &dec)> > _dqOnStatusCB;
recursive_mutex _mtxOnStatusCB;
//超时功能实现
std::shared_ptr<Timer> _pPublishTimer;
unordered_map<int, function<void(AMFDecoder &dec)> > _mapOnResultCB;
recursive_mutex _mtxOnResultCB;
deque<function<void(AMFValue &dec)> > _dqOnStatusCB;
recursive_mutex _mtxOnStatusCB;
//超时功能实现
std::shared_ptr<Timer> _pPublishTimer;
//源
std::weak_ptr<RtmpMediaSource> _pMediaSrc;
RtmpMediaSource::RingType::RingReader::Ptr _pRtmpReader;

View File

@@ -32,7 +32,7 @@
namespace mediakit {
RtmpSession::RtmpSession(const Socket::Ptr &pSock) : TcpSession(pSock) {
DebugP(this);
DebugP(this);
GET_CONFIG(uint32_t,keep_alive_sec,Rtmp::kKeepAliveSecond);
pSock->setSendTimeOutSecond(keep_alive_sec);
//起始接收buffer缓存设置为4K节省内存
@@ -65,42 +65,42 @@ void RtmpSession::onManager() {
GET_CONFIG(uint32_t,handshake_sec,Rtmp::kHandshakeSecond);
GET_CONFIG(uint32_t,keep_alive_sec,Rtmp::kKeepAliveSecond);
if (_ticker.createdTime() > handshake_sec * 1000) {
if (!_pRingReader && !_pPublisherSrc) {
shutdown(SockException(Err_timeout,"illegal connection"));
}
}
if (_pPublisherSrc) {
//publisher
if (_ticker.elapsedTime() > keep_alive_sec * 1000) {
shutdown(SockException(Err_timeout,"recv data from rtmp pusher timeout"));
}
}
if (_ticker.createdTime() > handshake_sec * 1000) {
if (!_pRingReader && !_pPublisherSrc) {
shutdown(SockException(Err_timeout,"illegal connection"));
}
}
if (_pPublisherSrc) {
//publisher
if (_ticker.elapsedTime() > keep_alive_sec * 1000) {
shutdown(SockException(Err_timeout,"recv data from rtmp pusher timeout"));
}
}
}
void RtmpSession::onRecv(const Buffer::Ptr &pBuf) {
_ticker.resetTime();
try {
_ticker.resetTime();
try {
_ui64TotalBytes += pBuf->size();
onParseRtmp(pBuf->data(), pBuf->size());
} catch (exception &e) {
shutdown(SockException(Err_shutdown, e.what()));
}
onParseRtmp(pBuf->data(), pBuf->size());
} catch (exception &e) {
shutdown(SockException(Err_shutdown, e.what()));
}
}
void RtmpSession::onCmd_connect(AMFDecoder &dec) {
auto params = dec.load<AMFValue>();
double amfVer = 0;
AMFValue objectEncoding = params["objectEncoding"];
if(objectEncoding){
amfVer = objectEncoding.as_number();
}
///////////set chunk size////////////////
sendChunkSize(60000);
////////////window Acknowledgement size/////
sendAcknowledgementSize(5000000);
///////////set peerBandwidth////////////////
sendPeerBandwidth(5000000);
auto params = dec.load<AMFValue>();
double amfVer = 0;
AMFValue objectEncoding = params["objectEncoding"];
if(objectEncoding){
amfVer = objectEncoding.as_number();
}
///////////set chunk size////////////////
sendChunkSize(60000);
////////////window Acknowledgement size/////
sendAcknowledgementSize(5000000);
///////////set peerBandwidth////////////////
sendPeerBandwidth(5000000);
_mediaInfo._app = params["app"].as_string();
_strTcUrl = params["tcUrl"].as_string();
@@ -108,27 +108,27 @@ void RtmpSession::onCmd_connect(AMFDecoder &dec) {
//defaultVhost:默认vhost
_strTcUrl = string(RTMP_SCHEMA) + "://" + DEFAULT_VHOST + "/" + _mediaInfo._app;
}
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.");
status.set("objectEncoding", amfVer);
sendReply(ok ? "_result" : "_error", version, status);
if (!ok) {
throw std::runtime_error("Unsupported application: " + _mediaInfo._app);
}
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.");
status.set("objectEncoding", amfVer);
sendReply(ok ? "_result" : "_error", version, status);
if (!ok) {
throw std::runtime_error("Unsupported application: " + _mediaInfo._app);
}
AMFEncoder invoke;
invoke << "onBWDone" << 0.0 << nullptr;
sendResponse(MSG_CMD, invoke.data());
AMFEncoder invoke;
invoke << "onBWDone" << 0.0 << nullptr;
sendResponse(MSG_CMD, invoke.data());
}
void RtmpSession::onCmd_createStream(AMFDecoder &dec) {
sendReply("_result", nullptr, double(STREAM_MEDIA));
sendReply("_result", nullptr, double(STREAM_MEDIA));
}
void RtmpSession::onCmd_publish(AMFDecoder &dec) {
@@ -140,7 +140,7 @@ void RtmpSession::onCmd_publish(AMFDecoder &dec) {
DebugP(strongSelf.get()) << "publish 回复时间:" << pTicker->elapsedTime() << "ms";
}
}));
dec.load<AMFValue>();/* NULL */
dec.load<AMFValue>();/* NULL */
_mediaInfo.parse(_strTcUrl + "/" + getStreamId(dec.load<std::string>()));
_mediaInfo._schema = RTMP_SCHEMA;
@@ -203,12 +203,12 @@ void RtmpSession::onCmd_publish(AMFDecoder &dec) {
}
void RtmpSession::onCmd_deleteStream(AMFDecoder &dec) {
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);
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);
}
@@ -355,7 +355,7 @@ void RtmpSession::doPlay(AMFDecoder &dec){
}
}
void RtmpSession::onCmd_play2(AMFDecoder &dec) {
doPlay(dec);
doPlay(dec);
}
string RtmpSession::getStreamId(const string &str){
@@ -390,49 +390,49 @@ string RtmpSession::getStreamId(const string &str){
}
void RtmpSession::onCmd_play(AMFDecoder &dec) {
dec.load<AMFValue>();/* NULL */
dec.load<AMFValue>();/* NULL */
_mediaInfo.parse(_strTcUrl + "/" + getStreamId(dec.load<std::string>()));
_mediaInfo._schema = RTMP_SCHEMA;
doPlay(dec);
doPlay(dec);
}
void RtmpSession::onCmd_pause(AMFDecoder &dec) {
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);
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);
//streamBegin
sendUserControl(paused ? CONTROL_STREAM_EOF : CONTROL_STREAM_BEGIN,
STREAM_MEDIA);
if (!_pRingReader) {
throw std::runtime_error("Rtmp not started yet!");
}
if (paused) {
_pRingReader->setReadCB(nullptr);
} else {
weak_ptr<RtmpSession> weakSelf = dynamic_pointer_cast<RtmpSession>(shared_from_this());
_pRingReader->setReadCB([weakSelf](const RtmpPacket::Ptr &pkt) {
auto strongSelf = weakSelf.lock();
if(!strongSelf) {
return;
}
sendUserControl(paused ? CONTROL_STREAM_EOF : CONTROL_STREAM_BEGIN,
STREAM_MEDIA);
if (!_pRingReader) {
throw std::runtime_error("Rtmp not started yet!");
}
if (paused) {
_pRingReader->setReadCB(nullptr);
} else {
weak_ptr<RtmpSession> weakSelf = dynamic_pointer_cast<RtmpSession>(shared_from_this());
_pRingReader->setReadCB([weakSelf](const RtmpPacket::Ptr &pkt) {
auto strongSelf = weakSelf.lock();
if(!strongSelf) {
return;
}
strongSelf->onSendMedia(pkt);
});
}
});
}
}
void RtmpSession::setMetaData(AMFDecoder &dec) {
if (!_pPublisherSrc) {
throw std::runtime_error("not a publisher");
}
std::string type = dec.load<std::string>();
if (type != "onMetaData") {
throw std::runtime_error("can only set metadata");
}
if (!_pPublisherSrc) {
throw std::runtime_error("not a publisher");
}
std::string type = dec.load<std::string>();
if (type != "onMetaData") {
throw std::runtime_error("can only set metadata");
}
auto metadata = dec.load<AMFValue>();
// dumpMetadata(metadata);
_pPublisherSrc->setMetaData(metadata);
@@ -453,42 +453,42 @@ void RtmpSession::onProcessCmd(AMFDecoder &dec) {
s_cmd_functions.emplace("pause",&RtmpSession::onCmd_pause);}, []() {});
std::string method = dec.load<std::string>();
auto it = s_cmd_functions.find(method);
if (it == s_cmd_functions.end()) {
auto it = s_cmd_functions.find(method);
if (it == s_cmd_functions.end()) {
// TraceP(this) << "can not support cmd:" << method;
return;
}
_dNowReqID = dec.load<double>();
auto fun = it->second;
(this->*fun)(dec);
return;
}
_dNowReqID = dec.load<double>();
auto fun = it->second;
(this->*fun)(dec);
}
void RtmpSession::onRtmpChunk(RtmpPacket &chunkData) {
switch (chunkData.typeId) {
case MSG_CMD:
case MSG_CMD3: {
AMFDecoder dec(chunkData.strBuf, chunkData.typeId == MSG_CMD3 ? 1 : 0);
onProcessCmd(dec);
}
break;
switch (chunkData.typeId) {
case MSG_CMD:
case MSG_CMD3: {
AMFDecoder dec(chunkData.strBuf, chunkData.typeId == MSG_CMD3 ? 1 : 0);
onProcessCmd(dec);
}
break;
case MSG_DATA:
case MSG_DATA3: {
AMFDecoder dec(chunkData.strBuf, chunkData.typeId == MSG_CMD3 ? 1 : 0);
std::string type = dec.load<std::string>();
if (type == "@setDataFrame") {
setMetaData(dec);
}else{
case MSG_DATA:
case MSG_DATA3: {
AMFDecoder dec(chunkData.strBuf, chunkData.typeId == MSG_CMD3 ? 1 : 0);
std::string type = dec.load<std::string>();
if (type == "@setDataFrame") {
setMetaData(dec);
}else{
TraceP(this) << "unknown notify:" << type;
}
}
break;
case MSG_AUDIO:
case MSG_VIDEO: {
if (!_pPublisherSrc) {
throw std::runtime_error("Not a rtmp publisher!");
}
GET_CONFIG(bool,rtmp_modify_stamp,Rtmp::kModifyStamp);
}
break;
case MSG_AUDIO:
case MSG_VIDEO: {
if (!_pPublisherSrc) {
throw std::runtime_error("Not a rtmp publisher!");
}
GET_CONFIG(bool,rtmp_modify_stamp,Rtmp::kModifyStamp);
if(rtmp_modify_stamp){
int64_t dts_out;
_stamp[chunkData.typeId % 2].revise(chunkData.timeStamp, chunkData.timeStamp, dts_out, dts_out, true);
@@ -500,12 +500,12 @@ void RtmpSession::onRtmpChunk(RtmpPacket &chunkData) {
_pPublisherSrc->setMetaData(TitleMeta().getMetadata());
}
_pPublisherSrc->onWrite(std::make_shared<RtmpPacket>(std::move(chunkData)));
}
break;
default:
WarnP(this) << "unhandled message:" << (int) chunkData.typeId << hexdump(chunkData.strBuf.data(), chunkData.strBuf.size());
break;
}
}
break;
default:
WarnP(this) << "unhandled message:" << (int) chunkData.typeId << hexdump(chunkData.strBuf.data(), chunkData.strBuf.size());
break;
}
}
void RtmpSession::onCmd_seek(AMFDecoder &dec) {
@@ -518,12 +518,12 @@ void RtmpSession::onCmd_seek(AMFDecoder &dec) {
_stamp[1].setPlayBack();
stongSrc->seekTo(milliSeconds);
}
AMFValue status(AMF_OBJECT);
AMFEncoder invoke;
status.set("level", "status");
status.set("code", "NetStream.Seek.Notify");
status.set("description", "Seeking.");
sendReply("onStatus", nullptr, status);
AMFValue status(AMF_OBJECT);
AMFEncoder invoke;
status.set("level", "status");
status.set("code", "NetStream.Seek.Notify");
status.set("description", "Seeking.");
sendReply("onStatus", nullptr, status);
}
void RtmpSession::onSendMedia(const RtmpPacket::Ptr &pkt) {

View File

@@ -45,65 +45,65 @@ namespace mediakit {
class RtmpSession: public TcpSession ,public RtmpProtocol , public MediaSourceEvent{
public:
typedef std::shared_ptr<RtmpSession> Ptr;
RtmpSession(const Socket::Ptr &_sock);
virtual ~RtmpSession();
void onRecv(const Buffer::Ptr &pBuf) override;
void onError(const SockException &err) override;
void onManager() override;
typedef std::shared_ptr<RtmpSession> Ptr;
RtmpSession(const Socket::Ptr &_sock);
virtual ~RtmpSession();
void onRecv(const Buffer::Ptr &pBuf) override;
void onError(const SockException &err) override;
void onManager() override;
private:
void onProcessCmd(AMFDecoder &dec);
void onCmd_connect(AMFDecoder &dec);
void onCmd_createStream(AMFDecoder &dec);
void onProcessCmd(AMFDecoder &dec);
void onCmd_connect(AMFDecoder &dec);
void onCmd_createStream(AMFDecoder &dec);
void onCmd_publish(AMFDecoder &dec);
void onCmd_deleteStream(AMFDecoder &dec);
void onCmd_publish(AMFDecoder &dec);
void onCmd_deleteStream(AMFDecoder &dec);
void onCmd_play(AMFDecoder &dec);
void onCmd_play2(AMFDecoder &dec);
void doPlay(AMFDecoder &dec);
void doPlayResponse(const string &err,const std::function<void(bool)> &cb);
void sendPlayResponse(const string &err,const RtmpMediaSource::Ptr &src);
void onCmd_play(AMFDecoder &dec);
void onCmd_play2(AMFDecoder &dec);
void doPlay(AMFDecoder &dec);
void doPlayResponse(const string &err,const std::function<void(bool)> &cb);
void sendPlayResponse(const string &err,const RtmpMediaSource::Ptr &src);
void onCmd_seek(AMFDecoder &dec);
void onCmd_pause(AMFDecoder &dec);
void setMetaData(AMFDecoder &dec);
void onCmd_seek(AMFDecoder &dec);
void onCmd_pause(AMFDecoder &dec);
void setMetaData(AMFDecoder &dec);
void onSendMedia(const RtmpPacket::Ptr &pkt);
void onSendRawData(const Buffer::Ptr &buffer) override{
void onSendMedia(const RtmpPacket::Ptr &pkt);
void onSendRawData(const Buffer::Ptr &buffer) override{
_ui64TotalBytes += buffer->size();
send(buffer);
}
void onRtmpChunk(RtmpPacket &chunkData) override;
send(buffer);
}
void onRtmpChunk(RtmpPacket &chunkData) override;
template<typename first, typename second>
inline void sendReply(const char *str, const first &reply, const second &status) {
AMFEncoder invoke;
invoke << str << _dNowReqID << reply << status;
sendResponse(MSG_CMD, invoke.data());
}
template<typename first, typename second>
inline void sendReply(const char *str, const first &reply, const second &status) {
AMFEncoder invoke;
invoke << str << _dNowReqID << reply << status;
sendResponse(MSG_CMD, invoke.data());
}
//MediaSourceEvent override
bool close(MediaSource &sender,bool force) override ;
//MediaSourceEvent override
bool close(MediaSource &sender,bool force) override ;
void onNoneReader(MediaSource &sender) override;
int totalReaderCount(MediaSource &sender) override;
int totalReaderCount(MediaSource &sender) override;
void setSocketFlags();
string getStreamId(const string &str);
void dumpMetadata(const AMFValue &metadata);
void setSocketFlags();
string getStreamId(const string &str);
void dumpMetadata(const AMFValue &metadata);
private:
std::string _strTcUrl;
MediaInfo _mediaInfo;
double _dNowReqID = 0;
bool _set_meta_data = false;
Ticker _ticker;//数据接收时间
RingBuffer<RtmpPacket::Ptr>::RingReader::Ptr _pRingReader;
std::shared_ptr<RtmpMediaSourceImp> _pPublisherSrc;
std::weak_ptr<RtmpMediaSource> _pPlayerSrc;
//时间戳修整器
Stamp _stamp[2];
//消耗的总流量
uint64_t _ui64TotalBytes = 0;
std::string _strTcUrl;
MediaInfo _mediaInfo;
double _dNowReqID = 0;
bool _set_meta_data = false;
Ticker _ticker;//数据接收时间
RingBuffer<RtmpPacket::Ptr>::RingReader::Ptr _pRingReader;
std::shared_ptr<RtmpMediaSourceImp> _pPublisherSrc;
std::weak_ptr<RtmpMediaSource> _pPlayerSrc;
//时间戳修整器
Stamp _stamp[2];
//消耗的总流量
uint64_t _ui64TotalBytes = 0;
};

View File

@@ -35,130 +35,130 @@ using namespace toolkit;
/////////////////////AMFValue/////////////////////////////
inline void AMFValue::destroy() {
switch (_type) {
case AMF_STRING:
if (_value.string) {
delete _value.string;
_value.string = nullptr;
}
break;
case AMF_OBJECT:
case AMF_ECMA_ARRAY:
if (_value.object) {
delete _value.object;
_value.object = nullptr;
}
break;
case AMF_STRICT_ARRAY:
if (_value.array) {
delete _value.array;
_value.array = nullptr;
}
break;
default:
break;
}
switch (_type) {
case AMF_STRING:
if (_value.string) {
delete _value.string;
_value.string = nullptr;
}
break;
case AMF_OBJECT:
case AMF_ECMA_ARRAY:
if (_value.object) {
delete _value.object;
_value.object = nullptr;
}
break;
case AMF_STRICT_ARRAY:
if (_value.array) {
delete _value.array;
_value.array = nullptr;
}
break;
default:
break;
}
}
inline void AMFValue::init() {
switch (_type) {
case AMF_OBJECT:
case AMF_ECMA_ARRAY:
_value.object = new mapType;
break;
case AMF_STRING:
_value.string = new std::string;
break;
case AMF_STRICT_ARRAY:
_value.array = new arrayType;
break;
switch (_type) {
case AMF_OBJECT:
case AMF_ECMA_ARRAY:
_value.object = new mapType;
break;
case AMF_STRING:
_value.string = new std::string;
break;
case AMF_STRICT_ARRAY:
_value.array = new arrayType;
break;
default:
break;
}
default:
break;
}
}
AMFValue::AMFValue(AMFType type) :
_type(type) {
init();
_type(type) {
init();
}
AMFValue::~AMFValue() {
destroy();
destroy();
}
AMFValue::AMFValue(const char *s) :
_type(AMF_STRING) {
init();
*_value.string = s;
_type(AMF_STRING) {
init();
*_value.string = s;
}
AMFValue::AMFValue(const std::string &s) :
_type(AMF_STRING) {
init();
*_value.string = s;
_type(AMF_STRING) {
init();
*_value.string = s;
}
AMFValue::AMFValue(double n) :
_type(AMF_NUMBER) {
init();
_value.number = n;
_type(AMF_NUMBER) {
init();
_value.number = n;
}
AMFValue::AMFValue(int i) :
_type(AMF_INTEGER) {
init();
_value.integer = i;
_type(AMF_INTEGER) {
init();
_value.integer = i;
}
AMFValue::AMFValue(bool b) :
_type(AMF_BOOLEAN) {
init();
_value.boolean = b;
_type(AMF_BOOLEAN) {
init();
_value.boolean = b;
}
AMFValue::AMFValue(const AMFValue &from) :
_type(AMF_NULL) {
*this = from;
_type(AMF_NULL) {
*this = from;
}
AMFValue::AMFValue(AMFValue &&from) {
*this = std::forward<AMFValue>(from);
*this = std::forward<AMFValue>(from);
}
AMFValue& AMFValue::operator =(const AMFValue &from) {
return *this = const_cast<AMFValue &&>(from);
return *this = const_cast<AMFValue &&>(from);
}
AMFValue& AMFValue::operator =(AMFValue &&from) {
destroy();
_type = from._type;
init();
switch (_type) {
case AMF_STRING:
*_value.string = (*from._value.string);
break;
case AMF_OBJECT:
case AMF_ECMA_ARRAY:
*_value.object = (*from._value.object);
break;
case AMF_STRICT_ARRAY:
*_value.array = (*from._value.array);
break;
case AMF_NUMBER:
_value.number = from._value.number;
break;
case AMF_INTEGER:
_value.integer = from._value.integer;
break;
case AMF_BOOLEAN:
_value.boolean = from._value.boolean;
break;
default:
break;
}
return *this;
destroy();
_type = from._type;
init();
switch (_type) {
case AMF_STRING:
*_value.string = (*from._value.string);
break;
case AMF_OBJECT:
case AMF_ECMA_ARRAY:
*_value.object = (*from._value.object);
break;
case AMF_STRICT_ARRAY:
*_value.array = (*from._value.array);
break;
case AMF_NUMBER:
_value.number = from._value.number;
break;
case AMF_INTEGER:
_value.integer = from._value.integer;
break;
case AMF_BOOLEAN:
_value.boolean = from._value.boolean;
break;
default:
break;
}
return *this;
}
@@ -305,156 +305,156 @@ const AMFValue::arrayType &AMFValue::getArr() const {
///////////////////////////////////////////////////////////////////////////
enum {
AMF0_NUMBER,
AMF0_BOOLEAN,
AMF0_STRING,
AMF0_OBJECT,
AMF0_MOVIECLIP,
AMF0_NULL,
AMF0_UNDEFINED,
AMF0_REFERENCE,
AMF0_ECMA_ARRAY,
AMF0_OBJECT_END,
AMF0_STRICT_ARRAY,
AMF0_DATE,
AMF0_LONG_STRING,
AMF0_UNSUPPORTED,
AMF0_RECORD_SET,
AMF0_XML_OBJECT,
AMF0_TYPED_OBJECT,
AMF0_SWITCH_AMF3,
AMF0_NUMBER,
AMF0_BOOLEAN,
AMF0_STRING,
AMF0_OBJECT,
AMF0_MOVIECLIP,
AMF0_NULL,
AMF0_UNDEFINED,
AMF0_REFERENCE,
AMF0_ECMA_ARRAY,
AMF0_OBJECT_END,
AMF0_STRICT_ARRAY,
AMF0_DATE,
AMF0_LONG_STRING,
AMF0_UNSUPPORTED,
AMF0_RECORD_SET,
AMF0_XML_OBJECT,
AMF0_TYPED_OBJECT,
AMF0_SWITCH_AMF3,
};
enum {
AMF3_UNDEFINED,
AMF3_NULL,
AMF3_FALSE,
AMF3_TRUE,
AMF3_INTEGER,
AMF3_NUMBER,
AMF3_STRING,
AMF3_LEGACY_XML,
AMF3_DATE,
AMF3_ARRAY,
AMF3_OBJECT,
AMF3_XML,
AMF3_BYTE_ARRAY,
AMF3_UNDEFINED,
AMF3_NULL,
AMF3_FALSE,
AMF3_TRUE,
AMF3_INTEGER,
AMF3_NUMBER,
AMF3_STRING,
AMF3_LEGACY_XML,
AMF3_DATE,
AMF3_ARRAY,
AMF3_OBJECT,
AMF3_XML,
AMF3_BYTE_ARRAY,
};
////////////////////////////////Encoder//////////////////////////////////////////
AMFEncoder & AMFEncoder::operator <<(const char *s) {
if (s) {
buf += char(AMF0_STRING);
uint16_t str_len = htons(strlen(s));
buf.append((char *) &str_len, 2);
buf += s;
} else {
buf += char(AMF0_NULL);
}
return *this;
if (s) {
buf += char(AMF0_STRING);
uint16_t str_len = htons(strlen(s));
buf.append((char *) &str_len, 2);
buf += s;
} else {
buf += char(AMF0_NULL);
}
return *this;
}
AMFEncoder & AMFEncoder::operator <<(const std::string &s) {
if (!s.empty()) {
buf += char(AMF0_STRING);
uint16_t str_len = htons(s.size());
buf.append((char *) &str_len, 2);
buf += s;
} else {
buf += char(AMF0_NULL);
}
return *this;
if (!s.empty()) {
buf += char(AMF0_STRING);
uint16_t str_len = htons(s.size());
buf.append((char *) &str_len, 2);
buf += s;
} else {
buf += char(AMF0_NULL);
}
return *this;
}
AMFEncoder & AMFEncoder::operator <<(std::nullptr_t) {
buf += char(AMF0_NULL);
return *this;
buf += char(AMF0_NULL);
return *this;
}
AMFEncoder & AMFEncoder::write_undefined() {
buf += char(AMF0_UNDEFINED);
return *this;
buf += char(AMF0_UNDEFINED);
return *this;
}
AMFEncoder & AMFEncoder::operator <<(const int n){
return (*this) << (double)n;
return (*this) << (double)n;
}
AMFEncoder & AMFEncoder::operator <<(const double n) {
buf += char(AMF0_NUMBER);
uint64_t encoded = 0;
memcpy(&encoded, &n, 8);
uint32_t val = htonl(encoded >> 32);
buf.append((char *) &val, 4);
val = htonl(encoded);
buf.append((char *) &val, 4);
return *this;
buf += char(AMF0_NUMBER);
uint64_t encoded = 0;
memcpy(&encoded, &n, 8);
uint32_t val = htonl(encoded >> 32);
buf.append((char *) &val, 4);
val = htonl(encoded);
buf.append((char *) &val, 4);
return *this;
}
AMFEncoder & AMFEncoder::operator <<(const bool b) {
buf += char(AMF0_BOOLEAN);
buf += char(b);
return *this;
buf += char(AMF0_BOOLEAN);
buf += char(b);
return *this;
}
AMFEncoder & AMFEncoder::operator <<(const AMFValue& value) {
switch ((int) value.type()) {
case AMF_STRING:
*this << value.as_string();
break;
case AMF_NUMBER:
*this << value.as_number();
break;
case AMF_INTEGER:
*this << value.as_integer();
break;
case AMF_BOOLEAN:
*this << value.as_boolean();
break;
case AMF_OBJECT: {
buf += char(AMF0_OBJECT);
for (auto &pr : value.getMap()) {
write_key(pr.first);
*this << pr.second;
}
write_key("");
buf += char(AMF0_OBJECT_END);
}
break;
case AMF_ECMA_ARRAY: {
buf += char(AMF0_ECMA_ARRAY);
uint32_t sz = htonl(value.getMap().size());
buf.append((char *) &sz, 4);
for (auto &pr : value.getMap()) {
write_key(pr.first);
*this << pr.second;
}
write_key("");
buf += char(AMF0_OBJECT_END);
}
break;
case AMF_NULL:
*this << nullptr;
break;
case AMF_UNDEFINED:
this->write_undefined();
break;
case AMF_STRICT_ARRAY: {
buf += char(AMF0_STRICT_ARRAY);
uint32_t sz = htonl(value.getArr().size());
buf.append((char *) &sz, 4);
for (auto &val : value.getArr()) {
*this << val;
}
//write_key("");
//buf += char(AMF0_OBJECT_END);
}
break;
}
return *this;
switch ((int) value.type()) {
case AMF_STRING:
*this << value.as_string();
break;
case AMF_NUMBER:
*this << value.as_number();
break;
case AMF_INTEGER:
*this << value.as_integer();
break;
case AMF_BOOLEAN:
*this << value.as_boolean();
break;
case AMF_OBJECT: {
buf += char(AMF0_OBJECT);
for (auto &pr : value.getMap()) {
write_key(pr.first);
*this << pr.second;
}
write_key("");
buf += char(AMF0_OBJECT_END);
}
break;
case AMF_ECMA_ARRAY: {
buf += char(AMF0_ECMA_ARRAY);
uint32_t sz = htonl(value.getMap().size());
buf.append((char *) &sz, 4);
for (auto &pr : value.getMap()) {
write_key(pr.first);
*this << pr.second;
}
write_key("");
buf += char(AMF0_OBJECT_END);
}
break;
case AMF_NULL:
*this << nullptr;
break;
case AMF_UNDEFINED:
this->write_undefined();
break;
case AMF_STRICT_ARRAY: {
buf += char(AMF0_STRICT_ARRAY);
uint32_t sz = htonl(value.getArr().size());
buf.append((char *) &sz, 4);
for (auto &val : value.getArr()) {
*this << val;
}
//write_key("");
//buf += char(AMF0_OBJECT_END);
}
break;
}
return *this;
}
void AMFEncoder::write_key(const std::string& s) {
uint16_t str_len = htons(s.size());
buf.append((char *) &str_len, 2);
buf += s;
uint16_t str_len = htons(s.size());
buf.append((char *) &str_len, 2);
buf += s;
}
void AMFEncoder::clear() {
@@ -468,237 +468,237 @@ const std::string& AMFEncoder::data() const {
//////////////////Decoder//////////////////
uint8_t AMFDecoder::front() {
if (pos >= buf.size()) {
throw std::runtime_error("Not enough data");
}
return uint8_t(buf[pos]);
if (pos >= buf.size()) {
throw std::runtime_error("Not enough data");
}
return uint8_t(buf[pos]);
}
uint8_t AMFDecoder::pop_front() {
if (version == 0 && front() == AMF0_SWITCH_AMF3) {
InfoL << "entering AMF3 mode";
pos++;
version = 3;
}
if (version == 0 && front() == AMF0_SWITCH_AMF3) {
InfoL << "entering AMF3 mode";
pos++;
version = 3;
}
if (pos >= buf.size()) {
throw std::runtime_error("Not enough data");
}
return uint8_t(buf[pos++]);
if (pos >= buf.size()) {
throw std::runtime_error("Not enough data");
}
return uint8_t(buf[pos++]);
}
template<>
double AMFDecoder::load<double>() {
if (pop_front() != AMF0_NUMBER) {
throw std::runtime_error("Expected a number");
}
if (pos + 8 > buf.size()) {
throw std::runtime_error("Not enough data");
}
uint64_t val = ((uint64_t) load_be32(&buf[pos]) << 32)
| load_be32(&buf[pos + 4]);
double n = 0;
memcpy(&n, &val, 8);
pos += 8;
return n;
if (pop_front() != AMF0_NUMBER) {
throw std::runtime_error("Expected a number");
}
if (pos + 8 > buf.size()) {
throw std::runtime_error("Not enough data");
}
uint64_t val = ((uint64_t) load_be32(&buf[pos]) << 32)
| load_be32(&buf[pos + 4]);
double n = 0;
memcpy(&n, &val, 8);
pos += 8;
return n;
}
template<>
bool AMFDecoder::load<bool>() {
if (pop_front() != AMF0_BOOLEAN) {
throw std::runtime_error("Expected a boolean");
}
return pop_front() != 0;
if (pop_front() != AMF0_BOOLEAN) {
throw std::runtime_error("Expected a boolean");
}
return pop_front() != 0;
}
template<>
unsigned int AMFDecoder::load<unsigned int>() {
unsigned int value = 0;
for (int i = 0; i < 4; ++i) {
uint8_t b = pop_front();
if (i == 3) {
/* use all bits from 4th byte */
value = (value << 8) | b;
break;
}
value = (value << 7) | (b & 0x7f);
if ((b & 0x80) == 0)
break;
}
return value;
unsigned int value = 0;
for (int i = 0; i < 4; ++i) {
uint8_t b = pop_front();
if (i == 3) {
/* use all bits from 4th byte */
value = (value << 8) | b;
break;
}
value = (value << 7) | (b & 0x7f);
if ((b & 0x80) == 0)
break;
}
return value;
}
template<>
int AMFDecoder::load<int>() {
if (version == 3) {
return load<unsigned int>();
} else {
return load<double>();
}
if (version == 3) {
return load<unsigned int>();
} else {
return load<double>();
}
}
template<>
std::string AMFDecoder::load<std::string>() {
size_t str_len = 0;
uint8_t type = pop_front();
if (version == 3) {
if (type != AMF3_STRING) {
throw std::runtime_error("Expected a string");
}
str_len = load<unsigned int>() / 2;
size_t str_len = 0;
uint8_t type = pop_front();
if (version == 3) {
if (type != AMF3_STRING) {
throw std::runtime_error("Expected a string");
}
str_len = load<unsigned int>() / 2;
} else {
if (type != AMF0_STRING) {
throw std::runtime_error("Expected a string");
}
if (pos + 2 > buf.size()) {
throw std::runtime_error("Not enough data");
}
str_len = load_be16(&buf[pos]);
pos += 2;
}
if (pos + str_len > buf.size()) {
throw std::runtime_error("Not enough data");
}
std::string s(buf, pos, str_len);
pos += str_len;
return s;
} else {
if (type != AMF0_STRING) {
throw std::runtime_error("Expected a string");
}
if (pos + 2 > buf.size()) {
throw std::runtime_error("Not enough data");
}
str_len = load_be16(&buf[pos]);
pos += 2;
}
if (pos + str_len > buf.size()) {
throw std::runtime_error("Not enough data");
}
std::string s(buf, pos, str_len);
pos += str_len;
return s;
}
template<>
AMFValue AMFDecoder::load<AMFValue>() {
uint8_t type = front();
if (version == 3) {
switch (type) {
case AMF3_STRING:
return load<std::string>();
case AMF3_NUMBER:
return load<double>();
case AMF3_INTEGER:
return load<int>();
case AMF3_FALSE:
pos++;
return false;
case AMF3_TRUE:
pos++;
return true;
case AMF3_OBJECT:
return load_object();
case AMF3_ARRAY:
return load_ecma();
case AMF3_NULL:
pos++;
return AMF_NULL;
case AMF3_UNDEFINED:
pos++;
return AMF_UNDEFINED;
default:
throw std::runtime_error(
StrPrinter << "Unsupported AMF3 type:" << (int) type << endl);
}
} else {
switch (type) {
case AMF0_STRING:
return load<std::string>();
case AMF0_NUMBER:
return load<double>();
case AMF0_BOOLEAN:
return load<bool>();
case AMF0_OBJECT:
return load_object();
case AMF0_ECMA_ARRAY:
return load_ecma();
case AMF0_NULL:
pos++;
return AMF_NULL;
case AMF0_UNDEFINED:
pos++;
return AMF_UNDEFINED;
case AMF0_STRICT_ARRAY:
return load_arr();
default:
throw std::runtime_error(
StrPrinter << "Unsupported AMF type:" << (int) type << endl);
}
}
uint8_t type = front();
if (version == 3) {
switch (type) {
case AMF3_STRING:
return load<std::string>();
case AMF3_NUMBER:
return load<double>();
case AMF3_INTEGER:
return load<int>();
case AMF3_FALSE:
pos++;
return false;
case AMF3_TRUE:
pos++;
return true;
case AMF3_OBJECT:
return load_object();
case AMF3_ARRAY:
return load_ecma();
case AMF3_NULL:
pos++;
return AMF_NULL;
case AMF3_UNDEFINED:
pos++;
return AMF_UNDEFINED;
default:
throw std::runtime_error(
StrPrinter << "Unsupported AMF3 type:" << (int) type << endl);
}
} else {
switch (type) {
case AMF0_STRING:
return load<std::string>();
case AMF0_NUMBER:
return load<double>();
case AMF0_BOOLEAN:
return load<bool>();
case AMF0_OBJECT:
return load_object();
case AMF0_ECMA_ARRAY:
return load_ecma();
case AMF0_NULL:
pos++;
return AMF_NULL;
case AMF0_UNDEFINED:
pos++;
return AMF_UNDEFINED;
case AMF0_STRICT_ARRAY:
return load_arr();
default:
throw std::runtime_error(
StrPrinter << "Unsupported AMF type:" << (int) type << endl);
}
}
}
std::string AMFDecoder::load_key() {
if (pos + 2 > buf.size()) {
throw std::runtime_error("Not enough data");
}
size_t str_len = load_be16(&buf[pos]);
pos += 2;
if (pos + str_len > buf.size()) {
throw std::runtime_error("Not enough data");
}
std::string s(buf, pos, str_len);
pos += str_len;
return s;
if (pos + 2 > buf.size()) {
throw std::runtime_error("Not enough data");
}
size_t str_len = load_be16(&buf[pos]);
pos += 2;
if (pos + str_len > buf.size()) {
throw std::runtime_error("Not enough data");
}
std::string s(buf, pos, str_len);
pos += str_len;
return s;
}
AMFValue AMFDecoder::load_object() {
AMFValue object(AMF_OBJECT);
if (pop_front() != AMF0_OBJECT) {
throw std::runtime_error("Expected an object");
}
while (1) {
std::string key = load_key();
if (key.empty())
break;
AMFValue value = load<AMFValue>();
object.set(key, value);
}
if (pop_front() != AMF0_OBJECT_END) {
throw std::runtime_error("expected object end");
}
return object;
AMFValue object(AMF_OBJECT);
if (pop_front() != AMF0_OBJECT) {
throw std::runtime_error("Expected an object");
}
while (1) {
std::string key = load_key();
if (key.empty())
break;
AMFValue value = load<AMFValue>();
object.set(key, value);
}
if (pop_front() != AMF0_OBJECT_END) {
throw std::runtime_error("expected object end");
}
return object;
}
AMFValue AMFDecoder::load_ecma() {
/* ECMA array is the same as object, with 4 extra zero bytes */
AMFValue object(AMF_ECMA_ARRAY);
if (pop_front() != AMF0_ECMA_ARRAY) {
throw std::runtime_error("Expected an ECMA array");
}
if (pos + 4 > buf.size()) {
throw std::runtime_error("Not enough data");
}
pos += 4;
while (1) {
std::string key = load_key();
if (key.empty())
break;
AMFValue value = load<AMFValue>();
object.set(key, value);
}
if (pop_front() != AMF0_OBJECT_END) {
throw std::runtime_error("expected object end");
}
return object;
/* ECMA array is the same as object, with 4 extra zero bytes */
AMFValue object(AMF_ECMA_ARRAY);
if (pop_front() != AMF0_ECMA_ARRAY) {
throw std::runtime_error("Expected an ECMA array");
}
if (pos + 4 > buf.size()) {
throw std::runtime_error("Not enough data");
}
pos += 4;
while (1) {
std::string key = load_key();
if (key.empty())
break;
AMFValue value = load<AMFValue>();
object.set(key, value);
}
if (pop_front() != AMF0_OBJECT_END) {
throw std::runtime_error("expected object end");
}
return object;
}
AMFValue AMFDecoder::load_arr() {
/* ECMA array is the same as object, with 4 extra zero bytes */
AMFValue object(AMF_STRICT_ARRAY);
if (pop_front() != AMF0_STRICT_ARRAY) {
throw std::runtime_error("Expected an STRICT array");
}
if (pos + 4 > buf.size()) {
throw std::runtime_error("Not enough data");
}
int arrSize = load_be32(&buf[pos]);
pos += 4;
while (arrSize--) {
AMFValue value = load<AMFValue>();
object.add(value);
}
/*pos += 2;
if (pop_front() != AMF0_OBJECT_END) {
throw std::runtime_error("expected object end");
}*/
return object;
/* ECMA array is the same as object, with 4 extra zero bytes */
AMFValue object(AMF_STRICT_ARRAY);
if (pop_front() != AMF0_STRICT_ARRAY) {
throw std::runtime_error("Expected an STRICT array");
}
if (pos + 4 > buf.size()) {
throw std::runtime_error("Not enough data");
}
int arrSize = load_be32(&buf[pos]);
pos += 4;
while (arrSize--) {
AMFValue value = load<AMFValue>();
object.add(value);
}
/*pos += 2;
if (pop_front() != AMF0_OBJECT_END) {
throw std::runtime_error("expected object end");
}*/
return object;
}
AMFDecoder::AMFDecoder(const std::string &buf_in, size_t pos_in, int version_in) :

View File

@@ -36,15 +36,15 @@
using namespace std;
enum AMFType {
AMF_NUMBER,
AMF_INTEGER,
AMF_BOOLEAN,
AMF_STRING,
AMF_OBJECT,
AMF_NULL,
AMF_UNDEFINED,
AMF_ECMA_ARRAY,
AMF_STRICT_ARRAY,
AMF_NUMBER,
AMF_INTEGER,
AMF_BOOLEAN,
AMF_STRING,
AMF_OBJECT,
AMF_NULL,
AMF_UNDEFINED,
AMF_ECMA_ARRAY,
AMF_STRICT_ARRAY,
};
class AMFValue;
@@ -55,79 +55,79 @@ public:
typedef std::map<std::string, AMFValue> mapType;
typedef std::vector<AMFValue> arrayType;
AMFValue(AMFType type = AMF_NULL);
AMFValue(const char *s);
AMFValue(const std::string &s);
AMFValue(double n);
AMFValue(int i);
AMFValue(bool b);
AMFValue(const AMFValue &from);
AMFValue(AMFValue &&from);
AMFValue &operator =(const AMFValue &from);
AMFValue &operator =(AMFValue &&from);
~AMFValue();
AMFValue(AMFType type = AMF_NULL);
AMFValue(const char *s);
AMFValue(const std::string &s);
AMFValue(double n);
AMFValue(int i);
AMFValue(bool b);
AMFValue(const AMFValue &from);
AMFValue(AMFValue &&from);
AMFValue &operator =(const AMFValue &from);
AMFValue &operator =(AMFValue &&from);
~AMFValue();
void clear();
AMFType type() const ;
const std::string &as_string() const;
double as_number() const;
int as_integer() const;
void clear();
AMFType type() const ;
const std::string &as_string() const;
double as_number() const;
int as_integer() const;
bool as_boolean() const;
string to_string() const;
const AMFValue &operator[](const char *str) const;
void object_for_each(const function<void(const string &key, const AMFValue &val)> &fun) const ;
operator bool() const;
void set(const std::string &s, const AMFValue &val);
void add(const AMFValue &val);
string to_string() const;
const AMFValue &operator[](const char *str) const;
void object_for_each(const function<void(const string &key, const AMFValue &val)> &fun) const ;
operator bool() const;
void set(const std::string &s, const AMFValue &val);
void add(const AMFValue &val);
private:
const mapType &getMap() const;
const arrayType &getArr() const;
void destroy();
void init();
private:
AMFType _type;
union {
std::string *string;
double number;
int integer;
bool boolean;
mapType *object;
arrayType *array;
} _value;
AMFType _type;
union {
std::string *string;
double number;
int integer;
bool boolean;
mapType *object;
arrayType *array;
} _value;
};
class AMFDecoder {
public:
AMFDecoder(const std::string &buf, size_t pos, int version = 0);
template<typename TP>
TP load();
AMFDecoder(const std::string &buf, size_t pos, int version = 0);
template<typename TP>
TP load();
private:
std::string load_key();
AMFValue load_object();
AMFValue load_ecma();
AMFValue load_arr();
uint8_t front();
uint8_t pop_front();
std::string load_key();
AMFValue load_object();
AMFValue load_ecma();
AMFValue load_arr();
uint8_t front();
uint8_t pop_front();
private:
const std::string &buf;
size_t pos;
int version;
const std::string &buf;
size_t pos;
int version;
};
class AMFEncoder {
public:
AMFEncoder & operator <<(const char *s);
AMFEncoder & operator <<(const std::string &s);
AMFEncoder & operator <<(std::nullptr_t);
AMFEncoder & operator <<(const int n);
AMFEncoder & operator <<(const double n);
AMFEncoder & operator <<(const bool b);
AMFEncoder & operator <<(const AMFValue &value);
const std::string& data() const ;
void clear() ;
AMFEncoder & operator <<(const char *s);
AMFEncoder & operator <<(const std::string &s);
AMFEncoder & operator <<(std::nullptr_t);
AMFEncoder & operator <<(const int n);
AMFEncoder & operator <<(const double n);
AMFEncoder & operator <<(const bool b);
AMFEncoder & operator <<(const AMFValue &value);
const std::string& data() const ;
void clear() ;
private:
void write_key(const std::string &s);
AMFEncoder &write_undefined();
void write_key(const std::string &s);
AMFEncoder &write_undefined();
private:
std::string buf;
};

View File

@@ -38,54 +38,54 @@ using namespace toolkit;
*/
uint32_t load_be32(const void *p)
{
uint32_t val;
memcpy(&val, p, sizeof val);
return ntohl(val);
uint32_t val;
memcpy(&val, p, sizeof val);
return ntohl(val);
}
uint16_t load_be16(const void *p)
{
uint16_t val;
memcpy(&val, p, sizeof val);
return ntohs(val);
uint16_t val;
memcpy(&val, p, sizeof val);
return ntohs(val);
}
uint32_t load_le32(const void *p)
{
const uint8_t *data = (const uint8_t *) p;
return data[0] | ((uint32_t) data[1] << 8) |
((uint32_t) data[2] << 16) | ((uint32_t) data[3] << 24);
const uint8_t *data = (const uint8_t *) p;
return data[0] | ((uint32_t) data[1] << 8) |
((uint32_t) data[2] << 16) | ((uint32_t) data[3] << 24);
}
uint32_t load_be24(const void *p)
{
const uint8_t *data = (const uint8_t *) p;
return data[2] | ((uint32_t) data[1] << 8) | ((uint32_t) data[0] << 16);
const uint8_t *data = (const uint8_t *) p;
return data[2] | ((uint32_t) data[1] << 8) | ((uint32_t) data[0] << 16);
}
void set_be24(void *p, uint32_t val)
{
uint8_t *data = (uint8_t *) p;
data[0] = val >> 16;
data[1] = val >> 8;
data[2] = val;
uint8_t *data = (uint8_t *) p;
data[0] = val >> 16;
data[1] = val >> 8;
data[2] = val;
}
void set_le32(void *p, uint32_t val)
{
uint8_t *data = (uint8_t *) p;
data[0] = val;
data[1] = val >> 8;
data[2] = val >> 16;
data[3] = val >> 24;
uint8_t *data = (uint8_t *) p;
data[0] = val;
data[1] = val >> 8;
data[2] = val >> 16;
data[3] = val >> 24;
}
void set_be32(void *p, uint32_t val)
{
uint8_t *data = (uint8_t *) p;
data[3] = val;
data[2] = val >> 8;
data[1] = val >> 16;
data[0] = val >> 24;
uint8_t *data = (uint8_t *) p;
data[3] = val;
data[2] = val >> 8;
data[1] = val >> 16;
data[0] = val >> 24;
}