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
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
use crate::runtime::BlockId;
use crate::runtime::Edge;
use crate::runtime::Error;
use crate::runtime::PortId;
use crate::runtime::PortName;
use crate::runtime::Result;
#[cfg(not(target_arch = "wasm32"))]
use crate::runtime::block_on;
use crate::runtime::buffer::BufferReader;
use crate::runtime::buffer::BufferWriter;
use crate::runtime::buffer::ThreadSafeConnect;
use super::Flowgraph;
use super::block_access;
use super::connector::FlowgraphConnector;
use super::types::BlockRef;
use super::types::StreamEdge;
impl Flowgraph {
pub(super) fn stream_ports_edge<B: BufferWriter>(
src_port: &mut B,
dst_port: &mut B::Reader,
) -> Edge {
Edge::new(
src_port.block_id(),
PortId::from(src_port.port_id()),
dst_port.block_id(),
PortId::from(dst_port.port_id()),
)
}
/// Connect stream ports through typed block handles owned by this flowgraph.
///
/// This is the typed block-level stream API used by the
/// [`connect`](crate::runtime::macros::connect) macro.
///
/// The selected writer must provide sendable cross-domain connection tokens.
/// Build local-only buffers inside [`LocalDomainContext`](crate::runtime::LocalDomainContext).
#[cfg(not(target_arch = "wasm32"))]
pub fn stream<KS, KD, B, FS, FD>(
&mut self,
src_block: &BlockRef<KS>,
src_port: FS,
dst_block: &BlockRef<KD>,
dst_port: FD,
) -> Result<(), Error>
where
KS: 'static,
KD: 'static,
B: ThreadSafeConnect + 'static,
FS: FnOnce(&mut KS) -> &mut B + Send + 'static,
FD: FnOnce(&mut KD) -> &mut B::Reader + Send + 'static,
{
block_on(self.stream_async::<KS, KD, B, FS, FD>(src_block, src_port, dst_block, dst_port))
}
/// Async counterpart to [`Flowgraph::stream`].
pub async fn stream_async<KS, KD, B, FS, FD>(
&mut self,
src_block: &BlockRef<KS>,
src_port: FS,
dst_block: &BlockRef<KD>,
dst_port: FD,
) -> Result<(), Error>
where
KS: 'static,
KD: 'static,
B: ThreadSafeConnect + 'static,
FS: FnOnce(&mut KS) -> &mut B + Send + 'static,
FD: FnOnce(&mut KD) -> &mut B::Reader + Send + 'static,
{
self.validate_block_ref(src_block)?;
self.validate_block_ref(dst_block)?;
let src_id = src_block.id;
let dst_id = dst_block.id;
let src = self.location(src_id)?;
let dst = self.location(dst_id)?;
let edge = if src.domain_id == dst.domain_id {
self.with_same_domain_two_blocks_mut(src, dst, move |src_block, dst_block| {
let src = block_access::typed_kernel_mut_from_object::<KS>(src_block, src_id)?;
let dst = block_access::typed_kernel_mut_from_object::<KD>(dst_block, dst_id)?;
Ok(Self::stream_ports_edge(src_port(src), dst_port(dst)))
})
.await?
} else {
FlowgraphConnector::new(self)
.cross_domain_stream_edge_async::<KS, KD, B, FS, FD>(src, src_port, dst, dst_port)
.await?
};
self.stream_edges.push(StreamEdge::from_edge(edge, false));
Ok(())
}
/// Connect stream ports by block id and port name.
///
/// This dynamic API skips the compile-time port type checks provided by
/// [`Flowgraph::stream`]. Port existence is validated immediately; buffer
/// compatibility is checked when the flowgraph applies the connection at startup.
///
/// Prefer the typed API when the concrete block types are known. The dynamic
/// API is useful when a runtime option selects between different block
/// implementations, for example switching a source between hardware and a
/// file.
///
/// ```
/// use anyhow::Result;
/// use futuresdr::blocks::Head;
/// use futuresdr::blocks::NullSink;
/// use futuresdr::blocks::NullSource;
/// use futuresdr::prelude::*;
///
/// fn main() -> Result<()> {
/// let mut fg = Flowgraph::new();
///
/// let src = NullSource::<u8>::new();
/// let head = Head::<u8>::new(1234);
/// let snk = NullSink::<u8>::new();
///
/// let src = fg.add(src)?;
/// let head = fg.add(head)?;
///
/// // dynamic stream connection by port name
/// fg.stream_dyn(src, "output", head, "input")?;
/// // typed connection through the `connect!` macro
/// connect!(fg, head > snk);
///
/// Runtime::new().run(fg)?;
/// Ok(())
/// }
/// ```
pub fn stream_dyn(
&mut self,
src_block_id: impl Into<BlockId>,
src_port_id: impl Into<PortName>,
dst_block_id: impl Into<BlockId>,
dst_port_id: impl Into<PortName>,
) -> Result<(), Error> {
let src_block_id = src_block_id.into();
let src_port_id = PortId::from(src_port_id.into());
let dst_block_id = dst_block_id.into();
let dst_port_id = PortId::from(dst_port_id.into());
let src = self.location(src_block_id)?;
let dst = self.location(dst_block_id)?;
let local_only = src.domain_id == dst.domain_id && src.is_local();
let edge = Edge::new(src_block_id, src_port_id, dst_block_id, dst_port_id);
let edge = self.named_stream_edge(&edge)?;
self.stream_edges
.push(StreamEdge::from_edge(edge, local_only));
Ok(())
}
}