lock the versions wtf

arbiter-engine/Cargo.toml

@@ -10,8 +10,8 @@ readme = "../README.md"
 
 [dependencies]
 ethers.workspace = true
-artemis-core = { git = "https://github.com/paradigmxyz/artemis.git" }
-# artemis-core = { path = "../../../artemis/crates/artemis-core" }
+# artemis-core = { git = "https://github.com/paradigmxyz/artemis.git" }
+artemis-core = { path = "../../../artemis/crates/artemis-core" }
 futures-util.workspace = true
 async-trait.workspace = true
 serde_json.workspace = true

arbiter-engine/src/agent.rs

@@ -87,15 +87,15 @@ impl Agent {
         self.dependents.push(dependent.to_owned());
     }
 
-    pub(crate) async fn run(&mut self) -> Vec<JoinSet<()>> {
-        let mut join_sets = vec![];
-        for behavior in self.behaviors.iter_mut() {
-            trace!("Running behavior");
-            let joinset = behavior.await.unwrap();
-            join_sets.push(joinset);
-        }
-        join_sets
-    }
+    // pub(crate) async fn run(&mut self) -> Vec<JoinSet<()>> {
+    //     let mut join_sets = vec![];
+    //     for behavior in self.behaviors.iter_mut() {
+    //         trace!("Running behavior");
+    //         let joinset = behavior.await.unwrap();
+    //         join_sets.push(joinset);
+    //     }
+    //     join_sets
+    // }
 }
 
 type Behavior = Pin<Box<dyn Future<Output = Result<JoinSet<()>, Box<dyn Error>>> + Send>>;

arbiter-engine/src/examples/timed_message.rs

@@ -76,7 +76,5 @@ async fn echoer() {
     let send_result = world.messager.execute(message).await;
     debug!("Start message sent {:?}", send_result);
 
-    for task in tasks {
-        task.await.unwrap();
-    }
+    world.join().await;
 }

arbiter-engine/src/examples/token_minter.rs

@@ -313,12 +313,13 @@ async fn token_minter_simulation() {
     requester_agent.add_behavior(mint_behavior);
 
     // Run the world and send the start message
-    let tasks = world.run().await;
+
     let message = Message {
         from: "host".to_owned(),
         to: To::Agent(REQUESTER_ID.to_owned()),
         data: "Start".to_owned(),
     };
+    // TODO: Messages like this could probably be put in the `world.run()`
     world.messager.execute(message).await;
 
     let message = Message {
@@ -328,5 +329,6 @@ async fn token_minter_simulation() {
     };
     world.messager.execute(message).await;
 
-    futures::future::join_all(tasks).await;
+    let tasks = world.run().await;
+    world.join().await;
 }

arbiter-engine/src/world.rs

@@ -15,6 +15,7 @@ use ethers::{
     abi::Hash,
     providers::{Provider, PubsubClient},
 };
+use tokio::task::JoinSet;
 
 use super::*;
 use crate::{
@@ -42,6 +43,8 @@ pub struct World<P> {
 
     /// The messaging layer for the world.
     pub messager: Messager, // TODO: Use this as the message executor that can be given to all agents and give each agent their specific collector.
+
+    pub joinsets: Option<Vec<JoinSet<()>>>,
 }
 
 // TODO: Can add a messager as an interconnect and have the manager give each
@@ -60,6 +63,7 @@ where
             agents: HashMap::new(),
             provider,
             messager: Messager::new(),
+            joinsets: None,
         }
     }
 
@@ -72,27 +76,28 @@ where
     }
 
     /// Runs the agents in the world.
-    pub async fn run(&mut self) -> Vec<tokio::task::JoinHandle<()>> {
+    pub async fn run(&mut self) {
         debug!("Running world: {}", self.id);
         debug!("Agents in world: {:?}", self.agents.keys());
 
-        let mut tasks = Vec::new();
-
+        let mut join_sets = vec![];
         for agent in self.agents.values_mut() {
             trace!("Running agent: {}", agent.id);
-            let join_sets = Box::leak(Box::new(agent.run().await));
-            for set in join_sets.iter_mut() {
-                let task = tokio::spawn(async move {
-                    while let Some(next) = set.join_next().await {
-                        if let Err(e) = next {
-                            panic!("Error: {:?}", e);
-                        }
-                    }
-                });
-                tasks.push(task);
+
+            for behavior in agent.behaviors.iter_mut() {
+                trace!("Running behavior");
+                let joinset = behavior.await.unwrap();
+                join_sets.push(joinset);
             }
         }
-        tasks
+        self.joinsets = Some(join_sets);
+    }
+
+    pub async fn join(&mut self) {
+        std::thread::sleep(std::time::Duration::from_secs(3));
+        for joinset in self.joinsets.as_mut().unwrap().iter_mut() {
+            while let Some(_) = joinset.join_next().await {}
+        }
     }
 }