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

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

Browse Source 
# Conflicts:
#	conf/config.ini
#	src/Codec/Transcode.cpp
#	src/Common/MediaSource.h
#	src/Common/MultiMediaSourceMuxer.cpp
#	src/Common/MultiMediaSourceMuxer.h
#	src/Common/macros.h
#	webrtc/WebRtcPusher.cpp
#	webrtc/WebRtcTransport.cpp
#	webrtc/WebRtcTransport.h
This commit is contained in:
cqm
2026-04-03 09:35:50 +08:00
parent 21cb962c39 c3c0fb4448
commit 03526f141e
283 changed files with 42056 additions and 13083 deletions
Download Patch File Download Diff File Expand all files Collapse all files

35
server/CMakeLists.txt
View File

@@ -1,6 +1,6 @@
# MIT License
#
# Copyright (c) 2016-2022 The ZLMediaKit project authors. All Rights Reserved.
# Copyright (c) 2016-present The ZLMediaKit project authors. All Rights Reserved.
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
@@ -50,10 +50,43 @@ target_compile_definitions(MediaServer
target_compile_options(MediaServer
PRIVATE ${COMPILE_OPTIONS_DEFAULT})
if(MINGW)
update_cached_list(MK_LINK_LIBRARIES dbghelp)
endif()
if(CMAKE_SYSTEM_NAME MATCHES "Linux")
target_link_libraries(MediaServer -Wl,--start-group ${MK_LINK_LIBRARIES} -Wl,--end-group)
else()
target_link_libraries(MediaServer ${MK_LINK_LIBRARIES})
endif()
if(MSVC)
set(RESOURCE_FILE "${CMAKE_SOURCE_DIR}/resource.rc")
set_source_files_properties(${RESOURCE_FILE} PROPERTIES LANGUAGE RC)
target_sources(MediaServer PRIVATE ${RESOURCE_FILE})
else()
# Android, IOS, macOS ...
# CLion, GCC ...
endif()
install(TARGETS MediaServer DESTINATION ${INSTALL_PATH_RUNTIME})
#relase 类型时额外输出debug调试信息
string(TOLOWER ${CMAKE_BUILD_TYPE} CMAKE_BUILD_TYPE_LOWER)
if(UNIX AND ENABLE_OBJCOPY)
if("${CMAKE_BUILD_TYPE_LOWER}" STREQUAL "release")
find_program(OBJCOPY_FOUND objcopy)
if (OBJCOPY_FOUND)
add_custom_command(TARGET MediaServer
POST_BUILD
COMMAND objcopy --only-keep-debug ${EXECUTABLE_OUTPUT_PATH}/MediaServer ${EXECUTABLE_OUTPUT_PATH}/MediaServer.debug
COMMAND objcopy --strip-all ${EXECUTABLE_OUTPUT_PATH}/MediaServer
COMMAND objcopy --add-gnu-debuglink=${EXECUTABLE_OUTPUT_PATH}/MediaServer.debug ${EXECUTABLE_OUTPUT_PATH}/MediaServer
)
install(FILES ${EXECUTABLE_OUTPUT_PATH}/MediaServer.debug DESTINATION ${INSTALL_PATH_RUNTIME})
else()
message(STATUS "not found objcopy, generate MediaServer.debug skip")
endif()
endif()
endif()

221
server/FFmpegSource.cpp
View File

@@ -84,86 +84,92 @@ void FFmpegSource::play(const string &ffmpeg_cmd_key, const string &src_url, con
try {
_media_info.parse(dst_url);
} catch (std::exception &ex) {
cb(SockException(Err_other, ex.what()));
return;
}
auto ffmpeg_cmd = ffmpeg_cmd_default;
if (!ffmpeg_cmd_key.empty()) {
auto cmd_it = mINI::Instance().find(ffmpeg_cmd_key);
if (cmd_it != mINI::Instance().end()) {
ffmpeg_cmd = cmd_it->second;
auto ffmpeg_cmd = ffmpeg_cmd_default;
if (!ffmpeg_cmd_key.empty()) {
auto cmd_it = mINI::Instance().find(ffmpeg_cmd_key);
if (cmd_it != mINI::Instance().end()) {
ffmpeg_cmd = cmd_it->second;
} else {
WarnL << "配置文件中,ffmpeg命令模板(" << ffmpeg_cmd_key << ")不存在,已采用默认模板(" << ffmpeg_cmd_default << ")";
}
}
if (!toolkit::start_with(ffmpeg_cmd, "%s")) {
throw std::invalid_argument("ffmpeg cmd template must start with '%s'");
}
char cmd[2048] = { 0 };
snprintf(cmd, sizeof(cmd), ffmpeg_cmd.data(), File::absolutePath("", ffmpeg_bin).data(), src_url.data(), dst_url.data());
auto log_file = ffmpeg_log.empty() ? "" : File::absolutePath("", ffmpeg_log);
_process.run(cmd, log_file);
_cmd = cmd;
InfoL << cmd;
if (is_local_ip(_media_info.host)) {
// 推流给自己的，通过判断流是否注册上来判断是否正常 [AUTO-TRANSLATED:423f2be6]
// Push stream to yourself, judge whether the stream is registered to determine whether it is normal
if (_media_info.schema != RTSP_SCHEMA && _media_info.schema != RTMP_SCHEMA && _media_info.schema != "srt") {
cb(SockException(Err_other, "本服务只支持rtmp/rtsp/srt推流"));
return;
}
weak_ptr<FFmpegSource> weakSelf = shared_from_this();
findAsync(timeout_ms, [cb, weakSelf, timeout_ms](const MediaSource::Ptr &src) {
auto strongSelf = weakSelf.lock();
if (!strongSelf) {
// 自己已经销毁 [AUTO-TRANSLATED:3d45c3b0]
// Self has been destroyed
return;
}
if (src) {
// 推流给自己成功 [AUTO-TRANSLATED:65dba71b]
// Push stream to yourself successfully
cb(SockException());
strongSelf->onGetMediaSource(src);
strongSelf->startTimer(timeout_ms);
return;
}
// 推流失败 [AUTO-TRANSLATED:4d8d226a]
// Push stream failed
if (!strongSelf->_process.wait(false)) {
// ffmpeg进程已经退出 [AUTO-TRANSLATED:04193893]
// ffmpeg process has exited
cb(SockException(Err_other, StrPrinter << "ffmpeg已经退出,exit code = " << strongSelf->_process.exit_code()));
return;
}
// ffmpeg进程还在线，但是等待推流超时 [AUTO-TRANSLATED:9f71f17b]
// ffmpeg process is still online, but waiting for the stream to timeout
cb(SockException(Err_other, "等待超时"));
});
} else {
WarnL << "配置文件中,ffmpeg命令模板(" << ffmpeg_cmd_key << ")不存在,已采用默认模板(" << ffmpeg_cmd_default << ")";
// 推流给其他服务器的，通过判断FFmpeg进程是否在线判断是否成功 [AUTO-TRANSLATED:9b963da5]
// Push stream to other servers, judge whether it is successful by judging whether the FFmpeg process is online
weak_ptr<FFmpegSource> weakSelf = shared_from_this();
_timer = std::make_shared<Timer>(
timeout_ms / 1000.0f,
[weakSelf, cb, timeout_ms]() {
auto strongSelf = weakSelf.lock();
if (!strongSelf) {
// 自身已经销毁 [AUTO-TRANSLATED:5f954f8a]
// Self has been destroyed
return false;
}
// FFmpeg还在线，那么我们认为推流成功 [AUTO-TRANSLATED:4330df49]
// FFmpeg is still online, so we think the push stream is successful
if (strongSelf->_process.wait(false)) {
cb(SockException());
strongSelf->startTimer(timeout_ms);
return false;
}
// ffmpeg进程已经退出 [AUTO-TRANSLATED:04193893]
// ffmpeg process has exited
cb(SockException(Err_other, StrPrinter << "ffmpeg已经退出,exit code = " << strongSelf->_process.exit_code()));
return false;
},
_poller);
}
}
char cmd[2048] = { 0 };
snprintf(cmd, sizeof(cmd), ffmpeg_cmd.data(), File::absolutePath("", ffmpeg_bin).data(), src_url.data(), dst_url.data());
auto log_file = ffmpeg_log.empty() ? "" : File::absolutePath("", ffmpeg_log);
_process.run(cmd, log_file);
_cmd = cmd;
InfoL << cmd;
if (is_local_ip(_media_info.host)) {
// 推流给自己的，通过判断流是否注册上来判断是否正常 [AUTO-TRANSLATED:423f2be6]
// Push stream to yourself, judge whether the stream is registered to determine whether it is normal
if (_media_info.schema != RTSP_SCHEMA && _media_info.schema != RTMP_SCHEMA) {
cb(SockException(Err_other, "本服务只支持rtmp/rtsp推流"));
return;
}
weak_ptr<FFmpegSource> weakSelf = shared_from_this();
findAsync(timeout_ms, [cb, weakSelf, timeout_ms](const MediaSource::Ptr &src) {
auto strongSelf = weakSelf.lock();
if (!strongSelf) {
// 自己已经销毁 [AUTO-TRANSLATED:3d45c3b0]
// Self has been destroyed
return;
}
if (src) {
// 推流给自己成功 [AUTO-TRANSLATED:65dba71b]
// Push stream to yourself successfully
cb(SockException());
strongSelf->onGetMediaSource(src);
strongSelf->startTimer(timeout_ms);
return;
}
// 推流失败 [AUTO-TRANSLATED:4d8d226a]
// Push stream failed
if (!strongSelf->_process.wait(false)) {
// ffmpeg进程已经退出 [AUTO-TRANSLATED:04193893]
// ffmpeg process has exited
cb(SockException(Err_other, StrPrinter << "ffmpeg已经退出,exit code = " << strongSelf->_process.exit_code()));
return;
}
// ffmpeg进程还在线，但是等待推流超时 [AUTO-TRANSLATED:9f71f17b]
// ffmpeg process is still online, but waiting for the stream to timeout
cb(SockException(Err_other, "等待超时"));
});
} else{
// 推流给其他服务器的，通过判断FFmpeg进程是否在线判断是否成功 [AUTO-TRANSLATED:9b963da5]
// Push stream to other servers, judge whether it is successful by judging whether the FFmpeg process is online
weak_ptr<FFmpegSource> weakSelf = shared_from_this();
_timer = std::make_shared<Timer>(timeout_ms / 1000.0f, [weakSelf, cb, timeout_ms]() {
auto strongSelf = weakSelf.lock();
if (!strongSelf) {
// 自身已经销毁 [AUTO-TRANSLATED:5f954f8a]
// Self has been destroyed
return false;
}
// FFmpeg还在线，那么我们认为推流成功 [AUTO-TRANSLATED:4330df49]
// FFmpeg is still online, so we think the push stream is successful
if (strongSelf->_process.wait(false)) {
cb(SockException());
strongSelf->startTimer(timeout_ms);
return false;
}
// ffmpeg进程已经退出 [AUTO-TRANSLATED:04193893]
// ffmpeg process has exited
cb(SockException(Err_other, StrPrinter << "ffmpeg已经退出,exit code = " << strongSelf->_process.exit_code()));
return false;
}, _poller);
} catch (std::exception &ex) {
WarnL << ex.what();
cb(SockException(Err_other, ex.what()));
}
}
@@ -341,15 +347,70 @@ void FFmpegSource::onGetMediaSource(const MediaSource::Ptr &src) {
setDelegate(listener);
muxer->setDelegate(shared_from_this());
if (_enable_hls) {
src->setupRecord(Recorder::type_hls, true, "", 0);
src->getOwnerPoller()->async([=]() mutable {
src->setupRecord(Recorder::type_hls, true, "", 0);
});
}
if (_enable_mp4) {
src->setupRecord(Recorder::type_mp4, true, "", 0);
src->getOwnerPoller()->async([=]() mutable {
src->setupRecord(Recorder::type_mp4, true, "", 0);
});
}
}
}
void FFmpegSnap::makeSnap(const string &play_url, const string &save_path, float timeout_sec, const onSnap &cb) {
#if defined(ENABLE_FFMPEG)
#include "Player/MediaPlayer.h"
#include "Codec/Transcode.h"
static void makeSnapAsync(const string &play_url, const string &save_path, float timeout_sec, const FFmpegSnap::onSnap &cb) {
struct Holder {
MediaPlayer::Ptr player;
};
auto holder = std::make_shared<Holder>();
auto player = std::make_shared<MediaPlayer>();
(*player)[mediakit::Client::kTimeoutMS] = timeout_sec * 1000;
player->setOnPlayResult([holder, save_path, cb, timeout_sec](const SockException &ex) mutable {
onceToken token(nullptr, [&]() { holder->player = nullptr; });
auto video = ex ? nullptr : dynamic_pointer_cast<VideoTrack>(holder->player->getTrack(TrackVideo, false));
if (!video) {
cb(false, ex ? ex.what() : "none video track");
return;
}
auto decoder = std::make_shared<FFmpegDecoder>(video);
auto new_holder = std::make_shared<Holder>(*holder);
auto timer = EventPollerPool::Instance().getPoller()->doDelayTask(1000 * timeout_sec, [cb, new_holder]() {
// 防止解码失败导致播放器无法释放
new_holder->player = nullptr;
cb(false, "decode frame timeout");
return 0;
});
auto done = false;
decoder->setOnDecode([save_path, new_holder, cb, done, timer](const FFmpegFrame::Ptr &frame) mutable {
if (done) {
return;
}
onceToken token(nullptr, [&]() { new_holder->player = nullptr; timer->cancel(); done = true; });
auto ret = FFmpegUtils::saveFrame(frame, save_path.data());
cb(std::get<0>(ret), std::get<1>(ret));
});
video->addDelegate([decoder](const Frame::Ptr &frame) { return decoder->inputFrame(frame, false, true); });
});
player->play(play_url);
holder->player = std::move(player);
}
#endif
void FFmpegSnap::makeSnap(bool async, const string &play_url, const string &save_path, float timeout_sec, const onSnap &cb) {
#if defined(ENABLE_FFMPEG)
if (async) {
makeSnapAsync(play_url, save_path, timeout_sec, cb);
return;
}
#endif
GET_CONFIG(string, ffmpeg_bin, FFmpeg::kBin);
GET_CONFIG(string, ffmpeg_snap, FFmpeg::kSnap);
GET_CONFIG(string, ffmpeg_log, FFmpeg::kLog);

25
server/FFmpegSource.h
View File

@@ -26,17 +26,20 @@ namespace FFmpeg {
class FFmpegSnap {
public:
using onSnap = std::function<void(bool success, const std::string &err_msg)>;
// / 创建截图 [AUTO-TRANSLATED:6d334c49]
// / Create a screenshot
// / \param play_url 播放url地址，只要FFmpeg支持即可 [AUTO-TRANSLATED:609d4de4]
// / \param play_url The playback URL address, as long as FFmpeg supports it
// / \param save_path 截图jpeg文件保存路径 [AUTO-TRANSLATED:0fc0ac0d]
// / \param save_path The path to save the screenshot JPEG file
// / \param timeout_sec 生成截图超时时间(防止阻塞太久) [AUTO-TRANSLATED:0dcc0095]
// / \param timeout_sec Timeout for generating the screenshot (to prevent blocking for too long)
// / \param cb 生成截图成功与否回调 [AUTO-TRANSLATED:5b4b93c9]
// / \param cb Callback for whether the screenshot was generated successfully
static void makeSnap(const std::string &play_url, const std::string &save_path, float timeout_sec, const onSnap &cb);
/**
* 创建截图 [AUTO-TRANSLATED:6d334c49]
* Create a screenshot
* @param async 是否使用异步截图方式(非ffmpeg命令行，而是使用zlm api，但是仅限于zlm播放器支持的拉流协议)
* @param play_url 播放url地址，只要FFmpeg支持即可 [AUTO-TRANSLATED:609d4de4]
* @param play_url The playback URL address, as long as FFmpeg supports it
* @param save_path 截图jpeg文件保存路径 [AUTO-TRANSLATED:0fc0ac0d]
* @param save_path The path to save the screenshot JPEG file
* @param timeout_sec 生成截图超时时间(防止阻塞太久) [AUTO-TRANSLATED:0dcc0095]
* @param timeout_sec Timeout for generating the screenshot (to prevent blocking for too long)
* @param cb 生成截图成功与否回调 [AUTO-TRANSLATED:5b4b93c9]
* @param cb Callback for whether the screenshot was generated successfully
*/
static void makeSnap(bool async, const std::string &play_url, const std::string &save_path, float timeout_sec, const onSnap &cb);
private:
FFmpegSnap() = delete;

27
server/Process.cpp
View File

@@ -10,6 +10,7 @@
#include <limits.h>
#include <sys/stat.h>
#include "ShellParser.h"
#ifndef _WIN32
#include <sys/resource.h>
#include <unistd.h>
@@ -86,20 +87,22 @@ static int runChildProcess(string cmd, string log_file) {
// Close log file.
::fclose(fp);
}
fprintf(stderr, "\r\n\r\n#### pid=%d,cmd=%s #####\r\n\r\n", getpid(), cmd.data());
fprintf(stderr, "\r\n#### pid=%d,cmd=%s #####\r\n", getpid(), cmd.data());
auto params = split(cmd, " ");
// memory leak in child process, it's ok.
char **charpv_params = new char *[params.size() + 1];
for (int i = 0; i < (int)params.size(); i++) {
std::string &p = params[i];
charpv_params[i] = (char *)p.data();
auto result = parse_shell_like(cmd);
if (!result.ok) {
fprintf(stderr, "parse cmd line failed: %s, pos: %ld", result.error_msg.data(), result.error_pos);
return -1;
}
// EOF: NULL
charpv_params[params.size()] = NULL;
// TODO: execv or execvp
auto ret = execv(params[0].c_str(), charpv_params);
delete[] charpv_params;
auto argv = make_argv(result.args);
auto argc = 0u;
fprintf(stderr, "\r\n#### args #####\r\n");
for (auto &arg : argv) {
fprintf(stderr, "arg[%d]: %s\r\n", argc++, arg ? arg : "null");
}
fprintf(stderr, "\r\n#### process log #####\r\n");
auto ret = execv(argv[0], (char * const *)(argv.data()));
if (ret < 0) {
fprintf(stderr, "execv process failed:%d(%s)\r\n", get_uv_error(), get_uv_errmsg());

207
server/ShellParser.h Normal file
View File

@@ -0,0 +1,207 @@
/*
* Copyright (c) 2016-present The ZLMediaKit project authors. All Rights Reserved.
*
* This file is part of ZLMediaKit(https://github.com/ZLMediaKit/ZLMediaKit).
*
* Use of this source code is governed by MIT-like license that can be found in the
* LICENSE file in the root of the source tree. All contributing project authors
* may be found in the AUTHORS file in the root of the source tree.
*/
#ifndef ZLMEDIAKIT_SHELLPARSER_H
#define ZLMEDIAKIT_SHELLPARSER_H
#include <iostream>
#include <string>
#include <vector>
#include <cctype>
// Shell-like command line parser.
// Features:
// - Whitespace splitting (space, tab, newline)
// - Quotes: single ('...') and double ("...")
// - Escapes with backslash (\\) outside quotes
// - In single quotes: backslash is literal (like POSIX shell)
// - In double quotes: backslash can escape " $ ` \\ and newline (line continuation)
// Additionally supports common C-style escapes: \n \t \r \0 .. outside and inside double quotes
// - Line continuation: backslash followed by newline is ignored
// - Produces argv pointers with stable lifetime backed by std::vector<std::string>
//
// Notes:
// - This is NOT a full shell (no variable expansion, no globbing, no command substitution).
// - Behavior aims to be practical and safe for exec* arguments building.
struct ParseResult {
ParseResult(bool ok, const char *err, size_t pos, std::vector<std::string> args)
: ok(ok)
, error_msg(err)
, error_pos(pos)
, args(std::move(args)) {}
bool ok;
std::string error_msg;
size_t error_pos = 0; // index in input when error happens
std::vector<std::string> args; // parsed arguments
};
namespace detail {
inline bool is_space(char c) {
return c == ' ' || c == '\t' || c == '\n';
}
// Returns true if it handled a line continuation ("\\\n").
inline bool handle_line_continuation(const std::string &s, size_t &i) {
if (i + 1 < s.size() && s[i] == '\\' && s[i + 1] == '\n') {
i += 2; // consume both and do nothing
return true;
}
return false;
}
inline bool hex_digit(char c) { return std::isxdigit(static_cast<unsigned char>(c)) != 0; }
inline int hex_val(char c) {
if (c >= '0' && c <= '9') return c - '0';
if (c >= 'a' && c <= 'f') return 10 + (c - 'a');
if (c >= 'A' && c <= 'F') return 10 + (c - 'A');
return 0;
}
// Parse C-style escapes: \n, \t, \r, \0..\377 (octal), \xHH (hex). Returns std::nullopt if not a known escape.
inline std::pair<bool, char> c_style_escape(const std::string &s, size_t &i) {
if (i >= s.size()) return std::make_pair(false, '\0');
char c = s[i];
switch (c) {
case 'n': ++i; return std::make_pair(true, '\n');
case 't': ++i; return std::make_pair(true, '\t');
case 'r': ++i; return std::make_pair(true, '\r');
case 'a': ++i; return std::make_pair(true, '\a');
case 'b': ++i; return std::make_pair(true, '\b');
case 'f': ++i; return std::make_pair(true, '\f');
case 'v': ++i; return std::make_pair(true, '\v');
case '\\': ++i; return std::make_pair(true, '\\');
case '"': ++i; return std::make_pair(true, '"');
case '\'': ++i; return std::make_pair(true, '\'');
case '0': {
// up to 3 octal digits total (including the first 0 already consumed here?)
// Here c=='0' means octal sequence starts at current '0'.
// We'll parse up to 3 octal digits starting at current pos.
int val = 0; int cnt = 0;
while (i < s.size() && cnt < 3 && (s[i] >= '0' && s[i] <= '7')) {
val = (val << 3) + (s[i] - '0');
++i; ++cnt;
}
return std::make_pair(true, static_cast<char>(val & 0xFF));
}
case 'x': {
++i; // consume 'x'
int val = 0; int cnt = 0;
while (i < s.size() && cnt < 2 && hex_digit(s[i])) {
val = (val << 4) + hex_val(s[i]);
++i; ++cnt;
}
if (cnt == 0) return std::make_pair(false, '\0'); // not actually a hex escape
return std::make_pair(true, static_cast<char>(val & 0xFF));
}
default:
return std::make_pair(false, '\0');
}
}
}
ParseResult parse_shell_like(const std::string &input) {
using namespace detail;
std::vector<std::string> args;
std::string cur;
enum class State { Normal, InSingle, InDouble };
State st = State::Normal;
size_t i = 0; const size_t N = input.size();
while (i < N) {
// line continuation check (\\\n) applies in all states
if (handle_line_continuation(input, i)) continue;
if (i >= N) break;
char c = input[i];
switch (st) {
case State::Normal: {
if (is_space(c)) {
if (!cur.empty()) { args.emplace_back(std::move(cur)); cur.clear(); }
++i;
} else if (c == '\'') {
st = State::InSingle; ++i;
} else if (c == '"') {
st = State::InDouble; ++i;
} else if (c == '\\') {
++i; // consume backslash
if (i >= N) {
return {false, "结尾处孤立的反斜杠（未转义任何字符）", i, {}};
}
// Try C-style escapes first
auto esc = c_style_escape(input, i);
if (esc.first) {
cur.push_back(esc.second);
} else {
// Not a known C escape: take the next char literally
cur.push_back(input[i]);
++i;
}
} else {
cur.push_back(c); ++i;
}
} break;
case State::InSingle: {
if (c == '\'') { st = State::Normal; ++i; }
else { cur.push_back(c); ++i; }
} break;
case State::InDouble: {
if (c == '"') { st = State::Normal; ++i; }
else if (c == '\\') {
++i; // consume backslash
if (i >= N) {
return {false, "双引号内以反斜杠结尾，缺少被转义字符", i, {}};
}
// In POSIX shell, within double quotes, only certain escapes are special.
// Here we support both POSIX subset and common C-style escapes for practicality.
auto esc = c_style_escape(input, i);
if (esc.first) {
cur.push_back(esc.second);
} else {
// If not a C-style escape, allow escaping one char literally (e.g., $ `)
cur.push_back(input[i]);
++i;
}
} else {
cur.push_back(c); ++i;
}
} break;
}
}
if (st == State::InSingle) {
return {false, "缺少配对的单引号（'）", i, {}};
}
if (st == State::InDouble) {
return {false, "缺少配对的双引号（\"）", i, {}};
}
if (!cur.empty()) args.emplace_back(std::move(cur));
return {true, "", 0, std::move(args)};
}
// Helper: build argv pointers backed by the strings' storage.
// The returned vector includes a trailing nullptr, suitable for execv*.
inline std::vector<const char*> make_argv(const std::vector<std::string>& args) {
std::vector<const char*> argv;
argv.reserve(args.size() + 1);
for (const auto &s : args) argv.push_back(s.c_str());
argv.push_back(nullptr);
return argv;
}
#endif // ZLMEDIAKIT_SHELLPARSER_H

48
server/System.cpp
View File

@@ -15,6 +15,12 @@
#if !defined(ANDROID)
#include <execinfo.h>
#endif//!defined(ANDROID)
#else
#include <fcntl.h>
#include <io.h>
#include <Windows.h>
#include <DbgHelp.h>
#pragma comment(lib, "DbgHelp.lib")
#endif//!defined(_WIN32)
#include <cstdlib>
@@ -213,6 +219,48 @@ void System::systemSetup(){
// Ignore the hang up signal
signal(SIGHUP, SIG_IGN);
#endif// ANDROID
#else
// 避免系统弹窗导致程序阻塞，适合无界面或后台服务场景。
SetErrorMode(SEM_FAILCRITICALERRORS | SEM_NOGPFAULTERRORBOX | SEM_NOOPENFILEERRORBOX);
#if !defined(__MINGW32__)
// 将assert和error时错误输出
_CrtSetReportMode(_CRT_ASSERT, _CRTDBG_MODE_DEBUG);
_CrtSetReportMode(_CRT_ERROR, _CRTDBG_MODE_DEBUG);
#endif
_setmode(0, _O_BINARY);
_setmode(1, _O_BINARY);
_setmode(2, _O_BINARY);
setvbuf(stdout, NULL, _IONBF, 0);
setvbuf(stderr, NULL, _IONBF, 0);
std::ios_base::sync_with_stdio(false);
// 注册crash自动生成dump（等价core dump）
SetUnhandledExceptionFilter([](EXCEPTION_POINTERS *pException) -> LONG {
// 生成 dump 文件名，带时间戳
char dumpPath[MAX_PATH];
std::time_t t = std::time(nullptr);
std::tm tm;
#ifdef _MSC_VER
localtime_s(&tm, &t);
#else
tm = *std::localtime(&t);
#endif
std::strftime(dumpPath, sizeof(dumpPath), "crash_%Y%m%d_%H%M%S.dmp", &tm);
HANDLE hFile = CreateFileA(dumpPath, GENERIC_WRITE, 0, nullptr, CREATE_ALWAYS, FILE_ATTRIBUTE_NORMAL, nullptr);
if (hFile != INVALID_HANDLE_VALUE) {
MINIDUMP_EXCEPTION_INFORMATION mdei;
mdei.ThreadId = GetCurrentThreadId();
mdei.ExceptionPointers = pException;
mdei.ClientPointers = FALSE;
MiniDumpWriteDump(GetCurrentProcess(), GetCurrentProcessId(), hFile, MiniDumpNormal, &mdei, nullptr, nullptr);
CloseHandle(hFile);
}
return EXCEPTION_EXECUTE_HANDLER;
});
#endif//!defined(_WIN32)
}

105
server/VideoStack.cpp
View File

@@ -21,6 +21,14 @@
#define RGB_TO_U(R, G, B) (((-26 * (R) - 87 * (G) + 112 * (B) + 128) >> 8) + 128)
#define RGB_TO_V(R, G, B) (((112 * (R) - 102 * (G) - 10 * (B) + 128) >> 8) + 128)
static void fill_yuv_func(const mediakit::FFmpegFrame::Ptr &frame, int y, int u, int v) {
const auto& yuv = frame->get();
memset(yuv->data[0], y, yuv->linesize[0] * yuv->height);
memset(yuv->data[1], u, yuv->linesize[1] * ((yuv->height + 1) / 2));
memset(yuv->data[2], v, yuv->linesize[2] * ((yuv->height + 1) / 2));
}
INSTANCE_IMP(VideoStackManager)
Param::~Param() {
@@ -31,6 +39,13 @@ Param::~Param() {
Channel::Channel(const std::string& id, int width, int height, AVPixelFormat pixfmt)
: _id(id), _width(width), _height(height), _pixfmt(pixfmt) {
#if defined(VIDEOSTACK_KEEP_ASPECT_RATIO)
_keepAspectRatio = true;
#else
_keepAspectRatio = false;
#endif
_lastWidht = 0;
_lastHeight = 0;
_tmp = std::make_shared<mediakit::FFmpegFrame>();
_tmp->get()->width = _width;
@@ -39,14 +54,9 @@ Channel::Channel(const std::string& id, int width, int height, AVPixelFormat pix
av_frame_get_buffer(_tmp->get(), 32);
memset(_tmp->get()->data[0], 0, _tmp->get()->linesize[0] * _height);
memset(_tmp->get()->data[1], 0, _tmp->get()->linesize[1] * _height / 2);
memset(_tmp->get()->data[2], 0, _tmp->get()->linesize[2] * _height / 2);
auto frame = VideoStackManager::Instance().getBgImg();
_sws = std::make_shared<mediakit::FFmpegSws>(_pixfmt, _width, _height);
_tmp = _sws->inputFrame(frame);
resizeFrame(frame);
}
void Channel::addParam(const std::weak_ptr<Param>& p) {
@@ -60,8 +70,7 @@ void Channel::onFrame(const mediakit::FFmpegFrame::Ptr& frame) {
_poller->async([weakSelf, frame]() {
auto self = weakSelf.lock();
if (!self) { return; }
self->_tmp = self->_sws->inputFrame(frame);
self->resizeFrame(frame);
self->forEachParam([self](const Param::Ptr& p) { self->fillBuffer(p); });
});
}
@@ -110,6 +119,78 @@ void Channel::copyData(const mediakit::FFmpegFrame::Ptr& buf, const Param::Ptr&
default: WarnL << "No support pixformat: " << av_get_pix_fmt_name(p->pixfmt); break;
}
}
void Channel::resizeFrame(const mediakit::FFmpegFrame::Ptr &frame) {
if (_keepAspectRatio) {
resizeFrameImplWithAspectRatio(frame);
} else {
resizeFrameImplWithoutAspectRatio(frame);
}
}
void Channel::resizeFrameImplWithAspectRatio(const mediakit::FFmpegFrame::Ptr &frame) {
int srcWidth = frame->get()->width;
int srcHeight = frame->get()->height;
if (srcWidth <= 0 || srcHeight <= 0) {
return;
}
// 当新frame宽高变化时，重新初始化sws
if (srcWidth != _lastWidht || srcHeight != _lastHeight) {
_lastWidht = srcWidth;
_lastHeight = srcHeight;
fill_yuv_func(_tmp, 16, 128, 128);
int dstWidth = _width;
int dstHeight = _height;
float srcAspectRatio = static_cast<float>(srcWidth) / srcHeight;
float dstAspectRatio = static_cast<float>(dstWidth) / dstHeight;
int scaledWidth, scaledHeight;
if (srcAspectRatio > dstAspectRatio) {
scaledWidth = dstWidth;
scaledHeight = static_cast<int>(dstWidth / srcAspectRatio);
} else {
scaledHeight = dstHeight;
scaledWidth = static_cast<int>(dstHeight * srcAspectRatio);
}
_offsetX = (dstWidth - scaledWidth) / 2;
_offsetY = (dstHeight - scaledHeight) / 2;
_sws = std::make_shared<mediakit::FFmpegSws>(_pixfmt, scaledWidth, scaledHeight);
}
auto scaledFrame = _sws->inputFrame(frame);
int copyWidth = ((_width) < (scaledFrame->get()->width) ? (_width) : (scaledFrame->get()->width));
int copyHeight = ((_height) < (scaledFrame->get()->height) ? (_height) : (scaledFrame->get()->height));
for (int i = 0; i < copyHeight; i++) {
memcpy(
_tmp->get()->data[0] + (i + _offsetY) * _tmp->get()->linesize[0] + _offsetX, scaledFrame->get()->data[0] + i * scaledFrame->get()->linesize[0],
copyWidth);
}
for (int i = 0; i < (copyHeight + 1) / 2; i++) {
memcpy(
_tmp->get()->data[1] + (i + _offsetY / 2) * _tmp->get()->linesize[1] + _offsetX / 2,
scaledFrame->get()->data[1] + i * scaledFrame->get()->linesize[1], copyWidth / 2);
memcpy(
_tmp->get()->data[2] + (i + _offsetY / 2) * _tmp->get()->linesize[2] + _offsetX / 2,
scaledFrame->get()->data[2] + i * scaledFrame->get()->linesize[2], copyWidth / 2);
}
}
void Channel::resizeFrameImplWithoutAspectRatio(const mediakit::FFmpegFrame::Ptr &frame) {
if (!_sws) {
fill_yuv_func(_tmp, 16, 128, 128);
_sws = std::make_shared<mediakit::FFmpegSws>(_pixfmt, _width, _height);
}
_tmp = _sws->inputFrame(frame);
}
void StackPlayer::addChannel(const std::weak_ptr<Channel>& chn) {
std::lock_guard<std::recursive_mutex> lock(_mx);
_channels.push_back(chn);
@@ -151,10 +232,10 @@ void StackPlayer::play() {
// auto audioTrack = std::dynamic_pointer_cast<mediakit::AudioTrack>(strongPlayer->getTrack(mediakit::TrackAudio, false));
if (videoTrack) {
// 如果每次不同 可以加个时间戳 time(NULL);
// TODO:添加使用显卡还是cpu解码的判断逻辑 [AUTO-TRANSLATED:44bef37a]
// TODO: Add logic to determine whether to use GPU or CPU decoding
auto decoder = std::make_shared<mediakit::FFmpegDecoder>(
videoTrack, 0, std::vector<std::string>{"h264", "hevc"});
auto decoder = std::make_shared<mediakit::FFmpegDecoder>(videoTrack, 0, std::vector<std::string> { "h264", "hevc" });
decoder->setOnDecode([weakSelf](const mediakit::FFmpegFrame::Ptr& frame) mutable {
auto self = weakSelf.lock();
@@ -300,9 +381,7 @@ void VideoStack::initBgColor() {
double U = RGB_TO_U(R, G, B);
double V = RGB_TO_V(R, G, B);
memset(_buffer->get()->data[0], Y, _buffer->get()->linesize[0] * _height);
memset(_buffer->get()->data[1], U, _buffer->get()->linesize[1] * _height / 2);
memset(_buffer->get()->data[2], V, _buffer->get()->linesize[2] * _height / 2);
fill_yuv_func(_buffer, Y, U, V);
}
Channel::Ptr VideoStackManager::getChannel(const std::string& id, int width, int height,

12
server/VideoStack.h
View File

@@ -62,12 +62,24 @@ protected:
void copyData(const mediakit::FFmpegFrame::Ptr& buf, const Param::Ptr& p);
void resizeFrame(const mediakit::FFmpegFrame::Ptr &frame);
void resizeFrameImplWithAspectRatio(const mediakit::FFmpegFrame::Ptr &frame);
void resizeFrameImplWithoutAspectRatio(const mediakit::FFmpegFrame::Ptr &frame);
private:
std::string _id;
int _width;
int _height;
AVPixelFormat _pixfmt;
int _lastWidht;
int _lastHeight;
bool _keepAspectRatio;
int _offsetX;
int _offsetY;
mediakit::FFmpegFrame::Ptr _tmp;
std::recursive_mutex _mx;

903
server/WebApi.cpp
View File

File diff suppressed because it is too large Load Diff

608
server/WebApi.h
View File

@@ -1,225 +1,383 @@
/*
* Copyright (c) 2016-present The ZLMediaKit project authors. All Rights Reserved.
*
* This file is part of ZLMediaKit(https://github.com/ZLMediaKit/ZLMediaKit).
*
* Use of this source code is governed by MIT-like license that can be found in the
* LICENSE file in the root of the source tree. All contributing project authors
* may be found in the AUTHORS file in the root of the source tree.
*/
#ifndef ZLMEDIAKIT_WEBAPI_H
#define ZLMEDIAKIT_WEBAPI_H
#include <string>
#include <functional>
#include "json/json.h"
#include "Common/Parser.h"
#include "Network/Socket.h"
#include "Http/HttpSession.h"
#include "Common/MultiMediaSourceMuxer.h"
// 配置文件路径 [AUTO-TRANSLATED:8a373c2f]
// Configuration file path
extern std::string g_ini_file;
namespace mediakit {
// //////////RTSP服务器配置/////////// [AUTO-TRANSLATED:950e1981]
// //////////RTSP server configuration///////////
namespace Rtsp {
extern const std::string kPort;
} //namespace Rtsp
// //////////RTMP服务器配置/////////// [AUTO-TRANSLATED:8de6f41f]
// //////////RTMP server configuration///////////
namespace Rtmp {
extern const std::string kPort;
} //namespace RTMP
} // namespace mediakit
namespace API {
typedef enum {
NotFound = -500,//未找到
Exception = -400,//代码抛异常
InvalidArgs = -300,//参数不合法
SqlFailed = -200,//sql执行失败
AuthFailed = -100,//鉴权失败
OtherFailed = -1,//业务代码执行失败，
Success = 0//执行成功
} ApiErr;
extern const std::string kSecret;
}//namespace API
class ApiRetException: public std::runtime_error {
public:
ApiRetException(const char *str = "success" ,int code = API::Success):runtime_error(str){
_code = code;
}
int code(){ return _code; }
private:
int _code;
};
class AuthException : public ApiRetException {
public:
AuthException(const char *str):ApiRetException(str,API::AuthFailed){}
};
class InvalidArgsException: public ApiRetException {
public:
InvalidArgsException(const char *str):ApiRetException(str,API::InvalidArgs){}
};
class SuccessException: public ApiRetException {
public:
SuccessException():ApiRetException("success",API::Success){}
};
using ApiArgsType = std::map<std::string, std::string, mediakit::StrCaseCompare>;
template<typename Args, typename First>
std::string getValue(Args &args, const First &first) {
return args[first];
}
template<typename First>
std::string getValue(Json::Value &args, const First &first) {
return args[first].asString();
}
template<typename First>
std::string getValue(std::string &args, const First &first) {
return "";
}
template<typename First>
std::string getValue(const mediakit::Parser &parser, const First &first) {
auto ret = parser.getUrlArgs()[first];
if (!ret.empty()) {
return ret;
}
return parser.getHeader()[first];
}
template<typename First>
std::string getValue(mediakit::Parser &parser, const First &first) {
return getValue((const mediakit::Parser &) parser, first);
}
template<typename Args, typename First>
std::string getValue(const mediakit::Parser &parser, Args &args, const First &first) {
auto ret = getValue(args, first);
if (!ret.empty()) {
return ret;
}
return getValue(parser, first);
}
template<typename Args>
class HttpAllArgs {
mediakit::Parser* _parser = nullptr;
Args* _args = nullptr;
public:
const mediakit::Parser& parser;
Args& args;
HttpAllArgs(const mediakit::Parser &p, Args &a): parser(p), args(a) {}
HttpAllArgs(const HttpAllArgs &that): _parser(new mediakit::Parser(that.parser)),
_args(new Args(that.args)),
parser(*_parser), args(*_args) {}
~HttpAllArgs() {
if (_parser) {
delete _parser;
}
if (_args) {
delete _args;
}
}
template<typename Key>
toolkit::variant operator[](const Key &key) const {
return (toolkit::variant)getValue(parser, args, key);
}
};
using ArgsMap = HttpAllArgs<ApiArgsType>;
using ArgsJson = HttpAllArgs<Json::Value>;
using ArgsString = HttpAllArgs<std::string>;
#define API_ARGS_MAP toolkit::SockInfo &sender, mediakit::HttpSession::KeyValue &headerOut, const ArgsMap &allArgs, Json::Value &val
#define API_ARGS_MAP_ASYNC API_ARGS_MAP, const mediakit::HttpSession::HttpResponseInvoker &invoker
#define API_ARGS_JSON toolkit::SockInfo &sender, mediakit::HttpSession::KeyValue &headerOut, const ArgsJson &allArgs, Json::Value &val
#define API_ARGS_JSON_ASYNC API_ARGS_JSON, const mediakit::HttpSession::HttpResponseInvoker &invoker
#define API_ARGS_STRING toolkit::SockInfo &sender, mediakit::HttpSession::KeyValue &headerOut, const ArgsString &allArgs, Json::Value &val
#define API_ARGS_STRING_ASYNC API_ARGS_STRING, const mediakit::HttpSession::HttpResponseInvoker &invoker
#define API_ARGS_VALUE sender, headerOut, allArgs, val
// 注册http请求参数是map<string, variant, StrCaseCompare>类型的http api [AUTO-TRANSLATED:8a273897]
// Register http request parameters as map<string, variant, StrCaseCompare> type http api
void api_regist(const std::string &api_path, const std::function<void(API_ARGS_MAP)> &func);
// 注册http请求参数是map<string, variant, StrCaseCompare>类型,但是可以异步回复的的http api [AUTO-TRANSLATED:9da5d5f5]
// Register http request parameters as map<string, variant, StrCaseCompare> type, but can be replied asynchronously http api
void api_regist(const std::string &api_path, const std::function<void(API_ARGS_MAP_ASYNC)> &func);
// 注册http请求参数是Json::Value类型的http api(可以支持多级嵌套的json参数对象) [AUTO-TRANSLATED:c4794456]
// Register http request parameters as Json::Value type http api (can support multi-level nested json parameter objects)
void api_regist(const std::string &api_path, const std::function<void(API_ARGS_JSON)> &func);
// 注册http请求参数是Json::Value类型，但是可以异步回复的的http api [AUTO-TRANSLATED:742e57fd]
// Register http request parameters as Json::Value type, but can be replied asynchronously http api
void api_regist(const std::string &api_path, const std::function<void(API_ARGS_JSON_ASYNC)> &func);
// 注册http请求参数是http原始请求信息的http api [AUTO-TRANSLATED:72d3fe93]
// Register http request parameters as http original request information http api
void api_regist(const std::string &api_path, const std::function<void(API_ARGS_STRING)> &func);
// 注册http请求参数是http原始请求信息的异步回复的http api [AUTO-TRANSLATED:49feefa8]
// Register http request parameters as http original request information asynchronous reply http api
void api_regist(const std::string &api_path, const std::function<void(API_ARGS_STRING_ASYNC)> &func);
template<typename Args, typename First>
bool checkArgs(Args &args, const First &first) {
return !args[first].empty();
}
template<typename Args, typename First, typename ...KeyTypes>
bool checkArgs(Args &args, const First &first, const KeyTypes &...keys) {
return checkArgs(args, first) && checkArgs(args, keys...);
}
// 检查http url中或body中或http header参数是否为空的宏 [AUTO-TRANSLATED:9de001a4]
// Check whether the http url, body or http header parameters are empty
#define CHECK_ARGS(...) \
if(!checkArgs(allArgs,##__VA_ARGS__)){ \
throw InvalidArgsException("Required parameter missed: " #__VA_ARGS__); \
}
// 检查http参数中是否附带secret密钥的宏，127.0.0.1的ip不检查密钥 [AUTO-TRANSLATED:7546956c]
// Check whether the http parameters contain the secret key, the ip of 127.0.0.1 does not check the key
// 同时检测是否在ip白名单内 [AUTO-TRANSLATED:d12f963d]
// Check whether it is in the ip whitelist at the same time
#define CHECK_SECRET() \
do { \
auto ip = sender.get_peer_ip(); \
if (!HttpFileManager::isIPAllowed(ip)) { \
throw AuthException("Your ip is not allowed to access the service."); \
} \
CHECK_ARGS("secret"); \
if (api_secret != allArgs["secret"]) { \
throw AuthException("Incorrect secret"); \
} \
} while(false);
void installWebApi();
void unInstallWebApi();
#if defined(ENABLE_RTPPROXY)
uint16_t openRtpServer(uint16_t local_port, const mediakit::MediaTuple &tuple, int tcp_mode, const std::string &local_ip, bool re_use_port, uint32_t ssrc, int only_track, bool multiplex=false);
#endif
Json::Value makeMediaSourceJson(mediakit::MediaSource &media);
void getStatisticJson(const std::function<void(Json::Value &val)> &cb);
void addStreamProxy(const mediakit::MediaTuple &tuple, const std::string &url, int retry_count,
const mediakit::ProtocolOption &option, int rtp_type, float timeout_sec, const toolkit::mINI &args,
const std::function<void(const toolkit::SockException &ex, const std::string &key)> &cb);
#endif //ZLMEDIAKIT_WEBAPI_H
/*
* Copyright (c) 2016-present The ZLMediaKit project authors. All Rights Reserved.
*
* This file is part of ZLMediaKit(https://github.com/ZLMediaKit/ZLMediaKit).
*
* Use of this source code is governed by MIT-like license that can be found in the
* LICENSE file in the root of the source tree. All contributing project authors
* may be found in the AUTHORS file in the root of the source tree.
*/
#ifndef ZLMEDIAKIT_WEBAPI_H
#define ZLMEDIAKIT_WEBAPI_H
#include <string>
#include <functional>
#include "json/json.h"
#include "Common/Parser.h"
#include "Network/Socket.h"
#include "Http/HttpSession.h"
#include "Common/MultiMediaSourceMuxer.h"
#if defined(ENABLE_WEBRTC)
#include "webrtc/WebRtcTransport.h"
#endif
#include "Http/HttpCookieManager.h"
// 配置文件路径 [AUTO-TRANSLATED:8a373c2f]
// Configuration file path
extern std::string g_ini_file;
namespace mediakit {
// //////////RTSP服务器配置/////////// [AUTO-TRANSLATED:950e1981]
// //////////RTSP server configuration///////////
namespace Rtsp {
extern const std::string kPort;
} //namespace Rtsp
// //////////RTMP服务器配置/////////// [AUTO-TRANSLATED:8de6f41f]
// //////////RTMP server configuration///////////
namespace Rtmp {
extern const std::string kPort;
} //namespace RTMP
} // namespace mediakit
namespace API {
typedef enum {
NotFound = -500,//未找到
Exception = -400,//代码抛异常
InvalidArgs = -300,//参数不合法
SqlFailed = -200,//sql执行失败
AuthFailed = -100,//鉴权失败
OtherFailed = -1,//业务代码执行失败，
Success = 0//执行成功
} ApiErr;
extern const std::string kSecret;
extern const std::string kApiDebug;
} // namespace API
class ApiRetException : public std::runtime_error {
public:
ApiRetException(const char *str = "success", int code = API::Success, mediakit::StrCaseMap headers = {}, Json::Value body = {})
: runtime_error(str) {
_code = code;
_headers = std::move(headers);
_body = std::move(body);
}
int code() { return _code; }
mediakit::StrCaseMap &getHeaders() { return _headers; }
Json::Value &getBody() { return _body; }
private:
int _code;
mediakit::StrCaseMap _headers;
Json::Value _body;
};
class AuthException : public ApiRetException {
public:
AuthException(const char *str, mediakit::StrCaseMap headers = {}, Json::Value body = {})
: ApiRetException(str, API::AuthFailed, std::move(headers), std::move(body)) {}
};
class InvalidArgsException : public ApiRetException {
public:
InvalidArgsException(const char *str, mediakit::StrCaseMap headers = {}, Json::Value body = {})
: ApiRetException(str, API::InvalidArgs, std::move(headers), std::move(body)) {}
};
class SuccessException : public ApiRetException {
public:
SuccessException(mediakit::StrCaseMap headers = {}, Json::Value body = {})
: ApiRetException("success", API::Success, std::move(headers), std::move(body)) {}
};
using ApiArgsType = std::map<std::string, std::string, mediakit::StrCaseCompare>;
template<typename Args, typename Key>
std::string getValue(Args &args, const Key &key) {
auto it = args.find(key);
if (it == args.end()) {
return "";
}
return it->second;
}
template<typename Key>
std::string getValue(Json::Value &args, const Key &key) {
auto value = args.find(key);
if (value == nullptr) {
return "";
}
return value->asString();
}
template<typename Key>
std::string getValue(std::string &args, const Key &key) {
return "";
}
template <typename Key>
std::string getValue(const mediakit::Parser &parser, const Key &key) {
auto ret = getValue(parser.getUrlArgs(), key);
if (!ret.empty()) {
return ret;
}
return getValue(parser.getHeader(), key);
}
template<typename Key>
std::string getValue(mediakit::Parser &parser, const Key &key) {
return getValue((const mediakit::Parser &) parser, key);
}
template<typename Args, typename Key>
std::string getValue(const mediakit::Parser &parser, Args &args, const Key &key) {
auto ret = getValue(args, key);
if (!ret.empty()) {
return ret;
}
return getValue(parser, key);
}
template<typename Args>
class HttpAllArgs {
mediakit::Parser* _parser = nullptr;
Args* _args = nullptr;
public:
const mediakit::Parser& parser;
Args& args;
HttpAllArgs(const mediakit::Parser &p, Args &a): parser(p), args(a) {}
HttpAllArgs(const HttpAllArgs &that): _parser(new mediakit::Parser(that.parser)),
_args(new Args(that.args)),
parser(*_parser), args(*_args) {}
~HttpAllArgs() {
if (_parser) {
delete _parser;
}
if (_args) {
delete _args;
}
}
template<typename Key>
toolkit::variant operator[](const Key &key) const {
return (toolkit::variant)getValue(parser, args, key);
}
const Args& getArgs() const {
return args;
}
const mediakit::Parser &getParser() const {
return parser;
}
};
using ArgsMap = HttpAllArgs<ApiArgsType>;
using ArgsJson = HttpAllArgs<Json::Value>;
using ArgsString = HttpAllArgs<std::string>;
#define API_ARGS_MAP toolkit::SockInfo &sender, mediakit::HttpSession::KeyValue &headerOut, const ArgsMap &allArgs, Json::Value &val
#define API_ARGS_MAP_ASYNC API_ARGS_MAP, const mediakit::HttpSession::HttpResponseInvoker &invoker
#define API_ARGS_JSON toolkit::SockInfo &sender, mediakit::HttpSession::KeyValue &headerOut, const ArgsJson &allArgs, Json::Value &val
#define API_ARGS_JSON_ASYNC API_ARGS_JSON, const mediakit::HttpSession::HttpResponseInvoker &invoker
#define API_ARGS_STRING toolkit::SockInfo &sender, mediakit::HttpSession::KeyValue &headerOut, const ArgsString &allArgs, Json::Value &val
#define API_ARGS_STRING_ASYNC API_ARGS_STRING, const mediakit::HttpSession::HttpResponseInvoker &invoker
#define API_ARGS_VALUE sender, headerOut, allArgs, val
// 注册http请求参数是map<string, variant, StrCaseCompare>类型的http api [AUTO-TRANSLATED:8a273897]
// Register http request parameters as map<string, variant, StrCaseCompare> type http api
void api_regist(const std::string &api_path, const std::function<void(API_ARGS_MAP)> &func);
// 注册http请求参数是map<string, variant, StrCaseCompare>类型,但是可以异步回复的的http api [AUTO-TRANSLATED:9da5d5f5]
// Register http request parameters as map<string, variant, StrCaseCompare> type, but can be replied asynchronously http api
void api_regist(const std::string &api_path, const std::function<void(API_ARGS_MAP_ASYNC)> &func);
// 注册http请求参数是Json::Value类型的http api(可以支持多级嵌套的json参数对象) [AUTO-TRANSLATED:c4794456]
// Register http request parameters as Json::Value type http api (can support multi-level nested json parameter objects)
void api_regist(const std::string &api_path, const std::function<void(API_ARGS_JSON)> &func);
// 注册http请求参数是Json::Value类型，但是可以异步回复的的http api [AUTO-TRANSLATED:742e57fd]
// Register http request parameters as Json::Value type, but can be replied asynchronously http api
void api_regist(const std::string &api_path, const std::function<void(API_ARGS_JSON_ASYNC)> &func);
// 注册http请求参数是http原始请求信息的http api [AUTO-TRANSLATED:72d3fe93]
// Register http request parameters as http original request information http api
void api_regist(const std::string &api_path, const std::function<void(API_ARGS_STRING)> &func);
// 注册http请求参数是http原始请求信息的异步回复的http api [AUTO-TRANSLATED:49feefa8]
// Register http request parameters as http original request information asynchronous reply http api
void api_regist(const std::string &api_path, const std::function<void(API_ARGS_STRING_ASYNC)> &func);
template<typename Args, typename Key>
bool checkArgs(Args &args, const Key &key) {
return !args[key].empty();
}
template<typename Args, typename Key, typename ...KeyTypes>
bool checkArgs(Args &args, const Key &key, const KeyTypes &...keys) {
return checkArgs(args, key) && checkArgs(args, keys...);
}
// 检查http url中或body中或http header参数是否为空的宏 [AUTO-TRANSLATED:9de001a4]
// Check whether the http url, body or http header parameters are empty
#define CHECK_ARGS(...) \
if(!checkArgs(allArgs,##__VA_ARGS__)){ \
throw InvalidArgsException("Required parameter missed: " #__VA_ARGS__); \
}
// 检查http参数中是否附带secret密钥的宏，127.0.0.1的ip不检查密钥 [AUTO-TRANSLATED:7546956c]
// Check whether the http parameters contain the secret key, the ip of 127.0.0.1 does not check the key
// 同时检测是否在ip白名单内 [AUTO-TRANSLATED:d12f963d]
// Check whether it is in the ip whitelist at the same time
template <typename T>
void check_secret(toolkit::SockInfo &sender, mediakit::HttpSession::KeyValue &headerOut, const HttpAllArgs<T> &allArgs, Json::Value &val);
#define CHECK_SECRET() check_secret(sender, headerOut, allArgs, val)
void installWebApi();
void unInstallWebApi();
#if defined(ENABLE_RTPPROXY)
uint16_t openRtpServer(uint16_t local_port, const mediakit::MediaTuple &tuple, int tcp_mode, const std::string &local_ip, bool re_use_port, uint32_t ssrc, int only_track, bool multiplex=false);
#endif
Json::Value makeMediaSourceJson(mediakit::MediaSource &media);
ApiArgsType getAllArgs(const mediakit::Parser &parser);
void getStatisticJson(const std::function<void(Json::Value &val)> &cb);
void addStreamProxy(const mediakit::MediaTuple &tuple, const std::string &url, int retry_count, bool force,
const mediakit::ProtocolOption &option, float timeout_sec, const toolkit::mINI &args,
const std::function<void(const toolkit::SockException &ex, const std::string &key)> &cb);
void updateStreamProxy(const mediakit::MediaTuple &tuple, const std::string &url, const toolkit::mINI &args);
template <typename Type>
class ServiceController {
public:
using Pointer = std::shared_ptr<Type>;
std::unordered_map<std::string, Pointer> _map;
mutable std::recursive_mutex _mtx;
void clear() {
decltype(_map) copy;
{
std::lock_guard<std::recursive_mutex> lck(_mtx);
copy.swap(_map);
}
}
size_t erase(const std::string &key) {
Pointer erase_ptr;
{
std::lock_guard<std::recursive_mutex> lck(_mtx);
auto itr = _map.find(key);
if (itr != _map.end()) {
erase_ptr = std::move(itr->second);
_map.erase(itr);
return 1;
}
}
return 0;
}
size_t size() {
std::lock_guard<std::recursive_mutex> lck(_mtx);
return _map.size();
}
Pointer find(const std::string &key) const {
std::lock_guard<std::recursive_mutex> lck(_mtx);
auto it = _map.find(key);
if (it == _map.end()) {
return nullptr;
}
return it->second;
}
void for_each(const std::function<void(const std::string &, const Pointer &)> &cb, const std::string &key = {}) {
std::lock_guard<std::recursive_mutex> lck(_mtx);
if (key.empty()) {
auto it = _map.begin();
while (it != _map.end()) {
cb(it->first, it->second);
++it;
}
} else {
auto it = _map.find(key);
if (it == _map.end()) {
throw std::invalid_argument("key not found: " + key);
}
cb(key, it->second);
}
}
template<class ..._Args>
Pointer make(const std::string &key, _Args&& ...__args) {
// assert(!find(key));
auto server = std::make_shared<Type>(std::forward<_Args>(__args)...);
std::lock_guard<std::recursive_mutex> lck(_mtx);
auto it = _map.emplace(key, server);
assert(it.second);
return server;
}
template<class ..._Args>
Pointer makeWithAction(const std::string &key, std::function<void(Pointer)> action, _Args&& ...__args) {
// assert(!find(key));
auto server = std::make_shared<Type>(std::forward<_Args>(__args)...);
action(server);
std::lock_guard<std::recursive_mutex> lck(_mtx);
auto it = _map.emplace(key, server);
assert(it.second);
return server;
}
template<class ..._Args>
Pointer emplace(const std::string &key, _Args&& ...__args) {
// assert(!find(key));
auto server = std::static_pointer_cast<Type>(std::forward<_Args>(__args)...);
std::lock_guard<std::recursive_mutex> lck(_mtx);
auto it = _map.emplace(key, server);
assert(it.second);
return server;
}
};
#if defined(ENABLE_WEBRTC)
template <typename Args>
class WebRtcArgsImp : public mediakit::WebRtcArgs {
public:
WebRtcArgsImp(const HttpAllArgs<Args> &args, std::string session_id)
: _args(args)
, _session_id(std::move(session_id)) {}
~WebRtcArgsImp() override = default;
toolkit::variant operator[](const std::string &key) const override {
if (key == "url") {
return getUrl();
}
return _args[key];
}
private:
std::string getUrl() const {
auto &allArgs = _args;
CHECK_ARGS("app", "stream");
return StrPrinter << RTC_SCHEMA << "://" << (_args["Host"].empty() ? DEFAULT_VHOST : _args["Host"].data()) << "/" << _args["app"] << "/"
<< _args["stream"] << "?" << _args.getParser().params() + "&session=" + _session_id;
}
private:
HttpAllArgs<Args> _args;
std::string _session_id;
};
#endif
#endif //ZLMEDIAKIT_WEBAPI_H

155
server/WebHook.cpp
View File

@@ -18,9 +18,14 @@
#include "Http/HttpRequester.h"
#include "Network/Session.h"
#include "Rtsp/RtspSession.h"
#include "Player/PlayerProxy.h"
#include "WebHook.h"
#include "WebApi.h"
#if defined(ENABLE_PYTHON)
#include "pyinvoker.h"
#endif
using namespace std;
using namespace Json;
using namespace toolkit;
@@ -226,9 +231,14 @@ void do_http_hook(const string &url, const ArgsType &body, const function<void(c
void dumpMediaTuple(const MediaTuple &tuple, Json::Value& item);
static ArgsType make_json(const MediaInfo &args) {
ArgsType make_json(const MediaInfo &args) {
ArgsType body;
body["schema"] = args.schema;
if(!args.protocol.empty()){
body["protocol"] = args.protocol;
}else{
body["protocol"] = args.schema;
}
dumpMediaTuple(args, body);
body["params"] = args.params;
return body;
@@ -296,7 +306,7 @@ static string getPullUrl(const string &origin_fmt, const MediaInfo &info) {
}
// 告知源站这是来自边沿站的拉流请求，如果未找到流请立即返回拉流失败 [AUTO-TRANSLATED:adf0d210]
// Inform the origin station that this is a pull stream request from the edge station, if the stream is not found, please return the pull stream failure immediately
return string(url) + '?' + kEdgeServerParam + '&' + VHOST_KEY + '=' + info.vhost + '&' + info.params;
return string(url) + (strchr(url, '?') ? '&' : '?') + kEdgeServerParam + '&' + VHOST_KEY + '=' + info.vhost + '&' + info.params;
}
static void pullStreamFromOrigin(const vector<string> &urls, size_t index, size_t failed_cnt, const MediaInfo &args, const function<void()> &closePlayer) {
@@ -311,7 +321,7 @@ static void pullStreamFromOrigin(const vector<string> &urls, size_t index, size_
option.enable_hls = option.enable_hls || (args.schema == HLS_SCHEMA);
option.enable_mp4 = false;
addStreamProxy(args, url, retry_count, option, Rtsp::RTP_TCP, timeout_sec, mINI{}, [=](const SockException &ex, const string &key) mutable {
addStreamProxy(args, url, retry_count, false, option, timeout_sec, mINI{}, [=](const SockException &ex, const string &key) mutable {
if (!ex) {
return;
}
@@ -334,6 +344,10 @@ static mINI jsonToMini(const Value &obj) {
mINI ret;
if (obj.isObject()) {
for (auto it = obj.begin(); it != obj.end(); ++it) {
if (it->isNull()) {
// 忽略null，修复wvp传null覆盖Protocol配置的问题
continue;
}
try {
auto str = (*it).asString();
ret[it.name()] = std::move(str);
@@ -345,10 +359,29 @@ static mINI jsonToMini(const Value &obj) {
return ret;
}
ArgsType getRecordInfo(const RecordInfo &info) {
ArgsType body;
body["start_time"] = (Json::UInt64)info.start_time;
body["file_size"] = (Json::UInt64)info.file_size;
body["time_len"] = info.time_len;
body["file_path"] = info.file_path;
body["file_name"] = info.file_name;
body["folder"] = info.folder;
body["url"] = info.url;
dumpMediaTuple(info, body);
return body;
}
void installWebHook() {
GET_CONFIG(bool, hook_enable, Hook::kEnable);
NoticeCenter::Instance().addListener(&web_hook_tag, Broadcast::kBroadcastMediaPublish, [](BroadcastMediaPublishArgs) {
#if defined(ENABLE_PYTHON)
if (PythonInvoker::Instance().on_publish(type, args, invoker, sender)) {
return;
}
#endif
GET_CONFIG(string, hook_publish, Hook::kOnPublish);
if (!hook_enable || hook_publish.empty()) {
invoker("", ProtocolOption());
@@ -378,6 +411,11 @@ void installWebHook() {
});
NoticeCenter::Instance().addListener(&web_hook_tag, Broadcast::kBroadcastMediaPlayed, [](BroadcastMediaPlayedArgs) {
#if defined(ENABLE_PYTHON)
if (PythonInvoker::Instance().on_play(args, invoker, sender)) {
return;
}
#endif
GET_CONFIG(string, hook_play, Hook::kOnPlay);
if (!hook_enable || hook_play.empty()) {
invoker("");
@@ -393,6 +431,11 @@ void installWebHook() {
});
NoticeCenter::Instance().addListener(&web_hook_tag, Broadcast::kBroadcastFlowReport, [](BroadcastFlowReportArgs) {
#if defined(ENABLE_PYTHON)
if (PythonInvoker::Instance().on_flow_report(args, totalBytes, totalDuration, isPlayer, sender)) {
return;
}
#endif
GET_CONFIG(string, hook_flowreport, Hook::kOnFlowReport);
if (!hook_enable || hook_flowreport.empty()) {
return;
@@ -414,6 +457,11 @@ void installWebHook() {
// 监听kBroadcastOnGetRtspRealm事件决定rtsp链接是否需要鉴权(传统的rtsp鉴权方案)才能访问 [AUTO-TRANSLATED:00dc9fa3]
// Listen to the kBroadcastOnGetRtspRealm event to determine whether the rtsp link needs authentication (traditional rtsp authentication scheme) to access
NoticeCenter::Instance().addListener(&web_hook_tag, Broadcast::kBroadcastOnGetRtspRealm, [](BroadcastOnGetRtspRealmArgs) {
#if defined(ENABLE_PYTHON)
if (PythonInvoker::Instance().on_get_rtsp_realm(args, invoker, sender)) {
return;
}
#endif
GET_CONFIG(string, hook_rtsp_realm, Hook::kOnRtspRealm);
if (!hook_enable || hook_rtsp_realm.empty()) {
// 无需认证 [AUTO-TRANSLATED:77728e07]
@@ -441,6 +489,11 @@ void installWebHook() {
// 监听kBroadcastOnRtspAuth事件返回正确的rtsp鉴权用户密码 [AUTO-TRANSLATED:bcf1754e]
// Listen to the kBroadcastOnRtspAuth event to return the correct rtsp authentication username and password
NoticeCenter::Instance().addListener(&web_hook_tag, Broadcast::kBroadcastOnRtspAuth, [](BroadcastOnRtspAuthArgs) {
#if defined(ENABLE_PYTHON)
if (PythonInvoker::Instance().on_rtsp_auth(args, realm, user_name, must_no_encrypt, invoker, sender)) {
return;
}
#endif
GET_CONFIG(string, hook_rtsp_auth, Hook::kOnRtspAuth);
if (unAuthedRealm == realm || !hook_enable || hook_rtsp_auth.empty()) {
// 认证失败 [AUTO-TRANSLATED:70cf56ff]
@@ -471,10 +524,6 @@ void installWebHook() {
// 监听rtsp、rtmp源注册或注销事件 [AUTO-TRANSLATED:6396afa8]
// Listen to rtsp, rtmp source registration or deregistration events
NoticeCenter::Instance().addListener(&web_hook_tag, Broadcast::kBroadcastMediaChanged, [](BroadcastMediaChangedArgs) {
GET_CONFIG(string, hook_stream_changed, Hook::kOnStreamChanged);
if (!hook_enable || hook_stream_changed.empty()) {
return;
}
GET_CONFIG_FUNC(std::set<std::string>, stream_changed_set, Hook::kStreamChangedSchemas, [](const std::string &str) {
std::set<std::string> ret;
auto vec = split(str, "/");
@@ -491,6 +540,15 @@ void installWebHook() {
// This protocol registration deregistration event is ignored
return;
}
#if defined(ENABLE_PYTHON)
if (PythonInvoker::Instance().on_media_changed(bRegist, sender)) {
return;
}
#endif
GET_CONFIG(string, hook_stream_changed, Hook::kOnStreamChanged);
if (!hook_enable || hook_stream_changed.empty()) {
return;
}
ArgsType body;
if (bRegist) {
@@ -536,6 +594,12 @@ void installWebHook() {
return;
}
#if defined(ENABLE_PYTHON)
if (PythonInvoker::Instance().on_stream_not_found(args, sender, closePlayer)) {
return;
}
#endif
GET_CONFIG(string, hook_stream_not_found, Hook::kOnStreamNotFound);
if (!hook_enable || hook_stream_not_found.empty()) {
return;
@@ -559,23 +623,15 @@ void installWebHook() {
do_http_hook(hook_stream_not_found, body, res_cb);
});
static auto getRecordInfo = [](const RecordInfo &info) {
ArgsType body;
body["start_time"] = (Json::UInt64)info.start_time;
body["file_size"] = (Json::UInt64)info.file_size;
body["time_len"] = info.time_len;
body["file_path"] = info.file_path;
body["file_name"] = info.file_name;
body["folder"] = info.folder;
body["url"] = info.url;
dumpMediaTuple(info, body);
return body;
};
#ifdef ENABLE_MP4
// 录制mp4文件成功后广播 [AUTO-TRANSLATED:479ec954]
// Broadcast after recording the mp4 file successfully
NoticeCenter::Instance().addListener(&web_hook_tag, Broadcast::kBroadcastRecordMP4, [](BroadcastRecordMP4Args) {
#if defined(ENABLE_PYTHON)
if (PythonInvoker::Instance().on_record_mp4(info)) {
return;
}
#endif
GET_CONFIG(string, hook_record_mp4, Hook::kOnRecordMp4);
if (!hook_enable || hook_record_mp4.empty()) {
return;
@@ -587,6 +643,11 @@ void installWebHook() {
#endif // ENABLE_MP4
NoticeCenter::Instance().addListener(&web_hook_tag, Broadcast::kBroadcastRecordTs, [](BroadcastRecordTsArgs) {
#if defined(ENABLE_PYTHON)
if (PythonInvoker::Instance().on_record_ts(info)) {
return;
}
#endif
GET_CONFIG(string, hook_record_ts, Hook::kOnRecordTs);
if (!hook_enable || hook_record_ts.empty()) {
return;
@@ -615,13 +676,31 @@ void installWebHook() {
});
NoticeCenter::Instance().addListener(&web_hook_tag, Broadcast::kBroadcastStreamNoneReader, [](BroadcastStreamNoneReaderArgs) {
auto auto_close = false;
auto muxer = sender.getMuxer();
if (muxer && muxer->getOption().auto_close) {
auto_close = true;
}
if (!origin_urls.empty() && sender.getOriginType() == MediaOriginType::pull) {
// 边沿站无人观看时如果是拉流的则立即停止溯源 [AUTO-TRANSLATED:a1429c77]
// If no one is watching at the edge station, stop tracing immediately if it is pulling
sender.close(false);
WarnL << "无人观看主动关闭流:" << sender.getOriginUrl();
if (!auto_close) {
auto ptr = sender.shared_from_this();
sender.getOwnerPoller()->async([ptr]() {
ptr->close(false);
});
WarnL << "Auto close stream when none reader: " << sender.getOriginUrl();
}
return;
}
#if defined(ENABLE_PYTHON)
if (PythonInvoker::Instance().on_stream_none_reader(sender)) {
return;
}
#endif
GET_CONFIG(string, hook_stream_none_reader, Hook::kOnStreamNoneReader);
if (!hook_enable || hook_stream_none_reader.empty()) {
return;
@@ -633,18 +712,27 @@ void installWebHook() {
weak_ptr<MediaSource> weakSrc = sender.shared_from_this();
// 执行hook [AUTO-TRANSLATED:1df68201]
// Execute hook
do_http_hook(hook_stream_none_reader, body, [weakSrc](const Value &obj, const string &err) {
do_http_hook(hook_stream_none_reader, body, [weakSrc, auto_close](const Value &obj, const string &err) {
if (auto_close) {
// 在上层已经关闭了
return;
}
bool flag = obj["close"].asBool();
auto strongSrc = weakSrc.lock();
if (!flag || !err.empty() || !strongSrc) {
return;
}
strongSrc->close(false);
strongSrc->getOwnerPoller()->async([strongSrc]() { strongSrc->close(false); });
WarnL << "无人观看主动关闭流:" << strongSrc->getOriginUrl();
});
});
NoticeCenter::Instance().addListener(&web_hook_tag, Broadcast::kBroadcastSendRtpStopped, [](BroadcastSendRtpStoppedArgs) {
#if defined(ENABLE_PYTHON)
if (PythonInvoker::Instance().on_send_rtp_stopped(sender, ssrc, ex)) {
return;
}
#endif
GET_CONFIG(string, hook_send_rtp_stopped, Hook::kOnSendRtpStopped);
if (!hook_enable || hook_send_rtp_stopped.empty()) {
return;
@@ -694,6 +782,11 @@ void installWebHook() {
// 追踪用户的目的是为了缓存上次鉴权结果，减少鉴权次数，提高性能 [AUTO-TRANSLATED:22827145]
// The purpose of tracking users is to cache the last authentication result, reduce the number of authentication times, and improve performance
NoticeCenter::Instance().addListener(&web_hook_tag, Broadcast::kBroadcastHttpAccess, [](BroadcastHttpAccessArgs) {
#if defined(ENABLE_PYTHON)
if (PythonInvoker::Instance().on_http_access(parser, path, file_path, is_dir, invoker, sender)) {
return;
}
#endif
GET_CONFIG(string, hook_http_access, Hook::kOnHttpAccess);
if (!hook_enable || hook_http_access.empty()) {
// 未开启http文件访问鉴权，那么允许访问，但是每次访问都要鉴权； [AUTO-TRANSLATED:deb3a0ae]
@@ -713,6 +806,7 @@ void installWebHook() {
body["port"] = sender.get_peer_port();
body["id"] = sender.getIdentifier();
body["path"] = path;
body["file_path"] = file_path;
body["is_dir"] = is_dir;
body["params"] = parser.params();
for (auto &pr : parser.getHeader()) {
@@ -738,6 +832,11 @@ void installWebHook() {
});
NoticeCenter::Instance().addListener(&web_hook_tag, Broadcast::kBroadcastRtpServerTimeout, [](BroadcastRtpServerTimeoutArgs) {
#if defined(ENABLE_PYTHON)
if (PythonInvoker::Instance().on_rtp_server_timeout(local_port, tuple, tcp_mode, re_use_port, ssrc)) {
return;
}
#endif
GET_CONFIG(string, rtp_server_timeout, Hook::kOnRtpServerTimeout);
if (!hook_enable || rtp_server_timeout.empty()) {
return;
@@ -754,6 +853,14 @@ void installWebHook() {
do_http_hook(rtp_server_timeout, body);
});
NoticeCenter::Instance().addListener(&web_hook_tag, Broadcast::kBroadcastPlayerProxyFailed, [](BroadcastPlayerProxyFailedArgs) {
#if defined(ENABLE_PYTHON)
if (PythonInvoker::Instance().on_player_proxy_failed(sender, ex)) {
return;
}
#endif
});
// 汇报服务器重新启动 [AUTO-TRANSLATED:bd7d83df]
// Report server restart
reportServerStarted();

76
server/main.cpp
View File

@@ -30,6 +30,8 @@
#if defined(ENABLE_WEBRTC)
#include "../webrtc/WebRtcTransport.h"
#include "../webrtc/WebRtcSession.h"
#include "../webrtc/WebRtcSignalingSession.h"
#include "../webrtc/IceSession.hpp"
#endif
#if defined(ENABLE_SRT)
@@ -41,9 +43,11 @@
#include "ZLMVersion.h"
#endif
#if !defined(_WIN32)
#if defined(ENABLE_PYTHON)
#include "pyinvoker.h"
#endif
#include "System.h"
#endif//!defined(_WIN32)
using namespace std;
using namespace toolkit;
@@ -59,7 +63,7 @@ const string kSSLPort = HTTP_FIELD"sslport";
onceToken token1([](){
mINI::Instance()[kPort] = 80;
mINI::Instance()[kSSLPort] = 443;
},nullptr);
});
}//namespace Http
// //////////SHELL配置/////////// [AUTO-TRANSLATED:f023ec45]
@@ -69,7 +73,7 @@ namespace Shell {
const string kPort = SHELL_FIELD"port";
onceToken token1([](){
mINI::Instance()[kPort] = 9000;
},nullptr);
});
} //namespace Shell
// //////////RTSP服务器配置/////////// [AUTO-TRANSLATED:950e1981]
@@ -81,7 +85,7 @@ const string kSSLPort = RTSP_FIELD"sslport";
onceToken token1([](){
mINI::Instance()[kPort] = 554;
mINI::Instance()[kSSLPort] = 332;
},nullptr);
});
} //namespace Rtsp
@@ -94,7 +98,7 @@ const string kSSLPort = RTMP_FIELD"sslport";
onceToken token1([](){
mINI::Instance()[kPort] = 1935;
mINI::Instance()[kSSLPort] = 19350;
},nullptr);
});
} //namespace RTMP
// //////////Rtp代理相关配置/////////// [AUTO-TRANSLATED:7b285587]
@@ -104,9 +108,17 @@ namespace RtpProxy {
const string kPort = RTP_PROXY_FIELD"port";
onceToken token1([](){
mINI::Instance()[kPort] = 10000;
},nullptr);
});
} //namespace RtpProxy
namespace Python {
#define Python_FIELD "python."
const string kPlugin = Python_FIELD"plugin";
onceToken token1([](){
mINI::Instance()[kPlugin] = "";
});
} //namespace Python
} // namespace mediakit
@@ -259,10 +271,21 @@ int start_main(int argc,char *argv[]) {
// Start daemon process
System::startDaemon(kill_parent_if_failed);
}
#endif //! defined(_WIN32)
// 设置poller线程数和cpu亲和性,该函数必须在使用ZLToolKit网络相关对象之前调用才能生效 [AUTO-TRANSLATED:7f03a1e5]
// Set the number of poller threads and CPU affinity. This function must be called before using ZLToolKit network related objects to take effect.
// 如果需要调用getSnap和addFFmpegSource接口，可以关闭cpu亲和性 [AUTO-TRANSLATED:7629f7bc]
// If you need to call the getSnap and addFFmpegSource interfaces, you can turn off CPU affinity
EventPollerPool::setPoolSize(threads);
WorkThreadPool::setPoolSize(threads);
EventPollerPool::enableCpuAffinity(affinity);
WorkThreadPool::enableCpuAffinity(affinity);
// 开启崩溃捕获等 [AUTO-TRANSLATED:9c7c759c]
// Enable crash capture, etc.
System::systemSetup();
#endif//!defined(_WIN32)
// 启动异步日志线程 [AUTO-TRANSLATED:c93cc6f4]
// Start asynchronous log thread
@@ -316,15 +339,6 @@ int start_main(int argc,char *argv[]) {
uint16_t httpsPort = mINI::Instance()[Http::kSSLPort];
uint16_t rtpPort = mINI::Instance()[RtpProxy::kPort];
// 设置poller线程数和cpu亲和性,该函数必须在使用ZLToolKit网络相关对象之前调用才能生效 [AUTO-TRANSLATED:7f03a1e5]
// Set the number of poller threads and CPU affinity. This function must be called before using ZLToolKit network related objects to take effect.
// 如果需要调用getSnap和addFFmpegSource接口，可以关闭cpu亲和性 [AUTO-TRANSLATED:7629f7bc]
// If you need to call the getSnap and addFFmpegSource interfaces, you can turn off CPU affinity
EventPollerPool::setPoolSize(threads);
WorkThreadPool::setPoolSize(threads);
EventPollerPool::enableCpuAffinity(affinity);
// 简单的telnet服务器，可用于服务器调试，但是不能使用23端口，否则telnet上了莫名其妙的现象 [AUTO-TRANSLATED:f9324c6e]
// Simple telnet server, can be used for server debugging, but cannot use port 23, otherwise telnet will have inexplicable phenomena
// 测试方法:telnet 127.0.0.1 9000 [AUTO-TRANSLATED:de0ac883]
@@ -369,8 +383,17 @@ int start_main(int argc,char *argv[]) {
}
return Socket::createSocket(new_poller, false);
});
auto signaleSrv = std::make_shared<TcpServer>();
auto signalsSrv = std::make_shared<TcpServer>();
auto iceTcpSrv = std::make_shared<TcpServer>();
auto iceSrv = std::make_shared<UdpServer>();
uint16_t rtcPort = mINI::Instance()[Rtc::kPort];
uint16_t rtcTcpPort = mINI::Instance()[Rtc::kTcpPort];
uint16_t signalingPort = mINI::Instance()[Rtc::kSignalingPort];
uint16_t signalSslPort = mINI::Instance()[Rtc::kSignalingSslPort];
uint16_t icePort = mINI::Instance()[Rtc::kIcePort];
uint16_t iceTcpPort = mINI::Instance()[Rtc::kIceTcpPort];
#endif//defined(ENABLE_WEBRTC)
@@ -436,6 +459,12 @@ int start_main(int argc,char *argv[]) {
if (rtcTcpPort) { rtcSrv_tcp->start<WebRtcSession>(rtcTcpPort, listen_ip);}
//webrtc 信令服务器
if (signalingPort) { signaleSrv->start<WebRtcWebcosktSignalingSession>(signalingPort);}
if (signalSslPort) { signalsSrv->start<WebRtcWebcosktSignalSslSession>(signalSslPort);}
//STUN/TURN服务
if (icePort) { iceSrv->start<IceSession>(icePort);}
if (iceTcpPort) { iceTcpSrv->start<IceSession>(iceTcpPort);}
#endif//defined(ENABLE_WEBRTC)
#if defined(ENABLE_SRT)
@@ -478,12 +507,25 @@ int start_main(int argc,char *argv[]) {
g_reload_certificates();
});
#endif
#if defined(ENABLE_PYTHON)
// 初始化python解释器
auto &ref = PythonInvoker::Instance();
auto py_plugin = mINI::Instance()[Python::kPlugin];
if (!py_plugin.empty()) {
ref.load(py_plugin);
}
#endif
sem.wait();
}
unInstallWebApi();
unInstallWebHook();
onProcessExited();
#if defined(ENABLE_PYTHON)
PythonInvoker::release();
#endif
// 休眠1秒再退出，防止资源释放顺序错误 [AUTO-TRANSLATED:1b11a74f]
// sleep for 1 second before exiting, to prevent resource release order errors
InfoL << "程序退出中,请等待...";

794
server/pyinvoker.cpp Normal file
View File

@@ -0,0 +1,794 @@
#if defined(ENABLE_PYTHON)
#include "pyinvoker.h"
#include <chrono>
#include <cstdlib>
#include <iostream>
#include <string>
#include <type_traits>
#include "WebApi.h"
#include "WebHook.h"
#include "Util/util.h"
#include "Util/File.h"
#include "Common/Parser.h"
#include "Common/macros.h"
#include "Http/HttpSession.h"
#include "Poller/EventPoller.h"
#include "WebApi.h"
using namespace toolkit;
using namespace mediakit;
extern ArgsType make_json(const MediaInfo &args);
extern void fillSockInfo(Json::Value & val, SockInfo* info);
extern ArgsType getRecordInfo(const RecordInfo &info);
extern std::string g_ini_file;
template <typename T>
typename std::enable_if<std::is_copy_constructible<T>::value, py::capsule>::type to_python(const T &obj) {
static auto name_str = toolkit::demangle(typeid(T).name());
auto p = new toolkit::Any(std::make_shared<T>(obj));
return py::capsule(p, name_str.data(), [](PyObject *capsule) {
auto p = reinterpret_cast<toolkit::Any *>(PyCapsule_GetPointer(capsule, name_str.data()));
delete p;
TraceL << "delete " << name_str << "(" << p << ")";
});
}
template <typename T>
typename std::enable_if<!std::is_copy_constructible<T>::value, py::capsule>::type to_python(const T &obj) {
static auto name_str = toolkit::demangle(typeid(T).name());
auto p = new toolkit::Any(std::shared_ptr<T>(const_cast<T *>(&obj), [](T *) {}));
return py::capsule(p, name_str.data(), [](PyObject *capsule) {
auto p = reinterpret_cast<toolkit::Any *>(PyCapsule_GetPointer(capsule, name_str.data()));
delete p;
TraceL << "unref " << name_str << "(" << p << ")";
});
}
static py::dict jsonToPython(const Json::Value &obj) {
py::dict ret;
if (obj.isObject()) {
for (auto it = obj.begin(); it != obj.end(); ++it) {
if (it->isNull()) {
// 忽略null，修复wvp传null覆盖Protocol配置的问题
continue;
}
try {
auto str = (*it).asString();
ret[it.name().data()] = std::move(str);
} catch (std::exception &) {
WarnL << "Json is not convertible to string, key: " << it.name() << ", value: " << (*it);
}
}
}
return ret;
}
py::dict to_python(const MediaInfo &args) {
auto json = make_json(args);
return jsonToPython(json);
}
py::dict to_python(const SockInfo &info) {
Json::Value json;
fillSockInfo(json, const_cast<SockInfo *>(&info));
return jsonToPython(json);
}
py::dict to_python(const RecordInfo &info) {
return jsonToPython(getRecordInfo(info));
}
template <typename T>
std::shared_ptr<T> to_python_ref(const T &t) {
return std::shared_ptr<T>(const_cast<T *>(&t), py::nodelete());
}
template <typename T>
T &to_native(const py::capsule &cap) {
static auto name_str = toolkit::demangle(typeid(T).name());
if (std::string(cap.name()) != name_str) {
throw std::runtime_error("Invalid capsule name!");
}
auto any = static_cast<toolkit::Any *>(cap.get_pointer());
return any->get<T>();
}
mINI to_native(const py::dict &opt) {
mINI ret;
for (auto &item : opt) {
// 转换为字符串（允许 int/float/bool 等）
ret.emplace(py::str(item.first).cast<std::string>(), py::str(item.second).cast<std::string>());
}
return ret;
}
void python_api_debug(const Parser &parser, const std::string &body) {
GET_CONFIG(bool, api_debug, API::kApiDebug);
if (!api_debug) {
return;
}
ssize_t size = body.size();
LogContextCapture log(getLogger(), toolkit::LDebug, __FILE__, "python http api debug", __LINE__);
log << "\r\n# request:\r\n" << parser.method() << " " << parser.fullUrl() << "\r\n";
log << "# header:\r\n";
for (auto &pr : parser.getHeader()) {
log << pr.first << " : " << pr.second << "\r\n";
}
auto &content = parser.content();
log << "# content:\r\n" << (content.size() > 4 * 1024 ? content.substr(0, 4 * 1024) : content) << "\r\n";
if (size > 0 && size < 4 * 1024) {
log << "# response:\r\n" << body << "\r\n";
} else {
log << "# response size:" << size << "\r\n";
}
}
void handle_http_request(const py::object &check_route, const py::object &submit_coro, const Parser &parser, const HttpSession::HttpResponseInvoker &invoker, bool &consumed, toolkit::SockInfo &sender) {
py::gil_scoped_acquire guard;
py::dict scope;
scope["type"] = "http";
scope["http_version"] = "1.1";
scope["method"] = parser.method();
scope["path"] = parser.url();
scope["query_string"] = parser.params();
py::list hdrs;
for (auto &kv : parser.getHeader()) {
// Starlette/ASGI 规范要求 headers 的 key 必须全小写字节串
hdrs.append(py::make_tuple(py::bytes(toolkit::strToLower(kv.first.data())), py::bytes(kv.second)));
}
scope["headers"] = hdrs;
bool ok = check_route(scope).cast<bool>();
if (!ok) {
return;
}
consumed = true;
Json::Value val;
HttpSession::KeyValue headerOut;
// http api被python拦截了，再api统一鉴权
try {
auto args = getAllArgs(parser);
auto allArgs = ArgsMap(parser, args);
// Python接口要求登录鉴权
CHECK_SECRET();
} catch (std::exception &ex) {
auto ex1 = dynamic_cast<ApiRetException *>(&ex);
if (ex1) {
val["code"] = ex1->code();
} else {
val["code"] = API::Exception;
}
val["msg"] = ex.what();
headerOut["Content-Type"] = "application/json";
invoker(200, headerOut, val.toStyledString());
return;
}
StrCaseMap resp_headers;
std::string resp_body;
int status = 500;
auto send = py::cpp_function([parser, invoker, status, resp_body, resp_headers](const py::dict &msg) mutable {
auto type = msg["type"].cast<std::string>();
if (type == "http.response.start") {
status = msg["status"].cast<int>();
for (auto tup : msg["headers"].cast<py::list>()) {
auto t = tup.cast<py::tuple>();
resp_headers[t[0].cast<std::string>()] = t[1].cast<std::string>();
}
return;
}
if (type == "http.response.body") {
resp_body += msg["body"].cast<std::string>();
// 💥 只在 more_body=False 时回调
bool more = msg.contains("more_body") && msg["more_body"].cast<bool>();
if (!more) {
python_api_debug(parser, resp_body);
invoker(status, resp_headers, resp_body);
}
}
});
submit_coro(scope, py::bytes(parser.content()), send);
}
class MuxerDelegatePython : public MediaSinkInterface {
public:
MuxerDelegatePython(py::object object) {
_py_muxer = std::move(object);
_input_frame = _py_muxer.attr("inputFrame");
_add_track = _py_muxer.attr("addTrack");
_add_track_completed = _py_muxer.attr("addTrackCompleted");
}
~MuxerDelegatePython() override {
py::gil_scoped_acquire guard;
try {
auto destroy = _py_muxer.attr("destroy");
destroy();
destroy = py::function();
} catch (std::exception &ex) {
ErrorL << "destroy python muxer failed: " << ex.what();
}
_input_frame = py::function();
_add_track = py::function();
_add_track_completed = py::function();
_py_muxer = py::function();
}
bool addTrack(const Track::Ptr &track) override {
py::gil_scoped_acquire guard;
return _add_track ? _add_track(track).cast<bool>() : false;
}
void addTrackCompleted() override {
py::gil_scoped_acquire guard;
if (_add_track_completed) {
_add_track_completed();
}
}
bool inputFrame(const Frame::Ptr &frame) override {
py::gil_scoped_acquire guard;
return _input_frame ? _input_frame(frame).cast<bool>() : false;
}
private:
py::object _py_muxer;
py::function _input_frame;
py::function _add_track;
py::function _add_track_completed;
};
PYBIND11_EMBEDDED_MODULE(mk_loader, m) {
m.def("log", [](int lev, const char *file, int line, const char *func, const char *content) {
py::gil_scoped_release release;
LoggerWrapper::printLog(::toolkit::getLogger(), lev, file, func, line, content);
});
m.def("get_config", [](const std::string &key) -> std::string {
py::gil_scoped_release release;
const auto it = mINI::Instance().find(key);
if (it != mINI::Instance().end()) {
return it->second;
}
return "";
});
m.def("get_full_path", [](const std::string &path, const std::string &current_path) -> std::string {
py::gil_scoped_release release;
switch (path.front()) {
case '/':
case '\\': return path;
default: return File::absolutePath(path, current_path);
}
}, py::arg("path"), py::arg("current_path") = "");
m.def("set_config", [](const std::string &key, const std::string &value) -> bool {
py::gil_scoped_release release;
mINI::Instance()[key]= value;
return true;
});
m.def("update_config", []() {
NOTICE_EMIT(BroadcastReloadConfigArgs, Broadcast::kBroadcastReloadConfig);
mINI::Instance().dumpFile(g_ini_file);
return true;
});
m.def("publish_auth_invoker_do", [](const py::capsule &cap, const std::string &err, const py::dict &opt) {
ProtocolOption option;
option.load(to_native(opt));
// 执行c++代码时释放gil锁
py::gil_scoped_release release;
auto &invoker = to_native<Broadcast::PublishAuthInvoker>(cap);
invoker(err, option);
});
m.def("play_auth_invoker_do", [](const py::capsule &cap, const std::string &err) {
// 执行c++代码时释放gil锁
py::gil_scoped_release release;
auto &invoker = to_native<Broadcast::AuthInvoker>(cap);
invoker(err);
});
m.def("rtsp_get_realm_invoker_do", [](const py::capsule &cap, const std::string &realm) {
// 执行c++代码时释放gil锁
py::gil_scoped_release release;
auto &invoker = to_native<RtspSession::onGetRealm>(cap);
invoker(realm);
});
m.def("rtsp_auth_invoker_do", [](const py::capsule &cap, bool encrypted, const std::string &pwd_or_md5) {
// 执行c++代码时释放gil锁
py::gil_scoped_release release;
auto &invoker = to_native<RtspSession::onAuth>(cap);
invoker(encrypted, pwd_or_md5);
});
m.def("close_player_invoker_do", [](const py::capsule &cap) {
// 执行c++代码时释放gil锁
py::gil_scoped_release release;
auto &invoker = to_native<std::function<void()>>(cap);
invoker();
});
m.def("http_access_invoker_do", [](const py::capsule &cap, const std::string &errMsg,const std::string &accessPath, int cookieLifeSecond) {
// 执行c++代码时释放gil锁
py::gil_scoped_release release;
auto &invoker = to_native<HttpSession::HttpAccessPathInvoker>(cap);
invoker(errMsg, accessPath, cookieLifeSecond);
});
// add_stream_proxy(vhost, app, stream, url, cb, retry_count=-1, force=False,
// rtp_type=0, timeout_sec=0, opt={})
// opt 字典可包含 ProtocolOption 的所有字段，以及其他透传给 Player 的 key-value 参数
m.def("add_stream_proxy",
[](const std::string &vhost, const std::string &app, const std::string &stream,
const std::string &url, const py::object &cb,
int retry_count, bool force, float timeout_sec,
const py::dict &opt) {
mINI args = to_native(opt);
ProtocolOption option;
option.load(args);
MediaTuple tuple { vhost.empty() ? DEFAULT_VHOST : vhost, app, stream, "" };
// 用 shared_ptr 包裹 py::object，使其析构（dec_ref）可在受控环境下执行。
// 必须在 GIL 持有期间创建该 shared_ptr（此处仍在 GIL 内）。
// 自定义 deleter 保证即使在非 Python 线程析构时也会先获取 GIL。
auto cb_ptr = std::shared_ptr<py::object>(
new py::object(cb),
[](py::object *p) {
// dec_ref / 析构 py::object 需要 GIL
py::gil_scoped_acquire guard;
delete p;
}
);
py::gil_scoped_release release;
EventPollerPool::Instance().getPoller(false)->async([=]() mutable {
addStreamProxy(tuple, url, retry_count, force, option, timeout_sec, args,
[cb_ptr](const SockException &ex, const std::string &key) {
// cb_ptr 按值捕获（shared_ptr 的拷贝，纯 C++ 操作，无需 GIL）
// inc_ref/dec_ref/调用 Python 对象均在 gil_scoped_acquire 保护下进行
py::gil_scoped_acquire guard;
try {
(*cb_ptr)(ex ? ex.what() : "", key);
} catch (py::error_already_set &e) {
WarnL << "Python exception in add_stream_proxy callback: " << e.what();
}
// cb_ptr 在此析构（局部副本），dec_ref 由自定义 deleter 在 GIL 下执行
});
});
},
py::arg("vhost"), py::arg("app"), py::arg("stream"), py::arg("url"), py::arg("cb"),
py::arg("retry_count") = -1, py::arg("force") = false,
py::arg("timeout_sec") = 0.0f, py::arg("opt") = py::dict()
);
// update_stream_proxy(vhost, app, stream, url, opt={})
// 更新已有拉流代理的 url 和参数，流不存在时抛出异常
m.def("update_stream_proxy",
[](const std::string &vhost, const std::string &app, const std::string &stream,
const std::string &url, const py::dict &opt) {
mINI args = to_native(opt);
MediaTuple tuple { vhost.empty() ? DEFAULT_VHOST : vhost, app, stream, "" };
py::gil_scoped_release release;
updateStreamProxy(tuple, url, args);
},
py::arg("vhost"), py::arg("app"), py::arg("stream"), py::arg("url"),
py::arg("opt") = py::dict()
);
m.def("set_fastapi", [](const py::object &check_route, const py::object &submit_coro) {
static void *fastapi_tag = nullptr;
NoticeCenter::Instance().delListener(&fastapi_tag, Broadcast::kBroadcastHttpRequest);
NoticeCenter::Instance().addListener(&fastapi_tag, Broadcast::kBroadcastHttpRequest, [check_route, submit_coro](BroadcastHttpRequestArgs) {
handle_http_request(check_route, submit_coro, parser, invoker, consumed, sender);
});
});
py::enum_<TrackType>(m, "TrackType")
.value("Invalid", TrackInvalid)
.value("Video", TrackVideo)
.value("Audio", TrackAudio)
.value("Title", TrackTitle)
.value("Application", TrackApplication)
.export_values();
py::class_<MediaSource, MediaSource::Ptr>(m, "MediaSource")
.def("getSchema", &MediaSource::getSchema)
.def("getUrl", &MediaSource::getUrl)
.def("getMediaTuple", &MediaSource::getMediaTuple)
.def("getTimeStamp", &MediaSource::getTimeStamp)
.def("setTimeStamp", &MediaSource::setTimeStamp)
.def("getBytesSpeed", &MediaSource::getBytesSpeed)
.def("getTotalBytes", &MediaSource::getTotalBytes)
.def("getCreateStamp", &MediaSource::getCreateStamp)
.def("getAliveSecond", &MediaSource::getAliveSecond)
.def("readerCount", &MediaSource::readerCount)
.def("totalReaderCount", &MediaSource::totalReaderCount)
.def("getOriginType", &MediaSource::getOriginType)
.def("getOriginUrl", &MediaSource::getOriginUrl)
.def("getOriginSock", &MediaSource::getOriginSock)
.def("seekTo", &MediaSource::seekTo)
.def("pause", &MediaSource::pause)
.def("speed", &MediaSource::speed)
.def("close", &MediaSource::close)
.def("setupRecord", &MediaSource::setupRecord)
.def("isRecording", &MediaSource::isRecording)
.def("stopSendRtp", &MediaSource::stopSendRtp)
.def("getLossRate", &MediaSource::getLossRate)
.def("getMuxer", &MediaSource::getMuxer);
py::class_<MediaTuple, std::shared_ptr<MediaTuple>>(m, "MediaTuple")
.def_readwrite("vhost", &MediaTuple::vhost)
.def_readwrite("app", &MediaTuple::app)
.def_readwrite("stream", &MediaTuple::stream)
.def_readwrite("params", &MediaTuple::params)
.def("shortUrl", &MediaTuple::shortUrl);
py::class_<SockException, std::shared_ptr<SockException>>(m, "SockException").def("what", &SockException::what).def("code", &SockException::getErrCode);
py::class_<Parser, std::shared_ptr<Parser>>(m, "Parser")
.def("method", &Parser::method)
.def("url", &Parser::url)
.def("status", &Parser::status)
.def("fullUrl", &Parser::fullUrl)
.def("protocol", &Parser::protocol)
.def("statusStr", &Parser::statusStr)
.def("content", &Parser::content)
.def("params", &Parser::params)
.def("getHeader", [](Parser *thiz) {
py::dict ret;
for (auto &pr : thiz->getHeader()) {
ret[pr.first.data()] = pr.second;
}
return ret;
});
py::enum_<Recorder::type>(m, "RecordType")
.value("hls", Recorder::type_hls)
.value("mp4", Recorder::type_mp4)
.value("hls_fmp4", Recorder::type_hls_fmp4)
.value("fmp4", Recorder::type_fmp4)
.value("ts", Recorder::type_ts)
.export_values();
#define OPT(key) .def_readwrite(#key, &ProtocolOption::key)
py::class_<ProtocolOption, std::shared_ptr<ProtocolOption>>(m, "ProtocolOption") OPT_VALUE(OPT);
#undef OPT
py::class_<MultiMediaSourceMuxer, std::shared_ptr<MultiMediaSourceMuxer>>(m, "MultiMediaSourceMuxer")
.def("totalReaderCount", static_cast<int (MultiMediaSourceMuxer::*)() const>(&MultiMediaSourceMuxer::totalReaderCount))
.def("isEnabled", &MultiMediaSourceMuxer::isEnabled)
.def("setupRecord", &MultiMediaSourceMuxer::setupRecord)
.def("startRecord", &MultiMediaSourceMuxer::startRecord)
.def("isRecording", &MultiMediaSourceMuxer::isRecording)
.def("startSendRtp", &MultiMediaSourceMuxer::startSendRtp)
.def("stopSendRtp", &MultiMediaSourceMuxer::stopSendRtp)
.def("getOption", &MultiMediaSourceMuxer::getOption)
.def("getMediaTuple", &MultiMediaSourceMuxer::getMediaTuple);
py::class_<Track, Track::Ptr>(m, "Track")
.def("getCodecId", &Track::getCodecId)
.def("getCodecName", &Track::getCodecName)
.def("getTrackType", &Track::getTrackType)
.def("getTrackTypeStr", &Track::getTrackTypeStr)
.def("setIndex", &Track::setIndex)
.def("getIndex", &Track::getIndex)
.def("getVideoKeyFrames", &Track::getVideoKeyFrames)
.def("getFrames", &Track::getFrames)
.def("getVideoGopSize", &Track::getVideoGopSize)
.def("getVideoGopInterval", &Track::getVideoGopInterval)
.def("getDuration", &Track::getDuration)
.def("ready", &Track::ready)
.def("update", &Track::update)
.def("getSdp", &Track::getSdp)
.def("getExtraData", &Track::getExtraData)
.def("setExtraData", &Track::setExtraData)
.def("getBitRate", &Track::getBitRate)
.def("setBitRate", &Track::setBitRate)
.def("getVideoHeight",[](Track *thiz) {
auto ptr = dynamic_cast<VideoTrack *>(thiz);
return ptr ? ptr->getVideoHeight() : 0;
})
.def("getVideoWidth", [](Track *thiz) {
auto ptr = dynamic_cast<VideoTrack *>(thiz);
return ptr ? ptr->getVideoWidth() : 0;
})
.def("getVideoFps", [](Track *thiz) {
auto ptr = dynamic_cast<VideoTrack *>(thiz);
return ptr ? ptr->getVideoFps() : 0;
})
.def("getAudioSampleRate",[](Track *thiz) {
auto ptr = dynamic_cast<AudioTrack *>(thiz);
return ptr ? ptr->getAudioSampleRate() : 0;
})
.def("getAudioSampleBit", [](Track *thiz) {
auto ptr = dynamic_cast<AudioTrack *>(thiz);
return ptr ? ptr->getAudioSampleBit() : 0;
})
.def("getAudioChannel", [](Track *thiz) {
auto ptr = dynamic_cast<AudioTrack *>(thiz);
return ptr ? ptr->getAudioChannel() : 0;
});
py::class_<Frame, Frame::Ptr>(m, "Frame")
.def("data", &Frame::data)
.def("size", &Frame::size)
.def("toString", &Frame::toString)
.def("getCapacity", &Frame::getCapacity)
.def("getCodecId", &Frame::getCodecId)
.def("getCodecName", &Frame::getCodecName)
.def("getTrackType", &Frame::getTrackType)
.def("getTrackTypeStr", &Frame::getTrackTypeStr)
.def("setIndex", &Frame::setIndex)
.def("getIndex", &Frame::getIndex)
.def("dts", &Frame::dts)
.def("pts", &Frame::pts)
.def("prefixSize", &Frame::prefixSize)
.def("keyFrame", &Frame::keyFrame)
.def("configFrame", &Frame::configFrame)
.def("cacheAble", &Frame::cacheAble)
.def("dropAble", &Frame::dropAble)
.def("decodeAble", &Frame::decodeAble);
}
namespace mediakit {
inline bool set_env(const char *name, const char *value) {
#if defined(_WIN32)
std::string env_str = std::string(name) + "=" + value;
return _putenv(env_str.c_str()) == 0;
#else
return setenv(name, value, 1) == 0; // overwrite = 1
#endif
}
bool set_python_path() {
const char *env_var = std::getenv("PYTHONPATH");
if (env_var && *env_var) {
PrintI("PYTHONPATH is already set to: %s", env_var);
return false;
}
auto default_path = exeDir() + "/python:" + exeDir() + "/pymkui/backend";
// 1 表示覆盖已存在的值
if (!set_env("PYTHONPATH", default_path.data())) {
PrintW("Failed to set PYTHONPATH");
return false;
}
PrintI("PYTHONPATH was not set. Set to default: %s", default_path.data());
return true;
}
static std::shared_ptr<PythonInvoker> g_instance;
PythonInvoker &PythonInvoker::Instance() {
static toolkit::onceToken s_token([]() {
g_instance.reset(new PythonInvoker);
});
return *g_instance;
}
void PythonInvoker::release() {
g_instance = nullptr;
}
PythonInvoker::PythonInvoker() {
// 确保日志一直可用
_logger = Logger::Instance().shared_from_this();
set_python_path(); // 确保 PYTHONPATH 在第一次调用时设置
_interpreter = new py::scoped_interpreter;
_rel = new py::gil_scoped_release;
NoticeCenter::Instance().addListener(this, Broadcast::kBroadcastReloadConfig, [this] (BroadcastReloadConfigArgs) {
py::gil_scoped_acquire guard;
if (_on_reload_config) {
_on_reload_config();
}
});
NoticeCenter::Instance().addListener(this, Broadcast::kBroadcastCreateMuxer, [this](BroadcastCreateMuxerArgs) {
py::gil_scoped_acquire guard;
if (_on_create_muxer) {
auto py_muxer = _on_create_muxer(to_python_ref(sender));
if (py_muxer && !py_muxer.is_none()) {
delegate = std::make_shared<MuxerDelegatePython>(std::move(py_muxer));
}
}
});
}
PythonInvoker::~PythonInvoker() {
NoticeCenter::Instance().delListener(this, Broadcast::kBroadcastReloadConfig);
{
py::gil_scoped_acquire gil; // 加锁
if (_on_exit) {
_on_exit();
}
_on_exit = py::function();
_on_publish = py::function();
_on_play = py::function();
_on_flow_report = py::function();
_on_reload_config = py::function();
_on_media_changed = py::function();
_on_player_proxy_failed = py::function();
_on_get_rtsp_realm = py::function();
_on_rtsp_auth = py::function();
_on_stream_not_found = py::function();
_on_record_mp4 = py::function();
_on_record_ts = py::function();
_on_stream_none_reader = py::function();
_on_send_rtp_stopped = py::function();
_on_http_access = py::function();
_on_rtp_server_timeout = py::function();
_on_create_muxer = py::function();
_module = py::module();
}
delete _rel;
delete _interpreter;
}
#define GET_FUNC(instance, name) \
if (hasattr(instance, #name)) { \
_##name = instance.attr(#name); \
}
void PythonInvoker::load(const std::string &module_name) {
try {
py::gil_scoped_acquire gil; // 加锁
_module = py::module::import(module_name.c_str());
GET_FUNC(_module, on_exit);
GET_FUNC(_module, on_publish);
GET_FUNC(_module, on_play);
GET_FUNC(_module, on_flow_report);
GET_FUNC(_module, on_reload_config);
GET_FUNC(_module, on_media_changed);
GET_FUNC(_module, on_player_proxy_failed);
GET_FUNC(_module, on_get_rtsp_realm);
GET_FUNC(_module, on_rtsp_auth);
GET_FUNC(_module, on_stream_not_found);
GET_FUNC(_module, on_record_mp4);
GET_FUNC(_module, on_record_ts);
GET_FUNC(_module, on_stream_none_reader);
GET_FUNC(_module, on_send_rtp_stopped);
GET_FUNC(_module, on_http_access);
GET_FUNC(_module, on_rtp_server_timeout);
GET_FUNC(_module, on_create_muxer);
if (hasattr(_module, "on_start")) {
py::object on_start = _module.attr("on_start");
if (on_start) {
on_start();
}
}
} catch (py::error_already_set &e) {
PrintE("Python exception:%s", e.what());
}
}
bool PythonInvoker::on_publish(BroadcastMediaPublishArgs) const {
py::gil_scoped_acquire gil; // 确保在 Python 调用期间持有 GIL
if (!_on_publish) {
return false;
}
return _on_publish(getOriginTypeString(type), to_python(args), to_python(invoker), to_python(sender)).cast<bool>();
}
bool PythonInvoker::on_play(BroadcastMediaPlayedArgs) const {
py::gil_scoped_acquire gil; // 确保在 Python 调用期间持有 GIL
if (!_on_play) {
return false;
}
return _on_play(to_python(args), to_python(invoker), to_python(sender)).cast<bool>();
}
bool PythonInvoker::on_flow_report(BroadcastFlowReportArgs) const {
py::gil_scoped_acquire gil; // 确保在 Python 调用期间持有 GIL
if (!_on_flow_report) {
return false;
}
return _on_flow_report(to_python(args), totalBytes, totalDuration, isPlayer, to_python(sender)).cast<bool>();
}
bool PythonInvoker::on_media_changed(BroadcastMediaChangedArgs) const {
py::gil_scoped_acquire gil; // 确保在 Python 调用期间持有 GIL
if (!_on_media_changed) {
return false;
}
return _on_media_changed(bRegist, to_python_ref(sender)).cast<bool>();
}
bool PythonInvoker::on_player_proxy_failed(BroadcastPlayerProxyFailedArgs) const {
py::gil_scoped_acquire gil; // 确保在 Python 调用期间持有 GIL
if (!_on_player_proxy_failed) {
return false;
}
return _on_player_proxy_failed(sender.getUrl(), to_python_ref(sender.getMediaTuple()), to_python_ref(ex)).cast<bool>();
}
bool PythonInvoker::on_get_rtsp_realm(BroadcastOnGetRtspRealmArgs) const {
py::gil_scoped_acquire gil; // 确保在 Python 调用期间持有 GIL
if (!_on_get_rtsp_realm) {
return false;
}
return _on_get_rtsp_realm(to_python(args), to_python(invoker), to_python(sender)).cast<bool>();
}
bool PythonInvoker::on_rtsp_auth(BroadcastOnRtspAuthArgs) const {
py::gil_scoped_acquire gil; // 确保在 Python 调用期间持有 GIL
if (!_on_rtsp_auth) {
return false;
}
return _on_rtsp_auth(to_python(args), realm, user_name, must_no_encrypt, to_python(invoker), to_python(sender)).cast<bool>();
}
bool PythonInvoker::on_stream_not_found(BroadcastNotFoundStreamArgs) const {
py::gil_scoped_acquire gil; // 确保在 Python 调用期间持有 GIL
if (!_on_stream_not_found) {
return false;
}
return _on_stream_not_found(to_python(args), to_python(sender), to_python(closePlayer)).cast<bool>();
}
bool PythonInvoker::on_record_mp4(BroadcastRecordMP4Args) const {
py::gil_scoped_acquire gil; // 确保在 Python 调用期间持有 GIL
if (!_on_record_mp4) {
return false;
}
return _on_record_mp4(to_python(info)).cast<bool>();
}
bool PythonInvoker::on_record_ts(BroadcastRecordTsArgs) const {
py::gil_scoped_acquire gil; // 确保在 Python 调用期间持有 GIL
if (!_on_record_ts) {
return false;
}
return _on_record_ts(to_python(info)).cast<bool>();
}
bool PythonInvoker::on_stream_none_reader(BroadcastStreamNoneReaderArgs) const {
py::gil_scoped_acquire gil; // 确保在 Python 调用期间持有 GIL
if (!_on_stream_none_reader) {
return false;
}
return _on_stream_none_reader(to_python_ref(sender)).cast<bool>();
}
bool PythonInvoker::on_send_rtp_stopped(BroadcastSendRtpStoppedArgs) const {
py::gil_scoped_acquire gil; // 确保在 Python 调用期间持有 GIL
if (!_on_send_rtp_stopped) {
return false;
}
return _on_send_rtp_stopped(to_python_ref(sender), ssrc, to_python_ref(ex)).cast<bool>();
}
bool PythonInvoker::on_http_access(BroadcastHttpAccessArgs) const {
py::gil_scoped_acquire gil; // 确保在 Python 调用期间持有 GIL
if (!_on_http_access) {
return false;
}
return _on_http_access(to_python_ref(parser), path, file_path, is_dir, to_python(invoker), to_python(sender)).cast<bool>();
}
bool PythonInvoker::on_rtp_server_timeout(BroadcastRtpServerTimeoutArgs) const {
py::gil_scoped_acquire gil; // 确保在 Python 调用期间持有 GIL
if (!_on_rtp_server_timeout) {
return false;
}
return _on_rtp_server_timeout(local_port, to_python_ref(tuple), tcp_mode, re_use_port, ssrc).cast<bool>();
}
} // namespace mediakit
#endif

95
server/pyinvoker.h Normal file
View File

@@ -0,0 +1,95 @@
#ifndef PYINVOKER_H
#define PYINVOKER_H
#if defined(ENABLE_PYTHON)
#include <map>
#include <string>
#include <pybind11/embed.h>
#include <pybind11/numpy.h>
#include "Util/logger.h"
#include "Common/config.h"
#include "Common/MediaSource.h"
#include "Player/PlayerProxy.h"
#include "Rtsp/RtspSession.h"
#include "Http/HttpSession.h"
namespace py = pybind11;
namespace mediakit {
class PythonInvoker : public std::enable_shared_from_this<PythonInvoker>{
public:
~PythonInvoker();
static PythonInvoker& Instance();
static void release();
void load(const std::string &module_name);
bool on_publish(BroadcastMediaPublishArgs) const;
bool on_play(BroadcastMediaPlayedArgs) const;
bool on_flow_report(BroadcastFlowReportArgs) const;
bool on_media_changed(BroadcastMediaChangedArgs) const;
bool on_player_proxy_failed(BroadcastPlayerProxyFailedArgs) const;
bool on_get_rtsp_realm(BroadcastOnGetRtspRealmArgs) const;
bool on_rtsp_auth(BroadcastOnRtspAuthArgs) const;
bool on_stream_not_found(BroadcastNotFoundStreamArgs) const;
bool on_record_mp4(BroadcastRecordMP4Args) const;
bool on_record_ts(BroadcastRecordTsArgs) const;
bool on_stream_none_reader(BroadcastStreamNoneReaderArgs) const;
bool on_send_rtp_stopped(BroadcastSendRtpStoppedArgs) const;
bool on_http_access(BroadcastHttpAccessArgs) const;
bool on_rtp_server_timeout(BroadcastRtpServerTimeoutArgs) const;
private:
PythonInvoker();
private:
py::gil_scoped_release *_rel;
py::scoped_interpreter *_interpreter;
std::shared_ptr<toolkit::Logger> _logger;
py::module _module;
// 程序退出
py::function _on_exit;
// 推流鉴权
py::function _on_publish;
// 播放鉴权
py::function _on_play;
// 流量汇报接口
py::function _on_flow_report;
// 配置文件热更新回调
py::function _on_reload_config;
// 媒体注册注销
py::function _on_media_changed;
// 拉流代理失败
py::function _on_player_proxy_failed;
// rtsp播放是否开启专属鉴权
py::function _on_get_rtsp_realm;
// rtsp播放或推流鉴权回调
py::function _on_rtsp_auth;
// 播放一个不存在的流时触发
py::function _on_stream_not_found;
// 生成mp4录制文件回调
py::function _on_record_mp4;
// 生成hls ts/fmp4切片文件回调
py::function _on_record_ts;
// 流无人观看事件
py::function _on_stream_none_reader;
// rtp转发失败事件
py::function _on_send_rtp_stopped;
// http访问鉴权事件
py::function _on_http_access;
// rtp服务收流超时事件
py::function _on_rtp_server_timeout;
// 创建Python muxer对象
py::function _on_create_muxer;
};
} // namespace mediakit
#endif
#endif // PYINVOKER_H