quickie 0.7.0

A simple, low-level, and customizable implementation of a QUIC P2P node.
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
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142

use std::{
    collections::HashMap,
    io,
    ops::Deref,
    sync::{Arc, OnceLock},
};

use parking_lot::{Mutex, RwLock};
use quinn::{ClientConfig, Endpoint, ServerConfig, StreamId};
use tokio::task::JoinHandle;

use crate::conn::{Conn, ConnId, Streams};

/// Contains objects providing P2P networking capabilities.
#[derive(Clone)]
pub struct Node(Arc<InnerNode>);

impl Deref for Node {
    type Target = Arc<InnerNode>;

    fn deref(&self) -> &Self::Target {
        &self.0
    }
}

#[doc(hidden)]
pub struct InnerNode {
    pub(crate) config: Config,
    pub(crate) endpoint: OnceLock<Endpoint>,
    pub(crate) conns: RwLock<HashMap<ConnId, Conn>>,
    pub(crate) tasks: Mutex<Vec<JoinHandle<()>>>,
}

impl Node {
    /// Creates an idle node with the given config.
    pub fn new(config: Config) -> Self {
        Self(Arc::new(InnerNode {
            config,
            endpoint: Default::default(),
            conns: Default::default(),
            tasks: Default::default(),
        }))
    }

    /// Returns a handle to the QUIC endpoint.
    pub(crate) fn get_endpoint(&self) -> io::Result<&Endpoint> {
        self.endpoint.get().ok_or_else(|| {
            io::Error::new(
                io::ErrorKind::Other,
                "no existing socket found; you must call `Node::start` first",
            )
        })
    }

    /// Registers the given node-level task.
    pub(crate) fn register_task(&self, handle: JoinHandle<()>) {
        self.tasks.lock().push(handle);
    }

    /// Registers the given connection-level task.
    pub(crate) fn register_conn_task(&self, conn_id: ConnId, handle: JoinHandle<()>) {
        if let Some(tasks) = self.conns.read().get(&conn_id).map(|c| c.tasks.clone()) {
            tasks.lock().push(handle);
        }
    }

    /// Returns the list of streams applicable to the given connection.
    pub(crate) fn get_streams(&self, conn_id: ConnId) -> Option<Arc<Streams>> {
        self.conns.read().get(&conn_id).map(|c| c.streams.clone())
    }

    /// Registers an inbound message from the given stream or a datagram from the connection.
    /// A missing `StreamId` indicates a datagram.
    pub(crate) fn register_msg_rx(
        &self,
        conn_id: ConnId,
        stream_id: Option<StreamId>,
        size: usize,
    ) {
        if let Some(stream_id) = stream_id {
            if let Some(stats) = self
                .get_streams(conn_id)
                .and_then(|streams| streams.read().get(&stream_id).map(|s| s.stats.clone()))
            {
                stats.register_msg_rx(size);
            }
        } else if let Some(stats) = self
            .conns
            .read()
            .get(&conn_id)
            .map(|c| c.datagram_stats.clone())
        {
            stats.register_msg_rx(size);
        }
    }

    /// Registers an outbound message to the given stream or a datagram to the connection.
    /// A missing `StreamId` indicates a datagram.
    pub(crate) fn register_msg_tx(
        &self,
        conn_id: ConnId,
        stream_id: Option<StreamId>,
        size: usize,
    ) {
        if let Some(stream_id) = stream_id {
            if let Some(stats) = self
                .get_streams(conn_id)
                .and_then(|streams| streams.read().get(&stream_id).map(|s| s.stats.clone()))
            {
                stats.register_msg_tx(size);
            }
        } else if let Some(stats) = self
            .conns
            .read()
            .get(&conn_id)
            .map(|c| c.datagram_stats.clone())
        {
            stats.register_msg_tx(size);
        }
    }
}

/// The configuration for the node.
#[derive(Debug, Clone)]
pub struct Config {
    /// Default client configuration (mandatory to allow outbound connections).
    pub(crate) client: Option<ClientConfig>,
    /// Server configuration (mandatory to allow inbound connections).
    pub(crate) server: Option<ServerConfig>,
}

impl Config {
    /// Creates a new node config object. The arguments can't be `None` at the same time
    /// (otherwise the node couldn't do anything).
    pub fn new(client: Option<ClientConfig>, server: Option<ServerConfig>) -> Self {
        if client.is_none() && server.is_none() {
            panic!("the node can't function without any config");
        }

        Self { client, server }
    }
}