melnet 0.2.11

Low-level RPC protocol used throughout Themelio for peer-to-peer communciations
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

use std::{collections::HashSet, sync::Arc, time::Duration};

use melnet::{NetState, Request};
use rand::prelude::*;
use serde::{Deserialize, Serialize};
use smol::prelude::*;

static EXEC: smol::Executor = smol::Executor::new();

#[derive(Debug, Serialize, Deserialize, Clone)]
struct GossipMsg {
    id: u128,
    body: String,
}

fn main() -> anyhow::Result<()> {
    env_logger::init();
    smol::block_on(EXEC.run(async {
        // create listener
        let tcp_listener = smol::net::TcpListener::bind("127.0.0.1:0").await?;
        println!("MY ADDRESS: {}", tcp_listener.local_addr()?);
        let nstate = NetState::new_with_name("gossip");
        let seen_msgs = Arc::new(parking_lot::RwLock::new(HashSet::new()));
        // register the verbs
        nstate.listen("gossip", move |req: Request<GossipMsg>| {
            let seen_msgs = seen_msgs.clone();
            async move {
                println!("received {:?}", req.body);
                if seen_msgs.read().get(&req.body.id).is_none() {
                    seen_msgs.write().insert(req.body.id);
                    let body = req.body.clone();
                    let state = req.state.clone();
                    // spam to all my neighbors
                    smolscale::spawn(async move {
                        if let Err(e) = spam_neighbors(&state, body).await {
                            println!("failed: {}", e);
                        }
                    })
                    .detach();
                }
                Ok(())
            }
        });
        // listen
        nstate
            .run_server(tcp_listener)
            .or(cmd_prompt(&nstate))
            .await;
        Ok(())
    }))
}

async fn spam_neighbors(nstate: &NetState, req: GossipMsg) -> anyhow::Result<()> {
    for &neigh in nstate.routes().iter() {
        melnet::request(neigh, "gossip", "gossip", req.clone()).await?;
    }
    Ok(())
}

async fn cmd_prompt(nstate: &NetState) {
    loop {
        spam_neighbors(
            nstate,
            GossipMsg {
                id: rand::thread_rng().gen(),
                body: "Hello World!".into(),
            },
        )
        .await
        .unwrap();
        smol::Timer::after(Duration::from_secs(10)).await;
    }
}