#![allow(unused_imports)]
use async_trait::async_trait;
use derive_builder::Builder;
use serde::{Deserialize, Serialize};
use serde_json::Value;
use std::{collections::HashMap, sync::Arc};
use crate::cms::websocket_streams::models;
use crate::common::{
models::ParamBuildError,
utils::replace_websocket_streams_placeholders,
websocket::{WebsocketBase, WebsocketStream, WebsocketStreams, create_stream_handler},
};
use crate::models::StreamId;
#[async_trait]
pub trait Api: Send + Sync {
async fn com_announcement_en(
&self,
) -> anyhow::Result<Arc<WebsocketStream<models::ComAnnouncementEnResponse>>>;
}
pub struct ApiClient {
websocket_streams_base: Arc<WebsocketStreams>,
}
impl ApiClient {
pub fn new(websocket_streams_base: Arc<WebsocketStreams>) -> Self {
Self {
websocket_streams_base,
}
}
}
#[async_trait]
impl Api for ApiClient {
async fn com_announcement_en(
&self,
) -> anyhow::Result<Arc<WebsocketStream<models::ComAnnouncementEnResponse>>> {
let pairs: &[(&str, Option<String>)] = &[];
let vars: HashMap<_, _> = pairs
.iter()
.filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
.collect();
let id_opt: Option<String> = vars.get("id").map(std::string::ToString::to_string);
let stream = replace_websocket_streams_placeholders("/com_announcement_en", &vars);
Ok(create_stream_handler::<models::ComAnnouncementEnResponse>(
WebsocketBase::WebsocketStreams(Arc::clone(&self.websocket_streams_base)),
stream,
id_opt.map(|s| {
if !s.is_empty() && s.bytes().all(|b| b.is_ascii_digit()) {
if let Ok(n) = s.parse::<u32>() {
return StreamId::Number(n);
}
}
StreamId::Str(s)
}),
None,
)
.await)
}
}
#[cfg(all(test, feature = "cms"))]
mod tests {
use super::*;
use crate::TOKIO_SHARED_RT;
use crate::{
common::websocket::{WebsocketConnection, WebsocketHandler},
config::ConfigurationWebsocketStreams,
};
use serde_json::json;
use std::sync::atomic::{AtomicBool, Ordering};
use tokio::task::yield_now;
async fn make_streams_base() -> (Arc<WebsocketStreams>, Arc<WebsocketConnection>) {
let conn = WebsocketConnection::new("test");
let config = ConfigurationWebsocketStreams::builder()
.api_key("test-api-key")
.api_secret("test-api-secret")
.build()
.expect("Failed to build configuration");
let streams_base = WebsocketStreams::new(config, vec![conn.clone()], vec![]);
streams_base.signed_streams.store(true, Ordering::Relaxed);
conn.set_handler(streams_base.clone() as Arc<dyn WebsocketHandler>)
.await;
(streams_base, conn)
}
#[test]
fn com_announcement_en_should_execute_successfully() {
TOKIO_SHARED_RT.block_on(async {
let (streams_base, _) = make_streams_base().await;
let api = ApiClient::new(streams_base.clone());
let pairs: &[(&str, Option<String>)] = &[];
let vars: HashMap<_, _> = pairs
.iter()
.filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
.collect();
let stream = replace_websocket_streams_placeholders("/com_announcement_en", &vars);
let ws_stream = api
.com_announcement_en()
.await
.expect("com_announcement_en should return a WebsocketStream");
assert!(
streams_base.is_subscribed(&stream).await,
"expected stream '{stream}' to be subscribed"
);
assert_eq!(ws_stream.id, None, "signed streams carry no request id");
});
}
#[test]
fn com_announcement_en_should_handle_incoming_message() {
TOKIO_SHARED_RT.block_on(async {
let (streams_base, conn) = make_streams_base().await;
let api = ApiClient::new(streams_base.clone());
let pairs: &[(&str, Option<String>)] = &[];
let vars: HashMap<_, _> = pairs
.iter()
.filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
.collect();
let stream = replace_websocket_streams_placeholders("/com_announcement_en", &vars);
let ws_stream = api.com_announcement_en().await.unwrap();
let called = Arc::new(AtomicBool::new(false));
let called_with_message = called.clone();
ws_stream.on_message(move |_payload: models::ComAnnouncementEnResponse| {
called_with_message.store(true, Ordering::SeqCst);
});
let payload: Value = serde_json::from_str(r#"{"catalogId":161,"catalogName":"Delisting","publishDate":1753257631403,"title":"Notice of...","body":"This is...","disclaimer":"Trade on-the-go..."}"#).unwrap_or_else(|_| serde_json::json!({}));
let msg = json!({
"type": "DATA",
"topic": stream,
"data": payload.to_string(),
});
streams_base.on_message(msg.to_string(), conn.clone()).await;
yield_now().await;
assert!(called.load(Ordering::SeqCst), "expected our callback to have been invoked");
});
}
#[test]
fn com_announcement_en_should_not_fire_after_unsubscribe() {
TOKIO_SHARED_RT.block_on(async {
let (streams_base, conn) = make_streams_base().await;
let api = ApiClient::new(streams_base.clone());
let pairs: &[(&str, Option<String>)] = &[];
let vars: HashMap<_, _> = pairs
.iter()
.filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
.collect();
let stream = replace_websocket_streams_placeholders("/com_announcement_en", &vars);
let ws_stream = api.com_announcement_en().await.unwrap();
let called = Arc::new(AtomicBool::new(false));
let called_clone = called.clone();
ws_stream.on_message(move |_payload: models::ComAnnouncementEnResponse| {
called_clone.store(true, Ordering::SeqCst);
});
assert!(streams_base.is_subscribed(&stream).await, "should be subscribed before unsubscribe");
ws_stream.unsubscribe().await;
let payload: Value = serde_json::from_str(r#"{"catalogId":161,"catalogName":"Delisting","publishDate":1753257631403,"title":"Notice of...","body":"This is...","disclaimer":"Trade on-the-go..."}"#).unwrap_or_else(|_| serde_json::json!({}));
let msg = json!({
"type": "DATA",
"topic": stream,
"data": payload.to_string(),
});
streams_base.on_message(msg.to_string(), conn.clone()).await;
yield_now().await;
assert!(!called.load(Ordering::SeqCst), "callback should not be invoked after unsubscribe");
});
}
}