Parallel SSH pools

sshpools.rs

// Here is the output for this gist
//
// Ok(Client { username: "beez", address: 192.168.122.89:22, connection_handle: "Handle<ClientHandler>" })
// Ok(Client { username: "beez", address: 192.168.122.204:22, connection_handle: "Handle<ClientHandler>" })
// killa-a2
// killa-a1
// killa-a2
// killa-a1
// killa-a1
// killa-a2
// killa-a1
// killa-a2


use futures::{
    stream,
    StreamExt
};

use async_ssh2_tokio::client as ssh;


#[tokio::main]
async fn main() -> Result<(), async_ssh2_tokio::Error> {
    let pool = make_pool().await;
    pool_exec(&pool, "ls").await;
    pool_exec(&pool, "ls").await;
    pool_exec(&pool, "ls").await;
    pool_exec(&pool, "ls").await;

    Ok(())
}


async fn pool_exec(pool: &Vec<Result<ssh::Client, async_ssh2_tokio::Error>>, cmd: &str) {
    let _r = stream::iter(pool)
        .map(|c| c.as_ref().unwrap().execute(cmd))
        .buffer_unordered(10)
        .collect::<Vec<Result<ssh::CommandExecutedResult, async_ssh2_tokio::Error>>>()
        .await;

    for r in _r.iter() {
        print!("{}", r.as_ref().unwrap().stdout);
    }
}


async fn make_pool() -> Vec<Result<ssh::Client, async_ssh2_tokio::Error>> {
    let concurrency: usize = 10;

    let results = stream::iter(vec!["192.168.122.89", "192.168.122.204"])
        .map(make_connection)
        .buffer_unordered(concurrency)
        .collect::<Vec<Result<ssh::Client, async_ssh2_tokio::Error>>>()
        .await;

    results
}


async fn make_connection(ip: &str) -> Result<ssh::Client, async_ssh2_tokio::Error> {
    let conn = ssh::Client::connect(
        (ip, 22),
        "beez",
        ssh::AuthMethod::with_password("killakilla"),
        ssh::ServerCheckMethod::NoCheck,
    ).await;

    println!("{:?}", conn);

    conn
}