feat(core): streams (#12596)

This allows resources to be "streams" by implementing read/write/shutdown. These streams are implicit since their nature (read/write/duplex) isn't known until called, but we could easily add another method to explicitly tag resources as streams.

`op_read/op_write/op_shutdown` are now builtin ops provided by `deno_core`

Note: this current implementation is simple & straightforward but it results in an additional alloc per read/write call

Closes #12556

1 files changed, 23 insertions, 0 deletions

diff --git a/core/resources.rs b/core/resources.rs
index c5e6684a4..33cabcad4 100644
--- a/core/resources.rs
+++ b/core/resources.rs
@@ -7,17 +7,25 @@
 // file descriptor (hence the different name).
 
 use crate::error::bad_resource_id;
+use crate::error::not_supported;
 use crate::error::AnyError;
+use crate::ZeroCopyBuf;
+use futures::Future;
 use std::any::type_name;
 use std::any::Any;
 use std::any::TypeId;
 use std::borrow::Cow;
 use std::collections::BTreeMap;
 use std::iter::Iterator;
+use std::pin::Pin;
 use std::rc::Rc;
 
+/// Returned by resource read/write/shutdown methods
+pub type AsyncResult<T> = Pin<Box<dyn Future<Output = Result<T, AnyError>>>>;
+
 /// All objects that can be store in the resource table should implement the
 /// `Resource` trait.
+/// TODO(@AaronO): investigate avoiding alloc on read/write/shutdown
 pub trait Resource: Any + 'static {
   /// Returns a string representation of the resource which is made available
   /// to JavaScript code through `op_resources`. The default implementation
@@ -27,6 +35,21 @@ pub trait Resource: Any + 'static {
     type_name::<Self>().into()
   }
 
+  /// Resources may implement `read()` to be a readable stream
+  fn read(self: Rc<Self>, _buf: ZeroCopyBuf) -> AsyncResult<usize> {
+    Box::pin(futures::future::err(not_supported()))
+  }
+
+  /// Resources may implement `write()` to be a writable stream
+  fn write(self: Rc<Self>, _buf: ZeroCopyBuf) -> AsyncResult<usize> {
+    Box::pin(futures::future::err(not_supported()))
+  }
+
+  /// Resources may implement `shutdown()` for graceful async shutdowns
+  fn shutdown(self: Rc<Self>) -> AsyncResult<()> {
+    Box::pin(futures::future::err(not_supported()))
+  }
+
   /// Resources may implement the `close()` trait method if they need to do
   /// resource specific clean-ups, such as cancelling pending futures, after a
   /// resource has been removed from the resource table.