streamx 2.28.0: Readable

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 streamx.Readable in pkg:npm/streamx@2.28.0
Kind: HOW

Use EXACTLY these public packages and versions:
  - pkg:npm/streamx@2.28.0
Demonstrate these symbols/APIs:
  - streamx.Readable

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:675e05389c716cf175135dc8a351bf6640fa250bb7a80058b1213615d3accc7e","contract":["Readable stream emits data and end events when chunks are pushed and ended with null","Readable class can be subclassed overriding _read to push items to the stream","Readable stream supports async iteration yielding each pushed chunk in order","Readable.from static method creates a readable stream from an iterable collection","Readable stream supports flow control with pause and resume and reflects state via Readable.isPaused","Readable stream supports unshift to prepend data back to the read queue","Readable stream supports open lifecycle hook for asynchronous initialization before reading","Readable stream supports map option to transform items during streaming","Readable stream supports pull-based consumption using readable event and read method","Readable stream emits error and close events when destroyed with an error"],"goal":"verify streamx.Readable in pkg:npm/streamx@2.28.0","kind":"HOW","packages":["pkg:npm/streamx@2.28.0"],"schemaVersion":1,"symbols":["streamx.Readable"]},"contractCommand":["node","test/contract.mjs"],"environment":{"arch":"x64","distro":"ubuntu","ecosystem":"npm","libc":"glibc","libcVersion":"2.39","os":"linux","osVersionBucket":"24","packageManager":"npm","schemaVersion":1},"license":"MIT-0","packages":["pkg:npm/streamx@2.28.0"],"schemaVersion":1,"subject":"pkg:npm/streamx@2.28.0","symbols":["streamx.Readable"],"verifierAdapter":"node-typescript@1"}

{
  "name": "sample-streamx-readable",
  "version": "1.0.0",
  "lockfileVersion": 3,
  "requires": true,
  "packages": {
    "": {
      "name": "sample-streamx-readable",
      "version": "1.0.0",
      "dependencies": {
        "streamx": "2.28.0"
      }
    },
    "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.1",
      "resolved": "https://registry.npmjs.org/bare-events/-/bare-events-2.9.1.tgz",
      "integrity": "sha512-Z0oHEHAFDZkffN8Qc39zNZjQlMDkPJRyyyZieU1VH7u8c5S+qHZ2S8ixdKIAxEjfHO7FJxXmJWgteOghVanIsg==",
      "license": "Apache-2.0",
      "peerDependencies": {
        "bare-abort-controller": "*"
      },
      "peerDependenciesMeta": {
        "bare-abort-controller": {
          "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.0",
      "resolved": "https://registry.npmjs.org/streamx/-/streamx-2.28.0.tgz",
      "integrity": "sha512-1Yowhzjf0ivGMrTIkY9hav5TxobO9qIVqUE41fiCGMGgc3CLlf4MY+9AHmZqBWgDTue0fY9zWjYFVyf6Diuobw==",
      "license": "MIT",
      "dependencies": {
        "events-universal": "^1.0.0",
        "fast-fifo": "^1.3.2",
        "text-decoder": "^1.1.0"
      }
    },
    "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-streamx-readable",
  "version": "1.0.0",
  "type": "module",
  "private": true,
  "dependencies": {
    "streamx": "2.28.0"
  }
}

{
  "schemaVersion": 1,
  "goal": "verify streamx.Readable in pkg:npm/streamx@2.28.0",
  "kind": "HOW",
  "packages": [
    "pkg:npm/streamx@2.28.0"
  ],
  "symbols": [
    "streamx.Readable"
  ]
}

import assert from 'node:assert/strict';
import { Readable } from 'streamx';

// 1. Readable stream emits data and end events when chunks are pushed and ended with null
await new Promise((resolve, reject) => {
  const chunks = [];
  const readable = new Readable({
    read(cb) {
      this.push('chunk1');
      this.push('chunk2');
      this.push(null);
      cb(null);
    }
  });

  readable.on('data', (chunk) => {
    chunks.push(chunk);
  });

  readable.on('end', () => {
    try {
      assert.deepEqual(chunks, ['chunk1', 'chunk2']);
      resolve();
    } catch (err) {
      reject(err);
    }
  });

  readable.on('error', reject);
});

// 2. Readable class can be subclassed overriding _read to push items to the stream
await new Promise((resolve, reject) => {
  class CustomReader extends Readable {
    constructor() {
      super();
      this.count = 0;
    }

    _read(cb) {
      if (this.count < 3) {
        this.push(`item-${this.count++}`);
      } else {
        this.push(null);
      }
      cb(null);
    }
  }

  const reader = new CustomReader();
  const results = [];

  reader.on('data', (chunk) => {
    results.push(chunk);
  });

  reader.on('end', () => {
    try {
      assert.deepEqual(results, ['item-0', 'item-1', 'item-2']);
      resolve();
    } catch (err) {
      reject(err);
    }
  });

  reader.on('error', reject);
});

// 3. Readable stream supports async iteration yielding each pushed chunk in order
await (async () => {
  const readable = new Readable({
    read(cb) {
      this.push('a');
      this.push('b');
      this.push('c');
      this.push(null);
      cb(null);
    }
  });

  const collected = [];
  for await (const item of readable) {
    collected.push(item);
  }

  assert.deepEqual(collected, ['a', 'b', 'c']);
})();

// 4. Readable.from static method creates a readable stream from an iterable collection
await (async () => {
  const fromStream = Readable.from(['alpha', 'beta', 'gamma']);
  const collected = [];
  for await (const item of fromStream) {
    collected.push(item);
  }

  assert.deepEqual(collected, ['alpha', 'beta', 'gamma']);
})();

// 5. Readable stream supports flow control with pause and resume and reflects state via Readable.isPaused
await new Promise((resolve, reject) => {
  const emitted = [];
  const readable = new Readable({
    read(cb) {
      this.push('first');
      this.push('second');
      this.push(null);
      cb(null);
    }
  });

  readable.on('data', (chunk) => {
    emitted.push(chunk);
    if (chunk === 'first') {
      readable.pause();
      assert.equal(Readable.isPaused(readable), true);
      setTimeout(() => {
        readable.resume();
      }, 30);
    }
  });

  readable.on('end', () => {
    try {
      assert.deepEqual(emitted, ['first', 'second']);
      resolve();
    } catch (err) {
      reject(err);
    }
  });

  readable.on('error', reject);
});

// 6. Readable stream supports unshift to prepend data back to the read queue
await new Promise((resolve, reject) => {
  const readable = new Readable({
    read(cb) {
      this.push('original');
      this.push(null);
      cb(null);
    }
  });

  readable.unshift('prepended');

  const items = [];
  readable.on('data', (chunk) => {
    items.push(chunk);
  });

  readable.on('end', () => {
    try {
      assert.deepEqual(items, ['prepended', 'original']);
      resolve();
    } catch (err) {
      reject(err);
    }
  });

  readable.on('error', reject);
});

// 7. Readable stream supports open lifecycle hook for asynchronous initialization before reading
await new Promise((resolve, reject) => {
  let opened = false;
  const events = [];

  const readable = new Readable({
    open(cb) {
      setTimeout(() => {
        opened = true;
        events.push('open');
        cb(null);
      }, 20);
    },
    read(cb) {
      assert.equal(opened, true);
      events.push('read');
      this.push('data');
      this.push(null);
      cb(null);
    }
  });

  readable.on('data', (chunk) => {
    events.push(`data:${chunk}`);
  });

  readable.on('end', () => {
    try {
      assert.deepEqual(events, ['open', 'read', 'data:data']);
      resolve();
    } catch (err) {
      reject(err);
    }
  });

  readable.on('error', reject);
});

// 8. Readable stream supports map option to transform items during streaming
await (async () => {
  const readable = new Readable({
    map(val) {
      return val.toUpperCase();
    },
    read(cb) {
      this.push('foo');
      this.push('bar');
      this.push(null);
      cb(null);
    }
  });

  const collected = [];
  for await (const chunk of readable) {
    collected.push(chunk);
  }

  assert.deepEqual(collected, ['FOO', 'BAR']);
})();

// 9. Readable stream supports pull-based consumption using readable event and read method
await new Promise((resolve, reject) => {
  const readable = new Readable({
    read(cb) {
      this.push('p1');
      this.push('p2');
      this.push(null);
      cb(null);
    }
  });

  const collected = [];
  readable.on('readable', () => {
    let chunk;
    while ((chunk = readable.read()) !== null) {
      collected.push(chunk);
    }
  });

  readable.on('end', () => {
    try {
      assert.deepEqual(collected, ['p1', 'p2']);
      resolve();
    } catch (err) {
      reject(err);
    }
  });

  readable.on('error', reject);
});

// 10. Readable stream emits error and close events when destroyed with an error
await new Promise((resolve, reject) => {
  const readable = new Readable({
    read(cb) {
      cb(null);
    }
  });

  let errorSeen = false;
  readable.on('error', (err) => {
    if (err.message === 'stream failure') {
      errorSeen = true;
    }
  });

  readable.on('close', () => {
    try {
      assert.equal(errorSeen, true);
      assert.equal(readable.destroyed, true);
      resolve();
    } catch (err) {
      reject(err);
    }
  });

  readable.destroy(new Error('stream failure'));
});