Proposal: Improve usability of bevy_matchbox with NetworkReader/NetworkWriters

[simbleau]

I think what's really missing from bevy_matchbox is something akin to an EventWriter and EventReader for Matchbox messages.

This proposal outlines how I would implement this for a known socket configuration, with 1 unreliable and 1 reliable channel. It would maybe be a little messier with generics, but follow my lead here, if you will:

Take an example payload:

#[derive(Serialize, Deserialize, Debug, Clone)]
pub enum ChatPayload {
    System { message: String },
    Chat { from: String, message: String },
}

An pair that with an API purposefully similar to EventReaders:

Client Receiver

pub fn my_system(
    mut chat_read: ClientReceiver<ChatPayload>,
) {
    for payload in chat_read.iter() {
        match payload {
            ChatPayload::Chat { from, message } => {
                chat_state.messages.push(format!("[{username}] said: {message}"));
            }
            ChatPayload::System { message } => {
                chat_state.messages.push(format!("SYSTEM: {message}"))
            }
        }
    }
}

FullMesh Receiver

pub fn my_system(
    mut chat_read: FullMeshReceiver<ChatPayload>,
) {
    for (peer_id, payload) in chat_read.iter() {
        match payload {
            ChatPayload::Chat { from, message } => {
                chat_state.messages.push(format!("[{username}/{peer_id:?}] said: {message}"));
            }
            ChatPayload::System { message } => {
                chat_state.messages.push(format!("SYSTEM: {message}"))
            }
        }
    }
}

(Server receiver not done for brevity)

(and Sending would be equally simple for the user)

pub fn broadcast(
    mut chat_send: ClientSender<ChatPayload>,
) {
    chat_send.reliable_to_host(ChatPayload::System { message: "System message!".to_string() }
}

---

Essentially we would need to perform the network read on CoreSet::First, load the buffer of messages, and then clear the buffers on CoreSet::Flush/Last. Do this for both sending and receiving.

Such an integration would be very flexible, since it would play with Bevy's systems quite well and the user's own custom systems, e.g.

struct ChatPlugin;
impl Plugin for ChatPlugin {
    fn build(&self, app: &mut App) {
        app.add_system(
            my_system
                .in_base_set(MyStage::NetworkRead)
                .in_schedule(MySchedule),
        )
        .add_network_message::<ChatPayload>();
    }
}

The key here is we would need something like .add_network_message<T>(), which would add the listeners and senders:

#[derive(Default, Debug, Resource)]
pub struct IncomingMatchboxQueue<M: Message> {
    pub messages: Vec<M>,
}
#[derive(Default, Debug, Resource)]
pub struct OutgoingMatchboxQueue<M: Message> {
    pub messages: Vec<M>,
}

pub trait AddNetworkMessageExt {
    fn add_network_message<M: Message>(&mut self) -> &mut Self;
}

impl AddNetworkMessageExt for App {
    fn add_network_message<M>(&mut self) -> &mut Self
    where
        M: Message,
    {
        if self.world.contains_resource::<IncomingMatchboxQueue<M>>()
            || self.world.contains_resource::<OutgoingMatchboxQueue<M>>()
        {
            panic!();
        }
        self.insert_resource(IncomingMatchboxQueue::<M> { messages: vec![] })
            .add_system(
                IncomingMessages::<M>::flush
                    .in_base_set(CoreSet::Flush),
            )
            .add_system(
                IncomingMessages::<M>::read
                    .in_base_set(CoreSet::First)),
            )
            .insert_resource(OutgoingMessages::<M> {
                reliable_to_host: vec![],
                unreliable_to_host: vec![],
            })
            .add_system(
                OutgoingMessages::<M>::write_system
                    .in_base_set(CoreSet::Last),
            );
        self
    }
}

The proposal hinges mostly on a shared trait, Message, which can de-serialize into a shared packet type like MatchboxPacket, as such:

struct MatchboxPacket<M: Message> {
    msg_id: u16,
    data: M
}

pub trait Message:
    Debug + Clone + Send + Sync + for<'a> Deserialize<'a> + Serialize + 'static
{
    fn id() -> u16;

    fn from_packet(packet: &Packet) -> Option<Self> {
        bincode::deserialize::<MatchboxPacket<Self>>(packet)
            .ok()
            .filter(|mb_packet| mb_packet.msg_id == Self::id())
            .map(|mb_packet| mb_packet.data)
    }

    fn to_packet(&self) -> Packet {
        let mb_packet = MatchboxPacket {
            msg_id: Self::id(),
            data: self.clone(),
        };
        bincode::serialize(&mb_packet).unwrap().into_boxed_slice()
    }
}

And to prevent against potential de-serialization collisions (an f32 can be interpreted as a u32, for example), Message needs an id() -> u16, but that can be easily derived with a hashing macro, based on the name of the Payload:

#[proc_macro_derive(MatchboxPayload)]
pub fn derive_payload_fn(item: TokenStream) -> TokenStream {
    let DeriveInput { ident, .. } = parse_macro_input!(item);
    let name = ident.to_token_stream();
    let mut s = DefaultHasher::new();
    name.to_string().hash(&mut s);
    let id = s.finish() as u16;
    quote! {
        impl ::bevy_matchbox::Message for #name {
            fn id() -> u16 {
                #id
            }

        }
    }
    .into()
}