protomux - npm Package Compare versions

Comparing version 3.10.1 to 3.10.3

index.js

@@ -12,3 +12,16 @@ const b4a = require('b4a')
 class Channel {
-  constructor (mux, info, userData, protocol, aliases, id, handshake, messages, onopen, onclose, ondestroy, ondrain) {
+  constructor(
+    mux,
+    info,
+    userData,
+    protocol,
+    aliases,
+    id,
+    handshake,
+    messages,
+    onopen,
+    onclose,
+    ondestroy,
+    ondrain
+  ) {
     this.userData = userData
@@ -50,7 +63,7 @@ this.protocol = protocol
 
-  get drained () {
+  get drained() {
     return this._mux.drained
   }
 
-  fullyOpened () {
+  fullyOpened() {
     if (this.opened) return Promise.resolve(true)
@@ -60,18 +73,20 @@ if (this.closed) return Promise.resolve(false)
 
-    this._openedPromise = new Promise((resolve) => { this._openedResolve = resolve })
+    this._openedPromise = new Promise((resolve) => {
+      this._openedResolve = resolve
+    })
     return this._openedPromise
   }
 
-  fullyClosed () {
+  fullyClosed() {
     if (this.destroyed) return Promise.resolve()
     if (this._destroyedPromise) return this._destroyedPromise
 
-    this._destroyedPromise = new Promise((resolve) => { this._destroyedResolve = resolve })
+    this._destroyedPromise = new Promise((resolve) => {
+      this._destroyedResolve = resolve
+    })
     return this._destroyedPromise
   }
 
-  open (handshake) {
-    const id = this._mux._free.length > 0
-      ? this._mux._free.pop()
-      : this._mux._local.push(null) - 1
+  open(handshake) {
+    const id = this._mux._free.length > 0 ? this._mux._free.pop() : this._mux._local.push(null) - 1
 
@@ -106,7 +121,7 @@ this._info.opened++
 
-  _dec () {
+  _dec() {
     if (--this._active === 0 && this.closed === true) this._destroy()
   }
 
-  _decAndDestroy (err) {
+  _decAndDestroy(err) {
     this._dec()
@@ -116,8 +131,16 @@ this._mux._safeDestroy(err)
 
-  _fullyOpenSoon () {
+  _fullyOpenSoon() {
     this._mux._remote[this._remoteId - 1].session = this
-    queueTick(this._fullyOpen.bind(this))
+    queueTick(this._fullyOpenOrDestroy.bind(this))
   }
 
-  _fullyOpen () {
+  _fullyOpenOrDestroy() {
+    try {
+      this._fullyOpen()
+    } catch (err) {
+      this._mux._safeDestroyBound(err)
+    }
+  }
+
+  _fullyOpen() {
     if (this.opened === true || this.closed === true) return
@@ -127,3 +150,2 @@
 
-    this.opened = true
     this.handshake = this._handshake ? this._handshake.decode(remote.state) : null
@@ -135,7 +157,9 @@ this._track(this.onopen(this.handshake, this))
     if (remote.pending !== null) this._drain(remote)
+    if (this._mux._destroying === true) return
 
+    this.opened = true
     this._resolveOpen(true)
   }
 
-  _resolveOpen (opened) {
+  _resolveOpen(opened) {
     if (this._openedResolve !== null) {
@@ -147,3 +171,3 @@ this._openedResolve(opened)
 
-  _resolveDestroyed () {
+  _resolveDestroyed() {
     if (this._destroyedResolve !== null) {
@@ -155,3 +179,3 @@ this._destroyedResolve()
 
-  _drain (remote) {
+  _drain(remote) {
     for (let i = 0; i < remote.pending.length; i++) {
@@ -161,2 +185,3 @@ const p = remote.pending[i]
       this._recv(p.type, p.state)
+      if (this._mux._destroying === true) return
     }
@@ -168,3 +193,3 @@
 
-  _track (p) {
+  _track(p) {
     if (isPromise(p) === true) {
@@ -178,3 +203,3 @@ this._active++
 
-  _close (isRemote) {
+  _close(isRemote) {
     if (this.closed === true) return
@@ -205,3 +230,3 @@ this.closed = true
 
-  _destroy () {
+  _destroy() {
     if (this.destroyed === true) return
@@ -213,3 +238,3 @@ this.destroyed = true
 
-  _recv (type, state) {
+  _recv(type, state) {
     if (type < this.messages.length) {
@@ -223,11 +248,11 @@ const m = this.messages[type]
 
-  cork () {
+  cork() {
     this._mux.cork()
   }
 
-  uncork () {
+  uncork() {
     this._mux.uncork()
   }
 
-  close () {
+  close() {
     if (this.closed === true) return
@@ -249,3 +274,3 @@
 
-  addMessage (opts) {
+  addMessage(opts) {
     if (!opts) return this._skipMessage()
@@ -266,6 +291,6 @@
       onmessage,
-      recv (state, session) {
+      recv(state, session) {
         return session._track(m.onmessage(encoding.decode(state), session))
       },
-      send (m, session = s) {
+      send(m, session = s) {
         if (session.closed === true) return false
@@ -307,3 +332,3 @@
 
-  _skipMessage () {
+  _skipMessage() {
     const type = this.messages.length
@@ -314,4 +339,4 @@ const m = {
       onmessage: noop,
-      recv (state, session) {},
-      send (m, session) {}
+      recv(state, session) {},
+      send(m, session) {}
     }
@@ -325,3 +350,3 @@
 module.exports = class Protomux {
-  constructor (stream, { alloc } = {}) {
+  constructor(stream, { alloc } = {}) {
     if (stream.userData === null) stream.userData = this
@@ -334,3 +359,4 @@
 
-    this._alloc = alloc || (typeof stream.alloc === 'function' ? stream.alloc.bind(stream) : b4a.allocUnsafe)
+    this._alloc =
+      alloc || (typeof stream.alloc === 'function' ? stream.alloc.bind(stream) : b4a.allocUnsafe)
     this._safeDestroyBound = this._safeDestroy.bind(this)
@@ -350,2 +376,4 @@ this._uncorkBound = this.uncork.bind(this)
     this._notify = new Map()
+    // stream.destroyed flips asynchronously on streamx-based transports.
+    this._destroying = false
 
@@ -359,3 +387,3 @@ this.stream.on('data', this._ondata.bind(this))
 
-  static from (stream, opts) {
+  static from(stream, opts) {
     if (stream.userData && stream.userData.isProtomux) return stream.userData
@@ -366,7 +394,7 @@ if (stream.isProtomux) return stream
 
-  static isProtomux (mux) {
+  static isProtomux(mux) {
     return typeof mux === 'object' && mux.isProtomux === true
   }
 
-  * [Symbol.iterator] () {
+  *[Symbol.iterator]() {
     for (const session of this._local) {
@@ -377,7 +405,7 @@ if (session !== null) yield session
 
-  isIdle () {
+  isIdle() {
     return this._local.length === this._free.length
   }
 
-  cork () {
+  cork() {
     if (++this.corked === 1) {
@@ -389,3 +417,3 @@ this._batch = []
 
-  uncork () {
+  uncork() {
     if (--this.corked === 0) {
@@ -398,3 +426,3 @@ this._sendBatch(this._batch, this._batchState)
 
-  getLastChannel ({ protocol, id = null }) {
+  getLastChannel({ protocol, id = null }) {
     const key = toKey(protocol, id)
@@ -406,11 +434,11 @@ const info = this._infos.get(key)
 
-  pair ({ protocol, id = null }, notify) {
+  pair({ protocol, id = null }, notify) {
     this._notify.set(toKey(protocol, id), notify)
   }
 
-  unpair ({ protocol, id = null }) {
+  unpair({ protocol, id = null }) {
     this._notify.delete(toKey(protocol, id))
   }
 
-  opened ({ protocol, id = null }) {
+  opened({ protocol, id = null }) {
     const key = toKey(protocol, id)
@@ -421,3 +449,15 @@ const info = this._infos.get(key)
 
-  createChannel ({ userData = null, protocol, aliases = [], id = null, unique = true, handshake = null, messages = [], onopen = noop, onclose = noop, ondestroy = noop, ondrain = noop }) {
+  createChannel({
+    userData = null,
+    protocol,
+    aliases = [],
+    id = null,
+    unique = true,
+    handshake = null,
+    messages = [],
+    onopen = noop,
+    onclose = noop,
+    ondestroy = noop,
+    ondrain = noop
+  }) {
     if (this.stream.destroyed) return null
@@ -429,3 +469,16 @@
     if (info.incoming.length === 0) {
-      return new Channel(this, info, userData, protocol, aliases, id, handshake, messages, onopen, onclose, ondestroy, ondrain)
+      return new Channel(
+        this,
+        info,
+        userData,
+        protocol,
+        aliases,
+        id,
+        handshake,
+        messages,
+        onopen,
+        onclose,
+        ondestroy,
+        ondrain
+      )
     }
@@ -439,3 +492,16 @@
 
-    const session = new Channel(this, info, userData, protocol, aliases, id, handshake, messages, onopen, onclose, ondestroy, ondrain)
+    const session = new Channel(
+      this,
+      info,
+      userData,
+      protocol,
+      aliases,
+      id,
+      handshake,
+      messages,
+      onopen,
+      onclose,
+      ondestroy,
+      ondrain
+    )
 
@@ -448,3 +514,3 @@ session._remoteId = remoteId
 
-  _pushBatch (localId, buffer) {
+  _pushBatch(localId, buffer) {
     if (this._batchState.end >= MAX_BATCH) {
@@ -464,3 +530,3 @@ this._sendBatch(this._batch, this._batchState)
 
-  _sendBatch (batch, state) {
+  _sendBatch(batch, state) {
     if (batch.length === 0) return
@@ -488,3 +554,3 @@
 
-  _get (protocol, id, aliases = []) {
+  _get(protocol, id, aliases = []) {
     const key = toKey(protocol, id)
@@ -495,3 +561,13 @@
 
-    info = { key, protocol, aliases: [], id, pairing: 0, opened: 0, incoming: [], outgoing: [], lastChannel: null }
+    info = {
+      key,
+      protocol,
+      aliases: [],
+      id,
+      pairing: 0,
+      opened: 0,
+      incoming: [],
+      outgoing: [],
+      lastChannel: null
+    }
     this._infos.set(key, info)
@@ -509,3 +585,3 @@
 
-  _gc (info) {
+  _gc(info) {
     if (info.opened === 0 && info.outgoing.length === 0 && info.incoming.length === 0) {
@@ -518,3 +594,3 @@ this._infos.delete(info.key)
 
-  _ondata (buffer) {
+  _ondata(buffer) {
     if (buffer.byteLength === 0) return // ignore empty frames...
@@ -529,3 +605,3 @@ try {
 
-  _ondrain () {
+  _ondrain() {
     this.drained = true
@@ -538,7 +614,8 @@
 
-  _onend () { // TODO: support half open mode for the users who wants that here
+  _onend() {
+    // TODO: support half open mode for the users who wants that here
     this.stream.end()
   }
 
-  _decode (remoteId, state) {
+  _decode(remoteId, state) {
     const type = c.uint.decode(state)
@@ -563,3 +640,3 @@
 
-  _oncontrolsession (type, state) {
+  _oncontrolsession(type, state) {
     switch (type) {
@@ -586,3 +663,3 @@ case 0:
 
-  _bufferMessage (r, type, { buffer, start, end }) {
+  _bufferMessage(r, type, { buffer, start, end }) {
     const state = { buffer, start, end } // copy
@@ -594,3 +671,3 @@ r.pending.push({ type, state })
 
-  _pauseMaybe () {
+  _pauseMaybe() {
     if (this._paused === true || this._buffered <= MAX_BUFFERED) return
@@ -601,3 +678,3 @@ this._paused = true
 
-  _resumeMaybe () {
+  _resumeMaybe() {
     if (this._paused === false || this._buffered > MAX_BUFFERED) return
@@ -608,3 +685,3 @@ this._paused = false
 
-  _onbatch (state) {
+  _onbatch(state) {
     const end = state.end
@@ -639,3 +716,3 @@ let remoteId = c.uint.decode(state)
 
-  _onopensession (state) {
+  _onopensession(state) {
     const remoteId = c.uint.decode(state)
@@ -668,3 +745,4 @@ const protocol = c.string.decode(state)
 
-      if (session === null) { // we already closed the channel - ignore
+      if (session === null) {
+        // we already closed the channel - ignore
         this._free.push(localId - 1)
@@ -694,3 +772,3 @@ return null
 
-  _onrejectsession (state) {
+  _onrejectsession(state) {
     const localId = c.uint.decode(state)
@@ -717,3 +795,3 @@
 
-  _onclosesession (state) {
+  _onclosesession(state) {
     const remoteId = c.uint.decode(state)
@@ -731,3 +809,3 @@
 
-  async _requestSession (protocol, id, info) {
+  async _requestSession(protocol, id, info) {
     const notify = this._notify.get(toKey(protocol, id)) || this._notify.get(toKey(protocol, null))
@@ -746,3 +824,3 @@
 
-  _rejectSession (info, remoteId) {
+  _rejectSession(info, remoteId) {
     if (remoteId > 0) {
@@ -774,3 +852,3 @@ const r = this._remote[remoteId - 1]
 
-  _write0 (buffer) {
+  _write0(buffer) {
     if (this._batch !== null) {
@@ -784,12 +862,15 @@ this._pushBatch(0, buffer.subarray(1))
 
-  destroy (err) {
+  destroy(err) {
+    this._destroying = true
     this.stream.destroy(err)
   }
 
-  _safeDestroy (err) {
+  _safeDestroy(err) {
     safetyCatch(err)
+    this._destroying = true
     this.stream.destroy(err)
   }
 
-  _shutdown () {
+  _shutdown() {
+    this._destroying = true
     for (const s of this._local) {
@@ -801,17 +882,17 @@ if (s !== null) s._close(true)
 
-function noop () {}
+function noop() {}
 
-function toKey (protocol, id) {
+function toKey(protocol, id) {
   return protocol + '##' + (id ? b4a.toString(id, 'hex') : '')
 }
 
-function byteSize (state) {
+function byteSize(state) {
   return 512 + (state.end - state.start)
 }
 
-function isPromise (p) {
+function isPromise(p) {
   return !!(p && typeof p.then === 'function')
 }
 
-function encodingLength (enc, val) {
+function encodingLength(enc, val) {
   const state = { buffer: null, start: 0, end: 0 }
@@ -818,0 +899,0 @@ enc.preencode(state, val)

package.json

 {
   "name": "protomux",
-  "version": "3.10.1",
+  "version": "3.10.3",
   "description": "Multiplex multiple message oriented protocols over a stream",
@@ -19,10 +19,16 @@ "main": "index.js",
     "brittle": "^3.0.0",
-    "standard": "^16.0.4"
+    "lunte": "^1.0.0",
+    "prettier": "^3.6.2",
+    "prettier-config-holepunch": "^2.0.0"
   },
   "scripts": {
-    "test": "standard && brittle test.js"
+    "format": "prettier --write .",
+    "lint": "prettier --check . && lunte",
+    "test": "npm run test:node && npm run test:bare",
+    "test:node": "brittle-node test.js",
+    "test:bare": "brittle-bare test.js"
   },
   "repository": {
     "type": "git",
-    "url": "https://github.com/mafintosh/protomux.git"
+    "url": "https://github.com/holepunchto/protomux.git"
   },
@@ -32,5 +38,5 @@ "author": "Mathias Buus (@mafintosh)",
   "bugs": {
-    "url": "https://github.com/mafintosh/protomux/issues"
+    "url": "https://github.com/holepunchto/protomux/issues"
   },
-  "homepage": "https://github.com/mafintosh/protomux"
+  "homepage": "https://github.com/holepunchto/protomux"
 }

README.md

@@ -11,3 +11,3 @@ # protomux
 
-``` js
+```js
 const Protomux = require('protomux')
@@ -26,6 +26,6 @@ const c = require('compact-encoding')
   id: Buffer.from('optional binary id'),
-  onopen () {
+  onopen() {
     console.log('the other side opened this protocol!')
   },
-  onclose () {
+  onclose() {
     console.log('either side closed the protocol')
@@ -39,3 +39,3 @@ }
   encoding: c.string,
-  onmessage (m) {
+  onmessage(m) {
     console.log('recv message (1)', m)
@@ -47,3 +47,3 @@ }
   encoding: c.bool,
-  onmessage (m) {
+  onmessage(m) {
     console.log('recv message (2)', m)
@@ -71,3 +71,3 @@ }
 
-``` js
+```js
 {
@@ -89,3 +89,3 @@ // Called when the muxer wants to allocate a message that is written, defaults to Buffer.allocUnsafe.
 
-``` js
+```js
 {
@@ -113,3 +113,3 @@ // Used to match the protocol
 
-__NOTE__: `mux.createChannel` returns `null` if the channel should not be opened, ie it's a duplicate channel or the remote has already closed this one.
+**NOTE**: `mux.createChannel` returns `null` if the channel should not be opened, ie it's a duplicate channel or the remote has already closed this one.
 
@@ -138,3 +138,3 @@ If you want multiple sessions with the same `protocol` and `id`, set `unique: false` as an option.
 
-``` js
+```js
 {
@@ -141,0 +141,0 @@ // compact-encoding specifying how to encode/decode this message

No alert changes

Improved metrics

Total package byte prevSize

25959

3.82%

Lines of code

667

13.63%

Worsened metrics

Dev dependency count

5

66.67%

No dependency changes