couple.js

/* jshint node: true */
'use strict';

var async = require('async');
var cleanup = require('./cleanup');
var monitor = require('./monitor');
var detect = require('./detect');
var CLOSED_STATES = [ 'closed', 'failed' ];

// track the various supported CreateOffer / CreateAnswer contraints
// that we recognize and allow
var OFFER_ANSWER_CONSTRAINTS = [
  'offerToReceiveVideo',
  'offerToReceiveAudio',
  'voiceActivityDetection',
  'iceRestart'
];

/**
  ### rtc/couple

  #### couple(pc, targetId, signaller, opts?)

  Couple a WebRTC connection with another webrtc connection identified by
  `targetId` via the signaller.

  The following options can be provided in the `opts` argument:

  - `sdpfilter` (default: null)

    A simple function for filtering SDP as part of the peer
    connection handshake (see the Using Filters details below).

  ##### Example Usage

  ```js
  var couple = require('rtc/couple');

  couple(pc, '54879965-ce43-426e-a8ef-09ac1e39a16d', signaller);
  ```

  ##### Using Filters

  In certain instances you may wish to modify the raw SDP that is provided
  by the `createOffer` and `createAnswer` calls.  This can be done by passing
  a `sdpfilter` function (or array) in the options.  For example:

  ```js
  // run the sdp from through a local tweakSdp function.
  couple(pc, '54879965-ce43-426e-a8ef-09ac1e39a16d', signaller, {
    sdpfilter: tweakSdp
  });
  ```

**/
function couple(pc, targetId, signaller, opts) {
  var debugLabel = (opts || {}).debugLabel || 'rtc';
  var debug = require('cog/logger')(debugLabel + '/couple');

  // create a monitor for the connection
  var mon = monitor(pc, targetId, signaller, opts);
  var queuedCandidates = [];
  var sdpFilter = (opts || {}).sdpfilter;
  var reactive = (opts || {}).reactive;
  var offerTimeout;
  var endOfCandidates = true;

  // configure the time to wait between receiving a 'disconnect'
  // iceConnectionState and determining that we are closed
  var disconnectTimeout = (opts || {}).disconnectTimeout || 10000;
  var disconnectTimer;

  // if the signaller does not support this isMaster function throw an
  // exception
  if (typeof signaller.isMaster != 'function') {
    throw new Error('rtc-signaller instance >= 0.14.0 required');
  }

  // initilaise the negotiation helpers
  var isMaster = signaller.isMaster(targetId);

  var createOffer = prepNegotiate(
    'createOffer',
    isMaster,
    [ checkStable ]
  );

  var createAnswer = prepNegotiate(
    'createAnswer',
    true,
    []
  );

  // initialise the processing queue (one at a time please)
  var q = async.queue(function(task, cb) {
    // if the task has no operation, then trigger the callback immediately
    if (typeof task.op != 'function') {
      return cb();
    }

    // process the task operation
    task.op(task, cb);
  }, 1);

  // initialise session description and icecandidate objects
  var RTCSessionDescription = (opts || {}).RTCSessionDescription ||
    detect('RTCSessionDescription');

  var RTCIceCandidate = (opts || {}).RTCIceCandidate ||
    detect('RTCIceCandidate');

  function abort(stage, sdp, cb) {
    return function(err) {
      // log the error
      console.error('rtc/couple error (' + stage + '): ', err);

      if (typeof cb == 'function') {
        cb(err);
      }
    };
  }

  function applyCandidatesWhenStable() {
    if (pc.signalingState == 'stable' && pc.remoteDescription) {
      debug('signaling state = stable, applying queued candidates');
      mon.removeListener('change', applyCandidatesWhenStable);

      // apply any queued candidates
      queuedCandidates.splice(0).forEach(function(data) {
        debug('applying queued candidate', data);

        try {
          pc.addIceCandidate(new RTCIceCandidate(data));
        }
        catch (e) {
          debug('invalidate candidate specified: ', data);
        }
      });
    }
  }

  function checkNotConnecting(negotiate) {
    if (pc.iceConnectionState != 'checking') {
      return true;
    }

    debug('connection state is checking, will wait to create a new offer');
    mon.once('connected', function() {
      q.push({ op: negotiate });
    });

    return false;
  }

  function checkStable(negotiate) {
    if (pc.signalingState === 'stable') {
      return true;
    }

    debug('cannot create offer, signaling state != stable, will retry');
    mon.on('change', function waitForStable() {
      if (pc.signalingState === 'stable') {
        q.push({ op: negotiate });
      }

      mon.removeListener('change', waitForStable);
    });

    return false;
  }

  function decouple() {
    debug('decoupling ' + signaller.id + ' from ' + targetId);

    // stop the monitor
    mon.removeAllListeners();
    mon.stop();

    // cleanup the peerconnection
    cleanup(pc);

    // remove listeners
    signaller.removeListener('sdp', handleSdp);
    signaller.removeListener('candidate', handleRemoteCandidate);
    signaller.removeListener('negotiate', handleNegotiateRequest);
  }

  function generateConstraints(methodName) {
    var constraints = {};

    function reformatConstraints() {
      var tweaked = {};

      Object.keys(constraints).forEach(function(param) {
        var sentencedCased = param.charAt(0).toUpperCase() + param.substr(1);
        tweaked[sentencedCased] = constraints[param];
      });

      // update the constraints to match the expected format
      constraints = {
        mandatory: tweaked
      };
    }

    // TODO: customize behaviour based on offer vs answer

    // pull out any valid
    OFFER_ANSWER_CONSTRAINTS.forEach(function(param) {
      var sentencedCased = param.charAt(0).toUpperCase() + param.substr(1);

      // if we have no opts, do nothing
      if (! opts) {
        return;
      }
      // if the parameter has been defined, then add it to the constraints
      else if (opts[param] !== undefined) {
        constraints[param] = opts[param];
      }
      // if the sentenced cased version has been added, then use that
      else if (opts[sentencedCased] !== undefined) {
        constraints[param] = opts[sentencedCased];
      }
    });

    // TODO: only do this for the older browsers that require it
    reformatConstraints();

    return constraints;
  }

  function prepNegotiate(methodName, allowed, preflightChecks) {
    var constraints = generateConstraints(methodName);

    // ensure we have a valid preflightChecks array
    preflightChecks = [].concat(preflightChecks || []);

    return function negotiate(task, cb) {
      var checksOK = true;

      // if the task is not allowed, then send a negotiate request to our
      // peer
      if (! allowed) {
        signaller.to(targetId).send('/negotiate');
        return cb();
      }

      // if the connection is closed, then abort
      if (isClosed()) {
        return cb(new Error('connection closed, cannot negotiate'));
      }

      // run the preflight checks
      preflightChecks.forEach(function(check) {
        checksOK = checksOK && check(negotiate);
      });

      // if the checks have not passed, then abort for the moment
      if (! checksOK) {
        debug('preflight checks did not pass, aborting ' + methodName);
        return cb();
      }

      // create the offer
      debug('calling ' + methodName);
      // debug('gathering state = ' + conn.iceGatheringState);
      // debug('connection state = ' + conn.iceConnectionState);
      // debug('signaling state = ' + conn.signalingState);

      pc[methodName](
        function(desc) {

          // if a filter has been specified, then apply the filter
          if (typeof sdpFilter == 'function') {
            desc.sdp = sdpFilter(desc.sdp, pc, methodName);
          }

          q.push({ op: queueLocalDesc(desc) });
          cb();
        },

        // on error, abort
        abort(methodName, '', cb),

        // include the appropriate constraints
        constraints
      );
    };
  }

  function handleConnectionClose() {
    debug('captured pc close, iceConnectionState = ' + pc.iceConnectionState);
    decouple();
  }

  function handleDisconnect() {
    debug('captured pc disconnect, monitoring connection status');

    // start the disconnect timer
    disconnectTimer = setTimeout(function() {
      debug('manually closing connection after disconnect timeout');
      pc.close();
    }, disconnectTimeout);

    mon.on('change', handleDisconnectAbort);
  }

  function handleDisconnectAbort() {
    debug('connection state changed to: ' + pc.iceConnectionState);
    resetDisconnectTimer();

    // if we have a closed or failed status, then close the connection
    if (CLOSED_STATES.indexOf(pc.iceConnectionState) >= 0) {
      return mon.emit('closed');
    }

    mon.once('disconnect', handleDisconnect);
  };

  function handleLocalCandidate(evt) {
    if (evt.candidate) {
      resetDisconnectTimer();

      signaller.to(targetId).send('/candidate', evt.candidate);
      endOfCandidates = false;
    }
    else if (! endOfCandidates) {
      endOfCandidates = true;
      debug('ice gathering state complete');
      signaller.to(targetId).send('/endofcandidates', {});
    }
  }

  function handleNegotiateRequest(src) {
    if (src.id === targetId) {
      debug('got negotiate request from ' + targetId + ', creating offer');
      q.push({ op: createOffer });
    }
  }

  function handleRemoteCandidate(data, src) {
    if ((! src) || (src.id !== targetId)) {
      return;
    }

    // queue candidates while the signaling state is not stable
    if (pc.signalingState != 'stable' || (! pc.remoteDescription)) {
      debug('queuing candidate');
      queuedCandidates.push(data);

      mon.removeListener('change', applyCandidatesWhenStable);
      mon.on('change', applyCandidatesWhenStable);
      return;
    }

    try {
      pc.addIceCandidate(new RTCIceCandidate(data));
    }
    catch (e) {
      debug('invalidate candidate specified: ', data);
    }
  }

  function handleSdp(data, src) {
    var abortType = data.type === 'offer' ? 'createAnswer' : 'createOffer';

    // if the source is unknown or not a match, then abort
    if ((! src) || (src.id !== targetId)) {
      return;
    }

    // prioritize setting the remote description operation
    q.push({ op: function(task, cb) {
      if (isClosed()) {
        return cb(new Error('pc closed: cannot set remote description'));
      }

      // update the remote description
      // once successful, send the answer
      debug('setting remote description');
      pc.setRemoteDescription(
        new RTCSessionDescription(data),

        function() {
          // create the answer
          if (data.type === 'offer') {
            queue(createAnswer)();
          }

          // trigger the callback
          cb();
        },

        abort(abortType, data.sdp, cb)
      );
    }});
  }

  function isClosed() {
    return CLOSED_STATES.indexOf(pc.iceConnectionState) >= 0;
  }

  function queue(negotiateTask) {
    return function() {
      q.push([
        { op: negotiateTask }
      ]);
    };
  }

  function queueLocalDesc(desc) {
    return function setLocalDesc(task, cb) {
      if (isClosed()) {
        return cb(new Error('connection closed, aborting'));
      }

      // initialise the local description
      debug('setting local description');
      pc.setLocalDescription(
        desc,

        // if successful, then send the sdp over the wire
        function() {
          // send the sdp
          signaller.to(targetId).send('/sdp', desc);

          // callback
          cb();
        },

        // abort('setLocalDesc', desc.sdp, cb)
        // on error, abort
        function(err) {
          debug('error setting local description', err);
          debug(desc.sdp);
          // setTimeout(function() {
          //   setLocalDesc(task, cb, (retryCount || 0) + 1);
          // }, 500);

          cb(err);
        }
      );
    };
  }

  function resetDisconnectTimer() {
    mon.removeListener('change', handleDisconnectAbort);

    // clear the disconnect timer
    debug('reset disconnect timer, state: ' + pc.iceConnectionState);
    clearTimeout(disconnectTimer);
  }

  // when regotiation is needed look for the peer
  if (reactive) {
    pc.onnegotiationneeded = function() {
      debug('renegotiation required, will create offer in 50ms');
      clearTimeout(offerTimeout);
      offerTimeout = setTimeout(queue(createOffer), 50);
    };
  }

  pc.onicecandidate = handleLocalCandidate;

  // when we receive sdp, then
  signaller.on('sdp', handleSdp);
  signaller.on('candidate', handleRemoteCandidate);

  // if this is a master connection, listen for negotiate events
  if (isMaster) {
    signaller.on('negotiate', handleNegotiateRequest);
  }

  // when the connection closes, remove event handlers
  mon.once('closed', handleConnectionClose);
  mon.once('disconnected', handleDisconnect);

  // patch in the create offer functions
  mon.createOffer = queue(createOffer);

  return mon;
}

module.exports = couple;