ferrodock 1.5.0

A fast, native container manager for Docker, written in Rust.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120

//! Terminal sessions: one task per open shell, fed keys and size changes
//! through a channel and sending back what the shell prints.

use std::collections::HashMap;
use std::sync::Arc;

use futures_util::StreamExt;
use tokio::io::AsyncWriteExt;
use tokio::sync::mpsc;

use super::{Response, Shared};
use crate::docker::{DockerClient, TermOp, exit_message};

enum Msg {
    Input(Vec<u8>),
    Resize(u16, u16),
}

#[derive(Default)]
pub(super) struct Sessions {
    open: HashMap<u64, (mpsc::UnboundedSender<Msg>, tokio::task::JoinHandle<()>)>,
}

impl Sessions {
    pub(super) fn handle(&mut self, client: &DockerClient, shared: &Arc<Shared>, op: TermOp) {
        self.open.retain(|_, (_, task)| !task.is_finished());
        match op {
            TermOp::Open {
                session,
                container,
                shell,
                cols,
                rows,
            } => {
                let (tx, rx) = mpsc::unbounded_channel();
                let (client, shared) = (client.clone(), shared.clone());
                let task = tokio::spawn(async move {
                    run(
                        &client,
                        &shared,
                        session,
                        &container,
                        &shell,
                        (cols, rows),
                        rx,
                    )
                    .await;
                });
                self.open.insert(session, (tx, task));
            }
            TermOp::Input { session, bytes } => self.send(session, Msg::Input(bytes)),
            TermOp::Resize {
                session,
                cols,
                rows,
            } => self.send(session, Msg::Resize(cols, rows)),
            TermOp::Close { session } => {
                if let Some((_, task)) = self.open.remove(&session) {
                    // Dropping the connection hangs up the shell.
                    task.abort();
                }
            }
        }
    }

    fn send(&self, session: u64, msg: Msg) {
        if let Some((tx, _)) = self.open.get(&session) {
            let _ = tx.send(msg);
        }
    }
}

async fn run(
    client: &DockerClient,
    shared: &Shared,
    session: u64,
    container: &str,
    shell: &str,
    (cols, rows): (u16, u16),
    mut keys: mpsc::UnboundedReceiver<Msg>,
) {
    let mut sh = match client.open_shell(container, shell, cols, rows).await {
        Ok(sh) => sh,
        Err(e) => {
            shared.send(Response::TermExit {
                session,
                message: Some(format!("Couldn't open a shell: {e}")),
            });
            return;
        }
    };
    loop {
        tokio::select! {
            chunk = sh.output.next() => match chunk {
                Some(Ok(output)) => shared.send(Response::TermOutput {
                    session,
                    bytes: output.into_bytes().to_vec(),
                }),
                Some(Err(_)) | None => break,
            },
            msg = keys.recv() => match msg {
                Some(Msg::Input(bytes)) => {
                    if sh.input.write_all(&bytes).await.is_err() {
                        break;
                    }
                    let _ = sh.input.flush().await;
                }
                Some(Msg::Resize(cols, rows)) => {
                    let _ = client.resize_shell(&sh.exec_id, cols, rows).await;
                }
                None => break,
            },
        }
    }
    let code = client.shell_exit_code(&sh.exec_id).await;
    shared.send(Response::TermExit {
        session,
        message: exit_message(code),
    });
}