minions 0.2.2

Experimental actor library, under constant development.
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
88
89
90
91
92
93
94

use lossyq::spsc::{Sender, Receiver, channel};
use super::super::common::{Task, Reporter, Message, Schedule, IdentifiedReceiver, Direction, new_id};
use super::super::connectable::{Connectable};
use super::super::scheduler::MeasureTime;

pub trait Scatter {
  type InputType   : Send;
  type OutputType  : Send;

  fn process(
    &mut self,
    input:   &mut Option<IdentifiedReceiver<Self::InputType>>,
    output:  &mut Vec<Sender<Message<Self::OutputType>>>) -> Schedule;
}

pub struct ScatterWrap<Input: Send, Output: Send> {
  name           : String,
  state          : Box<Scatter<InputType=Input,OutputType=Output>+Send>,
  input_rx       : Option<IdentifiedReceiver<Input>>,
  output_tx_vec  : Vec<Sender<Message<Output>>>,
  p1             : MeasureTime,
}

impl<Input: Send, Output: Send> Connectable for ScatterWrap<Input,Output> {
  type Input = Input;

  fn input(&mut self) -> &mut Option<IdentifiedReceiver<Input>> {
    &mut self.input_rx
  }
}

impl<Input: Send, Output: Send> Task for ScatterWrap<Input,Output> {
  fn execute(&mut self, reporter: &mut Reporter) -> Schedule {
    self.p1.start();
    // TODO : make this nicer. repetitive for all elems!
    let mut msg_ids = vec![];
    for otx in &self.output_tx_vec {
      msg_ids.push(otx.seqno());
    }
    let retval = self.state.process(&mut self.input_rx,
                                    &mut self.output_tx_vec);
    let otx_slice = self.output_tx_vec.as_slice();
    let ids_slice = msg_ids.as_slice();
    for i in 0..msg_ids.len() {
      let new_msg_id = otx_slice[i].seqno();
      if ids_slice[i] != new_msg_id {
        reporter.message_sent(i, new_msg_id);
      }
    }
    self.p1.end();
    self.p1.print("Scatter: p1");
    retval
  }
  fn name(&self) -> &String { &self.name }
}

pub fn new<Input: Send, Output: Send>(
    name            : &str,
    output_q_size   : usize,
    scatter         : Box<Scatter<InputType=Input,OutputType=Output>+Send>,
    n_channels      : usize)
      -> (Box<ScatterWrap<Input,Output>>, Vec<Box<Option<IdentifiedReceiver<Output>>>>)
{
  let mut tx_vec = vec![];
  let mut rx_vec = vec![];

  for i in 0..n_channels {
    let (output_tx, output_rx) = channel(output_q_size);
    tx_vec.push(output_tx);
    rx_vec.push(
      Box::new(
        Some(
          IdentifiedReceiver{
            id:     new_id(String::from(name), Direction::Out, i),
            input:  output_rx,
          }
        )
      )
    );
  }

  (
    Box::new(
      ScatterWrap{
        name           : String::from(name),
        state          : scatter,
        input_rx       : None,
        output_tx_vec  : tx_vec,
        p1             : MeasureTime::new(),
      }
    ),
    rx_vec
  )
}