Decouple Connection logic in ErizoJS (#1164)

erizo_controller/erizoJS/adapt_schemes/notify-slideshow.js

@@ -25,8 +25,7 @@ exports.MonitorSubscriber = function (log) {
   };
 
 
-  that.monitorMinVideoBw = function(mediaStream, callback, idPub,
-     idSub, options, erizoJsController) {
+  that.monitorMinVideoBw = function(mediaStream, callback) {
     mediaStream.bwValues = [];
     var ticks = 0;
     var retries = 0;
@@ -87,10 +86,10 @@ exports.MonitorSubscriber = function (log) {
             average = 0;
             lastAverage = 0;
             mediaStream.minVideoBW = false;
-            erizoJsController.setSlideShow(true, idSub, idPub);
+            callback('scheme-slideshow-change', {enabled: true});
             callback('callback', {type: 'bandwidthAlert',
-            message: 'slideshow',
-            bandwidth: average});
+                                  message: 'slideshow',
+                                  bandwidth: average});
             clearInterval(mediaStream.monitorInterval);
             break;
           default:

erizo_controller/erizoJS/models/Client.js

@@ -25,14 +25,14 @@ class Client {
     return id;
   }
 
-  getOrCreateConnection() {
+  getOrCreateConnection(options) {
     let connection = this.connections.values().next().value;
     log.info(`message: getOrCreateConnection for clientId ${this.id}`);
     if (!this.singlePc || !connection) {
-      let id = this._getNewConnectionClientId(); 
-      connection = new Connection(id, this.threadPool, this.ioThreadPool);
+      let id = this._getNewConnectionClientId();
+      connection = new Connection(id, this.threadPool, this.ioThreadPool, options);
       this.addConnection(connection);
-    } 
+    }
     return connection;
   }
 
@@ -52,7 +52,7 @@ class Client {
     log.debug(`message: maybeCloseConnection, connectionId: ${id}`);
     if (connection !== undefined) {
       // ExternalInputs don't have mediaStreams but have to be closed
-      if (!connection.mediaStreams || connection.mediaStreams.size === 0) {
+      if (connection.getNumMediaStreams() === 0) {
         log.info(`message: closing empty connection, clientId: ${this.id}` +
         ` connectionId: ${connection.id}`);
         connection.close();

erizo_controller/erizoJS/models/Connection.js

@@ -1,12 +1,24 @@
 /*global require, exports*/
 'use strict';
-var addon = require('./../../../erizoAPI/build/Release/addon');
-var logger = require('./../../common/logger').logger;
-var log = logger.getLogger('Connection');
+const events = require('events');
+const addon = require('./../../../erizoAPI/build/Release/addon');
+const logger = require('./../../common/logger').logger;
+const SessionDescription = require('./SessionDescription');
+const log = logger.getLogger('Connection');
 
-class Connection {
+const CONN_INITIAL        = 101,
+      // CONN_STARTED        = 102,
+      CONN_GATHERED       = 103,
+      CONN_READY          = 104,
+      CONN_FINISHED       = 105,
+      CONN_CANDIDATE      = 201,
+      CONN_SDP            = 202,
+      CONN_FAILED         = 500,
+      WARN_BAD_CONNECTION = 502;
 
-  constructor (id, threadPool, ioThreadPool) {
+class Connection extends events.EventEmitter {
+  constructor (id, threadPool, ioThreadPool, options = {}) {
+    super();
     log.info(`message: constructor, id: ${id}`);
     this.id = id;
     this.threadPool = threadPool;
@@ -15,6 +27,9 @@ class Connection {
     //  {id: stream}
     this.mediaStreams = new Map();
     this.wrtc = this._createWrtc();
+    this.initialized = false;
+    this.options = options;
+    this.metadata = this.options.metadata || {};
   }
 
   _getMediaConfiguration(mediaConfiguration = 'default') {
@@ -52,42 +67,124 @@ class Connection {
       global.config.erizo.turnpass,
       global.config.erizo.networkinterface);
 
+    if (this.metadata) {
+        wrtc.setMetadata(JSON.stringify(this.metadata));
+    }
     return wrtc;
   }
 
-  _createMediaStream (id) {
+  _createMediaStream (id, mediaStreamOptions) {
     log.debug(`message: _createMediaStream, connectionId: ${this.id}, mediaStreamId: ${id}`);
     const mediaStream = new addon.MediaStream(this.threadPool, this.wrtc, id,
       this._getMediaConfiguration(this.mediaConfiguration));
     mediaStream.id = id;
+    if (mediaStreamOptions) {
+      mediaStream.metadata = mediaStreamOptions;
+      mediaStream.setMetadata(JSON.stringify(mediaStreamOptions));
+    }
     return mediaStream;
   }
 
-  addMediaStream(id) {
+  init() {
+    if (this.initialized) {
+      return;
+    }
+    this.initialized = true;
+    log.debug(`message: Init Connection, connectionId: ${this.id} `+
+              `${logger.objectToLog(this.options)}`);
+
+    this.wrtc.init((newStatus, mess) => {
+      log.info('message: WebRtcConnection status update, ' +
+               'id: ' + this.id + ', status: ' + newStatus +
+                ', ' + logger.objectToLog(this.metadata));
+      switch(newStatus) {
+        case CONN_INITIAL:
+          this.emit('status_event', {type: 'started'}, newStatus);
+          break;
+
+        case CONN_SDP:
+        case CONN_GATHERED:
+          this.wrtc.localDescription = new SessionDescription(this.wrtc.getLocalDescription());
+          const sdp = this.wrtc.localDescription.getSdp();
+          mess = sdp.toString();
+          mess = mess.replace(this.options.privateRegexp, this.options.publicIP);
+
+          const info = {type: this.options.createOffer ? 'offer' : 'answer', sdp: mess};
+          this.emit('status_event', info, newStatus);
+          break;
+
+        case CONN_CANDIDATE:
+          mess = mess.replace(this.options.privateRegexp, this.options.publicIP);
+          this.emit('status_event', {type: 'candidate', candidate: mess}, newStatus);
+          break;
+
+        case CONN_FAILED:
+          log.warn('message: failed the ICE process, ' + 'code: ' + WARN_BAD_CONNECTION +
+                   ', id: ' + this.id);
+          this.emit('status_event', {type: 'failed', sdp: mess}, newStatus);
+          break;
+
+        case CONN_READY:
+          log.debug('message: connection ready, ' + 'id: ' + this.id +
+                    ', ' + 'status: ' + newStatus);
+          this.emit('status_event', {type: 'ready'}, newStatus);
+          break;
+      }
+    });
+    if (this.options.createOffer) {
+      log.debug('message: create offer requested, id:', this.id);
+      const audioEnabled = this.options.createOffer.audio;
+      const videoEnabled = this.options.createOffer.video;
+      const bundle = this.options.createOffer.bundle;
+      this.wrtc.createOffer(videoEnabled, audioEnabled, bundle);
+    }
+    this.emit('status_event', {type: 'initializing'});
+  }
+
+  addMediaStream(id, mediaStreamOptions) {
     log.info(`message: addMediaStream, connectionId: ${this.id}, mediaStreamId: ${id}`);
     if (this.mediaStreams.get(id) === undefined) {
-      const mediaStream = this._createMediaStream(id);
+      const mediaStream = this._createMediaStream(id, mediaStreamOptions);
       this.wrtc.addMediaStream(mediaStream);
       this.mediaStreams.set(id, mediaStream);
     }
   }
+
   removeMediaStream(id) {
-    if(this.mediaStreams.get(id) !== undefined) {
+    if (this.mediaStreams.get(id) !== undefined) {
       this.wrtc.removeMediaStream(id);
       this.mediaStreams.get(id).close();
       this.mediaStreams.delete(id);
-      log.debug(`removed mediaStreamId ${id}, remaining size ${this.mediaStreams.size}`);
+      log.debug(`removed mediaStreamId ${id}, remaining size ${this.getNumMediaStreams()}`);
     } else {
       log.error(`message: Trying to remove mediaStream not found, id: ${id}`);
     }
   }
-
+
+  setRemoteDescription(sdp) {
+    this.remoteDescription = new SessionDescription(sdp, this.mediaConfiguration);
+    this.wrtc.setRemoteDescription(this.remoteDescription.connectionDescription);
+  }
+
+  addRemoteCandidate(candidate) {
+    this.wrtc.addRemoteCandidate(candidate.sdpMid, candidate.sdpMLineIndex, candidate.candidate);
+  }
+
   getMediaStream(id) {
     return this.mediaStreams.get(id);
   }
 
+  getNumMediaStreams() {
+    return this.mediaStreams.size;
+  }
+
   close() {
     log.info(`message: Closing connection ${this.id}`);
+    log.info(`message: WebRtcConnection status update, id: ${this.id}, status: ${CONN_FINISHED}, ` +
+            `${logger.objectToLog(this.metadata)}`);
+    if (this._onConnectionStatusEventListener) {
+      this.removeListener('status_event', this._onConnectionStatusEventListener);
+    }
     this.mediaStreams.forEach((mediaStream, id) => {
       log.debug(`message: Closing mediaStream, connectionId : ${this.id}, `+
         `mediaStreamId: ${id}`);
@@ -97,6 +194,6 @@ class Connection {
     delete this.mediaStreams;
     delete this.wrtc;
   }
-  
+
 }
 exports.Connection = Connection;

erizo_controller/erizoJS/models/Node.js

@@ -0,0 +1,73 @@
+/*global require, exports*/
+'use strict';
+const EventEmitter = require('events').EventEmitter;
+const Helpers = require('./Helpers');
+const logger = require('./../../common/logger').logger;
+
+const log = logger.getLogger('Node');
+
+const WARN_PRECOND_FAILED = 412;
+
+class Node extends EventEmitter {
+  constructor(clientId, streamId, options = {}) {
+    super();
+    this.clientId = clientId;
+    this.streamId = streamId;
+    this.erizoStreamId = Helpers.getErizoStreamId(clientId, streamId);
+    this.options = options;
+  }
+
+  getStats(label, stats) {
+    const promise = new Promise((resolve) => {
+      if (!this.mediaStream || !this.connection) {
+        resolve();
+        return;
+      }
+      this.mediaStream.getStats((statsString) => {
+        const unfilteredStats = JSON.parse(statsString);
+        unfilteredStats.metadata = this.connection.metadata;
+        stats[label] = unfilteredStats;
+        resolve();
+      });
+    });
+    return promise;
+  }
+
+  _onMonitorMinVideoBWCallback(type, message) {
+    this.emit(type, message);
+  }
+
+  initMediaStream() {
+    if (!this.mediaStream) {
+      return;
+    }
+    const mediaStream = this.mediaStream;
+    if (mediaStream.minVideoBW) {
+      var monitorMinVideoBw = {};
+      if (mediaStream.scheme) {
+        try{
+          monitorMinVideoBw = require('../adapt_schemes/' + mediaStream.scheme)
+            .MonitorSubscriber(log);
+        } catch (e) {
+          log.warn('message: could not find custom adapt scheme, ' +
+                   'code: ' + WARN_PRECOND_FAILED + ', ' +
+                   'id:' + this.clientId + ', ' +
+                   'scheme: ' + mediaStream.scheme + ', ' +
+                   logger.objectToLog(this.options.metadata));
+        }
+      } else {
+        monitorMinVideoBw = require('../adapt_schemes/notify').MonitorSubscriber(log);
+      }
+      monitorMinVideoBw(mediaStream, this._onMonitorMinVideoBWCallback.bind(this), this.clientId);
+    }
+
+    if (global.config.erizoController.report.rtcp_stats) {  // jshint ignore:line
+        log.debug('message: RTCP Stat collection is active');
+        mediaStream.getPeriodicStats((newStats) => {
+            this.emit('periodic_stats', newStats);
+        });
+    }
+  }
+}
+
+exports.Node = Node;