calendar_client 0.1.1

use tokio::task::JoinHandle;
use tokio_tungstenite::{connect_async, tungstenite::Message};
use futures_util::{SinkExt, StreamExt};

use super::Client;

const WS_URL: &str = "wss://stream.tradingeconomics.com/";
const KEEP_ALIVE: &str = "{\"topic\":\"keepalive\"}";
const SUBSCRIPTION: &str = "{\"topic\":\"subscribe\",\"to\":\"calendar\"}";

impl Client {
    pub async fn subscribe_to_ws<S: Send + Clone + 'static>(
        &self,
        state: S,
        callback: impl Send + FnOnce(String, S) -> () + 'static + Copy,
    ) -> JoinHandle<()> {
        let ws_url = format!("{WS_URL}?client={}:{}", self.key, self.secret);
        tokio::spawn(async move {
            loop {
                let (stream, _) = connect_async(&ws_url)
                    .await
                    .expect("failed to connect to ws");
                println!("\n==== connected to ws ====\n");
                let (mut send, mut recv) = stream.split();
                send.send(Message::Text(SUBSCRIPTION.to_string()))
                    .await
                    .expect("failed to send subscription message");
                while let Some(msg) = recv.next().await {
                    match msg {
                        Ok(msg) => match msg {
                            Message::Text(msg) => {
                                if msg == KEEP_ALIVE {
                                    // println!("got keep alive, sending keep alive");
                                    let _ = send.send(Message::Text(KEEP_ALIVE.to_string())).await;
                                } else {
                                    callback(msg, state.clone());
                                }
                            }
                            Message::Close(close) => {
                                println!("WS CLOSE: {close:#?}");
                            }
                            _ => {}
                        },
                        Err(e) => {
                            eprintln!("RECV ERROR: {e:#?}");
                        }
                    }
                }
                println!("socket closed, reconnecting...");
            }
        })
    }
}