Add PacketEvent (#75)

* add PacketEvent

* docs and fixes

* Event::Packet works

azalea-client/src/events.rs

@@ -7,7 +7,7 @@ use azalea_ecs::{
     app::{App, Plugin},
     component::Component,
     event::EventReader,
-    query::{Added, Changed},
+    query::Added,
     system::Query,
     AppTickExt,
 };
@@ -21,7 +21,7 @@ use tokio::sync::mpsc;
 use crate::{
     chat::{ChatPacket, ChatReceivedEvent},
     packet_handling::{
-        AddPlayerEvent, DeathEvent, KeepAliveEvent, PacketReceiver, RemovePlayerEvent,
+        AddPlayerEvent, DeathEvent, KeepAliveEvent, PacketEvent, RemovePlayerEvent,
         UpdatePlayerEvent,
     },
     PlayerInfo,
@@ -62,6 +62,23 @@ pub enum Event {
     Chat(ChatPacket),
     /// Happens 20 times per second, but only when the world is loaded.
     Tick,
+    /// We received a packet from the server.
+    ///
+    /// ```
+    /// # use azalea_client::Event;
+    /// # use azalea_protocol::packets::game::ClientboundGamePacket;
+    /// # async fn example(event: Event) {
+    /// # match event {
+    /// Event::Packet(packet) => match *packet {
+    ///     ClientboundGamePacket::Login(_) => {
+    ///         println!("login packet");
+    ///     }
+    ///     _ => {}
+    /// },
+    /// # _ => {}
+    /// # }
+    /// # }
+    /// ```
     Packet(Arc<ClientboundGamePacket>),
     /// A player joined the game (or more specifically, was added to the tab
     /// list).
@@ -133,15 +150,16 @@ fn tick_listener(query: Query<&LocalPlayerEvents>) {
     }
 }
 
-fn packet_listener(query: Query<(&LocalPlayerEvents, &PacketReceiver), Changed<PacketReceiver>>) {
+fn packet_listener(query: Query<&LocalPlayerEvents>, mut events: EventReader<PacketEvent>) {
-    for (local_player_events, packet_receiver) in &query {
+    for event in events.iter() {
-        for packet in packet_receiver.packets.lock().iter() {
+        let local_player_events = query
+            .get(event.entity)
+            .expect("Non-localplayer entities shouldn't be able to receive add player events");
         local_player_events
-                .send(Event::Packet(packet.clone().into()))
+            .send(Event::Packet(Arc::new(event.packet.clone())))
             .unwrap();
     }
 }
-}
 
 fn add_player_listener(query: Query<&LocalPlayerEvents>, mut events: EventReader<AddPlayerEvent>) {
     for event in events.iter() {

azalea-client/src/packet_handling.rs

@@ -6,8 +6,8 @@ use azalea_ecs::{
     component::Component,
     ecs::Ecs,
     entity::Entity,
-    event::EventWriter,
+    event::{EventReader, EventWriter, Events},
-    query::Changed,
+    schedule::{StageLabel, SystemStage},
     system::{Commands, Query, ResMut, SystemState},
 };
 use azalea_protocol::{
@@ -42,11 +42,49 @@ use crate::{
     ClientInformation, PlayerInfo,
 };
 
+/// An event that's sent when we receive a packet.
+/// ```
+/// # use azalea_client::packet_handling::PacketEvent;
+/// # use azalea_protocol::packets::game::ClientboundGamePacket;
+/// # use azalea_ecs::event::EventReader;
+///
+/// fn handle_packets(mut events: EventReader<PacketEvent>) {
+///     for PacketEvent {
+///         entity,
+///         packet,
+///     } in events.iter() {
+///         match packet {
+///             ClientboundGamePacket::LevelParticles(p) => {
+///                 // ...
+///             }
+///             _ => {}
+///         }
+///     }
+/// }
+/// ```
+#[derive(Debug, Clone)]
+pub struct PacketEvent {
+    /// The client entity that received the packet.
+    pub entity: Entity,
+    /// The packet that was actually received.
+    pub packet: ClientboundGamePacket,
+}
+
 pub struct PacketHandlerPlugin;
 
+#[derive(StageLabel)]
+pub struct SendPacketEventsStage;
+
 impl Plugin for PacketHandlerPlugin {
     fn build(&self, app: &mut App) {
-        app.add_system_to_stage(CoreStage::PreUpdate, handle_packets)
+        app.add_stage_before(
+            CoreStage::PreUpdate,
+            SendPacketEventsStage,
+            SystemStage::parallel(),
+        )
+        .add_system_to_stage(SendPacketEventsStage, send_packet_events)
+        .add_system_to_stage(CoreStage::PreUpdate, process_packet_events)
+        .init_resource::<Events<PacketEvent>>()
         .add_event::<AddPlayerEvent>()
         .add_event::<RemovePlayerEvent>()
         .add_event::<UpdatePlayerEvent>()
@@ -107,26 +145,42 @@ pub struct PacketReceiver {
     pub run_schedule_sender: mpsc::UnboundedSender<()>,
 }
 
-pub fn handle_packets(ecs: &mut Ecs) {
+pub fn send_packet_events(
-    let mut events_owned = Vec::new();
+    query: Query<(Entity, &PacketReceiver)>,
-
+    mut packet_events: ResMut<Events<PacketEvent>>,
-    {
+) {
-        let mut system_state: SystemState<
+    // we manually clear and send the events at the beginning of each update
-            Query<(Entity, &PacketReceiver), Changed<PacketReceiver>>,
+    // since otherwise it'd cause issues with events in process_packet_events
-        > = SystemState::new(ecs);
+    // running twice
-        let query = system_state.get(ecs);
+    packet_events.clear();
-        for (player_entity, packet_events) in &query {
+    for (player_entity, packet_receiver) in &query {
-            let mut packets = packet_events.packets.lock();
+        let mut packets = packet_receiver.packets.lock();
         if !packets.is_empty() {
-                events_owned.push((player_entity, packets.clone()));
+            for packet in packets.iter() {
+                packet_events.send(PacketEvent {
+                    entity: player_entity,
+                    packet: packet.clone(),
+                });
+            }
             // clear the packets right after we read them
             packets.clear();
         }
     }
 }
 
-    for (player_entity, packets) in events_owned {
+fn process_packet_events(ecs: &mut Ecs) {
-        for packet in &packets {
+    let mut events_owned = Vec::new();
+    let mut system_state: SystemState<EventReader<PacketEvent>> = SystemState::new(ecs);
+    let mut events = system_state.get_mut(ecs);
+    for PacketEvent {
+        entity: player_entity,
+        packet,
+    } in events.iter()
+    {
+        // we do this so `ecs` isn't borrowed for the whole loop
+        events_owned.push((*player_entity, packet.clone()));
+    }
+    for (player_entity, packet) in events_owned {
         match packet {
             ClientboundGamePacket::Login(p) => {
                 debug!("Got login packet");
@@ -210,8 +264,7 @@ pub fn handle_packets(ecs: &mut Ecs) {
                     }
                     // add this world to the world_container (or don't if it's already
                     // there)
-                        let weak_world =
+                    let weak_world = world_container.insert(new_world_name.clone(), height, min_y);
-                            world_container.insert(new_world_name.clone(), height, min_y);
                     // set the partial_world to an empty world
                     // (when we add chunks or entities those will be in the
                     // world_container)
@@ -376,8 +429,7 @@ pub fn handle_packets(ecs: &mut Ecs) {
 
                 **position = new_pos;
 
-                    local_player
+                local_player.write_packet(ServerboundAcceptTeleportationPacket { id: p.id }.get());
-                        .write_packet(ServerboundAcceptTeleportationPacket { id: p.id }.get());
                 local_player.write_packet(
                     ServerboundMovePlayerPosRotPacket {
                         x: new_pos.x,
@@ -466,8 +518,7 @@ pub fn handle_packets(ecs: &mut Ecs) {
             ClientboundGamePacket::SetChunkCacheCenter(p) => {
                 debug!("Got chunk cache center packet {:?}", p);
 
-                    let mut system_state: SystemState<Query<&mut LocalPlayer>> =
+                let mut system_state: SystemState<Query<&mut LocalPlayer>> = SystemState::new(ecs);
-                        SystemState::new(ecs);
                 let mut query = system_state.get_mut(ecs);
                 let local_player = query.get_mut(player_entity).unwrap();
                 let mut partial_world = local_player.partial_world.write();
@@ -478,8 +529,7 @@ pub fn handle_packets(ecs: &mut Ecs) {
                 debug!("Got chunk with light packet {} {}", p.x, p.z);
                 let pos = ChunkPos::new(p.x, p.z);
 
-                    let mut system_state: SystemState<Query<&mut LocalPlayer>> =
+                let mut system_state: SystemState<Query<&mut LocalPlayer>> = SystemState::new(ecs);
-                        SystemState::new(ecs);
                 let mut query = system_state.get_mut(ecs);
                 let local_player = query.get_mut(player_entity).unwrap();
 
@@ -632,10 +682,8 @@ pub fn handle_packets(ecs: &mut Ecs) {
             ClientboundGamePacket::SetHealth(p) => {
                 debug!("Got set health packet {:?}", p);
 
-                    let mut system_state: SystemState<(
+                let mut system_state: SystemState<(Query<&mut Health>, EventWriter<DeathEvent>)> =
-                        Query<&mut Health>,
+                    SystemState::new(ecs);
-                        EventWriter<DeathEvent>,
-                    )> = SystemState::new(ecs);
                 let (mut query, mut death_events) = system_state.get_mut(ecs);
                 let mut health = query.get_mut(player_entity).unwrap();
 
@@ -803,8 +851,7 @@ pub fn handle_packets(ecs: &mut Ecs) {
             ClientboundGamePacket::BlockUpdate(p) => {
                 debug!("Got block update packet {:?}", p);
 
-                    let mut system_state: SystemState<Query<&mut LocalPlayer>> =
+                let mut system_state: SystemState<Query<&mut LocalPlayer>> = SystemState::new(ecs);
-                        SystemState::new(ecs);
                 let mut query = system_state.get_mut(ecs);
                 let local_player = query.get_mut(player_entity).unwrap();
 
@@ -817,8 +864,7 @@ pub fn handle_packets(ecs: &mut Ecs) {
             }
             ClientboundGamePacket::SectionBlocksUpdate(p) => {
                 debug!("Got section blocks update packet {:?}", p);
-                    let mut system_state: SystemState<Query<&mut LocalPlayer>> =
+                let mut system_state: SystemState<Query<&mut LocalPlayer>> = SystemState::new(ecs);
-                        SystemState::new(ecs);
                 let mut query = system_state.get_mut(ecs);
                 let local_player = query.get_mut(player_entity).unwrap();
 
@@ -937,7 +983,6 @@ pub fn handle_packets(ecs: &mut Ecs) {
         }
     }
 }
-}
 
 impl PacketReceiver {
     /// Loop that reads from the connection and adds the packets to the queue +
@@ -974,7 +1019,6 @@ impl PacketReceiver {
                 break;
             };
         }
-        println!("Write task finished");
         // receiver is automatically closed when it's dropped
     }
 }

azalea-ecs/azalea-ecs-macros/src/component.rs

@@ -19,7 +19,7 @@ pub fn derive_resource(input: TokenStream) -> TokenStream {
     let (impl_generics, type_generics, where_clause) = &ast.generics.split_for_impl();
 
     TokenStream::from(quote! {
-        impl #impl_generics #azalea_ecs_path::system::BevyResource for #struct_name #type_generics #where_clause {
+        impl #impl_generics #azalea_ecs_path::system::_BevyResource for #struct_name #type_generics #where_clause {
         }
     })
 }
@@ -44,7 +44,7 @@ pub fn derive_component(input: TokenStream) -> TokenStream {
     let (impl_generics, type_generics, where_clause) = &ast.generics.split_for_impl();
 
     TokenStream::from(quote! {
-        impl #impl_generics #azalea_ecs_path::component::BevyComponent for #struct_name #type_generics #where_clause {
+        impl #impl_generics #azalea_ecs_path::component::_BevyComponent for #struct_name #type_generics #where_clause {
             type Storage = #storage;
         }
     })

azalea-ecs/azalea-ecs-macros/src/lib.rs

@@ -151,13 +151,13 @@ pub fn derive_bundle(input: TokenStream) -> TokenStream {
         match field_kind {
             BundleFieldKind::Component => {
                 field_component_ids.push(quote! {
-                <#field_type as #ecs_path::bundle::BevyBundle>::component_ids(components, storages, &mut *ids);
+                <#field_type as #ecs_path::bundle::_BevyBundle>::component_ids(components, storages, &mut *ids);
                 });
                 field_get_components.push(quote! {
                     self.#field.get_components(&mut *func);
                 });
                 field_from_components.push(quote! {
-                    #field: <#field_type as #ecs_path::bundle::BevyBundle>::from_components(ctx, &mut *func),
+                    #field: <#field_type as #ecs_path::bundle::_BevyBundle>::from_components(ctx, &mut *func),
                 });
             }
 
@@ -174,7 +174,7 @@ pub fn derive_bundle(input: TokenStream) -> TokenStream {
 
     TokenStream::from(quote! {
         /// SAFETY: ComponentId is returned in field-definition-order. [from_components] and [get_components] use field-definition-order
-        unsafe impl #impl_generics #ecs_path::bundle::BevyBundle for #struct_name #ty_generics #where_clause {
+        unsafe impl #impl_generics #ecs_path::bundle::_BevyBundle for #struct_name #ty_generics #where_clause {
             fn component_ids(
                 components: &mut #ecs_path::component::Components,
                 storages: &mut #ecs_path::storage::Storages,
@@ -488,7 +488,9 @@ pub fn derive_stage_label(input: TokenStream) -> TokenStream {
     let input = parse_macro_input!(input as DeriveInput);
     let mut trait_path = azalea_ecs_path();
     trait_path.segments.push(format_ident!("schedule").into());
-    trait_path.segments.push(format_ident!("StageLabel").into());
+    trait_path
+        .segments
+        .push(format_ident!("_BevyStageLabel").into());
     derive_label(input, &trait_path, "stage_label")
 }
 

azalea-ecs/azalea-ecs-macros/src/utils/mod.rs

@@ -38,6 +38,7 @@ impl Default for BevyManifest {
 impl BevyManifest {
     pub fn maybe_get_path(&self, name: &str) -> Option<syn::Path> {
         const AZALEA: &str = "azalea";
+        const AZALEA_ECS: &str = "azalea_ecs";
         const BEVY_ECS: &str = "bevy_ecs";
         const BEVY: &str = "bevy";
 
@@ -57,6 +58,8 @@ impl BevyManifest {
                 return Some(Self::parse_str(dep_package(dep).unwrap_or(name)));
             } else if let Some(dep) = deps.get(AZALEA) {
                 dep_package(dep).unwrap_or(AZALEA)
+            } else if let Some(dep) = deps.get(AZALEA_ECS) {
+                dep_package(dep).unwrap_or(AZALEA_ECS)
             } else if let Some(dep) = deps.get(BEVY_ECS) {
                 dep_package(dep).unwrap_or(BEVY_ECS)
             } else if let Some(dep) = deps.get(BEVY) {

azalea-ecs/src/lib.rs

@@ -29,14 +29,14 @@ pub mod component {
     // we do this because re-exporting Component would re-export the macro as well,
     // which is bad (since we have our own Component macro)
     // instead, we have to do this so Component is a trait alias and the original
-    // impl-able trait is still available as BevyComponent
+    // impl-able trait is still available as _BevyComponent
     pub trait Component = bevy_ecs::component::Component;
-    pub use bevy_ecs::component::Component as BevyComponent;
+    pub use bevy_ecs::component::Component as _BevyComponent;
 }
 pub mod bundle {
     pub use azalea_ecs_macros::Bundle;
     pub trait Bundle = bevy_ecs::bundle::Bundle;
-    pub use bevy_ecs::bundle::Bundle as BevyBundle;
+    pub use bevy_ecs::bundle::Bundle as _BevyBundle;
 }
 pub mod system {
     pub use azalea_ecs_macros::Resource;
@@ -44,10 +44,19 @@ pub mod system {
         Command, Commands, EntityCommands, Query, Res, ResMut, SystemState,
     };
     pub trait Resource = bevy_ecs::system::Resource;
-    pub use bevy_ecs::system::Resource as BevyResource;
+    pub use bevy_ecs::system::Resource as _BevyResource;
+}
+pub mod schedule {
+    pub use azalea_ecs_macros::StageLabel;
+    pub use bevy_ecs::schedule::{
+        IntoRunCriteria, IntoSystemDescriptor, ReportExecutionOrderAmbiguities, Schedule, Stage,
+        SystemSet, SystemStage,
+    };
+    pub trait StageLabel = bevy_ecs::schedule::StageLabel;
+    pub use bevy_ecs::schedule::StageLabel as _BevyStageLabel;
 }
 pub use bevy_app as app;
-pub use bevy_ecs::{entity, event, ptr, query, schedule, storage};
+pub use bevy_ecs::{entity, event, ptr, query, storage};
 
 use app::{App, CoreStage, Plugin};
 use bevy_ecs::schedule::*;

azalea/examples/testbot.rs

@@ -9,6 +9,7 @@ use azalea::pathfinder::BlockPosGoal;
 use azalea::{prelude::*, swarm::prelude::*, BlockPos, GameProfileComponent, WalkDirection};
 use azalea::{Account, Client, Event};
 use azalea_protocol::packets::game::serverbound_client_command_packet::ServerboundClientCommandPacket;
+use azalea_protocol::packets::game::ClientboundGamePacket;
 use std::time::Duration;
 
 #[derive(Default, Clone, Component)]
@@ -148,6 +149,12 @@ async fn handle(mut bot: Client, event: Event, _state: State) -> anyhow::Result<
                 action: azalea_protocol::packets::game::serverbound_client_command_packet::Action::PerformRespawn,
             }.get());
         }
+        Event::Packet(packet) => match *packet {
+            ClientboundGamePacket::Login(_) => {
+                println!("login packet");
+            }
+            _ => {}
+        },
         _ => {}
     }