0-metaltrack.rs

use mio::{Poll, Token, Ready, PollOpt, Events, Evented};
use std::process::{Command, Stdio};
use mio_child_process::{CommandAsync, Process};
use std::sync::mpsc::TryRecvError;
use std::collections::HashMap;
use mio_extras::channel::{sync_channel, SyncSender};
use std::io::{BufRead, stdin};

fn read_commands<R: BufRead>(reader: R, output: SyncSender<String>) {
    for line in reader.lines() {
        output.send(line.unwrap()).unwrap();
    }
}

struct ProcessPoll {
    poll: Poll,
    processes: HashMap<Token, Process>,
    commands_finished: bool,
    next_command_id: usize,
}

fn main() {
    let (command_sender, command_receiver) = sync_channel(32);
    let reader = std::thread::spawn(|| {
        read_commands(stdin().lock(), command_sender)
    });

    let mut process_poll = ProcessPoll {
        poll: Poll::new().expect("Could not spawn poll"),
        processes: HashMap::new(),
        commands_finished: false,
        next_command_id: 1,
    };

    let reader_token = Token(0);
    command_receiver
        .register(&process_poll.poll, reader_token, Ready::all(), PollOpt::edge())
        .expect("Could no register");

    let mut events = Events::with_capacity(32);

    while !process_poll.commands_finished || !process_poll.processes.is_empty() {
        process_poll.poll.poll(&mut events, None).expect("Could not poll");
        for event in &events {
            if event.token() == reader_token {
                loop {
                    let command_line = match command_receiver.try_recv() {
                        Ok(r) => r,
                        Err(TryRecvError::Empty) => break,
                        Err(TryRecvError::Disconnected) => {
                            process_poll.commands_finished = true;
                            break;
                        },
                    };

                    let process = Command::new("cmd")
                        .arg("/C")
                        .arg(command_line)
                        .stdout(Stdio::piped())
                        .stderr(Stdio::piped())
                        .spawn_async()
                        .expect("cmd doesn't exist!");

                    let token = Token(process_poll.next_command_id);
                    process_poll.next_command_id += 1;

                    process
                        .register(&process_poll.poll, token, Ready::all(), PollOpt::edge())
                        .expect("Could not register");

                    println!("Process {}: Started on PID {}", token.0, process.id());

                    process_poll.processes.insert(token, process);
                }
            } else {
                loop {
                    let process = process_poll.processes.get_mut(&event.token()).unwrap();
                    let process_event = match process.try_recv() {
                        Ok(r) => r,
                        Err(TryRecvError::Empty) => break,
                        Err(TryRecvError::Disconnected) => {
                            process_poll.processes.remove(&event.token()).unwrap();
                            break;
                        },
                    };

                    println!("Process {}: {:?}", event.token().0, process_event);
                }
            }
        }
    }

    reader.join().expect("Reader thread panicked!");
}