follow-up article on protocol design dealing with subtasks & connection passing

[b5]

Fantastic feedback on @matheus23's initial article on protocol design that we should incorporate into a follow up, YouTube tutorial, or blog post:

#322 (comment)

[ValorZard]

For context, here is how I designed my own “custom protocol” for my Godot game

use std::collections::HashMap;
use std::default;
use std::hash::Hash;
use std::str::FromStr;

use crate::player::{self, PlayerData};
use bincode::config::Configuration;
use bincode::{Decode, Encode, config};
use godot::classes::INode;
use godot::classes::Node;
use godot::prelude::*;
use iroh::Endpoint;
use iroh::NodeId;
use iroh::endpoint::Connection;
use iroh::endpoint::RecvStream;
use iroh::endpoint::SendStream;
use iroh_base::ticket::{NodeTicket, Ticket};
use tokio::task::JoinSet;

const BINCODE_CONFIG: Configuration = config::standard();

use crate::NodeIdEncode;
use crate::async_runtime::AsyncRuntime;

type NetMessage = Vec<u8>;

type ConnectionData = (tokio::sync::mpsc::UnboundedSender<NetMessage>, Connection);

enum ChannelMessage {
    Print(String),
    Ticket(NodeTicket),
    NodeId(NodeId),
    Connection(NodeId, ConnectionData),
    NetMessage(NetMessageType),
}

#[derive(Encode, Decode, PartialEq, Debug)]
enum NetMessageType {
    PlayerData(PlayerData),
    AddPlayers(Vec<(NodeIdEncode, PlayerData)>),
}

#[derive(GodotClass)]
#[class(base=Node)]
pub struct NetSingleton {
    base: Base<Node>,
    #[var]
    pub node_id: GString,
    #[var]
    pub is_host: bool,
    #[var]
    pub log: GString,
    // native rust stuff not exposed to godots
    sender: async_channel::Sender<ChannelMessage>,
    receiver: async_channel::Receiver<ChannelMessage>,
    connections: HashMap<NodeId, ConnectionData>,
    world: HashMap<NodeId, PlayerData>,
}

#[godot_api]
impl INode for NetSingleton {
    fn init(base: Base<Node>) -> Self {
        godot_print!("Start net singleton"); // Prints to the Godot console

        let (sender, receiver) = async_channel::unbounded();

        Self {
            base,
            node_id: GString::new(),
            is_host: false,
            log: GString::new(),
            sender,
            receiver,
            connections: HashMap::new(),
            world: HashMap::new(),
        }
    }

    fn ready(&mut self) {
        /*
        let sender = self.sender.clone();
        AsyncRuntime::spawn(async move {
            let mut count = 0;
            loop {
                tokio::time::sleep(tokio::time::Duration::from_secs(1)).await;
                count += 1;
                let message = format!("Hello from Rust! Count: {}", count);
                if let Err(_) = sender.send(ChannelMessage::Print(message)).await {
                    break;
                }
            }
        });
        */
    }

    fn process(&mut self, delta: f64) {
        while let Ok(message) = self.receiver.try_recv() {
            match message {
                ChannelMessage::Print(msg) => {
                    godot_print!("{}", msg);
                }
                ChannelMessage::Ticket(ticket) => {
                    godot_print!("Received ticket: {}", ticket);
                    self.signals()
                        .ticket_received()
                        .emit(ticket.to_string().into());
                }
                ChannelMessage::NodeId(node_id) => {
                    self.signals()
                        .node_id_assigned()
                        .emit(node_id.to_string().into());
                    self.node_id = node_id.to_string().into();
                    // add self to the world
                    self.world.insert(
                        node_id,
                        PlayerData {
                            node_id: *node_id.as_bytes(),
                            position: Vector2::new(0.0, 0.0).into(),
                        },
                    );
                }
                ChannelMessage::Connection(node_id, (conn_send, conn)) => {
                    godot_print!("Received connection from: {}", node_id);
                    self.connections.insert(node_id, (conn_send, conn));
                    self.signals()
                        .connection_received()
                        .emit(node_id.to_string().into());

                    if self.is_host {
                        // update new player with all players
                        let players_info = self
                            .world
                            .iter()
                            .map(|(node_id, player_data)| (*node_id.as_bytes(), *player_data))
                            .collect::<Vec<(NodeIdEncode, PlayerData)>>();
                        // send all players to all the players
                        for (_, (sender, _)) in &self.connections {
                            sender
                                .send(
                                    bincode::encode_to_vec(
                                        NetMessageType::AddPlayers(players_info.clone()),
                                        BINCODE_CONFIG,
                                    )
                                    .unwrap(),
                                )
                                .unwrap();
                        }
                    }
                }
                ChannelMessage::NetMessage(net_message) => {
                    // Decode the net message
                    match net_message {
                        NetMessageType::PlayerData(player_data) => {
                            // Handle player data
                            let node_id = NodeId::from_bytes(&player_data.node_id).unwrap();
                            self.world.insert(node_id, player_data);

                            // send data to all connections if host
                            if self.is_host {
                                // send all players to all the players
                                for (_, (sender, _)) in &self.connections {
                                    sender
                                        .send(
                                            bincode::encode_to_vec(
                                                NetMessageType::PlayerData(player_data),
                                                BINCODE_CONFIG,
                                            )
                                            .unwrap(),
                                        )
                                        .unwrap();
                                }
                            }

                        }
                        NetMessageType::AddPlayers(players_info) => {
                            // Handle adding players
                            for (node_id, player_data) in players_info {
                                let node_id = NodeId::from_bytes(&node_id).unwrap();
                                // Check if the player is already in the world
                                if !self.world.contains_key(&node_id) {
                                    // Add new player data
                                    godot_print!("Adding new player: {:?}", node_id);
                                    self.signals()
                                        .connection_received()
                                        .emit(node_id.to_string().into());
                                }
                                // update player data otherwise
                                self.world.insert(node_id, player_data);
                            }
                        }
                    }
                }
            }
        }

        if self.is_host {
            // if we're the host, we need to send player data to all connections
            for (node_id, (conn_send, _)) in &self.connections {
                conn_send
                    .send(
                        bincode::encode_to_vec(
                            NetMessageType::PlayerData(self.world.get(node_id).unwrap().clone()),
                            BINCODE_CONFIG,
                        )
                        .unwrap(),
                    )
                    .unwrap();
            }
        }

        let mut str = String::new();
        for (node_id, player_data) in &self.world {
            str.push_str(&format!("{}: {:?}\n", node_id, player_data));
        }
        self.log = str.into();
    }
}

/// Each protocol is identified by its ALPN string.
///
/// The ALPN, or application-layer protocol negotiation, is exchanged in the connection handshake,
/// and the connection is aborted unless both nodes pass the same bytestring.
const ALPN: &[u8] = b"iroh-example/echo/0";

async fn setup_connection(
    mut send_stream: SendStream,
    mut recv_stream: RecvStream,
    connection: Connection,
    sender: async_channel::Sender<ChannelMessage>,
) {
    // we're makign this unbounded in case we have a lot of messages to send
    let (conn_sender, mut conn_receiver) = tokio::sync::mpsc::unbounded_channel::<Vec<u8>>();

    // spawn task to handle receiving messages from the connection
    let channel_sender = sender.clone();

    tokio::spawn(async move {
        loop {
            let mut buf = [0; 1024];
            let n = recv_stream.read(&mut buf).await.unwrap();
            if n.is_some() {
                let buffer = &buf[..n.unwrap()];
                let (net_message, _) = bincode::decode_from_slice(buffer, BINCODE_CONFIG).unwrap();
                channel_sender
                    .send(ChannelMessage::NetMessage(net_message))
                    .await
                    .unwrap();
            }
        }
    });

    // spawn task to handle sending messages to the connection
    tokio::spawn(async move {
        loop {
            if let Some(msg) = conn_receiver.recv().await {
                // send the message to the connection
                send_stream.write_all(&msg).await.unwrap();
            }
        }
    });

    sender
        .send(ChannelMessage::Connection(
            connection.remote_node_id().unwrap(),
            (conn_sender, connection),
        ))
        .await
        .unwrap();
}

#[godot_api]
impl NetSingleton {
    // have to have a different name than the class name because of autoload workaround
    pub const SINGLETON: &'static str = "NetAutoload";

    #[signal]
    fn ticket_received(ticket: GString);

    // for getting out own node id
    #[signal]
    fn node_id_assigned(node_id: GString);

    #[signal]
    fn connection_received(node_id: GString);

    #[func]
    // have to gain access to the scene tree to grab the autoload singleton
    // this is a workaround for the autoload singleton since its not natively supported by godot rust
    // if the name of the autoload singleton is different from SINGLETON, bad things will happen
    pub fn get_autoload_singleton(scene_tree: Option<Gd<SceneTree>>) -> Option<Gd<Self>> {
        if let Some(scene_tree) = scene_tree {
            let singleton = scene_tree
                .get_root()
                .unwrap()
                .try_get_node_as::<Self>(Self::SINGLETON);
            return singleton;
        }
        None
    }

    #[func]
    pub fn connect_side_start(&mut self, ticket: GString) {
        self.is_host = false;
        let ticket = NodeTicket::deserialize(ticket.to_string().as_str()).unwrap();
        let sender = self.sender.clone();
        AsyncRuntime::spawn(async move {
            sender
                .send(ChannelMessage::Print("Accept side started".to_string()))
                .await
                .unwrap();
            let addr = ticket.node_addr();

            let endpoint = Endpoint::builder().discovery_n0().bind().await.unwrap();

            sender
                .send(ChannelMessage::NodeId(endpoint.node_id()))
                .await
                .unwrap();

            // Open a connection to the accepting node
            let conn = endpoint.connect(addr.clone(), ALPN).await.unwrap();

            // Open a bidirectional QUIC stream
            let (send_stream, recv_stream) = conn.open_bi().await.unwrap();

            setup_connection(send_stream, recv_stream, conn, sender).await;
        });
    }

    #[func]
    pub fn accept_side_start(&mut self) {
        self.is_host = true;
        let sender = self.sender.clone();
        AsyncRuntime::spawn(async move {
            let endpoint = Endpoint::builder()
                .discovery_n0()
                .alpns(vec![ALPN.to_vec()])
                .bind()
                .await
                .unwrap();

            sender
                .send(ChannelMessage::NodeId(endpoint.node_id()))
                .await
                .unwrap();

            let ep = endpoint.clone();

            let sender_clone = sender.clone();
            // spawn a task so that `accept_side` returns immediately and we can continue in main()
            tokio::spawn(async move {
                // put connections in a set so we don't accidentally leak them
                let mut connection_set = JoinSet::new();
                while let Some(incoming) = endpoint.accept().await {
                    // spawn a task for each incoming connection, so we can serve multiple connections asynchronously
                    // clone sender for each task
                    let sender = sender_clone.clone();
                    connection_set.spawn(async move {
                        let connection = incoming.await.unwrap();

                        // We can get the remote's node id from the connection.
                        let node_id = connection.remote_node_id().unwrap();
                        sender
                            .send(ChannelMessage::Print(format!(
                                "server has accepted connection from {node_id}"
                            )))
                            .await
                            .unwrap();

                        // Our protocol is a simple request-response protocol, so we expect the
                        // connecting peer to open a single bi-directional stream.
                        let (send_stream, recv_stream) = connection.accept_bi().await.unwrap();

                        setup_connection(send_stream, recv_stream, connection, sender).await;
                    });
                }
            });

            let ticket = NodeTicket::new(ep.node_addr().await.unwrap());
            sender.send(ChannelMessage::Ticket(ticket)).await.unwrap();
        });
    }

    #[func]
    pub fn send_player_data(&self, node_id: GString, position: Vector2) {
        let node_id = NodeId::from_str(node_id.to_string().as_str()).unwrap();
        let player_data = PlayerData {
            node_id: *node_id.as_bytes(),
            position: position.into(),
        };
        let net_message = NetMessageType::PlayerData(player_data);
        //godot_print!("{:?}", player_data);
        let message = bincode::encode_to_vec(net_message, BINCODE_CONFIG).unwrap();
        for (node_id, (conn_send, _)) in &self.connections {
            conn_send.send(message.clone()).unwrap();
        }
    }

    #[func]
    pub fn get_player_data(&self, node_id: GString) -> Option<Gd<PlayerData>> {
        if let Ok(node_id) = NodeId::from_str(node_id.to_string().as_str()) {
            let data = self.world.get(&node_id);
            if let Some(data) = data {
                // wrap the data in a GD object to use it in GDScript
                return Some(Gd::from_object(*data));
            }
        }
        None
    }
}

It’s defiantly not perfect but I think it works alright?