qrpc_broker/
lib.rs

1//! An extensible rpc message broker for the libqaul ecosystem.
2
3#[macro_use]
4extern crate tracing;
5
6mod protocol;
7
8use async_std::{net::TcpStream, sync::RwLock, task};
9use identity::Identity;
10use qrpc_sdk::{
11    builders, default_socket_path,
12    error::RpcResult,
13    io::{self, Message},
14    RpcSocket,
15};
16use std::{collections::BTreeMap, sync::Arc};
17
18#[allow(unused)]
19pub(crate) struct ServiceEntry {
20    pub(crate) id: Identity,
21    pub(crate) addr: String,
22    pub(crate) io: TcpStream,
23}
24
25type ConnMap = Arc<RwLock<BTreeMap<String, ServiceEntry>>>;
26
27const ADDRESS: &'static str = "org.qaul._broker";
28
29/// Hold the main broker state
30pub struct Broker {
31    _sock: Arc<RpcSocket>,
32    _conns: ConnMap,
33}
34
35impl Broker {
36    pub async fn new() -> RpcResult<Arc<Self>> {
37        let (addr, port) = default_socket_path();
38        let _conns = Default::default();
39
40        let _sock = {
41            let con = Arc::clone(&_conns);
42            RpcSocket::server(addr, port, |stream, data| reader_loop(stream, data), con).await?
43        };
44
45        Ok(Arc::new(Self { _sock, _conns }))
46    }
47}
48
49fn reader_loop(mut stream: TcpStream, data: ConnMap) {
50    task::block_on(async {
51        loop {
52            if let Err(e) = handle_packet(&mut stream, &data).await {
53                warn!(
54                    "Error occured while accepting packet: {}; dropping stream",
55                    e
56                );
57                break;
58            }
59        }
60    });
61}
62
63async fn handle_packet(s: &mut TcpStream, conns: &ConnMap) -> RpcResult<()> {
64    let Message { id, to, from, data } = io::recv(s).await?;
65    match to.as_str() {
66        ADDRESS => {
67            debug!("Message addressed to broker; handling!");
68            let msg = protocol::broker_command(id, &s, data, &conns).await?;
69            io::send(s, msg).await?;
70            Ok(())
71        }
72        _ => {
73            debug!("Message addressed to bus component; looking up stream!");
74            debug!("Looking up stream: {}", to);
75            let mut t_stream = match conns.read().await.get(&to).map(|s| s.io.clone()) {
76                Some(s) => s,
77                None => {
78                    warn!("Requested component does not exist on qrpc bus!");
79                    io::send(
80                        s,
81                        Message {
82                            id,
83                            to: "<unknown>".into(),
84                            from: ADDRESS.into(),
85                            data: builders::resp_bool(false),
86                        },
87                    )
88                    .await?;
89                    return Ok(());
90                }
91            };
92
93            // If we reach this point, we can send the relay message
94            io::send(&mut t_stream, Message { id, to, from, data }).await?;
95            Ok(())
96        }
97    }
98}
qrpc_broker/lib.rs

qrpc_broker/
lib.rs
