egui_charts/data/http/
mod.rs

1//! HTTP symbol metadata, column history and polling subscriptions (feature `http`).
2//!
3//! See the builder guide's HTTP endpoint contract. The application supplies the server and
4//! its data. Requests run off the UI thread on native targets and through browser fetch on
5//! wasm32; HTTPS, redirects and response framing are handled by the HTTP client.
6
7mod protocol;
8#[cfg(test)]
9mod tests;
10
11use super::{
12    Datafeed, FeedCapabilities, FeedError, FeedStatus, HistoryPage, HistoryRequest, Reply,
13    Subscription, SubscriptionId,
14};
15use crate::model::{Instrument, SymbolType};
16use protocol::{Columns, Config, Symbol, history_url, url};
17use serde::de::DeserializeOwned;
18use std::collections::HashMap;
19use std::sync::atomic::{AtomicBool, Ordering};
20use std::sync::{Arc, Mutex};
21use std::time::Duration;
22
23const REFRESH: Duration = Duration::from_secs(5);
24
25/// A reusable feed for a server exposing `/config`, `/symbols`, `/search` and `/history`.
26///
27/// Implements both [`Datafeed`] and [`SymbolSearch`](crate::services::SymbolSearch), so the same `Arc` can supply chart data
28/// and the host's symbol search dialog. Subscriptions refresh bars every five seconds,
29/// retry recoverable failures with backoff and stop on unsubscribe or chart drop. An HTTP
30/// request already in flight may finish, but its answer is discarded after cancellation.
31///
32/// ```
33/// # #[cfg(feature = "widget")] {
34/// use egui_charts::{ChartBuilder, data::http::HttpDatafeed};
35/// use std::sync::Arc;
36/// let feed = Arc::new(HttpDatafeed::new("https://data.example.com"));
37/// let chart = ChartBuilder::new().with_symbol("SAMPLE").with_feed(feed).build();
38/// # }
39/// ```
40pub struct HttpDatafeed {
41    base: String,
42    subscriptions: Mutex<HashMap<SubscriptionId, Arc<AtomicBool>>>,
43}
44
45impl HttpDatafeed {
46    /// Connects to the endpoint root at `address`, which may include a path prefix.
47    /// A trailing slash is optional. Browser servers must allow the application's origin.
48    pub fn new(address: impl Into<String>) -> Self {
49        Self {
50            base: address.into().trim_end_matches('/').to_owned(),
51            subscriptions: Mutex::new(HashMap::new()),
52        }
53    }
54}
55
56impl Datafeed for HttpDatafeed {
57    fn capabilities(&self, reply: Reply<FeedCapabilities>) {
58        get(
59            url(&self.base, "config", &[]),
60            String::new(),
61            move |result| {
62                reply.send(result.and_then(Config::into_capabilities));
63            },
64        );
65    }
66
67    fn resolve(&self, symbol: &str, reply: Reply<Instrument>) {
68        get(
69            url(&self.base, "symbols", &[("symbol", symbol.to_owned())]),
70            symbol.to_owned(),
71            move |result| reply.send(result.and_then(Symbol::into_instrument)),
72        );
73    }
74
75    fn resolve_converted(&self, symbol: &str, conversion: &str, reply: Reply<Instrument>) {
76        get(
77            url(
78                &self.base,
79                "symbols",
80                &[
81                    ("symbol", symbol.to_owned()),
82                    ("conversion", conversion.to_owned()),
83                ],
84            ),
85            symbol.to_owned(),
86            move |result| reply.send(result.and_then(Symbol::into_instrument)),
87        );
88    }
89
90    fn history(&self, request: HistoryRequest, reply: Reply<HistoryPage>) {
91        let target = match history_url(&self.base, &request) {
92            Ok(target) => target,
93            Err(error) => return reply.send(Err(error)),
94        };
95        get(target, request.symbol.clone(), move |result| {
96            reply.send(result.and_then(|columns: Columns| columns.into_page(&request)));
97        });
98    }
99
100    fn subscribe(&self, subscription: Subscription) {
101        let cancelled = Arc::new(AtomicBool::new(false));
102        if let Some(old) = self
103            .subscriptions
104            .lock()
105            .unwrap_or_else(std::sync::PoisonError::into_inner)
106            .insert(subscription.id, cancelled.clone())
107        {
108            old.store(true, Ordering::Release);
109        }
110        Poll {
111            base: self.base.clone(),
112            subscription,
113            cancelled,
114            since: None,
115            retry: REFRESH,
116        }
117        .fetch();
118    }
119
120    fn unsubscribe(&self, id: SubscriptionId) {
121        if let Some(cancelled) = self
122            .subscriptions
123            .lock()
124            .unwrap_or_else(std::sync::PoisonError::into_inner)
125            .remove(&id)
126        {
127            cancelled.store(true, Ordering::Release);
128        }
129    }
130}
131
132impl Drop for HttpDatafeed {
133    fn drop(&mut self) {
134        for cancelled in self
135            .subscriptions
136            .get_mut()
137            .unwrap_or_else(std::sync::PoisonError::into_inner)
138            .values()
139        {
140            cancelled.store(true, Ordering::Release);
141        }
142    }
143}
144
145impl HttpDatafeed {
146    pub(crate) fn search_symbols(
147        &self,
148        text: String,
149        symbol_type: Option<SymbolType>,
150        exchange: Option<String>,
151        offset: usize,
152        limit: usize,
153        reply: Reply<Vec<Instrument>>,
154    ) {
155        let target = url(
156            &self.base,
157            "search",
158            &[
159                ("query", text),
160                (
161                    "type",
162                    symbol_type.map(|kind| kind.to_string()).unwrap_or_default(),
163                ),
164                ("exchange", exchange.unwrap_or_default()),
165                ("offset", offset.to_string()),
166                ("limit", limit.to_string()),
167            ],
168        );
169        get(target, String::new(), move |result| {
170            reply.send(result.and_then(|symbols: Vec<Symbol>| {
171                symbols
172                    .into_iter()
173                    .take(limit)
174                    .map(Symbol::into_instrument)
175                    .collect()
176            }));
177        });
178    }
179}
180
181fn get<T: DeserializeOwned>(
182    target: String,
183    symbol: String,
184    answer: impl FnOnce(Result<T, FeedError>) + Send + 'static,
185) {
186    ehttp::fetch(ehttp::Request::get(target), move |result| {
187        answer(
188            result
189                .map_err(FeedError::Network)
190                .and_then(|response| decode(&symbol, response)),
191        );
192    });
193}
194
195fn decode<T: DeserializeOwned>(symbol: &str, response: ehttp::Response) -> Result<T, FeedError> {
196    match response.status {
197        200..=299 => serde_json::from_slice(&response.bytes)
198            .map_err(|error| FeedError::InvalidData(error.to_string())),
199        404 => Err(FeedError::SymbolNotFound(symbol.to_owned())),
200        429 => Err(FeedError::RateLimited {
201            retry_after: response.headers.get("retry-after").and_then(|value| {
202                value
203                    .trim()
204                    .parse()
205                    .ok()
206                    .map(Duration::from_secs)
207                    .or_else(|| {
208                        chrono::DateTime::parse_from_rfc2822(value)
209                            .ok()
210                            .map(|time| {
211                                (time.with_timezone(&chrono::Utc) - chrono::Utc::now())
212                                    .to_std()
213                                    .unwrap_or(Duration::ZERO)
214                            })
215                    })
216            }),
217        }),
218        408 | 500..=599 => Err(FeedError::Network(format!("HTTP {}", response.status))),
219        _ => Err(FeedError::InvalidData(format!("HTTP {}", response.status))),
220    }
221}
222
223struct Poll {
224    base: String,
225    subscription: Subscription,
226    cancelled: Arc<AtomicBool>,
227    since: Option<chrono::DateTime<chrono::Utc>>,
228    retry: Duration,
229}
230
231impl Poll {
232    fn active(&self) -> bool {
233        !self.cancelled.load(Ordering::Acquire) && self.subscription.sink.is_active()
234    }
235
236    fn fetch(self) {
237        if !self.active() {
238            return;
239        }
240        let request = HistoryRequest {
241            symbol: self.subscription.symbol.clone(),
242            timeframe: self.subscription.timeframe,
243            clock: self.subscription.clock.clone(),
244            before: None,
245            since: self.since,
246            count: 500,
247            extended_hours: self.subscription.extended_hours,
248            adjusted: false,
249            conversion: self.subscription.conversion.clone(),
250        };
251        let target = match history_url(&self.base, &request) {
252            Ok(target) => target,
253            Err(error) => return self.subscription.sink.report_error(error),
254        };
255        self.subscription.sink.set_status(FeedStatus::Connecting);
256        get(target, request.symbol.clone(), move |result| {
257            let result = result.and_then(|columns: Columns| columns.into_page(&request));
258            if let Some((poll, delay)) = self.received(result) {
259                later(
260                    delay,
261                    poll.cancelled.clone(),
262                    Box::new(move || poll.fetch()),
263                );
264            }
265        });
266    }
267
268    fn received(mut self, result: Result<HistoryPage, FeedError>) -> Option<(Self, Duration)> {
269        if !self.active() {
270            return None;
271        }
272        let sink = &self.subscription.sink;
273        let result = result.and_then(|page| {
274            if page.bars.len() == 500 && self.since.is_some_and(|since| page.bars[0].time > since) {
275                Err(FeedError::InvalidData(
276                    "live history window exceeded; reload history".into(),
277                ))
278            } else {
279                Ok(page)
280            }
281        });
282        let delay = match result {
283            Ok(page) => {
284                for bar in page.bars {
285                    self.since = Some(bar.time);
286                    sink.push_bar(bar);
287                }
288                sink.set_status(FeedStatus::Live);
289                self.retry = REFRESH;
290                REFRESH
291            }
292            Err(error) => {
293                let recoverable = error.is_recoverable();
294                let delay = match &error {
295                    FeedError::RateLimited {
296                        retry_after: Some(delay),
297                    } => (*delay).max(self.retry),
298                    _ => self.retry,
299                };
300                sink.set_status(FeedStatus::Disconnected);
301                sink.report_error(error);
302                if !recoverable {
303                    return None;
304                }
305                self.retry = (self.retry * 2).min(Duration::from_secs(60));
306                delay
307            }
308        };
309        Some((self, delay))
310    }
311}
312
313#[cfg(not(target_arch = "wasm32"))]
314fn later(delay: Duration, cancelled: Arc<AtomicBool>, action: Box<dyn FnOnce() + Send>) {
315    std::thread::spawn(move || {
316        let start = std::time::Instant::now();
317        while !cancelled.load(Ordering::Acquire) {
318            let remaining = delay.saturating_sub(start.elapsed());
319            if remaining.is_zero() {
320                return action();
321            }
322            std::thread::sleep(remaining.min(Duration::from_secs(1)));
323        }
324    });
325}
326
327#[cfg(target_arch = "wasm32")]
328fn later(delay: Duration, cancelled: Arc<AtomicBool>, action: Box<dyn FnOnce() + Send>) {
329    use wasm_bindgen::{JsCast as _, closure::Closure};
330    if let Some(window) = web_sys::window() {
331        // Release cancelled retries promptly, even after a long Retry-After.
332        let wait = delay.min(Duration::from_secs(1));
333        let millis =
334            wait.as_millis() as i32 + i32::from(!wait.subsec_nanos().is_multiple_of(1_000_000));
335        let callback = Closure::once_into_js(move || {
336            if !cancelled.load(Ordering::Acquire) {
337                if delay > wait {
338                    later(delay - wait, cancelled, action);
339                } else {
340                    action();
341                }
342            }
343        });
344        let _ = window.set_timeout_with_callback_and_timeout_and_arguments_0(
345            callback.unchecked_ref(),
346            millis,
347        );
348    }
349}
egui_charts/data/http/mod.rs

egui_charts/data/http/
mod.rs
