Rework using protomux-bridging-relay

lib/connect.js

@@ -1,5 +1,6 @@
 const NoiseSecretStream = require('@hyperswarm/secret-stream')
 const b4a = require('b4a')
+const relay = require('protomux-bridging-relay')
 const DebuggingStream = require('debugging-stream')
 const { isPrivate } = require('bogon')
 const Semaphore = require('./semaphore')
@@ -24,7 +25,6 @@ const {
   SERVER_ERROR,
   SERVER_INCOMPATIBLE
 } = require('./errors')
-const connectionRelay = require('./connection-relay')
 
 module.exports = function connect (dht, publicKey, opts = {}) {
   const pool = opts.pool || null
@@ -64,7 +64,7 @@ module.exports = function connect (dht, publicKey, opts = {}) {
     onsocket: null,
     sleeper: new Sleeper(),
     relayThrough: opts.relayThrough || null,
-    relayToken: opts.relayThrough ? connectionRelay.token() : null,
+    relayToken: opts.relayThrough ? relay.token() : null,
     encryptedSocket
   }
 
@@ -389,9 +389,23 @@ async function connectThroughNode (c, address, socket) {
       token = c.relayToken
     }
 
-    const request = connectionRelay.Client
-      .from(c.dht.connect(publicKey, { pool: c.pool }))
-      .rendezvous(isInitiator, token, c.rawStream, c.onsocket)
+    const relaySocket = c.dht.connect(publicKey, { pool: c.pool })
+
+    const request = relay.Client
+      .from(relaySocket, { id: relaySocket.publicKey })
+      .pair(isInitiator, token, c.rawStream)
+      .on('error', () => { /* TODO */ })
+      .on('data', (remoteId) => {
+        const {
+          remotePort,
+          remoteHost,
+          socket
+        } = relaySocket.rawStream
+
+        c.rawStream.connect(socket, remoteId, remotePort, remoteHost)
+
+        c.onsocket(socket, remotePort, remoteHost)
+      })
 
     const onerror = () => request.destroy()
 

lib/connection-pool.js

@@ -64,7 +64,7 @@ module.exports = class ConnectionPool extends EventEmitter {
       return
     }
 
-    const session = new ConnectionSession(this, stream)
+    const session = new ConnectionRef(this, stream)
 
     const keyString = stream.remotePublicKey.toString('hex')
 
@@ -123,10 +123,23 @@ module.exports = class ConnectionPool extends EventEmitter {
   }
 }
 
-class ConnectionSession {
+class ConnectionRef {
   constructor (pool, stream) {
     this._pool = pool
     this._stream = stream
+    this._refs = 0
+  }
+
+  active () {
+    this._refs++
+  }
+
+  inactive () {
+    this._refs--
+  }
+
+  release () {
+    this._stream.destroy()
   }
 }