add support for ws verb:status event notifications (#196)

lib/session/call-session.js

@@ -63,6 +63,7 @@ class CallSession extends Emitter {
     assert(rootSpan);
 
     this._recordState = RecordState.RecordingOff;
+    this._notifyEvents = false;
 
     this.tmpFiles = new Set();
 
@@ -265,6 +266,9 @@ class CallSession extends Emitter {
 
   get recordState() { return this._recordState; }
 
+  get notifyEvents() { return this._notifyEvents; }
+  set notifyEvents(notify) { this._notifyEvents = !!notify; }
+
   set globalSttHints({hints, hintsBoost}) {
     this._globalSttHints = {hints, hintsBoost};
   }
@@ -612,6 +616,7 @@ class CallSession extends Emitter {
       const stackNum = this.stackIdx;
       const task = this.tasks.shift();
       this.logger.info(`CallSession:exec starting task #${stackNum}:${taskNum}: ${task.name}`);
+      this._notifyTaskStatus(task, {event: 'starting'});
       try {
         const resources = await this._evaluatePreconditions(task);
         let skip = false;
@@ -635,6 +640,7 @@ class CallSession extends Emitter {
         }
         this.currentTask = null;
         this.logger.info(`CallSession:exec completed task #${stackNum}:${taskNum}: ${task.name}`);
+        this._notifyTaskStatus(task, {event: 'finished'});
       } catch (err) {
         task.span?.end();
         this.currentTask = null;
@@ -1623,6 +1629,25 @@ class CallSession extends Emitter {
       .catch((err) => this.logger.error(err, 'redis error'));
   }
 
+  /**
+   * notifyTaskError - only used when websocket connection is used instead of webhooks
+   */
+
+  _notifyTaskError(obj) {
+    if (this.requestor instanceof WsRequestor) {
+      this.requestor.request('jambonz:error', '/error', obj)
+        .catch((err) => this.logger.debug({err}, 'CallSession:_notifyTaskError - Error sending'));
+    }
+  }
+
+  _notifyTaskStatus(task, evt) {
+    if (this.notifyEvents && this.requestor instanceof WsRequestor) {
+      const obj = {...evt, id: task.id, name: task.name};
+      this.requestor.request('verb:status', '/status', obj)
+        .catch((err) => this.logger.debug({err}, 'CallSession:_notifyTaskStatus - Error sending'));
+    }
+  }
+
   _awaitCommandsOrHangup() {
     assert(!this.wakeupResolver);
     return new Promise((resolve, reject) => {

lib/tasks/config.js

@@ -11,6 +11,10 @@ class TaskConfig extends Task {
       'record'
     ].forEach((k) => this[k] = this.data[k] || {});
 
+    if ('notifyEvents' in this.data) {
+      this.notifyEvents = !!this.data.notifyEvents;
+    }
+
     if (this.bargeIn.enable) {
       this.gatherOpts = {
         verb: 'gather',
@@ -51,12 +55,19 @@ class TaskConfig extends Task {
       phrase.push(`set recognizer${s}`);
     }
     if (this.data.amd) phrase.push('enable amd');
-    return  `${this.name}{${phrase.join(',')}`;
+    if (this.notifyEvents) phrase.push(`event notification ${this.notifyEvents ? 'on' : 'off'}`);
+    return `${this.name}{${phrase.join(',')}`;
   }
 
   async exec(cs, {ep} = {}) {
     await super.exec(cs);
 
+    if (this.notifyEvents) {
+      this.logger.debug(`turning event notification ${this.notifyEvents ? 'on' : 'off'}`);
+      cs.notifyEvents = !!this.data.notifEvents;
+
+    }
+
     if (this.data.amd) {
       this.startAmd = cs.startAmd;
       this.stopAmd = cs.stopAmd;

lib/tasks/say.js

@@ -156,7 +156,10 @@ class TaskSay extends Task {
           alert_type: AlertType.TTS_NOT_PROVISIONED,
           vendor
         }).catch((err) => this.logger.info({err}, 'Error generating alert for no tts'));
-        this.notifyError(`No speech credentials have been provisioned for ${vendor}`);
+        this.notifyError({
+          msg: 'TTS error',
+          details:`No speech credentials provisioned for selected vendor ${vendor}`
+        });
         throw new Error('no provisioned speech credentials for TTS');
       }
       // synthesize all of the text elements
@@ -174,7 +177,7 @@ class TaskSay extends Task {
           'tts.voice': voice
         });
         try {
-          const {filePath, servedFromCache} = await synthAudio(stats, {
+          const {filePath, servedFromCache, rtt} = await synthAudio(stats, {
             text,
             vendor,
             language,
@@ -193,6 +196,15 @@ class TaskSay extends Task {
           }
           span.setAttributes({'tts.cached': servedFromCache});
           span.end();
+          if (!servedFromCache && rtt) {
+            this.notifyStatus({
+              event: 'synthesized-audio',
+              vendor,
+              language,
+              characters: text.length,
+              elapsedTime: rtt
+            });
+          }
           return filePath;
         } catch (err) {
           this.logger.info({err}, 'Error synthesizing tts');
@@ -203,7 +215,7 @@ class TaskSay extends Task {
             vendor,
             detail: err.message
           }).catch((err) => this.logger.info({err}, 'Error generating alert for tts failure'));
-          this.notifyError(err.message || err);
+          this.notifyError({msg: 'TTS error', details: err.message || err});
           return;
         }
       };
@@ -211,6 +223,7 @@ class TaskSay extends Task {
       const arr = this.text.map((t) => generateAudio(t));
       const filepath = (await Promise.all(arr)).filter((fp) => fp && fp.length);
       this.logger.debug({filepath}, 'synthesized files for tts');
+      this.notifyStatus({event: 'start-playback'});
 
       while (!this.killed && (this.loop === 'forever' || this.loop--) && this.ep?.connected) {
         let segment = 0;
@@ -242,6 +255,7 @@ class TaskSay extends Task {
         this.killPlayToConfMember(this.ep, memberId, confName);
       }
       else {
+        this.notifyStatus({event: 'kill-playback'});
         this.ep.api('uuid_break', this.ep.uuid);
       }
     }

lib/tasks/specs.json

@@ -1,6 +1,7 @@
 {
   "sip:decline": {
     "properties": {
+      "id": "string",
       "status": "number",
       "reason": "string",
       "headers": "object"
@@ -11,6 +12,7 @@
   },
   "sip:request": {
     "properties": {
+      "id": "string",
       "method": "string",
       "body": "string",
       "headers": "object",
@@ -22,6 +24,7 @@
   },
   "sip:refer": {
     "properties": {
+      "id": "string",
       "referTo": "string",
       "referredBy": "string",
       "headers": "object",
@@ -34,11 +37,13 @@
   },
   "config": {
     "properties": {
+      "id": "string",
       "synthesizer": "#synthesizer",
       "recognizer": "#recognizer",
       "bargeIn": "#bargeIn",
       "record": "#recordOptions",
-      "amd": "#amd"
+      "amd": "#amd",
+      "notifyEvents": "boolean"
     },
     "required": []
   },
@@ -62,6 +67,7 @@
   },
   "dequeue": {
     "properties": {
+      "id": "string",
       "name": "string",
       "actionHook": "object|string",
       "timeout": "number",
@@ -73,6 +79,7 @@
   },
   "enqueue": {
     "properties": {
+      "id": "string",
       "name": "string",
       "actionHook": "object|string",
       "waitHook": "object|string",
@@ -84,11 +91,12 @@
   },
   "leave": {
     "properties": {
-
+      "id": "string"
     }
   },
   "hangup": {
     "properties": {
+      "id": "string",
       "headers": "object"
     },
     "required": [
@@ -96,6 +104,7 @@
   },
   "play": {
     "properties": {
+      "id": "string",
       "url": "string|array",
       "loop": "number|string",
       "earlyMedia": "boolean",
@@ -109,6 +118,7 @@
   },
   "say": {
     "properties": {
+      "id": "string",
       "text": "string|array",
       "loop": "number|string",
       "synthesizer": "#synthesizer",
@@ -120,6 +130,7 @@
   },
   "gather": {
     "properties": {
+      "id": "string",
       "actionHook": "object|string",
       "finishOnKey": "string",
       "input": "array",
@@ -143,6 +154,7 @@
   },
   "conference": {
     "properties": {
+      "id": "string",
       "name": "string",
       "beep": "boolean",
       "startConferenceOnEnter": "boolean",
@@ -162,6 +174,7 @@
   },
   "dial": {
     "properties": {
+      "id": "string",
       "actionHook": "object|string",
       "answerOnBridge": "boolean",
       "callerId": "string",
@@ -185,6 +198,7 @@
   },
   "dialogflow": {
     "properties": {
+      "id": "string",
       "credentials": "object|string",
       "project": "string",
       "environment": "string",
@@ -213,6 +227,7 @@
   },
   "dtmf": {
     "properties": {
+      "id": "string",
       "dtmf": "string",
       "duration": "number"
     },
@@ -222,6 +237,7 @@
   },
   "lex": {
     "properties": {
+      "id": "string",
       "botId": "string",
       "botAlias": "string",
       "credentials": "object",
@@ -246,6 +262,7 @@
   },
   "listen": {
     "properties": {
+      "id": "string",
       "actionHook": "object|string",
       "auth": "#auth",
       "finishOnKey": "string",
@@ -270,6 +287,7 @@
   },
   "message": {
     "properties": {
+      "id": "string",
       "carrier": "string",
       "account_sid": "string",
       "message_sid": "string",
@@ -286,6 +304,7 @@
   },
   "pause": {
     "properties": {
+      "id": "string",
       "length": "number"
     },
     "required": [
@@ -294,6 +313,7 @@
   },
   "rasa": {
     "properties": {
+      "id": "string",
       "url": "string",
       "recognizer": "#recognizer",
       "tts": "#synthesizer",
@@ -328,6 +348,7 @@
   },
   "redirect": {
     "properties": {
+      "id": "string",
       "actionHook": "object|string"
     },
     "required": [
@@ -336,6 +357,7 @@
   },
   "rest:dial": {
     "properties": {
+      "id": "string",
       "account_sid": "string",
       "application_sid": "string",
       "call_hook": "object|string",
@@ -360,6 +382,7 @@
   },
   "tag": {
     "properties": {
+      "id": "string",
       "data": "object"
     },
     "required": [
@@ -368,6 +391,7 @@
   },
   "transcribe": {
     "properties": {
+      "id": "string",
       "transcriptionHook": "string",
       "recognizer": "#recognizer",
       "earlyMedia": "boolean"

lib/tasks/task.js

@@ -4,6 +4,7 @@ const debug = require('debug')('jambonz:feature-server');
 const assert = require('assert');
 const {TaskPreconditions} = require('../utils/constants');
 const normalizeJambones = require('../utils/normalize-jambones');
+const WsRequestor = require('../utils/ws-requestor');
 const {trace} = require('@opentelemetry/api');
 const specs = new Map();
 const _specData = require('./specs');
@@ -21,6 +22,7 @@ class Task extends Emitter {
     this.logger = logger;
     this.data = data;
     this.actionHook = this.data.actionHook;
+    this.id = data.id;
 
     this._killInProgress = false;
     this._completionPromise = new Promise((resolve) => this._completionResolver = resolve);
@@ -137,10 +139,20 @@ class Task extends Emitter {
     return this.callSession.normalizeUrl(url, method, auth);
   }
 
-  notifyError(errMsg) {
+  notifyError(obj) {
-    const params = {error: errMsg, verb: this.name};
+    if (this.cs.requestor instanceof WsRequestor) {
-    this.cs.requestor.request('jambonz:error', '/error', params)
+      const params = {...obj, verb: this.name, id: this.id};
-      .catch((err) => this.logger.info({err}, 'Task:notifyError error sending error'));
+      this.cs.requestor.request('jambonz:error', '/error', params)
+        .catch((err) => this.logger.info({err}, 'Task:notifyError error sending error'));
+    }
+  }
+
+  notifyStatus(obj) {
+    if (this.cs.notifyEvents && this.cs.requestor instanceof WsRequestor) {
+      const params = {...obj, verb: this.name, id: this.id};
+      this.cs.requestor.request('verb:status', '/status', params)
+        .catch((err) => this.logger.info({err}, 'Task:notifyStatus error sending error'));
+    }
   }
 
   async performAction(results, expectResponse = true) {