cram 0.1.2

Cram realy awesome middleware
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
95
96
97
98
99
100
101
102
103
104
105
106
107

#![allow(dead_code)]

mod dns;

use crate::node::Node;
use crossbeam::Receiver;
use std::collections::HashMap;
use std::thread;
use dns::Dns;

use crate::serializers::*;

pub struct Io { 
    input_channel : Receiver<(i8,Vec<u8>,String)>,
    port : usize,
    option_dns : Option<Dns>,
}

impl Io {
    
    pub fn new(port: usize, input_channel: Receiver<(i8,Vec<u8>,String)> ) -> Io {
        
        Io {
            input_channel : input_channel,
            port : port,
            option_dns : None,
        }
    }

    pub fn new_with_dns(port: usize, input_channel: Receiver<(i8,Vec<u8>,String)>, dns_path : String ) -> Io {
        
        Io {
            input_channel : input_channel,
            port : port,
            option_dns : Some( Dns::new(dns_path)),
        }
    }

    pub fn start(mut self, node : Node){

        let mut context = zmq::Context::new();
                
        let pull_socket = context.socket(zmq::PULL).unwrap();
        pull_socket.bind(&format!("tcp://*:{}", self.port)).unwrap();
        
        let _read_from_sck = thread::spawn(move || loop {
            
            let mut vec;  
            match pull_socket.recv_bytes(0) {
                Ok(val) => vec = val,
                Err(_e) => break,
            }

            let message_type = get_type(&mut vec);
            
            match node.controllers.get(&message_type){
                None => (),
                Some(c) => c.send( vec ).unwrap(),
            };
        });

        let _write_to_sck = thread::spawn(move || {
            
            let mut push_sockets : HashMap<String, zmq::Socket> = HashMap::new();

            loop {
                
                let (controller_type, mut vec, mut to);
                match self.input_channel.recv() {
                    Ok( (v1,v2,v3) ) => {
                        controller_type = v1;
                        vec = v2;
                        to = v3;
                    }
                    Err(_e) => {
                        context.destroy().unwrap();
                        break;
                    }
                }
            
                put_type( controller_type, &mut vec);

                self.option_dns = match self.option_dns{
                    None => None,
                    Some(dns) =>{
                        match dns.get_address(&to) {
                            Some(addr) => 
                                to =  addr,
                            None => (),
                        };
                        Some(dns)
                    }
                };
                
                if !push_sockets.contains_key(&to) {
                    let sck = context.socket(zmq::PUSH).unwrap();
                    sck.connect(&format!("tcp://{}", to)).unwrap();
                    push_sockets.insert(to.clone(), sck);
                }

                let sck = push_sockets.get(&to).unwrap();
                sck.send(&vec, 0).unwrap();
            }
        });
        
    }   
}