stats.rs

use std::io::Write;
use std::net::TcpListener;
use std::sync::mpsc::{channel, Receiver, Sender};
use std::thread;

pub fn run_stats_server () -> Sender<Stats> {
    let (stats_tx, stats_rx) = channel();
    thread::Builder::new()
        .name("Stats:Collector".into())
        .spawn(move || {
            let new_socket_rx = stats_socket_server();

            let mut outputs = vec![];
            while let Ok(stats) = stats_rx.recv() {
                while let Ok(new_output) = new_socket_rx.try_recv() {
                    outputs.push(new_output);
                }

                let mut dead_ones = vec![];
                for (number, output) in outputs.iter().enumerate() {
                    if let Err(_) = output.send(stats) {
                        dead_ones.push(number);
                    }
                }

                for dead in dead_ones.into_iter() {
                    outputs.remove(dead);
                } 
            }
        })
        .unwrap();
    stats_tx
}

fn stats_socket_server() -> Receiver<Sender<Stats>> {
    let (new_socket_tx, new_socket_rx) = channel();
    thread::Builder::new()
        .name("Stats:SocketServer".into())
        .spawn(move || {
            let server = TcpListener::bind("127.0.0.1:6327").unwrap();
            let mut connection_id = 0;
            for stream in server.incoming() {
                if let Ok(mut stream) = stream {
                    let (tx, rx): (_, Receiver<Stats>) = channel();
                    new_socket_tx.send(tx).unwrap();
                    thread::Builder::new()
                        .name(format!("Stats:SocketServer:Socket:{}", connection_id))
                        .spawn(move || {
                            while let Ok(stats) = rx.recv() {
                                let message = format!("[{},{}]\n", stats.when, stats.duration).into_bytes();
                                if let Err(_) = stream.write(&message) {
                                    // Connection died;
                                    break;
                                }
                            }
                        })
                        .unwrap();
                    connection_id += 1;
                }
            }
        })
        .unwrap();
    new_socket_rx
}

#[derive(Clone, Copy, Debug)]
pub struct Stats {
    pub when: u64,
    pub duration: u64
}