fix(recorder): stop stream-gears losing the tail, default to mesio (#1733)

* fix(stream-gears): write the cached GOP on disconnect, flush before the segment hook

The httpflv loop kept the current GOP in `flv_tags_cache` and only wrote it
when the next keyframe arrived. On EOF, a 30 s read timeout or a parse error
the function returned and the cache was dropped, so every disconnect,
reconnect and stream end lost the last 1-10 s. The loop now runs in an inner
function; whatever way it exits, the complete tags in the cache are written
before the original result (including the error) is returned.

`FlvFile` / `TsFile` renamed the `.part` file and ran the segment hook before
the old `BufWriter` was flushed, so the hook (FileValidator, upload) could see
a truncated file, and a failed final flush (disk full) was swallowed. Both now
go through `finish()`: flush and check the error, then rename, then run the
hook. `create_new` returns the flush error; `Drop` logs it with `error!`.

Tests: the rolling-segment test now expects 120 keyframes on disk (was 119)
and checks the size seen by the hook equals the final size; new cases for a
stream that ends cleanly, a read that times out and a parse error; a TS hook
size test; a failed-flush test.

Ref #1726

* feat(recorder): default to mesio when no downloader is configured

Rooms without a `downloader` setting followed the plugin hint, which is
stream-gears for almost every platform. They now use mesio: its FLV output
carries `onMetaData.keyframes` and every segment starts at timestamp 0, so
players can seek; it also records HEVC / Enhanced-FLV and HLS fMP4, which
stream-gears cannot. Only the hint mapping in `core/live.rs` changes; an
explicit `stream-gears` keeps using stream-gears, and plugins that hint
streamlink / yt-dlp are unaffected.

`DownloaderRuntime::from_type` no longer has a catch-all arm. `ffmpeg-external`
and `ffmpeg-internal` used to fall into it and silently run stream-gears; both
now run ffmpeg (external segmenting, same as `ffmpeg`). The arm for
streamlink / yt-dlp / ytarchive, which need the live stream and are built in
`downloader_runtime`, is listed explicitly and logs a warning.

WebUI, README, sample configs and CHANGELOG now say mesio is the default and
list the behaviour differences: the container follows the source (HLS fMP4 is
saved as .mp4), files are written under their final name without `.part`, and
Douyu Wangsu reconnects leave a 2.6-5.8 s overlap between segments.

Ref #1726

* feat(mesio): pass segment duration and size to the segment callback

mesio's segment-complete callback reports the duration and byte size of each
closed file, but biliup only logged them and built `SegmentInfo` with the path
and index. `SegmentInfo` gains `duration_secs` / `size_bytes` (None for the
other downloaders) and mesio fills them in. The end-to-end mesio test checks
the reported size equals the file size on disk.

Ref #1726

* fix(recorder): parse segment_time the same way in every downloader

`segment_time` was read three different ways. stream-gears indexed
`parts[1]` / `parts[2]` directly, so "3600" or "01:00" panicked the download
task (and an unparsable hour silently became 1); mesio returned 0 for anything
that was not HH:MM:SS and silently recorded without time splits; ffmpeg got
the raw string and accepted both forms.

`parse_segment_time` follows ffmpeg's duration syntax (`[HH:]MM:SS[.frac]`
or plain seconds; minutes and seconds below 60 when colons are used), so all
three downloaders now agree. stream-gears and mesio share
`DownloadConfig::segment_time_limit`: 0 means no time split for both (mesio
already did this; stream-gears used to split at every keyframe), and a value
that still cannot be parsed logs a warning and records without time splits
instead of panicking or being ignored silently. The time-range clamp uses the
same parser, so "3600" is shortened near the window end like "01:00:00".

Ref #1726

* feat(webui): enter file_size as a number with an MB/GB unit

The config still stores bytes. The control converts with 1024 (the same
sizes Windows Explorer shows), keeps an untouched value byte-exact even
when it is not a round MB/GB, and re-picks the unit when the form value
is replaced from outside.

In the streamer override, clearing a value that was already set now
saves null (no size split for that streamer); before, Semi dropped the
emptied key and the old value stayed. Clearing something typed in the
same session still falls back to inheriting the global setting.

* fix(webui): keep the file_size unit beside the number in narrow modals
This commit is contained in:
Noah authored and GitHub committed 2026-09-24 06:24:17 +00:00
1 parent bff8815be4
commit 67e49354ad
23 files changed
+683 -139

No files matched your search

+1 -1
View File
@@ -161,7 +161,7 @@ biliup season sort <小节ID> <episode ID>... # 列出的稿件按顺序
- 详见[Wiki](https://github.com/biliup/biliup/wiki/Termux-%E4%B8%AD%E4%BD%BF%E7%94%A8-biliup)
> [!NOTE]
> 默认下载器 `stream-gears` 由 Rust 实现,无需外部依赖。若配置 `ffmpeg` 下载器或使用后处理,需要本机安装 `ffmpeg`;YouTube、niconico 等平台与通用适配器则依赖 `yt-dlp` 或 `streamlink`。Docker 镜像已内置 `ffmpeg`。
> 默认下载器 `mesio`(rust-srec 引擎)和可选的 `stream-gears` 都由 Rust 实现、编进 biliup,无需外部依赖。若配置 `ffmpeg` 下载器或使用后处理,需要本机安装 `ffmpeg`;YouTube、niconico 等平台与通用适配器则依赖 `yt-dlp` 或 `streamlink`。Docker 镜像已内置 `ffmpeg`。
### 🔓 远程访问(监听 0.0.0.0)
+127
View File
@@ -0,0 +1,127 @@
'use client'
import React, { useState } from 'react'
import { InputNumber, Select, Typography, withField } from '@douyinfe/semi-ui'
const MB = 1024 * 1024
const GB = 1024 * MB
export type SizeUnit = 'MB' | 'GB'
const UNIT_BYTES: Record<SizeUnit, number> = { MB, GB }
/** 选能用两位以内小数精确表示的最大单位;都不能精确表示时,1 GB 及以上用 GB,否则用 MB */
export const pickUnit = (bytes: number): SizeUnit => {
for (const unit of ['GB', 'MB'] as const) {
const size = UNIT_BYTES[unit]
if (bytes >= size && (bytes * 100) % size === 0) return unit
}
return bytes >= GB ? 'GB' : 'MB'
}
/** 显示用的数值:1 以上保留两位小数,更小的保留三位有效数字 */
export const formatAmount = (bytes: number, unit: SizeUnit): number => {
const amount = bytes / UNIT_BYTES[unit]
return amount >= 1 ? Math.round(amount * 100) / 100 : Number(amount.toPrecision(3))
}
export const toBytes = (amount: number, unit: SizeUnit): number => Math.round(amount * UNIT_BYTES[unit])
const bytesOf = (value: unknown): number | null => {
if (value === null || value === undefined || value === '') return null
const bytes = typeof value === 'number' ? value : Number(value)
return Number.isFinite(bytes) ? bytes : null
}
/** 用户最近一次输入:只对应它产生的那个字节数 */
type Draft = { bytes: number | null; unit: SizeUnit; amount: number | '' }
type Props = {
/** 字节数;空串 / undefined / null 表示没有值 */
value?: number | string | null
/**
* 清空时:外部带进来的值被清掉给 null(覆写里即「显式清空」,Semi 表单会保留这个键);
* 清掉的是自己刚输入的值则给空串,Semi 会把键移除,回到打开时的状态(覆写里即「跟随全局」)
*/
onChange?: (value: number | '' | null) => void
onBlur?: (e: React.FocusEvent) => void
disabled?: boolean
id?: string
placeholder?: string
validateStatus?: 'default' | 'error' | 'warning' | 'success'
}
/**
* 字节数的「数值 + 单位(MB / GB,1024 进制)」输入。表单里存的仍是字节整数:
* 没动过就原样保留(不经过换算,不丢字节),改了数值或单位才按新输入换算。
*/
const FileSizeInput: React.FC<Props> = ({
value,
onChange,
onBlur,
disabled,
id,
placeholder,
validateStatus,
}) => {
const bytes = bytesOf(value)
const [draft, setDraft] = useState<Draft | null>(null)
// 表单值被外部改写(读回配置、平台插件回填覆写)后草稿作废,按新值重新选单位
const current = draft && draft.bytes === bytes ? draft : null
const unit = current?.unit ?? (bytes === null ? 'GB' : pickUnit(bytes))
const amount = current ? current.amount : bytes === null ? '' : formatAmount(bytes, unit)
// 最近一次从外部带进来的值,用来区分「清空已有值」和「撤销自己的输入」
const [loaded, setLoaded] = useState<number | null>(bytes)
if (!current && bytes !== loaded) setLoaded(bytes)
const commit = (nextAmount: number | '', nextUnit: SizeUnit) => {
const nextBytes = nextAmount === '' ? null : toBytes(nextAmount, nextUnit)
setDraft({ bytes: nextBytes, unit: nextUnit, amount: nextAmount })
onChange?.(nextBytes !== null ? nextBytes : loaded === null ? '' : null)
}
return (
<div style={{ display: 'flex', flexWrap: 'wrap', alignItems: 'center', gap: 8, width: '100%' }}>
<InputNumber
id={id}
value={amount}
min={0}
placeholder={placeholder}
disabled={disabled}
validateStatus={validateStatus}
showClear
onBlur={onBlur}
onChange={next => {
if (next === '') commit('', unit)
else if (typeof next === 'number' && Number.isFinite(next)) commit(next, unit)
}}
style={{ flex: '1 1 120px', minWidth: 0 }}
/>
<Select
aria-label="单位"
value={unit}
disabled={disabled}
onChange={next => {
const nextUnit = next as SizeUnit
// 还没填数值时只记住单位,不写表单:保持「未设置」
if (amount === '') setDraft({ bytes, unit: nextUnit, amount: '' })
else commit(amount, nextUnit)
}}
optionList={[
{ value: 'MB', label: 'MB' },
{ value: 'GB', label: 'GB' },
]}
style={{ width: 88 }}
/>
{bytes !== null ? (
<Typography.Text type="tertiary" size="small">
= {bytes.toLocaleString('en-US')} 字节
</Typography.Text>
) : null}
</div>
)
}
/** 表单字段版:`field` 存字节数 */
export const FileSizeField = withField(FileSizeInput)
export default FileSizeInput
+6 -8
View File
@@ -12,6 +12,7 @@ import { useState } from 'react'
import { LiveStreamerEntity } from '../lib/api-streamer'
import { SupportedPlatforms } from '@/app/ui/plugins'
import { useBiliUsers } from '../lib/use-streamers'
import { FileSizeField } from './FileSizeInput'
type PluginProps = {
entity?: LiveStreamerEntity
@@ -181,7 +182,7 @@ const OverrideModal: React.FC<TemplateModalProps> = ({ children, entity, onOk })
<Form.Select
label="下载插件(downloader)"
field="downloader"
placeholder="stream-gears(默认)"
placeholder="mesio(默认)"
style={{ width: '100%' }}
fieldStyle={{
alignSelf: 'stretch',
@@ -191,22 +192,19 @@ const OverrideModal: React.FC<TemplateModalProps> = ({ children, entity, onOk })
>
<Select.Option value="streamlink">streamlink(hls多线程下载)</Select.Option>
<Select.Option value="ffmpeg">ffmpeg</Select.Option>
<Select.Option value="stream-gears">stream-gears(默认)</Select.Option>
<Select.Option value="stream-gears">stream-gears</Select.Option>
<Select.Option value="sync-downloader">sync-downloader(边录边传)</Select.Option>
<Select.Option value="mesio">mesio(内置流修复下载器)</Select.Option>
<Select.Option value="mesio">mesio(默认)</Select.Option>
</Form.Select>
<Form.InputNumber
<FileSizeField
label="视频分段大小(file_size)"
field="file_size"
placeholder=""
suffix={'Byte'}
style={{ width: '100%' }}
extraText="按 1024 进制:1 GB = 1024 MB。没填过的留空即跟随全局设置;把已有的值清空,则这个主播不按大小分段(边录边传仍约 2 GB 一段)。"
fieldStyle={{
alignSelf: 'stretch',
padding: 0,
}}
showClear={true}
/>
<Form.Input
+1 -1
View File
@@ -246,7 +246,7 @@ const TemplateModal: React.FC<TemplateModalProps> = ({ children, entity, onOk })
label="视频格式"
placeholder="flv"
style={{ width: 176 }}
helpText="视频保存格式。不支持stream-gears下载器和Youtube平台。"
helpText="视频保存格式(转封装)。mesio、stream-gears 下载器不转封装,按源站容器保存;YouTube 平台也不支持。"
/>
<Collapse keepDOM>
+2 -1
View File
@@ -150,7 +150,8 @@ const Bilibili: React.FC<Props> = props => {
选 hls_fmp4 时,开播后会先等待下方「hls_fmp4 转码等待时间」,仍拿不到 fmp4 流则本场回退为
FLV。
<br />
<strong>stream-gears 不支持 hls_fmp4</strong>,需把下载插件改为 ffmpeg 或 streamlink。
默认的下载插件 mesio 支持 hls_fmp4(保存为 .mp4);<strong>stream-gears 不支持 hls_fmp4</strong>,选了
stream-gears 的需改为 mesio、ffmpeg 或 streamlink。
</div>
}
label="直播流协议(bili_protocol)"
+2 -2
View File
@@ -114,8 +114,8 @@ const Douyin: React.FC<Props> = props => {
<div style={{ fontSize: '14px' }}>
默认关闭。开启后录制抖音的真原画流,<strong>仅在画质为原画、直播流协议为 flv 时生效</strong>。
<br />
真原画可能是 HEVC 编码,默认下载插件 stream-gears 不支持 HEVC 会导致录制失败,请先把下载插件换成
ffmpeg 或 streamlink。
真原画可能是 HEVC 编码。默认的下载插件 mesio 能录 HEVC;stream-gears 不支持 HEVC,会导致录制失败,选了
stream-gears 的请换成 mesio、ffmpeg 或 streamlink。浏览器里的直播预览放不了 HEVC,录制不受影响。
</div>
}
label="抖音真原画(douyin_true_origin)"
+21 -24
View File
@@ -4,6 +4,7 @@ import styles from '../../styles/dashboard.module.scss'
import SectionTitle from '../../(app)/components/SectionTitle'
import { Form, Select, Space, useFormState } from '@douyinfe/semi-ui'
import { IconUpload, IconDownload } from '@douyinfe/semi-icons'
import { FileSizeField } from '../FileSizeInput'
const Global: React.FC = () => {
// useFormApi 不订阅表单值变化,切换下拉框后条件渲染不会刷新;useFormState 会
@@ -18,14 +19,23 @@ const Global: React.FC = () => {
<Form.Select
label="下载插件(downloader)"
field="downloader"
placeholder="stream-gears(默认)"
// initValue="stream-gears"
placeholder="mesio(默认)"
extraText={
<div style={{ fontSize: '14px' }}>
全局默认的下载插件,可在单个主播的覆写设置里另选。可选:
<br />
1. <strong>stream-gears</strong>(默认):内置,无需额外安装,可防 FLV 流花屏;不支持 HEVC 编码和
hls_fmp4 流。
1. <strong>mesio</strong>(默认):内置 rust-srec 引擎,无需额外安装。进程内下载 FLV / HLS,修复时间戳(每段从
0 开始)、写入关键帧索引(onMetaData.keyframes),支持 HEVC 和 hls_fmp4;不转封装,按源站的容器保存(FLV /
TS,hls_fmp4 存为 .mp4)。详见{' '}
<a
href="https://github.com/hua0512/rust-srec"
target="_blank"
rel="noopener noreferrer"
style={{ color: 'var(--semi-color-link)' }}
>
项目主页
</a>
。
<br />
2. ffmpeg:非 Docker 用户需自行安装 FFmpeg。
<br />
@@ -46,17 +56,8 @@ const Global: React.FC = () => {
<br />
5. ytarchive:仅适用于 YouTube 直播。
<br />
6. mesio:内置 rust-srec 引擎,进程内下载 FLV / HLS 并修复时间戳、注入关键帧索引,支持按大小 /
时长分段,无需额外安装。详见{' '}
<a
href="https://github.com/hua0512/rust-srec"
target="_blank"
rel="noopener noreferrer"
style={{ color: 'var(--semi-color-link)' }}
>
项目主页
</a>
。
6. stream-gears:内置,无需额外安装,可防 FLV 流花屏;不支持 HEVC 编码和 hls_fmp4 流,FLV
时间戳沿用源站的原值(不从 0 开始)。
</div>
}
style={{ width: '100%' }}
@@ -68,10 +69,10 @@ const Global: React.FC = () => {
>
<Select.Option value="streamlink">streamlink(hls多线程下载)</Select.Option>
<Select.Option value="ffmpeg">ffmpeg</Select.Option>
<Select.Option value="stream-gears">stream-gears(默认)</Select.Option>
<Select.Option value="stream-gears">stream-gears</Select.Option>
<Select.Option value="sync-downloader">sync-downloader(边录边传)</Select.Option>
<Select.Option value="ytarchive">ytarchive(仅适用于 Youtube Live)</Select.Option>
<Select.Option value="mesio">mesio(内置流修复下载器)</Select.Option>
<Select.Option value="mesio">mesio(默认)</Select.Option>
</Form.Select>
{isSyncDownloader ? (
<>
@@ -115,24 +116,20 @@ const Global: React.FC = () => {
/>
</>
) : null}
<Form.InputNumber
<FileSizeField
label="视频分段大小(file_size)"
extraText={
<div style={{ fontSize: '14px' }}>
录像单文件大小限制,超过此大小触发文件分割。下载回放时无法使用。
录像单文件大小上限,超过后开始写下一个文件。下载回放时无法使用。留空表示不按大小分段。
<br />
单位:Byte,示例:4294967296(4GB)
按 1024 进制换算:1 GB = 1024 MB = 1073741824 字节,与 Windows 资源管理器显示的大小一致。配置文件里存的仍是字节数。
</div>
}
field="file_size"
placeholder=""
suffix={'Byte'}
style={{ width: '100%' }}
fieldStyle={{
alignSelf: 'stretch',
padding: 0,
}}
showClear={true}
/>
<Form.Input
field="segment_time"
+1 -1
View File
@@ -43,7 +43,7 @@ const Twitch: React.FC<Props> = props => {
<br />
更好的办法是开通 Twitch Turbo 会员并在下方填入 Twitch Cookie,可直接免广告。
<br />
<strong>仅下载插件为 streamlink 或 ffmpeg 时生效</strong>;默认的 stream-gears 直接拉流,此开关不起作用。
<strong>仅下载插件为 streamlink 或 ffmpeg 时生效</strong>;默认的 mesio 和 stream-gears 直接拉流,此开关不起作用。
</div>
}
label="去除广告(twitch_disable_ads)"
@@ -10,6 +10,7 @@
//! 比较的行为一致。若改用本地时间判定,UTC+8 用户配置的 16:00–20:00 会被当成
//! 08:00–12:00,整整错开 8 小时。
use crate::server::common::util::parse_segment_time;
use chrono::{DateTime, NaiveDateTime, Timelike, Utc};
use tracing::debug;
@@ -135,26 +136,17 @@ fn clamp_segment_time_at(
let remaining = range.seconds_until_end(now);
match segment_time {
// 分段时长无法解析时原样下发,交给下游工具自己报错
Some(configured) => match parse_hms(configured) {
Some(seconds) if seconds > remaining => Some(format_hms(remaining)),
// 分段时长无法解析时原样下发,由下载器自己报错或提示
Some(configured) => match parse_segment_time(configured) {
Some(limit) if limit.as_secs_f64() > f64::from(remaining) => {
Some(format_hms(remaining))
}
Some(_) | None => Some(configured.to_owned()),
},
None => Some(format_hms(remaining)),
}
}
/// 解析 `"HH:MM:SS"` 为秒数。
fn parse_hms(raw: &str) -> Option<u32> {
let [hours, minutes, seconds] =
<[&str; 3]>::try_from(raw.split(':').collect::<Vec<_>>()).ok()?;
Some(
hours.parse::<u32>().ok()? * 3600
+ minutes.parse::<u32>().ok()? * 60
+ seconds.parse::<u32>().ok()?,
)
}
fn format_hms(total: u32) -> String {
format!(
"{:02}:{:02}:{:02}",
@@ -170,7 +162,9 @@ mod tests {
/// 把 "HH:MM:SS" 写成当日秒数,让用例读起来贴近配置里的时刻
fn at(hms: &str) -> u32 {
parse_hms(hms).expect("测试时刻格式应合法")
parse_segment_time(hms)
.expect("测试时刻格式应合法")
.as_secs() as u32
}
/// 前端 `Date.toISOString()` 写出的形态
@@ -305,6 +299,19 @@ mod tests {
}
}
/// ffmpeg 也接受的 `MM:SS` 与纯秒数写法同样按窗口裁短
#[test]
fn seconds_and_minutes_forms_are_shortened_too() {
let tr = range("16:00:00", "20:00:00");
for configured in ["3600", "59:00"] {
assert_eq!(
clamp_segment_time_at(Some(configured), Some(&tr), at("19:40:00")),
Some("00:20:00".to_string()),
"configured={configured}"
);
}
}
#[test]
fn sitting_exactly_on_the_window_end_leaves_zero_remaining() {
let tr = range("16:00:00", "20:00:00");
+60 -7
View File
@@ -240,23 +240,76 @@ pub fn media_ext_from_url(input: &str) -> Option<String> {
None
}
pub fn parse_time(segment_time: &str) -> std::time::Duration {
let parts: Vec<&str> = segment_time.split(':').collect();
let h = parts[0].parse::<i32>().unwrap_or(1);
let m = parts[1].parse::<i32>().unwrap_or(0);
let s = parts[2].parse::<i32>().unwrap_or(0);
std::time::Duration::from_secs((h * 3600 + m * 60 + s) as u64)
/// 解析 `segment_time`,语法与 ffmpeg 的时长参数一致(ffmpeg 下载器把原字符串直接交给 `-to`):
/// `[HH:]MM:SS[.小数]` 或纯秒数 `S[.小数]`,例如 `"01:00:00"`、`"30:00"`(30 分钟)、`"3600"`。
/// 带冒号时分、秒必须小于 60。无法解析时返回 `None`。
pub fn parse_segment_time(raw: &str) -> Option<std::time::Duration> {
let parts: Vec<&str> = raw.trim().split(':').collect();
let (hours, minutes, seconds) = match parts.as_slice() {
[seconds] => ("0", "0", *seconds),
[minutes, seconds] => ("0", *minutes, *seconds),
[hours, minutes, seconds] => (*hours, *minutes, *seconds),
_ => return None,
};
let digits = |s: &str| !s.is_empty() && s.bytes().all(|b| b.is_ascii_digit());
let seconds_ok = seconds
.split_once('.')
.map_or(digits(seconds), |(whole, frac)| {
digits(whole) && (frac.is_empty() || digits(frac))
});
if !digits(hours) || !digits(minutes) || !seconds_ok {
return None;
}
let hours: u64 = hours.parse().ok()?;
let minutes: u64 = minutes.parse().ok()?;
let seconds: f64 = seconds.parse().ok()?;
if parts.len() > 1 && (minutes >= 60 || seconds >= 60.0) {
return None;
}
let whole = hours.checked_mul(3600)?.checked_add(minutes * 60)?;
std::time::Duration::try_from_secs_f64(whole as f64 + seconds).ok()
}
#[cfg(test)]
mod tests {
use crate::server::common::util::{
Recorder, media_ext_from_url, path_with_suffix, redact_secrets,
Recorder, media_ext_from_url, parse_segment_time, path_with_suffix, redact_secrets,
};
use crate::server::infrastructure::models::StreamerInfo;
use chrono::Utc;
use std::path::PathBuf;
/// 与 ffmpeg `-to` 接受的时长写法一致(ffmpeg 下载器把原字符串直接交给它)
#[test]
fn segment_time_accepts_what_ffmpeg_accepts() {
use std::time::Duration;
let secs = |s: u64| Some(Duration::from_secs(s));
assert_eq!(parse_segment_time("01:00:00"), secs(3600));
assert_eq!(parse_segment_time("1:2:3"), secs(3723));
assert_eq!(parse_segment_time("100:00:00"), secs(360_000));
assert_eq!(parse_segment_time("30:00"), secs(1800));
assert_eq!(parse_segment_time("3600"), secs(3600));
assert_eq!(parse_segment_time(" 01:00:00 "), secs(3600));
assert_eq!(parse_segment_time("1.5"), Some(Duration::from_millis(1500)));
assert_eq!(parse_segment_time("00:00:00"), secs(0));
for invalid in [
"",
"abc",
"90:00",
"1:70:00",
"01:00:60",
"01:00:00:00",
"-1:00:00",
"1h",
"inf",
"NaN",
"+5",
"1e3",
] {
assert_eq!(parse_segment_time(invalid), None, "{invalid:?}");
}
}
#[test]
fn path_with_suffix_keeps_dots_in_basename() {
assert_eq!(
@@ -12,7 +12,7 @@ pub mod ws_expire;
pub mod ytdlp;
use crate::server::common::timerange;
use crate::server::common::util::Recorder;
use crate::server::common::util::{Recorder, parse_segment_time};
use crate::server::core::downloader::ffmpeg_downloader::FfmpegDownloader;
use crate::server::core::downloader::mesio::Mesio;
use crate::server::core::downloader::stream_gears::StreamGears;
@@ -29,6 +29,7 @@ use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::path::PathBuf;
use std::sync::Mutex;
use tracing::warn;
/// 下载器配置
/// 包含下载过程中需要的各种参数和设置
@@ -87,6 +88,23 @@ impl DownloadConfig {
timerange::clamp_segment_time(self.segment_time.as_deref(), self.time_range.as_deref())
}
/// 进程内下载器(stream-gears、mesio)按时长分段的上限,已按录制时间范围裁短。
/// 未配置或为 0 时不按时长分段;无法解析时同样不分段,但打 warn 说明原因。
pub fn segment_time_limit(&self) -> Option<std::time::Duration> {
let raw = self.segment_duration()?;
match parse_segment_time(&raw) {
Some(limit) if !limit.is_zero() => Some(limit),
Some(_) => None,
None => {
warn!(
segment_time = raw,
"segment_time 无法解析(支持 HH:MM:SS、MM:SS 或秒数),本次录制不按时长分段"
);
None
}
}
}
/// 距录制时间范围结束还剩多久(`"HH:MM:SS"`);未配置录制时间范围时为 `None`。
pub fn time_range_remaining(&self) -> Option<String> {
timerange::remaining_until_end(self.time_range.as_deref())
@@ -134,13 +152,25 @@ impl DownloaderRuntime {
/// 从配置创建
pub fn from_type(downloader_type: DownloaderType) -> Self {
match downloader_type {
DownloaderType::Ffmpeg => Self::Ffmpeg(FfmpegDownloader::new(
// `ffmpeg-internal` 的 segment muxer 实现从未接通过,与 `ffmpeg` / `ffmpeg-external`
// 一样按外部分段跑,至少不再静默换成 stream-gears
DownloaderType::Ffmpeg
| DownloaderType::FfmpegExternal
| DownloaderType::FfmpegInternal => Self::Ffmpeg(FfmpegDownloader::new(
Vec::new(),
DownloaderType::FfmpegExternal,
)),
DownloaderType::SyncDownloader => Self::Sync(SyncDownloader::new()),
DownloaderType::Mesio => Self::Mesio(Mesio::new()),
_ => Self::StreamGears(StreamGears::new(None)),
DownloaderType::StreamGears => Self::StreamGears(StreamGears::new(None)),
// 这三种要用到直播流里的参数,由 `core::live::downloader_runtime` 构造,走不到这里
DownloaderType::Streamlink | DownloaderType::YtDlp | DownloaderType::Ytarchive => {
warn!(
?downloader_type,
"this downloader needs the live stream to be built, using stream-gears"
);
Self::StreamGears(StreamGears::new(None))
}
}
}
@@ -182,6 +212,10 @@ pub struct SegmentInfo {
pub next_file_path: Option<PathBuf>,
/// 分段序号
pub segment_index: usize,
/// 分段时长(秒)。下载器报告了才有,目前只有 mesio
pub duration_secs: Option<f64>,
/// 分段文件的字节数。下载器报告了才有,目前只有 mesio
pub size_bytes: Option<u64>,
// /// 分段开始时间戳
// start_time: std::time::SystemTime,
// /// 分段结束时间戳
@@ -200,8 +234,17 @@ impl SegmentInfo {
danmaku_file_path,
next_file_path,
segment_index,
duration_secs: None,
size_bytes: None,
}
}
/// 附上下载器报告的分段时长与字节数。
pub fn with_stats(mut self, duration_secs: f64, size_bytes: u64) -> Self {
self.duration_secs = Some(duration_secs);
self.size_bytes = Some(size_bytes);
self
}
}
/// 分段事件
@@ -373,6 +416,37 @@ mod tests {
assert!(matches!(gears, DownloaderRuntime::StreamGears(_)));
}
#[test]
fn segment_time_limit_is_shared_by_the_in_process_downloaders() {
let limit = |segment_time: &str| {
DownloadConfig {
segment_time: Some(segment_time.to_string()),
..Default::default()
}
.segment_time_limit()
};
let secs = |s| Some(std::time::Duration::from_secs(s));
assert_eq!(limit("01:00:00"), secs(3600));
assert_eq!(limit("30:00"), secs(1800));
assert_eq!(limit("3600"), secs(3600));
assert_eq!(limit("00:00:00"), None);
assert_eq!(limit("abc"), None);
assert_eq!(DownloadConfig::default().segment_time_limit(), None);
}
#[test]
fn ffmpeg_variants_are_not_silently_mapped_to_stream_gears() {
for configured in ["\"ffmpeg\"", "\"ffmpeg-external\"", "\"ffmpeg-internal\""] {
let downloader_type: DownloaderType = serde_json::from_str(configured).unwrap();
match DownloaderRuntime::from_type(downloader_type) {
DownloaderRuntime::Ffmpeg(ffmpeg) => {
assert_eq!(ffmpeg.downloader_type, DownloaderType::FfmpegExternal)
}
_ => panic!("{configured} must run ffmpeg"),
}
}
}
#[test]
fn mesio_maps_to_its_own_runtime_and_kebab_case_name() {
let runtime = DownloaderRuntime::from_type(DownloaderType::Mesio);
@@ -238,6 +238,8 @@ impl FfmpegDownloader {
danmaku_file_path: None,
segment_index: 0,
next_file_path: None,
duration_secs: None,
size_bytes: None,
}));
// 根据退出码判断状态
match status.code() {
@@ -303,6 +305,8 @@ impl FfmpegDownloader {
danmaku_file_path: None,
next_file_path: None,
segment_index,
duration_secs: None,
size_bytes: None,
// start_time: std::time::SystemTime::now(),
// end_time: std::time::SystemTime::now(),
}));
@@ -328,6 +332,8 @@ impl FfmpegDownloader {
danmaku_file_path: None,
next_file_path: None,
segment_index,
duration_secs: None,
size_bytes: None,
// start_time: std::time::SystemTime::now(),
// end_time: std::time::SystemTime::now(),
}));
@@ -42,8 +42,8 @@ use tracing::{debug, info, warn};
/// 管线内部通道容量(条目数),与 mesio-cli 默认值一致。
const CHANNEL_SIZE: usize = 64;
/// 分段回调载荷:已关闭的分段文件路径与 0 起始的序号。
type SegmentClosed = (PathBuf, u32);
/// 分段回调载荷:已关闭的分段文件路径、0 起始的序号、时长(秒)与字节数。
type SegmentClosed = (PathBuf, u32, f64, u64);
/// 直播预览旁路:写入端与「把一个管线条目旁路给它」的函数。
type PreviewTee<I> = (PreviewSink, fn(&mut PreviewSink, &I));
@@ -280,11 +280,8 @@ fn pipeline_config(download_config: &DownloadConfig) -> PipelineConfig {
let mut builder = PipelineConfig::builder()
.max_file_size(download_config.file_size.unwrap_or(0))
.channel_size(CHANNEL_SIZE);
if let Some(segment) = download_config.segment_duration() {
let secs = downloader::parse_duration(&segment);
if secs > 0 {
builder = builder.max_duration(Duration::from_secs(secs));
}
if let Some(limit) = download_config.segment_time_limit() {
builder = builder.max_duration(limit);
}
builder.build()
}
@@ -348,7 +345,7 @@ fn segment_complete_hook(
reason = ?reason,
"mesio 分段完成"
);
let _ = seg_tx.send((path.to_path_buf(), index));
let _ = seg_tx.send((path.to_path_buf(), index, duration_secs, size_bytes));
}
}
@@ -408,13 +405,11 @@ where
}
});
while let Some((path, index)) = seg_rx.recv().await {
callback(SegmentEvent::Segment(SegmentInfo::new(
path,
None,
None,
index as usize,
)));
while let Some((path, index, duration_secs, size_bytes)) = seg_rx.recv().await {
callback(SegmentEvent::Segment(
SegmentInfo::new(path, None, None, index as usize)
.with_stats(duration_secs, size_bytes),
));
}
if let Err(e) = forward.await {
@@ -560,6 +555,13 @@ mod tests {
let unlimited = pipeline_config(&DownloadConfig::default());
assert_eq!(unlimited.max_file_size, 0);
assert_eq!(unlimited.max_duration, None);
// 与 stream-gears 同一套解析:纯秒数不再被静默当成不分段
let seconds = pipeline_config(&DownloadConfig {
segment_time: Some("3600".to_string()),
..Default::default()
});
assert_eq!(seconds.max_duration, Some(Duration::from_secs(3600)));
}
/// 端到端:本地 HTTP 服务吐一段真实 FLV 录像,走完整的引擎拉流 → 修复管线 → 落盘,
@@ -659,6 +661,12 @@ mod tests {
assert!(name.ends_with(".flv"), "{name}");
let size = std::fs::metadata(&info.prev_file_path).unwrap().len();
assert!(size > 13, "segment {name} is empty");
assert_eq!(info.size_bytes, Some(size), "reported size of {name}");
assert!(
info.duration_secs.is_some_and(|d| d >= 0.0),
"reported duration of {name}: {:?}",
info.duration_secs
);
}
}
@@ -1,5 +1,4 @@
use crate::server::common::construct_headers;
use crate::server::common::util::parse_time;
use crate::server::core::downloader::{DownloadConfig, DownloadStatus, SegmentEvent, SegmentInfo};
use crate::server::errors::{AppError, AppResult};
use biliup::client::StatelessClient;
@@ -52,10 +51,7 @@ impl StreamGears {
let proxy = self.proxy.clone();
let segment = Segmentable::new(
// 快到录制时间范围结束时会被裁短,使录制停在窗口边界
download_config
.segment_duration()
.as_deref()
.map(parse_time),
download_config.segment_time_limit(),
download_config.file_size,
);
@@ -85,6 +81,8 @@ impl StreamGears {
danmaku_file_path: None,
next_file_path: None,
segment_index: i,
duration_secs: None,
size_bytes: None,
};
callback(SegmentEvent::Segment(event));
+50 -1
View File
@@ -173,7 +173,9 @@ pub fn downloader_runtime(
stream: &LiveStream,
) -> DownloaderRuntime {
let downloader_type = config_type.unwrap_or_else(|| match stream.downloader_hint {
DownloaderHint::StreamGears => DownloaderType::StreamGears,
// 未配置下载器时用 mesio:写出的 FLV 带 onMetaData.keyframes、每段时间戳从 0 开始,
// 能录 HEVC / Enhanced-FLV 与 HLS fMP4。显式配置 stream-gears 的不受影响
DownloaderHint::StreamGears => DownloaderType::Mesio,
DownloaderHint::Ffmpeg => DownloaderType::Ffmpeg,
DownloaderHint::Streamlink => DownloaderType::Streamlink,
DownloaderHint::YtDlp => DownloaderType::YtDlp,
@@ -323,3 +325,50 @@ pub fn danmaku_client(
Some(Arc::new(RustDanmakuClient::new(config)))
}
#[cfg(test)]
mod tests {
use super::*;
use std::collections::HashMap;
fn stream(hint: DownloaderHint) -> LiveStream {
LiveStream {
name: "room".into(),
url: "https://live.bilibili.com/1".into(),
title: "t".into(),
date: chrono::Utc::now(),
live_cover_url: String::new(),
avatar_url: None,
raw_stream_url: "https://cdn.example/live.flv".into(),
platform: "bilibili".into(),
stream_headers: HashMap::new(),
suffix: "flv".into(),
danmaku: None,
downloader_hint: hint,
runtime_options: None,
}
}
#[test]
fn unconfigured_rooms_default_to_mesio() {
let runtime = downloader_runtime(None, &stream(DownloaderHint::StreamGears));
assert!(matches!(runtime, DownloaderRuntime::Mesio(_)));
}
#[test]
fn an_explicit_stream_gears_choice_is_kept() {
let runtime = downloader_runtime(
Some(DownloaderType::StreamGears),
&stream(DownloaderHint::StreamGears),
);
assert!(matches!(runtime, DownloaderRuntime::StreamGears(_)));
}
#[test]
fn plugin_hints_for_external_tools_are_unchanged() {
let runtime = downloader_runtime(None, &stream(DownloaderHint::Streamlink));
assert!(matches!(runtime, DownloaderRuntime::StreamLink(_)));
let runtime = downloader_runtime(None, &stream(DownloaderHint::Ffmpeg));
assert!(matches!(runtime, DownloaderRuntime::Ffmpeg(_)));
}
}
+1
View File
@@ -47,6 +47,7 @@ tracing-subscriber = { workspace = true, features = ["env-filter"] } # 添加
tempfile = "3.14"
http = "1"
axum = "0.8"
tokio = { workspace = true, features = ["test-util"] }
[features]
cli = ["clap"]
+18 -3
View File
@@ -9,7 +9,7 @@ use serde::Serialize;
use std::fs::File;
use std::io::{BufWriter, Write};
use tracing::info;
use tracing::{error, info};
const FLV_HEADER: [u8; 9] = [
0x46, // 'F'
@@ -23,6 +23,8 @@ const FLV_HEADER: [u8; 9] = [
pub struct FlvFile<'a> {
pub buf_writer: BufWriter<File>,
pub file: LifecycleFile<'a>,
/// 当前分段已交给 [`LifecycleFile::finish`],`Drop` 不再重复改名、触发钩子。
finished: bool,
}
impl<'a> FlvFile<'a> {
@@ -32,16 +34,25 @@ impl<'a> FlvFile<'a> {
Ok(Self {
buf_writer: Self::create(path)?,
file,
finished: false,
})
}
/// 结束当前分段并开始下一个。当前分段 flush 失败时返回错误,不再开新文件。
pub fn create_new(&mut self) -> std::io::Result<()> {
self.file.rename();
self.finish()?;
let path = self.file.create()?;
self.buf_writer = Self::create(path)?;
self.finished = false;
Ok(())
}
/// flush 并检查错误 → 去掉 `.part` → 触发钩子,见 [`LifecycleFile::finish`]。
fn finish(&mut self) -> std::io::Result<()> {
self.finished = true;
self.file.finish(&mut self.buf_writer)
}
fn create<P: AsRef<std::path::Path>>(path: P) -> std::io::Result<BufWriter<File>> {
let path = path.as_ref();
let out = match File::create(path) {
@@ -99,7 +110,11 @@ impl<'a> FlvFile<'a> {
impl Drop for FlvFile<'_> {
fn drop(&mut self) {
self.file.rename()
if !self.finished
&& let Err(e) = self.finish()
{
error!("{e}");
}
}
}
+48 -3
View File
@@ -6,7 +6,7 @@ use m3u8_rs::{MediaPlaylist, Playlist};
use std::fs::File;
use std::io::{BufWriter, Write};
use std::time::{Duration, Instant};
use tracing::{debug, info, warn};
use tracing::{debug, error, info, warn};
use url::Url;
use crate::client::StatelessClient;
@@ -185,6 +185,8 @@ async fn download_to_file(
pub struct TsFile<'a> {
pub buf_writer: BufWriter<File>,
pub file: LifecycleFile<'a>,
/// 当前分段已交给 [`LifecycleFile::finish`],`Drop` 不再重复改名、触发钩子。
finished: bool,
}
impl<'a> TsFile<'a> {
@@ -193,16 +195,25 @@ impl<'a> TsFile<'a> {
Ok(Self {
buf_writer: Self::create(path)?,
file,
finished: false,
})
}
/// 结束当前分段并开始下一个。当前分段 flush 失败时返回错误,不再开新文件。
pub fn create_new(&mut self) -> std::io::Result<()> {
self.file.rename();
self.finish()?;
let path = self.file.create()?;
self.buf_writer = Self::create(path)?;
self.finished = false;
Ok(())
}
/// flush 并检查错误 → 去掉 `.part` → 触发钩子,见 [`LifecycleFile::finish`]。
fn finish(&mut self) -> std::io::Result<()> {
self.finished = true;
self.file.finish(&mut self.buf_writer)
}
fn create<P: AsRef<std::path::Path>>(path: P) -> std::io::Result<BufWriter<File>> {
let path = path.as_ref();
let out = match File::create(path) {
@@ -221,7 +232,11 @@ impl<'a> TsFile<'a> {
impl Drop for TsFile<'_> {
fn drop(&mut self) {
self.file.rename()
if !self.finished
&& let Err(e) = self.finish()
{
error!("{e}");
}
}
}
@@ -284,6 +299,36 @@ mod tests {
assert_eq!(playlist.segments.len(), 1);
}
/// 分段钩子触发时 `BufWriter` 里的数据已经写进文件:钩子看到的大小就是最终大小。
#[test]
fn the_segment_hook_sees_the_flushed_file() -> Result<(), Box<dyn std::error::Error>> {
use super::TsFile;
use crate::downloader::util::LifecycleFile;
use std::io::Write;
use std::sync::{Arc, Mutex};
let dir = tempfile::tempdir()?;
let seen: Arc<Mutex<Vec<u64>>> = Arc::default();
let file = LifecycleFile::with_hook(dir.path().join("rec").to_str().unwrap(), "ts", {
let seen = seen.clone();
let dir = dir.path().to_path_buf();
move |name: &str| {
let mut seen = seen.lock().unwrap();
seen.push(std::fs::metadata(name).unwrap().len());
std::fs::rename(name, dir.join(format!("seg-{}.ts", seen.len()))).unwrap();
}
});
let mut ts = TsFile::new(file)?;
ts.buf_writer.write_all(&[0x47; 188 * 3])?;
ts.create_new()?;
ts.buf_writer.write_all(&[0x47; 188])?;
drop(ts);
assert_eq!(*seen.lock().unwrap(), vec![188 * 3, 188]);
assert_eq!(std::fs::metadata(dir.path().join("seg-2.ts"))?.len(), 188);
Ok(())
}
#[test]
fn parse_media_playlist_returns_error_for_invalid_content() {
let error = parse_media_playlist(b"not a media playlist")
+153 -43
View File
@@ -40,7 +40,6 @@ pub async fn parse_flv(
mut segment: Segmentable,
mut preview: Option<PreviewSink>,
) -> crate::downloader::error::Result<()> {
let mut flv_tags_cache: Vec<(TagHeader, Bytes, Bytes)> = Vec::new();
// println!("parse_flv Segment: {:?}", segment);
let _previous_tag_size = connection.read_frame(4).await?;
@@ -52,11 +51,71 @@ pub async fn parse_flv(
Bytes::from_static(&preview::flv::FILE_HEADER),
);
}
// let mut downloaded_size = 9 + 4;
let mut gop = GopCache::default();
let result = read_tags(
&mut connection,
&mut out,
&mut segment,
&mut preview,
&mut gop,
)
.await;
// 连接因 EOF、读超时或解析错误结束时,缓存里还留着最后一个 GOP(都是完整读到并解析过的 tag):
// 先写出再返回原结果,否则每次断流、重连、下播都会丢掉最后 1–10 秒
let drained = gop.write_to(&mut out, &mut segment);
match (result, drained) {
(Ok(()), drained) => Ok(drained?),
(Err(e), Err(drained)) => {
warn!("writing the cached GOP after `{e}` failed: {drained}");
Err(e)
}
(Err(e), Ok(())) => Err(e),
}
}
/// 写盘循环的 GOP 缓存:tag 先攒着,遇到下一个关键帧或连接结束时整组落盘。
#[derive(Default)]
struct GopCache {
tags: Vec<(TagHeader, Bytes, Bytes)>,
prev_timestamp: u32,
}
impl GopCache {
fn write_to(
&mut self,
out: &mut FlvFile<'_>,
segment: &mut Segmentable,
) -> std::io::Result<()> {
let Self {
tags,
prev_timestamp,
} = self;
// drain:中途写失败时剩下的 tag 也一并丢弃,不会在收尾时再写一遍造成重复
for (tag_header, flv_tag_data, previous_tag_size_bytes) in tags.drain(..) {
if tag_header.timestamp < *prev_timestamp {
warn!(
"Non-monotonous DTS in output stream; previous: {prev_timestamp}, current: {};",
tag_header.timestamp
);
}
out.write_tag(&tag_header, &flv_tag_data, &previous_tag_size_bytes)?;
segment.increase_size((11 + tag_header.data_size + 4) as u64);
*prev_timestamp = tag_header.timestamp
}
Ok(())
}
}
async fn read_tags(
connection: &mut Connection,
out: &mut FlvFile<'_>,
segment: &mut Segmentable,
preview: &mut Option<PreviewSink>,
gop: &mut GopCache,
) -> crate::downloader::error::Result<()> {
let mut on_meta_data = None;
let mut aac_sequence_header = None;
let mut h264_sequence_header: Option<(TagHeader, Bytes, Bytes)> = None;
let mut prev_timestamp = 0;
let mut create_new = false;
loop {
let tag_header_bytes = connection.read_frame(11).await?;
@@ -178,24 +237,11 @@ pub async fn parse_flv(
..
} => {
let timestamp = flv_tag.header.timestamp as u64;
if prev_timestamp == 0 && timestamp != 0 {
if gop.prev_timestamp == 0 && timestamp != 0 {
segment.set_start_time(Duration::from_millis(timestamp));
}
segment.set_time_position(Duration::from_millis(timestamp));
for (tag_header, flv_tag_data, previous_tag_size_bytes) in &flv_tags_cache {
if tag_header.timestamp < prev_timestamp {
warn!(
"Non-monotonous DTS in output stream; previous: {prev_timestamp}, current: {};",
tag_header.timestamp
);
}
out.write_tag(tag_header, flv_tag_data, previous_tag_size_bytes)?;
segment.increase_size((11 + tag_header.data_size + 4) as u64);
// downloaded_size += (11 + tag_header.data_size + 4) as u64;
prev_timestamp = tag_header.timestamp
// println!("{downloaded_size}");
}
flv_tags_cache.clear();
gop.write_to(out, segment)?;
if segment.needed() || create_new {
segment.set_start_time(Duration::from_millis(timestamp));
@@ -208,7 +254,7 @@ pub async fn parse_flv(
if let Some((meta_header, meta_bytes, previous_meta_tag_size)) =
on_meta_data.as_ref()
{
flv_tags_cache.push((
gop.tags.push((
*meta_header,
meta_bytes.clone(),
previous_meta_tag_size.clone(),
@@ -220,16 +266,13 @@ pub async fn parse_flv(
if let Some((aac_header, aac_bytes, aac_prev_tag_size)) =
aac_sequence_header.as_ref()
{
flv_tags_cache.push((
*aac_header,
aac_bytes.clone(),
aac_prev_tag_size.clone(),
));
gop.tags
.push((*aac_header, aac_bytes.clone(), aac_prev_tag_size.clone()));
}
if !create_new {
// H264SequenceHeader
if let Some(h264_header) = h264_sequence_header.as_ref() {
flv_tags_cache.push(h264_header.clone());
gop.tags.push(h264_header.clone());
} else {
warn!(
"h264_sequence_header not found before segmenting; new segment may be unplayable."
@@ -240,10 +283,12 @@ pub async fn parse_flv(
out.create_new()?;
create_new = false;
}
flv_tags_cache.push((tag_header, bytes.clone(), previous_tag_size.clone()));
gop.tags
.push((tag_header, bytes.clone(), previous_tag_size.clone()));
}
_ => {
flv_tags_cache.push((tag_header, bytes.clone(), previous_tag_size.clone()));
gop.tags
.push((tag_header, bytes.clone(), previous_tag_size.clone()));
}
}
}
@@ -490,19 +535,19 @@ mod tests {
let connection = super::Connection::new(reqwest::Response::from(http_resp));
// 文件名模板不含时间占位符时每个分段都叫同一个名字,会互相覆盖;
// 在 rename 钩子里把每个分段改成带序号的名字留下来(钩子触发时 BufWriter
// 可能还没 flush,所以只改名、等全部结束后再读)
// 在 rename 钩子里把每个分段改成带序号的名字留下来,并记下钩子触发时的文件大小
let dir = tempfile::tempdir()?;
let file_stem = dir.path().join("rolling");
let segments: Arc<Mutex<Vec<std::path::PathBuf>>> = Arc::default();
let segments: Arc<Mutex<Vec<(std::path::PathBuf, u64)>>> = Arc::default();
let file = LifecycleFile::with_hook(file_stem.to_str().unwrap(), "flv", {
let segments = segments.clone();
let dir = dir.path().to_path_buf();
move |name: &str| {
let mut segments = segments.lock().unwrap();
let size_at_hook = std::fs::metadata(name).unwrap().len();
let kept = dir.join(format!("seg-{:04}.flv", segments.len()));
std::fs::rename(name, &kept).unwrap();
segments.push(kept);
segments.push((kept, size_at_hook));
}
});
@@ -557,7 +602,7 @@ mod tests {
);
// 预览 = 文件头 + 序列头 + 从第一个关键帧起的全部 tag = 源流逐字节(源流体不含 9 字节
// 文件头、含 PreviousTagSize0;也含写盘循环还留在缓存里、尚未落盘的最后一个 GOP——预览不等它)
// 文件头、含 PreviousTagSize0)
assert_eq!(
&received[13..],
&body[4..],
@@ -572,22 +617,87 @@ mod tests {
segments.len()
);
let mut written_tags = Vec::new();
for path in segments.iter() {
for (path, size_at_hook) in segments.iter() {
let data = std::fs::read(path)?;
assert_eq!(&data[..13], &crate::downloader::preview::flv::FILE_HEADER);
assert_eq!(
*size_at_hook,
data.len() as u64,
"the hook must see the flushed file ({})",
path.display()
);
written_tags.extend_from_slice(&data[13..]);
}
// 关键帧 NALU 的标记:写盘只有前 119 个(最后一个 GOP 留在缓存里没落盘),预览是全部 120 个
let keyframes = |data: &[u8]| {
data.windows(5)
.filter(|w| *w == [0x17, 0x01, 0, 0, 0])
.count()
};
assert_eq!(keyframes(&written_tags), 119);
// 关键帧 NALU 的标记:连接结束时缓存里的最后一个 GOP 也落盘,写盘与预览都是 120 个
assert_eq!(keyframes(&written_tags), 120);
assert_eq!(keyframes(&received), 120);
Ok(())
}
fn keyframes(data: &[u8]) -> usize {
data.windows(5)
.filter(|w| *w == [0x17, 0x01, 0, 0, 0])
.count()
}
/// 录一段不分段的流,返回 `parse_flv` 的结果与落盘文件内容。
async fn record_unsegmented(
body: reqwest::Body,
) -> (crate::downloader::error::Result<()>, Vec<u8>) {
use crate::downloader::util::{LifecycleFile, Segmentable};
let http_resp = http::Response::builder().status(200).body(body).unwrap();
let connection = super::Connection::new(reqwest::Response::from(http_resp));
let dir = tempfile::tempdir().unwrap();
let file_stem = dir.path().join("rec");
let file = LifecycleFile::new(file_stem.to_str().unwrap(), "flv");
let result = super::parse_flv(connection, file, Segmentable::new(None, None), None).await;
let data = std::fs::read(dir.path().join("rec.flv")).unwrap();
(result, data)
}
/// 下播:源站在 tag 边界干净地关掉连接,最后一个 GOP 照样落盘。
#[tokio::test]
async fn the_last_gop_is_written_when_the_stream_ends() {
let body = gop_flv_body(5);
let (result, data) = record_unsegmented(body.clone().into()).await;
result.unwrap();
assert_eq!(keyframes(&data), 5);
assert_eq!(&data[13..], &body[4..], "every source tag is on disk");
}
/// 断流:读到一半 30 秒没有新数据,读超时照样作为错误返回,但此前完整读到的 GOP 都已落盘。
#[tokio::test(start_paused = true)]
async fn the_last_gop_is_written_when_a_read_times_out() {
use futures::StreamExt;
let body = gop_flv_body(5);
let stalled = futures::stream::iter([Ok::<_, std::io::Error>(body.clone())])
.chain(futures::stream::pending());
let (result, data) = record_unsegmented(reqwest::Body::wrap_stream(stalled)).await;
assert!(
matches!(
result,
Err(crate::downloader::error::Error::ElapsedError(_))
),
"the read timeout must still be reported: {result:?}"
);
assert_eq!(keyframes(&data), 5);
assert_eq!(&data[13..], &body[4..]);
}
/// 解析错误:流里混进一个坏 tag,错误照样返回,坏 tag 之前的 GOP 都已落盘,坏字节不落盘。
#[tokio::test]
async fn the_last_gop_is_written_when_parsing_fails() {
let good = gop_flv_body(5);
let mut body = good.clone();
body.extend_from_slice(&[0xff; 32]);
let (result, data) = record_unsegmented(body.into()).await;
assert!(result.is_err(), "the parse error must still be reported");
assert_eq!(keyframes(&data), 5);
assert_eq!(&data[13..], &good[4..]);
}
/// 一个从不读取的订阅者在场时,录制照常结束、落盘内容与无订阅者时完全一致。
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn stalled_preview_subscriber_does_not_change_what_is_written()
@@ -652,8 +762,8 @@ mod tests {
/// 写盘字节计数:只统计真正经 `write_tag` 落盘的 tag(11 字节头 + 数据 + 4 字节 previous tag size)。
///
/// 这段流里只有 onMetaData 脚本标签会在遇到关键帧时被写出(14 字节数据 → 29 字节),
/// 关键帧本身留在缓存里等待下一个关键帧,流结束时并未落盘,因此不计入。
/// 这段流里 onMetaData 脚本标签在遇到关键帧时写出(14 字节数据 → 29 字节),
/// 关键帧(5 字节数据 → 20 字节)留在缓存里,流结束时一并落盘。
#[tokio::test]
async fn written_bytes_are_counted_on_the_shared_counter()
-> Result<(), Box<dyn std::error::Error>> {
@@ -671,7 +781,7 @@ mod tests {
LifecycleFile::new(file_stem.to_str().unwrap(), "flv").with_counter(counter.clone());
super::parse_flv(connection, file, Segmentable::new(None, None), None).await?;
assert_eq!(counter.total(), 11 + 14 + 4);
assert_eq!(counter.total(), (11 + 14 + 4) + (11 + 5 + 4));
Ok(())
}
}
+44
View File
@@ -1,5 +1,6 @@
use chrono::{DateTime, Local};
use std::fs;
use std::io::Write;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};
@@ -315,6 +316,19 @@ impl<'a> LifecycleFile<'a> {
}
}
}
/// 结束当前分段:先把 `writer` 缓冲的数据写进文件并检查错误,再去掉 `.part` 后缀、触发钩子,
/// 钩子(上传、`FileValidator`)看到的文件大小即最终大小。
///
/// flush 失败(如盘满)时,已经写进文件的部分照常改名交给钩子,不留下没人处理的 `.part`;
/// 错误返回给调用方。
pub fn finish(&mut self, writer: &mut impl Write) -> std::io::Result<()> {
let flushed = writer.flush().map_err(|e| {
std::io::Error::new(e.kind(), format!("flush {}: {e}", self.path.display()))
});
self.rename();
flushed
}
}
pub fn format_filename(file_name: &str) -> String {
@@ -357,6 +371,36 @@ mod tests {
assert_eq!(counter.total(), 16);
}
/// flush 失败时不再静默:错误返回给调用方,已写入的部分照常改名交给钩子。
#[test]
fn finish_reports_a_failed_flush_and_still_hands_the_file_over() {
struct FullDisk;
impl std::io::Write for FullDisk {
fn write(&mut self, _: &[u8]) -> std::io::Result<usize> {
Err(std::io::ErrorKind::StorageFull.into())
}
fn flush(&mut self) -> std::io::Result<()> {
Err(std::io::ErrorKind::StorageFull.into())
}
}
let dir = tempfile::tempdir().unwrap();
let hooked: Arc<std::sync::Mutex<Vec<String>>> = Arc::default();
let mut file = LifecycleFile::with_hook(dir.path().join("rec").to_str().unwrap(), "flv", {
let hooked = hooked.clone();
move |name: &str| hooked.lock().unwrap().push(name.to_string())
});
let part = file.create().unwrap().to_path_buf();
fs::write(&part, b"FLV").unwrap();
let err = file.finish(&mut FullDisk).unwrap_err();
assert_eq!(err.kind(), std::io::ErrorKind::StorageFull);
assert!(err.to_string().contains("rec.flv.part"), "{err}");
assert!(!part.exists());
assert_eq!(*hooked.lock().unwrap(), vec![file.file_name.clone()]);
assert_eq!(fs::read(&file.file_name).unwrap(), b"FLV");
}
#[test]
fn test_segmentation_logic() -> Result<(), Box<dyn std::error::Error>> {
// 测试时间分割
+11
View File
@@ -19,6 +19,17 @@ top = false
- 🔧已修复的问题
- ⚠️需要手动操作的更新信息
## 未发布
- ⚠️没有配置下载插件(`downloader`)的主播,默认下载插件从 stream-gears 改为 mesio。显式配置了 `downloader` 的不受影响;想保持原样,把 `downloader` 设为 `stream-gears` 即可。mesio 写出的 FLV 带关键帧索引(`onMetaData.keyframes`)、每段时间戳从 0 开始,播放器可以拖动进度条;能录 HEVC / Enhanced-FLV 和 B 站 `hls_fmp4`。行为上的变化:
- 不转封装,容器跟随源站:FLV 流存 `.flv`,HLS TS 存 `.ts`,HLS fMP4(如 B 站 `hls_fmp4`)存 `.mp4`;配置的 `format` 与实际容器不一致时只在日志里提示。
- 录制中直接写最终文件名,不再先写 `.part` 再改名。
- 斗鱼网宿 CDN 断开重连时,相邻两段之间会有 2.6–5.8 秒的重复画面(stream-gears 会丢掉这段回灌)。
- 🔧stream-gears:断流、读超时、下播时不再丢掉最后一个 GOP(通常 1–10 秒);分段钩子在文件写完(flush)之后才触发,盘满等写入错误不再被静默吞掉。
- 🔧`ffmpeg-external` / `ffmpeg-internal` 不再被静默换成 stream-gears,改为按 ffmpeg 录制。
- 🔧`segment_time` 在 stream-gears、mesio、ffmpeg 下按同一种写法解析:`HH:MM:SS`、`MM:SS` 或秒数(如 `3600`),和 ffmpeg 一致。以前 stream-gears 遇到 `3600`、`01:00` 这类写法下载任务会 panic 中断,mesio 会静默当成不分段;现在都能正确分段,确实写错的值会在日志里提示并按不分段录制。
- 💡WebUI 的「视频分段大小(file_size)」改成数值 + 单位(MB / GB)输入,按 1024 进制换算(1 GB = 1024 MB,与 Windows 资源管理器一致),配置文件里存的仍是字节数,旧配置原样可读。
- 🔧主播「配置覆写」里把已有的 `file_size` 清空后保存,现在会真正写成「不按大小分段」;以前清空后保存,原来的值原样留着。
## 1.2.1
**Full Changelog**:[v1.2.0...v1.2.1](https://github.com/biliup/biliup/compare/v1.2.0...v1.2.1)
+3 -3
View File
@@ -1,11 +1,11 @@
######### 全局录播与上传设置 #########
#------录制与下载------#
### 选择全局默认的下载插件,默认为stream-gears。可选:
### 选择全局默认的下载插件,默认为mesio。可选:
### 1.streamlink(streamlink配合ffmpeg混合下载模式,适合用于下载hls_fmp4与hls_ts流,因为streamlink支持多线程拉取)。
### 使用该模式下载flv流时,将会仅使用ffmpeg。请手动安装streamlink以及ffmpeg。
### 2.ffmpeg(纯ffmpeg下载),请手动安装ffmpeg。
### 3.stream-gears
### 4.mesio(内置 rust-srec 引擎,进程内下载 FLV/HLS 并修复时间戳、注入关键帧索引,支持按大小/时长分段,无需额外安装)
### 3.stream-gears(不支持 HEVC 与 hls_fmp4,FLV 时间戳沿用源站原值)
### 4.mesio(默认。内置 rust-srec 引擎,进程内下载 FLV/HLS 并修复时间戳、注入关键帧索引,支持按大小/时长分段,无需额外安装)
#downloader = "ffmpeg"
### 录像单文件大小限制,单位Byte,超过此大小分段下载,下载回放时无法使用
file_size = 2621440000
+3 -3
View File
@@ -1,11 +1,11 @@
######### 全局录播与上传设置 #########
#------录制与下载------#
### 选择全局默认的下载插件,默认为stream-gears。可选:
### 选择全局默认的下载插件,默认为mesio。可选:
### 1.streamlink(streamlink配合ffmpeg混合下载模式,适合用于下载hls_fmp4与hls_ts流,因为streamlink支持多线程拉取)。
### 使用该模式下载flv流时,将会仅使用ffmpeg。请手动安装streamlink以及ffmpeg。
### 2.ffmpeg(纯ffmpeg下载),请手动安装ffmpeg。
### 3.stream-gears
### 4.mesio(内置 rust-srec 引擎,进程内下载 FLV/HLS 并修复时间戳、注入关键帧索引,支持按大小/时长分段,无需额外安装)
### 3.stream-gears(不支持 HEVC 与 hls_fmp4,FLV 时间戳沿用源站原值)
### 4.mesio(默认。内置 rust-srec 引擎,进程内下载 FLV/HLS 并修复时间戳、注入关键帧索引,支持按大小/时长分段,无需额外安装)
#downloader: ffmpeg
### 录像单文件大小限制,单位Byte,超过此大小分段下载,下载回放时无法使用
file_size: 2621440000