ice_core 0.5.1

High performance Web engine
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

use container::{Container, ControlDispatcher};
use config::Config;
use lssa::control::Control;
use lssa::manager::AppManager;

use futures;
use futures::Future;
use futures::Stream;
use futures::Sink;
//use futures::{StreamExt, FutureExt};

pub struct Server {
    container: Container
}

impl Server {
    pub fn new(config: Config) -> Server {
        Server {
            container: Container::new(config)
        }
    }

    fn launch_manager(container: Container) -> futures::sync::mpsc::Sender<Control> {
        let (tx, rx) = futures::sync::mpsc::channel(4096);
        ::std::thread::spawn(move || {
            ::tokio::executor::current_thread::block_on_all(
                futures::future::ok(()).map(move |_| {
                    let mut manager = AppManager::new(container.clone());
                    load_apps_from_config(
                        &mut manager,
                        &container.config_state.read().unwrap().config
                    );
                    manager
                }).map(move |manager| {
                    rx.for_each(move |c| {
                        manager.dispatch_control(c);
                        Ok(())
                    })
                }).flatten().map_err(|_: ()| ())
            ).unwrap();
        });
        tx
    }

    pub fn run_apps(&self) -> impl Future<Item = (), Error = ()> {
        let (tx, rx) = futures::sync::mpsc::channel::<Control>(4096);
        self.container.set_control_dispatcher(ControlDispatcher::new(tx));

        let container = self.container.clone();
        let mut control_sender = Self::launch_manager(container);

        futures::future::ok(()).then(move |_: Result<(), ()>| {
            rx.for_each(move |c| {
                control_sender.start_send(c).unwrap();
                Ok(())
            }).map(|_| ()).map_err(|_| ())
        })
    }
}

fn load_apps_from_config(manager: &mut AppManager, config: &Config) {
    use std::fs::File;
    use std::io::Read;

    for (i, app) in config.applications.iter().enumerate() {
        let mut code_file = match File::open(
            &::std::path::Path::new(
                &app.path
            ).join(&app.metadata.bin).to_str().unwrap()
        ) {
            Ok(v) => v,
            Err(e) => {
                dwarning!(
                    logger!("load_apps_from_config"),
                    "Unable to load app `{}`: {:?}",
                    app.name,
                    e
                );
                continue;
            }
        };
        let mut code: Vec<u8> = Vec::new();
        code_file.read_to_end(&mut code).unwrap();

        manager.load(&code, i, app.clone());
    }
}