ext/node/polyfills/internal/js_stream_socket.js - main - githubmirror/deno

ext/node/polyfills/internal/js_stream_socket.js
333 строки10 KB
Kenta Moriuchi
chore: make prefer-primordials lint an internal plugin (#35978)
23 июл 2026, 11:53
23 июл 2026, 11:535b55d08
// Copyright 2018-2026 the Deno authors. MIT license.
// Copyright Joyent and Node contributors. All rights reserved. MIT license.
//
// Ported from Node.js lib/internal/js_stream_socket.js

(function () {
const { core, primordials } = __bootstrap;
const {
  Array,
  FunctionPrototypeCall,
  MapPrototypeGet,
  Symbol,
  SymbolFor,
} = primordials;

const lazyNet = core.createLazyLoader("node:net");
const lazyTimers = core.createLazyLoader("node:timers");

const { nextTick } = core.loadExtScript("ext:deno_node/_next_tick.ts");
const { codeMap, UV_ECANCELED } = core.loadExtScript(
  "ext:deno_node/internal_binding/uv.ts",
);
const {
  kArrayBufferOffset,
  kBytesWritten,
  kLastWriteWasAsync,
  kReadBytesOrError,
  streamBaseState,
} = core.loadExtScript("ext:deno_node/internal_binding/stream_wrap.ts");
const { Buffer } = core.loadExtScript("ext:deno_node/internal/buffer.mjs");
const { ERR_STREAM_WRAP } = core.loadExtScript(
  "ext:deno_node/internal/errors.ts",
);

const kCurrentWriteRequest = Symbol("kCurrentWriteRequest");
const kCurrentShutdownRequest = Symbol("kCurrentShutdownRequest");
const kPendingShutdownRequest = Symbol("kPendingShutdownRequest");
const kPendingClose = Symbol("kPendingClose");

// Mark JS stream handles so TLSWrap can detect them.
// Use Symbol.for so tls_wrap.ts can access it without importing
// (avoids circular dependency).
const kJSStreamHandle = SymbolFor("kJSStreamHandle");

function isClosing() {
  return this[kOwner].isClosing();
}
function onreadstart() {
  return this[kOwner].readStart();
}
function onreadstop() {
  return this[kOwner].readStop();
}

const kOwner = SymbolFor("kJSStreamOwner");

/* This class serves as a wrapper for when the Rust TLS layer wants access
 * to a standard JS stream. For example, TLS or HTTP2 do not operate on
 * network resources conceptually, although that is the common case; in
 * theory, they are completely composable and can work with any stream
 * resource they see.
 *
 * For the common case, i.e. a TLS socket wrapping around a net.Socket, we
 * can skip going through the JS layer and let TLS access the raw native
 * handle of a net.Socket. The flipside of this is that, to maintain
 * composability, we need a way to create "fake" net.Socket instances that
 * call back into a "real" JavaScript stream. JSStreamSocket is exactly this.
 */
class JSStreamSocket extends lazyNet().Socket {
  constructor(stream) {
    // Create a lightweight handle object that mimics what Node's
    // JSStream C++ binding provides. TLSWrap detects kJSStreamHandle
    // and uses attachJsStream() instead of attach().
    const handle = {
      [kJSStreamHandle]: true,
      [kOwner]: null, // set below
      close(cb) {
        handle[kOwner].doClose(cb);
      },
      isClosing,
      // Callbacks invoked by the owner (JSStreamSocket) methods
      onreadstart,
      onreadstop,
      // Methods called by net.Socket on the handle - delegate to owner
      readStart() {
        return handle[kOwner].readStart();
      },
      readStop() {
        return handle[kOwner].readStop();
      },
      // Write methods - delegate to owner.doWrite which writes to the
      // underlying stream and triggers req.oncomplete via finishWrite.
      writeBuffer(req, data) {
        // deno-lint-ignore deno-internal/prefer-primordials
        const len = data.byteLength ?? data.length ?? 0;
        streamBaseState[kBytesWritten] = len;
        streamBaseState[kLastWriteWasAsync] = 1;
        return handle[kOwner].doWrite(req, [data]);
      },
      writev(req, chunks, allBuffers) {
        let bufs;
        let total = 0;
        if (allBuffers) {
          bufs = chunks;
          for (let i = 0; i < bufs.length; i++) {
            // deno-lint-ignore deno-internal/prefer-primordials
            total += bufs[i].byteLength ?? bufs[i].length ?? 0;
          }
        } else {
          bufs = new Array(chunks.length >> 1);
          for (let i = 0; i < chunks.length; i += 2) {
            const chunk = chunks[i];
            const enc = chunks[i + 1];
            const buf = typeof chunk === "string"
              ? Buffer.from(chunk, enc)
              : chunk;
            bufs[i >> 1] = buf;
            // deno-lint-ignore deno-internal/prefer-primordials
            total += buf.byteLength ?? buf.length ?? 0;
          }
        }
        streamBaseState[kBytesWritten] = total;
        streamBaseState[kLastWriteWasAsync] = 1;
        return handle[kOwner].doWrite(req, bufs);
      },
      writeAsciiString(req, data) {
        return this.writeBuffer(req, Buffer.from(data, "ascii"));
      },
      writeUtf8String(req, data) {
        return this.writeBuffer(req, Buffer.from(data, "utf8"));
      },
      writeLatin1String(req, data) {
        return this.writeBuffer(req, Buffer.from(data, "latin1"));
      },
      writeUcs2String(req, data) {
        return this.writeBuffer(req, Buffer.from(data, "utf16le"));
      },
      shutdown(req) {
        return handle[kOwner].doShutdown(req);
      },
      // Default read path: forward bytes from the underlying Duplex into
      // the wrapping Socket's readable side via the standard onread
      // callback (set by net._initSocketHandle).
      // TLSWrap.attachJsStream() overrides these to route data through
      // the TLS engine instead.
      readBuffer(chunk) {
        if (!handle.onread) return;
        // deno-lint-ignore deno-internal/prefer-primordials
        const len = chunk.byteLength ?? chunk.length ?? 0;
        if (len === 0) return;
        // deno-lint-ignore deno-internal/prefer-primordials
        streamBaseState[kArrayBufferOffset] = chunk.byteOffset ?? 0;
        streamBaseState[kReadBytesOrError] = len;
        FunctionPrototypeCall(handle.onread, handle, chunk, len);
      },
      emitEOF() {
        if (!handle.onread) return;
        FunctionPrototypeCall(
          handle.onread,
          handle,
          null,
          MapPrototypeGet(codeMap, "EOF"),
        );
      },
      reading: false,
    };

    stream.pause();
    stream.on("error", (err) => this.emit("error", err));
    const ondata = (chunk) => {
      if (
        typeof chunk === "string" ||
        stream.readableObjectMode === true
      ) {
        // Make sure that no further `data` events will happen.
        stream.pause();
        stream.removeListener("data", ondata);
        this.emit("error", new ERR_STREAM_WRAP());
        return;
      }

      if (this._handle && this._handle.readBuffer) {
        this._handle.readBuffer(chunk);
      }
    };
    stream.on("data", ondata);
    stream.once("end", () => {
      if (this._handle && this._handle.emitEOF) {
        this._handle.emitEOF();
      }
    });
    // Some `Stream` don't pass `hasError` parameters when closed.
    stream.once("close", () => {
      this.destroy();
    });

    super({ handle, manualStart: true });
    handle[kOwner] = this;
    this.stream = stream;
    this[kCurrentWriteRequest] = null;
    this[kCurrentShutdownRequest] = null;
    this[kPendingShutdownRequest] = null;
    this[kPendingClose] = false;
    this.readable = stream.readable;
    this.writable = stream.writable;

    // Start reading.
    this.read(0);
  }

  isClosing() {
    return !this.readable || !this.writable;
  }

  readStart() {
    this.stream.resume();
    return 0;
  }

  readStop() {
    this.stream.pause();
    return 0;
  }

  doShutdown(req) {
    if (this[kCurrentWriteRequest] !== null) {
      this[kPendingShutdownRequest] = req;
      return 0;
    }

    this[kCurrentShutdownRequest] = req;

    if (this[kPendingClose]) {
      return 0;
    }

    const handle = this._handle;

    nextTick(() => {
      this.stream.end(() => {
        this.finishShutdown(handle, 0);
      });
    });
    return 0;
  }

  finishShutdown(_handle, errCode) {
    const req = this[kCurrentShutdownRequest];
    if (req === null) return;
    this[kCurrentShutdownRequest] = null;
    if (typeof req.oncomplete === "function") {
      req.oncomplete(errCode | 0);
    }
  }

  doWrite(req, bufs) {
    if (this[kPendingClose]) {
      this[kCurrentWriteRequest] = req;
      return 0;
    } else if (this._handle === null) {
      return 0;
    }

    const handle = this._handle;
    // deno-lint-ignore no-this-alias
    const self = this;

    let pending = bufs.length;

    this.stream.cork();
    for (let i = 0; i < bufs.length; ++i) {
      this.stream.write(bufs[i], done);
    }
    this.stream.uncork();

    this[kCurrentWriteRequest] = req;

    function done(err) {
      if (!err && --pending !== 0) return;
      pending = 0;

      let errCode = 0;
      if (err) {
        errCode = MapPrototypeGet(codeMap, err.code) ||
          MapPrototypeGet(codeMap, "EPIPE");
      }

      lazyTimers().setImmediate(() => {
        self.finishWrite(handle, errCode);
      });
    }

    return 0;
  }

  finishWrite(_handle, errCode) {
    const req = this[kCurrentWriteRequest];
    if (req === null) return;
    this[kCurrentWriteRequest] = null;
    if (typeof req.oncomplete === "function") {
      req.oncomplete(errCode | 0);
    }

    if (this[kPendingShutdownRequest]) {
      const sreq = this[kPendingShutdownRequest];
      this[kPendingShutdownRequest] = null;
      this.doShutdown(sreq);
    }
  }

  doClose(cb) {
    this[kPendingClose] = true;
    const handle = this._handle;

    this.stream.destroy();

    lazyTimers().setImmediate(() => {
      this.finishWrite(handle, UV_ECANCELED);
      this.finishShutdown(handle, UV_ECANCELED);
      this[kPendingClose] = false;
      if (cb) cb();
    });
  }
}

// Node's lib/internal/js_stream_socket exports the class as the module
// itself, with `StreamWrap` as a self-reference so destructuring works:
//   const StreamWrap = require('internal/js_stream_socket');
//   const { StreamWrap } = require('internal/js_stream_socket');
JSStreamSocket.StreamWrap = JSStreamSocket;

return { JSStreamSocket, kJSStreamHandle, kOwner, default: JSStreamSocket };
})();