phoenix_channel_runtime_web/
lib.rs

1#![doc = include_str!("../README.md")]
2#![forbid(unsafe_code)]
3#![warn(missing_docs)]
4#![warn(rustdoc::broken_intra_doc_links)]
5
6#[cfg(target_arch = "wasm32")]
7mod web {
8    use std::{
9        cell::{Cell, RefCell},
10        collections::{HashMap, VecDeque},
11        pin::Pin,
12        rc::Rc,
13        time::Duration,
14    };
15
16    use base64::{
17        Engine as _,
18        engine::general_purpose::{STANDARD, STANDARD_NO_PAD},
19    };
20    use futures::{FutureExt, Sink, SinkExt, StreamExt, channel::mpsc, future::poll_fn};
21    use gloo_net::http::Request;
22    use gloo_net::websocket::{Message, State, WebSocketError, futures::WebSocket};
23    use gloo_timers::future::TimeoutFuture;
24    use phoenix_channel_client::{
25        ConnectContext, Connector, Endpoint, ResolvedEndpoint, Socket, SocketStatus, Timer,
26    };
27    use phoenix_channel_runtime::{
28        Transport, TransportClose, TransportCloseRequest, TransportError, TransportErrorKind,
29        TransportEvent, WireMessage,
30    };
31    use serde::Deserialize;
32    use wasm_bindgen::{JsCast, JsValue, closure::Closure};
33    use web_sys::{AbortController, Event, VisibilityState, Window};
34
35    const AUTH_TOKEN_PREFIX: &str = "base64url.bearer.phx.";
36    const DEFAULT_LONG_POLL_TIMEOUT: Duration = Duration::from_secs(20);
37    const HEALTH_CHECK_REFERENCE: &str = "phoenix-channel-runtime-health";
38
39    /// Browser WebSocket transport implementing the runtime-neutral transport API.
40    pub struct WebTransport {
41        inner: Option<WebSocket>,
42        queued: VecDeque<TransportEvent>,
43    }
44
45    /// Phoenix LongPoll transport for browsers without a usable WebSocket path.
46    pub struct LongPollTransport {
47        endpoint: url::Url,
48        token: Rc<RefCell<Option<String>>>,
49        auth_token: Option<String>,
50        events: mpsc::UnboundedReceiver<Result<TransportEvent, TransportError>>,
51        closed: Rc<Cell<bool>>,
52        request_timeout: Duration,
53        requests: Rc<RefCell<HashMap<u64, AbortController>>>,
54        next_request_id: Rc<Cell<u64>>,
55    }
56
57    /// Browser page and network lifecycle listeners for a managed socket.
58    ///
59    /// Keep this value alive while the socket is active. Dropping it removes
60    /// all installed event listeners.
61    pub struct WebLifecycle {
62        window: Window,
63        pagehide: Closure<dyn FnMut(Event)>,
64        pageshow: Closure<dyn FnMut(Event)>,
65        visibilitychange: Closure<dyn FnMut(Event)>,
66        offline: Closure<dyn FnMut(Event)>,
67        online: Closure<dyn FnMut(Event)>,
68    }
69
70    impl WebLifecycle {
71        /// Attaches page hide/show, visibility, offline, and online listeners.
72        pub fn attach(socket: Socket) -> Result<Self, JsValue> {
73            let window =
74                web_sys::window().ok_or_else(|| JsValue::from_str("window unavailable"))?;
75            let resume_after_page_show = Rc::new(Cell::new(false));
76            let resume_after_online = Rc::new(Cell::new(false));
77
78            let pagehide = {
79                let socket = socket.clone();
80                let resume = resume_after_page_show.clone();
81                Closure::wrap(Box::new(move |_event: Event| {
82                    if is_active(socket.status()) {
83                        resume.set(true);
84                        let socket = socket.clone();
85                        wasm_bindgen_futures::spawn_local(async move {
86                            let _ = socket.disconnect().await;
87                        });
88                    }
89                }) as Box<dyn FnMut(Event)>)
90            };
91            let pageshow = {
92                let socket = socket.clone();
93                let resume = resume_after_page_show;
94                Closure::wrap(Box::new(move |_event: Event| {
95                    if resume.replace(false) {
96                        let socket = socket.clone();
97                        wasm_bindgen_futures::spawn_local(async move {
98                            let _ = socket.connect().await;
99                        });
100                    }
101                }) as Box<dyn FnMut(Event)>)
102            };
103            let visibilitychange = {
104                let socket = socket.clone();
105                let window = window.clone();
106                Closure::wrap(Box::new(move |_event: Event| {
107                    let visible = window.document().is_some_and(|document| {
108                        document.visibility_state() == VisibilityState::Visible
109                    });
110                    if visible && socket.status() == SocketStatus::WaitingToReconnect {
111                        let socket = socket.clone();
112                        wasm_bindgen_futures::spawn_local(async move {
113                            let _ = socket.reconnect().await;
114                        });
115                    }
116                }) as Box<dyn FnMut(Event)>)
117            };
118            let offline = {
119                let socket = socket.clone();
120                let resume = resume_after_online.clone();
121                Closure::wrap(Box::new(move |_event: Event| {
122                    if is_active(socket.status()) {
123                        resume.set(true);
124                        let socket = socket.clone();
125                        wasm_bindgen_futures::spawn_local(async move {
126                            let _ = socket.disconnect().await;
127                        });
128                    }
129                }) as Box<dyn FnMut(Event)>)
130            };
131            let online = {
132                let socket = socket.clone();
133                let resume = resume_after_online;
134                Closure::wrap(Box::new(move |_event: Event| {
135                    if resume.replace(false) {
136                        let socket = socket.clone();
137                        wasm_bindgen_futures::spawn_local(async move {
138                            let _ = socket.connect().await;
139                        });
140                    }
141                }) as Box<dyn FnMut(Event)>)
142            };
143
144            add_listener(&window, "pagehide", &pagehide)?;
145            add_listener(&window, "pageshow", &pageshow)?;
146            add_listener(&window, "visibilitychange", &visibilitychange)?;
147            add_listener(&window, "offline", &offline)?;
148            add_listener(&window, "online", &online)?;
149            Ok(Self {
150                window,
151                pagehide,
152                pageshow,
153                visibilitychange,
154                offline,
155                online,
156            })
157        }
158    }
159
160    impl Drop for WebLifecycle {
161        fn drop(&mut self) {
162            remove_listener(&self.window, "pagehide", &self.pagehide);
163            remove_listener(&self.window, "pageshow", &self.pageshow);
164            remove_listener(&self.window, "visibilitychange", &self.visibilitychange);
165            remove_listener(&self.window, "offline", &self.offline);
166            remove_listener(&self.window, "online", &self.online);
167        }
168    }
169
170    fn is_active(status: SocketStatus) -> bool {
171        matches!(
172            status,
173            SocketStatus::Connecting | SocketStatus::Connected | SocketStatus::WaitingToReconnect
174        )
175    }
176
177    fn add_listener(
178        window: &Window,
179        name: &str,
180        callback: &Closure<dyn FnMut(Event)>,
181    ) -> Result<(), JsValue> {
182        window.add_event_listener_with_callback(name, callback.as_ref().unchecked_ref())
183    }
184
185    fn remove_listener(window: &Window, name: &str, callback: &Closure<dyn FnMut(Event)>) {
186        let _ = window.remove_event_listener_with_callback(name, callback.as_ref().unchecked_ref());
187    }
188
189    /// Browser connector with optional Phoenix LongPoll fallback.
190    #[derive(Clone)]
191    pub struct WebConnector {
192        endpoint: WebEndpoint,
193        long_poll_fallback: Option<Duration>,
194        long_poll_timeout: Duration,
195    }
196
197    /// Browser connector that always uses Phoenix LongPoll.
198    #[derive(Clone)]
199    pub struct LongPollConnector {
200        endpoint: WebEndpoint,
201        request_timeout: Duration,
202    }
203
204    #[derive(Clone)]
205    enum WebEndpoint {
206        Url(String),
207        Phoenix(Endpoint),
208    }
209
210    impl WebConnector {
211        /// Creates a connector for an already-resolved WebSocket URL.
212        pub fn new(url: impl Into<String>) -> Self {
213            Self {
214                endpoint: WebEndpoint::Url(url.into()),
215                long_poll_fallback: None,
216                long_poll_timeout: DEFAULT_LONG_POLL_TIMEOUT,
217            }
218        }
219
220        /// Creates a connector that resolves a Phoenix endpoint each attempt.
221        pub fn from_endpoint(endpoint: Endpoint) -> Self {
222            Self {
223                endpoint: WebEndpoint::Phoenix(endpoint),
224                long_poll_fallback: None,
225                long_poll_timeout: DEFAULT_LONG_POLL_TIMEOUT,
226            }
227        }
228
229        /// Enables LongPoll fallback after this WebSocket open or health deadline.
230        pub fn long_poll_fallback(mut self, after: Duration) -> Self {
231            self.long_poll_fallback = Some(after);
232            self
233        }
234
235        /// Sets the timeout applied to each fallback LongPoll request.
236        pub fn long_poll_timeout(mut self, timeout: Duration) -> Self {
237            self.long_poll_timeout = timeout;
238            self
239        }
240    }
241
242    impl LongPollConnector {
243        /// Creates a LongPoll connector from an already-resolved endpoint URL.
244        pub fn new(url: impl Into<String>) -> Self {
245            Self {
246                endpoint: WebEndpoint::Url(url.into()),
247                request_timeout: DEFAULT_LONG_POLL_TIMEOUT,
248            }
249        }
250
251        /// Creates a LongPoll connector that resolves a Phoenix endpoint each attempt.
252        pub fn from_endpoint(endpoint: Endpoint) -> Self {
253            Self {
254                endpoint: WebEndpoint::Phoenix(endpoint),
255                request_timeout: DEFAULT_LONG_POLL_TIMEOUT,
256            }
257        }
258
259        /// Sets the timeout applied to every LongPoll request.
260        pub fn request_timeout(mut self, timeout: Duration) -> Self {
261            self.request_timeout = timeout;
262            self
263        }
264    }
265
266    async fn resolve_endpoint(
267        endpoint: WebEndpoint,
268        context: ConnectContext,
269    ) -> Result<ResolvedEndpoint, TransportError> {
270        match endpoint {
271            WebEndpoint::Url(url) => Ok(ResolvedEndpoint {
272                url,
273                protocols: Vec::new(),
274            }),
275            WebEndpoint::Phoenix(endpoint) => endpoint.resolve(context).await.map_err(|error| {
276                TransportError::with_kind(TransportErrorKind::Connect, error.to_string())
277            }),
278        }
279    }
280
281    impl Connector for WebConnector {
282        fn connect(
283            &self,
284            context: ConnectContext,
285        ) -> futures::future::LocalBoxFuture<'static, Result<Box<dyn Transport>, TransportError>>
286        {
287            let endpoint = self.endpoint.clone();
288            let fallback = self.long_poll_fallback;
289            let long_poll_timeout = self.long_poll_timeout;
290            Box::pin(async move {
291                let endpoint = resolve_endpoint(endpoint, context).await?;
292                if let Some(after) = fallback {
293                    if let Ok(transport) =
294                        WebTransport::connect_with_health_check(endpoint.clone(), after).await
295                    {
296                        return Ok(Box::new(transport) as Box<dyn Transport>);
297                    }
298                    let transport =
299                        LongPollTransport::connect_resolved(endpoint, long_poll_timeout).await?;
300                    Ok(Box::new(transport) as Box<dyn Transport>)
301                } else {
302                    let transport = WebTransport::connect_resolved(endpoint).await?;
303                    Ok(Box::new(transport) as Box<dyn Transport>)
304                }
305            })
306        }
307    }
308
309    impl Connector for LongPollConnector {
310        fn connect(
311            &self,
312            context: ConnectContext,
313        ) -> futures::future::LocalBoxFuture<'static, Result<Box<dyn Transport>, TransportError>>
314        {
315            let endpoint = self.endpoint.clone();
316            let request_timeout = self.request_timeout;
317            Box::pin(async move {
318                let endpoint = resolve_endpoint(endpoint, context).await?;
319                let transport =
320                    LongPollTransport::connect_resolved(endpoint, request_timeout).await?;
321                Ok(Box::new(transport) as Box<dyn Transport>)
322            })
323        }
324    }
325
326    /// Browser timer backed by `setTimeout` and the Performance API.
327    #[derive(Clone, Copy, Debug, Default)]
328    pub struct WebTimer;
329
330    impl Timer for WebTimer {
331        fn sleep(&self, duration: Duration) -> futures::future::LocalBoxFuture<'static, ()> {
332            let milliseconds = duration.as_millis().min(u128::from(u32::MAX)) as u32;
333            Box::pin(TimeoutFuture::new(milliseconds))
334        }
335
336        fn now(&self) -> Duration {
337            web_sys::window()
338                .and_then(|window| window.performance())
339                .map(|performance| Duration::from_secs_f64(performance.now() / 1000.0))
340                .unwrap_or_default()
341        }
342    }
343
344    impl WebTransport {
345        /// Opens an already-resolved browser WebSocket URL.
346        pub fn connect(url: &str) -> Result<Self, TransportError> {
347            let inner = WebSocket::open(url).map_err(|error| {
348                TransportError::with_kind(TransportErrorKind::Connect, format!("{error:?}"))
349            })?;
350            Ok(Self {
351                inner: Some(inner),
352                queued: VecDeque::new(),
353            })
354        }
355
356        async fn connect_with_health_check(
357            endpoint: ResolvedEndpoint,
358            threshold: Duration,
359        ) -> Result<Self, TransportError> {
360            let connect = Self::connect_resolved(endpoint).fuse();
361            let open_timeout = timeout(threshold).fuse();
362            futures::pin_mut!(connect, open_timeout);
363            let mut transport = futures::select! {
364                result = connect => result?,
365                () = open_timeout => return Err(connect_error("WebSocket open timed out")),
366            };
367
368            {
369                let health_check = transport.health_check().fuse();
370                let health_timeout = timeout(threshold).fuse();
371                futures::pin_mut!(health_check, health_timeout);
372                futures::select! {
373                    result = health_check => result?,
374                    () = health_timeout => return Err(connect_error("WebSocket health check timed out")),
375                }
376            }
377            Ok(transport)
378        }
379
380        async fn connect_resolved(endpoint: ResolvedEndpoint) -> Result<Self, TransportError> {
381            let mut inner = if endpoint.protocols.is_empty() {
382                WebSocket::open(&endpoint.url)
383            } else {
384                WebSocket::open_with_protocols(&endpoint.url, &endpoint.protocols)
385            }
386            .map_err(|error| {
387                TransportError::with_kind(TransportErrorKind::Connect, format!("{error:?}"))
388            })?;
389            poll_fn(|context| Pin::new(&mut inner).poll_ready(context))
390                .await
391                .map_err(|error| {
392                    TransportError::with_kind(TransportErrorKind::Connect, error.to_string())
393                })?;
394            if !matches!(inner.state(), State::Open) {
395                return Err(TransportError::with_kind(
396                    TransportErrorKind::Connect,
397                    "WebSocket closed before the open event",
398                ));
399            }
400            Ok(Self {
401                inner: Some(inner),
402                queued: VecDeque::new(),
403            })
404        }
405
406        async fn health_check(&mut self) -> Result<(), TransportError> {
407            let message =
408                serde_json::json!([null, HEALTH_CHECK_REFERENCE, "phoenix", "heartbeat", {}])
409                    .to_string();
410            let inner = self.inner.as_mut().ok_or_else(closed_error)?;
411            inner.send(Message::Text(message)).await.map_err(|error| {
412                TransportError::with_kind(TransportErrorKind::Send, format!("{error:?}"))
413            })?;
414
415            loop {
416                let Some(message) = inner.next().await else {
417                    return Err(connect_error("WebSocket ended during health check"));
418                };
419                let event = websocket_event(message)?;
420                if let TransportEvent::Message(WireMessage::Text(text)) = &event {
421                    if is_health_check_ack(text) {
422                        return Ok(());
423                    }
424                }
425                if matches!(event, TransportEvent::Closed(_)) {
426                    return Err(connect_error("WebSocket closed during health check"));
427                }
428                self.queued.push_back(event);
429            }
430        }
431    }
432
433    fn is_health_check_ack(text: &str) -> bool {
434        let Ok(frame) = serde_json::from_str::<serde_json::Value>(text) else {
435            return false;
436        };
437        let Some(frame) = frame.as_array() else {
438            return false;
439        };
440        frame.get(1).and_then(serde_json::Value::as_str) == Some(HEALTH_CHECK_REFERENCE)
441            && frame.get(2).and_then(serde_json::Value::as_str) == Some("phoenix")
442            && frame.get(3).and_then(serde_json::Value::as_str) == Some("phx_reply")
443            && frame
444                .get(4)
445                .and_then(|payload| payload.get("status"))
446                .and_then(serde_json::Value::as_str)
447                == Some("ok")
448    }
449
450    #[derive(Deserialize)]
451    struct PollResponse {
452        status: u16,
453        #[serde(default)]
454        token: Option<String>,
455        #[serde(default)]
456        messages: Vec<String>,
457    }
458
459    impl LongPollTransport {
460        async fn connect_resolved(
461            endpoint: ResolvedEndpoint,
462            request_timeout: Duration,
463        ) -> Result<Self, TransportError> {
464            let mut url = url::Url::parse(&endpoint.url).map_err(connect_error)?;
465            let scheme = match url.scheme() {
466                "ws" => "http".to_owned(),
467                "wss" => "https".to_owned(),
468                "http" | "https" => url.scheme().to_owned(),
469                scheme => {
470                    return Err(connect_error(format!(
471                        "unsupported LongPoll URL scheme {scheme}"
472                    )));
473                }
474            };
475            url.set_scheme(&scheme)
476                .map_err(|_| connect_error("failed to set LongPoll URL scheme"))?;
477            let path = url.path().trim_end_matches('/');
478            let path = path.strip_suffix("/websocket").map_or_else(
479                || format!("{path}/longpoll"),
480                |base| format!("{base}/longpoll"),
481            );
482            url.set_path(&path);
483
484            let auth_token = endpoint
485                .protocols
486                .get(1)
487                .and_then(|protocol| protocol.strip_prefix(AUTH_TOKEN_PREFIX))
488                .and_then(|encoded| STANDARD_NO_PAD.decode(encoded).ok())
489                .and_then(|bytes| String::from_utf8(bytes).ok());
490            let requests = Rc::new(RefCell::new(HashMap::new()));
491            let next_request_id = Rc::new(Cell::new(0));
492            let response = long_poll_request(
493                &url,
494                None,
495                auth_token.as_deref(),
496                None,
497                request_timeout,
498                &requests,
499                &next_request_id,
500                TransportErrorKind::Connect,
501            )
502            .await?;
503            match response.status {
504                200 | 204 | 410 => {
505                    let Some(initial_token) = response.token else {
506                        return Err(connect_error("LongPoll handshake did not return a token"));
507                    };
508                    let token = Rc::new(RefCell::new(Some(initial_token)));
509                    let closed = Rc::new(Cell::new(false));
510                    let (events_tx, events) = mpsc::unbounded();
511                    for message in response.messages {
512                        let _ = events_tx.unbounded_send(Ok(TransportEvent::Message(
513                            WireMessage::Text(message),
514                        )));
515                    }
516                    let poll_endpoint = url.clone();
517                    let poll_token = token.clone();
518                    let poll_auth_token = auth_token.clone();
519                    let poll_closed = closed.clone();
520                    let poll_requests = requests.clone();
521                    let poll_next_request_id = next_request_id.clone();
522                    wasm_bindgen_futures::spawn_local(async move {
523                        while !poll_closed.get() {
524                            let current_token = poll_token.borrow().clone();
525                            let response = long_poll_request(
526                                &poll_endpoint,
527                                current_token.as_deref(),
528                                poll_auth_token.as_deref(),
529                                None,
530                                request_timeout,
531                                &poll_requests,
532                                &poll_next_request_id,
533                                TransportErrorKind::Receive,
534                            )
535                            .await;
536                            let response = match response {
537                                Ok(response) => response,
538                                Err(error) => {
539                                    let _ = events_tx.unbounded_send(Err(error));
540                                    break;
541                                }
542                            };
543                            match response.status {
544                                200 => {
545                                    if let Some(next_token) = response.token {
546                                        poll_token.replace(Some(next_token));
547                                    }
548                                    for message in response.messages {
549                                        if events_tx
550                                            .unbounded_send(Ok(TransportEvent::Message(
551                                                WireMessage::Text(message),
552                                            )))
553                                            .is_err()
554                                        {
555                                            return;
556                                        }
557                                    }
558                                }
559                                204 => {}
560                                410 => {
561                                    poll_closed.set(true);
562                                    let _ = events_tx.unbounded_send(Ok(TransportEvent::Closed(
563                                        TransportClose::new(
564                                            Some(3410),
565                                            "LongPoll session is gone",
566                                            false,
567                                        ),
568                                    )));
569                                }
570                                403 => {
571                                    poll_closed.set(true);
572                                    let _ = events_tx.unbounded_send(Ok(TransportEvent::Closed(
573                                        TransportClose::new(
574                                            Some(1008),
575                                            "LongPoll request was forbidden",
576                                            false,
577                                        ),
578                                    )));
579                                }
580                                status => {
581                                    let _ =
582                                        events_tx.unbounded_send(Err(TransportError::with_kind(
583                                            TransportErrorKind::Receive,
584                                            format!("LongPoll GET returned status {status}"),
585                                        )));
586                                    break;
587                                }
588                            }
589                        }
590                    });
591                    Ok(Self {
592                        endpoint: url,
593                        token,
594                        auth_token,
595                        events,
596                        closed,
597                        request_timeout,
598                        requests,
599                        next_request_id,
600                    })
601                }
602                403 => Err(connect_error("LongPoll handshake was forbidden")),
603                status => Err(connect_error(format!(
604                    "LongPoll handshake returned status {status}"
605                ))),
606            }
607        }
608
609        async fn post(&self, body: String) -> Result<(), TransportError> {
610            let token = self.token.borrow().clone();
611            let response = long_poll_request(
612                &self.endpoint,
613                token.as_deref(),
614                self.auth_token.as_deref(),
615                Some(body),
616                self.request_timeout,
617                &self.requests,
618                &self.next_request_id,
619                TransportErrorKind::Send,
620            )
621            .await?;
622            validate_post_status(response.status)
623        }
624    }
625
626    fn validate_post_status(status: u16) -> Result<(), TransportError> {
627        match status {
628            200 => Ok(()),
629            403 => Err(TransportError::with_kind(
630                TransportErrorKind::Send,
631                "LongPoll POST was forbidden",
632            )),
633            408 => Err(TransportError::with_kind(
634                TransportErrorKind::Send,
635                "LongPoll POST dispatch timed out",
636            )),
637            410 => Err(TransportError::with_kind(
638                TransportErrorKind::Send,
639                "LongPoll POST session is gone",
640            )),
641            status => Err(TransportError::with_kind(
642                TransportErrorKind::Send,
643                format!("LongPoll POST returned status {status}"),
644            )),
645        }
646    }
647
648    fn endpoint_url(endpoint: &url::Url, token: Option<&str>) -> String {
649        let mut endpoint = endpoint.clone();
650        if let Some(token) = token {
651            endpoint.query_pairs_mut().append_pair("token", token);
652        }
653        endpoint.into()
654    }
655
656    async fn long_poll_request(
657        endpoint: &url::Url,
658        token: Option<&str>,
659        auth_token: Option<&str>,
660        body: Option<String>,
661        request_timeout: Duration,
662        requests: &Rc<RefCell<HashMap<u64, AbortController>>>,
663        next_request_id: &Rc<Cell<u64>>,
664        error_kind: TransportErrorKind,
665    ) -> Result<PollResponse, TransportError> {
666        let controller = AbortController::new()
667            .map_err(|error| TransportError::with_kind(error_kind, format!("{error:?}")))?;
668        let mut request_id = next_request_id.get().wrapping_add(1);
669        if request_id == 0 {
670            request_id = 1;
671        }
672        next_request_id.set(request_id);
673        requests.borrow_mut().insert(request_id, controller.clone());
674
675        let endpoint = endpoint_url(endpoint, token);
676        let mut request = if body.is_some() {
677            Request::post(&endpoint).header("Content-Type", "application/x-ndjson")
678        } else {
679            Request::get(&endpoint).header("Accept", "application/json")
680        }
681        .abort_signal(Some(&controller.signal()));
682        if let Some(token) = auth_token {
683            request = request.header("X-Phoenix-AuthToken", token);
684        }
685        let request = match match body {
686            Some(body) => request.body(body),
687            None => request.build(),
688        } {
689            Ok(request) => request,
690            Err(error) => {
691                requests.borrow_mut().remove(&request_id);
692                return Err(TransportError::with_kind(error_kind, error.to_string()));
693            }
694        };
695        let response = async move {
696            let response = request
697                .send()
698                .await
699                .map_err(|error| TransportError::with_kind(error_kind, error.to_string()))?;
700            let http_status = response.status();
701            let text = response
702                .text()
703                .await
704                .map_err(|error| TransportError::with_kind(error_kind, error.to_string()))?;
705            if text.is_empty() {
706                return Ok(PollResponse {
707                    status: http_status,
708                    token: token.map(str::to_owned),
709                    messages: Vec::new(),
710                });
711            }
712            serde_json::from_str(&text).map_err(|error| {
713                TransportError::with_kind(error_kind, format!("invalid LongPoll response: {error}"))
714            })
715        }
716        .fuse();
717        let timeout = timeout(request_timeout).fuse();
718        futures::pin_mut!(response, timeout);
719        let result = futures::select! {
720            result = response => result,
721            () = timeout => {
722                controller.abort();
723                Err(TransportError::with_kind(error_kind, format!(
724                    "LongPoll request timed out after {request_timeout:?}"
725                )))
726            }
727        };
728        requests.borrow_mut().remove(&request_id);
729        result
730    }
731
732    fn timeout(duration: Duration) -> TimeoutFuture {
733        TimeoutFuture::new(duration.as_millis().min(u128::from(u32::MAX)) as u32)
734    }
735
736    fn connect_error(error: impl std::fmt::Display) -> TransportError {
737        TransportError::with_kind(TransportErrorKind::Connect, error.to_string())
738    }
739
740    impl Transport for WebTransport {
741        fn send<'a>(
742            &'a mut self,
743            message: WireMessage,
744        ) -> futures::future::LocalBoxFuture<'a, Result<(), TransportError>> {
745            Box::pin(async move {
746                let message = match message {
747                    WireMessage::Text(text) => Message::Text(text),
748                    WireMessage::Binary(bytes) => Message::Bytes(bytes),
749                };
750                self.inner
751                    .as_mut()
752                    .ok_or_else(closed_error)?
753                    .send(message)
754                    .await
755                    .map_err(|error| {
756                        TransportError::with_kind(TransportErrorKind::Send, format!("{error:?}"))
757                    })
758            })
759        }
760
761        fn receive<'a>(
762            &'a mut self,
763        ) -> futures::future::LocalBoxFuture<'a, Result<TransportEvent, TransportError>> {
764            Box::pin(async move {
765                if let Some(event) = self.queued.pop_front() {
766                    return Ok(event);
767                }
768                let Some(message) = self.inner.as_mut().ok_or_else(closed_error)?.next().await
769                else {
770                    return Ok(TransportEvent::Closed(TransportClose::connection_ended()));
771                };
772                websocket_event(message)
773            })
774        }
775
776        fn close<'a>(
777            &'a mut self,
778        ) -> futures::future::LocalBoxFuture<'a, Result<(), TransportError>> {
779            Box::pin(async move {
780                let Some(inner) = self.inner.take() else {
781                    return Ok(());
782                };
783                inner.close(None, None).map_err(|error| {
784                    TransportError::with_kind(TransportErrorKind::Close, format!("{error:?}"))
785                })
786            })
787        }
788
789        fn close_with<'a>(
790            &'a mut self,
791            request: TransportCloseRequest,
792        ) -> futures::future::LocalBoxFuture<'a, Result<(), TransportError>> {
793            Box::pin(async move {
794                let Some(inner) = self.inner.take() else {
795                    return Ok(());
796                };
797                inner
798                    .close(Some(request.code), Some(&request.reason))
799                    .map_err(|error| {
800                        TransportError::with_kind(TransportErrorKind::Close, format!("{error:?}"))
801                    })
802            })
803        }
804    }
805
806    fn websocket_event(
807        message: Result<Message, WebSocketError>,
808    ) -> Result<TransportEvent, TransportError> {
809        match message {
810            Ok(Message::Text(text)) => Ok(TransportEvent::Message(WireMessage::Text(text))),
811            Ok(Message::Bytes(bytes)) => Ok(TransportEvent::Message(WireMessage::Binary(bytes))),
812            Err(WebSocketError::ConnectionClose(close)) => Ok(TransportEvent::Closed(
813                TransportClose::new(Some(close.code), close.reason, close.was_clean),
814            )),
815            Err(error) => Err(TransportError::with_kind(
816                TransportErrorKind::Receive,
817                error.to_string(),
818            )),
819        }
820    }
821
822    impl Transport for LongPollTransport {
823        fn supports_binary(&self) -> bool {
824            false
825        }
826
827        fn send<'a>(
828            &'a mut self,
829            message: WireMessage,
830        ) -> futures::future::LocalBoxFuture<'a, Result<(), TransportError>> {
831            Box::pin(async move {
832                if self.closed.get() {
833                    return Err(closed_error());
834                }
835                let body = match message {
836                    WireMessage::Text(text) => text,
837                    WireMessage::Binary(bytes) => STANDARD.encode(bytes),
838                };
839                self.post(body).await
840            })
841        }
842
843        fn receive<'a>(
844            &'a mut self,
845        ) -> futures::future::LocalBoxFuture<'a, Result<TransportEvent, TransportError>> {
846            Box::pin(async move {
847                if self.closed.get() {
848                    return Ok(TransportEvent::Closed(TransportClose::new(
849                        Some(1000),
850                        "LongPoll transport closed",
851                        true,
852                    )));
853                }
854                self.events.next().await.unwrap_or_else(|| {
855                    Ok(TransportEvent::Closed(TransportClose::connection_ended()))
856                })
857            })
858        }
859
860        fn close<'a>(
861            &'a mut self,
862        ) -> futures::future::LocalBoxFuture<'a, Result<(), TransportError>> {
863            Box::pin(async move {
864                self.closed.set(true);
865                for (_, controller) in self.requests.borrow_mut().drain() {
866                    controller.abort();
867                }
868                self.events.close();
869                Ok(())
870            })
871        }
872    }
873
874    fn closed_error() -> TransportError {
875        TransportError::with_kind(TransportErrorKind::Other, "WebSocket is closed")
876    }
877
878    #[cfg(test)]
879    mod tests {
880        use super::*;
881        use wasm_bindgen_test::wasm_bindgen_test;
882
883        wasm_bindgen_test::wasm_bindgen_test_configure!(run_in_browser);
884
885        #[wasm_bindgen_test]
886        fn classifies_long_poll_post_statuses() {
887            assert!(validate_post_status(200).is_ok());
888            for (status, message) in [
889                (403, "forbidden"),
890                (408, "timed out"),
891                (410, "session is gone"),
892            ] {
893                let error = validate_post_status(status).unwrap_err();
894                assert_eq!(error.kind(), TransportErrorKind::Send);
895                assert!(error.message().contains(message));
896            }
897        }
898    }
899}
900
901#[cfg(target_arch = "wasm32")]
902pub use web::{
903    LongPollConnector, LongPollTransport, WebConnector, WebLifecycle, WebTimer, WebTransport,
904};
phoenix_channel_runtime_web/lib.rs

phoenix_channel_runtime_web/
lib.rs
