Optimize FrameCodec.js with zero-copy varint encoding, drain handling, and reusable buffers

- Add varint.encodeTo() for zero-copy encoding into reusable buffer
- Implement waitForDrain() in FrameEncoder for backpressure handling
- Optimize _take() method with zero-copy fast path for single chunks
- Add detailed debug logging for performance analysis

Author: Your Name

src/FrameCodec.js

@@ -3,16 +3,22 @@ const { Transform } = require('stream');
 const DEBUG_FRAME_DECODER = process.env.FRAME_DECODER_DEBUG === '1';
 
 const varint = {
-  encode: (n) => {
+  // Write varint-encoded `n` into `target` at `offset`. Returns number of bytes written.
+  encodeTo: (target, offset, n) => {
     if (n < 0) throw new RangeError('varint unsigned only');
-    const o = [];
+    let i = 0;
     do {
       let b = n & 0x7f;
       n = Math.floor(n / 128);
       if (n > 0) b |= 0x80;
-      o.push(b);
+      target[offset + (i++)] = b;
     } while (n > 0);
-    return Buffer.from(o);
+    return i;
+  },
+  encode: (n) => {
+    const buf = Buffer.allocUnsafe(10);
+    const len = varint.encodeTo(buf, 0, n);
+    return buf.slice(0, len);
   },
   decodeFrom: (buf, offset = 0) => {
     let r = 0, s = 0, i = offset;
@@ -30,11 +36,35 @@ const varint = {
 class FrameEncoder extends Transform {
   constructor() {
     super({ writableObjectMode: true });
+    let drainDeferred = null;
+    // per-instance varint buffer to avoid allocating a small header Buffer per frame
+    this._varintBuf = Buffer.allocUnsafe(10);
+    this.waitForDrain = () => {
+      if (!drainDeferred) {
+        drainDeferred = {};
+        drainDeferred.promise = new Promise(resolve => {
+          drainDeferred.resolve = resolve;
+        });
+        this.once('drain', () => {
+          if (drainDeferred) {
+            drainDeferred.resolve();
+            drainDeferred = null;
+          }
+        });
+      }
+      return drainDeferred.promise;
+    };
   }
   _transform(f, e, cb) {
     try {
       if (!Buffer.isBuffer(f)) f = Buffer.from(f);
-      this.push(Buffer.concat([varint.encode(f.length), f]));
+      // encode varint header into reusable buffer then copy into final frame
+      const payloadLen = f.length;
+      const hdrLen = varint.encodeTo(this._varintBuf, 0, payloadLen);
+      const frame = Buffer.allocUnsafe(hdrLen + payloadLen);
+      this._varintBuf.copy(frame, 0, 0, hdrLen);
+      f.copy(frame, hdrLen);
+      this.push(frame);
       cb();
     } catch (err) {
       cb(err);
@@ -129,9 +159,25 @@ class FrameDecoder extends Transform {
   }
 
   _take(n, label = 'bytes') {
+    this._log('take_start', { label, bytes: n, buffered: this._l });
+
+    // Zero-copy fast path: single chunk contains all needed bytes
+    if (this._q.length > 0 && this._q[0].length >= n) {
+      const head = this._q[0];
+      const slice = head.slice(0, n);
+      this._l -= n;
+      if (n === head.length) {
+        this._q.shift();
+      } else {
+        this._q[0] = head.slice(n);
+      }
+      this._log('take_complete', { label, bytes: n, buffered: this._l, zeroCopy: true });
+      return slice;
+    }
+
+    // Multi-chunk path: allocate and copy
     const f = Buffer.allocUnsafe(n);
     let w = 0;
-    this._log('take_start', { label, bytes: n, buffered: this._l });
     while (w < n && this._q.length > 0) {
       const next = this._q[0];
       const t = Math.min(next.length, n - w);
@@ -145,7 +191,7 @@ class FrameDecoder extends Transform {
       }
       this._log('take_progress', { label, copied: t, written: w, buffered: this._l });
     }
-    this._log('take_complete', { label, bytes: n, buffered: this._l });
+    this._log('take_complete', { label, bytes: n, buffered: this._l, zeroCopy: false });
     return f;
   }
 }
@@ -155,9 +201,31 @@ FrameDecoder._nextId = 1;
 function createLibp2pStream(socket) {
   const d = new FrameDecoder(), e = new FrameEncoder();
   socket.pipe(d); e.pipe(socket);
-  const s = { source: (async function* () { for await (const c of d) yield c; })(), sink: async (src) => { for await (const c of src) { if (!e.write(c)) await new Promise(r => e.once('drain', r)); } e.end(); } };
+  const s = {
+    source: (async function* () { for await (const c of d) yield c; })(),
+    sink: async (src) => {
+      for await (const c of src) {
+        if (!e.write(c)) await e.waitForDrain();
+      }
+      e.end();
+    }
+  };
   s[Symbol.asyncIterator] = () => s.source[Symbol.asyncIterator]();
   return s;
 }
 
-module.exports = { FrameEncoder, FrameDecoder, createLibp2pStream, encodeFrame: (b) => { const buf = Buffer.isBuffer(b) ? b : Buffer.from(b); return Buffer.concat([varint.encode(buf.length), buf]); } };
+module.exports = {
+  FrameEncoder,
+  FrameDecoder,
+  createLibp2pStream,
+  encodeFrame: (b) => {
+    const buf = Buffer.isBuffer(b) ? b : Buffer.from(b);
+    // Avoid Buffer.concat by preallocating exact size and writing header then payload
+    const tmp = Buffer.allocUnsafe(10);
+    const hdrLen = varint.encodeTo(tmp, 0, buf.length);
+    const out = Buffer.allocUnsafe(hdrLen + buf.length);
+    tmp.copy(out, 0, 0, hdrLen);
+    buf.copy(out, hdrLen);
+    return out;
+  }
+};

src/Socket.js

@@ -1,16 +1,35 @@
 'use strict';
 
 import { NativeModules } from 'react-native';
+/**
+ * Provide a local JSDoc alias for stream.Transform so the JS file's type
+ * annotations don't require @types/node at the project root. This keeps the
+ * IntelliSense/types consistent while avoiding a hard dependency on Node types
+ * for React Native consumers.
+ * @typedef {import('stream').Transform} _Transform
+ */
 import EventEmitter from 'eventemitter3';
 import { Buffer } from 'buffer';
 const Sockets = NativeModules.TcpSockets;
 import { nativeEventEmitter, getNextId } from './Globals';
+import { FrameEncoder, FrameDecoder } from './FrameCodec';
 
 /**
  * @typedef {"ascii" | "utf8" | "utf-8" | "utf16le" | "ucs2" | "ucs-2" | "base64" | "latin1" | "binary" | "hex"} BufferEncoding
  *
  * @typedef {import('react-native').NativeEventEmitter} NativeEventEmitter
  *
+ * Minimal interface for the frame encoder/decoder streams used in this
+ * module. We declare only the members the Socket uses so the JSDoc types
+ * don't require @types/node.
+ * @typedef {{
+ *   write: (chunk: Buffer | Uint8Array) => boolean;
+ *   end: () => void;
+ *   once: (event: string, cb: (...args: any[]) => void) => void;
+ *   on: (event: string, cb: (...args: any[]) => void) => void;
+ *   pipe?: (dest: any) => any;
+ * }} FrameStream
+ *
  * @typedef {{address: string, family: string, port: number}} AddressInfo
  *
  * @typedef {{localAddress: string, localPort: number, remoteAddress: string, remotePort: number, remoteFamily: string}} NativeConnectionInfo
@@ -25,6 +44,7 @@ import { nativeEventEmitter, getNextId } from './Globals';
  * tls?: boolean,
  * tlsCheckValidity?: boolean,
  * tlsCert?: any,
+ * frameMode?: boolean,
  * }} ConnectionOptions
  *
  * @typedef {object} ReadableEvents
@@ -39,6 +59,7 @@ import { nativeEventEmitter, getNextId } from './Globals';
  * @property {(err: Error) => void} error
  * @property {() => void} timeout
  * @property {() => void} secureConnect
+ * @property {() => void} end
  *
  * @extends {EventEmitter<SocketEvents & ReadableEvents, any>}
  */
@@ -95,6 +116,12 @@ export default class Socket extends EventEmitter {
         this.remoteAddress = undefined;
         this.remotePort = undefined;
         this.remoteFamily = undefined;
+        /** @type {boolean} @private */
+        this._frameMode = false;
+    /** @type {any | null} @private */
+    this._frameEncoder = null;
+    /** @type {any | null} @private */
+    this._frameDecoder = null;
         this._registerEvents();
     }
 
@@ -159,6 +186,15 @@ export default class Socket extends EventEmitter {
         // Normalize args
         customOptions.host = customOptions.host || 'localhost';
         customOptions.port = Number(customOptions.port) || 0;
+
+        // Enable frame mode if requested
+        if (customOptions.frameMode) {
+            this._frameMode = true;
+            this._frameEncoder = /** @type {any} */ (new FrameEncoder());
+            this._frameDecoder = /** @type {any} */ (new FrameDecoder());
+            this._setupFrameCodec();
+        }
+
         this.once('connect', () => {
             if (callback) callback();
         });
@@ -309,6 +345,11 @@ export default class Socket extends EventEmitter {
         if (this._destroyed) return this;
         this._destroyed = true;
         this._clearTimeout();
+
+    // Clean up frame codec references
+    this._frameDecoder = null;
+    this._frameEncoder = null;
+
         Sockets.destroy(this._id);
         return this;
     }
@@ -351,7 +392,15 @@ export default class Socket extends EventEmitter {
     write(buffer, encoding, cb) {
         if (this._pending || this._destroyed) throw new Error('Socket is closed.');
 
-        const generatedBuffer = this._generateSendBuffer(buffer, encoding);
+        let generatedBuffer = this._generateSendBuffer(buffer, encoding);
+
+        // Apply frame encoding if in frame mode
+        if (this._frameMode) {
+            // Use varint encoding for libp2p compatibility
+            const varint = this._encodeVarint(generatedBuffer.byteLength);
+            generatedBuffer = Buffer.concat([varint, generatedBuffer]);
+        }
+
         this._writeBufferSize += generatedBuffer.byteLength;
         const currentMsgId = this._msgId;
         this._msgId = (this._msgId + 1) % Number.MAX_SAFE_INTEGER;
@@ -450,6 +499,22 @@ export default class Socket extends EventEmitter {
         Sockets.resume(this._id);
     }
 
+    /**
+     * @private
+     */
+    _setupFrameCodec() {
+        if (!this._frameDecoder || !this._frameEncoder) return;
+
+        // Wire up decoder to emit framed data
+        this._frameDecoder.on('data', (/** @type {Buffer} */ frame) => {
+            this.emit('data', this._encoding ? frame.toString(this._encoding) : frame);
+        });
+
+        this._frameDecoder.on('error', (/** @type {Error} */ err) => {
+            this.emit('error', err);
+        });
+    }
+
     /**
      * @private
      */
@@ -459,8 +524,14 @@ export default class Socket extends EventEmitter {
         if (!this._paused) {
             const bufferData = Buffer.from(evt.data, 'base64');
             this._bytesRead += bufferData.byteLength;
-            const finalData = this._encoding ? bufferData.toString(this._encoding) : bufferData;
-            this.emit('data', finalData);
+
+            if (this._frameMode && this._frameDecoder) {
+                // Feed raw data into frame decoder
+                this._frameDecoder.write(bufferData);
+            } else {
+                const finalData = this._encoding ? bufferData.toString(this._encoding) : bufferData;
+                this.emit('data', finalData);
+            }
         } else {
             // If the socket is paused, save the data events for later
             this._pausedDataEvents.push(evt);
@@ -525,6 +596,22 @@ export default class Socket extends EventEmitter {
         }
     }
 
+    /**
+     * @private
+     * @param {number} n
+     */
+    _encodeVarint(n) {
+        if (n < 0) throw new RangeError('varint unsigned only');
+        const o = [];
+        do {
+            let b = n & 0x7f;
+            n = Math.floor(n / 128);
+            if (n > 0) b |= 0x80;
+            o.push(b);
+        } while (n > 0);
+        return Buffer.from(o);
+    }
+
     /**
      * @private
      */