1 use anyhow::Result; 2 use wasmtime::component::{ 3 Accessor, AccessorTask, GuardedStreamWriter, Resource, StreamReader, StreamWriter, 4 }; 5 6 use super::Ctx; 7 8 pub mod bindings { 9 wasmtime::component::bindgen!({ 10 path: "wit", 11 world: "read-resource-stream", 12 with: { 13 "local:local/resource-stream/x": super::ResourceStreamX, 14 }, 15 imports: { 16 "local:local/resource-stream/foo": async | store | trappable, 17 default: trappable, 18 }, 19 }); 20 } 21 22 pub struct ResourceStreamX; 23 24 impl bindings::local::local::resource_stream::HostX for Ctx { 25 fn foo(&mut self, x: Resource<ResourceStreamX>) -> Result<()> { 26 self.table.get(&x)?; 27 Ok(()) 28 } 29 30 fn drop(&mut self, x: Resource<ResourceStreamX>) -> Result<()> { 31 self.table.delete(x)?; 32 Ok(()) 33 } 34 } 35 36 impl bindings::local::local::resource_stream::HostWithStore for Ctx { 37 async fn foo<T: 'static>( 38 accessor: &Accessor<T, Self>, 39 count: u32, 40 ) -> wasmtime::Result<StreamReader<Resource<ResourceStreamX>>> { 41 struct Task { 42 tx: StreamWriter<Resource<ResourceStreamX>>, 43 44 count: u32, 45 } 46 47 impl<T> AccessorTask<T, Ctx, Result<()>> for Task { 48 async fn run(self, accessor: &Accessor<T, Ctx>) -> Result<()> { 49 let mut tx = GuardedStreamWriter::new(accessor, self.tx); 50 for _ in 0..self.count { 51 let item = accessor.with(|mut view| view.get().table.push(ResourceStreamX))?; 52 tx.write_all(Some(item)).await; 53 } 54 Ok(()) 55 } 56 } 57 58 let (tx, rx) = accessor.with(|mut view| { 59 let instance = view.instance(); 60 instance.stream(&mut view) 61 })?; 62 accessor.spawn(Task { tx, count }); 63 Ok(rx) 64 } 65 } 66 67 impl bindings::local::local::resource_stream::Host for Ctx {} 68