chore: use stream registry map

Author: vasco-santos

src/connection.js

@@ -4,7 +4,6 @@ const PeerId = require('peer-id')
 const multiaddr = require('multiaddr')
 
 const withIs = require('class-is')
-const Stream = require('./stream')
 
 const assert = require('assert')
 const errCode = require('err-code')
@@ -23,6 +22,7 @@ class Connection {
    * @param {PeerId} properties.remotePeer remote peer-id.
    * @param {function} properties.newStream new stream muxer function.
    * @param {function} properties.close close raw connection function.
+   * @param {function} properties.getStreams get streams from muxer function.
    * @param {Object} properties.stat metadata of the connection.
    * @param {string} properties.stat.direction connection establishment direction ("inbound" or "outbound").
    * @param {Object} properties.stat.timeline connection relevant events timestamp.
@@ -31,13 +31,14 @@ class Connection {
    * @param {string} [properties.stat.multiplexer] connection multiplexing identifier.
    * @param {string} [properties.stat.encryption] connection encryption method identifier.
    */
-  constructor ({ localAddr, remoteAddr, localPeer, remotePeer, newStream, close, stat }) {
+  constructor ({ localAddr, remoteAddr, localPeer, remotePeer, newStream, close, getStreams, stat }) {
     assert(multiaddr.isMultiaddr(localAddr), 'localAddr must be an instance of multiaddr')
     assert(multiaddr.isMultiaddr(remoteAddr), 'remoteAddr must be an instance of multiaddr')
     assert(PeerId.isPeerId(localPeer), 'localPeer must be an instance of peer-id')
     assert(PeerId.isPeerId(remotePeer), 'remotePeer must be an instance of peer-id')
     assert(typeof newStream === 'function', 'new stream must be a function')
     assert(typeof close === 'function', 'close must be a function')
+    assert(typeof getStreams === 'function', 'getStreams must be a function')
     assert(stat, 'connection metadata object must be provided')
     assert(stat.direction === 'inbound' || stat.direction === 'outbound', 'direction must be "inbound" or "outbound"')
     assert(stat.timeline, 'connection timeline object must be provided in the stat object')
@@ -92,9 +93,14 @@ class Connection {
     this._close = close
 
     /**
-     * Connection streams
+     * Reference to the getStreams function of the muxer
      */
-    this._streams = []
+    this._getStreams = getStreams
+
+    /**
+     * Connection streams registry
+     */
+    this.streamRegistry = new Map()
 
     /**
      * User provided tags
@@ -111,17 +117,24 @@ class Connection {
   }
 
   /**
-   * Get all the streams associated with the connection.
-   * @return {Array<Stream>}
+   * Get all the streams of the muxer.
+   * @return {Array<*>}
+   */
+  get streams () {
+    return this._getStreams()
+  }
+
+  /**
+   * Get stream registry map with stream metadata indexed by id.
    */
-  getStreams () {
-    return this._streams
+  get registry () {
+    return this.streamRegistry
   }
 
   /**
    * Create a new stream from this connection
    * @param {string[]} protocols intended protocol for the stream
-   * @return {Stream} new muxed+multistream-selected stream
+   * @return {*} new muxed+multistream-selected stream
    */
   async newStream (protocols) {
     if (this.stat.status === 'closing') {
@@ -134,40 +147,37 @@ class Connection {
 
     if (!Array.isArray(protocols)) protocols = [protocols]
 
-    const { stream: duplexStream, protocol } = await this._newStream(protocols)
+    const { stream: muxedStream, protocol } = await this._newStream(protocols)
 
-    return this.addStream({
-      stream: duplexStream,
-      protocol,
-      direction: 'outbound'
-    })
+    this.addStream(muxedStream, protocol)
+    return muxedStream
   }
 
   /**
-   * Add an inbound stream when it is opened.
-   * @param {object} options
-   * @param {*} options.stream an Iterable Duplex stream
-   * @param {string} options.protocol the protocol the stream is using
-   * @param {string} [options.direction = 'inbound'] stream establishment direction ("inbound" or "outbound")
-   * @return {Stream} new stream within the connection
+   * Add a stream when it is opened to the registry.
+   * @param {*} muxedStream a muxed stream
+   * @param {string} protocol the protocol the stream is using
+   * @param {Array<String>} tags tags of the stream
+   * @return {void}
    */
-  addStream ({ stream, protocol, direction = 'inbound' }) {
-    assert(direction === 'inbound' || direction === 'outbound', 'direction must be "inbound" or "outbound"')
-
-    const s = new Stream({
-      iterableDuplex: stream,
-      conn: this,
-      direction,
+  addStream (muxedStream, protocol, tags = []) {
+    // Add metadata for the stream
+    this.streamRegistry.set(muxedStream.id, {
+      tags,
       protocol
     })
+  }
 
-    this._streams.push(s)
-
-    return s
+  /**
+   * Remove stream registry after it is closed.
+   * @param {string} id identifier of the stream
+   */
+  removeStream (id) {
+    this.streamRegistry.delete(id)
   }
 
   /**
-   * Close the connection, as well as all its associated streams.
+   * Close the connection.
    * @return {Promise}
    */
   async close () {
@@ -184,9 +194,6 @@ class Connection {
     // Close raw connection
     this._closing = await this._close()
 
-    // All streams closed
-    this._streams.map((stream) => stream.close())
-
     this._stat.timeline.close = Date.now()
     this.stat.status = 'closed'
   }

src/stream.js

@@ -3,9 +3,7 @@
 'use strict'
 
 const connectionSuite = require('../test/connection')
-const streamSuite = require('../test/stream')
 
 module.exports = (test) => {
   connectionSuite(test)
-  streamSuite(test)
 }

src/tests.js

@@ -17,6 +17,8 @@ describe('compliance tests', () => {
         PeerId.createFromJSON(peers[0]),
         PeerId.createFromJSON(peers[1])
       ])
+      const openStreams = []
+      let streamId = 0
 
       return new Connection({
         localPeer,
@@ -33,14 +35,21 @@ describe('compliance tests', () => {
           multiplexer: '/mplex/6.7.0'
         },
         newStream: (protocols) => {
+          const id = streamId++
           const stream = pair()
+
           stream.close = () => stream.sink([])
+          stream.id = id
+
+          openStreams.push(stream)
+
           return {
             stream,
             protocol: protocols[0]
           }
         },
-        close: () => {}
+        close: () => {},
+        getStreams: () => openStreams
       })
     },
     async teardown () {

test/compliance.spec.js

@@ -35,7 +35,7 @@ module.exports = (test) => {
         expect(connection.stat.timeline.upgraded).to.exist()
         expect(connection.stat.timeline.close).to.not.exist()
         expect(connection.stat.direction).to.exist()
-        expect(connection.getStreams()).to.eql([])
+        expect(connection.streams).to.eql([])
         expect(connection.tags).to.eql([])
       })
 
@@ -50,33 +50,21 @@ module.exports = (test) => {
       })
 
       it('should return an empty array of streams', () => {
-        const streams = connection.getStreams()
+        const streams = connection.streams
 
         expect(streams).to.eql([])
       })
 
       it('should be able to create a new stream', async () => {
         const protocol = '/echo/0.0.1'
         const stream = await connection.newStream(protocol)
-        const connStreams = await connection.getStreams()
+        const connStreams = await connection.streams
 
         expect(stream).to.exist()
         expect(connStreams).to.exist()
         expect(connStreams).to.have.lengthOf(1)
         expect(connStreams[0]).to.equal(stream)
       })
-
-      it('should be able to add an inbound stream', () => {
-        const s = goodbye({ source: ['hey'], sink: collect })
-        s.close = () => s.sink([])
-
-        connection.addStream({ stream: s, protocol: '/echo/0.0.1' })
-
-        const connStreams = connection.getStreams()
-        expect(connStreams).to.exist()
-        expect(connStreams).to.have.lengthOf(1)
-        expect(connStreams[0].protocol).to.equal('/echo/0.0.1')
-      })
     })
 
     describe('close connection', () => {