amass/ZLMediaKit
1
0
Fork 0
mirror of https://github.com/ZLMediaKit/ZLMediaKit.git synced 2024-11-26 04:31:37 +08:00
Code Issues Actions 6 Packages Projects Releases Wiki Activity

Add rtp server

Browse Source
This commit is contained in:
Gemfield 2019-12-05 19:20:12 +08:00
parent 10dc254510
commit 55d258fafc
16 changed files with 1194 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

49
src/Rtp/PSDecoder.cpp Normal file
View File

@ -0,0 +1,49 @@
/*
* MIT License
*
* Copyright (c) 2016-2019 Gemfield <gemfield@civilnet.cn>
*
* This file is part of ZLMediaKit(https://github.com/xiongziliang/ZLMediaKit).
*
* Permission is hereby granted, free of charge, to any person obtaining a copy
* of this software and associated documentation files (the "Software"), to deal
* in the Software without restriction, including without limitation the rights
* to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
* copies of the Software, and to permit persons to whom the Software is
* furnished to do so, subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in all
* copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
* AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
* LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
* OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
* SOFTWARE.
*/
#include "PSDecoder.h"
PSDecoder::PSDecoder() {
_ps_demuxer = ps_demuxer_create([](void* param,
int stream,
int codecid,
int flags,
int64_t pts,
int64_t dts,
const void* data,
size_t bytes){
PSDecoder *thiz = (PSDecoder *)param;
thiz->onPSDecode(stream, codecid, flags, pts, dts, data, bytes);
},this);
}
PSDecoder::~PSDecoder() {
ps_demuxer_destroy(_ps_demuxer);
}
int PSDecoder::decodePS(const uint8_t *data, size_t bytes) {
return ps_demuxer_input(_ps_demuxer,data,bytes);
}

57
src/Rtp/PSDecoder.h Normal file
View File

@ -0,0 +1,57 @@
/*
* MIT License
*
* Copyright (c) 2016-2019 Gemfield <gemfield@civilnet.cn>
*
* This file is part of ZLMediaKit(https://github.com/xiongziliang/ZLMediaKit).
*
* Permission is hereby granted, free of charge, to any person obtaining a copy
* of this software and associated documentation files (the "Software"), to deal
* in the Software without restriction, including without limitation the rights
* to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
* copies of the Software, and to permit persons to whom the Software is
* furnished to do so, subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in all
* copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
* AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
* LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
* OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
* SOFTWARE.
*/
#ifndef RTPPROXY_PSDECODER_H
#define RTPPROXY_PSDECODER_H
#ifdef __cplusplus
extern "C" {
#endif
#include "mpeg-ps.h"
#ifdef __cplusplus
}
#endif
class PSDecoder {
public:
PSDecoder();
virtual ~PSDecoder();
int decodePS(const uint8_t *data, size_t bytes);
protected:
virtual void onPSDecode(int stream,
int codecid,
int flags,
int64_t pts,
int64_t dts,
const void *data,
size_t bytes) = 0;
private:
struct ps_demuxer_t *_ps_demuxer = nullptr;
};
#endif //RTPPROXY_PSDECODER_H

70
src/Rtp/RtpDecoder.cpp Normal file
View File

@ -0,0 +1,70 @@
/*
* MIT License
*
* Copyright (c) 2016-2019 Gemfield <gemfield@civilnet.cn>
*
* This file is part of ZLMediaKit(https://github.com/xiongziliang/ZLMediaKit).
*
* Permission is hereby granted, free of charge, to any person obtaining a copy
* of this software and associated documentation files (the "Software"), to deal
* in the Software without restriction, including without limitation the rights
* to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
* copies of the Software, and to permit persons to whom the Software is
* furnished to do so, subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in all
* copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
* AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
* LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
* OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
* SOFTWARE.
*/
#include <assert.h>
#include "Util/logger.h"
#include "RtpDecoder.h"
using namespace toolkit;
RtpDecoder::RtpDecoder() {
_buffer = std::make_shared<BufferRaw>();
}
RtpDecoder::~RtpDecoder() {
if(_rtp_decoder){
rtp_payload_decode_destroy(_rtp_decoder);
_rtp_decoder = nullptr;
}
}
void RtpDecoder::decodeRtp(const void *data, int bytes,const char *type_name) {
if(!_rtp_decoder){
static rtp_payload_t s_func= {
[](void* param, int bytes){
RtpDecoder *obj = (RtpDecoder *)param;
obj->_buffer->setCapacity(bytes);
return (void *)obj->_buffer->data();
},
[](void* param, void* packet){
//do nothing
},
[](void* param, const void *packet, int bytes, uint32_t timestamp, int flags){
RtpDecoder *obj = (RtpDecoder *)param;
obj->onRtpDecode(packet, bytes, timestamp, flags);
}
};
uint8_t rtp_type = 0x7F & ((uint8_t *) data)[1];
InfoL << "rtp type:" << (int) rtp_type;
_rtp_decoder = rtp_payload_decode_create(rtp_type, type_name, &s_func, this);
if (!_rtp_decoder) {
WarnL << "unsupported rtp type:" << (int) rtp_type << ",size:" << bytes << ",hexdump" << hexdump(data, bytes > 16 ? 16 : bytes);
}
}
if(_rtp_decoder){
rtp_payload_decode_input(_rtp_decoder,data,bytes);
}
}

55
src/Rtp/RtpDecoder.h Normal file
View File

@ -0,0 +1,55 @@
/*
* MIT License
*
* Copyright (c) 2016-2019 Gemfield <gemfield@civilnet.cn>
*
* This file is part of ZLMediaKit(https://github.com/xiongziliang/ZLMediaKit).
*
* Permission is hereby granted, free of charge, to any person obtaining a copy
* of this software and associated documentation files (the "Software"), to deal
* in the Software without restriction, including without limitation the rights
* to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
* copies of the Software, and to permit persons to whom the Software is
* furnished to do so, subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in all
* copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
* AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
* LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
* OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
* SOFTWARE.
*/
#ifndef IPTV_RTPTOTS_H
#define IPTV_RTPTOTS_H
#include "Network/Buffer.h"
using namespace toolkit;
#ifdef __cplusplus
extern "C" {
#endif
#include "rtp-payload.h"
#include "mpeg-ts.h"
#ifdef __cplusplus
}
#endif
class RtpDecoder {
public:
RtpDecoder();
virtual ~RtpDecoder();
protected:
void decodeRtp(const void *data, int bytes,const char *type_name);
virtual void onRtpDecode(const void *packet, int bytes, uint32_t timestamp, int flags) = 0;
private:
void *_rtp_decoder = nullptr;
BufferRaw::Ptr _buffer;
};
#endif //IPTV_RTPTOTS_H

79
src/Rtp/RtpFileLoader.cpp Normal file
View File

@ -0,0 +1,79 @@
/*
* MIT License
*
* Copyright (c) 2016-2019 Gemfield <gemfield@civilnet.cn>
*
* This file is part of ZLMediaKit(https://github.com/xiongziliang/ZLMediaKit).
*
* Permission is hereby granted, free of charge, to any person obtaining a copy
* of this software and associated documentation files (the "Software"), to deal
* in the Software without restriction, including without limitation the rights
* to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
* copies of the Software, and to permit persons to whom the Software is
* furnished to do so, subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in all
* copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
* AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
* LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
* OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
* SOFTWARE.
*/
#include <cstdint>
#include <cstdio>
#include <sys/socket.h>
#include "RtpFileLoader.h"
#include "Util/logger.h"
#include "RtpSelector.h"
using namespace toolkit;
bool RtpFileLoader::loadFile(const char *path) {
FILE *fp = fopen(path, "rb");
if (!fp) {
WarnL << "open file failed:" << path;
return false;
}
uint32_t timeStamp_last = 0;
uint16_t len;
char rtp[2 * 1024];
while (true) {
if (2 != fread(&len, 1, 2, fp)) {
WarnL;
break;
}
len = ntohs(len);
if (len < 12 || len > sizeof(rtp)) {
WarnL << len;
break;
}
if (len != fread(rtp, 1, len, fp)) {
WarnL;
break;
}
uint32_t timeStamp;
memcpy(&timeStamp, rtp + 4, 4);
timeStamp = ntohl(timeStamp);
timeStamp /= 90;
if(timeStamp_last){
auto diff = timeStamp - timeStamp_last;
if(diff > 0){
usleep(diff * 1000);
}
}
timeStamp_last = timeStamp;
RtpSelector::Instance().inputRtp(rtp,len, nullptr);
}
fclose(fp);
return true;
}

39
src/Rtp/RtpFileLoader.h Normal file
View File

@ -0,0 +1,39 @@
/*
* MIT License
*
* Copyright (c) 2016-2019 Gemfield <gemfield@civilnet.cn>
*
* This file is part of ZLMediaKit(https://github.com/xiongziliang/ZLMediaKit).
*
* Permission is hereby granted, free of charge, to any person obtaining a copy
* of this software and associated documentation files (the "Software"), to deal
* in the Software without restriction, including without limitation the rights
* to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
* copies of the Software, and to permit persons to whom the Software is
* furnished to do so, subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in all
* copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
* AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
* LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
* OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
* SOFTWARE.
*/
#ifndef RTPPROXY_RTPFILELOADER_H
#define RTPPROXY_RTPFILELOADER_H
class RtpFileLoader {
public:
RtpFileLoader(){};
~RtpFileLoader(){};
static bool loadFile(const char *path);
};
#endif //RTPPROXY_RTPFILELOADER_H

286
src/Rtp/RtpProcess.cpp Normal file
View File

@ -0,0 +1,286 @@
/*
* MIT License
*
* Copyright (c) 2016-2019 Gemfield <gemfield@civilnet.cn>
*
* This file is part of ZLMediaKit(https://github.com/xiongziliang/ZLMediaKit).
*
* Permission is hereby granted, free of charge, to any person obtaining a copy
* of this software and associated documentation files (the "Software"), to deal
* in the Software without restriction, including without limitation the rights
* to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
* copies of the Software, and to permit persons to whom the Software is
* furnished to do so, subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in all
* copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
* AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
* LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
* OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
* SOFTWARE.
*/
#include "RtpProcess.h"
#include "Util/File.h"
#include "Util/logger.h"
#include "Extension/H265.h"
#include "Extension/H264.h"
#include "Extension/AAC.h"
using namespace toolkit;
namespace dump {
const string kEnable = "dump.enable";
const string kDir = "dump.dir";
const string kChcekSource = "dump.checkSource";
onceToken token([](){
mINI::Instance()[kEnable] = 0;
mINI::Instance()[kDir] = "./dump/";
mINI::Instance()[kChcekSource] = 1;
});
}//namespace dump
namespace Rtp {
const string kRtpType = "rtp.rtp_type";
const string kTimeoutSec = "rtp.timeoutSec";
onceToken token([](){
mINI::Instance()[kRtpType] = "MP2P";
mINI::Instance()[kTimeoutSec] = 15;
});
}//namespace dump
/**
* 合并一些时间戳相同的frame
*/
class FrameMerger {
public:
FrameMerger() = default;
virtual ~FrameMerger() = default;
void inputFrame(const Frame::Ptr &frame,const function<void(uint32_t dts,uint32_t pts,const Buffer::Ptr &buffer)> &cb){
if (!_frameCached.empty() && _frameCached.back()->dts() != frame->dts()) {
Frame::Ptr back = _frameCached.back();
Buffer::Ptr merged_frame = back;
if(_frameCached.size() != 1){
string merged;
_frameCached.for_each([&](const Frame::Ptr &frame){
merged.append(frame->data(),frame->size());
});
merged_frame = std::make_shared<BufferString>(std::move(merged));
}
cb(back->dts(),back->pts(),merged_frame);
_frameCached.clear();
}
_frameCached.emplace_back(Frame::getCacheAbleFrame(frame));
}
private:
List<Frame::Ptr> _frameCached;
};
string printSSRC(uint32_t ui32Ssrc) {
char tmp[9] = { 0 };
ui32Ssrc = htonl(ui32Ssrc);
uint8_t *pSsrc = (uint8_t *) &ui32Ssrc;
for (int i = 0; i < 4; i++) {
sprintf(tmp + 2 * i, "%02X", pSsrc[i]);
}
return tmp;
}
static string printAddress(const struct sockaddr *addr){
return StrPrinter << inet_ntoa(((struct sockaddr_in *) addr)->sin_addr) << ":" << ntohs(((struct sockaddr_in *) addr)->sin_port);
}
RtpProcess::RtpProcess(uint32_t ssrc) {
_ssrc = ssrc;
_track = std::make_shared<SdpTrack>();
_track = std::make_shared<SdpTrack>();
_track->_interleaved = 0;
_track->_samplerate = 90000;
_track->_type = TrackVideo;
_track->_ssrc = _ssrc;
DebugL << printSSRC(_ssrc);
_muxer = std::make_shared<MultiMediaSourceMuxer>(DEFAULT_VHOST,"rtp",printSSRC(_ssrc));
GET_CONFIG(bool,dump_enable,dump::kEnable);
GET_CONFIG(string,dump_dir,dump::kDir);
{
FILE *fp = dump_enable ? File::createfile_file(File::absolutePath(printSSRC(_ssrc) + ".rtp",dump_dir).data(),"wb") : nullptr;
if(fp){
_save_file_rtp.reset(fp,[](FILE *fp){
fclose(fp);
});
}
}
{
FILE *fp = dump_enable ? File::createfile_file(File::absolutePath(printSSRC(_ssrc) + ".mp2",dump_dir).data(),"wb") : nullptr;
if(fp){
_save_file_ps.reset(fp,[](FILE *fp){
fclose(fp);
});
}
}
{
FILE *fp = dump_enable ? File::createfile_file(File::absolutePath(printSSRC(_ssrc) + ".video",dump_dir).data(),"wb") : nullptr;
if(fp){
_save_file_video.reset(fp,[](FILE *fp){
fclose(fp);
});
}
}
_merger = std::make_shared<FrameMerger>();
}
RtpProcess::~RtpProcess() {
if(_addr){
DebugL << printSSRC(_ssrc) << " " << printAddress(_addr);
delete _addr;
}else{
DebugL << printSSRC(_ssrc);
}
}
bool RtpProcess::inputRtp(const char *data, int data_len,const struct sockaddr *addr) {
GET_CONFIG(bool,check_source,dump::kChcekSource);
//检查源是否合法
if(!_addr){
_addr = new struct sockaddr;
memcpy(_addr,addr, sizeof(struct sockaddr));
DebugL << "RtpProcess(" << printSSRC(_ssrc) << ") bind to address:" << printAddress(_addr);
}
if(check_source && memcmp(_addr,addr,sizeof(struct sockaddr)) != 0){
DebugL << "RtpProcess(" << printSSRC(_ssrc) << ") address dismatch:" << printAddress(addr) << " != " << printAddress(_addr);
return false;
}
_last_rtp_time.resetTime();
return handleOneRtp(0,_track,(unsigned char *)data,data_len);
}
void RtpProcess::onRtpSorted(const RtpPacket::Ptr &rtp, int) {
if(rtp->sequence != _sequence + 1){
WarnL << rtp->sequence << " != " << _sequence << "+1";
}
_sequence = rtp->sequence;
if(_save_file_rtp){
uint16_t size = rtp->size() - 4;
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());
}
GET_CONFIG(string,rtp_type,::Rtp::kRtpType);
decodeRtp(rtp->data() + 4 ,rtp->size() - 4,rtp_type.data());
}
void RtpProcess::onRtpDecode(const void *packet, int bytes, uint32_t, int flags) {
if(_save_file_ps){
fwrite((uint8_t *)packet,bytes, 1, _save_file_ps.get());
}
auto ret = decodePS((uint8_t *)packet,bytes);
if(ret != bytes){
WarnL << ret << " != " << bytes << " " << flags;
}
}
void RtpProcess::onPSDecode(int stream,
int codecid,
int flags,
int64_t pts,
int64_t dts,
const void *data,
size_t bytes) {
switch (codecid) {
case STREAM_VIDEO_H264: {
if (!_codecid_video) {
//获取到视频
_codecid_video = codecid;
InfoL << "got video track: H264";
auto track = std::make_shared<H264Track>();
_muxer->addTrack(track);
}
if (codecid != _codecid_video) {
WarnL << "video track change to H264 from codecid:" << _codecid_video;
return;
}
if(_save_file_video){
fwrite((uint8_t *)data,bytes, 1, _save_file_video.get());
}
auto frame = std::make_shared<H264FrameNoCacheAble>((char *) data, bytes, dts / 90, pts / 90,0);
_merger->inputFrame(frame,[this](uint32_t dts, uint32_t pts, const Buffer::Ptr &buffer) {
_muxer->inputFrame(std::make_shared<H264FrameNoCacheAble>(buffer->data(), buffer->size(), dts, pts,4));
});
break;
}
case STREAM_VIDEO_H265: {
if (!_codecid_video) {
//获取到视频
_codecid_video = codecid;
InfoL << "got video track: H265";
auto track = std::make_shared<H265Track>();
_muxer->addTrack(track);
}
if (codecid != _codecid_video) {
WarnL << "video track change to H265 from codecid:" << _codecid_video;
return;
}
if(_save_file_video){
fwrite((uint8_t *)data,bytes, 1, _save_file_video.get());
}
auto frame = std::make_shared<H265FrameNoCacheAble>((char *) data, bytes, dts / 90, pts / 90, 0);
_merger->inputFrame(frame,[this](uint32_t dts, uint32_t pts, const Buffer::Ptr &buffer) {
_muxer->inputFrame(std::make_shared<H265FrameNoCacheAble>(buffer->data(), buffer->size(), dts, pts, 4));
});
break;
}
case STREAM_AUDIO_AAC: {
if (!_codecid_audio) {
//获取到音频
_codecid_audio = codecid;
InfoL << "got audio track: AAC";
auto track = std::make_shared<AACTrack>();
_muxer->addTrack(track);
}
if (codecid != _codecid_audio) {
WarnL << "audio track change to AAC from codecid:" << _codecid_audio;
return;
}
_muxer->inputFrame(std::make_shared<AACFrameNoCacheAble>((char *) data, bytes, dts / 90, 7));
break;
}
default:
WarnL << "unsupported codec type:" << codecid;
return;
}
}
bool RtpProcess::alive() {
GET_CONFIG(int,timeoutSec,::Rtp::kTimeoutSec)
if(_last_rtp_time.elapsedTime() / 1000 < timeoutSec){
return true;
}
return false;
}
string RtpProcess::get_peer_ip() {
return inet_ntoa(((struct sockaddr_in *) _addr)->sin_addr);
}
uint16_t RtpProcess::get_peer_port() {
return ntohs(((struct sockaddr_in *) _addr)->sin_port);
}

74
src/Rtp/RtpProcess.h Normal file
View File

@ -0,0 +1,74 @@
/*
* MIT License
*
* Copyright (c) 2016-2019 Gemfield <gemfield@civilnet.cn>
*
* This file is part of ZLMediaKit(https://github.com/xiongziliang/ZLMediaKit).
*
* Permission is hereby granted, free of charge, to any person obtaining a copy
* of this software and associated documentation files (the "Software"), to deal
* in the Software without restriction, including without limitation the rights
* to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
* copies of the Software, and to permit persons to whom the Software is
* furnished to do so, subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in all
* copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
* AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
* LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
* OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
* SOFTWARE.
*/
#ifndef RTPPROXY_RTPDECODER_H
#define RTPPROXY_RTPDECODER_H
#include "Rtsp/RtpReceiver.h"
#include "RtpDecoder.h"
#include "PSDecoder.h"
#include "Common/Device.h"
using namespace mediakit;
string printSSRC(uint32_t ui32Ssrc);
class FrameMerger;
class RtpProcess : public RtpReceiver , public RtpDecoder , public PSDecoder {
public:
typedef std::shared_ptr<RtpProcess> Ptr;
RtpProcess(uint32_t ssrc);
~RtpProcess();
bool inputRtp(const char *data,int data_len, const struct sockaddr *addr);
bool alive();
string get_peer_ip();
uint16_t get_peer_port();
protected:
void onRtpSorted(const RtpPacket::Ptr &rtp, int track_index) override ;
void onRtpDecode(const void *packet, int bytes, uint32_t timestamp, int flags) override;
void onPSDecode(int stream,
int codecid,
int flags,
int64_t pts,
int64_t dts,
const void *data,
size_t bytes) override ;
private:
std::shared_ptr<FILE> _save_file_rtp;
std::shared_ptr<FILE> _save_file_ps;
std::shared_ptr<FILE> _save_file_video;
uint32_t _ssrc;
SdpTrack::Ptr _track;
struct sockaddr *_addr = nullptr;
uint16_t _sequence = 0;
int _codecid_video = 0;
int _codecid_audio = 0;
MultiMediaSourceMuxer::Ptr _muxer;
std::shared_ptr<FrameMerger> _merger;
Ticker _last_rtp_time;
};
#endif //RTPPROXY_RTPDECODER_H

99
src/Rtp/RtpSelector.cpp Normal file
View File

@ -0,0 +1,99 @@
/*
* MIT License
*
* Copyright (c) 2016-2019 Gemfield <gemfield@civilnet.cn>
*
* This file is part of ZLMediaKit(https://github.com/xiongziliang/ZLMediaKit).
*
* Permission is hereby granted, free of charge, to any person obtaining a copy
* of this software and associated documentation files (the "Software"), to deal
* in the Software without restriction, including without limitation the rights
* to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
* copies of the Software, and to permit persons to whom the Software is
* furnished to do so, subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in all
* copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
* AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
* LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
* OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
* SOFTWARE.
*/
#include "RtpSelector.h"
INSTANCE_IMP(RtpSelector);
bool RtpSelector::inputRtp(const char *data, int data_len,const struct sockaddr *addr) {
if(_last_rtp_time.elapsedTime() > 3000){
_last_rtp_time.resetTime();
onManager();
}
auto ssrc = getSSRC(data,data_len);
if(!ssrc){
WarnL << "get ssrc from rtp failed:" << data_len;
return false;
}
auto process = getProcess(ssrc, true);
if(process){
return process->inputRtp(data,data_len, addr);
}
return false;
}
uint32_t RtpSelector::getSSRC(const char *data, int data_len) {
if(data_len < 12){
return 0;
}
uint32_t *ssrc_ptr = (uint32_t *)(data + 8);
return ntohl(*ssrc_ptr);
}
RtpProcess::Ptr RtpSelector::getProcess(uint32_t ssrc,bool makeNew) {
lock_guard<decltype(_mtx_map)> lck(_mtx_map);
auto it = _map_rtp_process.find(ssrc);
if(it == _map_rtp_process.end() && !makeNew){
return nullptr;
}
RtpProcess::Ptr &ref = _map_rtp_process[ssrc];
if(!ref){
ref = std::make_shared<RtpProcess>(ssrc);
}
return ref;
}
void RtpSelector::delProcess(uint32_t ssrc,const RtpProcess *ptr) {
lock_guard<decltype(_mtx_map)> lck(_mtx_map);
auto it = _map_rtp_process.find(ssrc);
if(it == _map_rtp_process.end()){
return;
}
if(it->second.get() != ptr){
return;
}
_map_rtp_process.erase(it);
}
void RtpSelector::onManager() {
lock_guard<decltype(_mtx_map)> lck(_mtx_map);
for (auto it = _map_rtp_process.begin(); it != _map_rtp_process.end();) {
if (it->second->alive()) {
++it;
continue;
}
WarnL << "RtpProcess timeout:" << printSSRC(it->first);
it = _map_rtp_process.erase(it);
}
}
RtpSelector::RtpSelector() {
}
RtpSelector::~RtpSelector() {
}

55
src/Rtp/RtpSelector.h Normal file
View File

@ -0,0 +1,55 @@
/*
* MIT License
*
* Copyright (c) 2016-2019 Gemfield <gemfield@civilnet.cn>
*
* This file is part of ZLMediaKit(https://github.com/xiongziliang/ZLMediaKit).
*
* Permission is hereby granted, free of charge, to any person obtaining a copy
* of this software and associated documentation files (the "Software"), to deal
* in the Software without restriction, including without limitation the rights
* to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
* copies of the Software, and to permit persons to whom the Software is
* furnished to do so, subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in all
* copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
* AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
* LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
* OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
* SOFTWARE.
*/
#ifndef RTPPROXY_RTPSELECTOR_H
#define RTPPROXY_RTPSELECTOR_H
#include <cstdint>
#include <mutex>
#include <unordered_map>
#include "RtpProcess.h"
class RtpSelector : public std::enable_shared_from_this<RtpSelector>{
public:
RtpSelector();
~RtpSelector();
static RtpSelector &Instance();
bool inputRtp(const char *data,int data_len,const struct sockaddr *addr);
static uint32_t getSSRC(const char *data,int data_len);
RtpProcess::Ptr getProcess(uint32_t ssrc,bool makeNew);
void delProcess(uint32_t ssrc,const RtpProcess *ptr);
private:
void onManager();
private:
unordered_map<uint32_t,RtpProcess::Ptr> _map_rtp_process;
recursive_mutex _mtx_map;
Ticker _last_rtp_time;
};
#endif //RTPPROXY_RTPSELECTOR_H

72
src/Rtp/RtpSession.cpp Normal file
View File

@ -0,0 +1,72 @@
/*
* MIT License
*
* Copyright (c) 2016-2019 Gemfield <gemfield@civilnet.cn>
*
* This file is part of ZLMediaKit(https://github.com/xiongziliang/ZLMediaKit).
*
* Permission is hereby granted, free of charge, to any person obtaining a copy
* of this software and associated documentation files (the "Software"), to deal
* in the Software without restriction, including without limitation the rights
* to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
* copies of the Software, and to permit persons to whom the Software is
* furnished to do so, subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in all
* copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
* AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
* LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
* OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
* SOFTWARE.
*/
#include "RtpSession.h"
#include "RtpSelector.h"
RtpSession::RtpSession(const Socket::Ptr &sock) : TcpSession(sock) {
DebugP(this);
socklen_t addr_len = sizeof(addr);
getpeername(sock->rawFD(), &addr, &addr_len);
}
RtpSession::~RtpSession() {
DebugP(this);
if(_ssrc){
RtpSelector::Instance().delProcess(_ssrc,_process.get());
}
}
void RtpSession::onRecv(const Buffer::Ptr &data) {
try {
RtpSplitter::input(data->data(), data->size());
} catch (SockException &ex) {
shutdown(ex);
} catch (std::exception &ex) {
shutdown(SockException(Err_other, ex.what()));
}
}
void RtpSession::onError(const SockException &err) {
WarnL << _ssrc << " " << err.what();
}
void RtpSession::onManager() {
if(_process && !_process->alive()){
shutdown(SockException(Err_timeout, "receive rtp timeout"));
}
if(!_process && _ticker.createdTime() > 10 * 1000){
shutdown(SockException(Err_timeout, "illegal connection"));
}
}
void RtpSession::onRtpPacket(const char *data, uint64_t len) {
if(!_ssrc){
_ssrc = RtpSelector::getSSRC(data,len);
_process = RtpSelector::Instance().getProcess(_ssrc, true);
}
_process->inputRtp(data + 2,len - 2,&addr);
_ticker.resetTime();
}

54
src/Rtp/RtpSession.h Normal file
View File

@ -0,0 +1,54 @@
/*
* MIT License
*
* Copyright (c) 2016-2019 Gemfield <gemfield@civilnet.cn>
*
* This file is part of ZLMediaKit(https://github.com/xiongziliang/ZLMediaKit).
*
* Permission is hereby granted, free of charge, to any person obtaining a copy
* of this software and associated documentation files (the "Software"), to deal
* in the Software without restriction, including without limitation the rights
* to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
* copies of the Software, and to permit persons to whom the Software is
* furnished to do so, subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in all
* copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
* AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
* LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
* OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
* SOFTWARE.
*/
#ifndef RTPPROXY_RTPSESSION_H
#define RTPPROXY_RTPSESSION_H
#include "Network/TcpSession.h"
#include "RtpSplitter.h"
#include "RtpProcess.h"
#include "Util/TimeTicker.h"
using namespace toolkit;
class RtpSession : public TcpSession , public RtpSplitter{
public:
RtpSession(const Socket::Ptr &sock);
~RtpSession() override;
void onRecv(const Buffer::Ptr &) override;
void onError(const SockException &err) override;
void onManager() override;
private:
void onRtpPacket(const char *data,uint64_t len) override;
private:
uint32_t _ssrc = 0;
RtpProcess::Ptr _process;
Ticker _ticker;
struct sockaddr addr;
};
#endif //RTPPROXY_RTPSESSION_H

53
src/Rtp/RtpSplitter.cpp Normal file
View File

@ -0,0 +1,53 @@
/*
* MIT License
*
* Copyright (c) 2016-2019 Gemfield <gemfield@civilnet.cn>
*
* This file is part of ZLMediaKit(https://github.com/xiongziliang/ZLMediaKit).
*
* Permission is hereby granted, free of charge, to any person obtaining a copy
* of this software and associated documentation files (the "Software"), to deal
* in the Software without restriction, including without limitation the rights
* to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
* copies of the Software, and to permit persons to whom the Software is
* furnished to do so, subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in all
* copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
* AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
* LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
* OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
* SOFTWARE.
*/
#include "RtpSplitter.h"
RtpSplitter::RtpSplitter() {
}
RtpSplitter::~RtpSplitter() {
}
const char *RtpSplitter::onSearchPacketTail(const char *data, int len) {
//这是rtp包
if(len < 2){
//数据不够
return nullptr;
}
uint16_t length = (((uint8_t *)data)[0] << 8) | ((uint8_t *)data)[1];
if(len < length + 2){
//数据不够
return nullptr;
}
//返回rtp包末尾
return data + 2 + length;
}
int64_t RtpSplitter::onRecvHeader(const char *data, uint64_t len) {
onRtpPacket(data,len);
return 0;
}

50
src/Rtp/RtpSplitter.h Normal file
View File

@ -0,0 +1,50 @@
/*
* MIT License
*
* Copyright (c) 2016-2019 Gemfield <gemfield@civilnet.cn>
*
* This file is part of ZLMediaKit(https://github.com/xiongziliang/ZLMediaKit).
*
* Permission is hereby granted, free of charge, to any person obtaining a copy
* of this software and associated documentation files (the "Software"), to deal
* in the Software without restriction, including without limitation the rights
* to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
* copies of the Software, and to permit persons to whom the Software is
* furnished to do so, subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in all
* copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
* AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
* LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
* OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
* SOFTWARE.
*/
#ifndef RTPPROXY_RTPSPLITTER_H
#define RTPPROXY_RTPSPLITTER_H
#include "Http/HttpRequestSplitter.h"
using namespace mediakit;
class RtpSplitter : public HttpRequestSplitter{
public:
RtpSplitter();
virtual ~RtpSplitter();
protected:
/**
* 收到rtp包回调
* @param data
* @param len
*/
virtual void onRtpPacket(const char *data,uint64_t len) = 0;
protected:
const char *onSearchPacketTail(const char *data,int len) override ;
int64_t onRecvHeader(const char *data,uint64_t len) override;
};
#endif //RTPPROXY_RTPSPLITTER_H

50
src/Rtp/UdpRecver.cpp Normal file
View File

@ -0,0 +1,50 @@
/*
* MIT License
*
* Copyright (c) 2016-2019 Gemfield <gemfield@civilnet.cn>
*
* This file is part of ZLMediaKit(https://github.com/xiongziliang/ZLMediaKit).
*
* Permission is hereby granted, free of charge, to any person obtaining a copy
* of this software and associated documentation files (the "Software"), to deal
* in the Software without restriction, including without limitation the rights
* to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
* copies of the Software, and to permit persons to whom the Software is
* furnished to do so, subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in all
* copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
* AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
* LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
* OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
* SOFTWARE.
*/
#include "UdpRecver.h"
#include "RtpSelector.h"
UdpRecver::UdpRecver() {
}
UdpRecver::~UdpRecver() {
}
bool UdpRecver::initSock(uint16_t local_port,const char *local_ip) {
_sock.reset(new Socket(nullptr, false));
onceToken token(nullptr,[&](){
SockUtil::setRecvBuf(_sock->rawFD(),4 * 1024 * 1024);
});
auto &ref = RtpSelector::Instance();
_sock->setOnRead([&ref](const Buffer::Ptr &buf, struct sockaddr *addr, int ){
ref.inputRtp(buf->data(),buf->size(),addr);
});
return _sock->bindUdpSock(local_port,local_ip);
}
EventPoller::Ptr UdpRecver::getPoller() {
return _sock->getPoller();
}

52
src/Rtp/UdpRecver.h Normal file
View File

@ -0,0 +1,52 @@
/*
* MIT License
*
* Copyright (c) 2016-2019 Gemfield <gemfield@civilnet.cn>
*
* This file is part of ZLMediaKit(https://github.com/xiongziliang/ZLMediaKit).
*
* Permission is hereby granted, free of charge, to any person obtaining a copy
* of this software and associated documentation files (the "Software"), to deal
* in the Software without restriction, including without limitation the rights
* to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
* copies of the Software, and to permit persons to whom the Software is
* furnished to do so, subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in all
* copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
* AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
* LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
* OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
* SOFTWARE.
*/
#ifndef ZLMEDIAKIT_UDPRECVER_H
#define ZLMEDIAKIT_UDPRECVER_H
#include <memory>
#include "Network/Socket.h"
using namespace std;
using namespace toolkit;
/**
* 组播接收器
*/
class UdpRecver {
public:
typedef std::shared_ptr<UdpRecver> Ptr;
typedef function<void(const Buffer::Ptr &buf)> onRecv;
UdpRecver();
virtual ~UdpRecver();
bool initSock(uint16_t local_port,const char *local_ip = "0.0.0.0");
EventPoller::Ptr getPoller();
protected:
Socket::Ptr _sock;
};
#endif //ZLMEDIAKIT_UDPRECVER_H