*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

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
 {

package.json

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

test.js

@@ -345,6 +345,91 @@ test('alias', function (t) {
   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)
 }