protomux (updated, working now)

protomux-pair.js

const Protomux = require('protomux')
const SecretStream = require('@hyperswarm/secret-stream')
const c = require('compact-encoding')

const duplexify = require('duplexify')
const stream = require('stream')

const opts = { protocol: 'foo'}

const alice = {
  onconnect: async (conn, isInitiator) => {
    const mux = new Protomux(new SecretStream(isInitiator))
    conn.pipe(mux.stream.rawStream).pipe(conn)
    
    make_protocol(mux, opts, cb)
    
    function cb (isChannelOpen) {
      if (isChannelOpen) return
      const channel = create_and_open_channel(mux, {...opts, onopen, onclose })
      const msg = add_string_msg(channel, onmessage)
    }
    
  }
}

const bob = {
  onconnect: async (conn, isInitiator) => {
    const mux = new Protomux(new SecretStream(isInitiator))
    conn.pipe(mux.stream.rawStream).pipe(conn)
    
    make_protocol(mux, opts, cb)
    
    function cb (isChannelOpen) {
      if (isChannelOpen) return
      const channel = create_and_open_channel(mux, {...opts, onopen, onclose })
      const msg = add_string_msg(channel, onmessage)

      msg.send('hello world')
      msg.send('hola world')  
    }
  }
}

/// helpers

async function make_protocol (mux, opts, cb) {
  mux.pair(opts, callback)
  const opened = await mux.stream.opened
  console.log({ text: 'mux proto', mux_stream_opened: opened, mux_opened: mux.opened(opts) })
  if (opened) callback()
  function callback () {
    cb(mux.opened(opts))
  }
}

function create_and_open_channel (mux, opts) {
  const channel = mux.createChannel(opts)
  if (!channel) return
  channel.open()
  return channel
}

function add_string_msg (channel, onmessage) {
  const msg = channel.addMessage({
    encoding: c.string,
    onmessage
  })
  return msg
}

function onopen () { console.log('onopen') }
function onclose () { console.log('onclose') }
function onmessage (message) { console.log(message) }

async function simulate () {
  const PassThroughStream = stream.PassThrough
  const alicePass = new PassThroughStream()
  const bobPass = new PassThroughStream()
  const bobChannel = duplexify()
  const aliceChannel = duplexify()  
  bobChannel.setReadable(bobPass)
  bobChannel.setWritable(alicePass)
  aliceChannel.setReadable(alicePass)
  aliceChannel.setWritable(bobPass)
  bob.onconnect(aliceChannel, false)
  const delay = 0
  // const delay = 5000
  await new Promise(r => setTimeout(r, delay))
  alice.onconnect(bobChannel, true)  
}

simulate()