ZLMediaKit/src/Rtp/RtpSelector.cpp

162 lines
4.6 KiB
C++
Raw Normal View History

2019-12-05 19:20:12 +08:00
/*
2020-04-04 20:30:09 +08:00
* Copyright (c) 2016 The ZLMediaKit project authors. All Rights Reserved.
2019-12-06 11:54:10 +08:00
*
* This file is part of ZLMediaKit(https://github.com/xia-chu/ZLMediaKit).
2019-12-06 11:54:10 +08:00
*
2020-04-04 20:30:09 +08:00
* 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.
2019-12-06 11:54:10 +08:00
*/
2019-12-05 19:20:12 +08:00
2019-12-06 11:54:10 +08:00
#if defined(ENABLE_RTPPROXY)
2020-10-24 23:33:13 +08:00
#include <stddef.h>
2019-12-05 19:20:12 +08:00
#include "RtpSelector.h"
2020-10-24 23:33:13 +08:00
#include "RtpSplitter.h"
2019-12-05 19:20:12 +08:00
2019-12-06 11:54:10 +08:00
namespace mediakit{
2019-12-05 19:20:12 +08:00
INSTANCE_IMP(RtpSelector);
2020-10-24 23:29:44 +08:00
void RtpSelector::clear(){
lock_guard<decltype(_mtx_map)> lck(_mtx_map);
_map_rtp_process.clear();
}
bool RtpSelector::inputRtp(const Socket::Ptr &sock, const char *data, size_t data_len,
const struct sockaddr *addr,uint32_t *dts_out) {
2020-07-08 09:36:10 +08:00
uint32_t ssrc = 0;
if (!getSSRC(data, data_len, ssrc)) {
WarnL << "get ssrc from rtp failed:" << data_len;
return false;
2019-12-05 19:20:12 +08:00
}
2020-07-08 09:36:10 +08:00
auto process = getProcess(printSSRC(ssrc), true);
if (process) {
2020-10-24 23:33:13 +08:00
try {
return process->inputRtp(true, sock, data, data_len, addr, dts_out);
} catch (...) {
delProcess(printSSRC(ssrc), process.get());
throw;
}
2019-12-05 19:20:12 +08:00
}
return false;
}
bool RtpSelector::getSSRC(const char *data, size_t data_len, uint32_t &ssrc){
if (data_len < 12) {
2020-02-23 12:16:20 +08:00
return false;
2019-12-05 19:20:12 +08:00
}
uint32_t *ssrc_ptr = (uint32_t *) (data + 8);
2020-02-23 12:16:20 +08:00
ssrc = ntohl(*ssrc_ptr);
return true;
2019-12-05 19:20:12 +08:00
}
RtpProcess::Ptr RtpSelector::getProcess(const string &stream_id,bool makeNew) {
2019-12-05 19:20:12 +08:00
lock_guard<decltype(_mtx_map)> lck(_mtx_map);
auto it = _map_rtp_process.find(stream_id);
if (it == _map_rtp_process.end() && !makeNew) {
2019-12-05 19:20:12 +08:00
return nullptr;
}
RtpProcessHelper::Ptr &ref = _map_rtp_process[stream_id];
if (!ref) {
ref = std::make_shared<RtpProcessHelper>(stream_id, shared_from_this());
2020-02-28 16:25:14 +08:00
ref->attachEvent();
2020-03-05 11:36:31 +08:00
createTimer();
2019-12-05 19:20:12 +08:00
}
2020-02-28 16:25:14 +08:00
return ref->getProcess();
2019-12-05 19:20:12 +08:00
}
2020-03-05 11:36:31 +08:00
void RtpSelector::createTimer() {
if (!_timer) {
//创建超时管理定时器
weak_ptr<RtpSelector> weakSelf = shared_from_this();
_timer = std::make_shared<Timer>(3.0f, [weakSelf] {
2020-03-05 11:36:31 +08:00
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) {
2020-09-20 10:43:25 +08:00
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);
2019-12-05 19:20:12 +08:00
}
process->onDetach();
2019-12-05 19:20:12 +08:00
}
void RtpSelector::onManager() {
2020-09-20 10:43:25 +08:00
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);
2019-12-05 19:20:12 +08:00
}
}
2020-09-20 10:43:25 +08:00
clear_list.for_each([](const RtpProcess::Ptr &process) {
process->onDetach();
});
2019-12-05 19:20:12 +08:00
}
RtpSelector::RtpSelector() {
}
RtpSelector::~RtpSelector() {
}
2019-12-06 11:54:10 +08:00
RtpProcessHelper::RtpProcessHelper(const string &stream_id, const weak_ptr<RtpSelector> &parent) {
_stream_id = stream_id;
2020-02-28 16:25:14 +08:00
_parent = parent;
_process = std::make_shared<RtpProcess>(stream_id);
2020-02-28 16:25:14 +08:00
}
RtpProcessHelper::~RtpProcessHelper() {
}
void RtpProcessHelper::attachEvent() {
_process->setListener(shared_from_this());
}
bool RtpProcessHelper::close(MediaSource &sender, bool force) {
//此回调在其他线程触发
2021-01-17 20:15:08 +08:00
if (!_process || (!force && _process->getTotalReaderCount())) {
2020-02-28 16:25:14 +08:00
return false;
}
auto parent = _parent.lock();
if (!parent) {
2020-02-28 16:25:14 +08:00
return false;
}
parent->delProcess(_stream_id, _process.get());
2020-03-05 18:04:34 +08:00
WarnL << "close media:" << sender.getSchema() << "/" << sender.getVhost() << "/" << sender.getApp() << "/" << sender.getId() << " " << force;
2020-02-28 16:25:14 +08:00
return true;
}
int RtpProcessHelper::totalReaderCount(MediaSource &sender) {
2021-01-17 20:15:08 +08:00
return _process ? _process->getTotalReaderCount() : sender.totalReaderCount();
2020-02-28 16:25:14 +08:00
}
RtpProcess::Ptr &RtpProcessHelper::getProcess() {
return _process;
}
2019-12-06 11:54:10 +08:00
}//namespace mediakit
#endif//defined(ENABLE_RTPPROXY)