Hyper example

Cargo.toml

[package]
name = "hyperexample"
version = "0.1.0"
authors = ["Alice Ryhl <[email protected]>"]
edition = "2018"

[dependencies]
hyper = "0.12"
hyper-tls = "0.3"
tokio = "0.1"
futures = "0.1"
http = "0.1"

main.rs

use hyper::body::Body;
use hyper::{Uri, Client};
use hyper_tls::HttpsConnector;
use http::response::Parts;

use std::sync::mpsc;

use tokio::runtime::Runtime;
use futures::prelude::*;

fn main() {
    let mut runtime = Runtime::new().unwrap();
    let https = HttpsConnector::new(1).unwrap();
    let client: Client<_, Body> = Client::builder().build(https);

    let (send, recv) = mpsc::channel::<Result<(Parts, Vec<u8>), hyper::error::Error>>();
    let mut results = Vec::new();

    let uri = Uri::builder()
        .scheme("https")
        .authority("ryhl.io") // <- my website
        .path_and_query("/")
        .build()
        .unwrap();

    let mut requests_to_go = 100;
    let mut active_requests = 0;
    while requests_to_go > 0 {
        while active_requests < 12 {
            let send_clone = send.clone();
            let send_clone_err = send.clone();
            // start 12 requests
            let response_future = client.get(uri.clone());
            let future = response_future.and_then(move |response| {
                let (parts, body) = response.into_parts();

                // concat2 is a future, so this is inside and_then
                body.concat2().map(move |chunk| (parts, chunk.to_vec()))
            }).map(move |(parts, body_vec)| {
                // Here we have both body and headers, so we could handle them now.
                // Let's just send them to the main thread.
                send_clone.send(Ok((parts, body_vec))).unwrap();
            }).map_err(move |err| {
                // Send the error. Notice that the two closures need different clones of the sender.
                // This is because each closure takes ownership of it's own sender.
                send_clone_err.send(Err(err)).unwrap();
            });

            runtime.spawn(future);
            active_requests += 1;
            requests_to_go -= 1;
            println!("Spawned request");
        }
        let result = recv.recv().unwrap();
        results.push(result);
        active_requests -= 1;
    }
    // Dropping send ensures that the only senders are the ones we cloned in the loop,
    // which are currently found inside a running future.
    //
    // This means that once the last future completes and is dropped, there won't
    // be any senders left, meaning the receiver will return an error when we are done.
    drop(send);
    while let Ok(result) = recv.recv() {
        results.push(result);
    }


    // you must drop the client before waiting for the runtime to shut down
    drop(client);
    runtime.shutdown_on_idle().wait().unwrap();
    for res in results {
        match res {
            Ok((parts, body)) => {
                println!("Body len: {}", body.len());
                println!("Header count: {}", parts.headers.len());
            },
            Err(err) => {
                println!("Err: {}", err);
            },
        }
    }
}