Mirror/ZLMediaKit
1
0
Fork 0
mirror of https://github.com/ZLMediaKit/ZLMediaKit.git synced 2026-07-07 04:08:13 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'master' of github.com:ZLMediaKit/ZLMediaKit into transcode2

Browse Source 
# Conflicts:
#	CMakeLists.txt
#	conf/config.ini
#	src/Common/MediaSink.cpp
#	src/Common/MediaSink.h
#	src/Common/MediaSource.cpp
#	src/Common/MultiMediaSourceMuxer.h
#	src/Common/config.cpp
#	src/Common/config.h
#	src/Extension/AAC.cpp
#	src/Extension/AAC.h
#	src/Rtsp/RtpCodec.h
#	src/Rtsp/RtspMuxer.cpp
#	src/Rtsp/RtspMuxer.h
#	webrtc/Nack.cpp
#	webrtc/WebRtcTransport.cpp
This commit is contained in:
cqm
2024-07-12 09:48:01 +08:00
parent 9bf48567b2 3e7e1a317e
commit 2ffdfab71e
465 changed files with 18124 additions and 9008 deletions
Download Patch File Download Diff File Expand all files Collapse all files

203
src/Rtp/Decoder.cpp
View File

@@ -1,9 +1,9 @@
/*
* Copyright (c) 2016 The ZLMediaKit project authors. All Rights Reserved.
* Copyright (c) 2016-present The ZLMediaKit project authors. All Rights Reserved.
*
* This file is part of ZLMediaKit(https://github.com/xia-chu/ZLMediaKit).
* This file is part of ZLMediaKit(https://github.com/ZLMediaKit/ZLMediaKit).
*
* Use of this source code is governed by MIT license that can be found in the
* Use of this source code is governed by MIT-like 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.
*/
@@ -11,11 +11,7 @@
#include "Decoder.h"
#include "PSDecoder.h"
#include "TSDecoder.h"
#include "Extension/H264.h"
#include "Extension/H265.h"
#include "Extension/AAC.h"
#include "Extension/G711.h"
#include "Extension/Opus.h"
#include "Extension/Factory.h"
#if defined(ENABLE_RTPPROXY) || defined(ENABLE_HLS)
#include "mpeg-ts.h"
@@ -66,7 +62,9 @@ DecoderImp::Ptr DecoderImp::createDecoder(Type type, MediaSinkInterface *sink){
}
void DecoderImp::flush() {
_merger.flush();
for (auto &pr : _tracks) {
pr.second.second.flush();
}
}
ssize_t DecoderImp::input(const uint8_t *data, size_t bytes){
@@ -85,164 +83,71 @@ DecoderImp::DecoderImp(const Decoder::Ptr &decoder, MediaSinkInterface *sink){
}
#if defined(ENABLE_RTPPROXY) || defined(ENABLE_HLS)
#define SWITCH_CASE(codec_id) case codec_id : return #codec_id
static const char *getCodecName(int codec_id) {
switch (codec_id) {
SWITCH_CASE(PSI_STREAM_MPEG1);
SWITCH_CASE(PSI_STREAM_MPEG2);
SWITCH_CASE(PSI_STREAM_AUDIO_MPEG1);
SWITCH_CASE(PSI_STREAM_MP3);
SWITCH_CASE(PSI_STREAM_AAC);
SWITCH_CASE(PSI_STREAM_MPEG4);
SWITCH_CASE(PSI_STREAM_MPEG4_AAC_LATM);
SWITCH_CASE(PSI_STREAM_H264);
SWITCH_CASE(PSI_STREAM_MPEG4_AAC);
SWITCH_CASE(PSI_STREAM_H265);
SWITCH_CASE(PSI_STREAM_AUDIO_AC3);
SWITCH_CASE(PSI_STREAM_AUDIO_EAC3);
SWITCH_CASE(PSI_STREAM_AUDIO_DTS);
SWITCH_CASE(PSI_STREAM_VIDEO_DIRAC);
SWITCH_CASE(PSI_STREAM_VIDEO_VC1);
SWITCH_CASE(PSI_STREAM_VIDEO_SVAC);
SWITCH_CASE(PSI_STREAM_AUDIO_SVAC);
SWITCH_CASE(PSI_STREAM_AUDIO_G711A);
SWITCH_CASE(PSI_STREAM_AUDIO_G711U);
SWITCH_CASE(PSI_STREAM_AUDIO_G722);
SWITCH_CASE(PSI_STREAM_AUDIO_G723);
SWITCH_CASE(PSI_STREAM_AUDIO_G729);
SWITCH_CASE(PSI_STREAM_AUDIO_OPUS);
default : return "unknown codec";
void DecoderImp::onStream(int stream, int codecid, const void *extra, size_t bytes, int finish) {
// G711传统只支持 8000/1/16的规格，FFmpeg貌似做了扩展，但是这里不管它了
auto track = Factory::getTrackByCodecId(getCodecByMpegId(codecid), 8000, 1, 16);
if (track) {
onTrack(stream, std::move(track));
}
}
void DecoderImp::onStream(int stream, int codecid, const void *extra, size_t bytes, int finish){
switch (codecid) {
case PSI_STREAM_H264: {
onTrack(std::make_shared<H264Track>());
break;
}
case PSI_STREAM_H265: {
onTrack(std::make_shared<H265Track>());
break;
}
case PSI_STREAM_MPEG4_AAC :
case PSI_STREAM_AAC: {
onTrack(std::make_shared<AACTrack>());
break;
}
case PSI_STREAM_AUDIO_G711A:
case PSI_STREAM_AUDIO_G711U: {
auto codec = codecid == PSI_STREAM_AUDIO_G711A ? CodecG711A : CodecG711U;
//G711传统只支持 8000/1/16的规格，FFmpeg貌似做了扩展，但是这里不管它了
onTrack(std::make_shared<G711Track>(codec, 8000, 1, 16));
break;
}
case PSI_STREAM_AUDIO_OPUS: {
onTrack(std::make_shared<OpusTrack>());
break;
}
default:
if(codecid != 0){
WarnL<< "unsupported codec type:" << getCodecName(codecid) << " " << (int)codecid;
}
break;
}
//防止未获取视频track提前complete导致忽略后续视频的问题，用于兼容一些不太规范的ps流
if (finish && _tracks[TrackVideo] ) {
// 防止未获取视频track提前complete导致忽略后续视频的问题，用于兼容一些不太规范的ps流
if (finish && _have_video) {
_sink->addTrackCompleted();
InfoL << "add track finished";
InfoL << "Add track finished";
}
}
void DecoderImp::onDecode(int stream,int codecid,int flags,int64_t pts,int64_t dts,const void *data,size_t bytes) {
void DecoderImp::onDecode(int stream, int codecid, int flags, int64_t pts, int64_t dts, const void *data, size_t bytes) {
pts /= 90;
dts /= 90;
switch (codecid) {
case PSI_STREAM_H264: {
if (!_tracks[TrackVideo]) {
onTrack(std::make_shared<H264Track>());
}
auto frame = std::make_shared<H264FrameNoCacheAble>((char *) data, bytes, (uint64_t)dts, (uint64_t)pts, prefixSize((char *) data, bytes));
_merger.inputFrame(frame,[this](uint64_t dts, uint64_t pts, const Buffer::Ptr &buffer, bool) {
onFrame(std::make_shared<FrameWrapper<H264FrameNoCacheAble> >(buffer, dts, pts, prefixSize(buffer->data(), buffer->size()), 0));
});
break;
}
case PSI_STREAM_H265: {
if (!_tracks[TrackVideo]) {
onTrack(std::make_shared<H265Track>());
}
auto frame = std::make_shared<H265FrameNoCacheAble>((char *) data, bytes, (uint64_t)dts, (uint64_t)pts, prefixSize((char *) data, bytes));
_merger.inputFrame(frame,[this](uint64_t dts, uint64_t pts, const Buffer::Ptr &buffer, bool) {
onFrame(std::make_shared<FrameWrapper<H265FrameNoCacheAble> >(buffer, dts, pts, prefixSize(buffer->data(), buffer->size()), 0));
});
break;
}
case PSI_STREAM_MPEG4_AAC :
case PSI_STREAM_AAC: {
uint8_t *ptr = (uint8_t *)data;
if(!(bytes > 7 && ptr[0] == 0xFF && (ptr[1] & 0xF0) == 0xF0)){
//这不是aac
break;
}
if (!_tracks[TrackAudio]) {
onTrack(std::make_shared<AACTrack>());
}
onFrame(std::make_shared<FrameFromPtr>(CodecAAC, (char *) data, bytes, (uint64_t)dts, 0, ADTS_HEADER_LEN));
break;
}
case PSI_STREAM_AUDIO_G711A:
case PSI_STREAM_AUDIO_G711U: {
auto codec = codecid == PSI_STREAM_AUDIO_G711A ? CodecG711A : CodecG711U;
if (!_tracks[TrackAudio]) {
//G711传统只支持 8000/1/16的规格，FFmpeg貌似做了扩展，但是这里不管它了
onTrack(std::make_shared<G711Track>(codec, 8000, 1, 16));
}
onFrame(std::make_shared<FrameFromPtr>(codec, (char *) data, bytes, (uint64_t)dts));
break;
}
case PSI_STREAM_AUDIO_OPUS: {
if (!_tracks[TrackAudio]) {
onTrack(std::make_shared<OpusTrack>());
}
onFrame(std::make_shared<FrameFromPtr>(CodecOpus, (char *) data, bytes, (uint64_t)dts));
break;
}
default:
// 海康的 PS 流中会有 codecid 为 0xBD 的包
if (codecid != 0 && codecid != 0xBD) {
WarnL << "unsupported codec type:" << getCodecName(codecid) << " " << (int) codecid;
}
break;
auto codec = getCodecByMpegId(codecid);
if (codec == CodecInvalid) {
return;
}
auto &ref = _tracks[stream];
if (!ref.first) {
onTrack(stream, Factory::getTrackByCodecId(codec, 8000, 1, 16));
}
if (!ref.first) {
WarnL << "Unsupported codec :" << getCodecName(codec);
return;
}
auto frame = Factory::getFrameFromPtr(codec, (char *)data, bytes, dts, pts);
if (getTrackType(codec) != TrackVideo) {
onFrame(stream, frame);
return;
}
ref.second.inputFrame(frame, [this, stream, codec](uint64_t dts, uint64_t pts, const Buffer::Ptr &buffer, bool) {
onFrame(stream, Factory::getFrameFromBuffer(codec, buffer, dts, pts));
});
}
#else
void DecoderImp::onDecode(int stream,int codecid,int flags,int64_t pts,int64_t dts,const void *data,size_t bytes) {}
void DecoderImp::onStream(int stream,int codecid,const void *extra,size_t bytes,int finish) {}
#endif
void DecoderImp::onTrack(const Track::Ptr &track) {
if (!_tracks[track->getTrackType()]) {
_tracks[track->getTrackType()] = track;
_sink->addTrack(track);
InfoL << "got track: " << track->getCodecName();
void DecoderImp::onTrack(int index, const Track::Ptr &track) {
if (!track) {
return;
}
track->setIndex(index);
auto &ref = _tracks[index];
if (ref.first) {
WarnL << "Already existed a same track: " << index << ", codec: " << track->getCodecName();
return;
}
ref.first = track;
_sink->addTrack(track);
InfoL << "Got track: " << track->getCodecName();
_have_video = track->getTrackType() == TrackVideo ? true : _have_video;
}
void DecoderImp::onFrame(const Frame::Ptr &frame) {
_sink->inputFrame(frame);
void DecoderImp::onFrame(int index, const Frame::Ptr &frame) {
if (frame) {
frame->setIndex(index);
_sink->inputFrame(frame);
}
}
}//namespace mediakit

22
src/Rtp/Decoder.h
View File

@@ -1,9 +1,9 @@
/*
* Copyright (c) 2016 The ZLMediaKit project authors. All Rights Reserved.
* Copyright (c) 2016-present The ZLMediaKit project authors. All Rights Reserved.
*
* This file is part of ZLMediaKit(https://github.com/xia-chu/ZLMediaKit).
* This file is part of ZLMediaKit(https://github.com/ZLMediaKit/ZLMediaKit).
*
* Use of this source code is governed by MIT license that can be found in the
* Use of this source code is governed by MIT-like 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.
*/
@@ -37,20 +37,19 @@ protected:
onStream _on_stream;
};
class DecoderImp{
class DecoderImp {
public:
typedef enum { decoder_ts = 0, decoder_ps } Type;
using Ptr = std::shared_ptr<DecoderImp>;
~DecoderImp() = default;
static Ptr createDecoder(Type type, MediaSinkInterface *sink);
ssize_t input(const uint8_t *data, size_t bytes);
void flush();
protected:
void onTrack(const Track::Ptr &track);
void onFrame(const Frame::Ptr &frame);
void onTrack(int index, const Track::Ptr &track);
void onFrame(int index, const Frame::Ptr &frame);
private:
DecoderImp(const Decoder::Ptr &decoder, MediaSinkInterface *sink);
@@ -58,10 +57,15 @@ private:
void onStream(int stream, int codecid, const void *extra, size_t bytes, int finish);
private:
bool _have_video = false;
Decoder::Ptr _decoder;
MediaSinkInterface *_sink;
FrameMerger _merger{FrameMerger::none};
Track::Ptr _tracks[TrackMax];
class FrameMergerImp : public FrameMerger {
public:
FrameMergerImp() : FrameMerger(FrameMerger::none) {}
};
std::unordered_map<int, std::pair<Track::Ptr, FrameMergerImp> > _tracks;
};
}//namespace mediakit

50
src/Rtp/GB28181Process.cpp
View File

@@ -1,9 +1,9 @@
/*
* Copyright (c) 2016 The ZLMediaKit project authors. All Rights Reserved.
* Copyright (c) 2016-present The ZLMediaKit project authors. All Rights Reserved.
*
* This file is part of ZLMediaKit(https://github.com/xia-chu/ZLMediaKit).
* This file is part of ZLMediaKit(https://github.com/ZLMediaKit/ZLMediaKit).
*
* Use of this source code is governed by MIT license that can be found in the
* Use of this source code is governed by MIT-like 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.
*/
@@ -12,11 +12,6 @@
#include "GB28181Process.h"
#include "Extension/CommonRtp.h"
#include "Extension/Factory.h"
#include "Extension/G711.h"
#include "Extension/H264.h"
#include "Extension/H265.h"
#include "Extension/Opus.h"
#include "Extension/JPEG.h"
#include "Http/HttpTSPlayer.h"
#include "Util/File.h"
#include "Common/config.h"
@@ -45,8 +40,6 @@ public:
setNtpStamp(0, 0);
}
~RtpReceiverImp() override = default;
bool inputRtp(TrackType type, uint8_t *ptr, size_t len) {
return RtpTrack::inputRtp(type, _sample_rate, ptr, len).operator bool();
}
@@ -85,48 +78,58 @@ bool GB28181Process::inputRtp(bool, const char *data, size_t data_len) {
if (!ref) {
if (_rtp_receiver.size() > 2) {
// 防止pt类型太多导致内存溢出
throw std::invalid_argument("rtp pt类型不得超过2种!");
WarnL << "Rtp payload type more than 2 types: " << _rtp_receiver.size();
}
switch (pt) {
case Rtsp::PT_PCMA:
case Rtsp::PT_PCMU: {
// CodecG711U or CodecG711A
ref = std::make_shared<RtpReceiverImp>(8000, [this](RtpPacket::Ptr rtp) { onRtpSorted(std::move(rtp)); });
auto track = std::make_shared<G711Track>(pt == Rtsp::PT_PCMU ? CodecG711U : CodecG711A, 8000, 1, 16);
auto track = Factory::getTrackByCodecId(pt == Rtsp::PT_PCMU ? CodecG711U : CodecG711A, 8000, 1, 16);
CHECK(track);
track->setIndex(pt);
_interface->addTrack(track);
_rtp_decoder[pt] = Factory::getRtpDecoderByTrack(track);
_rtp_decoder[pt] = Factory::getRtpDecoderByCodecId(track->getCodecId());
break;
}
case Rtsp::PT_JPEG: {
// mjpeg
ref = std::make_shared<RtpReceiverImp>(90000, [this](RtpPacket::Ptr rtp) { onRtpSorted(std::move(rtp)); });
auto track = std::make_shared<JPEGTrack>();
auto track = Factory::getTrackByCodecId(CodecJPEG);
CHECK(track);
track->setIndex(pt);
_interface->addTrack(track);
_rtp_decoder[pt] = Factory::getRtpDecoderByTrack(track);
_rtp_decoder[pt] = Factory::getRtpDecoderByCodecId(track->getCodecId());
break;
}
default: {
if (pt == opus_pt) {
// opus负载
ref = std::make_shared<RtpReceiverImp>(48000, [this](RtpPacket::Ptr rtp) { onRtpSorted(std::move(rtp)); });
auto track = std::make_shared<OpusTrack>();
auto track = Factory::getTrackByCodecId(CodecOpus);
CHECK(track);
track->setIndex(pt);
_interface->addTrack(track);
_rtp_decoder[pt] = Factory::getRtpDecoderByTrack(track);
_rtp_decoder[pt] = Factory::getRtpDecoderByCodecId(track->getCodecId());
} else if (pt == h265_pt) {
// H265负载
ref = std::make_shared<RtpReceiverImp>(90000, [this](RtpPacket::Ptr rtp) { onRtpSorted(std::move(rtp)); });
auto track = std::make_shared<H265Track>();
auto track = Factory::getTrackByCodecId(CodecH265);
CHECK(track);
track->setIndex(pt);
_interface->addTrack(track);
_rtp_decoder[pt] = Factory::getRtpDecoderByTrack(track);
_rtp_decoder[pt] = Factory::getRtpDecoderByCodecId(track->getCodecId());
} else if (pt == h264_pt) {
// H264负载
ref = std::make_shared<RtpReceiverImp>(90000, [this](RtpPacket::Ptr rtp) { onRtpSorted(std::move(rtp)); });
auto track = std::make_shared<H264Track>();
auto track = Factory::getTrackByCodecId(CodecH264);
CHECK(track);
track->setIndex(pt);
_interface->addTrack(track);
_rtp_decoder[pt] = Factory::getRtpDecoderByTrack(track);
_rtp_decoder[pt] = Factory::getRtpDecoderByCodecId(track->getCodecId());
} else {
if (pt != Rtsp::PT_MP2T && pt != ps_pt) {
WarnL << "rtp payload type未识别(" << (int)pt << "),已按ts或ps负载处理";
WarnL << "Unknown rtp payload type(" << (int)pt << "), decode it as mpeg-ps or mpeg-ts";
}
ref = std::make_shared<RtpReceiverImp>(90000, [this](RtpPacket::Ptr rtp) { onRtpSorted(std::move(rtp)); });
// ts或ps负载
@@ -146,7 +149,8 @@ bool GB28181Process::inputRtp(bool, const char *data, size_t data_len) {
}
}
// 设置frame回调
_rtp_decoder[pt]->addDelegate([this](const Frame::Ptr &frame) {
_rtp_decoder[pt]->addDelegate([this, pt](const Frame::Ptr &frame) {
frame->setIndex(pt);
onRtpDecode(frame);
return true;
});

7
src/Rtp/GB28181Process.h
View File

@@ -1,9 +1,9 @@
/*
* Copyright (c) 2016 The ZLMediaKit project authors. All Rights Reserved.
* Copyright (c) 2016-present The ZLMediaKit project authors. All Rights Reserved.
*
* This file is part of ZLMediaKit(https://github.com/xia-chu/ZLMediaKit).
* This file is part of ZLMediaKit(https://github.com/ZLMediaKit/ZLMediaKit).
*
* Use of this source code is governed by MIT license that can be found in the
* Use of this source code is governed by MIT-like 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.
*/
@@ -27,7 +27,6 @@ public:
using Ptr = std::shared_ptr<GB28181Process>;
GB28181Process(const MediaInfo &media_info, MediaSinkInterface *sink);
~GB28181Process() override = default;
/**
* 输入rtp

6
src/Rtp/PSDecoder.cpp
View File

@@ -1,9 +1,9 @@
/*
* Copyright (c) 2016 The ZLMediaKit project authors. All Rights Reserved.
* Copyright (c) 2016-present The ZLMediaKit project authors. All Rights Reserved.
*
* This file is part of ZLMediaKit(https://github.com/xia-chu/ZLMediaKit).
* This file is part of ZLMediaKit(https://github.com/ZLMediaKit/ZLMediaKit).
*
* Use of this source code is governed by MIT license that can be found in the
* Use of this source code is governed by MIT-like 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.
*/

6
src/Rtp/PSDecoder.h
View File

@@ -1,9 +1,9 @@
/*
* Copyright (c) 2016 The ZLMediaKit project authors. All Rights Reserved.
* Copyright (c) 2016-present The ZLMediaKit project authors. All Rights Reserved.
*
* This file is part of ZLMediaKit(https://github.com/xia-chu/ZLMediaKit).
* This file is part of ZLMediaKit(https://github.com/ZLMediaKit/ZLMediaKit).
*
* Use of this source code is governed by MIT license that can be found in the
* Use of this source code is governed by MIT-like 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.
*/

35
src/Rtp/PSEncoder.cpp
View File

@@ -1,9 +1,9 @@
/*
* Copyright (c) 2016 The ZLMediaKit project authors. All Rights Reserved.
* Copyright (c) 2016-present The ZLMediaKit project authors. All Rights Reserved.
*
* This file is part of ZLMediaKit(https://github.com/xia-chu/ZLMediaKit).
* This file is part of ZLMediaKit(https://github.com/ZLMediaKit/ZLMediaKit).
*
* Use of this source code is governed by MIT license that can be found in the
* Use of this source code is governed by MIT-like 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.
*/
@@ -12,7 +12,6 @@
#include "PSEncoder.h"
#include "Common/config.h"
#include "Extension/H264.h"
#include "Extension/CommonRtp.h"
#include "Rtsp/RtspMuxer.h"
@@ -20,25 +19,33 @@ using namespace toolkit;
namespace mediakit{
PSEncoderImp::PSEncoderImp(uint32_t ssrc, uint8_t payload_type) : MpegMuxer(true) {
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),is_key);
}));
InfoL << this << " " << printSSRC(_rtp_encoder->getSsrc());
PSEncoderImp::PSEncoderImp(uint32_t ssrc, uint8_t payload_type, bool ps_or_ts) : MpegMuxer(ps_or_ts) {
GET_CONFIG(uint32_t, s_video_mtu, Rtp::kVideoMtuSize);
_rtp_encoder = std::make_shared<CommonRtpEncoder>();
auto video_mtu = s_video_mtu;
if (!ps_or_ts) {
// 确保ts rtp负载部分长度是188的倍数
video_mtu = RtpPacket::kRtpHeaderSize + (s_video_mtu - (s_video_mtu % 188));
if (video_mtu > s_video_mtu) {
video_mtu -= 188;
}
}
_rtp_encoder->setRtpInfo(ssrc, video_mtu, 90000, payload_type);
auto ring = std::make_shared<RtpRing::RingType>();
ring->setDelegate(std::make_shared<RingDelegateHelper>([this](RtpPacket::Ptr rtp, bool is_key) { onRTP(std::move(rtp), is_key); }));
_rtp_encoder->setRtpRing(std::move(ring));
InfoL << this << " " << ssrc;
}
PSEncoderImp::~PSEncoderImp() {
InfoL << this << " " << printSSRC(_rtp_encoder->getSsrc());
InfoL << this;
}
void PSEncoderImp::onWrite(std::shared_ptr<Buffer> buffer, uint64_t stamp, bool key_pos) {
if (!buffer) {
return;
}
_rtp_encoder->inputFrame(std::make_shared<FrameFromPtr>(buffer->data(), buffer->size(), stamp, stamp,0,key_pos));
_rtp_encoder->inputFrame(std::make_shared<FrameFromPtr>(CodecH264/*只用于识别为视频*/, buffer->data(), buffer->size(), stamp, stamp, 0, key_pos));
}
}//namespace mediakit

22
src/Rtp/PSEncoder.h
View File

@@ -1,9 +1,9 @@
/*
* Copyright (c) 2016 The ZLMediaKit project authors. All Rights Reserved.
* Copyright (c) 2016-present The ZLMediaKit project authors. All Rights Reserved.
*
* This file is part of ZLMediaKit(https://github.com/xia-chu/ZLMediaKit).
* This file is part of ZLMediaKit(https://github.com/ZLMediaKit/ZLMediaKit).
*
* Use of this source code is governed by MIT license that can be found in the
* Use of this source code is governed by MIT-like 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.
*/
@@ -16,16 +16,24 @@
#include "Record/MPEG.h"
#include "Common/MediaSink.h"
namespace mediakit{
namespace mediakit {
class CommonRtpEncoder;
class PSEncoderImp : public MpegMuxer{
class PSEncoderImp : public MpegMuxer {
public:
PSEncoderImp(uint32_t ssrc, uint8_t payload_type = 96);
/**
* 创建psh或ts rtp编码器
* @param ssrc rtp的ssrc
* @param payload_type rtp的pt
* @param ps_or_ts true: ps, false: ts
*/
PSEncoderImp(uint32_t ssrc, uint8_t payload_type = 96, bool ps_or_ts = true);
~PSEncoderImp() override;
protected:
//rtp打包后回调
virtual void onRTP(toolkit::Buffer::Ptr rtp,bool is_key = false) = 0;
virtual void onRTP(toolkit::Buffer::Ptr rtp, bool is_key = false) = 0;
protected:
void onWrite(std::shared_ptr<toolkit::Buffer> buffer, uint64_t stamp, bool key_pos) override;

7
src/Rtp/ProcessInterface.h
View File

@@ -1,9 +1,9 @@
/*
* Copyright (c) 2016 The ZLMediaKit project authors. All Rights Reserved.
* Copyright (c) 2016-present The ZLMediaKit project authors. All Rights Reserved.
*
* This file is part of ZLMediaKit(https://github.com/xia-chu/ZLMediaKit).
* This file is part of ZLMediaKit(https://github.com/ZLMediaKit/ZLMediaKit).
*
* Use of this source code is governed by MIT license that can be found in the
* Use of this source code is governed by MIT-like 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.
*/
@@ -20,7 +20,6 @@ namespace mediakit {
class ProcessInterface {
public:
using Ptr = std::shared_ptr<ProcessInterface>;
ProcessInterface() = default;
virtual ~ProcessInterface() = default;
/**

35
src/Rtp/RawEncoder.cpp
View File

@@ -1,9 +1,9 @@
/*
* Copyright (c) 2016 The ZLMediaKit project authors. All Rights Reserved.
* Copyright (c) 2016-present The ZLMediaKit project authors. All Rights Reserved.
*
* This file is part of ZLMediaKit(https://github.com/xia-chu/ZLMediaKit).
* This file is part of ZLMediaKit(https://github.com/ZLMediaKit/ZLMediaKit).
*
* Use of this source code is governed by MIT license that can be found in the
* Use of this source code is governed by MIT-like 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.
*/
@@ -13,6 +13,7 @@
#include "RawEncoder.h"
#include "Extension/Factory.h"
#include "Rtsp/RtspMuxer.h"
#include "Common//config.h"
using namespace toolkit;
@@ -30,17 +31,23 @@ RawEncoderImp::~RawEncoderImp() {
bool RawEncoderImp::addTrack(const Track::Ptr &track) {
if (_send_audio && track->getTrackType() == TrackType::TrackAudio && !_rtp_encoder) { // audio
_rtp_encoder = createRtpEncoder(track);
_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), true); }));
auto ring = std::make_shared<RtpRing::RingType>();
ring->setDelegate(std::make_shared<RingDelegateHelper>([this](RtpPacket::Ptr rtp, bool is_key) { onRTP(std::move(rtp), true); }));
_rtp_encoder->setRtpRing(std::move(ring));
if (track->getCodecId() == CodecG711A || track->getCodecId() == CodecG711U) {
GET_CONFIG(uint32_t, dur_ms, RtpProxy::kRtpG711DurMs);
Any param;
param.set<uint32_t>(dur_ms);
_rtp_encoder->setOpt(RtpCodec::RTP_ENCODER_PKT_DUR_MS, param);
}
return true;
}
if (!_send_audio && track->getTrackType() == TrackType::TrackVideo && !_rtp_encoder) {
_rtp_encoder = createRtpEncoder(track);
_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), is_key); }));
auto ring = std::make_shared<RtpRing::RingType>();
ring->setDelegate(std::make_shared<RingDelegateHelper>([this](RtpPacket::Ptr rtp, bool is_key) { onRTP(std::move(rtp), is_key); }));
_rtp_encoder->setRtpRing(std::move(ring));
return true;
}
return true;
@@ -62,11 +69,17 @@ bool RawEncoderImp::inputFrame(const Frame::Ptr &frame) {
}
RtpCodec::Ptr RawEncoderImp::createRtpEncoder(const Track::Ptr &track) {
uint32_t sample_rate = 90000;
GET_CONFIG(uint32_t, audio_mtu, Rtp::kAudioMtuSize);
GET_CONFIG(uint32_t, video_mtu, Rtp::kVideoMtuSize);
auto sample_rate = 90000u;
auto mtu = video_mtu;
if (track->getTrackType() == TrackType::TrackAudio) {
mtu = audio_mtu;
sample_rate = std::static_pointer_cast<AudioTrack>(track)->getAudioSampleRate();
}
return Factory::getRtpEncoderByCodecId(track->getCodecId(), sample_rate, _payload_type, _ssrc);
auto ret = Factory::getRtpEncoderByCodecId(track->getCodecId(), _payload_type);
ret->setRtpInfo(_ssrc, mtu, sample_rate, _payload_type);
return ret;
}
} // namespace mediakit

6
src/Rtp/RawEncoder.h
View File

@@ -1,9 +1,9 @@
/*
* Copyright (c) 2016 The ZLMediaKit project authors. All Rights Reserved.
* Copyright (c) 2016-present The ZLMediaKit project authors. All Rights Reserved.
*
* This file is part of ZLMediaKit(https://github.com/xia-chu/ZLMediaKit).
* This file is part of ZLMediaKit(https://github.com/ZLMediaKit/ZLMediaKit).
*
* Use of this source code is governed by MIT license that can be found in the
* Use of this source code is governed by MIT-like 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.
*/

6
src/Rtp/RtpCache.cpp
View File

@@ -1,9 +1,9 @@
/*
* Copyright (c) 2016 The ZLMediaKit project authors. All Rights Reserved.
* Copyright (c) 2016-present The ZLMediaKit project authors. All Rights Reserved.
*
* This file is part of ZLMediaKit(https://github.com/xia-chu/ZLMediaKit).
* This file is part of ZLMediaKit(https://github.com/ZLMediaKit/ZLMediaKit).
*
* Use of this source code is governed by MIT license that can be found in the
* Use of this source code is governed by MIT-like 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.
*/

16
src/Rtp/RtpCache.h
View File

@@ -1,9 +1,9 @@
/*
* Copyright (c) 2016 The ZLMediaKit project authors. All Rights Reserved.
* Copyright (c) 2016-present The ZLMediaKit project authors. All Rights Reserved.
*
* This file is part of ZLMediaKit(https://github.com/xia-chu/ZLMediaKit).
* This file is part of ZLMediaKit(https://github.com/ZLMediaKit/ZLMediaKit).
*
* Use of this source code is governed by MIT license that can be found in the
* Use of this source code is governed by MIT-like 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.
*/
@@ -23,7 +23,6 @@ class RtpCache : protected PacketCache<toolkit::Buffer> {
public:
using onFlushed = std::function<void(std::shared_ptr<toolkit::List<toolkit::Buffer::Ptr> >)>;
RtpCache(onFlushed cb);
~RtpCache() override = default;
protected:
/**
@@ -41,8 +40,9 @@ private:
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;
RtpCachePS(onFlushed cb, uint32_t ssrc, uint8_t payload_type = 96, bool ps_or_ts = true) :
RtpCache(std::move(cb)), PSEncoderImp(ssrc, ps_or_ts ? payload_type : Rtsp::PT_MP2T, ps_or_ts) {};
void flush() override;
protected:
@@ -52,13 +52,13 @@ protected:
class RtpCacheRaw : public RtpCache, public RawEncoderImp {
public:
RtpCacheRaw(onFlushed cb, uint32_t ssrc, uint8_t payload_type = 96, bool send_audio = true) : RtpCache(std::move(cb)), RawEncoderImp(ssrc, payload_type, send_audio) {};
~RtpCacheRaw() override = default;
void flush() override;
protected:
void onRTP(toolkit::Buffer::Ptr rtp, bool is_key = false) override;
};
}//namespace mediakit
} //namespace mediakit
#endif//ENABLE_RTPPROXY
#endif //ZLMEDIAKIT_RTPCACHE_H

94
src/Rtp/RtpProcess.cpp
View File

@@ -1,9 +1,9 @@
/*
* Copyright (c) 2016 The ZLMediaKit project authors. All Rights Reserved.
* Copyright (c) 2016-present The ZLMediaKit project authors. All Rights Reserved.
*
* This file is part of ZLMediaKit(https://github.com/xia-chu/ZLMediaKit).
* This file is part of ZLMediaKit(https://github.com/ZLMediaKit/ZLMediaKit).
*
* Use of this source code is governed by MIT license that can be found in the
* Use of this source code is governed by MIT-like 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.
*/
@@ -11,29 +11,30 @@
#if defined(ENABLE_RTPPROXY)
#include "GB28181Process.h"
#include "RtpProcess.h"
#include "Http/HttpTSPlayer.h"
#include "Util/File.h"
#include "Common/config.h"
using namespace std;
using namespace toolkit;
static constexpr char kRtpAppName[] = "rtp";
//在创建_muxer对象前(也就是推流鉴权成功前)，需要先缓存frame，这样可以防止丢包，提高体验
//但是同时需要控制缓冲长度，防止内存溢出。200帧数据，大概有10秒数据，应该足矣等待鉴权hook返回
static constexpr size_t kMaxCachedFrame = 200;
//但是同时需要控制缓冲长度，防止内存溢出。最多缓存10秒数据，应该足矣等待鉴权hook返回
static constexpr size_t kMaxCachedFrameMS = 10 * 1000;
namespace mediakit {
RtpProcess::RtpProcess(const string &stream_id) {
_media_info.schema = kRtpAppName;
_media_info.vhost = DEFAULT_VHOST;
_media_info.app = kRtpAppName;
_media_info.stream = stream_id;
RtpProcess::Ptr RtpProcess::createProcess(const MediaTuple &tuple) {
RtpProcess::Ptr ret(new RtpProcess(tuple));
ret->createTimer();
return ret;
}
RtpProcess::RtpProcess(const MediaTuple &tuple) {
static_cast<MediaTuple &>(_media_info) = tuple;
GET_CONFIG(string, dump_dir, RtpProxy::kDumpDir);
{
FILE *fp = !dump_dir.empty() ? File::create_file(File::absolutePath(_media_info.stream + ".rtp", dump_dir).data(), "wb") : nullptr;
FILE *fp = !dump_dir.empty() ? File::create_file(File::absolutePath(_media_info.stream + ".rtp", dump_dir), "wb") : nullptr;
if (fp) {
_save_file_rtp.reset(fp, [](FILE *fp) {
fclose(fp);
@@ -42,7 +43,7 @@ RtpProcess::RtpProcess(const string &stream_id) {
}
{
FILE *fp = !dump_dir.empty() ? File::create_file(File::absolutePath(_media_info.stream + ".video", dump_dir).data(), "wb") : nullptr;
FILE *fp = !dump_dir.empty() ? File::create_file(File::absolutePath(_media_info.stream + ".video", dump_dir), "wb") : nullptr;
if (fp) {
_save_file_video.reset(fp, [](FILE *fp) {
fclose(fp);
@@ -67,18 +68,40 @@ RtpProcess::~RtpProcess() {
GET_CONFIG(uint32_t, iFlowThreshold, General::kFlowThreshold);
if (_total_bytes >= iFlowThreshold * 1024) {
try {
NoticeCenter::Instance().emitEvent(Broadcast::kBroadcastFlowReport, _media_info, _total_bytes, duration, false, static_cast<SockInfo &>(*this));
NOTICE_EMIT(BroadcastFlowReportArgs, Broadcast::kBroadcastFlowReport, _media_info, _total_bytes, duration, false, *this);
} catch (std::exception &ex) {
WarnL << "Exception occurred: " << ex.what();
}
}
}
void RtpProcess::onManager() {
if (!alive()) {
onDetach(SockException(Err_timeout, "RtpProcess timeout"));
}
}
void RtpProcess::createTimer() {
//创建超时管理定时器
weak_ptr<RtpProcess> weakSelf = shared_from_this();
_timer = std::make_shared<Timer>(3.0f, [weakSelf] {
auto strongSelf = weakSelf.lock();
if (!strongSelf) {
return false;
}
strongSelf->onManager();
return true;
}, EventPollerPool::Instance().getPoller());
}
bool RtpProcess::inputRtp(bool is_udp, const Socket::Ptr &sock, const char *data, size_t len, const struct sockaddr *addr, uint64_t *dts_out) {
if (!isRtp(data, len)) {
WarnP(this) << "Not rtp packet";
return false;
}
if (!_auth_err.empty()) {
throw toolkit::SockException(toolkit::Err_other, _auth_err);
}
if (_sock != sock) {
// 第一次运行本函数
bool first = !_sock;
@@ -86,6 +109,7 @@ bool RtpProcess::inputRtp(bool is_udp, const Socket::Ptr &sock, const char *data
_addr.reset(new sockaddr_storage(*((sockaddr_storage *)addr)));
if (first) {
emitOnPublish();
_cache_ticker.resetTime();
}
}
@@ -126,8 +150,8 @@ bool RtpProcess::inputFrame(const Frame::Ptr &frame) {
_last_frame_time.resetTime();
return _muxer->inputFrame(frame);
}
if (_cached_func.size() > kMaxCachedFrame) {
WarnL << "cached frame of track(" << frame->getCodecName() << ") is too much, now dropped, please check your on_publish hook url in config.ini file";
if (_cache_ticker.elapsedTime() > kMaxCachedFrameMS) {
WarnL << "Cached frame of stream(" << _media_info.stream << ") is too much, your on_publish hook responded too late!";
return false;
}
auto frame_cached = Frame::getCacheAbleFrame(frame);
@@ -195,17 +219,18 @@ void RtpProcess::setStopCheckRtp(bool is_check){
}
}
void RtpProcess::setOnlyAudio(bool only_audio){
_only_audio = only_audio;
void RtpProcess::setOnlyTrack(OnlyTrack only_track) {
_only_track = only_track;
}
void RtpProcess::onDetach() {
void RtpProcess::onDetach(const SockException &ex) {
if (_on_detach) {
_on_detach();
WarnL << ex << ", stream_id: " << getIdentifier();
_on_detach(ex);
}
}
void RtpProcess::setOnDetach(function<void()> cb) {
void RtpProcess::setOnDetach(onDetachCB cb) {
_on_detach = std::move(cb);
}
@@ -251,24 +276,26 @@ void RtpProcess::emitOnPublish() {
return;
}
if (err.empty()) {
strong_self->_muxer = std::make_shared<MultiMediaSourceMuxer>(strong_self->_media_info, 0.0f,
option);
if (strong_self->_only_audio) {
strong_self->_muxer->setOnlyAudio();
strong_self->_muxer = std::make_shared<MultiMediaSourceMuxer>(strong_self->_media_info, 0.0f, option);
switch (strong_self->_only_track) {
case kOnlyAudio: strong_self->_muxer->setOnlyAudio(); break;
case kOnlyVideo: strong_self->_muxer->enableAudio(false); break;
default: break;
}
strong_self->_muxer->setMediaListener(strong_self);
strong_self->doCachedFunc();
InfoP(strong_self) << "允许RTP推流";
} else {
strong_self->_auth_err = err;
WarnP(strong_self) << "禁止RTP推流:" << err;
}
});
};
//触发推流鉴权事件
auto flag = NoticeCenter::Instance().emitEvent(Broadcast::kBroadcastMediaPublish, MediaOriginType::rtp_push, _media_info, invoker, static_cast<SockInfo &>(*this));
auto flag = NOTICE_EMIT(BroadcastMediaPublishArgs, Broadcast::kBroadcastMediaPublish, MediaOriginType::rtp_push, _media_info, invoker, *this);
if (!flag) {
//该事件无人监听,默认不鉴权
// 该事件无人监听,默认不鉴权
invoker("", ProtocolOption());
}
}
@@ -285,6 +312,15 @@ std::shared_ptr<SockInfo> RtpProcess::getOriginSock(MediaSource &sender) const {
return const_cast<RtpProcess *>(this)->shared_from_this();
}
RtpProcess::Ptr RtpProcess::getRtpProcess(mediakit::MediaSource &sender) const {
return const_cast<RtpProcess *>(this)->shared_from_this();
}
bool RtpProcess::close(mediakit::MediaSource &sender) {
onDetach(SockException(Err_shutdown, "close media"));
return true;
}
toolkit::EventPoller::Ptr RtpProcess::getOwnerPoller(MediaSource &sender) {
if (_sock) {
return _sock->getPoller();
@@ -301,4 +337,4 @@ float RtpProcess::getLossRate(MediaSource &sender, TrackType type) {
}
}//namespace mediakit
#endif//defined(ENABLE_RTPPROXY)
#endif//defined(ENABLE_RTPPROXY)

42
src/Rtp/RtpProcess.h
View File

@@ -1,9 +1,9 @@
/*
* Copyright (c) 2016 The ZLMediaKit project authors. All Rights Reserved.
* Copyright (c) 2016-present The ZLMediaKit project authors. All Rights Reserved.
*
* This file is part of ZLMediaKit(https://github.com/xia-chu/ZLMediaKit).
* This file is part of ZLMediaKit(https://github.com/ZLMediaKit/ZLMediaKit).
*
* Use of this source code is governed by MIT license that can be found in the
* Use of this source code is governed by MIT-like 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.
*/
@@ -18,12 +18,16 @@
namespace mediakit {
class RtpProcess final : public RtcpContextForRecv, public toolkit::SockInfo, public MediaSinkInterface, public MediaSourceEventInterceptor, public std::enable_shared_from_this<RtpProcess>{
static constexpr char kRtpAppName[] = "rtp";
class RtpProcess final : public RtcpContextForRecv, public toolkit::SockInfo, public MediaSinkInterface, public MediaSourceEvent, public std::enable_shared_from_this<RtpProcess>{
public:
using Ptr = std::shared_ptr<RtpProcess>;
friend class RtpProcessHelper;
RtpProcess(const std::string &stream_id);
using onDetachCB = std::function<void(const toolkit::SockException &ex)>;
static Ptr createProcess(const MediaTuple &tuple);
~RtpProcess();
enum OnlyTrack { kAll = 0, kOnlyAudio = 1, kOnlyVideo = 2 };
/**
* 输入rtp
@@ -37,20 +41,16 @@ public:
*/
bool inputRtp(bool is_udp, const toolkit::Socket::Ptr &sock, const char *data, size_t len, const struct sockaddr *addr , uint64_t *dts_out = nullptr);
/**
* 是否超时，用于超时移除对象
*/
bool alive();
/**
* 超时时被RtpSelector移除时触发
*/
void onDetach();
void onDetach(const toolkit::SockException &ex);
/**
* 设置onDetach事件回调
*/
void setOnDetach(std::function<void()> cb);
void setOnDetach(onDetachCB cb);
/**
* 设置onDetach事件回调,false检查RTP超时，true停止
@@ -58,10 +58,10 @@ public:
void setStopCheckRtp(bool is_check=false);
/**
* 设置为单track，单音频时可以加快媒体注册速度
* 设置为单track，单音频/单视频时可以加快媒体注册速度
* 请在inputRtp前调用此方法，否则可能会是空操作
*/
void setOnlyAudio(bool only_audio);
void setOnlyTrack(OnlyTrack only_track);
/**
* flush输出缓存
@@ -87,27 +87,37 @@ protected:
std::shared_ptr<SockInfo> getOriginSock(MediaSource &sender) const override;
toolkit::EventPoller::Ptr getOwnerPoller(MediaSource &sender) override;
float getLossRate(MediaSource &sender, TrackType type) override;
Ptr getRtpProcess(mediakit::MediaSource &sender) const override;
bool close(mediakit::MediaSource &sender) override;
private:
RtpProcess(const MediaTuple &tuple);
void emitOnPublish();
void doCachedFunc();
bool alive();
void onManager();
void createTimer();
private:
bool _only_audio = false;
OnlyTrack _only_track = kAll;
std::string _auth_err;
uint64_t _dts = 0;
uint64_t _total_bytes = 0;
std::unique_ptr<sockaddr_storage> _addr;
toolkit::Socket::Ptr _sock;
MediaInfo _media_info;
toolkit::Ticker _last_frame_time;
std::function<void()> _on_detach;
onDetachCB _on_detach;
std::shared_ptr<FILE> _save_file_rtp;
std::shared_ptr<FILE> _save_file_video;
ProcessInterface::Ptr _process;
MultiMediaSourceMuxer::Ptr _muxer;
std::atomic_bool _stop_rtp_check{false};
toolkit::Timer::Ptr _timer;
toolkit::Ticker _last_check_alive;
std::recursive_mutex _func_mtx;
toolkit::Ticker _cache_ticker;
std::deque<std::function<void()> > _cached_func;
};

145
src/Rtp/RtpSelector.cpp
View File

@@ -1,145 +0,0 @@
/*
* Copyright (c) 2016 The ZLMediaKit project authors. All Rights Reserved.
*
* This file is part of ZLMediaKit(https://github.com/xia-chu/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 <stddef.h>
#include "RtpSelector.h"
#include "RtpSplitter.h"
using namespace std;
using namespace toolkit;
namespace mediakit{
INSTANCE_IMP(RtpSelector);
void RtpSelector::clear(){
lock_guard<decltype(_mtx_map)> lck(_mtx_map);
_map_rtp_process.clear();
}
bool RtpSelector::getSSRC(const char *data, size_t data_len, uint32_t &ssrc){
if (data_len < 12) {
return false;
}
uint32_t *ssrc_ptr = (uint32_t *) (data + 8);
ssrc = ntohl(*ssrc_ptr);
return true;
}
RtpProcess::Ptr RtpSelector::getProcess(const string &stream_id,bool makeNew) {
lock_guard<decltype(_mtx_map)> lck(_mtx_map);
auto it = _map_rtp_process.find(stream_id);
if (it == _map_rtp_process.end() && !makeNew) {
return nullptr;
}
if (it != _map_rtp_process.end() && makeNew) {
//已经被其他线程持有了，不得再被持有，否则会存在线程安全的问题
throw ProcessExisted(StrPrinter << "RtpProcess(" << stream_id << ") already existed");
}
RtpProcessHelper::Ptr &ref = _map_rtp_process[stream_id];
if (!ref) {
ref = std::make_shared<RtpProcessHelper>(stream_id, shared_from_this());
ref->attachEvent();
createTimer();
}
return ref->getProcess();
}
void RtpSelector::createTimer() {
if (!_timer) {
//创建超时管理定时器
weak_ptr<RtpSelector> weakSelf = shared_from_this();
_timer = std::make_shared<Timer>(3.0f, [weakSelf] {
auto strongSelf = weakSelf.lock();
if (!strongSelf) {
return false;
}
strongSelf->onManager();
return true;
}, EventPollerPool::Instance().getPoller());
}
}
void RtpSelector::delProcess(const string &stream_id,const RtpProcess *ptr) {
RtpProcess::Ptr process;
{
lock_guard<decltype(_mtx_map)> lck(_mtx_map);
auto it = _map_rtp_process.find(stream_id);
if (it == _map_rtp_process.end()) {
return;
}
if (it->second->getProcess().get() != ptr) {
return;
}
process = it->second->getProcess();
_map_rtp_process.erase(it);
}
process->onDetach();
}
void RtpSelector::onManager() {
List<RtpProcess::Ptr> clear_list;
{
lock_guard<decltype(_mtx_map)> lck(_mtx_map);
for (auto it = _map_rtp_process.begin(); it != _map_rtp_process.end();) {
if (it->second->getProcess()->alive()) {
++it;
continue;
}
WarnL << "RtpProcess timeout:" << it->first;
clear_list.emplace_back(it->second->getProcess());
it = _map_rtp_process.erase(it);
}
}
clear_list.for_each([](const RtpProcess::Ptr &process) {
process->onDetach();
});
}
RtpProcessHelper::RtpProcessHelper(const string &stream_id, const weak_ptr<RtpSelector> &parent) {
_stream_id = stream_id;
_parent = parent;
_process = std::make_shared<RtpProcess>(stream_id);
}
RtpProcessHelper::~RtpProcessHelper() {
auto process = std::move(_process);
try {
// flush时，确保线程安全
process->getOwnerPoller(MediaSource::NullMediaSource())->async([process]() { process->flush(); });
} catch (...) {
// 忽略getOwnerPoller可能抛出的异常
}
}
void RtpProcessHelper::attachEvent() {
//主要目的是close回调触发时能把对象从RtpSelector中删除
_process->setDelegate(shared_from_this());
}
bool RtpProcessHelper::close(MediaSource &sender) {
//此回调在其他线程触发
auto parent = _parent.lock();
if (!parent) {
return false;
}
parent->delProcess(_stream_id, _process.get());
WarnL << "close media: " << sender.getUrl();
return true;
}
RtpProcess::Ptr &RtpProcessHelper::getProcess() {
return _process;
}
}//namespace mediakit
#endif//defined(ENABLE_RTPPROXY)

89
src/Rtp/RtpSelector.h
View File

@@ -1,89 +0,0 @@
/*
* Copyright (c) 2016 The ZLMediaKit project authors. All Rights Reserved.
*
* This file is part of ZLMediaKit(https://github.com/xia-chu/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_RTPSELECTOR_H
#define ZLMEDIAKIT_RTPSELECTOR_H
#if defined(ENABLE_RTPPROXY)
#include <stdint.h>
#include <mutex>
#include <unordered_map>
#include "RtpProcess.h"
#include "Common/MediaSource.h"
namespace mediakit{
class RtpSelector;
class RtpProcessHelper : public MediaSourceEvent , public std::enable_shared_from_this<RtpProcessHelper> {
public:
using Ptr = std::shared_ptr<RtpProcessHelper>;
RtpProcessHelper(const std::string &stream_id, const std::weak_ptr<RtpSelector > &parent);
~RtpProcessHelper();
void attachEvent();
RtpProcess::Ptr & getProcess();
protected:
// 通知其停止推流
bool close(MediaSource &sender) override;
private:
std::string _stream_id;
RtpProcess::Ptr _process;
std::weak_ptr<RtpSelector> _parent;
};
class RtpSelector : public std::enable_shared_from_this<RtpSelector>{
public:
RtpSelector() = default;
~RtpSelector() = default;
class ProcessExisted : public std::runtime_error {
public:
template<typename ...T>
ProcessExisted(T && ...args) : std::runtime_error(std::forward<T>(args)...) {}
~ProcessExisted() override = default;
};
static bool getSSRC(const char *data,size_t data_len, uint32_t &ssrc);
static RtpSelector &Instance();
/**
* 清空所有对象
*/
void clear();
/**
* 获取一个rtp处理器
* @param stream_id 流id
* @param makeNew 不存在时是否新建, 该参数为true时，必须确保之前未创建同名对象
* @return rtp处理器
*/
RtpProcess::Ptr getProcess(const std::string &stream_id, bool makeNew);
/**
* 删除rtp处理器
* @param stream_id 流id
* @param ptr rtp处理器指针
*/
void delProcess(const std::string &stream_id, const RtpProcess *ptr);
private:
void onManager();
void createTimer();
private:
toolkit::Timer::Ptr _timer;
std::recursive_mutex _mtx_map;
std::unordered_map<std::string,RtpProcessHelper::Ptr> _map_rtp_process;
};
}//namespace mediakit
#endif//defined(ENABLE_RTPPROXY)
#endif //ZLMEDIAKIT_RTPSELECTOR_H

34
src/Rtp/RtpSender.cpp
View File

@@ -1,9 +1,9 @@
/*
* Copyright (c) 2016 The ZLMediaKit project authors. All Rights Reserved.
* Copyright (c) 2016-present The ZLMediaKit project authors. All Rights Reserved.
*
* This file is part of ZLMediaKit(https://github.com/xia-chu/ZLMediaKit).
* This file is part of ZLMediaKit(https://github.com/ZLMediaKit/ZLMediaKit).
*
* Use of this source code is governed by MIT license that can be found in the
* Use of this source code is governed by MIT-like 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.
*/
@@ -40,10 +40,11 @@ void RtpSender::startSend(const MediaSourceEvent::SendRtpArgs &args, const funct
if (!_interface) {
//重连时不重新创建对象
auto lam = [this](std::shared_ptr<List<Buffer::Ptr>> list) { onFlushRtpList(std::move(list)); };
if (args.use_ps) {
_interface = std::make_shared<RtpCachePS>(lam, atoi(args.ssrc.data()), args.pt);
} else {
_interface = std::make_shared<RtpCacheRaw>(lam, atoi(args.ssrc.data()), args.pt, args.only_audio);
switch (args.type) {
case MediaSourceEvent::SendRtpArgs::kRtpPS: _interface = std::make_shared<RtpCachePS>(lam, atoi(args.ssrc.data()), args.pt, true); break;
case MediaSourceEvent::SendRtpArgs::kRtpTS: _interface = std::make_shared<RtpCachePS>(lam, atoi(args.ssrc.data()), args.pt, false); break;
case MediaSourceEvent::SendRtpArgs::kRtpRAW: _interface = std::make_shared<RtpCacheRaw>(lam, atoi(args.ssrc.data()), args.pt, args.only_audio); break;
default: CHECK(0, "invalid rtp type:" + to_string(args.type)); break;
}
}
@@ -140,7 +141,7 @@ void RtpSender::startSend(const MediaSourceEvent::SendRtpArgs &args, const funct
cb(0, SockException(Err_other, ex.what()));
return;
}
strong_self->_socket_rtp->bindPeerAddr((struct sockaddr *)&addr);
strong_self->_socket_rtp->bindPeerAddr((struct sockaddr *)&addr, 0, true);
strong_self->onConnect();
cb(strong_self->_socket_rtp->get_local_port(), SockException());
});
@@ -173,16 +174,9 @@ void RtpSender::createRtcpSocket() {
return;
}
struct sockaddr_storage addr;
//目标rtp端口
SockUtil::get_sock_peer_addr(_socket_rtp->rawFD(), addr);
//绑定目标rtcp端口(目标rtp端口 + 1)
switch (addr.ss_family) {
case AF_INET: ((sockaddr_in *)&addr)->sin_port = htons(ntohs(((sockaddr_in *)&addr)->sin_port) + 1); break;
case AF_INET6: ((sockaddr_in6 *)&addr)->sin6_port = htons(ntohs(((sockaddr_in6 *)&addr)->sin6_port) + 1); break;
default: assert(0); break;
}
_socket_rtcp->bindPeerAddr((struct sockaddr *)&addr);
// 绑定目标rtcp端口(目标rtp端口 + 1)
auto addr = SockUtil::make_sockaddr(_socket_rtp->get_peer_ip().data(), _socket_rtp->get_peer_port() + 1);
_socket_rtcp->bindPeerAddr((struct sockaddr *)&addr, 0, true);
_rtcp_context = std::make_shared<RtcpContextForSend>();
weak_ptr<RtpSender> weak_self = shared_from_this();
@@ -286,7 +280,7 @@ void RtpSender::onSendRtpUdp(const toolkit::Buffer::Ptr &buf, bool check) {
return;
}
auto rtp = static_pointer_cast<RtpPacket>(buf);
_rtcp_context->onRtp(rtp->getSeq(), rtp->getStamp(), rtp->getStampMS(), 90000 /*not used*/, rtp->size());
_rtcp_context->onRtp(rtp->getSeq(), rtp->getStamp(), rtp->ntp_stamp, 90000 /*not used*/, rtp->size());
if (!check) {
//减少判断次数
@@ -349,4 +343,4 @@ void RtpSender::setOnClose(std::function<void(const toolkit::SockException &ex)>
}
}//namespace mediakit
#endif// defined(ENABLE_RTPPROXY)
#endif// defined(ENABLE_RTPPROXY)

6
src/Rtp/RtpSender.h
View File

@@ -1,9 +1,9 @@
/*
* Copyright (c) 2016 The ZLMediaKit project authors. All Rights Reserved.
* Copyright (c) 2016-present The ZLMediaKit project authors. All Rights Reserved.
*
* This file is part of ZLMediaKit(https://github.com/xia-chu/ZLMediaKit).
* This file is part of ZLMediaKit(https://github.com/ZLMediaKit/ZLMediaKit).
*
* Use of this source code is governed by MIT license that can be found in the
* Use of this source code is governed by MIT-like 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.
*/

165
src/Rtp/RtpServer.cpp
View File

@@ -1,9 +1,9 @@
/*
* Copyright (c) 2016 The ZLMediaKit project authors. All Rights Reserved.
* Copyright (c) 2016-present The ZLMediaKit project authors. All Rights Reserved.
*
* This file is part of ZLMediaKit(https://github.com/xia-chu/ZLMediaKit).
* This file is part of ZLMediaKit(https://github.com/ZLMediaKit/ZLMediaKit).
*
* Use of this source code is governed by MIT license that can be found in the
* Use of this source code is governed by MIT-like 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.
*/
@@ -11,7 +11,7 @@
#if defined(ENABLE_RTPPROXY)
#include "Util/uv_errno.h"
#include "RtpServer.h"
#include "RtpSelector.h"
#include "RtpProcess.h"
#include "Rtcp/RtcpContext.h"
#include "Common/config.h"
@@ -30,43 +30,39 @@ class RtcpHelper: public std::enable_shared_from_this<RtcpHelper> {
public:
using Ptr = std::shared_ptr<RtcpHelper>;
RtcpHelper(Socket::Ptr rtcp_sock, std::string stream_id) {
RtcpHelper(Socket::Ptr rtcp_sock, MediaTuple tuple) {
_rtcp_sock = std::move(rtcp_sock);
_stream_id = std::move(stream_id);
_tuple = std::move(tuple);
}
~RtcpHelper() {
if (_process) {
// 删除rtp处理器
RtpSelector::Instance().delProcess(_stream_id, _process.get());
}
}
void setRtpServerInfo(uint16_t local_port,RtpServer::TcpMode mode,bool re_use_port,uint32_t ssrc, bool only_audio) {
_local_port = local_port;
_tcp_mode = mode;
_re_use_port = re_use_port;
void setRtpServerInfo(uint16_t local_port, RtpServer::TcpMode mode, bool re_use_port, uint32_t ssrc, int only_track) {
_ssrc = ssrc;
_only_audio = only_audio;
_process = RtpProcess::createProcess(_tuple);
_process->setOnlyTrack((RtpProcess::OnlyTrack)only_track);
_timeout_cb = [=]() mutable {
NOTICE_EMIT(BroadcastRtpServerTimeoutArgs, Broadcast::kBroadcastRtpServerTimeout, local_port, _tuple, (int)mode, re_use_port, ssrc);
};
weak_ptr<RtcpHelper> weak_self = shared_from_this();
_process->setOnDetach([weak_self](const SockException &ex) {
if (auto strong_self = weak_self.lock()) {
if (strong_self->_on_detach) {
strong_self->_on_detach(ex);
}
if (ex.getErrCode() == Err_timeout) {
strong_self->_timeout_cb();
}
}
});
}
void setOnDetach(function<void()> cb) {
if (_process) {
_process->setOnDetach(std::move(cb));
} else {
_on_detach = std::move(cb);
}
}
void setOnDetach(RtpProcess::onDetachCB cb) { _on_detach = std::move(cb); }
RtpProcess::Ptr getProcess() const { return _process; }
void onRecvRtp(const Socket::Ptr &sock, const Buffer::Ptr &buf, struct sockaddr *addr) {
if (!_process) {
_process = RtpSelector::Instance().getProcess(_stream_id, true);
_process->setOnlyAudio(_only_audio);
_process->setOnDetach(std::move(_on_detach));
cancelDelayTask();
}
_process->inputRtp(true, sock, buf->data(), buf->size(), addr);
// 统计rtp接受情况，用于发送rr包
auto header = (RtpHeader *)buf->data();
sendRtcp(ntohl(header->ssrc), addr);
@@ -89,38 +85,15 @@ public:
for (auto &rtcp : rtcps) {
strong_self->_process->onRtcp(rtcp);
}
// 收到sr rtcp后驱动返回rr rtcp
strong_self->sendRtcp(strong_self->_ssrc, (struct sockaddr *)(strong_self->_rtcp_addr.get()));
});
GET_CONFIG(uint64_t, timeoutSec, RtpProxy::kTimeoutSec);
_delay_task = _rtcp_sock->getPoller()->doDelayTask(timeoutSec * 1000, [weak_self]() {
if (auto strong_self = weak_self.lock()) {
auto process = RtpSelector::Instance().getProcess(strong_self->_stream_id, false);
if (!process && strong_self->_on_detach) {
strong_self->_on_detach();
}
if(process && strong_self->_on_detach){// tcp 链接防止断开不删除rtpServer
process->setOnDetach(std::move(strong_self->_on_detach));
}
if (!process) { // process 未创建，触发rtp server 超时事件
NoticeCenter::Instance().emitEvent(Broadcast::KBroadcastRtpServerTimeout, strong_self->_local_port, strong_self->_stream_id,
(int)strong_self->_tcp_mode, strong_self->_re_use_port, strong_self->_ssrc);
}
}
return 0;
});
}
void cancelDelayTask() {
if (_delay_task) {
_delay_task->cancel();
_delay_task = nullptr;
}
}
private:
void sendRtcp(uint32_t rtp_ssrc, struct sockaddr *addr) {
// 每5秒发送一次rtcp
if (_ticker.elapsedTime() < 5000 || !_process) {
if (_ticker.elapsedTime() < 5000) {
return;
}
_ticker.resetTime();
@@ -139,25 +112,21 @@ private:
}
private:
bool _re_use_port = false;
bool _only_audio = false;
uint16_t _local_port = 0;
uint32_t _ssrc = 0;
RtpServer::TcpMode _tcp_mode = RtpServer::NONE;
std::function<void()> _timeout_cb;
Ticker _ticker;
Socket::Ptr _rtcp_sock;
RtpProcess::Ptr _process;
std::string _stream_id;
function<void()> _on_detach;
MediaTuple _tuple;
RtpProcess::onDetachCB _on_detach;
std::shared_ptr<struct sockaddr_storage> _rtcp_addr;
EventPoller::DelayTask::Ptr _delay_task;
};
void RtpServer::start(uint16_t local_port, const string &stream_id, TcpMode tcp_mode, const char *local_ip, bool re_use_port, uint32_t ssrc, bool only_audio) {
void RtpServer::start(uint16_t local_port, const MediaTuple &tuple, TcpMode tcp_mode, const char *local_ip, bool re_use_port, uint32_t ssrc, int only_track, bool multiplex) {
//创建udp服务器
Socket::Ptr rtp_socket = Socket::createSocket(nullptr, true);
Socket::Ptr rtcp_socket = Socket::createSocket(nullptr, true);
auto poller = EventPollerPool::Instance().getPoller();
Socket::Ptr rtp_socket = Socket::createSocket(poller, true);
Socket::Ptr rtcp_socket = Socket::createSocket(poller, true);
if (local_port == 0) {
//随机端口，rtp端口采用偶数
auto pair = std::make_pair(rtp_socket, rtcp_socket);
@@ -172,32 +141,18 @@ void RtpServer::start(uint16_t local_port, const string &stream_id, TcpMode tcp_
}
//设置udp socket读缓存
SockUtil::setRecvBuf(rtp_socket->rawFD(), 4 * 1024 * 1024);
TcpServer::Ptr tcp_server;
_tcp_mode = tcp_mode;
if (tcp_mode == PASSIVE || tcp_mode == ACTIVE) {
//创建tcp服务器
tcp_server = std::make_shared<TcpServer>(rtp_socket->getPoller());
(*tcp_server)[RtpSession::kStreamID] = stream_id;
(*tcp_server)[RtpSession::kSSRC] = ssrc;
(*tcp_server)[RtpSession::kOnlyAudio] = only_audio;
if (tcp_mode == PASSIVE) {
tcp_server->start<RtpSession>(local_port, local_ip);
} else if (stream_id.empty()) {
// tcp主动模式时只能一个端口一个流，必须指定流id; 创建TcpServer对象也仅用于传参
throw std::runtime_error(StrPrinter << "tcp主动模式时必需指定流id");
}
}
GET_CONFIG(int, udpRecvSocketBuffer, RtpProxy::kUdpRecvSocketBuffer);
SockUtil::setRecvBuf(rtp_socket->rawFD(), udpRecvSocketBuffer);
//创建udp服务器
UdpServer::Ptr udp_server;
RtcpHelper::Ptr helper;
if (!stream_id.empty()) {
//增加了多路复用判断，如果多路复用为true，就走else逻辑，同时保留了原来stream_id为空走else逻辑
if (!tuple.stream.empty() && !multiplex) {
//指定了流id，那么一个端口一个流(不管是否包含多个ssrc的多个流，绑定rtp源后，会筛选掉ip端口不匹配的流)
helper = std::make_shared<RtcpHelper>(std::move(rtcp_socket), stream_id);
helper = std::make_shared<RtcpHelper>(std::move(rtcp_socket), tuple);
helper->startRtcp();
helper->setRtpServerInfo(local_port, tcp_mode, re_use_port, ssrc, only_audio);
helper->setRtpServerInfo(local_port, tcp_mode, re_use_port, ssrc, only_track);
bool bind_peer_addr = false;
auto ssrc_ptr = std::make_shared<uint32_t>(ssrc);
_ssrc = ssrc_ptr;
@@ -218,13 +173,35 @@ void RtpServer::start(uint16_t local_port, const string &stream_id, TcpMode tcp_
});
} else {
//单端口多线程接收多个流，根据ssrc区分流
udp_server = std::make_shared<UdpServer>(rtp_socket->getPoller());
(*udp_server)[RtpSession::kOnlyAudio] = only_audio;
udp_server = std::make_shared<UdpServer>();
(*udp_server)[RtpSession::kOnlyTrack] = only_track;
(*udp_server)[RtpSession::kUdpRecvBuffer] = udpRecvSocketBuffer;
udp_server->start<RtpSession>(local_port, local_ip);
rtp_socket = nullptr;
}
_on_cleanup = [rtp_socket, stream_id]() {
TcpServer::Ptr tcp_server;
if (tcp_mode == PASSIVE || tcp_mode == ACTIVE) {
auto processor = helper ? helper->getProcess() : nullptr;
// 如果共享同一个processor对象，那么tcp server深圳为单线程模式确保线程安全
tcp_server = std::make_shared<TcpServer>(processor ? poller : nullptr);
(*tcp_server)[RtpSession::kVhost] = tuple.vhost;
(*tcp_server)[RtpSession::kApp] = tuple.app;
(*tcp_server)[RtpSession::kStreamID] = tuple.stream;
(*tcp_server)[RtpSession::kSSRC] = ssrc;
(*tcp_server)[RtpSession::kOnlyTrack] = only_track;
if (tcp_mode == PASSIVE) {
weak_ptr<RtpServer> weak_self = shared_from_this();
tcp_server->start<RtpSession>(local_port, local_ip, 1024, [weak_self, processor](std::shared_ptr<RtpSession> &session) {
session->setRtpProcess(processor);
});
} else if (tuple.stream.empty()) {
// tcp主动模式时只能一个端口一个流，必须指定流id; 创建TcpServer对象也仅用于传参
throw std::runtime_error(StrPrinter << "tcp主动模式时必需指定流id");
}
}
_on_cleanup = [rtp_socket]() {
if (rtp_socket) {
//去除循环引用
rtp_socket->setOnRead(nullptr);
@@ -235,9 +212,10 @@ void RtpServer::start(uint16_t local_port, const string &stream_id, TcpMode tcp_
_udp_server = udp_server;
_rtp_socket = rtp_socket;
_rtcp_helper = helper;
_tcp_mode = tcp_mode;
}
void RtpServer::setOnDetach(function<void()> cb) {
void RtpServer::setOnDetach(RtpProcess::onDetachCB cb) {
if (_rtcp_helper) {
_rtcp_helper->setOnDetach(std::move(cb));
}
@@ -272,6 +250,7 @@ void RtpServer::connectToServer(const std::string &url, uint16_t port, const fun
void RtpServer::onConnect() {
auto rtp_session = std::make_shared<RtpSession>(_rtp_socket);
rtp_session->setRtpProcess(_rtcp_helper->getProcess());
rtp_session->attachServer(*_tcp_server);
_rtp_socket->setOnRead([rtp_session](const Buffer::Ptr &buf, struct sockaddr *addr, int addr_len) {
rtp_session->onRecv(buf);

16
src/Rtp/RtpServer.h
View File

@@ -1,9 +1,9 @@
/*
* Copyright (c) 2016 The ZLMediaKit project authors. All Rights Reserved.
* Copyright (c) 2016-present The ZLMediaKit project authors. All Rights Reserved.
*
* This file is part of ZLMediaKit(https://github.com/xia-chu/ZLMediaKit).
* This file is part of ZLMediaKit(https://github.com/ZLMediaKit/ZLMediaKit).
*
* Use of this source code is governed by MIT license that can be found in the
* Use of this source code is governed by MIT-like 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.
*/
@@ -31,7 +31,6 @@ public:
using onRecv = std::function<void(const toolkit::Buffer::Ptr &buf)>;
enum TcpMode { NONE = 0, PASSIVE, ACTIVE };
RtpServer() = default;
~RtpServer();
/**
@@ -42,9 +41,10 @@ public:
* @param local_ip 绑定的本地网卡ip
* @param re_use_port 是否设置socket为re_use属性
* @param ssrc 指定的ssrc
* @param multiplex 多路复用
*/
void start(uint16_t local_port, const std::string &stream_id = "", TcpMode tcp_mode = PASSIVE,
const char *local_ip = "::", bool re_use_port = true, uint32_t ssrc = 0, bool only_audio = false);
void start(uint16_t local_port, const MediaTuple &tuple = MediaTuple{DEFAULT_VHOST, kRtpAppName, "", ""}, TcpMode tcp_mode = PASSIVE,
const char *local_ip = "::", bool re_use_port = true, uint32_t ssrc = 0, int only_track = 0, bool multiplex = false);
/**
* 连接到tcp服务(tcp主动模式)
@@ -62,7 +62,7 @@ public:
/**
* 设置RtpProcess onDetach事件回调
*/
void setOnDetach(std::function<void()> cb);
void setOnDetach(RtpProcess::onDetachCB cb);
/**
* 更新ssrc
@@ -81,7 +81,7 @@ protected:
std::shared_ptr<RtcpHelper> _rtcp_helper;
std::function<void()> _on_cleanup;
bool _only_audio = false;
int _only_track = 0;
//用于tcp主动模式
TcpMode _tcp_mode = NONE;
};

205
src/Rtp/RtpSession.cpp
View File

@@ -1,16 +1,16 @@
/*
* Copyright (c) 2016 The ZLMediaKit project authors. All Rights Reserved.
* Copyright (c) 2016-present The ZLMediaKit project authors. All Rights Reserved.
*
* This file is part of ZLMediaKit(https://github.com/xia-chu/ZLMediaKit).
* This file is part of ZLMediaKit(https://github.com/ZLMediaKit/ZLMediaKit).
*
* Use of this source code is governed by MIT license that can be found in the
* Use of this source code is governed by MIT-like 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 "RtpSession.h"
#include "RtpSelector.h"
#include "RtpProcess.h"
#include "Network/TcpServer.h"
#include "Rtsp/Rtsp.h"
#include "Rtsp/RtpReceiver.h"
@@ -21,35 +21,39 @@ using namespace toolkit;
namespace mediakit{
const string RtpSession::kVhost = "vhost";
const string RtpSession::kApp = "app";
const string RtpSession::kStreamID = "stream_id";
const string RtpSession::kSSRC = "ssrc";
const string RtpSession::kOnlyAudio = "only_audio";
const string RtpSession::kOnlyTrack = "only_track";
const string RtpSession::kUdpRecvBuffer = "udp_recv_socket_buffer";
void RtpSession::attachServer(const Server &server) {
setParams(const_cast<Server &>(server));
}
void RtpSession::setParams(mINI &ini) {
_stream_id = ini[kStreamID];
_tuple.vhost = ini[kVhost];
_tuple.app = ini[kApp];
_tuple.stream = ini[kStreamID];
_ssrc = ini[kSSRC];
_only_audio = ini[kOnlyAudio];
_only_track = ini[kOnlyTrack];
int udp_socket_buffer = ini[kUdpRecvBuffer];
if (_is_udp) {
// 设置udp socket读缓存
SockUtil::setRecvBuf(getSock()->rawFD(),
(udp_socket_buffer > 0) ? udp_socket_buffer : (4 * 1024 * 1024));
}
}
RtpSession::RtpSession(const Socket::Ptr &sock) : Session(sock) {
RtpSession::RtpSession(const Socket::Ptr &sock)
: Session(sock) {
socklen_t addr_len = sizeof(_addr);
getpeername(sock->rawFD(), (struct sockaddr *)&_addr, &addr_len);
_is_udp = sock->sockType() == SockNum::Sock_UDP;
if (_is_udp) {
// 设置udp socket读缓存
SockUtil::setRecvBuf(getSock()->rawFD(), 4 * 1024 * 1024);
}
}
RtpSession::~RtpSession() {
if(_process){
RtpSelector::Instance().delProcess(_stream_id,_process.get());
}
}
RtpSession::~RtpSession() = default;
void RtpSession::onRecv(const Buffer::Ptr &data) {
if (_is_udp) {
@@ -60,24 +64,24 @@ void RtpSession::onRecv(const Buffer::Ptr &data) {
}
void RtpSession::onError(const SockException &err) {
WarnP(this) << _stream_id << " " << err;
if (_emit_detach) {
_process->onDetach(err);
}
WarnP(this) << _tuple.shortUrl() << " " << err;
}
void RtpSession::onManager() {
if(_process && !_process->alive()){
shutdown(SockException(Err_timeout, "receive rtp timeout"));
}
if(!_process && _ticker.createdTime() > 10 * 1000){
if (!_process && _ticker.createdTime() > 10 * 1000) {
shutdown(SockException(Err_timeout, "illegal connection"));
}
}
void RtpSession::setRtpProcess(RtpProcess::Ptr process) {
_emit_detach = (bool)process;
_process = std::move(process);
}
void RtpSession::onRtpPacket(const char *data, size_t len) {
if (_delay_close) {
// 正在延时关闭中，忽略所有数据
return;
}
if (!isRtp(data, len)) {
// 忽略非rtp数据
WarnP(this) << "Not rtp packet";
@@ -100,33 +104,31 @@ void RtpSession::onRtpPacket(const char *data, size_t len) {
return;
}
}
// 未设置ssrc时，尝试获取ssrc
if (!_ssrc && !getSSRC(data, len, _ssrc)) {
return;
}
// 未指定流id就使用ssrc为流id
if (_tuple.stream.empty()) {
_tuple.stream = printSSRC(_ssrc);
}
if (!_process) {
//未设置ssrc时，尝试获取ssrc
if (!_ssrc && !RtpSelector::getSSRC(data, len, _ssrc)) {
return;
}
if (_stream_id.empty()) {
//未指定流id就使用ssrc为流id
_stream_id = printSSRC(_ssrc);
}
try {
_process = RtpSelector::Instance().getProcess(_stream_id, true);
} catch (RtpSelector::ProcessExisted &ex) {
if (!_is_udp) {
// tcp情况下立即断开连接
throw;
_process = RtpProcess::createProcess(_tuple);
_process->setOnlyTrack((RtpProcess::OnlyTrack)_only_track);
weak_ptr<RtpSession> weak_self = static_pointer_cast<RtpSession>(shared_from_this());
_process->setOnDetach([weak_self](const SockException &ex) {
if (auto strong_self = weak_self.lock()) {
strong_self->_process = nullptr;
strong_self->shutdown(ex);
}
// udp情况下延时断开连接(等待超时自动关闭)，防止频繁创建销毁RtpSession对象
WarnP(this) << ex.what();
_delay_close = true;
return;
}
_process->setOnlyAudio(_only_audio);
_process->setDelegate(static_pointer_cast<RtpSession>(shared_from_this()));
});
}
try {
uint32_t rtp_ssrc = 0;
RtpSelector::getSSRC(data, len, rtp_ssrc);
getSSRC(data, len, rtp_ssrc);
if (rtp_ssrc != _ssrc) {
WarnP(this) << "ssrc mismatched, rtp dropped: " << rtp_ssrc << " != " << _ssrc;
return;
@@ -139,74 +141,121 @@ void RtpSession::onRtpPacket(const char *data, size_t len) {
} else {
throw;
}
} catch (...) {
throw;
}
_ticker.resetTime();
}
bool RtpSession::close(MediaSource &sender) {
//此回调在其他线程触发
string err = StrPrinter << "close media: " << sender.getUrl();
safeShutdown(SockException(Err_shutdown, err));
return true;
}
static const char *findSSRC(const char *data, ssize_t len, uint32_t ssrc) {
//rtp前面必须预留两个字节的长度字段
// rtp前面必须预留两个字节的长度字段
for (ssize_t i = 2; i <= len - 4; ++i) {
auto ptr = (const uint8_t *) data + i;
if (ptr[0] == (ssrc >> 24) && ptr[1] == ((ssrc >> 16) & 0xFF) &&
ptr[2] == ((ssrc >> 8) & 0xFF) && ptr[3] == (ssrc & 0xFF)) {
return (const char *) ptr;
auto ptr = (const uint8_t *)data + i;
if (ptr[0] == (ssrc >> 24) && ptr[1] == ((ssrc >> 16) & 0xFF) && ptr[2] == ((ssrc >> 8) & 0xFF)
&& ptr[3] == (ssrc & 0xFF)) {
return (const char *)ptr;
}
}
return nullptr;
}
//rtp长度到ssrc间的长度固定为10
static const char *findPsHeaderFlag(const char *data, ssize_t len) {
for (ssize_t i = 2; i <= len - 4; ++i) {
auto ptr = (const uint8_t *)data + i;
// PsHeader 0x000001ba、PsSystemHeader0x000001bb（关键帧标识）
if (ptr[0] == (0x00) && ptr[1] == (0x00) && ptr[2] == (0x01) && ptr[3] == (0xbb)) {
return (const char *)ptr;
}
}
return nullptr;
}
// rtp长度到ssrc间的长度固定为10
static size_t constexpr kSSRCOffset = 2 + 4 + 4;
// rtp长度到ps header间的长度固定为14 （暂时不采用找ps header,采用找system header代替）
// rtp长度到ps system header间的长度固定为20 (关键帧标识)
static size_t constexpr kPSHeaderOffset = 2 + 4 + 4 + 4 + 20;
const char *RtpSession::onSearchPacketTail(const char *data, size_t len) {
if (!_search_rtp) {
//tcp上下文正常，不用搜索ssrc
// tcp上下文正常，不用搜索ssrc
return RtpSplitter::onSearchPacketTail(data, len);
}
if (!_process) {
throw SockException(Err_shutdown, "ssrc未获取到，无法通过ssrc恢复tcp上下文");
InfoL << "ssrc未获取到，无法通过ssrc恢复tcp上下文；尝试搜索PsSystemHeader恢复tcp上下文。";
auto rtp_ptr1 = searchByPsHeaderFlag(data, len);
return rtp_ptr1;
}
//搜索第一个rtp的ssrc
auto rtp_ptr0 = searchBySSRC(data, len);
if (rtp_ptr0) {
return rtp_ptr0;
}
// ssrc搜索失败继续尝试搜索ps header flag
auto rtp_ptr2 = searchByPsHeaderFlag(data, len);
return rtp_ptr2;
}
const char *RtpSession::searchBySSRC(const char *data, size_t len) {
InfoL << "尝试rtp搜索ssrc..._ssrc=" << _ssrc;
// 搜索第一个rtp的ssrc
auto ssrc_ptr0 = findSSRC(data, len, _ssrc);
if (!ssrc_ptr0) {
//未搜索到任意rtp，返回数据不够
// 未搜索到任意rtp，返回数据不够
InfoL << "rtp搜索ssrc失败（第一个数据不够），丢弃rtp数据为：" << len;
return nullptr;
}
//这两个字节是第一个rtp的长度字段
// 这两个字节是第一个rtp的长度字段
auto rtp_len_ptr = (ssrc_ptr0 - kSSRCOffset);
auto rtp_len = ((uint8_t *)rtp_len_ptr)[0] << 8 | ((uint8_t *)rtp_len_ptr)[1];
//搜索第二个rtp的ssrc
auto ssrc_ptr1 = findSSRC(ssrc_ptr0 + rtp_len, data + (ssize_t) len - ssrc_ptr0 - rtp_len, _ssrc);
// 搜索第二个rtp的ssrc
auto ssrc_ptr1 = findSSRC(ssrc_ptr0 + rtp_len, data + (ssize_t)len - ssrc_ptr0 - rtp_len, _ssrc);
if (!ssrc_ptr1) {
//未搜索到第二个rtp，返回数据不够
// 未搜索到第二个rtp，返回数据不够
InfoL << "rtp搜索ssrc失败(第二个数据不够)，丢弃rtp数据为：" << len;
return nullptr;
}
//两个ssrc的间隔正好等于rtp的长度(外加rtp长度字段)，那么说明找到rtp
// 两个ssrc的间隔正好等于rtp的长度(外加rtp长度字段)，那么说明找到rtp
auto ssrc_offset = ssrc_ptr1 - ssrc_ptr0;
if (ssrc_offset == rtp_len + 2 || ssrc_offset == rtp_len + 4) {
InfoL << "rtp搜索成功，tcp上下文恢复成功，丢弃的rtp残余数据为：" << rtp_len_ptr - data;
InfoL << "rtp搜索ssrc成功，tcp上下文恢复成功，丢弃的rtp残余数据为：" << rtp_len_ptr - data;
_search_rtp_finished = true;
if (rtp_len_ptr == data) {
//停止搜索rtp，否则会进入死循环
// 停止搜索rtp，否则会进入死循环
_search_rtp = false;
}
//前面的数据都需要丢弃，这个是rtp的起始
// 前面的数据都需要丢弃，这个是rtp的起始
return rtp_len_ptr;
}
//第一个rtp长度不匹配，说明第一个找到的ssrc不是rtp，丢弃之，我们从第二个ssrc所在rtp开始搜索
// 第一个rtp长度不匹配，说明第一个找到的ssrc不是rtp，丢弃之，我们从第二个ssrc所在rtp开始搜索
return ssrc_ptr1 - kSSRCOffset;
}
const char *RtpSession::searchByPsHeaderFlag(const char *data, size_t len) {
InfoL << "尝试rtp搜索PsSystemHeaderFlag..._ssrc=" << _ssrc;
// 搜索rtp中的第一个PsHeaderFlag
auto ps_header_flag_ptr = findPsHeaderFlag(data, len);
if (!ps_header_flag_ptr) {
InfoL << "rtp搜索flag失败，丢弃rtp数据为：" << len;
return nullptr;
}
auto rtp_ptr = ps_header_flag_ptr - kPSHeaderOffset;
_search_rtp_finished = true;
if (rtp_ptr == data) {
// 停止搜索rtp，否则会进入死循环
_search_rtp = false;
}
InfoL << "rtp搜索flag成功，tcp上下文恢复成功，丢弃的rtp残余数据为：" << rtp_ptr - data;
// TODO or Not ? 更新设置ssrc
uint32_t rtp_ssrc = 0;
getSSRC(rtp_ptr + 2, len, rtp_ssrc);
_ssrc = rtp_ssrc;
InfoL << "设置_ssrc为：" << _ssrc;
// RtpServer::updateSSRC(uint32_t ssrc)
return rtp_ptr;
}
}//namespace mediakit
#endif//defined(ENABLE_RTPPROXY)
#endif//defined(ENABLE_RTPPROXY)

26
src/Rtp/RtpSession.h
View File

@@ -1,9 +1,9 @@
/*
* Copyright (c) 2016 The ZLMediaKit project authors. All Rights Reserved.
* Copyright (c) 2016-present The ZLMediaKit project authors. All Rights Reserved.
*
* This file is part of ZLMediaKit(https://github.com/xia-chu/ZLMediaKit).
* This file is part of ZLMediaKit(https://github.com/ZLMediaKit/ZLMediaKit).
*
* Use of this source code is governed by MIT license that can be found in the
* Use of this source code is governed by MIT-like 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.
*/
@@ -21,11 +21,14 @@
namespace mediakit{
class RtpSession : public toolkit::Session, public RtpSplitter, public MediaSourceEvent {
class RtpSession : public toolkit::Session, public RtpSplitter {
public:
static const std::string kVhost;
static const std::string kApp;
static const std::string kStreamID;
static const std::string kSSRC;
static const std::string kOnlyAudio;
static const std::string kOnlyTrack;
static const std::string kUdpRecvBuffer;
RtpSession(const toolkit::Socket::Ptr &sock);
~RtpSession() override;
@@ -34,24 +37,27 @@ public:
void onManager() override;
void setParams(toolkit::mINI &ini);
void attachServer(const toolkit::Server &server) override;
void setRtpProcess(RtpProcess::Ptr process);
protected:
// 通知其停止推流
bool close(MediaSource &sender) override;
// 收到rtp回调
void onRtpPacket(const char *data, size_t len) override;
// RtpSplitter override
const char *onSearchPacketTail(const char *data, size_t len) override;
// 搜寻SSRC
const char *searchBySSRC(const char *data, size_t len);
// 搜寻PS包里的关键帧标头
const char *searchByPsHeaderFlag(const char *data, size_t len);
private:
bool _delay_close = false;
bool _is_udp = false;
bool _search_rtp = false;
bool _search_rtp_finished = false;
bool _only_audio = false;
bool _emit_detach = false;
int _only_track = 0;
uint32_t _ssrc = 0;
toolkit::Ticker _ticker;
std::string _stream_id;
MediaTuple _tuple;
struct sockaddr_storage _addr;
RtpProcess::Ptr _process;
};

6
src/Rtp/RtpSplitter.cpp
View File

@@ -1,9 +1,9 @@
/*
* Copyright (c) 2016 The ZLMediaKit project authors. All Rights Reserved.
* Copyright (c) 2016-present The ZLMediaKit project authors. All Rights Reserved.
*
* This file is part of ZLMediaKit(https://github.com/xia-chu/ZLMediaKit).
* This file is part of ZLMediaKit(https://github.com/ZLMediaKit/ZLMediaKit).
*
* Use of this source code is governed by MIT license that can be found in the
* Use of this source code is governed by MIT-like 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.
*/

10
src/Rtp/RtpSplitter.h
View File

@@ -1,9 +1,9 @@
/*
* Copyright (c) 2016 The ZLMediaKit project authors. All Rights Reserved.
* Copyright (c) 2016-present The ZLMediaKit project authors. All Rights Reserved.
*
* This file is part of ZLMediaKit(https://github.com/xia-chu/ZLMediaKit).
* This file is part of ZLMediaKit(https://github.com/ZLMediaKit/ZLMediaKit).
*
* Use of this source code is governed by MIT license that can be found in the
* Use of this source code is governed by MIT-like 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.
*/
@@ -17,10 +17,6 @@
namespace mediakit{
class RtpSplitter : public HttpRequestSplitter{
public:
RtpSplitter() = default;
~RtpSplitter() override = default;
protected:
/**
* 收到rtp包回调

6
src/Rtp/TSDecoder.cpp
View File

@@ -1,9 +1,9 @@
/*
* Copyright (c) 2016 The ZLMediaKit project authors. All Rights Reserved.
* Copyright (c) 2016-present The ZLMediaKit project authors. All Rights Reserved.
*
* This file is part of ZLMediaKit(https://github.com/xia-chu/ZLMediaKit).
* This file is part of ZLMediaKit(https://github.com/ZLMediaKit/ZLMediaKit).
*
* Use of this source code is governed by MIT license that can be found in the
* Use of this source code is governed by MIT-like 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.
*/

7
src/Rtp/TSDecoder.h
View File

@@ -1,9 +1,9 @@
/*
* Copyright (c) 2016 The ZLMediaKit project authors. All Rights Reserved.
* Copyright (c) 2016-present The ZLMediaKit project authors. All Rights Reserved.
*
* This file is part of ZLMediaKit(https://github.com/xia-chu/ZLMediaKit).
* This file is part of ZLMediaKit(https://github.com/ZLMediaKit/ZLMediaKit).
*
* Use of this source code is governed by MIT license that can be found in the
* Use of this source code is governed by MIT-like 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.
*/
@@ -25,7 +25,6 @@ class TSSegment : public HttpRequestSplitter {
public:
typedef std::function<void(const char *data,size_t len)> onSegment;
TSSegment(size_t size = TS_PACKET_SIZE) : _size(size){}
~TSSegment() = default;
void setOnSegment(onSegment cb);
static bool isTSPacket(const char *data, size_t len);