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
use crate::runtime::dev::prelude::*;
/// Push received messages into a channel.
///
/// # Message Inputs
///
/// `in`: Messages to send through the channel.
///
/// # Message Outputs
///
/// No message outputs.
///
/// # Usage
/// ```
/// use futuresdr::blocks::MessagePipe;
/// use futuresdr::prelude::*;
///
/// let (tx, rx) = mpsc::channel(8);
/// let pipe = MessagePipe::new(tx);
/// ```
#[derive(Block)]
#[message_inputs(r#in)]
#[null_kernel]
pub struct MessagePipe {
sender: mpsc::Sender<Pmt>,
}
impl MessagePipe {
/// Create MessagePipe block
pub fn new(sender: mpsc::Sender<Pmt>) -> Self {
Self { sender }
}
async fn r#in(
&mut self,
_io: &mut WorkIo,
_mo: &mut MessageOutputs,
_meta: &mut BlockMeta,
p: Pmt,
) -> Result<Pmt> {
if self.sender.send(p).await.is_ok() {
Ok(Pmt::Ok)
} else {
// Channel Receiver dropped
Ok(Pmt::Finished)
}
}
}