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

use crate::ws::{error::SharedWsError, response::SharedWsResponseTrait};
use std::{collections::HashMap, fmt::Debug, hash::Hash};
use tokio::sync::mpsc;

#[async_trait::async_trait]
pub trait SharedWsStreamTrait<
    WsStreamId: Hash + Eq + Clone + Debug + Send,
    Response: SharedWsResponseTrait<WsStreamId> + Send + Debug,
>
{
    fn get_stream_tx_map(
        &mut self,
    ) -> &mut HashMap<WsStreamId, mpsc::Sender<Result<Response, SharedWsError>>>;

    async fn recv_stream_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 stream response");
                        (id, Err(err))
                    }
                };

                let tx_map = self.get_stream_tx_map();

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