bare-stream 2.13.3: pipeline

Clean-room public code sample — generation instructions

Write a brand-new, minimal, self-contained code sample in this clean-room directory.
Do not copy, paraphrase, or reference any existing project source. Work only from this spec.

A csx.json manifest scaffold already exists. Do not recreate it from memory. Preserve its case.goal, packages and symbols; fill its empty case.contract with exact assertions and correct its environment, commands and verifierAdapter for the files you generate.

Goal: verify bare-stream.pipeline in pkg:npm/bare-stream@2.13.3
Kind: HOW

Use EXACTLY these public packages and versions:
  - pkg:npm/bare-stream@2.13.3
Demonstrate these symbols/APIs:
  - bare-stream.pipeline
Constraints:
  - executionContext: node
Required runtime conditions:
  - ecosystem: npm
  - language: javascript
  - moduleSystem: cjs
  - packageManager: npm@10.9.8
  - runtime: node@22.23.2

Rules:
  - One focused purpose; the smallest project that proves the goal.
  - Include a contract test (test/contract.*) that runs OFFLINE and exits 0 exactly when the goal behavior works.
  - Pin every dependency with a lockfile so resolution is reproducible.
  - No secrets, credentials, or tokens. No real URLs (only example.com or localhost). No absolute paths.
  - No personal names, emails, company names, or project identifiers of any kind.
  - No binaries and no generated output (node_modules, dist, target, venv, .git, .env).
  - Keep it under 200 files and 256KB packed.

{"case":{"caseId":"case:sha256:85c306333e3da4698495b582f5eb3960d36ee3599d7c37523a685c340cd4fc3e","constraints":{"executionContext":"node"},"contract":["pipeline pipes data from a Readable source to a Writable destination and executes completion callback","pipeline chains a Readable stream through intermediate Transform streams to a Writable destination","pipeline accepts stream arguments either as consecutive parameters or as a single array of streams","pipeline returns the destination stream which is the last stream in the pipeline chain","pipeline throws a BAD_ARGUMENT error when fewer than two streams are provided","pipeline destroys all streams in the pipeline and forwards error to callback when the source stream fails","pipeline destroys all streams in the pipeline and forwards error to callback when an intermediate transform stream fails"],"goal":"verify bare-stream.pipeline in pkg:npm/bare-stream@2.13.3","kind":"HOW","packages":["pkg:npm/bare-stream@2.13.3"],"schemaVersion":1,"symbols":["bare-stream.pipeline"]},"contractCommand":["node","test/contract.mjs"],"environment":{"arch":"x64","distro":"ubuntu","ecosystem":"npm","executionContext":"node","language":"javascript","libc":"glibc","libcVersion":"2.39","moduleSystem":"cjs","os":"linux","osVersionBucket":"24","packageManager":"npm","packageManagerVersion":"10.9.8","runtime":"node","runtimeVersion":"22.23.2","schemaVersion":1},"license":"MIT-0","packages":["pkg:npm/bare-stream@2.13.3"],"schemaVersion":1,"subject":"pkg:npm/bare-stream@2.13.3","symbols":["bare-stream.pipeline"],"verifierAdapter":"node-typescript@1"}

{
  "name": "sample-bare-stream-pipeline",
  "version": "1.0.0",
  "lockfileVersion": 3,
  "requires": true,
  "packages": {
    "": {
      "name": "sample-bare-stream-pipeline",
      "version": "1.0.0",
      "license": "MIT-0",
      "dependencies": {
        "bare-stream": "2.13.3"
      }
    },
    "node_modules/b4a": {
      "version": "1.8.1",
      "resolved": "https://registry.npmjs.org/b4a/-/b4a-1.8.1.tgz",
      "integrity": "sha512-aiqre1Nr0B/6DgE2N5vwTc+2/oQZ4Wh1t4NznYY4E00y8LCt6NqdRv81so00oo27D8MVKTpUa/MwUUtBLXCoDw==",
      "license": "Apache-2.0",
      "peerDependencies": {
        "react-native-b4a": "*"
      },
      "peerDependenciesMeta": {
        "react-native-b4a": {
          "optional": true
        }
      }
    },
    "node_modules/bare-events": {
      "version": "2.9.2",
      "resolved": "https://registry.npmjs.org/bare-events/-/bare-events-2.9.2.tgz",
      "integrity": "sha512-AIPKioV7/Y/8KfZ3AAhjPJxLLbY49S64Ym5DakZlUg75qQiTgUq9hEJoEwa4eUezPUlXRy/i5NpsKvo9jgKmoA==",
      "license": "Apache-2.0",
      "peerDependencies": {
        "bare-abort-controller": "*"
      },
      "peerDependenciesMeta": {
        "bare-abort-controller": {
          "optional": true
        }
      }
    },
    "node_modules/bare-stream": {
      "version": "2.13.3",
      "resolved": "https://registry.npmjs.org/bare-stream/-/bare-stream-2.13.3.tgz",
      "integrity": "sha512-Kc+brLqvEqGkjyfiwJmImAOqLZL7OsoLKuavx+hJjgVV3nLTOjloJyPMFxjUPerGGHrNH0fLU06jjykMLWrERQ==",
      "license": "Apache-2.0",
      "dependencies": {
        "b4a": "^1.8.1",
        "streamx": "^2.25.0",
        "teex": "^1.0.1"
      },
      "peerDependencies": {
        "bare-abort-controller": "*",
        "bare-buffer": "*",
        "bare-events": "*"
      },
      "peerDependenciesMeta": {
        "bare-abort-controller": {
          "optional": true
        },
        "bare-buffer": {
          "optional": true
        },
        "bare-events": {
          "optional": true
        }
      }
    },
    "node_modules/events-universal": {
      "version": "1.0.1",
      "resolved": "https://registry.npmjs.org/events-universal/-/events-universal-1.0.1.tgz",
      "integrity": "sha512-LUd5euvbMLpwOF8m6ivPCbhQeSiYVNb8Vs0fQ8QjXo0JTkEHpz8pxdQf0gStltaPpw0Cca8b39KxvK9cfKRiAw==",
      "license": "Apache-2.0",
      "dependencies": {
        "bare-events": "^2.7.0"
      }
    },
    "node_modules/fast-fifo": {
      "version": "1.3.2",
      "resolved": "https://registry.npmjs.org/fast-fifo/-/fast-fifo-1.3.2.tgz",
      "integrity": "sha512-/d9sfos4yxzpwkDkuN7k2SqFKtYNmCTzgfEpz82x34IM9/zc8KGxQoXg1liNC/izpRM/MBdt44Nmx41ZWqk+FQ==",
      "license": "MIT"
    },
    "node_modules/streamx": {
      "version": "2.28.1",
      "resolved": "https://registry.npmjs.org/streamx/-/streamx-2.28.1.tgz",
      "integrity": "sha512-zEzXb0s5Cds7tqMH6rhZ05lcJydCWiQPEwiNngVqzsxCc962vLY4Uw+mW7od8kDH258k2Uz/JrOkdIAAhSh9VA==",
      "license": "MIT",
      "dependencies": {
        "events-universal": "^1.0.0",
        "fast-fifo": "^1.3.2",
        "text-decoder": "^1.1.0"
      }
    },
    "node_modules/teex": {
      "version": "1.0.1",
      "resolved": "https://registry.npmjs.org/teex/-/teex-1.0.1.tgz",
      "integrity": "sha512-eYE6iEI62Ni1H8oIa7KlDU6uQBtqr4Eajni3wX7rpfXD8ysFx8z0+dri+KWEPWpBsxXfxu58x/0jvTVT1ekOSg==",
      "license": "MIT",
      "dependencies": {
        "streamx": "^2.12.5"
      }
    },
    "node_modules/text-decoder": {
      "version": "1.2.7",
      "resolved": "https://registry.npmjs.org/text-decoder/-/text-decoder-1.2.7.tgz",
      "integrity": "sha512-vlLytXkeP4xvEq2otHeJfSQIRyWxo/oZGEbXrtEEF9Hnmrdly59sUbzZ/QgyWuLYHctCHxFF4tRQZNQ9k60ExQ==",
      "license": "Apache-2.0",
      "dependencies": {
        "b4a": "^1.6.4"
      }
    }
  }
}

{
  "name": "sample-bare-stream-pipeline",
  "version": "1.0.0",
  "type": "module",
  "private": true,
  "license": "MIT-0",
  "dependencies": {
    "bare-stream": "2.13.3"
  }
}

{
  "schemaVersion": 1,
  "goal": "verify bare-stream.pipeline in pkg:npm/bare-stream@2.13.3",
  "kind": "HOW",
  "packages": [
    "pkg:npm/bare-stream@2.13.3"
  ],
  "symbols": [
    "bare-stream.pipeline"
  ],
  "constraints": {
    "executionContext": "node"
  },
  "runtimeConditions": {
    "ecosystem": "npm",
    "language": "javascript",
    "moduleSystem": "cjs",
    "packageManager": "npm@10.9.8",
    "runtime": "node@22.23.2"
  }
}

import assert from "node:assert/strict";
import { pipeline, Readable, Writable, Transform, PassThrough } from "bare-stream";

// 1. pipeline pipes data from a Readable source to a Writable destination and executes completion callback
{
  const input = ["hello", " ", "world"];
  const output = [];
  const source = Readable.from(input);
  const dest = new Writable({
    write(chunk, encoding, cb) {
      output.push(chunk.toString());
      cb(null);
    }
  });

  let completed = false;
  let receivedError = undefined;

  await new Promise((resolve) => {
    pipeline(source, dest, (err) => {
      completed = true;
      receivedError = err;
      resolve();
    });
  });

  assert.equal(completed, true, "pipeline callback must be invoked");
  assert.equal(receivedError, null, "pipeline callback error must be null on clean completion");
  assert.equal(output.join(""), "hello world", "pipeline must deliver all source data to destination");
}

// 2. pipeline chains a Readable stream through intermediate Transform streams to a Writable destination
{
  const input = ["alpha", "beta", "gamma"];
  const output = [];
  const source = Readable.from(input);
  const upperTransform = new Transform({
    transform(chunk, encoding, cb) {
      cb(null, chunk.toString().toUpperCase());
    }
  });
  const dest = new Writable({
    write(chunk, encoding, cb) {
      output.push(chunk.toString());
      cb(null);
    }
  });

  let completed = false;
  let receivedError = undefined;

  await new Promise((resolve) => {
    pipeline(source, upperTransform, dest, (err) => {
      completed = true;
      receivedError = err;
      resolve();
    });
  });

  assert.equal(completed, true, "pipeline callback must be invoked");
  assert.equal(receivedError, null, "pipeline callback error must be null on clean completion");
  assert.deepEqual(output, ["ALPHA", "BETA", "GAMMA"], "pipeline must transform all chunks along the stream chain");
}

// 3. pipeline accepts stream arguments either as consecutive parameters or as a single array of streams
{
  const input = ["foo", "bar"];
  const output = [];
  const source = Readable.from(input);
  const pass = new PassThrough();
  const dest = new Writable({
    write(chunk, encoding, cb) {
      output.push(chunk.toString());
      cb(null);
    }
  });

  let completed = false;
  let receivedError = undefined;

  await new Promise((resolve) => {
    pipeline([source, pass, dest], (err) => {
      completed = true;
      receivedError = err;
      resolve();
    });
  });

  assert.equal(completed, true, "pipeline callback must be invoked for array form");
  assert.equal(receivedError, null, "pipeline callback error must be null on clean completion");
  assert.deepEqual(output, ["foo", "bar"], "pipeline with array of streams must deliver all chunks");
}

// 4. pipeline returns the destination stream which is the last stream in the pipeline chain
{
  const source = Readable.from(["a", "b"]);
  const dest = new PassThrough();
  const returned = pipeline(source, dest);

  assert.equal(returned, dest, "pipeline must return the destination stream");
  source.resume();
}

// 5. pipeline throws a BAD_ARGUMENT error when fewer than two streams are provided
{
  const singleStream = new PassThrough();
  assert.throws(
    () => pipeline(singleStream),
    (err) => err.message.includes("Pipeline requires at least 2 streams"),
    "pipeline with one stream argument must throw"
  );

  assert.throws(
    () => pipeline([singleStream]),
    (err) => err.message.includes("Pipeline requires at least 2 streams"),
    "pipeline with array of one stream must throw"
  );
}

// 6. pipeline destroys all streams in the pipeline and forwards error to callback when the source stream fails
{
  const expectedErr = new Error("source failure");
  const source = new PassThrough();
  const transform = new PassThrough();
  const dest = new PassThrough();

  let callbackErr = null;
  await new Promise((resolve) => {
    pipeline(source, transform, dest, (err) => {
      callbackErr = err;
      resolve();
    });
    source.destroy(expectedErr);
  });

  assert.equal(callbackErr, expectedErr, "pipeline callback must receive the source error");
  assert.equal(source.destroyed, true, "source stream must be destroyed");
  assert.equal(transform.destroyed, true, "transform stream must be destroyed on source error");
  assert.equal(dest.destroyed, true, "destination stream must be destroyed on source error");
}

// 7. pipeline destroys all streams in the pipeline and forwards error to callback when an intermediate transform stream fails
{
  const expectedErr = new Error("transform failure");
  const source = new PassThrough();
  const transform = new PassThrough();
  const dest = new PassThrough();

  let callbackErr = null;
  await new Promise((resolve) => {
    pipeline(source, transform, dest, (err) => {
      callbackErr = err;
      resolve();
    });
    transform.destroy(expectedErr);
  });

  assert.equal(callbackErr, expectedErr, "pipeline callback must receive the transform error");
  assert.equal(source.destroyed, true, "source stream must be destroyed on transform error");
  assert.equal(dest.destroyed, true, "destination stream must be destroyed on transform error");
}

console.log("Contract tests passed successfully.");