dvm_run/
client.rs

1//! The connection to dvm.run's API: its address, the key every request
2//! bears, what a refusal means, and server-sent events.
3
4use std::sync::Arc;
5use std::time::Duration;
6
7use dvm_api::{ApiError, SandboxId};
8use reqwest::{Method, RequestBuilder, Response};
9use serde::Serialize;
10use serde::de::DeserializeOwned;
11
12use crate::credentials;
13use crate::error::{Error, Result};
14use crate::execution::Execution;
15use crate::sandbox::{Sandbox, SandboxBuilder};
16
17pub const DEFAULT_API_URL: &str = "https://sandbox.dvm.run";
18
19/// The controller answers a `?newer_than=` read within 30 s, and keeps a
20/// quiet event stream alive with comments more often than this.
21const QUIET_LIMIT: Duration = Duration::from_secs(60);
22
23/// dvm.run's API, at an address, as the user whose key it bears. Cheap to
24/// clone: clones share their connections.
25#[derive(Clone, Debug)]
26pub struct Client {
27    inner: Arc<Connection>,
28}
29
30#[derive(Debug)]
31struct Connection {
32    url: String,
33    api: String,
34    key: Option<String>,
35    http: reqwest::Client,
36}
37
38pub(crate) type Query<'a> = &'a [(&'a str, String)];
39
40impl Client {
41    /// dvm.run at `DVM_API_URL`, else `https://sandbox.dvm.run`, with the key
42    /// in `DVM_API_KEY`, else the one `dvm-run login` saved for it.
43    pub fn from_env() -> Result<Self> {
44        let api_url = api_url_of(None);
45        let key = match std::env::var("DVM_API_KEY") {
46            Ok(key) if !key.is_empty() => Some(key),
47            _ => credentials::saved_key(&api_url)?,
48        };
49        let key = key.ok_or_else(|| Error::NoApiKey {
50            api_url: api_url.clone(),
51        })?;
52        Ok(Self::new(api_url, key))
53    }
54
55    /// dvm.run at `api_url`, with `api_key`.
56    pub fn new(api_url: impl Into<String>, api_key: impl Into<String>) -> Self {
57        Self::with_key(api_url.into(), Some(api_key.into()))
58    }
59
60    /// As no one, for what needs no key: a login.
61    pub(crate) fn with_key(api_url: String, key: Option<String>) -> Self {
62        let url = api_url.trim_end_matches('/').to_owned();
63        Self {
64            inner: Arc::new(Connection {
65                api: format!("{url}/api"),
66                url,
67                key,
68                http: reqwest::Client::new(),
69            }),
70        }
71    }
72
73    /// The dvm.run it calls.
74    pub fn api_url(&self) -> &str {
75        &self.inner.url
76    }
77
78    /// A sandbox of `image`, as `docker run` takes it, once it's running.
79    pub fn create_sandbox(&self, image: impl Into<String>) -> SandboxBuilder {
80        SandboxBuilder::new(self.clone(), image.into())
81    }
82
83    /// A handle to a sandbox made elsewhere.
84    pub async fn connect_sandbox(&self, id: SandboxId) -> Result<Sandbox> {
85        let sandbox = self
86            .json(Method::GET, &format!("/sandboxes/{id}"), &[], None::<&()>)
87            .await?;
88        Ok(Sandbox::new(self.clone(), sandbox))
89    }
90
91    /// A sandbox's execution: everything it ran, under its id.
92    pub async fn execution(&self, id: SandboxId) -> Result<Execution> {
93        let execution = self
94            .json(Method::GET, &format!("/executions/{id}"), &[], None::<&()>)
95            .await?;
96        Ok(Execution::new(self.clone(), execution))
97    }
98
99    /// The answer's JSON; an empty answer is `null`.
100    pub(crate) async fn json<T: DeserializeOwned>(
101        &self,
102        method: Method,
103        path: &str,
104        query: Query<'_>,
105        body: Option<&impl Serialize>,
106    ) -> Result<T> {
107        let mut request = self.request(method, path, query).timeout(QUIET_LIMIT);
108        if let Some(body) = body {
109            request = request.json(body);
110        }
111        let bytes = self.send(path, request).await?.bytes().await;
112        let bytes = bytes.map_err(|error| Error::network(format!("reading {path}"), error))?;
113        let bytes = if bytes.is_empty() {
114            &b"null"[..]
115        } else {
116            &bytes
117        };
118        serde_json::from_slice(bytes)
119            .map_err(|error| Error::Unexpected(format!("{path}'s answer: {error}")))
120    }
121
122    /// The answer's body, as it is.
123    pub(crate) async fn bytes(
124        &self,
125        method: Method,
126        path: &str,
127        query: Query<'_>,
128        content: Option<Vec<u8>>,
129    ) -> Result<Vec<u8>> {
130        let mut request = self.request(method, path, query);
131        if let Some(content) = content {
132            request = request.body(content);
133        }
134        let bytes = self.send(path, request).await?.bytes().await;
135        Ok(bytes
136            .map_err(|error| Error::network(format!("reading {path}"), error))?
137            .to_vec())
138    }
139
140    /// The server-sent events at `path`.
141    pub(crate) async fn events(&self, path: &str, query: Query<'_>) -> Result<EventStream> {
142        let request = self.request(Method::GET, path, query);
143        let answered = tokio::time::timeout(QUIET_LIMIT, self.send(path, request)).await;
144        let response = answered.map_err(|_| Error::Network {
145            context: format!("{path} didn't answer"),
146            source: None,
147        })??;
148        Ok(EventStream {
149            path: path.to_owned(),
150            response,
151            pending: Vec::new(),
152            data: Vec::new(),
153        })
154    }
155
156    fn request(&self, method: Method, path: &str, query: Query<'_>) -> RequestBuilder {
157        let request = self
158            .inner
159            .http
160            .request(method, format!("{}{path}", self.inner.api));
161        let request = if query.is_empty() {
162            request
163        } else {
164            request.query(query)
165        };
166        match &self.inner.key {
167            Some(key) => request.bearer_auth(key),
168            None => request,
169        }
170    }
171
172    /// The response, refused unless it's a success.
173    async fn send(&self, path: &str, request: RequestBuilder) -> Result<Response> {
174        let response = request.send().await.map_err(|error| {
175            Error::network(format!("couldn't reach {}{path}", self.inner.api), error)
176        })?;
177        let status = response.status();
178        if status.is_success() {
179            return Ok(response);
180        }
181        let text = response.text().await.unwrap_or_default();
182        let reason = match serde_json::from_str::<ApiError>(&text) {
183            Ok(refusal) => refusal.message,
184            Err(_) if text.is_empty() => status.canonical_reason().unwrap_or("").to_owned(),
185            Err(_) => text,
186        };
187        Err(Error::refusal(status.as_u16(), reason))
188    }
189}
190
191/// Each server-sent event's data, as JSON, until the stream ends. A stream
192/// that goes quiet for longer than the controller lets one is dropped.
193pub(crate) struct EventStream {
194    path: String,
195    response: Response,
196    pending: Vec<u8>,
197    data: Vec<u8>,
198}
199
200impl EventStream {
201    pub(crate) async fn next<T: DeserializeOwned>(&mut self) -> Result<Option<T>> {
202        loop {
203            while let Some(end) = self.pending.iter().position(|&byte| byte == b'\n') {
204                let mut line: Vec<u8> = self.pending.drain(..=end).collect();
205                line.pop();
206                if line.last() == Some(&b'\r') {
207                    line.pop();
208                }
209                if let Some(data) = line.strip_prefix(b"data:") {
210                    if !self.data.is_empty() {
211                        self.data.push(b'\n');
212                    }
213                    self.data
214                        .extend_from_slice(data.strip_prefix(b" ").unwrap_or(data));
215                } else if line.is_empty() && !self.data.is_empty() {
216                    let data = std::mem::take(&mut self.data);
217                    let event = serde_json::from_slice(&data).map_err(|error| {
218                        Error::Unexpected(format!("{}'s event: {error}", self.path))
219                    })?;
220                    return Ok(Some(event));
221                }
222            }
223            let chunk = tokio::time::timeout(QUIET_LIMIT, self.response.chunk()).await;
224            let chunk = chunk.map_err(|_| Error::Network {
225                context: format!("{}'s stream went quiet", self.path),
226                source: None,
227            })?;
228            match chunk.map_err(|error| Error::network(format!("{}'s stream", self.path), error))? {
229                Some(chunk) => self.pending.extend_from_slice(&chunk),
230                None => return Ok(None),
231            }
232        }
233    }
234}
235
236/// dvm.run's address: `given`, else `DVM_API_URL`, else dvm.run's.
237pub(crate) fn api_url_of(given: Option<&str>) -> String {
238    let from_environment = std::env::var("DVM_API_URL").ok();
239    let url = given
240        .filter(|url| !url.is_empty())
241        .or(from_environment.as_deref().filter(|url| !url.is_empty()))
242        .unwrap_or(DEFAULT_API_URL);
243    url.trim_end_matches('/').to_owned()
244}
dvm_run/client.rs

dvm_run/
client.rs
