js-libp2p/test/switch/stream-muxers.node.js

/* eslint-env mocha */
/* eslint max-nested-callbacks: ["error", 8] */
'use strict'

const chai = require('chai')
const dirtyChai = require('dirty-chai')
const expect = chai.expect
chai.use(dirtyChai)

const parallel = require('async/parallel')
const TCP = require('libp2p-tcp')
const multiplex = require('libp2p-mplex')
const pullMplex = require('pull-mplex')
const spdy = require('libp2p-spdy')
const pull = require('pull-stream')
const PeerBook = require('peer-book')
const utils = require('./utils')
const createInfos = utils.createInfos
const tryEcho = utils.tryEcho

const Switch = require('../../src/switch')

describe('Stream Multiplexing', () => {
  [
    multiplex,
    pullMplex,
    spdy
  ].forEach((sm) => describe(sm.multicodec, () => {
    let switchA
    let switchB
    let switchC

    before((done) => createInfos(3, (err, peerInfos) => {
      expect(err).to.not.exist()
      function maGen (port) { return `/ip4/127.0.0.1/tcp/${port}` }

      const peerA = peerInfos[0]
      const peerB = peerInfos[1]
      const peerC = peerInfos[2]

      peerA.multiaddrs.add(maGen(9001))
      peerB.multiaddrs.add(maGen(9002))
      peerC.multiaddrs.add(maGen(9003))

      switchA = new Switch(peerA, new PeerBook())
      switchB = new Switch(peerB, new PeerBook())
      switchC = new Switch(peerC, new PeerBook())

      switchA.transport.add('tcp', new TCP())
      switchB.transport.add('tcp', new TCP())
      switchC.transport.add('tcp', new TCP())

      parallel([
        (cb) => switchA.transport.listen('tcp', {}, null, cb),
        (cb) => switchB.transport.listen('tcp', {}, null, cb),
        (cb) => switchC.transport.listen('tcp', {}, null, cb)
      ], done)
    }))

    after((done) => parallel([
      (cb) => switchA.stop(cb),
      (cb) => switchB.stop(cb)
    ], done))

    it('switch.connection.addStreamMuxer', (done) => {
      switchA.connection.addStreamMuxer(sm)
      switchB.connection.addStreamMuxer(sm)
      switchC.connection.addStreamMuxer(sm)
      done()
    })

    it('handle + dial on protocol', (done) => {
      switchB.handle('/abacaxi/1.0.0', (protocol, conn) => pull(conn, conn))

      switchA.dial(switchB._peerInfo, '/abacaxi/1.0.0', (err, conn) => {
        expect(err).to.not.exist()
        expect(switchA.connection.getAll()).to.have.length(1)

        tryEcho(conn, done)
      })
    })

    it('dial to warm conn', (done) => {
      switchB.dial(switchA._peerInfo, (err) => {
        expect(err).to.not.exist()

        expect(Object.keys(switchB.conns).length).to.equal(0)
        expect(switchB.connection.getAll()).to.have.length(1)
        done()
      })
    })

    it('dial on protocol, reuse warmed conn', (done) => {
      switchA.handle('/papaia/1.0.0', (protocol, conn) => pull(conn, conn))

      switchB.dial(switchA._peerInfo, '/papaia/1.0.0', (err, conn) => {
        expect(err).to.not.exist()
        expect(Object.keys(switchB.conns).length).to.equal(0)
        expect(switchB.connection.getAll()).to.have.length(1)

        tryEcho(conn, done)
      })
    })

    it('enable identify to reuse incomming muxed conn', (done) => {
      switchA.connection.reuse()
      switchC.connection.reuse()

      switchC.dial(switchA._peerInfo, (err) => {
        expect(err).to.not.exist()
        setTimeout(() => {
          expect(switchC.connection.getAll()).to.have.length(1)
          expect(switchA.connection.getAll()).to.have.length(2)
          done()
        }, 500)
      })
    })

    it('with Identify enabled, do getPeerInfo', (done) => {
      switchA.handle('/banana/1.0.0', (protocol, conn) => {
        conn.getPeerInfo((err, pi) => {
          expect(err).to.not.exist()
          expect(switchC._peerInfo.id.toB58String()).to.equal(pi.id.toB58String())
        })

        pull(conn, conn)
      })

      switchC.dial(switchA._peerInfo, '/banana/1.0.0', (err, conn) => {
        expect(err).to.not.exist()
        setTimeout(() => {
          expect(switchC.connection.getAll()).to.have.length(1)
          expect(switchA.connection.getAll()).to.have.length(2)

          conn.getPeerInfo((err, pi) => {
            expect(err).to.not.exist()
            expect(switchA._peerInfo.id.toB58String()).to.equal(pi.id.toB58String())
            tryEcho(conn, done)
          })
        }, 500)
      })
    })

    it('closing one side cleans out in the other', (done) => {
      switchC.stop((err) => {
        expect(err).to.not.exist()

        setTimeout(() => {
          expect(switchA.connection.getAll()).to.have.length(1)
          done()
        }, 500)
      })
    })
  }))
})