futuresdr 0.0.40

An Experimental Async SDR Runtime for Heterogeneous Architectures.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87

use anyhow::Result;
use anyhow::anyhow;
use futuresdr::blocks::ChannelSource;
use futuresdr::blocks::VectorSink;
use futuresdr::prelude::*;

#[test]
fn channel_source_min() -> Result<()> {
    let mut fg = Flowgraph::new();
    let (tx, rx) = mpsc::channel(10);

    let cs = ChannelSource::<u32>::new(rx);
    let snk = VectorSink::<u32>::new(1024);
    connect!(fg, cs > snk);

    let rt = Runtime::new();
    let running = rt.start(fg)?;
    let fg = Runtime::block_on(async move {
        tx.send(vec![0, 1, 2].into_boxed_slice()).await?;
        drop(tx);
        running
            .wait_async()
            .await
            .map_err(|e| anyhow!("Flowgraph error, {e}"))
    })?;

    let snk = fg.block(&snk)?;
    assert_eq!(*snk.items(), vec![0, 1, 2]);
    Ok(())
}

#[test]
fn channel_source_small() -> Result<()> {
    let mut fg = Flowgraph::new();
    let (tx, rx) = mpsc::channel(10);

    let cs = ChannelSource::<u32>::new(rx);
    let snk = VectorSink::<u32>::new(1024);
    connect!(fg, cs > snk);

    let rt = Runtime::new();
    let running = rt.start(fg)?;
    let fg = Runtime::block_on(async move {
        tx.send(vec![0, 1, 2].into_boxed_slice()).await?;
        tx.send(vec![3, 4].into_boxed_slice()).await?;
        tx.send(vec![].into_boxed_slice()).await?;
        tx.send(vec![5].into_boxed_slice()).await?;
        drop(tx);
        running
            .wait_async()
            .await
            .map_err(|e| anyhow!("Flowgraph error, {e}"))
    })?;

    let snk = fg.block(&snk)?;
    assert_eq!(*snk.items(), vec![0, 1, 2, 3, 4, 5]);
    Ok(())
}

#[test]
fn channel_source_big() -> Result<()> {
    let mut fg = Flowgraph::new();
    let (tx, rx) = mpsc::channel(10);

    let cs = ChannelSource::<u32>::new(rx);
    let snk = VectorSink::<u32>::new(1024);
    connect!(fg, cs > snk);

    let rt = Runtime::new();
    let running = rt.start(fg)?;
    let fg = Runtime::block_on(async move {
        tx.send(vec![0; 99999].into_boxed_slice()).await?;
        tx.send(vec![1; 88888].into_boxed_slice()).await?;
        drop(tx);
        running
            .wait_async()
            .await
            .map_err(|e| anyhow!("Flowgraph error, {e}"))
    })?;

    let snk = fg.block(&snk)?;
    let mut expected = vec![0; 99999];
    expected.extend_from_slice(&[1; 88888]);
    assert_eq!(*snk.items(), expected);

    Ok(())
}