feat(stdlib): remove pipeline from streamToBuffer

packages/stdlib/src/node.js

@@ -2,11 +2,9 @@ import { exec as cpExec, spawn as cpSpawn } from "child_process";
 import { lstatSync, readdirSync } from "fs";
 import { lstat, readdir } from "fs/promises";
 import { posix } from "path";
-import { pipeline } from "stream";
 import { promisify } from "util";
 
 const internalExec = promisify(cpExec);
-const internalPipeline = promisify(pipeline);
 
 /**
  * Join all arguments together and normalize the resulting path. Arguments must be
@@ -77,19 +75,27 @@ export function spawn(command, args, opts = {}) {
  *
  * @since 0.1.0
  *
- * @param {ReadableStream} stream
+ * @param {NodeJS.ReadableStream} stream
  * @returns {Promise<Buffer>}
  */
 export async function streamToBuffer(stream) {
-  const buffers = [];
-  await internalPipeline(stream, async function* (transform) {
-    for await (const chunk of transform) {
+  if (!stream || typeof stream._read !== "function") {
+    return Buffer.from([]);
+  }
+
+  return new Promise((resolve, reject) => {
+    const buffers = [];
+
+    stream.on("data", function (chunk) {
       buffers.push(chunk);
-      yield chunk;
-    }
+    });
+    stream.on("end", function () {
+      resolve(Buffer.concat(buffers));
+    });
+    stream.on("error", function (err) {
+      reject(err);
+    });
   });
-
-  return Buffer.concat(buffers);
 }
 
 /**

packages/store/src/file-cache.js

@@ -1,8 +1,8 @@
 import { once } from "events";
-import { mkdirSync, createReadStream, createWriteStream } from "fs";
+import { createReadStream, createWriteStream, mkdirSync } from "fs";
 import { pipeline as pipelineCallbacks, Readable } from "stream";
 import { promisify } from "util";
-import { isNil, pathJoin, uuid } from "@compas/stdlib";
+import { isNil, pathJoin, streamToBuffer, uuid } from "@compas/stdlib";
 import { getFileStream } from "./files.js";
 
 const pipeline = promisify(pipelineCallbacks);
@@ -126,18 +126,10 @@ export class FileCache {
    * @param end
    */
   async cacheFileInMemory(key, id, start, end) {
-    const buffers = [];
-    await pipeline(
-      await getFileStream(this.minio, this.bucketName, id),
-      async function* (transform) {
-        for await (const chunk of transform) {
-          buffers.push(chunk);
-          yield chunk;
-        }
-      },
-    );
+    const stream = await getFileStream(this.minio, this.bucketName, id);
+    const buffer = await streamToBuffer(stream);
 
-    this.memoryCache.set(key, Buffer.concat(buffers));
+    this.memoryCache.set(key, buffer);
 
     return {
       stream: Readable.from(this.memoryCache.get(key).slice(start, end)),