#209 bounty implementation

Both Ring based and LL Deque implementations of the same queue strategy to run the p2p FSM and

A/B Results:
A (Queue-based original): 19,909ms
B (Circular buffer): 20,984ms
Circular buffer is 5.4% slower. The queue-based approach with .shift() and .slice() is already optimized by V8. The circular buffer adds modulo arithmetic overhead and buffer resizing complexity that doesn't pay off for this workload. Recommendation: Keep the original queue-based implementation.

Author: Your Name

__tests__/framecodec.test.js

@@ -0,0 +1,156 @@
+const { FrameEncoder, FrameDecoder, encodeFrame } = require('../src/FrameCodec');
+const { once } = require('events');
+const crypto = require('crypto');
+
+let _seed = 0xdeadbeef;
+function seededRandom() {
+  _seed = (_seed * 1664525 + 1013904223) >>> 0;
+  return _seed / 0x100000000;
+}
+
+function chunkBuffer(buf, cuts) {
+  const out = [];
+  let offset = 0;
+  for (const len of cuts) {
+    const end = Math.min(buf.length, offset + len);
+    out.push(buf.slice(offset, end));
+    offset = end;
+    if (offset >= buf.length) break;
+  }
+  if (offset < buf.length) out.push(buf.slice(offset));
+  return out;
+}
+
+test('reconstructs frames from arbitrary chunks', (done) => {
+  const encoder = new FrameEncoder();
+  const decoder = new FrameDecoder();
+
+  const messages = [];
+  for (let i = 0; i < 50; i++) {
+    const len = Math.floor(Math.random() * 200);
+    const payload = Buffer.alloc(len, i % 256);
+    messages.push(payload);
+  }
+
+  const wire = Buffer.concat(messages.map(m => encodeFrame(m)));
+
+  // create random chunk sizes that simulate fragmentation
+  const cuts = [];
+  let remaining = wire.length;
+  while (remaining > 0) {
+    const c = Math.max(1, Math.floor(Math.random() * 20));
+    cuts.push(c);
+    remaining -= c;
+  }
+
+  const chunks = chunkBuffer(wire, cuts);
+
+  const out = [];
+  decoder.on('data', (frame) => out.push(frame));
+  decoder.on('end', () => {
+    try {
+      expect(out.length).toBe(messages.length);
+      for (let i = 0; i < messages.length; i++) {
+        expect(out[i]).toEqual(messages[i]);
+      }
+      done();
+    } catch (err) {
+      done(err);
+    }
+  });
+
+  // feed fragments with a small async schedule to simulate streaming
+  (async () => {
+    for (const c of chunks) {
+      decoder.write(c);
+      // occasionally wait
+      if (Math.random() < 0.1) await new Promise(r => setTimeout(r, 0));
+    }
+    decoder.end();
+  })();
+});
+
+function randomInt(min, max) {
+  return Math.floor(seededRandom() * (max - min + 1)) + min;
+}
+
+function randomFramePayload(length) {
+  return crypto.randomBytes(length);
+}
+
+async function decodeWithFrameDecoder(chunks) {
+  const decoder = new FrameDecoder();
+  const frames = [];
+
+  decoder.on('data', (frame) => {
+    // Ensure we never mutate the emitted Buffer instance
+    frames.push(Buffer.from(frame));
+  });
+
+  const errorPromise = once(decoder, 'error').then(([err]) => {
+    throw err;
+  });
+
+  const endPromise = once(decoder, 'end').then(() => frames);
+
+  (async () => {
+    for (const chunk of chunks) {
+      if (!decoder.write(chunk)) {
+        await once(decoder, 'drain');
+      }
+    }
+    decoder.end();
+  })().catch((err) => decoder.emit('error', err));
+
+  return Promise.race([endPromise, errorPromise]);
+}
+
+test('fuzzes frame decoding with heavy fragmentation', async () => {
+  const warmup = 1000;
+  const iterations = 1000000;
+
+  for (let i = 0; i < warmup; i++) {
+    const frameCount = randomInt(1, 12);
+    const frames = [];
+    for (let j = 0; j < frameCount; j++) frames.push(randomFramePayload(randomInt(0, 512)));
+    const wire = Buffer.concat(frames.map((frame) => encodeFrame(frame)));
+    const chunks = [];
+    let offset = 0;
+    while (offset < wire.length) {
+      const chunkLen = randomInt(1, Math.min(64, wire.length - offset));
+      chunks.push(wire.slice(offset, offset + chunkLen));
+      offset += chunkLen;
+    }
+    await decodeWithFrameDecoder(chunks);
+  }
+
+  const start = Date.now();
+  for (let i = 0; i < iterations; i++) {
+    const frameCount = randomInt(1, 12);
+    const frames = [];
+
+    for (let j = 0; j < frameCount; j++) {
+      const payloadLength = randomInt(0, 512);
+      frames.push(randomFramePayload(payloadLength));
+    }
+
+    const wire = Buffer.concat(frames.map((frame) => encodeFrame(frame)));
+
+    const chunks = [];
+    let offset = 0;
+    while (offset < wire.length) {
+      const chunkLen = randomInt(1, Math.min(64, wire.length - offset));
+      chunks.push(wire.slice(offset, offset + chunkLen));
+      offset += chunkLen;
+    }
+
+    const decoded = await decodeWithFrameDecoder(chunks);
+
+    expect(decoded.length).toBe(frames.length);
+    for (let k = 0; k < frames.length; k++) {
+      expect(decoded[k]).toEqual(frames[k]);
+    }
+  }
+
+  console.log(`1M rounds: ${Date.now() - start}ms`);
+});

bench-ab.js

@@ -0,0 +1,108 @@
+const { FrameDecoder, encodeFrame } = require('./src/FrameCodec');
+const { FrameDecoderCirc } = require('./src/FrameCodecCirc');
+const { once } = require('events');
+const crypto = require('crypto');
+
+let _seed = 0xdeadbeef;
+function seededRandom() {
+  _seed = (_seed * 1664525 + 1013904223) >>> 0;
+  return _seed / 0x100000000;
+}
+
+function randomInt(min, max) {
+  return Math.floor(seededRandom() * (max - min + 1)) + min;
+}
+
+function randomFramePayload(length) {
+  return crypto.randomBytes(length);
+}
+
+async function decodeWithFrameDecoder(chunks, DecoderClass) {
+  const decoder = new DecoderClass();
+  const frames = [];
+
+  decoder.on('data', (frame) => {
+    frames.push(Buffer.from(frame));
+  });
+
+  const errorPromise = once(decoder, 'error').then(([err]) => {
+    throw err;
+  });
+
+  const endPromise = once(decoder, 'end').then(() => frames);
+
+  (async () => {
+    for (const chunk of chunks) {
+      if (!decoder.write(chunk)) {
+        await once(decoder, 'drain');
+      }
+    }
+    decoder.end();
+  })().catch((err) => decoder.emit('error', err));
+
+  return Promise.race([endPromise, errorPromise]);
+}
+
+async function runBench(DecoderClass, name) {
+  const warmup = 1000;
+  const iterations = 1000000;
+
+  for (let i = 0; i < warmup; i++) {
+    const frameCount = randomInt(1, 12);
+    const frames = [];
+    for (let j = 0; j < frameCount; j++) frames.push(randomFramePayload(randomInt(0, 512)));
+    const wire = Buffer.concat(frames.map((frame) => encodeFrame(frame)));
+    const chunks = [];
+    let offset = 0;
+    while (offset < wire.length) {
+      const chunkLen = randomInt(1, Math.min(64, wire.length - offset));
+      chunks.push(wire.slice(offset, offset + chunkLen));
+      offset += chunkLen;
+    }
+    await decodeWithFrameDecoder(chunks, DecoderClass);
+  }
+
+  const start = Date.now();
+  for (let i = 0; i < iterations; i++) {
+    const frameCount = randomInt(1, 12);
+    const frames = [];
+
+    for (let j = 0; j < frameCount; j++) {
+      const payloadLength = randomInt(0, 512);
+      frames.push(randomFramePayload(payloadLength));
+    }
+
+    const wire = Buffer.concat(frames.map((frame) => encodeFrame(frame)));
+
+    const chunks = [];
+    let offset = 0;
+    while (offset < wire.length) {
+      const chunkLen = randomInt(1, Math.min(64, wire.length - offset));
+      chunks.push(wire.slice(offset, offset + chunkLen));
+      offset += chunkLen;
+    }
+
+    const decoded = await decodeWithFrameDecoder(chunks, DecoderClass);
+
+    if (decoded.length !== frames.length) throw new Error('length mismatch');
+    for (let k = 0; k < frames.length; k++) {
+      if (!decoded[k].equals(frames[k])) throw new Error('frame mismatch');
+    }
+  }
+
+  const elapsed = Date.now() - start;
+  console.log(`${name}: ${elapsed}ms`);
+  return elapsed;
+}
+
+(async () => {
+  console.log('A/B Benchmark: FrameDecoder vs FrameDecoderCirc\n');
+
+  _seed = 0xdeadbeef;
+  const timeA = await runBench(FrameDecoder, 'A: Queue-based (original)');
+
+  _seed = 0xdeadbeef;
+  const timeB = await runBench(FrameDecoderCirc, 'B: Circular buffer');
+
+  console.log(`\nDelta: ${timeB - timeA > 0 ? '+' : ''}${timeB - timeA}ms (${((timeB / timeA - 1) * 100).toFixed(1)}%)`);
+})();

examples/tcpsockets/examples/libp2p-framing.js

@@ -0,0 +1,115 @@
+const net = require('react-native-tcp-socket');
+const { createLibp2pStream, FrameEncoder, FrameDecoder } = require('react-native-tcp-socket');
+
+/**
+ * Example 1: Using createLibp2pStream for async iterator compatibility
+ * This is the recommended approach for libp2p integration
+ */
+function exampleLibp2pStream() {
+  const socket = net.createConnection({ host: '127.0.0.1', port: 4001 }, () => {
+    console.log('Connected to libp2p peer');
+  });
+
+  // Create a libp2p-compatible stream with source/sink pattern
+  const stream = createLibp2pStream(socket);
+
+  // Consume frames via async iterator
+  (async () => {
+    for await (const frame of stream) {
+      console.log('Received frame:', frame.length, 'bytes');
+      // Handle libp2p protocol frames here
+    }
+  })();
+
+  // Send frames via sink
+  (async () => {
+    await stream.sink((async function* () {
+      // libp2p multistream-select protocol
+      yield Buffer.from('/multistream/1.0.0\n');
+      yield Buffer.from('/libp2p/circuit/relay/0.2.0/hop\n');
+    })());
+  })();
+
+  socket.on('error', (err) => {
+    console.error('Socket error:', err);
+  });
+
+  socket.on('close', () => {
+    console.log('Connection closed');
+  });
+}
+
+/**
+ * Example 2: Using FrameEncoder/FrameDecoder directly with streams
+ * This approach gives you more control over the framing process
+ */
+function exampleDirectFraming() {
+  const socket = net.createConnection({ host: '127.0.0.1', port: 4001 }, () => {
+    console.log('Connected using direct framing');
+  });
+
+  const encoder = new FrameEncoder();
+  const decoder = new FrameDecoder();
+
+  // Pipe socket through decoder/encoder
+  socket.pipe(decoder);
+  encoder.pipe(socket);
+
+  // Handle incoming frames
+  decoder.on('data', (frame) => {
+    console.log('Received frame:', frame.length, 'bytes');
+    // Process libp2p protocol messages
+  });
+
+  // Send frames
+  encoder.write(Buffer.from('/multistream/1.0.0\n'));
+  encoder.write(Buffer.from('/libp2p/circuit/relay/0.2.0/hop\n'));
+
+  socket.on('error', (err) => {
+    console.error('Socket error:', err);
+  });
+
+  socket.on('close', () => {
+    console.log('Connection closed');
+    encoder.end();
+  });
+}
+
+/**
+ * Example 3: Server-side libp2p framing
+ * Shows how to handle framed connections on the server side
+ */
+function exampleServer() {
+  const server = net.createServer((socket) => {
+    console.log('Client connected');
+
+    const encoder = new FrameEncoder();
+    const decoder = new FrameDecoder();
+
+    socket.pipe(decoder);
+    encoder.pipe(socket);
+
+    decoder.on('data', (frame) => {
+      console.log('Server received frame:', frame.toString());
+      // Echo back the frame
+      encoder.write(frame);
+    });
+
+    socket.on('error', (err) => {
+      console.error('Server socket error:', err);
+    });
+
+    socket.on('close', () => {
+      console.log('Client disconnected');
+    });
+  });
+
+  server.listen(4002, '127.0.0.1', () => {
+    console.log('Server listening on port 4002');
+  });
+}
+
+// Uncomment to run the examples:
+// exampleLibp2pStream();
+// exampleDirectFraming();
+// exampleServer();