GithubFollow/One-KVM
1
0
Fork 0
mirror of https://github.com/mofeng-git/One-KVM.git synced 2026-10-28 00:10:50 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

fix: 完善 RK3588 HDMI RX 信号检测与自动恢复

Browse Source 
- 统一使用 QUERY_DV_TIMINGS 判断 HDMI RX 输入状态
  - 增加 source-following 设备状态 API 与前端只读展示
  - 支持长时间无信号后的 MJPEG/WebRTC 自动恢复
  - 修复模式切换时采集设备尚未释放导致的 EBUSY
  - 取消陈旧 WebRTC 重连并统一采集恢复策略
  - 移除视频输入状态区域的冗余标题
This commit is contained in:
mofeng-git
2026-07-30 19:41:44 +08:00
parent 971c263bf8
commit 9fb23476ac
36 changed files with 1883 additions and 1492 deletions
Download Patch File Download Diff File Expand all files Collapse all files

482
src/video/pipeline/shared.rs
View File

@@ -32,10 +32,7 @@ use super::encoder_state::{build_encoder_state, EncoderThreadState};
const AUTO_STOP_GRACE_PERIOD_SECS: u64 = 3;
/// After this many consecutive timeouts, log a prominent warning.
const CAPTURE_TIMEOUT_RESTART_THRESHOLD: u32 = 5;
const CAPTURE_TIMEOUT_STOP_THRESHOLD: u32 = 60;
const CAPTURE_TIMEOUT_SOFT_RESTART_THRESHOLD: u32 = 3;
const CSI_BRIDGE_NOSIGNAL_INTERVAL_MS: u64 = 500;
const NOSIGNAL_POLL_MAX: Duration = Duration::from_secs(20);
/// Throttle repeated encoding errors to avoid log flooding
const ENCODE_ERROR_THROTTLE_SECS: u64 = 5;
@@ -53,10 +50,11 @@ use crate::video::capture::status::{
use crate::video::capture::{is_source_changed_error, BridgeContext, CaptureStream};
use crate::video::codec::h264_bitstream;
use crate::video::codec::registry::{EncoderBackend, VideoEncoderType};
use crate::video::device::bridge::{self as csi_bridge, ProbeResult};
use crate::video::device::parse_bridge_kind;
use crate::video::device::VideoControlMode;
use crate::video::format::{PixelFormat, Resolution};
use crate::video::frame::{FrameBuffer, FrameBufferPool, VideoFrame};
use crate::video::recovery::{wait_for_source_change, CaptureRecoveryPolicy};
use crate::video::signal::SignalStatus;
const MIN_CAPTURE_FRAME_SIZE: usize = 128;
@@ -90,14 +88,27 @@ pub struct PipelineStateNotification {
pub state: &'static str,
pub reason: Option<&'static str>,
pub next_retry_ms: Option<u64>,
pub applied_config: Option<PipelineAppliedConfig>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct PipelineAppliedConfig {
pub resolution: Resolution,
pub format: PixelFormat,
pub fps: u32,
}
impl PipelineStateNotification {
fn streaming() -> Self {
fn streaming(resolution: Resolution, format: PixelFormat, fps: u32) -> Self {
Self {
state: "streaming",
reason: None,
next_retry_ms: None,
applied_config: Some(PipelineAppliedConfig {
resolution,
format,
fps,
}),
}
}
@@ -106,14 +117,7 @@ impl PipelineStateNotification {
state: "no_signal",
reason: Some(status.as_str()),
next_retry_ms,
}
}
fn device_busy(reason: &'static str) -> Self {
Self {
state: "device_busy",
reason: Some(reason),
next_retry_ms: None,
applied_config: None,
}
}
}
@@ -121,6 +125,8 @@ impl PipelineStateNotification {
/// Shared video pipeline configuration
#[derive(Debug, Clone)]
pub struct SharedVideoPipelineConfig {
/// Whether the capture mode is configured by the client or follows HDMI.
pub control_mode: VideoControlMode,
/// Input resolution
pub resolution: Resolution,
/// Input pixel format
@@ -138,6 +144,7 @@ pub struct SharedVideoPipelineConfig {
impl Default for SharedVideoPipelineConfig {
fn default() -> Self {
Self {
control_mode: VideoControlMode::Configurable,
resolution: Resolution::HD720,
input_format: PixelFormat::Yuyv,
output_codec: VideoEncoderType::H264,
@@ -285,81 +292,6 @@ pub struct SharedVideoPipeline {
last_state_notification: ParkingMutex<Option<PipelineStateNotification>>,
}
fn poll_bridge_subdev_after_no_signal(bridge_ctx: &BridgeContext, pipeline: &SharedVideoPipeline) {
let Some(subdev_path) = bridge_ctx.subdev_path.as_ref() else {
return;
};
let kind = bridge_ctx
.kind
.unwrap_or(csi_bridge::CsiBridgeKind::Unknown);
let deadline = Instant::now() + NOSIGNAL_POLL_MAX;
let mut poll_count: u32 = 0;
info!(
"No-signal poll: scanning subdev {:?} every {} ms (max {:?})",
subdev_path, CSI_BRIDGE_NOSIGNAL_INTERVAL_MS, NOSIGNAL_POLL_MAX
);
loop {
if !pipeline.running_flag.load(Ordering::Acquire) {
return;
}
if Instant::now() >= deadline {
info!(
"No-signal poll: stopped after {:?} ({} attempts)",
NOSIGNAL_POLL_MAX, poll_count
);
return;
}
let fd = match csi_bridge::open_subdev(subdev_path) {
Ok(f) => f,
Err(e) => {
debug!(
"No-signal poll: open subdev {:?} failed: {}",
subdev_path, e
);
std::thread::sleep(Duration::from_millis(CSI_BRIDGE_NOSIGNAL_INTERVAL_MS));
continue;
}
};
match csi_bridge::probe_signal_thread_timeout(
&fd,
kind,
csi_bridge::RK628_SUBDEV_PROBE_TIMEOUT,
) {
Some(ProbeResult::Locked(mode)) => {
info!(
"No-signal poll: locked {}x{} @ {} Hz — proceeding to capture re-open",
mode.width, mode.height, mode.pixelclock
);
return;
}
Some(other) => {
poll_count = poll_count.saturating_add(1);
if poll_count == 1 || poll_count.is_multiple_of(8) {
debug!(
"No-signal poll: attempt {} — still {:?}",
poll_count,
other.as_status()
);
}
if let Some(st) = other.as_status() {
pipeline.notify_state(PipelineStateNotification::no_signal(
st,
Some(CSI_BRIDGE_NOSIGNAL_INTERVAL_MS.saturating_add(50)),
));
}
}
None => {
poll_count = poll_count.saturating_add(1);
debug!(
"No-signal poll: attempt {} — probe ioctl timed out",
poll_count
);
}
}
std::thread::sleep(Duration::from_millis(CSI_BRIDGE_NOSIGNAL_INTERVAL_MS));
}
}
impl SharedVideoPipeline {
/// Create a new shared video pipeline
pub fn new(config: SharedVideoPipelineConfig) -> Result<Arc<Self>> {
@@ -575,7 +507,6 @@ impl SharedVideoPipeline {
_jpeg_quality: u8,
subdev_path: Option<std::path::PathBuf>,
bridge_kind: Option<String>,
_v4l2_driver: Option<String>,
) -> Result<()> {
if *self.running_rx.borrow() {
warn!("Pipeline already running");
@@ -601,24 +532,33 @@ impl SharedVideoPipeline {
buffer_count.max(1),
Duration::from_secs(2),
bridge_ctx_probe,
config.control_mode,
) {
Ok(s) => {
let negotiated_res = s.resolution();
let negotiated_fmt = s.format();
if negotiated_res != config.resolution || negotiated_fmt != config.input_format {
let previous = (config.resolution, config.input_format, config.fps);
if config.control_mode == VideoControlMode::SourceFollowing {
if let Some(source_fps) = s.source_fps() {
config.fps = source_fps.round().clamp(1.0, 120.0) as u32;
}
}
config.resolution = negotiated_res;
config.input_format = negotiated_fmt;
if previous != (config.resolution, config.input_format, config.fps) {
info!(
"Negotiated capture {}x{} {:?} (configured {}x{} {:?}) — aligning encoder to source",
"Negotiated capture {}x{} {:?} @ {} fps (configured {}x{} {:?} @ {} fps) — aligning encoder to source",
negotiated_res.width,
negotiated_res.height,
negotiated_fmt,
config.resolution.width,
config.resolution.height,
config.input_format
config.fps,
previous.0.width,
previous.0.height,
previous.1,
previous.2,
);
config.resolution = negotiated_res;
config.input_format = negotiated_fmt;
*self.config.write().await = config.clone();
}
*self.config.write().await = config.clone();
Some(s)
}
Err(AppError::CaptureNoSignal { kind }) => {
@@ -628,7 +568,11 @@ impl SharedVideoPipeline {
let status = signal_status_from_capture_kind(&kind);
self.notify_state(PipelineStateNotification::no_signal(
status,
Some(Duration::from_secs(2).as_millis() as u64),
Some(
CaptureRecoveryPolicy::new(config.control_mode)
.retry_delay(1)
.as_millis() as u64,
),
));
None
}
@@ -754,74 +698,19 @@ impl SharedVideoPipeline {
let mut initial_geometry: Option<(Resolution, PixelFormat)> = None;
let mut resolution = config.resolution;
let mut pixel_format = config.input_format;
let mut active_fps = config.fps;
let mut stride: u32 = 0;
match preopened {
Some(s) => {
resolution = s.resolution();
pixel_format = s.format();
stride = s.stride();
initial_geometry = Some((resolution, pixel_format));
stream = Some(s);
}
None => {
match open_capture_stream(
&device_path,
config.resolution,
config.input_format,
config.fps,
buffer_count.max(1),
Duration::from_secs(2),
bridge_ctx.clone(),
) {
Ok(s) => {
resolution = s.resolution();
pixel_format = s.format();
stride = s.stride();
if resolution != config.resolution
|| pixel_format != config.input_format
{
info!(
"First capture open negotiated {}x{} {:?} but encoder expects {}x{} {:?} — stopping for dimension resync",
resolution.width,
resolution.height,
pixel_format,
config.resolution.width,
config.resolution.height,
config.input_format
);
pipeline.notify_state(PipelineStateNotification::device_busy(
"config_changing",
));
*pipeline.pending_sync_geometry.lock() =
Some((resolution, pixel_format));
let _ = pipeline.running.send(false);
pipeline.running_flag.store(false, Ordering::Release);
let _ = frame_seq_tx.send(1);
return;
}
initial_geometry = Some((resolution, pixel_format));
stream = Some(s);
}
Err(AppError::CaptureNoSignal { kind }) => {
warn!(
"Capture stream open reports no signal ({}) — pipeline will retry",
kind
);
pipeline.notify_state(PipelineStateNotification::no_signal(
signal_status_from_capture_kind(&kind),
Some(CSI_BRIDGE_NOSIGNAL_INTERVAL_MS),
));
}
Err(e) => {
error!("Failed to open capture stream: {}", e);
let _ = pipeline.running.send(false);
pipeline.running_flag.store(false, Ordering::Release);
let _ = frame_seq_tx.send(1);
return;
}
}
}
if let Some(s) = preopened {
resolution = s.resolution();
pixel_format = s.format();
active_fps = s
.source_fps()
.map(|fps| fps.round().clamp(1.0, 120.0) as u32)
.unwrap_or(config.fps);
stride = s.stride();
initial_geometry = Some((resolution, pixel_format));
stream = Some(s);
}
fn open_or_retry(
@@ -838,6 +727,7 @@ impl SharedVideoPipeline {
buffer_count.max(1),
Duration::from_secs(2),
bridge_ctx,
config.control_mode,
is_device_lost_message,
) {
CaptureOpenResult::NoSignal(status) => {
@@ -860,6 +750,7 @@ impl SharedVideoPipeline {
let grace_period = Duration::from_secs(AUTO_STOP_GRACE_PERIOD_SECS);
let mut sequence: u64 = 0;
let mut consecutive_timeouts: u32 = 0;
let recovery_policy = CaptureRecoveryPolicy::new(config.control_mode);
let capture_error_throttler = LogThrottler::with_secs(5);
let mut suppressed_capture_errors: HashMap<String, u64> = HashMap::new();
@@ -877,9 +768,7 @@ impl SharedVideoPipeline {
"No subscribers for {}s, auto-stopping video pipeline",
grace_period.as_secs()
);
let _ = pipeline.running.send(false);
pipeline.running_flag.store(false, Ordering::Release);
let _ = frame_seq_tx.send(sequence.wrapping_add(1));
break;
}
}
@@ -899,6 +788,10 @@ impl SharedVideoPipeline {
let new_res = new_stream.resolution();
let new_fmt = new_stream.format();
let new_stride = new_stream.stride();
let new_fps = new_stream
.source_fps()
.map(|fps| fps.round().clamp(1.0, 120.0) as u32)
.unwrap_or(config.fps);
// Pre-probe was skipped (no signal at pipeline start) but the
// encoder was sized to saved settings — if DV timings now
@@ -916,14 +809,13 @@ impl SharedVideoPipeline {
config.resolution.height,
config.input_format
);
pipeline.notify_state(PipelineStateNotification::device_busy(
"config_changing",
pipeline.notify_state(PipelineStateNotification::no_signal(
SignalStatus::NoSignal,
Some(recovery_policy.retry_delay(1).as_millis() as u64),
));
*pipeline.pending_sync_geometry.lock() =
Some((new_res, new_fmt));
let _ = pipeline.running.send(false);
pipeline.running_flag.store(false, Ordering::Release);
let _ = frame_seq_tx.send(sequence.wrapping_add(1));
break;
}
@@ -944,15 +836,15 @@ impl SharedVideoPipeline {
orig_res, orig_fmt, new_res, new_fmt
);
pipeline.notify_state(
PipelineStateNotification::device_busy(
"config_changing",
PipelineStateNotification::no_signal(
SignalStatus::NoSignal,
Some(recovery_policy.retry_delay(1).as_millis()
as u64),
),
);
*pipeline.pending_sync_geometry.lock() =
Some((new_res, new_fmt));
let _ = pipeline.running.send(false);
pipeline.running_flag.store(false, Ordering::Release);
let _ = frame_seq_tx.send(sequence.wrapping_add(1));
break;
}
_ => {}
@@ -963,6 +855,7 @@ impl SharedVideoPipeline {
}
resolution = new_res;
pixel_format = new_fmt;
active_fps = new_fps;
stride = new_stride;
stream = Some(new_stream);
consecutive_timeouts = 0;
@@ -973,36 +866,34 @@ impl SharedVideoPipeline {
}
CaptureOpenResult::NoSignal(status) => {
consecutive_timeouts = consecutive_timeouts.saturating_add(1);
if consecutive_timeouts >= CAPTURE_TIMEOUT_STOP_THRESHOLD {
if !recovery_policy.should_retry(consecutive_timeouts) {
warn!(
"Capture soft-restart gave up after {} attempts, \
stopping pipeline",
consecutive_timeouts
);
let _ = pipeline.running.send(false);
pipeline.running_flag.store(false, Ordering::Release);
let _ = frame_seq_tx.send(sequence.wrapping_add(1));
break;
}
let wait_ms = CSI_BRIDGE_NOSIGNAL_INTERVAL_MS;
let delay = recovery_policy.retry_delay(consecutive_timeouts);
pipeline.notify_state(PipelineStateNotification::no_signal(
status,
Some(wait_ms),
Some(delay.as_millis() as u64),
));
std::thread::sleep(Duration::from_millis(wait_ms));
if wait_for_source_change(&bridge_ctx, delay, || {
pipeline.running_flag.load(Ordering::Acquire)
}) {
info!("SOURCE_CHANGE woke capture retry");
}
continue;
}
CaptureOpenResult::DeviceLost(reason) => {
pipeline.mark_device_lost(reason);
let _ = pipeline.running.send(false);
pipeline.running_flag.store(false, Ordering::Release);
let _ = frame_seq_tx.send(sequence.wrapping_add(1));
break;
}
CaptureOpenResult::Fatal => {
let _ = pipeline.running.send(false);
pipeline.running_flag.store(false, Ordering::Release);
let _ = frame_seq_tx.send(sequence.wrapping_add(1));
break;
}
}
@@ -1024,100 +915,37 @@ impl SharedVideoPipeline {
// drop the stream so the next iteration re-opens
// via a fresh DV_TIMINGS probe. This is the fast
// path for source-side resolution switches on
// RK628 / rkcif — sub-second recovery vs. the ~8 s
// timeout fallback.
// RK628 / rkcif; the retry policy is only a fallback
// when a driver does not provide usable events.
if is_source_changed_error(&e) {
info!(
"Capture reported SOURCE_CHANGE — \
dropping stream for immediate re-open"
);
if recovery_policy.control_mode()
== VideoControlMode::SourceFollowing
{
pipeline.notify_state(PipelineStateNotification::no_signal(
SignalStatus::NoSignal,
Some(recovery_policy.retry_delay(1).as_millis() as u64),
));
}
consecutive_timeouts = 0;
stream = None;
continue;
}
if e.kind() == std::io::ErrorKind::TimedOut {
consecutive_timeouts = consecutive_timeouts.saturating_add(1);
let probe_result = {
let sr = stream.as_mut().expect("stream is Some above");
sr.probe_bridge_signal_with_timeout(
csi_bridge::RK628_SUBDEV_PROBE_TIMEOUT,
)
};
match probe_result {
Some(ProbeResult::Locked(mode)) => {
let probed_resolution =
Resolution::new(mode.width, mode.height);
if probed_resolution == resolution {
info!(
"Capture timeout but bridge is locked at {}x{} — soft-restarting capture without encoder rebuild",
probed_resolution.width,
probed_resolution.height
);
} else {
info!(
"Capture timeout probe detected geometry change {}x{} -> {}x{} — soft-restarting capture for encoder rebuild",
resolution.width,
resolution.height,
probed_resolution.width,
probed_resolution.height
);
pipeline.notify_state(
PipelineStateNotification::device_busy(
"config_changing",
),
);
}
consecutive_timeouts = 0;
stream = None;
continue;
}
Some(other) => {
let status =
other.as_status().unwrap_or(SignalStatus::NoSignal);
warn!(
"Capture timeout probe reports no signal ({})",
status.as_str()
);
pipeline.notify_state(
PipelineStateNotification::no_signal(
status,
Some(Duration::from_secs(2).as_millis() as u64),
),
);
// Drop capture so RK628 / rkcif can release the queue,
// then poll subdev on a fresh fd until timings lock (or
// timeout). Avoids sitting on DQBUF 2s × N with a dead
// stream while `v4l2-ctl --query-dv-timings` already shows
// a real mode.
stream = None;
consecutive_timeouts = 0;
if bridge_ctx.has_subdev()
&& matches!(
other,
ProbeResult::NoSignal
| ProbeResult::NoSync
| ProbeResult::OutOfRange
)
{
poll_bridge_subdev_after_no_signal(
&bridge_ctx,
&pipeline,
);
}
continue;
}
None if bridge_ctx.has_subdev() => {
warn!(
"DV-timings probe timed out or failed — forcing stream re-open (RK628 / rkcif)"
);
consecutive_timeouts = 0;
stream = None;
poll_bridge_subdev_after_no_signal(&bridge_ctx, &pipeline);
continue;
}
None => {
warn!("Capture timeout - no signal?");
}
if recovery_policy.control_mode()
== VideoControlMode::SourceFollowing
{
let delay = recovery_policy.retry_delay(consecutive_timeouts);
pipeline.notify_state(PipelineStateNotification::no_signal(
SignalStatus::NoSignal,
Some(delay.as_millis() as u64),
));
stream = None;
continue;
}
if consecutive_timeouts >= CAPTURE_TIMEOUT_SOFT_RESTART_THRESHOLD {
@@ -1145,17 +973,6 @@ impl SharedVideoPipeline {
consecutive_timeouts
);
}
if consecutive_timeouts >= CAPTURE_TIMEOUT_STOP_THRESHOLD {
warn!(
"Capture timed out {} consecutive times, stopping video pipeline",
consecutive_timeouts
);
let _ = pipeline.running.send(false);
pipeline.running_flag.store(false, Ordering::Release);
let _ = frame_seq_tx.send(sequence.wrapping_add(1));
break;
}
} else {
consecutive_timeouts = 0;
// EIO (5) / EPIPE (32) / EPROTO (71) in next_into generally
@@ -1187,9 +1004,7 @@ impl SharedVideoPipeline {
CaptureIoErrorKind::DeviceLost => {
error!("Capture device lost: {}", e);
pipeline.mark_device_lost(e.to_string());
let _ = pipeline.running.send(false);
pipeline.running_flag.store(false, Ordering::Release);
let _ = frame_seq_tx.send(sequence.wrapping_add(1));
break;
}
CaptureIoErrorKind::Other => {}
@@ -1223,7 +1038,11 @@ impl SharedVideoPipeline {
owned.truncate(frame_size);
// Notify streaming only after the short-frame guard passes.
pipeline.notify_state(PipelineStateNotification::streaming());
pipeline.notify_state(PipelineStateNotification::streaming(
resolution,
pixel_format,
active_fps,
));
let frame = Arc::new(VideoFrame::from_pooled(
Arc::new(FrameBuffer::new(owned, Some(buffer_pool.clone()))),
resolution,
@@ -1240,10 +1059,14 @@ impl SharedVideoPipeline {
let _ = frame_seq_tx.send(sequence);
}
// `running` represents completed lifecycle state, not a stop request.
// Drop the V4L2 stream first so STREAMOFF, buffer teardown and FD close
// have all completed before another consumer is told the device is free.
drop(stream);
pipeline.running_flag.store(false, Ordering::Release);
let _ = pipeline.running.send(false);
let _ = frame_seq_tx.send(sequence.wrapping_add(1));
info!("Video pipeline stopped");
let _ = pipeline.running.send(false);
info!("Video pipeline stopped and capture device released");
});
}
@@ -1458,9 +1281,7 @@ impl SharedVideoPipeline {
/// Stop the pipeline (non-blocking, does not wait for capture thread to exit)
pub fn stop(&self) {
if *self.running_rx.borrow() {
let _ = self.running.send(false);
self.running_flag.store(false, Ordering::Release);
if self.running_flag.swap(false, Ordering::AcqRel) {
self.clear_cmd_tx();
info!("Stopping video pipeline");
}
@@ -1471,32 +1292,38 @@ impl SharedVideoPipeline {
/// This ensures the V4L2 device is released before returning, which is
/// necessary when another consumer (e.g. MJPEG streamer) needs to open
/// the same device immediately after.
pub async fn stop_and_wait(&self, timeout: std::time::Duration) {
pub async fn stop_and_wait(&self, timeout: std::time::Duration) -> Result<()> {
self.stop();
let mut rx = self.running_watch();
if !*rx.borrow() {
// Capture thread may still be running from a previous `stop()` call.
// Wait for the "Video pipeline stopped" log (thread sets running=false
// at exit), unless it already happened.
}
let deadline = tokio::time::Instant::now() + timeout;
loop {
if !self.running_flag.load(Ordering::Acquire) {
// Flag is cleared, but the capture thread may still be unwinding
// (dropping the V4L2 stream). Give it a brief moment.
tokio::time::sleep(std::time::Duration::from_millis(50)).await;
break;
}
while *rx.borrow() {
let remaining = deadline.saturating_duration_since(tokio::time::Instant::now());
if remaining.is_zero() {
warn!(
"Timed out waiting for video pipeline to stop after {:?}",
return Err(AppError::VideoError(format!(
"Timed out waiting {:?} for video pipeline to release capture device",
timeout
);
break;
)));
}
match tokio::time::timeout(remaining, rx.changed()).await {
Ok(Ok(())) => {}
Ok(Err(_)) if !*rx.borrow() => break,
Ok(Err(_)) => {
return Err(AppError::VideoError(
"Video pipeline lifecycle channel closed before capture device release"
.to_string(),
));
}
Err(_) => {
return Err(AppError::VideoError(format!(
"Timed out waiting {:?} for video pipeline to release capture device",
timeout
)));
}
}
let _ = tokio::time::timeout(remaining, rx.changed()).await;
}
Ok(())
}
/// Set bitrate using preset
@@ -1696,7 +1523,7 @@ fn copy_rows(
impl Drop for SharedVideoPipeline {
fn drop(&mut self) {
let _ = self.running.send(false);
self.running_flag.store(false, Ordering::Release);
}
}
@@ -1765,4 +1592,49 @@ mod tests {
let h265 = SharedVideoPipelineConfig::h265(Resolution::HD720, BitratePreset::Speed);
assert_eq!(h265.output_codec, VideoEncoderType::H265);
}
#[test]
fn stop_request_does_not_publish_worker_exit() {
let pipeline = SharedVideoPipeline::new(SharedVideoPipelineConfig::h264(
Resolution::HD720,
BitratePreset::Balanced,
))
.unwrap();
let _ = pipeline.running.send(true);
pipeline.running_flag.store(true, Ordering::Release);
pipeline.stop();
assert!(!pipeline.running_flag.load(Ordering::Acquire));
assert!(pipeline.is_running());
// Simulate the capture thread's common cleanup tail.
let _ = pipeline.running.send(false);
assert!(!pipeline.is_running());
}
#[tokio::test]
async fn stop_and_wait_observes_completed_worker_cleanup() {
let pipeline = SharedVideoPipeline::new(SharedVideoPipelineConfig::h264(
Resolution::HD720,
BitratePreset::Balanced,
))
.unwrap();
let _ = pipeline.running.send(true);
pipeline.running_flag.store(true, Ordering::Release);
let worker = pipeline.clone();
tokio::spawn(async move {
tokio::time::sleep(Duration::from_millis(30)).await;
let _ = worker.running.send(false);
});
let started = Instant::now();
pipeline
.stop_and_wait(Duration::from_secs(1))
.await
.unwrap();
assert!(started.elapsed() >= Duration::from_millis(20));
assert!(!pipeline.is_running());
}
}