inter_thread/
inter_thread.rs

1// Demo of inter-thread communication (with no RPC layer) with a UNIX socket for external clients
2use busrt::broker::{Broker, ServerConfig};
3use busrt::client::AsyncClient;
4use busrt::QoS;
5use std::time::Duration;
6use tokio::time::sleep;
7
8const SLEEP_STEP: Duration = Duration::from_secs(1);
9
10#[tokio::main]
11async fn main() -> Result<(), Box<dyn std::error::Error>> {
12    // create a new broker instance
13    let mut broker = Broker::new();
14    // init the default broker RPC API, optional
15    broker.init_default_core_rpc().await?;
16    // spawn unix server for external clients
17    broker
18        .spawn_unix_server("/tmp/busrt.sock", ServerConfig::default())
19        .await?;
20    // worker 1 will send to worker2 direct "hello" message
21    let mut client1 = broker.register_client("worker.1").await?;
22    // worker 2 will listen to incoming frames only
23    let mut client2 = broker.register_client("worker.2").await?;
24    // worker 3 will send broadcasts to all workers, an external client with a name "worker.N" can
25    // connect the broker via unix socket and receive them as well or send a message to "worker.2"
26    // to print it
27    let mut client3 = broker.register_client("worker.3").await?;
28    let rx = client2.take_event_channel().unwrap();
29    tokio::spawn(async move {
30        loop {
31            client1
32                .send("worker.2", "hello".as_bytes().into(), QoS::No)
33                .await
34                .unwrap();
35            sleep(SLEEP_STEP).await;
36        }
37    });
38    tokio::spawn(async move {
39        loop {
40            client3
41                .send_broadcast(
42                    "worker.*",
43                    "this is a broadcast message".as_bytes().into(),
44                    QoS::No,
45                )
46                .await
47                .unwrap();
48            sleep(SLEEP_STEP).await;
49        }
50    });
51    while let Ok(frame) = rx.recv().await {
52        println!(
53            "{}: {}",
54            frame.sender(),
55            std::str::from_utf8(frame.payload()).unwrap_or("something unreadable")
56        );
57    }
58    Ok(())
59}
inter_thread/inter_thread.rs

inter_thread/
inter_thread.rs
