ws 8.21.1: WebSocket.createWebSocketStream

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 WebSocket.createWebSocketStream in pkg:npm/ws@8.21.1
Kind: HOW

Use EXACTLY these public packages and versions:
  - pkg:npm/ws@8.21.1
Demonstrate these symbols/APIs:
  - WebSocket.createWebSocketStream

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:e56bc3b62ee387cc0c9993b498636d257dd23c9def1216efc345a2752c23b1bb","contract":["WebSocket.createWebSocketStream wraps a WebSocket connection in a standard Node.js Duplex stream.","WebSocket.createWebSocketStream transmits data written to the stream over the WebSocket and pushes received frames to the readable stream.","WebSocket.createWebSocketStream supports Duplex stream options such as encoding utf8 to emit string chunks on data events.","WebSocket.createWebSocketStream integrates with Node.js stream pipelining utilities to pipe readable stream sources through WebSockets.","WebSocket.createWebSocketStream handles graceful stream termination on end and closes or terminates the underlying WebSocket on stream destroy."],"goal":"verify WebSocket.createWebSocketStream in pkg:npm/ws@8.21.1","kind":"HOW","packages":["pkg:npm/ws@8.21.1"],"schemaVersion":1,"symbols":["WebSocket.createWebSocketStream"]},"contractCommand":["node","test/contract.mjs"],"environment":{"arch":"x64","distro":"ubuntu","ecosystem":"npm","language":"javascript","libc":"glibc","libcVersion":"2.39","os":"linux","osVersionBucket":"24","packageManager":"npm","runtime":"node","runtimeVersion":"22.23.2","schemaVersion":1},"license":"MIT-0","packages":["pkg:npm/ws@8.21.1"],"schemaVersion":1,"subject":"pkg:npm/ws@8.21.1","symbols":["WebSocket.createWebSocketStream"],"verifierAdapter":"node-typescript@1"}

const WebSocket = require('ws');
const { createWebSocketStream } = WebSocket;
const { pipeline } = require('stream');

/**
 * Creates a Duplex stream wrapping the provided WebSocket instance.
 *
 * @param {WebSocket} ws - The WebSocket connection to wrap
 * @param {import('stream').DuplexOptions} [options] - Optional stream configuration options
 * @returns {import('stream').Duplex} Duplex stream wrapping the WebSocket
 */
function createStream(ws, options) {
  return createWebSocketStream(ws, options);
}

/**
 * Pipes a readable source stream into a WebSocket via createWebSocketStream.
 *
 * @param {WebSocket} ws - Destination WebSocket
 * @param {import('stream').Readable} sourceStream - Readable stream source
 * @param {import('stream').DuplexOptions} [options] - Optional stream configuration options
 * @returns {Promise<void>} Resolves when piping finishes
 */
function pipeToWebSocket(ws, sourceStream, options) {
  const wsStream = createWebSocketStream(ws, options);
  return new Promise((resolve, reject) => {
    pipeline(sourceStream, wsStream, (err) => {
      if (err) reject(err);
      else resolve();
    });
  });
}

/**
 * Collects all data chunks emitted by a stream into a single Buffer.
 *
 * @param {import('stream').Readable} stream - The readable stream to collect data from
 * @returns {Promise<Buffer>} Collected stream payload as a Buffer
 */
function collectStream(stream) {
  return new Promise((resolve, reject) => {
    const chunks = [];
    stream.on('data', (chunk) => {
      chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk));
    });
    stream.once('end', () => {
      resolve(Buffer.concat(chunks));
    });
    stream.once('error', reject);
  });
}

module.exports = {
  createStream,
  createWebSocketStream,
  pipeToWebSocket,
  collectStream,
  WebSocket
};

{
  "name": "sample-ws-create-websocket-stream",
  "version": "1.0.0",
  "lockfileVersion": 3,
  "requires": true,
  "packages": {
    "": {
      "name": "sample-ws-create-websocket-stream",
      "version": "1.0.0",
      "license": "MIT-0",
      "dependencies": {
        "ws": "8.21.1"
      }
    },
    "node_modules/ws": {
      "version": "8.21.1",
      "resolved": "https://registry.npmjs.org/ws/-/ws-8.21.1.tgz",
      "integrity": "sha512-+0NTnW77fFN/DjQi6k/Sq/Yvk4Sgajw7urW8V+asjXnRgDs9gyGkdb7EzgfhA4goXsRIZKE28fzIXBHEzhuiWw==",
      "license": "MIT",
      "engines": {
        "node": ">=10.0.0"
      },
      "peerDependencies": {
        "bufferutil": "^4.0.1",
        "utf-8-validate": ">=5.0.2"
      },
      "peerDependenciesMeta": {
        "bufferutil": {
          "optional": true
        },
        "utf-8-validate": {
          "optional": true
        }
      }
    }
  }
}

{
  "name": "sample-ws-create-websocket-stream",
  "version": "1.0.0",
  "private": true,
  "description": "Clean-room verification of WebSocket.createWebSocketStream in ws",
  "main": "index.js",
  "scripts": {
    "test": "node test/contract.mjs"
  },
  "license": "MIT-0",
  "dependencies": {
    "ws": "8.21.1"
  }
}

{
  "schemaVersion": 1,
  "goal": "verify WebSocket.createWebSocketStream in pkg:npm/ws@8.21.1",
  "kind": "HOW",
  "packages": [
    "pkg:npm/ws@8.21.1"
  ],
  "symbols": [
    "WebSocket.createWebSocketStream"
  ]
}

import assert from 'node:assert';
import { createRequire } from 'node:module';
import { Readable, Duplex } from 'node:stream';

const require = createRequire(import.meta.url);
const WebSocket = require('ws');
const { createWebSocketStream } = WebSocket;
const {
  createStream,
  pipeToWebSocket,
  collectStream
} = require('../index.js');

async function runContractTests() {
  // 1. Symbol and export verification
  assert.strictEqual(
    typeof WebSocket.createWebSocketStream,
    'function',
    'WebSocket.createWebSocketStream must be exported as a function'
  );
  assert.strictEqual(
    typeof createWebSocketStream,
    'function',
    'createWebSocketStream must be accessible from ws package'
  );
  assert.strictEqual(
    createWebSocketStream,
    WebSocket.createWebSocketStream,
    'createWebSocketStream and WebSocket.createWebSocketStream must be identical'
  );

  // Helper to create a local WebSocket server on ephemeral port
  function createLocalServer() {
    return new Promise((resolve, reject) => {
      const server = new WebSocket.Server({ port: 0, host: '127.0.0.1' }, () => {
        const address = server.address();
        resolve({ server, port: address.port });
      });
      server.once('error', reject);
    });
  }

  // Helper to establish an open client connection and return duplex stream
  function connectClient(port, options) {
    return new Promise((resolve, reject) => {
      const clientWs = new WebSocket(`ws://127.0.0.1:${port}`);
      const clientStream = createWebSocketStream(clientWs, options);
      clientWs.once('open', () => {
        resolve({ clientWs, clientStream });
      });
      clientWs.once('error', reject);
    });
  }

  // Helper to safely close server
  function closeServer(server) {
    return new Promise((resolve) => server.close(resolve));
  }

  // 2. Stream instantiation & prototype verification
  {
    const { server, port } = await createLocalServer();

    let serverWsStream = null;
    const clientConnectedPromise = new Promise((resolve) => {
      server.once('connection', (ws) => {
        serverWsStream = createStream(ws);
        assert(serverWsStream instanceof Duplex, 'Wrapped server socket stream must be a Duplex stream');
        resolve();
      });
    });

    const { clientWs, clientStream } = await connectClient(port);
    assert(clientStream instanceof Duplex, 'Wrapped client socket stream must be a Duplex stream');

    await clientConnectedPromise;

    clientWs.close();
    await closeServer(server);
  }

  // 3. Bidirectional data transfer through duplex streams
  {
    const { server, port } = await createLocalServer();

    const serverReceivedPromise = new Promise((resolve, reject) => {
      server.once('connection', (serverWs) => {
        try {
          const serverStream = createWebSocketStream(serverWs);
          const received = [];
          serverStream.on('data', (chunk) => {
            const text = chunk.toString('utf8');
            received.push(text);
            serverStream.write(Buffer.from(`ECHO:${text}`));
          });
          serverStream.once('end', () => {
            resolve(received);
          });
          serverStream.once('error', reject);
        } catch (err) {
          reject(err);
        }
      });
    });

    const { clientWs, clientStream } = await connectClient(port);

    const clientReceived = [];
    clientStream.on('data', (chunk) => {
      clientReceived.push(chunk.toString('utf8'));
    });

    clientStream.write('chunk-alpha\n');
    clientStream.write('chunk-beta\n');

    // Wait until echo response for chunk-beta is received
    await new Promise((resolve) => {
      const check = () => {
        if (clientReceived.join('').includes('ECHO:chunk-beta\n')) resolve();
      };
      clientStream.on('data', check);
      check();
    });

    assert.strictEqual(
      clientReceived.join(''),
      'ECHO:chunk-alpha\nECHO:chunk-beta\n',
      'Client duplex stream must receive echoed response chunks'
    );

    clientStream.end();
    const serverChunks = await serverReceivedPromise;
    assert.deepStrictEqual(
      serverChunks.join(''),
      'chunk-alpha\nchunk-beta\n',
      'Server duplex stream must receive all chunks sent by client stream'
    );

    clientWs.close();
    await closeServer(server);
  }

  // 4. Stream options support (encoding option yields string chunks on data events)
  {
    const { server, port } = await createLocalServer();

    const serverReceivedStringPromise = new Promise((resolve, reject) => {
      server.once('connection', (serverWs) => {
        const serverStream = createWebSocketStream(serverWs, { encoding: 'utf8' });
        serverStream.on('data', (chunk) => {
          assert.strictEqual(typeof chunk, 'string', 'Chunk must be decoded to string when encoding is utf8');
          resolve(chunk);
        });
        serverStream.once('error', reject);
      });
    });

    const { clientWs, clientStream } = await connectClient(port);
    clientStream.write(Buffer.from('utf8-encoded-payload'));

    const chunk = await serverReceivedStringPromise;
    assert.strictEqual(chunk, 'utf8-encoded-payload', 'Decoded utf8 payload must match sent Buffer');

    clientWs.close();
    await closeServer(server);
  }

  // 5. Integration with stream pipelining and collectStream helper
  {
    const { server, port } = await createLocalServer();

    const serverCollectedPromise = new Promise((resolve, reject) => {
      server.once('connection', (serverWs) => {
        const serverStream = createWebSocketStream(serverWs);
        collectStream(serverStream).then(resolve, reject);
      });
    });

    const { clientWs } = await connectClient(port);
    const sourceStream = Readable.from(['pipe-part-1', '-', 'pipe-part-2']);

    await pipeToWebSocket(clientWs, sourceStream);

    const collectedBuffer = await serverCollectedPromise;
    assert.strictEqual(
      collectedBuffer.toString('utf8'),
      'pipe-part-1-pipe-part-2',
      'Piping readable stream into WebSocket duplex stream must transmit entire stream contents'
    );

    clientWs.close();
    await closeServer(server);
  }

  // 6. Graceful close and EOF signaling on stream end
  {
    const { server, port } = await createLocalServer();

    const serverEndPromise = new Promise((resolve) => {
      server.once('connection', (serverWs) => {
        const serverStream = createWebSocketStream(serverWs);
        serverStream.on('data', () => {});
        serverStream.once('end', () => {
          resolve(true);
        });
      });
    });

    const { clientWs, clientStream } = await connectClient(port);

    clientStream.write('final-data');
    clientStream.end();

    const endEmitted = await serverEndPromise;
    assert.strictEqual(endEmitted, true, 'Stream end must trigger EOF end event on receiving peer');

    clientWs.close();
    await closeServer(server);
  }

  // 7. Stream destroy terminates the underlying WebSocket
  {
    const { server, port } = await createLocalServer();

    const serverClosedPromise = new Promise((resolve) => {
      server.once('connection', (serverWs) => {
        serverWs.once('close', () => {
          resolve(true);
        });
      });
    });

    const { clientWs, clientStream } = await connectClient(port);
    assert.strictEqual(clientWs.readyState, WebSocket.OPEN, 'WebSocket must be in OPEN state');

    clientStream.destroy();
    assert.strictEqual(clientStream.destroyed, true, 'Duplex stream must be destroyed');

    const serverClosed = await serverClosedPromise;
    assert.strictEqual(serverClosed, true, 'Destroying client stream must terminate the WebSocket connection');

    await closeServer(server);
  }

  console.log('Contract tests passed successfully.');
}

runContractTests().catch((err) => {
  console.error('Contract test failed:', err);
  process.exit(1);
});