senax_common/
linker.rs

1use anyhow::{Context as _, Result, bail};
2use bytes::Bytes;
3use std::marker::PhantomData;
4use tokio::sync::mpsc::{self, UnboundedReceiver, UnboundedSender};
5
6#[cfg(all(feature = "uring", target_os = "linux"))]
7pub mod common;
8pub mod stream;
9#[cfg(all(feature = "uring", target_os = "linux"))]
10mod tcp_client;
11#[cfg(all(feature = "uring", target_os = "linux"))]
12mod unix_client;
13
14#[derive(Debug, Clone)]
15pub struct Sender<T> {
16    tx: UnboundedSender<Bytes>,
17    _phantom: PhantomData<T>,
18}
19impl<T> Sender<T>
20where
21    T: senax_encoder::Encoder,
22{
23    pub fn send(&self, data: &T) -> Result<()> {
24        let bytes = senax_encoder::encode(data)?;
25        self.tx.send(bytes)?;
26        Ok(())
27    }
28}
29#[derive(Debug)]
30pub struct Receiver<T> {
31    rx: UnboundedReceiver<Bytes>,
32    _phantom: PhantomData<T>,
33}
34impl<T> Receiver<T>
35where
36    T: senax_encoder::Decoder,
37{
38    /// Receive data from the Linker.
39    /// If the connection with the Linker is disconnected, return None.
40    /// If there is an abnormal disconnection between Linkers and it reconnects, return Some(None).
41    pub async fn recv(&mut self) -> Option<Option<Result<T>>> {
42        match self.rx.recv().await {
43            Some(mut v) => {
44                if v.is_empty() {
45                    Some(None)
46                } else {
47                    Some(Some(senax_encoder::decode(&mut v).context("parse error")))
48                }
49            }
50            None => None,
51        }
52    }
53}
54
55pub fn link<T>(
56    stream_id: u64,
57    port: &str,
58    pw: &str,
59    exit_tx: mpsc::Sender<i32>,
60    send_only: bool,
61) -> Result<(Sender<T>, Receiver<T>)>
62where
63    T: senax_encoder::Encoder + senax_encoder::Decoder,
64{
65    let (to_linker, from_linker) = LinkerClient::start(port, stream_id, pw, exit_tx, send_only)?;
66    Ok((
67        Sender {
68            tx: to_linker,
69            _phantom: Default::default(),
70        },
71        Receiver {
72            rx: from_linker,
73            _phantom: Default::default(),
74        },
75    ))
76}
77
78pub struct LinkerClient;
79#[cfg(all(feature = "uring", target_os = "linux"))]
80#[allow(clippy::type_complexity)]
81impl LinkerClient {
82    pub fn start(
83        port: &str,
84        stream_id: u64,
85        pw: &str,
86        exit_tx: mpsc::Sender<i32>,
87        send_only: bool,
88    ) -> Result<(UnboundedSender<Bytes>, UnboundedReceiver<Bytes>)> {
89        let (to_linker, from_local) = mpsc::unbounded_channel();
90        let (to_local, from_linker) = mpsc::unbounded_channel();
91        if port.starts_with('/') {
92            match unix_client::run(
93                port,
94                stream_id,
95                from_local,
96                to_local,
97                pw.to_string(),
98                exit_tx,
99                send_only,
100            ) {
101                Ok(_) => {
102                    return Ok((to_linker, from_linker));
103                }
104                Err(e) => {
105                    log::warn!("{}", e);
106                }
107            }
108        } else {
109            match tcp_client::run(
110                port,
111                stream_id,
112                from_local,
113                to_local,
114                pw.to_string(),
115                exit_tx,
116                send_only,
117            ) {
118                Ok(_) => {
119                    return Ok((to_linker, from_linker));
120                }
121                Err(e) => {
122                    log::warn!("{}", e);
123                }
124            }
125        }
126        bail!("linker connection failed");
127    }
128}
129
130#[cfg(not(all(feature = "uring", target_os = "linux")))]
131#[allow(unused_variables)]
132#[allow(clippy::type_complexity)]
133impl LinkerClient {
134    pub fn start(
135        _port: &str,
136        _stream_id: u64,
137        _pw: &str,
138        _exit_tx: mpsc::Sender<i32>,
139        _send_only: bool,
140    ) -> Result<(UnboundedSender<Bytes>, UnboundedReceiver<Bytes>)> {
141        bail!("linker is not supported");
142    }
143}
senax_common/linker.rs

senax_common/
linker.rs
