More elaborate SSH pools

main.rs

use async_ssh2_tokio::{
    Error,
    client::AuthMethod
};

mod pools;


#[tokio::main]
async fn main() -> Result<(), Error> {
    let username = "beez";
    let auth = AuthMethod::with_password("killakilla");

    let ips = vec!["192.168.122.89", "192.168.122.204"];
    let pool = pools::SSHPool::new(ips, username, &auth).await;

    let cmd = "sleep $(shuf -i 1-3 -n 1) && ls";
    pools::print_results(pool.exec(cmd).await);
    pools::print_results(pool.exec(cmd).await);
    pools::print_results(pool.exec(cmd).await);
    pools::print_results(pool.exec(cmd).await);

    Ok(())
}

pools.rs

use futures::{
    stream,
    StreamExt
};

use async_ssh2_tokio::Error;
use async_ssh2_tokio::client::{
    Client,
    CommandExecutedResult,
    AuthMethod,
    ServerCheckMethod
};

pub struct SSHConnection {
    client: Client,
    ip: String,
    username: String,
}

impl SSHConnection {
    pub async fn open(ip: &str, username: &str, auth: &AuthMethod) -> Self {
        let conn = Client::connect(
            (ip, 22),
            username,
            auth.clone(),
            ServerCheckMethod::NoCheck,
        ).await;

        SSHConnection {
            client: conn.unwrap(),
            ip: String::from(ip),
            username: String::from(username)
        }
    }
}

pub struct SSHPool {
    conns: Vec<SSHConnection>
}

impl SSHPool {
    pub async fn new(ips: Vec<&str>, username: &str, auth: &AuthMethod) -> SSHPool {
        let concurrency: usize = 10;
    
        let results = stream::iter(ips)
            .map(|ip| SSHConnection::open(ip, username, &auth))
            .buffer_unordered(concurrency)
            .collect::<Vec<SSHConnection>>()
            .await;
    
        SSHPool {
            conns: results
        }
    }

    pub async fn exec(&self, cmd: &str) -> Vec<CommandExecutedResult> {
        let results = stream::iter(self.conns.iter())
            .map(|c| c.client.execute(cmd))
            .buffer_unordered(10)
            .collect::<Vec<Result<CommandExecutedResult, Error>>>()
            .await;

        let mut output = Vec::new();
        for r in results.iter() {
            output.push(r.as_ref().unwrap().clone());
        }

        output
    }
}


pub fn print_results(results: Vec<CommandExecutedResult>) {
    for r in results.iter() {
        print!("{}", r.stdout);
    }
}