THENG
/
ZLMediaKit
mirror of https://gitcode.com/gh_mirrors/zlme/ZLMediaKit.git
6
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Rtmp头文件重构

This commit is contained in:
cqm 2022-11-23 10:05:33 +08:00
parent 2889335803
commit 27fd74d896
16 changed files with 375 additions and 305 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
src/Rtmp/FlvMuxer.h
View File

@ -13,8 +13,7 @@
#include "Rtmp/Rtmp.h"
#include "Rtmp/RtmpMediaSource.h"
#include "Network/Socket.h"
#include "Common/Stamp.h"
#include "Poller/EventPoller.h"
namespace mediakit {

112
src/Rtmp/Rtmp.cpp
View File

@ -12,6 +12,16 @@
#include "Extension/Factory.h"
namespace mediakit{
TitleMeta::TitleMeta(float dur_sec, size_t fileSize, const std::map<std::string, std::string> &header)
{
_metadata.set("duration", dur_sec);
_metadata.set("fileSize", (int)fileSize);
_metadata.set("server", kServerName);
for (auto &pr : header) {
_metadata.set(pr.first, pr.second);
}
}
VideoMeta::VideoMeta(const VideoTrack::Ptr &video){
if(video->getVideoWidth() > 0 ){
_metadata.set("width", video->getVideoWidth());
@ -146,6 +156,108 @@ RtmpPacket::Ptr RtmpPacket::create(){
#endif
}
void RtmpPacket::clear()
{
is_abs_stamp = false;
time_stamp = 0;
ts_field = 0;
body_size = 0;
buffer.clear();
}
bool RtmpPacket::isVideoKeyFrame() const
{
return type_id == MSG_VIDEO && (uint8_t)buffer[0] >> 4 == FLV_KEY_FRAME && (uint8_t)buffer[1] == 1;
}
bool RtmpPacket::isCfgFrame() const
{
switch (type_id) {
case MSG_VIDEO: return buffer[1] == 0;
case MSG_AUDIO: {
switch (getMediaType()) {
case FLV_CODEC_AAC: return buffer[1] == 0;
default: return false;
}
}
default: return false;
}
}
int RtmpPacket::getMediaType() const
{
switch (type_id) {
case MSG_VIDEO: return (uint8_t)buffer[0] & 0x0F;
case MSG_AUDIO: return (uint8_t)buffer[0] >> 4;
default: return 0;
}
}
int RtmpPacket::getAudioSampleRate() const
{
if (type_id != MSG_AUDIO) {
return 0;
}
int flvSampleRate = ((uint8_t)buffer[0] & 0x0C) >> 2;
const static int sampleRate[] = { 5512, 11025, 22050, 44100 };
return sampleRate[flvSampleRate];
}
int RtmpPacket::getAudioSampleBit() const
{
if (type_id != MSG_AUDIO) {
return 0;
}
int flvSampleBit = ((uint8_t)buffer[0] & 0x02) >> 1;
const static int sampleBit[] = { 8, 16 };
return sampleBit[flvSampleBit];
}
int RtmpPacket::getAudioChannel() const
{
if (type_id != MSG_AUDIO) {
return 0;
}
int flvStereoOrMono = (uint8_t)buffer[0] & 0x01;
const static int channel[] = { 1, 2 };
return channel[flvStereoOrMono];
}
RtmpPacket & RtmpPacket::operator=(const RtmpPacket &that)
{
is_abs_stamp = that.is_abs_stamp;
stream_index = that.stream_index;
body_size = that.body_size;
type_id = that.type_id;
ts_field = that.ts_field;
time_stamp = that.time_stamp;
return *this;
}
RtmpHandshake::RtmpHandshake(uint32_t _time, uint8_t *_random /*= nullptr*/)
{
_time = htonl(_time);
memcpy(time_stamp, &_time, 4);
if (!_random) {
random_generate((char *)random, sizeof(random));
}
else {
memcpy(random, _random, sizeof(random));
}
}
void RtmpHandshake::random_generate(char *bytes, int size)
{
static char cdata[] = { 0x73, 0x69, 0x6d, 0x70, 0x6c, 0x65, 0x2d, 0x72,
0x74, 0x6d, 0x70, 0x2d, 0x73, 0x65, 0x72, 0x76, 0x65, 0x72,
0x2d, 0x77, 0x69, 0x6e, 0x6c, 0x69, 0x6e, 0x2d, 0x77, 0x69,
0x6e, 0x74, 0x65, 0x72, 0x73, 0x65, 0x72, 0x76, 0x65, 0x72,
0x40, 0x31, 0x32, 0x36, 0x2e, 0x63, 0x6f, 0x6d };
for (int i = 0; i < size; i++) {
bytes[i] = cdata[rand() % (sizeof(cdata) - 1)];
}
}
}//namespace mediakit
namespace toolkit {

108
src/Rtmp/Rtmp.h
View File

@ -14,11 +14,8 @@
#include <memory>
#include <string>
#include <cstdlib>
#include "Util/util.h"
#include "Util/logger.h"
#include "Network/Buffer.h"
#include "Network/sockutil.h"
#include "amf.h"
#include "Network/Buffer.h"
#include "Extension/Track.h"
#if !defined(_WIN32)
@ -86,30 +83,13 @@ namespace mediakit {
class RtmpHandshake {
public:
RtmpHandshake(uint32_t _time, uint8_t *_random = nullptr) {
_time = htonl(_time);
memcpy(time_stamp, &_time, 4);
if (!_random) {
random_generate((char *) random, sizeof(random));
} else {
memcpy(random, _random, sizeof(random));
}
}
RtmpHandshake(uint32_t _time, uint8_t *_random = nullptr);
uint8_t time_stamp[4];
uint8_t zero[4] = {0};
uint8_t random[RANDOM_LEN];
void random_generate(char *bytes, int size) {
static char cdata[] = {0x73, 0x69, 0x6d, 0x70, 0x6c, 0x65, 0x2d, 0x72,
0x74, 0x6d, 0x70, 0x2d, 0x73, 0x65, 0x72, 0x76, 0x65, 0x72,
0x2d, 0x77, 0x69, 0x6e, 0x6c, 0x69, 0x6e, 0x2d, 0x77, 0x69,
0x6e, 0x74, 0x65, 0x72, 0x73, 0x65, 0x72, 0x76, 0x65, 0x72,
0x40, 0x31, 0x32, 0x36, 0x2e, 0x63, 0x6f, 0x6d};
for (int i = 0; i < size; i++) {
bytes[i] = cdata[rand() % (sizeof(cdata) - 1)];
}
}
void random_generate(char *bytes, int size);
void create_complex_c0c1();
@ -196,65 +176,16 @@ public:
return buffer.size();
}
void clear(){
is_abs_stamp = false;
time_stamp = 0;
ts_field = 0;
body_size = 0;
buffer.clear();
}
void clear();
bool isVideoKeyFrame() const {
return type_id == MSG_VIDEO && (uint8_t) buffer[0] >> 4 == FLV_KEY_FRAME && (uint8_t) buffer[1] == 1;
}
bool isVideoKeyFrame() const;
bool isCfgFrame() const;
bool isCfgFrame() const {
switch (type_id){
case MSG_VIDEO : return buffer[1] == 0;
case MSG_AUDIO : {
switch (getMediaType()){
case FLV_CODEC_AAC : return buffer[1] == 0;
default : return false;
}
}
default : return false;
}
}
int getMediaType() const;
int getMediaType() const {
switch (type_id) {
case MSG_VIDEO : return (uint8_t) buffer[0] & 0x0F;
case MSG_AUDIO : return (uint8_t) buffer[0] >> 4;
default : return 0;
}
}
int getAudioSampleRate() const {
if (type_id != MSG_AUDIO) {
return 0;
}
int flvSampleRate = ((uint8_t) buffer[0] & 0x0C) >> 2;
const static int sampleRate[] = { 5512, 11025, 22050, 44100 };
return sampleRate[flvSampleRate];
}
int getAudioSampleBit() const {
if (type_id != MSG_AUDIO) {
return 0;
}
int flvSampleBit = ((uint8_t) buffer[0] & 0x02) >> 1;
const static int sampleBit[] = { 8, 16 };
return sampleBit[flvSampleBit];
}
int getAudioChannel() const {
if (type_id != MSG_AUDIO) {
return 0;
}
int flvStereoOrMono = (uint8_t) buffer[0] & 0x01;
const static int channel[] = { 1, 2 };
return channel[flvStereoOrMono];
}
int getAudioSampleRate() const;
int getAudioSampleBit() const;
int getAudioChannel() const;
private:
friend class toolkit::ResourcePool_l<RtmpPacket>;
@ -262,15 +193,7 @@ private:
clear();
}
RtmpPacket &operator=(const RtmpPacket &that) {
is_abs_stamp = that.is_abs_stamp;
stream_index = that.stream_index;
body_size = that.body_size;
type_id = that.type_id;
ts_field = that.ts_field;
time_stamp = that.time_stamp;
return *this;
}
RtmpPacket &operator=(const RtmpPacket &that);
private:
//对象个数统计
@ -304,14 +227,7 @@ public:
TitleMeta(float dur_sec = 0,
size_t fileSize = 0,
const std::map<std::string, std::string> &header = std::map<std::string, std::string>()){
_metadata.set("duration", dur_sec);
_metadata.set("fileSize", (int)fileSize);
_metadata.set("server",kServerName);
for (auto &pr : header){
_metadata.set(pr.first, pr.second);
}
}
const std::map<std::string, std::string> &header = std::map<std::string, std::string>());
CodecId getCodecId() const override{
return CodecInvalid;

2
src/Rtmp/RtmpCodec.h
View File

@ -53,7 +53,7 @@ protected:
RingType::Ptr _ring;
};
class RtmpCodec : public RtmpRing, public FrameDispatcher , public CodecInfo{
class RtmpCodec : public RtmpRing, public FrameDispatcher, public CodecInfo {
public:
typedef std::shared_ptr<RtmpCodec> Ptr;
RtmpCodec() = default;

64
src/Rtmp/RtmpMediaSource.h
View File

@ -18,15 +18,9 @@
#include <unordered_map>
#include "amf.h"
#include "Rtmp.h"
#include "RtmpDemuxer.h"
#include "Common/config.h"
#include "Common/MediaSource.h"
#include "Common/PacketCache.h"
#include "Util/RingBuffer.h"
#include "Util/TimeTicker.h"
#include "Util/ResourcePool.h"
#include "Util/NoticeCenter.h"
#include "Thread/ThreadPool.h"
#define RTMP_GOP_SIZE 512
@ -125,66 +119,12 @@ public:
* 输入rtmp包
* @param pkt rtmp包
*/
void onWrite(RtmpPacket::Ptr pkt, bool = true) override {
bool is_video = pkt->type_id == MSG_VIDEO;
_speed[is_video ? TrackVideo : TrackAudio] += pkt->size();
//保存当前时间戳
switch (pkt->type_id) {
case MSG_VIDEO : _track_stamps[TrackVideo] = pkt->time_stamp, _have_video = true; break;
case MSG_AUDIO : _track_stamps[TrackAudio] = pkt->time_stamp, _have_audio = true; break;
default : break;
}
if (pkt->isCfgFrame()) {
std::lock_guard<std::recursive_mutex> lock(_mtx);
_config_frame_map[pkt->type_id] = pkt;
if (!_ring) {
//注册后收到config帧更新到各播放器
return;
}
}
if (!_ring) {
std::weak_ptr<RtmpMediaSource> weakSelf = std::dynamic_pointer_cast<RtmpMediaSource>(shared_from_this());
auto lam = [weakSelf](int size) {
auto strongSelf = weakSelf.lock();
if (!strongSelf) {
return;
}
strongSelf->onReaderChanged(size);
};
//GOP默认缓冲512组RTMP包，每组RTMP包时间戳相同(如果开启合并写了，那么每组为合并写时间内的RTMP包),
//每次遇到关键帧第一个RTMP包，则会清空GOP缓存(因为有新的关键帧了，同样可以实现秒开)
_ring = std::make_shared<RingType>(_ring_size,std::move(lam));
if(_metadata){
regist();
}
}
bool key = pkt->isVideoKeyFrame();
auto stamp = pkt->time_stamp;
PacketCache<RtmpPacket>::inputPacket(stamp, is_video, std::move(pkt), key);
}
void onWrite(RtmpPacket::Ptr pkt, bool = true) override;
/**
* 获取当前时间戳
*/
uint32_t getTimeStamp(TrackType trackType) override {
assert(trackType >= TrackInvalid && trackType < TrackMax);
if (trackType != TrackInvalid) {
//获取某track的时间戳
return _track_stamps[trackType];
}
//获取所有track的最小时间戳
uint32_t ret = UINT32_MAX;
for (auto &stamp : _track_stamps) {
if (stamp > 0 && stamp < ret) {
ret = stamp;
}
}
return ret;
}
uint32_t getTimeStamp(TrackType trackType) override;
void clearCache() override{
PacketCache<RtmpPacket>::clearCache();

166
src/Rtmp/RtmpMediaSourceImp.cpp Normal file
View File

@ -0,0 +1,166 @@
#include "RtmpDemuxer.h"
#include "RtmpMediaSourceImp.h"
namespace mediakit {
uint32_t RtmpMediaSource::getTimeStamp(TrackType trackType)
{
assert(trackType >= TrackInvalid && trackType < TrackMax);
if (trackType != TrackInvalid) {
//获取某track的时间戳
return _track_stamps[trackType];
}
//获取所有track的最小时间戳
uint32_t ret = UINT32_MAX;
for (auto &stamp : _track_stamps) {
if (stamp > 0 && stamp < ret) {
ret = stamp;
}
}
return ret;
}
void RtmpMediaSource::onWrite(RtmpPacket::Ptr pkt, bool /*= true*/)
{
bool is_video = pkt->type_id == MSG_VIDEO;
_speed[is_video ? TrackVideo : TrackAudio] += pkt->size();
//保存当前时间戳
switch (pkt->type_id) {
case MSG_VIDEO: _track_stamps[TrackVideo] = pkt->time_stamp, _have_video = true; break;
case MSG_AUDIO: _track_stamps[TrackAudio] = pkt->time_stamp, _have_audio = true; break;
default: break;
}
if (pkt->isCfgFrame()) {
std::lock_guard<std::recursive_mutex> lock(_mtx);
_config_frame_map[pkt->type_id] = pkt;
if (!_ring) {
//注册后收到config帧更新到各播放器
return;
}
}
if (!_ring) {
std::weak_ptr<RtmpMediaSource> weakSelf = std::dynamic_pointer_cast<RtmpMediaSource>(shared_from_this());
auto lam = [weakSelf](int size) {
auto strongSelf = weakSelf.lock();
if (!strongSelf) {
return;
}
strongSelf->onReaderChanged(size);
};
//GOP默认缓冲512组RTMP包，每组RTMP包时间戳相同(如果开启合并写了，那么每组为合并写时间内的RTMP包),
//每次遇到关键帧第一个RTMP包，则会清空GOP缓存(因为有新的关键帧了，同样可以实现秒开)
_ring = std::make_shared<RingType>(_ring_size, std::move(lam));
if (_metadata) {
regist();
}
}
bool key = pkt->isVideoKeyFrame();
auto stamp = pkt->time_stamp;
PacketCache<RtmpPacket>::inputPacket(stamp, is_video, std::move(pkt), key);
}
RtmpMediaSourceImp::RtmpMediaSourceImp(const std::string &vhost, const std::string &app, const std::string &id, int ringSize) : RtmpMediaSource(vhost, app, id, ringSize)
{
_demuxer = std::make_shared<RtmpDemuxer>();
_demuxer->setTrackListener(this);
}
void RtmpMediaSourceImp::setMetaData(const AMFValue &metadata)
{
if (!_demuxer->loadMetaData(metadata)) {
//该metadata无效，需要重新生成
_metadata = metadata;
_recreate_metadata = true;
}
RtmpMediaSource::setMetaData(metadata);
}
void RtmpMediaSourceImp::onWrite(RtmpPacket::Ptr pkt, bool /*= true*/)
{
if (!_all_track_ready || _muxer->isEnabled()) {
//未获取到所有Track后，或者开启转协议，那么需要解复用rtmp
_demuxer->inputRtmp(pkt);
}
RtmpMediaSource::onWrite(std::move(pkt));
}
int RtmpMediaSourceImp::totalReaderCount()
{
return readerCount() + (_muxer ? _muxer->totalReaderCount() : 0);
}
void RtmpMediaSourceImp::setProtocolOption(const ProtocolOption &option)
{
//不重复生成rtmp
_option = option;
//不重复生成rtmp协议
_option.enable_rtmp = false;
_muxer = std::make_shared<MultiMediaSourceMuxer>(getVhost(), getApp(), getId(), _demuxer->getDuration(), _option);
_muxer->setMediaListener(getListener());
_muxer->setTrackListener(std::static_pointer_cast<RtmpMediaSourceImp>(shared_from_this()));
//让_muxer对象拦截一部分事件(比如说录像相关事件)
MediaSource::setListener(_muxer);
for (auto &track : _demuxer->getTracks(false)) {
_muxer->addTrack(track);
track->addDelegate(_muxer);
}
}
bool RtmpMediaSourceImp::addTrack(const Track::Ptr &track)
{
if (_muxer) {
if (_muxer->addTrack(track)) {
track->addDelegate(_muxer);
return true;
}
}
return false;
}
void RtmpMediaSourceImp::addTrackCompleted()
{
if (_muxer) {
_muxer->addTrackCompleted();
}
}
void RtmpMediaSourceImp::resetTracks()
{
if (_muxer) {
_muxer->resetTracks();
}
}
void RtmpMediaSourceImp::onAllTrackReady()
{
_all_track_ready = true;
if (_recreate_metadata) {
//更新metadata
for (auto &track : _muxer->getTracks()) {
Metadata::addTrack(_metadata, track);
}
RtmpMediaSource::updateMetaData(_metadata);
}
}
void RtmpMediaSourceImp::setListener(const std::weak_ptr<MediaSourceEvent> &listener)
{
if (_muxer) {
//_muxer对象不能处理的事件再给listener处理
_muxer->setMediaListener(listener);
}
else {
//未创建_muxer对象，事件全部给listener处理
MediaSource::setListener(listener);
}
}
}

94
src/Rtmp/RtmpMediaSourceImp.h
View File

@ -16,16 +16,13 @@
#include <memory>
#include <functional>
#include <unordered_map>
#include "Util/util.h"
#include "Util/logger.h"
#include "amf.h"
#include "Rtmp.h"
#include "RtmpMediaSource.h"
#include "RtmpDemuxer.h"
#include "Common/MultiMediaSourceMuxer.h"
namespace mediakit {
class RtmpDemuxer;
class RtmpMediaSourceImp final : public RtmpMediaSource, private TrackListener, public MultiMediaSourceMuxer::Listener {
public:
using Ptr = std::shared_ptr<RtmpMediaSourceImp>;
@ -37,62 +34,29 @@ public:
* @param id 流id
* @param ringSize 环形缓存大小
*/
RtmpMediaSourceImp(const std::string &vhost, const std::string &app, const std::string &id, int ringSize = RTMP_GOP_SIZE) : RtmpMediaSource(vhost, app, id, ringSize) {
_demuxer = std::make_shared<RtmpDemuxer>();
_demuxer->setTrackListener(this);
}
RtmpMediaSourceImp(const std::string &vhost, const std::string &app, const std::string &id, int ringSize = RTMP_GOP_SIZE);
~RtmpMediaSourceImp() override = default;
/**
* 设置metadata
*/
void setMetaData(const AMFValue &metadata) override{
if(!_demuxer->loadMetaData(metadata)){
//该metadata无效，需要重新生成
_metadata = metadata;
_recreate_metadata = true;
}
RtmpMediaSource::setMetaData(metadata);
}
void setMetaData(const AMFValue &metadata) override;
/**
* 输入rtmp并解析
*/
void onWrite(RtmpPacket::Ptr pkt, bool = true) override {
if (!_all_track_ready || _muxer->isEnabled()) {
//未获取到所有Track后，或者开启转协议，那么需要解复用rtmp
_demuxer->inputRtmp(pkt);
}
RtmpMediaSource::onWrite(std::move(pkt));
}
void onWrite(RtmpPacket::Ptr pkt, bool = true) override;
/**
* 获取观看总人数，包括(hls/rtsp/rtmp)
*/
int totalReaderCount() override{
return readerCount() + (_muxer ? _muxer->totalReaderCount() : 0);
}
int totalReaderCount() override;
/**
* 设置协议转换
*/
void setProtocolOption(const ProtocolOption &option) {
//不重复生成rtmp
_option = option;
//不重复生成rtmp协议
_option.enable_rtmp = false;
_muxer = std::make_shared<MultiMediaSourceMuxer>(getVhost(), getApp(), getId(), _demuxer->getDuration(), _option);
_muxer->setMediaListener(getListener());
_muxer->setTrackListener(std::static_pointer_cast<RtmpMediaSourceImp>(shared_from_this()));
//让_muxer对象拦截一部分事件(比如说录像相关事件)
MediaSource::setListener(_muxer);
for (auto &track : _demuxer->getTracks(false)) {
_muxer->addTrack(track);
track->addDelegate(_muxer);
}
}
void setProtocolOption(const ProtocolOption &option);
const ProtocolOption &getProtocolOption() const {
return _option;
@ -101,66 +65,32 @@ public:
/**
* _demuxer触发的添加Track事件
*/
bool addTrack(const Track::Ptr &track) override {
if (_muxer) {
if (_muxer->addTrack(track)) {
track->addDelegate(_muxer);
return true;
}
}
return false;
}
bool addTrack(const Track::Ptr &track) override;
/**
* _demuxer触发的Track添加完毕事件
*/
void addTrackCompleted() override {
if (_muxer) {
_muxer->addTrackCompleted();
}
}
void addTrackCompleted() override;
void resetTracks() override {
if (_muxer) {
_muxer->resetTracks();
}
}
void resetTracks() override;
/**
* _muxer触发的所有Track就绪的事件
*/
void onAllTrackReady() override{
_all_track_ready = true;
if (_recreate_metadata) {
//更新metadata
for (auto &track : _muxer->getTracks()) {
Metadata::addTrack(_metadata, track);
}
RtmpMediaSource::updateMetaData(_metadata);
}
}
void onAllTrackReady() override;
/**
* 设置事件监听器
* @param listener 监听器
*/
void setListener(const std::weak_ptr<MediaSourceEvent> &listener) override{
if (_muxer) {
//_muxer对象不能处理的事件再给listener处理
_muxer->setMediaListener(listener);
} else {
//未创建_muxer对象，事件全部给listener处理
MediaSource::setListener(listener);
}
}
void setListener(const std::weak_ptr<MediaSourceEvent> &listener) override;
private:
bool _all_track_ready = false;
bool _recreate_metadata = false;
ProtocolOption _option;
AMFValue _metadata;
RtmpDemuxer::Ptr _demuxer;
std::shared_ptr<RtmpDemuxer> _demuxer;
MultiMediaSourceMuxer::Ptr _muxer;
};

56
src/Rtmp/RtmpPlayer.cpp
View File

@ -16,6 +16,9 @@
#include "Common/config.h"
#include "Common/Parser.h"
#include "RtmpDemuxer.h"
#include "RtmpPlayerImp.h"
using namespace toolkit;
using namespace std;
@ -174,7 +177,7 @@ void RtmpPlayer::speed(float speed) {
//todo
}
inline void RtmpPlayer::send_connect() {
void RtmpPlayer::send_connect() {
AMFValue obj(AMF_OBJECT);
obj.set("app", _app);
obj.set("tcUrl", _tc_url);
@ -202,7 +205,7 @@ inline void RtmpPlayer::send_connect() {
});
}
inline void RtmpPlayer::send_createStream() {
void RtmpPlayer::send_createStream() {
AMFValue obj(AMF_NULL);
sendInvoke("createStream", obj);
addOnResultCB([this](AMFDecoder &dec) {
@ -213,7 +216,7 @@ inline void RtmpPlayer::send_createStream() {
});
}
inline void RtmpPlayer::send_play() {
void RtmpPlayer::send_play() {
AMFEncoder enc;
enc << "play" << ++_send_req_id << nullptr << _stream_id << -2000;
sendRequest(MSG_CMD, enc.data());
@ -229,7 +232,7 @@ inline void RtmpPlayer::send_play() {
addOnStatusCB(fun);
}
inline void RtmpPlayer::send_pause(bool pause) {
void RtmpPlayer::send_pause(bool pause) {
AMFEncoder enc;
enc << "pause" << ++_send_req_id << nullptr << pause;
sendRequest(MSG_CMD, enc.data());
@ -417,4 +420,49 @@ void RtmpPlayer::seekToMilliSecond(uint32_t seekMS){
});
}
////////////////////////////////////////////
float RtmpPlayerImp::getDuration() const
{
return _demuxer ? _demuxer->getDuration() : 0;
}
std::vector<mediakit::Track::Ptr> RtmpPlayerImp::getTracks(bool ready /*= true*/) const
{
return _demuxer ? _demuxer->getTracks(ready) : Super::getTracks(ready);
}
bool RtmpPlayerImp::onCheckMeta(const AMFValue &val)
{
//无metadata或metadata中无track信息时，需要从数据包中获取track
_wait_track_ready = (*this)[Client::kWaitTrackReady].as<bool>() || RtmpDemuxer::trackCount(val) == 0;
onCheckMeta_l(val);
return true;
}
void RtmpPlayerImp::onMediaData(RtmpPacket::Ptr chunkData)
{
if (!_demuxer) {
//有些rtmp流没metadata
onCheckMeta_l(TitleMeta().getMetadata());
}
_demuxer->inputRtmp(chunkData);
if (_rtmp_src) {
_rtmp_src->onWrite(std::move(chunkData));
}
}
void RtmpPlayerImp::onCheckMeta_l(const AMFValue &val)
{
_rtmp_src = std::dynamic_pointer_cast<RtmpMediaSource>(_media_src);
if (_rtmp_src) {
_rtmp_src->setMetaData(val);
}
if (_demuxer) {
return;
}
_demuxer = std::make_shared<RtmpDemuxer>();
//TraceL<<" _wait_track_ready "<<_wait_track_ready;
_demuxer->setTrackListener(this, _wait_track_ready);
_demuxer->loadMetaData(val);
}
} /* namespace mediakit */

2
src/Rtmp/RtmpPlayer.h
View File

@ -18,8 +18,6 @@
#include "Rtmp.h"
#include "RtmpProtocol.h"
#include "Player/PlayerBase.h"
#include "Util/util.h"
#include "Util/logger.h"
#include "Util/TimeTicker.h"
#include "Network/Socket.h"
#include "Network/TcpClient.h"

48
src/Rtmp/RtmpPlayerImp.h
View File

@ -13,15 +13,11 @@
#include <memory>
#include <functional>
#include "Common/config.h"
#include "RtmpPlayer.h"
#include "RtmpMediaSource.h"
#include "RtmpDemuxer.h"
#include "Poller/Timer.h"
#include "Util/TimeTicker.h"
namespace mediakit {
class RtmpDemuxer;
class RtmpPlayerImp: public PlayerImp<RtmpPlayer,PlayerBase>, private TrackListener {
public:
using Ptr = std::shared_ptr<RtmpPlayerImp>;
@ -50,33 +46,15 @@ public:
seekToMilliSecond(pos);
}
float getDuration() const override {
return _demuxer ? _demuxer->getDuration() : 0;
}
float getDuration() const override;
std::vector<Track::Ptr> getTracks(bool ready = true) const override {
return _demuxer ? _demuxer->getTracks(ready) : Super::getTracks(ready);
}
std::vector<Track::Ptr> getTracks(bool ready = true) const override;
private:
//派生类回调函数
bool onCheckMeta(const AMFValue &val) override {
//无metadata或metadata中无track信息时，需要从数据包中获取track
_wait_track_ready = (*this)[Client::kWaitTrackReady].as<bool>() || RtmpDemuxer::trackCount(val) == 0;
onCheckMeta_l(val);
return true;
}
bool onCheckMeta(const AMFValue &val) override;
void onMediaData(RtmpPacket::Ptr chunkData) override {
if (!_demuxer) {
//有些rtmp流没metadata
onCheckMeta_l(TitleMeta().getMetadata());
}
_demuxer->inputRtmp(chunkData);
if (_rtmp_src) {
_rtmp_src->onWrite(std::move(chunkData));
}
}
void onMediaData(RtmpPacket::Ptr chunkData) override;
void onPlayResult(const toolkit::SockException &ex) override {
if (!_wait_track_ready || ex) {
@ -94,23 +72,11 @@ private:
}
private:
void onCheckMeta_l(const AMFValue &val) {
_rtmp_src = std::dynamic_pointer_cast<RtmpMediaSource>(_media_src);
if (_rtmp_src) {
_rtmp_src->setMetaData(val);
}
if(_demuxer){
return;
}
_demuxer = std::make_shared<RtmpDemuxer>();
//TraceL<<" _wait_track_ready "<<_wait_track_ready;
_demuxer->setTrackListener(this, _wait_track_ready);
_demuxer->loadMetaData(val);
}
void onCheckMeta_l(const AMFValue &val);
private:
bool _wait_track_ready = true;
RtmpDemuxer::Ptr _demuxer;
std::shared_ptr<RtmpDemuxer> _demuxer;
RtmpMediaSource::Ptr _rtmp_src;
};

1
src/Rtmp/RtmpProtocol.cpp
View File

@ -12,6 +12,7 @@
#include "RtmpProtocol.h"
#include "Rtmp/utils.h"
#include "RtmpMediaSource.h"
#include "Util/util.h"
using namespace std;
using namespace toolkit;

5
src/Rtmp/RtmpProtocol.h
View File

@ -14,12 +14,9 @@
#include <memory>
#include <string>
#include <functional>
#include <unordered_map>
#include "amf.h"
#include "Rtmp.h"
#include "Util/util.h"
#include "Util/logger.h"
#include "Util/TimeTicker.h"
#include "Network/Socket.h"
#include "Util/ResourcePool.h"
#include "Http/HttpRequestSplitter.h"

8
src/Rtmp/RtmpPusher.cpp
View File

@ -128,7 +128,7 @@ void RtmpPusher::onRecv(const Buffer::Ptr &buf){
}
}
inline void RtmpPusher::send_connect() {
void RtmpPusher::send_connect() {
AMFValue obj(AMF_OBJECT);
obj.set("app", _app);
obj.set("type", "nonprivate");
@ -148,7 +148,7 @@ inline void RtmpPusher::send_connect() {
});
}
inline void RtmpPusher::send_createStream() {
void RtmpPusher::send_createStream() {
AMFValue obj(AMF_NULL);
sendInvoke("createStream", obj);
addOnResultCB([this](AMFDecoder &dec) {
@ -160,7 +160,7 @@ inline void RtmpPusher::send_createStream() {
}
#define RTMP_STREAM_LIVE "live"
inline void RtmpPusher::send_publish() {
void RtmpPusher::send_publish() {
AMFEncoder enc;
enc << "publish" << ++_send_req_id << nullptr << _stream_id << RTMP_STREAM_LIVE;
sendRequest(MSG_CMD, enc.data());
@ -176,7 +176,7 @@ inline void RtmpPusher::send_publish() {
});
}
inline void RtmpPusher::send_metaData(){
void RtmpPusher::send_metaData(){
auto src = _publish_src.lock();
if (!src) {
throw std::runtime_error("the media source was released");

2
src/Rtmp/RtmpSession.h
View File

@ -15,10 +15,8 @@
#include "amf.h"
#include "Rtmp.h"
#include "utils.h"
#include "Common/config.h"
#include "RtmpProtocol.h"
#include "RtmpMediaSourceImp.h"
#include "Util/util.h"
#include "Util/TimeTicker.h"
#include "Network/Session.h"
#include "Common/Stamp.h"

2
src/Rtmp/amf.cpp
View File

@ -15,7 +15,7 @@
#include "Util/util.h"
#include "Util/logger.h"
#include "Network/sockutil.h"
#include "Util/util.h"
#include "Network/Buffer.h"
using namespace std;
using namespace toolkit;

7
src/Rtmp/amf.h
View File

@ -14,12 +14,11 @@
#include <assert.h>
#include <string>
#include <vector>
#include <unordered_map>
#include <map>
#include <stdexcept>
#include <functional>
#include "Network/Buffer.h"
namespace toolkit {
class BufferLikeString;
}
enum AMFType {
AMF_NUMBER,
AMF_INTEGER,