<commit_before>//! A "tiny" example of HTTP request/response handling using just tokio-core
//!
//! This example is intended for *learning purposes* to see how various pieces
//! hook up together and how HTTP can get up and running. Note that this example
//! is written with the restriction that it *can't* use any "big" library other
//! than tokio-core, if you'd like a "real world" HTTP library you likely want a
//! crate like Hyper.
//!
//! Code here is based on the `echo-threads` example and implements two paths,
//! the `/plaintext` and `/json` routes to respond with some text and json,
//! respectively. By default this will run I/O on all the cores your system has
//! available, and it doesn't support HTTP request bodies.

extern crate bytes;
extern crate futures;
extern crate http;
extern crate httparse;
extern crate num_cpus;
#[macro_use]
extern crate serde_derive;
extern crate serde_json;
extern crate time;
extern crate tokio_core;
extern crate tokio_io;

use std::env;
use std::fmt;
use std::io;
use std::net::{self, SocketAddr};
use std::thread;

use bytes::BytesMut;
use futures::future;
use futures::sync::mpsc;
use futures::{Stream, Future, Sink};
use http::{Request, Response, StatusCode};
use http::header::HeaderValue;
use tokio_core::net::TcpStream;
use tokio_core::reactor::Core;
use tokio_io::codec::{Encoder, Decoder};
use tokio_io::{AsyncRead};

fn main() {
    // Parse the arguments, bind the TCP socket we'll be listening to, spin up
    // our worker threads, and start shipping sockets to those worker threads.
    let addr = env::args().nth(1).unwrap_or("127.0.0.1:8080".to_string());
    let addr = addr.parse::<SocketAddr>().unwrap();
    let num_threads = env::args().nth(2).and_then(|s| s.parse().ok())
        .unwrap_or(num_cpus::get());

    let listener = net::TcpListener::bind(&addr).expect("failed to bind");
    println!("Listening on: {}", addr);

    let mut channels = Vec::new();
    for _ in 0..num_threads {
        let (tx, rx) = mpsc::unbounded();
        channels.push(tx);
        thread::spawn(|| worker(rx));
    }
    let mut next = 0;
    for socket in listener.incoming() {
        let socket = socket.expect("failed to accept");
        channels[next].unbounded_send(socket).expect("worker thread died");
        next = (next + 1) % channels.len();
    }
}

fn worker(rx: mpsc::UnboundedReceiver<net::TcpStream>) {
    let mut core = Core::new().unwrap();
    let handle = core.handle();

    let done = rx.for_each(move |socket| {
        // Associate each socket we get with our local event loop, and then use
        // the codec support in the tokio-io crate to deal with discrete
        // request/response types instead of bytes. Here we'll just use our
        // framing defined below and then use the `send_all` helper to send the
        // responses back on the socket after we've processed them
        let socket = future::result(TcpStream::from_stream(socket, &handle));
        let req = socket.and_then(|socket| {<commit_msg>Don't unwrap accepted connections<commit_after>//! A "tiny" example of HTTP request/response handling using just tokio-core
//!
//! This example is intended for *learning purposes* to see how various pieces
//! hook up together and how HTTP can get up and running. Note that this example
//! is written with the restriction that it *can't* use any "big" library other
//! than tokio-core, if you'd like a "real world" HTTP library you likely want a
//! crate like Hyper.
//!
//! Code here is based on the `echo-threads` example and implements two paths,
//! the `/plaintext` and `/json` routes to respond with some text and json,
//! respectively. By default this will run I/O on all the cores your system has
//! available, and it doesn't support HTTP request bodies.

extern crate bytes;
extern crate futures;
extern crate http;
extern crate httparse;
extern crate num_cpus;
#[macro_use]
extern crate serde_derive;
extern crate serde_json;
extern crate time;
extern crate tokio_core;
extern crate tokio_io;

use std::env;
use std::fmt;
use std::io;
use std::net::{self, SocketAddr};
use std::thread;

use bytes::BytesMut;
use futures::future;
use futures::sync::mpsc;
use futures::{Stream, Future, Sink};
use http::{Request, Response, StatusCode};
use http::header::HeaderValue;
use tokio_core::net::TcpStream;
use tokio_core::reactor::Core;
use tokio_io::codec::{Encoder, Decoder};
use tokio_io::{AsyncRead};

fn main() {
    // Parse the arguments, bind the TCP socket we'll be listening to, spin up
    // our worker threads, and start shipping sockets to those worker threads.
    let addr = env::args().nth(1).unwrap_or("127.0.0.1:8080".to_string());
    let addr = addr.parse::<SocketAddr>().unwrap();
    let num_threads = env::args().nth(2).and_then(|s| s.parse().ok())
        .unwrap_or(num_cpus::get());

    let listener = net::TcpListener::bind(&addr).expect("failed to bind");
    println!("Listening on: {}", addr);

    let mut channels = Vec::new();
    for _ in 0..num_threads {
        let (tx, rx) = mpsc::unbounded();
        channels.push(tx);
        thread::spawn(|| worker(rx));
    }
    let mut next = 0;
    for socket in listener.incoming() {
        if let Ok(socket) = socket {
            channels[next].unbounded_send(socket).expect("worker thread died");
            next = (next + 1) % channels.len();
        }
    }
}

fn worker(rx: mpsc::UnboundedReceiver<net::TcpStream>) {
    let mut core = Core::new().unwrap();
    let handle = core.handle();

    let done = rx.for_each(move |socket| {
        // Associate each socket we get with our local event loop, and then use
        // the codec support in the tokio-io crate to deal with discrete
        // request/response types instead of bytes. Here we'll just use our
        // framing defined below and then use the `send_all` helper to send the
        // responses back on the socket after we've processed them
        let socket = future::result(TcpStream::from_stream(socket, &handle));
        let req = socket.and_then(|socket| {
{'fraction_non_alphanumeric': 0.12829001820890581, 'fraction_numerical': 0.004966065220989902, 'mean_word_length': 3.6263399693721285, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 2, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4584811', 'n_tokens_mistral': 1856, 'n_tokens_neox': 1774, 'n_words': 674}