chore: Stop writing and close reader when consumer closes writer

Author: smessie

example/pipeline.ttl

@@ -7,9 +7,9 @@
 @prefix rml: <https://w3id.org/conn/rml#>.
 @prefix ql: <http://semweb.mmlab.be/ns/ql#>.
 
-<> owl:imports <../node_modules/@ajuvercr/js-runner/ontology.ttl>.
-<> owl:imports <../node_modules/@ajuvercr/js-runner/channels/file.ttl>.
-<> owl:imports <../node_modules/@ajuvercr/js-runner/channels/http.ttl>.
+<> owl:imports <../node_modules/@rdfc/js-runner/ontology.ttl>.
+<> owl:imports <../node_modules/@rdfc/js-runner/channels/file.ttl>.
+<> owl:imports <../node_modules/@rdfc/js-runner/channels/http.ttl>.
 <> owl:imports <../processor.ttl>.
 [ ] a js:JsChannel;
   :writer <json/writer>.

lib/client.ts

@@ -20,7 +20,7 @@ import { FetchedPage, Fetcher, longPromise, resetPromise } from "./pageFetcher";
 import { Manager } from "./memberManager";
 import { OrderedStrategy, StrategyEvents, UnorderedStrategy } from "./strategy";
 import debug from "debug";
-import type { Writer } from "@ajuvercr/js-runner";
+import type { Writer } from "@rdfc/js-runner";
 
 // import { ReadableStream } from "stream/web";
 export { intoConfig } from "./config";
@@ -468,8 +468,14 @@ export async function processor(
     client.on("fragment", () => console.error("Fragment!"));
   }
 
+  const reader = client.stream({ highWaterMark: 10 }).getReader();
+
+  writer.on("end", async () => {
+      await reader.cancel();
+      console.log("Writer closed, so closing reader as well.");
+  });
+
   return async () => {
-    const reader = client.stream({ highWaterMark: 10 }).getReader();
     let el = await reader.read();
     const seen = new Set();
     while (el) {