wrpc_transport/frame/conn/
client.rs

1use core::marker::PhantomData;
2
3use anyhow::Context as _;
4use bytes::{BufMut as _, BytesMut};
5use tokio::io::{AsyncRead, AsyncWrite, AsyncWriteExt as _};
6use tokio_util::codec::Encoder;
7use tracing::{instrument, trace};
8use wasm_tokio::{CoreNameEncoder, CoreVecEncoderBytes};
9
10use crate::frame::conn::{Incoming, Outgoing};
11use crate::frame::{ConnHandler, PROTOCOL};
12
13/// Encodes an invocation to a `BytesMut`
14///
15/// This is low-level API, most users should use [`invoke`].
16#[instrument(level = "trace", skip_all)]
17pub fn encode_invocation(
18    buf: &mut BytesMut,
19    instance: &str,
20    func: &str,
21    params: &[u8],
22) -> std::io::Result<()> {
23    buf.reserve(
24        17_usize // len(PROTOCOL) + len(instance) + len(func) + len([]) + len(params)
25            .saturating_add(instance.len())
26            .saturating_add(func.len())
27            .saturating_add(params.len()),
28    );
29    buf.put_u8(PROTOCOL);
30    CoreNameEncoder.encode(instance, buf)?;
31    CoreNameEncoder.encode(func, buf)?;
32    buf.put_u8(0);
33    CoreVecEncoderBytes.encode(params, buf)?;
34    Ok(())
35}
36
37/// Defines invocation behavior
38#[derive(Clone)]
39pub struct InvokeBuilder<H = ()>(PhantomData<H>)
40where
41    H: ?Sized;
42
43impl<H> InvokeBuilder<H> {
44    /// Invoke function `func` on instance `instance`
45    #[instrument(level = "trace", skip_all)]
46    pub async fn invoke<P, I, O>(
47        self,
48        mut tx: O,
49        rx: I,
50        instance: &str,
51        func: &str,
52        params: impl AsRef<[u8]>,
53        paths: impl AsRef<[P]> + Send,
54    ) -> anyhow::Result<(Outgoing, Incoming)>
55    where
56        P: AsRef<[Option<usize>]> + Send + Sync,
57        I: AsyncRead + Unpin + Send + 'static,
58        O: AsyncWrite + Unpin + Send + 'static,
59        H: ConnHandler<I, O>,
60    {
61        let mut buf = BytesMut::default();
62        encode_invocation(&mut buf, instance, func, params.as_ref())
63            .context("failed to encode invocation")?;
64        trace!(?buf, "writing invocation");
65        tx.write_all(&buf)
66            .await
67            .context("failed to initialize connection")?;
68        tx.flush().await.context("failed to flush invocation")?;
69
70        let tx = Outgoing::new(tx, |tx, res| H::on_egress(tx, res));
71        let rx = Incoming::new(rx, paths.as_ref(), |rx, res| H::on_ingress(rx, res));
72        Ok((tx, rx))
73    }
74}
75
76impl<H> Default for InvokeBuilder<H> {
77    fn default() -> Self {
78        Self(PhantomData)
79    }
80}
81
82/// Invoke function `func` on instance `instance`
83#[instrument(level = "trace", skip_all)]
84pub async fn invoke<P, I, O>(
85    tx: O,
86    rx: I,
87    instance: &str,
88    func: &str,
89    params: impl AsRef<[u8]>,
90    paths: impl AsRef<[P]> + Send,
91) -> anyhow::Result<(Outgoing, Incoming)>
92where
93    P: AsRef<[Option<usize>]> + Send + Sync,
94    I: AsyncRead + Unpin + Send + 'static,
95    O: AsyncWrite + Unpin + Send + 'static,
96{
97    InvokeBuilder::<()>::default()
98        .invoke(tx, rx, instance, func, params, paths)
99        .await
100}
wrpc_transport/frame/conn/client.rs

wrpc_transport/frame/conn/
client.rs
