feat: add onStreamEnd, muxer.streams and timeline (#56)

BREAKING CHANGE: This adds new validations to the stream muxer, which will cause existing tests to fail.

src/base-test.js

@@ -8,11 +8,28 @@ const pair = require('it-pair/duplex')
 const pipe = require('it-pipe')
 const { collect, map, consume } = require('streaming-iterables')
 
+function close (stream) {
+  return pipe([], stream, consume)
+}
+
 async function closeAndWait (stream) {
-  await pipe([], stream, consume)
+  await close(stream)
   expect(true).to.be.true.mark()
 }
 
+/**
+ * A tick is considered valid if it happened between now
+ * and `ms` milliseconds ago
+ * @param {number} date Time in ticks
+ * @param {number} ms max milliseconds that should have expired
+ * @returns {boolean}
+ */
+function isValidTick (date, ms = 5000) {
+  const now = Date.now()
+  if (date > now - ms && date <= now) return true
+  return false
+}
+
 module.exports = (common) => {
   describe('base', () => {
     let Muxer
@@ -25,25 +42,44 @@ module.exports = (common) => {
       const p = pair()
       const dialer = new Muxer()
 
-      const listener = new Muxer(stream => {
-        expect(stream).to.exist.mark()
-        closeAndWait(stream)
+      const listener = new Muxer({
+        onStream: stream => {
+          expect(stream).to.exist.mark() // 1st check
+          expect(isValidTick(stream.timeline.open)).to.equal(true)
+          // Make sure the stream is being tracked
+          expect(listener.streams).to.include(stream)
+          close(stream)
+        },
+        onStreamEnd: stream => {
+          expect(stream).to.exist.mark() // 2nd check
+          expect(listener.streams).to.not.include(stream)
+          // Make sure the stream is removed from tracking
+          expect(isValidTick(stream.timeline.close)).to.equal(true)
+        }
       })
 
       pipe(p[0], dialer, p[0])
       pipe(p[1], listener, p[1])
 
-      expect(3).checks(done)
+      expect(3).checks(() => {
+        // ensure we have no streams left
+        expect(dialer.streams).to.have.length(0)
+        expect(listener.streams).to.have.length(0)
+        done()
+      })
 
       const conn = dialer.newStream()
+      expect(dialer.streams).to.include(conn)
+      expect(isValidTick(conn.timeline.open)).to.equal(true)
 
-      closeAndWait(conn)
+      closeAndWait(conn) // 3rd check
     })
 
     it('Open a stream from the listener', (done) => {
       const p = pair()
       const dialer = new Muxer(stream => {
         expect(stream).to.exist.mark()
+        expect(isValidTick(stream.timeline.open)).to.equal(true)
         closeAndWait(stream)
       })
       const listener = new Muxer()
@@ -54,6 +90,8 @@ module.exports = (common) => {
       expect(3).check(done)
 
       const conn = listener.newStream()
+      expect(listener.streams).to.include(conn)
+      expect(isValidTick(conn.timeline.open)).to.equal(true)
 
       closeAndWait(conn)
     })