nandi.uk
Distributed Erlang over iroh P2P + TLS, with a bot powers SDK
1//! Clonable duplex byte streams that implement `futures_io` traits.
2//!
3//! `erl_dist` requires `AsyncRead + AsyncWrite + Unpin + Clone` on the carrier
4//! (smol's `TcpStream` is clonable; iroh bi-streams and tokio-rustls streams are
5//! not). We wrap any split read/write half in a shared duplex.
6//!
7//! Layout: `async_dup::Arc<async_dup::Mutex<Combined>>` — Arc is `Clone`, and
8//! both Arc and Mutex forward futures-io traits when the inner type does.
9
10use async_dup::{Arc as DupArc, Mutex as DupMutex};
11use futures::io::{AsyncRead, AsyncWrite};
12use std::io;
13use std::pin::Pin;
14use std::task::{Context, Poll};
15use tokio::io::{AsyncRead as TokioRead, AsyncWrite as TokioWrite, ReadBuf};
16
17/// A clonable futures-io duplex built from independent tokio read/write halves.
18#[derive(Clone)]
19pub struct DistStream {
20 inner: DupArc<DupMutex<Combined>>,
21}
22
23/// Type-erased tokio read half.
24type DynRead = Box<dyn TokioRead + Unpin + Send>;
25/// Type-erased tokio write half.
26type DynWrite = Box<dyn TokioWrite + Unpin + Send>;
27
28struct Combined {
29 read: DynRead,
30 write: DynWrite,
31}
32
33impl DistStream {
34 /// Wrap independent tokio async read/write halves.
35 pub fn from_split<R, W>(read: R, write: W) -> Self
36 where
37 R: TokioRead + Unpin + Send + 'static,
38 W: TokioWrite + Unpin + Send + 'static,
39 {
40 Self {
41 inner: DupArc::new(DupMutex::new(Combined {
42 read: Box::new(read),
43 write: Box::new(write),
44 })),
45 }
46 }
47
48 /// Wrap a single tokio type that is both `AsyncRead` and `AsyncWrite`.
49 pub fn from_rw<T>(stream: T) -> Self
50 where
51 T: TokioRead + TokioWrite + Unpin + Send + 'static,
52 {
53 let (read, write) = tokio::io::split(stream);
54 Self::from_split(read, write)
55 }
56}
57
58impl TokioRead for Combined {
59 fn poll_read(
60 mut self: Pin<&mut Self>,
61 cx: &mut Context<'_>,
62 buf: &mut ReadBuf<'_>,
63 ) -> Poll<io::Result<()>> {
64 Pin::new(&mut self.read).poll_read(cx, buf)
65 }
66}
67
68impl TokioWrite for Combined {
69 fn poll_write(
70 mut self: Pin<&mut Self>,
71 cx: &mut Context<'_>,
72 buf: &[u8],
73 ) -> Poll<io::Result<usize>> {
74 Pin::new(&mut self.write).poll_write(cx, buf)
75 }
76
77 fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
78 Pin::new(&mut self.write).poll_flush(cx)
79 }
80
81 fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
82 Pin::new(&mut self.write).poll_shutdown(cx)
83 }
84}
85
86// Bridge tokio traits → futures_io for erl_dist.
87impl AsyncRead for Combined {
88 fn poll_read(
89 self: Pin<&mut Self>,
90 cx: &mut Context<'_>,
91 buf: &mut [u8],
92 ) -> Poll<io::Result<usize>> {
93 let mut read_buf = ReadBuf::new(buf);
94 match TokioRead::poll_read(self, cx, &mut read_buf) {
95 Poll::Ready(Ok(())) => Poll::Ready(Ok(read_buf.filled().len())),
96 Poll::Ready(Err(e)) => Poll::Ready(Err(e)),
97 Poll::Pending => Poll::Pending,
98 }
99 }
100}
101
102impl AsyncWrite for Combined {
103 fn poll_write(
104 self: Pin<&mut Self>,
105 cx: &mut Context<'_>,
106 buf: &[u8],
107 ) -> Poll<io::Result<usize>> {
108 TokioWrite::poll_write(self, cx, buf)
109 }
110
111 fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
112 TokioWrite::poll_flush(self, cx)
113 }
114
115 fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
116 TokioWrite::poll_shutdown(self, cx)
117 }
118}
119
120impl AsyncRead for DistStream {
121 fn poll_read(
122 mut self: Pin<&mut Self>,
123 cx: &mut Context<'_>,
124 buf: &mut [u8],
125 ) -> Poll<io::Result<usize>> {
126 Pin::new(&mut self.inner).poll_read(cx, buf)
127 }
128}
129
130impl AsyncWrite for DistStream {
131 fn poll_write(
132 mut self: Pin<&mut Self>,
133 cx: &mut Context<'_>,
134 buf: &[u8],
135 ) -> Poll<io::Result<usize>> {
136 Pin::new(&mut self.inner).poll_write(cx, buf)
137 }
138
139 fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
140 Pin::new(&mut self.inner).poll_flush(cx)
141 }
142
143 fn poll_close(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
144 Pin::new(&mut self.inner).poll_close(cx)
145 }
146}
147
148impl std::fmt::Debug for DistStream {
149 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
150 f.debug_struct("DistStream").finish_non_exhaustive()
151 }
152}