Rust で並列HTTP Request (非同期IO版)

前回 はスレッドで並列HTTP Request を実装しましたが、 hyper が v0.11.0 で非同期IOに対応したので、非同期IO版も書いてみました。

extern crate futures;
extern crate hyper;
extern crate tokio_core;

use std::io::{ self, Write };
use std::time::{ Duration, SystemTime };
use futures::{ Future };
use futures::future::{ join_all };
use hyper::Client;
use hyper::client::{ HttpConnector };
use tokio_core::reactor::Core;

trait DurationExtended {
    fn to_seconds(&self) -> f64;
}

impl DurationExtended for Duration {
    fn to_seconds(&self) -> f64 {
        let nanos = self.subsec_nanos() as u64;
        let secs  = self.as_secs();
        ((1000_000_000 * secs + nanos) as f64) / 1000_000_000.
    }
}

fn http_request_job(client: &Client<HttpConnector>,
                    index: usize,
                    url: String) -> Box<Future<Item=(), Error=()>> {
    let uri: hyper::Uri = url.parse().unwrap();
    let time = SystemTime::now();

    let job = client.get(uri).then(move |result| -> Result<(), ()> {
        let duration = time.elapsed().map(|t| t.to_seconds()).unwrap();
        match result {
            Ok(response) => {
                println!("{:2}: {:5.3}: {} => {}", 
                         index, duration, url, response.status());
            },
            Err(error) => {
                let _ = writeln!(&mut io::stderr(), "{:2}: Error: {}", index, error);
            }
        }
        Ok(())
    });
    Box::new(job)
}

fn main() {
    let gen_url = |q: &str| -> String { format!("http://www.google.com/search?q={}", q) };
    let urls: Vec<String> =
        vec![].into_iter()
            .chain(vec!["clojure", "haskell", "ocaml", "rust", "scala"]
                        .iter().map(|q| gen_url(q)))
            .chain(vec!["http://127.0.0.1:5000/".to_owned()].into_iter())
            .chain(vec!["javascript", "python", "perl", "php", "ruby"]
                        .iter().map(|q| gen_url(q)))
            .collect();

    let mut core = Core::new().unwrap();
    let client   = Client::new(&core.handle());

    let jobs = urls.into_iter().enumerate()
        .map(|(i, url)| http_request_job(&client, i + 1, url));

    core.run(join_all(jobs)).unwrap();
}

//  To check parallel http requests, prepare a long time request:
//  $ rackup -s webrick --port 5000 -b 'run(proc { |env| sleep 5; ["200", {"Content-Type" => "text/html"}, ["Hello World"]] })'

// $ cargo run
// ...
//  1: 0.086: http://www.google.com/search?q=clojure => 302 Found
//  4: 0.086: http://www.google.com/search?q=rust => 302 Found
//  3: 0.087: http://www.google.com/search?q=ocaml => 302 Found
//  7: 0.088: http://www.google.com/search?q=javascript => 302 Found
// 10: 0.089: http://www.google.com/search?q=php => 302 Found
//  2: 0.091: http://www.google.com/search?q=haskell => 302 Found
// 11: 0.091: http://www.google.com/search?q=ruby => 302 Found
//  5: 0.092: http://www.google.com/search?q=scala => 302 Found
//  9: 0.092: http://www.google.com/search?q=perl => 302 Found
//  8: 0.112: http://www.google.com/search?q=python => 302 Found
//  6: 5.012: http://127.0.0.1:5000/ => 200 OK

Cargo.toml

[package]
name = "http_request2"
version = "0.1.0"

[dependencies]
futures = "*"
hyper = "0.11.0"
tokio-core = "*"