web_rpc/lib.rs
1//! Bidirectional RPC for browsing contexts, web workers, and message channels.
2//!
3//! This crate allows you to define a service as a trait and annotate it with
4//! [`#[web_rpc::service]`](macro@service). The macro then produces a `*Client`, a `*Service`,
5//! and a forwarding trait that you can implement on the server side.
6//!
7//! Routing is inferred from each type: anything implementing
8//! [`AsRef<JsValue>`](https://docs.rs/wasm-bindgen/latest/wasm_bindgen/struct.JsValue.html) is
9//! posted through `postMessage` directly and everything that is serializable is first encoded
10//! via bincode. There is special support for `Option<T>` and `Result<T, E>` to allow Javascript
11//! types to be embedded within these types. This behaviour is recursive.
12//!
13//! # Quickstart
14//! ```rust
15//! #[web_rpc::service]
16//! pub trait Calculator {
17//! fn add(&self, left: u32, right: u32) -> u32;
18//! }
19//! struct Calc;
20//! impl Calculator for Calc {
21//! fn add(&self, left: u32, right: u32) -> u32 { left + right }
22//! }
23//! ```
24//! Wire up over a `MessageChannel`, [`Worker`](https://docs.rs/web-sys/latest/web_sys/struct.Worker.html),
25//! or any [`MessagePort`](https://docs.rs/web-sys/latest/web_sys/struct.MessagePort.html). Each
26//! Each call to [`Interface::new`] is async because temporary listeners need to detect when
27//! both ends are ready.
28//! ```rust,no_run
29//! # #[web_rpc::service]
30//! # pub trait Calculator { fn add(&self, l: u32, r: u32) -> u32; }
31//! # struct Calc;
32//! # impl Calculator for Calc { fn add(&self, l: u32, r: u32) -> u32 { l + r } }
33//! # async fn run() {
34//! let channel = web_sys::MessageChannel::new().unwrap();
35//! let (server_iface, client_iface) = futures_util::future::join(
36//! web_rpc::Interface::new(channel.port1()),
37//! web_rpc::Interface::new(channel.port2()),
38//! ).await;
39//!
40//! let server = web_rpc::Builder::new(server_iface)
41//! .with_service::<CalculatorService<_>>(Calc)
42//! .build();
43//! wasm_bindgen_futures::spawn_local(server);
44//!
45//! let client = web_rpc::Builder::new(client_iface)
46//! .with_client::<CalculatorClient>()
47//! .build();
48//! assert_eq!(client.add(41, 1).await, 42);
49//! # }
50//! ```
51//!
52//! # Routing
53//! ```rust
54//! #[web_rpc::service]
55//! pub trait Routing {
56//! // Plain types that implement Serialize go through bincode.
57//! fn add(&self, l: u32, r: u32) -> u32;
58//! // Anything `AsRef<JsValue>` is posted through the JS array.
59//! fn echo(&self, s: js_sys::JsString) -> js_sys::JsString;
60//! // `Option`/`Result` recurse: Ok(Some(_)) is posted, Ok(None) is one byte,
61//! // Err carries a bincoded `String`.
62//! fn lookup(&self, k: u32) -> Result<Option<js_sys::JsString>, String>;
63//! // `&str` / `&[u8]` deserialize zero-copy on the server.
64//! fn count(&self, data: &[u8]) -> usize;
65//! // References to JS types are accepted too and are decoded via JsCast::dyn_ref.
66//! fn len(&self, s: &js_sys::JsString) -> u32;
67//! }
68//! ```
69//!
70//! # Async, notifications, streaming
71//! ```rust
72//! use futures_core::Stream;
73//!
74//! #[web_rpc::service]
75//! pub trait Misc {
76//! // `async` here makes the server impl async; the client side is also async because we return a u32.
77//! async fn slow(&self, ms: u32) -> u32;
78//! // No return type means the method is a notification.
79//! fn fire(&self, msg: String);
80//! // `impl Stream<Item = T>` makes the method a streaming RPC.
81//! fn items(&self, n: u32) -> impl Stream<Item = u32>;
82//! }
83//! ```
84//! On the client side, RPC methods that have a return type are async and yield a
85//! [`client::RequestFuture<T>`] which you await for the response. Methods without a return type
86//! are sync and act as fire-and-forget notifications. This is independent of whether the trait
87//! method itself is marked `async`, which only affects the server implementation. Dropping the
88//! `RequestFuture` cancels the request, so notifications cannot be cancelled.
89//!
90//! Streaming methods return a [`client::StreamReceiver<T>`] that yields each item the server
91//! produces. Dropping the receiver aborts the stream on the server, while
92//! [`close`](client::StreamReceiver::close) lets buffered items finish arriving instead.
93//! Streaming methods can also be `async` and the items they yield can be wrapper types like
94//! `Result<JsT, E>`.
95//!
96//! # Transfer
97//! Anything that should be transferred to the other side rather than copied with the structured
98//! clone algorithm can be specified inside a `#[transfer(...)]` attribute as a comma-separated
99//! list. The simplest case is to list the parameter that holds the transferable value, but if
100//! that value is wrapped or derived from a parameter, you can use a parameter-name expression
101//! (`name => expr`, evaluated with `name` in scope), a closure with a refutable pattern
102//! (`name => |pat| body`), or a match-block (`name => match { arm, ... }`). The same forms also
103//! work for the return value via `return`.
104//! ```rust
105//! # use wasm_bindgen::JsCast;
106//! #[web_rpc::service]
107//! pub trait Transfer {
108//! // Bare param + derived expression + return closure.
109//! #[transfer(
110//! canvas,
111//! data => data.buffer(),
112//! return => |Ok(buf)| buf.buffer(),
113//! )]
114//! fn render(
115//! &self,
116//! canvas: web_sys::OffscreenCanvas,
117//! data: js_sys::Uint8Array,
118//! ) -> Result<js_sys::Uint8Array, String>;
119//!
120//! // Match-block: useful when several variants need transferring.
121//! #[transfer(return => match { Some(buf) => buf.buffer(), })]
122//! fn maybe(&self) -> Option<js_sys::Uint8Array>;
123//! }
124//! ```
125//!
126//! # Conditional methods
127//! Methods can be gated with `#[cfg(...)]` or `#[cfg_attr(...)]`. The macro propagates these
128//! attributes to every generated artifact for that method, so rustc strips them in lockstep.
129//! ```rust
130//! #[web_rpc::service]
131//! pub trait Conditional {
132//! fn always_on(&self, x: u32) -> u32;
133//! #[cfg(feature = "extra")]
134//! fn extra(&self, s: &str) -> String;
135//! }
136//! ```
137//! Bincode encodes enum variants by their positional discriminant, so the set of methods
138//! that survive cfg evaluation must match on both ends of a channel. If one side has a gated
139//! method enabled and the other does not, the wire format will silently desync.
140//!
141//! # Bi-directional
142//! Both sides of a channel can be set up to act as both client and server at the same time. To
143//! do this, stack [`with_service`](Builder::with_service) and
144//! [`with_client`](Builder::with_client) on the same [`Builder`] before calling `build()`, which
145//! then returns a `(C, Server)` tuple instead of one or the other.
146//! ```rust,no_run
147//! # #[web_rpc::service]
148//! # pub trait Calculator { fn add(&self, l: u32, r: u32) -> u32; }
149//! # struct Calc;
150//! # impl Calculator for Calc { fn add(&self, l: u32, r: u32) -> u32 { l + r } }
151//! # async fn run() {
152//! # let channel = web_sys::MessageChannel::new().unwrap();
153//! # let (iface, _) = futures_util::future::join(
154//! # web_rpc::Interface::new(channel.port1()),
155//! # web_rpc::Interface::new(channel.port2()),
156//! # ).await;
157//! let (client, server) = web_rpc::Builder::new(iface)
158//! .with_service::<CalculatorService<_>>(Calc)
159//! .with_client::<CalculatorClient>()
160//! .build();
161//! # }
162//! ```
163
164use std::{
165 cell::RefCell,
166 marker::PhantomData,
167 pin::Pin,
168 rc::Rc,
169 task::{Context, Poll},
170};
171
172use futures_channel::mpsc;
173use futures_core::{future::LocalBoxFuture, Future};
174use futures_util::{FutureExt, StreamExt};
175use gloo_events::EventListener;
176use js_sys::{ArrayBuffer, Uint8Array};
177use serde::{de::DeserializeOwned, Deserialize, Serialize};
178use wasm_bindgen::JsCast;
179
180#[doc(hidden)]
181pub use bincode;
182#[doc(hidden)]
183pub use futures_channel;
184#[doc(hidden)]
185pub use futures_core;
186#[doc(hidden)]
187pub use futures_util;
188#[doc(hidden)]
189pub use gloo_events;
190#[doc(hidden)]
191pub use js_sys;
192#[doc(hidden)]
193pub use pin_utils;
194#[doc(hidden)]
195pub use serde;
196#[doc(hidden)]
197pub use wasm_bindgen;
198
199pub use web_rpc_macro::service;
200
201pub mod client;
202#[doc(hidden)]
203pub mod codec;
204pub mod interface;
205pub mod port;
206#[doc(hidden)]
207pub mod service;
208
209pub use interface::Interface;
210
211#[doc(hidden)]
212#[derive(Serialize, Deserialize)]
213pub enum MessageHeader {
214 Request(usize),
215 Abort(usize),
216 Response(usize),
217 StreamItem(usize),
218 StreamEnd(usize),
219}
220
221/// This struct allows one to configure the RPC interface prior to creating it.
222/// To get an instance of this struct, call [`Builder<C, S>::new`] with
223/// an [`Interface`].
224pub struct Builder<C, S> {
225 client: PhantomData<C>,
226 service: S,
227 interface: Interface,
228}
229
230impl Builder<(), ()> {
231 /// Create a new builder from an [`Interface`]
232 pub fn new(interface: Interface) -> Self {
233 Self {
234 interface,
235 client: PhantomData::<()>,
236 service: (),
237 }
238 }
239}
240
241impl<C> Builder<C, ()> {
242 /// Configure the RPC interface with a service that implements methods
243 /// that can be called from the other side of the channel. To use this method,
244 /// you need to specify the type `S` which is the service type generated by the
245 /// attribute macro [`macro@service`]. The implementation parameter is then an
246 /// instance of something that implements the trait to which to applied the
247 /// [`macro@service`] macro. For example, if you have a trait `Calculator` to
248 /// which you have applied [`macro@service`], you would use this method as follows:
249 /// ```rust,no_run
250 /// # #[web_rpc::service]
251 /// # pub trait Calculator {
252 /// # fn add(&self, left: u32, right: u32) -> u32;
253 /// # }
254 /// # struct CalculatorServiceImpl;
255 /// # impl Calculator for CalculatorServiceImpl {
256 /// # fn add(&self, left: u32, right: u32) -> u32 { left + right }
257 /// # }
258 /// # fn example(some_interface: web_rpc::Interface) {
259 /// let server = web_rpc::Builder::new(some_interface)
260 /// .with_service::<CalculatorService<_>>(CalculatorServiceImpl)
261 /// .build();
262 /// # }
263 /// ```
264 pub fn with_service<S: service::Service>(self, implementation: impl Into<S>) -> Builder<C, S> {
265 let service = implementation.into();
266 let Builder {
267 interface, client, ..
268 } = self;
269 Builder {
270 interface,
271 client,
272 service,
273 }
274 }
275}
276
277impl<S> Builder<(), S> {
278 /// Configure the RPC interface with a client that allows you to execute RPCs on the
279 /// server. The builder will automatically instansiate the client for you, you just
280 /// need to provide the type which is generated via the [`macro@service`] attribute
281 /// macro. For example, if you had a trait `Calculator` to which you applied the
282 /// [`macro@service`] attribute macro, the macro would have generated a `CalculatorClient`
283 /// struct which you can use as the `C` in this function.
284 pub fn with_client<C: client::Client>(self) -> Builder<C, S> {
285 let Builder {
286 interface, service, ..
287 } = self;
288 Builder {
289 interface,
290 client: PhantomData::<C>,
291 service,
292 }
293 }
294}
295
296/// `Server` is the server that is returned from the [`Builder::build`] method given
297/// you configured the RPC interface with a service. Note that `Server` implements future and needs
298/// to be polled in order to execute and respond to inbound RPC requests.
299#[must_use = "Server must be polled in order for RPC requests to be executed"]
300pub struct Server {
301 _listener: Rc<EventListener>,
302 task: LocalBoxFuture<'static, ()>,
303}
304
305impl Future for Server {
306 type Output = ();
307
308 fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
309 self.task.poll_unpin(cx)
310 }
311}
312
313impl<C> Builder<C, ()>
314where
315 C: client::Client + From<client::Configuration<C::Response>> + 'static,
316 <C as client::Client>::Response: DeserializeOwned,
317{
318 /// Build function for client-only RPC interfaces.
319 pub fn build(self) -> C {
320 let Builder {
321 interface:
322 Interface {
323 port,
324 listener,
325 mut messages_rx,
326 },
327 ..
328 } = self;
329 let client_callback_map: Rc<RefCell<client::CallbackMap<C::Response>>> = Default::default();
330 let client_callback_map_cloned = client_callback_map.clone();
331 let stream_callback_map: Rc<RefCell<client::StreamCallbackMap<C::Response>>> =
332 Default::default();
333 let stream_callback_map_cloned = stream_callback_map.clone();
334 let dispatcher = async move {
335 while let Some(array) = messages_rx.next().await {
336 let header_bytes =
337 Uint8Array::new(&array.shift().dyn_into::<ArrayBuffer>().unwrap()).to_vec();
338 let header: MessageHeader = bincode::deserialize(&header_bytes).unwrap();
339 match header {
340 MessageHeader::Response(seq_id) => {
341 let payload_bytes =
342 Uint8Array::new(&array.shift().dyn_into::<ArrayBuffer>().unwrap())
343 .to_vec();
344 let response: C::Response = bincode::deserialize(&payload_bytes).unwrap();
345 if let Some(callback_tx) =
346 client_callback_map_cloned.borrow_mut().remove(&seq_id)
347 {
348 let _ = callback_tx.send((response, array));
349 }
350 }
351 MessageHeader::StreamItem(seq_id) => {
352 let payload_bytes =
353 Uint8Array::new(&array.shift().dyn_into::<ArrayBuffer>().unwrap())
354 .to_vec();
355 let response: C::Response = bincode::deserialize(&payload_bytes).unwrap();
356 if let Some(tx) = stream_callback_map_cloned.borrow().get(&seq_id) {
357 let _ = tx.unbounded_send((response, array));
358 }
359 }
360 MessageHeader::StreamEnd(seq_id) => {
361 stream_callback_map_cloned.borrow_mut().remove(&seq_id);
362 }
363 _ => panic!("client received a server message"),
364 }
365 }
366 }
367 .boxed_local()
368 .shared();
369 let port_cloned = port.clone();
370 let abort_sender = move |seq_id: usize| {
371 let header = MessageHeader::Abort(seq_id);
372 let header_bytes = bincode::serialize(&header).unwrap();
373 let buffer = js_sys::Uint8Array::from(&header_bytes[..]).buffer();
374 let post_args = js_sys::Array::of1(&buffer);
375 let transfer_args = js_sys::Array::of1(&buffer);
376 port_cloned
377 .post_message(&post_args, &transfer_args)
378 .unwrap();
379 };
380 C::from((
381 client_callback_map,
382 stream_callback_map,
383 port,
384 Rc::new(listener),
385 dispatcher,
386 Rc::new(abort_sender),
387 ))
388 }
389}
390
391impl<S> Builder<(), S>
392where
393 S: service::Service + 'static,
394 <S as service::Service>::Response: Serialize,
395{
396 /// Build function for server-only RPC interfaces.
397 pub fn build(self) -> Server {
398 let Builder {
399 service,
400 interface:
401 Interface {
402 port,
403 listener,
404 mut messages_rx,
405 },
406 ..
407 } = self;
408 let (server_requests_tx, server_requests_rx) = mpsc::unbounded();
409 let (abort_requests_tx, abort_requests_rx) = mpsc::unbounded();
410 let dispatcher = async move {
411 while let Some(array) = messages_rx.next().await {
412 let header_bytes =
413 Uint8Array::new(&array.shift().dyn_into::<ArrayBuffer>().unwrap()).to_vec();
414 let header: MessageHeader = bincode::deserialize(&header_bytes).unwrap();
415 match header {
416 MessageHeader::Request(seq_id) => {
417 let payload =
418 Uint8Array::new(&array.shift().dyn_into::<ArrayBuffer>().unwrap())
419 .to_vec();
420 server_requests_tx
421 .unbounded_send((seq_id, payload, array))
422 .unwrap();
423 }
424 MessageHeader::Abort(seq_id) => {
425 abort_requests_tx.unbounded_send(seq_id).unwrap();
426 }
427 _ => panic!("server received a client message"),
428 }
429 }
430 }
431 .boxed_local()
432 .shared();
433 Server {
434 _listener: Rc::new(listener),
435 task: service::task::<S>(
436 service,
437 port,
438 dispatcher,
439 server_requests_rx,
440 abort_requests_rx,
441 )
442 .boxed_local(),
443 }
444 }
445}
446
447impl<C, S> Builder<C, S>
448where
449 C: client::Client + From<client::Configuration<C::Response>> + 'static,
450 S: service::Service + 'static,
451 <S as service::Service>::Response: Serialize,
452 <C as client::Client>::Response: DeserializeOwned,
453{
454 /// Build function for client-server RPC interfaces.
455 pub fn build(self) -> (C, Server) {
456 let Builder {
457 service: server,
458 interface:
459 Interface {
460 port,
461 listener,
462 mut messages_rx,
463 },
464 ..
465 } = self;
466 let client_callback_map: Rc<RefCell<client::CallbackMap<C::Response>>> = Default::default();
467 let stream_callback_map: Rc<RefCell<client::StreamCallbackMap<C::Response>>> =
468 Default::default();
469 let (server_requests_tx, server_requests_rx) = mpsc::unbounded();
470 let (abort_requests_tx, abort_requests_rx) = mpsc::unbounded();
471 let client_callback_map_cloned = client_callback_map.clone();
472 let stream_callback_map_cloned = stream_callback_map.clone();
473 let dispatcher = async move {
474 while let Some(array) = messages_rx.next().await {
475 let header_bytes =
476 Uint8Array::new(&array.shift().dyn_into::<ArrayBuffer>().unwrap()).to_vec();
477 let header: MessageHeader = bincode::deserialize(&header_bytes).unwrap();
478 match header {
479 MessageHeader::Response(seq_id) => {
480 let payload_bytes =
481 Uint8Array::new(&array.shift().dyn_into::<ArrayBuffer>().unwrap())
482 .to_vec();
483 let response: C::Response = bincode::deserialize(&payload_bytes).unwrap();
484 if let Some(callback_tx) =
485 client_callback_map_cloned.borrow_mut().remove(&seq_id)
486 {
487 let _ = callback_tx.send((response, array));
488 }
489 }
490 MessageHeader::StreamItem(seq_id) => {
491 let payload_bytes =
492 Uint8Array::new(&array.shift().dyn_into::<ArrayBuffer>().unwrap())
493 .to_vec();
494 let response: C::Response = bincode::deserialize(&payload_bytes).unwrap();
495 if let Some(tx) = stream_callback_map_cloned.borrow().get(&seq_id) {
496 let _ = tx.unbounded_send((response, array));
497 }
498 }
499 MessageHeader::StreamEnd(seq_id) => {
500 stream_callback_map_cloned.borrow_mut().remove(&seq_id);
501 }
502 MessageHeader::Request(seq_id) => {
503 let payload =
504 Uint8Array::new(&array.shift().dyn_into::<ArrayBuffer>().unwrap())
505 .to_vec();
506 server_requests_tx
507 .unbounded_send((seq_id, payload, array))
508 .unwrap();
509 }
510 MessageHeader::Abort(seq_id) => {
511 abort_requests_tx.unbounded_send(seq_id).unwrap();
512 }
513 }
514 }
515 }
516 .boxed_local()
517 .shared();
518 let port_cloned = port.clone();
519 let abort_sender = move |seq_id: usize| {
520 let header = MessageHeader::Abort(seq_id);
521 let header_bytes = bincode::serialize(&header).unwrap();
522 let buffer = js_sys::Uint8Array::from(&header_bytes[..]).buffer();
523 let post_args = js_sys::Array::of1(&buffer);
524 let transfer_args = js_sys::Array::of1(&buffer);
525 port_cloned
526 .post_message(&post_args, &transfer_args)
527 .unwrap();
528 };
529 let listener = Rc::new(listener);
530 let client = C::from((
531 client_callback_map,
532 stream_callback_map,
533 port.clone(),
534 listener.clone(),
535 dispatcher.clone(),
536 Rc::new(abort_sender),
537 ));
538 let server = Server {
539 _listener: listener,
540 task: service::task::<S>(
541 server,
542 port,
543 dispatcher,
544 server_requests_rx,
545 abort_requests_rx,
546 )
547 .boxed_local(),
548 };
549 (client, server)
550 }
551}