1 files changed, 29 insertions, 31 deletions

diff --git a/p2p/examples/monitor.rs b/p2p/examples/monitor.rs
index cd4defc..fc48c2f 100644
--- a/p2p/examples/monitor.rs
+++ b/p2p/examples/monitor.rs
@@ -1,13 +1,16 @@
+mod shared;
+
 use std::sync::Arc;
 
 use clap::Parser;
-use easy_parallel::Parallel;
-use smol::{channel, future, Executor};
+use smol::{channel, Executor};
 
 use karyons_net::{Endpoint, Port};
 
 use karyons_p2p::{Backend, Config, PeerID};
 
+use shared::run_executor;
+
 #[derive(Parser)]
 #[command(author, version, about, long_about = None)]
 struct Cli {
@@ -50,44 +53,39 @@ fn main() {
         ..Default::default()
     };
 
+    // Create a new Executor
+    let ex = Arc::new(Executor::new());
+
     // Create a new Backend
-    let backend = Backend::new(peer_id, config);
+    let backend = Backend::new(peer_id, config, ex.clone());
 
     let (ctrlc_s, ctrlc_r) = channel::unbounded();
     let handle = move || ctrlc_s.try_send(()).unwrap();
     ctrlc::set_handler(handle).unwrap();
 
-    let (signal, shutdown) = channel::unbounded::<()>();
-
-    // Create a new Executor
-    let ex = Arc::new(Executor::new());
-
-    let task = async {
-        let monitor = backend.monitor().await;
+    let exc = ex.clone();
+    run_executor(
+        async {
+            let monitor = backend.monitor().await;
 
-        let monitor_task = ex.spawn(async move {
-            loop {
-                let event = monitor.recv().await.unwrap();
-                println!("{}", event);
-            }
-        });
+            let monitor_task = exc.spawn(async move {
+                loop {
+                    let event = monitor.recv().await.unwrap();
+                    println!("{}", event);
+                }
+            });
 
-        // Run the backend
-        backend.run(ex.clone()).await.unwrap();
+            // Run the backend
+            backend.run().await.unwrap();
 
-        // Wait for ctrlc signal
-        ctrlc_r.recv().await.unwrap();
+            // Wait for ctrlc signal
+            ctrlc_r.recv().await.unwrap();
 
-        // Shutdown the backend
-        backend.shutdown().await;
-
-        monitor_task.cancel().await;
-
-        drop(signal);
-    };
+            // Shutdown the backend
+            backend.shutdown().await;
 
-    // Run four executor threads.
-    Parallel::new()
-        .each(0..4, |_| future::block_on(ex.run(shutdown.recv())))
-        .finish(|| future::block_on(task));
+            monitor_task.cancel().await;
+        },
+        ex,
+    );
 }