euzu/tuliprox
1
0
Fork 0
mirror of https://github.com/euzu/tuliprox.git synced 2026-10-01 21:42:06 +02:00
Code Issues Packages Projects Releases 5 Wiki Activity

Merge pull request #362 from euzu/feature/epg_fix_and_resource_handling_optimization

Browse Source 
Some little fixes for epg_view and stream task yielding for better async operations
This commit is contained in:
euzu
2025-10-10 17:52:12 +02:00
committed by GitHub
parent 3dea155475 c286a88f50
commit 2cabec79ea
6 changed files with 49 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+1
View File
@@ -1197,6 +1197,7 @@ dependencies = [
"cron",
"futures",
"futures-signals",
"gloo-render 0.2.0",
"gloo-storage 0.3.0",
"gloo-timers 0.3.0",
"gloo-utils 0.2.0",
backend/src/api/model/streams/buffered_stream.rs
+11 -11
View File
@@ -30,20 +30,20 @@ impl BufferedStream {
mut stream: BoxedProviderStream,
client_close_signal: Arc<AtomicOnceFlag>,
) {
loop {
if !client_close_signal.is_active() {
break;
}
while client_close_signal.is_active() {
match stream.next().await {
Some(Ok(chunk)) => {
match tx.reserve().await {
Ok(permit) => permit.send(Ok(chunk)),
Err(_err) => {
// Receiver dropped, notify and exit
client_close_signal.notify();
break;
}
match tx.reserve().await {
Ok(permit) => {
permit.send(Ok(chunk));
tokio::task::yield_now().await;
},
Err(_err) => {
// Receiver dropped, notify and exit
client_close_signal.notify();
break;
}
}
}
Some(Err(err)) => {
//trace!("Buffered Stream Error: {err:?}");
backend/src/api/model/streams/shared_stream_manager.rs
+1
View File
@@ -107,6 +107,7 @@ impl SharedStreamState {
debug!("Shared stream client send error: {address} {err}");
break;
}
tokio::task::yield_now().await;
}
Err(tokio::sync::broadcast::error::RecvError::Lagged(skipped)) => {
trace!("Client lagged behind. Skipped {skipped} messages. {address}");
backend/src/api/model/streams/throttled_stream.rs
+5 -2
View File
@@ -56,7 +56,8 @@ where
match Pin::new(&mut this.inner).poll_next(cx) {
Poll::Ready(Some(Ok(bytes))) => {
let len = bytes.len() as f64;
let delay_duration = Duration::from_secs_f64(len / this.rate_bytes_per_sec);
let delay_secs = (len / this.rate_bytes_per_sec).max(0.001);
let delay_duration = Duration::from_secs_f64(delay_secs);
// Schedule the next delay
this.next_delay = Some(Box::pin(sleep(delay_duration)));
@@ -71,4 +72,6 @@ where
Poll::Pending => Poll::Pending,
}
}
}
}
impl<S: Unpin> Unpin for ThrottledStream<S> {}
frontend/Cargo.toml
+3 -2
View File
@@ -18,8 +18,9 @@ thiserror = "2"
reqwasm = {version = "0.5", features = ["json"]}
log = "0.4"
wasm-logger = "0.2"
gloo-storage = "0"
gloo-utils = "0"
gloo-storage = "0.3"
gloo-utils = "0.2"
gloo-render = "0.2"
gloo-timers = { version = "0", features = ["futures"] }
implicit-clone = "0"
futures-signals = "0.3"
frontend/src/app/components/playlist/epg_view.rs
+28 -19
View File
@@ -99,7 +99,7 @@ pub fn EpgView() -> Html {
.unwrap_or_else(|_| String::new()); // fallback if not set
row_height.trim_end_matches("px").parse::<usize>().unwrap_or(60)
row_height.trim_end_matches("px").parse::<usize>().unwrap_or(60).max(1)
});
// Add scroll listener to calculate visible channels
@@ -108,37 +108,46 @@ pub fn EpgView() -> Html {
let visible_range = visible_range.clone();
let channel_row_height = *row_height;
use_effect_with((), move |_| {
let debounce_handle: Rc<RefCell<Option<Timeout>>> = Rc::new(RefCell::new(None));
let onscroll_handle: Rc<RefCell<Option<Closure<dyn FnMut(_)>>>> = Rc::new(RefCell::new(None));
if let Some(div) = container_ref.cast::<HtmlElement>() {
let visible_range = visible_range.clone();
// Store debounce timer in Rc<RefCell>
let debounce_handle: Rc<RefCell<Option<Timeout>>> = Rc::new(RefCell::new(None));
let debounce_handle_clone = debounce_handle.clone();
let onscroll_handle_clone = onscroll_handle.clone();
let onscroll = Closure::wrap(Box::new(move |_event: web_sys::Event| {
// Cancel previous scheduled update
if let Some(prev) = debounce_handle_clone.borrow_mut().take() {
prev.cancel();
}
if let Some(div) = container_ref.cast::<HtmlElement>() {
let scroll_top = div.scroll_top();
let client_height = div.client_height(); // Calculate which channel rows are visible
let vr = visible_range.clone();
// Schedule a new update after X ms (debounce)
let container_ref = container_ref.clone();
let vr = visible_range.clone();
let handle = Timeout::new(16, move || {
if let Some(div) = container_ref.cast::<HtmlElement>() {
let scroll_top = div.scroll_top();
let client_height = div.client_height(); // Calculate which channel rows are visible
// Schedule a new update after 80ms (debounce)
let handle = Timeout::new(80, move || {
let start_index = (scroll_top / (channel_row_height as i32)).max(0);
let end_index = ((scroll_top + client_height) / (channel_row_height as i32) + 1).max(0);
// render 10 + 10 more lines
let start_index = (scroll_top / (channel_row_height as i32) - 10).max(0);
let end_index = ((scroll_top + client_height) / (channel_row_height as i32) + 10).max(0);
vr.set((start_index as usize, end_index as usize));
});
}
});
*debounce_handle_clone.borrow_mut() = Some(handle);
}
*debounce_handle_clone.borrow_mut() = Some(handle);
}) as Box<dyn FnMut(_)>);
div.add_event_listener_with_callback("scroll", onscroll.as_ref().unchecked_ref()).unwrap();
onscroll.forget();
*onscroll_handle_clone.borrow_mut() = Some(onscroll);
}
move || {
if let Some(prev) = debounce_handle.borrow_mut().take() {
prev.cancel();
}
if let Some(onscroll) = onscroll_handle.borrow_mut().take() {
drop(onscroll);
}
}
|| {}
});
}
@@ -174,7 +183,7 @@ pub fn EpgView() -> Html {
<div style={format!("height:{}px", start_index * channel_row_height)}></div>
{ for tv.channels.iter().enumerate().skip(start_index).take(end_index - start_index).map(|(_i, ch)| {
html! {
<div class="tp__epg__channel" style={format!("max-height:{channel_row_height}px;min-height={channel_row_height}px;height:{channel_row_height}px")}>
<div class="tp__epg__channel" style={format!("max-height:{channel_row_height}px;min-height:{channel_row_height}px;height:{channel_row_height}px")}>
<div class="tp__epg__channel-icon">
{ if let Some(icon) = &ch.icon {
html! { <img src={icon.clone()} alt={ch.title.clone()} /> }
@@ -213,7 +222,7 @@ pub fn EpgView() -> Html {
<div style={format!("height:{}px", start_index * channel_row_height)}></div>
{ for tv.channels.iter().enumerate().skip(start_index).take(end_index - start_index).map(|(_i, ch)| {
html! {
<div class="tp__epg__channel-programs" style={format!("max-height:{channel_row_height}px;min-height={channel_row_height}px;height:{channel_row_height}px")}>
<div class="tp__epg__channel-programs" style={format!("max-height:{channel_row_height}px;min-height:{channel_row_height}px;height:{channel_row_height}px")}>
{ for ch.programmes.iter().map(|p| {
let is_active = now >= p.start && now < p.stop;
let left = get_pos(p.start, start_window);