libsurfer/remote/
client.rs

1use std::fmt::Write as _;
2use std::sync::Arc;
3use std::sync::OnceLock;
4use std::sync::mpsc::Sender;
5
6use bincode::Options;
7use eyre::{Result, WrapErr as _, anyhow, bail, eyre};
8use reqwest::StatusCode;
9use thiserror::Error;
10use tracing::{error, info, warn};
11use wellen::CompressedTimeTable;
12
13use surver::{
14    BINCODE_OPTIONS, HTTP_SERVER_KEY, HTTP_SERVER_VALUE_SURFER, SURFER_VERSION, SurverStatus,
15    WELLEN_VERSION, X_SURFER_VERSION, X_WELLEN_VERSION,
16};
17
18use super::HierarchyResponse;
19#[cfg(all(not(target_arch = "wasm32"), feature = "https"))]
20use crate::async_util::ensure_rustls_crypto_provider;
21use crate::async_util::{perform_async_work, sleep_ms};
22use crate::channels::checked_send;
23use crate::message::Message;
24use crate::wave_source::{LoadOptions, WaveSource};
25use crate::wellen::{BodyResult, HeaderResult};
26
27/// Returns a shared reqwest client to reuse HTTP connections and reduce TLS overhead.
28fn get_client() -> &'static reqwest::Client {
29    static CLIENT: OnceLock<reqwest::Client> = OnceLock::new();
30    #[cfg(all(not(target_arch = "wasm32"), feature = "https"))]
31    ensure_rustls_crypto_provider();
32    CLIENT.get_or_init(reqwest::Client::new)
33}
34
35#[derive(Debug, Error)]
36pub enum ReloadError {
37    #[error("File unchanged since last reload")]
38    FileUnchanged,
39    #[error("Unexpected response code: {0}")]
40    UnexpectedStatus(StatusCode),
41    #[error("Network error: {0}")]
42    Network(#[from] reqwest::Error),
43    #[error("Parse error: {0}")]
44    Parse(#[from] serde_json::Error),
45    #[error("Response validation error: {0}")]
46    Validation(#[from] eyre::Report),
47}
48
49fn check_response(server_url: &str, response: &reqwest::Response) -> Result<()> {
50    let server = response
51        .headers()
52        .get(HTTP_SERVER_KEY)
53        .ok_or(eyre!("no server header"))?
54        .to_str()?;
55    if server != HTTP_SERVER_VALUE_SURFER {
56        bail!("Unexpected server {server} from {server_url}");
57    }
58    let surfer_version = response
59        .headers()
60        .get(X_SURFER_VERSION)
61        .ok_or(eyre!("no surfer version header"))?
62        .to_str()?;
63    if surfer_version != SURFER_VERSION {
64        // this mismatch may be OK as long as the wellen version matches
65        info!(
66            "Surfer version on the server: {surfer_version} does not match client version {SURFER_VERSION}"
67        );
68    }
69    let wellen_version = response
70        .headers()
71        .get(X_WELLEN_VERSION)
72        .ok_or(eyre!("no wellen version header"))?
73        .to_str()?;
74    if wellen_version != WELLEN_VERSION {
75        bail!(
76            "Version incompatibility! The server uses wellen {wellen_version}, our client uses wellen {WELLEN_VERSION}"
77        );
78    }
79    Ok(())
80}
81
82async fn get_status(server: String) -> Result<SurverStatus> {
83    let client = get_client();
84    let response = client.get(format!("{server}/get_status")).send().await?;
85    check_response(&server, &response)?;
86    let body = response.text().await?;
87    let status = serde_json::from_str::<SurverStatus>(&body)?;
88    Ok(status)
89}
90
91async fn reload(
92    server: String,
93    file_index: usize,
94) -> std::result::Result<SurverStatus, ReloadError> {
95    let client = get_client();
96    let response = client
97        .get(format!("{server}/{file_index}/reload"))
98        .send()
99        .await?;
100    check_response(&server, &response)?;
101    let status_code = response.status();
102    let body = response.text().await?;
103    match status_code {
104        StatusCode::NOT_MODIFIED => {
105            info!("File unchanged, no reload needed");
106            Err(ReloadError::FileUnchanged)
107        }
108        StatusCode::ACCEPTED => {
109            info!("File reloaded at server");
110            let status = serde_json::from_str::<SurverStatus>(&body)?;
111            Ok(status)
112        }
113        code => {
114            warn!("Unexpected response code: {code}");
115            Err(ReloadError::UnexpectedStatus(code))
116        }
117    }
118}
119
120async fn get_hierarchy(server: String, file_index: usize) -> Result<HierarchyResponse> {
121    let client = get_client();
122    let response = client
123        .get(format!("{server}/{file_index}/get_hierarchy"))
124        .send()
125        .await?;
126    check_response(&server, &response)?;
127    let compressed = response.bytes().await?;
128    let raw = lz4_flex::decompress_size_prepended(&compressed)?;
129    let mut reader = std::io::Cursor::new(raw);
130    // first we read a value, expecting there to be more bytes
131    let opts = BINCODE_OPTIONS.allow_trailing_bytes();
132    let file_format: wellen::FileFormat = opts.deserialize_from(&mut reader)?;
133    // the last value should consume all remaining bytes
134    let hierarchy: wellen::Hierarchy = BINCODE_OPTIONS.deserialize_from(&mut reader)?;
135    Ok(HierarchyResponse {
136        hierarchy,
137        file_format,
138    })
139}
140
141async fn get_time_table(server: String, file_index: usize) -> Result<Vec<wellen::Time>> {
142    let client = get_client();
143    let response = client
144        .get(format!("{server}/{file_index}/get_time_table"))
145        .send()
146        .await?;
147    check_response(&server, &response)?;
148    let compressed_data = response.bytes().await?;
149    let compressed: CompressedTimeTable = BINCODE_OPTIONS.deserialize(&compressed_data)?;
150    let table = compressed.uncompress();
151    Ok(table)
152}
153
154// Helper to calculate URL length for a signal index
155// Much more efficient than string conversion
156// Extracted for testing
157#[inline]
158fn signal_url_len(index: usize) -> usize {
159    index.checked_ilog10().unwrap_or(0) as usize + 2 // +1 for '/', +1 as ilog10 rounds down
160}
161
162pub async fn get_signals(
163    server: String,
164    signals: &[wellen::SignalRef],
165    max_url_length: u16,
166    file_index: usize,
167) -> Result<Vec<wellen::Signal>> {
168    if signals.is_empty() {
169        return Ok(vec![]);
170    }
171
172    let max_url_length = max_url_length as usize;
173    let base_url = format!("{server}/{file_index}/get_signals");
174    let base_len = base_url.len();
175
176    let mut all_results = Vec::with_capacity(signals.len());
177    let mut current_batch = Vec::new();
178    let mut current_url_len = base_len;
179
180    for signal in signals {
181        // Each signal adds: "/" + digits
182        let signal_len = signal_url_len(signal.index());
183
184        // Check if adding this signal would exceed the limit
185        if current_url_len + signal_len > max_url_length && !current_batch.is_empty() {
186            info!(
187                "Fetching batch of {} signals due to URL length limit",
188                current_batch.len()
189            );
190            // Fetch current batch
191            let batch_results = get_signals_batch(&base_url, &current_batch).await?;
192            all_results.extend(batch_results);
193
194            // Start new batch
195            current_batch.clear();
196            current_url_len = base_len;
197        }
198
199        current_batch.push(*signal);
200        current_url_len += signal_len;
201    }
202
203    // Fetch remaining batch
204    if !current_batch.is_empty() {
205        let batch_results = get_signals_batch(&base_url, &current_batch).await?;
206        all_results.extend(batch_results);
207    }
208
209    Ok(all_results)
210}
211
212// Helper to format signal URL
213// Extracted for testing
214#[inline]
215fn format_signal_url(base_url: &str, signals: &[wellen::SignalRef]) -> String {
216    let mut url = base_url.to_string();
217    for signal in signals {
218        write!(url, "/{}", signal.index()).unwrap();
219    }
220    url
221}
222
223async fn get_signals_batch(
224    base_url: &str,
225    signals: &[wellen::SignalRef],
226) -> Result<Vec<wellen::Signal>> {
227    let client = get_client();
228    let url = format_signal_url(base_url, signals);
229
230    let response = client.get(url).send().await?;
231    check_response(base_url, &response)?;
232    let data = response.bytes().await?;
233    let mut reader = std::io::Cursor::new(data);
234    let num_ids: u64 = leb128::read::unsigned(&mut reader)?;
235    if num_ids > signals.len() as u64 {
236        bail!(
237            "Too many signals in response: {num_ids}, expected {}",
238            signals.len()
239        );
240    }
241    if num_ids == 0 {
242        return Ok(vec![]);
243    }
244
245    let opts = BINCODE_OPTIONS.allow_trailing_bytes();
246    let mut out = Vec::with_capacity(num_ids as usize);
247    for _ in 0..(num_ids - 1) {
248        let compressed: wellen::CompressedSignal = opts.deserialize_from(&mut reader)?;
249        let signal = compressed.uncompress();
250        out.push(signal);
251    }
252    // for the final signal, we expect to consume all bytes
253    let compressed: wellen::CompressedSignal = BINCODE_OPTIONS.deserialize_from(&mut reader)?;
254    let signal = compressed.uncompress();
255    out.push(signal);
256    Ok(out)
257}
258
259pub fn get_hierarchy_from_server(
260    sender: Sender<Message>,
261    server: String,
262    load_options: LoadOptions,
263    file_index: usize,
264) {
265    let start = web_time::Instant::now();
266    let source = WaveSource::Url(server.clone());
267
268    perform_async_work(async move {
269        let res = get_hierarchy(server.clone(), file_index)
270            .await
271            .map_err(|e| anyhow!("{e:?}"))
272            .with_context(|| format!("Failed to retrieve hierarchy from remote server {server}"));
273
274        let h = match res {
275            Ok(h) => h,
276            Err(e) => {
277                error!("{e:?}");
278                return;
279            }
280        };
281        let header = HeaderResult::Remote(Arc::new(h.hierarchy), h.file_format, server, file_index);
282        checked_send(
283            &sender,
284            Message::WaveHeaderLoaded(start, source, load_options, header),
285        );
286    });
287}
288
289pub fn get_time_table_from_server(sender: Sender<Message>, server: String, file_index: usize) {
290    let start = web_time::Instant::now();
291    let source = WaveSource::Url(server.clone());
292
293    perform_async_work(async move {
294        let res = get_time_table(server.clone(), file_index)
295            .await
296            .map_err(|e| anyhow!("{e:?}"))
297            .with_context(|| format!("Failed to retrieve time table from remote server {server}"));
298
299        let table = match res {
300            Ok(table) => table,
301            Err(e) => {
302                error!("{e:?}");
303                return;
304            }
305        };
306        checked_send(
307            &sender,
308            Message::WaveBodyLoaded(start, source, BodyResult::Remote(table, server)),
309        );
310    });
311}
312
313pub fn get_server_status(sender: Sender<Message>, server: String, delay_ms: u64) {
314    let start = web_time::Instant::now();
315    perform_async_work(async move {
316        sleep_ms(delay_ms).await;
317        let res = get_status(server.clone())
318            .await
319            .map_err(|e| anyhow!("{e:?}"))
320            .with_context(|| format!("Failed to retrieve status from remote server {server}"));
321
322        let status = match res {
323            Ok(status) => status,
324            Err(e) => {
325                error!("{e:?}");
326                return;
327            }
328        };
329        checked_send(&sender, Message::SetSurverStatus(start, server, status));
330    });
331}
332
333pub fn server_reload(
334    sender: Sender<Message>,
335    server: String,
336    load_options: LoadOptions,
337    file_index: usize,
338) {
339    let start = web_time::Instant::now();
340    perform_async_work(async move {
341        let res = reload(server.clone(), file_index).await;
342        let request_hierarchy = match res {
343            Ok(status) => {
344                checked_send(
345                    &sender,
346                    Message::SetSurverStatus(start, server.clone(), status),
347                );
348                true
349            }
350            Err(crate::remote::ReloadError::FileUnchanged) => {
351                checked_send(&sender, Message::StopProgressTracker);
352                false
353            }
354            Err(e) => {
355                error!("{e:?}");
356                false
357            }
358        };
359        if request_hierarchy {
360            get_hierarchy_from_server(sender, server, load_options, file_index);
361        }
362    });
363}
364
365mod tests {
366    #[test]
367    fn test_signal_url_length_calculation() {
368        use crate::remote::client::signal_url_len;
369        // Test edge cases for digit calculation
370        assert_eq!(signal_url_len(0), 2); // "/0" -> 2 chars
371        assert_eq!(signal_url_len(1), 2); // "/1" -> 2 chars
372        assert_eq!(signal_url_len(9), 2); // "/9" -> 2 chars
373        assert_eq!(signal_url_len(10), 3); // "/10" -> 3 chars
374        assert_eq!(signal_url_len(99), 3); // "/99" -> 3 chars
375        assert_eq!(signal_url_len(100), 4); // "/100" -> 4 chars
376        assert_eq!(signal_url_len(999), 4); // "/999" -> 4 chars
377        assert_eq!(signal_url_len(1000), 5); // "/1000" -> 5 chars
378        assert_eq!(signal_url_len(65535), 6); // "/65535" -> 6 chars
379    }
380
381    #[test]
382    fn test_empty_signals_returns_empty() {
383        use crate::remote::get_signals;
384        // Create a mock async runtime for testing
385        let rt = tokio::runtime::Runtime::new().unwrap();
386        rt.block_on(async {
387            let signals: Vec<wellen::SignalRef> = vec![];
388            let result = get_signals("http://localhost:8080".to_string(), &signals, 1000, 0).await;
389
390            // Should return Ok with empty vec without making any network calls
391            assert!(result.is_ok());
392            assert_eq!(result.unwrap().len(), 0);
393        });
394    }
395
396    #[test]
397    fn test_boundary_signal_indices() {
398        use crate::remote::client::signal_url_len;
399        // Test that we handle boundary cases correctly
400        let boundary_indices = vec![0, 1, 9, 10, 99, 100, 999, 1000, 9999, 10000];
401
402        for idx in boundary_indices {
403            let sig_ref = wellen::SignalRef::from_index(idx);
404            let len = signal_url_len(sig_ref.unwrap().index());
405
406            // Verify the calculated length matches actual string length
407            let actual = format!("/{idx}");
408            assert_eq!(
409                len,
410                actual.len(),
411                "URL length calculation mismatch for index {}: expected {}, got {}",
412                idx,
413                actual.len(),
414                len
415            );
416        }
417    }
418
419    #[test]
420    fn test_url_construction_format() {
421        use crate::remote::client::format_signal_url;
422        // Verify URL format matches expected pattern
423        let base_url = "http://localhost:8080/get_signals";
424        let signals: Vec<wellen::SignalRef> = vec![
425            wellen::SignalRef::from_index(1),
426            wellen::SignalRef::from_index(42),
427            wellen::SignalRef::from_index(999),
428        ]
429        .into_iter()
430        .flatten()
431        .collect();
432
433        let url = format_signal_url(base_url, &signals);
434
435        assert_eq!(url, "http://localhost:8080/get_signals/1/42/999");
436    }
437}
libsurfer/remote/client.rs

libsurfer/remote/
client.rs
