fix: Dockerfile to reduce vulnerabilities

The following vulnerabilities are fixed with an upgrade:
- https://snyk.io/vuln/SNYK-UPSTREAM-NODE-2946423
- https://snyk.io/vuln/SNYK-UPSTREAM-NODE-2946427
- https://snyk.io/vuln/SNYK-UPSTREAM-NODE-2946428
- https://snyk.io/vuln/SNYK-UPSTREAM-NODE-2946723
- https://snyk.io/vuln/SNYK-UPSTREAM-NODE-2946727

Dockerfile

@@ -1,10 +1,23 @@
-FROM node:lts-slim
+FROM --platform=linux/amd64 node:18.6.0-alpine as base
+
+RUN apk --update --no-cache add --virtual .builds-deps build-base python3
+
 WORKDIR /opt/app/
+
+FROM base as build
+
 COPY package.json package-lock.json ./
+
 RUN npm ci
-RUN npm prune
-COPY . /opt/app
+
+COPY . .
+
+FROM base
+
+COPY --from=build /opt/app /opt/app/
+
 ARG NODE_ENV
+
 ENV NODE_ENV $NODE_ENV
 
-CMD [ "npm", "start" ]
+CMD [ "node", "app.js" ]

lib/http-routes/api/create-call.js

@@ -197,9 +197,10 @@ router.post('/', async(req, res) => {
           });
           cs.exec(req);
 
-          res.status(201).json({sid: cs.callSid});
+          res.status(201).json({sid: cs.callSid, callId: inviteReq.get('Call-ID')});
 
-          sipLogger.info(`outbound REST call attempt to ${JSON.stringify(target)} has been sent`);
+          sipLogger.info({sid: cs.callSid, callId: inviteReq.get('Call-ID')},
+            `outbound REST call attempt to ${JSON.stringify(target)} has been sent`);
         },
         cbProvisional: (prov) => {
           const callStatus = prov.body ? CallStatus.EarlyMedia : CallStatus.Ringing;

lib/session/call-session.js

@@ -1,6 +1,13 @@
 const Emitter = require('events');
 const fs = require('fs');
-const {CallDirection, TaskPreconditions, CallStatus, TaskName, KillReason} = require('../utils/constants');
+const {
+  CallDirection,
+  TaskPreconditions,
+  CallStatus,
+  TaskName,
+  KillReason,
+  RecordState
+} = require('../utils/constants');
 const moment = require('moment');
 const assert = require('assert');
 const sessionTracker = require('./session-tracker');
@@ -54,6 +61,8 @@ class CallSession extends Emitter {
 
     assert(rootSpan);
 
+    this._recordState = RecordState.RecordingOff;
+
     this.tmpFiles = new Set();
 
     if (!this.isSmsCallSession) {
@@ -85,6 +94,10 @@ class CallSession extends Emitter {
     return this.callInfo.direction;
   }
 
+  get applicationSid() {
+    return this.callInfo.applicationSid;
+  }
+
   /**
    * SIP call-id for the call
    */
@@ -234,6 +247,153 @@ class CallSession extends Emitter {
     return this.rootSpan?.getTracingPropagation();
   }
 
+  get recordState() { return this._recordState; }
+
+  async notifyRecordOptions(opts) {
+    const {action} = opts;
+    this.logger.debug({opts}, 'CallSession:notifyRecordOptions');
+
+    /* if we have not answered yet, just save the details for later */
+    if (!this.dlg) {
+      if (action === 'startCallRecording') {
+        this.recordOptions = opts;
+        return true;
+      }
+      return false;
+    }
+
+    /* check validity of request */
+    if (action == 'startCallRecording' && this.recordState !== RecordState.RecordingOff) {
+      this.logger.info({recordState: this.recordState},
+        'CallSession:notifyRecordOptions: recording is already started, ignoring request');
+      return false;
+    }
+    if (action == 'stopCallRecording' && this.recordState === RecordState.RecordingOff) {
+      this.logger.info({recordState: this.recordState},
+        'CallSession:notifyRecordOptions: recording is already stopped, ignoring request');
+      return false;
+    }
+    if (action == 'pauseCallRecording' && this.recordState !== RecordState.RecordingOn) {
+      this.logger.info({recordState: this.recordState},
+        'CallSession:notifyRecordOptions: cannot pause recording, ignoring request ');
+      return false;
+    }
+    if (action == 'resumeCallRecording' && this.recordState !== RecordState.RecordingPaused) {
+      this.logger.info({recordState: this.recordState},
+        'CallSession:notifyRecordOptions: cannot resume recording, ignoring request ');
+      return false;
+    }
+
+    this.recordOptions = opts;
+
+    switch (action) {
+      case 'startCallRecording':
+        return await this.startRecording();
+      case 'stopCallRecording':
+        return await this.stopRecording();
+      case 'pauseCallRecording':
+        return await this.pauseRecording();
+      case 'resumeCallRecording':
+        return await this.resumeRecording();
+      default:
+        throw new Error(`invalid record action ${action}`);
+    }
+  }
+
+  async startRecording() {
+    const {recordingID, siprecServerURL} = this.recordOptions;
+    assert(this.dlg);
+    this.logger.debug(`CallSession:startRecording - sending to ${siprecServerURL}`);
+    try {
+      const res = await this.dlg.request({
+        method: 'INFO',
+        headers: {
+          'X-Reason': 'startCallRecording',
+          'X-Srs-Url': siprecServerURL,
+          'X-Srs-Recording-ID': recordingID,
+          'X-Call-Sid': this.callSid,
+          'X-Account-Sid': this.accountSid,
+          'X-Application-Sid': this.applicationSid,
+        }
+      });
+      if (res.status === 200) {
+        this._recordState = RecordState.RecordingOn;
+        return true;
+      }
+      this.logger.info(`CallSession:startRecording - ${res.status} failure sending to ${siprecServerURL}`);
+      return false;
+    } catch (err) {
+      this.logger.info({err}, `CallSession:startRecording - failure sending to ${siprecServerURL}`);
+      return false;
+    }
+  }
+
+  async stopRecording() {
+    assert(this.dlg);
+    this.logger.debug('CallSession:stopRecording');
+    try {
+      const res = await this.dlg.request({
+        method: 'INFO',
+        headers: {
+          'X-Reason': 'stopCallRecording',
+        }
+      });
+      if (res.status === 200) {
+        this._recordState = RecordState.RecordingOff;
+        return true;
+      }
+      this.logger.info(`CallSession:stopRecording - ${res.status} failure`);
+      return false;
+    } catch (err) {
+      this.logger.info({err}, 'CallSession:startRecording - failure sending');
+      return false;
+    }
+  }
+
+  async pauseRecording() {
+    assert(this.dlg);
+    this.logger.debug('CallSession:pauseRecording');
+    try {
+      const res = await this.dlg.request({
+        method: 'INFO',
+        headers: {
+          'X-Reason': 'pauseCallRecording',
+        }
+      });
+      if (res.status === 200) {
+        this._recordState = RecordState.RecordingPaused;
+        return true;
+      }
+      this.logger.info(`CallSession:pauseRecording - ${res.status} failure`);
+      return false;
+    } catch (err) {
+      this.logger.info({err}, 'CallSession:pauseRecording - failure sending');
+      return false;
+    }
+  }
+
+  async resumeRecording() {
+    assert(this.dlg);
+    this.logger.debug('CallSession:resumeRecording');
+    try {
+      const res = await this.dlg.request({
+        method: 'INFO',
+        headers: {
+          'X-Reason': 'resumeCallRecording',
+        }
+      });
+      if (res.status === 200) {
+        this._recordState = RecordState.RecordingOn;
+        return true;
+      }
+      this.logger.info(`CallSession:resumeRecording - ${res.status} failure`);
+      return false;
+    } catch (err) {
+      this.logger.info({err}, 'CallSession:resumeRecording - failure sending');
+      return false;
+    }
+  }
+
   async enableBotMode(gather, autoEnable) {
     try {
       const t = normalizeJambones(this.logger, [gather]);
@@ -724,6 +884,9 @@ class CallSession extends Emitter {
       const res = await this._lccSipRequest(opts, callSid);
       return {status: res.status, reason: res.reason};
     }
+    else if (opts.record) {
+      await this.notifyRecordOptions(opts.record);
+    }
 
     // whisper may be the only thing we are asked to do, or it may that
     // we are doing a whisper after having muted, paused reccording etc..
@@ -1076,6 +1239,9 @@ class CallSession extends Emitter {
       this.dlg.callSid = this.callSid;
       this.emit('callStatusChange', {sipStatus: 200, sipReason: 'OK', callStatus: CallStatus.InProgress});
 
+      if (this.recordOptions && this.recordState === RecordState.RecordingOff) {
+        this.startRecording();
+      }
       this.dlg.on('modify', this._onReinvite.bind(this));
       this.dlg.on('refer', this._onRefer.bind(this));
 
@@ -1258,7 +1424,7 @@ class CallSession extends Emitter {
         this.emit('callStatusChange', {callStatus: CallStatus.Completed, duration});
         this.logger.debug('CallSession: call terminated by jambones');
         this.rootSpan.setAttributes({'call.termination': 'hangup by jambonz'});
-        origDestroy();
+        origDestroy().catch((err) => this.logger.info({err}, 'CallSession - error destroying dialog'));
         if (this.wakeupResolver) {
           this.wakeupResolver({reason: 'session ended'});
           this.wakeupResolver = null;

lib/tasks/conference.js

@@ -453,7 +453,7 @@ class Conference extends Task {
         this._playSession = null;
         break;
       }
-    } while (!this.killed && this.conf_hold_status !== 'hold');
+    } while (!this.killed && this.conf_hold_status === 'hold');
   }
 
   /**

lib/tasks/config.js

@@ -9,7 +9,8 @@ class TaskConfig extends Task {
     [
       'synthesizer',
       'recognizer',
-      'bargeIn'
+      'bargeIn',
+      'record'
     ].forEach((k) => this[k] = this.data[k] || {});
 
     if (this.bargeIn.enable) {
@@ -74,7 +75,15 @@ class TaskConfig extends Task {
       cs.speechRecognizerLanguage = this.recognizer.language !== 'default'
         ? this.recognizer.language
         : cs.speechRecognizerLanguage;
-      this.logger.info({recognizer: this.recognizer}, 'Config: updated recognizer');
+      cs.isContinuousAsr = typeof this.recognizer.asrTimeout === 'number' ? true : false;
+      if (cs.isContinuousAsr) {
+        cs.asrTimeout = this.recognizer.asrTimeout;
+        cs.asrDtmfTerminationDigit = this.recognizer.asrDtmfTerminationDigit;
+      }
+      this.logger.info({
+        recognizer: this.recognizer,
+        isContinuousAsr: cs.isContinuousAsr
+      }, 'Config: updated recognizer');
     }
     if ('enable' in this.bargeIn) {
       if (this.gatherOpts) {
@@ -92,6 +101,7 @@ class TaskConfig extends Task {
         cs.disableBotMode();
       }
     }
+    if (this.record.action) cs.notifyRecordOptions(this.record);
   }
 
   async kill(cs) {

lib/tasks/dial.js

@@ -404,6 +404,11 @@ class TaskDial extends Task {
       this.logger.info(`Dial:_attemptCall: ring no answer timer ${this.timeout}s exceeded`);
       this.timerRing = null;
       this._killOutdials();
+      this.result = {
+        dialCallStatus: CallStatus.NoAnswer,
+        dialSipStatus: 487
+      };
+      this.kill(cs);
     }, this.timeout * 1000);
 
     this.span.setAttributes({'dial.target': JSON.stringify(this.target)});

lib/tasks/gather.js

@@ -23,7 +23,7 @@ class TaskGather extends Task {
     ].forEach((k) => this[k] = this.data[k]);
 
     /* when collecting dtmf, bargein on dtmf is true unless explicitly set to false */
-    if (this.dtmfBargein !== false && this.input.includes('digits')) this.dtmfBargein = true;
+    if (this.dtmfBargein !== false  && this.input.includes('digits')) this.dtmfBargein = true;
 
     /* timeout of zero means no timeout */
     this.timeout = this.timeout === 0 ? 0 : (this.timeout || 15) * 1000;
@@ -49,6 +49,11 @@ class TaskGather extends Task {
       this.naicsCode = recognizer.naicsCode || 0;
       this.altLanguages = recognizer.altLanguages || [];
 
+      /* continuous ASR (i.e. compile transcripts until a special timeout or dtmf key) */
+      this.asrTimeout = typeof recognizer.asrTimeout === 'number' ? recognizer.asrTimeout * 1000 : 0;
+      if (this.asrTimeout > 0) this.asrDtmfTerminationDigit = recognizer.asrDtmfTerminationDigit;
+      this.isContinuousAsr = this.asrTimeout > 0;
+
       /* vad: if provided, we dont connect to recognizer until voice activity is detected */
       const {enable, voiceMs = 0, mode = -1} = recognizer.vad || {};
       this.vad = {enable, voiceMs, mode};
@@ -65,6 +70,10 @@ class TaskGather extends Task {
       this.initialSpeechTimeoutMs = recognizer.initialSpeechTimeoutMs || 0;
       this.azureServiceEndpoint = recognizer.azureServiceEndpoint;
     }
+    else {
+      this.hints = [];
+      this.altLanguages = [];
+    }
 
     this.digitBuffer = '';
     this._earlyMedia = this.data.earlyMedia === true;
@@ -77,6 +86,9 @@ class TaskGather extends Task {
     }
     if (!this.sayTask && !this.playTask) this.listenDuringPrompt = false;
 
+    /* buffer speech for continueous asr */
+    this._bufferedTranscripts = [];
+
     this.parentTask = parentTask;
   }
 
@@ -109,6 +121,15 @@ class TaskGather extends Task {
     await super.exec(cs);
     const {updateSpeechCredentialLastUsed} = require('../utils/db-utils')(this.logger, cs.srf);
 
+    if (!this.isContinuousAsr && cs.isContinuousAsr) {
+      this.isContinuousAsr = true;
+      this.asrTimeout = cs.asrTimeout * 1000;
+      this.asrDtmfTerminationDigit = cs.asrDtmfTerminationDigit;
+      this.logger.debug({
+        asrTimeout: this.asrTimeout,
+        asrDtmfTerminationDigit: this.asrDtmfTerminationDigit
+      }, 'Gather:exec - enabling continuous ASR since it is turned on for the session');
+    }
     this.ep = ep;
     if ('default' === this.vendor || !this.vendor) this.vendor = cs.speechRecognizerVendor;
     if ('default' === this.language || !this.language) this.language = cs.speechRecognizerLanguage;
@@ -127,6 +148,7 @@ class TaskGather extends Task {
 
     const startListening = (cs, ep) => {
       this._startTimer();
+      if (this.isContinuousAsr && 0 === this.timeout) this._startAsrTimer();
       if (this.input.includes('speech') && !this.listenDuringPrompt) {
         this._initSpeech(cs, ep)
           .then(() => {
@@ -171,7 +193,7 @@ class TaskGather extends Task {
           .catch(() => {/*already logged error */});
       }
 
-      if (this.input.includes('digits') || this.dtmfBargein) {
+      if (this.input.includes('digits') || this.dtmfBargein || this.asrDtmfTerminationDigit) {
         ep.on('dtmf', this._onDtmf.bind(this, cs, ep));
       }
 
@@ -209,12 +231,15 @@ class TaskGather extends Task {
     this.logger.debug(evt, 'TaskGather:_onDtmf');
     clearTimeout(this.interDigitTimer);
     let resolved = false;
-    if (this.dtmfBargein) this._killAudio(cs);
+    if (this.dtmfBargein) {
+      this._killAudio(cs);
+      this.emit('dtmf', evt);
+    }
     if (evt.dtmf === this.finishOnKey && this.input.includes('digits')) {
       resolved = true;
       this._resolve('dtmf-terminator-key');
     }
-    else {
+    else if (this.input.includes('digits')) {
       this.digitBuffer += evt.dtmf;
       const len = this.digitBuffer.length;
       if (len === this.numDigits || len === this.maxDigits) {
@@ -222,6 +247,13 @@ class TaskGather extends Task {
         this._resolve('dtmf-num-digits');
       }
     }
+    else if (this.isContinuousAsr && evt.dtmf === this.asrDtmfTerminationDigit) {
+      this.logger.info(`continuousAsr triggered with dtmf ${this.asrDtmfTerminationDigit}`);
+      this._clearAsrTimer();
+      this._clearTimer();
+      this._startFinalAsrTimer();
+      return;
+    }
     if (!resolved && this.interDigitTimeout > 0 && this.digitBuffer.length >= this.minDigits) {
       /* start interDigitTimer */
       const ms = this.interDigitTimeout * 1000;
@@ -343,14 +375,10 @@ class TaskGather extends Task {
 
   _startTimer() {
     if (0 === this.timeout) return;
-    if (this._timeoutTimer) {
-      clearTimeout(this._timeoutTimer);
-      this._timeoutTimer = null;
-    }
-    assert(!this._timeoutTimer);
-    this.logger.debug(`Gather:_startTimer: timeout ${this.timeout}`);
+    this._clearTimer();
     this._timeoutTimer = setTimeout(() => {
-      this._resolve(this.digitBuffer.length >= this.minDigits ? 'dtmf-num-digits' : 'timeout');
+      if (this.isContinuousAsr) this._startAsrTimer();
+      else this._resolve(this.digitBuffer.length >= this.minDigits ? 'dtmf-num-digits' : 'timeout');
     }, this.timeout);
   }
 
@@ -361,6 +389,35 @@ class TaskGather extends Task {
     }
   }
 
+  _startAsrTimer() {
+    assert(this.isContinuousAsr);
+    this._clearAsrTimer();
+    this._asrTimer = setTimeout(() => {
+      this.logger.debug('_startAsrTimer - asr timer went off');
+      this._resolve(this._bufferedTranscripts.length > 0 ? 'speech' : 'timeout');
+    }, this.asrTimeout);
+    this.logger.debug(`_startAsrTimer: set for ${this.asrTimeout}ms`);
+  }
+
+  _clearAsrTimer() {
+    if (this._asrTimer) clearTimeout(this._asrTimer);
+    this._asrTimer = null;
+  }
+
+  _startFinalAsrTimer() {
+    this._clearFinalAsrTimer();
+    this._finalAsrTimer = setTimeout(() => {
+      this.logger.debug('_startFinalAsrTimer - final asr timer went off');
+      this._resolve(this._bufferedTranscripts.length > 0 ? 'speech' : 'timeout');
+    }, 1000);
+    this.logger.debug('_startFinalAsrTimer: set for 1 second');
+  }
+
+  _clearFinalAsrTimer() {
+    if (this._finalAsrTimer) clearTimeout(this._finalAsrTimer);
+    this._finalAsrTimer = null;
+  }
+
   _killAudio(cs) {
     if (!this.sayTask && !this.playTask && this.bargein) {
       if (this.ep?.connected && !this.playComplete) {
@@ -419,7 +476,19 @@ class TaskGather extends Task {
         this.logger.info({evt}, 'TaskGather:_onTranscription - got empty transcript, listen again');
         return this._startTranscribing(ep);
       }
-      this._resolve('speech', evt);
+      if (this.isContinuousAsr) {
+        /* append the transcript and start listening again for asrTimeout */
+        this.logger.info({evt}, 'TaskGather:_onTranscription - got transcript during continous asr');
+        this._bufferedTranscripts.push(evt);
+        this._clearTimer();
+        if (this._finalAsrTimer) {
+          this._clearFinalAsrTimer();
+          return this._resolve(this._bufferedTranscripts.length > 0 ? 'speech' : 'timeout');
+        }
+        this._startAsrTimer();
+        return this._startTranscribing(ep);
+      }
+      else this._resolve('speech', evt);
     }
     else {
       /* google has a measure of stability:
@@ -476,6 +545,15 @@ class TaskGather extends Task {
 
     this.resolved = true;
     clearTimeout(this.interDigitTimer);
+    this._clearTimer();
+
+    if (this.isContinuousAsr && reason.startsWith('speech')) {
+      evt = {
+        is_final: true,
+        transcripts: this._bufferedTranscripts
+      };
+      this.logger.debug({evt}, 'TaskGather:resolve continuous asr');
+    }
 
     this.span.setAttributes({'stt.resolve': reason, 'stt.result': JSON.stringify(evt)});
     if (this.ep && this.ep.connected) {
@@ -483,8 +561,6 @@ class TaskGather extends Task {
         .catch((err) => this.logger.error({err}, 'Error stopping transcription'));
     }
 
-    this._clearTimer();
-
     if (this.callSession && this.callSession.callGone) {
       this.logger.debug('TaskGather:_resolve - call is gone, not invoking web callback');
       this.notifyTaskDone();

lib/tasks/listen.js

@@ -22,8 +22,6 @@ class TaskListen extends Task {
     this.results = {};
 
     if (this.transcribe) this.transcribeTask = makeTask(logger, {'transcribe': opts.transcribe}, this);
-
-    this._dtmfHandler = this._onDtmf.bind(this);
   }
 
   get name() { return TaskName.Listen; }
@@ -31,6 +29,7 @@ class TaskListen extends Task {
   async exec(cs, ep) {
     await super.exec(cs);
     this.ep = ep;
+    this._dtmfHandler = this._onDtmf.bind(this, ep);
 
     try {
       this.hook = this.normalizeUrl(this.url, 'GET', this.wsAuth);
@@ -148,7 +147,13 @@ class TaskListen extends Task {
 
   }
 
-  _onDtmf(evt) {
+  _onDtmf(ep, evt) {
+    this.logger.debug({evt}, `TaskListen:_onDtmf received dtmf ${evt.dtmf}`);
+    if (this.passDtmf && this.ep?.connected) {
+      const obj = {event: 'dtmf', dtmf: evt.dtmf, duration: evt.duration};
+      this.ep.forkAudioSendText(obj)
+        .catch((err) => this.logger.info({err}, 'TaskListen:_onDtmf error sending dtmf'));
+    }
     if (evt.dtmf === this.finishOnKey) {
       this.logger.info(`TaskListen:_onDtmf terminating task due to dtmf ${evt.dtmf}`);
       this.results.digits = evt.dtmf;

lib/tasks/sip_request.js

@@ -12,6 +12,7 @@ class TaskSipRequest extends Task {
     this.method = this.data.method.toUpperCase();
     this.headers = this.data.headers || {};
     this.body = this.data.body;
+    if (this.body) this.body = `${this.body}\n`;
   }
 
   get name() { return TaskName.SipRequest; }

lib/tasks/specs.json

@@ -36,7 +36,8 @@
     "properties": {
       "synthesizer": "#synthesizer",
       "recognizer": "#recognizer",
-      "bargeIn": "#bargeIn"
+      "bargeIn": "#bargeIn",
+      "record": "#recordOptions"
     },
     "required": []
   },
@@ -307,6 +308,19 @@
       "path"
     ]
   },
+  "recordOptions": {
+    "properties": {
+      "action": {
+        "type": "string",
+        "enum": ["startCallRecording", "stopCallRecording", "pauseCallRecording", "resumeCallRecording"]
+      },
+      "recordingID": "string",
+      "siprecServerURL": "string"
+    },
+    "required": [
+      "action"
+    ]
+  },
   "redirect": {
     "properties": {
       "actionHook": "object|string"
@@ -477,7 +491,9 @@
       },
       "requestSnr": "boolean",
       "initialSpeechTimeoutMs": "number",
-      "azureServiceEndpoint": "string"
+      "azureServiceEndpoint": "string",
+      "asrDtmfTerminationDigit": "string",
+      "asrTimeout": "number"
     },
     "required": [
       "vendor"

lib/tasks/task.js

@@ -145,7 +145,7 @@ class Task extends Emitter {
 
   async performAction(results, expectResponse = true) {
     if (this.actionHook) {
-      const params = results ? Object.assign(results, this.cs.callInfo.toJSON()) : this.cs.callInfo.toJSON();
+      const params = results ? Object.assign(this.cs.callInfo.toJSON(), results) : this.cs.callInfo.toJSON();
       const span = this.startSpan('verb:hook', {'hook.url': this.actionHook});
       const b3 = this.getTracingPropagation('b3', span);
       const httpHeaders = b3 && {b3};

lib/utils/constants.json

@@ -120,6 +120,11 @@
     "verb:hook",
     "jambonz:error"
   ],
+  "RecordState": {
+    "RecordingOn": "recording_on",
+    "RecordingOff": "recording_off",
+    "RecordingPaused": "recording_paused"
+  },
   "MAX_SIMRINGS": 10,
   "BONG_TONE": "tone_stream://v=-7;%(100,0,941.0,1477.0);v=-7;>=2;+=.1;%(1400,0,350,440)",
   "FS_UUID_SET_NAME": "fsUUIDs"

lib/utils/http-requestor.js

@@ -1,9 +1,11 @@
-const bent = require('bent');
+const {Client, Pool} = require('undici');
 const parseUrl = require('parse-url');
 const assert = require('assert');
 const BaseRequestor = require('./base-requestor');
 const {HookMsgTypes} = require('./constants.json');
 const snakeCaseKeys = require('./snakecase-keys');
+const pools = new Map();
+const HTTP_TIMEOUT = 10000;
 
 const toBase64 = (str) => Buffer.from(str || '', 'utf8').toString('base64');
 
@@ -22,22 +24,41 @@ class HttpRequestor extends BaseRequestor {
     this.method = hook.method || 'POST';
     this.authHeader = basicAuth(hook.username, hook.password);
 
-    const u = parseUrl(this.url);
-    const myPort = u.port ? `:${u.port}` : '';
-    const baseUrl = this._baseUrl = `${u.protocol}://${u.resource}${myPort}`;
-
-    this.get = bent(baseUrl, 'GET', 'buffer', 200, 201);
-    this.post = bent(baseUrl, 'POST', 'buffer', 200, 201);
-
-
     assert(this._isAbsoluteUrl(this.url));
     assert(['GET', 'POST'].includes(this.method));
+
+    const u = this._parsedUrl = parseUrl(this.url);
+    this._baseUrl = `${u.protocol}://${u.resource}`;
+    this._resource = u.resource;
+    this._protocol = u.protocol;
+    this._usePools = process.env.HTTP_POOL && parseInt(process.env.HTTP_POOL);
+
+    if (this._usePools) {
+      if (pools.has(this._baseUrl)) {
+        this.client = pools.get(this._baseUrl);
+      }
+      else {
+        const connections = process.env.HTTP_POOLSIZE ? parseInt(process.env.HTTP_POOLSIZE) : 10;
+        const pipelining = process.env.HTTP_PIPELINING ? parseInt(process.env.HTTP_PIPELINING) : 1;
+        const pool = this.client = new Pool(this._baseUrl, {
+          connections,
+          pipelining
+        });
+        pools.set(this._baseUrl, pool);
+        this.logger.debug(`HttpRequestor:created pool for ${this._baseUrl}`);
+      }
+    }
+    else this.client = new Client(`${u.protocol}://${u.resource}`);
   }
 
   get baseUrl() {
     return this._baseUrl;
   }
 
+  close() {
+    if (!this._usePools && !this.client?.closed) this.client.close();
+  }
+
   /**
    * Make an HTTP request.
    * All requests use json bodies.
@@ -58,6 +79,7 @@ class HttpRequestor extends BaseRequestor {
     const payload = params ? snakeCaseKeys(params, ['customerData', 'sip']) : null;
     const url = hook.url || hook;
     const method = hook.method || 'POST';
+    let buf = '';
 
     assert.ok(url, 'HttpRequestor:request url was not provided');
     assert.ok, (['GET', 'POST'].includes(method), `HttpRequestor:request method must be 'GET' or 'POST' not ${method}`);
@@ -65,14 +87,46 @@ class HttpRequestor extends BaseRequestor {
     this.logger.debug({url: urlInfo, method: methodInfo, payload}, `HttpRequestor:request ${method} ${url}`);
     const startAt = process.hrtime();
 
-    let buf;
+    let newClient;
     try {
+      let client, path;
+      if (this._isRelativeUrl(url)) {
+        client = this.client;
+        path = url;
+      }
+      else {
+        const u = parseUrl(url);
+        if (u.resource === this._resource && u.protocol === this._protocol) {
+          client = this.client;
+          path = u.pathname;
+        }
+        else {
+          client = newClient = new Client(`${u.protocol}://${u.resource}`);
+          path = u.pathname;
+        }
+      }
       const sigHeader = this._generateSigHeader(payload, this.secret);
-      const headers = {...sigHeader, ...this.authHeader, ...httpHeaders};
-      this.logger.debug({url, headers}, 'send webhook');
-      buf = this._isRelativeUrl(url) ?
-        await this.post(url, payload, headers) :
-        await bent(method, 'buffer', 200, 201, 202)(url, payload, headers);
+      const hdrs = {
+        ...sigHeader,
+        ...this.authHeader,
+        ...httpHeaders,
+        ...('POST' === method && {'Content-Type': 'application/json'})
+      };
+      const absUrl = this._isRelativeUrl(url) ? `${this.baseUrl}${url}` : url;
+      this.logger.debug({url, absUrl, hdrs}, 'send webhook');
+      const {statusCode, headers, body} =  await client.request({
+        path,
+        method,
+        headers: hdrs,
+        ...('POST' === method && {body: JSON.stringify(payload)}),
+        timeout: HTTP_TIMEOUT,
+        followRedirects: false
+      });
+      if (![200, 202, 204].includes(statusCode)) throw new Error({statusCode});
+      if (headers['content-type'].includes('application/json')) {
+        buf = await body.json();
+      }
+      if (newClient) newClient.close();
     } catch (err) {
       if (err.statusCode) {
         this.logger.info({baseUrl: this.baseUrl, url},
@@ -94,20 +148,15 @@ class HttpRequestor extends BaseRequestor {
       }
       this.Alerter.writeAlerts(opts).catch((err) => this.logger.info({err, opts}, 'Error writing alert'));
 
+      if (newClient) newClient.close();
       throw err;
     }
     const rtt = this._roundTrip(startAt);
     if (buf) this.stats.histogram('app.hook.response_time', rtt, ['hook_type:app']);
 
-    if (buf && buf.toString().length > 0) {
-      try {
-        const json = JSON.parse(buf.toString());
-        this.logger.info({response: json}, `HttpRequestor:request ${method} ${url} succeeded in ${rtt}ms`);
-        return json;
-      }
-      catch (err) {
-        //this.logger.debug({err, url, method}, `HttpRequestor:request returned non-JSON content: '${buf.toString()}'`);
-      }
+    if (buf && Array.isArray(buf)) {
+      this.logger.info({response: buf}, `HttpRequestor:request ${method} ${url} succeeded in ${rtt}ms`);
+      return buf;
     }
   }
 }

lib/utils/ws-requestor.js

@@ -158,7 +158,7 @@ class WsRequestor extends BaseRequestor {
 
   close() {
     this.closedGracefully = true;
-    this.logger.info('WsRequestor:close closing socket');
+    this.logger.debug('WsRequestor:close closing socket');
     try {
       if (this.ws) {
         this.ws.close();

package.json

@@ -21,7 +21,7 @@
   },
   "scripts": {
     "start": "node app",
-    "test": "NODE_ENV=test JAMBONES_HOSTING=1 DRACHTIO_HOST=127.0.0.1 DRACHTIO_PORT=9060 DRACHTIO_SECRET=cymru JAMBONES_MYSQL_HOST=127.0.0.1 JAMBONES_MYSQL_PORT=3360 JAMBONES_MYSQL_USER=jambones_test JAMBONES_MYSQL_PASSWORD=jambones_test JAMBONES_MYSQL_DATABASE=jambones_test JAMBONES_REDIS_HOST=127.0.0.1 JAMBONES_REDIS_PORT=16379 JAMBONES_LOGLEVEL=info ENABLE_METRICS=0 HTTP_PORT=3000 JAMBONES_SBCS=172.38.0.10 JAMBONES_FREESWITCH=127.0.0.1:8022:ClueCon:docker-host JAMBONES_TIME_SERIES_HOST=127.0.0.1 JAMBONES_NETWORK_CIDR=172.38.0.0/16 node test/ ",
+    "test": "NODE_ENV=test JAMBONES_HOSTING=1 HTTP_POOL=1 DRACHTIO_HOST=127.0.0.1 DRACHTIO_PORT=9060 DRACHTIO_SECRET=cymru JAMBONES_MYSQL_HOST=127.0.0.1 JAMBONES_MYSQL_PORT=3360 JAMBONES_MYSQL_USER=jambones_test JAMBONES_MYSQL_PASSWORD=jambones_test JAMBONES_MYSQL_DATABASE=jambones_test JAMBONES_REDIS_HOST=127.0.0.1 JAMBONES_REDIS_PORT=16379 JAMBONES_LOGLEVEL=info ENABLE_METRICS=0 HTTP_PORT=3000 JAMBONES_SBCS=172.38.0.10 JAMBONES_FREESWITCH=127.0.0.1:8022:ClueCon:docker-host JAMBONES_TIME_SERIES_HOST=127.0.0.1 JAMBONES_NETWORK_CIDR=172.38.0.0/16 node test/ ",
     "coverage": "./node_modules/.bin/nyc --reporter html --report-dir ./coverage npm run test",
     "jslint": "eslint app.js lib"
   },
@@ -33,7 +33,7 @@
     "@jambonz/time-series": "^0.1.9",
     "@opentelemetry/api": "^1.1.0",
     "@opentelemetry/exporter-jaeger": "^1.3.1",
-    "@opentelemetry/exporter-trace-otlp-http": "^0.29.0",
+    "@opentelemetry/exporter-trace-otlp-http": "^0.27.0",
     "@opentelemetry/exporter-zipkin": "^1.3.1",
     "@opentelemetry/instrumentation": "^0.27.0",
     "@opentelemetry/resources": "^1.3.1",
@@ -44,19 +44,20 @@
     "bent": "^7.3.12",
     "debug": "^4.3.4",
     "deepcopy": "^2.1.0",
-    "drachtio-fsmrf": "^3.0.0",
-    "drachtio-srf": "^4.5.0",
+    "drachtio-fsmrf": "^3.0.1",
+    "drachtio-srf": "^4.5.1",
     "express": "^4.18.1",
     "helmet": "^5.1.0",
     "ip": "^1.1.8",
     "moment": "^2.29.3",
-    "parse-url": "^5.0.8",
+    "parse-url": "^7.0.2",
     "pino": "^6.14.0",
     "sdp-transform": "^2.14.1",
     "short-uuid": "^4.2.0",
     "to-snake-case": "^1.0.0",
+    "undici": "^5.7.0",
     "uuid": "^8.3.2",
-    "verify-aws-sns-signature": "^0.0.6",
+    "verify-aws-sns-signature": "^0.0.7",
     "ws": "^8.8.0",
     "xml2js": "^0.4.23"
   },