radion_sdk/realtime/
client.rs

1//! The realtime (WebSocket) client and its background connection manager.
2
3use std::pin::Pin;
4use std::sync::atomic::{AtomicBool, Ordering};
5use std::sync::{Arc, Mutex};
6use std::task::{Context, Poll};
7use std::time::Duration;
8
9use futures_util::{Sink, SinkExt, Stream, StreamExt};
10use tokio::sync::{broadcast, mpsc, oneshot};
11use tokio::task::JoinHandle;
12use tokio_stream::wrappers::BroadcastStream;
13use tokio_stream::wrappers::errors::BroadcastStreamRecvError;
14use tokio_tungstenite::tungstenite::client::IntoClientRequest;
15use tokio_tungstenite::tungstenite::http::header::{HeaderName, HeaderValue};
16use tokio_tungstenite::tungstenite::{Error as WsError, Message};
17
18use super::auth::{TokenProvider, build_auth_query_url};
19#[cfg(feature = "compression")]
20use super::compression::{inflate, with_compress_query};
21use super::protocol::{
22    ChannelEvent, InboundFrame, OutboundFrame, Subscription, parse_inbound_frame,
23};
24use super::reconnect::{ReconnectOptions, ReconnectPolicy};
25use super::subscription::SubscriptionManager;
26use crate::config::DEFAULT_WS_URL;
27use crate::error::{RadionError, Result};
28
29/// Buffered events retained per [`broadcast`] receiver before lagging.
30const EVENT_BUFFER: usize = 1024;
31/// Buffered lifecycle events retained per receiver before lagging.
32const LIFECYCLE_BUFFER: usize = 64;
33
34/// Options controlling heartbeat / stale-connection detection.
35#[derive(Debug, Clone, Copy)]
36pub struct HeartbeatOptions {
37    /// Interval between client pings.
38    pub interval: Duration,
39    /// How long to wait for any inbound traffic after a ping before declaring
40    /// the connection stale.
41    pub timeout: Duration,
42}
43
44impl Default for HeartbeatOptions {
45    fn default() -> Self {
46        Self {
47            interval: Duration::from_secs(15),
48            timeout: Duration::from_secs(10),
49        }
50    }
51}
52
53/// Configuration for a [`RealtimeClient`].
54#[derive(Debug, Clone)]
55pub struct RealtimeOptions {
56    /// Radion API key, sent as the `X-API-Key` header.
57    pub api_key: String,
58    /// WebSocket endpoint. Defaults to [`DEFAULT_WS_URL`].
59    pub url: String,
60    /// Reconnect policy, or `None` to disable auto-reconnect.
61    pub reconnect: Option<ReconnectOptions>,
62    /// Heartbeat policy, or `None` to disable heartbeats.
63    pub heartbeat: Option<HeartbeatOptions>,
64    /// User JWT provider for the public-key (`pk_jwt_`) flow. `None` = secret
65    /// key. Resolved on every (re)connect.
66    pub token_provider: Option<TokenProvider>,
67    /// Send credentials in the URL query instead of headers. Defaults to
68    /// `false`; enable for header-stripping proxies or gateways.
69    pub auth_in_query: bool,
70    /// Ask the server for zlib-compressed binary frames. Defaults to `false`.
71    #[cfg(feature = "compression")]
72    #[cfg_attr(docsrs, doc(cfg(feature = "compression")))]
73    pub compression: bool,
74}
75
76impl RealtimeOptions {
77    /// Options for the given API key with default URL, reconnect, and heartbeat.
78    pub fn new(api_key: impl Into<String>) -> Self {
79        Self {
80            api_key: api_key.into(),
81            url: DEFAULT_WS_URL.to_string(),
82            reconnect: Some(ReconnectOptions::default()),
83            heartbeat: Some(HeartbeatOptions::default()),
84            token_provider: None,
85            auth_in_query: false,
86            #[cfg(feature = "compression")]
87            compression: false,
88        }
89    }
90
91    /// Override the WebSocket endpoint.
92    #[must_use]
93    pub fn url(mut self, url: impl Into<String>) -> Self {
94        self.url = url.into();
95        self
96    }
97
98    /// Tune the reconnect policy.
99    #[must_use]
100    pub fn reconnect(mut self, options: ReconnectOptions) -> Self {
101        self.reconnect = Some(options);
102        self
103    }
104
105    /// Disable auto-reconnect.
106    #[must_use]
107    pub fn disable_reconnect(mut self) -> Self {
108        self.reconnect = None;
109        self
110    }
111
112    /// Tune the heartbeat policy.
113    #[must_use]
114    pub fn heartbeat(mut self, options: HeartbeatOptions) -> Self {
115        self.heartbeat = Some(options);
116        self
117    }
118
119    /// Disable heartbeats.
120    #[must_use]
121    pub fn disable_heartbeat(mut self) -> Self {
122        self.heartbeat = None;
123        self
124    }
125
126    /// Set a static user JWT for the public-key (`pk_jwt_`) flow.
127    #[must_use]
128    pub fn token(mut self, token: impl Into<String>) -> Self {
129        self.token_provider = Some(TokenProvider::from_static(token));
130        self
131    }
132
133    /// Set a user JWT provider, called on every (re)connect for a fresh token.
134    #[must_use]
135    pub fn token_provider(mut self, provider: TokenProvider) -> Self {
136        self.token_provider = Some(provider);
137        self
138    }
139
140    /// Send credentials in the URL query string instead of headers.
141    #[must_use]
142    pub fn auth_in_query(mut self, enabled: bool) -> Self {
143        self.auth_in_query = enabled;
144        self
145    }
146
147    /// Ask the server for zlib-compressed binary frames.
148    ///
149    /// Adds `compress=zlib` to the connect URL. The server then sends event
150    /// frames as binary zlib, which the client inflates before parsing. Text
151    /// frames still work, so mixed traffic is fine.
152    #[cfg(feature = "compression")]
153    #[cfg_attr(docsrs, doc(cfg(feature = "compression")))]
154    #[must_use]
155    pub fn compression(mut self, enabled: bool) -> Self {
156        self.compression = enabled;
157        self
158    }
159
160    /// Decode a binary frame into the JSON text it carries: inflate it when
161    /// compression is on, otherwise read it as plain UTF-8.
162    fn decode_binary(&self, bytes: &[u8]) -> Result<String> {
163        #[cfg(feature = "compression")]
164        {
165            if self.compression {
166                return inflate(bytes);
167            }
168        }
169        std::str::from_utf8(bytes)
170            .map(str::to_owned)
171            .map_err(RadionError::transport)
172    }
173}
174
175/// A connection lifecycle event, delivered on [`RealtimeClient::lifecycle`].
176#[derive(Debug, Clone)]
177#[non_exhaustive]
178pub enum LifecycleEvent {
179    /// The connection opened (initial connect or successful reconnect).
180    Open,
181    /// The connection closed.
182    Close {
183        /// WebSocket close code.
184        code: u16,
185        /// Close reason, if any.
186        reason: String,
187    },
188    /// A reconnect attempt was scheduled.
189    Reconnect {
190        /// Number of retries since the last successful connection.
191        attempt: u32,
192        /// Delay before the attempt.
193        delay: Duration,
194    },
195    /// A non-fatal server `warning` frame — for example `mempool_unavailable`,
196    /// sent after a pending (`confirmed=false`) subscribe when the node has no
197    /// pending stream. Delivery continues; this is not an error.
198    Warning {
199        /// Machine-readable warning code (e.g. `mempool_unavailable`).
200        code: String,
201        /// Subscription id the warning refers to, if any.
202        id: Option<String>,
203        /// Human-readable message.
204        message: String,
205    },
206    /// An error occurred: a server `error` frame, a transport failure, or a
207    /// stale connection.
208    Error(RadionError),
209}
210
211/// Commands sent from a [`RealtimeClient`] handle to its manager task.
212enum Command {
213    Subscribe(Subscription),
214    Unsubscribe(String),
215    Close { code: u16, reason: String },
216}
217
218/// Async WebSocket client for the Radion realtime API.
219///
220/// Owns the connection lifecycle: it transparently reconnects with exponential
221/// backoff after unexpected drops, restores subscriptions on reconnect, and
222/// fans inbound channel frames out to [`events`](Self::events) and
223/// per-subscription streams.
224///
225/// Usually reached as [`Radion::realtime`](crate::Radion::realtime), but can be
226/// constructed standalone with [`RealtimeClient::new`].
227#[derive(Debug)]
228pub struct RealtimeClient {
229    options: RealtimeOptions,
230    cmd_tx: mpsc::UnboundedSender<Command>,
231    cmd_rx: Mutex<Option<mpsc::UnboundedReceiver<Command>>>,
232    events_tx: broadcast::Sender<ChannelEvent>,
233    lifecycle_tx: broadcast::Sender<LifecycleEvent>,
234    connected: Arc<AtomicBool>,
235    task: Mutex<Option<JoinHandle<()>>>,
236}
237
238impl RealtimeClient {
239    /// Construct a standalone realtime client.
240    pub fn new(options: RealtimeOptions) -> Self {
241        let (cmd_tx, cmd_rx) = mpsc::unbounded_channel();
242        let (events_tx, _) = broadcast::channel(EVENT_BUFFER);
243        let (lifecycle_tx, _) = broadcast::channel(LIFECYCLE_BUFFER);
244        Self {
245            options,
246            cmd_tx,
247            cmd_rx: Mutex::new(Some(cmd_rx)),
248            events_tx,
249            lifecycle_tx,
250            connected: Arc::new(AtomicBool::new(false)),
251            task: Mutex::new(None),
252        }
253    }
254
255    /// Whether the underlying socket is currently open.
256    pub fn connected(&self) -> bool {
257        self.connected.load(Ordering::SeqCst)
258    }
259
260    /// Open the connection.
261    ///
262    /// Resolves once the socket is established. Calling it again after a
263    /// successful connect is a no-op.
264    ///
265    /// # Errors
266    ///
267    /// Returns an error if the first connection attempt fails.
268    pub async fn connect(&self) -> Result<()> {
269        if self.connected() {
270            return Ok(());
271        }
272        let Some(cmd_rx) = self.cmd_rx.lock().expect("cmd_rx mutex poisoned").take() else {
273            // Manager already started by a previous connect() call.
274            return Ok(());
275        };
276
277        let (ready_tx, ready_rx) = oneshot::channel();
278        let task = tokio::spawn(run(
279            self.options.clone(),
280            cmd_rx,
281            self.events_tx.clone(),
282            self.lifecycle_tx.clone(),
283            Arc::clone(&self.connected),
284            ready_tx,
285        ));
286        *self.task.lock().expect("task mutex poisoned") = Some(task);
287
288        match ready_rx.await {
289            Ok(result) => result,
290            Err(_) => Err(RadionError::connection(
291                "connection task ended before connecting",
292            )),
293        }
294    }
295
296    /// Subscribe to a channel, returning a stream of its events.
297    ///
298    /// The subscription is resent automatically after a reconnect. The returned
299    /// stream yields only events for this subscription's `id`; use
300    /// [`events`](Self::events) for the firehose across all subscriptions.
301    ///
302    /// # Errors
303    ///
304    /// Returns [`RadionError::Connection`] if the subscription is missing a
305    /// filter its channel requires, or if the client has been closed.
306    pub async fn subscribe(&self, subscription: Subscription) -> Result<ChannelEventStream> {
307        subscription.validate()?;
308        let id = subscription.id.clone();
309        // Subscribe to the broadcast before sending the command so no event
310        // delivered between the two is missed.
311        let rx = self.events_tx.subscribe();
312        self.cmd_tx
313            .send(Command::Subscribe(subscription))
314            .map_err(|_| RadionError::connection("client has been closed"))?;
315        Ok(ChannelEventStream {
316            inner: BroadcastStream::new(rx),
317            filter_id: Some(id),
318        })
319    }
320
321    /// Unsubscribe by subscription id.
322    ///
323    /// # Errors
324    ///
325    /// Returns [`RadionError::Connection`] if the client has been closed.
326    pub async fn unsubscribe(&self, id: impl Into<String>) -> Result<()> {
327        self.cmd_tx
328            .send(Command::Unsubscribe(id.into()))
329            .map_err(|_| RadionError::connection("client has been closed"))
330    }
331
332    /// Stream of every channel event across all subscriptions (the firehose).
333    pub fn events(&self) -> ChannelEventStream {
334        ChannelEventStream {
335            inner: BroadcastStream::new(self.events_tx.subscribe()),
336            filter_id: None,
337        }
338    }
339
340    /// Stream of connection lifecycle events.
341    pub fn lifecycle(&self) -> LifecycleStream {
342        LifecycleStream {
343            inner: BroadcastStream::new(self.lifecycle_tx.subscribe()),
344        }
345    }
346
347    /// Gracefully close the connection and stop reconnect attempts.
348    ///
349    /// Waits for the manager task to finish. Subsequent calls are no-ops.
350    pub async fn close(&self, code: u16, reason: impl Into<String>) {
351        let _ = self.cmd_tx.send(Command::Close {
352            code,
353            reason: reason.into(),
354        });
355        let handle = self.task.lock().expect("task mutex poisoned").take();
356        if let Some(handle) = handle {
357            let _ = handle.await;
358        }
359    }
360}
361
362/// A stream of [`ChannelEvent`]s. Lagged events (slow consumer) are skipped.
363pub struct ChannelEventStream {
364    inner: BroadcastStream<ChannelEvent>,
365    filter_id: Option<String>,
366}
367
368impl Stream for ChannelEventStream {
369    type Item = ChannelEvent;
370
371    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
372        loop {
373            match self.inner.poll_next_unpin(cx) {
374                Poll::Ready(Some(Ok(event))) => {
375                    if self.filter_id.as_ref().is_none_or(|id| *id == event.id) {
376                        return Poll::Ready(Some(event));
377                    }
378                }
379                Poll::Ready(Some(Err(BroadcastStreamRecvError::Lagged(_)))) => {}
380                Poll::Ready(None) => return Poll::Ready(None),
381                Poll::Pending => return Poll::Pending,
382            }
383        }
384    }
385}
386
387/// A stream of [`LifecycleEvent`]s. Lagged events are skipped.
388pub struct LifecycleStream {
389    inner: BroadcastStream<LifecycleEvent>,
390}
391
392impl Stream for LifecycleStream {
393    type Item = LifecycleEvent;
394
395    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
396        loop {
397            match self.inner.poll_next_unpin(cx) {
398                Poll::Ready(Some(Ok(event))) => return Poll::Ready(Some(event)),
399                Poll::Ready(Some(Err(BroadcastStreamRecvError::Lagged(_)))) => {}
400                Poll::Ready(None) => return Poll::Ready(None),
401                Poll::Pending => return Poll::Pending,
402            }
403        }
404    }
405}
406
407/// Outcome of a single connected session.
408enum SessionOutcome {
409    /// The consumer requested a graceful shutdown.
410    Shutdown { code: u16, reason: String },
411    /// The connection dropped unexpectedly.
412    Disconnected { code: u16, reason: String },
413}
414
415/// The background manager task: connect, run a session, reconnect on drop.
416async fn run(
417    options: RealtimeOptions,
418    mut cmd_rx: mpsc::UnboundedReceiver<Command>,
419    events_tx: broadcast::Sender<ChannelEvent>,
420    lifecycle_tx: broadcast::Sender<LifecycleEvent>,
421    connected: Arc<AtomicBool>,
422    ready_tx: oneshot::Sender<Result<()>>,
423) {
424    let mut ready = Some(ready_tx);
425    let mut policy = ReconnectPolicy::new(options.reconnect.unwrap_or_default());
426    let mut subscriptions = SubscriptionManager::default();
427
428    loop {
429        match connect_ws(&options).await {
430            Ok(ws) => {
431                connected.store(true, Ordering::SeqCst);
432                policy.reset();
433                if let Some(tx) = ready.take() {
434                    let _ = tx.send(Ok(()));
435                }
436                let _ = lifecycle_tx.send(LifecycleEvent::Open);
437                #[cfg(feature = "tracing")]
438                tracing::debug!(url = %options.url, "radion realtime connected");
439
440                let outcome = session(
441                    ws,
442                    &options,
443                    &mut cmd_rx,
444                    &events_tx,
445                    &lifecycle_tx,
446                    &mut subscriptions,
447                )
448                .await;
449                connected.store(false, Ordering::SeqCst);
450
451                match outcome {
452                    SessionOutcome::Shutdown { code, reason } => {
453                        let _ = lifecycle_tx.send(LifecycleEvent::Close { code, reason });
454                        return;
455                    }
456                    SessionOutcome::Disconnected { code, reason } => {
457                        let _ = lifecycle_tx.send(LifecycleEvent::Close { code, reason });
458                        if options.reconnect.is_none() {
459                            return;
460                        }
461                    }
462                }
463            }
464            Err(error) => {
465                if let Some(tx) = ready.take() {
466                    // First attempt failed: surface to connect() and stop.
467                    let _ = tx.send(Err(error));
468                    return;
469                }
470                let _ = lifecycle_tx.send(LifecycleEvent::Error(error));
471                if options.reconnect.is_none() {
472                    return;
473                }
474            }
475        }
476
477        // Back off before the next attempt; a Close command stops reconnecting.
478        let delay = policy.next_delay();
479        let _ = lifecycle_tx.send(LifecycleEvent::Reconnect {
480            attempt: policy.attempts(),
481            delay,
482        });
483        #[cfg(feature = "tracing")]
484        tracing::debug!(
485            ?delay,
486            attempt = policy.attempts(),
487            "radion realtime reconnecting"
488        );
489
490        tokio::select! {
491            () = tokio::time::sleep(delay) => {}
492            cmd = cmd_rx.recv() => match cmd {
493                Some(Command::Subscribe(subscription)) => {
494                    subscriptions.add(subscription);
495                }
496                Some(Command::Unsubscribe(id)) => {
497                    subscriptions.remove(&id);
498                }
499                Some(Command::Close { .. }) | None => return,
500            },
501        }
502    }
503}
504
505/// Open a WebSocket connection, presenting the API key (and user JWT, if a token
506/// provider is set) either as headers or in the URL query string.
507async fn connect_ws(
508    options: &RealtimeOptions,
509) -> Result<
510    impl Stream<Item = std::result::Result<Message, WsError>> + Sink<Message, Error = WsError> + Unpin,
511> {
512    let token = match &options.token_provider {
513        Some(provider) => Some(provider.fetch().await?),
514        None => None,
515    };
516
517    let base_url = connect_url(options);
518
519    let (ws, _response) = if options.auth_in_query {
520        let url = build_auth_query_url(&base_url, &options.api_key, token.as_deref());
521        let request = url.into_client_request().map_err(RadionError::transport)?;
522        tokio_tungstenite::connect_async(request)
523            .await
524            .map_err(RadionError::transport)?
525    } else {
526        let mut request = base_url
527            .as_str()
528            .into_client_request()
529            .map_err(RadionError::transport)?;
530        let api_key = HeaderValue::from_str(&options.api_key).map_err(RadionError::transport)?;
531        request
532            .headers_mut()
533            .insert(HeaderName::from_static("x-api-key"), api_key);
534        if let Some(token) = &token {
535            let bearer = HeaderValue::from_str(&format!("Bearer {token}"))
536                .map_err(RadionError::transport)?;
537            request
538                .headers_mut()
539                .insert(HeaderName::from_static("authorization"), bearer);
540        }
541        tokio_tungstenite::connect_async(request)
542            .await
543            .map_err(RadionError::transport)?
544    };
545    Ok(ws)
546}
547
548/// The URL to connect to, carrying `compress=zlib` when compression is on.
549fn connect_url(options: &RealtimeOptions) -> String {
550    #[cfg(feature = "compression")]
551    {
552        if options.compression {
553            return with_compress_query(&options.url);
554        }
555    }
556    options.url.clone()
557}
558
559/// Run one connected session until it shuts down or drops.
560async fn session<S>(
561    mut ws: S,
562    options: &RealtimeOptions,
563    cmd_rx: &mut mpsc::UnboundedReceiver<Command>,
564    events_tx: &broadcast::Sender<ChannelEvent>,
565    lifecycle_tx: &broadcast::Sender<LifecycleEvent>,
566    subscriptions: &mut SubscriptionManager,
567) -> SessionOutcome
568where
569    S: Stream<Item = std::result::Result<Message, WsError>>
570        + Sink<Message, Error = WsError>
571        + Unpin,
572{
573    // Restore desired subscriptions after a (re)connect.
574    let replay: Vec<_> = subscriptions
575        .desired()
576        .map(OutboundFrame::subscribe)
577        .collect();
578    for frame in replay {
579        send(&mut ws, frame).await;
580    }
581
582    let mut ping = options
583        .heartbeat
584        .map(|hb| tokio::time::interval(hb.interval));
585    let mut stale_deadline: Option<tokio::time::Instant> = None;
586
587    loop {
588        let stale = async {
589            match stale_deadline {
590                Some(deadline) => tokio::time::sleep_until(deadline).await,
591                None => std::future::pending().await,
592            }
593        };
594
595        tokio::select! {
596            message = ws.next() => match message {
597                Some(Ok(message)) => {
598                    stale_deadline = None;
599                    if let Some(outcome) = handle_message(&message, options, events_tx, lifecycle_tx) {
600                        return outcome;
601                    }
602                }
603                Some(Err(error)) => {
604                    let _ = lifecycle_tx.send(LifecycleEvent::Error(RadionError::transport(error)));
605                    return SessionOutcome::Disconnected { code: 1006, reason: String::new() };
606                }
607                None => return SessionOutcome::Disconnected { code: 1006, reason: String::new() },
608            },
609            command = cmd_rx.recv() => match command {
610                Some(Command::Subscribe(subscription)) => {
611                    if subscriptions.add(subscription.clone()) {
612                        send(&mut ws, OutboundFrame::subscribe(&subscription)).await;
613                    }
614                }
615                Some(Command::Unsubscribe(id)) => {
616                    if subscriptions.remove(&id) {
617                        send(&mut ws, OutboundFrame::Unsubscribe { id }).await;
618                    }
619                }
620                Some(Command::Close { code, reason }) => {
621                    let _ = ws.close().await;
622                    return SessionOutcome::Shutdown { code, reason };
623                }
624                None => {
625                    // Handle dropped: client gone, shut down quietly.
626                    let _ = ws.close().await;
627                    return SessionOutcome::Shutdown { code: 1000, reason: String::from("client dropped") };
628                }
629            },
630            () = next_ping(&mut ping) => {
631                send(&mut ws, OutboundFrame::Ping).await;
632                if stale_deadline.is_none() {
633                    if let Some(hb) = options.heartbeat {
634                        stale_deadline = Some(tokio::time::Instant::now() + hb.timeout);
635                    }
636                }
637            }
638            () = stale => {
639                let _ = lifecycle_tx.send(LifecycleEvent::Error(RadionError::connection("stale connection")));
640                return SessionOutcome::Disconnected { code: 1006, reason: String::from("stale connection") };
641            }
642        }
643    }
644}
645
646/// Await the next heartbeat tick, or never if heartbeats are disabled.
647async fn next_ping(ping: &mut Option<tokio::time::Interval>) {
648    match ping {
649        Some(interval) => {
650            interval.tick().await;
651        }
652        None => std::future::pending().await,
653    }
654}
655
656/// Route an inbound message. Returns `Some` to end the session on a close frame.
657///
658/// Text frames are parsed as JSON. Binary frames are inflated first when
659/// compression is on, so a server may mix both on one connection.
660fn handle_message(
661    message: &Message,
662    options: &RealtimeOptions,
663    events_tx: &broadcast::Sender<ChannelEvent>,
664    lifecycle_tx: &broadcast::Sender<LifecycleEvent>,
665) -> Option<SessionOutcome> {
666    match message {
667        Message::Text(text) => {
668            route_text(text, events_tx, lifecycle_tx);
669            None
670        }
671        Message::Binary(bytes) => {
672            match options.decode_binary(bytes) {
673                Ok(text) => route_text(&text, events_tx, lifecycle_tx),
674                Err(error) => {
675                    let _ = lifecycle_tx.send(LifecycleEvent::Error(error));
676                }
677            }
678            None
679        }
680        Message::Close(frame) => {
681            let (code, reason) = frame
682                .as_ref()
683                .map(|frame| (u16::from(frame.code), frame.reason.to_string()))
684                .unwrap_or((1005, String::new()));
685            Some(SessionOutcome::Disconnected { code, reason })
686        }
687        Message::Ping(_) | Message::Pong(_) | Message::Frame(_) => None,
688    }
689}
690
691/// Parse a text frame and route it to the event / lifecycle broadcasts.
692fn route_text(
693    text: &str,
694    events_tx: &broadcast::Sender<ChannelEvent>,
695    lifecycle_tx: &broadcast::Sender<LifecycleEvent>,
696) {
697    let Some(frame) = parse_inbound_frame(text) else {
698        return;
699    };
700    match frame {
701        frame @ InboundFrame::Event { .. } => {
702            if let Some(event) = frame.into_channel_event() {
703                let _ = events_tx.send(event);
704            }
705        }
706        InboundFrame::Warning { code, id, message } => {
707            let _ = lifecycle_tx.send(LifecycleEvent::Warning { code, id, message });
708        }
709        InboundFrame::Error {
710            message,
711            code,
712            id,
713            channel,
714            ..
715        } => {
716            let _ = lifecycle_tx.send(LifecycleEvent::Error(RadionError::Server {
717                message,
718                code,
719                channel,
720                id,
721            }));
722        }
723        InboundFrame::Pong
724        | InboundFrame::Subscribed { .. }
725        | InboundFrame::Unsubscribed { .. } => {}
726    }
727}
728
729/// Serialize and send an outbound frame, dropping it on a transport error.
730async fn send<S>(ws: &mut S, frame: OutboundFrame)
731where
732    S: Sink<Message, Error = WsError> + Unpin,
733{
734    if let Ok(text) = serde_json::to_string(&frame) {
735        let _ = ws.send(Message::text(text)).await;
736    }
737}
738
739#[cfg(test)]
740mod auth_wiring_tests {
741    use super::*;
742
743    #[test]
744    fn defaults_have_no_token_and_header_mode() {
745        let options = RealtimeOptions::new("k");
746        assert!(options.token_provider.is_none());
747        assert!(!options.auth_in_query);
748    }
749
750    #[tokio::test]
751    async fn static_token_builder_sets_provider() {
752        let options = RealtimeOptions::new("k").token("jwt");
753        let provider = options.token_provider.expect("provider set");
754        assert_eq!(provider.fetch().await.unwrap(), "jwt");
755    }
756
757    #[test]
758    fn auth_in_query_builder_flips_flag() {
759        assert!(RealtimeOptions::new("k").auth_in_query(true).auth_in_query);
760    }
761
762    #[test]
763    fn accepts_async_provider() {
764        let _ = RealtimeOptions::new("k")
765            .token_provider(TokenProvider::new(|| async { Ok("x".into()) }));
766    }
767}
768
769#[cfg(all(test, feature = "compression"))]
770mod compression_wiring_tests {
771    use std::io::Write;
772
773    use flate2::Compression;
774    use flate2::write::ZlibEncoder;
775
776    use super::*;
777
778    const PONG: &str = r#"{"type":"pong"}"#;
779    const EVENT: &str = r#"{"type":"event","id":"t","channel":"trading","confirmed":true,"seq":1,"sent_at_ms":1721818200123,"data":{"type":"order_cancelled"}}"#;
780
781    fn deflate(text: &str) -> Vec<u8> {
782        let mut encoder = ZlibEncoder::new(Vec::new(), Compression::default());
783        encoder.write_all(text.as_bytes()).expect("writes");
784        encoder.finish().expect("finishes")
785    }
786
787    #[test]
788    fn compression_is_off_by_default() {
789        assert!(!RealtimeOptions::new("k").compression);
790    }
791
792    #[test]
793    fn builder_flips_the_flag() {
794        assert!(RealtimeOptions::new("k").compression(true).compression);
795    }
796
797    #[test]
798    fn connect_url_is_untouched_when_compression_is_off() {
799        let options = RealtimeOptions::new("k").url("wss://example.test/ws");
800        assert_eq!(connect_url(&options), "wss://example.test/ws");
801    }
802
803    #[test]
804    fn connect_url_asks_for_zlib_when_compression_is_on() {
805        let options = RealtimeOptions::new("k")
806            .url("wss://example.test/ws")
807            .compression(true);
808        assert_eq!(connect_url(&options), "wss://example.test/ws?compress=zlib");
809    }
810
811    #[test]
812    fn connect_url_keeps_an_existing_query() {
813        let options = RealtimeOptions::new("k")
814            .url("wss://example.test/ws?v=1")
815            .compression(true);
816        assert_eq!(
817            connect_url(&options),
818            "wss://example.test/ws?v=1&compress=zlib"
819        );
820    }
821
822    #[test]
823    fn binary_frames_inflate_when_compression_is_on() {
824        let options = RealtimeOptions::new("k").compression(true);
825        assert_eq!(options.decode_binary(&deflate(PONG)).unwrap(), PONG);
826    }
827
828    #[test]
829    fn binary_frames_stay_plain_when_compression_is_off() {
830        let options = RealtimeOptions::new("k");
831        assert_eq!(options.decode_binary(PONG.as_bytes()).unwrap(), PONG);
832    }
833
834    #[test]
835    fn inflate_failure_surfaces_on_the_lifecycle_stream() {
836        let options = RealtimeOptions::new("k").compression(true);
837        let (events_tx, _events_rx) = broadcast::channel(EVENT_BUFFER);
838        let (lifecycle_tx, mut lifecycle_rx) = broadcast::channel(LIFECYCLE_BUFFER);
839
840        let outcome = handle_message(
841            &Message::binary(b"not zlib at all".to_vec()),
842            &options,
843            &events_tx,
844            &lifecycle_tx,
845        );
846
847        assert!(outcome.is_none());
848        match lifecycle_rx.try_recv().expect("lifecycle event") {
849            LifecycleEvent::Error(RadionError::Decompression(_)) => {}
850            other => panic!("expected a decompression error, got {other:?}"),
851        }
852    }
853
854    #[test]
855    fn text_and_compressed_binary_both_deliver_events() {
856        let options = RealtimeOptions::new("k").compression(true);
857        let (events_tx, mut events_rx) = broadcast::channel(EVENT_BUFFER);
858        let (lifecycle_tx, _lifecycle_rx) = broadcast::channel(LIFECYCLE_BUFFER);
859
860        handle_message(&Message::text(EVENT), &options, &events_tx, &lifecycle_tx);
861        handle_message(
862            &Message::binary(deflate(EVENT)),
863            &options,
864            &events_tx,
865            &lifecycle_tx,
866        );
867
868        assert_eq!(events_rx.try_recv().expect("text event").channel, "trading");
869        assert_eq!(
870            events_rx.try_recv().expect("binary event").channel,
871            "trading"
872        );
873    }
874}
radion_sdk/realtime/client.rs

radion_sdk/realtime/
client.rs
