feat(media,messaging,network): deepen streaming and transport APIs

Origin hybrid weighted select, latency path stats, live-edge listing;
topic channel batch publish and retained clear; wakeup reschedule and
cancelAll; electron bridge handler registry helpers.

Co-authored-by: Cursor <[email protected]>

media-streaming/hyper-p2p-latency-optimizer/index.js

@@ -57,6 +57,21 @@ class HyperP2PLatencyOptimizer extends EventEmitter {
     return [...this._paths.keys()]
   }
 
+  clearPath (pathId) {
+    return this._paths.delete(String(pathId))
+  }
+
+  averageRtt () {
+    const vals = [...this._paths.values()]
+    if (!vals.length) return 0
+    return vals.reduce((s, p) => s + p.rttMs, 0) / vals.length
+  }
+
+  isWithinTarget (plan) {
+    const glass = plan?.estimatedGlassMs ?? this.averageRtt() + 200
+    return glass <= this.targetGlassMs
+  }
+
   getStats () {
     return mediaStats(this._stats, PROTOCOL, { paths: this._paths.size })
   }

media-streaming/hyper-p2p-live-edge-manager/index.js

@@ -61,6 +61,23 @@ class HyperP2PLiveEdgeManager extends EventEmitter {
     return this.setLiveEdge(streamId, e.seq + (delta | 0), Date.now())
   }
 
+  removeEdge (streamId) {
+    return this._edges.delete(assertStreamId(streamId))
+  }
+
+  listEdges () {
+    return [...this._edges.values()]
+  }
+
+  maxLagAcrossStreams (now = Date.now()) {
+    let max = 0
+    for (const sid of this._edges.keys()) {
+      const lag = this.edgeLagMs(sid, now)
+      if (lag != null && lag > max) max = lag
+    }
+    return max
+  }
+
   getStats () {
     return mediaStats(this._stats, PROTOCOL, { streams: this._edges.size })
   }

media-streaming/hyper-p2p-origin-hybrid-bridge/index.js

@@ -48,6 +48,33 @@ class HyperP2POriginHybridBridge extends EventEmitter {
     return [...(this._origins.get(assertStreamId(streamId)) || [])]
   }
 
+  removeOrigin (streamId, url) {
+    const sid = assertStreamId(streamId)
+    const list = this._origins.get(sid)
+    if (!list) return false
+    const idx = list.findIndex((o) => o.url === url)
+    if (idx < 0) return false
+    list.splice(idx, 1)
+    if (!list.length) this._origins.delete(sid)
+    return true
+  }
+
+  listStreamIds () {
+    return [...this._origins.keys()]
+  }
+
+  weightedSelect (streamId) {
+    const list = this._origins.get(assertStreamId(streamId))
+    if (!list || !list.length) return null
+    const total = list.reduce((s, o) => s + o.weight, 0)
+    let r = Math.random() * total
+    for (const o of list) {
+      r -= o.weight
+      if (r <= 0) return o
+    }
+    return list[0]
+  }
+
   totalBytes () {
     return this._stats.originBytes + this._stats.p2pBytes
   }

messaging-pubsub/hyper-p2p-topic-channel/index.js

@@ -79,6 +79,23 @@ class HyperP2PTopicChannel extends EventEmitter {
 
   retainedCount () { return this._retained.size }
 
+  clearRetained (channel) {
+    if (channel) return this._retained.delete(String(channel))
+    const n = this._retained.size
+    this._retained.clear()
+    return n
+  }
+
+  publishBatch (channel, payloads) {
+    assertNonEmpty(channel, 'channel')
+    if (!Array.isArray(payloads)) throw new Error('payloads array required')
+    return payloads.map((p) => this.publish(channel, p))
+  }
+
+  subscriberCount () {
+    return this._subs.size
+  }
+
   _deliverLocal (channel, payload, meta) {
     const handler = this._subs.get(channel)
     if (handler) {

network-transport/hyper-p2p-wakeup-channel/index.js

@@ -55,6 +55,26 @@ class HyperP2PWakeupChannel extends EventEmitter {
       .sort((a, b) => a.at - b.at)
   }
 
+  nextWakeup () {
+    const list = this.pending()
+    return list.length ? list[0] : null
+  }
+
+  hasPending (peerId) {
+    return this._pending.has(String(peerId))
+  }
+
+  reschedule (peerId, at) {
+    if (!this.hasPending(peerId)) return null
+    return this.scheduleWakeup(peerId, at)
+  }
+
+  cancelAll () {
+    const ids = [...this._pending.keys()]
+    for (const id of ids) this.cancel(id)
+    return ids.length
+  }
+
   cancel (peerId, opts = {}) {
     assertNonEmpty(peerId, 'peerId')
     const had = this._pending.has(peerId)

pear-platform/hyper-pear-electron-bridge/index.js

@@ -30,6 +30,16 @@ class HyperPearElectronBridge extends EventEmitter {
     return [...this._channels.keys()]
   }
 
+  unregisterAll () {
+    const n = this._channels.size
+    this._channels.clear()
+    return n
+  }
+
+  handlerCount () {
+    return this._channels.size
+  }
+
   invoke (name, payload) {
     const h = this._channels.get(assertId(name))
     if (!h) throw new Error(`ipc not registered: ${name}`)