drachtio-srf/lib/dialog.js

const Emitter = require('events') ;
const assert = require('assert') ;
const only = require('only') ;
const methods = require('sip-methods') ;
const debug = require('debug')('drachtio:srf') ;
const SipError = require('./sip_error');
const { parseUri } = require('drachtio-sip').parser ;
const sdpTransform = require('sdp-transform');


/**
 * Class representing a SIP Dialog.
 *
 * Note that instances of this class are not created directly by your code;
 * rather they are returned from the {@link Srf#createUAC}, {@link Srf#createUAC}, and {@link Srf#createB2BUA}
 * @class
 * @extends EventEmitter
 */
class Dialog extends Emitter {

  /**
   * Constructor that is called internally by Srf when generating a Dialog instance.
   * @param {Srf} srf - Srf instance that created this dialog
   * @param {string} type - type of SIP dialog: 'uac', or 'uas'
   * @param {Dialog~Options} opts
   */
  constructor(srf, type, opts) {
    super() ;

    const types = ['uas', 'uac'] ;
    assert.ok(-1 !== types.indexOf(type), 'argument \'type\' must be one of ' + types.join(',')) ;

    this.srf = srf ;
    this.type = type ;
    this.req = opts.req ;
    this.res = opts.res ;
    this.agent = this.res.agent ;
    this.onHold = false ;
    this.connected = true ;
    this.queuedRequests = [];
    this._queueRequests = false;

    /**
     * sip properties that uniquely identify this Dialog
     * @type {Object}
     * @property {String} callId - SIP Call-ID
     * @property {String} localTag - tag generated by local side of the Dialog
     * @property {String} remoteTag  - tag generated by the remote side of the Dialog
     */
    this.sip = {
      callId: this.res.get('Call-ID'),
      remoteTag: 'uas' === type ?
        this.req.getParsedHeader('from').params.tag : this.res.getParsedHeader('to').params.tag,
      localTag: 'uas' === type ?
        opts.sent.getParsedHeader('to').params.tag : this.req.getParsedHeader('from').params.tag
    } ;

    /**
     * local side of the Dialog
     * @type {Object}
     * @property {String} uri - sip
     * @property {String} sdp - session description protocol
     */
    this.local = {
      uri: 'uas' === type ? opts.sent.getParsedHeader('Contact')[0].uri : this.req.uri,
      sdp: 'uas' === type ? opts.sent.body : this.req.body,
      contact: 'uas' === type ? opts.sent.get('Contact') : this.req.get('Contact')
    } ;

    /**
     * remote side of the Dialog
     * @type {Object}
     * @property {String} uri - sip
     * @property {String} sdp - session description protocol
     */
    this.remote = {
      uri: 'uas' === type ? this.req.getParsedHeader('Contact')[0].uri : this.res.getParsedHeader('Contact')[0].uri,
      sdp: 'uas' === type ? this.req.body : this.res.body
    } ;

    /*
     * subscriptions created by this dialog
     */
    this.subscriptions = [];

    // if this is a SUBSCRIBE, then we have our first subscription
    if (this.req.method === 'SUBSCRIBE') {
      this.addSubscription(this.req);
    }
  }

  get id() {
    return this.res.stackDialogId ;
  }

  get dialogType() {
    return this.req.method ;
  }

  get subscribeEvent() {
    return this.dialogType === 'SUBSCRIBE' ? this.req.get('Event') : null ;
  }

  get socket() {
    return this.req.socket;
  }

  set queueRequests(enqueue) {
    debug(`dialog ${this.id}: queueing requests: ${enqueue ? 'ON' : 'OFF'}`);
    this._queueRequests = enqueue;
    if (!enqueue) {
      // process any queued messages
      if (this.queuedRequests.length > 0) {
        setImmediate(() => {
          debug(`dialog ${this.id}: processing ${this.queuedRequests.length} queued requests`);
          this.queuedRequests.forEach(({req, res}) => this.handle(req, res));
          this.queueRequests = [];
        });
      }
    }
  }

  toJSON() {
    return only(this, 'id type sip local remote onHold') ;
  }

  toString() {
    return this.toJSON().toString() ;
  }

  getCountOfSubscriptions() {
    return this.subscriptions.length;
  }

  addSubscription(req) {
    const to = req.getParsedHeader('To');
    const u = parseUri(to.uri);
    debug(`Dialog#addSubscription: to header: ${JSON.stringify(to)}, uri ${JSON.stringify(u)}`);
    const entity = `${u.user}@${u.host}:${req.get('Event')}`;
    this.subscriptions.push(entity);
    debug(`Dialog#addSubscription: adding subscription ${entity}; current count ${this.subscriptions.length}`);
    return this.subscriptions.length;
  }

  removeSubscription(uri, event) {
    const u = parseUri(uri);
    const entity = `${u.user}@${u.host}:${event}`;
    const idx = this.subscriptions.indexOf(entity);
    if (-1 === idx) {
      console.error(`Dialog#removeSubscription: no subscription found for ${entity}: subs: ${this.subscriptions}`);
    }
    else {
      this.subscriptions.splice(idx, 1);
    }
    return this.subscriptions.length;
  }

  /**
   * destroy the sip dialog by generating a BYE request (in the case of INVITE dialog),
   * or NOTIFY (in the case of SUBSCRIBE)
   * @param {Object} [opts] configuration options
   * @param {Object} [opts.headers] SIP headers to add to the outgoing BYE or NOTIFY
   * @param  {function} [callback] if provided, callback with signature <code>(err, msg)</code>
   * that provides the BYE or NOTIFY message that was sent to terminate the dialog
   * @return {Promise|Dialog} if no callback is supplied, otherwise a reference to the Dialog
   */
  destroy(opts, callback) {
    opts = opts || {} ;
    if (typeof opts === 'function') {
      callback = opts ;
      opts = {} ;
    }
    this.queuedRequests = [];

    const __x = (callback) => {
      if (this.dialogType === 'INVITE') {
        this.agent.request({
          method: 'BYE',
          headers: opts.headers || {},
          stackDialogId: this.id,
          _socket: this.socket
        }, (err, bye) => {
          this.connected = false ;
          this.srf.removeDialog(this) ;
          callback(err, bye) ;
          this.removeAllListeners();
        }) ;
      }
      else if (this.dialogType === 'SUBSCRIBE') {
        opts.headers = opts.headers || {} ;
        opts.headers['subscription-state'] = 'terminated';
        opts.headers['event'] = this.subscribeEvent ;
        this.agent.request({
          method: 'NOTIFY',
          headers: opts.headers || {},
          stackDialogId: this.id,
          _socket: this.socket
        }, (err, notify) => {
          this.connected = false ;
          this.srf.removeDialog(this) ;
          callback(err, notify) ;
          this.removeAllListeners();
        }) ;
      }
    };

    if (callback) {
      __x(callback) ;
      return this ;
    }

    return new Promise((resolve, reject) => {
      __x((err, msg) => {
        if (err) return reject(err);
        resolve(msg);
      });
    });
  }

  /**
   * modify the dialog session by changing attributes of the media connection
   * @param  {string} sdp - 'hold', 'unhold', or a session description protocol
   * @param  {function} [callback] - callback invoked with signature <code>(err)</code> when operation has completed
   * @return {Promise|Dialog} if no callback is supplied, otherwise the function returns a reference to the Dialog
   */
  modify(sdp, callback) {

    const __x = (callback) => {
      switch (sdp) {
        case 'hold':
          this.local.sdp = this.local.sdp.replace(/a=sendrecv/, 'a=inactive') ;
          this.onHold = true ;
          break ;
        case 'unhold':
          if (this.onHold) {
            this.local.sdp = this.local.sdp.replace(/a=inactive/, 'a=sendrecv') ;
          }
          else {
            console.error('Dialog#modify: attempt to \'unhold\' session which is not on hold');
            return process.nextTick(() => {
              callback(new Error('attempt to unhold session that is not on hold'));
            }) ;
          }
          break ;
        default:
          this.local.sdp = sdp ;
          break ;
      }

      debug(`Dialog#modify: sending reINVITE for dialog id: ${this.id}, sdp: ${this.local.sdp}`) ;
      this.agent.request({
        method: 'INVITE',
        stackDialogId: this.id,
        body: this.local.sdp,
        _socket: this.socket,
        headers: {
          'Contact': this.local.contact
        }
      }, (err, req) => {
        if (err) return callback(err);
        req.on('response', (res, ack) => {
          debug(`Dialog#modifySession: received response to reINVITE with status ${res.status}`) ;
          if (res.status >= 200) {
            ack() ;
            if (200 === res.status) {
              this.remote.sdp = res.body ;
              return callback(null, res.body);
            }
            callback(new SipError(res.status, res.reason)) ;
          }
        }) ;
      }) ;
    };

    if (callback) {
      __x(callback) ;
      return this ;
    }

    return new Promise((resolve, reject) => {
      __x((err, sdp) => {
        if (err) return reject(err);
        resolve(sdp);
      });
    });
  }

  /**
   * send a request within a dialog.
   * Note that you may also call <code>request.info(..)</code> as a shortcut
   * to send an INFO message, <code>request.notify(..)</code>
   * to send a NOTIFY, etc..
   * @param {Object} [opts]
   * @param {string} opts.method - SIP method to use for the request
   * @param {Object} [opts.headers] - SIP headers to apply to the request
   * @param {string} [opts.body] - body of the SIP request
   * @param {function} [callback]  - callback invoked with signature <code>(err, req)</code>
   * when operation has completed
   * @return {Promise|Dialog} if no callback is supplied a Promise that resolves to the response received,
   * otherwise the function returns a reference to the Dialog
   */
  request(opts, callback) {
    assert.ok(typeof opts.method === 'string' &&
      -1 !== methods.indexOf(opts.method), '\'opts.method\' is required and must be a SIP method') ;

    const __x = (callback) => {
      const method = opts.method.toUpperCase() ;

      this.agent.request({
        method: method,
        stackDialogId: this.id,
        headers: opts.headers || {},
        auth: opts.auth,
        _socket: this.socket,
        body: opts.body
      }, (err, req) => {
        if (err) {
          return callback(err) ;
        }

        req.on('response', (res, ack) => {
          if ('BYE' === method) {
            this.srf.removeDialog(this) ;
          }
          if (res.status >= 200) {
            if ('INVITE' === method) ack() ;

            if (this.dialogType === 'SUBSCRIBE' && 'NOTIFY' === method &&
              /terminated/.test(req.get('Subscription-State'))) {
              debug('received response to a NOTIFY we sent terminating final subscription; dialog is ended') ;

              const from = req.getParsedHeader('From');
              if (this.removeSubscription(from.uri, req.get('Event')) === 0) {
                this.connected = false ;
                this.srf.removeDialog(this) ;
                this.emit('destroy', req) ;
                this.removeAllListeners();
              }
            }
            callback(null, res) ;
          }
        }) ;
      }) ;
    } ;

    if (callback) {
      __x(callback) ;
      return this ;
    }

    return new Promise((resolve, reject) => {
      __x((err, res) => {
        if (err) return reject(err);
        resolve(res);
      });
    });
  }

  handle(req, res) {
    debug(`dialog ${this.id}: handle: ${req.method}`) ;
    if (this._queueRequests === true) {
      debug(`dialog ${this.id}: queueing incoming request: ${req.method}`) ;
      this.queuedRequests.push({req, res});
      return;
    }
    const eventName = req.method.toLowerCase() ;
    switch (req.method) {
      case 'BYE':
        let reason = 'normal release';
        if (req.meta.source === 'application') {
          if (req.has('Reason')) {
            reason = req.get('Reason');
            const arr = /text=\"(.*)\"/.exec(reason);
            if (arr) reason = arr[1];
          }
        }
        this.connected = false ;
        this.srf.removeDialog(this) ;
        res.send(200) ;
        this.emit('destroy', req, reason) ;
        this.removeAllListeners();
        break ;

      case 'INVITE':
        const origRedacted = this.remote.sdp.replace(/^o=.*$/m, 'o=REDACTED') ;
        const newRedacted = req.body.replace(/^o=.*$/m, 'o=REDACTED') ;
        let refresh = false;
        try {
          if (this.listeners('refresh').length > 0) {
            const sdp1 = sdpTransform.parse(this.remote.sdp);
            const sdp2 = sdpTransform.parse(req.body);
            refresh = sdp1.origin.sessionId === sdp2.origin.sessionId &&
              sdp1.origin.sessionVersion === sdp2.origin.sessionVersion;
          }
        } catch (err) {
        }
        const hold = origRedacted.replace(/a=sendrecv\r\n/g, 'a=sendonly\r\n') === newRedacted &&
          this.listeners('hold').length > 0;
        const unhold = this.onHold === true &&
          origRedacted.replace(/a=sendonly\r\n/g, 'a=sendrecv\r\n') === newRedacted &&
          this.listeners('unhold').length > 0;
        const modify = !hold && !unhold && !refresh ;
        this.remote.sdp = req.body ;

        if (refresh) {
          this.emit('refresh', req);
        }
        else if (hold) {
          this.local.sdp = this.local.sdp.replace(/a=sendrecv\r\n/g, 'a=recvonly\r\n') ;
          this.onHold = true;
          this.emit('hold', req) ;
        }
        else if (unhold) {
          this.onHold = false;
          this.local.sdp = this.local.sdp.replace(/a=recvonly\r\n/g, 'a=sendrecv\r\n') ;
          this.emit('unhold', req) ;
        }
        if ((refresh || hold || unhold) || (modify && 0 === this.listeners('modify').length)) {
          debug('responding with 200 OK to reINVITE') ;
          res.send(200, {
            body: this.local.sdp,
            headers: {
              'Contact': this.local.contact,
              'Content-Type': 'application/sdp'
            }
          }) ;
        }
        else if (modify) {
          this.emit('modify', req, res) ;
        }
        break ;

      case 'NOTIFY':
        // if this is a subscribe dialog and subscription-state: terminated, then remove the subscription
        // and if there are no more subscriptions then remove the dialog
        if (this.dialogType === 'SUBSCRIBE' &&
          req.has('subscription-state') &&
          /terminated/.test(req.get('subscription-state'))) {

          setImmediate(() => {
            const to = req.getParsedHeader('to');
            if (this.removeSubscription(to.uri, req.get('Event')) === 0) {
              debug('received a NOTIFY with Subscription-State terminated for final subscription; dialog is ended') ;
              this.connected = false ;
              this.srf.removeDialog(this) ;
              this.emit('destroy', req) ;
            }
          }) ;
        }
        if (0 === this.listeners(eventName).length) {
          res.send(200) ;
        }
        else {
          this.emit(eventName, req, res) ;
        }
        break ;

      case 'INFO':
      case 'REFER':
      case 'OPTIONS':
      case 'MESSAGE':
      case 'PUBLISH':
      case 'UPDATE':

        // N.B.: this is because an app may be using the Promises version
        // of Srf#createUAS or Srf#createB2B and if so the 'then()' code
        // may be enqueued at the back of the job queue right now if a very
        // quick INFO or other request within the dialog just arrived.
        // We need the dialog to be resolved in the calling app first so it
        // has time to attach event handlers.
        setImmediate(() => {
          if (0 === this.listeners(eventName).length) res.send(200) ;
          else this.emit(eventName, req, res);
        });
        break ;

      case 'SUBSCRIBE':
        if (req.has('Expires') && 0 === parseInt(req.get('Expires'))) {
          res.send(202) ;
          this.emit('unsubscribe', req, 'unsubscribe') ;
        }
        else {
          if (0 === this.listeners('subscribe').length) {
            res.send(489, 'Bad Event - no dialog handler');
          }
          else this.emit('subscribe', req, res);
        }
        break;

      case 'ACK':
        setImmediate(() => this.emit('ack', req));
        break ;

      default:
        console.error(`Dialog#handle received invalid method within an INVITE dialog: ${req.method}`) ;
        res.send(501) ;
        break ;
    }

  }
}

module.exports = exports = Dialog ;

methods.forEach((method) => {
  Dialog.prototype[method.toLowerCase()] = (opts, cb) => {
    opts = opts || {} ;
    opts.method = method ;
    return this.request(opts, cb) ;
  };
}) ;

/**
 * a <code>destroy</code> event is triggered when the Dialog is torn down from the far end
 * @event Dialog#destroy
 * @param {Object} msg - incoming BYE request message
 */
/**
 * a <code>modify</code> event is triggered when the far end modifies the session by sending a re-INVITE.
 * When an application adds a handler for this event it must generate
 * the SIP response by calling <code>res.send</code> on the provided drachtio response object.
 * When no handler is found for this event a 200 OK with the current local SDP
 * will be automatically generated.
 *
 * @event Dialog#modify
 * @param {Object} req - drachtio request object
 * @param {Object} res - drachtio response object
 * @memberOf Dialog
 */
/**
 * a <code>refresh</code> event is triggered when the far end sends a session refresh.
 * There is no need for the application to respond to this event; this is purely
 * a notification.
 * @event Dialog#refresh
 * @param {Object} msg - incoming re-INVITE request message
 */
/**
 * an <code>info</code> event is triggered when the far end sends an INFO message.
 * When an application adds a handler for this event it must generate
 * the SIP response by calling <code>res.send</code> on the provided drachtio response object.
 * When no handler is found for this event a 200 OK will be automatically generated.
 * @event Dialog#info
 * @param {Object} req - drachtio request object
 * @param {Object} res - drachtio response object
 */
/**
 * a <code>notify</code> event is triggered when the far end sends a NOTIFY message.
 * When an application adds a handler for this event it must generate
 * the SIP response by calling <code>res.send</code> on the provided drachtio response object.
 * When no handler is found for this event a 200 OK will be automatically generated.
 * @event Dialog#notify
 * @param {Object} req - drachtio request object
 * @param {Object} res - drachtio response object
 */
/**
 * an <code>options</code> event is triggered when the far end sends an OPTIONS message.
 * When an application adds a handler for this event it must generate
 * the SIP response by calling <code>res.send</code> on the provided drachtio response object.
 * When no handler is found for this event a 200 OK will be automatically generated.
 * @event Dialog#options
 * @param {Object} req - drachtio request object
 * @param {Object} res - drachtio response object
 */
/**
 * an <code>update</code> event is triggered when the far end sends an UPDATE message.
 * When an application adds a handler for this event it must generate
 * the SIP response by calling <code>res.send</code> on the provided drachtio response object.
 * When no handler is found for this event a 200 OK will be automatically generated.
 * @event Dialog#update
 * @param {Object} req - drachtio request object
 * @param {Object} res - drachtio response object
 */
/**
 * a <code>refer</code> event is triggered when the far end sends a REFER message.
 * When an application adds a handler for this event it must generate
 * the SIP response by calling <code>res.send</code> on the provided drachtio response object.
 * When no handler is found for this event a 200 OK will be automatically generated.
 * @event Dialog#refer
 * @param {Object} req - drachtio request object
 * @param {Object} res - drachtio response object
 */
/**
 * a <code>message</code> event is triggered when the far end sends a MESSAGE message.
 * When an application adds a handler for this event it must generate
 * the SIP response by calling <code>res.send</code> on the provided drachtio response object.
 * When no handler is found for this event a 200 OK will be automatically generated.
 * @event Dialog#message
 * @param {Object} req - drachtio request object
 * @param {Object} res - drachtio response object
 */