Skip to content

Z
ZLMediaKit
Commit ab9a605a by xiongziliang
Options

优化代码结构

parent 289e4f69
正在显示 包含 498 行增加 和 227 行删除
src/Common/MediaSource.cpp
......@@ -179,7 +179,7 @@ bool MediaSource::isRecording(Recorder::type type){
return listener->isRecording(*this, type);
}
void MediaSource::startSendRtp(const string &dst_url, uint16_t dst_port, uint32_t ssrc, bool is_udp, const function<void(const SockException &ex)> &cb){
void MediaSource::startSendRtp(const string &dst_url, uint16_t dst_port, const string &ssrc, bool is_udp, const function<void(const SockException &ex)> &cb){
auto listener = _listener.lock();
if (!listener) {
cb(SockException(Err_other, "尚未设置事件监听器"));
......@@ -638,7 +638,7 @@ vector<Track::Ptr> MediaSourceEventInterceptor::getTracks(MediaSource &sender, b
return listener->getTracks(sender, trackReady);
}
void MediaSourceEventInterceptor::startSendRtp(MediaSource &sender, const string &dst_url, uint16_t dst_port, uint32_t ssrc, bool is_udp, const function<void(const SockException &ex)> &cb){
void MediaSourceEventInterceptor::startSendRtp(MediaSource &sender, const string &dst_url, uint16_t dst_port, const string &ssrc, bool is_udp, const function<void(const SockException &ex)> &cb){
auto listener = _listener.lock();
if (listener) {
listener->startSendRtp(sender, dst_url, dst_port, ssrc, is_udp, cb);
......
src/Common/MediaSource.h
......@@ -83,7 +83,7 @@ public:
// 获取所有track相关信息
virtual vector<Track::Ptr> getTracks(MediaSource &sender, bool trackReady = true) const { return vector<Track::Ptr>(); };
// 开始发送ps-rtp
virtual void startSendRtp(MediaSource &sender, const string &dst_url, uint16_t dst_port, uint32_t ssrc, bool is_udp, const function<void(const SockException &ex)> &cb) { cb(SockException(Err_other, "not implemented"));};
virtual void startSendRtp(MediaSource &sender, const string &dst_url, uint16_t dst_port, const string &ssrc, bool is_udp, const function<void(const SockException &ex)> &cb) { cb(SockException(Err_other, "not implemented"));};
// 停止发送ps-rtp
virtual bool stopSendRtp(MediaSource &sender) {return false; }
......@@ -112,7 +112,7 @@ public:
bool setupRecord(MediaSource &sender, Recorder::type type, bool start, const string &custom_path) override;
bool isRecording(MediaSource &sender, Recorder::type type) override;
vector<Track::Ptr> getTracks(MediaSource &sender, bool trackReady = true) const override;
void startSendRtp(MediaSource &sender, const string &dst_url, uint16_t dst_port, uint32_t ssrc, bool is_udp, const function<void(const SockException &ex)> &cb) override;
void startSendRtp(MediaSource &sender, const string &dst_url, uint16_t dst_port, const string &ssrc, bool is_udp, const function<void(const SockException &ex)> &cb) override;
bool stopSendRtp(MediaSource &sender) override;
private:
......@@ -256,7 +256,7 @@ public:
// 获取录制状态
bool isRecording(Recorder::type type);
// 开始发送ps-rtp
void startSendRtp(const string &dst_url, uint16_t dst_port, uint32_t ssrc, bool is_udp, const function<void(const SockException &ex)> &cb);
void startSendRtp(const string &dst_url, uint16_t dst_port, const string &ssrc, bool is_udp, const function<void(const SockException &ex)> &cb);
// 停止发送ps-rtp
bool stopSendRtp();
......
src/Common/MultiMediaSourceMuxer.cpp
......@@ -328,21 +328,21 @@ bool MultiMediaSourceMuxer::isRecording(MediaSource &sender, Recorder::type type
return _muxer->isRecording(sender,type);
}
void MultiMediaSourceMuxer::startSendRtp(MediaSource &sender, const string &dst_url, uint16_t dst_port, uint32_t ssrc, bool is_udp, const function<void(const SockException &ex)> &cb){
void MultiMediaSourceMuxer::startSendRtp(MediaSource &sender, const string &dst_url, uint16_t dst_port, const string &ssrc, bool is_udp, const function<void(const SockException &ex)> &cb){
#if defined(ENABLE_RTPPROXY)
auto ps_rtp_sender = std::make_shared<PSRtpSender>(ssrc);
RtpSender::Ptr rtp_sender = std::make_shared<RtpSender>(atoi(ssrc.data()));
weak_ptr<MultiMediaSourceMuxer> weak_self = shared_from_this();
ps_rtp_sender->startSend(dst_url, dst_port, is_udp, [weak_self, ps_rtp_sender, cb](const SockException &ex) {
rtp_sender->startSend(dst_url, dst_port, is_udp, [weak_self, rtp_sender, cb](const SockException &ex) {
cb(ex);
auto strong_self = weak_self.lock();
if (!strong_self || ex) {
return;
}
for (auto &track : strong_self->_muxer->getTracks(false)) {
ps_rtp_sender->addTrack(track);
rtp_sender->addTrack(track);
}
ps_rtp_sender->addTrackCompleted();
strong_self->_ps_rtp_sender = ps_rtp_sender;
rtp_sender->addTrackCompleted();
strong_self->_rtp_sender = rtp_sender;
});
#else
cb(SockException(Err_other, "该功能未启用，编译时请打开ENABLE_RTPPROXY宏"));
......@@ -351,8 +351,8 @@ void MultiMediaSourceMuxer::startSendRtp(MediaSource &sender, const string &dst_
bool MultiMediaSourceMuxer::stopSendRtp(MediaSource &sender){
#if defined(ENABLE_RTPPROXY)
if (_ps_rtp_sender) {
_ps_rtp_sender = nullptr;
if (_rtp_sender) {
_rtp_sender = nullptr;
return true;
}
#endif//ENABLE_RTPPROXY
......@@ -441,9 +441,9 @@ void MultiMediaSourceMuxer::inputFrame(const Frame::Ptr &frame_in) {
_muxer->inputFrame(frame);
#if defined(ENABLE_RTPPROXY)
auto ps_rtp_sender = _ps_rtp_sender;
if (ps_rtp_sender) {
ps_rtp_sender->inputFrame(frame);
auto rtp_sender = _rtp_sender;
if (rtp_sender) {
rtp_sender->inputFrame(frame);
}
#endif //ENABLE_RTPPROXY
......@@ -451,7 +451,7 @@ void MultiMediaSourceMuxer::inputFrame(const Frame::Ptr &frame_in) {
bool MultiMediaSourceMuxer::isEnabled(){
#if defined(ENABLE_RTPPROXY)
return (_muxer->isEnabled() || _ps_rtp_sender);
return (_muxer->isEnabled() || _rtp_sender);
#else
return _muxer->isEnabled();
#endif //ENABLE_RTPPROXY
......
src/Common/MultiMediaSourceMuxer.h
......@@ -12,7 +12,7 @@
#define ZLMEDIAKIT_MULTIMEDIASOURCEMUXER_H
#include "Common/Stamp.h"
#include "Rtp/PSRtpSender.h"
#include "Rtp/RtpSender.h"
#include "Record/Recorder.h"
#include "Record/HlsRecorder.h"
#include "Record/HlsMediaSource.h"
......@@ -142,7 +142,7 @@ public:
* @param is_udp 是否为udp
* @param cb 启动成功或失败回调
*/
void startSendRtp(MediaSource &sender, const string &dst_url, uint16_t dst_port, uint32_t ssrc, bool is_udp, const function<void(const SockException &ex)> &cb) override;
void startSendRtp(MediaSource &sender, const string &dst_url, uint16_t dst_port, const string &ssrc, bool is_udp, const function<void(const SockException &ex)> &cb) override;
/**
* 停止ps-rtp发送
......@@ -187,7 +187,7 @@ private:
MultiMuxerPrivate::Ptr _muxer;
std::weak_ptr<MultiMuxerPrivate::Listener> _track_listener;
#if defined(ENABLE_RTPPROXY)
PSRtpSender::Ptr _ps_rtp_sender;
RtpSender::Ptr _rtp_sender;
#endif //ENABLE_RTPPROXY
};
......
src/Rtp/GB28181Process.cpp 0 → 100644
/*
* Copyright (c) 2016 The ZLMediaKit project authors. All Rights Reserved.
*
* This file is part of ZLMediaKit(https://github.com/xiongziliang/ZLMediaKit).
*
* Use of this source code is governed by MIT license that can be found in the
* LICENSE file in the root of the source tree. All contributing project authors
* may be found in the AUTHORS file in the root of the source tree.
*/
#if defined(ENABLE_RTPPROXY)
#include "GB28181Process.h"
#include "Util/File.h"
#include "Http/HttpTSPlayer.h"
#include "Extension/CommonRtp.h"
namespace mediakit{
//判断是否为ts负载
static inline bool checkTS(const uint8_t *packet, int bytes){
return bytes % TS_PACKET_SIZE == 0 && packet[0] == TS_SYNC_BYTE;
}
GB28181Process::GB28181Process(const MediaInfo &media_info, MediaSinkInterface *interface) {
assert(interface);
_media_info = media_info;
_interface = interface;
}
GB28181Process::~GB28181Process() {}
bool GB28181Process::inputRtp(bool, const char *data, int data_len) {
return handleOneRtp(0, TrackVideo, 90000, (unsigned char *) data, data_len);
}
void GB28181Process::onRtpSorted(const RtpPacket::Ptr &rtp, int) {
if (!_rtp_decoder) {
switch (rtp->PT) {
case 33:
case 96: {
//ts或ps负载
_rtp_decoder = std::make_shared<CommonRtpDecoder>(CodecInvalid, 256 * 1024);
//设置dump目录
GET_CONFIG(string, dump_dir, RtpProxy::kDumpDir);
if (!dump_dir.empty()) {
auto save_path = File::absolutePath(_media_info._streamid + ".mp2", dump_dir);
_save_file_ps.reset(File::create_file(save_path.data(), "wb"), [](FILE *fp) {
if (fp) {
fclose(fp);
}
});
}
break;
}
default:
WarnL << "不支持的rtp负载类型:" << (int) rtp->PT;
return;
}
//设置frame回调
_rtp_decoder->addDelegate(std::make_shared<FrameWriterInterfaceHelper>([this](const Frame::Ptr &frame) {
onRtpDecode(frame);
}));
}
//解码rtp
_rtp_decoder->inputRtp(rtp, false);
}
const char *GB28181Process::onSearchPacketTail(const char *packet,int bytes){
try {
auto ret = _decoder->input((uint8_t *) packet, bytes);
if (ret > 0) {
return packet + ret;
}
return nullptr;
} catch (std::exception &ex) {
InfoL << "解析ps或ts异常: bytes=" << bytes
<< " ,exception=" << ex.what()
<< " ,hex=" << hexdump((uint8_t *) packet, bytes);
if (remainDataSize() > 256 * 1024) {
//缓存太多数据无法处理则上抛异常
throw;
}
return nullptr;
}
}
void GB28181Process::onRtpDecode(const Frame::Ptr &frame) {
//这是TS或PS
if (_save_file_ps) {
fwrite(frame->data(), frame->size(), 1, _save_file_ps.get());
}
if (!_decoder) {
//创建解码器
if (checkTS((uint8_t *) frame->data(), frame->size())) {
//猜测是ts负载
InfoL << _media_info._streamid << " judged to be TS";
_decoder = DecoderImp::createDecoder(DecoderImp::decoder_ts, _interface);
} else {
//猜测是ps负载
InfoL << _media_info._streamid << " judged to be PS";
_decoder = DecoderImp::createDecoder(DecoderImp::decoder_ps, _interface);
}
}
if (_decoder) {
HttpRequestSplitter::input(frame->data(), frame->size());
}
}
}//namespace mediakit
#endif//defined(ENABLE_RTPPROXY)
\ No newline at end of file
src/Rtp/GB28181Process.h 0 → 100644
/*
* Copyright (c) 2016 The ZLMediaKit project authors. All Rights Reserved.
*
* This file is part of ZLMediaKit(https://github.com/xiongziliang/ZLMediaKit).
*
* Use of this source code is governed by MIT license that can be found in the
* LICENSE file in the root of the source tree. All contributing project authors
* may be found in the AUTHORS file in the root of the source tree.
*/
#ifndef ZLMEDIAKIT_GB28181ROCESS_H
#define ZLMEDIAKIT_GB28181ROCESS_H
#if defined(ENABLE_RTPPROXY)
#include "Decoder.h"
#include "ProcessInterface.h"
#include "Rtsp/RtpCodec.h"
#include "Rtsp/RtpReceiver.h"
#include "Http/HttpRequestSplitter.h"
namespace mediakit{
class GB28181Process : public HttpRequestSplitter, public RtpReceiver, public ProcessInterface{
public:
typedef std::shared_ptr<GB28181Process> Ptr;
GB28181Process(const MediaInfo &media_info, MediaSinkInterface *interface);
~GB28181Process() override;
/**
* 输入rtp
* @param data rtp数据指针
* @param data_len rtp数据长度
* @return 是否解析成功
*/
bool inputRtp(bool, const char *data, int data_len) override;
protected:
void onRtpSorted(const RtpPacket::Ptr &rtp, int track_index) override ;
const char *onSearchPacketTail(const char *data,int len) override;
int64_t onRecvHeader(const char *data,uint64_t len) override { return 0; };
private:
void onRtpDecode(const Frame::Ptr &frame);
private:
MediaInfo _media_info;
DecoderImp::Ptr _decoder;
MediaSinkInterface *_interface;
std::shared_ptr<FILE> _save_file_ps;
std::shared_ptr<RtpCodec> _rtp_decoder;
};
}//namespace mediakit
#endif//defined(ENABLE_RTPPROXY)
#endif //ZLMEDIAKIT_GB28181ROCESS_H
src/Rtp/PSEncoder.cpp
......@@ -167,5 +167,43 @@ void PSEncoder::inputFrame(const Frame::Ptr &frame) {
}
}
////////////////////////////////////////////////////////////////////////////////////////////////
class RingDelegateHelper : public RingDelegate<RtpPacket::Ptr> {
public:
typedef function<void(RtpPacket::Ptr in, bool is_key)> onRtp;
~RingDelegateHelper() override{}
RingDelegateHelper(onRtp on_rtp){
_on_rtp = std::move(on_rtp);
}
void onWrite(RtpPacket::Ptr in, bool is_key) override{
_on_rtp(std::move(in), is_key);
}
private:
onRtp _on_rtp;
};
////////////////////////////////////////////////////////////////////////////////////////////////////////////////
PSEncoderImp::PSEncoderImp(uint32_t ssrc, uint8_t payload_type) {
GET_CONFIG(uint32_t,video_mtu,Rtp::kVideoMtuSize);
_rtp_encoder = std::make_shared<CommonRtpEncoder>(CodecInvalid, ssrc, video_mtu, 90000, payload_type, 0);
_rtp_encoder->setRtpRing(std::make_shared<RtpRing::RingType>());
_rtp_encoder->getRtpRing()->setDelegate(std::make_shared<RingDelegateHelper>([this](RtpPacket::Ptr rtp, bool is_key){
onRTP(std::move(rtp));
}));
InfoL << this << " " << printSSRC(_rtp_encoder->getSsrc());
}
PSEncoderImp::~PSEncoderImp() {
InfoL << this << " " << printSSRC(_rtp_encoder->getSsrc());
}
void PSEncoderImp::onPS(uint32_t stamp, void *packet, size_t bytes) {
_rtp_encoder->inputFrame(std::make_shared<FrameFromPtr>((char *) packet, bytes, stamp));
}
}//namespace mediakit
#endif//defined(ENABLE_RTPPROXY)
src/Rtp/PSEncoder.h
......@@ -14,6 +14,7 @@
#include "mpeg-ps.h"
#include "Common/MediaSink.h"
#include "Common/Stamp.h"
#include "Extension/CommonRtp.h"
namespace mediakit{
//该类实现mpeg-ps容器格式的打包
......@@ -65,6 +66,22 @@ private:
unordered_map<int, track_info> _codec_to_trackid;
};
class PSEncoderImp : public PSEncoder{
public:
PSEncoderImp(uint32_t ssrc, uint8_t payload_type = 96);
~PSEncoderImp() override;
protected:
//rtp打包后回调
virtual void onRTP(Buffer::Ptr rtp) = 0;
protected:
void onPS(uint32_t stamp, void *packet, size_t bytes) override;
private:
std::shared_ptr<CommonRtpEncoder> _rtp_encoder;
};
}//namespace mediakit
#endif //ENABLE_RTPPROXY
#endif //ZLMEDIAKIT_PSENCODER_H
src/Rtp/ProcessInterface.h 0 → 100644
/*
* Copyright (c) 2016 The ZLMediaKit project authors. All Rights Reserved.
*
* This file is part of ZLMediaKit(https://github.com/xiongziliang/ZLMediaKit).
*
* Use of this source code is governed by MIT license that can be found in the
* LICENSE file in the root of the source tree. All contributing project authors
* may be found in the AUTHORS file in the root of the source tree.
*/
#ifndef ZLMEDIAKIT_PROCESSINTERFACE_H
#define ZLMEDIAKIT_PROCESSINTERFACE_H
#include <stdint.h>
#include <memory>
namespace mediakit {
class ProcessInterface {
public:
using Ptr = std::shared_ptr<ProcessInterface>;
ProcessInterface() = default;
virtual ~ProcessInterface() = default;
/**
* 输入rtp
* @param is_udp 是否为udp模式
* @param data rtp数据指针
* @param data_len rtp数据长度
* @return 是否解析成功
*/
virtual bool inputRtp(bool is_udp, const char *data, int data_len) = 0;
};
}//namespace mediakit
#endif //ZLMEDIAKIT_PROCESSINTERFACE_H
src/Rtp/RtpCache.cpp 0 → 100644
/*
* Copyright (c) 2016 The ZLMediaKit project authors. All Rights Reserved.
*
* This file is part of ZLMediaKit(https://github.com/xiongziliang/ZLMediaKit).
*
* Use of this source code is governed by MIT license that can be found in the
* LICENSE file in the root of the source tree. All contributing project authors
* may be found in the AUTHORS file in the root of the source tree.
*/
#include "RtpCache.h"
namespace mediakit{
RtpCache::RtpCache(onFlushed cb) {
_cb = std::move(cb);
}
void RtpCache::onFlush(std::shared_ptr<List<Buffer::Ptr> > rtp_list, bool) {
_cb(std::move(rtp_list));
}
void RtpCache::input(uint64_t stamp, Buffer::Ptr buffer) {
inputPacket(stamp, true, std::move(buffer), false);
}
void RtpCachePS::onRTP(Buffer::Ptr buffer) {
auto rtp = static_pointer_cast<RtpPacket>(buffer);
auto stamp = rtp->timeStamp;
input(stamp, std::move(buffer));
}
}//namespace mediakit
src/Rtp/RtpCache.h 0 → 100644
/*
* Copyright (c) 2016 The ZLMediaKit project authors. All Rights Reserved.
*
* This file is part of ZLMediaKit(https://github.com/xiongziliang/ZLMediaKit).
*
* Use of this source code is governed by MIT license that can be found in the
* LICENSE file in the root of the source tree. All contributing project authors
* may be found in the AUTHORS file in the root of the source tree.
*/
#ifndef ZLMEDIAKIT_RTPCACHE_H
#define ZLMEDIAKIT_RTPCACHE_H
#include "PSEncoder.h"
#include "Extension/CommonRtp.h"
namespace mediakit{
class RtpCache : public PacketCache<Buffer> {
public:
using onFlushed = function<void(std::shared_ptr<List<Buffer::Ptr> >)>;
RtpCache(onFlushed cb);
~RtpCache() override = default;
protected:
/**
* 输入rtp(目的是为了合并写)
* @param buffer rtp数据
*/
void input(uint64_t stamp, Buffer::Ptr buffer);
protected:
void onFlush(std::shared_ptr<List<Buffer::Ptr> > rtp_list, bool) override;
private:
onFlushed _cb;
};
class RtpCachePS : public RtpCache, public PSEncoderImp{
public:
RtpCachePS(onFlushed cb, uint32_t ssrc, uint8_t payload_type = 96) : RtpCache(std::move(cb)), PSEncoderImp(ssrc, payload_type) {};
~RtpCachePS() override = default;
protected:
void onRTP(Buffer::Ptr rtp) override;
};
}//namespace mediakit
#endif //ZLMEDIAKIT_RTPCACHE_H
src/Rtp/RtpProcess.cpp
......@@ -9,14 +9,17 @@
*/
#if defined(ENABLE_RTPPROXY)
#include "GB28181Process.h"
#include "RtpProcess.h"
#include "RtpSplitter.h"
#include "Util/File.h"
#include "Http/HttpTSPlayer.h"
#define RTP_APP_NAME "rtp"
namespace mediakit{
namespace mediakit {
static string printAddress(const struct sockaddr *addr){
static string printAddress(const struct sockaddr *addr) {
return StrPrinter << SockUtil::inet_ntoa(((struct sockaddr_in *) addr)->sin_addr) << ":" << ntohs(((struct sockaddr_in *) addr)->sin_port);
}
......@@ -26,20 +29,11 @@ RtpProcess::RtpProcess(const string &stream_id) {
_media_info._app = RTP_APP_NAME;
_media_info._streamid = stream_id;
GET_CONFIG(string,dump_dir,RtpProxy::kDumpDir);
GET_CONFIG(string, dump_dir, RtpProxy::kDumpDir);
{
FILE *fp = !dump_dir.empty() ? File::create_file(File::absolutePath(_media_info._streamid + ".rtp", dump_dir).data(), "wb") : nullptr;
if(fp){
_save_file_rtp.reset(fp,[](FILE *fp){
fclose(fp);
});
}
}
{
FILE *fp = !dump_dir.empty() ? File::create_file(File::absolutePath(_media_info._streamid + ".mp2", dump_dir).data(), "wb") : nullptr;
if(fp){
_save_file_ps.reset(fp,[](FILE *fp){
if (fp) {
_save_file_rtp.reset(fp, [](FILE *fp) {
fclose(fp);
});
}
......@@ -47,20 +41,16 @@ RtpProcess::RtpProcess(const string &stream_id) {
{
FILE *fp = !dump_dir.empty() ? File::create_file(File::absolutePath(_media_info._streamid + ".video", dump_dir).data(), "wb") : nullptr;
if(fp){
_save_file_video.reset(fp,[](FILE *fp){
if (fp) {
_save_file_video.reset(fp, [](FILE *fp) {
fclose(fp);
});
}
}
_rtp_decoder = std::make_shared<CommonRtpDecoder>(CodecInvalid, 256 * 1024);
_rtp_decoder->addDelegate(std::make_shared<FrameWriterInterfaceHelper>([this](const Frame::Ptr &frame){
onRtpDecode((uint8_t *) frame->data(), frame->size(), frame->dts());
}));
}
RtpProcess::~RtpProcess() {
uint64_t duration = (_last_rtp_time.createdTime() - _last_rtp_time.elapsedTime()) / 1000;
uint64_t duration = (_last_frame_time.createdTime() - _last_frame_time.elapsedTime()) / 1000;
WarnP(this) << "RTP推流器("
<< _media_info._vhost << "/"
<< _media_info._app << "/"
......@@ -79,95 +69,47 @@ RtpProcess::~RtpProcess() {
}
}
bool RtpProcess::inputRtp(const Socket::Ptr &sock, const char *data, int data_len,const struct sockaddr *addr,uint32_t *dts_out) {
GET_CONFIG(bool,check_source,RtpProxy::kCheckSource);
bool RtpProcess::inputRtp(bool is_udp, const Socket::Ptr &sock, const char *data, int len, const struct sockaddr *addr, uint32_t *dts_out) {
GET_CONFIG(bool, check_source, RtpProxy::kCheckSource);
//检查源是否合法
if(!_addr){
if (!_addr) {
_addr = new struct sockaddr;
_sock = sock;
memcpy(_addr,addr, sizeof(struct sockaddr));
memcpy(_addr, addr, sizeof(struct sockaddr));
DebugP(this) << "bind to address:" << printAddress(_addr);
//推流鉴权
emitOnPublish();
}
if(!_muxer){
if (!_muxer) {
//无权限推流
return false;
}
if(check_source && memcmp(_addr,addr,sizeof(struct sockaddr)) != 0){
if (check_source && memcmp(_addr, addr, sizeof(struct sockaddr)) != 0) {
DebugP(this) << "address dismatch:" << printAddress(addr) << " != " << printAddress(_addr);
return false;
}
_total_bytes += data_len;
bool ret = handleOneRtp(0, TrackVideo, 90000, (unsigned char *) data, data_len);
if(dts_out){
*dts_out = _dts;
}
return ret;
}
//判断是否为ts负载
static inline bool checkTS(const uint8_t *packet, int bytes){
return bytes % TS_PACKET_SIZE == 0 && packet[0] == TS_SYNC_BYTE;
}
void RtpProcess::onRtpSorted(const RtpPacket::Ptr &rtp, int) {
if(rtp->sequence != (uint16_t)(_sequence + 1) && _sequence != 0){
WarnP(this) << "rtp丢包:" << rtp->sequence << " != " << _sequence << "+1" << ",公网环境下请使用tcp方式推流";
}
_sequence = rtp->sequence;
if(_save_file_rtp){
uint16_t size = rtp->size() - 4;
_total_bytes += len;
if (_save_file_rtp) {
uint16_t size = len;
size = htons(size);
fwrite((uint8_t *) &size, 2, 1, _save_file_rtp.get());
fwrite((uint8_t *) rtp->data() + 4, rtp->size() - 4, 1, _save_file_rtp.get());
fwrite((uint8_t *) data, len, 1, _save_file_rtp.get());
}
_rtp_decoder->inputRtp(rtp);
}
const char *RtpProcess::onSearchPacketTail(const char *packet,int bytes){
try {
auto ret = _decoder->input((uint8_t *) packet, bytes);
if (ret > 0) {
return packet + ret;
}
return nullptr;
} catch (std::exception &ex) {
InfoL << "解析ps或ts异常: bytes=" << bytes
<< " ,exception=" << ex.what()
<< " ,hex=" << hexdump((uint8_t *) packet, bytes);
return nullptr;
}
}
void RtpProcess::onRtpDecode(const uint8_t *packet, int bytes, uint32_t timestamp) {
if(_save_file_ps){
fwrite((uint8_t *)packet,bytes, 1, _save_file_ps.get());
}
if (!_decoder) {
//创建解码器
if (checkTS(packet, bytes)) {
//猜测是ts负载
InfoP(this) << "judged to be TS";
_decoder = DecoderImp::createDecoder(DecoderImp::decoder_ts, this);
} else {
//猜测是ps负载
InfoP(this) << "judged to be PS";
_decoder = DecoderImp::createDecoder(DecoderImp::decoder_ps, this);
}
if (!_process) {
_process = std::make_shared<GB28181Process>(_media_info, this);
}
if (_decoder) {
HttpRequestSplitter::input((char *) packet, bytes);
bool ret = _process ? _process->inputRtp(is_udp, data, len) : false;
if (dts_out) {
*dts_out = _dts;
}
return ret;
}
void RtpProcess::inputFrame(const Frame::Ptr &frame){
_last_rtp_time.resetTime();
void RtpProcess::inputFrame(const Frame::Ptr &frame) {
_last_frame_time.resetTime();
_dts = frame->dts();
if (_save_file_video && frame->getTrackType() == TrackVideo) {
fwrite((uint8_t *) frame->data(), frame->size(), 1, _save_file_video.get());
......@@ -175,20 +117,20 @@ void RtpProcess::inputFrame(const Frame::Ptr &frame){
_muxer->inputFrame(frame);
}
void RtpProcess::addTrack(const Track::Ptr & track){
void RtpProcess::addTrack(const Track::Ptr &track) {
_muxer->addTrack(track);
}
bool RtpProcess::alive() {
GET_CONFIG(int,timeoutSec,RtpProxy::kTimeoutSec)
if(_last_rtp_time.elapsedTime() / 1000 < timeoutSec){
GET_CONFIG(int, timeoutSec, RtpProxy::kTimeoutSec)
if (_last_frame_time.elapsedTime() / 1000 < timeoutSec) {
return true;
}
return false;
}
void RtpProcess::onDetach(){
if(_on_detach){
void RtpProcess::onDetach() {
if (_on_detach) {
_on_detach();
}
}
......@@ -198,45 +140,45 @@ void RtpProcess::setOnDetach(const function<void()> &cb) {
}
string RtpProcess::get_peer_ip() {
if(_addr){
if (_addr) {
return SockUtil::inet_ntoa(((struct sockaddr_in *) _addr)->sin_addr);
}
return "0.0.0.0";
}
uint16_t RtpProcess::get_peer_port() {
if(!_addr){
if (!_addr) {
return 0;
}
return ntohs(((struct sockaddr_in *) _addr)->sin_port);
}
string RtpProcess::get_local_ip() {
if(_sock){
if (_sock) {
return _sock->get_local_ip();
}
return "0.0.0.0";
}
uint16_t RtpProcess::get_local_port() {
if(_sock){
return _sock->get_local_port();
if (_sock) {
return _sock->get_local_port();
}
return 0;
}
string RtpProcess::getIdentifier() const{
string RtpProcess::getIdentifier() const {
return _media_info._streamid;
}
int RtpProcess::totalReaderCount(){
int RtpProcess::totalReaderCount() {
return _muxer ? _muxer->totalReaderCount() : 0;
}
void RtpProcess::setListener(const std::weak_ptr<MediaSourceEvent> &listener){
if(_muxer){
void RtpProcess::setListener(const std::weak_ptr<MediaSourceEvent> &listener) {
if (_muxer) {
_muxer->setMediaListener(listener);
}else{
} else {
_listener = listener;
}
}
......@@ -262,7 +204,7 @@ void RtpProcess::emitOnPublish() {
//触发推流鉴权事件
auto flag = NoticeCenter::Instance().emitEvent(Broadcast::kBroadcastMediaPublish, _media_info, invoker, static_cast<SockInfo &>(*this));
if(!flag){
if (!flag) {
//该事件无人监听,默认不鉴权
GET_CONFIG(bool, toHls, General::kPublishToHls);
GET_CONFIG(bool, toMP4, General::kPublishToMP4);
......
src/Rtp/RtpProcess.h
......@@ -12,18 +12,14 @@
#define ZLMEDIAKIT_RTPPROCESS_H
#if defined(ENABLE_RTPPROXY)
#include "ProcessInterface.h"
#include "Common/MultiMediaSourceMuxer.h"
#include "Rtsp/RtpReceiver.h"
#include "Decoder.h"
#include "Common/Device.h"
#include "Common/Stamp.h"
#include "Http/HttpRequestSplitter.h"
#include "Extension/CommonRtp.h"
using namespace mediakit;
namespace mediakit{
namespace mediakit {
class RtpProcess : public HttpRequestSplitter, public RtpReceiver, public SockInfo, public MediaSinkInterface, public std::enable_shared_from_this<RtpProcess>{
class RtpProcess : public SockInfo, public MediaSinkInterface, public std::enable_shared_from_this<RtpProcess>{
public:
typedef std::shared_ptr<RtpProcess> Ptr;
friend class RtpProcessHelper;
......@@ -32,14 +28,15 @@ public:
/**
* 输入rtp
* @param is_udp 是否为udp模式
* @param sock 本地监听的socket
* @param data rtp数据指针
* @param data_len rtp数据长度
* @param len rtp数据长度
* @param addr 数据源地址
* @param dts_out 解析出最新的dts
* @return 是否解析成功
*/
bool inputRtp(const Socket::Ptr &sock, const char *data,int data_len, const struct sockaddr *addr , uint32_t *dts_out = nullptr);
bool inputRtp(bool is_udp, const Socket::Ptr &sock, const char *data, int len, const struct sockaddr *addr , uint32_t *dts_out = nullptr);
/**
* 是否超时，用于超时移除对象
......@@ -67,34 +64,26 @@ public:
void setListener(const std::weak_ptr<MediaSourceEvent> &listener);
protected:
void onRtpSorted(const RtpPacket::Ptr &rtp, int track_index) override ;
void inputFrame(const Frame::Ptr &frame) override;
void addTrack(const Track::Ptr & track) override;
void resetTracks() override {};
const char *onSearchPacketTail(const char *data,int len) override;
int64_t onRecvHeader(const char *data,uint64_t len) override { return 0; };
private:
void emitOnPublish();
void onRtpDecode(const uint8_t *packet, int bytes, uint32_t timestamp);
private:
std::shared_ptr<CommonRtpDecoder> _rtp_decoder;
std::shared_ptr<FILE> _save_file_rtp;
std::shared_ptr<FILE> _save_file_ps;
std::shared_ptr<FILE> _save_file_video;
struct sockaddr *_addr = nullptr;
uint16_t _sequence = 0;
MultiMediaSourceMuxer::Ptr _muxer;
Ticker _last_rtp_time;
uint32_t _dts = 0;
DecoderImp::Ptr _decoder;
std::weak_ptr<MediaSourceEvent> _listener;
MediaInfo _media_info;
uint64_t _total_bytes = 0;
struct sockaddr *_addr = nullptr;
Socket::Ptr _sock;
MediaInfo _media_info;
Ticker _last_frame_time;
function<void()> _on_detach;
std::shared_ptr<FILE> _save_file_rtp;
std::shared_ptr<FILE> _save_file_video;
std::weak_ptr<MediaSourceEvent> _listener;
ProcessInterface::Ptr _process;
MultiMediaSourceMuxer::Ptr _muxer;
};
}//namespace mediakit
......
src/Rtp/RtpSelector.cpp
......@@ -9,7 +9,9 @@
*/
#if defined(ENABLE_RTPPROXY)
#include <stddef.h>
#include "RtpSelector.h"
#include "RtpSplitter.h"
namespace mediakit{
......@@ -22,17 +24,19 @@ void RtpSelector::clear(){
bool RtpSelector::inputRtp(const Socket::Ptr &sock, const char *data, int data_len,
const struct sockaddr *addr,uint32_t *dts_out) {
//使用ssrc为流id
uint32_t ssrc = 0;
if (!getSSRC(data, data_len, ssrc)) {
WarnL << "get ssrc from rtp failed:" << data_len;
return false;
}
//假定指定了流id，那么通过流id来区分是否为一路流(哪怕可能同时收到多路流)
auto process = getProcess(printSSRC(ssrc), true);
if (process) {
return process->inputRtp(sock, data, data_len, addr, dts_out);
try {
return process->inputRtp(true, sock, data, data_len, addr, dts_out);
} catch (...) {
delProcess(printSSRC(ssrc), process.get());
throw;
}
}
return false;
}
......
src/Rtp/PSRtpSender.cpp → src/Rtp/RtpSender.cpp
......@@ -9,33 +9,29 @@
*/
#if defined(ENABLE_RTPPROXY)
#include "PSRtpSender.h"
#include "RtpSender.h"
#include "Rtsp/RtspSession.h"
#include "Thread/WorkThreadPool.h"
#include "RtpCache.h"
namespace mediakit{
PSRtpSender::PSRtpSender(uint32_t ssrc, uint8_t payload_type) {
GET_CONFIG(uint32_t,video_mtu,Rtp::kVideoMtuSize);
_rtp_encoder = std::make_shared<CommonRtpEncoder>(CodecInvalid, ssrc, video_mtu, 90000, payload_type, 0);
_rtp_encoder->setRtpRing(std::make_shared<RtpRing::RingType>());
_rtp_encoder->getRtpRing()->setDelegate(std::make_shared<RingDelegateHelper>([this](const RtpPacket::Ptr &rtp, bool is_key){
onRtp(rtp, is_key);
}));
RtpSender::RtpSender(uint32_t ssrc, uint8_t payload_type) {
_poller = EventPollerPool::Instance().getPoller();
InfoL << this << " " << printSSRC(_rtp_encoder->getSsrc());
_interface = std::make_shared<RtpCachePS>([this](std::shared_ptr<List<Buffer::Ptr> > list) {
onFlushRtpList(std::move(list));
}, ssrc, payload_type);
}
PSRtpSender::~PSRtpSender() {
InfoL << this << " " << printSSRC(_rtp_encoder->getSsrc());
RtpSender::~RtpSender() {
}
void PSRtpSender::startSend(const string &dst_url, uint16_t dst_port, bool is_udp, const function<void(const SockException &ex)> &cb){
void RtpSender::startSend(const string &dst_url, uint16_t dst_port, bool is_udp, const function<void(const SockException &ex)> &cb){
_is_udp = is_udp;
_socket = Socket::createSocket(_poller, false);
_dst_url = dst_url;
_dst_port = dst_port;
weak_ptr<PSRtpSender> weak_self = shared_from_this();
weak_ptr<RtpSender> weak_self = shared_from_this();
if (is_udp) {
_socket->bindUdpSock(0);
auto poller = _poller;
......@@ -73,7 +69,7 @@ void PSRtpSender::startSend(const string &dst_url, uint16_t dst_port, bool is_ud
}
}
void PSRtpSender::onConnect(){
void RtpSender::onConnect(){
_is_connect = true;
//加大发送缓存,防止udp丢包之类的问题
SockUtil::setSendBuf(_socket->rawFD(), 4 * 1024 * 1024);
......@@ -83,37 +79,38 @@ void PSRtpSender::onConnect(){
_socket->setSendFlags(SOCKET_DEFAULE_FLAGS | FLAG_MORE);
}
//连接建立成功事件
weak_ptr<PSRtpSender> weak_self = shared_from_this();
weak_ptr<RtpSender> weak_self = shared_from_this();
_socket->setOnErr([weak_self](const SockException &err) {
auto strong_self = weak_self.lock();
if (strong_self) {
strong_self->onErr(err);
}
});
InfoL << "开始发送 ps rtp:" << _socket->get_peer_ip() << ":" << _socket->get_peer_port() << ", 是否为udp方式:" << _is_udp;
InfoL << "开始发送 rtp:" << _socket->get_peer_ip() << ":" << _socket->get_peer_port() << ", 是否为udp方式:" << _is_udp;
}
//此函数在其他线程执行
void PSRtpSender::inputFrame(const Frame::Ptr &frame) {
if (_is_connect) {
//连接成功后才做实质操作(节省cpu资源)
PSEncoder::inputFrame(frame);
}
void RtpSender::addTrack(const Track::Ptr &track){
_interface->addTrack(track);
}
//此函数在其他线程执行
void PSRtpSender::onPS(uint32_t stamp, void *packet, size_t bytes) {
_rtp_encoder->inputFrame(std::make_shared<FrameFromPtr>((char *) packet, bytes, stamp));
void RtpSender::addTrackCompleted(){
_interface->addTrackCompleted();
}
void RtpSender::resetTracks(){
_interface->resetTracks();
}
//此函数在其他线程执行
void PSRtpSender::onRtp(const RtpPacket::Ptr &rtp, bool) {
//开启合并写提高发送性能
PacketCache<RtpPacket>::inputPacket(true, rtp, false);
void RtpSender::inputFrame(const Frame::Ptr &frame) {
if (_is_connect) {
//连接成功后才做实质操作(节省cpu资源)
_interface->inputFrame(frame);
}
}
//此函数在其他线程执行
void PSRtpSender::onFlush(shared_ptr<List<RtpPacket::Ptr> > rtp_list, bool) {
void RtpSender::onFlushRtpList(shared_ptr<List<Buffer::Ptr> > rtp_list) {
if(!_is_connect){
//连接成功后才能发送数据
return;
......@@ -124,29 +121,29 @@ void PSRtpSender::onFlush(shared_ptr<List<RtpPacket::Ptr> > rtp_list, bool) {
_poller->async([rtp_list, is_udp, socket]() {
int i = 0;
int size = rtp_list->size();
rtp_list->for_each([&](const RtpPacket::Ptr &packet) {
rtp_list->for_each([&](Buffer::Ptr &packet) {
if (is_udp) {
//udp模式，rtp over tcp前4个字节可以忽略
socket->send(std::make_shared<BufferRtp>(packet, 4), nullptr, 0, ++i == size);
socket->send(std::make_shared<BufferRtp>(std::move(packet), 4), nullptr, 0, ++i == size);
} else {
//tcp模式, rtp over tcp前2个字节可以忽略,只保留后续rtp长度的2个字节
socket->send(std::make_shared<BufferRtp>(packet, 2), nullptr, 0, ++i == size);
socket->send(std::make_shared<BufferRtp>(std::move(packet), 2), nullptr, 0, ++i == size);
}
});
});
}
void PSRtpSender::onErr(const SockException &ex, bool is_connect) {
void RtpSender::onErr(const SockException &ex, bool is_connect) {
_is_connect = false;
//监听socket断开事件，方便重连
if (is_connect) {
WarnL << "重连" << _dst_url << ":" << _dst_port << "失败, 原因为:" << ex.what();
} else {
WarnL << "停止发送 ps rtp:" << _dst_url << ":" << _dst_port << ", 原因为:" << ex.what();
WarnL << "停止发送 rtp:" << _dst_url << ":" << _dst_port << ", 原因为:" << ex.what();
}
weak_ptr<PSRtpSender> weak_self = shared_from_this();
weak_ptr<RtpSender> weak_self = shared_from_this();
_connect_timer = std::make_shared<Timer>(10.0, [weak_self]() {
auto strong_self = weak_self.lock();
if (!strong_self) {
......
src/Rtp/PSRtpSender.h → src/Rtp/RtpSender.h
......@@ -8,42 +8,27 @@
* may be found in the AUTHORS file in the root of the source tree.
*/
#ifndef ZLMEDIAKIT_PSRTPSENDER_H
#define ZLMEDIAKIT_PSRTPSENDER_H
#ifndef ZLMEDIAKIT_RTPSENDER_H
#define ZLMEDIAKIT_RTPSENDER_H
#if defined(ENABLE_RTPPROXY)
#include "PSEncoder.h"
#include "Extension/CommonRtp.h"
namespace mediakit{
class RingDelegateHelper : public RingDelegate<RtpPacket::Ptr> {
//rtp发送客户端，支持发送GB28181协议
class RtpSender : public MediaSinkInterface, public std::enable_shared_from_this<RtpSender>{
public:
typedef function<void(const RtpPacket::Ptr &in, bool is_key)> onRtp;
typedef std::shared_ptr<RtpSender> Ptr;
~RingDelegateHelper() override{}
RingDelegateHelper(onRtp on_rtp){
_on_rtp = std::move(on_rtp);
}
void onWrite(RtpPacket::Ptr in, bool is_key) override{
_on_rtp(in, is_key);
}
private:
onRtp _on_rtp;
};
//该类在PSEncoder的基础上，实现了mpeg-ps的rtp打包以及发送
class PSRtpSender : public PSEncoder, public std::enable_shared_from_this<PSRtpSender>, public PacketCache<RtpPacket>{
public:
typedef std::shared_ptr<PSRtpSender> Ptr;
~RtpSender() override;
/**
* 构造函数
* 构造函数，创建GB28181 RTP发送客户端
* @param ssrc rtp的ssrc
* @param payload_type 国标中ps-rtp的pt一般为96
*/
PSRtpSender(uint32_t ssrc, uint8_t payload_type = 96);
~PSRtpSender() override;
RtpSender(uint32_t ssrc, uint8_t payload_type = 96);
/**
* 开始发送ps-rtp包
......@@ -59,20 +44,26 @@ public:
*/
void inputFrame(const Frame::Ptr &frame) override;
protected:
//mpeg-ps回调
void onPS(uint32_t stamp, void *packet, size_t bytes) override;
/**
* 添加track，内部会调用Track的clone方法
* 只会克隆sps pps这些信息 ，而不会克隆Delegate相关关系
* @param track
*/
virtual void addTrack(const Track::Ptr & track) override;
/**
* 添加所有Track完毕
*/
virtual void addTrackCompleted() override;
/**
* 批量flush rtp包时触发该函数
* @param rtp_list rtp包列表
* @param key_pos 是否包含关键帧
* 重置track
*/
void onFlush(std::shared_ptr<List<RtpPacket::Ptr> > rtp_list, bool key_pos) override;
virtual void resetTracks() override;
private:
//rtp打包后回调
void onRtp(const RtpPacket::Ptr &in, bool is_key);
//合并写输出
void onFlushRtpList(std::shared_ptr<List<Buffer::Ptr> > rtp_list);
//udp/tcp连接成功回调
void onConnect();
//异常断开socket事件
......@@ -86,9 +77,9 @@ private:
Socket::Ptr _socket;
EventPoller::Ptr _poller;
Timer::Ptr _connect_timer;
std::shared_ptr<CommonRtpEncoder> _rtp_encoder;
MediaSinkInterface::Ptr _interface;
};
}//namespace mediakit
#endif// defined(ENABLE_RTPPROXY)
#endif //ZLMEDIAKIT_PSRTPSENDER_H
#endif //ZLMEDIAKIT_RTPSENDER_H
src/Rtp/RtpServer.cpp
......@@ -52,7 +52,7 @@ void RtpServer::start(uint16_t local_port, const string &stream_id, bool enable
//指定了流id，那么一个端口一个流(不管是否包含多个ssrc的多个流，绑定rtp源后，会筛选掉ip端口不匹配的流)
process = RtpSelector::Instance().getProcess(stream_id, true);
udp_server->setOnRead([udp_server, process](const Buffer::Ptr &buf, struct sockaddr *addr, int) {
process->inputRtp(udp_server, buf->data(), buf->size(), addr);
process->inputRtp(true, udp_server, buf->data(), buf->size(), addr);
});
} else {
//未指定流id，一个端口多个流，通过ssrc来分流
......
src/Rtp/RtpSession.cpp
......@@ -73,7 +73,7 @@ void RtpSession::onRtpPacket(const char *data, uint64_t len) {
_process = RtpSelector::Instance().getProcess(_stream_id, true);
_process->setListener(dynamic_pointer_cast<RtpSession>(shared_from_this()));
}
_process->inputRtp(getSock(), data, len, &addr);
_process->inputRtp(false, getSock(), data, len, &addr);
_ticker.resetTime();
}
......
src/Rtp/RtpSession.h
......@@ -35,13 +35,14 @@ protected:
bool close(MediaSource &sender,bool force) override;
// 观看总人数
int totalReaderCount(MediaSource &sender) override;
void onRtpPacket(const char *data,uint64_t len) override;
// 收到rtp回调
void onRtpPacket(const char *data, uint64_t len) override;
private:
RtpProcess::Ptr _process;
Ticker _ticker;
struct sockaddr addr;
string _stream_id;
struct sockaddr addr;
RtpProcess::Ptr _process;
};
}//namespace mediakit
......
Markdown 格式
0% 或
您添加了 0 人 到此讨论。请谨慎行事。
请先完成此评论的编辑！
请 注册 或者 后发表评论