bsdkrun_sdk/
client.rs

1//! A client for a remote `bsdkrund` daemon's GraphQL API.
2//!
3//! [`crate::Sandbox`] talks to a *local* `bsdkrun` binary by shelling out to
4//! it. [`Client`] is the network sibling: it drives the exact same operations
5//! against a daemon over HTTP (queries/mutations) and one shared
6//! `graphql-transport-ws` socket (subscriptions — exec output, live shells,
7//! log follow), so a program can target either a machine with the CLI
8//! installed or a remote host running `bsdkrund` with the same calls.
9//!
10//! The GraphQL documents below are deliberately minimal string literals
11//! rather than a generated client: this SDK has no code-generation step, and
12//! the schema is small and stable enough (`daemon/src/graphql.rs`) that
13//! hand-typed queries stay easy to keep in sync.
14
15use std::sync::{mpsc, Arc, Mutex};
16
17use base64::engine::general_purpose::STANDARD as B64;
18use base64::Engine as _;
19use serde_json::{json, Value};
20
21use crate::args::{strvec, NetOpts};
22use crate::error::{Error, Result};
23use crate::transport::{http_request, normalize_url, ws_url, WsTransport, TOKEN_ENV, URL_ENV};
24use crate::types::{CommandResult, RemoteExecResult, SandboxInfo, ShellSessionInfo};
25
26// ---------------------------------------------------------------------------
27// GraphQL documents
28// ---------------------------------------------------------------------------
29
30const MACHINE_FIELDS: &str = "id name image kind command status running exitCode pid detached \
31     cpus mem volume stateDir createdAt finishedAt network netIp \
32     ports { bind host guest }";
33const CMD_RESULT_FIELDS: &str = "exitCode stdout stderr";
34const SESSION_FIELDS: &str = "id machineId finished truncated";
35
36fn list_query() -> String {
37    format!("query($all: Boolean!) {{ machines(all: $all) {{ {MACHINE_FIELDS} }} }}")
38}
39fn get_query() -> String {
40    format!("query($id: String!) {{ machine(id: $id) {{ {MACHINE_FIELDS} }} }}")
41}
42const LOGS_QUERY: &str =
43    "query($id: String!, $boot: Boolean!) { machineLogs(id: $id, boot: $boot) }";
44
45fn stop_mutation() -> String {
46    format!("mutation($id: String!) {{ stopMachine(id: $id) {{ {CMD_RESULT_FIELDS} }} }}")
47}
48fn start_mutation() -> String {
49    format!("mutation($id: String!) {{ startMachine(id: $id) {{ {CMD_RESULT_FIELDS} }} }}")
50}
51fn remove_mutation() -> String {
52    format!(
53        "mutation($ids: [String!]!, $force: Boolean!) {{ \
54         removeMachines(ids: $ids, force: $force) {{ {CMD_RESULT_FIELDS} }} }}"
55    )
56}
57fn update_mutation() -> String {
58    format!(
59        "mutation($id: String!, $cpus: Int, $mem: Int) {{ \
60         updateMachine(id: $id, cpus: $cpus, mem: $mem) {{ {CMD_RESULT_FIELDS} }} }}"
61    )
62}
63fn commit_mutation() -> String {
64    format!(
65        "mutation($id: String!, $name: String!, $description: String!) {{ \
66         commitMachine(id: $id, name: $name, description: $description) {{ {CMD_RESULT_FIELDS} }} }}"
67    )
68}
69
70const RUN_LINUX_MUTATION: &str = "mutation($input: RunLinuxInput!) { runLinux(input: $input) }";
71const RUN_BSD_MUTATION: &str = "mutation($input: RunBsdInput!) { runBsd(input: $input) }";
72const RUN_NANOS_MUTATION: &str = "mutation($input: RunNanosInput!) { runNanos(input: $input) }";
73const RUN_UNIKRAFT_MUTATION: &str =
74    "mutation($input: RunUnikraftInput!) { runUnikraft(input: $input) }";
75const RUN_SOLO5_MUTATION: &str = "mutation($input: RunSolo5Input!) { runSolo5(input: $input) }";
76const RUN_OSV_MUTATION: &str = "mutation($input: RunOsvInput!) { runOsv(input: $input) }";
77const RUN_FLAVOR_MUTATION: &str = "mutation($input: RunFlavorInput!) { runFlavor(input: $input) }";
78
79const MACHINE_LOGS_SUBSCRIPTION: &str =
80    "subscription($id: String!, $follow: Boolean!, $boot: Boolean!) { \
81     machineLogs(id: $id, follow: $follow, boot: $boot) { dataBase64 exitCode } }";
82
83fn open_shell_mutation() -> String {
84    format!(
85        "mutation($machineId: String!, $command: [String!]!, $env: [String!]!, \
86         $rows: Int!, $cols: Int!) {{ \
87         openShell(machineId: $machineId, command: $command, env: $env, \
88         rows: $rows, cols: $cols) {{ {SESSION_FIELDS} }} }}"
89    )
90}
91const SHELL_OUTPUT_SUBSCRIPTION: &str = "subscription($sessionId: String!) { \
92     shellOutput(sessionId: $sessionId) { dataBase64 exitCode } }";
93const SEND_INPUT_MUTATION: &str = "mutation($sessionId: String!, $dataBase64: String!) { \
94     sendShellInput(sessionId: $sessionId, dataBase64: $dataBase64) }";
95const RESIZE_MUTATION: &str = "mutation($sessionId: String!, $rows: Int!, $cols: Int!) { \
96     resizeShell(sessionId: $sessionId, rows: $rows, cols: $cols) }";
97const CLOSE_MUTATION: &str = "mutation($sessionId: String!) { closeShell(sessionId: $sessionId) }";
98
99// ---------------------------------------------------------------------------
100// Client
101// ---------------------------------------------------------------------------
102
103struct ClientInner {
104    url: String,
105    token: String,
106    /// The one lazily opened WS transport every subscription shares. It drops
107    /// its socket when the last subscription ends and reconnects on the next,
108    /// so the `Arc` never needs replacing.
109    ws: Mutex<Option<Arc<WsTransport>>>,
110}
111
112/// A client for a remote `bsdkrund`'s GraphQL API.
113///
114/// Queries and mutations go over HTTP; subscriptions (used internally by
115/// [`Client::exec`], [`Client::shell`] and [`Client::follow_logs`]) share one
116/// lazily opened `graphql-transport-ws` socket per client, torn down once the
117/// last subscription ends. Cloning is cheap and shares that socket.
118#[derive(Clone)]
119pub struct Client {
120    inner: Arc<ClientInner>,
121}
122
123impl std::fmt::Debug for Client {
124    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
125        f.debug_struct("Client")
126            .field("url", &self.inner.url)
127            .finish()
128    }
129}
130
131impl Client {
132    /// Build a client from a daemon URL and its bearer token.
133    ///
134    /// A URL configured without a token is refused rather than silently
135    /// making an unauthenticated request — the daemon has no anonymous tier.
136    pub fn new(url: impl Into<String>, token: impl Into<String>) -> Result<Client> {
137        let url = normalize_url(&url.into());
138        if url.is_empty() {
139            return Err(Error::InvalidInput("the daemon URL is empty".into()));
140        }
141        let token = token.into().trim().to_string();
142        if token.is_empty() {
143            return Err(Error::InvalidInput(
144                "a daemon URL without a token is a configuration error; pass the bearer token"
145                    .into(),
146            ));
147        }
148        Ok(Client {
149            inner: Arc::new(ClientInner {
150                url,
151                token,
152                ws: Mutex::new(None),
153            }),
154        })
155    }
156
157    /// Build a client from `BSDKRUN_URL` / `BSDKRUN_TOKEN`.
158    ///
159    /// Errors if `BSDKRUN_URL` is unset (nothing to connect to), or if it is
160    /// set but `BSDKRUN_TOKEN` is not — a host configured without a token is
161    /// a configuration error, never a silent fall-back to an unauthenticated
162    /// request.
163    pub fn from_env() -> Result<Client> {
164        let url = std::env::var(URL_ENV)
165            .unwrap_or_default()
166            .trim()
167            .to_string();
168        if url.is_empty() {
169            return Err(Error::InvalidInput(format!(
170                "{URL_ENV} is not set; nothing to connect to"
171            )));
172        }
173        let token = std::env::var(TOKEN_ENV)
174            .unwrap_or_default()
175            .trim()
176            .to_string();
177        if token.is_empty() {
178            return Err(Error::InvalidInput(format!(
179                "{URL_ENV} is set but {TOKEN_ENV} is not"
180            )));
181        }
182        Client::new(url, token)
183    }
184
185    /// The normalized GraphQL endpoint URL.
186    pub fn url(&self) -> &str {
187        &self.inner.url
188    }
189
190    // -- transport (escape hatch) ------------------------------------------
191
192    /// Run a raw query or mutation and return its `data`.
193    pub fn request(&self, query: &str, variables: Value) -> Result<Value> {
194        http_request(&self.inner.url, &self.inner.token, query, &variables)
195    }
196
197    /// Start a raw subscription; each `next` payload's `data` goes to
198    /// `on_next`. Returns a [`Subscription`] handle to end it with.
199    pub fn subscribe(
200        &self,
201        query: &str,
202        variables: Value,
203        on_next: impl FnMut(Value) + Send + 'static,
204    ) -> Result<Subscription> {
205        self.subscribe_with(query, variables, on_next, |_| {}, || {})
206    }
207
208    /// [`Client::subscribe`] with error/completion callbacks.
209    pub fn subscribe_with(
210        &self,
211        query: &str,
212        variables: Value,
213        on_next: impl FnMut(Value) + Send + 'static,
214        on_error: impl FnMut(Error) + Send + 'static,
215        on_complete: impl FnMut() + Send + 'static,
216    ) -> Result<Subscription> {
217        let transport = self.ws();
218        let id = transport.subscribe(
219            query,
220            variables,
221            Box::new(on_next),
222            Box::new(on_error),
223            Box::new(on_complete),
224        )?;
225        Ok(Subscription { transport, id })
226    }
227
228    fn ws(&self) -> Arc<WsTransport> {
229        let mut guard = self.inner.ws.lock().unwrap();
230        guard
231            .get_or_insert_with(|| {
232                Arc::new(WsTransport::new(
233                    ws_url(&self.inner.url),
234                    self.inner.token.clone(),
235                ))
236            })
237            .clone()
238    }
239
240    // -- lifecycle / listing -----------------------------------------------
241
242    /// List machines. `all` includes exited ones.
243    pub fn list(&self, all: bool) -> Result<Vec<SandboxInfo>> {
244        let data = self.request(&list_query(), json!({"all": all}))?;
245        Ok(data
246            .get("machines")
247            .and_then(Value::as_array)
248            .map(|machines| machines.iter().map(SandboxInfo::from_graphql).collect())
249            .unwrap_or_default())
250    }
251
252    /// Fetch one machine by id (a unique prefix) or name, or `None`.
253    pub fn get(&self, id: &str) -> Result<Option<SandboxInfo>> {
254        let data = self.request(&get_query(), json!({"id": id}))?;
255        Ok(data
256            .get("machine")
257            .filter(|m| !m.is_null())
258            .map(SandboxInfo::from_graphql))
259    }
260
261    pub fn stop(&self, id: &str) -> Result<CommandResult> {
262        let data = self.request(&stop_mutation(), json!({"id": id}))?;
263        Ok(CommandResult::from_graphql(&data["stopMachine"]))
264    }
265
266    pub fn start(&self, id: &str) -> Result<CommandResult> {
267        let data = self.request(&start_mutation(), json!({"id": id}))?;
268        Ok(CommandResult::from_graphql(&data["startMachine"]))
269    }
270
271    pub fn remove<S: AsRef<str>>(&self, ids: &[S], force: bool) -> Result<CommandResult> {
272        let ids: Vec<&str> = ids.iter().map(AsRef::as_ref).collect();
273        let data = self.request(&remove_mutation(), json!({"ids": ids, "force": force}))?;
274        Ok(CommandResult::from_graphql(&data["removeMachines"]))
275    }
276
277    /// Change a machine's recorded vCPU / memory; applies on its next start.
278    pub fn update(&self, id: &str, cpus: Option<u32>, mem: Option<u32>) -> Result<CommandResult> {
279        let data = self.request(
280            &update_mutation(),
281            json!({"id": id, "cpus": cpus, "mem": mem}),
282        )?;
283        Ok(CommandResult::from_graphql(&data["updateMachine"]))
284    }
285
286    /// Snapshot a machine into a named flavor, like `docker commit`.
287    pub fn commit(&self, id: &str, name: &str, description: &str) -> Result<CommandResult> {
288        let data = self.request(
289            &commit_mutation(),
290            json!({"id": id, "name": name, "description": description}),
291        )?;
292        Ok(CommandResult::from_graphql(&data["commitMachine"]))
293    }
294
295    /// One-shot read of a machine's console log (bsdkrun's boot log with
296    /// `boot`).
297    pub fn logs(&self, id: &str, boot: bool) -> Result<String> {
298        let data = self.request(LOGS_QUERY, json!({"id": id, "boot": boot}))?;
299        Ok(data
300            .get("machineLogs")
301            .and_then(Value::as_str)
302            .unwrap_or_default()
303            .to_string())
304    }
305
306    /// Stream a machine's console log live.
307    ///
308    /// ```no_run
309    /// # let client = bsdkrun_sdk::Client::new("localhost:50052", "tok")?;
310    /// let sub = client
311    ///     .follow_logs("abc123")
312    ///     .on_data(|bytes| print!("{}", String::from_utf8_lossy(&bytes)))
313    ///     .start()?;
314    /// # Ok::<(), bsdkrun_sdk::Error>(())
315    /// ```
316    pub fn follow_logs(&self, id: &str) -> FollowLogsBuilder {
317        FollowLogsBuilder {
318            client: self.clone(),
319            id: id.to_string(),
320            follow: true,
321            boot: false,
322            on_data: None,
323            on_error: None,
324            on_complete: None,
325        }
326    }
327
328    // -- booting -----------------------------------------------------------
329
330    /// Boot a Linux machine on the daemon — `runLinux`.
331    pub fn run_linux(&self) -> RunLinuxBuilder {
332        RunLinuxBuilder {
333            client: self.clone(),
334            image: None,
335            cpus: None,
336            mem: None,
337            net: NetOpts::default(),
338            volume: None,
339            mounts: Vec::new(),
340            attach_disk: Vec::new(),
341            env: Vec::new(),
342            entrypoint: None,
343            initramfs: false,
344            kernel: None,
345            kernel_version: None,
346            console: None,
347            repo: None,
348            command: Vec::new(),
349        }
350    }
351
352    /// Boot FreeBSD or NetBSD on the daemon — `runBsd`.
353    pub fn run_bsd(&self, os: BsdOs) -> RunBsdBuilder {
354        RunBsdBuilder {
355            client: self.clone(),
356            os,
357            version: None,
358            cpus: None,
359            mem: None,
360            net: NetOpts::default(),
361            volume: None,
362            persist: false,
363            force: false,
364            firmware: None,
365            attach_disk: Vec::new(),
366            disk_size: None,
367            repo: None,
368            command: Vec::new(),
369        }
370    }
371
372    /// Boot a Nanos unikernel on the daemon — `runNanos`.
373    pub fn run_nanos(&self) -> RunNanosBuilder {
374        RunNanosBuilder {
375            client: self.clone(),
376            image: None,
377            cpus: None,
378            mem: None,
379            net: NetOpts::default(),
380            kernel: None,
381            cmdline: None,
382            persist: false,
383        }
384    }
385
386    /// Boot a Unikraft unikernel on the daemon — `runUnikraft`.
387    pub fn run_unikraft(&self) -> RunUnikraftBuilder {
388        RunUnikraftBuilder {
389            client: self.clone(),
390            path: None,
391            cpus: None,
392            mem: None,
393            net: NetOpts::default(),
394            cmdline: None,
395            initramfs: None,
396            mounts: Vec::new(),
397        }
398    }
399
400    /// Boot a Solo5 (MirageOS) unikernel on the daemon — `runSolo5`. Runs
401    /// under the `solo5-hvt` tender rather than libkrun; the unikernel
402    /// declares its own devices in its `MFT1` manifest, so only block
403    /// backings and its own args are passed.
404    pub fn run_solo5(&self) -> RunSolo5Builder {
405        RunSolo5Builder {
406            client: self.clone(),
407            path: None,
408            cpus: None,
409            mem: None,
410            net: NetOpts::default(),
411            block: Vec::new(),
412            args: Vec::new(),
413        }
414    }
415
416    /// Boot an OSv unikernel on the daemon — `runOsv`.
417    pub fn run_osv(&self) -> RunOsvBuilder {
418        RunOsvBuilder {
419            client: self.clone(),
420            image: None,
421            cpus: None,
422            mem: None,
423            net: NetOpts::default(),
424            cmdline: None,
425            disk: None,
426            no_disk: false,
427            attach_disk: Vec::new(),
428            gic: None,
429            persist: false,
430            volume: None,
431        }
432    }
433
434    /// Boot a named flavor on the daemon — `runFlavor`.
435    pub fn run_flavor(&self, name: impl Into<String>) -> RunFlavorBuilder {
436        RunFlavorBuilder {
437            client: self.clone(),
438            name: name.into(),
439            cpus: None,
440            mem: None,
441            ports: Vec::new(),
442            volume: None,
443            repo: None,
444        }
445    }
446
447    // -- exec / interactive shell ------------------------------------------
448
449    /// Run a command to completion via the machine's shell agent.
450    pub fn exec<I, S>(&self, id: &str, command: I) -> Result<RemoteExecResult>
451    where
452        I: IntoIterator<Item = S>,
453        S: Into<String>,
454    {
455        self.exec_with_env(id, command, Vec::<String>::new())
456    }
457
458    /// [`Client::exec`] with per-command `"K=V"` environment entries.
459    ///
460    /// Sequenced exactly as `daemon/README.md` describes: `openShell` (with
461    /// `command` set, so the session runs it instead of a login shell), THEN
462    /// subscribe to `shellOutput` (output is buffered from the moment the
463    /// session opened, so nothing is lost even though the subscribe
464    /// necessarily happens after the mutation), collecting bytes until an
465    /// event carries a non-null exit code, THEN `closeShell` — called
466    /// unconditionally, including on error, since it is idempotent and a
467    /// session must never be left dangling.
468    pub fn exec_with_env<I, S, E, T>(
469        &self,
470        id: &str,
471        command: I,
472        env: E,
473    ) -> Result<RemoteExecResult>
474    where
475        I: IntoIterator<Item = S>,
476        S: Into<String>,
477        E: IntoIterator<Item = T>,
478        T: Into<String>,
479    {
480        let transport = self.ws();
481        let data = self.request(
482            &open_shell_mutation(),
483            json!({
484                "machineId": id,
485                "command": strvec(command),
486                "env": strvec(env),
487                "rows": 24,
488                "cols": 80,
489            }),
490        )?;
491        let session = ShellSessionInfo::from_graphql(&data["openShell"]);
492
493        let chunks = Arc::new(Mutex::new(Vec::<u8>::new()));
494        // The reader thread delivers `shellOutput` events via callbacks; this
495        // channel is how the calling thread blocks until the one it cares
496        // about (an exit code, or a terminal error) arrives, keeping exec() a
497        // synchronous call.
498        let (done_tx, done_rx) = mpsc::channel::<Result<i32>>();
499
500        let chunk_sink = Arc::clone(&chunks);
501        let exit_tx = done_tx.clone();
502        let error_tx = done_tx.clone();
503        let complete_tx = done_tx;
504
505        let outcome: Result<i32> = (|| {
506            let sub_id = transport.subscribe(
507                SHELL_OUTPUT_SUBSCRIPTION,
508                json!({"sessionId": session.id}),
509                Box::new(move |data: Value| {
510                    let payload = &data["shellOutput"];
511                    if let Some(b64) = payload["dataBase64"].as_str() {
512                        if let Ok(bytes) = B64.decode(b64) {
513                            chunk_sink.lock().unwrap().extend_from_slice(&bytes);
514                        }
515                    }
516                    if let Some(code) = payload["exitCode"].as_i64() {
517                        let _ = exit_tx.send(Ok(code as i32));
518                    }
519                }),
520                Box::new(move |err: Error| {
521                    let _ = error_tx.send(Err(err));
522                }),
523                Box::new(move || {
524                    // The subscription ended without ever delivering an exit
525                    // code (e.g. the daemon tore the session down) — surface
526                    // that instead of blocking forever.
527                    let _ = complete_tx.send(Err(Error::GraphQL {
528                        message: "shell session ended before an exit code arrived".to_string(),
529                        code: None,
530                    }));
531                }),
532            )?;
533            let outcome = done_rx.recv().unwrap_or_else(|_| {
534                Err(Error::GraphQL {
535                    message: "the shell output subscription was dropped".to_string(),
536                    code: None,
537                })
538            });
539            transport.unsubscribe(&sub_id);
540            outcome
541        })();
542
543        // closeShell runs unconditionally — including on error — since it is
544        // idempotent and a session must never be left dangling.
545        let _ = self.request(CLOSE_MUTATION, json!({"sessionId": session.id}));
546
547        let exit_code = outcome?;
548        let output = chunks.lock().unwrap().clone();
549        Ok(RemoteExecResult { exit_code, output })
550    }
551
552    /// Open a live interactive session — output/exit arrive via callbacks.
553    ///
554    /// ```no_run
555    /// # let client = bsdkrun_sdk::Client::new("localhost:50052", "tok")?;
556    /// let session = client.shell("abc123").rows(50).cols(120).open()?;
557    /// session.on_output(|bytes| print!("{}", String::from_utf8_lossy(bytes)));
558    /// session.on_exit(|code| println!("exited {code}"));
559    /// session.write("ls -la\n")?;
560    /// # Ok::<(), bsdkrun_sdk::Error>(())
561    /// ```
562    pub fn shell(&self, id: &str) -> ShellBuilder {
563        ShellBuilder {
564            client: self.clone(),
565            machine_id: id.to_string(),
566            command: Vec::new(),
567            env: Vec::new(),
568            rows: 24,
569            cols: 80,
570        }
571    }
572}
573
574/// A raw subscription handle returned by [`Client::subscribe`]. Dropping it
575/// does *not* unsubscribe — call [`Subscription::unsubscribe`], matching the
576/// Python SDK's explicit unsubscribe function.
577pub struct Subscription {
578    transport: Arc<WsTransport>,
579    id: String,
580}
581
582impl Subscription {
583    /// The graphql-transport-ws subscription id.
584    pub fn id(&self) -> &str {
585        &self.id
586    }
587
588    /// End the subscription.
589    pub fn unsubscribe(self) {
590        self.transport.unsubscribe(&self.id);
591    }
592}
593
594// ---------------------------------------------------------------------------
595// follow_logs
596// ---------------------------------------------------------------------------
597
598type DataFn = Box<dyn FnMut(Vec<u8>) + Send>;
599type ErrFn = Box<dyn FnMut(Error) + Send>;
600type DoneFn = Box<dyn FnMut() + Send>;
601
602/// A live log stream being assembled — see [`Client::follow_logs`].
603pub struct FollowLogsBuilder {
604    client: Client,
605    id: String,
606    follow: bool,
607    boot: bool,
608    on_data: Option<DataFn>,
609    on_error: Option<ErrFn>,
610    on_complete: Option<DoneFn>,
611}
612
613impl FollowLogsBuilder {
614    /// Keep following after the backlog (default true; false replays and ends).
615    pub fn follow(mut self, follow: bool) -> Self {
616        self.follow = follow;
617        self
618    }
619
620    /// Stream bsdkrun's boot log instead of the console.
621    pub fn boot(mut self, boot: bool) -> Self {
622        self.boot = boot;
623        self
624    }
625
626    /// Receive each chunk of log bytes.
627    pub fn on_data(mut self, cb: impl FnMut(Vec<u8>) + Send + 'static) -> Self {
628        self.on_data = Some(Box::new(cb));
629        self
630    }
631
632    /// Receive the terminal error, if the stream fails.
633    pub fn on_error(mut self, cb: impl FnMut(Error) + Send + 'static) -> Self {
634        self.on_error = Some(Box::new(cb));
635        self
636    }
637
638    /// Notified when the stream ends cleanly.
639    pub fn on_complete(mut self, cb: impl FnMut() + Send + 'static) -> Self {
640        self.on_complete = Some(Box::new(cb));
641        self
642    }
643
644    /// Start streaming. Returns the [`Subscription`] to stop with.
645    pub fn start(self) -> Result<Subscription> {
646        let mut on_data = self.on_data.unwrap_or_else(|| Box::new(|_| {}));
647        let on_error = self.on_error.unwrap_or_else(|| Box::new(|_| {}));
648        let on_complete = self.on_complete.unwrap_or_else(|| Box::new(|| {}));
649        let transport = self.client.ws();
650        let id = transport.subscribe(
651            MACHINE_LOGS_SUBSCRIPTION,
652            json!({"id": self.id, "follow": self.follow, "boot": self.boot}),
653            Box::new(move |data: Value| {
654                if let Some(b64) = data
655                    .pointer("/machineLogs/dataBase64")
656                    .and_then(Value::as_str)
657                {
658                    if let Ok(bytes) = B64.decode(b64) {
659                        on_data(bytes);
660                    }
661                }
662                // exitCode marks the stream's end; graphql-transport-ws
663                // follows it with its own "complete" message, which fires
664                // on_complete.
665            }),
666            on_error,
667            on_complete,
668        )?;
669        Ok(Subscription { transport, id })
670    }
671}
672
673// ---------------------------------------------------------------------------
674// run builders
675// ---------------------------------------------------------------------------
676
677/// The BSD to boot with [`Client::run_bsd`].
678#[derive(Debug, Clone, Copy, PartialEq, Eq)]
679pub enum BsdOs {
680    Freebsd,
681    Netbsd,
682}
683
684impl BsdOs {
685    fn graphql(self) -> &'static str {
686        match self {
687            BsdOs::Freebsd => "FREEBSD",
688            BsdOs::Netbsd => "NETBSD",
689        }
690    }
691}
692
693fn net_input(net: &NetOpts) -> Value {
694    if !net.touched {
695        return Value::Null;
696    }
697    json!({
698        "noNet": net.no_net,
699        "ports": net.ports,
700        "mac": net.mac,
701        "network": net.network,
702        "name": net.name,
703    })
704}
705
706fn launch_mutation(client: &Client, mutation: &str, key: &str, input: Value) -> Result<String> {
707    let data = client.request(mutation, json!({"input": input}))?;
708    data.get(key)
709        .and_then(Value::as_str)
710        .map(str::to_string)
711        .ok_or_else(|| Error::GraphQL {
712            message: format!("the daemon's {key} response carried no machine id"),
713            code: None,
714        })
715}
716
717// The remote builders share the same net/vm option groups as the local create
718// builders; the macros keep them from drifting apart between the seven run_*
719// mutations, exactly as `NetInput` is one shared input object in the schema.
720macro_rules! remote_net_vm_setters {
721    () => {
722        /// vCPU count.
723        pub fn cpus(mut self, cpus: u32) -> Self {
724            self.cpus = Some(cpus);
725            self
726        }
727
728        /// Guest RAM in MiB.
729        pub fn mem(mut self, mib: u32) -> Self {
730            self.mem = Some(mib);
731            self
732        }
733
734        /// Add a host->guest TCP port forward, `"HOST:GUEST"`.
735        pub fn port(mut self, forward: impl Into<String>) -> Self {
736            self.net.touched = true;
737            self.net.ports.push(forward.into());
738            self
739        }
740
741        /// Add a port forward from numbers instead of a string.
742        pub fn forward(self, host: u16, guest: u16) -> Self {
743            self.port(format!("{host}:{guest}"))
744        }
745
746        /// Pin the guest MAC address.
747        pub fn mac(mut self, mac: impl Into<String>) -> Self {
748            self.net.touched = true;
749            self.net.mac = Some(mac.into());
750            self
751        }
752
753        /// Join a global network.
754        pub fn network(mut self, network: impl Into<String>) -> Self {
755            self.net.touched = true;
756            self.net.network = Some(network.into());
757            self
758        }
759
760        /// Name the machine (the `NetInput.name` field).
761        pub fn name(mut self, name: impl Into<String>) -> Self {
762            self.net.touched = true;
763            self.net.name = Some(name.into());
764            self
765        }
766
767        /// Disable guest networking entirely.
768        pub fn no_net(mut self) -> Self {
769            self.net.touched = true;
770            self.net.no_net = true;
771            self
772        }
773    };
774}
775
776/// A `runLinux` mutation being assembled — see [`Client::run_linux`].
777pub struct RunLinuxBuilder {
778    client: Client,
779    image: Option<String>,
780    cpus: Option<u32>,
781    mem: Option<u32>,
782    net: NetOpts,
783    volume: Option<String>,
784    mounts: Vec<String>,
785    attach_disk: Vec<String>,
786    env: Vec<String>,
787    entrypoint: Option<String>,
788    initramfs: bool,
789    kernel: Option<String>,
790    kernel_version: Option<String>,
791    console: Option<String>,
792    repo: Option<String>,
793    command: Vec<String>,
794}
795
796impl RunLinuxBuilder {
797    remote_net_vm_setters!();
798
799    /// The OCI image to boot (required).
800    pub fn image(mut self, image: impl Into<String>) -> Self {
801        self.image = Some(image.into());
802        self
803    }
804
805    /// Use a persistent CoW volume as the rootfs.
806    pub fn volume(mut self, name: impl Into<String>) -> Self {
807        self.volume = Some(name.into());
808        self
809    }
810
811    /// Share a host directory into the guest, `"HOST:GUEST"` (repeatable).
812    pub fn mount(mut self, mount: impl Into<String>) -> Self {
813        self.mounts.push(mount.into());
814        self
815    }
816
817    /// Attach a raw disk image as virtio-blk, `"PATH"` or `"PATH:ro"`
818    /// (repeatable).
819    pub fn attach_disk(mut self, disk: impl Into<String>) -> Self {
820        self.attach_disk.push(disk.into());
821        self
822    }
823
824    /// Set a guest environment variable.
825    pub fn env(mut self, key: impl Into<String>, value: impl Into<String>) -> Self {
826        self.env.push(format!("{}={}", key.into(), value.into()));
827        self
828    }
829
830    /// Override the image entrypoint.
831    pub fn entrypoint(mut self, entrypoint: impl Into<String>) -> Self {
832        self.entrypoint = Some(entrypoint.into());
833        self
834    }
835
836    /// Boot through an initramfs.
837    pub fn initramfs(mut self) -> Self {
838        self.initramfs = true;
839        self
840    }
841
842    /// Custom kernel image.
843    pub fn kernel(mut self, kernel: impl Into<String>) -> Self {
844        self.kernel = Some(kernel.into());
845        self
846    }
847
848    /// Kernel version to fetch.
849    pub fn kernel_version(mut self, version: impl Into<String>) -> Self {
850        self.kernel_version = Some(version.into());
851        self
852    }
853
854    /// Console device.
855    pub fn console(mut self, console: impl Into<String>) -> Self {
856        self.console = Some(console.into());
857        self
858    }
859
860    /// Clone a git repo into the guest before running.
861    pub fn repo(mut self, repo: impl Into<String>) -> Self {
862        self.repo = Some(repo.into());
863        self
864    }
865
866    /// The command to run in the guest.
867    pub fn command<I, S>(mut self, command: I) -> Self
868    where
869        I: IntoIterator<Item = S>,
870        S: Into<String>,
871    {
872        self.command = strvec(command);
873        self
874    }
875
876    /// Boot the machine and return its id.
877    pub fn launch(self) -> Result<String> {
878        let Some(image) = self.image else {
879            return Err(Error::InvalidInput("run_linux requires an image".into()));
880        };
881        let input = json!({
882            "image": image,
883            "cpus": self.cpus,
884            "mem": self.mem,
885            "net": net_input(&self.net),
886            "volume": self.volume,
887            "mounts": self.mounts,
888            "attachDisk": self.attach_disk,
889            "env": self.env,
890            "entrypoint": self.entrypoint,
891            "initramfs": self.initramfs,
892            "kernel": self.kernel,
893            "kernelVersion": self.kernel_version,
894            "console": self.console,
895            "repo": self.repo,
896            "command": self.command,
897        });
898        launch_mutation(&self.client, RUN_LINUX_MUTATION, "runLinux", input)
899    }
900}
901
902/// A `runBsd` mutation being assembled — see [`Client::run_bsd`].
903pub struct RunBsdBuilder {
904    client: Client,
905    os: BsdOs,
906    version: Option<String>,
907    cpus: Option<u32>,
908    mem: Option<u32>,
909    net: NetOpts,
910    volume: Option<String>,
911    persist: bool,
912    force: bool,
913    firmware: Option<String>,
914    attach_disk: Vec<String>,
915    disk_size: Option<String>,
916    repo: Option<String>,
917    command: Vec<String>,
918}
919
920impl RunBsdBuilder {
921    remote_net_vm_setters!();
922
923    /// The release to boot.
924    pub fn version(mut self, version: impl Into<String>) -> Self {
925        self.version = Some(version.into());
926        self
927    }
928
929    /// Use a persistent CoW volume as the root disk.
930    pub fn volume(mut self, name: impl Into<String>) -> Self {
931        self.volume = Some(name.into());
932        self
933    }
934
935    /// Keep the root disk across `rm`.
936    pub fn persist(mut self) -> Self {
937        self.persist = true;
938        self
939    }
940
941    /// Re-fetch the image even if cached.
942    pub fn force(mut self) -> Self {
943        self.force = true;
944        self
945    }
946
947    /// Custom EFI firmware.
948    pub fn firmware(mut self, firmware: impl Into<String>) -> Self {
949        self.firmware = Some(firmware.into());
950        self
951    }
952
953    /// Attach an extra raw disk, `"PATH"` or `"PATH:ro"` (repeatable).
954    pub fn attach_disk(mut self, disk: impl Into<String>) -> Self {
955        self.attach_disk.push(disk.into());
956        self
957    }
958
959    /// Root disk size, e.g. `"20G"`.
960    pub fn disk_size(mut self, size: impl Into<String>) -> Self {
961        self.disk_size = Some(size.into());
962        self
963    }
964
965    /// Clone a git repo into the guest before running.
966    pub fn repo(mut self, repo: impl Into<String>) -> Self {
967        self.repo = Some(repo.into());
968        self
969    }
970
971    /// The command to run in the guest.
972    pub fn command<I, S>(mut self, command: I) -> Self
973    where
974        I: IntoIterator<Item = S>,
975        S: Into<String>,
976    {
977        self.command = strvec(command);
978        self
979    }
980
981    /// Boot the machine and return its id.
982    pub fn launch(self) -> Result<String> {
983        let input = json!({
984            "os": self.os.graphql(),
985            "version": self.version,
986            "cpus": self.cpus,
987            "mem": self.mem,
988            "net": net_input(&self.net),
989            "volume": self.volume,
990            "persist": self.persist,
991            "force": self.force,
992            "firmware": self.firmware,
993            "attachDisk": self.attach_disk,
994            "diskSize": self.disk_size,
995            "repo": self.repo,
996            "command": self.command,
997        });
998        launch_mutation(&self.client, RUN_BSD_MUTATION, "runBsd", input)
999    }
1000}
1001
1002/// A `runNanos` mutation being assembled — see [`Client::run_nanos`].
1003pub struct RunNanosBuilder {
1004    client: Client,
1005    image: Option<String>,
1006    cpus: Option<u32>,
1007    mem: Option<u32>,
1008    net: NetOpts,
1009    kernel: Option<String>,
1010    cmdline: Option<String>,
1011    persist: bool,
1012}
1013
1014impl RunNanosBuilder {
1015    remote_net_vm_setters!();
1016
1017    /// A path, or a bare name in `~/.ops/images` (required).
1018    pub fn image(mut self, image: impl Into<String>) -> Self {
1019        self.image = Some(image.into());
1020        self
1021    }
1022
1023    /// Nanos kernel override (Linux hosts).
1024    pub fn kernel(mut self, kernel: impl Into<String>) -> Self {
1025        self.kernel = Some(kernel.into());
1026        self
1027    }
1028
1029    /// Kernel command line.
1030    pub fn cmdline(mut self, cmdline: impl Into<String>) -> Self {
1031        self.cmdline = Some(cmdline.into());
1032        self
1033    }
1034
1035    /// Keep the root disk across `rm`.
1036    pub fn persist(mut self) -> Self {
1037        self.persist = true;
1038        self
1039    }
1040
1041    /// Boot the unikernel and return its machine id.
1042    pub fn launch(self) -> Result<String> {
1043        let Some(image) = self.image else {
1044            return Err(Error::InvalidInput("run_nanos requires an image".into()));
1045        };
1046        let input = json!({
1047            "image": image,
1048            "cpus": self.cpus,
1049            "mem": self.mem,
1050            "net": net_input(&self.net),
1051            "kernel": self.kernel,
1052            "cmdline": self.cmdline,
1053            "persist": self.persist,
1054        });
1055        launch_mutation(&self.client, RUN_NANOS_MUTATION, "runNanos", input)
1056    }
1057}
1058
1059/// A `runUnikraft` mutation being assembled — see [`Client::run_unikraft`].
1060pub struct RunUnikraftBuilder {
1061    client: Client,
1062    path: Option<String>,
1063    cpus: Option<u32>,
1064    mem: Option<u32>,
1065    net: NetOpts,
1066    cmdline: Option<String>,
1067    initramfs: Option<String>,
1068    mounts: Vec<String>,
1069}
1070
1071impl RunUnikraftBuilder {
1072    remote_net_vm_setters!();
1073
1074    /// A `kraft` project directory or a built unikernel image (defaults to `.`).
1075    pub fn path(mut self, path: impl Into<String>) -> Self {
1076        self.path = Some(path.into());
1077        self
1078    }
1079
1080    /// Kernel command line; Unikraft hands it to the application as argv.
1081    pub fn cmdline(mut self, cmdline: impl Into<String>) -> Self {
1082        self.cmdline = Some(cmdline.into());
1083        self
1084    }
1085
1086    /// Initramfs image path.
1087    pub fn initramfs(mut self, path: impl Into<String>) -> Self {
1088        self.initramfs = Some(path.into());
1089        self
1090    }
1091
1092    /// A virtio-fs share, `"HOST:GUEST"` with an absolute guest path
1093    /// (repeatable). Needs a unikernel built for it.
1094    pub fn mount(mut self, mount: impl Into<String>) -> Self {
1095        self.mounts.push(mount.into());
1096        self
1097    }
1098
1099    /// Boot the unikernel and return its machine id.
1100    pub fn launch(self) -> Result<String> {
1101        let input = json!({
1102            "path": self.path,
1103            "cpus": self.cpus,
1104            "mem": self.mem,
1105            "net": net_input(&self.net),
1106            "cmdline": self.cmdline,
1107            "initramfs": self.initramfs,
1108            "mounts": self.mounts,
1109        });
1110        launch_mutation(&self.client, RUN_UNIKRAFT_MUTATION, "runUnikraft", input)
1111    }
1112}
1113
1114/// A `runSolo5` mutation being assembled — see [`Client::run_solo5`].
1115pub struct RunSolo5Builder {
1116    client: Client,
1117    path: Option<String>,
1118    cpus: Option<u32>,
1119    mem: Option<u32>,
1120    net: NetOpts,
1121    block: Vec<String>,
1122    args: Vec<String>,
1123}
1124
1125impl RunSolo5Builder {
1126    remote_net_vm_setters!();
1127
1128    /// A `.hvt` binary, or a project directory whose `dist/` holds one
1129    /// (defaults to `.`).
1130    pub fn path(mut self, path: impl Into<String>) -> Self {
1131        self.path = Some(path.into());
1132        self
1133    }
1134
1135    /// Backing file for a declared block device, `"NAME=FILE"` (repeatable).
1136    pub fn block(mut self, block: impl Into<String>) -> Self {
1137        self.block.push(block.into());
1138        self
1139    }
1140
1141    /// Arguments passed to the unikernel itself (e.g. `--ipv4=10.0.0.2/24`).
1142    pub fn args<I, S>(mut self, args: I) -> Self
1143    where
1144        I: IntoIterator<Item = S>,
1145        S: Into<String>,
1146    {
1147        self.args = strvec(args);
1148        self
1149    }
1150
1151    /// Boot the unikernel and return its machine id.
1152    pub fn launch(self) -> Result<String> {
1153        let input = json!({
1154            "path": self.path,
1155            "cpus": self.cpus,
1156            "mem": self.mem,
1157            "net": net_input(&self.net),
1158            "block": self.block,
1159            "args": self.args,
1160        });
1161        launch_mutation(&self.client, RUN_SOLO5_MUTATION, "runSolo5", input)
1162    }
1163}
1164
1165/// A `runOsv` mutation being assembled — see [`Client::run_osv`].
1166pub struct RunOsvBuilder {
1167    client: Client,
1168    image: Option<String>,
1169    cpus: Option<u32>,
1170    mem: Option<u32>,
1171    net: NetOpts,
1172    cmdline: Option<String>,
1173    disk: Option<String>,
1174    no_disk: bool,
1175    attach_disk: Vec<String>,
1176    gic: Option<String>,
1177    persist: bool,
1178    volume: Option<String>,
1179}
1180
1181impl RunOsvBuilder {
1182    remote_net_vm_setters!();
1183
1184    /// An aarch64 `loader.img`, or on x86_64 the loader ELF (required).
1185    pub fn image(mut self, image: impl Into<String>) -> Self {
1186        self.image = Some(image.into());
1187        self
1188    }
1189
1190    /// The application to run and its arguments, e.g. `"/hello.so"`.
1191    pub fn cmdline(mut self, cmdline: impl Into<String>) -> Self {
1192        self.cmdline = Some(cmdline.into());
1193        self
1194    }
1195
1196    /// Root disk (raw). Required on x86_64.
1197    pub fn disk(mut self, disk: impl Into<String>) -> Self {
1198        self.disk = Some(disk.into());
1199        self
1200    }
1201
1202    /// Boot the kernel alone, with no root filesystem to mount.
1203    pub fn no_disk(mut self) -> Self {
1204        self.no_disk = true;
1205        self
1206    }
1207
1208    /// Extra disks as virtio-blk, `"PATH"` or `"PATH:ro"` (repeatable).
1209    pub fn attach_disk(mut self, disk: impl Into<String>) -> Self {
1210        self.attach_disk.push(disk.into());
1211        self
1212    }
1213
1214    /// `"v2"` (the default) or `"v3"`. aarch64 only.
1215    pub fn gic(mut self, gic: impl Into<String>) -> Self {
1216        self.gic = Some(gic.into());
1217        self
1218    }
1219
1220    /// Keep the root disk across `rm`.
1221    pub fn persist(mut self) -> Self {
1222        self.persist = true;
1223        self
1224    }
1225
1226    /// Use a persistent CoW volume as the root disk.
1227    pub fn volume(mut self, name: impl Into<String>) -> Self {
1228        self.volume = Some(name.into());
1229        self
1230    }
1231
1232    /// Boot the unikernel and return its machine id.
1233    pub fn launch(self) -> Result<String> {
1234        let Some(image) = self.image else {
1235            return Err(Error::InvalidInput("run_osv requires an image".into()));
1236        };
1237        let input = json!({
1238            "image": image,
1239            "cpus": self.cpus,
1240            "mem": self.mem,
1241            "net": net_input(&self.net),
1242            "cmdline": self.cmdline,
1243            "disk": self.disk,
1244            "noDisk": self.no_disk,
1245            "attachDisk": self.attach_disk,
1246            "gic": self.gic,
1247            "persist": self.persist,
1248            "volume": self.volume,
1249        });
1250        launch_mutation(&self.client, RUN_OSV_MUTATION, "runOsv", input)
1251    }
1252}
1253
1254/// A `runFlavor` mutation being assembled — see [`Client::run_flavor`].
1255pub struct RunFlavorBuilder {
1256    client: Client,
1257    name: String,
1258    cpus: Option<u32>,
1259    mem: Option<u32>,
1260    ports: Vec<String>,
1261    volume: Option<String>,
1262    repo: Option<String>,
1263}
1264
1265impl RunFlavorBuilder {
1266    /// vCPU count.
1267    pub fn cpus(mut self, cpus: u32) -> Self {
1268        self.cpus = Some(cpus);
1269        self
1270    }
1271
1272    /// Guest RAM in MiB.
1273    pub fn mem(mut self, mib: u32) -> Self {
1274        self.mem = Some(mib);
1275        self
1276    }
1277
1278    /// Add a host->guest TCP port forward, `"HOST:GUEST"`.
1279    pub fn port(mut self, forward: impl Into<String>) -> Self {
1280        self.ports.push(forward.into());
1281        self
1282    }
1283
1284    /// Use a persistent CoW volume as the root disk.
1285    pub fn volume(mut self, name: impl Into<String>) -> Self {
1286        self.volume = Some(name.into());
1287        self
1288    }
1289
1290    /// Clone a git repo into the guest before running.
1291    pub fn repo(mut self, repo: impl Into<String>) -> Self {
1292        self.repo = Some(repo.into());
1293        self
1294    }
1295
1296    /// Boot the flavor and return its machine id.
1297    pub fn launch(self) -> Result<String> {
1298        let input = json!({
1299            "name": self.name,
1300            "cpus": self.cpus,
1301            "mem": self.mem,
1302            "ports": self.ports,
1303            "volume": self.volume,
1304            "repo": self.repo,
1305        });
1306        launch_mutation(&self.client, RUN_FLAVOR_MUTATION, "runFlavor", input)
1307    }
1308}
1309
1310// ---------------------------------------------------------------------------
1311// interactive shell sessions
1312// ---------------------------------------------------------------------------
1313
1314/// An `openShell` mutation being assembled — see [`Client::shell`].
1315pub struct ShellBuilder {
1316    client: Client,
1317    machine_id: String,
1318    command: Vec<String>,
1319    env: Vec<String>,
1320    rows: u32,
1321    cols: u32,
1322}
1323
1324impl ShellBuilder {
1325    /// Run this command instead of the machine's login shell.
1326    pub fn command<I, S>(mut self, command: I) -> Self
1327    where
1328        I: IntoIterator<Item = S>,
1329        S: Into<String>,
1330    {
1331        self.command = strvec(command);
1332        self
1333    }
1334
1335    /// Set a session environment variable.
1336    pub fn env(mut self, key: impl Into<String>, value: impl Into<String>) -> Self {
1337        self.env.push(format!("{}={}", key.into(), value.into()));
1338        self
1339    }
1340
1341    /// Terminal rows (default 24).
1342    pub fn rows(mut self, rows: u32) -> Self {
1343        self.rows = rows;
1344        self
1345    }
1346
1347    /// Terminal columns (default 80).
1348    pub fn cols(mut self, cols: u32) -> Self {
1349        self.cols = cols;
1350        self
1351    }
1352
1353    /// Open the session and start streaming its output.
1354    pub fn open(self) -> Result<ShellSession> {
1355        let data = self.client.request(
1356            &open_shell_mutation(),
1357            json!({
1358                "machineId": self.machine_id,
1359                "command": self.command,
1360                "env": self.env,
1361                "rows": self.rows,
1362                "cols": self.cols,
1363            }),
1364        )?;
1365        let info = ShellSessionInfo::from_graphql(&data["openShell"]);
1366        ShellSession::start(self.client, info.id)
1367    }
1368}
1369
1370type OutputFn = Box<dyn FnMut(&[u8]) + Send>;
1371type ExitFn = Box<dyn FnMut(i32) + Send>;
1372
1373struct ShellShared {
1374    output_cb: Option<OutputFn>,
1375    exit_cb: Option<ExitFn>,
1376    /// Anything that arrives *before* a callback is registered — a real
1377    /// possibility, since the subscription starts inside `open()` and the
1378    /// daemon can reply before the caller registers anything — is buffered
1379    /// and flushed the moment a callback is set, so no frame is silently
1380    /// lost.
1381    buffered_output: Vec<Vec<u8>>,
1382    buffered_exit: Option<i32>,
1383    exit_fired: bool,
1384}
1385
1386/// A live interactive session opened by [`Client::shell`].
1387///
1388/// Output and exit events arrive on the shared WS transport's reader thread
1389/// and are handed to whatever callbacks are registered via
1390/// [`ShellSession::on_output`] / [`ShellSession::on_exit`] at the time they
1391/// arrive. Callbacks run holding the session's internal lock, so they must
1392/// not call `on_output`/`on_exit` themselves (writing and resizing is fine).
1393pub struct ShellSession {
1394    id: String,
1395    client: Client,
1396    transport: Arc<WsTransport>,
1397    sub_id: String,
1398    shared: Arc<Mutex<ShellShared>>,
1399    closed: bool,
1400}
1401
1402impl std::fmt::Debug for ShellSession {
1403    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1404        f.debug_struct("ShellSession")
1405            .field("id", &self.id)
1406            .finish()
1407    }
1408}
1409
1410impl ShellSession {
1411    fn start(client: Client, session_id: String) -> Result<ShellSession> {
1412        let shared = Arc::new(Mutex::new(ShellShared {
1413            output_cb: None,
1414            exit_cb: None,
1415            buffered_output: Vec::new(),
1416            buffered_exit: None,
1417            exit_fired: false,
1418        }));
1419        let transport = client.ws();
1420
1421        let on_next_shared = Arc::clone(&shared);
1422        let on_error_shared = Arc::clone(&shared);
1423        let sub_id = transport.subscribe(
1424            SHELL_OUTPUT_SUBSCRIPTION,
1425            json!({"sessionId": session_id}),
1426            Box::new(move |data: Value| {
1427                let payload = &data["shellOutput"];
1428                if let Some(b64) = payload["dataBase64"].as_str() {
1429                    if let Ok(bytes) = B64.decode(b64) {
1430                        emit_output(&on_next_shared, bytes);
1431                    }
1432                }
1433                if let Some(code) = payload["exitCode"].as_i64() {
1434                    emit_exit(&on_next_shared, code as i32);
1435                }
1436            }),
1437            Box::new(move |_err: Error| {
1438                // A dropped connection ends the session the same way an exit
1439                // would, so a caller has one place (on_exit) to notice the
1440                // session is gone. -1 has no exit-code meaning of its own; it
1441                // just isn't 0.
1442                emit_exit(&on_error_shared, -1);
1443            }),
1444            Box::new(|| {}),
1445        )?;
1446
1447        Ok(ShellSession {
1448            id: session_id,
1449            client,
1450            transport,
1451            sub_id,
1452            shared,
1453            closed: false,
1454        })
1455    }
1456
1457    /// The session id, as `openShell` returned it.
1458    pub fn id(&self) -> &str {
1459        &self.id
1460    }
1461
1462    /// Register the output callback; anything buffered so far is flushed to
1463    /// it immediately.
1464    pub fn on_output(&self, mut cb: impl FnMut(&[u8]) + Send + 'static) {
1465        let mut shared = self.shared.lock().unwrap();
1466        for chunk in std::mem::take(&mut shared.buffered_output) {
1467            cb(&chunk);
1468        }
1469        shared.output_cb = Some(Box::new(cb));
1470    }
1471
1472    /// Register the exit callback; a buffered exit fires immediately.
1473    pub fn on_exit(&self, mut cb: impl FnMut(i32) + Send + 'static) {
1474        let mut shared = self.shared.lock().unwrap();
1475        if let Some(code) = shared.buffered_exit.take() {
1476            cb(code);
1477        }
1478        shared.exit_cb = Some(Box::new(cb));
1479    }
1480
1481    /// Send keystrokes (arbitrary bytes) to the session.
1482    pub fn write(&self, data: impl AsRef<[u8]>) -> Result<()> {
1483        self.client.request(
1484            SEND_INPUT_MUTATION,
1485            json!({"sessionId": self.id, "dataBase64": B64.encode(data.as_ref())}),
1486        )?;
1487        Ok(())
1488    }
1489
1490    /// Apply a terminal resize, so full-screen programs in the guest redraw.
1491    pub fn resize(&self, rows: u32, cols: u32) -> Result<()> {
1492        self.client.request(
1493            RESIZE_MUTATION,
1494            json!({"sessionId": self.id, "rows": rows, "cols": cols}),
1495        )?;
1496        Ok(())
1497    }
1498
1499    /// Close the session and kill its command. Idempotent.
1500    pub fn close(&mut self) {
1501        if self.closed {
1502            return;
1503        }
1504        self.closed = true;
1505        self.transport.unsubscribe(&self.sub_id);
1506        // closeShell is idempotent; an already-gone session is not a failure.
1507        let _ = self
1508            .client
1509            .request(CLOSE_MUTATION, json!({"sessionId": self.id}));
1510    }
1511}
1512
1513fn emit_output(shared: &Arc<Mutex<ShellShared>>, data: Vec<u8>) {
1514    let mut guard = shared.lock().unwrap();
1515    match &mut guard.output_cb {
1516        Some(cb) => cb(&data),
1517        None => guard.buffered_output.push(data),
1518    }
1519}
1520
1521fn emit_exit(shared: &Arc<Mutex<ShellShared>>, code: i32) {
1522    let mut guard = shared.lock().unwrap();
1523    if guard.exit_fired {
1524        return;
1525    }
1526    guard.exit_fired = true;
1527    match &mut guard.exit_cb {
1528        Some(cb) => cb(code),
1529        None => guard.buffered_exit = Some(code),
1530    }
1531}
bsdkrun_sdk/client.rs

bsdkrun_sdk/
client.rs
