xapi-shared 0.0.1

XAPI Shared Library
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81

use crate::ws::{
    error::SharedWsError, payload::SharedWsApiPayloadTrait, response::SharedWsResponseTrait,
};
use ezsockets::{Client, ClientExt};
use std::{collections::HashMap, fmt::Debug, hash::Hash};
use tokio::sync::oneshot;

pub trait SharedWsApiTrait<
    WsApiId: Hash + Eq + Clone + Debug,
    Payload: SharedWsApiPayloadTrait<WsApiId>,
    Response: SharedWsResponseTrait<WsApiId> + Debug,
> where
    Self: Sized + ClientExt,
{
    fn get_client(&self) -> &Client<Self>;

    fn get_oneshot_tx_map(
        &mut self,
    ) -> &mut HashMap<WsApiId, oneshot::Sender<Result<Response, SharedWsError>>>;

    fn send_oneshot(
        &mut self,
        payload: Payload,
        tx: oneshot::Sender<Result<Response, SharedWsError>>,
    ) -> Result<(), SharedWsError> {
        let id = payload.get_id();

        let oneshot_tx_map = self.get_oneshot_tx_map();

        if oneshot_tx_map.contains_key(id) {
            tracing::error!(?id, "duplicated id in ws api request");
            return Err(SharedWsError::InvalidIdError(format!("{id:?}")));
        }

        let text = payload.serialize()?;

        oneshot_tx_map.insert(id.clone(), tx);

        tracing::info!(?text, "sending ws api request");
        self.get_client()
            .text(text)
            .inspect_err(|err| {
                tracing::error!(?err, "failed to send ws api request");
            })
            .map_err(|err| SharedWsError::ChannelClosedError(err.to_string()))?;

        Ok(())
    }

    fn recv_oneshot_resp(&mut self, text: &str) -> Option<Result<(), SharedWsError>> {
        match Response::try_parse(text) {
            None => None,
            Some(result) => {
                let (id, msg) = match result {
                    Ok(resp) => {
                        let id = resp.get_id().clone();

                        (id, Ok(resp))
                    }
                    Err((id, err)) => {
                        tracing::error!(?id, ?err, "failed to parse ws api response");
                        (id, Err(err))
                    }
                };

                let tx_map = self.get_oneshot_tx_map();

                Some(match tx_map.remove(&id) {
                    None => {
                        tracing::error!(?id, ?msg, "missing ws api resp channel");
                        Err(SharedWsError::InvalidIdError(format!("{id:?}")))
                    }
                    Some(tx) => tx
                        .send(msg)
                        .inspect_err(|err| tracing::error!(?err, "failed to send to resp channel"))
                        .map_err(|err| SharedWsError::ChannelClosedError(format!("{err:?}"))),
                })
            }
        }
    }
}