web_rpc/
interface.rs

1use futures_channel::{mpsc, oneshot};
2use futures_util::future;
3use wasm_bindgen::{JsCast, JsValue};
4
5/// An interface represents a [`crate::port::Port`] that has been fully initialised and has
6/// verified that the other end of the channel is ready to receive messages.
7pub struct Interface {
8    pub(crate) port: crate::port::Port,
9    pub(crate) listener: gloo_events::EventListener,
10    pub(crate) messages_rx: mpsc::UnboundedReceiver<js_sys::Array>,
11}
12
13impl Interface {
14    /// Create a new interface from anything that implements `Into<Port>`, for example, a
15    /// [`web_sys::MessagePort`], a [`web_sys::Worker`], or a
16    /// [`web_sys::DedicatedWorkerGlobalScope`]. This function is async and resolves to the new
17    /// interface instance once the other side of the channel is ready.
18    ///
19    /// The transport is used, not owned: a [`web_sys::MessagePort`] must have been
20    /// [`start`](web_sys::MessagePort::start)ed by its owner before it is passed here, or the
21    /// handshake never completes.
22    pub async fn new(port: impl Into<crate::port::Port>) -> Self {
23        let port = port.into();
24        let (dispatcher_tx, dispatcher_rx) = mpsc::unbounded();
25        let (ready_tx, ready_rx) = oneshot::channel();
26        let mut ready_tx = Option::from(ready_tx);
27        let listener =
28            gloo_events::EventListener::new(port.event_target(), "message", move |event| {
29                let message = event.unchecked_ref::<web_sys::MessageEvent>().data();
30                match message.dyn_into::<js_sys::Array>() {
31                    /* default path, enqueue the message for deserialization by the dispatcher */
32                    Ok(array) => {
33                        let _ = dispatcher_tx.unbounded_send(array);
34                    }
35                    /* handshake path */
36                    Err(_) => {
37                        if let Some(ready_tx) = ready_tx.take() {
38                            let _ = ready_tx.send(());
39                        }
40                    }
41                }
42            });
43        /* poll other end of the channel */
44        let port_cloned = port.clone();
45        let poll = std::pin::pin!(async move {
46            loop {
47                port_cloned
48                    .post_message(&JsValue::NULL, &JsValue::UNDEFINED)
49                    .unwrap();
50                gloo_timers::future::TimeoutFuture::new(10).await;
51            }
52        });
53        future::select(ready_rx, poll).await;
54        /* at this point we know the other end's listener is available, but we may
55        need to send one last message to indicate that we are available */
56        port.post_message(&JsValue::NULL, &JsValue::UNDEFINED)
57            .unwrap();
58        /* return the interface */
59        Self {
60            messages_rx: dispatcher_rx,
61            listener,
62            port,
63        }
64    }
65}
web_rpc/interface.rs

web_rpc/
interface.rs
