Updates

messaging-streams/hyper-p2p-stream-multiplex/index.js

@@ -1,37 +1,106 @@
 require('bare-process/global')
 const EventEmitter = require('bare-events')
+const b4a = require('b4a')
 const { assertNonEmpty } = require('../../_shared/lib/errors.js')
-const { attachGossip, sendGossip } = require('../../_shared/storage-gossip-base.js')
 
 const PROTOCOL = 'stream-multiplex/v1'
 
+class StreamHandle {
+  constructor (mux, id) {
+    this.mux = mux
+    this.id = id
+    this._listeners = []
+    this.closed = false
+  }
+
+  write (chunk) {
+    if (this.closed) throw new Error('stream closed')
+    return this.mux._sendFrame(this.id, chunk, false)
+  }
+
+  end (chunk) {
+    if (chunk != null) this.write(chunk)
+    this.mux._sendFrame(this.id, null, true)
+    this.closed = true
+  }
+
+  ondata (fn) {
+    this._listeners.push(fn)
+    return () => {
+      this._listeners = this._listeners.filter((f) => f !== fn)
+    }
+  }
+
+  _emitData (chunk, fin) {
+    for (const fn of this._listeners) fn(chunk, fin)
+    if (fin) this.closed = true
+  }
+}
+
 class HyperP2PStreamMultiplex extends EventEmitter {
   constructor (opts = {}) {
     super()
-    this._chunks = []
-    this._stats = { chunks: 0, bytes: 0 }
     this.highWaterMark = opts.highWaterMark || 65536
+    this._streams = new Map()
+    this._nextId = 1
+    this._stats = { streams: 0, frames: 0, bytes: 0 }
   }
 
-  write (chunk) {
-    if (chunk == null) throw new Error('chunk required')
-    const b4a = require('b4a')
-    const buf = typeof chunk === 'string' ? b4a.from(chunk) : chunk
-    this._chunks.push(buf)
-    this._stats.chunks++
-    this._stats.bytes += buf.length || buf.byteLength || 0
-    this.emit('data', buf)
-    return this._stats.bytes <= this.highWaterMark
+  openStream (id = null) {
+    const sid = id != null ? String(id) : String(this._nextId++)
+    if (this._streams.has(sid)) throw new Error('stream id already open')
+    const handle = new StreamHandle(this, sid)
+    this._streams.set(sid, handle)
+    this._stats.streams++
+    this.emit('open', { streamId: sid })
+    return handle
   }
 
-  read () { return this._chunks.shift() || null }
+  _sendFrame (streamId, chunk, fin) {
+    const size = chunk ? (chunk.length || chunk.byteLength || 0) : 0
+    if (this._stats.bytes + size > this.highWaterMark && !fin) {
+      return false
+    }
+    const frame = { type: 'frame', streamId, chunk, fin: !!fin }
+    this._stats.frames++
+    this._stats.bytes += size
+    this.emit('frame', frame)
+    const local = this._streams.get(streamId)
+    if (local) local._emitData(chunk, fin)
+    return true
+  }
 
-  pending () { return this._chunks.length }
+  receiveFrame (frame) {
+    if (!frame || frame.type !== 'frame') return false
+    const h = this._streams.get(frame.streamId)
+    if (!h) return false
+    h._emitData(frame.chunk, frame.fin)
+    if (frame.fin) this._streams.delete(frame.streamId)
+    return true
+  }
 
-  getStats () { return { ...this._stats, protocol: PROTOCOL } }
+  closeStream (streamId) {
+    const h = this._streams.get(streamId)
+    if (!h) return false
+    h.end()
+    this._streams.delete(streamId)
+    return true
+  }
+
+  getStats () {
+    return {
+      ...this._stats,
+      open: this._streams.size,
+      protocol: PROTOCOL
+    }
+  }
 
   async ready () { return this }
-  async close () { this._chunks = [] }
+
+  async close () {
+    for (const id of [...this._streams.keys()]) this.closeStream(id)
+    this.emit('closed')
+  }
 }
 
-module.exports = { HyperP2PStreamMultiplex, PROTOCOL }
+module.exports = { HyperP2PStreamMultiplex, StreamHandle, PROTOCOL }

messaging-streams/hyper-p2p-stream-multiplex/test/test.js

@@ -4,23 +4,50 @@ const { HyperP2PStreamMultiplex, PROTOCOL } = require('../index.js')
 
 test('exports', (t) => {
   t.ok(HyperP2PStreamMultiplex)
-  t.ok(PROTOCOL)
+  t.is(PROTOCOL, 'stream-multiplex/v1')
 })
 
-test('basic operation', async (t) => {
-  const m = new HyperP2PStreamMultiplex()
-  m.write('hi'); t.ok(m.read())
-  await m.close()
+test('openStream write read via ondata', async (t) => {
+  const mux = new HyperP2PStreamMultiplex()
+  const a = mux.openStream('a')
+  const chunks = []
+  a.ondata((c, fin) => { if (c) chunks.push(c); if (fin) t.pass() })
+  a.write(b4aFrom('hi'))
+  a.end()
+  t.is(chunks.length, 1)
+  await mux.close()
 })
 
-test('validation', async (t) => {
-  const m = new HyperP2PStreamMultiplex()
-  try { m.write(null) } catch (e) { t.ok(e) }
-  await m.close()
+test('receiveFrame remote', async (t) => {
+  const mux = new HyperP2PStreamMultiplex()
+  const b = mux.openStream('b')
+  let got = false
+  b.ondata((c) => { if (c) got = true })
+  mux.receiveFrame({ type: 'frame', streamId: 'b', chunk: b4aFrom('x'), fin: false })
+  mux.receiveFrame({ type: 'frame', streamId: 'b', chunk: null, fin: true })
+  t.ok(got)
+  await mux.close()
+})
+
+function b4aFrom (s) {
+  return require('b4a').from(s)
+}
+
+test('validation duplicate stream id', async (t) => {
+  const mux = new HyperP2PStreamMultiplex()
+  mux.openStream('dup')
+  try {
+    mux.openStream('dup')
+    t.fail('expected throw')
+  } catch (e) {
+    t.ok(e instanceof Error)
+  }
+  await mux.close()
 })
 
 test('getStats', async (t) => {
-  const m = new HyperP2PStreamMultiplex()
-  t.ok(m.getStats().protocol)
-  await m.close()
+  const mux = new HyperP2PStreamMultiplex()
+  mux.openStream()
+  t.ok(mux.getStats().protocol)
+  await mux.close()
 })