diff --git a/Cargo.toml b/Cargo.toml
index 22ded0a..ddde2c6 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -20,7 +20,7 @@ members = [
 ]
 
 [dependencies]
-futuresdr = { git = "https://github.com/FutureSDR/FutureSDR", branch = "main", version = "0.0.40-dev"}
+futuresdr = { version = "0.0.40" }
 async-channel = { version = "2.5", optional = true }
 crossbeam-channel = { version = "0.5", optional = true }
 bimap = { version = "0.6", optional = true }
diff --git a/crates/sigmf-utilities/Cargo.toml b/crates/sigmf-utilities/Cargo.toml
index 759ae38..a725a4a 100644
--- a/crates/sigmf-utilities/Cargo.toml
+++ b/crates/sigmf-utilities/Cargo.toml
@@ -15,7 +15,7 @@ readme = "README.md"
 anyhow = "1.0"
 clap = { version = "4.5", features = ["derive"] }
 fsdr-blocks = {path="../../"}
-futuresdr = { git = "https://github.com/FutureSDR/FutureSDR", branch = "main" }
+futuresdr = { version = "0.0.40" }
 serde = "^1.0"
 serde_derive = "^1.0"
 serde_json = "^1.0"
diff --git a/crates/sigmf-utilities/src/sigmf_col.rs b/crates/sigmf-utilities/src/sigmf_col.rs
index 8f0a61b..f47407e 100644
--- a/crates/sigmf-utilities/src/sigmf_col.rs
+++ b/crates/sigmf-utilities/src/sigmf_col.rs
@@ -89,7 +89,7 @@ impl Commands {
             // .author(self.author())
             .build()?
             .create_pretty(output)
-            .with_context(|| format!("Error writing to {}", &output.display()))?;
+            .with_context(|| format!("Error writing to {}", output.display()))?;
         Ok(())
     }
 }
diff --git a/crates/sigmf-utilities/src/sigmf_convert.rs b/crates/sigmf-utilities/src/sigmf_convert.rs
index c3a760d..9822c64 100644
--- a/crates/sigmf-utilities/src/sigmf_convert.rs
+++ b/crates/sigmf-utilities/src/sigmf_convert.rs
@@ -6,10 +6,10 @@ use fsdr_blocks::sigmf::{SigMFSinkBuilder, SigMFSourceBuilder};
 use fsdr_blocks::type_converters::TypeConvertersBuilder;
 use futuresdr::blocks::Apply;
 use futuresdr::blocks::TagDebug;
-use futuresdr::macros::connect;
 use futuresdr::runtime::Flowgraph;
 use futuresdr::runtime::Result;
 use futuresdr::runtime::Runtime;
+use futuresdr::runtime::macros::connect;
 use std::path::PathBuf;
 
 #[derive(Parser)]
@@ -28,7 +28,7 @@ impl Cli {
         let mut fg = Flowgraph::new();
 
         let mut src_builder = SigMFSourceBuilder::from(&self.input);
-        let src = fg.add_block(src_builder.build::<f32>().await?);
+        let src = fg.add(src_builder.build::<f32>().await?);
 
         let snk = SigMFSinkBuilder::from(self.output);
 
@@ -36,32 +36,27 @@ impl Cli {
             RI8 => {
                 let conv = TypeConvertersBuilder::lossy_scale_convert_f32_i8().build();
                 let snk = snk.datatype(self.target).build::<i8>().await?;
-                let src_ref = src.clone();
-                connect!(fg, src_ref > conv > snk);
+                connect!(fg, src > conv > snk);
             }
             RU8 => {
                 let conv = TypeConvertersBuilder::lossy_scale_convert_f32_u8().build();
                 let snk = snk.datatype(self.target).build::<u8>().await?;
-                let src_ref = src.clone();
-                connect!(fg, src_ref > conv > snk);
+                connect!(fg, src > conv > snk);
             }
             Rf32Be | Rf32Le => {
                 let conv: Apply<fn(&f32) -> f32, f32, f32> = Apply::new(|x: &f32| *x);
                 let snk = snk.datatype(self.target).build::<f32>().await?;
-                let src_ref = src.clone();
-                connect!(fg, src_ref > conv > snk);
+                connect!(fg, src > conv > snk);
             }
             Rf64Be | Rf64Le => {
                 let conv = TypeConvertersBuilder::convert::<f32, f64>().build();
                 let snk = snk.datatype(self.target).build::<f64>().await?;
-                let src_ref = src.clone();
-                connect!(fg, src_ref > conv > snk);
+                connect!(fg, src > conv > snk);
             }
             Ri16Be | Ri16Le => {
                 let conv = TypeConvertersBuilder::lossy_scale_convert_f32_i16().build();
                 let snk = snk.datatype(self.target).build::<i16>().await?;
-                let src_ref = src.clone();
-                connect!(fg, src_ref > conv > snk);
+                connect!(fg, src > conv > snk);
             }
             _ => return Err(anyhow!("Unsupported target type: {}", self.target)),
         };
diff --git a/crates/sigmf-utilities/src/sigmf_hash.rs b/crates/sigmf-utilities/src/sigmf_hash.rs
index 05f684b..a18c26c 100644
--- a/crates/sigmf-utilities/src/sigmf_hash.rs
+++ b/crates/sigmf-utilities/src/sigmf_hash.rs
@@ -88,7 +88,7 @@ fn update_sigmf(basename: PathBuf) -> Result<()> {
         basename.set_extension("sigmf-meta");
         desc.global_mut()?.sha512 = Some(computed_sha512);
         desc.create_pretty(&basename)
-            .with_context(|| format!("Error writing to {}", &basename.display()))?;
+            .with_context(|| format!("Error writing to {}", basename.display()))?;
     }
     Ok(())
 }
diff --git a/examples/agc/Cargo.toml b/examples/agc/Cargo.toml
index a4cbb2e..8e8377f 100644
--- a/examples/agc/Cargo.toml
+++ b/examples/agc/Cargo.toml
@@ -8,5 +8,5 @@ edition = "2024"
 #path = "main"
 
 [dependencies]
-futuresdr = { git = "https://github.com/FutureSDR/FutureSDR", branch = "main", version = "0.0.40-dev", features = ["audio"] }
+futuresdr = { version = "0.0.40", features = ["audio"] }
 fsdr-blocks = { path = "../../" }
diff --git a/examples/agc/src/main.rs b/examples/agc/src/main.rs
index 8468ed4..131d71a 100644
--- a/examples/agc/src/main.rs
+++ b/examples/agc/src/main.rs
@@ -1,5 +1,4 @@
 use fsdr_blocks::agc::AgcBuilder;
-use futuresdr::async_io;
 use futuresdr::blocks::{Combine, SignalSourceBuilder, audio::AudioSink};
 use futuresdr::prelude::*;
 use std::thread::sleep;
@@ -23,42 +22,42 @@ fn main() -> Result<()> {
         .build();
 
     // Audiosink to output the modulated tone
-    let audio_snk = AudioSink::new(48_000, 1);
+    let audio_snk = AudioSink::new(48_000, 1)?;
 
     connect!(fg,
              src > in0.combine;
              gain_change > in1.combine;
              combine > agc > audio_snk;
     );
-    let agc = agc.get()?.id;
+    let agc = agc.id();
 
     // Start the flowgraph and save the handle
     let rt = Runtime::new();
-    let (_res, mut handle) = rt.start_sync(fg)?;
+    let handle = rt.start(fg)?;
 
     // Keep changing gain and gain lock.
     loop {
         // Reference power of 1.0 is the power level we want to achieve
         println!("Setting reference power to 1.0");
-        async_io::block_on(handle.call(agc, "reference_power", Pmt::F32(1.0)))?;
+        Runtime::block_on(handle.call(agc, "reference_power", Pmt::F32(1.0)))?;
 
         // A high max gain allows to amplify a signal
         println!("Setting Max Gain to 65536.0");
-        async_io::block_on(handle.call(agc, "max_gain", Pmt::F32(65536.0)))?;
+        Runtime::block_on(handle.call(agc, "max_gain", Pmt::F32(65536.0)))?;
         sleep(Duration::from_secs(5));
 
         // Setting a gain lock prevents gain changes from happening
         println!("Setting gain lock for 5s");
-        async_io::block_on(handle.call(agc, "gain_lock", Pmt::Bool(true)))?;
+        Runtime::block_on(handle.call(agc, "gain_lock", Pmt::Bool(true)))?;
 
         // Audio should get quiet faster, but gain is still locked here. It will be released after 5 seconds.
         println!("Setting reference power to 0.2");
-        async_io::block_on(handle.call(agc, "reference_power", Pmt::F32(0.2)))?;
+        Runtime::block_on(handle.call(agc, "reference_power", Pmt::F32(0.2)))?;
         sleep(Duration::from_secs(5));
 
         // Gain lock released! Audio should get more quiet here for 10 seconds
         println!("Releasing gain lock");
-        async_io::block_on(handle.call(agc, "gain_lock", Pmt::Bool(false)))?;
+        Runtime::block_on(handle.call(agc, "gain_lock", Pmt::Bool(false)))?;
         sleep(Duration::from_secs(10));
     }
 }
diff --git a/src/agc.rs b/src/agc.rs
index 484cfde..b6ca3ab 100644
--- a/src/agc.rs
+++ b/src/agc.rs
@@ -1,5 +1,5 @@
 use futuresdr::num_complex::ComplexFloat;
-use futuresdr::prelude::*;
+use futuresdr::runtime::dev::prelude::*;
 
 /// Automatic Gain Control Block
 #[derive(Block)]
diff --git a/src/async_channel/async_channel_sink.rs b/src/async_channel/async_channel_sink.rs
index 8b65196..2652c81 100644
--- a/src/async_channel/async_channel_sink.rs
+++ b/src/async_channel/async_channel_sink.rs
@@ -1,5 +1,5 @@
 use async_channel::Sender;
-use futuresdr::prelude::*;
+use futuresdr::runtime::dev::prelude::*;
 
 /// Push samples originating from a stream in a flowgraph into an async channel.
 ///
diff --git a/src/async_channel/async_channel_source.rs b/src/async_channel/async_channel_source.rs
index 680f35e..945e390 100644
--- a/src/async_channel/async_channel_source.rs
+++ b/src/async_channel/async_channel_source.rs
@@ -1,5 +1,5 @@
 use async_channel::Receiver;
-use futuresdr::prelude::*;
+use futuresdr::runtime::dev::prelude::*;
 
 /// Push samples through a channel into a stream connection.
 ///
diff --git a/src/channel/crossbeam_sink.rs b/src/channel/crossbeam_sink.rs
index e5ba96f..0be23b2 100644
--- a/src/channel/crossbeam_sink.rs
+++ b/src/channel/crossbeam_sink.rs
@@ -1,5 +1,5 @@
 use crossbeam_channel::Sender;
-use futuresdr::prelude::*;
+use futuresdr::runtime::dev::prelude::*;
 
 /// Push samples originating from a stream in a flowgraph into a crossbeam channel.
 ///
diff --git a/src/channel/crossbeam_source.rs b/src/channel/crossbeam_source.rs
index 068cafa..46b2226 100644
--- a/src/channel/crossbeam_source.rs
+++ b/src/channel/crossbeam_source.rs
@@ -1,5 +1,5 @@
 use crossbeam_channel::Receiver;
-use futuresdr::prelude::*;
+use futuresdr::runtime::dev::prelude::*;
 
 /// Pull samples from a crossbeam channel into a stream in a flowgraph.
 ///
diff --git a/src/cw/baseband_to_cw.rs b/src/cw/baseband_to_cw.rs
index 36312a9..2631d79 100644
--- a/src/cw/baseband_to_cw.rs
+++ b/src/cw/baseband_to_cw.rs
@@ -1,6 +1,6 @@
 use std::ops::RangeInclusive;
 
-use futuresdr::prelude::*;
+use futuresdr::runtime::dev::prelude::*;
 
 use crate::cw::shared::CWAlphabet::{self, *};
 
diff --git a/src/cw/cw_to_char.rs b/src/cw/cw_to_char.rs
index 2ff1683..2cc7db6 100644
--- a/src/cw/cw_to_char.rs
+++ b/src/cw/cw_to_char.rs
@@ -1,4 +1,4 @@
-use futuresdr::prelude::*;
+use futuresdr::runtime::dev::prelude::*;
 
 use crate::cw::shared::CWAlphabet::{self, LetterSpace, WordSpace};
 use crate::cw::shared::get_alphabet;
diff --git a/src/math/freq_shift.rs b/src/math/freq_shift.rs
index ea31957..469adeb 100644
--- a/src/math/freq_shift.rs
+++ b/src/math/freq_shift.rs
@@ -1,7 +1,7 @@
 use futuresdr::blocks::signal_source::FixedPointPhase;
 use futuresdr::blocks::signal_source::NCO;
 use futuresdr::num_complex::Complex32;
-use futuresdr::prelude::*;
+use futuresdr::runtime::dev::prelude::*;
 
 /// This blocks shift the signal in the frequency domain based on the [`NCO`] implementation.
 /// Currently implemented only for float and [`Complex32`]
diff --git a/src/sigmf/sigmf_sink.rs b/src/sigmf/sigmf_sink.rs
index cf5cb23..6b43c92 100644
--- a/src/sigmf/sigmf_sink.rs
+++ b/src/sigmf/sigmf_sink.rs
@@ -2,7 +2,7 @@ use std::ffi::OsStr;
 use std::io::Write;
 use std::path::PathBuf;
 
-use futuresdr::prelude::*;
+use futuresdr::runtime::dev::prelude::*;
 
 use sigmf::Annotation;
 use sigmf::{DatasetFormat, DescriptionBuilder};
diff --git a/src/sigmf/sigmf_source.rs b/src/sigmf/sigmf_source.rs
index 38774fa..7592ed5 100644
--- a/src/sigmf/sigmf_source.rs
+++ b/src/sigmf/sigmf_source.rs
@@ -3,7 +3,7 @@ use std::path::PathBuf;
 
 use futuresdr::futures::AsyncRead;
 use futuresdr::futures::AsyncReadExt;
-use futuresdr::prelude::*;
+use futuresdr::runtime::dev::prelude::*;
 
 use sigmf::RecordingBuilder;
 use sigmf::{Annotation, Description};
diff --git a/src/stream/deinterleave.rs b/src/stream/deinterleave.rs
index 389c3d6..e7c02c9 100644
--- a/src/stream/deinterleave.rs
+++ b/src/stream/deinterleave.rs
@@ -1,4 +1,4 @@
-use futuresdr::prelude::*;
+use futuresdr::runtime::dev::prelude::*;
 
 /// This blocks deinterleave a unique stream into two separate stream.
 /// Typically used to deinterleave iq stream into of stream for `i` and one for `q`.
diff --git a/tests/async_channel/async_channel_sink.rs b/tests/async_channel/async_channel_sink.rs
index 0ef9a3e..4d7cd85 100644
--- a/tests/async_channel/async_channel_sink.rs
+++ b/tests/async_channel/async_channel_sink.rs
@@ -1,9 +1,9 @@
 use fsdr_blocks::async_channel::AsyncChannelSink;
 use futuresdr::blocks::VectorSource;
-use futuresdr::macros::connect;
 use futuresdr::runtime::Flowgraph;
 use futuresdr::runtime::Result;
 use futuresdr::runtime::Runtime;
+use futuresdr::runtime::macros::connect;
 
 #[test]
 fn run_async_channel_sink_f32() -> Result<()> {
diff --git a/tests/async_channel/async_channel_source.rs b/tests/async_channel/async_channel_source.rs
index 704ad08..8d93ef6 100644
--- a/tests/async_channel/async_channel_source.rs
+++ b/tests/async_channel/async_channel_source.rs
@@ -1,7 +1,7 @@
 use fsdr_blocks::async_channel::AsyncChannelSource;
 use futuresdr::blocks::{Head, VectorSink};
-use futuresdr::macros::connect;
 use futuresdr::runtime::Result;
+use futuresdr::runtime::macros::connect;
 use futuresdr::runtime::{Flowgraph, Runtime};
 
 #[test]
@@ -25,10 +25,10 @@ async fn async_channel_source_u32() -> Result<()> {
     tx.send(orig.clone().into_boxed_slice()).await.unwrap();
     tx.close();
 
-    Runtime::new().run(fg)?;
+    let fg = Runtime::new().run(fg)?;
 
-    let snk = vector_snk.get()?;
-    let received = snk.items();
+    let binding = vector_snk.get(&fg)?;
+    let received = binding.items();
 
     // debug!("{}", received.len());
     // debug!("{}", orig.len());
diff --git a/tests/channel/crossbeam_sink.rs b/tests/channel/crossbeam_sink.rs
index d284d46..67cee5e 100644
--- a/tests/channel/crossbeam_sink.rs
+++ b/tests/channel/crossbeam_sink.rs
@@ -1,9 +1,9 @@
 use fsdr_blocks::channel::CrossbeamSink;
 use futuresdr::blocks::VectorSource;
-use futuresdr::macros::connect;
 use futuresdr::runtime::Flowgraph;
 use futuresdr::runtime::Result;
 use futuresdr::runtime::Runtime;
+use futuresdr::runtime::macros::connect;
 
 #[test]
 fn crossbeam_sink_f32() -> Result<()> {
diff --git a/tests/channel/crossbeam_source.rs b/tests/channel/crossbeam_source.rs
index c090d6a..ad0b37d 100644
--- a/tests/channel/crossbeam_source.rs
+++ b/tests/channel/crossbeam_source.rs
@@ -1,7 +1,7 @@
 use fsdr_blocks::channel::CrossbeamSource;
 use futuresdr::blocks::{Head, VectorSink};
-use futuresdr::macros::connect;
 use futuresdr::runtime::Result;
+use futuresdr::runtime::macros::connect;
 use futuresdr::runtime::{Flowgraph, Runtime};
 
 #[test]
@@ -20,10 +20,10 @@ fn crossbeam_source_u32() -> Result<()> {
 
     tx.try_send(orig.clone().into_boxed_slice()).unwrap();
 
-    Runtime::new().run(fg)?;
+    let fg = Runtime::new().run(fg)?;
 
-    let snk = vector_sink.get()?;
-    let received = snk.items();
+    let binding = vector_sink.get(&fg)?;
+    let received = binding.items();
 
     // debug!("{}", received.len());
     // debug!("{}", orig.len());
diff --git a/tests/cw/baseband_to_cw.rs b/tests/cw/baseband_to_cw.rs
index d369cf1..41564ec 100644
--- a/tests/cw/baseband_to_cw.rs
+++ b/tests/cw/baseband_to_cw.rs
@@ -2,8 +2,8 @@ use fsdr_blocks::cw::baseband_to_cw::BaseBandToCWBuilder;
 use fsdr_blocks::cw::shared::CWAlphabet::*;
 use fsdr_blocks::cw::shared::{CWAlphabet, char_to_baseband};
 use futuresdr::blocks::{VectorSink, VectorSource};
-use futuresdr::macros::connect;
 use futuresdr::runtime::Result;
+use futuresdr::runtime::macros::connect;
 use futuresdr::runtime::{Flowgraph, Runtime};
 
 // cargo nextest run test_baseband_to_cw --no-capture
@@ -32,10 +32,10 @@ fn test_baseband_to_cw() -> Result<()> {
         vector_src > baseband_to_cw > vector_snk;
     );
 
-    Runtime::new().run(fg)?;
+    let fg = Runtime::new().run(fg)?;
 
-    let snk = vector_snk.get()?;
-    let received = snk.items();
+    let binding = vector_snk.get(&fg)?;
+    let received = binding.items();
 
     println!(
         "CW-Alphabet Vector Length: {}, Content: {:?}",
diff --git a/tests/cw/cw_to_char.rs b/tests/cw/cw_to_char.rs
index 2f7100d..33d101e 100644
--- a/tests/cw/cw_to_char.rs
+++ b/tests/cw/cw_to_char.rs
@@ -1,10 +1,9 @@
 use fsdr_blocks::cw::cw_to_char::CWToCharBuilder;
 use fsdr_blocks::cw::shared::{CWAlphabet, msg_to_cw};
-use futuresdr::async_io::block_on;
 use futuresdr::blocks::{ChannelSource, VectorSink, VectorSource};
-use futuresdr::futures::SinkExt;
-use futuresdr::macros::connect;
 use futuresdr::runtime::Result;
+use futuresdr::runtime::channel::mpsc;
+use futuresdr::runtime::macros::connect;
 use futuresdr::runtime::{Flowgraph, Runtime};
 
 // cargo test --features="cw"
@@ -26,10 +25,10 @@ fn test_cw_to_char_vector() -> Result<()> {
         cw_to_char > vector_snk;
     );
 
-    Runtime::new().run(fg)?;
+    let fg = Runtime::new().run(fg)?;
 
-    let snk = vector_snk.get()?;
-    let received: Vec<char> = snk
+    let binding = vector_snk.get(&fg)?;
+    let received: Vec<char> = binding
         .items()
         .iter()
         .map(|&c| char::from_u32(c).unwrap_or('_'))
@@ -50,7 +49,7 @@ fn test_cw_to_char_vector() -> Result<()> {
 fn test_cw_to_char_channel() -> Result<()> {
     let mut fg = Flowgraph::new();
 
-    let (mut tx, rx) = futuresdr::futures::channel::mpsc::channel::<Box<[CWAlphabet]>>(10);
+    let (tx, rx) = mpsc::channel::<Box<[CWAlphabet]>>(10);
 
     let channel_src = ChannelSource::<CWAlphabet>::new(rx);
     let cw_to_char = CWToCharBuilder::new().build();
@@ -61,8 +60,9 @@ fn test_cw_to_char_channel() -> Result<()> {
     );
 
     let rt = Runtime::new();
-    let _fg = block_on(async move {
-        let (fg, _) = rt.start(fg).await.unwrap();
+    let running = rt.start(fg)?;
+
+    Runtime::block_on(async move {
         let c = msg_to_cw(['S'].as_slice()).into_boxed_slice();
         tx.send(c).await.unwrap();
         let c = msg_to_cw([' '].as_slice()).into_boxed_slice();
@@ -76,11 +76,12 @@ fn test_cw_to_char_channel() -> Result<()> {
         let c = msg_to_cw(['S'].as_slice()).into_boxed_slice();
         tx.send(c).await.unwrap();
         tx.close().await.unwrap();
-        fg.await // as Result<Flowgraph>
-    })?;
+    });
+
+    let fg = running.wait()?;
 
-    let snk = vector_snk.get()?;
-    let received: Vec<char> = snk
+    let binding = vector_snk.get(&fg)?;
+    let received: Vec<char> = binding
         .items()
         .iter()
         .map(|&c| char::from_u32(c).unwrap_or('_'))
diff --git a/tests/math/freq_shift.rs b/tests/math/freq_shift.rs
index 9dae7cd..2eff67d 100644
--- a/tests/math/freq_shift.rs
+++ b/tests/math/freq_shift.rs
@@ -1,11 +1,11 @@
 use fsdr_blocks::math::FrequencyShifter;
 use futuresdr::blocks::VectorSink;
 use futuresdr::blocks::VectorSource;
-use futuresdr::macros::connect;
 use futuresdr::num_complex::Complex32;
 use futuresdr::runtime::Flowgraph;
 use futuresdr::runtime::Result;
 use futuresdr::runtime::Runtime;
+use futuresdr::runtime::macros::connect;
 
 #[test]
 fn freq_shift_f32() -> Result<()> {
@@ -20,18 +20,14 @@ fn freq_shift_f32() -> Result<()> {
     connect!(fg,
         src > freq_shifter > vect_sink;
     );
-    Runtime::new().run(fg)?;
+    let fg = Runtime::new().run(fg)?;
 
-    let snk_0 = vect_sink.get()?;
-    let snk_0 = snk_0.items();
+    let binding = vect_sink.get(&fg)?;
+    let snk_0 = binding.items();
 
     assert_eq!(snk_0.len(), orig.len());
     let expected: Vec<f32> = vec![0.0, 0.0, -2.0, 0.0, 4.0, 0.0, -6.0, 0.0, 8.0, 0.0, -10.0];
 
-    // assert!(snk_0
-    //     .iter()
-    //     .zip(expected.iter())
-    //     .all(|(v, e)| (*v - *e).abs() < 0.0000001));
     for (v, e) in snk_0.iter().zip(expected.iter()) {
         assert!(
             (v - e).abs() < 0.0000001,
@@ -68,10 +64,10 @@ fn freq_shift_c32() -> Result<()> {
     connect!(fg,
         src > freq_shifter > vect_sink;
     );
-    Runtime::new().run(fg)?;
+    let fg = Runtime::new().run(fg)?;
 
-    let snk_0 = vect_sink.get()?;
-    let snk_0 = snk_0.items();
+    let binding = vect_sink.get(&fg)?;
+    let snk_0 = binding.items();
 
     assert_eq!(snk_0.len(), orig.len());
     for (i, (v, e)) in snk_0.iter().zip(expected.iter()).enumerate() {
diff --git a/tests/sigmf/sigmf_graph.rs b/tests/sigmf/sigmf_graph.rs
index 8a3f989..8ebc786 100644
--- a/tests/sigmf/sigmf_graph.rs
+++ b/tests/sigmf/sigmf_graph.rs
@@ -1,8 +1,8 @@
 use fsdr_blocks::sigmf::{BytesConveter, SigMFSink, SigMFSourceBuilder};
 use futuresdr::{
     blocks::{VectorSink, VectorSource},
-    macros::connect,
     runtime::Result,
+    runtime::macros::connect,
     runtime::{Flowgraph, Runtime},
 };
 
@@ -37,11 +37,11 @@ where
     connect!(fg,
         src1 > snk1;
     );
-    Runtime::new().run(fg)?;
-    let snk1 = snk1.get()?;
-    let desc = snk1.description.build()?;
+    let fg = Runtime::new().run(fg)?;
+    let snk1_guard = snk1.get(&fg)?;
+    let desc = snk1_guard.description.build()?;
+    let data_file = snk1_guard.writer.to_owned().into_inner();
     let mut fg = Flowgraph::new();
-    let data_file = snk1.writer.to_owned().into_inner();
     let data_file = futuresdr::futures::io::Cursor::new(data_file);
     let src2 = futuresdr::futures::executor::block_on(
         SigMFSourceBuilder::with_data_and_description(data_file, desc).build::<T>(),
@@ -50,8 +50,8 @@ where
     connect!(fg,
         src2 > snk2;
     );
-    Runtime::new().run(fg)?;
-    let snk2 = snk2.get()?.items().clone();
+    let fg = Runtime::new().run(fg)?;
+    let snk2 = snk2.get(&fg)?.items().clone();
     assert_eq!(data.len(), snk2.len());
     for (o, i) in data.iter().zip(snk2) {
         assert_eq!(o, &i);
@@ -116,11 +116,10 @@ fn sigmf_read_write_annotation() -> Result<()> {
         src1 > snk1;
     );
     // Now run the flowgraph
-    Runtime::new().run(fg)?;
+    let fg = Runtime::new().run(fg)?;
 
     // Time to verify
-    let snk1 = snk1.get()?;
-    let tgt_desc = snk1.description.build()?;
+    let tgt_desc = snk1.get(&fg)?.description.build()?;
     let annotations = tgt_desc.annotations()?;
     assert_eq!(2, annotations.len());
     let annot1 = annotations
diff --git a/tests/sigmf/sigmf_source.rs b/tests/sigmf/sigmf_source.rs
index dc074df..736cebf 100644
--- a/tests/sigmf/sigmf_source.rs
+++ b/tests/sigmf/sigmf_source.rs
@@ -3,11 +3,11 @@ use fsdr_blocks::sigmf::SigMFSourceBuilder;
 use futuresdr::blocks::VectorSink;
 use futuresdr::futures::io::BufReader;
 use futuresdr::futures::io::Cursor;
-use futuresdr::macros::connect;
 use futuresdr::num_complex::Complex;
 use futuresdr::runtime::Flowgraph;
 use futuresdr::runtime::Result;
 use futuresdr::runtime::Runtime;
+use futuresdr::runtime::macros::connect;
 use sigmf::DatasetFormat;
 use sigmf::DescriptionBuilder;
 
@@ -36,10 +36,10 @@ where
         src > snk;
     );
 
-    Runtime::new().run(fg)?;
+    let fg = Runtime::new().run(fg)?;
 
-    let snk = snk.get()?;
-    Ok(snk.items().clone())
+    let binding = snk.get(&fg)?;
+    Ok(binding.items().clone())
 }
 
 #[test]
diff --git a/tests/stream/deinterleave.rs b/tests/stream/deinterleave.rs
index d60f284..f7d8de6 100644
--- a/tests/stream/deinterleave.rs
+++ b/tests/stream/deinterleave.rs
@@ -1,10 +1,10 @@
 use fsdr_blocks::stream::*;
 use futuresdr::blocks::VectorSink;
 use futuresdr::blocks::VectorSource;
-use futuresdr::macros::connect;
 use futuresdr::runtime::Flowgraph;
 use futuresdr::runtime::Result;
 use futuresdr::runtime::Runtime;
+use futuresdr::runtime::macros::connect;
 
 #[test]
 fn deinterleave_u8() -> Result<()> {
@@ -22,13 +22,13 @@ fn deinterleave_u8() -> Result<()> {
         deinterleaver.out0 > vect_sink_0;
         deinterleaver.out1 > vect_sink_1;
     );
-    Runtime::new().run(fg)?;
+    let fg = Runtime::new().run(fg)?;
 
-    let snk_0 = vect_sink_0.get()?;
-    let snk_0 = snk_0.items();
+    let binding_0 = vect_sink_0.get(&fg)?;
+    let snk_0 = binding_0.items();
 
-    let snk_1 = vect_sink_1.get()?;
-    let snk_1 = snk_1.items();
+    let binding_1 = vect_sink_1.get(&fg)?;
+    let snk_1 = binding_1.items();
 
     assert_eq!(snk_0.len(), orig.len() / 2);
     assert_eq!(snk_0.len(), snk_1.len());
diff --git a/tests/type_converters.rs b/tests/type_converters.rs
index ba1cdf5..8c494b8 100644
--- a/tests/type_converters.rs
+++ b/tests/type_converters.rs
@@ -1,10 +1,10 @@
 use fsdr_blocks::type_converters::*;
 use futuresdr::blocks::VectorSink;
 use futuresdr::blocks::VectorSource;
-use futuresdr::macros::connect;
 use futuresdr::runtime::Flowgraph;
 use futuresdr::runtime::Result;
 use futuresdr::runtime::Runtime;
+use futuresdr::runtime::macros::connect;
 
 #[test]
 fn convert_u8_f32() -> Result<()> {
@@ -19,10 +19,10 @@ fn convert_u8_f32() -> Result<()> {
     connect!(fg,
         src > convert_u8_f32 > vect_sink;
     );
-    Runtime::new().run(fg)?;
+    let fg = Runtime::new().run(fg)?;
 
-    let snk = vect_sink.get()?;
-    let v = snk.items();
+    let binding = vect_sink.get(&fg)?;
+    let v = binding.items();
 
     assert_eq!(v.len(), orig.len());
     for (v_before, v_after) in orig.iter().zip(v) {