foo.rs

extern crate futures;
extern crate tokio;

mod environment;

use environment::{Environment, load_environment};

use std::io;
use std::path::Path;
use std::sync::Arc;

use futures::future::join_all;

use tokio::prelude::*;
use tokio::fs::OpenOptions;
use tokio::io::{read_to_end, write_all};
use tokio::net::{TcpListener, TcpStream};

fn main() -> io::Result<()> {
    let env = load_environment();

    let addr = "0.0.0.0:8000".parse().unwrap();
    let listener = TcpListener::bind(&addr).expect("unable to bind TCP listener");

    let server = listener.incoming()
        .map_err(|e| eprintln!("accept failed = {:?}", e))
        .for_each(move |sock| {
            let Environment { dir_path, other_nodes: nodes, .. } = env.clone();
            let (reader, _writer) = sock.split();

            let read_buf = Vec::new();

            let dir_path = Path::new(&dir_path);
            let file_name = "output.txt";
            let file_path = dir_path.join(file_name);
            let mut open_options = OpenOptions::new();

            let file = open_options
                .append(true)
                .create(true)
                .open(file_path);

            let vec_conns = nodes.into_iter()
                .map(|ip| {
                    let addr = ip.parse().unwrap();
                    TcpStream::connect(&addr)
                });
            let conns = join_all(vec_conns);

            let future = read_to_end(reader, read_buf)
                .join3(file, conns)
                .and_then(|((_, buf), file, nodes)| {
                    write_all(file, buf)
                        .and_then(move |(_, buf)| {
                            join_all(nodes.into_iter().map(move |node| {
                                let buf_clone = buf.clone();
                                write_all(node, buf_clone)
                            }))
                        })
                });

            let handle_conn = future.map(|amt| {
                println!("wrote {:?} bytes", amt)
            }).map_err(|err| {
                eprintln!("IO error {:?}", err)
            });

            tokio::spawn(handle_conn)
        });

    tokio::run(server);

    Ok(())
}