BridgeSwarm/extension/framed-stream.js

/**
 * Framed stream adapter for BridgeSwarm connections.
 * Presents a stream interface compatible with Protomux: message boundaries are preserved
 * (each chunk from the native host is one decrypted frame from NoiseSecretStream).
 * Buffers and parses 3-byte little-endian length-prefixed frames in case the bridge
 * ever delivers coalesced or split frames.
 */
(function (global) {
  var FRAME_HEADER = 3; // 3-byte LE length prefix (same as @hyperswarm/secret-stream)

  function readUint24LE(buf, offset) {
    return (buf[offset] | (buf[offset + 1] << 8) | (buf[offset + 2] << 16)) >>> 0;
  }

  function writeUint24LE(n, buf, offset) {
    buf[offset] = n & 255;
    buf[offset + 1] = (n >>> 8) & 255;
    buf[offset + 2] = (n >>> 16) & 255;
  }

  function EventEmitter() {
    this._listeners = {};
  }
  EventEmitter.prototype.on = function (ev, fn) {
    if (!this._listeners[ev]) this._listeners[ev] = [];
    this._listeners[ev].push(fn);
    return this;
  };
  EventEmitter.prototype.off = function (ev, fn) {
    if (!this._listeners[ev]) return this;
    this._listeners[ev] = this._listeners[ev].filter(function (f) { return f !== fn; });
    return this;
  };
  EventEmitter.prototype.emit = function (ev) {
    var args = Array.prototype.slice.call(arguments, 1);
    (this._listeners[ev] || []).forEach(function (fn) {
      try { fn.apply(null, args); } catch (_) {}
    });
    return this;
  };

  /**
   * Wraps a BridgeSwarmConnection (or any duplex with write/on('data')/on('end')/destroy)
   * and exposes a Protomux-compatible stream: preserves message boundaries, and implements
   * .destroyed, .end(), .pause(), .resume(), 'drain', 'close'.
   * Incoming data is buffered and split by 3-byte LE length prefix; if the underlying
   * connection already delivers one chunk per frame (as with the native host), each chunk
   * is treated as one frame when it's not length-prefixed (length prefix is only used when
   * we have enough bytes to parse it and the chunk doesn't look like a raw frame).
   */
  function FramedStreamAdapter(conn) {
    EventEmitter.call(this);
    this.conn = conn;
    this.destroyed = false;
    this._paused = false;
    this._buffer = null;
    this._bufferOffset = 0;
    this._frameLength = 0;
    this._readingLength = true;
    this._boundOnData = this._onData.bind(this);
    this._boundOnEnd = this._onEnd.bind(this);
    this._boundOnError = this._onError.bind(this);

    var self = this;
    this.conn.on('data', this._boundOnData);
    this.conn.on('end', this._boundOnEnd);
    this.conn.on('error', this._boundOnError);
  }

  FramedStreamAdapter.prototype = Object.create(EventEmitter.prototype);

  FramedStreamAdapter.prototype._onData = function (chunk) {
    if (this.destroyed || this._paused) return;
    var buf = chunk instanceof Uint8Array ? chunk : new Uint8Array(chunk);
    if (buf.length === 0) return;

    if (!this._buffer) {
      this._buffer = buf;
      this._bufferOffset = 0;
    } else {
      var total = this._bufferOffset + (this._buffer.length - this._bufferOffset) + buf.length;
      var combined = new Uint8Array(total);
      combined.set(this._buffer.subarray(this._bufferOffset));
      combined.set(buf, this._buffer.length - this._bufferOffset);
      this._buffer = combined;
      this._bufferOffset = 0;
    }

    while (this._bufferOffset < this._buffer.length && !this.destroyed) {
      if (this._readingLength) {
        if (this._buffer.length - this._bufferOffset < FRAME_HEADER) break;
        this._frameLength = readUint24LE(this._buffer, this._bufferOffset);
        this._bufferOffset += FRAME_HEADER;
        this._readingLength = false;
      }
      if (this._buffer.length - this._bufferOffset < this._frameLength) break;
      var frame = this._buffer.slice(this._bufferOffset, this._bufferOffset + this._frameLength);
      this._bufferOffset += this._frameLength;
      this._readingLength = true;
      this._frameLength = 0;
      this.emit('data', frame);
    }

    if (this._bufferOffset >= this._buffer.length) {
      this._buffer = null;
      this._bufferOffset = 0;
    } else if (this._bufferOffset > 0) {
      this._buffer = this._buffer.subarray(this._bufferOffset);
      this._bufferOffset = 0;
    }
  };

  FramedStreamAdapter.prototype._onEnd = function () {
    if (this.destroyed) return;
    this.emit('end');
  };

  FramedStreamAdapter.prototype._onError = function (err) {
    if (this.destroyed) return;
    this.emit('error', err);
  };

  FramedStreamAdapter.prototype.write = function (data) {
    if (this.destroyed) return false;
    var buf = data instanceof Uint8Array ? data : new Uint8Array(data);
    this.conn.write(buf);
    var self = this;
    if (typeof setImmediate !== 'undefined') {
      setImmediate(function () { self.emit('drain'); });
    } else {
      setTimeout(function () { self.emit('drain'); }, 0);
    }
    return true;
  };

  FramedStreamAdapter.prototype.end = function () {
    if (this.destroyed) return;
    this.conn.destroy();
  };

  FramedStreamAdapter.prototype.pause = function () {
    this._paused = true;
  };

  FramedStreamAdapter.prototype.resume = function () {
    this._paused = false;
  };

  FramedStreamAdapter.prototype.destroy = function (err) {
    if (this.destroyed) return;
    this.destroyed = true;
    this.conn.off('data', this._boundOnData);
    this.conn.off('end', this._boundOnEnd);
    this.conn.off('error', this._boundOnError);
    this.conn.destroy();
    if (err) this.emit('error', err);
    this.emit('close');
  };

  /**
   * Wrap a connection that delivers raw frames (one chunk = one frame, no length prefix).
   * Use this when the native host sends one event per NoiseSecretStream message.
   */
  function wrapRawFrames(conn) {
    var stream = new EventEmitter();
    stream.conn = conn;
    stream.destroyed = false;
    stream._paused = false;
    stream.write = function (data) {
      if (stream.destroyed) return false;
      var buf = data instanceof Uint8Array ? data : new Uint8Array(data);
      conn.write(buf);
      var s = stream;
      setTimeout(function () { s.emit('drain'); }, 0);
      return true;
    };
    stream.end = function () {
      if (stream.destroyed) return;
      conn.destroy();
    };
    stream.pause = function () { stream._paused = true; };
    stream.resume = function () { stream._paused = false; };
    stream.destroy = function (err) {
      if (stream.destroyed) return;
      stream.destroyed = true;
      conn.destroy();
      if (err) stream.emit('error', err);
      stream.emit('close');
    };
    conn.on('data', function (chunk) {
      if (!stream._paused && !stream.destroyed) stream.emit('data', chunk);
    });
    conn.on('end', function () {
      if (stream.destroyed) return;
      stream.emit('end');
      stream.emit('close');
    });
    conn.on('error', function (e) { if (!stream.destroyed) stream.emit('error', e); });
    return stream;
  }

  /**
   * Wrap a connection with length-prefixed framing (3-byte LE).
   * Use when you need to send/receive framed messages; outgoing writes get a length prefix.
   */
  function wrapLengthPrefixed(conn) {
    return new FramedStreamAdapter(conn);
  }

  if (typeof module !== 'undefined' && module.exports) {
    module.exports = { FramedStreamAdapter, wrapRawFrames, wrapLengthPrefixed };
  } else {
    global.BridgeSwarmFramedStream = {
      FramedStreamAdapter: FramedStreamAdapter,
      wrapRawFrames: wrapRawFrames,
      wrapLengthPrefixed: wrapLengthPrefixed
    };
  }
})(typeof window !== 'undefined' ? window : typeof self !== 'undefined' ? self : this);