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

2 files changed, 3 insertions, 4 deletions

diff --git a/cli/tests/unit/metrics_test.ts b/cli/tests/unit/metrics_test.ts
index 70efeac21..f020d69e8 100644
--- a/cli/tests/unit/metrics_test.ts
+++ b/cli/tests/unit/metrics_test.ts
@@ -16,7 +16,7 @@ unitTest(async function metrics() {
   assert(m1.bytesSentControl === 0);
   assert(m1.bytesSentData === 0);
   assert(m1.bytesReceived === 0);
-  const m1OpWrite = m1.ops["op_write_async"];
+  const m1OpWrite = m1.ops["op_write"];
   assert(m1OpWrite.opsDispatchedAsync > 0);
   assert(m1OpWrite.opsCompletedAsync > 0);
   assert(m1OpWrite.bytesSentControl === 0);
@@ -31,7 +31,7 @@ unitTest(async function metrics() {
   assert(m2.bytesSentControl === m1.bytesSentControl);
   assert(m2.bytesSentData === 0);
   assert(m2.bytesReceived === m1.bytesReceived);
-  const m2OpWrite = m2.ops["op_write_async"];
+  const m2OpWrite = m2.ops["op_write"];
   assert(m2OpWrite.opsDispatchedAsync > m1OpWrite.opsDispatchedAsync);
   assert(m2OpWrite.opsCompletedAsync > m1OpWrite.opsCompletedAsync);
   assert(m2OpWrite.bytesSentControl === m1OpWrite.bytesSentControl);
diff --git a/cli/tests/unit/opcall_test.ts b/cli/tests/unit/opcall_test.ts
index 63871cd4c..2cf086220 100644
--- a/cli/tests/unit/opcall_test.ts
+++ b/cli/tests/unit/opcall_test.ts
@@ -35,8 +35,7 @@ declare global {
 unitTest(async function opsAsyncBadResource() {
   try {
     const nonExistingRid = 9999;
-    await Deno.core.opAsync(
-      "op_read_async",
+    await Deno.core.read(
       nonExistingRid,
       new Uint8Array(0),
     );