11use std:: {
22 collections:: VecDeque ,
33 path:: Path ,
4+ pin:: Pin ,
45 sync:: {
56 Arc , Mutex ,
67 atomic:: { self , AtomicUsize } ,
78 } ,
9+ task:: { Context , Poll } ,
810} ;
911
1012use bstr:: ByteSlice ;
13+ use tokio:: io:: { self , AsyncWrite } ;
1114use wasmtime_wasi:: {
12- DirPerms , FilePerms , OutputStream , Pollable , StdoutStream , StreamError , WasiCtxBuilder ,
15+ DirPerms , FilePerms , WasiCtxBuilder ,
16+ cli:: { IsTerminal , StdoutStream } ,
17+ p2:: { OutputStream , Pollable , StreamError } ,
1318 preview1:: WasiP1Ctx ,
1419} ;
1520
@@ -28,11 +33,17 @@ struct Buf {
2833}
2934
3035impl StdoutStream for StdErr {
31- fn stream ( & self ) -> Box < dyn OutputStream > {
36+ fn async_stream ( & self ) -> Box < dyn AsyncWrite + Send + Sync > {
3237 Box :: new ( self . clone ( ) )
3338 }
3439
35- fn isatty ( & self ) -> bool {
40+ fn p2_stream ( & self ) -> Box < dyn OutputStream > {
41+ Box :: new ( self . clone ( ) )
42+ }
43+ }
44+
45+ impl IsTerminal for StdErr {
46+ fn is_terminal ( & self ) -> bool {
3647 false
3748 }
3849}
@@ -61,10 +72,8 @@ impl StdErr {
6172 timer. log_auto_splitter ( format_args ! ( "{}" , to_print. trim( ) . as_bstr( ) ) ) ;
6273 buf. drain ( ..flush_idx) ;
6374 }
64- }
6575
66- impl OutputStream for StdErr {
67- fn write ( & mut self , bytes : bytes:: Bytes ) -> Result < ( ) , StreamError > {
76+ fn write ( & mut self , bytes : & [ u8 ] ) -> Result < ( ) , StreamError > {
6877 let buffer = & mut * self . buffer . buf . lock ( ) . unwrap ( ) ;
6978 if bytes. len ( ) > ERR_CAPACITY - buffer. len ( ) {
7079 return Err ( StreamError :: Trap ( anyhow:: format_err!(
@@ -77,9 +86,15 @@ impl OutputStream for StdErr {
7786 atomic:: Ordering :: Relaxed ,
7887 ) ;
7988
80- buffer. extend ( bytes. as_ref ( ) ) ;
89+ buffer. extend ( bytes) ;
8190 Ok ( ( ) )
8291 }
92+ }
93+
94+ impl OutputStream for StdErr {
95+ fn write ( & mut self , bytes : bytes:: Bytes ) -> Result < ( ) , StreamError > {
96+ self . write ( & bytes)
97+ }
8398
8499 fn flush ( & mut self ) -> Result < ( ) , StreamError > {
85100 let len = self . buffer . buf . lock ( ) . unwrap ( ) . len ( ) ;
@@ -93,6 +108,25 @@ impl OutputStream for StdErr {
93108 }
94109}
95110
111+ impl AsyncWrite for StdErr {
112+ fn poll_write (
113+ mut self : Pin < & mut Self > ,
114+ _cx : & mut Context ,
115+ buf : & [ u8 ] ,
116+ ) -> Poll < io:: Result < usize > > {
117+ match self . as_mut ( ) . write ( buf) {
118+ Ok ( ( ) ) => Poll :: Ready ( Ok ( buf. len ( ) ) ) ,
119+ Err ( e) => Poll :: Ready ( Err ( io:: Error :: other ( e) ) ) ,
120+ }
121+ }
122+ fn poll_flush ( mut self : Pin < & mut Self > , _cx : & mut Context ) -> Poll < io:: Result < ( ) > > {
123+ Poll :: Ready ( self . as_mut ( ) . flush ( ) . map_err ( io:: Error :: other) )
124+ }
125+ fn poll_shutdown ( self : Pin < & mut Self > , _cx : & mut Context ) -> Poll < io:: Result < ( ) > > {
126+ Poll :: Ready ( Ok ( ( ) ) )
127+ }
128+ }
129+
96130#[ async_trait:: async_trait]
97131impl Pollable for StdErr {
98132 async fn ready ( & mut self ) { }
0 commit comments