-
Notifications
You must be signed in to change notification settings - Fork 2
/
gather.rs
49 lines (45 loc) · 1.47 KB
/
gather.rs
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
use lossyq::spsc::{Sender, channel};
use super::super::{Message, ChannelWrapper, SenderName, SenderChannelId,
ReceiverChannelId, ReceiverName
};
use super::wrap::gather_wrap;
pub trait Gather {
type InputValue : Send;
type InputError : Send;
type OutputValue : Send;
type OutputError : Send;
fn process(
&mut self,
input: &mut Vec<ChannelWrapper<Self::InputValue, Self::InputError>>,
output: &mut Sender<Message<Self::OutputValue, Self::OutputError>>,
stop: &mut bool);
}
pub fn new<InputValue: Send, InputError: Send, OutputValue: Send, OutputError: Send>(
name : &str,
output_q_size : usize,
gather : Box<Gather<InputValue=InputValue, InputError=InputError,
OutputValue=OutputValue, OutputError=OutputError>+Send>,
n_channels : usize)
-> (Box<gather_wrap::GatherWrap<InputValue, InputError, OutputValue, OutputError>>,
Box<ChannelWrapper<OutputValue, OutputError>>)
{
let (output_tx, output_rx) = channel(output_q_size);
let name = String::from(name);
let mut inputs = vec![];
for i in 0..n_channels {
inputs.push(ChannelWrapper::ReceiverNotConnected(
ReceiverChannelId(i),
ReceiverName (name.clone())
));
}
(
Box::new(gather_wrap::new( name.clone(), gather, inputs, output_tx)),
Box::new(
ChannelWrapper::SenderNotConnected(
SenderChannelId(0),
output_rx,
SenderName(name)
)
)
)
}