*make message encoding/decoding async for use with the kernel

* update readme with mux.pair and mux.opened

* add opts to .pair, .unpair and .opened

.gitignore

@@ -1,4 +1,5 @@
 node_modules
+package-lock.json
 sandbox.js
 sandbox.mjs
 sandbox

README.md

@@ -46,10 +46,9 @@ const two = cool.addMessage({
   }
 })
 
-// open the channels
+// open the channel
 
-one.open()
-two.open()
+cool.open()
 
 // And send some data
 
@@ -109,13 +108,25 @@ __NOTE__: `mux.createChannel` returns `null` if the channel should not be opened
 
 If you want multiple sessions with the same `protocol` and `id`, set `unique: false` as an option.
 
+#### `const opened = mux.opened({ protocol, id })`
+
+Boolean that indicates if the channel is opened.
+
+#### `mux.pair({ protocol, id }, callback)`
+
+Register a callback to be called everytime a new channel is requested.
+
+#### `mux.unpair({ protocol, id })`
+
+Unregisters the pair callback.
+
 #### `channel.open([handshake])`
 
 Open the channel.
 
 #### `const m = channel.addMessage(opts)`
 
-Add a message. Options include:
+Add/register a message type for a certain encoding. Options include:
 
 ``` js
 {
@@ -159,6 +170,10 @@ Same as `channel.cork` but on the muxer instance.
 
 Same as `channel.uncork` but on the muxer instance.
 
+#### `for (const channel of muxer) { ... }`
+
+The muxer instance is iterable, so you can iterate over all the channels.
+
 ## License
 
 MIT

package.json

@@ -1,6 +1,6 @@
 {
   "name": "protomux",
-  "version": "3.2.1",
+  "version": "3.4.1",
   "description": "Multiplex multiple message oriented protocols over a stream",
   "main": "index.js",
   "dependencies": {
@@ -10,8 +10,9 @@
     "safety-catch": "^1.0.1"
   },
   "devDependencies": {
-    "@hyperswarm/secret-stream": "^5.1.4",
-    "brittle": "^2.0.1",
+    "@hyperswarm/secret-stream": "^6.0.0",
+    "brittle": "^3.0.0",
+    "prettier": "^2.8.7",
     "standard": "^16.0.4"
   },
   "scripts": {

test.js

@@ -2,6 +2,7 @@ const Protomux = require('./')
 const SecretStream = require('@hyperswarm/secret-stream')
 const test = require('brittle')
 const c = require('compact-encoding')
+const b4a = require('b4a')
 
 test('basic', function (t) {
   const a = new Protomux(new SecretStream(true))
@@ -189,6 +190,52 @@ test('corks', function (t) {
   }
 })
 
+test('mega cork', function (t) {
+  const a = new Protomux(new SecretStream(true))
+
+  a.cork()
+
+  const ap = a.createChannel({
+    protocol: 'mega'
+  })
+
+  ap.open()
+
+  const a1 = ap.addMessage({ encoding: c.buffer })
+
+  const b = new Protomux(new SecretStream(false))
+
+  const bp = b.createChannel({
+    protocol: 'mega'
+  })
+
+  bp.open()
+
+  const b1 = bp.addMessage({ encoding: c.buffer })
+
+  replicate(a, b)
+
+  t.plan(32 + 4)
+
+  const buf = b4a.alloc(1024 * 1024)
+  const expected = []
+
+  for (let i = 0; i < 32; i++) {
+    a1.send(buf)
+    expected.push(buf)
+  }
+
+  a.uncork()
+
+  b.stream.on('data', function (data) {
+    t.ok(data.byteLength > 8000000, 'got big message')
+  })
+
+  b1.onmessage = function (message) {
+    t.alike(message, expected.shift())
+  }
+})
+
 test('handshake', function (t) {
   const a = new Protomux(new SecretStream(true))
   const b = new Protomux(new SecretStream(false))
@@ -265,6 +312,124 @@ test('pipeline close and rejections', function (t) {
   }
 })
 
+test('alias', function (t) {
+  const a = new Protomux(new SecretStream(true))
+  const b = new Protomux(new SecretStream(false))
+
+  replicate(a, b)
+
+  const p = a.createChannel({
+    protocol: 'foo',
+    aliases: ['bar'],
+    onopen () {
+      t.pass('a remote opened')
+    }
+  })
+
+  p.open()
+
+  p.addMessage({
+    encoding: c.string,
+    onmessage (message) {
+      t.is(message, 'hello world')
+    }
+  })
+
+  const bp = b.createChannel({
+    protocol: 'bar'
+  })
+
+  t.plan(2)
+
+  bp.open()
+  bp.addMessage({ encoding: c.string }).send('hello world')
+})
+
+test('deduplicate muxers', function (t) {
+  const sa = new SecretStream(true)
+  const sb = new SecretStream(false)
+
+  replicate({ stream: sa }, { stream: sb })
+
+  const a = Protomux.from(sa)
+  const foo = a.createChannel({
+    protocol: 'foo',
+    onopen () { t.pass('a remote opened') }
+  })
+
+  foo.open()
+
+  foo.addMessage({
+    encoding: c.string,
+    onmessage (message) { t.is(message, 'hello foo') }
+  })
+
+  const bfoo = Protomux.from(sb).createChannel({ protocol: 'foo' })
+
+  // Another Protomux instance for another protocol
+  const a2 = Protomux.from(sa)
+  const bar = a2.createChannel({
+    protocol: 'bar',
+    onopen () { t.pass('a remote opened') }
+  })
+
+  bar.open()
+
+  bar.addMessage({
+    encoding: c.string,
+    onmessage (message) { t.is(message, 'hello bar') }
+  })
+
+  const bbar = Protomux.from(sb).createChannel({ protocol: 'bar' })
+
+  t.plan(4)
+
+  bfoo.open()
+  bfoo.addMessage({ encoding: c.string }).send('hello foo')
+
+  bbar.open()
+  bbar.addMessage({ encoding: c.string }).send('hello bar')
+})
+
+test('open + send + close on same tick', async function (t) {
+  t.plan(4)
+
+  const a = new Protomux(new SecretStream(true))
+  const b = new Protomux(new SecretStream(false))
+
+  replicate(a, b)
+
+  const ac = a.createChannel({
+    protocol: 'foo',
+    onopen () {
+      t.pass('a opened')
+    },
+    onclose () {
+      t.pass('a closed')
+    }
+  })
+
+  ac.open()
+  ac.addMessage({
+    encoding: c.string,
+    onmessage (message) { t.is(message, 'hello') }
+  })
+
+  const bc = b.createChannel({
+    protocol: 'foo',
+    onopen () {
+      t.fail('b opened')
+    },
+    onclose () {
+      t.pass('b closed')
+    }
+  })
+
+  bc.open()
+  bc.addMessage({ encoding: c.string }).send('hello')
+  bc.close()
+})
+
 function replicate (a, b) {
   a.stream.rawStream.pipe(b.stream.rawStream).pipe(a.stream.rawStream)
 }