better connection handling

backend/src/api/api_utils.rs

@@ -509,7 +509,7 @@ async fn create_stream_response_details(
             }
 
             // if we have no stream, we should release the provider
-            if stream.is_none() {
+            let provider_handle = if stream.is_none() {
                 let provider_handle = streaming_strategy.provider_handle.take();
                 app_state.connection_manager.release_provider_handle(provider_handle).await;
                 error!("Cant open stream {}", sanitize_sensitive_info(&request_url));
@@ -524,7 +524,7 @@ async fn create_stream_response_details(
                 provider_name: guard_provider_name.clone(),
                 grace_period_millis,
                 reconnect_flag,
-                provider_handle: streaming_strategy.provider_handle,
+                provider_handle,
             }
         }
     }

backend/src/api/endpoints/m3u_api.rs

@@ -169,7 +169,6 @@ async fn m3u_api_stream(
 
     let connection_permission = user.connection_permission(app_state).await;
     if connection_permission == UserConnectionPermission::Exhausted {
-        app_state.connection_manager.update_stream_detail(&fingerprint.addr, CustomVideoStreamType::UserConnectionsExhausted).await;
         return create_custom_video_stream_response(
             app_state, &fingerprint.addr,
             CustomVideoStreamType::UserConnectionsExhausted,

backend/src/api/model/event_manager.rs

@@ -1,4 +1,4 @@
-use log::{error, trace};
+use log::{trace};
 use shared::model::{ActiveUserConnectionChange, ConfigType, PlaylistUpdateState};
 
 #[allow(clippy::large_enum_variant)]
@@ -6,7 +6,7 @@ use shared::model::{ActiveUserConnectionChange, ConfigType, PlaylistUpdateState}
 pub enum EventMessage {
     ServerError(String),
     ActiveUser(ActiveUserConnectionChange), // user_count, connection count
-    ActiveProvider(Option<String>, usize), // provider name, connections
+    ActiveProvider(String, usize), // provider name, connections
     ConfigChange(ConfigType),
     PlaylistUpdate(PlaylistUpdateState),
     PlaylistUpdateProgress(String, String),
@@ -42,8 +42,8 @@ impl EventManager {
     }
 
     pub fn send_provider_event(&self, provider: &str, connection_count: usize) {
-        if !self.send_event(EventMessage::ActiveProvider(Some(String::from(provider)), connection_count)) {
-            error!("Failed to send connection change: {provider}: {connection_count}");
+        if !self.send_event(EventMessage::ActiveProvider(String::from(provider), connection_count)) {
+            trace!("Failed to send connection change: {provider}: {connection_count}");
         }
     }
 }

backend/src/utils/network/request.rs

@@ -107,7 +107,7 @@ pub async fn get_input_text_content(client: Arc<reqwest::Client>, input: &InputS
         match download_text_content(client, None, input, None, persist_filepath).await {
             Ok((content, _response_url)) => Ok(content),
             Err(e) => {
-                error!("cant download input url: {}  => {}", sanitize_sensitive_info(&input.name), sanitize_sensitive_info(e.to_string().as_str()));
+                error!("cant download input name: {}  => {}", &input.name, sanitize_sensitive_info(e.to_string().as_str()));
                 create_tuliprox_error_result!(TuliproxErrorKind::Notify, "Failed to download")
             }
         }

frontend/src/app/components/source_editor/layout.rs

@@ -90,8 +90,8 @@ pub fn cluster_layout(blocks: &mut [Block], connections: &[Connection]) {
 
         for &input_id in &cluster.inputs {
             let connected_targets: Vec<BlockId> = connections.iter()
-                .filter(|c| c.to == input_id && cluster.targets.contains(&c.from))
-                .map(|c| c.from)
+                .filter(|c| c.from == input_id && cluster.targets.contains(&c.to))
+                .map(|c| c.to)
                 .collect();
 
             input_to_targets.insert(input_id, connected_targets.clone());

frontend/src/hooks/use_server_status.rs

@@ -61,7 +61,7 @@ pub fn use_server_status(
                         *status_holder_signal.borrow_mut() = Some(Rc::clone(&new_status));
                         status_signal.set(Some(new_status));
                     }
-                    EventMessage::ActiveProvider(Some(provider), connections) => {
+                    EventMessage::ActiveProvider(provider, connections) => {
                         let mut server_status = {
                             if let Some(old_status) = status_holder_signal.borrow().as_ref() {
                                 (**old_status).clone()

frontend/src/model/event_message.rs

@@ -9,7 +9,7 @@ pub enum EventMessage {
     ServerError(String),
     ServerStatus(Rc<StatusCheck>),
     ActiveUser(ActiveUserConnectionChange),
-    ActiveProvider(Option<String>, usize), // single provider
+    ActiveProvider(String, usize), // single provider
     ActiveProviderCount(usize), // all provider
     ConfigChange(ConfigType),
     Busy(BusyStatus),

shared/src/model/web_socket.rs

@@ -83,7 +83,7 @@ pub enum ProtocolMessage {
     // Responses
     StatusResponse(StatusCheck),
     ActiveUserResponse(ActiveUserConnectionChange),
-    ActiveProviderResponse(Option<String>, usize), // single provider
+    ActiveProviderResponse(String, usize), // single provider
     ActiveProviderCountRequest(String),
     ActiveProviderCountResponse(usize),
     ConfigChangeResponse(ConfigType),