新增支持Python混合编程模式 (#4579)
Some checks failed
Android / build (push) Has been cancelled
CodeQL / Analyze (cpp) (push) Has been cancelled
CodeQL / Analyze (javascript) (push) Has been cancelled
Docker / build (push) Has been cancelled
Linux / build (push) Has been cancelled
Linux_Python / build (push) Has been cancelled
macOS / build (push) Has been cancelled
macOS_Python / build (push) Has been cancelled
Windows / build (push) Has been cancelled
Windows_Python / build (push) Has been cancelled

This commit is contained in:
夏楚 2026-02-10 13:28:42 +08:00 committed by GitHub
parent da9deb352c
commit 6d520ea6a3
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
25 changed files with 1553 additions and 55 deletions

172
.github/workflows/linux_py.yml vendored Normal file
View File

@ -0,0 +1,172 @@
name: Linux_Python
on: [push, pull_request]
jobs:
build:
runs-on: ubuntu-24.04
steps:
- uses: actions/checkout@v1
- name: 下载submodule源码
run: mv -f .gitmodules_github .gitmodules && git submodule sync && git submodule update --init
- name: 下载 SRTP
uses: actions/checkout@v2
with:
repository: cisco/libsrtp
fetch-depth: 1
ref: v2.3.0
path: 3rdpart/libsrtp
- name: 下载 openssl
uses: actions/checkout@v2
with:
repository: openssl/openssl
fetch-depth: 1
ref: OpenSSL_1_1_1
path: 3rdpart/openssl
- name: 下载 usrsctp
uses: actions/checkout@v2
with:
repository: sctplab/usrsctp
fetch-depth: 1
ref: 0.9.5.0
path: 3rdpart/usrsctp
- name: 启动 Docker 容器, 在Docker 容器中执行脚本
run: |
docker pull centos:7
docker run -v $(pwd):/root -w /root --rm centos:7 sh -c "
#!/bin/bash
set -x
# Backup original CentOS-Base.repo file
cp /etc/yum.repos.d/CentOS-Base.repo /etc/yum.repos.d/CentOS-Base.repo.backup
# Define new repository configuration
cat <<EOF > /etc/yum.repos.d/CentOS-Base.repo
[base]
name=CentOS-7 - Base - mirrors.aliyun.com
baseurl=http://mirrors.aliyun.com/centos/7/os/x86_64/
gpgcheck=1
gpgkey=http://mirrors.aliyun.com/centos/RPM-GPG-KEY-CentOS-7
[updates]
name=CentOS-7 - Updates - mirrors.aliyun.com
baseurl=http://mirrors.aliyun.com/centos/7/updates/x86_64/
gpgcheck=1
gpgkey=http://mirrors.aliyun.com/centos/RPM-GPG-KEY-CentOS-7
EOF
cat > /etc/yum.repos.d/epel-aliyun.repo <<EOF
[epel]
name=Extra Packages for Enterprise Linux 7 - x86_64
baseurl=http://mirrors.aliyun.com/epel/7/x86_64/
enabled=1
gpgcheck=0
EOF
cat > /etc/yum.repos.d/CentOS-SCLo-aliyun.repo <<EOF
[C7-SCLo-rh]
name=CentOS-7 SCLo RH - x86_64
baseurl=http://mirrors.aliyun.com/centos/7/sclo/x86_64/rh/
enabled=1
gpgcheck=0
EOF
# Clean yum cache and recreate it
yum clean all
yum makecache
echo \"CentOS 7 软件源已成功切换\"
yum install -y git wget gcc gcc-c++ make which devtoolset-11
# === 1. 下载并静默安装 Miniconda ===
wget -q https://repo.anaconda.com/miniconda/Miniconda3-py39_23.3.1-0-Linux-x86_64.sh -O miniconda.sh
bash miniconda.sh -b -p "$HOME/miniconda" # -b 表示 batch静默安装
export PATH="$HOME/miniconda/bin:$PATH"
# === 2. 初始化 conda非交互模式 ===
source "$HOME/miniconda/etc/profile.d/conda.sh"
# === 3. 创建 Python 3.11 环境 ===
conda create -n py11 python=3.11 -y
# === 4. 激活环境 ===
conda activate py11
# === 5. 验证环境 ===
python --version
# === 6. 安装必要模块 ===
conda install -y pip setuptools jinja2 wheel
mkdir -p /root/install
cd 3rdpart/openssl
./config no-shared --prefix=/root/install
make -j $(nproc)
make install
cd ../../
wget https://github.com/Kitware/CMake/releases/download/v3.29.5/cmake-3.29.5.tar.gz
tar -xf cmake-3.29.5.tar.gz
cd cmake-3.29.5
OPENSSL_ROOT_DIR=/root/install ./configure
make -j $(nproc)
make install
cd ..
cd 3rdpart/usrsctp
mkdir build
cd build
cmake -DCMAKE_BUILD_TYPE=Release -DCMAKE_POSITION_INDEPENDENT_CODE=ON ..
make -j $(nproc)
make install
cd ../../../
cd 3rdpart/libsrtp && ./configure --enable-openssl --with-openssl-dir=/root/install && make -j $(nproc) && make install
cd ../../
source /opt/rh/devtoolset-11/enable
gcc --version
mkdir -p linux_build && cd linux_build && cmake .. -DENABLE_PYTHON=ON -DPYTHON_EXECUTABLE=$(which python3.14) -DOPENSSL_ROOT_DIR=/root/install -DCMAKE_BUILD_TYPE=Release && make -j $(nproc)
"
- name: 设置环境变量
run: |
echo "BRANCH=$(echo ${GITHUB_REF#refs/heads/} | tr -s "/\?%*:|\"<>" "_")" >> $GITHUB_ENV
echo "BRANCH2=$(echo ${GITHUB_REF#refs/heads/} )" >> $GITHUB_ENV
echo "DATE=$(date +%Y-%m-%d)" >> $GITHUB_ENV
- name: 打包二进制
id: upload
uses: actions/upload-artifact@v4
with:
name: ${{ github.workflow }}_${{ env.BRANCH }}_${{ env.DATE }}
path: release/*
if-no-files-found: error
retention-days: 90
- name: issue评论
if: github.event_name != 'pull_request' && github.ref != 'refs/heads/feature/test'
uses: actions/github-script@v7
with:
github-token: ${{ secrets.GITHUB_TOKEN }}
script: |
github.rest.issues.createComment({
issue_number: ${{vars.VERSION_ISSUE_NO}},
owner: context.repo.owner,
repo: context.repo.repo,
body: '- 下载地址: [${{ github.workflow }}_${{ env.BRANCH }}_${{ env.DATE }}](${{ steps.upload.outputs.artifact-url }})\n'
+ '- 分支: ${{ env.BRANCH2 }}\n'
+ '- git hash: ${{ github.sha }} \n'
+ '- 编译日期: ${{ env.DATE }}\n'
+ '- 编译记录: [${{ github.run_id }}](https://github.com/${{ github.repository }}/actions/runs/${{ github.run_id }})\n'
+ '- 打包ci名: ${{ github.workflow }}\n'
+ '- 开启特性: openssl/webrtc/datachannel\n'
+ '- 说明: 本二进制在centos7(x64)上编译请确保您的机器系统不低于此版本本程序依赖python3.11, 运行前请miniconda安装python3.11\n'
})

80
.github/workflows/macos_py.yml vendored Normal file
View File

@ -0,0 +1,80 @@
name: macOS_Python
on: [push, pull_request]
jobs:
build:
runs-on: macOS-latest
steps:
- uses: actions/checkout@v1
- name: 下载submodule源码
run: mv -f .gitmodules_github .gitmodules && git submodule sync && git submodule update --init
- name: 配置 vcpkg
uses: lukka/run-vcpkg@v7
with:
vcpkgDirectory: '${{github.workspace}}/vcpkg'
vcpkgTriplet: arm64-osx
# 2025.07.11
vcpkgGitCommitId: 'efcfaaf60d7ec57a159fc3110403d939bfb69729'
vcpkgArguments: 'openssl libsrtp[openssl] usrsctp'
- name: 安装指定 CMake
uses: jwlawson/actions-setup-cmake@v2
with:
cmake-version: '3.30.5'
- name: 检查并设置 Python 3
run: |
PYTHON_ROOT=$(python3 -c "import sys; print(sys.prefix)")
echo "PYTHON_ROOT=$PYTHON_ROOT" >> $GITHUB_ENV
PYTHON_EXECUTABLE=$(which python3)
echo "PYTHON_EXECUTABLE=$PYTHON_EXECUTABLE" >> $GITHUB_ENV
- name: 编译
uses: lukka/run-cmake@v3
with:
useVcpkgToolchainFile: true
cmakeBuildType: Release
cmakeListsOrSettingsJson: CMakeListsTxtAdvanced
buildDirectory: '${{github.workspace}}/build'
buildWithCMakeArgs: '--config Release'
cmakeAppendedArgs: '-DPYTHON_EXECUTABLE=${{ env.PYTHON_EXECUTABLE }} -DENABLE_PYTHON=ON -DENABLE_API=OFF -DENABLE_TESTS=OFF -DCMAKE_BUILD_TYPE=Release'
- name: 设置环境变量
run: |
echo "BRANCH=$(echo ${GITHUB_REF#refs/heads/} | tr -s "/\?%*:|\"<>" "_")" >> $GITHUB_ENV
echo "BRANCH2=$(echo ${GITHUB_REF#refs/heads/} )" >> $GITHUB_ENV
echo "DATE=$(date +%Y-%m-%d)" >> $GITHUB_ENV
- name: 打包二进制
id: upload
uses: actions/upload-artifact@v4
with:
name: ${{ github.workflow }}_${{ env.BRANCH }}_${{ env.DATE }}
path: release/*
if-no-files-found: error
retention-days: 90
- name: issue评论
if: github.event_name != 'pull_request' && github.ref != 'refs/heads/feature/test'
uses: actions/github-script@v7
with:
github-token: ${{ secrets.GITHUB_TOKEN }}
script: |
github.rest.issues.createComment({
issue_number: ${{vars.VERSION_ISSUE_NO}},
owner: context.repo.owner,
repo: context.repo.repo,
body: '- 下载地址: [${{ github.workflow }}_${{ env.BRANCH }}_${{ env.DATE }}](${{ steps.upload.outputs.artifact-url }})\n'
+ '- 分支: ${{ env.BRANCH2 }}\n'
+ '- git hash: ${{ github.sha }} \n'
+ '- 编译日期: ${{ env.DATE }}\n'
+ '- 编译记录: [${{ github.run_id }}](https://github.com/${{ github.repository }}/actions/runs/${{ github.run_id }})\n'
+ '- 打包ci名: ${{ github.workflow }}\n'
+ '- 开启特性: openssl/webrtc/datachannel\n'
+ '- 说明: 此二进制为arm64版本; 本程序依赖python3.14, 运行前请brew install python@3.14安装\n'
})

86
.github/workflows/windows_py.yml vendored Normal file
View File

@ -0,0 +1,86 @@
name: Windows_Python
on: [push, pull_request]
jobs:
build:
runs-on: windows-2022
steps:
- uses: actions/checkout@v1
- name: 下载submodule源码
run: mv -Force .gitmodules_github .gitmodules && git submodule sync && git submodule update --init
- name: 配置 vcpkg
uses: lukka/run-vcpkg@v7
with:
vcpkgDirectory: '${{github.workspace}}/vcpkg'
vcpkgTriplet: x64-windows-static
# 2025.07.11
vcpkgGitCommitId: 'efcfaaf60d7ec57a159fc3110403d939bfb69729'
vcpkgArguments: 'openssl libsrtp[openssl] usrsctp'
- name: Setup Python 3.14
uses: actions/setup-python@v4
with:
python-version: 3.14
architecture: x64
- name: Set PYTHON_EXECUTABLE
shell: pwsh
run: |
$pythonExe = python -c "import sys; print(sys.executable)"
Add-Content -Path $Env:GITHUB_ENV -Value "PYTHON_EXECUTABLE=$pythonExe"
- name: Check PYTHON_EXECUTABLE
run: echo $Env:PYTHON_EXECUTABLE
shell: pwsh
- name: 编译
uses: lukka/run-cmake@v3
with:
useVcpkgToolchainFile: true
cmakeBuildType: Release
cmakeListsOrSettingsJson: CMakeListsTxtAdvanced
buildDirectory: '${{github.workspace}}/build'
buildWithCMakeArgs: '--config Release'
cmakeAppendedArgs: '-DPYTHON_EXECUTABLE=${{ env.PYTHON_EXECUTABLE }} -DENABLE_PYTHON=ON -DENABLE_API=OFF -DENABLE_TESTS=OFF -DCMAKE_BUILD_TYPE=Release'
- name: 设置环境变量
run: |
$dateString = Get-Date -Format "yyyy-MM-dd"
$branch = $env:GITHUB_REF -replace "refs/heads/", "" -replace "[\\/\\\?\%\*:\|\x22<>]", "_"
$branch2 = $env:GITHUB_REF -replace "refs/heads/", ""
echo "BRANCH=$branch" >> $env:GITHUB_ENV
echo "BRANCH2=$branch2" >> $env:GITHUB_ENV
echo "DATE=$dateString" >> $env:GITHUB_ENV
- name: 打包二进制
id: upload
uses: actions/upload-artifact@v4
with:
name: ${{ github.workflow }}_${{ env.BRANCH }}_${{ env.DATE }}
path: release/*
if-no-files-found: error
retention-days: 90
- name: issue评论
if: github.event_name != 'pull_request' && github.ref != 'refs/heads/feature/test'
uses: actions/github-script@v7
with:
github-token: ${{ secrets.GITHUB_TOKEN }}
script: |
github.rest.issues.createComment({
issue_number: ${{vars.VERSION_ISSUE_NO}},
owner: context.repo.owner,
repo: context.repo.repo,
body: '- 下载地址: [${{ github.workflow }}_${{ env.BRANCH }}_${{ env.DATE }}](${{ steps.upload.outputs.artifact-url }})\n'
+ '- 分支: ${{ env.BRANCH2 }}\n'
+ '- git hash: ${{ github.sha }} \n'
+ '- 编译日期: ${{ env.DATE }}\n'
+ '- 编译记录: [${{ github.run_id }}](https://github.com/${{ github.repository }}/actions/runs/${{ github.run_id }})\n'
+ '- 打包ci名: ${{ github.workflow }}\n'
+ '- 开启特性: openssl/webrtc/datachannel\n'
+ '- 说明: 此二进制为x64版本;本程序依赖python3.14, 运行前请先安装python3.14\n'
})

6
.gitmodules vendored
View File

@ -10,3 +10,9 @@
[submodule "www/webassist"]
path = www/webassist
url = https://gitee.com/victor1002/zlm_webassist
[submodule "3rdpart/pybind11"]
path = 3rdpart/pybind11
url = https://gitee.com/mirrors/pybind11.git
[submodule "python/StreamUI"]
path = python/StreamUI
url = https://gitee.com/xia-chu/StreamUI.git

View File

@ -10,3 +10,9 @@
[submodule "www/webassist"]
path = www/webassist
url = https://github.com/1002victor/zlm_webassist
[submodule "3rdpart/pybind11"]
path = 3rdpart/pybind11
url = https://github.com/pybind/pybind11.git
[submodule "python/StreamUI"]
path = python/StreamUI
url = https://github.com/xia-chu/StreamUI.git

View File

@ -121,3 +121,13 @@ add_subdirectory(ZLToolKit)
add_library(ZLMediaKit::ToolKit ALIAS ZLToolKit)
#
update_cached_list(MK_LINK_LIBRARIES ZLMediaKit::ToolKit)
##############################################################################
if (ENABLE_PYTHON)
# ============ pybind11 lib ============
add_subdirectory(pybind11)
update_cached_list(MK_LINK_LIBRARIES pybind11::embed)
include_directories(${CMAKE_CURRENT_SOURCE_DIR}/pybind11/include)
update_cached_list(MK_COMPILE_DEFINITIONS ENABLE_PYTHON)
endif ()

1
3rdpart/pybind11 Submodule

@ -0,0 +1 @@
Subproject commit ed5057ded698e305210269dafa57574ecf964483

View File

@ -65,6 +65,7 @@ option(USE_SOLUTION_FOLDERS "Enable solution dir supported" ON)
option(ENABLE_OBJCOPY "Enable use objcopy to generate debug info file" ON)
#
option(BUILD_SHARED_LIBS "Build shared instead of static" OFF)
option(ENABLE_PYTHON "Enable python plugin" OFF)
##############################################################################
# socket256k.0socket,使用系统内核默认值(设置为0仅对linux有效)
@ -603,6 +604,16 @@ endif ()
file(COPY "${CMAKE_CURRENT_SOURCE_DIR}/www" DESTINATION ${EXECUTABLE_OUTPUT_PATH})
file(COPY "${CMAKE_CURRENT_SOURCE_DIR}/conf/config.ini" DESTINATION ${EXECUTABLE_OUTPUT_PATH})
file(COPY "${CMAKE_CURRENT_SOURCE_DIR}/default.pem" DESTINATION ${EXECUTABLE_OUTPUT_PATH})
if (ENABLE_PYTHON)
file(COPY "${CMAKE_CURRENT_SOURCE_DIR}/python" DESTINATION ${EXECUTABLE_OUTPUT_PATH})
file(GLOB FRONTEND_FILES
"${CMAKE_CURRENT_SOURCE_DIR}/python/StreamUI/frontend/*"
)
file(COPY ${FRONTEND_FILES}
DESTINATION "${EXECUTABLE_OUTPUT_PATH}/www/StreamUI/"
)
endif ()
if (ENABLE_FFMPEG)
file(COPY "${CMAKE_CURRENT_SOURCE_DIR}/DejaVuSans.ttf" DESTINATION ${EXECUTABLE_OUTPUT_PATH})
endif ()

View File

@ -1,5 +1,5 @@
FROM ubuntu:20.04 AS build
ARG MODEL
FROM ubuntu:24.04 AS build
ARG MODEL=Release
#shell,rtmp,rtsp,rtsps,http,https,rtp
EXPOSE 1935/tcp
EXPOSE 554/tcp
@ -27,6 +27,7 @@ RUN apt-get update && \
libssl-dev \
gcc \
g++ \
python3-dev \
gdb && \
apt-get autoremove -y && \
apt-get clean -y && \
@ -41,17 +42,17 @@ WORKDIR /opt/media/ZLMediaKit/3rdpart
RUN wget https://github.com/cisco/libsrtp/archive/v2.3.0.tar.gz -O libsrtp-2.3.0.tar.gz && \
tar xfv libsrtp-2.3.0.tar.gz && \
mv libsrtp-2.3.0 libsrtp && \
cd libsrtp && ./configure --enable-openssl && make -j $(nproc) && make install
cd libsrtp && CFLAGS="-fcommon" ./configure --enable-openssl && make -j $(nproc) && make install
#RUN git submodule update --init --recursive && \
RUN mkdir -p build release/linux/${MODEL}/
WORKDIR /opt/media/ZLMediaKit/build
RUN cmake -DCMAKE_BUILD_TYPE=${MODEL} -DENABLE_WEBRTC=true -DENABLE_FFMPEG=true -DENABLE_TESTS=false -DENABLE_API=false .. && \
RUN cmake -DENABLE_PYTHON=true -DCMAKE_BUILD_TYPE=${MODEL} -DENABLE_WEBRTC=true -DENABLE_FFMPEG=true -DENABLE_TESTS=false -DENABLE_API=false .. && \
make -j $(nproc)
FROM ubuntu:20.04
ARG MODEL
FROM ubuntu:24.04
ARG MODEL=Release
# ADD sources.list /etc/apt/sources.list
@ -67,6 +68,10 @@ RUN apt-get update && \
ffmpeg \
gcc \
g++ \
python3 \
python3-dev \
python3-venv \
python3-pip \
gdb && \
apt-get autoremove -y && \
apt-get clean -y && \

1
python/StreamUI Submodule

@ -0,0 +1 @@
Subproject commit d055e925a5c061d3ce785f60f07b59ded08b4e55

27
python/mk_logger.py Normal file
View File

@ -0,0 +1,27 @@
import inspect
try:
import mk_loader
USE_PLUGIN_LOGGER = True
except ImportError:
USE_PLUGIN_LOGGER = False
def _do_log(level: int, *args):
frame_info = inspect.stack()[2]
filename = frame_info.filename
lineno = frame_info.lineno
funcname = frame_info.function
# 把所有参数转成字符串后用空格拼接
msg = " ".join(str(arg) for arg in args)
if USE_PLUGIN_LOGGER:
mk_loader.log(level, filename, lineno, funcname, msg)
else:
print(f"[{filename}:{lineno}] {funcname} | {msg}")
def log_trace(*args): _do_log(0, *args)
def log_debug(*args): _do_log(1, *args)
def log_info(*args): _do_log(2, *args)
def log_warn(*args): _do_log(3, *args)
def log_error(*args): _do_log(4, *args)

168
python/mk_plugin.py Normal file
View File

@ -0,0 +1,168 @@
import mk_logger
import mk_loader
import asyncio
import threading
from StreamUI.backend.main import app
from starlette.routing import Match
def start_background_loop(loop):
asyncio.set_event_loop(loop)
loop.run_forever()
loop = asyncio.new_event_loop()
threading.Thread(target=start_background_loop, args=(loop,), daemon=True).start()
def submit_coro(scope, body, send):
async def run():
# 包装 send 函数,确保它总是可等待的
async def async_send(message):
# 调用原始的 send 函数,它现在应该返回一个协程
result = send(message)
if result is not None:
await result
async def receive():
return {
"type": "http.request",
"body": body,
"more_body": False,
}
try:
await app(scope, receive, async_send)
except Exception as e:
mk_logger.log_warn(f"FastAPI failed: {e}")
# 发送错误响应
await async_send({
"type": "http.response.start",
"status": 500,
"headers": [(b"content-type", b"text/plain")],
})
await async_send({
"type": "http.response.body",
"body": b"Internal Server Error",
"more_body": False,
})
return asyncio.run_coroutine_threadsafe(run(), loop)
def check_route(scope) -> bool:
for route in app.routes:
if hasattr(route, "matches"):
match, _ = route.matches(scope)
if match == Match.FULL:
return True
return False
def on_start():
mk_logger.log_info(f"on_start, secret: {mk_loader.get_config('api.secret')}")
# mk_loader.set_config('api.secret', "new_secret_from_python")
# mk_loader.update_config()
mk_loader.set_fastapi(check_route, submit_coro)
def on_exit():
mk_logger.log_info("on_exit")
def on_publish(type: str, args: dict, invoker, sender: dict) -> bool:
mk_logger.log_info(f"type: {type}, args: {args}, sender: {sender}")
# opt 控制转协议,请参考配置文件[protocol]下字段
opt = {
"enable_rtmp": "1"
}
# 响应推流鉴权结果
mk_loader.publish_auth_invoker_do(invoker, "", opt)
# 返回True代表此事件被python拦截
return True
def on_play(args: dict, invoker, sender: dict) -> bool:
mk_logger.log_info(f"args: {args}, sender: {sender}")
# 响应播放鉴权结果
mk_loader.play_auth_invoker_do(invoker, "")
# 返回True代表此事件被python拦截
return True
def on_flow_report(args: dict, totalBytes: int, totalDuration: int, isPlayer: bool, sender: dict) -> bool:
mk_logger.log_info(f"args: {args}, totalBytes: {totalBytes}, totalDuration: {totalDuration}, isPlayer: {isPlayer}, sender: {sender}")
# 返回True代表此事件被python拦截
return True
def on_media_changed(is_register: bool, sender: mk_loader.MediaSource) -> bool:
mk_logger.log_info(f"is_register: {is_register}, sender: {sender.getUrl()}")
# 该事件在c++中也处理下
return False
def on_player_proxy_failed(url: str, media_tuple: mk_loader.MediaTuple , ex: mk_loader.SockException) -> bool:
mk_logger.log_info(f"on_player_proxy_failed: {url}, {media_tuple.shortUrl()}, {ex.what()}")
# 该事件在c++中也处理下
return False
def on_get_rtsp_realm(args: dict, invoker, sender: dict) -> bool:
mk_logger.log_info(f"on_get_rtsp_realm, args: {args}, sender: {sender}")
mk_loader.rtsp_get_realm_invoker_do(invoker, "zlmediakit")
# 返回True代表此事件被python拦截
return True
def on_rtsp_auth(args: dict, realm: str, user_name: str, must_no_encrypt: bool, invoker, sender:dict) -> bool:
mk_logger.log_info(f"on_rtsp_auth, args: {args}, realm: {realm}, user_name: {user_name}, must_no_encrypt: {must_no_encrypt}, sender: {sender}")
mk_loader.rtsp_auth_invoker_do(invoker, False, "zlmediakit")
# 返回True代表此事件被python拦截
return True
def on_stream_not_found(args: dict, sender:dict, invoker) -> bool:
mk_logger.log_info(f"on_stream_not_found, args: {args}, sender: {sender}")
# 立即通知播放器流不存在并关闭
mk_loader.close_player_invoker_do(invoker)
# 返回True代表此事件被python拦截
return True
def on_record_mp4(info: dict) -> bool:
mk_logger.log_info(f"on_record_mp4, info: {info}")
# 返回True代表此事件被python拦截
return True
def on_record_ts(info: dict) -> bool:
mk_logger.log_info(f"on_record_ts, info: {info}")
# 返回True代表此事件被python拦截
return True
def on_stream_none_reader(sender: mk_loader.MediaSource) -> bool:
mk_logger.log_info(f"on_stream_none_reader: {sender.getUrl()}")
# 无人观看自动关闭
sender.close(False)
# 返回True代表此事件被python拦截
return True
def on_send_rtp_stopped(sender: mk_loader.MultiMediaSourceMuxer, ssrc: str, ex: mk_loader.SockException) -> bool:
mk_logger.log_info(f"on_send_rtp_stopped, ssrc: {ssrc}, ex: {ex.what()}, url: {sender.getMediaTuple().getUrl()}")
# 返回True代表此事件被python拦截
return True
def on_http_access(parser: mk_loader.Parser, path: str, is_dir: bool, invoker, sender: dict) -> bool:
mk_logger.log_info(f"on_http_access, path: {path}, is_dir: {is_dir}, sender: {sender}, http header: {parser.getHeader()}")
# 允许访问该文件/目录1小时, cookie有效期内访问该目录下的文件或路径不再触发该事件
mk_loader.http_access_invoker_do(invoker, "", path, 60 * 60)
# 返回True代表此事件被python拦截
return True
def on_rtp_server_timeout(local_port: int, tuple: mk_loader.MediaTuple, tcp_mode: int, re_use_port: bool, ssrc: int) -> bool:
mk_logger.log_info(f"on_rtp_server_timeout, local_port: {local_port}, tuple: {tuple.shortUrl()}, tcp_mode: {tcp_mode}, re_use_port: {re_use_port}, ssrc: {ssrc}")
# 返回True代表此事件被python拦截
return True
def on_reload_config():
mk_logger.log_info(f"on_reload_config")
class PyMultiMediaSourceMuxer:
def __init__(self, sender: mk_loader.MultiMediaSourceMuxer):
mk_logger.log_info(f"PyMultiMediaSourceMuxer: {sender.getMediaTuple().shortUrl()}")
def destroy(self):
mk_logger.log_info(f"~PyMultiMediaSourceMuxer")
def addTrack(self, track: mk_loader.Track):
mk_logger.log_info(f"addTrack: {track.getCodecName()}")
return True
def addTrackCompleted(self):
mk_logger.log_info(f"addTrackCompleted")
def inputFrame(self, frame: mk_loader.Frame):
mk_logger.log_info(f"inputFrame: {frame.getCodecName()} {frame.dts()}")
return True
def on_create_muxer(sender: mk_loader.MultiMediaSourceMuxer):
return PyMultiMediaSourceMuxer(sender)

View File

@ -115,8 +115,6 @@ static void responseApi(int code, const string &msg, const HttpSession::HttpResp
responseApi(res, invoker);
}
static ApiArgsType getAllArgs(const Parser &parser);
static HttpApi toApi(const function<void(API_ARGS_MAP_ASYNC)> &cb) {
return [cb](const Parser &parser, const HttpSession::HttpResponseInvoker &invoker, SockInfo &sender) {
GET_CONFIG(string, charSet, Http::kCharSet);
@ -215,7 +213,7 @@ void api_regist(const string &api_path, const function<void(API_ARGS_STRING_ASYN
// 获取HTTP请求中url参数、content参数 [AUTO-TRANSLATED:d161a1e1]
// Get URL parameters and content parameters from the HTTP request
static ApiArgsType getAllArgs(const Parser &parser) {
ApiArgsType getAllArgs(const Parser &parser) {
ApiArgsType allArgs;
if (parser["Content-Type"].find("application/x-www-form-urlencoded") == 0) {
auto contentArgs = parser.parseArgs(parser.content());
@ -345,7 +343,7 @@ static inline string getPusherKey(const string &schema, const string &vhost, con
return schema + "/" + vhost + "/" + app + "/" + stream + "/" + MD5(dst_url).hexdigest();
}
static void fillSockInfo(Value& val, SockInfo* info) {
void fillSockInfo(Value& val, SockInfo* info) {
val["peer_ip"] = info->get_peer_ip();
val["peer_port"] = info->get_peer_port();
val["local_port"] = info->get_local_port();

View File

@ -238,6 +238,7 @@ uint16_t openRtpServer(uint16_t local_port, const mediakit::MediaTuple &tuple, i
#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,
const mediakit::ProtocolOption &option, int rtp_type, float timeout_sec, const toolkit::mINI &args,

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,7 +231,7 @@ 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()){
@ -354,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());
@ -387,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("");
@ -402,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;
@ -423,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]
@ -450,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]
@ -480,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, "/");
@ -500,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) {
@ -545,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;
@ -568,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;
@ -596,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;
@ -643,6 +695,12 @@ void installWebHook() {
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;
@ -670,6 +728,11 @@ void installWebHook() {
});
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;
@ -719,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, 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]
@ -763,6 +831,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;
@ -779,6 +852,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();

View File

@ -43,6 +43,10 @@
#include "ZLMVersion.h"
#endif
#if defined(ENABLE_PYTHON)
#include "pyinvoker.h"
#endif
#include "System.h"
using namespace std;
@ -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
@ -261,6 +273,16 @@ int start_main(int argc,char *argv[]) {
}
#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();
@ -317,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]
@ -494,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 << "程序退出中,请等待...";

697
server/pyinvoker.cpp Normal file
View File

@ -0,0 +1,697 @@
#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 "Http/HttpSession.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 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()) {
hdrs.append(py::make_tuple(py::bytes(kv.first), py::bytes(kv.second)));
}
scope["headers"] = hdrs;
bool ok = check_route(scope).cast<bool>();
if (!ok) {
return;
}
consumed = true;
// http api被python拦截了再api统一鉴权
try {
auto args = getAllArgs(parser);
auto allArgs = ArgsMap(parser, args);
GET_CONFIG(std::string, api_secret, API::kSecret);
// TODO python http api暂不开启secret鉴权
// CHECK_SECRET(); // 检测secret
} catch (std::exception &ex) {
Json::Value val;
val["code"] = API::Exception;
val["msg"] = ex.what();
HttpSession::KeyValue headerOut;
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([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) {
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;
return File::absolutePath(path, 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);
});
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";
// 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(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, 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

View File

@ -243,6 +243,8 @@ MultiMediaSourceMuxer::MultiMediaSourceMuxer(const MediaTuple& tuple, float dur_
// Audio related settings
enableAudio(option.enable_audio);
enableMuteAudio(option.add_mute_audio);
NOTICE_EMIT(BroadcastCreateMuxerArgs, Broadcast::kBroadcastCreateMuxer, _delegate, *this);
}
void MultiMediaSourceMuxer::setMediaListener(const std::weak_ptr<MediaSourceEvent> &listener) {
@ -705,6 +707,9 @@ bool MultiMediaSourceMuxer::onTrackReady(const Track::Ptr &track) {
if (_mp4) {
ret = _mp4->addTrack(track) ? true : ret;
}
if (_delegate) {
_delegate->addTrack(track);
}
return ret;
}
@ -764,6 +769,9 @@ void MultiMediaSourceMuxer::onAllTrackReady() {
pr.second.syncTo(*first);
}
}
if (_delegate) {
_delegate->addTrackCompleted();
}
InfoL << "stream: " << shortUrl() << " , codec info: " << getTrackInfoStr(this);
}
@ -847,6 +855,9 @@ bool MultiMediaSourceMuxer::onTrackFrame_l(const Frame::Ptr &frame_in) {
if (_fmp4) {
ret = _fmp4->inputFrame(frame) ? true : ret;
}
if (_delegate) {
_delegate->inputFrame(frame);
}
if (_ring) {
// 此场景由于直接转发可能存在切换线程引起的数据被缓存在管道所以需要CacheAbleFrame [AUTO-TRANSLATED:528afbb7]
// In this scenario, due to direct forwarding, there may be data cached in the pipeline due to thread switching, so CacheAbleFrame is needed

View File

@ -29,6 +29,7 @@ class MultiMediaSourceMuxer : public MediaSourceEventInterceptor, public MediaSi
public:
using Ptr = std::shared_ptr<MultiMediaSourceMuxer>;
using RingType = toolkit::RingBuffer<Frame::Ptr>;
using onCreateMuxer = std::function<MediaSinkInterface::Ptr()>;
class Listener {
public:
@ -249,6 +250,8 @@ private:
toolkit::EventPoller::Ptr _poller;
RingType::Ptr _ring;
MediaSinkInterface::Ptr _delegate;
// 对象个数统计 [AUTO-TRANSLATED:3b43e8c2]
// Object count statistics
toolkit::ObjectStatistic<MultiMediaSourceMuxer> _statistic;

View File

@ -81,6 +81,8 @@ const string kBroadcastRtcSctpClosed = "kBroadcastRtcSctpClosed";
const string kBroadcastRtcSctpSend = "kBroadcastRtcSctpSend";
const string kBroadcastRtcSctpReceived = "kBroadcastRtcSctpReceived";
const string kBroadcastPlayerCountChanged = "kBroadcastPlayerCountChanged";
const string kBroadcastPlayerProxyFailed = "kBroadcastPlayerProxyFailed";
const string kBroadcastCreateMuxer = "kBroadcastCreateMuxer";
} // namespace Broadcast

View File

@ -126,7 +126,7 @@ extern const std::string kBroadcastStreamNoneReader;
// rtp推流被动停止时触发 [AUTO-TRANSLATED:43881965]
// Triggered when rtp push stream is passively stopped.
extern const std::string kBroadcastSendRtpStopped;
#define BroadcastSendRtpStoppedArgs MultiMediaSourceMuxer &sender, const std::string &ssrc, const SockException &ex
#define BroadcastSendRtpStoppedArgs MultiMediaSourceMuxer &sender, const std::string &ssrc, const toolkit::SockException &ex
// 更新配置文件事件广播,执行loadIniConfig函数加载配置文件成功后会触发该广播 [AUTO-TRANSLATED:ad4e167d]
// Update configuration file event broadcast. This broadcast will be triggered after the loadIniConfig function loads the configuration file successfully.
@ -161,6 +161,12 @@ extern const std::string kBroadcastRtcSctpReceived;
extern const std::string kBroadcastPlayerCountChanged;
#define BroadcastPlayerCountChangedArgs const MediaTuple& args, const int& count
extern const std::string kBroadcastPlayerProxyFailed;
#define BroadcastPlayerProxyFailedArgs const PlayerProxy& sender, const toolkit::SockException &ex
extern const std::string kBroadcastCreateMuxer;
#define BroadcastCreateMuxerArgs MediaSinkInterface::Ptr &delegate, const MultiMediaSourceMuxer &sender
#define ReloadConfigTag ((void *)(0xFF))
#define RELOAD_KEY(arg, key) \
do { \

View File

@ -110,7 +110,9 @@ void PlayerProxy::play(const string &strUrlTmp) {
if (!strongSelf) {
return;
}
if (err) {
NOTICE_EMIT(BroadcastPlayerProxyFailedArgs, Broadcast::kBroadcastPlayerProxyFailed, *strongSelf, err);
}
if (strongSelf->_on_play) {
strongSelf->_on_play(err);
strongSelf->_on_play = nullptr;
@ -146,6 +148,9 @@ void PlayerProxy::play(const string &strUrlTmp) {
if (!strongSelf) {
return;
}
if (err) {
NOTICE_EMIT(BroadcastPlayerProxyFailedArgs, Broadcast::kBroadcastPlayerProxyFailed, *strongSelf, err);
}
// 注销直接拉流代理产生的流:#532 [AUTO-TRANSLATED:c6343a3b]
// Unregister the stream generated by the direct stream proxy: #532

View File

@ -18,8 +18,7 @@
namespace mediakit {
struct StreamInfo
{
struct StreamInfo {
TrackType codec_type;
std::string codec_name;
int bitrate;
@ -30,8 +29,7 @@ struct StreamInfo
int video_height;
float video_fps;
StreamInfo()
{
StreamInfo() {
codec_type = TrackInvalid;
codec_name = "none";
bitrate = -1;
@ -44,14 +42,12 @@ struct StreamInfo
}
};
struct TranslationInfo
{
struct TranslationInfo {
std::vector<StreamInfo> stream_info;
int byte_speed;
uint64_t start_time_stamp;
TranslationInfo()
{
TranslationInfo() {
byte_speed = -1;
start_time_stamp = 0;
}

View File

@ -19,7 +19,11 @@ using namespace toolkit;
namespace mediakit {
MP4Muxer::~MP4Muxer() {
try {
closeMP4();
} catch (std::exception &e) {
WarnL << e.what();
}
}
void MP4Muxer::openMP4(const string &file) {