mitrid_core 0.9.4

Core library of the Mitrid framework
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

//! # Server
//!
//! `server` is the module providing the trait implemented by network servers.

use std::thread;
use std::sync::{Arc, Mutex};

use base::Result;
use base::size::{ConstantSize, VariableSize};
use base::Serializable;
use base::Datable;
use base::{Eval, EvalMut};
use io::store::Store;
use io::network::transport::{ClientTransport, ServerTransport};
use io::network::server::Handler;
use io::network::server::Router;
use io::network::message::{Request, Response};

/// Trait implemented by network servers.
pub trait Server<St, StS, ST, CT, Ad, H, R, S, D, MP>
    where   St: Store<StS>,
            StS: Datable + Serializable,
            ST: ServerTransport<Ad, CT>,
            CT: ClientTransport<Ad>,
            Ad: Ord + Datable + VariableSize + Serializable,
            H: Handler<St, StS, S, D, MP>,
            R: Router<St, StS, S, D, MP, H>,
            S: Datable + Serializable,
            D: Ord + Datable + ConstantSize + Serializable,
            MP: Datable + Serializable
{
    /// Serves incoming requests.
    fn serve<Ev, EvM>(store: St,
                      handler: H,
                      router: R,
                      address: &Ad,
                      thread_limit: u64,
                      evaluator: Ev,
                      evaluator_mut: EvM)
                -> Result<()>
        where   Ev: 'static + Send + Sync + Eval<St, Request<S, D, MP>, Response<S, D, MP>>,
                EvM: 'static + Send + EvalMut<St, Request<S, D, MP>, Response<S, D, MP>>
    {
        address.check()?;
        address.check_size()?;

        let mut transport = ST::listen(address)?;

        let threads_num = Arc::new(Mutex::new(0));
        let store = Arc::new(Mutex::new(store));
        let handler = Arc::new(Mutex::new(handler));
        let router = Arc::new(Mutex::new(router));
        let evaluator = Arc::new(evaluator);
        let evaluator_mut = Arc::new(Mutex::new(evaluator_mut));

        loop {
            while *threads_num.lock().unwrap() < thread_limit {

                transport.accept()
                    .and_then(|(mut transport, _)| {

                        let ser_req = transport.recv()?;
                        let req = Request::from_bytes(ser_req.as_slice())?;
                        let mut transport = transport.clone();
                        let store = store.clone();
                        let handler = handler.clone(); 
                        let router = router.clone();
                        let evaluator = evaluator.clone();
                        let evaluator_mut = evaluator_mut.clone();
                        let threads_num = threads_num.clone();
                        
                        let _ = thread::spawn(move || {
                            *threads_num.lock().unwrap() += 1;

                            let router = &mut *router.lock().unwrap();
                            let store = &mut *store.lock().unwrap();
                            let handler = &mut *handler.lock().unwrap();
                            let evaluator: &Ev = &*evaluator;
                            let evaluator_mut: &mut EvM = &mut *evaluator_mut.lock().unwrap();

                            router.route(store, handler, &req, &*evaluator, evaluator_mut)
                                .and_then(|res| {
                                    transport.send(&res.to_bytes().unwrap())
                                })
                                .or_else(|e| Err(format!("{:}", e)))
                        })
                        .join()
                        .map_err(|e| format!("{:?}", e))
                        .unwrap();

                        Ok(())
                    })
                    .or_else(|e| Err(format!("{:?}", e)))?;
                    
            }

            *threads_num.lock().unwrap() = 0;
        }
    }
}