Better chat events (#67)

* Better chat events

* add a comment explaining why SendChatKindEvent is only one event

azalea-client/src/chat.rs

@@ -1,6 +1,12 @@
 //! Implementations of chat-related features.
 
 use azalea_chat::FormattedText;
+use azalea_ecs::{
+    app::{App, Plugin},
+    entity::Entity,
+    event::{EventReader, EventWriter},
+    schedule::IntoSystemDescriptor,
+};
 use azalea_protocol::packets::game::{
     clientbound_player_chat_packet::ClientboundPlayerChatPacket,
     clientbound_system_chat_packet::ClientboundSystemChatPacket,
@@ -13,7 +19,7 @@ use std::{
 };
 use uuid::Uuid;
 
-use crate::client::Client;
+use crate::{client::Client, local_player::SendPacketEvent};
 
 /// A chat packet, either a system message or a chat message.
 #[derive(Debug, Clone, PartialEq)]
@@ -107,42 +113,23 @@ impl Client {
     /// whether the message is a command and using the proper packet for you,
     /// so you should use that instead.
     pub fn send_chat_packet(&self, message: &str) {
-        // TODO: chat signing
+        self.ecs.lock().send_event(SendChatKindEvent {
-        // let signature = sign_message();
+            entity: self.entity,
-        let packet = ServerboundChatPacket {
+            content: message.to_string(),
-            message: message.to_string(),
+            kind: ChatPacketKind::Message,
-            timestamp: SystemTime::now()
+        });
-                .duration_since(UNIX_EPOCH)
+        self.run_schedule_sender.send(()).unwrap();
-                .expect("Time shouldn't be before epoch")
-                .as_millis()
-                .try_into()
-                .expect("Instant should fit into a u64"),
-            salt: azalea_crypto::make_salt(),
-            signature: None,
-            last_seen_messages: LastSeenMessagesUpdate::default(),
-        }
-        .get();
-        self.write_packet(packet);
     }
 
     /// Send a command packet to the server. The `command` argument should not
     /// include the slash at the front.
     pub fn send_command_packet(&self, command: &str) {
-        // TODO: chat signing
+        self.ecs.lock().send_event(SendChatKindEvent {
-        let packet = ServerboundChatCommandPacket {
+            entity: self.entity,
-            command: command.to_string(),
+            content: command.to_string(),
-            timestamp: SystemTime::now()
+            kind: ChatPacketKind::Command,
-                .duration_since(UNIX_EPOCH)
+        });
-                .expect("Time shouldn't be before epoch")
+        self.run_schedule_sender.send(()).unwrap();
-                .as_millis()
-                .try_into()
-                .expect("Instant should fit into a u64"),
-            salt: azalea_crypto::make_salt(),
-            argument_signatures: vec![],
-            last_seen_messages: LastSeenMessagesUpdate::default(),
-        }
-        .get();
-        self.write_packet(packet);
     }
 
     /// Send a message in chat.
@@ -154,13 +141,130 @@ impl Client {
     /// # Ok(())
     /// # }
     /// ```
-    pub fn chat(&self, message: &str) {
+    pub fn chat(&self, content: &str) {
-        if let Some(command) = message.strip_prefix('/') {
+        self.ecs.lock().send_event(SendChatEvent {
-            self.send_command_packet(command);
+            entity: self.entity,
-        } else {
+            content: content.to_string(),
-            self.send_chat_packet(message);
+        });
     }
 }
+
+pub struct ChatPlugin;
+impl Plugin for ChatPlugin {
+    fn build(&self, app: &mut App) {
+        app.add_event::<SendChatEvent>()
+            .add_event::<SendChatKindEvent>()
+            .add_event::<ChatReceivedEvent>()
+            .add_system(
+                handle_send_chat_event
+                    .label("handle_send_chat_event")
+                    .after("packet"),
+            )
+            .add_system(
+                handle_send_chat_kind_event
+                    .label("handle_send_chat_kind_event")
+                    .after("handle_send_chat_event"),
+            );
+    }
+}
+
+/// A client received a chat message packet.
+#[derive(Debug, Clone)]
+pub struct ChatReceivedEvent {
+    pub entity: Entity,
+    pub packet: ChatPacket,
+}
+
+/// Send a chat message (or command, if it starts with a slash) to the server.
+pub struct SendChatEvent {
+    pub entity: Entity,
+    pub content: String,
+}
+
+fn handle_send_chat_event(
+    mut events: EventReader<SendChatEvent>,
+    mut send_chat_kind_events: EventWriter<SendChatKindEvent>,
+) {
+    for event in events.iter() {
+        if event.content.starts_with('/') {
+            send_chat_kind_events.send(SendChatKindEvent {
+                entity: event.entity,
+                content: event.content[1..].to_string(),
+                kind: ChatPacketKind::Command,
+            });
+        } else {
+            send_chat_kind_events.send(SendChatKindEvent {
+                entity: event.entity,
+                content: event.content.clone(),
+                kind: ChatPacketKind::Message,
+            });
+        }
+    }
+}
+
+/// Send a chat packet to the server of a specific kind (chat message or
+/// command). Usually you just want [`SendChatEvent`] instead.
+///
+/// Usually setting the kind to `Message` will make it send a chat message even
+/// if it starts with a slash, but some server implementations will always do a
+/// command if it starts with a slash.
+///
+/// If you're wondering why this isn't two separate events, it's so ordering is
+/// preserved if multiple chat messages and commands are sent at the same time.
+pub struct SendChatKindEvent {
+    pub entity: Entity,
+    pub content: String,
+    pub kind: ChatPacketKind,
+}
+
+/// A kind of chat packet, either a chat message or a command.
+pub enum ChatPacketKind {
+    Message,
+    Command,
+}
+
+fn handle_send_chat_kind_event(
+    mut events: EventReader<SendChatKindEvent>,
+    mut send_packet_events: EventWriter<SendPacketEvent>,
+) {
+    for event in events.iter() {
+        let packet = match event.kind {
+            ChatPacketKind::Message => ServerboundChatPacket {
+                message: event.content.clone(),
+                timestamp: SystemTime::now()
+                    .duration_since(UNIX_EPOCH)
+                    .expect("Time shouldn't be before epoch")
+                    .as_millis()
+                    .try_into()
+                    .expect("Instant should fit into a u64"),
+                salt: azalea_crypto::make_salt(),
+                signature: None,
+                last_seen_messages: LastSeenMessagesUpdate::default(),
+            }
+            .get(),
+            ChatPacketKind::Command => {
+                // TODO: chat signing
+                ServerboundChatCommandPacket {
+                    command: event.content.clone(),
+                    timestamp: SystemTime::now()
+                        .duration_since(UNIX_EPOCH)
+                        .expect("Time shouldn't be before epoch")
+                        .as_millis()
+                        .try_into()
+                        .expect("Instant should fit into a u64"),
+                    salt: azalea_crypto::make_salt(),
+                    argument_signatures: vec![],
+                    last_seen_messages: LastSeenMessagesUpdate::default(),
+                }
+                .get()
+            }
+        };
+
+        send_packet_events.send(SendPacketEvent {
+            entity: event.entity,
+            packet,
+        });
+    }
 }
 
 // TODO

azalea-client/src/client.rs

@@ -1,8 +1,9 @@
-pub use crate::chat::ChatPacket;
 use crate::{
+    chat::ChatPlugin,
     events::{Event, EventPlugin, LocalPlayerEvents},
     local_player::{
-        death_event, update_in_loaded_chunk, GameProfileComponent, LocalPlayer, PhysicsState,
+        death_event, handle_send_packet_event, update_in_loaded_chunk, GameProfileComponent,
+        LocalPlayer, PhysicsState, SendPacketEvent,
     },
     movement::{local_player_ai_step, send_position, sprint_listener, walk_listener},
     packet_handling::{self, PacketHandlerPlugin},
@@ -80,6 +81,9 @@ pub struct Client {
     /// directly. Note that if you're using a shared world (i.e. a swarm), this
     /// will contain all entities in all worlds.
     pub ecs: Arc<Mutex<Ecs>>,
+
+    /// Use this to force the client to run the schedule outside of a tick.
+    pub run_schedule_sender: mpsc::UnboundedSender<()>,
 }
 
 /// An error that happened while joining the server.
@@ -107,7 +111,12 @@ impl Client {
     /// Create a new client from the given GameProfile, Connection, and World.
     /// You should only use this if you want to change these fields from the
     /// defaults, otherwise use [`Client::join`].
-    pub fn new(profile: GameProfile, entity: Entity, ecs: Arc<Mutex<Ecs>>) -> Self {
+    pub fn new(
+        profile: GameProfile,
+        entity: Entity,
+        ecs: Arc<Mutex<Ecs>>,
+        run_schedule_sender: mpsc::UnboundedSender<()>,
+    ) -> Self {
         Self {
             profile,
             // default our id to 0, it'll be set later
@@ -115,6 +124,8 @@ impl Client {
             world: Arc::new(RwLock::new(PartialWorld::default())),
 
             ecs,
+
+            run_schedule_sender,
         }
     }
 
@@ -146,7 +157,7 @@ impl Client {
         let resolved_address = resolver::resolve_address(&address).await?;
 
         // An event that causes the schedule to run. This is only used internally.
-        let (run_schedule_sender, run_schedule_receiver) = mpsc::channel(1);
+        let (run_schedule_sender, run_schedule_receiver) = mpsc::unbounded_channel();
         let app = init_ecs_app();
         let ecs_lock = start_ecs(app, run_schedule_receiver, run_schedule_sender.clone());
 
@@ -167,7 +178,7 @@ impl Client {
         account: &Account,
         address: &ServerAddress,
         resolved_address: &SocketAddr,
-        run_schedule_sender: mpsc::Sender<()>,
+        run_schedule_sender: mpsc::UnboundedSender<()>,
     ) -> Result<(Self, mpsc::UnboundedReceiver<Event>), JoinError> {
         let conn = Connection::new(resolved_address).await?;
         let (conn, game_profile) = Self::handshake(conn, account, address).await?;
@@ -182,7 +193,12 @@ impl Client {
         let entity = entity_mut.id();
 
         // we got the GameConnection, so the server is now connected :)
-        let client = Client::new(game_profile.clone(), entity, ecs_lock.clone());
+        let client = Client::new(
+            game_profile.clone(),
+            entity,
+            ecs_lock.clone(),
+            run_schedule_sender.clone(),
+        );
 
         let (packet_writer_sender, packet_writer_receiver) = mpsc::unbounded_channel();
 
@@ -458,8 +474,6 @@ impl Plugin for AzaleaPlugin {
         app.add_event::<StartWalkEvent>()
             .add_event::<StartSprintEvent>();
 
-        app.add_plugins(DefaultPlugins);
-
         app.add_tick_system_set(
             SystemSet::new()
                 .with_system(send_position)
@@ -490,6 +504,9 @@ impl Plugin for AzaleaPlugin {
                 .after("packet"),
         );
 
+        app.add_event::<SendPacketEvent>()
+            .add_system(handle_send_packet_event.after("tick").after("packet"));
+
         app.init_resource::<WorldContainer>();
     }
 }
@@ -509,7 +526,7 @@ pub fn init_ecs_app() -> App {
     // you might be able to just drop the lock or put it in its own scope to fix
 
     let mut app = App::new();
-    app.add_plugin(AzaleaPlugin);
+    app.add_plugins(DefaultPlugins);
     app
 }
 
@@ -518,8 +535,8 @@ pub fn init_ecs_app() -> App {
 #[doc(hidden)]
 pub fn start_ecs(
     app: App,
-    run_schedule_receiver: mpsc::Receiver<()>,
+    run_schedule_receiver: mpsc::UnboundedReceiver<()>,
-    run_schedule_sender: mpsc::Sender<()>,
+    run_schedule_sender: mpsc::UnboundedSender<()>,
 ) -> Arc<Mutex<Ecs>> {
     // all resources should have been added by now so we can take the ecs from the
     // app
@@ -538,7 +555,7 @@ pub fn start_ecs(
 async fn run_schedule_loop(
     ecs: Arc<Mutex<Ecs>>,
     mut schedule: Schedule,
-    mut run_schedule_receiver: mpsc::Receiver<()>,
+    mut run_schedule_receiver: mpsc::UnboundedReceiver<()>,
 ) {
     loop {
         // whenever we get an event from run_schedule_receiver, run the schedule
@@ -549,14 +566,14 @@ async fn run_schedule_loop(
 
 /// Send an event to run the schedule every 50 milliseconds. It will stop when
 /// the receiver is dropped.
-pub async fn tick_run_schedule_loop(run_schedule_sender: mpsc::Sender<()>) {
+pub async fn tick_run_schedule_loop(run_schedule_sender: mpsc::UnboundedSender<()>) {
     let mut game_tick_interval = time::interval(time::Duration::from_millis(50));
     // TODO: Minecraft bursts up to 10 ticks and then skips, we should too
     game_tick_interval.set_missed_tick_behavior(time::MissedTickBehavior::Burst);
 
     loop {
         game_tick_interval.tick().await;
-        if let Err(e) = run_schedule_sender.send(()).await {
+        if let Err(e) = run_schedule_sender.send(()) {
             println!("tick_run_schedule_loop error: {e}");
             // the sender is closed so end the task
             return;
@@ -572,10 +589,12 @@ impl PluginGroup for DefaultPlugins {
     fn build(self) -> PluginGroupBuilder {
         PluginGroupBuilder::start::<Self>()
             .add(TickPlugin::default())
+            .add(AzaleaPlugin)
             .add(PacketHandlerPlugin)
             .add(EntityPlugin)
             .add(PhysicsPlugin)
             .add(EventPlugin)
             .add(TaskPoolPlugin::default())
+            .add(ChatPlugin)
     }
 }

azalea-client/src/events.rs

@@ -19,11 +19,12 @@ use derive_more::{Deref, DerefMut};
 use tokio::sync::mpsc;
 
 use crate::{
+    chat::{ChatPacket, ChatReceivedEvent},
     packet_handling::{
-        AddPlayerEvent, ChatReceivedEvent, DeathEvent, KeepAliveEvent, PacketReceiver,
+        AddPlayerEvent, DeathEvent, KeepAliveEvent, PacketReceiver, RemovePlayerEvent,
-        RemovePlayerEvent, UpdatePlayerEvent,
+        UpdatePlayerEvent,
     },
-    ChatPacket, PlayerInfo,
+    PlayerInfo,
 };
 
 // (for contributors):

azalea-client/src/lib.rs

@@ -12,7 +12,7 @@
 #![feature(type_alias_impl_trait)]
 
 mod account;
-mod chat;
+pub mod chat;
 mod client;
 mod entity_query;
 mod events;
@@ -26,7 +26,7 @@ pub mod task_pool;
 
 pub use account::Account;
 pub use azalea_ecs as ecs;
-pub use client::{init_ecs_app, start_ecs, ChatPacket, Client, ClientInformation, JoinError};
+pub use client::{init_ecs_app, start_ecs, Client, ClientInformation, JoinError};
 pub use events::Event;
 pub use local_player::{GameProfileComponent, LocalPlayer};
 pub use movement::{SprintDirection, StartSprintEvent, StartWalkEvent, WalkDirection};

azalea-client/src/local_player.rs

@@ -4,6 +4,7 @@ use azalea_auth::game_profile::GameProfile;
 use azalea_core::ChunkPos;
 use azalea_ecs::component::Component;
 use azalea_ecs::entity::Entity;
+use azalea_ecs::event::EventReader;
 use azalea_ecs::{query::Added, system::Query};
 use azalea_protocol::packets::game::ServerboundGamePacket;
 use azalea_world::{
@@ -168,3 +169,20 @@ impl<T> From<std::sync::PoisonError<T>> for HandlePacketError {
         HandlePacketError::Poison(e.to_string())
     }
 }
+
+/// Event for sending a packet to the server.
+pub struct SendPacketEvent {
+    pub entity: Entity,
+    pub packet: ServerboundGamePacket,
+}
+
+pub fn handle_send_packet_event(
+    mut send_packet_events: EventReader<SendPacketEvent>,
+    mut query: Query<&mut LocalPlayer>,
+) {
+    for event in send_packet_events.iter() {
+        if let Ok(mut local_player) = query.get_mut(event.entity) {
+            local_player.write_packet(event.packet.clone());
+        }
+    }
+}

azalea-client/src/packet_handling.rs

@@ -37,8 +37,9 @@ use parking_lot::Mutex;
 use tokio::sync::mpsc;
 
 use crate::{
+    chat::{ChatPacket, ChatReceivedEvent},
     local_player::{GameProfileComponent, LocalPlayer},
-    ChatPacket, ClientInformation, PlayerInfo,
+    ClientInformation, PlayerInfo,
 };
 
 pub struct PacketHandlerPlugin;
@@ -82,13 +83,6 @@ pub struct UpdatePlayerEvent {
     pub info: PlayerInfo,
 }
 
-/// A client received a chat message packet.
-#[derive(Debug, Clone)]
-pub struct ChatReceivedEvent {
-    pub entity: Entity,
-    pub packet: ChatPacket,
-}
-
 /// Event for when an entity dies. dies. If it's a local player and there's a
 /// reason in the death screen, the [`ClientboundPlayerCombatKillPacket`] will
 /// be included.
@@ -112,7 +106,7 @@ pub struct KeepAliveEvent {
 #[derive(Component, Clone)]
 pub struct PacketReceiver {
     pub packets: Arc<Mutex<Vec<ClientboundGamePacket>>>,
-    pub run_schedule_sender: mpsc::Sender<()>,
+    pub run_schedule_sender: mpsc::UnboundedSender<()>,
 }
 
 fn handle_packets(ecs: &mut Ecs) {
@@ -950,7 +944,7 @@ impl PacketReceiver {
                 Ok(packet) => {
                     self.packets.lock().push(packet);
                     // tell the client to run all the systems
-                    self.run_schedule_sender.send(()).await.unwrap();
+                    self.run_schedule_sender.send(()).unwrap();
                 }
                 Err(error) => {
                     if !matches!(*error, ReadPacketError::ConnectionClosed) {

azalea/src/lib.rs

@@ -135,7 +135,7 @@ where
         let resolved_address = resolver::resolve_address(&address).await?;
 
         // An event that causes the schedule to run. This is only used internally.
-        let (run_schedule_sender, run_schedule_receiver) = mpsc::channel(1);
+        let (run_schedule_sender, run_schedule_receiver) = mpsc::unbounded_channel();
         let ecs_lock = start_ecs(self.app, run_schedule_receiver, run_schedule_sender.clone());
 
         let (bot, mut rx) = Client::start_client(

azalea/src/swarm/chat.rs

@@ -13,7 +13,7 @@
 // in Swarm that's set to the smallest index of all the bots, and we remove all
 // messages from the queue that are before that index.
 
-use azalea_client::{packet_handling::ChatReceivedEvent, ChatPacket};
+use azalea_client::chat::{ChatPacket, ChatReceivedEvent};
 use azalea_ecs::{
     app::{App, Plugin},
     component::Component,

azalea/src/swarm/mod.rs

@@ -5,7 +5,7 @@ mod events;
 pub mod prelude;
 
 use crate::{bot::DefaultBotPlugins, HandleFn};
-use azalea_client::{init_ecs_app, start_ecs, Account, ChatPacket, Client, Event, JoinError};
+use azalea_client::{chat::ChatPacket, init_ecs_app, start_ecs, Account, Client, Event, JoinError};
 use azalea_ecs::{
     app::{App, Plugin, PluginGroup, PluginGroupBuilder},
     component::Component,
@@ -47,7 +47,7 @@ pub struct Swarm {
     bots_tx: mpsc::UnboundedSender<(Option<Event>, Client)>,
     swarm_tx: mpsc::UnboundedSender<SwarmEvent>,
 
-    run_schedule_sender: mpsc::Sender<()>,
+    run_schedule_sender: mpsc::UnboundedSender<()>,
 }
 
 /// Create a new [`Swarm`].
@@ -253,7 +253,7 @@ where
         let (bots_tx, mut bots_rx) = mpsc::unbounded_channel();
         let (swarm_tx, mut swarm_rx) = mpsc::unbounded_channel();
 
-        let (run_schedule_sender, run_schedule_receiver) = mpsc::channel(1);
+        let (run_schedule_sender, run_schedule_receiver) = mpsc::unbounded_channel();
         let ecs_lock = start_ecs(self.app, run_schedule_receiver, run_schedule_sender.clone());
 
         let swarm = Swarm {