diff --git a/frontend/src/app/components/authentication.rs b/frontend/src/app/components/authentication.rs index 4324bf1ac..63bd4ac4b 100644 --- a/frontend/src/app/components/authentication.rs +++ b/frontend/src/app/components/authentication.rs @@ -24,7 +24,7 @@ pub fn Authentication(props: &AuthenticationProps) -> Html { &mut |success| { authenticated_state.set(success); if success { - services_ctx.websocket.connect_ws(); + services_ctx.websocket.connect_ws_with_backoff(); } future::ready(()) } diff --git a/frontend/src/app/components/login.rs b/frontend/src/app/components/login.rs index 9d6e48a95..018fc98be 100644 --- a/frontend/src/app/components/login.rs +++ b/frontend/src/app/components/login.rs @@ -39,7 +39,7 @@ pub fn Login() -> Html { let password = password_input.value(); let result = services_ctx.auth.authenticate(username, password).await; match &result { - Ok(token) => { + Ok(_token) => { authorized_state.set(true) } Err(_) => {authorized_state.set(false);} diff --git a/frontend/src/services/websocket_service.rs b/frontend/src/services/websocket_service.rs index 898e4b086..5345e18df 100644 --- a/frontend/src/services/websocket_service.rs +++ b/frontend/src/services/websocket_service.rs @@ -1,25 +1,48 @@ -use wasm_bindgen::prelude::*; use wasm_bindgen::JsCast; use web_sys::{WebSocket, MessageEvent, Event, ErrorEvent, CloseEvent}; use std::cell::RefCell; use std::rc::Rc; -use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicU16, Ordering}; use web_sys::js_sys::{Uint8Array, ArrayBuffer}; -use log::{debug, error, trace}; +use log::{error, trace, warn}; use shared::model::{ProtocolMessage, PROTOCOL_VERSION}; use shared::utils::{concat_path_leading_slash}; use crate::model::EventMessage; use crate::services::{get_base_href, get_token, EventService, StatusService}; use crate::utils::set_timeout; +use wasm_bindgen::closure::Closure; -const WS_RECONNECT_MS:i32 = 2000; +const WS_RECONNECT_BASE_MS: u32 = 300; +const WS_RECONNECT_MAX_MS: u32 = 2000; +const WS_RECONNECT_MAX_ATTEMPTS: u32 = 20; + +fn reconnect_delay(attempt: u32) -> u32 { + if attempt < 6 { + let d = WS_RECONNECT_BASE_MS * (attempt+1); + d.min(WS_RECONNECT_MAX_MS) + } else { + WS_RECONNECT_MAX_MS + } +} + +type JsOnMessageCallback = Option>; +type JsOnCloseCallback = Option>; +type JsOnErrorCallback = Option>; +type JsOnOpenCallback = Option>; pub struct WebSocketService { connected: Rc, + attempt_counter: Rc, ws: Rc>>, status_service: Rc, event_service: Rc, ws_path: String, + + // store closures so they live as long as we want and can be dropped on close + ws_onmessage: Rc>, + ws_onopen: Rc>, + ws_onclose: Rc>, + ws_onerror: Rc>, } impl WebSocketService { @@ -27,24 +50,35 @@ impl WebSocketService { let base_href = get_base_href(); Self { connected: Rc::new(AtomicBool::new(false)), + attempt_counter: Rc::new(AtomicU16::new(0)), ws: Rc::new(RefCell::new(None)), status_service, event_service, ws_path: concat_path_leading_slash(&base_href, "ws"), + ws_onmessage: Rc::new(RefCell::new(None)), + ws_onopen: Rc::new(RefCell::new(None)), + ws_onclose: Rc::new(RefCell::new(None)), + ws_onerror: Rc::new(RefCell::new(None)), } } + /// Helper function to allow cloning the service into JS closures for reconnect fn clone_for_reconnect(&self) -> Self { Self { connected: self.connected.clone(), + attempt_counter: self.attempt_counter.clone(), ws: self.ws.clone(), status_service: self.status_service.clone(), event_service: self.event_service.clone(), ws_path: self.ws_path.clone(), + ws_onmessage: self.ws_onmessage.clone(), + ws_onopen: self.ws_onopen.clone(), + ws_onclose: self.ws_onclose.clone(), + ws_onerror: self.ws_onerror.clone(), } } - pub fn connect_ws(&self) { + pub fn connect_ws_with_backoff(&self) { if self.connected.load(Ordering::SeqCst) { return; } @@ -54,126 +88,125 @@ impl WebSocketService { socket.set_binary_type(web_sys::BinaryType::Arraybuffer); let ws_clone = self.ws.clone(); *ws_clone.borrow_mut() = Some(socket.clone()); - let event_service = self.event_service.clone(); - let ws_onmessage_clone = Rc::clone(&ws_clone); // onmessage - let onmessage_callback = Closure::::wrap(Box::new(move |event: MessageEvent| { - trace!("WebSocket received message: {event:?}"); - if let Ok(buf) = event.data().dyn_into::() { - let array = Uint8Array::new(&buf); - let bytes = bytes::Bytes::from(array.to_vec()); - match ProtocolMessage::from_bytes(bytes) { - Ok(message) => { - match message { - ProtocolMessage::Unauthorized => { - event_service.broadcast(EventMessage::Unauthorized); - }, - ProtocolMessage::Error(err) => { - error!("{err}"); - }, - ProtocolMessage::ActiveUserResponse(event) => { - event_service.broadcast(EventMessage::ActiveUser(event)); - }, - ProtocolMessage::ActiveProviderResponse(provider_name, connections) => { - event_service.broadcast(EventMessage::ActiveProvider(provider_name, connections)); - if let Some(token) = get_token() { - Self::try_send_message(ws_onmessage_clone.borrow().as_ref(), ProtocolMessage::ActiveProviderCountRequest(token)); - } - }, - ProtocolMessage::ActiveProviderCountResponse(connections) => { - event_service.broadcast(EventMessage::ActiveProviderCount(connections)); - } - ProtocolMessage::StatusResponse(status) => { - let data = Rc::new(status); - event_service.broadcast(EventMessage::ServerStatus(data)); - } - ProtocolMessage::ConfigChangeResponse(config_type) => { - if !event_service.is_config_change_message_blocked() { - event_service.broadcast(EventMessage::ConfigChange(config_type)); - } - } - ProtocolMessage::ServerError(error) => { - event_service.broadcast(EventMessage::ServerError(error)); - } - ProtocolMessage::PlaylistUpdateResponse(update_state) => { - event_service.broadcast(EventMessage::PlaylistUpdate(update_state)); - } - ProtocolMessage::PlaylistUpdateProgressResponse(target, msg) => { - event_service.broadcast(EventMessage::PlaylistUpdateProgress(target, msg)); - } - ProtocolMessage::Version(_) => { - if let Some(token) = get_token() { - Self::try_send_message(ws_onmessage_clone.borrow().as_ref(), ProtocolMessage::Auth(token)); - } - } - ProtocolMessage::UserActionResponse(_success) => { - // Success is already handled in the UI component that initiated the action - } - ProtocolMessage::Auth(_) - | ProtocolMessage::Authorized - | ProtocolMessage::ActiveProviderCountRequest(_) - | ProtocolMessage::StatusRequest(_) - | ProtocolMessage::UserAction(_) => {} - } - } - Err(err) => error!("Failed to decode websocket message: {err}") + { + let ws_onmessage_ref = self.ws_onmessage.clone(); + let ws_onmessage_clone = ws_clone.clone(); + let event_service = self.event_service.clone(); + let attempt_counter = self.attempt_counter.clone(); + let onmessage_callback = Closure::::wrap(Box::new(move |event: MessageEvent| { + trace!("WebSocket received message: {event:?}"); + if let Some(response) = handle_socket_protocol_msg(event, &event_service, &attempt_counter) { + Self::try_send_message(ws_onmessage_clone.borrow().as_ref(), response); } - } - })); - socket.set_onmessage(Some(onmessage_callback.as_ref().unchecked_ref())); - onmessage_callback.forget(); // Important: leak the closure to keep it alive + })); + socket.set_onmessage(Some(onmessage_callback.as_ref().unchecked_ref())); + // store closure so we can drop it later (no forget) + ws_onmessage_ref.borrow_mut().replace(onmessage_callback); + } - let ws_open_clone = Rc::clone(&ws_clone); - let connected_clone = self.connected.clone(); - let event_service_clone = Rc::clone(&self.event_service); // onopen - let onopen_callback = Closure::::wrap(Box::new(move |_event: Event| { - trace!("WebSocket connection opened."); - connected_clone.store(true, Ordering::SeqCst); - Self::try_send_message(ws_open_clone.borrow().as_ref(), ProtocolMessage::Version(PROTOCOL_VERSION)); - event_service_clone.broadcast(EventMessage::WebSocketStatus(true)); - })); - socket.set_onopen(Some(onopen_callback.as_ref().unchecked_ref())); - onopen_callback.forget(); + { + let ws_onopen_ref = self.ws_onopen.clone(); + let ws_open_clone = ws_clone.clone(); + let connected_clone = self.connected.clone(); + let onopen_callback = Closure::::wrap(Box::new(move |_event: Event| { + // on open is called on a connect attempt, it does not mean it is connected! + trace!("WebSocket connection opened."); + if Self::try_send_message(ws_open_clone.borrow().as_ref(), ProtocolMessage::Version(PROTOCOL_VERSION)) { + connected_clone.store(true, Ordering::SeqCst); + } + })); + socket.set_onopen(Some(onopen_callback.as_ref().unchecked_ref())); + ws_onopen_ref.borrow_mut().replace(onopen_callback); + } - let ws_close_rc = self.ws.clone(); - let connected_clone = self.connected.clone(); + // prepare a reconnect-holder (shared clone) let ws_service_reconnect = Rc::new(self.clone_for_reconnect()); - let event_service_clone = Rc::clone(&self.event_service); - let onclose_callback = Closure::::wrap(Box::new(move |e: CloseEvent| { - debug!("Websocket closed, Code: {}, Reason: {}, WasClean: {}", e.code(), e.reason(), e.was_clean()); - *ws_close_rc.borrow_mut() = None; - connected_clone.store(false, Ordering::SeqCst); - event_service_clone.broadcast(EventMessage::WebSocketStatus(false)); - // schedule reconnect after 3 seconds - let ws_service_inner = ws_service_reconnect.clone(); - set_timeout(move || { ws_service_inner.connect_ws(); }, WS_RECONNECT_MS); - })); - socket.set_onclose(Some(onclose_callback.as_ref().unchecked_ref())); - onclose_callback.forget(); + // onclose + { + let ws_onclose_ref = self.ws_onclose.clone(); + let ws_onmessage_ref = self.ws_onmessage.clone(); + let ws_onopen_ref = self.ws_onopen.clone(); + let ws_onerror_ref = self.ws_onerror.clone(); + let ws_close_rc = self.ws.clone(); + let connected_clone = self.connected.clone(); + let event_service_clone = Rc::clone(&self.event_service); + let ws_service_reconnect_clone = Rc::clone(&ws_service_reconnect); + + let onclose_callback = Closure::::wrap(Box::new(move |e: CloseEvent| { + trace!( + "WebSocket closed (Code {}, Reason: {}, Clean: {})", + e.code(), e.reason(), e.was_clean() + ); + + // clear JS handlers on the underlying socket if present + if let Some(s) = ws_close_rc.borrow().as_ref() { + s.set_onmessage(None); + s.set_onopen(None); + s.set_onerror(None); + s.set_onclose(None); + } + + // drop stored closures so they are freed + ws_onmessage_ref.borrow_mut().take(); + ws_onopen_ref.borrow_mut().take(); + ws_onerror_ref.borrow_mut().take(); + // Note: we keep ws_onclose alive (this closure) until function returns; + // it will be dropped when the service or field is taken elsewhere if desired. + + *ws_close_rc.borrow_mut() = None; + connected_clone.store(false, Ordering::SeqCst); + event_service_clone.broadcast(EventMessage::WebSocketStatus(false)); + + // schedule reconnect + ws_service_reconnect_clone.schedule_reconnect(); + })); + socket.set_onclose(Some(onclose_callback.as_ref().unchecked_ref())); + ws_onclose_ref.borrow_mut().replace(onclose_callback); + } - let connected_clone = self.connected.clone(); - let ws_service_reconnect = Rc::new(self.clone_for_reconnect()); - let event_service_clone = Rc::clone(&self.event_service); // onerror - let onerror_callback = Closure::::wrap(Box::new(move |e: ErrorEvent| { - error!("WebSocket error"); - connected_clone.store(false, Ordering::SeqCst); - event_service_clone.broadcast(EventMessage::WebSocketStatus(false)); - web_sys::console::error_1(&e); + { + let ws_onerror_ref = self.ws_onerror.clone(); + let connected_clone = self.connected.clone(); + let event_service_clone = Rc::clone(&self.event_service); + //let ws_service_reconnect_clone = Rc::clone(&ws_service_reconnect); - // schedule reconnect after 3 seconds - let ws_service_inner = ws_service_reconnect.clone(); - set_timeout(move || { ws_service_inner.connect_ws(); }, WS_RECONNECT_MS); - })); - socket.set_onerror(Some(onerror_callback.as_ref().unchecked_ref())); - onerror_callback.forget(); + let onerror_callback = Closure::::wrap(Box::new(move |e: ErrorEvent| { + error!("WebSocket error: {:?}", e); + connected_clone.store(false, Ordering::SeqCst); + event_service_clone.broadcast(EventMessage::WebSocketStatus(false)); + // ws_service_reconnect_clone.schedule_reconnect(); + })); + socket.set_onerror(Some(onerror_callback.as_ref().unchecked_ref())); + ws_onerror_ref.borrow_mut().replace(onerror_callback); + } } } } + fn schedule_reconnect(&self) { + // increment attempts atomically and get the previous value + let attempt = self.attempt_counter.fetch_add(1, Ordering::SeqCst) as u32; + + if attempt >= WS_RECONNECT_MAX_ATTEMPTS { + warn!("WebSocket reconnect attempts exceeded ({attempt}). Giving up."); + return; + } + let delay = reconnect_delay(attempt); + + warn!("WebSocket reconnect attempt #{attempt} scheduled in {} ms", delay); + + // clone_for_reconnect returns a Service with shared internals (Atomic, etc.) + let ws_clone = Rc::new(self.clone_for_reconnect()); + set_timeout(move || { + ws_clone.connect_ws_with_backoff(); + }, delay as i32); + } + fn try_send_message(ws_opt: Option<&WebSocket>, msg: ProtocolMessage) -> bool { if let Some(ws) = ws_opt { match msg.to_bytes() { @@ -212,30 +245,71 @@ impl WebSocketService { Err(err) => {error!("Failed to get server status: {err:?}");} } } - - - // TODO - // on no wesocket connection - - // let fetch_status = { - // let status = status_signal.clone(); - // let services_ctx = services_ctx.clone(); - // move || { - // let status = status.clone(); - // let services_ctx = services_ctx.clone(); - // spawn_local(async move { - // status.set(services_ctx.status.get_server_status().await.ok()); - // }); - // } - // }; - // - // fetch_status(); - // // all 5 seconds - // let interval = Interval::new(5000, move || { - // fetch_status(); - // }); - // - // // Cleanup function - // || drop(interval) } -} \ No newline at end of file +} + +fn handle_socket_protocol_msg(event: MessageEvent, event_service: &Rc, attempt_counter: &Rc) -> Option{ + if let Ok(buf) = event.data().dyn_into::() { + let array = Uint8Array::new(&buf); + let bytes = bytes::Bytes::from(array.to_vec()); + match ProtocolMessage::from_bytes(bytes) { + Ok(message) => { + match message { + ProtocolMessage::Unauthorized => { + event_service.broadcast(EventMessage::Unauthorized); + }, + ProtocolMessage::Error(err) => { + error!("{err}"); + }, + ProtocolMessage::ActiveUserResponse(event) => { + event_service.broadcast(EventMessage::ActiveUser(event)); + }, + ProtocolMessage::ActiveProviderResponse(provider_name, connections) => { + event_service.broadcast(EventMessage::ActiveProvider(provider_name, connections)); + if let Some(token) = get_token() { + return Some(ProtocolMessage::ActiveProviderCountRequest(token)); + } + }, + ProtocolMessage::ActiveProviderCountResponse(connections) => { + event_service.broadcast(EventMessage::ActiveProviderCount(connections)); + } + ProtocolMessage::StatusResponse(status) => { + let data = Rc::new(status); + event_service.broadcast(EventMessage::ServerStatus(data)); + } + ProtocolMessage::ConfigChangeResponse(config_type) => { + if !event_service.is_config_change_message_blocked() { + event_service.broadcast(EventMessage::ConfigChange(config_type)); + } + } + ProtocolMessage::ServerError(error) => { + event_service.broadcast(EventMessage::ServerError(error)); + } + ProtocolMessage::PlaylistUpdateResponse(update_state) => { + event_service.broadcast(EventMessage::PlaylistUpdate(update_state)); + } + ProtocolMessage::PlaylistUpdateProgressResponse(target, msg) => { + event_service.broadcast(EventMessage::PlaylistUpdateProgress(target, msg)); + } + ProtocolMessage::Version(_) => { + attempt_counter.store(0, Ordering::SeqCst); + event_service.broadcast(EventMessage::WebSocketStatus(true)); + if let Some(token) = get_token() { + return Some(ProtocolMessage::Auth(token)); + } + } + ProtocolMessage::UserActionResponse(_success) => { + // Success is already handled in the UI component that initiated the action + } + ProtocolMessage::Auth(_) + | ProtocolMessage::Authorized + | ProtocolMessage::ActiveProviderCountRequest(_) + | ProtocolMessage::StatusRequest(_) + | ProtocolMessage::UserAction(_) => {} + } + } + Err(err) => error!("Failed to decode websocket message: {err}") + } + } + None +}