diff --git a/src/api/model/buffered_stream.rs b/src/api/model/buffered_stream.rs index 8f5f1c224..9e5235839 100644 --- a/src/api/model/buffered_stream.rs +++ b/src/api/model/buffered_stream.rs @@ -72,7 +72,7 @@ impl Drop for BufferedReceiverStream { } } -type ProviderStreamResponse = (Option> + Unpin + 'static>>>, Option<(Vec<(String, String)>, StatusCode)>); +type ProviderStreamResponse = (Option> + Unpin + 'static>>>, Option<(Vec<(String, String)>, StatusCode)>); pub async fn get_provider_stream(http_client: &Arc, stream_url: &Url, @@ -181,7 +181,7 @@ pub async fn get_buffered_stream(http_client: &Arc, // We need some header information from the provider, we extract the neccessary headers and forard them to the client debug_if_enabled!("Provider response headers: {:?}", response.headers_mut()); first_run = false; - let response_headers: Vec<(String, String)> = get_response_headers(&mut response); + let response_headers: Vec<(String, String)> = get_response_headers(&mut response); // debug!("First headers {headers:?} {} {}", mask_sensitive_info(url.as_str())); let mut status = response.status(); if partial_content && status.is_success() { @@ -195,8 +195,8 @@ pub async fn get_buffered_stream(http_client: &Arc, match byte_stream.next().await { Some(Ok(chunk)) => { if chunk.is_empty() { - debug!("Provider finished stream {masked_url}"); - break; + actix_web::rt::time::sleep(Duration::from_millis(20)).await; + continue; } // this is for backpressure, we fill the buffer and wait for the receiver if let Ok(permit) = tx.reserve().await { diff --git a/src/utils/lru_cache.rs b/src/utils/lru_cache.rs index b111e9911..a8615f41f 100644 --- a/src/utils/lru_cache.rs +++ b/src/utils/lru_cache.rs @@ -2,7 +2,7 @@ use crate::repository::storage::hash_string_as_hex; use crate::utils::file_utils::traverse_dir; use crate::utils::size_utils::human_readable_byte_size; use async_std::sync::RwLock; -use log::{debug, error, info}; +use log::{debug, error, info, trace}; use std::collections::{HashMap, VecDeque}; use std::fs; use std::path::{Path, PathBuf}; @@ -60,7 +60,7 @@ impl LRUResourceCache { { // insert_to_cache let mut path = self.cache_dir.clone(); path.push(&key); - debug!("Added file to cache: {}", &path.to_string_lossy()); + trace!("Added file to cache: {}", &path.to_string_lossy()); self.cache.insert(key.clone(), (path.clone(), file_size)); self.usage_order.push_back(key); self.current_size += file_size; diff --git a/src/utils/multi_file_reader.rs b/src/utils/multi_file_reader.rs index d4f013769..a35538d3f 100644 --- a/src/utils/multi_file_reader.rs +++ b/src/utils/multi_file_reader.rs @@ -34,6 +34,11 @@ impl Read for MultiFileReader { return Ok(0); } self.current_reader = Some(BufReader::new(self.files.pop().unwrap())); + // we put a newline if the config does not have one + if !buf.is_empty() && buf[0] != b'\n' { + buf[0] = b'\n'; + return Ok(1); + } } let reader = self.current_reader.as_mut().unwrap(); match reader.read(buf) {