lib/utils/ws-requestor.js
/**
   * Send a JSON payload over the websocket.  If this is the first request,
   * open the websocket.
   * All requests expect an ack message in response
   * @param {object|string} hook - may be a absolute or relative url, or an object
   * @param {string} [hook.url] - an absolute or relative url
   * @param {string} [hook.method] - 'GET' or 'POST'
   * @param {string} [hook.username] - if basic auth is protecting the endpoint
   * @param {string} [hook.password] - if basic auth is protecting the endpoint
   * @param {object} [params] - request parameters
   */
Send a JSON payload over the websocket.  If this is the first request,
open the websocket.
Send a JSON payload over the websocket . If this is the first request open the websocket .
async request(type, hook, params) {
    assert(HookMsgTypes.includes(type));
    const url = hook.url || hook;
    if (this.maliciousClient) {
      this.logger.info({url: this.url}, 'WsRequestor:request - discarding msg to malicious client');
      return;
    }
    if (type === 'session:new') this.call_sid = params.callSid;
    if (this._isAbsoluteUrl(url) && url.startsWith('http')) {
      this.logger.debug({hook}, 'WsRequestor: sending a webhook (HTTP)');
      const requestor = new HttpRequestor(this.logger, this.account_sid, hook, this.secret);
      return requestor.request(type, hook, params);
    }
    if (!this.ws) {
      if (this.connections >= MAX_RECONNECTS) {
        throw new Error(`max attempts connecting to ${this.url}`);
      }
      try {
        const startAt = process.hrtime();
        await this._connect();
        const rtt = this._roundTrip(startAt);
        this.stats.histogram('app.hook.connect_time', rtt, ['hook_type:app']);
      } catch (err) {
        this.logger.info({url, err}, 'WsRequestor:request - failed connecting');
        throw err;
      }
    }
    assert(this.ws);
    const payload = params ? snakeCaseKeys(params, ['customerData', 'sip']) : null;
    assert.ok(url, 'WsRequestor:request url was not provided');
    const msgid = short.generate();
    const obj = {
      type,
      msgid,
      call_sid: this.call_sid,
      hook: type === 'verb:hook' ? url : undefined,
      data: {...payload}
    };
    if (['call:status', 'jambonz:error'].includes(type)) {
      this.ws.send(JSON.stringify(obj), () => {
        this.logger.debug({obj}, `WsRequestor:request websocket: sent (${url})`);
      });
      return;
    }
    return new Promise((resolve, reject) => {
      const timer = setTimeout(() => {
        const {failure} = this.messagesInFlight.get(msgid);
        failure && failure(`timeout from far end for msgid ${msgid}`);
        this.messagesInFlight.delete(msgid);
      }, RESPONSE_TIMEOUT_MS);
      const startAt = process.hrtime();
      this.messagesInFlight.set(msgid, {
        success: (response) => {
          clearTimeout(timer);
          const rtt = this._roundTrip(startAt);
          this.logger.info({response}, `WsRequestor:request ${url} succeeded in ${rtt}ms`);
          this.stats.histogram('app.hook.ws_response_time', rtt, ['hook_type:app']);
          resolve(response);
        },
        failure: (err) => {
          clearTimeout(timer);
          reject(err);
        }
      });
      this.ws.send(JSON.stringify(obj), () => {
        this.logger.debug({obj}, `WsRequestor:request websocket: sent (${url})`);
      });
    });
  }
async request ( type , hook , params ) { assert ( HookMsgTypes . includes ( type ) ) ; const url = hook . url || hook ; if ( this . maliciousClient ) { this . logger . info ( { url : this . url } , 'WsRequestor:request - discarding msg to malicious client' ) ; return ; } if ( type === 'session:new' ) this . call_sid = params . callSid ; if ( this . _isAbsoluteUrl ( url ) && url . startsWith ( 'http' ) ) { this . logger . debug ( { hook } , 'WsRequestor: sending a webhook (HTTP)' ) ; const requestor = new HttpRequestor ( this . logger , this . account_sid , hook , this . secret ) ; return requestor . request ( type , hook , params ) ; } if ( ! this . ws ) { if ( this . connections >= MAX_RECONNECTS ) { throw new Error ( ` ${ this . url } ` ) ; } try { const startAt = process . hrtime ( ) ; await this . _connect ( ) ; const rtt = this . _roundTrip ( startAt ) ; this . stats . histogram ( 'app.hook.connect_time' , rtt , [ 'hook_type:app' ] ) ; } catch ( err ) { this . logger . info ( { url , err } , 'WsRequestor:request - failed connecting' ) ; throw err ; } } assert ( this . ws ) ; const payload = params ? snakeCaseKeys ( params , [ 'customerData' , 'sip' ] ) : null ; assert . ok ( url , 'WsRequestor:request url was not provided' ) ; const msgid = short . generate ( ) ; const obj = { type , msgid , call_sid : this . call_sid , hook : type === 'verb:hook' ? url : undefined , data : { ... payload } } ; if ( [ 'call:status' , 'jambonz:error' ] . includes ( type ) ) { this . ws . send ( JSON . stringify ( obj ) , ( ) => { this . logger . debug ( { obj } , ` ${ url } ` ) ; } ) ; return ; } return new Promise ( ( resolve , reject ) => { const timer = setTimeout ( ( ) => { const { failure } = this . messagesInFlight . get ( msgid ) ; failure && failure ( ` ${ msgid } ` ) ; this . messagesInFlight . delete ( msgid ) ; } , RESPONSE_TIMEOUT_MS ) ; const startAt = process . hrtime ( ) ; this . messagesInFlight . set ( msgid , { success : ( response ) => { clearTimeout ( timer ) ; const rtt = this . _roundTrip ( startAt ) ; this . logger . info ( { response } , ` ${ url } ${ rtt } ` ) ; this . stats . histogram ( 'app.hook.ws_response_time' , rtt , [ 'hook_type:app' ] ) ; resolve ( response ) ; } , failure : ( err ) => { clearTimeout ( timer ) ; reject ( err ) ; } } ) ; this . ws . send ( JSON . stringify ( obj ) , ( ) => { this . logger . debug ( { obj } , ` ${ url } ` ) ; } ) ; } ) ; }
Send a JSON payload over the websocket.
Send a JSON payload over the websocket .
/* if we have an absolute url, and it is http then do a standard webhook */ /* connect if necessary */ /* prepare and send message */ //this.logger.debug({obj}, `websocket: sending (${url})`); /* simple notifications */ /* messages that require an ack */ /* give the far end a reasonable amount of time to ack our message */ /* save the message info for reply */ /* send the message */