Update for FutureSDR 0.0.40

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 }

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"

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(())
     }
 }

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,40 +28,35 @@ 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);
 
         match self.target {
             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)),
         };

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(())
 }

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 = "../../" }

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));
     }
 }

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)]

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.
 ///

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.
 ///

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.
 ///

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.
 ///

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, *};
 

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;

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`]

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};

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};

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`.

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<()> {

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());

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<()> {

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());

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: {:?}",

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('_'))