harlequinn 0.1.1

A real-time networking library primarily aimed at games.
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
143
144

mod connection;
mod handshake;

pub use connection::ConnectionCommand;

use std::net::SocketAddr;

use {
    bytes::Bytes,
    crossbeam_channel::Sender,
    futures::StreamExt,
    quinn::{ClientConfigBuilder, Connection, Endpoint, EndpointDriver, Incoming},
    tokio::sync::{
        mpsc::{Receiver as TokioReceiver, Sender as TokioSender},
        oneshot::Sender as OneshotSender,
    },
};

use crate::{worker::connection::handle_connecting, Certificate, PeerId};

#[derive(Debug)]
pub enum WorkerCommand {
    Connect {
        server_addr: SocketAddr,
        server_name: String,
        certificate: Certificate,
        assigned_peer_id: PeerId,
    },
    Stop,
}

#[derive(Debug)]
pub enum WorkerEvent {
    /// A connection initiated by this endpoint has failed.
    ConnectionFailed {
        peer_id: PeerId,
    },
    /// The start of a new connection, negotiate a peer ID with the synchronous thread.
    ConnectionStarted {
        connection: Connection,
        /// The command channel through which the worker receives connection commands.
        connection_sender: TokioSender<ConnectionCommand>,
        /// Previously established peer ID if applicable.
        server_peer_id: Option<PeerId>,
        peer_id_sender: OneshotSender<PeerId>,
    },
    /// The connection handshake has succeeded, get the caller to confirm it.
    ConnectionRequested {
        peer_id: PeerId,
        socket_addr: SocketAddr,
        confirm_sender: OneshotSender<bool>,
    },
    Disconnected {
        peer_id: PeerId,
        reason: Option<String>,
    },
    ReceivedDatagram {
        peer_id: PeerId,
        bytes: Bytes,
    },
    ReceivedMessage {
        peer_id: PeerId,
        bytes: Bytes,
    },
    Stopped,
}

pub async fn driver_worker(driver: EndpointDriver, sender: Sender<WorkerEvent>) {
    let result = driver.await;

    sender.send(WorkerEvent::Stopped).unwrap();

    if let Err(error) = result {
        // TODO: Better error handling
        panic!("IO Error: {}", error);
    }
}

pub async fn network_worker(
    endpoint: Endpoint,
    incoming: Incoming,
    sender: Sender<WorkerEvent>,
    mut receiver: TokioReceiver<WorkerCommand>,
    protocol_checksum: u32,
    is_listening: bool,
) {
    // If we've got an open endpoint we're listening on, spawn the listen task
    if is_listening {
        tokio::spawn(listen_incoming(incoming, sender.clone(), protocol_checksum));
    }

    // Listen for commands
    while let Some(command) = receiver.recv().await {
        match command {
            WorkerCommand::Connect {
                server_addr,
                server_name,
                certificate,
                assigned_peer_id,
            } => {
                // Create the client config
                let mut config_builder = ClientConfigBuilder::default();
                config_builder
                    .add_certificate_authority(certificate)
                    .unwrap();
                let client_config = config_builder.build();

                // Start the connection attempt
                let connecting = endpoint
                    .connect_with(client_config, &server_addr, &server_name)
                    .unwrap();

                // Spawn a handler for this connection
                tokio::spawn(handle_connecting(
                    connecting,
                    sender.clone(),
                    Some(assigned_peer_id),
                    protocol_checksum,
                ));
            }
            WorkerCommand::Stop => {
                // The driver worker will send an event back when it's stopped
                endpoint.close(0u32.into(), b"Endpoint was dropped");
                return;
            }
        }
    }
}

async fn listen_incoming(
    mut incoming: Incoming,
    sender: Sender<WorkerEvent>,
    protocol_checksum: u32,
) {
    while let Some(connecting) = incoming.next().await {
        // Spawn a handler for this connection
        tokio::spawn(handle_connecting(
            connecting,
            sender.clone(),
            None,
            protocol_checksum,
        ));
    }
}