Fix existing tests

Signed-off-by: Michael X. Grey <mxgrey@intrinsic.ai>

Author: mxgrey

src/callback.rs

@@ -18,8 +18,8 @@
 use crate::{
     BlockingCallback, AsyncCallback, Channel, InnerChannel, ChannelQueue,
     OperationRoster, StreamPack, Input, Provider, ProvideOnce,
-    AddOperation, OperateCallback, ManageInput, OperationError, SetupFailure,
-    OrBroken, OperateTask, Operation, OperationSetup,
+    AddOperation, OperateCallback, ManageInput, OperationError,
+    OrBroken, OperateTask,
 };
 
 use bevy::{

src/channel.rs

@@ -160,6 +160,7 @@ impl<T: Stream> StreamChannel<T> {
 mod tests {
     use crate::{*, testing::*};
     use bevy::ecs::system::EntityCommands;
+    use std::time::Duration;
 
     #[test]
     fn test_channel_request() {
@@ -192,8 +193,14 @@ mod tests {
                     repeat,
                 ).take().response
             });
-            context.run_while_pending(&mut promise);
+
+            context.run_with_conditions(
+                &mut promise,
+                FlushConditions::new().with_timeout(Duration::from_secs(5)),
+            );
+
             assert!(promise.peek().is_available());
+            assert!(context.no_unhandled_errors());
         }
 
         let count = context.app.world.get::<RunCount>(hello.provider()).unwrap().0;

src/flush.rs

@@ -131,7 +131,14 @@ fn collect_from_channels(
         // TODO(@mxgrey): Make sure this works for services which are spawned by
         // providers that are being flushed.
         for (e, mut hook) in new_service_query.iter_mut(world) {
-            hook.lifecycle = Some(ServiceLifecycle::new(e, lifecycles.sender.clone()));
+            if hook.lifecycle.is_none() {
+                // Check if the lifecycle is none, because collect_from_channels
+                // can be run multiple times per flush, in which case we will
+                // iterate over the query again, and end up dropping the lifecycle
+                // managers that we just created. When that happens, the service
+                // gets treated as despawned prematurely.
+                hook.lifecycle = Some(ServiceLifecycle::new(e, lifecycles.sender.clone()));
+            }
         }
     });
 

src/impulse.rs

@@ -89,7 +89,7 @@ where
     pub fn take(self) -> Recipient<Response, Streams> {
         let (response_sender, response_promise) = Promise::<Response>::new();
         self.commands.add(AddImpulse::new(
-            dbg!(self.target),
+            self.target,
             TakenResponse::<Response>::new(response_sender),
         ));
         let mut map = StreamTargetMap::default();
@@ -107,7 +107,7 @@ where
     pub fn take_response(self) -> Promise<Response> {
         let (response_sender, response_promise) = Promise::<Response>::new();
         self.commands.add(AddImpulse::new(
-            dbg!(self.target),
+            self.target,
             TakenResponse::<Response>::new(response_sender),
         ));
         response_promise
@@ -327,39 +327,115 @@ impl<T> Default for Collection<T> {
 #[cfg(test)]
 mod tests {
     use crate::{*, testing::*};
-    use std::time::Duration;
+    use std::time::{Instant, Duration};
 
     #[test]
-    fn test_provide() {
+    fn test_blocking_map() {
         let mut context = TestingContext::minimal_plugins();
 
         let mut promise = context.build(|commands| {
-            commands.provide("hello".to_owned()).take_response()
+            commands
+            .request("hello".to_owned(), to_uppercase.into_blocking_map())
+            .take_response()
         });
-        assert!(promise.peek().available().is_some_and(|v| v == "hello"));
+
+        context.run_while_pending(&mut promise);
+        assert!(promise.peek().available().is_some_and(|v| v == "HELLO"));
+        assert!(context.no_unhandled_errors());
+
+        let mut promise = context.build(|commands| {
+            commands
+            .request("hello".to_owned(), to_uppercase.into_blocking_map_once())
+            .take_response()
+        });
+
+        context.run_while_pending(&mut promise);
+        assert!(promise.peek().available().is_some_and(|v| v == "HELLO"));
+        assert!(context.no_unhandled_errors());
+
+        let mut promise = context.build(|commands| {
+            commands
+            .provide("hello".to_owned())
+            .map_block(to_uppercase)
+            .take_response()
+        });
+
+        context.run_while_pending(&mut promise);
+        assert!(promise.peek().available().is_some_and(|v| v == "HELLO"));
+        assert!(context.no_unhandled_errors());
+
+        let mut promise = context.build(|commands| {
+            commands
+            .provide("hello".to_owned())
+            .map_block(|request| request.to_uppercase())
+            .take_response()
+        });
+
+        context.run_while_pending(&mut promise);
+        assert!(promise.peek().available().is_some_and(|v| v == "HELLO"));
+        assert!(context.no_unhandled_errors());
     }
 
     #[test]
     fn test_async_map() {
         let mut context = TestingContext::minimal_plugins();
 
+        let request = WaitRequest {
+            duration: Duration::from_secs_f64(0.001),
+            value: "hello".to_owned(),
+        };
+
+        let conditions = FlushConditions::new()
+            .with_timeout(Duration::from_secs_f64(5.0));
+
         let mut promise = context.build(|commands| {
             commands
-            .request(
-                WaitRequest {
-                    duration: Duration::from_secs_f64(0.001),
-                    value: "hello".to_owned(),
-                },
-                wait.into_async_map(),
-            )
+            .request(request.clone(), wait.into_async_map())
             .take_response()
         });
 
-        context.run_with_conditions(
-            &mut promise,
-            FlushConditions::new()
-            .with_timeout(Duration::from_secs_f64(5.0)),
-        );
+        assert!(context.run_with_conditions(&mut promise, conditions.clone()));
+        assert!(promise.peek().available().is_some_and(|v| v == "hello"));
+        assert!(context.no_unhandled_errors());
+
+        let mut promise = context.build(|commands| {
+            commands
+            .request(request.clone(), wait.into_async_map_once())
+            .take_response()
+        });
+
+        assert!(context.run_with_conditions(&mut promise, conditions.clone()));
+        assert!(promise.peek().available().is_some_and(|v| v == "hello"));
+        assert!(context.no_unhandled_errors());
+
+        let mut promise = context.build(|commands| {
+            commands
+            .provide(request.clone())
+            .map_async(wait)
+            .take_response()
+        });
+
+        assert!(context.run_with_conditions(&mut promise, conditions.clone()));
+        assert!(promise.peek().available().is_some_and(|v| v == "hello"));
+        assert!(context.no_unhandled_errors());
+
+        let mut promise = context.build(|commands| {
+            commands
+            .provide(request.clone())
+            .map_async(|request| {
+                async move {
+                    let t = Instant::now();
+                    while t.elapsed() < request.duration {
+                        // Busy wait
+                    }
+                    request.value
+                }
+            })
+            .take_response()
+        });
+
+        assert!(context.run_with_conditions(&mut promise, conditions.clone()));
         assert!(promise.peek().available().is_some_and(|v| v == "hello"));
+        assert!(context.no_unhandled_errors());
     }
 }

src/operation/operate_map.rs

@@ -190,7 +190,7 @@ where
 
         let task_source = world.spawn(()).id();
         OperateTask::new(
-            dbg!(task_source), session, source, dbg!(target), task, None, sender
+            task_source, session, source, target, task, None, sender,
         ).add(world, roster);
         Ok(())
     }

src/request.rs

@@ -97,9 +97,9 @@ impl<'w, 's> RequestExt<'w, 's> for Commands<'w, 's> {
             .set_parent(target)
             .id();
 
-        provider.connect(source, dbg!(target), self);
+        provider.connect(source, target, self);
 
-        self.add(InputCommand { session: dbg!(source), target: dbg!(source), data: request });
+        self.add(InputCommand { session: source, target: source, data: request });
 
         Impulse {
             source,

src/testing.rs

@@ -210,6 +210,7 @@ pub fn to_uppercase(value: String) -> String {
     value.to_uppercase()
 }
 
+#[derive(Clone, Copy)]
 pub struct WaitRequest<Value> {
     pub duration: std::time::Duration,
     pub value: Value