wasmrs_runtime/
runtime.rs

1#[cfg(target_family = "wasm")]
2mod wasm;
3use std::task::{Context, Poll};
4
5use futures::{Future, FutureExt};
6#[cfg(target_family = "wasm")]
7pub use wasm::*;
8
9#[cfg(not(target_family = "wasm"))]
10mod native;
11#[cfg(not(target_family = "wasm"))]
12pub use native::*;
13
14use crate::Error;
15
16#[must_use]
17/// Create an unbounded channel.
18pub fn unbounded_channel<Item>() -> (UnboundedSender<Item>, UnboundedReceiver<Item>)
19where
20  Item: ConditionallySendSync,
21{
22  let (tx, rx) = tokio::sync::mpsc::unbounded_channel();
23
24  (UnboundedSender(tx), UnboundedReceiver(rx))
25}
26
27#[allow(missing_debug_implementations)]
28/// A Unbounded Sender that works the same way in single-threaded WebAssembly as multi-threaded native.
29pub struct UnboundedSender<Item>(tokio::sync::mpsc::UnboundedSender<Item>)
30where
31  Item: ConditionallySendSync;
32
33impl<Item> Clone for UnboundedSender<Item>
34where
35  Item: ConditionallySendSync,
36{
37  fn clone(&self) -> Self {
38    Self(self.0.clone())
39  }
40}
41
42impl<Item> UnboundedSender<Item>
43where
44  Item: ConditionallySendSync,
45{
46  /// Send an `Item` to the channel.
47  pub fn send(&self, message: Item) -> Result<(), Error> {
48    self.0.send(message).map_err(|_| Error::SendFailed(0))
49  }
50
51  #[must_use]
52  /// Check if the channel is closed.
53  pub fn is_closed(&self) -> bool {
54    self.0.is_closed()
55  }
56}
57
58#[allow(missing_debug_implementations)]
59/// A Unbounded Receiver that works the same way in single-threaded WebAssembly as multi-threaded native.
60pub struct UnboundedReceiver<Item>(tokio::sync::mpsc::UnboundedReceiver<Item>)
61where
62  Item: ConditionallySendSync;
63
64impl<Item> UnboundedReceiver<Item>
65where
66  Item: ConditionallySendSync,
67{
68  /// Receive the next `Item` on the channel.
69  pub async fn recv(&mut self) -> Option<Item> {
70    self.0.recv().await
71  }
72
73  /// Poll the channel to see if an `Item` is ready.
74  pub fn poll_recv(&mut self, cx: &mut Context) -> Poll<Option<Item>> {
75    self.0.poll_recv(cx)
76  }
77}
78
79impl<T> futures::Stream for UnboundedReceiver<T>
80where
81  T: ConditionallySendSync,
82{
83  type Item = T;
84
85  fn poll_next(mut self: std::pin::Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
86    self.0.poll_recv(cx)
87  }
88}
89
90#[must_use]
91/// A oneshot channel similar to [tokio::sync::oneshot::channel] but works the same way in single-threaded WebAssembly as multi-threaded native.
92pub fn oneshot<Item>() -> (OneShotSender<Item>, OneShotReceiver<Item>)
93where
94  Item: ConditionallySendSync,
95{
96  let (tx, rx) = tokio::sync::oneshot::channel();
97
98  (OneShotSender(tx), OneShotReceiver(rx))
99}
100
101#[allow(missing_debug_implementations)]
102/// A Unbounded Sender that works the same way in single-threaded WebAssembly as multi-threaded native.
103pub struct OneShotSender<Item>(tokio::sync::oneshot::Sender<Item>)
104where
105  Item: ConditionallySendSync;
106
107impl<Item> OneShotSender<Item>
108where
109  Item: ConditionallySendSync,
110{
111  /// Send an item on the channel.
112  pub fn send(self, message: Item) -> Result<(), Error> {
113    self.0.send(message).map_err(|_| Error::SendFailed(0))
114  }
115
116  #[must_use]
117  /// Check if the channel is closed.
118  pub fn is_closed(&self) -> bool {
119    self.0.is_closed()
120  }
121}
122
123#[allow(missing_debug_implementations)]
124/// A OneShort Receiver that works the same way in single-threaded WebAssembly as multi-threaded native.
125pub struct OneShotReceiver<Item>(tokio::sync::oneshot::Receiver<Item>)
126where
127  Item: ConditionallySendSync;
128
129impl<Item> OneShotReceiver<Item>
130where
131  Item: ConditionallySendSync,
132{
133  /// Receive the next item on the channel.
134  pub async fn recv(self) -> Result<Item, Error> {
135    self.0.await.map_err(|_e| Error::RecvFailed(80))
136  }
137}
138
139impl<Item> Future for OneShotReceiver<Item>
140where
141  Item: ConditionallySendSync,
142{
143  type Output = Result<Item, Error>;
144
145  fn poll(self: std::pin::Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
146    let poll = self.get_mut().0.poll_unpin(cx);
147    poll.map_err(|_e| Error::RecvFailed(95))
148  }
149}
150
151impl<T> std::fmt::Debug for MutRc<T>
152where
153  T: std::fmt::Debug,
154{
155  fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
156    f.debug_tuple("MutRc").field(&self.0).finish()
157  }
158}
159
160impl<T> Clone for MutRc<T> {
161  fn clone(&self) -> Self {
162    Self(self.0.clone())
163  }
164}
165
166#[cfg(test)]
167mod test {
168  use super::*;
169  use anyhow::Result;
170
171  #[test]
172  fn test_rc() -> Result<()> {
173    let one = RtRc::new("Hello World".to_owned());
174    let two = RtRc::new("Hello World".to_owned());
175
176    assert_eq!(one, two);
177    Ok(())
178  }
179}
wasmrs_runtime/runtime.rs

wasmrs_runtime/
runtime.rs
