ZLMediaKit/src/Rtsp/RtpBroadCaster.cpp

188 lines
6.3 KiB
C++
Raw Normal View History

2017-10-09 22:11:01 +08:00
/*
2017-09-27 16:20:30 +08:00
* MIT License
2017-04-01 16:35:56 +08:00
*
2017-09-27 16:20:30 +08:00
* Copyright (c) 2016 xiongziliang <771730766@qq.com>
*
* This file is part of ZLMediaKit(https://github.com/xiongziliang/ZLMediaKit).
*
* Permission is hereby granted, free of charge, to any person obtaining a copy
* of this software and associated documentation files (the "Software"), to deal
* in the Software without restriction, including without limitation the rights
* to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
* copies of the Software, and to permit persons to whom the Software is
* furnished to do so, subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in all
* copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
* AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
* LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
* OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
* SOFTWARE.
2017-04-01 16:35:56 +08:00
*/
#include <list>
2017-10-09 11:24:00 +08:00
#include <type_traits>
2017-04-25 11:35:41 +08:00
#include "RtpBroadCaster.h"
#include "Util/util.h"
#include "Network/sockutil.h"
2018-01-30 09:35:54 +08:00
#include "RtspSession.h"
2017-10-09 11:24:00 +08:00
using namespace std;
2018-03-05 10:41:15 +08:00
using namespace ZL::Network;
2017-04-01 16:35:56 +08:00
namespace ZL {
namespace Rtsp {
2018-02-07 13:43:44 +08:00
static uint32_t addressToInt(const string &ip){
struct in_addr addr;
bzero(&addr,sizeof(addr));
2018-03-05 10:41:15 +08:00
addr.s_addr = inet_addr(ip.data());
2018-02-07 13:43:44 +08:00
return (uint32_t)ntohl((uint32_t &)addr.s_addr);
}
2017-04-01 16:35:56 +08:00
std::shared_ptr<uint32_t> MultiCastAddressMaker::obtain(uint32_t iTry) {
lock_guard<recursive_mutex> lck(m_mtx);
2018-02-09 11:42:55 +08:00
GET_CONFIG_AND_REGISTER(string,addrMinStr,Config::MultiCast::kAddrMin);
GET_CONFIG_AND_REGISTER(string,addrMaxStr,Config::MultiCast::kAddrMax);
uint32_t addrMin = addressToInt(addrMinStr);
uint32_t addrMax = addressToInt(addrMaxStr);
2018-02-07 13:43:44 +08:00
if(m_iAddr > addrMax || m_iAddr == 0){
2017-04-01 16:35:56 +08:00
m_iAddr = addrMin;
}
auto iGotAddr = m_iAddr++;
if(m_setBadAddr.find(iGotAddr) != m_setBadAddr.end()){
//已经分配过了
if(iTry){
return obtain(--iTry);
}
//分配完了,应该不可能到这里
2018-04-09 11:26:39 +08:00
ErrorL;
2017-04-01 16:35:56 +08:00
return nullptr;
}
m_setBadAddr.emplace(iGotAddr);
std::shared_ptr<uint32_t> ret(new uint32_t(iGotAddr),[](uint32_t *ptr){
2018-02-07 13:43:44 +08:00
MultiCastAddressMaker::Instance().release(*ptr);
2017-04-01 16:35:56 +08:00
delete ptr;
});
return ret;
}
void MultiCastAddressMaker::release(uint32_t iAddr){
lock_guard<recursive_mutex> lck(m_mtx);
m_setBadAddr.erase(iAddr);
}
recursive_mutex RtpBroadCaster::g_mtx;
unordered_map<string, weak_ptr<RtpBroadCaster> > RtpBroadCaster::g_mapBroadCaster;
void RtpBroadCaster::setDetachCB(void* listener, const onDetach& cb) {
lock_guard<recursive_mutex> lck(m_mtx);
if(cb){
m_mapDetach.emplace(listener,cb);
}else{
m_mapDetach.erase(listener);
}
}
RtpBroadCaster::~RtpBroadCaster() {
2017-05-04 09:47:02 +08:00
m_pReader->setReadCB(nullptr);
m_pReader->setDetachCB(nullptr);
2017-04-01 16:35:56 +08:00
DebugL;
}
RtpBroadCaster::RtpBroadCaster(const string &strLocalIp,const string &strVhost,const string &strApp,const string &strStream) {
auto src = dynamic_pointer_cast<RtspMediaSource>(MediaSource::find(RTSP_SCHEMA,strVhost,strApp, strStream));
2017-04-01 16:35:56 +08:00
if(!src){
auto strErr = StrPrinter << "未找到媒体源:" << strVhost << " " << strApp << " " << strStream << endl;
2017-04-01 16:35:56 +08:00
throw std::runtime_error(strErr);
}
m_multiAddr = MultiCastAddressMaker::Instance().obtain();
for(auto i = 0; i < 2; i++){
m_apUdpSock[i].reset(new Socket());
if(!m_apUdpSock[i]->bindUdpSock(0, strLocalIp.data())){
auto strErr = StrPrinter << "绑定UDP端口失败:" << strLocalIp << endl;
throw std::runtime_error(strErr);
}
auto fd = m_apUdpSock[i]->rawFD();
2018-02-09 11:42:55 +08:00
GET_CONFIG_AND_REGISTER(uint32_t,udpTTL,Config::MultiCast::kUdpTTL);
SockUtil::setMultiTTL(fd, udpTTL);
2017-04-01 16:35:56 +08:00
SockUtil::setMultiLOOP(fd, false);
SockUtil::setMultiIF(fd, strLocalIp.data());
struct sockaddr_in &peerAddr = m_aPeerUdpAddr[i];
peerAddr.sin_family = AF_INET;
peerAddr.sin_port = htons(m_apUdpSock[i]->get_local_port());
peerAddr.sin_addr.s_addr = htonl(*m_multiAddr);
bzero(&(peerAddr.sin_zero), sizeof peerAddr.sin_zero);
}
m_pReader = src->getRing()->attach();
m_pReader->setReadCB([this](const RtpPacket::Ptr &pkt){
int i = (pkt->interleaved/2)%2;
auto &pSock = m_apUdpSock[i];
auto &peerAddr = m_aPeerUdpAddr[i];
2018-01-30 09:35:54 +08:00
BufferRtp::Ptr buffer(new BufferRtp(pkt,4));
pSock->send(buffer,SOCKET_DEFAULE_FLAGS,(struct sockaddr *)(&peerAddr));
2017-04-01 16:35:56 +08:00
});
m_pReader->setDetachCB([this](){
2017-05-04 09:47:02 +08:00
unordered_map<void * , onDetach > m_mapDetach_copy;
{
lock_guard<recursive_mutex> lck(m_mtx);
2017-10-09 11:24:00 +08:00
m_mapDetach_copy = std::move(m_mapDetach);
2017-04-01 16:35:56 +08:00
}
2017-05-04 09:47:02 +08:00
for(auto &pr : m_mapDetach_copy){
pr.second();
2017-04-01 16:35:56 +08:00
}
});
DebugL << MultiCastAddressMaker::toString(*m_multiAddr) << " "
<< m_apUdpSock[0]->get_local_port() << " "
<< m_apUdpSock[1]->get_local_port() << " "
<< strVhost << " "
2017-04-01 16:35:56 +08:00
<< strApp << " " << strStream;
}
uint16_t RtpBroadCaster::getPort(int iTrackId){
int i = iTrackId%2;
return m_apUdpSock[i]->get_local_port();
}
string RtpBroadCaster::getIP(){
return inet_ntoa(m_aPeerUdpAddr[0].sin_addr);
}
RtpBroadCaster::Ptr RtpBroadCaster::make(const string &strLocalIp,const string &strVhost,const string &strApp,const string &strStream){
2017-04-01 16:35:56 +08:00
try{
auto ret = Ptr(new RtpBroadCaster(strLocalIp,strVhost,strApp,strStream));
2017-04-01 16:35:56 +08:00
lock_guard<recursive_mutex> lck(g_mtx);
string strKey = StrPrinter << strLocalIp << " " << strVhost << " " << strApp << " " << strStream << endl;
2017-04-01 16:35:56 +08:00
weak_ptr<RtpBroadCaster> weakPtr = ret;
g_mapBroadCaster.emplace(strKey,weakPtr);
return ret;
}catch (std::exception &ex) {
WarnL << ex.what();
return nullptr;
}
}
RtpBroadCaster::Ptr RtpBroadCaster::get(const string &strLocalIp,const string &strVhost,const string &strApp,const string &strStream) {
string strKey = StrPrinter << strLocalIp << " " << strVhost << " " << strApp << " " << strStream << endl;
2017-04-01 16:35:56 +08:00
lock_guard<recursive_mutex> lck(g_mtx);
auto it = g_mapBroadCaster.find(strKey);
if (it == g_mapBroadCaster.end()) {
return make(strLocalIp,strVhost,strApp, strStream);
2017-04-01 16:35:56 +08:00
}
auto ret = it->second.lock();
if (!ret) {
g_mapBroadCaster.erase(it);
return make(strLocalIp,strVhost,strApp, strStream);
2017-04-01 16:35:56 +08:00
}
return ret;
}
} /* namespace Rtsp */
} /* namespace ZL */