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

Rtsp头文件重构

This commit is contained in:
cqm 2022-11-23 10:46:00 +08:00
parent 27fd74d896
commit 16c48e967d
13 changed files with 213 additions and 185 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
src/Rtsp/RtpCodec.h
View File

@ -59,7 +59,7 @@ protected:
RingType::Ptr _ring;
};
class RtpInfo{
class RtpInfo {
public:
using Ptr = std::shared_ptr<RtpInfo>;

2
src/Rtsp/RtpMultiCaster.h
View File

@ -15,9 +15,7 @@
#include <memory>
#include <unordered_set>
#include <unordered_map>
#include "Common/config.h"
#include "RtspMediaSource.h"
#include "Util/mini.h"
#include "Network/Socket.h"
namespace mediakit{

76
src/Rtsp/RtspMediaSource.h
View File

@ -15,15 +15,9 @@
#include <string>
#include <memory>
#include <functional>
#include <unordered_map>
#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 RTP_GOP_SIZE 512
@ -37,7 +31,6 @@ namespace mediakit {
*/
class RtspMediaSource : public MediaSource, public toolkit::RingDelegate<RtpPacket::Ptr>, private PacketCache<RtpPacket> {
public:
using PoolType = toolkit::ResourcePool<RtpPacket>;
using Ptr = std::shared_ptr<RtspMediaSource>;
using RingDataType = std::shared_ptr<toolkit::List<RtpPacket::Ptr> >;
using RingType = toolkit::RingBuffer<RingDataType>;
@ -110,85 +103,24 @@ public:
/**
* 获取相应轨道的时间戳，单位毫秒
*/
uint32_t getTimeStamp(TrackType trackType) override {
assert(trackType >= TrackInvalid && trackType < TrackMax);
if (trackType != TrackInvalid) {
//获取某track的时间戳
auto &track = _tracks[trackType];
if (track) {
return track->_time_stamp;
}
}
//获取所有track的最小时间戳
uint32_t ret = UINT32_MAX;
for (auto &track : _tracks) {
if (track && track->_time_stamp < ret) {
ret = track->_time_stamp;
}
}
return ret;
}
uint32_t getTimeStamp(TrackType trackType) override;
/**
* 更新时间戳
*/
void setTimeStamp(uint32_t stamp) override {
for (auto &track : _tracks) {
if (track) {
track->_time_stamp = stamp;
}
}
}
void setTimeStamp(uint32_t stamp) override;
/**
* 设置sdp
*/
virtual void setSdp(const std::string &sdp) {
SdpParser sdp_parser(sdp);
_tracks[TrackVideo] = sdp_parser.getTrack(TrackVideo);
_tracks[TrackAudio] = sdp_parser.getTrack(TrackAudio);
_have_video = (bool) _tracks[TrackVideo];
_sdp = sdp_parser.toString();
if (_ring) {
regist();
}
}
virtual void setSdp(const std::string &sdp);
/**
* 输入rtp
* @param rtp rtp包
* @param keyPos 该包是否为关键帧的第一个包
*/
void onWrite(RtpPacket::Ptr rtp, bool keyPos) override {
_speed[rtp->type] += rtp->size();
assert(rtp->type >= 0 && rtp->type < TrackMax);
auto &track = _tracks[rtp->type];
auto stamp = rtp->getStampMS();
if (track) {
track->_seq = rtp->getSeq();
track->_time_stamp = rtp->getStamp() * uint64_t(1000) / rtp->sample_rate;
track->_ssrc = rtp->getSSRC();
}
if (!_ring) {
std::weak_ptr<RtspMediaSource> weakSelf = std::dynamic_pointer_cast<RtspMediaSource>(shared_from_this());
auto lam = [weakSelf](int size) {
auto strongSelf = weakSelf.lock();
if (!strongSelf) {
return;
}
strongSelf->onReaderChanged(size);
};
//GOP默认缓冲512组RTP包，每组RTP包时间戳相同(如果开启合并写了，那么每组为合并写时间内的RTP包),
//每次遇到关键帧第一个RTP包，则会清空GOP缓存(因为有新的关键帧了，同样可以实现秒开)
_ring = std::make_shared<RingType>(_ring_size, std::move(lam));
if (!_sdp.empty()) {
regist();
}
}
bool is_video = rtp->type == TrackVideo;
PacketCache<RtpPacket>::inputPacket(stamp, is_video, std::move(rtp), keyPos);
}
void onWrite(RtpPacket::Ptr rtp, bool keyPos) override;
void clearCache() override{
PacketCache<RtpPacket>::clearCache();

131
src/Rtsp/RtspMediaSourceImp.cpp Normal file
View File

@ -0,0 +1,131 @@
#include "RtspMediaSourceImp.h"
#include "RtspDemuxer.h"
#include "Common/config.h"
namespace mediakit {
void RtspMediaSource::setSdp(const std::string &sdp) {
SdpParser sdp_parser(sdp);
_tracks[TrackVideo] = sdp_parser.getTrack(TrackVideo);
_tracks[TrackAudio] = sdp_parser.getTrack(TrackAudio);
_have_video = (bool)_tracks[TrackVideo];
_sdp = sdp_parser.toString();
if (_ring) {
regist();
}
}
uint32_t RtspMediaSource::getTimeStamp(TrackType trackType) {
assert(trackType >= TrackInvalid && trackType < TrackMax);
if (trackType != TrackInvalid) {
//获取某track的时间戳
auto &track = _tracks[trackType];
if (track) {
return track->_time_stamp;
}
}
//获取所有track的最小时间戳
uint32_t ret = UINT32_MAX;
for (auto &track : _tracks) {
if (track && track->_time_stamp < ret) {
ret = track->_time_stamp;
}
}
return ret;
}
/**
* 更新时间戳
*/
void RtspMediaSource::setTimeStamp(uint32_t stamp) {
for (auto &track : _tracks) {
if (track) {
track->_time_stamp = stamp;
}
}
}
void RtspMediaSource::onWrite(RtpPacket::Ptr rtp, bool keyPos) {
_speed[rtp->type] += rtp->size();
assert(rtp->type >= 0 && rtp->type < TrackMax);
auto &track = _tracks[rtp->type];
auto stamp = rtp->getStampMS();
if (track) {
track->_seq = rtp->getSeq();
track->_time_stamp = rtp->getStamp() * uint64_t(1000) / rtp->sample_rate;
track->_ssrc = rtp->getSSRC();
}
if (!_ring) {
std::weak_ptr<RtspMediaSource> weakSelf = std::dynamic_pointer_cast<RtspMediaSource>(shared_from_this());
auto lam = [weakSelf](int size) {
auto strongSelf = weakSelf.lock();
if (!strongSelf) {
return;
}
strongSelf->onReaderChanged(size);
};
//GOP默认缓冲512组RTP包，每组RTP包时间戳相同(如果开启合并写了，那么每组为合并写时间内的RTP包),
//每次遇到关键帧第一个RTP包，则会清空GOP缓存(因为有新的关键帧了，同样可以实现秒开)
_ring = std::make_shared<RingType>(_ring_size, std::move(lam));
if (!_sdp.empty()) {
regist();
}
}
bool is_video = rtp->type == TrackVideo;
PacketCache<RtpPacket>::inputPacket(stamp, is_video, std::move(rtp), keyPos);
}
RtspMediaSourceImp::RtspMediaSourceImp(const std::string &vhost, const std::string &app, const std::string &id, int ringSize)
: RtspMediaSource(vhost, app, id, ringSize)
{
_demuxer = std::make_shared<RtspDemuxer>();
_demuxer->setTrackListener(this);
}
void RtspMediaSourceImp::setSdp(const std::string &strSdp)
{
if (!getSdp().empty()) {
return;
}
_demuxer->loadSdp(strSdp);
RtspMediaSource::setSdp(strSdp);
}
void RtspMediaSourceImp::onWrite(RtpPacket::Ptr rtp, bool key_pos)
{
if (_all_track_ready && !_muxer->isEnabled()) {
//获取到所有Track后，并且未开启转协议，那么不需要解复用rtp
//在关闭rtp解复用后，无法知道是否为关键帧，这样会导致无法秒开，或者开播花屏
key_pos = rtp->type == TrackVideo;
} else {
//需要解复用rtp
key_pos = _demuxer->inputRtp(rtp);
}
GET_CONFIG(bool, directProxy, Rtsp::kDirectProxy);
if (directProxy) {
//直接代理模式才直接使用原始rtp
RtspMediaSource::onWrite(std::move(rtp), key_pos);
}
}
void RtspMediaSourceImp::setProtocolOption(const ProtocolOption &option)
{
GET_CONFIG(bool, direct_proxy, Rtsp::kDirectProxy);
//开启直接代理模式时，rtsp直接代理，不重复产生；但是有些rtsp推流端，由于sdp中已有sps pps，rtp中就不再包括sps pps,
//导致rtc无法播放，所以在rtsp推流rtc播放时，建议关闭直接代理模式
_option = option;
_option.enable_rtsp = !direct_proxy;
_muxer = std::make_shared<MultiMediaSourceMuxer>(getVhost(), getApp(), getId(), _demuxer->getDuration(), _option);
_muxer->setMediaListener(getListener());
_muxer->setTrackListener(std::static_pointer_cast<RtspMediaSourceImp>(shared_from_this()));
//让_muxer对象拦截一部分事件(比如说录像相关事件)
MediaSource::setListener(_muxer);
for (auto &track : _demuxer->getTracks(false)) {
_muxer->addTrack(track);
track->addDelegate(_muxer);
}
}
}

54
src/Rtsp/RtspMediaSourceImp.h
View File

@ -11,12 +11,11 @@
#ifndef SRC_RTSP_RTSPTORTMPMEDIASOURCE_H_
#define SRC_RTSP_RTSPTORTMPMEDIASOURCE_H_
#include "Rtmp/amf.h"
#include "RtspMediaSource.h"
#include "RtspDemuxer.h"
#include "Common/MultiMediaSourceMuxer.h"
namespace mediakit {
class RtspDemuxer;
class RtspMediaSourceImp final : public RtspMediaSource, private TrackListener, public MultiMediaSourceMuxer::Listener {
public:
using Ptr = std::shared_ptr<RtspMediaSourceImp>;
@ -28,70 +27,31 @@ public:
* @param id 流id
* @param ringSize 环形缓存大小
*/
RtspMediaSourceImp(const std::string &vhost, const std::string &app, const std::string &id, int ringSize = RTP_GOP_SIZE) : RtspMediaSource(vhost, app, id,ringSize) {
_demuxer = std::make_shared<RtspDemuxer>();
_demuxer->setTrackListener(this);
}
RtspMediaSourceImp(const std::string &vhost, const std::string &app, const std::string &id, int ringSize = RTP_GOP_SIZE);
~RtspMediaSourceImp() override = default;
/**
* 设置sdp
*/
void setSdp(const std::string &strSdp) override {
if (!getSdp().empty()) {
return;
}
_demuxer->loadSdp(strSdp);
RtspMediaSource::setSdp(strSdp);
}
void setSdp(const std::string &strSdp) override;
/**
* 输入rtp并解析
*/
void onWrite(RtpPacket::Ptr rtp, bool key_pos) override {
if (_all_track_ready && !_muxer->isEnabled()) {
//获取到所有Track后，并且未开启转协议，那么不需要解复用rtp
//在关闭rtp解复用后，无法知道是否为关键帧，这样会导致无法秒开，或者开播花屏
key_pos = rtp->type == TrackVideo;
} else {
//需要解复用rtp
key_pos = _demuxer->inputRtp(rtp);
}
GET_CONFIG(bool, directProxy, Rtsp::kDirectProxy);
if (directProxy) {
//直接代理模式才直接使用原始rtp
RtspMediaSource::onWrite(std::move(rtp), key_pos);
}
}
void onWrite(RtpPacket::Ptr rtp, bool key_pos) override;
/**
* 获取观看总人数，包括(hls/rtsp/rtmp)
*/
int totalReaderCount() override{
int totalReaderCount() override {
return readerCount() + (_muxer ? _muxer->totalReaderCount() : 0);
}
/**
* 设置协议转换选项
*/
void setProtocolOption(const ProtocolOption &option) {
GET_CONFIG(bool, direct_proxy, Rtsp::kDirectProxy);
//开启直接代理模式时，rtsp直接代理，不重复产生；但是有些rtsp推流端，由于sdp中已有sps pps，rtp中就不再包括sps pps,
//导致rtc无法播放，所以在rtsp推流rtc播放时，建议关闭直接代理模式
_option = option;
_option.enable_rtsp = !direct_proxy;
_muxer = std::make_shared<MultiMediaSourceMuxer>(getVhost(), getApp(), getId(), _demuxer->getDuration(), _option);
_muxer->setMediaListener(getListener());
_muxer->setTrackListener(std::static_pointer_cast<RtspMediaSourceImp>(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;
@ -149,7 +109,7 @@ public:
private:
bool _all_track_ready = false;
ProtocolOption _option;
RtspDemuxer::Ptr _demuxer;
std::shared_ptr<RtspDemuxer> _demuxer;
MultiMediaSourceMuxer::Ptr _muxer;
};
} /* namespace mediakit */

50
src/Rtsp/RtspPlayer.cpp
View File

@ -17,6 +17,11 @@
#include "Util/MD5.h"
#include "Util/base64.h"
#include "Rtcp/Rtcp.h"
#include "Rtcp/RtcpContext.h"
#include "RtspMediaSource.h"
#include "RtspDemuxer.h"
#include "RtspPlayerImp.h"
using namespace toolkit;
using namespace std;
@ -745,4 +750,49 @@ int RtspPlayer::getTrackIndexByTrackType(TrackType track_type) const {
throw SockException(Err_shutdown, StrPrinter << "no such track with type:" << getTrackString(track_type));
}
///////////////////////////////////////////////////
// RtspPlayerImp
float RtspPlayerImp::getDuration() const
{
return _demuxer ? _demuxer->getDuration() : 0;
}
void RtspPlayerImp::onPlayResult(const toolkit::SockException &ex) {
if (!(*this)[Client::kWaitTrackReady].as<bool>() || ex) {
Super::onPlayResult(ex);
return;
}
}
void RtspPlayerImp::addTrackCompleted() {
if ((*this)[Client::kWaitTrackReady].as<bool>()) {
Super::onPlayResult(toolkit::SockException(toolkit::Err_success, "play success"));
}
}
std::vector<Track::Ptr> RtspPlayerImp::getTracks(bool ready /*= true*/) const
{
return _demuxer ? _demuxer->getTracks(ready) : Super::getTracks(ready);
}
bool RtspPlayerImp::onCheckSDP(const std::string &sdp)
{
_rtsp_media_src = std::dynamic_pointer_cast<RtspMediaSource>(_media_src);
if (_rtsp_media_src) {
_rtsp_media_src->setSdp(sdp);
}
_demuxer = std::make_shared<RtspDemuxer>();
_demuxer->setTrackListener(this, (*this)[Client::kWaitTrackReady].as<bool>());
_demuxer->loadSdp(sdp);
return true;
}
void RtspPlayerImp::onRecvRTP(RtpPacket::Ptr rtp, const SdpTrack::Ptr &track) {
//rtp解复用时可以判断是否为关键帧起始位置
auto key_pos = _demuxer->inputRtp(rtp);
if (_rtsp_media_src) {
_rtsp_media_src->onWrite(std::move(rtp), key_pos);
}
}
} /* namespace mediakit */

12
src/Rtsp/RtspPlayer.h
View File

@ -13,22 +13,16 @@
#include <string>
#include <memory>
#include "RtspSession.h"
#include "RtspMediaSource.h"
#include "Player/PlayerBase.h"
#include "Util/util.h"
#include "Util/logger.h"
#include "Util/TimeTicker.h"
#include "Poller/Timer.h"
#include "Network/Socket.h"
#include "Player/PlayerBase.h"
#include "Network/TcpClient.h"
#include "RtspSplitter.h"
#include "RtpReceiver.h"
#include "Common/Stamp.h"
#include "Rtcp/RtcpContext.h"
namespace mediakit {
class RtcpContext;
//实现了rtsp播放器协议部分的功能，及数据接收功能
class RtspPlayer : public PlayerBase, public toolkit::TcpClient, public RtspSplitter, public RtpReceiver {
public:
@ -149,7 +143,7 @@ private:
//rtcp发送时间,trackid idx 为数组下标
toolkit::Ticker _rtcp_send_ticker[2];
//统计rtp并发送rtcp
std::vector<RtcpContext::Ptr> _rtcp_context;
std::vector<std::shared_ptr<RtcpContext>> _rtcp_context;
};
} /* namespace mediakit */

51
src/Rtsp/RtspPlayerImp.h
View File

@ -14,14 +14,11 @@
#include <memory>
#include <algorithm>
#include <functional>
#include "Common/config.h"
#include "RtspPlayer.h"
#include "RtspDemuxer.h"
#include "Poller/Timer.h"
#include "Util/TimeTicker.h"
namespace mediakit {
class RtspDemuxer;
class RtspMediaSource;
class RtspPlayerImp : public PlayerImp<RtspPlayer, PlayerBase> ,private TrackListener {
public:
using Ptr = std::shared_ptr<RtspPlayerImp>;
@ -57,53 +54,25 @@ 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 onCheckSDP(const std::string &sdp) override {
_rtsp_media_src = std::dynamic_pointer_cast<RtspMediaSource>(_media_src);
if (_rtsp_media_src) {
_rtsp_media_src->setSdp(sdp);
}
_demuxer = std::make_shared<RtspDemuxer>();
_demuxer->setTrackListener(this, (*this)[Client::kWaitTrackReady].as<bool>());
_demuxer->loadSdp(sdp);
return true;
}
bool onCheckSDP(const std::string &sdp) override;
void onRecvRTP(RtpPacket::Ptr rtp, const SdpTrack::Ptr &track) override {
//rtp解复用时可以判断是否为关键帧起始位置
auto key_pos = _demuxer->inputRtp(rtp);
if (_rtsp_media_src) {
_rtsp_media_src->onWrite(std::move(rtp), key_pos);
}
}
void onRecvRTP(RtpPacket::Ptr rtp, const SdpTrack::Ptr &track) override;
void onPlayResult(const toolkit::SockException &ex) override {
if (!(*this)[Client::kWaitTrackReady].as<bool>() || ex) {
Super::onPlayResult(ex);
return;
}
}
void onPlayResult(const toolkit::SockException &ex) override;
bool addTrack(const Track::Ptr &track) override { return true; }
void addTrackCompleted() override {
if ((*this)[Client::kWaitTrackReady].as<bool>()) {
Super::onPlayResult(toolkit::SockException(toolkit::Err_success, "play success"));
}
}
void addTrackCompleted() override;
private:
RtspDemuxer::Ptr _demuxer;
RtspMediaSource::Ptr _rtsp_media_src;
std::shared_ptr<RtspDemuxer> _demuxer;
std::shared_ptr<RtspMediaSource> _rtsp_media_src;
};
} /* namespace mediakit */

1
src/Rtsp/RtspPusher.cpp
View File

@ -12,6 +12,7 @@
#include "Util/base64.h"
#include "RtspPusher.h"
#include "RtspSession.h"
#include "Rtcp/RtcpContext.h"
using namespace std;
using namespace toolkit;

7
src/Rtsp/RtspPusher.h
View File

@ -14,17 +14,14 @@
#include <string>
#include <memory>
#include "RtspMediaSource.h"
#include "Util/util.h"
#include "Util/logger.h"
#include "Poller/Timer.h"
#include "Network/Socket.h"
#include "Network/TcpClient.h"
#include "RtspSplitter.h"
#include "Pusher/PusherBase.h"
#include "Rtcp/RtcpContext.h"
namespace mediakit {
class RtcpContext;
class RtspPusher : public toolkit::TcpClient, public RtspSplitter, public PusherBase {
public:
typedef std::shared_ptr<RtspPusher> Ptr;
@ -96,7 +93,7 @@ private:
//rtcp发送时间,trackid idx 为数组下标
toolkit::Ticker _rtcp_send_ticker[2];
//统计rtp并发送rtcp
std::vector<RtcpContext::Ptr> _rtcp_context;
std::vector<std::shared_ptr<RtcpContext>> _rtcp_context;
};
using RtspPusherImp = PusherImp<RtspPusher, PusherBase>;

1
src/Rtsp/RtspSession.cpp
View File

@ -16,6 +16,7 @@
#include "Util/MD5.h"
#include "Util/base64.h"
#include "RtpMultiCaster.h"
#include "Rtcp/RtcpContext.h"
using namespace std;
using namespace toolkit;

8
src/Rtsp/RtspSession.h
View File

@ -14,18 +14,16 @@
#include <set>
#include <vector>
#include <unordered_set>
#include <unordered_map>
#include "Network/Session.h"
#include "RtspMediaSource.h"
#include "RtspSplitter.h"
#include "RtpReceiver.h"
#include "RtspMediaSource.h"
#include "RtspMediaSourceImp.h"
#include "Rtcp/RtcpContext.h"
namespace mediakit {
class RtpMultiCaster;
class RtspSession;
class RtcpContext;
using BufferRtp = toolkit::BufferOffset<toolkit::Buffer::Ptr>;
class RtspSession : public toolkit::Session, public RtspSplitter, public RtpReceiver, public MediaSourceEvent {
@ -195,7 +193,7 @@ private:
//rtcp发送时间,trackid idx 为数组下标
toolkit::Ticker _rtcp_send_tickers[2];
//统计rtp并发送rtcp
std::vector<RtcpContext::Ptr> _rtcp_context;
std::vector<std::shared_ptr<RtcpContext>> _rtcp_context;
};
/**

3
src/Rtsp/UDPServer.h
View File

@ -15,9 +15,6 @@
#include <mutex>
#include <memory>
#include <unordered_map>
#include <unordered_set>
#include "Util/util.h"
#include "Util/logger.h"
#include "Network/Socket.h"
namespace mediakit {