xapi_shared/ws/
api.rs

1use crate::ws::{
2    error::SharedWsError, payload::SharedWsApiPayloadTrait, response::SharedWsResponseTrait,
3};
4use ezsockets::{Client, ClientExt};
5use std::{collections::HashMap, fmt::Debug, hash::Hash};
6use tokio::sync::oneshot;
7
8pub trait SharedWsApiTrait<
9    WsApiId: Hash + Eq + Clone + Debug,
10    Payload: SharedWsApiPayloadTrait<WsApiId>,
11    Response: SharedWsResponseTrait<WsApiId> + Debug,
12> where
13    Self: Sized + ClientExt,
14{
15    fn get_client(&self) -> &Client<Self>;
16
17    fn get_oneshot_tx_map(
18        &mut self,
19    ) -> &mut HashMap<WsApiId, oneshot::Sender<Result<Response, SharedWsError>>>;
20
21    fn send_oneshot(
22        &mut self,
23        payload: Payload,
24        tx: oneshot::Sender<Result<Response, SharedWsError>>,
25    ) -> Result<(), SharedWsError> {
26        let id = payload.get_id();
27
28        let oneshot_tx_map = self.get_oneshot_tx_map();
29
30        if oneshot_tx_map.contains_key(id) {
31            tracing::error!(?id, "duplicated id in ws api request");
32            return Err(SharedWsError::InvalidIdError(format!("{id:?}")));
33        }
34
35        let text = payload.serialize()?;
36
37        oneshot_tx_map.insert(id.clone(), tx);
38
39        tracing::info!(?text, "sending ws api request");
40        self.get_client()
41            .text(text)
42            .inspect_err(|err| {
43                tracing::error!(?err, "failed to send ws api request");
44            })
45            .map_err(|err| SharedWsError::ChannelClosedError(err.to_string()))?;
46
47        Ok(())
48    }
49
50    fn recv_oneshot_resp(&mut self, text: &str) -> Option<Result<(), SharedWsError>> {
51        match Response::try_parse(text) {
52            None => None,
53            Some(result) => {
54                let (id, msg) = match result {
55                    Ok(resp) => {
56                        let id = resp.get_id().clone();
57
58                        (id, Ok(resp))
59                    }
60                    Err((id, err)) => {
61                        tracing::error!(?id, ?err, "failed to parse ws api response");
62                        (id, Err(err))
63                    }
64                };
65
66                let tx_map = self.get_oneshot_tx_map();
67
68                Some(match tx_map.remove(&id) {
69                    None => {
70                        tracing::error!(?id, ?msg, "missing ws api resp channel");
71                        Err(SharedWsError::InvalidIdError(format!("{id:?}")))
72                    }
73                    Some(tx) => tx
74                        .send(msg)
75                        .inspect_err(|err| tracing::error!(?err, "failed to send to resp channel"))
76                        .map_err(|err| SharedWsError::ChannelClosedError(format!("{err:?}"))),
77                })
78            }
79        }
80    }
81}
xapi_shared/ws/api.rs

xapi_shared/ws/
api.rs
