fix

* feat siprec custom header

* wip

* update verb specification

* add newline to info siprec body

* add newline to info siprec body

app.js

@@ -120,10 +120,15 @@ function handle(signal) {
   srf.locals.disabled = true;
   logger.info(`got signal ${signal}`);
   const setName = `${(JAMBONES_CLUSTER_ID || 'default')}:active-fs`;
+  const fsServiceUrlSetName = `${(JAMBONES_CLUSTER_ID || 'default')}:fs-service-url`;
   if (setName && srf.locals.localSipAddress) {
     logger.info(`got signal ${signal}, removing ${srf.locals.localSipAddress} from set ${setName}`);
     removeFromSet(setName, srf.locals.localSipAddress);
   }
+  if (fsServiceUrlSetName && srf.locals.serviceUrl) {
+    logger.info(`got signal ${signal}, removing ${srf.locals.serviceUrl} from set ${fsServiceUrlSetName}`);
+    removeFromSet(fsServiceUrlSetName, srf.locals.serviceUrl);
+  }
   removeFromSet(FS_UUID_SET_NAME, srf.locals.fsUUID);
   if (K8S) {
     srf.locals.lifecycleEmitter.operationalState = LifeCycleEvents.ScaleIn;

lib/session/call-info.js

@@ -145,6 +145,9 @@ class CallInfo {
 
     if (this._customerData) {
       Object.assign(obj, {customerData: this._customerData});
+      // Duplicate customer data here to respect API contract that was created by bug
+      //https://github.com/jambonz/jambonz-feature-server/issues/406
+      Object.assign(obj, {customerdata: this._customerData});
     }
 
     if (JAMBONES_API_BASE_URL) {

lib/session/call-session.js

@@ -423,7 +423,10 @@ class CallSession extends Emitter {
           'X-Call-Sid': this.callSid,
           'X-Account-Sid': this.accountSid,
           'X-Application-Sid': this.applicationSid,
-        }
+          ...(this.recordOptions.headers && {'Content-Type': 'application/json'})
+        },
+        // Siprect Client is initiated from startCallRecording, so just need to pass custom headers in startRecording
+        ...(this.recordOptions.headers && {body: JSON.stringify(this.recordOptions.headers) + '\n'})
       });
       if (res.status === 200) {
         this._recordState = RecordState.RecordingOn;
@@ -444,7 +447,7 @@ class CallSession extends Emitter {
       const res = await this.dlg.request({
         method: 'INFO',
         headers: {
-          'X-Reason': 'stopCallRecording',
+          'X-Reason': 'stopCallRecording'
         }
       });
       if (res.status === 200) {
@@ -466,7 +469,7 @@ class CallSession extends Emitter {
       const res = await this.dlg.request({
         method: 'INFO',
         headers: {
-          'X-Reason': 'pauseCallRecording',
+          'X-Reason': 'pauseCallRecording'
         }
       });
       if (res.status === 200) {
@@ -488,7 +491,7 @@ class CallSession extends Emitter {
       const res = await this.dlg.request({
         method: 'INFO',
         headers: {
-          'X-Reason': 'resumeCallRecording',
+          'X-Reason': 'resumeCallRecording'
         }
       });
       if (res.status === 200) {
@@ -513,6 +516,8 @@ class CallSession extends Emitter {
       const t = normalizeJambones(this.logger, [opts]);
       this.backgroundListenTask = makeTask(this.logger, t[0]);
       this.backgroundListenTask.bugname = bugname;
+      // Remove unneeded customer data to be sent to api server.
+      this.backgroundListenTask.ignoreCustomerData = true;
       const resources = await this._evaluatePreconditions(this.backgroundListenTask);
       const {span, ctx} = this.rootSpan.startChildSpan(`background-listen:${this.backgroundListenTask.summary}`);
       this.backgroundListenTask.span = span;
@@ -1779,6 +1784,15 @@ class CallSession extends Emitter {
 
     this.callInfo.updateCallStatus(callStatus, sipStatus, sipReason);
     if (typeof duration === 'number') this.callInfo.duration = duration;
+    this.executeStatusCallback(callStatus, sipStatus);
+
+    // update calls db
+    //this.logger.debug(`updating redis with ${JSON.stringify(this.callInfo)}`);
+    this.updateCallStatus(Object.assign({}, this.callInfo.toJSON()), this.serviceUrl)
+      .catch((err) => this.logger.error(err, 'redis error'));
+  }
+
+  async executeStatusCallback(callStatus, sipStatus) {
     const {span} = this.rootSpan.startChildSpan(`call-status:${this.callInfo.callStatus}`);
     span.setAttributes(this.callInfo.toJSON());
     try {
@@ -1790,11 +1804,6 @@ class CallSession extends Emitter {
       span.end();
       this.logger.info(err, `CallSession:_notifyCallStatusChange error sending ${callStatus} ${sipStatus}`);
     }
-
-    // update calls db
-    //this.logger.debug(`updating redis with ${JSON.stringify(this.callInfo)}`);
-    this.updateCallStatus(Object.assign({}, this.callInfo.toJSON()), this.serviceUrl)
-      .catch((err) => this.logger.error(err, 'redis error'));
   }
 
   async enableRecordAllCall() {

lib/tasks/conference.js

@@ -114,7 +114,12 @@ class Conference extends Task {
     }
     this.emitter.emit('kill');
     await this._doFinalMemberCheck(cs);
-    if (this.ep && this.ep.connected) this.ep.conn.removeAllListeners('esl::event::CUSTOM::*') ;
+    if (this.ep && this.ep.connected) {
+      this.ep.conn.removeAllListeners('esl::event::CUSTOM::*');
+      this.ep.api(`conference ${this.confName} kick ${this.memberId}`)
+        .catch((err) => this.logger.info({err}, 'Error kicking participant'));
+    }
+    cs.clearConferenceDetails();
     this.notifyTaskDone();
   }
 

lib/tasks/listen.js

@@ -32,6 +32,8 @@ class TaskListen extends Task {
 
   set bugname(name) { this._bugname = name; }
 
+  set ignoreCustomerData(val) { this._ignoreCustomerData = val; }
+
   async exec(cs, {ep}) {
     await super.exec(cs);
     this.ep = ep;
@@ -111,9 +113,13 @@ class TaskListen extends Task {
 
   async _startListening(cs, ep) {
     this._initListeners(ep);
+    const ci = this.nested ? this.parentTask.sd.callInfo : cs.callInfo.toJSON();
+    if (this._ignoreCustomerData) {
+      delete ci.customerData;
+    }
     const metadata = Object.assign(
       {sampleRate: this.sampleRate, mixType: this.mixType},
-      this.nested ? this.parentTask.sd.callInfo : cs.callInfo.toJSON(),
+      ci,
       this.metadata);
     if (this.hook.auth) {
       this.logger.debug({username: this.hook.auth.username, password: this.hook.auth.password},

lib/tasks/rest_dial.js

@@ -68,7 +68,7 @@ class TaskRestDial extends Task {
       const b3 = this.getTracingPropagation();
       const httpHeaders = b3 && {b3};
       const params = {
-        ...cs.callInfo,
+        ...cs.callInfo.toJSON(),
         defaults: {
           synthesizer: {
             vendor: cs.speechSynthesisVendor,

lib/utils/aws-sns-lifecycle.js

@@ -11,15 +11,20 @@ const {LifeCycleEvents} = require('./constants');
 const express = require('express');
 const app = express();
 const getString = bent('string');
-const AWS = require('aws-sdk');
-const sns = new AWS.SNS({apiVersion: '2010-03-31'});
-const autoscaling = new AWS.AutoScaling({apiVersion: '2011-01-01'});
+const {
+  SNSClient,
+  SubscribeCommand,
+  UnsubscribeCommand } = require('@aws-sdk/client-sns');
+const snsClient = new SNSClient({ region: AWS_REGION, apiVersion: '2010-03-31' });
+const {
+  AutoScalingClient,
+  DescribeAutoScalingGroupsCommand,
+  CompleteLifecycleActionCommand } = require('@aws-sdk/client-auto-scaling');
+const autoScalingClient = new AutoScalingClient({ region: AWS_REGION, apiVersion: '2011-01-01' });
 const {Parser} = require('xml2js');
 const parser = new Parser();
 const {validatePayload} = require('verify-aws-sns-signature');
 
-AWS.config.update({region: AWS_REGION});
-
 class SnsNotifier extends Emitter {
   constructor(logger) {
     super();
@@ -69,7 +74,7 @@ class SnsNotifier extends Emitter {
             subscriptionRequestId: this.subscriptionRequestId
           }, 'response from SNS SubscribeURL');
           const data = await this.describeInstance();
-          this.lifecycleState = data.AutoScalingInstances[0].LifecycleState;
+          this.lifecycleState = data.AutoScalingGroups[0].Instances[0].LifecycleState;
           this.emit('SubscriptionConfirmation', {publicIp: this.publicIp});
           break;
 
@@ -135,11 +140,12 @@ class SnsNotifier extends Emitter {
 
   async subscribe() {
     try {
-      const response = await sns.subscribe({
+      const params = {
         Protocol: 'http',
         TopicArn: AWS_SNS_TOPIC_ARM,
         Endpoint: this.snsEndpoint
-      }).promise();
+      };
+      const response = await snsClient.send(new SubscribeCommand(params));
       this.logger.info({response}, `response to SNS subscribe to ${AWS_SNS_TOPIC_ARM}`);
     } catch (err) {
       this.logger.error({err}, `Error subscribing to SNS topic arn ${AWS_SNS_TOPIC_ARM}`);
@@ -149,9 +155,10 @@ class SnsNotifier extends Emitter {
   async unsubscribe() {
     if (!this.subscriptionArn) throw new Error('SnsNotifier#unsubscribe called without an active subscription');
     try {
-      const response = await sns.unsubscribe({
+      const params = {
         SubscriptionArn: this.subscriptionArn
-      }).promise();
+      };
+      const response = await snsClient.send(new UnsubscribeCommand(params));
       this.logger.info({response}, `response to SNS unsubscribe to ${AWS_SNS_TOPIC_ARM}`);
     } catch (err) {
       this.logger.error({err}, `Error unsubscribing to SNS topic arn ${AWS_SNS_TOPIC_ARM}`);
@@ -160,26 +167,29 @@ class SnsNotifier extends Emitter {
 
   completeScaleIn() {
     assert(this.scaleInParams);
-    autoscaling.completeLifecycleAction(this.scaleInParams, (err, response) => {
-      if (err) return this.logger.error({err}, 'Error completing scale-in');
-      this.logger.info({response}, 'Successfully completed scale-in action');
-    });
+    autoScalingClient.send(new CompleteLifecycleActionCommand(this.scaleInParams))
+      .then((data) => {
+        return this.logger.info({data}, 'Successfully completed scale-in action');
+      })
+      .catch((err) => {
+        this.logger.error({err}, 'Error completing scale-in');
+      });
   }
 
   describeInstance() {
     return new Promise((resolve, reject) => {
       if (!this.instanceId) return reject('instance-id unknown');
-      autoscaling.describeAutoScalingInstances({
+      autoScalingClient.send(new DescribeAutoScalingGroupsCommand({
         InstanceIds: [this.instanceId]
-      }, (err, data) => {
-        if (err) {
+      }))
+        .then((data) => {
+          this.logger.info({data}, 'SnsNotifier: describeInstance');
+          return resolve(data);
+        })
+        .catch((err) => {
           this.logger.error({err}, 'Error describing instances');
           reject(err);
-        } else {
-          this.logger.info({data}, 'SnsNotifier: describeInstance');
-          resolve(data);
-        }
-      });
+        });
     });
   }
 
@@ -193,7 +203,7 @@ module.exports = async function(logger) {
   process.on('SIGHUP', async() => {
     try {
       const data = await notifier.describeInstance();
-      const state = data.AutoScalingInstances[0].LifecycleState;
+      const state = data.AutoScalingGroups[0].Instances[0].LifecycleState;
       if (state !== notifier.lifecycleState) {
         notifier.lifecycleState = state;
         switch (state) {

lib/utils/http-requestor.js

@@ -87,7 +87,7 @@ class HttpRequestor extends BaseRequestor {
 
     assert(HookMsgTypes.includes(type));
 
-    const payload = params ? snakeCaseKeys(params, ['customerData', 'sip']) : null;
+    const payload = params ? snakeCaseKeys(params, ['customerdata', 'customerData', 'sip']) : null;
     const url = hook.url || hook;
     const method = hook.method || 'POST';
     let buf = '';

lib/utils/sbc-pinger.js

@@ -101,7 +101,8 @@ module.exports = (logger) => {
           method: 'OPTIONS',
           headers: {
             'X-FS-Status': ms && !dryUpCalls ? 'open' : 'closed',
-            'X-FS-Calls': srf.locals.sessionTracker.count
+            'X-FS-Calls': srf.locals.sessionTracker.count,
+            'X-FS-ServiceUrl': srf.locals.serviceUrl
           }
         });
         req.on('response', (res) => {

lib/utils/ws-requestor.js

@@ -138,7 +138,7 @@ class WsRequestor extends BaseRequestor {
 
     /* simple notifications */
     if (['call:status', 'verb:status', 'jambonz:error'].includes(type) || reconnectingWithoutAck) {
-      this.ws.send(JSON.stringify(obj), () => {
+      this.ws?.send(JSON.stringify(obj), () => {
         this.logger.debug({obj}, `WsRequestor:request websocket: sent (${url})`);
         sendQueuedMsgs();
       });

package.json

@@ -1,6 +1,6 @@
 {
   "name": "jambonz-feature-server",
-  "version": "0.8.3",
+  "version": "0.8.4",
   "main": "app.js",
   "engines": {
     "node": ">= 10.16.0"
@@ -19,19 +19,19 @@
   "bugs": {},
   "scripts": {
     "start": "node app",
-    "test": "NODE_ENV=test JAMBONES_HOSTING=1 HTTP_POOL=1 ENCRYPTION_SECRET=foobar 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=error ENABLE_METRICS=0 HTTP_PORT=3000 JAMBONES_SBCS=172.38.0.10 JAMBONES_FREESWITCH=127.0.0.1:8022:JambonzR0ck$: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 JAMBONES_TTS_TRIM_SILENCE=1 ENCRYPTION_SECRET=foobar 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=error ENABLE_METRICS=0 HTTP_PORT=3000 JAMBONES_SBCS=172.38.0.10 JAMBONES_FREESWITCH=127.0.0.1:8022:JambonzR0ck$: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 tracer.js lib",
     "jslint:fix": "eslint app.js tracer.js lib --fix"
   },
   "dependencies": {
-    "@jambonz/db-helpers": "^0.9.0",
+    "@jambonz/db-helpers": "^0.9.1",
     "@jambonz/http-health-check": "^0.0.1",
     "@jambonz/realtimedb-helpers": "^0.8.6",
-    "@jambonz/speech-utils": "^0.0.15",
+    "@jambonz/speech-utils": "^0.0.18",
     "@jambonz/stats-collector": "^0.1.8",
-    "@jambonz/time-series": "^0.2.7",
-    "@jambonz/verb-specifications": "^0.0.24",
+    "@jambonz/time-series": "^0.2.8",
+    "@jambonz/verb-specifications": "^0.0.26",
     "@opentelemetry/api": "^1.4.0",
     "@opentelemetry/exporter-jaeger": "^1.9.0",
     "@opentelemetry/exporter-trace-otlp-http": "^0.35.0",
@@ -41,7 +41,8 @@
     "@opentelemetry/sdk-trace-base": "^1.9.0",
     "@opentelemetry/sdk-trace-node": "^1.9.0",
     "@opentelemetry/semantic-conventions": "^1.9.0",
-    "aws-sdk": "^2.1313.0",
+    "@aws-sdk/client-sns": "^3.360.0",
+    "@aws-sdk/client-auto-scaling": "^3.360.0",
     "bent": "^7.3.12",
     "debug": "^4.3.4",
     "deepcopy": "^2.1.0",

test/create-call-test.js

@@ -80,6 +80,19 @@ test('test create-call call-hook basic authentication', async(t) => {
         "username": "username",
         "password": "password"
       },
+      "call_status_hook": {
+        "url": "http://127.0.0.1:3100/callStatus",
+        "method": "POST"
+      },
+      "tag": {
+        "callCount": 10,
+        "env": "DEVELOPMENT",
+        "processId": "processId",
+        "Customer": "Customer",
+        "Vehicle": "Vehicle",
+        "Event_Camel": "Event_Camel",
+        "CamelCase": "CamelCase"
+      },
       "from": from,
       "to": {
         "type": "phone",
@@ -96,9 +109,19 @@ test('test create-call call-hook basic authentication', async(t) => {
     //THEN
     await p;
 
-    let obj = await getJSON(`http:127.0.0.1:3100/lastRequest/${from}`)
+    let obj = await getJSON(`http:127.0.0.1:3100/lastRequest/${from}`);
     t.ok(obj.headers.Authorization = 'Basic dXNlcm5hbWU6cGFzc3dvcmQ=',
       'create-call: call-hook contains basic authentication header');
+    t.ok(obj.body.customerdata.callCount === 10,
+      'create-call: call-hook contains correct format for customerData');
+      t.ok(obj.body.customerData.callCount === 10,
+        'create-call: call-hook contains correct format for customerData');
+
+      obj = await getJSON(`http:127.0.0.1:3100/lastRequest/${from}_callStatus`);
+      t.ok(obj.body.customerdata.callCount === 10,
+        'create-call: status-hook contains correct format for customerData');
+        t.ok(obj.body.customerData.callCount === 10,
+          'create-call: status-hook contains correct format for customerData');
     disconnect();
   } catch (err) {
     console.log(`error received: ${err}`);

test/dial-tests.js

@@ -5,6 +5,8 @@ const getJSON = bent('json')
 const clearModule = require('clear-module');
 const {provisionCallHook} = require('./utils')
 
+const sleepFor = (ms) => new Promise((r) => setTimeout(r, ms));
+
 process.on('unhandledRejection', (reason, p) => {
   console.log('Unhandled Rejection at: Promise', p, 'reason:', reason);
 });
@@ -47,6 +49,7 @@ test('\'dial-phone\'', async(t) => {
 
     // THEN
     const p = sippUac('uas-dial.xml', '172.38.0.10', undefined, undefined, 2);
+    await sleepFor(1000);
 
     let account_sid = '622f62e4-303a-49f2-bbe0-eb1e1714e37a';
     let post = bent('http://127.0.0.1:3000/', 'POST', 'json', 201);
@@ -84,7 +87,7 @@ test('\'dial-sip\'', async(t) => {
     try {
       await connect(srf);
       // wait for fs connected to drachtio server.
-      await new Promise(r => setTimeout(r, 1000));
+      await sleepFor(1000);
       // GIVEN
       const from = "dial_sip";
       let verbs = [

test/docker-compose-testbed.yaml

@@ -42,7 +42,7 @@ services:
         ipv4_address: 172.38.0.7
 
   drachtio:
-    image: drachtio/drachtio-server:latest
+    image: drachtio/drachtio-server:0.8.22
     restart: always
     command: drachtio --contact "sip:*;transport=udp" --mtu 4096 --address 0.0.0.0 --port 9022
     ports:
@@ -57,7 +57,7 @@ services:
         condition: service_healthy
 
   freeswitch:
-    image: drachtio/drachtio-freeswitch-mrf:0.4.18
+    image: drachtio/drachtio-freeswitch-mrf:0.4.33
     restart: always
     command: freeswitch --rtp-range-start 20000 --rtp-range-end 20100
     environment:

test/say-tests.js

@@ -2,6 +2,8 @@ const test = require('tape');
 const { sippUac } = require('./sipp')('test_fs');
 const clearModule = require('clear-module');
 const {provisionCallHook} = require('./utils')
+const bent = require('bent');
+const getJSON = bent('json')
 
 process.on('unhandledRejection', (reason, p) => {
   console.log('Unhandled Rejection at: Promise', p, 'reason:', reason);
@@ -27,6 +29,18 @@ test('\'say\' tests', async(t) => {
       {
         verb: 'say',
         text: 'hello'
+      },
+      {
+        verb: "tag",
+        data: {
+          callCount: 10,
+          env: "DEVELOPMENT",
+          processId: "processId",
+          Customer: "Customer",
+          Vehicle: "Vehicle",
+          Event_Camel: "Event_Camel",
+          CamelCase: "CamelCase"
+        }
       }
     ];
 
@@ -36,6 +50,12 @@ test('\'say\' tests', async(t) => {
     // THEN
     await sippUac('uac-success-received-bye.xml', '172.38.0.10', from);
     t.pass('say: succeeds when using using account credentials');
+    let obj = await getJSON(`http:127.0.0.1:3100/lastRequest/${from}_callStatus`);
+    console.log(obj);
+    t.ok(obj.body.customerdata.callCount === 10,
+      'create-call: status-hook contains correct format for customerData');
+      t.ok(obj.body.customerData.callCount === 10,
+        'create-call: status-hook contains correct format for customerData');
     disconnect();
   } catch (err) {
     console.log(`error received: ${err}`);