Skip to content

Z
ZLMediaKit
Commit 5cc684ac by lyg1949 Committed by GitHub
Options

Merge pull request #2 from xiongziliang/master 

Merge from zlmediakit
parents 0c9dd568 7bce2127
正在显示 包含 489 行增加 和 194 行删除
ZLToolKit @ ed47015f
Subproject commit 5030af90126ea8f01ded6744ae8abdf549d00a81 Subproject commit ed47015f92cc79dfe3344b3666aafb54f1bbc2f4
media-server @ 7f7906b0
Subproject commit 576216c64bf3bcdc5e787da2adb3e169bdd97118 Subproject commit 7f7906b05d84c5efeceecb8d6f540a71c8153431
CMakeLists.txt
...@@ -41,13 +41,17 @@ INCLUDE_DIRECTORIES(${ToolKit_Root}) ...@@ -41,13 +41,17 @@ INCLUDE_DIRECTORIES(${ToolKit_Root})
INCLUDE_DIRECTORIES(${MediaKit_Root}) INCLUDE_DIRECTORIES(${MediaKit_Root})
INCLUDE_DIRECTORIES(${CMAKE_CURRENT_SOURCE_DIR}/3rdpart) INCLUDE_DIRECTORIES(${CMAKE_CURRENT_SOURCE_DIR}/3rdpart)
set(ENABLE_HLS true) option(ENABLE_HLS "Enable HLS" true)
set(ENABLE_OPENSSL true) option(ENABLE_OPENSSL "Enable OpenSSL" true)
set(ENABLE_MYSQL false) option(ENABLE_MYSQL "Enable MySQL" false)
set(ENABLE_FAAC false) option(ENABLE_FAAC "Enable FAAC" false)
set(ENABLE_X264 false) option(ENABLE_X264 "Enable x264" false)
set(ENABLE_MP4 true) option(ENABLE_MP4 "Enable MP4" true)
set(ENABLE_RTPPROXY true) option(ENABLE_RTPPROXY "Enable RTPPROXY" true)
option(ENABLE_API "Enable C API SDK" true)
option(ENABLE_CXX_API "Enable C++ API SDK" false)
option(ENABLE_TESTS "Enable Tests" true)
option(ENABLE_SERVER "Enable Server" true)
set(LINK_LIB_LIST zlmediakit zltoolkit) set(LINK_LIB_LIST zlmediakit zltoolkit)
...@@ -175,6 +179,22 @@ endif () ...@@ -175,6 +179,22 @@ endif ()
add_library(zltoolkit STATIC ${ToolKit_src_list}) add_library(zltoolkit STATIC ${ToolKit_src_list})
add_library(zlmediakit STATIC ${MediaKit_src_list}) add_library(zlmediakit STATIC ${MediaKit_src_list})
#安装目录
if (WIN32)
set(INSTALL_PATH_LIB $ENV{HOME}/${CMAKE_PROJECT_NAME}/lib)
set(INSTALL_PATH_INCLUDE $ENV{HOME}/${CMAKE_PROJECT_NAME}/include)
else ()
set(INSTALL_PATH_LIB lib)
set(INSTALL_PATH_INCLUDE include)
endif ()
if(ENABLE_CXX_API)
# 保留目录结构
install(DIRECTORY ${ToolKit_Root}/ DESTINATION ${INSTALL_PATH_INCLUDE}/ZLToolKit REGEX "(.*[.](md|cpp)|win32)$" EXCLUDE)
install(DIRECTORY ${MediaKit_Root}/ DESTINATION ${INSTALL_PATH_INCLUDE}/ZLMediaKit REGEX ".*[.](md|cpp)$" EXCLUDE)
install(TARGETS zltoolkit zlmediakit DESTINATION ${INSTALL_PATH_LIB})
endif()
if (WIN32) if (WIN32)
list(APPEND LINK_LIB_LIST WS2_32 Iphlpapi shlwapi) list(APPEND LINK_LIB_LIST WS2_32 Iphlpapi shlwapi)
set_target_properties(zltoolkit PROPERTIES COMPILE_FLAGS ${VS_FALGS} ) set_target_properties(zltoolkit PROPERTIES COMPILE_FLAGS ${VS_FALGS} )
...@@ -188,11 +208,17 @@ execute_process(COMMAND cp -r ${CMAKE_CURRENT_SOURCE_DIR}/www ${EXECUTABLE_OUTPU ...@@ -188,11 +208,17 @@ execute_process(COMMAND cp -r ${CMAKE_CURRENT_SOURCE_DIR}/www ${EXECUTABLE_OUTPU
execute_process(COMMAND cp ${CMAKE_CURRENT_SOURCE_DIR}/conf/config.ini ${EXECUTABLE_OUTPUT_PATH}/) execute_process(COMMAND cp ${CMAKE_CURRENT_SOURCE_DIR}/conf/config.ini ${EXECUTABLE_OUTPUT_PATH}/)
#添加c库 #添加c库
add_subdirectory(api) if(ENABLE_API)
add_subdirectory(api)
endif()
if (NOT IOS) if (NOT IOS)
#测试程序 #测试程序
add_subdirectory(tests) if(ENABLE_TESTS)
add_subdirectory(tests)
endif()
#主服务器 #主服务器
add_subdirectory(server) if(ENABLE_SERVER)
add_subdirectory(server)
endif()
endif () endif ()
api/CMakeLists.txt
...@@ -22,15 +22,6 @@ else () ...@@ -22,15 +22,6 @@ else ()
target_link_libraries(mk_api ${LINK_LIB_LIST}) target_link_libraries(mk_api ${LINK_LIB_LIST})
add_subdirectory(tests) add_subdirectory(tests)
#安装目录
if (WIN32)
set(INSTALL_PATH_LIB $ENV{HOME}/${CMAKE_PROJECT_NAME}/lib)
set(INSTALL_PATH_INCLUDE $ENV{HOME}/${CMAKE_PROJECT_NAME}/include)
else ()
set(INSTALL_PATH_LIB lib)
set(INSTALL_PATH_INCLUDE include)
endif ()
file(GLOB api_header_list include/*.h) file(GLOB api_header_list include/*.h)
install(FILES ${api_header_list} DESTINATION ${INSTALL_PATH_INCLUDE}) install(FILES ${api_header_list} DESTINATION ${INSTALL_PATH_INCLUDE})
install(TARGETS mk_api ARCHIVE DESTINATION ${INSTALL_PATH_LIB} LIBRARY DESTINATION ${INSTALL_PATH_LIB}) install(TARGETS mk_api ARCHIVE DESTINATION ${INSTALL_PATH_LIB} LIBRARY DESTINATION ${INSTALL_PATH_LIB})
......
api/include/mk_media.h
...@@ -12,6 +12,7 @@ ...@@ -12,6 +12,7 @@
#define MK_MEDIA_H_ #define MK_MEDIA_H_
#include "mk_common.h" #include "mk_common.h"
#include "mk_events_objects.h"
#ifdef __cplusplus #ifdef __cplusplus
extern "C" { extern "C" {
...@@ -158,6 +159,22 @@ API_EXPORT void API_CALL mk_media_set_on_seek(mk_media ctx, on_mk_media_seek cb, ...@@ -158,6 +159,22 @@ API_EXPORT void API_CALL mk_media_set_on_seek(mk_media ctx, on_mk_media_seek cb,
*/ */
API_EXPORT int API_CALL mk_media_total_reader_count(mk_media ctx); API_EXPORT int API_CALL mk_media_total_reader_count(mk_media ctx);
/**
* 生成的MediaSource注册或注销事件
* @param user_data 设置回调时的用户数据指针
* @param sender 生成的MediaSource对象
* @param regist 1为注册事件，0为注销事件
*/
typedef void(API_CALL *on_mk_media_source_regist)(void *user_data, mk_media_source sender, int regist);
/**
* 设置MediaSource注册或注销事件回调函数
* @param ctx 对象指针
* @param cb 回调指针
* @param user_data 用户数据指针
*/
API_EXPORT void API_CALL mk_media_set_on_regist(mk_media ctx, on_mk_media_source_regist cb, void *user_data);
#ifdef __cplusplus #ifdef __cplusplus
} }
#endif #endif
......
api/source/mk_common.cpp
...@@ -30,10 +30,8 @@ static TcpServer::Ptr http_server[2]; ...@@ -30,10 +30,8 @@ static TcpServer::Ptr http_server[2];
static TcpServer::Ptr shell_server; static TcpServer::Ptr shell_server;
#ifdef ENABLE_RTPPROXY #ifdef ENABLE_RTPPROXY
#include "Rtp/UdpRecver.h" #include "Rtp/RtpServer.h"
#include "Rtp/RtpSession.h" static std::shared_ptr<RtpServer> rtpServer;
static std::shared_ptr<UdpRecver> udpRtpServer;
static TcpServer::Ptr tcpRtpServer;
#endif #endif
//////////////////////////environment init/////////////////////////// //////////////////////////environment init///////////////////////////
...@@ -57,8 +55,7 @@ API_EXPORT void API_CALL mk_stop_all_server(){ ...@@ -57,8 +55,7 @@ API_EXPORT void API_CALL mk_stop_all_server(){
CLEAR_ARR(rtmp_server); CLEAR_ARR(rtmp_server);
CLEAR_ARR(http_server); CLEAR_ARR(http_server);
#ifdef ENABLE_RTPPROXY #ifdef ENABLE_RTPPROXY
udpRtpServer = nullptr; rtpServer = nullptr;
tcpRtpServer = nullptr;
#endif #endif
stopAllTcpServer(); stopAllTcpServer();
} }
...@@ -184,18 +181,12 @@ API_EXPORT uint16_t API_CALL mk_rtmp_server_start(uint16_t port, int ssl) { ...@@ -184,18 +181,12 @@ API_EXPORT uint16_t API_CALL mk_rtmp_server_start(uint16_t port, int ssl) {
API_EXPORT uint16_t API_CALL mk_rtp_server_start(uint16_t port){ API_EXPORT uint16_t API_CALL mk_rtp_server_start(uint16_t port){
#ifdef ENABLE_RTPPROXY #ifdef ENABLE_RTPPROXY
try { try {
//创建rtp tcp服务器 //创建rtp 服务器
tcpRtpServer = std::make_shared<TcpServer>(); rtpServer = std::make_shared<RtpServer>();
tcpRtpServer->start<RtpSession>(port); rtpServer->start(port);
return rtpServer->getPort();
//创建rtp udp服务器
auto ret = tcpRtpServer->getPort();
udpRtpServer = std::make_shared<UdpRecver>();
udpRtpServer->initSock(port);
return ret;
} catch (std::exception &ex) { } catch (std::exception &ex) {
tcpRtpServer.reset(); rtpServer.reset();
udpRtpServer.reset();
WarnL << ex.what(); WarnL << ex.what();
return 0; return 0;
} }
......
api/source/mk_media.cpp
...@@ -42,6 +42,12 @@ public: ...@@ -42,6 +42,12 @@ public:
_on_seek = cb; _on_seek = cb;
_on_seek_data = user_data; _on_seek_data = user_data;
} }
void setOnRegist(on_mk_media_source_regist cb, void *user_data){
_on_regist = cb;
_on_regist_data = user_data;
}
protected: protected:
// 通知其停止推流 // 通知其停止推流
bool close(MediaSource &sender,bool force) override{ bool close(MediaSource &sender,bool force) override{
...@@ -70,12 +76,21 @@ protected: ...@@ -70,12 +76,21 @@ protected:
int totalReaderCount(MediaSource &sender) override{ int totalReaderCount(MediaSource &sender) override{
return _channel->totalReaderCount(); return _channel->totalReaderCount();
} }
void onRegist(MediaSource &sender, bool regist) override{
if (_on_regist) {
_on_regist(_on_regist_data, &sender, regist);
}
}
private: private:
DevChannel::Ptr _channel; DevChannel::Ptr _channel;
on_mk_media_close _on_close = nullptr; on_mk_media_close _on_close = nullptr;
on_mk_media_seek _on_seek = nullptr; on_mk_media_seek _on_seek = nullptr;
on_mk_media_source_regist _on_regist = nullptr;
void *_on_seek_data; void *_on_seek_data;
void *_on_close_data; void *_on_close_data;
void *_on_regist_data;
}; };
API_EXPORT void API_CALL mk_media_set_on_close(mk_media ctx, on_mk_media_close cb, void *user_data){ API_EXPORT void API_CALL mk_media_set_on_close(mk_media ctx, on_mk_media_close cb, void *user_data){
...@@ -90,6 +105,12 @@ API_EXPORT void API_CALL mk_media_set_on_seek(mk_media ctx, on_mk_media_seek cb, ...@@ -90,6 +105,12 @@ API_EXPORT void API_CALL mk_media_set_on_seek(mk_media ctx, on_mk_media_seek cb,
(*obj)->setOnSeek(cb, user_data); (*obj)->setOnSeek(cb, user_data);
} }
API_EXPORT void API_CALL mk_media_set_on_regist(mk_media ctx, on_mk_media_source_regist cb, void *user_data){
assert(ctx);
MediaHelper::Ptr *obj = (MediaHelper::Ptr *) ctx;
(*obj)->setOnRegist(cb, user_data);
}
API_EXPORT int API_CALL mk_media_total_reader_count(mk_media ctx){ API_EXPORT int API_CALL mk_media_total_reader_count(mk_media ctx){
assert(ctx); assert(ctx);
MediaHelper::Ptr *obj = (MediaHelper::Ptr *) ctx; MediaHelper::Ptr *obj = (MediaHelper::Ptr *) ctx;
......
server/FFmpegSource.cpp
...@@ -249,6 +249,13 @@ void FFmpegSource::onNoneReader(MediaSource &sender){ ...@@ -249,6 +249,13 @@ void FFmpegSource::onNoneReader(MediaSource &sender){
MediaSourceEvent::onNoneReader(sender); MediaSourceEvent::onNoneReader(sender);
} }
void FFmpegSource::onRegist(MediaSource &sender, bool regist){
auto listener = _listener.lock();
if(listener){
listener->onRegist(sender, regist);
}
}
void FFmpegSource::onGetMediaSource(const MediaSource::Ptr &src) { void FFmpegSource::onGetMediaSource(const MediaSource::Ptr &src) {
_listener = src->getListener(); _listener = src->getListener();
src->setListener(shared_from_this()); src->setListener(shared_from_this());
......
server/FFmpegSource.h
...@@ -62,6 +62,7 @@ private: ...@@ -62,6 +62,7 @@ private:
bool close(MediaSource &sender,bool force) override; bool close(MediaSource &sender,bool force) override;
int totalReaderCount(MediaSource &sender) override; int totalReaderCount(MediaSource &sender) override;
void onNoneReader(MediaSource &sender) override; void onNoneReader(MediaSource &sender) override;
void onRegist(MediaSource &sender, bool regist) override;
private: private:
Process _process; Process _process;
......
server/WebApi.cpp
...@@ -34,6 +34,9 @@ ...@@ -34,6 +34,9 @@
#include "Thread/WorkThreadPool.h" #include "Thread/WorkThreadPool.h"
#include "Rtp/RtpSelector.h" #include "Rtp/RtpSelector.h"
#include "FFmpegSource.h" #include "FFmpegSource.h"
#if defined(ENABLE_RTPPROXY)
#include "Rtp/RtpServer.h"
#endif
using namespace Json; using namespace Json;
using namespace toolkit; using namespace toolkit;
using namespace mediakit; using namespace mediakit;
...@@ -244,15 +247,24 @@ bool checkArgs(Args &&args,First &&first,KeyTypes && ...keys){ ...@@ -244,15 +247,24 @@ bool checkArgs(Args &&args,First &&first,KeyTypes && ...keys){
} \ } \
} }
//拉流代理器列表
static unordered_map<string ,PlayerProxy::Ptr> s_proxyMap; static unordered_map<string ,PlayerProxy::Ptr> s_proxyMap;
static recursive_mutex s_proxyMapMtx; static recursive_mutex s_proxyMapMtx;
static inline string getProxyKey(const string &vhost,const string &app,const string &stream){
return vhost + "/" + app + "/" + stream;
}
//FFmpeg拉流代理器列表
static unordered_map<string ,FFmpegSource::Ptr> s_ffmpegMap; static unordered_map<string ,FFmpegSource::Ptr> s_ffmpegMap;
static recursive_mutex s_ffmpegMapMtx; static recursive_mutex s_ffmpegMapMtx;
#if defined(ENABLE_RTPPROXY)
//rtp服务器列表
static unordered_map<uint16_t, RtpServer::Ptr> s_rtpServerMap;
static recursive_mutex s_rtpServerMapMtx;
#endif
static inline string getProxyKey(const string &vhost,const string &app,const string &stream){
return vhost + "/" + app + "/" + stream;
}
/** /**
* 安装api接口 * 安装api接口
* 所有api都支持GET和POST两种方式 * 所有api都支持GET和POST两种方式
...@@ -729,15 +741,12 @@ void installWebApi() { ...@@ -729,15 +741,12 @@ void installWebApi() {
}); });
#if defined(ENABLE_RTPPROXY) #if defined(ENABLE_RTPPROXY)
api_regist1("/index/api/getSsrcInfo",[](API_ARGS1){ api_regist1("/index/api/getRtpInfo",[](API_ARGS1){
CHECK_SECRET(); CHECK_SECRET();
CHECK_ARGS("ssrc"); CHECK_ARGS("stream_id");
uint32_t ssrc = 0;
stringstream ss(allArgs["ssrc"]);
ss >> std::hex >> ssrc;
auto process = RtpSelector::Instance().getProcess(ssrc,false); auto process = RtpSelector::Instance().getProcess(allArgs["stream_id"], false);
if(!process){ if (!process) {
val["exist"] = false; val["exist"] = false;
return; return;
} }
...@@ -745,6 +754,46 @@ void installWebApi() { ...@@ -745,6 +754,46 @@ void installWebApi() {
val["peer_ip"] = process->get_peer_ip(); val["peer_ip"] = process->get_peer_ip();
val["peer_port"] = process->get_peer_port(); val["peer_port"] = process->get_peer_port();
}); });
api_regist1("/index/api/openRtpServer",[](API_ARGS1){
CHECK_SECRET();
CHECK_ARGS("port", "enable_tcp", "stream_id");
RtpServer::Ptr server = std::make_shared<RtpServer>();
server->start(allArgs["port"], allArgs["stream_id"], allArgs["enable_tcp"].as<bool>());
auto port = server->getPort();
server->setOnDetach([port]() {
//设置rtp超时移除事件
lock_guard<recursive_mutex> lck(s_rtpServerMapMtx);
s_rtpServerMap.erase(port);
});
//保存对象
lock_guard<recursive_mutex> lck(s_rtpServerMapMtx);
s_rtpServerMap.emplace(port, server);
//回复json
val["port"] = port;
});
api_regist1("/index/api/closeRtpServer",[](API_ARGS1){
CHECK_SECRET();
CHECK_ARGS("port");
lock_guard<recursive_mutex> lck(s_rtpServerMapMtx);
val["hit"] = (int)s_rtpServerMap.erase(allArgs["port"].as<uint16_t>());
});
api_regist1("/index/api/listRtpServer",[](API_ARGS1){
CHECK_SECRET();
lock_guard<recursive_mutex> lck(s_rtpServerMapMtx);
for(auto &pr : s_rtpServerMap){
val["data"].append(pr.first);
}
});
#endif//ENABLE_RTPPROXY #endif//ENABLE_RTPPROXY
// 开始录制hls或MP4 // 开始录制hls或MP4
...@@ -1045,4 +1094,10 @@ void unInstallWebApi(){ ...@@ -1045,4 +1094,10 @@ void unInstallWebApi(){
lock_guard<recursive_mutex> lck(s_ffmpegMapMtx); lock_guard<recursive_mutex> lck(s_ffmpegMapMtx);
s_ffmpegMap.clear(); s_ffmpegMap.clear();
} }
{
#if defined(ENABLE_RTPPROXY)
lock_guard<recursive_mutex> lck(s_rtpServerMapMtx);
s_rtpServerMap.clear();
#endif
}
} }
\ No newline at end of file
server/main.cpp
...@@ -20,13 +20,11 @@ ...@@ -20,13 +20,11 @@
#include "Network/TcpServer.h" #include "Network/TcpServer.h"
#include "Poller/EventPoller.h" #include "Poller/EventPoller.h"
#include "Common/config.h" #include "Common/config.h"
#include "Rtsp/UDPServer.h"
#include "Rtsp/RtspSession.h" #include "Rtsp/RtspSession.h"
#include "Rtp/RtpSession.h"
#include "Rtmp/RtmpSession.h" #include "Rtmp/RtmpSession.h"
#include "Shell/ShellSession.h" #include "Shell/ShellSession.h"
#include "Http/WebSocketSession.h" #include "Http/WebSocketSession.h"
#include "Rtp/UdpRecver.h" #include "Rtp/RtpServer.h"
#include "WebApi.h" #include "WebApi.h"
#include "WebHook.h" #include "WebHook.h"
...@@ -283,8 +281,7 @@ int start_main(int argc,char *argv[]) { ...@@ -283,8 +281,7 @@ int start_main(int argc,char *argv[]) {
#if defined(ENABLE_RTPPROXY) #if defined(ENABLE_RTPPROXY)
//GB28181 rtp推流端口，支持UDP/TCP //GB28181 rtp推流端口，支持UDP/TCP
UdpRecver recver; RtpServer::Ptr rtpServer = std::make_shared<RtpServer>();
TcpServer::Ptr tcpRtpServer(new TcpServer());
#endif//defined(ENABLE_RTPPROXY) #endif//defined(ENABLE_RTPPROXY)
try { try {
...@@ -307,12 +304,8 @@ int start_main(int argc,char *argv[]) { ...@@ -307,12 +304,8 @@ int start_main(int argc,char *argv[]) {
if(shellPort) { shellSrv->start<ShellSession>(shellPort); } if(shellPort) { shellSrv->start<ShellSession>(shellPort); }
#if defined(ENABLE_RTPPROXY) #if defined(ENABLE_RTPPROXY)
if(rtpPort){ //创建rtp服务器
//创建rtp udp服务器 if(rtpPort){ rtpServer->start(rtpPort); }
recver.initSock(rtpPort);
//创建rtp tcp服务器
tcpRtpServer->start<RtpSession>(rtpPort);
}
#endif//defined(ENABLE_RTPPROXY) #endif//defined(ENABLE_RTPPROXY)
}catch (std::exception &ex){ }catch (std::exception &ex){
......
src/Common/MediaSource.cpp
...@@ -295,7 +295,7 @@ void MediaSource::regist() { ...@@ -295,7 +295,7 @@ void MediaSource::regist() {
//注册该源，注册后服务器才能找到该源 //注册该源，注册后服务器才能找到该源
{ {
lock_guard<recursive_mutex> lock(g_mtxMediaSrc); lock_guard<recursive_mutex> lock(g_mtxMediaSrc);
g_mapMediaSrc[_strSchema][_strVhost][_strApp][_strId] = shared_from_this(); g_mapMediaSrc[_strSchema][_strVhost][_strApp][_strId] = shared_from_this();
} }
_StrPrinter codec_info; _StrPrinter codec_info;
auto tracks = getTracks(true); auto tracks = getTracks(true);
...@@ -326,6 +326,11 @@ void MediaSource::regist() { ...@@ -326,6 +326,11 @@ void MediaSource::regist() {
InfoL << _strSchema << " " << _strVhost << " " << _strApp << " " << _strId << " " << codec_info; InfoL << _strSchema << " " << _strVhost << " " << _strApp << " " << _strId << " " << codec_info;
NoticeCenter::Instance().emitEvent(Broadcast::kBroadcastMediaChanged, true, *this); NoticeCenter::Instance().emitEvent(Broadcast::kBroadcastMediaChanged, true, *this);
auto listener = _listener.lock();
if (listener) {
listener->onRegist(*this, true);
}
} }
//反注册该源 //反注册该源
...@@ -352,6 +357,11 @@ bool MediaSource::unregist() { ...@@ -352,6 +357,11 @@ bool MediaSource::unregist() {
if(ret){ if(ret){
InfoL << _strSchema << " " << _strVhost << " " << _strApp << " " << _strId; InfoL << _strSchema << " " << _strVhost << " " << _strApp << " " << _strId;
NoticeCenter::Instance().emitEvent(Broadcast::kBroadcastMediaChanged, false, *this); NoticeCenter::Instance().emitEvent(Broadcast::kBroadcastMediaChanged, false, *this);
auto listener = _listener.lock();
if (listener) {
listener->onRegist(*this, false);
}
} }
return ret; return ret;
} }
......
src/Common/MediaSource.h
...@@ -55,6 +55,8 @@ public: ...@@ -55,6 +55,8 @@ public:
virtual bool isRecording(MediaSource &sender, Recorder::type type) { return false; }; virtual bool isRecording(MediaSource &sender, Recorder::type type) { return false; };
// 通知无人观看 // 通知无人观看
virtual void onNoneReader(MediaSource &sender); virtual void onNoneReader(MediaSource &sender);
//流注册或注销事件
virtual void onRegist(MediaSource &sender, bool regist) {};
private: private:
Timer::Ptr _async_close_timer; Timer::Ptr _async_close_timer;
......
src/Common/MultiMediaSourceMuxer.cpp
...@@ -287,6 +287,13 @@ void MultiMediaSourceMuxer::onNoneReader(MediaSource &sender){ ...@@ -287,6 +287,13 @@ void MultiMediaSourceMuxer::onNoneReader(MediaSource &sender){
listener->onNoneReader(sender); listener->onNoneReader(sender);
} }
void MultiMediaSourceMuxer::onRegist(MediaSource &sender, bool regist){
auto listener = _listener.lock();
if (listener) {
listener->onRegist(sender, regist);
}
}
bool MultiMediaSourceMuxer::setupRecord(MediaSource &sender, Recorder::type type, bool start, const string &custom_path) { bool MultiMediaSourceMuxer::setupRecord(MediaSource &sender, Recorder::type type, bool start, const string &custom_path) {
return _muxer->setupRecord(sender,type,start,custom_path); return _muxer->setupRecord(sender,type,start,custom_path);
} }
......
src/Common/MultiMediaSourceMuxer.h
...@@ -138,6 +138,13 @@ public: ...@@ -138,6 +138,13 @@ public:
void onNoneReader(MediaSource &sender) override; void onNoneReader(MediaSource &sender) override;
/** /**
* 媒体注册注销事件
* @param sender 触发者
* @param regist 是否为注册事件
*/
void onRegist(MediaSource &sender, bool regist) override;
/**
* 设置录制状态 * 设置录制状态
* @param type 录制类型 * @param type 录制类型
* @param start 开始或停止 * @param start 开始或停止
......
src/Rtp/RtpProcess.cpp
...@@ -16,32 +16,21 @@ ...@@ -16,32 +16,21 @@
namespace mediakit{ namespace mediakit{
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){ static string printAddress(const struct sockaddr *addr){
return StrPrinter << SockUtil::inet_ntoa(((struct sockaddr_in *) addr)->sin_addr) << ":" << ntohs(((struct sockaddr_in *) addr)->sin_port); return StrPrinter << SockUtil::inet_ntoa(((struct sockaddr_in *) addr)->sin_addr) << ":" << ntohs(((struct sockaddr_in *) addr)->sin_port);
} }
RtpProcess::RtpProcess(uint32_t ssrc) { RtpProcess::RtpProcess(const string &stream_id) {
_ssrc = ssrc;
_track = std::make_shared<SdpTrack>(); _track = std::make_shared<SdpTrack>();
_track->_interleaved = 0; _track->_interleaved = 0;
_track->_samplerate = 90000; _track->_samplerate = 90000;
_track->_type = TrackVideo; _track->_type = TrackVideo;
_track->_ssrc = _ssrc; _track->_ssrc = 0;
_media_info._schema = RTP_APP_NAME; _media_info._schema = RTP_APP_NAME;
_media_info._vhost = DEFAULT_VHOST; _media_info._vhost = DEFAULT_VHOST;
_media_info._app = RTP_APP_NAME; _media_info._app = RTP_APP_NAME;
_media_info._streamid = printSSRC(_ssrc); _media_info._streamid = stream_id;
GET_CONFIG(string,dump_dir,RtpProxy::kDumpDir); GET_CONFIG(string,dump_dir,RtpProxy::kDumpDir);
{ {
...@@ -188,6 +177,16 @@ bool RtpProcess::alive() { ...@@ -188,6 +177,16 @@ bool RtpProcess::alive() {
return false; return false;
} }
void RtpProcess::onDetach(){
if(_on_detach){
_on_detach();
}
}
void RtpProcess::setOnDetach(const function<void()> &cb) {
_on_detach = cb;
}
string RtpProcess::get_peer_ip() { string RtpProcess::get_peer_ip() {
if(_addr){ if(_addr){
return SockUtil::inet_ntoa(((struct sockaddr_in *) _addr)->sin_addr); return SockUtil::inet_ntoa(((struct sockaddr_in *) _addr)->sin_addr);
......
src/Rtp/RtpProcess.h
...@@ -22,15 +22,39 @@ using namespace mediakit; ...@@ -22,15 +22,39 @@ using namespace mediakit;
namespace mediakit{ namespace mediakit{
string printSSRC(uint32_t ui32Ssrc);
class RtpProcess : public RtpReceiver , public RtpDecoder, public SockInfo, public MediaSinkInterface, public std::enable_shared_from_this<RtpProcess>{ class RtpProcess : public RtpReceiver , public RtpDecoder, public SockInfo, public MediaSinkInterface, public std::enable_shared_from_this<RtpProcess>{
public: public:
typedef std::shared_ptr<RtpProcess> Ptr; typedef std::shared_ptr<RtpProcess> Ptr;
RtpProcess(uint32_t ssrc); RtpProcess(const string &stream_id);
~RtpProcess(); ~RtpProcess();
/**
* 输入rtp
* @param sock 本地监听的socket
* @param data rtp数据指针
* @param data_len rtp数据长度
* @param addr 数据源地址
* @param dts_out 解析出最新的dts
* @return 是否解析成功
*/
bool inputRtp(const Socket::Ptr &sock, const char *data,int data_len, const struct sockaddr *addr , uint32_t *dts_out = nullptr); bool inputRtp(const Socket::Ptr &sock, const char *data,int data_len, const struct sockaddr *addr , uint32_t *dts_out = nullptr);
/**
* 是否超时，用于超时移除对象
*/
bool alive(); bool alive();
/**
* 超时时被RtpSelector移除时触发
*/
void onDetach();
/**
* 设置onDetach事件回调
*/
void setOnDetach(const function<void()> &cb);
/// SockInfo override
string get_local_ip() override; string get_local_ip() override;
uint16_t get_local_port() override; uint16_t get_local_port() override;
string get_peer_ip() override; string get_peer_ip() override;
...@@ -54,7 +78,6 @@ private: ...@@ -54,7 +78,6 @@ private:
std::shared_ptr<FILE> _save_file_rtp; std::shared_ptr<FILE> _save_file_rtp;
std::shared_ptr<FILE> _save_file_ps; std::shared_ptr<FILE> _save_file_ps;
std::shared_ptr<FILE> _save_file_video; std::shared_ptr<FILE> _save_file_video;
uint32_t _ssrc;
SdpTrack::Ptr _track; SdpTrack::Ptr _track;
struct sockaddr *_addr = nullptr; struct sockaddr *_addr = nullptr;
uint16_t _sequence = 0; uint16_t _sequence = 0;
...@@ -66,6 +89,7 @@ private: ...@@ -66,6 +89,7 @@ private:
MediaInfo _media_info; MediaInfo _media_info;
uint64_t _total_bytes = 0; uint64_t _total_bytes = 0;
Socket::Ptr _sock; Socket::Ptr _sock;
function<void()> _on_detach;
}; };
}//namespace mediakit }//namespace mediakit
......
src/Rtp/RtpSelector.cpp
...@@ -15,37 +15,41 @@ namespace mediakit{ ...@@ -15,37 +15,41 @@ namespace mediakit{
INSTANCE_IMP(RtpSelector); INSTANCE_IMP(RtpSelector);
bool RtpSelector::inputRtp(const Socket::Ptr &sock, const char *data, int data_len,const struct sockaddr *addr,uint32_t *dts_out) { bool RtpSelector::inputRtp(const Socket::Ptr &sock, const char *data, int data_len,
const struct sockaddr *addr,uint32_t *dts_out) {
//使用ssrc为流id
uint32_t ssrc = 0; uint32_t ssrc = 0;
if(!getSSRC(data,data_len,ssrc)){ if (!getSSRC(data, data_len, ssrc)) {
WarnL << "get ssrc from rtp failed:" << data_len; WarnL << "get ssrc from rtp failed:" << data_len;
return false; return false;
} }
auto process = getProcess(ssrc, true);
if(process){ //假定指定了流id，那么通过流id来区分是否为一路流(哪怕可能同时收到多路流)
return process->inputRtp(sock, data,data_len, addr,dts_out); auto process = getProcess(printSSRC(ssrc), true);
if (process) {
return process->inputRtp(sock, data, data_len, addr, dts_out);
} }
return false; return false;
} }
bool RtpSelector::getSSRC(const char *data,int data_len, uint32_t &ssrc){ bool RtpSelector::getSSRC(const char *data,int data_len, uint32_t &ssrc){
if(data_len < 12){ if (data_len < 12) {
return false; return false;
} }
uint32_t *ssrc_ptr = (uint32_t *)(data + 8); uint32_t *ssrc_ptr = (uint32_t *) (data + 8);
ssrc = ntohl(*ssrc_ptr); ssrc = ntohl(*ssrc_ptr);
return true; return true;
} }
RtpProcess::Ptr RtpSelector::getProcess(uint32_t ssrc,bool makeNew) { RtpProcess::Ptr RtpSelector::getProcess(const string &stream_id,bool makeNew) {
lock_guard<decltype(_mtx_map)> lck(_mtx_map); lock_guard<decltype(_mtx_map)> lck(_mtx_map);
auto it = _map_rtp_process.find(ssrc); auto it = _map_rtp_process.find(stream_id);
if(it == _map_rtp_process.end() && !makeNew){ if (it == _map_rtp_process.end() && !makeNew) {
return nullptr; return nullptr;
} }
RtpProcessHelper::Ptr &ref = _map_rtp_process[ssrc]; RtpProcessHelper::Ptr &ref = _map_rtp_process[stream_id];
if(!ref){ if (!ref) {
ref = std::make_shared<RtpProcessHelper>(ssrc,shared_from_this()); ref = std::make_shared<RtpProcessHelper>(stream_id, shared_from_this());
ref->attachEvent(); ref->attachEvent();
createTimer(); createTimer();
} }
...@@ -67,17 +71,15 @@ void RtpSelector::createTimer() { ...@@ -67,17 +71,15 @@ void RtpSelector::createTimer() {
} }
} }
void RtpSelector::delProcess(uint32_t ssrc,const RtpProcess *ptr) { void RtpSelector::delProcess(const string &stream_id,const RtpProcess *ptr) {
lock_guard<decltype(_mtx_map)> lck(_mtx_map); lock_guard<decltype(_mtx_map)> lck(_mtx_map);
auto it = _map_rtp_process.find(ssrc); auto it = _map_rtp_process.find(stream_id);
if(it == _map_rtp_process.end()){ if (it == _map_rtp_process.end()) {
return; return;
} }
if (it->second->getProcess().get() != ptr) {
if(it->second->getProcess().get() != ptr){
return; return;
} }
_map_rtp_process.erase(it); _map_rtp_process.erase(it);
} }
...@@ -88,8 +90,10 @@ void RtpSelector::onManager() { ...@@ -88,8 +90,10 @@ void RtpSelector::onManager() {
++it; ++it;
continue; continue;
} }
WarnL << "RtpProcess timeout:" << printSSRC(it->first); WarnL << "RtpProcess timeout:" << it->first;
auto process = it->second->getProcess();
it = _map_rtp_process.erase(it); it = _map_rtp_process.erase(it);
process->onDetach();
} }
} }
...@@ -99,10 +103,10 @@ RtpSelector::RtpSelector() { ...@@ -99,10 +103,10 @@ RtpSelector::RtpSelector() {
RtpSelector::~RtpSelector() { RtpSelector::~RtpSelector() {
} }
RtpProcessHelper::RtpProcessHelper(uint32_t ssrc, const weak_ptr<RtpSelector> &parent) { RtpProcessHelper::RtpProcessHelper(const string &stream_id, const weak_ptr<RtpSelector> &parent) {
_ssrc = ssrc; _stream_id = stream_id;
_parent = parent; _parent = parent;
_process = std::make_shared<RtpProcess>(_ssrc); _process = std::make_shared<RtpProcess>(stream_id);
} }
RtpProcessHelper::~RtpProcessHelper() { RtpProcessHelper::~RtpProcessHelper() {
...@@ -114,14 +118,14 @@ void RtpProcessHelper::attachEvent() { ...@@ -114,14 +118,14 @@ void RtpProcessHelper::attachEvent() {
bool RtpProcessHelper::close(MediaSource &sender, bool force) { bool RtpProcessHelper::close(MediaSource &sender, bool force) {
//此回调在其他线程触发 //此回调在其他线程触发
if(!_process || (!force && _process->totalReaderCount())){ if (!_process || (!force && _process->totalReaderCount())) {
return false; return false;
} }
auto parent = _parent.lock(); auto parent = _parent.lock();
if(!parent){ if (!parent) {
return false; return false;
} }
parent->delProcess(_ssrc,_process.get()); parent->delProcess(_stream_id, _process.get());
WarnL << "close media:" << sender.getSchema() << "/" << sender.getVhost() << "/" << sender.getApp() << "/" << sender.getId() << " " << force; WarnL << "close media:" << sender.getSchema() << "/" << sender.getVhost() << "/" << sender.getApp() << "/" << sender.getId() << " " << force;
return true; return true;
} }
......
src/Rtp/RtpSelector.h
...@@ -24,19 +24,21 @@ class RtpSelector; ...@@ -24,19 +24,21 @@ class RtpSelector;
class RtpProcessHelper : public MediaSourceEvent , public std::enable_shared_from_this<RtpProcessHelper> { class RtpProcessHelper : public MediaSourceEvent , public std::enable_shared_from_this<RtpProcessHelper> {
public: public:
typedef std::shared_ptr<RtpProcessHelper> Ptr; typedef std::shared_ptr<RtpProcessHelper> Ptr;
RtpProcessHelper(uint32_t ssrc,const weak_ptr<RtpSelector > &parent); RtpProcessHelper(const string &stream_id, const weak_ptr<RtpSelector > &parent);
~RtpProcessHelper(); ~RtpProcessHelper();
void attachEvent(); void attachEvent();
RtpProcess::Ptr & getProcess(); RtpProcess::Ptr & getProcess();
protected: protected:
// 通知其停止推流 // 通知其停止推流
bool close(MediaSource &sender,bool force) override; bool close(MediaSource &sender,bool force) override;
// 观看总人数 // 观看总人数
int totalReaderCount(MediaSource &sender) override; int totalReaderCount(MediaSource &sender) override;
private: private:
weak_ptr<RtpSelector > _parent; weak_ptr<RtpSelector > _parent;
RtpProcess::Ptr _process; RtpProcess::Ptr _process;
uint32_t _ssrc = 0; string _stream_id;
}; };
class RtpSelector : public std::enable_shared_from_this<RtpSelector>{ class RtpSelector : public std::enable_shared_from_this<RtpSelector>{
...@@ -44,16 +46,42 @@ public: ...@@ -44,16 +46,42 @@ public:
RtpSelector(); RtpSelector();
~RtpSelector(); ~RtpSelector();
static RtpSelector &Instance();
bool inputRtp(const Socket::Ptr &sock, const char *data,int data_len,const struct sockaddr *addr ,uint32_t *dts_out = nullptr );
static bool getSSRC(const char *data,int data_len, uint32_t &ssrc); static bool getSSRC(const char *data,int data_len, uint32_t &ssrc);
RtpProcess::Ptr getProcess(uint32_t ssrc,bool makeNew); static RtpSelector &Instance();
void delProcess(uint32_t ssrc,const RtpProcess *ptr);
/**
* 输入多个rtp流，根据ssrc分流
* @param sock 本地socket
* @param data 收到的数据
* @param data_len 收到的数据长度
* @param addr rtp流源地址
* @param dts_out 解析出最新的dts
* @return 是否成功
*/
bool inputRtp(const Socket::Ptr &sock, const char *data, int data_len,
const struct sockaddr *addr, uint32_t *dts_out = nullptr);
/**
* 获取一个rtp处理器
* @param stream_id 流id
* @param makeNew 不存在时是否新建
* @return rtp处理器
*/
RtpProcess::Ptr getProcess(const string &stream_id, bool makeNew);
/**
* 删除rtp处理器
* @param stream_id 流id
* @param ptr rtp处理器指针
*/
void delProcess(const string &stream_id, const RtpProcess *ptr);
private: private:
void onManager(); void onManager();
void createTimer(); void createTimer();
private: private:
unordered_map<uint32_t,RtpProcessHelper::Ptr> _map_rtp_process; unordered_map<string,RtpProcessHelper::Ptr> _map_rtp_process;
recursive_mutex _mtx_map; recursive_mutex _mtx_map;
Timer::Ptr _timer; Timer::Ptr _timer;
}; };
......
src/Rtp/RtpServer.cpp 0 → 100644
/*
* Copyright (c) 2016 The ZLMediaKit project authors. All Rights Reserved.
*
* This file is part of ZLMediaKit(https://github.com/xiongziliang/ZLMediaKit).
*
* Use of this source code is governed by MIT license that can be found in the
* LICENSE file in the root of the source tree. All contributing project authors
* may be found in the AUTHORS file in the root of the source tree.
*/
#if defined(ENABLE_RTPPROXY)
#include "RtpServer.h"
#include "RtpSelector.h"
namespace mediakit{
RtpServer::RtpServer() {
}
RtpServer::~RtpServer() {
if(_on_clearup){
_on_clearup();
}
}
void RtpServer::start(uint16_t local_port, const string &stream_id, bool enable_tcp, const char *local_ip) {
//创建udp服务器
Socket::Ptr udp_server = std::make_shared<Socket>(nullptr, false);
if (!udp_server->bindUdpSock(local_port, local_ip)) {
throw std::runtime_error(StrPrinter << "bindUdpSock on " << local_ip << ":" << local_port << " failed:" << get_uv_errmsg(true));
}
//设置udp socket读缓存
SockUtil::setRecvBuf(udp_server->rawFD(), 4 * 1024 * 1024);
TcpServer::Ptr tcp_server;
if (enable_tcp) {
try {
//创建tcp服务器
tcp_server = std::make_shared<TcpServer>(udp_server->getPoller());
(*tcp_server)[RtpSession::kStreamID] = stream_id;
tcp_server->start<RtpSession>(udp_server->get_local_port(), local_ip);
} catch (...) {
throw;
}
}
RtpProcess::Ptr process;
if (!stream_id.empty()) {
//指定了流id，那么一个端口一个流(不管是否包含多个ssrc的多个流，绑定rtp源后，会筛选掉ip端口不匹配的流)
process = RtpSelector::Instance().getProcess(stream_id, true);
udp_server->setOnRead([udp_server, process](const Buffer::Ptr &buf, struct sockaddr *addr, int) {
process->inputRtp(udp_server, buf->data(), buf->size(), addr);
});
} else {
//未指定流id，一个端口多个流，通过ssrc来分流
auto &ref = RtpSelector::Instance();
udp_server->setOnRead([&ref, udp_server](const Buffer::Ptr &buf, struct sockaddr *addr, int) {
ref.inputRtp(udp_server, buf->data(), buf->size(), addr);
});
}
_on_clearup = [udp_server, process, stream_id]() {
//去除循环引用
udp_server->setOnRead(nullptr);
if (process) {
//删除rtp处理器
RtpSelector::Instance().delProcess(stream_id, process.get());
}
};
_tcp_server = tcp_server;
_udp_server = udp_server;
_rtp_process = process;
}
void RtpServer::setOnDetach(const function<void()> &cb){
if(_rtp_process){
_rtp_process->setOnDetach(cb);
}
}
EventPoller::Ptr RtpServer::getPoller() {
return _udp_server->getPoller();
}
uint16_t RtpServer::getPort() {
return _udp_server ? _udp_server->get_local_port() : 0;
}
}//namespace mediakit
#endif//defined(ENABLE_RTPPROXY)
\ No newline at end of file
src/Rtp/UdpRecver.h → src/Rtp/RtpServer.h
...@@ -8,33 +8,62 @@ ...@@ -8,33 +8,62 @@
* may be found in the AUTHORS file in the root of the source tree. * may be found in the AUTHORS file in the root of the source tree.
*/ */
#ifndef ZLMEDIAKIT_UDPRECVER_H #ifndef ZLMEDIAKIT_RTPSERVER_H
#define ZLMEDIAKIT_UDPRECVER_H #define ZLMEDIAKIT_RTPSERVER_H
#if defined(ENABLE_RTPPROXY) #if defined(ENABLE_RTPPROXY)
#include <memory> #include <memory>
#include "Network/Socket.h" #include "Network/Socket.h"
#include "Network/TcpServer.h"
#include "RtpSession.h"
using namespace std; using namespace std;
using namespace toolkit; using namespace toolkit;
namespace mediakit{ namespace mediakit{
/** /**
* 组播接收器 * RTP服务器，支持UDP/TCP
*/ */
class UdpRecver { class RtpServer {
public: public:
typedef std::shared_ptr<UdpRecver> Ptr; typedef std::shared_ptr<RtpServer> Ptr;
typedef function<void(const Buffer::Ptr &buf)> onRecv; typedef function<void(const Buffer::Ptr &buf)> onRecv;
UdpRecver(); RtpServer();
virtual ~UdpRecver(); ~RtpServer();
bool initSock(uint16_t local_port,const char *local_ip = "0.0.0.0");
/**
* 开启服务器，可能抛异常
* @param local_port 本地端口，0时为随机端口
* @param stream_id 流id，置空则使用ssrc
* @param enable_tcp 是否启用tcp服务器
* @param local_ip 绑定的本地网卡ip
*/
void start(uint16_t local_port, const string &stream_id = "", bool enable_tcp = true, const char *local_ip = "0.0.0.0");
/**
* 获取绑定的本地端口
*/
uint16_t getPort();
/**
* 获取绑定的线程
*/
EventPoller::Ptr getPoller(); EventPoller::Ptr getPoller();
/**
* 设置RtpProcess onDetach事件回调
*/
void setOnDetach(const function<void()> &cb);
protected: protected:
Socket::Ptr _sock; Socket::Ptr _udp_server;
TcpServer::Ptr _tcp_server;
RtpProcess::Ptr _rtp_process;
function<void()> _on_clearup;
}; };
}//namespace mediakit }//namespace mediakit
#endif//defined(ENABLE_RTPPROXY) #endif//defined(ENABLE_RTPPROXY)
#endif //ZLMEDIAKIT_UDPRECVER_H #endif //ZLMEDIAKIT_RTPSERVER_H
src/Rtp/RtpSession.cpp
...@@ -11,8 +11,15 @@ ...@@ -11,8 +11,15 @@
#if defined(ENABLE_RTPPROXY) #if defined(ENABLE_RTPPROXY)
#include "RtpSession.h" #include "RtpSession.h"
#include "RtpSelector.h" #include "RtpSelector.h"
#include "Network/TcpServer.h"
namespace mediakit{ namespace mediakit{
const string RtpSession::kStreamID = "stream_id";
void RtpSession::attachServer(const TcpServer &server) {
_stream_id = const_cast<TcpServer &>(server)[kStreamID];
}
RtpSession::RtpSession(const Socket::Ptr &sock) : TcpSession(sock) { RtpSession::RtpSession(const Socket::Ptr &sock) : TcpSession(sock) {
DebugP(this); DebugP(this);
socklen_t addr_len = sizeof(addr); socklen_t addr_len = sizeof(addr);
...@@ -21,7 +28,7 @@ RtpSession::RtpSession(const Socket::Ptr &sock) : TcpSession(sock) { ...@@ -21,7 +28,7 @@ RtpSession::RtpSession(const Socket::Ptr &sock) : TcpSession(sock) {
RtpSession::~RtpSession() { RtpSession::~RtpSession() {
DebugP(this); DebugP(this);
if(_process){ if(_process){
RtpSelector::Instance().delProcess(_ssrc,_process.get()); RtpSelector::Instance().delProcess(_stream_id,_process.get());
} }
} }
...@@ -36,7 +43,7 @@ void RtpSession::onRecv(const Buffer::Ptr &data) { ...@@ -36,7 +43,7 @@ void RtpSession::onRecv(const Buffer::Ptr &data) {
} }
void RtpSession::onError(const SockException &err) { void RtpSession::onError(const SockException &err) {
WarnL << _ssrc << " " << err.what(); WarnL << _stream_id << " " << err.what();
} }
void RtpSession::onManager() { void RtpSession::onManager() {
...@@ -51,13 +58,19 @@ void RtpSession::onManager() { ...@@ -51,13 +58,19 @@ void RtpSession::onManager() {
void RtpSession::onRtpPacket(const char *data, uint64_t len) { void RtpSession::onRtpPacket(const char *data, uint64_t len) {
if (!_process) { if (!_process) {
if (!RtpSelector::getSSRC(data + 2, len - 2, _ssrc)) { uint32_t ssrc;
if (!RtpSelector::getSSRC(data + 2, len - 2, ssrc)) {
return; return;
} }
_process = RtpSelector::Instance().getProcess(_ssrc, true); if (_stream_id.empty()) {
//未指定流id就使用ssrc为流id
_stream_id = printSSRC(ssrc);
}
//tcp情况下，一个tcp链接只可能是一路流，不需要通过多个ssrc来区分，所以不需要频繁getProcess
_process = RtpSelector::Instance().getProcess(_stream_id, true);
_process->setListener(dynamic_pointer_cast<RtpSession>(shared_from_this())); _process->setListener(dynamic_pointer_cast<RtpSession>(shared_from_this()));
} }
_process->inputRtp(_sock,data + 2, len - 2, &addr); _process->inputRtp(_sock, data + 2, len - 2, &addr);
_ticker.resetTime(); _ticker.resetTime();
} }
......
src/Rtp/RtpSession.h
...@@ -22,22 +22,26 @@ namespace mediakit{ ...@@ -22,22 +22,26 @@ namespace mediakit{
class RtpSession : public TcpSession , public RtpSplitter , public MediaSourceEvent{ class RtpSession : public TcpSession , public RtpSplitter , public MediaSourceEvent{
public: public:
static const string kStreamID;
RtpSession(const Socket::Ptr &sock); RtpSession(const Socket::Ptr &sock);
~RtpSession() override; ~RtpSession() override;
void onRecv(const Buffer::Ptr &) override; void onRecv(const Buffer::Ptr &) override;
void onError(const SockException &err) override; void onError(const SockException &err) override;
void onManager() override; void onManager() override;
void attachServer(const TcpServer &server) override;
protected: protected:
// 通知其停止推流 // 通知其停止推流
bool close(MediaSource &sender,bool force) override; bool close(MediaSource &sender,bool force) override;
// 观看总人数 // 观看总人数
int totalReaderCount(MediaSource &sender) override; int totalReaderCount(MediaSource &sender) override;
void onRtpPacket(const char *data,uint64_t len) override; void onRtpPacket(const char *data,uint64_t len) override;
private: private:
uint32_t _ssrc = 0;
RtpProcess::Ptr _process; RtpProcess::Ptr _process;
Ticker _ticker; Ticker _ticker;
struct sockaddr addr; struct sockaddr addr;
string _stream_id;
}; };
}//namespace mediakit }//namespace mediakit
......
src/Rtp/UdpRecver.cpp deleted 100644 → 0
/*
* Copyright (c) 2016 The ZLMediaKit project authors. All Rights Reserved.
*
* This file is part of ZLMediaKit(https://github.com/xiongziliang/ZLMediaKit).
*
* Use of this source code is governed by MIT license that can be found in the
* LICENSE file in the root of the source tree. All contributing project authors
* may be found in the AUTHORS file in the root of the source tree.
*/
#if defined(ENABLE_RTPPROXY)
#include "UdpRecver.h"
#include "RtpSelector.h"
namespace mediakit{
UdpRecver::UdpRecver() {
}
UdpRecver::~UdpRecver() {
if(_sock){
_sock->setOnRead(nullptr);
}
}
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();
auto sock = _sock;
_sock->setOnRead([&ref, sock](const Buffer::Ptr &buf, struct sockaddr *addr, int ){
ref.inputRtp(sock, buf->data(),buf->size(),addr);
});
return _sock->bindUdpSock(local_port,local_ip);
}
EventPoller::Ptr UdpRecver::getPoller() {
return _sock->getPoller();
}
}//namespace mediakit
#endif//defined(ENABLE_RTPPROXY)
\ No newline at end of file
src/Rtsp/Rtsp.cpp
...@@ -404,4 +404,14 @@ std::pair<Socket::Ptr, Socket::Ptr> makeSockPair(const EventPoller::Ptr &poller, ...@@ -404,4 +404,14 @@ std::pair<Socket::Ptr, Socket::Ptr> makeSockPair(const EventPoller::Ptr &poller,
} }
} }
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;
}
}//namespace mediakit }//namespace mediakit
\ No newline at end of file
src/Rtsp/Rtsp.h
...@@ -272,6 +272,7 @@ private: ...@@ -272,6 +272,7 @@ private:
}; };
std::pair<Socket::Ptr, Socket::Ptr> makeSockPair(const EventPoller::Ptr &poller, const string &local_ip); std::pair<Socket::Ptr, Socket::Ptr> makeSockPair(const EventPoller::Ptr &poller, const string &local_ip);
string printSSRC(uint32_t ui32Ssrc);
} //namespace mediakit } //namespace mediakit
#endif //RTSP_RTSP_H_ #endif //RTSP_RTSP_H_
src/Rtsp/RtspSession.cpp
...@@ -218,7 +218,7 @@ void RtspSession::handleReq_Options(const Parser &parser) { ...@@ -218,7 +218,7 @@ void RtspSession::handleReq_Options(const Parser &parser) {
} }
void RtspSession::handleReq_ANNOUNCE(const Parser &parser) { void RtspSession::handleReq_ANNOUNCE(const Parser &parser) {
auto src = dynamic_pointer_cast<RtmpMediaSource>(MediaSource::find(RTSP_SCHEMA, auto src = dynamic_pointer_cast<RtspMediaSource>(MediaSource::find(RTSP_SCHEMA,
_mediaInfo._vhost, _mediaInfo._vhost,
_mediaInfo._app, _mediaInfo._app,
_mediaInfo._streamid)); _mediaInfo._streamid));
...@@ -1068,15 +1068,6 @@ bool RtspSession::sendRtspResponse(const string &res_code, ...@@ -1068,15 +1068,6 @@ bool RtspSession::sendRtspResponse(const string &res_code,
return sendRtspResponse(res_code,header_map,sdp,protocol); return sendRtspResponse(res_code,header_map,sdp,protocol);
} }
inline string RtspSession::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;
}
inline int RtspSession::getTrackIndexByTrackType(TrackType type) { inline int RtspSession::getTrackIndexByTrackType(TrackType type) {
for (unsigned int i = 0; i < _aTrackInfo.size(); i++) { for (unsigned int i = 0; i < _aTrackInfo.size(); i++) {
if (type == _aTrackInfo[i]->_type) { if (type == _aTrackInfo[i]->_type) {
......
src/Rtsp/RtspSession.h
...@@ -136,8 +136,6 @@ private: ...@@ -136,8 +136,6 @@ private:
void inline send_SessionNotFound(); void inline send_SessionNotFound();
//一般rtsp服务器打开端口失败时触发 //一般rtsp服务器打开端口失败时触发
void inline send_NotAcceptable(); void inline send_NotAcceptable();
//ssrc转字符串
inline string printSSRC(uint32_t ui32Ssrc);
//获取track下标 //获取track下标
inline int getTrackIndexByTrackType(TrackType type); inline int getTrackIndexByTrackType(TrackType type);
......
tests/CMakeLists.txt
...@@ -8,20 +8,38 @@ if (SDL2_FOUND) ...@@ -8,20 +8,38 @@ if (SDL2_FOUND)
message(STATUS "found library:${SDL2_LIBRARY}") message(STATUS "found library:${SDL2_LIBRARY}")
endif (SDL2_FOUND) endif (SDL2_FOUND)
find_package(PkgConfig QUIET)
#查找ffmpeg/libutil是否安装 #查找ffmpeg/libutil是否安装
find_package(AVUTIL QUIET) if(PKG_CONFIG_FOUND)
if(AVUTIL_FOUND) pkg_check_modules(AVUTIL QUIET IMPORTED_TARGET libavutil)
include_directories(${AVUTIL_INCLUDE_DIR}) if(AVUTIL_FOUND)
list(APPEND LINK_LIB_LIST ${AVUTIL_LIBRARIES}) list(APPEND LINK_LIB_LIST PkgConfig::AVUTIL)
message(STATUS "found library:${AVUTIL_LIBRARIES}") message(STATUS "found library:${AVUTIL_LIBRARY}")
endif()
else()
find_package(AVUTIL QUIET)
if(AVUTIL_FOUND)
include_directories(${AVUTIL_INCLUDE_DIR})
list(APPEND LINK_LIB_LIST ${AVUTIL_LIBRARIES})
message(STATUS "found library:${AVUTIL_LIBRARIES}")
endif()
endif() endif()
#查找ffmpeg/libavcodec是否安装 #查找ffmpeg/libavcodec是否安装
find_package(AVCODEC QUIET) if(PKG_CONFIG_FOUND)
if(AVCODEC_FOUND) pkg_check_modules(AVCODEC QUIET IMPORTED_TARGET libavcodec)
include_directories(${AVCODEC_INCLUDE_DIR}) if(AVCODEC_FOUND)
list(APPEND LINK_LIB_LIST ${AVCODEC_LIBRARIES}) list(APPEND LINK_LIB_LIST PkgConfig::AVCODEC)
message(STATUS "found library:${AVCODEC_LIBRARIES}") message(STATUS "found library:${AVCODEC_LIBRARY}")
endif()
else()
find_package(AVCODEC QUIET)
if(AVCODEC_FOUND)
include_directories(${AVCODEC_INCLUDE_DIR})
list(APPEND LINK_LIB_LIST ${AVCODEC_LIBRARIES})
message(STATUS "found library:${AVCODEC_LIBRARIES}")
endif()
endif() endif()
aux_source_directory(. TEST_SRC_LIST) aux_source_directory(. TEST_SRC_LIST)
......
tests/test_rtp.cpp
...@@ -55,7 +55,7 @@ static bool loadFile(const char *path){ ...@@ -55,7 +55,7 @@ static bool loadFile(const char *path){
} }
uint32_t timeStamp; uint32_t timeStamp;
RtpSelector::Instance().inputRtp(nullptr,rtp,len, &addr,&timeStamp); RtpSelector::Instance().inputRtp(nullptr, rtp, len, &addr, &timeStamp);
if(timeStamp_last){ if(timeStamp_last){
auto diff = timeStamp - timeStamp_last; auto diff = timeStamp - timeStamp_last;
if(diff > 0 && diff < 500){ if(diff > 0 && diff < 500){
......
Markdown 格式
0% 或
您添加了 0 人 到此讨论。请谨慎行事。
请先完成此评论的编辑！
请 注册 或者 后发表评论