1pub mod traits {
2 use std::io::{Read, Seek};
3 use symphonia::core::io::MediaSource;
4 pub trait AudioSource: Read + Seek + MediaSource + Send {
5 fn content_type(&self) -> Option<String> {
6 None
7 }
8 fn seekable(&self) -> bool {
9 self.is_seekable()
10 }
11 }
12}
13pub mod segmented {
14 use super::AudioSource;
15 use crate::{
16 audio::constants::{
17 CHUNK_SIZE, FETCH_WAIT_MS, MAX_CONCURRENT_FETCHES, MAX_FETCH_RETRIES, PREFETCH_CHUNKS,
18 PROBE_TIMEOUT_SECS, WORKER_IDLE_MS,
19 },
20 common::types::AnyResult,
21 };
22 use bytes::Bytes;
23 use parking_lot::{Condvar, Mutex};
24 use std::{
25 collections::HashMap,
26 io::{Read, Seek, SeekFrom},
27 sync::Arc,
28 time::Duration,
29 };
30 use symphonia::core::io::MediaSource;
31 use tracing::{debug, trace, warn};
32 #[derive(Clone)]
33 enum ChunkState {
34 Empty(u32),
35 Downloading,
36 Ready(Bytes),
37 }
38 struct ReaderState {
39 chunks: HashMap<usize, ChunkState>,
40 current_pos: u64,
41 total_len: u64,
42 is_terminated: bool,
43 fatal_error: Option<String>,
44 }
45 pub struct SegmentedSource {
46 pos: u64,
47 len: u64,
48 content_type: Option<Arc<str>>,
49 shared: Arc<(Mutex<ReaderState>, Condvar)>,
50 }
51 impl SegmentedSource {
52 pub async fn new(client: reqwest::Client, url: &str) -> AnyResult<Self> {
53 let probe = client
54 .get(url)
55 .header("Range", "bytes=0-0")
56 .header("Connection", "close")
57 .timeout(Duration::from_secs(PROBE_TIMEOUT_SECS))
58 .send()
59 .await?;
60 let len = probe
61 .headers()
62 .get(reqwest::header::CONTENT_RANGE)
63 .and_then(|v| v.to_str().ok())
64 .and_then(|v| v.split('/').next_back())
65 .and_then(|v| v.parse::<u64>().ok())
66 .or_else(|| probe.content_length())
67 .ok_or_else(|| {
68 std::io::Error::new(
69 std::io::ErrorKind::InvalidData,
70 "SegmentedSource: could not determine content length",
71 )
72 })?;
73 let content_type: Option<Arc<str>> = probe
74 .headers()
75 .get(reqwest::header::CONTENT_TYPE)
76 .and_then(|v| v.to_str().ok())
77 .map(Arc::from);
78 debug!(
79 "SegmentedSource opened: len={}, type={:?}",
80 len, content_type
81 );
82 let mut chunks = HashMap::new();
83 chunks.insert(0, ChunkState::Empty(0));
84 let shared = Arc::new((
85 Mutex::new(ReaderState {
86 chunks,
87 current_pos: 0,
88 total_len: len,
89 is_terminated: false,
90 fatal_error: None,
91 }),
92 Condvar::new(),
93 ));
94 for worker_id in 0..MAX_CONCURRENT_FETCHES {
95 let shared_clone = shared.clone();
96 let client_clone = client.clone();
97 let url_str = url.to_string();
98 tokio::spawn(async move {
99 fetch_worker(worker_id, shared_clone, client_clone, url_str).await;
100 });
101 }
102 Ok(Self {
103 pos: 0,
104 len,
105 content_type,
106 shared,
107 })
108 }
109 }
110 impl AudioSource for SegmentedSource {
111 fn content_type(&self) -> Option<String> {
112 self.content_type.as_deref().map(str::to_string)
113 }
114 }
115 impl Read for SegmentedSource {
116 fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
117 if self.pos >= self.len {
118 return Ok(0);
119 }
120 let (lock, cvar) = &*self.shared;
121 let mut state = lock.lock();
122 state.current_pos = self.pos;
123 loop {
124 if let Some(ref err) = state.fatal_error {
125 return Err(std::io::Error::other(err.clone()));
126 }
127 let chunk_idx = (self.pos / CHUNK_SIZE as u64) as usize;
128 let offset_in_chunk = (self.pos % CHUNK_SIZE as u64) as usize;
129 match state.chunks.get(&chunk_idx) {
130 Some(ChunkState::Ready(bytes)) => {
131 let bytes = bytes.clone();
132 let available = bytes.len().saturating_sub(offset_in_chunk);
133 if available == 0 {
134 self.pos = ((chunk_idx + 1) * CHUNK_SIZE) as u64;
135 state.current_pos = self.pos;
136 continue;
137 }
138 let n = buf.len().min(available);
139 buf[..n].copy_from_slice(&bytes[offset_in_chunk..offset_in_chunk + n]);
140 self.pos += n as u64;
141 state.current_pos = self.pos;
142 if chunk_idx > 1 {
143 state.chunks.retain(|&idx, _| idx >= chunk_idx - 1);
144 }
145 return Ok(n);
146 }
147 Some(ChunkState::Downloading) | Some(ChunkState::Empty(_)) => {
148 cvar.notify_all();
149 trace!("SegmentedSource: waiting for chunk {}", chunk_idx);
150 cvar.wait_for(&mut state, Duration::from_millis(FETCH_WAIT_MS));
151 }
152 None => {
153 state.chunks.insert(chunk_idx, ChunkState::Empty(0));
154 cvar.notify_all();
155 }
156 }
157 }
158 }
159 }
160 impl Seek for SegmentedSource {
161 fn seek(&mut self, pos: SeekFrom) -> std::io::Result<u64> {
162 let new_pos = match pos {
163 SeekFrom::Start(p) => p,
164 SeekFrom::Current(delta) => self.pos.saturating_add_signed(delta),
165 SeekFrom::End(delta) => self.len.saturating_add_signed(delta),
166 };
167 self.pos = new_pos.min(self.len);
168 debug!("SegmentedSource: seek → {}", self.pos);
169 let (lock, cvar) = &*self.shared;
170 let mut state = lock.lock();
171 state.current_pos = self.pos;
172 cvar.notify_all();
173 Ok(self.pos)
174 }
175 }
176 impl MediaSource for SegmentedSource {
177 fn is_seekable(&self) -> bool {
178 true
179 }
180 fn byte_len(&self) -> Option<u64> {
181 Some(self.len)
182 }
183 }
184 impl Drop for SegmentedSource {
185 fn drop(&mut self) {
186 let (lock, cvar) = &*self.shared;
187 let mut state = lock.lock();
188 state.is_terminated = true;
189 cvar.notify_all();
190 }
191 }
192 async fn fetch_chunk(
193 client: &reqwest::Client,
194 url: &str,
195 offset: u64,
196 size: u64,
197 ) -> AnyResult<Bytes> {
198 let range = format!("bytes={}-{}", offset, offset + size - 1);
199 let res = client
200 .get(url)
201 .header("Range", range)
202 .header("Accept", "*/*")
203 .send()
204 .await?;
205 if !res.status().is_success() {
206 return Err(format!("fetch_chunk: HTTP {}", res.status()).into());
207 }
208 Ok(res.bytes().await?)
209 }
210 async fn fetch_worker(
211 worker_id: usize,
212 shared: Arc<(Mutex<ReaderState>, Condvar)>,
213 client: reqwest::Client,
214 url: String,
215 ) {
216 let (lock, cvar) = &*shared;
217 loop {
218 let target = {
219 let mut state = lock.lock();
220 if state.is_terminated {
221 break;
222 }
223 let current_chunk = (state.current_pos / CHUNK_SIZE as u64) as usize;
224 let total_len = state.total_len;
225 let claimed = try_claim_chunk(&mut state, current_chunk, total_len);
226 if claimed.is_none() {
227 let cursor_ready =
228 matches!(state.chunks.get(¤t_chunk), Some(ChunkState::Ready(_)));
229 let window = if cursor_ready { PREFETCH_CHUNKS } else { 2 };
230 let mut found = None;
231 for j in 1..window {
232 let idx = current_chunk + j;
233 if (idx * CHUNK_SIZE) as u64 >= total_len {
234 break;
235 }
236 if let Some(c) = try_claim_chunk(&mut state, idx, total_len) {
237 found = Some(c);
238 break;
239 }
240 }
241 found
242 } else {
243 claimed
244 }
245 .map(|(idx, retries)| {
246 debug!(
247 "Worker {}: claiming chunk {} (retry={})",
248 worker_id, idx, retries
249 );
250 state.chunks.insert(idx, ChunkState::Downloading);
251 (idx, retries, total_len)
252 })
253 };
254 let (idx, prior_retries, total_len) = match target {
255 Some(t) => t,
256 None => {
257 tokio::time::sleep(Duration::from_millis(WORKER_IDLE_MS)).await;
258 continue;
259 }
260 };
261 let offset = (idx * CHUNK_SIZE) as u64;
262 let size = CHUNK_SIZE.min((total_len - offset) as usize) as u64;
263 trace!(
264 "Worker {}: requesting chunk {} (offset={}, size={})",
265 worker_id, idx, offset, size
266 );
267 match fetch_chunk(&client, &url, offset, size).await {
268 Ok(bytes) => {
269 let actual = bytes.len() as u64;
270 if actual != size {
271 warn!(
272 "Worker {}: partial fetch for chunk {} (got {}/{} bytes)",
273 worker_id, idx, actual, size
274 );
275 requeue_or_fatal(
276 lock,
277 cvar,
278 idx,
279 prior_retries,
280 &format!("partial fetch: {}/{} bytes", actual, size),
281 );
282 tokio::time::sleep(Duration::from_millis(FETCH_WAIT_MS)).await;
283 continue;
284 }
285 let mut state = lock.lock();
286 state.chunks.insert(idx, ChunkState::Ready(bytes));
287 trace!(
288 "Worker {}: filled chunk {} ({} bytes)",
289 worker_id, idx, actual
290 );
291 cvar.notify_all();
292 }
293 Err(e) => {
294 warn!(
295 "Worker {}: fetch failed for chunk {}: {}",
296 worker_id, idx, e
297 );
298 requeue_or_fatal(lock, cvar, idx, prior_retries, &e.to_string());
299 tokio::time::sleep(Duration::from_millis(FETCH_WAIT_MS)).await;
300 }
301 }
302 }
303 }
304 #[inline]
305 fn try_claim_chunk(
306 state: &mut ReaderState,
307 idx: usize,
308 total_len: u64,
309 ) -> Option<(usize, u32)> {
310 if (idx * CHUNK_SIZE) as u64 >= total_len {
311 return None;
312 }
313 match state.chunks.get(&idx) {
314 Some(ChunkState::Empty(r)) => Some((idx, *r)),
315 None => Some((idx, 0)),
316 _ => None,
317 }
318 }
319 #[inline]
320 fn requeue_or_fatal(
321 lock: &Mutex<ReaderState>,
322 cvar: &Condvar,
323 idx: usize,
324 prior_retries: u32,
325 error: &str,
326 ) {
327 let mut state = lock.lock();
328 if prior_retries >= MAX_FETCH_RETRIES {
329 let msg = format!(
330 "Chunk {}: permanently failed after {} retries: {}",
331 idx, prior_retries, error
332 );
333 warn!("SegmentedSource: fatal error - {}", msg);
334 state.fatal_error = Some(msg);
335 } else {
336 state
337 .chunks
338 .insert(idx, ChunkState::Empty(prior_retries + 1));
339 }
340 cvar.notify_all();
341 }
342}
343pub mod client {
344 use crate::{common::types::AnyResult, config::HttpProxyConfig};
345 use reqwest::{Client, Proxy, header::HeaderMap};
346 use std::{net::IpAddr, time::Duration};
347 use tracing::warn;
348 pub fn create_client(
349 user_agent: String,
350 local_addr: Option<IpAddr>,
351 proxy: Option<HttpProxyConfig>,
352 headers: Option<HeaderMap>,
353 ) -> AnyResult<Client> {
354 let mut builder = Client::builder()
355 .user_agent(user_agent)
356 .connect_timeout(Duration::from_secs(5))
357 .read_timeout(Duration::from_secs(8))
358 .tcp_nodelay(true)
359 .tcp_keepalive(Duration::from_secs(25))
360 .pool_max_idle_per_host(64)
361 .pool_idle_timeout(Duration::from_secs(70));
362 if let Some(headers) = headers {
363 builder = builder.default_headers(headers);
364 }
365 if let Some(ip) = local_addr {
366 builder = builder.local_address(ip);
367 }
368 if let Some(p_cfg) = proxy
369 && let Some(p_url) = p_cfg.url
370 {
371 match Proxy::all(&p_url) {
372 Ok(mut p) => {
373 if let (Some(u), Some(pw)) = (p_cfg.username, p_cfg.password) {
374 p = p.basic_auth(&u, &pw);
375 }
376 builder = builder.proxy(p);
377 }
378 Err(e) => warn!("Failed to parse proxy URL '{}': {}", p_url, e),
379 }
380 }
381 Ok(builder.build()?)
382 }
383}
384pub mod http {
385 pub mod prefetcher {
386 use super::HttpSource;
387 use crate::audio::constants::{MAX_FETCH_RETRIES, MAX_HTTP_BUF_BYTES};
388 use bytes::Bytes;
389 use parking_lot::{Condvar, Mutex};
390 use std::{sync::Arc, time::Duration};
391 use tracing::{debug, warn};
392 #[derive(Debug)]
393 pub enum PrefetchCommand {
394 Continue,
395 Seek(u64),
396 Stop,
397 }
398 pub struct SharedState {
399 pub chunks: std::collections::VecDeque<Bytes>,
400 pub buffered: usize,
401 pub done: bool,
402 pub error: Option<String>,
403 pub command: PrefetchCommand,
404 }
405 impl Default for SharedState {
406 fn default() -> Self {
407 Self::new()
408 }
409 }
410 impl SharedState {
411 pub fn new() -> Self {
412 Self {
413 chunks: std::collections::VecDeque::with_capacity(64),
414 buffered: 0,
415 done: false,
416 error: None,
417 command: PrefetchCommand::Continue,
418 }
419 }
420 pub fn drain_into(&mut self, dst: &mut [u8]) -> usize {
421 let mut written = 0;
422 while written < dst.len() {
423 let Some(front) = self.chunks.front_mut() else {
424 break;
425 };
426 let want = dst.len() - written;
427 let have = front.len();
428 if have <= want {
429 dst[written..written + have].copy_from_slice(front);
430 written += have;
431 self.buffered -= have;
432 self.chunks.pop_front();
433 } else {
434 dst[written..].copy_from_slice(&front[..want]);
435 *front = front.slice(want..);
436 self.buffered -= want;
437 written += want;
438 break;
439 }
440 }
441 written
442 }
443 pub fn skip(&mut self, mut n: usize) -> usize {
444 let total = n;
445 while n > 0 {
446 let Some(front) = self.chunks.front_mut() else {
447 break;
448 };
449 let have = front.len();
450 if have <= n {
451 n -= have;
452 self.buffered -= have;
453 self.chunks.pop_front();
454 } else {
455 *front = front.slice(n..);
456 self.buffered -= n;
457 n = 0;
458 }
459 }
460 total - n
461 }
462 }
463 const SLEEP_SLICE_MS: u64 = 50;
464 async fn interruptible_sleep(
465 shared: &Arc<(Mutex<SharedState>, Condvar)>,
466 total_ms: u64,
467 ) -> bool {
468 let slices = (total_ms / SLEEP_SLICE_MS).max(1);
469 for _ in 0..slices {
470 tokio::time::sleep(Duration::from_millis(SLEEP_SLICE_MS)).await;
471 if matches!(shared.0.lock().command, PrefetchCommand::Stop) {
472 return true;
473 }
474 }
475 false
476 }
477 pub async fn prefetch_loop(
478 shared: Arc<(Mutex<SharedState>, Condvar)>,
479 client: reqwest::Client,
480 url: String,
481 mut current_pos: u64,
482 mut response: Option<reqwest::Response>,
483 total_len: Option<u64>,
484 ) {
485 let mut retry_count: u32 = 0;
486 'outer: loop {
487 let seek_target: Option<u64> = {
488 let (lock, cvar) = &*shared;
489 let mut state = lock.lock();
490 loop {
491 match std::mem::replace(&mut state.command, PrefetchCommand::Continue) {
492 PrefetchCommand::Stop => break 'outer,
493 PrefetchCommand::Seek(pos) => {
494 state.done = false;
495 state.chunks.clear();
496 state.buffered = 0;
497 cvar.notify_all();
498 break Some(pos);
499 }
500 PrefetchCommand::Continue => {
501 if state.buffered >= MAX_HTTP_BUF_BYTES || state.done {
502 cvar.wait_for(&mut state, Duration::from_millis(200));
503 continue;
504 }
505 break None;
506 }
507 }
508 }
509 };
510 if let Some(target) = seek_target {
511 let forward = target.saturating_sub(current_pos);
512 if forward > 0 && forward <= 256 * 1024 && response.is_some() {
513 debug!("prefetch: socket-skip {} bytes", forward);
514 let mut leftover: Option<Bytes> = None;
515 let res = response.take().unwrap();
516 let skip_result = async {
517 let mut res = res;
518 let mut remaining = forward;
519 while remaining > 0 {
520 match res.chunk().await {
521 Ok(Some(chunk)) => {
522 if chunk.len() as u64 <= remaining {
523 remaining -= chunk.len() as u64;
524 } else {
525 leftover = Some(chunk.slice(remaining as usize..));
526 remaining = 0;
527 }
528 }
529 _ => return Err(()),
530 }
531 }
532 Ok(res)
533 }
534 .await;
535 match skip_result {
536 Ok(r) => {
537 current_pos = target;
538 response = Some(r);
539 if let Some(lo) = leftover {
540 let (lock, cvar) = &*shared;
541 let mut state = lock.lock();
542 state.buffered += lo.len();
543 state.chunks.push_front(lo);
544 cvar.notify_all();
545 }
546 }
547 Err(_) => {
548 current_pos = target;
549 response = None;
550 }
551 }
552 } else {
553 current_pos = target;
554 response = None;
555 }
556 retry_count = 0;
557 }
558 if response.is_none() {
559 match HttpSource::fetch_stream(&client, &url, current_pos, None).await {
560 Ok(r) => {
561 response = Some(r);
562 retry_count = 0;
563 }
564 Err(e) => {
565 let msg = e.to_string();
566 if msg.contains("416") {
567 debug!("prefetch: 416 – reached end of stream");
568 let (lock, cvar) = &*shared;
569 let mut state = lock.lock();
570 state.done = true;
571 cvar.notify_all();
572 while state.done
573 && matches!(state.command, PrefetchCommand::Continue)
574 {
575 cvar.wait_for(&mut state, Duration::from_millis(200));
576 }
577 continue;
578 }
579 retry_count += 1;
580 if retry_count > MAX_FETCH_RETRIES {
581 warn!(
582 "prefetch: fetch failed fatally after {} retries: {}",
583 MAX_FETCH_RETRIES, e
584 );
585 let (lock, cvar) = &*shared;
586 let mut state = lock.lock();
587 state.error = Some(msg);
588 cvar.notify_all();
589 break 'outer;
590 }
591 let backoff_ms = 100u64 << (retry_count - 1).min(5);
592 warn!(
593 "prefetch: fetch failed (retry {}/{}): {} — backing off {}ms",
594 retry_count, MAX_FETCH_RETRIES, e, backoff_ms
595 );
596 if interruptible_sleep(&shared, backoff_ms).await {
597 break 'outer;
598 }
599 continue;
600 }
601 }
602 }
603 {
604 let (lock, cvar) = &*shared;
605 let mut state = lock.lock();
606 while state.buffered >= MAX_HTTP_BUF_BYTES
607 && matches!(state.command, PrefetchCommand::Continue)
608 && !state.done
609 {
610 cvar.wait_for(&mut state, Duration::from_millis(100));
611 }
612 if !matches!(state.command, PrefetchCommand::Continue) {
613 continue;
614 }
615 }
616 let res = response.as_mut().unwrap();
617 match res.chunk().await {
618 Ok(Some(chunk)) => {
619 let n = chunk.len();
620 let (lock, cvar) = &*shared;
621 let mut state = lock.lock();
622 if !matches!(state.command, PrefetchCommand::Continue) {
623 continue;
624 }
625 current_pos += n as u64;
626 state.buffered += n;
627 state.chunks.push_back(chunk);
628 cvar.notify_all();
629 }
630 Ok(None) => {
631 response = None;
632 retry_count = 0;
633 let is_eof = total_len.is_none_or(|l| current_pos >= l);
634 if is_eof {
635 let (lock, cvar) = &*shared;
636 let mut state = lock.lock();
637 state.done = true;
638 cvar.notify_all();
639 while state.done && matches!(state.command, PrefetchCommand::Continue) {
640 cvar.wait_for(&mut state, Duration::from_millis(200));
641 }
642 }
643 }
644 Err(e) => {
645 response = None;
646 retry_count += 1;
647 if retry_count > MAX_FETCH_RETRIES {
648 warn!(
649 "prefetch: read failed fatally after {} retries: {}",
650 MAX_FETCH_RETRIES, e
651 );
652 let (lock, cvar) = &*shared;
653 let mut state = lock.lock();
654 state.error = Some(e.to_string());
655 cvar.notify_all();
656 break 'outer;
657 }
658 let backoff_ms = 50u64 << (retry_count - 1).min(5);
659 warn!(
660 "prefetch: read error (retry {}/{}): {} — backing off {}ms",
661 retry_count, MAX_FETCH_RETRIES, e, backoff_ms
662 );
663 if interruptible_sleep(&shared, backoff_ms).await {
664 break 'outer;
665 }
666 }
667 }
668 }
669 }
670 }
671 use super::AudioSource;
672 use crate::common::types::AnyResult;
673 use parking_lot::{Condvar, Mutex};
674 use prefetcher::{PrefetchCommand, SharedState, prefetch_loop};
675 use std::{
676 io::{Read, Seek, SeekFrom},
677 sync::Arc,
678 thread,
679 };
680 use symphonia::core::io::MediaSource;
681 use tracing::debug;
682 pub struct HttpSource {
683 pos: u64,
684 len: Option<u64>,
685 content_type: Option<String>,
686 shared: Arc<(Mutex<SharedState>, Condvar)>,
687 }
688 impl HttpSource {
689 pub async fn new(client: reqwest::Client, url: &str) -> AnyResult<Self> {
690 let response = Self::fetch_stream(&client, url, 0, None).await?;
691 let len = response
692 .headers()
693 .get(reqwest::header::CONTENT_RANGE)
694 .and_then(|v| v.to_str().ok())
695 .and_then(|s| s.split('/').next_back())
696 .and_then(|s| s.parse::<u64>().ok())
697 .or_else(|| response.content_length());
698 let content_type = response
699 .headers()
700 .get(reqwest::header::CONTENT_TYPE)
701 .and_then(|v| v.to_str().ok())
702 .map(str::to_string);
703 debug!("HttpSource opened: {} (len={:?})", url, len);
704 let shared = Arc::new((Mutex::new(SharedState::new()), Condvar::new()));
705 let shared_clone = Arc::clone(&shared);
706 let url_clone = url.to_string();
707 thread::Builder::new()
708 .name("http-prefetch".into())
709 .spawn(move || {
710 let rt = tokio::runtime::Builder::new_current_thread()
711 .enable_all()
712 .build()
713 .unwrap();
714 rt.block_on(prefetch_loop(
715 shared_clone,
716 client,
717 url_clone,
718 0,
719 Some(response),
720 len,
721 ));
722 })?;
723 Ok(Self {
724 pos: 0,
725 len,
726 content_type,
727 shared,
728 })
729 }
730 pub(crate) async fn fetch_stream(
731 client: &reqwest::Client,
732 url: &str,
733 offset: u64,
734 limit: Option<u64>,
735 ) -> AnyResult<reqwest::Response> {
736 let range = match limit {
737 Some(l) => format!("bytes={}-{}", offset, offset + l - 1),
738 None => format!("bytes={}-", offset),
739 };
740 let res = client
741 .get(url)
742 .header("Accept", "*/*")
743 .header("Accept-Encoding", "identity")
744 .header("Connection", "keep-alive")
745 .header("Range", &range)
746 .send()
747 .await?;
748 if !res.status().is_success() {
749 return Err(format!("HTTP {} for {}", res.status(), url).into());
750 }
751 Ok(res)
752 }
753 }
754 impl AudioSource for HttpSource {
755 fn content_type(&self) -> Option<String> {
756 self.content_type.clone()
757 }
758 }
759 impl Read for HttpSource {
760 fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
761 let (lock, cvar) = &*self.shared;
762 let mut state = lock.lock();
763 loop {
764 if !state.chunks.is_empty() || state.done || state.error.is_some() {
765 break;
766 }
767 cvar.wait_for(&mut state, std::time::Duration::from_millis(100));
768 }
769 if let Some(err) = state.error.take() {
770 return Err(std::io::Error::other(err));
771 }
772 let n = state.drain_into(buf);
773 if state.buffered < crate::audio::constants::HTTP_PREFETCH_BUFFER_SIZE {
774 cvar.notify_one();
775 }
776 self.pos += n as u64;
777 Ok(n)
778 }
779 }
780 impl Seek for HttpSource {
781 fn seek(&mut self, pos: SeekFrom) -> std::io::Result<u64> {
782 let new_pos = match pos {
783 SeekFrom::Start(p) => p,
784 SeekFrom::Current(d) => self.pos.saturating_add_signed(d),
785 SeekFrom::End(d) => {
786 let l = self.len.ok_or_else(|| {
787 std::io::Error::new(
788 std::io::ErrorKind::Unsupported,
789 "stream length unknown",
790 )
791 })?;
792 l.saturating_add_signed(d)
793 }
794 };
795 if new_pos == self.pos {
796 return Ok(self.pos);
797 }
798 let (lock, cvar) = &*self.shared;
799 let mut state = lock.lock();
800 let forward = new_pos.saturating_sub(self.pos);
801 if forward > 0 && forward <= state.buffered as u64 {
802 debug!("HttpSource: in-memory seek +{} bytes", forward);
803 state.skip(forward as usize);
804 self.pos = new_pos;
805 return Ok(self.pos);
806 }
807 debug!("HttpSource: hard seek {} → {}", self.pos, new_pos);
808 state.chunks.clear();
809 state.buffered = 0;
810 state.done = false;
811 state.error = None;
812 state.command = PrefetchCommand::Seek(new_pos);
813 cvar.notify_all();
814 self.pos = new_pos;
815 Ok(self.pos)
816 }
817 }
818 impl MediaSource for HttpSource {
819 fn is_seekable(&self) -> bool {
820 self.len.is_some()
821 }
822 fn byte_len(&self) -> Option<u64> {
823 self.len
824 }
825 }
826 impl Drop for HttpSource {
827 fn drop(&mut self) {
828 let (lock, cvar) = &*self.shared;
829 let mut state = lock.lock();
830 state.command = PrefetchCommand::Stop;
831 cvar.notify_all();
832 }
833 }
834}
835pub use client::create_client;
836pub use http::HttpSource;
837pub use segmented::SegmentedSource;
838pub use traits::AudioSource;