Node.js API compatibility - Stream

/**
 * An example of using Node.js Stream API in an Azion Function.
 * Support:
 * - Partially supported (Extended by library `stream-browserify`)
 * @module runtime-apis/nodejs/stream/main
 * @example
 * // Execute with Azion Bundler:
 * npx edge-functions build
 * npx edge-functions dev
 */
import stream from "node:stream";

/**
 * An example of using the Node.js Stream API in an Azion Function.
 * @param {*} event
 * @returns {Promise<Response>}
 */
const main = async (event) => {
  return new Promise((resolve, reject) => {
    const chunks = ["chunk1", "chunk2", "chunk3", "chunk4", "chunk5"];

    const nextChunk = () => {
      const chunk = chunks.shift();
      if (chunk) {
        console.log("Chunk", chunk);
        nextChunk();
      }
    };

    const readable = new stream.Readable({
      encoding: "utf8",
      read() {
        nextChunk();
      },
    });

    const writable = new stream.Writable({
      write(chunk, encoding, callback) {
        console.log("Chunk", chunk.toString());
        callback();
      },
    });

    readable.pipe(writable);

    resolve(new Response("Done"));
  });
};

export default main;

import { Transform, pipeline } from "node:stream";

const main = async (event) => {
  // Create a transform stream that uppercases text
  const upperCaseTransform = new Transform({
    transform(chunk, encoding, callback) {
      const uppercased = chunk.toString().toUpperCase();
      this.push(uppercased);
      callback();
    }
  });

  // Create a transform stream that adds line numbers
  let lineNumber = 0;
  const addLineNumbers = new Transform({
    transform(chunk, encoding, callback) {
      lineNumber++;
      const numbered = `${lineNumber}: ${chunk.toString()}`;
      this.push(numbered);
      callback();
    }
  });

  // Collect output
  const output = [];
  const collector = new Transform({
    transform(chunk, encoding, callback) {
      output.push(chunk.toString());
      callback();
    }
  });

  // Simulate input data
  const inputData = ["hello\n", "world\n", "azion\n", "runtime\n"];

  // Process each chunk through the pipeline
  for (const data of inputData) {
    upperCaseTransform.write(data);
  }
  upperCaseTransform.end();

  return new Promise((resolve) => {
    pipeline(
      upperCaseTransform,
      addLineNumbers,
      collector,
      (err) => {
        if (err) {
          console.error("Pipeline error:", err);
        }
        console.log("Output:", output.join(""));
        resolve(new Response(output.join("")));
      }
    );
  });
};

export default main;

import { PassThrough, pipeline } from "node:stream";

const main = async (event) => {
  // Create a PassThrough stream
  const passThrough = new PassThrough();

  // Collect data that passes through
  const collected = [];
  passThrough.on("data", (chunk) => {
    collected.push(chunk.toString());
    console.log("Data passed through:", chunk.toString());
  });

  // Write data to the stream
  const data = ["First chunk\n", "Second chunk\n", "Third chunk\n"];
  data.forEach(chunk => passThrough.write(chunk));
  passThrough.end();

  // Wait for all data to be processed
  await new Promise(resolve => passThrough.on("finish", resolve));

  return new Response(JSON.stringify({
    message: "Data passed through successfully",
    chunks: collected,
    totalChunks: collected.length
  }), {
    headers: { "Content-Type": "application/json" }
  });
};

export default main;

import { Duplex } from "node:stream";

const main = async (event) => {
  // Create a duplex stream
  const duplex = new Duplex({
    read(size) {
      // Simulate reading data
      if (this._readCounter === undefined) this._readCounter = 0;
      this._readCounter++;

      if (this._readCounter <= 3) {
        this.push(`Read chunk ${this._readCounter}\n`);
      } else {
        this.push(null); // End the stream
      }
    },
    write(chunk, encoding, callback) {
      console.log("Written:", chunk.toString().trim());
      callback();
    }
  });

  // Write to the duplex stream
  duplex.write("Data to write\n");
  duplex.write("More data\n");

  // Read from the duplex stream
  const output = [];
  duplex.on("data", (chunk) => {
    output.push(chunk.toString());
  });

  // Wait for stream to end
  await new Promise(resolve => duplex.on("end", resolve));

  return new Response(JSON.stringify({
    readOutput: output.join(""),
    message: "Duplex stream processed"
  }), {
    headers: { "Content-Type": "application/json" }
  });
};

export default main;

import { Transform } from "node:stream";
import { Buffer } from "node:buffer";

const main = async (event) => {
  const request = event.request;

  // Create a transform stream to process chunks
  let totalBytes = 0;
  const byteCounter = new Transform({
    transform(chunk, encoding, callback) {
      totalBytes += chunk.length;
      this.push(chunk);
      callback();
    }
  });

  // Create a transform to collect data
  const chunks = [];
  const collector = new Transform({
    transform(chunk, encoding, callback) {
      chunks.push(chunk);
      this.push(chunk);
      callback();
    }
  });

  // Get request body as stream (if available)
  if (request.body) {
    const reader = request.body.getReader();

    // Process the stream
    while (true) {
      const { done, value } = await reader.read();
      if (done) break;

      byteCounter.write(Buffer.from(value));
      collector.write(Buffer.from(value));
    }

    byteCounter.end();
    collector.end();
  }

  // Combine collected chunks
  const bodyContent = chunks.length > 0
    ? Buffer.concat(chunks).toString("utf8")
    : "No body content";

  return new Response(JSON.stringify({
    totalBytes,
    chunkCount: chunks.length,
    bodyPreview: bodyContent.substring(0, 200)
  }), {
    headers: { "Content-Type": "application/json" }
  });
};

export default main;