diff --git a/src/api/model/active_provider_manager.rs b/src/api/model/active_provider_manager.rs index 711656de6..a775da15f 100644 --- a/src/api/model/active_provider_manager.rs +++ b/src/api/model/active_provider_manager.rs @@ -3,6 +3,7 @@ use std::cell::RefCell; use std::collections::HashMap; use std::sync::atomic::{AtomicU16, AtomicUsize, Ordering}; +#[derive(Debug)] pub enum ProviderAllocation<'a> { Exhausted, Available(&'a ProviderConfig), @@ -78,12 +79,12 @@ impl ProviderConfig { // !self.is_exhausted() // } - pub fn try_allocate(&self) -> ProviderAllocation { + pub fn try_allocate(&self, grace: bool) -> ProviderAllocation { let connections = self.current_connections.load(Ordering::Acquire); if self.max_connections == 0 { return ProviderAllocation::Available(self); } - if connections <= self.max_connections { + if (!grace && connections < self.max_connections) || (grace && connections <= self.max_connections) { self.current_connections.fetch_add(1, Ordering::AcqRel); return if connections < self.max_connections { ProviderAllocation::Available(self) } else { ProviderAllocation::GracePeriod(self) }; } @@ -142,7 +143,7 @@ impl SingleProviderLineup { } fn acquire(&self) -> ProviderAllocation { - self.provider.try_allocate() + self.provider.try_allocate(true) } fn release(&self, provider_name: &str) { @@ -221,17 +222,14 @@ impl MultiProviderLineup { /// Attempts to acquire the next available provider from a specific priority group. /// /// # Parameters - /// - `group_index`: The index of the provider group to search within. - /// - `force`: A boolean flag indicating whether to return a provider even if all are exhausted. + /// - `priority_group`: Thep rovider group to search within. /// /// # Returns - /// - `Some(&ProviderConfig)`: A reference to the next available provider in the specified group. - /// - `None`: If no providers are available in the group and `force` is `false`. + /// - `ProviderAllocation`: A reference to the next available provider in the specified group. /// /// # Behavior /// - Iterates through the providers in the given group in a round-robin manner. /// - Checks if a provider has available capacity before selecting it. - /// - If `force` is `true`, returns a provider even if no capacity is available. /// - Uses atomic operations to maintain fair provider selection. /// /// # Thread Safety @@ -241,16 +239,17 @@ impl MultiProviderLineup { /// # Example Usage /// ```rust /// let lineup = MultiProviderLineup::new(&config); - /// if let Some(provider) = lineup.acquire_next_provider_from_group(0, false) { - /// println!("Acquired provider: {}", provider.name); - /// } else { - /// println!("No available providers in group 0."); + /// match lineup.acquire_next_provider_from_group(priority_group) { + /// ProviderAllocation::Exhausted => println!("All providers exhausted"), + /// ProviderAllocation::Available(provider) => println!("Provider available {}", provider.name), + /// ProviderAllocation::GracePeriodprovider) => println!("Provider with grace period {}", provider.name), + /// } /// } /// ``` - fn acquire_next_provider_from_group(priority_group: &ProviderPriorityGroup) -> ProviderAllocation { + fn acquire_next_provider_from_group(priority_group: &ProviderPriorityGroup, grace: bool) -> ProviderAllocation { match priority_group { ProviderPriorityGroup::SingleProviderGroup(p) => { - let result = p.try_allocate(); + let result = p.try_allocate(grace); match result { ProviderAllocation::Exhausted => {} ProviderAllocation::Available(_) | ProviderAllocation::GracePeriod(_) => return result @@ -262,7 +261,7 @@ impl MultiProviderLineup { for _ in idx..provider_count { let p = pg.get(idx).unwrap(); idx = (idx + 1) % provider_count; - let result = p.try_allocate(); + let result = p.try_allocate(grace); match result { ProviderAllocation::Exhausted => {} ProviderAllocation::Available(_) | ProviderAllocation::GracePeriod(_) => { @@ -280,15 +279,13 @@ impl MultiProviderLineup { /// Attempts to acquire a provider from the lineup based on priority and availability. /// /// # Returns - /// - `Some(&ProviderConfig)`: A reference to the acquired provider if allocation was successful. - /// - `None`: If no providers are available and `force` is `false`. + /// - `ProviderAllocation`: A reference to the acquired provider if allocation was successful. /// /// # Behavior /// - The method iterates through provider priority groups in a round-robin fashion. /// - It attempts to allocate a provider from the highest priority group first. /// - If a provider has available capacity, it is returned. /// - If all providers in a group are exhausted, it moves to the next group. - /// - If `force` is `true`, it will return a provider even if all are exhausted. /// - Updates the internal index to ensure fair distribution of requests. /// /// # Thread Safety @@ -298,10 +295,10 @@ impl MultiProviderLineup { /// # Example Usage /// ```rust /// let lineup = MultiProviderLineup::new(&config); - /// if let Some(provider) = lineup.acquire(false) { - /// println!("Acquired provider: {}", provider.name); - /// } else { - /// println!("No available providers."); + /// match lineup.acquire() { + /// ProviderAllocation::Exhausted => println!("All providers exhausted"), + /// ProviderAllocation::Available(provider) => println!("Provider available {}", provider.name), + /// ProviderAllocation::GracePeriodprovider) => println!("Provider with grace period {}", provider.name), /// } /// ``` fn acquire(&self) -> ProviderAllocation { @@ -310,33 +307,41 @@ impl MultiProviderLineup { for index in main_idx..provider_count { let priority_group = &self.providers[index]; - let allocation = Self::acquire_next_provider_from_group(priority_group); + let allocation = { + let without_grace_allocation = Self::acquire_next_provider_from_group(priority_group, false); + if matches!(without_grace_allocation, ProviderAllocation::Exhausted) { + Self::acquire_next_provider_from_group(priority_group, true) + } else { + without_grace_allocation + } + }; match allocation { ProviderAllocation::Exhausted => {} ProviderAllocation::Available(_) | ProviderAllocation::GracePeriod(_) => { if priority_group.is_exhausted() { - self.index.store((index + 1) % provider_count, Ordering::Release); + self.index.store((index + 1) % provider_count, Ordering::Release); } return allocation; } } } - let provider = &self.providers[main_idx]; - self.index.store((main_idx + 1) % provider_count, Ordering::Release); - - match provider { - ProviderPriorityGroup::SingleProviderGroup(p) => ProviderAllocation::Available(p), - ProviderPriorityGroup::MultiProviderGroup(gindex, group) => { - let idx = gindex.load(Ordering::Acquire); - gindex.store((idx + 1) % group.len(), Ordering::Release); - match group.get(idx) { - None => ProviderAllocation::Exhausted, - Some(p) => ProviderAllocation::Available(p) - } - } - } + ProviderAllocation::Exhausted + // let provider = &self.providers[main_idx]; + // self.index.store((main_idx + 1) % provider_count, Ordering::Release); + // + // match provider { + // ProviderPriorityGroup::SingleProviderGroup(p) => ProviderAllocation::Available(p), + // ProviderPriorityGroup::MultiProviderGroup(gindex, group) => { + // let idx = gindex.load(Ordering::Acquire); + // gindex.store((idx + 1) % group.len(), Ordering::Release); + // match group.get(idx) { + // None => ProviderAllocation::Exhausted, + // Some(p) => ProviderAllocation::Available(p) + // } + // } + // } } @@ -468,7 +473,7 @@ impl ActiveProviderManager { } } - pub fn is_over_limit(&self, provider_name : &str) -> bool { + pub fn is_over_limit(&self, provider_name: &str) -> bool { if let Some((_, config)) = self.get_provider_config(provider_name) { config.is_over_limit() } else { @@ -483,6 +488,38 @@ mod tests { use crate::Arc; use std::thread; + + macro_rules! should_available { + ($lineup:expr, $provider_id:expr) => { + match $lineup.acquire() { + ProviderAllocation::Exhausted => assert!(false, "Should available and not exhausted"), + ProviderAllocation::Available(provider) => assert_eq!(provider.id, $provider_id), + ProviderAllocation::GracePeriod(provider) => assert!(false, "Should available and not grace period: {}", provider.id), + } + }; + } + macro_rules! should_grace_period { + ($lineup:expr, $provider_id:expr) => { + match $lineup.acquire() { + ProviderAllocation::Exhausted => assert!(false, "Should grace period and not exhausted"), + ProviderAllocation::Available(provider) => assert!(false, "Should grace period and not available: {}", provider.id), + ProviderAllocation::GracePeriod(provider) => assert_eq!(provider.id, $provider_id), + } + }; + } + + macro_rules! should_exhausted { + ($lineup:expr) => { + match $lineup.acquire() { + ProviderAllocation::Exhausted => {}, + ProviderAllocation::Available(provider) => assert!(false, "Should exhausted and not available: {}", provider.id), + ProviderAllocation::GracePeriod(provider) => assert!(false, "Should exhausted and not grace period: {}", provider.id), + } + }; + } + + + // Helper function to create a ConfigInput instance fn create_config_input(id: u16, name: &str, priority: i16, max_connections: u16) -> ConfigInput { ConfigInput { @@ -509,6 +546,7 @@ mod tests { fn create_config_input_alias(id: u16, url: &str, priority: i16, max_connections: u16) -> ConfigInputAlias { ConfigInputAlias { id, + name: format!("alias_{id}"), url: url.to_string(), username: Some("alias_user".to_string()), password: Some("alias_pass".to_string()), @@ -530,211 +568,159 @@ mod tests { let lineup = MultiProviderLineup::new(&input); // Test that the alias provider is available - match lineup.acquire() { - ProviderAllocation::Exhausted => assert!(false, "Should not Exhausted"), - ProviderAllocation::Available(provider) => { - assert_eq!(provider.id, 1); - } - ProviderAllocation::GracePeriod(_) => assert!(false, "Exceeded grace period unexpectedly"), - } - + should_available!(lineup, 1); // Try acquiring again - match lineup.acquire() { - ProviderAllocation::Exhausted => assert!(false, "Should not Exhausted"), - ProviderAllocation::Available(provider) => { - assert_eq!(provider.id, 2); - assert_eq!(provider.name, "provider1_1"); - } - ProviderAllocation::GracePeriod(_) => assert!(false, "Exceeded grace period unexpectedly"), - } + should_available!(lineup, 2); + should_available!(lineup, 2); + should_grace_period!(lineup, 1); + should_grace_period!(lineup, 2); + should_exhausted!(lineup); + } - - // Try acquiring with force (should succeed as force allows even exhausted providers) - match lineup.acquire() { - ProviderAllocation::Exhausted => {}, - ProviderAllocation::Available(_) => assert!(false, "Should not available"), - ProviderAllocation::GracePeriod(_) => assert!(false, "Exceeded grace period unexpectedly"), + // // Test acquiring from a MultiProviderLineup where the alias has a different priority + #[test] + fn test_provider_with_priority_alias() { + let mut input = create_config_input(1, "provider2_1", 1, 2); + let alias = create_config_input_alias(2, "http://alias.com", 0, 2); + // Adding alias with different priority + input.aliases = Some(vec![alias]); + let lineup = MultiProviderLineup::new(&input); + // The alias has a higher priority, so the alias should be acquired first + for _ in 0..2 { + should_available!(lineup, 2); } + should_available!(lineup, 1); + } + + // Test provider when there are multiple aliases, all with distinct priorities + #[test] + fn test_provider_with_multiple_aliases() { + let mut input = create_config_input(1, "provider3_1", 1, 1); + let alias1 = create_config_input_alias(2, "http://alias1.com", 1, 2); + let alias2 = create_config_input_alias(3, "http://alias2.com", 0, 1); + + // Adding multiple aliases + input.aliases = Some(vec![alias1, alias2]); + + let lineup = MultiProviderLineup::new(&input); + + // The alias with priority 0 should be acquired first (higher priority) + should_available!(lineup, 3); + // Acquire again, and provider should still be available (with remaining capacity) + should_available!(lineup, 1); + // // Check that the second alias with priority 2 is considered next + should_available!(lineup, 2); + should_available!(lineup, 2); + + should_grace_period!(lineup, 3); + should_grace_period!(lineup, 1); + should_grace_period!(lineup, 2); + + should_exhausted!(lineup); } - // TOD fix this tests - // Test acquiring from a MultiProviderLineup where the alias has a different priority - // #[test] - // fn test_provider_with_priority_alias() { - // let mut input = create_config_input(1, "provider2_1", 1, 2); - // let alias = create_config_input_alias(2, "http://alias.com", 0, 2); - // - // // Adding alias with different priority - // input.aliases = Some(vec![alias]); - // - // let lineup = MultiProviderLineup::new(&input); - // - // // The alias has a higher priority, so the alias should be acquired first - // let provider = lineup.acquire(false); - // assert!(provider.is_some()); - // assert_eq!(provider.unwrap().id, 2); - // - // let provider = lineup.acquire(false); - // assert!(provider.is_some()); - // assert_eq!(provider.unwrap().id, 2); - // - // let provider = lineup.acquire(false); - // assert!(provider.is_some()); - // assert_eq!(provider.unwrap().id, 1); - // } - // - // // Test provider when there are multiple aliases, all with distinct priorities - // #[test] - // fn test_provider_with_multiple_aliases() { - // let mut input = create_config_input(1, "provider3_1", 1, 1); - // let alias1 = create_config_input_alias(2, "http://alias1.com", 1, 2); - // let alias2 = create_config_input_alias(3, "http://alias2.com", 0, 1); - // - // // Adding multiple aliases - // input.aliases = Some(vec![alias1, alias2]); - // - // let lineup = MultiProviderLineup::new(&input); - // - // // The alias with priority 0 should be acquired first (higher priority) - // let provider = lineup.acquire(false); - // assert!(provider.is_some()); - // assert_eq!(provider.unwrap().id, 3); - // - // // Acquire again, and provider should still be available (with remaining capacity) - // let provider = lineup.acquire(false); - // assert!(provider.is_some()); - // assert_eq!(provider.unwrap().id, 1); - // - // // Check that the second alias with priority 2 is considered next - // let provider = lineup.acquire(false); - // assert!(provider.is_some()); - // assert_eq!(provider.unwrap().id, 2); - // } - // - // // // Test acquiring when all aliases are exhausted - // #[test] - // fn test_provider_with_exhausted_aliases() { - // let mut input = create_config_input(1, "provider4_1", 1, 1); - // let alias1 = create_config_input_alias(2, "http://alias.com", 2, 1); - // let alias2 = create_config_input_alias(3, "http://alias.com", -2, 1); - // - // // Adding alias - // input.aliases = Some(vec![alias1, alias2]); - // - // let lineup = MultiProviderLineup::new(&input); - // - // // Acquire connection from alias2 - // let provider = lineup.acquire(false); - // assert!(provider.is_some()); - // assert_eq!(provider.unwrap().id, 3); - // - // // Acquire connection from provider1 - // let provider = lineup.acquire(false); - // assert!(provider.is_some()); - // assert_eq!(provider.unwrap().id, 1); - // - // // Acquire connection from alias1 - // let provider = lineup.acquire(false); - // assert!(provider.is_some()); - // assert_eq!(provider.unwrap().id, 2); - // - // // Now, all are exhausted - // assert!(lineup.acquire(false).is_none()); - // } - // - // // Test acquiring a connection when there is available capacity - // #[test] - // fn test_acquire_when_capacity_available() { - // let cfg = create_config_input(1, "provider5_1", 1, 2); - // let lineup = SingleProviderLineup::new(&cfg); - // - // // First acquire attempt should succeed - // assert!(lineup.acquire(false).is_some()); - // - // // Second acquire attempt should succeed as well - // assert!(lineup.acquire(false).is_some()); - // - // // Third acquire attempt should fail as the provider is exhausted - // assert!(lineup.acquire(false).is_none()); - // } - // - // // Test acquiring a connection with the force flag - // #[test] - // fn test_acquire_with_force_flag() { - // let cfg = create_config_input(1, "provider6_1", 1, 1); - // let lineup = SingleProviderLineup::new(&cfg); - // - // // First acquire attempt should succeed - // assert!(lineup.acquire(false).is_some()); - // - // // Second acquire attempt should fail without force - // assert!(lineup.acquire(false).is_none()); - // - // // Third acquire attempt should succeed because force is true - // assert!(lineup.acquire(true).is_some()); - // } - // - // // Test releasing a connection - // #[test] - // fn test_release_connection() { - // let cfg = create_config_input(1, "provider7_1", 1, 2); - // let lineup = SingleProviderLineup::new(&cfg); - // - // // Acquire two connections - // assert!(lineup.acquire(false).is_some()); - // assert!(lineup.acquire(false).is_some()); - // - // // Release one connection - // lineup.release("provider7_1"); - // - // // After release, one connection should be available - // assert!(lineup.acquire(false).is_some()); - // - // // Release again, no connections should be available now - // assert!(lineup.acquire(false).is_none()); - // } - // - // // Test acquiring with MultiProviderLineup and round-robin allocation - // #[test] - // fn test_multi_provider_acquire() { - // let mut cfg1 = create_config_input(1, "provider8_1", 1, 2); - // let alias = create_config_input_alias(2, "http://alias1", 1, 1); - // - // // Adding alias to the provider - // cfg1.aliases = Some(vec![alias]); - // - // // Create MultiProviderLineup with the provider and alias - // let lineup = MultiProviderLineup::new(&cfg1); - // - // // Test acquiring the first provider - // let provider = lineup.acquire(false); - // assert!(provider.is_some()); - // assert_eq!(provider.unwrap().id, 1); - // - // // Test acquiring the second provider - // let provider = lineup.acquire(false); - // assert!(provider.is_some()); - // assert_eq!(provider.unwrap().id, 2); - // - // // Test acquiring the first provider - // let provider = lineup.acquire(false); - // assert!(provider.is_some()); - // assert_eq!(provider.unwrap().id, 1); - // - // // Test no more providers available - // assert!(lineup.acquire(false).is_none()); - // - // // Force flag should still allow allocation, round robin 2 because last was 1 - // let provider = lineup.acquire(true); - // assert!(provider.is_some()); - // assert_eq!(provider.unwrap().id, 2); - // - // // Force flag should still allow allocation, round robin 1 - // let provider = lineup.acquire(true); - // assert!(provider.is_some()); - // assert_eq!(provider.unwrap().id, 1); - // } + + // // Test acquiring when all aliases are exhausted + #[test] + fn test_provider_with_exhausted_aliases() { + let mut input = create_config_input(1, "provider4_1", 1, 1); + let alias1 = create_config_input_alias(2, "http://alias.com", 2, 1); + let alias2 = create_config_input_alias(3, "http://alias.com", -2, 1); + + // Adding alias + input.aliases = Some(vec![alias1, alias2]); + + let lineup = MultiProviderLineup::new(&input); + + // Acquire connection from alias2 + should_available!(lineup, 3); + // Acquire connection from provider1 + should_available!(lineup, 1); + // Acquire connection from alias1 + should_available!(lineup, 2); + + // Acquire connection from alias2 + should_grace_period!(lineup, 3); + // Acquire connection from provider1 + should_grace_period!(lineup, 1); + // Acquire connection from alias1 + should_grace_period!(lineup, 2); + + // Now, all are exhausted + should_exhausted!(lineup); + } + + // Test acquiring a connection when there is available capacity + #[test] + fn test_acquire_when_capacity_available() { + let cfg = create_config_input(1, "provider5_1", 1, 2); + let lineup = SingleProviderLineup::new(&cfg); + + // First acquire attempt should succeed + should_available!(lineup, 1); + // Second acquire attempt should succeed as well + should_available!(lineup, 1); + // Third with grace time + should_grace_period!(lineup, 1); + // Fourth acquire attempt should fail as the provider is exhausted + should_exhausted!(lineup); + } + + + // Test releasing a connection + #[test] + fn test_release_connection() { + let cfg = create_config_input(1, "provider7_1", 1, 2); + let lineup = SingleProviderLineup::new(&cfg); + + // Acquire two connections + should_available!(lineup, 1); + should_available!(lineup, 1); + should_grace_period!(lineup, 1); + lineup.release("provider7_1"); + should_grace_period!(lineup, 1); + lineup.release("provider7_1"); + lineup.release("provider7_1"); + should_available!(lineup, 1); + should_grace_period!(lineup, 1); + should_exhausted!(lineup); + } + + // Test acquiring with MultiProviderLineup and round-robin allocation + #[test] + fn test_multi_provider_acquire() { + let mut cfg1 = create_config_input(1, "provider8_1", 1, 2); + let alias = create_config_input_alias(2, "http://alias1", 1, 1); + + // Adding alias to the provider + cfg1.aliases = Some(vec![alias]); + + // Create MultiProviderLineup with the provider and alias + let lineup = MultiProviderLineup::new(&cfg1); + + // Test acquiring the first provider + should_available!(lineup, 1); + + // Test acquiring the second provider + should_available!(lineup, 2); + + // Test acquiring the first provider + should_available!(lineup, 1); + + should_grace_period!(lineup, 1); + should_grace_period!(lineup, 2); + + lineup.release("provider8_1"); + lineup.release("alias_2"); + lineup.release("provider8_1"); + + should_available!(lineup, 1); + should_grace_period!(lineup, 1); + should_grace_period!(lineup, 2); + + should_exhausted!(lineup); + } // Test concurrent access to `acquire` using multiple threads #[test] @@ -744,11 +730,23 @@ mod tests { let mut handles = vec![]; + let available_count= Arc::new(AtomicU16::new(2)); + let grace_period_count= Arc::new(AtomicU16::new(1)); + let exhausted_count= Arc::new(AtomicU16::new(2)); + for _ in 0..5 { let lineup_clone = Arc::clone(&lineup); + let available = Arc::clone(&available_count); + let grace_period = Arc::clone(&grace_period_count); + let exhausted = Arc::clone(&exhausted_count); let handle = thread::spawn(move || { // Each thread tries to acquire a connection - let _result = lineup_clone.acquire(); + match lineup_clone.acquire() { + ProviderAllocation::Exhausted => exhausted.fetch_sub(1, Ordering::Release), + ProviderAllocation::Available(_) => available.fetch_sub(1, Ordering::Release), + ProviderAllocation::GracePeriod(_) => grace_period.fetch_sub(1, Ordering::Release), + } + }); handles.push(handle); } @@ -758,8 +756,9 @@ mod tests { handle.join().unwrap(); } - // Verify that only the capacity of the provider was utilized (2 connections) - assert_eq!(lineup.provider.current_connections.load(Ordering::Acquire), 2); + assert_eq!(exhausted_count.load(Ordering::Acquire), 0); + assert_eq!(available_count.load(Ordering::Acquire), 0); + assert_eq!(grace_period_count.load(Ordering::Acquire), 0); } } diff --git a/src/processing/parser/xmltv.rs b/src/processing/parser/xmltv.rs index 35d9433b8..74bca9a00 100644 --- a/src/processing/parser/xmltv.rs +++ b/src/processing/parser/xmltv.rs @@ -174,20 +174,21 @@ mod tests { #[test] fn parse_test() -> io::Result<()> { - let file_path = "/tmp/epg.xml.gz"; + let file_path = PathBuf::from("/tmp/epg.xml.gz"); - let tv_guide = TVGuide { file: PathBuf::from(file_path) }; + if file_path.exists() { + let tv_guide = TVGuide { file: file_path}; - let channel_ids = vec!["channel.1", "channel.2", "channel.3"]; - let channel_ids : HashSet = channel_ids.into_iter().map(|s| s.to_string()).collect(); + let channel_ids = vec!["channel.1", "channel.2", "channel.3"]; + let channel_ids : HashSet = channel_ids.into_iter().map(|s| s.to_string()).collect(); - match tv_guide.filter(&channel_ids) { - None => assert!(false, "No epg filtered"), - Some(epg) => { - assert_eq!(epg.children.len(), channel_ids.len() * 2, "Epg size does not match") + match tv_guide.filter(&channel_ids) { + None => assert!(false, "No epg filtered"), + Some(epg) => { + assert_eq!(epg.children.len(), channel_ids.len() * 2, "Epg size does not match") + } } } - Ok(()) } } \ No newline at end of file diff --git a/src/repository/indexed_document.rs b/src/repository/indexed_document.rs index 4534c89b6..a3cf7afa7 100644 --- a/src/repository/indexed_document.rs +++ b/src/repository/indexed_document.rs @@ -512,7 +512,7 @@ mod tests { use std::path::PathBuf; use serde::{Deserialize, Serialize}; - use crate::model::playlist::XtreamPlaylistItem; + // use crate::model::playlist::XtreamPlaylistItem; use crate::repository::indexed_document::{IndexedDocumentGarbageCollector, IndexedDocumentIterator, IndexedDocumentWriter}; // Example usage with a simple struct @@ -585,14 +585,14 @@ mod tests { Ok(()) } - #[test] - fn test_read_xt() -> io::Result<()> { - let main_path = PathBuf::from("../m3u-test/settings/m3u-silver/data/xt_m3u/xtream/live.db"); - let index_path = PathBuf::from("../m3u-test/settings/m3u-silver/data/xt_m3u/xtream/live.idx"); - let reader = IndexedDocumentIterator::::new(&main_path, &index_path)?; - for doc in reader { - println!("{doc:?}"); - } - Ok(()) - } + // #[test] + // fn test_read_xt() -> io::Result<()> { + // let main_path = PathBuf::from("../m3u-test/settings/m3u-silver/data/xt_m3u/xtream/live.db"); + // let index_path = PathBuf::from("../m3u-test/settings/m3u-silver/data/xt_m3u/xtream/live.idx"); + // let reader = IndexedDocumentIterator::::new(&main_path, &index_path)?; + // for doc in reader { + // println!("{doc:?}"); + // } + // Ok(()) + // } } diff --git a/src/repository/target_id_mapping.rs b/src/repository/target_id_mapping.rs index 94590c9e6..7f2511ec7 100644 --- a/src/repository/target_id_mapping.rs +++ b/src/repository/target_id_mapping.rs @@ -126,18 +126,18 @@ impl Drop for TargetIdMapping { #[cfg(test)] mod tests { - use std::path::PathBuf; - use crate::repository::bplustree::BPlusTree; - use crate::repository::target_id_mapping::{VirtualIdRecord}; + // use std::path::PathBuf; + // use crate::repository::bplustree::BPlusTree; + // use crate::repository::target_id_mapping::{VirtualIdRecord}; - #[test] - fn test_id_mapping() { - let path = PathBuf::from("../m3u-test/settings/m3u-silver/data/xt_m3u/id_mapping.db"); - let mapping = BPlusTree::::load(&path); - mapping.unwrap().traverse(|keys, values| { - for (key, value) in keys.iter().zip(values.iter()) { - println!("{key:?} {value:?}\n"); - } - }); - } + // #[test] + // fn test_id_mapping() { + // let path = PathBuf::from("../m3u-test/settings/m3u-silver/data/xt_m3u/id_mapping.db"); + // let mapping = BPlusTree::::load(&path); + // mapping.unwrap().traverse(|keys, values| { + // for (key, value) in keys.iter().zip(values.iter()) { + // println!("{key:?} {value:?}\n"); + // } + // }); + // } } diff --git a/src/repository/user_repository.rs b/src/repository/user_repository.rs index bdb55ccc9..564ccdd2a 100644 --- a/src/repository/user_repository.rs +++ b/src/repository/user_repository.rs @@ -99,12 +99,13 @@ pub fn merge_api_user(cfg: &Config, target_users: &[TargetUser]) -> Result {} - Err(err) => { error!("Could not backup file {}:{}", &backup_path.to_str().unwrap_or("?"), err) } + if let Some(backup_dir) = cfg.backup_dir.as_ref() { + let backup_path = PathBuf::from(backup_dir).join(format!("{API_USER_DB_FILE}_{}", Local::now().format("%Y%m%d_%H%M%S"))); + let _lock = cfg.file_locks.read_lock(path); + match std::fs::copy(path, &backup_path) { + Ok(_) => {} + Err(err) => { error!("Could not backup file {}:{}", &backup_path.to_str().unwrap_or("?"), err) } + } } } diff --git a/src/utils/file/config_reader.rs b/src/utils/file/config_reader.rs index 1b2310cbb..5fd9d43f1 100644 --- a/src/utils/file/config_reader.rs +++ b/src/utils/file/config_reader.rs @@ -1,17 +1,19 @@ -use std::env; -use std::fs::File; -use std::path::PathBuf; -use std::sync::LazyLock; +use crate::m3u_filter_error::{create_m3u_filter_error, create_m3u_filter_error_result, handle_m3u_filter_error_result, info_err, str_to_io_error, to_io_error, M3uFilterError, M3uFilterErrorKind}; +use crate::model::api_proxy::ApiProxyConfig; +use crate::model::config::{Config, ConfigDto, ConfigInputAlias, InputType}; +use crate::model::mapping::Mappings; +use crate::utils::file::{file_utils, multi_file_reader}; +use crate::utils::sys_utils::exit; use chrono::Local; use log::{debug, error, info, warn}; use regex::Regex; use serde::Serialize; -use crate::utils::sys_utils::exit; -use crate::m3u_filter_error::{to_io_error, M3uFilterError, M3uFilterErrorKind, create_m3u_filter_error, create_m3u_filter_error_result, info_err, handle_m3u_filter_error_result}; -use crate::model::api_proxy::ApiProxyConfig; -use crate::model::config::{Config, ConfigDto}; -use crate::model::mapping::Mappings; -use crate::utils::file::{file_utils, multi_file_reader}; +use std::env; +use std::fs::File; +use std::io::{self, BufRead}; +use std::path::PathBuf; +use std::sync::LazyLock; +use url::Url; pub fn read_mappings(args_mapping: Option, cfg: &mut Config) -> Result, M3uFilterError> { let mappings_file: String = args_mapping.unwrap_or_else(|| file_utils::get_default_mappings_path(cfg.t_config_path.as_str())); @@ -80,7 +82,7 @@ pub fn read_mapping(mapping_file: &str) -> Result, M3uFilterErr Ok(mut result) => { handle_m3u_filter_error_result!(M3uFilterErrorKind::Info, result.prepare()); return Ok(Some(result)); - }, + } Err(err) => { return Err(info_err!(err.to_string())); } @@ -92,29 +94,30 @@ pub fn read_mapping(mapping_file: &str) -> Result, M3uFilterErr pub fn read_api_proxy(config: &Config, api_proxy_file: &str, resolve_var: bool) -> Option { file_utils::open_file(&std::path::PathBuf::from(api_proxy_file)).map_or(None, |file| { - let mapping: Result = serde_yaml::from_reader(file); - match mapping { - Ok(mut result) => { - match result.prepare(config, resolve_var) { - Err(err) => { - exit!("cant read api-proxy-config file: {}", err); - } - _ => { - Some(result) - } + let mapping: Result = serde_yaml::from_reader(file); + match mapping { + Ok(mut result) => { + match result.prepare(config, resolve_var) { + Err(err) => { + exit!("cant read api-proxy-config file: {}", err); + } + _ => { + Some(result) } } - Err(err) => { - error!("cant read api-proxy-config file: {}", err); - None - } } - }) + Err(err) => { + error!("cant read api-proxy-config file: {}", err); + None + } + } + }) } fn write_config_file(file_path: &str, backup_dir: &str, config: &T, default_name: &str) -> Result<(), M3uFilterError> - where - T: ?Sized + Serialize { +where + T: ?Sized + Serialize, +{ let path = PathBuf::from(file_path); let filename = path.file_name().map_or(default_name.to_string(), |f| f.to_string_lossy().to_string()); let backup_path = PathBuf::from(backup_dir).join(format!("{filename}_{}", Local::now().format("%Y%m%d_%H%M%S"))); @@ -129,8 +132,6 @@ fn write_config_file(file_path: &str, backup_dir: &str, config: &T, default_n File::create(&path) .and_then(|f| serde_yaml::to_writer(f, &config).map_err(to_io_error)) .map_err(|err| create_m3u_filter_error!(M3uFilterErrorKind::Info, "Could not write file {}: {}", &path.to_str().unwrap_or("?"), err)) - - } pub fn save_api_proxy(file_path: &str, backup_dir: &str, config: &ApiProxyConfig) -> Result<(), M3uFilterError> { @@ -153,13 +154,110 @@ pub fn resolve_env_var(value: &str) -> String { }).to_string() } +// +// pub fn read_yaml_inputs(input_type: InputType, file_path: &str) -> Result { +// +// } +// +// +// pub fn read_json_inputs(input_type: InputType, file_path: &str) -> Result { +// +// } + +const CSV_SEPARATOR: char = ','; +const HEADER_PREFIX: char = '#'; + +const FIELD_MAX_CON: &str = "max_connections"; +const FIELD_PRIO: &str = "priority"; +const FIELD_URL: &str = "url"; + +fn csv_assign_config_input_column(config_input: &mut ConfigInputAlias, input_type: &InputType, header: &str, value: &str) -> Result<(), io::Error> { + match header { + FIELD_URL => { + let url = Url::parse(value.trim()).map_err(to_io_error)?; + + match input_type { + InputType::Xtream => { + let username = url.query_pairs().find(|(key, _)| key == "username").map(|(_, value)| value); + if let Some(uname) = username { + config_input.username = Some(uname.to_string()); + } + let password = url.query_pairs().find(|(key, _)| key == "password").map(|(_, value)| value); + if let Some(pwd) = password { + config_input.password = Some(pwd.to_string()); + } + config_input.url = url.origin().ascii_serialization() + } + + InputType::M3u => { + config_input.url = url.to_string(); + } + }; + }, + FIELD_MAX_CON => { + let max_connections = value.parse::().unwrap_or(1); + config_input.max_connections = max_connections; + }, + FIELD_PRIO => { + let priority = value.parse::().unwrap_or(0); + config_input.priority = priority; + }, + _ => {} + } + Ok(()) +} + +pub fn csv_read_inputs(input_type: InputType, file_path: &str) -> Result, io::Error> { + let inputs_file = std::path::PathBuf::from(file_path); + if let Ok(file) = file_utils::open_file(&inputs_file) { + let mut result = vec![]; + let default_columns = vec![FIELD_URL, FIELD_MAX_CON, FIELD_PRIO]; + let reader = io::BufReader::new(file); + for line in reader.lines().into_iter() { + let line = line?; + if line.is_empty() || line.starts_with(HEADER_PREFIX) { + continue; + } + + let mut config_input = ConfigInputAlias { + id: 0, + name: "".to_string(), + url: "".to_string(), + username: None, + password: None, + priority: 0, + max_connections: 1, + }; + + let columns: Vec<&str> = line.split(CSV_SEPARATOR).collect(); + for (&header, &value) in default_columns.iter().zip(columns.iter()).into_iter() { + if let Err(err) = csv_assign_config_input_column(&mut config_input, &input_type, header, value) { + error!("Could not parse input line: {} err: {err}", line); + continue; + } + } + result.push(config_input); + } + return Ok(result); + } + Err(str_to_io_error(&format!("Could not open {file_path}"))) +} + #[cfg(test)] mod tests { - use crate::utils::file::config_reader::resolve_env_var; + use crate::model::config::InputType; + use crate::utils::file::config_reader::{csv_read_inputs, resolve_env_var}; + + #[test] + fn test_read_inputs() { + let result = csv_read_inputs(InputType::Xtream, &resolve_env_var("${env:HOME}/projects/m3u-test/settings/inputs.txt")); + assert_eq!(result.is_ok(), true); + println!("result: {:?}", result.unwrap()); + } #[test] fn test_resolve() { - let resolved = resolve_env_var("${env:HOME}"); + let resolved = resolve_env_var("${env:HOME}"); assert_eq!(resolved, std::env::var("HOME").unwrap()); } } \ No newline at end of file