👍 Use Streams API instead

denops_test/plugin.ts

@@ -1,20 +1,19 @@
-import { copy } from "https://deno.land/[email protected]/streams/copy.ts";
 import {
-  WorkerReader,
-  WorkerWriter,
-} from "https://deno.land/x/workerio@v1.4.4/mod.ts#^";
+  readableStreamFromWorker,
+  writableStreamFromWorker,
+} from "https://deno.land/x/workerio@v3.0.1/mod.ts#^";
 import { orElse } from "./or_else.ts";
 
 const worker = self as unknown as Worker;
-const reader = new WorkerReader(worker);
-const writer = new WorkerWriter(worker);
+const reader = readableStreamFromWorker(worker);
+const writer = writableStreamFromWorker(worker);
 
 const addr = JSON.parse(orElse(Deno.env.get("DENOPS_TEST_ADDRESS"), () => {
   throw new Error("Environment variable `DENOPS_TEST_ADDRESS` is required");
 }));
 const conn = await Deno.connect(addr);
 
 await Promise.race([
-  copy(conn, writer).finally(() => conn.close()),
-  copy(reader, conn),
+  reader.pipeTo(conn.writable),
+  conn.readable.pipeTo(writer),
 ]);

denops_test/with.ts

@@ -1,6 +1,8 @@
 import * as path from "https://deno.land/[email protected]/path/mod.ts";
-import { Session } from "https://deno.land/x/[email protected]/mod.ts#^";
-import { using } from "https://deno.land/x/[email protected]/mod.ts#^";
+import {
+  Client,
+  Session,
+} from "https://deno.land/x/[email protected]/mod.ts#^";
 import type {
   Denops,
   Meta,
@@ -78,39 +80,30 @@ export async function withDenops(
   });
   const conn = await listener.accept();
   try {
-    await using(
-      new Session(conn, conn, {}, {
-        errorCallback(e) {
-          if (e.name === "Interrupted") {
-            return;
-          }
-          console.error("Unexpected error occurred", e);
-        },
-      }),
-      async (session) => {
-        const meta = await session.call(
-          "call",
-          "denops#_internal#meta#get",
-        ) as Meta;
-        const denops = await newDenopsImpl(meta, session, pluginName);
-        // Workaround for unexpected "leaking async ops"
-        // https://github.com/denoland/deno/issues/15425#issuecomment-1368245954
-        await new Promise((resolve) => setTimeout(resolve, 0));
-        await main(denops);
-      },
-    );
+    const session = new Session(conn.readable, conn.writable);
+    await session.start(async (client) => {
+      const meta = await client.request(
+        "call",
+        "denops#_internal#meta#get",
+      ) as Meta;
+      const denops = await newDenopsImpl(meta, session, client, pluginName);
+      // Workaround for unexpected "leaking async ops"
+      // https://github.com/denoland/deno/issues/15425#issuecomment-1368245954
+      await new Promise((resolve) => setTimeout(resolve, 0));
+      await main(denops);
+    });
   } finally {
     proc.stdin.close();
     proc.kill();
     await proc.status;
-    conn.close();
     listener.close();
   }
 }
 
 async function newDenopsImpl(
   meta: Meta,
   session: Session,
+  client: Client,
   pluginName: string,
 ): Promise<Denops> {
   const url = path.toFileUrl(path.join(
@@ -120,5 +113,5 @@ async function newDenopsImpl(
     "impl.ts",
   ));
   const { DenopsImpl } = await import(url.href);
-  return new DenopsImpl(pluginName, meta, session);
+  return new DenopsImpl(pluginName, meta, session, client);
 }