adjust moc latencies upon messages, closes and connections

bin/n2n-wrtc.bundle.js

@@ -785,25 +785,26 @@ class N2N extends EventEmitter {
             this._signalDisconnect(peerId, true, false)
             return this.send(this.options.n2n.protocol, peerId, {
               type: events.n2n.DISCONNECT
+            }).catch(e => {
+              console.warn('[%s] cannot send the message to %s', this.id, peerId, e)
             }).then(() => {
               return p.socket.disconnect(this.options.socket)
-            }).catch(e => {
-              return p.socket.disconnect()
             })
           } else {
             this._signalDisconnect(peerId, true, false) // signal disconnect
             return Promise.resolve()
           }
         }).catch(e => {
+          console.error(e)
           console.warn('[%s] cannot send the message to %s', this.id, peerId)
           if (this.livingOutview.get(peerId).occurences === 0) {
             this._signalDisconnect(peerId, true, false)
             return this.send(this.options.n2n.protocol, peerId, {
               type: events.n2n.DISCONNECT
-            }).then(() => {
-              return p.socket.disconnect()
             }).catch(e => {
-              return p.socket.disconnect()
+              console.warn('[%s] cannot send the message to %s', this.id, peerId, e)
+            }).then(() => {
+              return p.socket.disconnect(this.options.socket)
             })
           } else {
             this._signalDisconnect(peerId, true, false) // signal disconnect
@@ -2284,10 +2285,7 @@ class DirectSignaling extends SignalingAPI {
 
         const socket = this.parent.createNewSocket(this.parent.options.socket, peerId, true)
         socket.on('error', (error) => {
-          this.parent._manageError(error, peerId, true, (e) => {
-            this._debug('[%s][%s][_connectToUs] receive an error during the connection %s...', this.parent.id, jobId, id, e)
-            reject(e)
-          }, 'direct')
+          this.parent._manageError(error, peerId, true, reject, 'direct')
         })
         socket.on(events.socket.EMIT_OFFER, (offer) => {
           const off = {
@@ -3209,7 +3207,8 @@ class Manager {
     }
     this.manager = new Map()
     this._options = {
-      latency: (send) => { setTimeout(send, 0) }
+      latency: 0,
+      connections: 50
     }
     debugManager('manager initialized')
   }
@@ -3228,17 +3227,22 @@ class Manager {
   connect (from, to) {
     debugManager('peer connected from/to: ', from, to)
     this.manager.get(to)._connectWith(from)
-    this.manager.get(from)._connectWith(to)
+    setTimeout(() => {
+      this.manager.get(from)._connectWith(to)
+    }, this._options.connections)
   }
   // @private
   destroy (from, to) {
     debugManager('peer disconnected from/to: ', from, to)
-    if (this.manager.get(from)) {
+    if (this.manager.has(from)) {
       this.manager.get(from)._close()
     }
-    if (this.manager.get(to)) {
-      this.manager.get(to)._close()
-    }
+    // disconnect the overside of the socket after one second
+    setTimeout(() => {
+      if (this.manager.has(to)) {
+        this.manager.get(to)._close()
+      }
+    }, this._options.connections)
   }
   // @private
   send (from, to, msg, retry = 0) {
@@ -3247,9 +3251,12 @@ class Manager {
   // @private
   _send (from, to, msg, retry = 0) {
     try {
-      if (!this.manager.has(from) || !this.manager.has(to)) throw new Error('need a (from) and (to) peer.')
-      this.manager.get(to).emit('data', msg)
-      this._statistics.message++
+      if (!this.manager.has(from) || !this.manager.has(to)) {
+        throw new Error('need a (from) and (to) peer.')
+      } else {
+        this._statistics.message++
+        this.manager.get(to)._receive(msg)
+      }
     } catch (e) {
       throw new Error('cannot send the message. perhaps your destination is not reachable.', e)
     }
@@ -3284,14 +3291,16 @@ module.exports = class SimplePeerAbstract extends EventEmitter {
       })
     }
     this._manager.set(this.id, this)
-    this.on('internal_close', () => {
-      this._manager.manager.delete(this.id)
-    })
   }
   // @private
   static get manager () {
     return manager
   }
+  _receive (data) {
+    setTimeout(() => {
+      this.emit('data', data)
+    }, this._manager._options.latency)
+  }
   send (data) {
     if (!this.connectedWith) {
       this.messageBuffer.push(data)
@@ -3327,7 +3336,7 @@ module.exports = class SimplePeerAbstract extends EventEmitter {
   }
   // @private
   _close () {
-    this.emit('internal_close')
+    this._manager.manager.delete(this.id)
     debugManager('[%s] is closed.', this.id)
     this.emit('close')
   }

lib/main.js

@@ -555,25 +555,26 @@ class N2N extends EventEmitter {
             this._signalDisconnect(peerId, true, false)
             return this.send(this.options.n2n.protocol, peerId, {
               type: events.n2n.DISCONNECT
+            }).catch(e => {
+              console.warn('[%s] cannot send the message to %s', this.id, peerId, e)
             }).then(() => {
               return p.socket.disconnect(this.options.socket)
-            }).catch(e => {
-              return p.socket.disconnect()
             })
           } else {
             this._signalDisconnect(peerId, true, false) // signal disconnect
             return Promise.resolve()
           }
         }).catch(e => {
+          console.error(e)
           console.warn('[%s] cannot send the message to %s', this.id, peerId)
           if (this.livingOutview.get(peerId).occurences === 0) {
             this._signalDisconnect(peerId, true, false)
             return this.send(this.options.n2n.protocol, peerId, {
               type: events.n2n.DISCONNECT
-            }).then(() => {
-              return p.socket.disconnect()
             }).catch(e => {
-              return p.socket.disconnect()
+              console.warn('[%s] cannot send the message to %s', this.id, peerId, e)
+            }).then(() => {
+              return p.socket.disconnect(this.options.socket)
             })
           } else {
             this._signalDisconnect(peerId, true, false) // signal disconnect

lib/signaling/direct.js

@@ -192,10 +192,7 @@ class DirectSignaling extends SignalingAPI {
 
         const socket = this.parent.createNewSocket(this.parent.options.socket, peerId, true)
         socket.on('error', (error) => {
-          this.parent._manageError(error, peerId, true, (e) => {
-            this._debug('[%s][%s][_connectToUs] receive an error during the connection %s...', this.parent.id, jobId, id, e)
-            reject(e)
-          }, 'direct')
+          this.parent._manageError(error, peerId, true, reject, 'direct')
         })
         socket.on(events.socket.EMIT_OFFER, (offer) => {
           const off = {

lib/sockets/webrtc-moc.js

@@ -26,7 +26,8 @@ class Manager {
     }
     this.manager = new Map()
     this._options = {
-      latency: (send) => { setTimeout(send, 0) }
+      latency: 0,
+      connections: 50
     }
     debugManager('manager initialized')
   }
@@ -45,17 +46,22 @@ class Manager {
   connect (from, to) {
     debugManager('peer connected from/to: ', from, to)
     this.manager.get(to)._connectWith(from)
-    this.manager.get(from)._connectWith(to)
+    setTimeout(() => {
+      this.manager.get(from)._connectWith(to)
+    }, this._options.connections)
   }
   // @private
   destroy (from, to) {
     debugManager('peer disconnected from/to: ', from, to)
-    if (this.manager.get(from)) {
+    if (this.manager.has(from)) {
       this.manager.get(from)._close()
     }
-    if (this.manager.get(to)) {
-      this.manager.get(to)._close()
-    }
+    // disconnect the overside of the socket after one second
+    setTimeout(() => {
+      if (this.manager.has(to)) {
+        this.manager.get(to)._close()
+      }
+    }, this._options.connections)
   }
   // @private
   send (from, to, msg, retry = 0) {
@@ -64,9 +70,12 @@ class Manager {
   // @private
   _send (from, to, msg, retry = 0) {
     try {
-      if (!this.manager.has(from) || !this.manager.has(to)) throw new Error('need a (from) and (to) peer.')
-      this.manager.get(to).emit('data', msg)
-      this._statistics.message++
+      if (!this.manager.has(from) || !this.manager.has(to)) {
+        throw new Error('need a (from) and (to) peer.')
+      } else {
+        this._statistics.message++
+        this.manager.get(to)._receive(msg)
+      }
     } catch (e) {
       throw new Error('cannot send the message. perhaps your destination is not reachable.', e)
     }
@@ -101,14 +110,16 @@ module.exports = class SimplePeerAbstract extends EventEmitter {
       })
     }
     this._manager.set(this.id, this)
-    this.on('internal_close', () => {
-      this._manager.manager.delete(this.id)
-    })
   }
   // @private
   static get manager () {
     return manager
   }
+  _receive (data) {
+    setTimeout(() => {
+      this.emit('data', data)
+    }, this._manager._options.latency)
+  }
   send (data) {
     if (!this.connectedWith) {
       this.messageBuffer.push(data)
@@ -144,7 +155,7 @@ module.exports = class SimplePeerAbstract extends EventEmitter {
   }
   // @private
   _close () {
-    this.emit('internal_close')
+    this._manager.manager.delete(this.id)
     debugManager('[%s] is closed.', this.id)
     this.emit('close')
   }

tests/connect4u-null-peer-test.js

@@ -1,5 +1,6 @@
 const N2N = require('../lib').N2N
 const assert = require('assert')
+const utils = require('../lib/utils')
 
 describe('[N2N] Offline connection', function () {
   this.timeout(2 * 60 * 1000)
@@ -30,6 +31,7 @@ describe('[N2N] Offline connection', function () {
     assert.strictEqual(c.livingOutview.has(b.id), false)
     assert.strictEqual(c.livingInview.get(b.id).occurences, 1)
     await b.connect4u(null, c.id)
+    await utils.timeout(500)
     assert.strictEqual(b.livingOutview.get(c.id).occurences, 2)
     assert.strictEqual(c.livingOutview.has(b.id), false)
     assert.strictEqual(c.livingInview.get(b.id).occurences, 2)

tests/crash-test.js

@@ -1,5 +1,6 @@
 const N2N = require('../lib').N2N
 const assert = require('assert')
+const utils = require('../lib/utils')
 
 describe('[N2N] Crash test', function () {
   this.timeout(10 * 1000)
@@ -31,6 +32,8 @@ describe('[N2N] Crash test', function () {
     await a.connect(b)
     await a.connect(b)
     await a.connect(b)
+    // because the moc is enabled, and the code is synchronous we need to wait for messages
+    await utils.timeout(500)
     return new Promise((resolve, reject) => {
       let o = 0
       const done = (occ) => {

tests/n2n-test.js

@@ -1,5 +1,6 @@
 const N2N = require('../lib').N2N
 const assert = require('assert')
+const utils = require('../lib/utils')
 const socket = {
   moc: true,
   tricle: true
@@ -142,6 +143,7 @@ describe('N2N connection', function () {
     assert.strictEqual(b.livingOutview.get(a.id).occurences, 1)
     assert.strictEqual(b.livingOutview.get(a.id).lock, 0)
     await a.disconnect()
+    await utils.timeout(500)
     assert.strictEqual(a.getNeighboursIds().length, 0)
     assert.strictEqual(b.getNeighboursIds().length, 1)
     assert.strictEqual(a.livingInview.size, 1)
@@ -187,13 +189,15 @@ describe('N2N connection', function () {
     assert.strictEqual(b.livingOutview.get(a.id).occurences, 1)
     assert.strictEqual(b.livingOutview.get(a.id).lock, 0)
     await a.disconnect()
+    await utils.timeout(500)
     assert.strictEqual(a.getNeighboursIds().length, 0)
     assert.strictEqual(b.getNeighboursIds().length, 1)
     assert.strictEqual(a.livingInview.size, 1)
     assert.strictEqual(a.livingOutview.size, 0)
     assert.strictEqual(b.livingInview.size, 0)
     assert.strictEqual(b.livingOutview.size, 1)
     await b.disconnect()
+    await utils.timeout(500)
     assert.strictEqual(a.getNeighboursIds().length, 0)
     assert.strictEqual(b.getNeighboursIds().length, 0)
     assert.strictEqual(a.livingInview.size, 0)