diff --git a/src/app.rs b/src/app.rs index bdcc77e..028de29 100644 --- a/src/app.rs +++ b/src/app.rs @@ -15,9 +15,13 @@ use std::sync::Arc; use std::time::Duration; use wasm_bindgen::JsCast; +use std::time::Duration; + use crate::event_utils::{EventOptions, wheel_event_options, window_event_listener_with_options}; use crate::global_signals; -use crate::global_state::{ensure_chart, get_chart_signal, set_chart_in_ecs, view_state}; +use crate::global_state::{ + connection_id, domain_state, ensure_chart, get_chart_signal, set_chart_in_ecs, view_state, +}; use crate::{ domain::{ chart::Chart, @@ -1202,6 +1206,13 @@ pub async fn start_websocket_stream(set_status: WriteSignal) { } let interval = current_interval().get_untracked(); + let conn_id = connection_id().get_untracked() + 1; + connection_id().set(conn_id); + domain_state().update(|ds| { + ds.timeframe = Duration::from_millis(interval.duration_ms()); + ds.candles = Arc::new(Vec::new()); + ds.indicators = Arc::new(Vec::new()); + }); let rest_client_arc = Arc::new(Mutex::new(BinanceWebSocketClient::new(symbol.clone(), interval))); @@ -1214,8 +1225,11 @@ pub async fn start_websocket_stream(set_status: WriteSignal) { let hist_res = { let client = rest_client_arc.lock().await; - client.fetch_historical_data(500).await + client.fetch_historical_data(1000).await }; + if conn_id != connection_id().get_untracked() { + return; + } match hist_res { Ok(historical_candles) => { get_logger().info( @@ -1225,6 +1239,10 @@ pub async fn start_websocket_stream(set_status: WriteSignal) { chart.update(|ch| ch.set_historical_data(historical_candles.clone())); chart.with_untracked(|c| set_chart_in_ecs(&symbol, c.clone())); + domain_state().update(|ds| { + ds.candles = Arc::new(historical_candles.clone()); + ds.indicators = Arc::new(Vec::new()); + }); chart.with_untracked(|c| { if c.get_candle_count() > 0 && with_global_renderer(|r| { @@ -1295,8 +1313,11 @@ pub async fn start_websocket_stream(set_status: WriteSignal) { let fut = futures::future::Abortable::new( async move { let handler_handle = handle_check.clone(); + let connection_guard = conn_id; let handler = move |candle: Candle| { - if handler_handle.is_aborted() { + if handler_handle.is_aborted() + || connection_guard != connection_id().get_untracked() + { return; } global_current_price().set(candle.ohlcv.close.value()); @@ -1313,6 +1334,11 @@ pub async fn start_websocket_stream(set_status: WriteSignal) { }); chart.with_untracked(|c| set_chart_in_ecs(&symbol, c.clone())); crate::global_state::push_realtime_candle(candle.clone()); + domain_state().update(|ds| { + let mut v = (*ds.candles).clone(); + v.push(candle.clone()); + ds.candles = Arc::new(v); + }); let count = chart.with(|c| c.get_candle_count()); global_candle_count().set(count); diff --git a/src/global_state.rs b/src/global_state.rs index b9280fb..d44229e 100644 --- a/src/global_state.rs +++ b/src/global_state.rs @@ -35,6 +35,7 @@ pub struct Globals { pub line_visibility: RwSignal, pub domain_state: RwSignal, pub view_state: RwSignal, + pub connection_id: RwSignal, } // The `OnceCell` ensures this state is created at most once on demand. @@ -63,6 +64,7 @@ pub fn globals() -> &'static Globals { Arc::new(Vec::new()), )), view_state: create_rw_signal(ViewState::new(5.0, 1.0, 20.0)), + connection_id: create_rw_signal(0), }) } @@ -103,6 +105,10 @@ pub fn view_state() -> RwSignal { globals().view_state } +pub fn connection_id() -> RwSignal { + globals().connection_id +} + /// Add a candle to the ECS world and process systems. pub fn push_realtime_candle(candle: Candle) { use crate::ecs::components::CandleComponent; diff --git a/tests/switch_symbol_loop.rs b/tests/switch_symbol_loop.rs new file mode 100644 index 0000000..54d5997 --- /dev/null +++ b/tests/switch_symbol_loop.rs @@ -0,0 +1,43 @@ +use futures::future::AbortHandle; +use leptos::*; +use price_chart_wasm::app::{abort_other_streams, current_symbol, stream_abort_handles}; +use price_chart_wasm::domain::market_data::Symbol; +use price_chart_wasm::global_state::domain_state; +use std::sync::Arc; +use wasm_bindgen::prelude::*; +use wasm_bindgen_test::*; + +wasm_bindgen_test_configure!(run_in_browser); + +fn heap_usage() -> f64 { + js_sys::Reflect::get(&js_sys::global(), &JsValue::from_str("performance")) + .ok() + .and_then(|perf| js_sys::Reflect::get(&perf, &JsValue::from_str("memory")).ok()) + .and_then(|mem| js_sys::Reflect::get(&mem, &JsValue::from_str("usedJSHeapSize")).ok()) + .and_then(|val| val.as_f64()) + .unwrap_or(0.0) +} + +#[wasm_bindgen_test] +fn rapid_symbol_switch_no_leak() { + let start = heap_usage(); + for i in 0..20 { + let sym = if i % 2 == 0 { "BTCUSDT" } else { "ETHUSDT" }; + current_symbol().set(Symbol::from(sym)); + let (handle, _) = AbortHandle::new_pair(); + stream_abort_handles().update(|m| { + m.insert(Symbol::from(sym), handle); + }); + abort_other_streams(&Symbol::from(sym)); + domain_state().update(|ds| { + ds.candles = Arc::new(Vec::new()); + ds.indicators = Arc::new(Vec::new()); + }); + } + assert_eq!(stream_abort_handles().with(|m| m.len()), 1); + let end = heap_usage(); + if start > 0.0 && end > 0.0 { + let growth = (end - start) / start; + assert!(growth <= 0.05); + } +}