Eliminate buffering in transform

A `TransformStream` currently always requires an internal queue
(until whatwg/streams#1158 is resolved).

Therefore, don't use `TransformStream` in `transform` anymore, but
create a new `ReadableStream` with the transformed chunks directly,
and with a `highWaterMark` of 0, such that the internal queue is
always empty.

Author: twiss

lib/streams.js

@@ -248,24 +248,45 @@ function transform(input, process = () => undefined, finish = () => undefined) {
     return output;
   }
   if (isStream(input)) {
-    return transformRaw(input, {
-      async transform(value, controller) {
-        try {
-          const result = await process(value);
-          if (result !== undefined) controller.enqueue(result);
-        } catch(e) {
-          controller.error(e);
-        }
+    let reader;
+    let allDone = false;
+    return new ReadableStream({
+      start() {
+        reader = input.getReader();
       },
-      async flush(controller) {
+      async pull(controller) {
+        if (allDone) {
+          controller.close();
+          return;
+        }
         try {
-          const result = await finish();
-          if (result !== undefined) controller.enqueue(result);
-        } catch(e) {
+          // Read repeatedly until we have a chunk to enqueue or until
+          // we can close the stream, as `pull` won't get called again
+          // until we call `enqueue` or `close`.
+          while (true) {
+            const { value, done } = await reader.read();
+            allDone = done;
+            const result = await (done ? finish : process)(value);
+            if (result !== undefined) {
+              controller.enqueue(result);
+              return; // `pull` will get called again
+            }
+            if (done) {
+              // If `finish` didn't return a chunk to enqueue, call
+              // `close` here. Otherwise, it will get called in the
+              // next call to `pull`, above (since `allDone == true`).
+              controller.close();
+              return;
+            }
+          }
+        } catch (e) {
           controller.error(e);
         }
+      },
+      async cancel(reason) {
+        await reader.cancel(reason);
       }
-    });
+    }, { highWaterMark: 0 });
   }
   const result1 = process(input);
   const result2 = finish();