Skip to content

Z
ZLMediaKit
Commit 820da438 by xiongziliang
Options

添加MediaSink类，简化重复逻辑代码

parent ce5c71c9
正在显示 包含 274 行增加 和 173 行删除
src/Common/MediaSink.cpp 0 → 100644
/*
* MIT License
*
* Copyright (c) 2016 xiongziliang <771730766@qq.com>
*
* This file is part of ZLMediaKit(https://github.com/xiongziliang/ZLMediaKit).
*
* Permission is hereby granted, free of charge, to any person obtaining a copy
* of this software and associated documentation files (the "Software"), to deal
* in the Software without restriction, including without limitation the rights
* to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
* copies of the Software, and to permit persons to whom the Software is
* furnished to do so, subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in all
* copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
* AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
* LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
* OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
* SOFTWARE.
*/
#include "MediaSink.h"
namespace mediakit{
void MediaSink::addTrack(const Track::Ptr &track_in) {
lock_guard<mutex> lck(_mtx);
//克隆Track，只拷贝其数据，不拷贝其数据转发关系
auto track = track_in->clone();
weak_ptr<MediaSink> weakSelf = shared_from_this();
track->addDelegate(std::make_shared<FrameWriterInterfaceHelper>([weakSelf](const Frame::Ptr &frame){
auto strongSelf = weakSelf.lock();
if(!strongSelf){
return;
}
if(strongSelf->_allTrackReady){
strongSelf->onTrackFrame(frame);
}
}));
auto codec_id = track->getCodecId();
_track_map[codec_id] = track;
auto lam = [this,track](){
onTrackReady(track);
};
if(track->ready()){
lam();
}else{
_allTrackReady = false;
_trackReadyCallback[codec_id] = lam;
}
}
void MediaSink::inputFrame(const Frame::Ptr &frame) {
lock_guard<mutex> lck(_mtx);
auto codec_id = frame->getCodecId();
auto it = _track_map.find(codec_id);
if (it == _track_map.end()) {
return;
}
it->second->inputFrame(frame);
if(!_allTrackReady && !_trackReadyCallback.empty() && it->second->ready()){
//Track由未就绪状态装换成就绪状态，我们就生成sdp以及rtp编码器
auto it_callback = _trackReadyCallback.find(codec_id);
if(it_callback != _trackReadyCallback.end()){
it_callback->second();
_trackReadyCallback.erase(it_callback);
}
}
if(!_allTrackReady && _trackReadyCallback.empty()){
_allTrackReady = true;
onAllTrackReady();
}
}
bool MediaSink::isAllTrackReady() const {
return _allTrackReady;
}
}//namespace mediakit
src/Common/MediaSink.h 0 → 100644
/*
* MIT License
*
* Copyright (c) 2016 xiongziliang <771730766@qq.com>
*
* This file is part of ZLMediaKit(https://github.com/xiongziliang/ZLMediaKit).
*
* Permission is hereby granted, free of charge, to any person obtaining a copy
* of this software and associated documentation files (the "Software"), to deal
* in the Software without restriction, including without limitation the rights
* to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
* copies of the Software, and to permit persons to whom the Software is
* furnished to do so, subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in all
* copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
* AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
* LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
* OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
* SOFTWARE.
*/
#ifndef ZLMEDIAKIT_MEDIASINK_H
#define ZLMEDIAKIT_MEDIASINK_H
#include <mutex>
#include <memory>
#include "Player/Frame.h"
#include "Player/Track.h"
using namespace std;
using namespace toolkit;
namespace mediakit{
class MediaSink : public FrameWriterInterface , public std::enable_shared_from_this<MediaSink>{
public:
typedef std::shared_ptr<MediaSink> Ptr;
MediaSink(){}
virtual ~MediaSink(){}
/**
* 输入frame
* @param frame
*/
void inputFrame(const Frame::Ptr &frame) override ;
/**
* 添加track，内部会调用Track的clone方法
* 只会克隆sps pps这些信息 ，而不会克隆Delegate相关关系
* @param track
*/
void addTrack(const Track::Ptr & track);
/**
* 全部Track是否都准备好了
* @return
*/
bool isAllTrackReady() const ;
protected:
/**
* 某track已经准备好，其ready()状态返回true，
* 此时代表可以获取其例如sps pps等相关信息了
* @param track
*/
virtual void onTrackReady(const Track::Ptr & track) {};
/**
* 所有Track已经准备好，
*/
virtual void onAllTrackReady() {};
/**
* 某Track输出frame，在onAllTrackReady触发后才会调用此方法
* @param frame
*/
virtual void onTrackFrame(const Frame::Ptr &frame) {};
private:
mutex _mtx;
map<int,Track::Ptr> _track_map;
map<int,function<void()> > _trackReadyCallback;
bool _allTrackReady = false;
};
}//namespace mediakit
#endif //ZLMEDIAKIT_MEDIASINK_H
src/Common/MultiMediaSourceMuxer.h
......@@ -30,7 +30,7 @@
#include "RtspMuxer/RtspMediaSourceMuxer.h"
#include "RtmpMuxer/RtmpMediaSourceMuxer.h"
class MultiMediaSourceMuxer : public FrameRingWriterInterface{
class MultiMediaSourceMuxer : public FrameWriterInterface{
public:
typedef std::shared_ptr<MultiMediaSourceMuxer> Ptr;
......
src/Device/PlayerProxy.cpp
......@@ -59,9 +59,6 @@ static uint8_t s_mute_adts[] = {0xff, 0xf1, 0x6c, 0x40, 0x2d, 0x3f, 0xfc, 0x00,
0x5c, 0xa7, 0x17, 0xcf, 0x34, 0x57, 0xc9, 0x58, 0xc5, 0x20, 0x09, 0xee, 0xa5, 0xf2, 0x9c, 0x6c,
0x39, 0x1a, 0x77, 0x92, 0x9b, 0xff, 0xc6, 0xae, 0xf8, 0x36, 0xba, 0xa8, 0xaa, 0x6b, 0x1e, 0x8c,
0xc5, 0x97, 0x39, 0x6a, 0xb8, 0xa2, 0x55, 0xa8, 0xf8};
#define MUTE_ADTS_CHN_CNT 1
#define MUTE_ADTS_SAMPLE_BIT 16
#define MUTE_ADTS_SAMPLE_RATE 8000
#define MUTE_ADTS_DATA s_mute_adts
#define MUTE_ADTS_DATA_LEN sizeof(s_mute_adts)
#define MUTE_ADTS_DATA_MS 130
......
src/Player/Frame.h
......@@ -28,6 +28,7 @@
#define ZLMEDIAKIT_FRAME_H
#include <mutex>
#include <functional>
#include "Util/RingBuffer.h"
#include "Network/Socket.h"
......@@ -106,12 +107,12 @@ private:
ResourcePool<T> _pool;
};
class FrameRingWriterInterface {
class FrameWriterInterface {
public:
typedef std::shared_ptr<FrameRingWriterInterface> Ptr;
typedef std::shared_ptr<FrameWriterInterface> Ptr;
FrameRingWriterInterface(){}
virtual ~FrameRingWriterInterface(){}
FrameWriterInterface(){}
virtual ~FrameWriterInterface(){}
/**
* 写入帧数据
* @param frame 帧
......@@ -119,10 +120,31 @@ public:
virtual void inputFrame(const Frame::Ptr &frame) = 0;
};
class FrameWriterInterfaceHelper : public FrameWriterInterface {
public:
typedef std::shared_ptr<FrameWriterInterfaceHelper> Ptr;
typedef std::function<void(const Frame::Ptr &frame)> onWriteFrame;
FrameWriterInterfaceHelper(const onWriteFrame& cb){
_writeCallback = cb;
}
virtual ~FrameWriterInterfaceHelper(){}
/**
* 写入帧数据
* @param frame 帧
*/
void inputFrame(const Frame::Ptr &frame) override {
_writeCallback(frame);
}
private:
onWriteFrame _writeCallback;
};
/**
* 帧环形缓存接口类
*/
class FrameRingInterface : public FrameRingWriterInterface{
class FrameRingInterface : public FrameWriterInterface{
public:
typedef RingBuffer<Frame::Ptr> RingType;
typedef std::shared_ptr<FrameRingInterface> Ptr;
......@@ -187,7 +209,7 @@ public:
FrameRingInterfaceDelegate(){}
virtual ~FrameRingInterfaceDelegate(){}
void addDelegate(const FrameRingWriterInterface::Ptr &delegate){
void addDelegate(const FrameWriterInterface::Ptr &delegate){
lock_guard<mutex> lck(_mtx);
_delegateMap.emplace(delegate.get(),delegate);
}
......@@ -210,7 +232,7 @@ public:
}
private:
mutex _mtx;
map<void *,FrameRingWriterInterface::Ptr> _delegateMap;
map<void *,FrameWriterInterface::Ptr> _delegateMap;
};
......
src/RtmpMuxer/RtmpMediaSourceMuxer.h
......@@ -52,7 +52,7 @@ public:
return _mediaSouce->getRing()->readerCount();
}
private:
void onInited() override {
void onAllTrackReady() override {
_mediaSouce->onGetMetaData(getMetedata());
}
private:
......
src/RtmpMuxer/RtmpMuxer.cpp
......@@ -28,19 +28,21 @@
namespace mediakit {
void RtmpMuxer::addTrack(const Track::Ptr &track_in) {
//克隆Track，只拷贝其数据，不拷贝其数据转发关系
auto track = track_in->clone();
auto codec_id = track->getCodecId();
_track_map[codec_id] = track;
RtmpMuxer::RtmpMuxer(const TitleMete::Ptr &title) {
if(!title){
_metedata = std::make_shared<TitleMete>()->getMetedata();
}else{
_metedata = title->getMetedata();
}
_rtmpRing = std::make_shared<RtmpRingInterface::RingType>();
}
auto lam = [this,track](){
//异步生成Rtmp编码器
void RtmpMuxer::onTrackReady(const Track::Ptr &track) {
//生成rtmp编码器
auto encoder = Factory::getRtmpCodecByTrack(track);
if (!encoder) {
return;
}
//根据track生产metedata
Metedata::Ptr metedate;
switch (track->getTrackType()){
......@@ -64,17 +66,11 @@ void RtmpMuxer::addTrack(const Track::Ptr &track_in) {
track->addDelegate(encoder);
//Rtmp编码器共用同一个环形缓存
encoder->setRtmpRing(_rtmpRing);
};
if(track->ready()){
lam();
}else{
_trackReadyCallback[codec_id] = lam;
}
}
const AMFValue &RtmpMuxer::getMetedata() const {
if(!_trackReadyCallback.empty()){
if(!isAllTrackReady()){
//尚未就绪
static AMFValue s_amf;
return s_amf;
......@@ -82,36 +78,8 @@ const AMFValue &RtmpMuxer::getMetedata() const {
return _metedata;
}
void RtmpMuxer::inputFrame(const Frame::Ptr &frame) {
auto codec_id = frame->getCodecId();
auto it = _track_map.find(codec_id);
if (it == _track_map.end()) {
return;
}
it->second->inputFrame(frame);
if(!_inited && !_trackReadyCallback.empty() && it->second->ready()){
//Track由未就绪状态装换成就绪状态，我们就生成metedata以及Rtmp编码器
auto it_callback = _trackReadyCallback.find(codec_id);
if(it_callback != _trackReadyCallback.end()){
it_callback->second();
_trackReadyCallback.erase(it_callback);
}
}
if(!_inited && _trackReadyCallback.empty()){
_inited = true;
onInited();
}
}
bool RtmpMuxer::inputRtmp(const RtmpPacket::Ptr &rtmp , bool key_pos) {
_rtmpRing->write(rtmp,key_pos);
return key_pos;
}
RtmpRingInterface::RingType::Ptr RtmpMuxer::getRtmpRing() const {
return _rtmpRing;
}
}
\ No newline at end of file
}/* namespace mediakit */
\ No newline at end of file
src/RtmpMuxer/RtmpMuxer.h
......@@ -29,65 +29,41 @@
#include "RtmpMetedata.h"
#include "Player/Frame.h"
#include "Common/MediaSink.h"
namespace mediakit{
class RtmpMuxer : public FrameRingWriterInterface{
class RtmpMuxer : public MediaSink{
public:
typedef std::shared_ptr<RtmpMuxer> Ptr;
/**
* 构造函数
*/
RtmpMuxer(const TitleMete::Ptr &title = nullptr) : _metedata(AMF_OBJECT){
if(!title){
_metedata = std::make_shared<TitleMete>()->getMetedata();
}else{
_metedata = title->getMetedata();
}
_rtmpRing = std::make_shared<RtmpRingInterface::RingType>();
}
RtmpMuxer(const TitleMete::Ptr &title);
virtual ~RtmpMuxer(){}
/**
* 添加音视频媒体
* @param track 媒体描述
*/
void addTrack(const Track::Ptr & track) ;
/**
* 获取完整的SDP字符串
* @return SDP字符串
*/
const AMFValue &getMetedata() const ;
/**
* 写入帧数据然后打包rtmp
* @param frame 帧数据
*/
void inputFrame(const Frame::Ptr &frame) override ;
/**
* 也可以在外部打包好rtmp然后再写入
* @param rtmp rtmp包
* @param key_pos 是否为关键帧
*/
bool inputRtmp(const RtmpPacket::Ptr &rtmp, bool key_pos = true);
/**
* 获取rtmp环形缓存
* @return
*/
RtmpRingInterface::RingType::Ptr getRtmpRing() const;
protected:
virtual void onInited(){};
/**
* 某track已经准备好，其ready()状态返回true，
* 此时代表可以获取其例如sps pps等相关信息了
* @param track
*/
void onTrackReady(const Track::Ptr & track) override ;
private:
map<int,Track::Ptr> _track_map;
map<int,function<void()> > _trackReadyCallback;
RtmpRingInterface::RingType::Ptr _rtmpRing;
AMFValue _metedata;
bool _inited = false;
};
......
src/RtspMuxer/RtspMediaSourceMuxer.h
......@@ -55,7 +55,7 @@ public:
_mediaSouce->setTimeStamp(stamp);
}
private:
void onInited() override {
void onAllTrackReady() override {
_mediaSouce->onGetSDP(getSdp());
}
private:
......
src/RtspMuxer/RtspMuxer.cpp
......@@ -29,24 +29,25 @@
namespace mediakit {
void RtspMuxer::addTrack(const Track::Ptr &track_in, uint32_t ssrc, int mtu) {
//克隆Track，只拷贝其数据，不拷贝其数据转发关系
auto track = track_in->clone();
auto codec_id = track->getCodecId();
_track_map[codec_id] = track;
if(mtu == 0){
mtu = (track->getTrackType() == TrackVideo ? 1400 : 600);
RtspMuxer::RtspMuxer(const TitleSdp::Ptr &title){
if(!title){
_sdp = std::make_shared<TitleSdp>()->getSdp();
} else{
_sdp = title->getSdp();
}
auto lam = [this,ssrc,mtu,track](){
//异步生成rtp编码器
_rtpRing = std::make_shared<RtpRingInterface::RingType>();
}
void RtspMuxer::onTrackReady(const Track::Ptr &track) {
//根据track生产sdp
Sdp::Ptr sdp = Factory::getSdpByTrack(track);
if (!sdp) {
return;
}
// 根据sdp生成rtp编码器
auto encoder = sdp->createRtpEncoder(ssrc ? ssrc : ((uint64_t) sdp.get()) & 0xFFFFFFFF, mtu);
uint32_t ssrc = ((uint64_t) sdp.get()) & 0xFFFFFFFF;
auto mtu = (track->getTrackType() == TrackVideo ? 1400 : 600);
// 根据sdp生成rtp编码器ssrc
auto encoder = sdp->createRtpEncoder(ssrc, mtu);
if (!encoder) {
return;
}
......@@ -56,52 +57,19 @@ void RtspMuxer::addTrack(const Track::Ptr &track_in, uint32_t ssrc, int mtu) {
track->addDelegate(encoder);
//rtp编码器共用同一个环形缓存
encoder->setRtpRing(_rtpRing);
};
if(track->ready()){
lam();
}else{
_trackReadyCallback[codec_id] = lam;
}
}
string RtspMuxer::getSdp() {
if(!_trackReadyCallback.empty()){
if(!isAllTrackReady()){
//尚未就绪
return "";
}
return _sdp;
}
void RtspMuxer::inputFrame(const Frame::Ptr &frame) {
auto codec_id = frame->getCodecId();
auto it = _track_map.find(codec_id);
if (it == _track_map.end()) {
return;
}
it->second->inputFrame(frame);
if(!_inited && !_trackReadyCallback.empty() && it->second->ready()){
//Track由未就绪状态装换成就绪状态，我们就生成sdp以及rtp编码器
auto it_callback = _trackReadyCallback.find(codec_id);
if(it_callback != _trackReadyCallback.end()){
it_callback->second();
_trackReadyCallback.erase(it_callback);
}
}
if(!_inited && _trackReadyCallback.empty()){
_inited = true;
onInited();
}
}
bool RtspMuxer::inputRtp(const RtpPacket::Ptr &rtp, bool key_pos) {
_rtpRing->write(rtp,key_pos);
return key_pos;
}
RtpRingInterface::RingType::Ptr RtspMuxer::getRtpRing() const {
return _rtpRing;
}
} /* namespace mediakit */
\ No newline at end of file
src/RtspMuxer/RtspMuxer.h
......@@ -29,35 +29,22 @@
#include "RtspSdp.h"
#include "Player/Frame.h"
#include "Common/MediaSink.h"
namespace mediakit{
/**
* rtsp生成器
*/
class RtspMuxer : public FrameRingWriterInterface{
class RtspMuxer : public MediaSink{
public:
typedef std::shared_ptr<RtspMuxer> Ptr;
/**
* 构造函数
*/
RtspMuxer(const TitleSdp::Ptr &title = nullptr){
if(!title){
_sdp = std::make_shared<TitleSdp>()->getSdp();
} else{
_sdp = title->getSdp();
}
_rtpRing = std::make_shared<RtpRingInterface::RingType>();
}
virtual ~RtspMuxer(){}
RtspMuxer(const TitleSdp::Ptr &title = nullptr);
/**
* 添加音视频媒体
* @param track 媒体描述
* @param ssrc 媒体rtp ssrc
* @param mtu 媒体rtp mtu
*/
void addTrack(const Track::Ptr & track,uint32_t ssrc = 0,int mtu = 0) ;
virtual ~RtspMuxer(){}
/**
* 获取完整的SDP字符串
......@@ -66,32 +53,20 @@ public:
string getSdp() ;
/**
* 写入帧数据然后打包rtp
* @param frame 帧数据
*/
void inputFrame(const Frame::Ptr &frame) override ;
/**
* 也可以在外部打包好rtp然后再写入
* @param rtp rtp包
* @param key_pos 是否为关键帧的第一个rtp包
*/
bool inputRtp(const RtpPacket::Ptr &rtp, bool key_pos = true);
/**
* 获取rtp环形缓存
* @return
*/
RtpRingInterface::RingType::Ptr getRtpRing() const;
protected:
virtual void onInited(){};
/**
* 某track已经准备好，其ready()状态返回true，
* 此时代表可以获取其例如sps pps等相关信息了
* @param track
*/
void onTrackReady(const Track::Ptr & track) override ;
private:
map<int,Track::Ptr> _track_map;
map<int,function<void()> > _trackReadyCallback;
RtpRingInterface::RingType::Ptr _rtpRing;
string _sdp;
bool _inited = false;
};
......
Markdown 格式
0% 或
您添加了 0 人 到此讨论。请谨慎行事。
请先完成此评论的编辑！
请 注册 或者 后发表评论