feat: refactor, simplify, make handling of logs nicer

src/constants.js

@@ -1,5 +1,4 @@
 'use strict'
 
-module.exports = {
-  PROTOCOL_ID: '/multistream/1.0.0'
-}
+exports = module.exports
+exports.PROTOCOL_ID = '/multistream/1.0.0'

src/dialer.js → src/dialer/index.js

@@ -1,76 +1,71 @@
 'use strict'
 
-const lp = require('pull-length-prefixed')
 const varint = require('varint')
 const pull = require('pull-stream')
+const pullLP = require('pull-length-prefixed')
 const Connection = require('interface-connection').Connection
-const debug = require('debug')
-const log = debug('multistream:dialer')
+const util = require('../util')
+const select = require('../select')
 
-const PROTOCOL_ID = require('./constants').PROTOCOL_ID
-const agrmt = require('./agreement')
-
-function getRandomId () {
-  return ((~~(Math.random() * 1e9)).toString(36))
-}
+const PROTOCOL_ID = require('./../constants').PROTOCOL_ID
 
 module.exports = class Dialer {
   constructor () {
     this.conn = null
-    this.msThreadId = getRandomId()
+    this.log = util.log.dialer()
   }
 
   // perform the multistream handshake
   handle (rawConn, callback) {
-    log('(%s) dialer handle conn', this.msThreadId)
-    const ms = agrmt.select(PROTOCOL_ID, (err, conn) => {
+    this.log('dialer handle conn')
+    const s = select(PROTOCOL_ID, (err, conn) => {
       if (err) {
         return callback(err)
       }
-      log('(%s) handshake success', this.msThreadId)
+      this.log('handshake success')
 
       this.conn = new Connection(conn, rawConn)
 
       callback()
-    }, this.msThreadId)
+    }, this.log)
 
     pull(
       rawConn,
-      ms,
+      s,
       rawConn
     )
   }
 
   select (protocol, callback) {
-    log('(%s) dialer select %s', this.msThreadId, protocol)
+    this.log('dialer select ' + protocol)
     if (!this.conn) {
       return callback(new Error('multistream handshake has not finalized yet'))
     }
 
-    const selectStream = agrmt.select(protocol, (err, conn) => {
+    const s = select(protocol, (err, conn) => {
       if (err) {
         this.conn = new Connection(conn, this.conn)
         return callback(err)
       }
       callback(null, new Connection(conn, this.conn))
-    }, this.msThreadId)
+    }, this.log)
 
     pull(
       this.conn,
-      selectStream,
+      s,
       this.conn
     )
   }
 
   ls (callback) {
-    const lsStream = agrmt.select('ls', (err, conn) => {
+    const lsStream = select('ls', (err, conn) => {
       if (err) {
         return callback(err)
       }
 
       pull(
         conn,
-        lp.decode(),
+        pullLP.decode(),
         collectLs(conn),
         pull.map(stringify),
         pull.collect((err, list) => {
@@ -80,7 +75,7 @@ module.exports = class Dialer {
           callback(null, list.slice(1))
         })
       )
-    })
+    }, this.log)
 
     pull(
       this.conn,

src/listener.js → src/listener/index.js

@@ -1,47 +1,46 @@
 'use strict'
 
-const lp = require('pull-length-prefixed')
 const pull = require('pull-stream')
+const pullLP = require('pull-length-prefixed')
 const varint = require('varint')
 const isFunction = require('lodash.isfunction')
 const assert = require('assert')
-const debug = require('debug')
-const log = debug('multistream:listener')
+const select = require('../select')
+const selectHandler = require('./selectHandler')
+const util = require('./../util')
 const Connection = require('interface-connection').Connection
 
-const PROTOCOL_ID = require('./constants').PROTOCOL_ID
-const agrmt = require('./agreement')
+const PROTOCOL_ID = require('./../constants').PROTOCOL_ID
 
 module.exports = class Listener {
   constructor () {
     this.handlers = {
       ls: (conn) => this._ls(conn)
     }
+    this.log = util.log.listener()
   }
 
   // perform the multistream handshake
   handle (rawConn, callback) {
-    const msThreadId = getRandomId()
-    log('(%s) listener handle conn', msThreadId)
+    this.log('listener handle conn')
 
-    const selectStream = agrmt.select(PROTOCOL_ID, (err, conn) => {
+    const selectStream = select(PROTOCOL_ID, (err, conn) => {
       if (err) {
         return callback(err)
       }
 
-      const hsConn = new Connection(conn, rawConn)
+      const shConn = new Connection(conn, rawConn)
 
-      const handlerSelector =
-        agrmt.handlerSelector(hsConn, this.handlers, msThreadId)
+      const sh = selectHandler(shConn, this.handlers, this.log)
 
       pull(
-        hsConn,
-        handlerSelector,
-        hsConn
+        shConn,
+        sh,
+        shConn
       )
 
       callback()
-    }, msThreadId)
+    }, this.log)
 
     pull(
       rawConn,
@@ -52,12 +51,11 @@ module.exports = class Listener {
 
   // be ready for a given `protocol`
   addHandler (protocol, handler) {
-    log('adding handler: %s', protocol)
-
+    this.log('adding handler: ' + protocol)
     assert(isFunction(handler), 'handler must be a function')
 
     if (this.handlers[protocol]) {
-      log('overwriting handler for %s', protocol)
+      this.log('overwriting handler for ' + protocol)
     }
 
     this.handlers[protocol] = handler
@@ -88,12 +86,8 @@ module.exports = class Listener {
 
     pull(
       pull.values(values),
-      lp.encode(),
+      pullLP.encode(),
       conn
     )
   }
 }
-
-function getRandomId () {
-  return ((~~(Math.random() * 1e9)).toString(36))
-}

src/listener/selectHandler.js

@@ -0,0 +1,43 @@
+'use strict'
+
+const handshake = require('pull-handshake')
+const lp = require('pull-length-prefixed')
+const Connection = require('interface-connection').Connection
+const writeEncoded = require('../util.js').writeEncoded
+
+function selectHandler (rawConn, handlersMap, log) {
+  const cb = (err) => {
+    // incoming errors are irrelevant for the app
+    log.error(err)
+  }
+
+  const stream = handshake({ timeout: 60 * 1000 }, cb)
+  const shake = stream.handshake
+
+  next()
+  return stream
+
+  function next () {
+    lp.decodeFromReader(shake, (err, data) => {
+      if (err) {
+        return cb(err)
+      }
+      log('received:', data.toString())
+      const protocol = data.toString().slice(0, -1)
+      const result = Object.keys(handlersMap).filter((id) => id === protocol)
+      const key = result && result[0]
+
+      if (key) {
+        log('send ack back of: ' + protocol)
+        writeEncoded(shake, data, cb)
+        handlersMap[key](new Connection(shake.rest(), rawConn))
+      } else {
+        log('not supported protocol: ' + protocol)
+        writeEncoded(shake, new Buffer('na\n'))
+        next()
+      }
+    })
+  }
+}
+
+module.exports = selectHandler

src/select.js

@@ -0,0 +1,35 @@
+'use strict'
+
+const handshake = require('pull-handshake')
+const pullLP = require('pull-length-prefixed')
+const util = require('./util')
+const writeEncoded = util.writeEncoded
+
+function select (multicodec, callback, log) {
+  const stream = handshake({
+    timeout: 60 * 1000
+  }, callback)
+
+  const shake = stream.handshake
+
+  log('writing multicodec: ' + multicodec)
+  writeEncoded(shake, new Buffer(multicodec + '\n'), callback)
+
+  pullLP.decodeFromReader(shake, (err, data) => {
+    if (err) {
+      return callback(err)
+    }
+    const protocol = data.toString().slice(0, -1)
+
+    if (protocol !== multicodec) {
+      return callback(new Error(`"${multicodec}" not supported`), shake.rest())
+    }
+
+    log('received ack: ' + protocol)
+    callback(null, shake.rest())
+  })
+
+  return stream
+}
+
+module.exports = select