feat(pull): migration to pull streams. Upgrade tests to use mocha as

well

src/base-test.js

@@ -0,0 +1,146 @@
+/* eslint-env mocha */
+'use strict'
+
+const chai = require('chai')
+chai.use(require('chai-checkmark'))
+const expect = chai.expect
+const pair = require('pull-pair/duplex')
+const pull = require('pull-stream')
+
+function closeAndWait (stream) {
+  pull(
+    pull.empty(),
+    stream,
+    pull.onEnd((err) => {
+      expect(err).to.not.exist.mark()
+    })
+  )
+}
+
+module.exports = (common) => {
+  describe('base', () => {
+    let muxer
+
+    beforeEach((done) => {
+      common.setup((err, _muxer) => {
+        if (err) return done(err)
+        muxer = _muxer
+        done()
+      })
+    })
+
+    it('Open a stream from the dialer', (done) => {
+      const p = pair()
+      const dialer = muxer.dial(p[0])
+      const listener = muxer.listen(p[1])
+
+      expect(4).checks(done)
+
+      listener.on('stream', (stream) => {
+        expect(stream).to.exist.mark()
+        closeAndWait(stream)
+      })
+
+      const conn = dialer.newStream((err) => {
+        expect(err).to.not.exist.mark()
+      })
+
+      closeAndWait(conn)
+    })
+
+    it('Open a stream from the listener', (done) => {
+      const p = pair()
+      const dialer = muxer.dial(p[0])
+      const listener = muxer.listen(p[1])
+
+      expect(4).check(done)
+
+      dialer.on('stream', (stream) => {
+        expect(stream).to.exist.mark()
+        closeAndWait(stream)
+      })
+
+      const conn = listener.newStream((err) => {
+        expect(err).to.not.exist.mark()
+      })
+
+      closeAndWait(conn)
+    })
+
+    it('Open a stream on both sides', (done) => {
+      const p = pair()
+      const dialer = muxer.dial(p[0])
+      const listener = muxer.listen(p[1])
+
+      expect(8).check(done)
+
+      dialer.on('stream', (stream) => {
+        expect(stream).to.exist.mark()
+        closeAndWait(stream)
+      })
+
+      const listenerConn = listener.newStream((err) => {
+        expect(err).to.not.exist.mark()
+      })
+
+      listener.on('stream', (stream) => {
+        expect(stream).to.exist.mark()
+        closeAndWait(stream)
+      })
+
+      const dialerConn = dialer.newStream((err) => {
+        expect(err).to.not.exist.mark()
+      })
+
+      closeAndWait(dialerConn)
+      closeAndWait(listenerConn)
+    })
+
+    it('Open a stream on one side, write, open a stream in the other side', (done) => {
+      const p = pair()
+      const dialer = muxer.dial(p[0])
+      const listener = muxer.listen(p[1])
+
+      expect(6).check(done)
+
+      const dialerConn = dialer.newStream((err) => {
+        expect(err).to.not.exist.mark()
+      })
+
+      pull(
+        pull.values(['hey']),
+        dialerConn
+      )
+
+      listener.on('stream', (stream) => {
+        pull(
+          stream,
+          pull.collect((err, chunks) => {
+            expect(err).to.not.exist.mark()
+            expect(chunks).to.be.eql([Buffer('hey')]).mark()
+          })
+        )
+
+        const listenerConn = listener.newStream((err) => {
+          expect(err).to.not.exist.mark()
+        })
+
+        pull(
+          pull.values(['hello']),
+          listenerConn
+        )
+
+        dialer.on('stream', onDialerStream)
+        function onDialerStream (stream) {
+          pull(
+            stream,
+            pull.collect((err, chunks) => {
+              expect(err).to.not.exist.mark()
+              expect(chunks).to.be.eql([Buffer('hello')]).mark()
+            })
+          )
+        }
+      })
+    })
+  })
+}

src/index.js

@@ -0,0 +1,14 @@
+/* eslint-env mocha */
+'use strict'
+
+const baseTest = require('./base-test')
+const stressTest = require('./stress-test')
+const megaStressTest = require('./mega-stress-test')
+
+module.exports = (common) => {
+  describe('interface-stream-muxer', () => {
+    baseTest(common)
+    stressTest(common)
+    megaStressTest(common)
+  })
+}

src/mega-stress-test.js

@@ -0,0 +1,23 @@
+/* eslint-env mocha */
+'use strict'
+
+const spawn = require('./spawner')
+
+module.exports = (common) => {
+  describe.skip('mega stress test', function () {
+    this.timeout(100 * 200 * 1000)
+    let muxer
+
+    beforeEach((done) => {
+      common.setup((err, _muxer) => {
+        if (err) return done(err)
+        muxer = _muxer
+        done()
+      })
+    })
+
+    it('10000 messages of 10000 streams', (done) => {
+      spawn(muxer, 10000, 10000, done, 5000)
+    })
+  })
+}

src/spawner.js

@@ -0,0 +1,88 @@
+'use strict'
+
+const expect = require('chai').expect
+
+const pair = require('pull-pair/duplex')
+const pull = require('pull-stream')
+const generate = require('pull-generate')
+const each = require('async/each')
+const eachLimit = require('async/eachLimit')
+
+module.exports = (muxer, nStreams, nMsg, done, limit) => {
+  const p = pair()
+  const dialerSocket = p[0]
+  const listenerSocket = p[1]
+
+  const check = marker((6 * nStreams) + (nStreams * nMsg), done)
+
+  const msg = 'simple msg'
+
+  const listener = muxer.listen(listenerSocket)
+  const dialer = muxer.dial(dialerSocket)
+
+  listener.on('stream', (stream) => {
+    expect(stream).to.exist
+    check()
+    pull(
+      stream,
+      pull.through((chunk) => {
+        expect(chunk).to.exist
+        check()
+      }),
+      pull.onEnd((err) => {
+        expect(err).to.not.exist
+        check()
+        pull(pull.empty(), stream)
+      })
+    )
+  })
+
+  const numbers = []
+  for (let i = 0; i < nStreams; i++) {
+    numbers.push(i)
+  }
+
+  const spawnStream = (n, cb) => {
+    const stream = dialer.newStream((err) => {
+      expect(err).to.not.exist
+      check()
+      expect(stream).to.exist
+      check()
+      pull(
+        generate(0, (s, cb) => {
+          cb(s === nMsg ? true : null, msg, s + 1)
+        }),
+        stream,
+        pull.collect((err, res) => {
+          expect(err).to.not.exist
+          check()
+          expect(res).to.be.eql([])
+          check()
+          cb()
+        })
+      )
+    })
+  }
+
+  if (limit) {
+    eachLimit(numbers, limit, spawnStream, () => {})
+  } else {
+    each(numbers, spawnStream, () => {})
+  }
+}
+
+function marker (n, done) {
+  let i = 0
+  return (err) => {
+    i++
+
+    if (err) {
+      console.error('Failed after %s iterations', i)
+      return done(err)
+    }
+
+    if (i === n) {
+      done()
+    }
+  }
+}

src/stress-test.js

@@ -0,0 +1,66 @@
+/* eslint-env mocha */
+'use strict'
+
+const spawn = require('./spawner')
+
+module.exports = (common) => {
+  describe('stress test', () => {
+    let muxer
+
+    beforeEach((done) => {
+      common.setup((err, _muxer) => {
+        if (err) return done(err)
+        muxer = _muxer
+        done()
+      })
+    })
+
+    it('1 stream with 1 msg', (done) => {
+      spawn(muxer, 1, 1, done)
+    })
+
+    it('1 stream with 10 msg', (done) => {
+      spawn(muxer, 1, 10, done)
+    })
+
+    it('1 stream with 100 msg', (done) => {
+      spawn(muxer, 1, 100, done)
+    })
+
+    it('10 streams with 1 msg', (done) => {
+      spawn(muxer, 10, 1, done)
+    })
+
+    it('10 streams with 10 msg', (done) => {
+      spawn(muxer, 10, 10, done)
+    })
+
+    it('10 streams with 100 msg', (done) => {
+      spawn(muxer, 10, 100, done)
+    })
+
+    it('100 streams with 1 msg', (done) => {
+      spawn(muxer, 100, 1, done)
+    })
+
+    it('100 streams with 10 msg', (done) => {
+      spawn(muxer, 100, 10, done)
+    })
+
+    it('100 streams with 100 msg', (done) => {
+      spawn(muxer, 100, 100, done)
+    })
+
+    it('1000 streams with 1 msg', (done) => {
+      spawn(muxer, 1000, 1, done)
+    })
+
+    it('1000 streams with 10 msg', (done) => {
+      spawn(muxer, 1000, 10, done)
+    })
+
+    it('1000 streams with 100 msg', (done) => {
+      spawn(muxer, 1000, 100, done)
+    })
+  })
+}