wasmtime_wasi/p2/host/
io.rs

1use crate::p2::{
2    StreamError, StreamResult,
3    bindings::sync::io::poll::Pollable,
4    bindings::sync::io::streams::{self, InputStream, OutputStream},
5};
6use crate::runtime::in_tokio;
7use wasmtime::component::{Resource, ResourceTable};
8use wasmtime_wasi_io::bindings::wasi::io::streams::{
9    self as async_streams, Host as AsyncHost, HostInputStream as AsyncHostInputStream,
10    HostOutputStream as AsyncHostOutputStream,
11};
12
13impl From<async_streams::StreamError> for streams::StreamError {
14    fn from(other: async_streams::StreamError) -> Self {
15        match other {
16            async_streams::StreamError::LastOperationFailed(e) => Self::LastOperationFailed(e),
17            async_streams::StreamError::Closed => Self::Closed,
18        }
19    }
20}
21
22impl streams::Host for ResourceTable {
23    fn convert_stream_error(&mut self, err: StreamError) -> anyhow::Result<streams::StreamError> {
24        Ok(AsyncHost::convert_stream_error(self, err)?.into())
25    }
26}
27
28impl streams::HostOutputStream for ResourceTable {
29    fn drop(&mut self, stream: Resource<OutputStream>) -> anyhow::Result<()> {
30        in_tokio(async { AsyncHostOutputStream::drop(self, stream).await })
31    }
32
33    fn check_write(&mut self, stream: Resource<OutputStream>) -> StreamResult<u64> {
34        Ok(AsyncHostOutputStream::check_write(self, stream)?)
35    }
36
37    fn write(&mut self, stream: Resource<OutputStream>, bytes: Vec<u8>) -> StreamResult<()> {
38        Ok(AsyncHostOutputStream::write(self, stream, bytes)?)
39    }
40
41    fn blocking_write_and_flush(
42        &mut self,
43        stream: Resource<OutputStream>,
44        bytes: Vec<u8>,
45    ) -> StreamResult<()> {
46        in_tokio(async {
47            AsyncHostOutputStream::blocking_write_and_flush(self, stream, bytes).await
48        })
49    }
50
51    fn blocking_write_zeroes_and_flush(
52        &mut self,
53        stream: Resource<OutputStream>,
54        len: u64,
55    ) -> StreamResult<()> {
56        in_tokio(async {
57            AsyncHostOutputStream::blocking_write_zeroes_and_flush(self, stream, len).await
58        })
59    }
60
61    fn subscribe(&mut self, stream: Resource<OutputStream>) -> anyhow::Result<Resource<Pollable>> {
62        Ok(AsyncHostOutputStream::subscribe(self, stream)?)
63    }
64
65    fn write_zeroes(&mut self, stream: Resource<OutputStream>, len: u64) -> StreamResult<()> {
66        Ok(AsyncHostOutputStream::write_zeroes(self, stream, len)?)
67    }
68
69    fn flush(&mut self, stream: Resource<OutputStream>) -> StreamResult<()> {
70        Ok(AsyncHostOutputStream::flush(
71            self,
72            Resource::new_borrow(stream.rep()),
73        )?)
74    }
75
76    fn blocking_flush(&mut self, stream: Resource<OutputStream>) -> StreamResult<()> {
77        in_tokio(async {
78            AsyncHostOutputStream::blocking_flush(self, Resource::new_borrow(stream.rep())).await
79        })
80    }
81
82    fn splice(
83        &mut self,
84        dst: Resource<OutputStream>,
85        src: Resource<InputStream>,
86        len: u64,
87    ) -> StreamResult<u64> {
88        AsyncHostOutputStream::splice(self, dst, src, len)
89    }
90
91    fn blocking_splice(
92        &mut self,
93        dst: Resource<OutputStream>,
94        src: Resource<InputStream>,
95        len: u64,
96    ) -> StreamResult<u64> {
97        in_tokio(async { AsyncHostOutputStream::blocking_splice(self, dst, src, len).await })
98    }
99}
100
101impl streams::HostInputStream for ResourceTable {
102    fn drop(&mut self, stream: Resource<InputStream>) -> anyhow::Result<()> {
103        in_tokio(async { AsyncHostInputStream::drop(self, stream).await })
104    }
105
106    fn read(&mut self, stream: Resource<InputStream>, len: u64) -> StreamResult<Vec<u8>> {
107        AsyncHostInputStream::read(self, stream, len)
108    }
109
110    fn blocking_read(&mut self, stream: Resource<InputStream>, len: u64) -> StreamResult<Vec<u8>> {
111        in_tokio(async { AsyncHostInputStream::blocking_read(self, stream, len).await })
112    }
113
114    fn skip(&mut self, stream: Resource<InputStream>, len: u64) -> StreamResult<u64> {
115        AsyncHostInputStream::skip(self, stream, len)
116    }
117
118    fn blocking_skip(&mut self, stream: Resource<InputStream>, len: u64) -> StreamResult<u64> {
119        in_tokio(async { AsyncHostInputStream::blocking_skip(self, stream, len).await })
120    }
121
122    fn subscribe(&mut self, stream: Resource<InputStream>) -> anyhow::Result<Resource<Pollable>> {
123        AsyncHostInputStream::subscribe(self, stream)
124    }
125}
wasmtime_wasi/p2/host/io.rs

wasmtime_wasi/p2/host/
io.rs
