Add broadcast messages and events (#684)

Co-authored-by: Hennadii Chernyshchyk <genaloner@gmail.com>

Author: Morb0

CHANGELOG.md

@@ -9,6 +9,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- Broadcast messages and events via `BroadcastMessageAppExt` and `BroadcastEventAppExt`. Emitted as `Broadcast<M>` on both the sender and the receiver to allow shared logic for client-side prediction.
 - `iter_received` and `iter_sent` methods on `ClientMessages` and `ServerMessages` to inspect inbound and outbound messages on a channel without consuming them.
 
 ### Changed

src/client/message.rs

@@ -51,6 +51,37 @@ impl Plugin for ClientMessagePlugin {
             .build_state(app.world_mut())
             .build_system(send);
 
+        let broadcast_send_fn = (
+            FilteredResourcesMutParamBuilder::new(|builder| {
+                for message in registry.iter_all_broadcast() {
+                    builder.add_write_by_id(message.messages_id());
+                }
+            }),
+            FilteredResourcesMutParamBuilder::new(|builder| {
+                for message in registry.iter_all_broadcast() {
+                    builder.add_write_by_id(message.broadcast_id());
+                }
+            }),
+            ParamBuilder,
+            ParamBuilder,
+            ParamBuilder,
+            ParamBuilder,
+        )
+            .build_state(app.world_mut())
+            .build_system(broadcast_send);
+
+        let broadcast_trigger_fn = (
+            FilteredResourcesMutParamBuilder::new(|builder| {
+                for event in registry.iter_broadcast_events() {
+                    builder.add_write_by_id(event.message().broadcast_id());
+                }
+            }),
+            ParamBuilder,
+            ParamBuilder,
+        )
+            .build_state(app.world_mut())
+            .build_system(broadcast_trigger);
+
         let receive_builder = (
             FilteredResourcesMutParamBuilder::new(|builder| {
                 for message in registry.iter_all_server() {
@@ -113,6 +144,22 @@ impl Plugin for ClientMessagePlugin {
             .build_state(app.world_mut())
             .build_system(send_locally);
 
+        let broadcast_send_locally_fn = (
+            FilteredResourcesMutParamBuilder::new(|builder| {
+                for message in registry.iter_all_broadcast() {
+                    builder.add_write_by_id(message.broadcast_id());
+                }
+            }),
+            FilteredResourcesMutParamBuilder::new(|builder| {
+                for message in registry.iter_all_broadcast() {
+                    builder.add_write_by_id(message.messages_id());
+                }
+            }),
+            ParamBuilder,
+        )
+            .build_state(app.world_mut())
+            .build_system(broadcast_send_locally);
+
         let reset_fn = (
             FilteredResourcesMutParamBuilder::new(|builder| {
                 for message in registry.iter_all_client() {
@@ -124,6 +171,11 @@ impl Plugin for ClientMessagePlugin {
                     builder.add_write_by_id(message.queue_id());
                 }
             }),
+            FilteredResourcesMutParamBuilder::new(|builder| {
+                for message in registry.iter_all_broadcast() {
+                    builder.add_write_by_id(message.messages_id());
+                }
+            }),
             ParamBuilder,
         )
             .build_state(app.world_mut())
@@ -154,8 +206,10 @@ impl Plugin for ClientMessagePlugin {
             .add_systems(
                 PostUpdate,
                 (
-                    send_fn.run_if(in_state(ClientState::Connected)),
-                    send_locally_fn.run_if(in_state(ClientState::Disconnected)),
+                    (send_fn, broadcast_send_fn).run_if(in_state(ClientState::Connected)),
+                    (send_locally_fn, broadcast_send_locally_fn)
+                        .run_if(in_state(ClientState::Disconnected)),
+                    broadcast_trigger_fn,
                 )
                     .chain()
                     .in_set(ClientSystems::Send),
@@ -197,6 +251,40 @@ fn send(
     }
 }
 
+fn broadcast_send(
+    mut messages: FilteredResourcesMut,
+    mut broadcasts: FilteredResourcesMut,
+    mut client_messages: ResMut<ClientMessages>,
+    type_registry: Res<AppTypeRegistry>,
+    entity_map: Res<ServerEntityMap>,
+    registry: Res<RemoteMessageRegistry>,
+) {
+    let mut ctx = ClientSendCtx {
+        entity_map: &entity_map,
+        type_registry: &type_registry,
+        invalid_entities: Vec::new(),
+    };
+
+    for message in registry.iter_all_broadcast() {
+        let messages = messages
+            .get_mut_by_id(message.messages_id())
+            .expect("messages resource should be accessible");
+        let broadcasts = broadcasts
+            .get_mut_by_id(message.broadcast_id())
+            .expect("broadcast messages resource should be accessible");
+
+        // SAFETY: passed pointers were obtained using this message data.
+        unsafe {
+            message.send(
+                &mut ctx,
+                messages.into_inner(),
+                broadcasts.into_inner(),
+                &mut client_messages,
+            );
+        }
+    }
+}
+
 fn receive(
     mut messages: FilteredResourcesMut,
     mut queues: FilteredResourcesMut,
@@ -246,6 +334,20 @@ fn trigger(
     }
 }
 
+fn broadcast_trigger(
+    mut broadcasts: FilteredResourcesMut,
+    mut commands: Commands,
+    registry: Res<RemoteMessageRegistry>,
+) {
+    for event in registry.iter_broadcast_events() {
+        let broadcasts = broadcasts
+            .get_mut_by_id(event.message().broadcast_id())
+            .expect("broadcast messages resource should be accessible");
+        // SAFETY: passed pointer was obtained using this event data.
+        unsafe { event.trigger(&mut commands, broadcasts.into_inner()) };
+    }
+}
+
 fn send_locally(
     mut from_messages: FilteredResourcesMut,
     mut messages: FilteredResourcesMut,
@@ -264,9 +366,28 @@ fn send_locally(
     }
 }
 
+fn broadcast_send_locally(
+    mut broadcasts: FilteredResourcesMut,
+    mut messages: FilteredResourcesMut,
+    registry: Res<RemoteMessageRegistry>,
+) {
+    for message in registry.iter_all_broadcast() {
+        let broadcasts = broadcasts
+            .get_mut_by_id(message.broadcast_id())
+            .expect("broadcast messages resource should be accessible");
+        let messages = messages
+            .get_mut_by_id(message.messages_id())
+            .expect("messages resource should be accessible");
+
+        // SAFETY: passed pointers were obtained using this message data.
+        unsafe { message.send_locally(broadcasts.into_inner(), messages.into_inner()) };
+    }
+}
+
 fn reset(
     mut messages: FilteredResourcesMut,
     mut queues: FilteredResourcesMut,
+    mut broadcast_messages: FilteredResourcesMut,
     registry: Res<RemoteMessageRegistry>,
 ) {
     for message in registry.iter_all_client() {
@@ -286,4 +407,13 @@ fn reset(
         // SAFETY: passed pointer was obtained using this message data.
         unsafe { messages.reset(queue.into_inner()) };
     }
+
+    for message in registry.iter_all_broadcast() {
+        let messages = broadcast_messages
+            .get_mut_by_id(message.messages_id())
+            .expect("broadcast messages resource should be accessible");
+
+        // SAFETY: passed pointer was obtained using this message data.
+        unsafe { message.reset(messages.into_inner()) };
+    }
 }

src/lib.rs

@@ -352,6 +352,32 @@ fn receive(ping: On<FromClient<Ping>>) {
 For events with entities inside use [`ClientEventAppExt::add_mapped_client_event`].
 Similar to messages, serialization can also be customized with [`ClientEventAppExt::add_client_event_with`].
 
+If you need to react to the same message on both the client and the server (for example,
+to share logic between client-side prediction and authoritative server processing), use
+[`BroadcastMessageAppExt::add_broadcast_message`] or
+[`BroadcastEventAppExt::add_broadcast_event`]. The message is emitted as [`Broadcast<M>`]
+on both sides, with the [`Broadcaster`] field indicating its origin.
+
+```
+# use bevy::{prelude::*, state::app::StatesPlugin};
+# use bevy_replicon::prelude::*;
+# use serde::{Deserialize, Serialize};
+# let mut app = App::new();
+# app.add_plugins((StatesPlugin, RepliconPlugins));
+app.add_broadcast_event::<Attack>(Channel::Ordered)
+    .add_observer(on_attack);
+
+fn on_attack(on: On<Broadcast<Attack>>) {
+    info!("attack from `{:?}`", on.broadcaster);
+}
+
+#[derive(Event, Deserialize, Serialize)]
+struct Attack;
+```
+
+Similar to regular client messages, we also provide [`BroadcastEventAppExt::add_mapped_broadcast_event`].
+and [`BroadcastEventAppExt::add_broadcast_event_with`].
+
 ### From server to client
 
 A similar technique is used to send messages from server to clients. To do this,
@@ -705,6 +731,8 @@ pub mod prelude {
             },
             client_id::ClientId,
             message::{
+                broadcast_event::{BroadcastEventAppExt, BroadcastTriggerExt},
+                broadcast_message::{Broadcast, BroadcastMessageAppExt, Broadcaster},
                 client_event::{ClientEventAppExt, ClientTriggerExt},
                 client_message::{ClientMessageAppExt, FromClient},
                 server_event::{ServerEventAppExt, ServerTriggerExt},

src/server/message.rs

@@ -61,6 +61,19 @@ impl Plugin for ServerMessagePlugin {
             .build_state(app.world_mut())
             .build_system(receive);
 
+        let broadcast_receive_fn = (
+            FilteredResourcesMutParamBuilder::new(|builder| {
+                for message in registry.iter_all_broadcast() {
+                    builder.add_write_by_id(message.broadcast_id());
+                }
+            }),
+            ParamBuilder,
+            ParamBuilder,
+            ParamBuilder,
+        )
+            .build_state(app.world_mut())
+            .build_system(broadcast_receive);
+
         let trigger_fn = (
             FilteredResourcesMutParamBuilder::new(|builder| {
                 for event in registry.iter_client_events() {
@@ -73,6 +86,18 @@ impl Plugin for ServerMessagePlugin {
             .build_state(app.world_mut())
             .build_system(trigger);
 
+        let broadcast_trigger_fn = (
+            FilteredResourcesMutParamBuilder::new(|builder| {
+                for event in registry.iter_broadcast_events() {
+                    builder.add_write_by_id(event.message().broadcast_id());
+                }
+            }),
+            ParamBuilder,
+            ParamBuilder,
+        )
+            .build_state(app.world_mut())
+            .build_system(broadcast_trigger);
+
         let send_locally_fn = (
             FilteredResourcesMutParamBuilder::new(|builder| {
                 for message in registry.iter_all_server() {
@@ -93,8 +118,11 @@ impl Plugin for ServerMessagePlugin {
             .add_systems(
                 PreUpdate,
                 (
-                    receive_fn.run_if(in_state(ServerState::Running)),
-                    trigger_fn.run_if(in_state(ClientState::Disconnected)),
+                    (receive_fn, broadcast_receive_fn).run_if(in_state(ServerState::Running)),
+                    (
+                        trigger_fn.run_if(in_state(ClientState::Disconnected)),
+                        broadcast_trigger_fn,
+                    ),
                 )
                     .chain()
                     .in_set(ServerSystems::Receive),
@@ -176,6 +204,26 @@ fn receive(
     }
 }
 
+fn broadcast_receive(
+    mut broadcasts: FilteredResourcesMut,
+    mut server_messages: ResMut<ServerMessages>,
+    type_registry: Res<AppTypeRegistry>,
+    message_registry: Res<RemoteMessageRegistry>,
+) {
+    let mut ctx = ServerReceiveCtx {
+        type_registry: &type_registry,
+    };
+
+    for message in message_registry.iter_all_broadcast() {
+        let broadcasts = broadcasts
+            .get_mut_by_id(message.broadcast_id())
+            .expect("broadcast messages resource should be accessible");
+
+        // SAFETY: passed pointer was obtained using this message data.
+        unsafe { message.receive(&mut ctx, broadcasts.into_inner(), &mut server_messages) };
+    }
+}
+
 fn trigger(
     mut from_messages: FilteredResourcesMut,
     mut commands: Commands,
@@ -190,6 +238,20 @@ fn trigger(
     }
 }
 
+fn broadcast_trigger(
+    mut broadcasts: FilteredResourcesMut,
+    mut commands: Commands,
+    registry: Res<RemoteMessageRegistry>,
+) {
+    for event in registry.iter_broadcast_events() {
+        let broadcasts = broadcasts
+            .get_mut_by_id(event.message().broadcast_id())
+            .expect("broadcast messages resource should be accessible");
+        // SAFETY: passed pointer was obtained using this event data.
+        unsafe { event.trigger(&mut commands, broadcasts.into_inner()) };
+    }
+}
+
 fn send_locally(
     mut to_messages: FilteredResourcesMut,
     mut messages: FilteredResourcesMut,

src/shared/message.rs

@@ -1,3 +1,5 @@
+pub mod broadcast_event;
+pub mod broadcast_message;
 pub mod client_event;
 pub mod client_message;
 pub mod ctx;