fix #673 - Listener::stop(&self) and in msg ctx

guest/rust/api_core/src/client/input.rs

@@ -59,17 +59,17 @@ pub fn set_cursor_lock(locked: bool) {
 /// Will unlock the cursor when dropped.
 pub struct CursorLockGuard {
     inner: Arc<Mutex<CursorLockGuardInner>>,
-    listener: Option<Listener>,
+    listener: Listener,
 }
 impl CursorLockGuard {
     /// Creates a new [CursorLockGuard] with the given lock state.
     pub fn new() -> Self {
         let inner = Arc::new(Mutex::new(CursorLockGuardInner::new()));
         Self {
             inner: inner.clone(),
-            listener: Some(WindowFocusChange::subscribe(move |msg| {
+            listener: WindowFocusChange::subscribe(move |msg| {
                 inner.lock().unwrap().set_locked(msg.focused)
-            })),
+            }),
         }
     }
 
@@ -113,7 +113,7 @@ impl Default for CursorLockGuard {
 impl Drop for CursorLockGuard {
     fn drop(&mut self) {
         self.set_locked(false);
-        self.listener.take().unwrap().stop();
+        self.listener.stop();
     }
 }
 

guest/rust/api_core/src/internal/executor.rs

@@ -15,7 +15,7 @@ use rand::random;
 use super::wit;
 
 pub type EventFuture = Pin<Box<dyn Future<Output = ResultEmpty>>>;
-type EventCallbackFn = Box<dyn FnMut(&wit::guest::Source, &[u8]) -> ResultEmpty>;
+type EventCallbackFn = Box<dyn FnMut(&wit::guest::Source, u128, &[u8]) -> ResultEmpty>;
 
 // the function is too general to be passed in directly
 #[allow(clippy::redundant_closure)]
@@ -71,8 +71,8 @@ impl Executor {
         {
             let mut callbacks = self.current_callbacks.borrow_mut();
             if let Some(callbacks) = callbacks.on.get_mut(&message_name) {
-                for callback in callbacks.values_mut() {
-                    callback(&source, &message_data).unwrap();
+                for (listener_id, callback) in callbacks.iter_mut() {
+                    callback(&source, *listener_id, &message_data).unwrap();
                 }
             }
         }

guest/rust/api_core/src/message.rs

@@ -1,3 +1,5 @@
+use std::fmt::Debug;
+
 use crate::{
     global::{CallbackReturn, EntityId},
     internal::{conversion::FromBindgen, executor::EXECUTOR, wit},
@@ -240,25 +242,81 @@ pub fn send<T: Message>(target: Target, data: &T) {
 pub struct Listener(String, u128);
 impl Listener {
     /// Stops listening.
-    pub fn stop(self) {
+    pub fn stop(&self) {
         EXECUTOR.unregister_callback(&self.0, self.1);
     }
 }
 
+/// Message context.
+pub struct MessageContext {
+    /// Where the message came from.
+    pub source: Source,
+    /// The listener that can be used to stop listening.
+    pub listener: Listener,
+}
+impl MessageContext {
+    /// Is this message from the runtime?
+    pub fn runtime(&self) -> bool {
+        self.source.runtime()
+    }
+
+    #[cfg(feature = "client")]
+    /// Is this message from the corresponding serverside package?
+    pub fn server(&self) -> bool {
+        self.source.server()
+    }
+
+    #[cfg(feature = "server")]
+    /// The user that sent this message, if any.
+    pub fn client_user_id(&self) -> Option<String> {
+        self.source.client_user_id()
+    }
+
+    #[cfg(feature = "server")]
+    /// The entity ID of the player that sent this message, if any.
+    pub fn client_entity_id(&self) -> Option<EntityId> {
+        self.source.client_entity_id()
+    }
+
+    /// The module on this side that sent this message, if any.
+    pub fn local(&self) -> Option<EntityId> {
+        self.source.local()
+    }
+
+    /// Stops listening.
+    pub fn stop(&self) {
+        self.listener.stop()
+    }
+}
+impl Debug for MessageContext {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        self.source.fmt(f)
+    }
+}
+
 /// Subscribes to a message.
+///
+/// To unsubscribe from a message, call [Listener::stop] on the returned [Listener],
+/// or on the [MessageContext] that is passed to the callback.
 #[allow(clippy::collapsible_else_if)]
 pub fn subscribe<R: CallbackReturn, T: Message>(
-    mut callback: impl FnMut(Source, T) -> R + 'static,
+    mut callback: impl FnMut(MessageContext, T) -> R + 'static,
 ) -> Listener {
     let id = T::id();
     wit::message::subscribe(id);
     Listener(
         id.to_string(),
         EXECUTOR.register_callback(
             id.to_string(),
-            Box::new(move |source, data| {
-                callback(source.clone().from_bindgen(), T::deserialize_message(data)?)
-                    .into_result()?;
+            Box::new(move |source, listener_id, data| {
+                callback(
+                    MessageContext {
+                        source: source.clone().from_bindgen(),
+                        listener: Listener(id.to_string(), listener_id),
+                    },
+                    T::deserialize_message(data)?,
+                )
+                .into_result()?;
                 Ok(())
             }),
         ),
@@ -357,7 +415,9 @@ pub trait ModuleMessage: Message {
     }
 
     /// Subscribes to this [Message]. Wrapper around [self::subscribe].
-    fn subscribe<R: CallbackReturn>(callback: impl FnMut(Source, Self) -> R + 'static) -> Listener {
+    fn subscribe<R: CallbackReturn>(
+        callback: impl FnMut(MessageContext, Self) -> R + 'static,
+    ) -> Listener {
         self::subscribe(callback)
     }
 }
@@ -366,7 +426,7 @@ pub trait ModuleMessage: Message {
 pub trait RuntimeMessage: Message {
     /// Subscribes to this [Message]. Wrapper around [self::subscribe].
     fn subscribe<R: CallbackReturn>(mut callback: impl FnMut(Self) -> R + 'static) -> Listener {
-        self::subscribe(move |_source, msg| callback(msg))
+        self::subscribe(move |_ctx, msg| callback(msg))
     }
 }
 

guest/rust/examples/basics/physics/src/client.rs

@@ -5,7 +5,7 @@ use packages::this::{assets, messages::Bonk};
 pub fn main() {
     let spatial_audio_player = audio::SpatialAudioPlayer::new();
 
-    Bonk::subscribe(move |_source, data| {
+    Bonk::subscribe(move |_ctx, data| {
         spatial_audio_player.set_listener(data.listener);
         spatial_audio_player.play_sound_on_entity(assets::url("bonk.ogg"), data.emitter);
     });

guest/rust/examples/controllers/first_person_camera/src/server.rs

@@ -76,8 +76,8 @@ pub fn main() {
         }
     });
 
-    Input::subscribe(move |source, msg| {
-        let Some(player_id) = source.client_entity_id() else {
+    Input::subscribe(move |ctx, msg| {
+        let Some(player_id) = ctx.client_entity_id() else {
             return;
         };
 

guest/rust/examples/controllers/third_person_camera/src/server.rs

@@ -72,8 +72,8 @@ pub fn main() {
         }
     });
 
-    Input::subscribe(move |source, msg| {
-        let Some(player_id) = source.client_entity_id() else {
+    Input::subscribe(move |ctx, msg| {
+        let Some(player_id) = ctx.client_entity_id() else {
             return;
         };
 

guest/rust/examples/intermediate/messaging/src/client.rs

@@ -47,8 +47,8 @@ pub fn main() {
     Hello::new("Hello, world from the client!", true).send_server_reliable();
 
     // 2
-    Hello::subscribe(|source, data| {
-        println!("{source:?}: {:?}", data);
+    Hello::subscribe(|ctx, data| {
+        println!("{ctx:?}: {:?}", data);
 
         let source_reliable = data.source_reliable;
         Entity::new()
@@ -74,9 +74,9 @@ pub fn main() {
     let handled = Arc::new(AtomicBool::new(false));
     Local::subscribe({
         let handled = handled.clone();
-        move |source, data| {
+        move |ctx, data| {
             handled.store(true, Ordering::SeqCst);
-            println!("{source:?}: {data:?}");
+            println!("{ctx:?}: {data:?}");
         }
     });
 

guest/rust/examples/intermediate/messaging/src/client_two.rs

@@ -4,9 +4,9 @@ use packages::this::messages::Local;
 #[main]
 pub fn main() {
     // This module will receive messages from `client.rs`, and respond to them.
-    Local::subscribe(move |source, data| {
-        println!("{source:?}: {data:?}");
-        if let Some(id) = source.local() {
+    Local::subscribe(move |ctx, data| {
+        println!("{ctx:?}: {data:?}");
+        if let Some(id) = ctx.local() {
             Local::new("Hi, back!").send_local(id);
         }
     });

guest/rust/examples/intermediate/messaging/src/server.rs

@@ -30,8 +30,8 @@ pub fn main() {
     // See <https://github.com/AmbientRun/Ambient/issues/590> for more details.
 
     // 1
-    Hello::subscribe(|source, data| {
-        let Some(user_id) = source.client_user_id() else {
+    Hello::subscribe(|ctx, data| {
+        let Some(user_id) = ctx.client_user_id() else {
             return;
         };
         println!("{user_id}: {:?}", data);
@@ -80,9 +80,9 @@ pub fn main() {
     let handled = Arc::new(AtomicBool::new(false));
     Local::subscribe({
         let handled = handled.clone();
-        move |source, data| {
+        move |ctx, data| {
             handled.store(true, Ordering::SeqCst);
-            println!("{source:?}: {data:?}");
+            println!("{ctx:?}: {data:?}");
         }
     });
 

guest/rust/examples/intermediate/messaging/src/server_two.rs

@@ -4,14 +4,14 @@ use packages::this::messages::{HelloWithoutBody, Local};
 #[main]
 pub fn main() {
     // This module will receive messages from `server.rs`, and respond to them.
-    Local::subscribe(move |source, data| {
-        println!("{source:?}: {data:?}");
-        if let Some(id) = source.local() {
+    Local::subscribe(move |ctx, data| {
+        println!("{ctx:?}: {data:?}");
+        if let Some(id) = ctx.local() {
             Local::new("Hi, back!").send_local(id);
         }
     });
 
-    HelloWithoutBody::subscribe(move |source, _| {
-        println!("HelloWithoutBody: {source:?}");
+    HelloWithoutBody::subscribe(move |ctx, _| {
+        println!("HelloWithoutBody: {ctx:?}");
     });
 }

guest/rust/examples/intermediate/screen_ray/src/server.rs

@@ -21,7 +21,7 @@ pub fn main() {
         .with(cube(), ())
         .spawn();
 
-    Input::subscribe(move |_source, msg| {
+    Input::subscribe(move |_ctx, msg| {
         if let Some(hit) = physics::raycast_first(msg.ray_origin, msg.ray_dir) {
             // Set position of cube to the raycast hit position
             entity::set_component(cube_id, translation(), hit.position);

guest/rust/examples/ui/todo/src/server.rs

@@ -6,13 +6,13 @@ use packages::this::{
 
 #[main]
 pub async fn main() {
-    NewItem::subscribe(|_source, data| {
+    NewItem::subscribe(|_ctx, data| {
         Entity::new()
             .with(todo_item(), data.description)
             .with(todo_time(), game_time())
             .spawn();
     });
-    DeleteItem::subscribe(|_source, data| {
+    DeleteItem::subscribe(|_ctx, data| {
         entity::despawn(data.id);
     });
 }

guest/rust/packages/games/afps/core/fpsmovement/src/server.rs

@@ -29,9 +29,9 @@ pub fn main() {
     });
 
     let mut last_walk = game_time();
-    Input::subscribe(move |source, msg| {
+    Input::subscribe(move |ctx, msg| {
         // receive movement and send this for further processing
-        let player_id = source.client_entity_id();
+        let player_id = ctx.client_entity_id();
         if player_id.is_none() {
             return;
         }

guest/rust/packages/games/afps/core/fpsrule/src/server.rs

@@ -41,7 +41,7 @@ pub fn main() {
         }
     });
 
-    Shoot::subscribe(move |_source, msg| {
+    Shoot::subscribe(move |_ctx, msg| {
         let result = physics::raycast_first(msg.ray_origin, msg.ray_dir);
 
         if let Some(hit) = result {

guest/rust/packages/games/afps/core/fpsui/src/server.rs

@@ -4,8 +4,8 @@ use packages::afps_schema::{components::player_name, messages::StartGame};
 
 #[main]
 pub fn main() {
-    StartGame::subscribe(|source, msg| {
-        if let Some(id) = source.client_entity_id() {
+    StartGame::subscribe(|ctx, msg| {
+        if let Some(id) = ctx.client_entity_id() {
             entity::add_component(id, player_name(), msg.name);
         }
     });

guest/rust/packages/games/afps/mods/spraypaint/src/server.rs

@@ -27,7 +27,7 @@ use packages::{
 #[main]
 
 pub fn main() {
-    Spraypaint::subscribe(move |_source, msg| {
+    Spraypaint::subscribe(move |_ctx, msg| {
         if let Some(hit) = physics::raycast_first(msg.origin, msg.dir) {
             let player_pos = entity::get_component(msg.source, translation()).unwrap();
             let distance = (player_pos - hit.position).length();

guest/rust/packages/games/afps/mods/world_latency/src/server.rs

@@ -23,8 +23,8 @@ pub fn main() {
         mutate_component(server_entity_id, server_frame(), |frame| *frame += 1);
     });
 
-    FrameSeen::subscribe(move |source, msg| {
-        let Some(player_entity_id) = source.client_entity_id() else {
+    FrameSeen::subscribe(move |ctx, msg| {
+        let Some(player_entity_id) = ctx.client_entity_id() else {
             eprintln!("Received message from unknown client");
             return;
         };

guest/rust/packages/games/arkanoid/src/client.rs

@@ -31,7 +31,7 @@ async fn main() {
     let bgm_player = audio::AudioPlayer::new();
     bgm_player.set_amplitude(0.2);
 
-    Ping::subscribe(move |_source, _data| {
+    Ping::subscribe(move |_ctx, _data| {
         bgm_player.play(url_from_server.clone());
     });
 

guest/rust/packages/games/arkanoid/src/server.rs

@@ -138,8 +138,8 @@ pub fn main() {
         }
     });
 
-    Input::subscribe(move |source, msg| {
-        let Some(player_id) = source.client_entity_id() else {
+    Input::subscribe(move |ctx, msg| {
+        let Some(player_id) = ctx.client_entity_id() else {
             return;
         };
 

guest/rust/packages/games/minigolf/src/client.rs

@@ -26,7 +26,7 @@ fn main() {
 
     let ball_hit_player = audio::AudioPlayer::new();
     let ball_drop_player = audio::AudioPlayer::new();
-    Hit::subscribe(move |_source, data| {
+    Hit::subscribe(move |_ctx, data| {
         let ball = data.ball;
         let vel = entity::get_component(ball, linear_velocity()).unwrap_or_default();
         let mut amp = (vel.x.abs() / 5.0).powf(2.0)
@@ -38,7 +38,7 @@ fn main() {
         ball_hit_player.play(assets::url("ball-hit.ogg"));
     });
 
-    Bonk::subscribe(move |_source, data| {
+    Bonk::subscribe(move |_ctx, data| {
         let ball = data.ball;
         let vel = entity::get_component(ball, linear_velocity()).unwrap_or_default();
         let mut amp = (vel.x.abs() / 5.0).powf(2.0)

guest/rust/packages/games/minigolf/src/server.rs

@@ -250,8 +250,8 @@ pub fn main() {
         }
     });
 
-    Input::subscribe(|source, msg| {
-        let Some(user_id) = source.client_entity_id() else {
+    Input::subscribe(|ctx, msg| {
+        let Some(user_id) = ctx.client_entity_id() else {
             return;
         };