diff --git a/crates/tinybus/src/connection.rs b/crates/tinybus/src/connection.rs index e637bc35..a2bf84f6 100644 --- a/crates/tinybus/src/connection.rs +++ b/crates/tinybus/src/connection.rs @@ -1581,7 +1581,10 @@ async fn dispatch(inner: &Inner, header: &Header, body: Value, brokered: bool) - }; // Snapshot admission under the lock, then release it before user code: // callbacks can register/unserve interfaces while another call is parked. - let target = inner.objects.read().await.lookup(path, interface)?; + let target = { + let objects = inner.objects.read().await; + objects.lookup(path, interface)? + }; ObjectTree::invoke( target, interface, diff --git a/crates/tinybus/src/connection_tests.rs b/crates/tinybus/src/connection_tests.rs index 426da268..54ca5ed6 100644 --- a/crates/tinybus/src/connection_tests.rs +++ b/crates/tinybus/src/connection_tests.rs @@ -77,6 +77,88 @@ async fn brokered_pair() -> (Connection, Connection) { (client, service) } +struct MutatesObjectTree { + service: Connection, + extra_path: ObjectPath, +} + +#[async_trait] +impl Interface for MutatesObjectTree { + fn name(&self) -> InterfaceName { + InterfaceName::new("ai.tinyhumans.Test").unwrap() + } + + fn members(&self) -> Vec { + vec![ + MemberName::new("Serve").unwrap(), + MemberName::new("Unserve").unwrap(), + ] + } + + async fn call(&self, member: &MemberName, _args: Value) -> Result { + match member.as_str() { + "Serve" => { + self.service.serve_at(self.extra_path.clone(), Echo).await?; + Ok(Value::Bool(true)) + } + "Unserve" => Ok(Value::Bool(self.service.unserve(&self.extra_path).await)), + _ => unreachable!("members() lists only the supported operations"), + } + } +} + +#[tokio::test] +async fn a_service_can_export_and_remove_objects_during_a_method_call() { + let transport = MemoryBus::new(); + Broker::new().spawn(transport.clone()); + let service = Connection::connect(transport.connect().await.unwrap()) + .await + .unwrap(); + service.request_name("ai.tinyhumans.Test").await.unwrap(); + let client = Connection::connect(transport.connect().await.unwrap()) + .await + .unwrap(); + let extra_path = ObjectPath::new("/ai/tinyhumans/TestExtra").unwrap(); + service + .serve_at( + path(), + MutatesObjectTree { + service: service.clone(), + extra_path: extra_path.clone(), + }, + ) + .await + .unwrap(); + let proxy = client + .proxy("ai.tinyhumans.Test", path().as_str(), "ai.tinyhumans.Test") + .unwrap() + .with_timeout(Duration::from_millis(250)); + + assert!(proxy.call::("Serve", ()).await.unwrap()); + let extra = client + .proxy( + "ai.tinyhumans.Test", + extra_path.as_str(), + "ai.tinyhumans.Test", + ) + .unwrap(); + assert_eq!( + extra.call::("Echo", ("served",)).await.unwrap(), + serde_json::json!(["served"]) + ); + + assert!(proxy.call::("Unserve", ()).await.unwrap()); + let removed_call = extra.call::("Echo", ("removed",)).await; + assert!( + matches!( + removed_call, + Err(Error::MethodFailed { ref name, .. }) + if name == "ai.tinyhumans.tinybus.Error.UnknownObject" + ), + "expected removal to reject the object, got {removed_call:?}" + ); +} + #[tokio::test] async fn a_call_reaches_the_service_and_the_reply_comes_back() { let (client, _service) = pair().await;