diff --git a/flow/src/bin/first_ten_distributed.rs b/flow/src/bin/first_ten_distributed.rs
index ab70f18..d448b63 100644
--- a/flow/src/bin/first_ten_distributed.rs
+++ b/flow/src/bin/first_ten_distributed.rs
@@ -1,7 +1,7 @@
 #[tokio::main]
 async fn main() {
-    hydroflow_plus::util::cli::launch(|ports| {
-        flow::first_ten_distributed::first_ten_distributed_runtime!(&ports)
+    hydroflow_plus::util::cli::launch!(|ports| {
+        flow::first_ten_distributed::first_ten_distributed_runtime!(ports)
     })
     .await;
 }
diff --git a/flow/src/first_ten.rs b/flow/src/first_ten.rs
index c1bf647..e6b325f 100644
--- a/flow/src/first_ten.rs
+++ b/flow/src/first_ten.rs
@@ -9,7 +9,7 @@ pub fn first_ten<'a, D: LocalDeploy<'a>>(
 ) {
     let process = flow.process(process_spec);
 
-    let numbers = process.source_iter(q!(0..10));
+    let numbers = flow.source_iter(&process, q!(0..10));
     numbers.for_each(q!(|n| println!("{}", n)));
 }
 
diff --git a/flow/src/first_ten_distributed.rs b/flow/src/first_ten_distributed.rs
index 4f3b2c8..5ea73aa 100644
--- a/flow/src/first_ten_distributed.rs
+++ b/flow/src/first_ten_distributed.rs
@@ -8,7 +8,7 @@ pub fn first_ten_distributed<'a, D: Deploy<'a>>(
     let process = flow.process(process_spec);
     let second_process = flow.process(process_spec);
 
-    let numbers = process.source_iter(q!(0..10));
+    let numbers = flow.source_iter(&process, q!(0..10));
     numbers
         .send_bincode(&second_process)
         .for_each(q!(|n| println!("{}", n)));
diff --git a/flow/src/lib.rs b/flow/src/lib.rs
index 8aa2800..145246c 100644
--- a/flow/src/lib.rs
+++ b/flow/src/lib.rs
@@ -1,5 +1,11 @@
 stageleft::stageleft_crate!(flow_macro);
 
+#[cfg(stageleft_macro)]
+pub(crate) mod first_ten;
+#[cfg(not(stageleft_macro))]
 pub mod first_ten;
 
+#[cfg(stageleft_macro)]
+pub(crate) mod first_ten_distributed;
+#[cfg(not(stageleft_macro))]
 pub mod first_ten_distributed;
diff --git a/flow_macro/Cargo.toml b/flow_macro/Cargo.toml
index 49ea9f9..09848c7 100644
--- a/flow_macro/Cargo.toml
+++ b/flow_macro/Cargo.toml
@@ -6,7 +6,7 @@ edition = "2021"
 
 [lib]
 proc-macro = true
-path = "src/lib.rs"
+path = "../flow/src/lib.rs"
 
 [features]
 default = ["macro"]
diff --git a/flow_macro/src/lib.rs b/flow_macro/src/lib.rs
deleted file mode 100644
index da4d796..0000000
--- a/flow_macro/src/lib.rs
+++ /dev/null
@@ -1 +0,0 @@
-stageleft::stageleft_macro_crate!();