From fccd03c3dfee551de118c9d3cb5b7b679f7a2987 Mon Sep 17 00:00:00 2001 From: Klesti Selimaj Date: Fri, 31 Jul 2026 01:17:35 +0200 Subject: [PATCH] Strategy split ws --- Cargo.lock | 29 +++++++++++++++-------------- Cargo.toml | 1 + src/engine/engine/mod.rs | 14 +++++++++++--- src/engine/engine/strategy.rs | 14 +++++++------- 4 files changed, 34 insertions(+), 24 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 2242481..36a13bf 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2432,9 +2432,9 @@ checksum = "e6d5a32815ae3f33302d95fdcb2ce17862f8c65363dcfd29360480ba1001fc9c" [[package]] name = "futures" -version = "0.3.32" +version = "0.3.33" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8b147ee9d1f6d097cef9ce628cd2ee62288d963e16fb287bd9286455b241382d" +checksum = "a88cf1f829d945f548cf8fec32c61b1f202b6d93b45848602fc02af4b12ad218" dependencies = [ "futures-channel", "futures-core", @@ -2447,9 +2447,9 @@ dependencies = [ [[package]] name = "futures-channel" -version = "0.3.32" +version = "0.3.33" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "07bbe89c50d7a535e539b8c17bc0b49bdb77747034daa8087407d655f3f7cc1d" +checksum = "262590f4fe6afeb0bc83be1daa64e52657fe185690a958af7f3ad0e92085c5ae" dependencies = [ "futures-core", "futures-sink", @@ -2457,15 +2457,15 @@ dependencies = [ [[package]] name = "futures-core" -version = "0.3.32" +version = "0.3.33" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7e3450815272ef58cec6d564423f6e755e25379b217b0bc688e295ba24df6b1d" +checksum = "2cd50c473c80f6d7c3670a752354b8e569b1a7cbfdc0419ec88e5edad85e0dc7" [[package]] name = "futures-executor" -version = "0.3.32" +version = "0.3.33" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "baf29c38818342a3b26b5b923639e7b1f4a61fc5e76102d4b1981c6dc7a7579d" +checksum = "6754879cc9f2c66f88c6e5c35344bb0bdb0708b0352b1201815667c7eabc7458" dependencies = [ "futures-core", "futures-task", @@ -2480,9 +2480,9 @@ checksum = "4577ecaa3c4f96589d473f679a71b596316f6641bc350038b962a5daf0085d7a" [[package]] name = "futures-macro" -version = "0.3.32" +version = "0.3.33" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e835b70203e41293343137df5c0664546da5745f82ec9b84d40be8336958447b" +checksum = "2d6d3cde68c518367be28956066ddfef33813991b77a55005a69dae04bf3b10b" dependencies = [ "proc-macro2", "quote", @@ -2497,15 +2497,15 @@ checksum = "e34418ac499d6305c2fb5ad0ed2f6ac998c5f8ca209b4510f7f94242c647e307" [[package]] name = "futures-task" -version = "0.3.32" +version = "0.3.33" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "037711b3d59c33004d3856fbdc83b99d4ff37a24768fa1be9ce3538a1cde4393" +checksum = "b231ed28831efb4a61a08580c4bc233ec56bc009f4cd8f52da2c3cb97df0c109" [[package]] name = "futures-util" -version = "0.3.32" +version = "0.3.33" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "389ca41296e6190b48053de0321d02a77f32f8a5d2461dd38762c0593805c6d6" +checksum = "a77a90a256fce34da66415271e30f94ee91c57b04b8a2c042d9cf3220179deaa" dependencies = [ "futures-channel", "futures-core", @@ -3959,6 +3959,7 @@ dependencies = [ "anyhow", "chrono", "crossterm", + "futures", "hypersdk", "postcard", "pulse-sdk", diff --git a/Cargo.toml b/Cargo.toml index c1d3ab1..2209230 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -25,6 +25,7 @@ chrono = "0.4.45" serde_json = "1" rand = "0.8.7" toml = "1.1.3" +futures = "0.3.33" [workspace] members = ["pulse-ui", "pulse-sdk"] diff --git a/src/engine/engine/mod.rs b/src/engine/engine/mod.rs index b9cc3f6..89bc936 100644 --- a/src/engine/engine/mod.rs +++ b/src/engine/engine/mod.rs @@ -7,6 +7,7 @@ use crate::{ store::{accounts::AccountList, config::Config}, terminal::TerminalServer, }; +use hypersdk::hypercore::ws::ConnectionStream; use pulse_sdk::prelude::*; use std::{collections::HashMap, sync::Arc}; use tokio::{sync::Mutex, task::JoinHandle}; @@ -19,8 +20,12 @@ pub struct WatchList { #[derive(Clone)] pub struct Engine { + // engine pub terminal_server: Arc, pub strategy_engine: Arc, + pub ws_stream: Arc>, + + // data pub config: Arc>, pub accounts: Arc>, pub watch_list: Arc>, @@ -31,16 +36,19 @@ impl Engine { pub async fn new() -> tokio::io::Result> { let config = Config::new().await?; - let strategy = StrategyEngine::new(&config.strategy).await?; + let (ws_handle, ws_stream) = hypersdk::hypercore::mainnet_ws().split(); + + let strategy = StrategyEngine::new(&config.strategy, ws_handle).await?; let accounts = Arc::new(Mutex::new(AccountList::new().await?)); let config = Arc::new(Mutex::new(config)); Ok(Arc::new_cyclic(|engine| Self { - terminal_server: TerminalServer::new(engine.clone()), - strategy_engine: strategy.initialize(engine.clone()), config, accounts, + ws_stream: Arc::new(Mutex::new(ws_stream)), + terminal_server: TerminalServer::new(engine.clone()), + strategy_engine: strategy.initialize(engine.clone()), watch_list: Arc::new(Mutex::new(WatchList { name_to_index: HashMap::new(), items: Vec::new(), diff --git a/src/engine/engine/strategy.rs b/src/engine/engine/strategy.rs index 6c1d372..90da13c 100644 --- a/src/engine/engine/strategy.rs +++ b/src/engine/engine/strategy.rs @@ -1,5 +1,5 @@ use anyhow::Context; -use hypersdk::hypercore::{self, CandleInterval, Subscription, WebSocket}; +use hypersdk::hypercore::{self, CandleInterval, Subscription, ws::ConnectionHandle}; use pulse_sdk::prelude::*; use std::{ collections::HashSet, @@ -19,12 +19,12 @@ use crate::{ pub struct StrategyEngine { pub engine: Weak, pub strategy: Mutex, - pub ws: WebSocket, + pub ws_handle: ConnectionHandle, pub subscriptions: Mutex>, } impl StrategyEngine { - pub async fn new(strategy_id: &str) -> tokio::io::Result { + pub async fn new(strategy_id: &str, ws_handle: ConnectionHandle) -> tokio::io::Result { let strategy = pulse_strategy(strategy_id)?; let (strategy, strategy_manifest) = get_manifest( @@ -42,7 +42,7 @@ impl StrategyEngine { Ok(Self { strategy: Mutex::new(StrategyChild::new(strategy, strategy_manifest)), engine: Weak::new(), - ws: hypercore::mainnet_ws(), + ws_handle, subscriptions: Mutex::new(HashSet::new()), }) } @@ -142,18 +142,18 @@ impl StrategyEngine { } Some(StrategyMessage::Subscribe(subscription)) => { - self.ws.subscribe(subscription.clone()); + self.ws_handle.subscribe(subscription.clone()); self.subscriptions.lock().await.insert(subscription); } Some(StrategyMessage::Unsubscribe(subscription)) => { self.subscriptions.lock().await.remove(&subscription); - self.ws.unsubscribe(subscription); + self.ws_handle.unsubscribe(subscription); } Some(StrategyMessage::UnsubscribeAll) => { for sub in self.subscriptions.lock().await.drain() { - self.ws.unsubscribe(sub); + self.ws_handle.unsubscribe(sub); } }