feat: upgrade to Bevy 0.17 and migrate to Messages/Observers API

Author: foxzool

Cargo.toml

@@ -23,7 +23,7 @@ inspect = ["bevy-inspector-egui"]
 
 [dependencies]
 async-std = { version = "1.12.0" }
-bevy = { version = "0.16.0", default-features = false, features = ["bevy_log"] }
+bevy = { version = "0.17.0", default-features = false, features = ["bevy_log"] }
 bytes = "1.6.0"
 kanal = "0.1.0-pre8"
 futures = "0.3.30"
@@ -32,12 +32,12 @@ thiserror = { version = "1.0.59" }
 
 bincode = { version = "1.3.3", optional = true }
 serde_json = { version = "1.0.68", optional = true }
-bevy-inspector-egui = { version = "0.25.0", optional = true }
+bevy-inspector-egui = { version = "0.26.0", optional = true }
 
 
 
 [dev-dependencies]
-bevy = { version = "0.16.0", default-features = false, features = [
+bevy = { version = "0.17.0", default-features = false, features = [
     "bevy_asset",
     "bevy_audio",
     "vorbis",

bevy_octopus_websocket/Cargo.toml

@@ -18,7 +18,7 @@ documentation = "https://docs.rs/bevy_octopus_websocket"
 
 [dependencies]
 bevy_octopus = { version = "*", path = ".." }
-bevy = { version = "0.16.0", default-features = false, features = [] }
+bevy = { version = "0.17.0", default-features = false, features = [] }
 async-std = "1.12.0"
 async-tungstenite = { version = "0.27.0", features = ["async-std-runtime", "async-tls"] }
 tungstenite = "0.23.0"

bevy_octopus_websocket/examples/common/lib.rs

@@ -41,12 +41,13 @@ pub struct PlayerInformation {
     pub position: (u32, u32, u32),
 }
 
-pub fn on_node_event(trigger: Trigger<NetworkEvent>) {
-    info!("{:?} trigger {:?}", trigger.target(), trigger.event());
+pub fn on_node_event(on: On<NodeEvent>) {
+    let e = on.event();
+    info!("{:?} trigger {:?}", e.entity, e.event);
 }
 
 pub fn handle_message_events(
-    mut ev_channels: EventReader<ReceiveChannelMessage<PlayerInformation>>,
+    mut ev_channels: MessageReader<ReceiveChannelMessage<PlayerInformation>>,
 ) {
     for event in ev_channels.read() {
         info!("{} Received: {:?}", event.channel_id, &event.message);
@@ -61,7 +62,9 @@ pub fn handle_raw_packet(q_server: Query<(&ChannelId, &NetworkNode)>) {
     }
 }
 
-pub fn send_json_message(mut channel_messages: EventWriter<SendChannelMessage<PlayerInformation>>) {
+pub fn send_json_message(
+    mut channel_messages: MessageWriter<SendChannelMessage<PlayerInformation>>,
+) {
     channel_messages.write(SendChannelMessage {
         channel_id: JSON_CHANNEL,
         message: PlayerInformation {
@@ -73,7 +76,7 @@ pub fn send_json_message(mut channel_messages: EventWriter<SendChannelMessage<Pl
 
 /// send bincode message
 pub fn send_bincode_message(
-    mut channel_messages: EventWriter<SendChannelMessage<PlayerInformation>>,
+    mut channel_messages: MessageWriter<SendChannelMessage<PlayerInformation>>,
 ) {
     channel_messages.write(SendChannelMessage {
         channel_id: BINCODE_CHANNEL,

bevy_octopus_websocket/examples/server.rs

@@ -42,7 +42,7 @@ fn setup_server(mut commands: Commands) {
 }
 
 /// broadcast message to all connected clients in channel
-fn send_channel_packet(mut channel_events: EventWriter<ChannelPacket>) {
+fn send_channel_packet(mut channel_events: MessageWriter<ChannelPacket>) {
     channel_events.write(ChannelPacket::new(RAW_CHANNEL, b"channel 1 message\r\n"));
 }
 

bevy_octopus_websocket/src/lib.rs

@@ -85,10 +85,11 @@ impl WebsocketAddress {
 }
 
 fn on_start_server(
-    trigger: Trigger<StartServer>,
+    on: On<StartServer>,
     q_ws_server: Query<(&NetworkNode, &ServerNode<WebsocketAddress>)>,
 ) {
-    if let Ok((net_node, server_node)) = q_ws_server.get(trigger.target()) {
+    let ev = on.event();
+    if let Ok((net_node, server_node)) = q_ws_server.get(ev.entity) {
         let local_addr = server_node.url.parse().expect("Invalid address");
         let event_tx = net_node.event_channel.sender.clone_async();
         let shutdown_clone = net_node.shutdown_channel.receiver.clone_async();
@@ -116,10 +117,11 @@ fn on_start_server(
 
 #[allow(clippy::type_complexity)]
 fn on_start_client(
-    trigger: Trigger<StartClient>,
+    on: On<StartClient>,
     q_ws_client: Query<(&NetworkNode, &ClientNode<WebsocketAddress>), Without<NetworkPeer>>,
 ) {
-    if let Ok((net_node, remote_addr)) = q_ws_client.get(trigger.target()) {
+    let ev = on.event();
+    if let Ok((net_node, remote_addr)) = q_ws_client.get(ev.entity) {
         let url = remote_addr.url.clone();
         debug!("try connect to {}", url);
         let recv_tx = net_node.recv_message_channel.sender.clone_async();
@@ -319,7 +321,7 @@ fn handle_endpoint(
 
             // Add the client to the server's children
             commands.entity(entity).add_child(child_ws_client);
-            commands.trigger_targets(NetworkEvent::Connected, vec![child_ws_client]);
+            commands.trigger(NodeEvent { entity: child_ws_client, event: NetworkEvent::Connected });
         }
     }
 }

examples/common/lib.rs

@@ -52,12 +52,13 @@ pub struct PlayerInformation {
     pub position: (u32, u32, u32),
 }
 
-pub fn on_node_event(trigger: Trigger<NetworkEvent>) {
-    info!("{:?} trigger {:?}", trigger.target(), trigger.event());
+pub fn on_node_event(on: On<NodeEvent>) {
+    let e = on.event();
+    info!("{:?} trigger {:?}", e.entity, e.event);
 }
 
 pub fn handle_message_events(
-    mut ev_channels: EventReader<ReceiveChannelMessage<PlayerInformation>>,
+    mut ev_channels: MessageReader<ReceiveChannelMessage<PlayerInformation>>,
 ) {
     for event in ev_channels.read() {
         info!("{} Received: {:?}", event.channel_id, &event.message);
@@ -75,7 +76,9 @@ pub fn handle_raw_packet(q_server: Query<(Entity, &ChannelId, &NetworkNode)>) {
     }
 }
 
-pub fn send_json_message(mut channel_messages: EventWriter<SendChannelMessage<PlayerInformation>>) {
+pub fn send_json_message(
+    mut channel_messages: MessageWriter<SendChannelMessage<PlayerInformation>>,
+) {
     channel_messages.write(SendChannelMessage {
         channel_id: JSON_CHANNEL,
         message: PlayerInformation {
@@ -87,7 +90,7 @@ pub fn send_json_message(mut channel_messages: EventWriter<SendChannelMessage<Pl
 
 /// send bincode message
 pub fn send_bincode_message(
-    mut channel_messages: EventWriter<SendChannelMessage<PlayerInformation>>,
+    mut channel_messages: MessageWriter<SendChannelMessage<PlayerInformation>>,
 ) {
     channel_messages.write(SendChannelMessage {
         channel_id: BINCODE_CHANNEL,

examples/tcp/server.rs

@@ -43,7 +43,7 @@ fn setup_server(mut commands: Commands) {
 }
 
 /// broadcast message to all connected clients in channel
-fn send_channel_packet(mut channel_events: EventWriter<ChannelPacket>) {
+fn send_channel_packet(mut channel_events: MessageWriter<ChannelPacket>) {
     channel_events.write(ChannelPacket::new(RAW_CHANNEL, b"channel 1 message\r\n"));
 }
 

src/channels.rs

@@ -2,7 +2,7 @@ use std::fmt::Display;
 
 use bevy::{
     ecs::reflect::ReflectComponent,
-    prelude::{Component, Event, EventReader, Query, Reflect},
+    prelude::{Component, Message, MessageReader, Query, Reflect},
 };
 use bytes::Bytes;
 
@@ -19,7 +19,7 @@ impl Display for ChannelId {
     }
 }
 
-#[derive(Event)]
+#[derive(Message)]
 pub struct ChannelPacket {
     pub channel_id: ChannelId,
     pub bytes: Bytes,
@@ -36,7 +36,7 @@ impl ChannelPacket {
     }
 }
 
-#[derive(Event, Debug)]
+#[derive(Message, Debug)]
 pub struct SendChannelMessage<M> {
     pub channel_id: ChannelId,
     pub message: M,
@@ -51,7 +51,7 @@ impl<M> SendChannelMessage<M> {
     }
 }
 
-#[derive(Event, Debug)]
+#[derive(Message, Debug)]
 pub struct ReceiveChannelMessage<M> {
     pub channel_id: ChannelId,
     pub message: M,
@@ -68,7 +68,7 @@ impl<M> ReceiveChannelMessage<M> {
 
 pub(crate) fn send_channel_message_system(
     q_net: Query<(&ChannelId, &NetworkNode)>,
-    mut channel_events: EventReader<ChannelPacket>,
+    mut channel_events: MessageReader<ChannelPacket>,
 ) {
     for channel_ev in channel_events.read() {
         q_net.par_iter().for_each(|(channel_id, net_node)| {

src/client.rs

@@ -1,15 +1,11 @@
 use crate::{
     error::NetworkError,
-    network_node::{NetworkAddress, NetworkEvent, NetworkPeer},
-};
-use bevy::{
-    ecs::component::{ComponentHooks, HookContext, Immutable, StorageType},
-    prelude::*,
+    network_node::{NetworkAddress, NetworkEvent, NetworkPeer, NodeEvent},
 };
+use bevy::{ecs::component::{Immutable, StorageType}, prelude::*};
 
 pub(super) fn plugin(app: &mut App) {
     app.register_type::<ReconnectSetting>()
-        .add_event::<StartClient>()
         .add_systems(Update, handle_reconnect_timer)
         .add_observer(cleanup_client_session)
         .add_observer(client_reconnect);
@@ -25,15 +21,18 @@ impl<T: NetworkAddress + 'static> Component for ClientNode<T> {
     const STORAGE_TYPE: StorageType = StorageType::Table;
     type Mutability = Immutable;
 
-    fn register_component_hooks(hooks: &mut ComponentHooks) {
-        hooks.on_insert(|mut world, HookContext { entity, .. }| {
+    fn on_insert() -> Option<bevy::ecs::lifecycle::ComponentHook> {
+        Some(|mut world, ctx| {
+            let entity = ctx.entity;
             world.commands().entity(entity).insert(ClientTag);
-            world.trigger_targets(StartClient, entity);
-        });
+            world.trigger(StartClient { entity });
+        })
     }
 }
-#[derive(Event, Clone)]
-pub struct StartClient;
+#[derive(EntityEvent, Clone)]
+pub struct StartClient {
+    pub entity: Entity,
+}
 
 #[derive(Debug, Component, Reflect)]
 #[reflect(Component)]
@@ -55,12 +54,13 @@ impl Default for ReconnectSetting {
 }
 
 pub(crate) fn client_reconnect(
-    trigger: Trigger<NetworkEvent>,
+    on: On<NodeEvent>,
     mut commands: Commands,
     mut q_net: Query<&mut ReconnectSetting, Without<NetworkPeer>>,
 ) {
-    if let Ok(mut reconnect) = q_net.get_mut(trigger.target()) {
-        let event = trigger.event();
+    let ev = on.event();
+    if let Ok(mut reconnect) = q_net.get_mut(ev.entity) {
+        let event = &ev.event;
         if reconnect.retries < reconnect.max_retries {
             reconnect.retries += 1;
         } else {
@@ -70,7 +70,7 @@ pub(crate) fn client_reconnect(
             NetworkEvent::Listen | NetworkEvent::Connected => reconnect.retries = 0,
             NetworkEvent::Disconnected | NetworkEvent::Error(NetworkError::Connection(_)) => {
                 commands
-                    .entity(trigger.target())
+                    .entity(ev.entity)
                     .insert(ReconnectTimer(Timer::from_seconds(
                         reconnect.delay,
                         TimerMode::Once,
@@ -92,18 +92,19 @@ pub(crate) fn handle_reconnect_timer(
     for (entity, mut timer) in q_reconnect.iter_mut() {
         if timer.tick(time.delta()).just_finished() {
             commands.entity(entity).remove::<ReconnectTimer>();
-            commands.trigger_targets(StartClient, entity);
+            commands.trigger(StartClient { entity });
         }
     }
 }
 
 pub(crate) fn cleanup_client_session(
-    trigger: Trigger<NetworkEvent>,
+    on: On<NodeEvent>,
     mut commands: Commands,
     q_net: Query<Entity, With<NetworkPeer>>,
 ) {
-    if let Ok(entity) = q_net.get(trigger.target()) {
-        let event = trigger.event();
+    let ev = on.event();
+    if let Ok(entity) = q_net.get(ev.entity) {
+        let event = &ev.event;
 
         match event {
             NetworkEvent::Disconnected | NetworkEvent::Error(NetworkError::Connection(_)) => {

src/network_node.rs

@@ -1,8 +1,5 @@
 use crate::{client::ReconnectSetting, error::NetworkError, prelude::ChannelId};
-use bevy::{
-    ecs::component::{ComponentHooks, HookContext, Mutable, StorageType},
-    prelude::*,
-};
+use bevy::{ecs::component::{Mutable, StorageType}, prelude::*};
 use bytes::Bytes;
 use kanal::{Receiver, Sender, unbounded};
 use std::{
@@ -73,12 +70,13 @@ impl Component for NetworkNode {
     const STORAGE_TYPE: StorageType = StorageType::Table;
 
     type Mutability = Mutable;
-    fn register_component_hooks(hooks: &mut ComponentHooks) {
-        hooks.on_remove(|world, HookContext { entity, .. }| {
-            if let Some(node) = world.get::<NetworkNode>(entity) {
+
+    fn on_remove() -> Option<bevy::ecs::lifecycle::ComponentHook> {
+        Some(|world, ctx| {
+            if let Some(node) = world.get::<NetworkNode>(ctx.entity) {
                 node.shutdown_channel.sender.try_send(()).unwrap();
             }
-        });
+        })
     }
 }
 
@@ -142,15 +140,22 @@ impl<T> AsyncChannel<T> {
     }
 }
 
-#[derive(Debug, Event)]
-/// A network event originating from a network node
+#[derive(Debug)]
+/// 来自网络节点后台的原始事件（线程通道）
 pub enum NetworkEvent {
     Listen,
     Connected,
     Disconnected,
     Error(NetworkError),
 }
 
+#[derive(EntityEvent, Debug)]
+/// ECS 观察者用的实体事件，携带目标实体
+pub struct NodeEvent {
+    pub entity: Entity,
+    pub event: NetworkEvent,
+}
+
 /// send network node error channel to events
 #[allow(clippy::type_complexity)]
 pub(crate) fn network_node_event(
@@ -167,7 +172,7 @@ pub(crate) fn network_node_event(
                     net_node.stop();
                 }
             }
-            commands.trigger_targets(event, vec![entity]);
+            commands.trigger(NodeEvent { entity, event });
         }
     }
 }