wrpc_transport/
lib.rs

1#![allow(clippy::type_complexity)]
2#![deny(missing_docs)]
3
4//! wRPC transport abstractions, codec and framing
5//!
6//! wRPC is an RPC framework based on [WIT](https://component-model.bytecodealliance.org/design/wit.html).
7//! It follows client-server model, where peers (servers) may serve function and method calls invoked by the other peers (clients).
8//!
9//! The two main abstractions on top of which wRPC is built are:
10//! - [Invoke] - the client-side handle to a wRPC transport, allowing clients to *invoke* WIT functions over wRPC transport
11//! - [Serve] - the server-side handle to a wRPC transport, allowing servers to *serve* WIT functions over wRPC transport
12//!
13//! [Invoke] and [Serve] establish a single bidirectional byte stream per invocation, over which
14//! wRPC layers its [framing](frame) protocol to multiplex the nested async parameter and result
15//! sub-streams. Both therefore yield the framed [`frame::Outgoing`] and [`frame::Incoming`]
16//! streams regardless of the underlying transport, which only needs to provide any
17//! [`AsyncWrite`](tokio::io::AsyncWrite)/[`AsyncBufRead`](tokio::io::AsyncBufRead) byte stream.
18
19pub mod frame;
20pub mod invoke;
21pub mod serve;
22
23mod pack;
24mod value;
25
26pub use frame::{Decoder as FrameDecoder, Encoder as FrameEncoder, Frame, FrameRef, Server};
27pub use invoke::{Invoke, InvokeExt};
28pub use pack::{Packer, pack, pack_ready, unpack};
29pub use serve::{Serve, ServeExt};
30pub use value::*;
31
32#[cfg(any(all(target_family = "wasm", feature = "wasi"), feature = "net"))]
33pub use frame::tcp;
34#[cfg(all(unix, feature = "net"))]
35pub use frame::unix;
36
37use core::mem;
38use core::pin::{Pin, pin};
39use core::task::{Context, Poll};
40
41use bytes::{BufMut as _, BytesMut};
42use tokio::io::{AsyncRead, ReadBuf};
43use tokio_util::codec::Encoder as _;
44use tracing::trace;
45use wasm_tokio::Leb128Encoder;
46
47/// Buffered incoming stream used for decoding values
48///
49/// This wraps the multiplexed framed [`frame::Incoming`] stream with a read buffer used to
50/// hold bytes that were read ahead while decoding synchronous values.
51pub struct BufferedIncoming {
52    buffer: BytesMut,
53    inner: frame::Incoming,
54}
55
56impl BufferedIncoming {
57    /// Index the incoming stream using a structural `path`, returning a handle to the
58    /// multiplexed sub-stream addressed by it.
59    pub fn index(&self, path: &[usize]) -> anyhow::Result<Self> {
60        let inner = self.inner.index(path)?;
61        Ok(Self {
62            buffer: BytesMut::default(),
63            inner,
64        })
65    }
66}
67
68impl Drop for BufferedIncoming {
69    fn drop(&mut self) {
70        if !self.buffer.is_empty() {
71            self.inner.mark_unprocessed();
72        }
73    }
74}
75
76impl AsyncRead for BufferedIncoming {
77    fn poll_read(
78        mut self: Pin<&mut Self>,
79        cx: &mut Context<'_>,
80        buf: &mut ReadBuf<'_>,
81    ) -> Poll<std::io::Result<()>> {
82        let cap = buf.remaining();
83        if cap == 0 {
84            trace!("attempt to read empty buffer");
85            return Poll::Ready(Ok(()));
86        }
87        if !self.buffer.is_empty() {
88            if self.buffer.len() > cap {
89                trace!(cap, len = self.buffer.len(), "reading part of buffer");
90                buf.put_slice(&self.buffer.split_to(cap));
91            } else {
92                trace!(cap, len = self.buffer.len(), "reading full buffer");
93                buf.put_slice(&mem::take(&mut self.buffer));
94            }
95            return Poll::Ready(Ok(()));
96        }
97        pin!(&mut self.inner).poll_read(cx, buf)
98    }
99}
100
101fn encode_ready_bytes_header(n: usize, dst: &mut BytesMut) -> std::io::Result<()> {
102    let n = u32::try_from(n)
103        .map_err(|err| std::io::Error::new(std::io::ErrorKind::InvalidInput, err))?;
104    dst.reserve(1);
105    dst.put_u8(0x01);
106    Leb128Encoder.encode(n, dst)
107}
108
109fn copy_io_error(err: &std::io::Error) -> std::io::Error {
110    std::io::Error::new(err.kind(), err.to_string())
111}
112
113/// Returns the error for data received for a closed receiver
114fn error_receiver_closed() -> std::io::Error {
115    std::io::Error::new(std::io::ErrorKind::BrokenPipe, "stream receiver closed")
116}
wrpc_transport/lib.rs

wrpc_transport/
lib.rs
