update to azure 1.22.0

* issue# 107: on gather timeout, if minDigits are collected, resolve with dtmf-num-digits

* gather timeout: use conditional instead of if/else

Co-authored-by: Prashanth Gujjeti <prashanth@minervacq.com>

Dockerfile

@@ -1,4 +1,4 @@
-FROM node:slim
+FROM node:lts-slim
 WORKDIR /opt/app/
 COPY package.json package-lock.json ./
 RUN npm ci

README.md

@@ -2,6 +2,8 @@
 
 This application implements the core feature server of the jambones platform.
 
+> Note: If you are a developer looking to work on the code please read  our [how-to for that](./docs/contributing.md).
+
 ## Configuration
 
 Configuration is provided via environment variables:
@@ -84,7 +86,5 @@ module.exports = {
 ```
 
 #### Running the test suite
-The test suite currently only consists of JSON-parsing unit tests.  A full end-to-end sip test suite should be added.
-```
-npm test
-```
+
+Please [see this]](./docs/contributing.md#run-the-regression-test-suite).

docs/contributing.md

@@ -0,0 +1,123 @@
+# Contributors are welcome!
+
+So, you want to hack on jambonz?  Maybe add some features, maybe help fix some bugs?  Awesome, welcome aboard! 
+
+This brief document should get you started.  Here you will find instructions showing how to set up your laptop to run the regression test suite (which you should always run before committing any changes), as well as some basic info on the structure of the code.
+
+## Getting oriented
+
+First of all, you are in the right place to begin hacking on jambonz. The jambonz-feature-server app is kinda the center of the universe for jambonz.  Most of the core logic in jambonz is implemented here: things like the [webhook verbs](../lib/tasks), [session management](../lib/session), and the [client-side webhook implementation](../lib/utils/http-requestor.js). A common thing you might want to do, for instance, is to add support for an all-new verb, and this code base is where would do that.
+
+This jambonz-feature-server app works together quite closely with a [drachtio server](https://github.com/drachtio/drachtio-server) and a Freeswitch.  In fact, these three components are bundled together into a single VM/instance (or a Deployment, in Kubernetes) that we more generally refer to as "Feature Server".  The Feature Server is a horizontally-scalable unit that is deployed behind the public-facing SBC elements of a jambonz cluster (the SBC is itself a separately scalable unit).  The drachtio-server handles the SIP signaling, the Freeswitch handles media operations and speech vendor integration, and the jambonz-feature-server app orchestrates all of it via the use of [drachtio-srf](https://github.com/drachtio/drachtio-srf) and [drachtio-fsmrf](https://github.com/drachtio/drachtio-fsmrf).
+
+## How to do things
+
+First of all, please join our [slack channel](https://joinslack.jambonz.org) in order to coordinate with us on the work, i.e. to notify us of what you are doing and make sure that no one else is already working on the same thing.
+
+To prepare to make changes, please fork the repo to your own Github account, make changes, test them on your own running jambonz cluster, then run the regression test suite and lint check before giving us a PR.
+
+### lint
+
+We have some opinionated conventions that you must follow - see our [eslintrc.json](../.eslintrc.json) for details. Make sure your code passes by running:
+
+```bash
+npm run jslint
+```
+
+### test suite
+
+#### Generate speech credentials and create run-tests.sh
+
+The test suite also requires you to provide speech credentials for both GCP and AWS.  You will want to create a new file named `run-tests.sh` in the project folder. Make the file executable and then copy in the text below, substituting your speech credentials where indicated:
+
+```bash
+#!/bin/bash
+GCP_JSON_KEY='{"type":"service_account","project_id":"...etc"}' \
+AWS_ACCESS_KEY_ID='your-aws-access-key-id' \
+AWS_SECRET_ACCESS_KEY='your-aws-secret-access-key' \
+AWS_REGION='us-east-1' \
+JWT_SECRET='foobar' \
+npm test
+```
+>> Note: The project's .gitignore file prevents this file from being sent to Github, so you do not need to worry about exposing your credentials.  Just make sure you name if run-tests.sh and create it in the project folder
+
+The GCP credential is the JSON service key in stringified format.
+
+#### Install Docker
+
+The test suite ralso equires [Docker](https://www.docker.com/) and docker-compose to be installed on your laptop.  Docker is used to set up a network with all of the elements required to test the jambonz-feature-server in a black-box type of fashion.
+
+Once you have docker installed, you can optionally make sure everything Docker-wise is working properly by running this command from the project folder:
+
+```bash
+docker-compose -f test/docker-compose-testbed.yaml up -d
+```
+
+This may take several minutes to complete, mainly because the mysql schema needs to be installed and seeded, but if successful the output should look like this:
+
+```bash
+$ docker-compose -f test/docker-compose-testbed.yaml up -d
+Creating network "test_fs" with driver "bridge"
+Creating test_webhook-transcribe_1 ... done
+Creating test_webhook-decline_1    ... done
+Creating test_mysql_1              ... done
+Creating test_docker-host_1        ... done
+Creating test_webhook-gather_1     ... done
+Creating test_webhook-say_1        ... done
+Creating test_freeswitch_1         ... done
+Creating test_influxdb_1           ... done
+Creating test_redis_1              ... done
+Creating test_drachtio_1           ... done
+```
+
+At that point, you can run `docker ps` to see all of the containers running
+
+```bash
+docker ps
+CONTAINER ID   IMAGE                                           COMMAND                  CREATED              STATUS                   PORTS                               NAMES
+abbc3594f390   drachtio/drachtio-server:latest                 "/entrypoint.sh drac…"   About a minute ago   Up About a minute        0.0.0.0:9060->9022/tcp              test_drachtio_1
+1f384a274f87   redis:5-alpine                                  "docker-entrypoint.s…"   2 minutes ago        Up 2 minutes             0.0.0.0:16379->6379/tcp             test_redis_1
+78d0bb6ec9b1   influxdb:1.8                                    "/entrypoint.sh infl…"   2 minutes ago        Up 2 minutes             0.0.0.0:8086->8086/tcp              test_influxdb_1
+9616ff790709   jambonz/webhook-test-scaffold:latest            "/entrypoint.sh"         2 minutes ago        Up 2 minutes             0.0.0.0:3102->3000/tcp              test_webhook-gather_1
+7323ab273ff4   drachtio/drachtio-freeswitch-mrf:v1.10.1-full   "/entrypoint.sh free…"   2 minutes ago        Up 2 minutes (healthy)   0.0.0.0:8022->8021/tcp              test_freeswitch_1
+e45e7d28dbc7   mysql:5.7                                       "docker-entrypoint.s…"   2 minutes ago        Up 2 minutes (healthy)   33060/tcp, 0.0.0.0:3360->3306/tcp   test_mysql_1
+b626e5f3067e   qoomon/docker-host                              "/entrypoint.sh"         2 minutes ago        Up 2 minutes                                                 test_docker-host_1
+b0a94b5e8941   jambonz/webhook-test-scaffold:latest            "/entrypoint.sh"         2 minutes ago        Up 2 minutes             0.0.0.0:3101->3000/tcp              test_webhook-say_1
+f80adda48eb5   jambonz/webhook-test-scaffold:latest            "/entrypoint.sh"         2 minutes ago        Up 2 minutes             0.0.0.0:3103->3000/tcp              test_webhook-transcribe_1
+223db4a9c670   jambonz/webhook-test-scaffold:latest            "/entrypoint.sh"         2 minutes ago        Up 2 minutes             0.0.0.0:3100->3000/tcp              test_webhook-decline_1
+```
+
+#### Run the regression test suite
+
+The test suite has a dependency that the mysql client is installed on your laptop/machine where the test will be run.  This is needed in order to seed the mysql database that is running in the docker network.
+
+Assuming you have installed the mysql client, and done the above steps, you should now be able to run the tests:
+
+```bash
+./run-tests.sh
+```
+
+If the docker network has not been started (as described above) it will start now, and this will take a minute or two.  Otherwise, the test suite will start running immediately.
+
+In evaluating the test results, be advised that the output is fairly verbose, and also in the process of shutting down once the tests are complete you will see a bunch of errors from redis (`@jambonz/realtimedb-helpers - redis error`).  You can ignore these errors, they are just spit out by jambonz-feature-server as the test environment is torn down and it tries and fails to reconnect to redis.
+
+The final output will indicate the number of tests run and passed:
+
+```bash
+1..28
+# tests 28
+# pass  28
+
+# ok
+```
+
+#### Adding your own tests
+
+Running a successful regression test means you haven't broken anything - Great!  
+
+It doesn't, of course, mean that your shiny new feature or bugfix works.  Adding a new test case to the suite is (unfortunately) non-trivial.  We will add more documentation in the future with a how-to guide on that, but be advised it does require knowledge of the SIP protocol and the [SIPp](http://sipp.sourceforge.net/doc/reference.html) tool.
+
+For now, if you are unable to add tests to the regression suite, please do test your feature as thoroughly as you can on your own jambonz cluster before giving us a pull request.
+
+
+

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

@@ -14,12 +14,12 @@ const dbUtils = require('../../utils/db-utils');
 router.post('/', async(req, res) => {
   const {logger} = req.app.locals;
   const accountSid = req.body.account_sid;
+  const {srf} = require('../../..');
 
   logger.debug({body: req.body}, 'got createCall request');
   try {
     let uri, cs, to;
     const restDial = makeTask(logger, {'rest:dial': req.body});
-    const {srf} = require('../../..');
     const {lookupAccountDetails} = dbUtils(logger, srf);
     const {getSBC, getFreeswitch} = srf.locals;
     const sbcAddress = getSBC();
@@ -77,7 +77,7 @@ router.post('/', async(req, res) => {
     if (target.type === 'phone' && target.trunk) {
       const {lookupCarrier} = dbUtils(this.logger, srf);
       const voip_carrier_sid = await lookupCarrier(req.body.account_sid, target.trunk);
-      this.logger.info(
+      logger.info(
         `createCall: selected ${voip_carrier_sid} for requested carrier: ${target.trunk || 'unspecified'})`);
       if (voip_carrier_sid) {
         opts.headers['X-Requested-Carrier-Sid'] = voip_carrier_sid;
@@ -118,15 +118,25 @@ router.post('/', async(req, res) => {
      * attach our requestor and notifier objects
      * these will be used for all http requests we make during this call
      */
-    if ('WS' === app.call_hook?.method) {
+    if ('WS' === app.call_hook?.method || /^wss?:/.test(app.call_hook.url)) {
+      logger.debug({call_hook: app.call_hook}, 'creating websocket for call hook');
       app.requestor = new WsRequestor(logger, account.account_sid, app.call_hook, account.webhook_secret) ;
-      app.notifier = app.requestor;
+      if (app.call_hook.url === app.call_status_hook.url  || !app.call_status_hook?.url) {
+        logger.debug('reusing websocket for call status hook');
+        app.notifier = app.requestor;
+      }
     }
     else {
+      logger.debug({call_hook: app.call_hook}, 'creating http client for call hook');
       app.requestor = new HttpRequestor(logger, account.account_sid, app.call_hook, account.webhook_secret);
-      if (app.call_status_hook) app.notifier = new HttpRequestor(logger, account.account_sid, app.call_status_hook,
-        account.webhook_secret);
-      else app.notifier = {request: () => {}};
+    }
+    if (!app.notifier && app.call_status_hook) {
+      app.notifier = new HttpRequestor(logger, account.account_sid, app.call_status_hook, account.webhook_secret);
+      logger.debug({call_hook: app.call_hook}, 'creating http client for call status hook');
+    }
+    else if (!app.notifier) {
+      logger.debug('creating null call status hook');
+      app.notifier = {request: () => {}};
     }
 
     /* now launch the outdial */
@@ -230,6 +240,7 @@ router.post('/', async(req, res) => {
         else console.error(err);
       }
       ep.destroy();
+      setTimeout(restDial.kill.bind(restDial), 5000);
     }
   } catch (err) {
     sysError(logger, res, err);

lib/session/call-session.js

@@ -239,10 +239,10 @@ class CallSession extends Emitter {
       const t = normalizeJambones(this.logger, [gather]);
       this.backgroundGatherTask = makeTask(this.logger, t[0]);
       this.backgroundGatherTask
-        .on('dtmf', this._clearTasks.bind(this))
-        .on('vad', this._clearTasks.bind(this))
-        .on('transcription', this._clearTasks.bind(this))
-        .on('timeout', this._clearTasks.bind(this));
+        .once('dtmf', this._clearTasks.bind(this))
+        .once('vad', this._clearTasks.bind(this))
+        .once('transcription', this._clearTasks.bind(this))
+        .once('timeout', this._clearTasks.bind(this));
       this.logger.info({gather}, 'CallSession:enableBotMode - starting background gather');
       const resources = await this._evaluatePreconditions(this.backgroundGatherTask);
       const {span, ctx} = this.rootSpan.startChildSpan(`background-gather:${this.backgroundGatherTask.summary}`);

lib/tasks/conference.js

@@ -551,7 +551,8 @@ class Conference extends Task {
         accountInfo: cs.accountInfo,
         memberId: this.memberId,
         confName: this.confName,
-        tasks
+        tasks,
+        rootSpan: cs.rootSpan
       });
       await this._playSession.exec();
       this._playSession = null;

lib/tasks/dial.js

@@ -606,8 +606,8 @@ class TaskDial extends Task {
     if (this.parentDtmfCollector) this._installDtmfDetection(cs, cs.dlg);
     if (this.childDtmfCollector) this._installDtmfDetection(cs, this.dlg);
 
-    if (this.transcribeTask) this.transcribeTask.exec(cs, this.ep);
-    if (this.listenTask) this.listenTask.exec(cs, this.ep);
+    if (this.transcribeTask) this.transcribeTask.exec(cs, this.epOther, this.ep);
+    if (this.listenTask) this.listenTask.exec(cs, this.epOther);
 
     /* if we can release the media back to the SBC, do so now */
     if (this.canReleaseMedia) this._releaseMedia(cs, sd);

lib/tasks/enqueue.js

@@ -349,7 +349,8 @@ class TaskEnqueue extends Task {
         ep: cs.ep,
         callInfo: cs.callInfo,
         accountInfo: cs.accountInfo,
-        tasks: tasksToRun
+        tasks: tasksToRun,
+        rootSpan: cs.rootSpan
       });
       await this._playSession.exec();
       this._playSession = null;

lib/tasks/gather.js

@@ -36,8 +36,18 @@ class TaskGather extends Task {
       this.language = recognizer.language;
       this.hints = recognizer.hints || [];
       this.hintsBoost = recognizer.hintsBoost;
-      this.altLanguages = recognizer.altLanguages || [];
+      this.profanityFilter = recognizer.profanityFilter;
       this.punctuation = !!recognizer.punctuation;
+      this.enhancedModel = !!recognizer.enhancedModel;
+      this.model = recognizer.model || 'command_and_search';
+      this.words = !!recognizer.words;
+      this.singleUtterance = recognizer.singleUtterance || true;
+      this.diarization = !!recognizer.diarization;
+      this.diarizationMinSpeakers = recognizer.diarizationMinSpeakers || 0;
+      this.diarizationMaxSpeakers = recognizer.diarizationMaxSpeakers || 0;
+      this.interactionType = recognizer.interactionType || 'unspecified';
+      this.naicsCode = recognizer.naicsCode || 0;
+      this.altLanguages = recognizer.altLanguages || [];
 
       /* vad: if provided, we dont connect to recognizer until voice activity is detected */
       const {enable, voiceMs = 0, mode = -1} = recognizer.vad || {};
@@ -232,24 +242,35 @@ class TaskGather extends Task {
 
     if ('google' === this.vendor) {
       if (this.sttCredentials) opts.GOOGLE_APPLICATION_CREDENTIALS = JSON.stringify(this.sttCredentials.credentials);
-      Object.assign(opts, {
-        GOOGLE_SPEECH_USE_ENHANCED: true,
-        GOOGLE_SPEECH_SINGLE_UTTERANCE: true,
-        GOOGLE_SPEECH_MODEL: 'command_and_search',
-        GOOGLE_SPEECH_ENABLE_AUTOMATIC_PUNCTUATION: !!this.punctuation
+      [
+        ['enhancedModel', 'GOOGLE_SPEECH_USE_ENHANCED'],
+        ['separateRecognitionPerChannel', 'GOOGLE_SPEECH_SEPARATE_RECOGNITION_PER_CHANNEL'],
+        ['profanityFilter', 'GOOGLE_SPEECH_PROFANITY_FILTER'],
+        ['punctuation', 'GOOGLE_SPEECH_ENABLE_AUTOMATIC_PUNCTUATION'],
+        ['words', 'GOOGLE_SPEECH_ENABLE_WORD_TIME_OFFSETS'],
+        ['singleUtterance', 'GOOGLE_SPEECH_SINGLE_UTTERANCE'],
+        ['diarization', 'GOOGLE_SPEECH_PROFANITY_FILTER']
+      ].forEach((arr) => {
+        if (this[arr[0]]) opts[arr[1]] = true;
       });
-      if (this.hints && this.hints.length > 1) {
-        opts.GOOGLE_SPEECH_HINTS = this.hints.map((h) => h.trim()).join(',');
+      if (this.hints.length > 1) {
+        opts.GOOGLE_SPEECH_HINTS = this.hints.join(',');
         if (typeof this.hintsBoost === 'number') {
           opts.GOOGLE_SPEECH_HINTS_BOOST = this.hintsBoost;
         }
       }
-      if (this.altLanguages && this.altLanguages.length > 0) {
-        opts.GOOGLE_SPEECH_ALTERNATIVE_LANGUAGE_CODES = this.altLanguages.join(',');
+      if (this.altLanguages.length > 1) opts.GOOGLE_SPEECH_ALTERNATIVE_LANGUAGE_CODES = this.altLanguages.join(',');
+      if ('unspecified' !== this.interactionType) {
+        opts.GOOGLE_SPEECH_METADATA_INTERACTION_TYPE = this.interactionType;
       }
-      if (this.profanityFilter === true) {
-        Object.assign(opts, {'GOOGLE_SPEECH_PROFANITY_FILTER': true});
+      opts.GOOGLE_SPEECH_MODEL = this.model;
+      if (this.diarization && this.diarizationMinSpeakers > 0) {
+        opts.GOOGLE_SPEECH_SPEAKER_DIARIZATION_MIN_SPEAKER_COUNT = this.diarizationMinSpeakers;
       }
+      if (this.diarization && this.diarizationMaxSpeakers > 0) {
+        opts.GOOGLE_SPEECH_SPEAKER_DIARIZATION_MAX_SPEAKER_COUNT = this.diarizationMaxSpeakers;
+      }
+      if (this.naicsCode > 0) opts.GOOGLE_SPEECH_METADATA_INDUSTRY_NAICS_CODE = this.naicsCode;
       ep.addCustomEventListener(GoogleTranscriptionEvents.Transcription, this._onTranscription.bind(this, cs, ep));
       ep.addCustomEventListener(GoogleTranscriptionEvents.EndOfUtterance, this._onEndOfUtterance.bind(this, cs, ep));
       ep.addCustomEventListener(GoogleTranscriptionEvents.VadDetected, this._onVadDetected.bind(this, cs, ep));
@@ -280,6 +301,9 @@ class TaskGather extends Task {
       if (this.hints && this.hints.length > 1) {
         opts.AZURE_SPEECH_HINTS = this.hints.map((h) => h.trim()).join(',');
       }
+      if (this.altLanguages && this.altLanguages.length > 0) {
+        opts.AZURE_SPEECH_ALTERNATIVE_LANGUAGE_CODES = this.altLanguages.join(',');
+      }
       if (this.requestSnr) opts.AZURE_REQUEST_SNR = 1;
       if (this.profanityOption && this.profanityOption !== 'raw') opts.AZURE_PROFANITY_OPTION = this.profanityOption;
       if (this.azureServiceEndpoint) opts.AZURE_SERVICE_ENDPOINT = this.azureServiceEndpoint;
@@ -325,7 +349,9 @@ class TaskGather extends Task {
     }
     assert(!this._timeoutTimer);
     this.logger.debug(`Gather:_startTimer: timeout ${this.timeout}`);
-    this._timeoutTimer = setTimeout(() => this._resolve('timeout'), this.timeout);
+    this._timeoutTimer = setTimeout(() => {
+      this._resolve(this.digitBuffer.length >= this.minDigits ? 'dtmf-num-digits' : 'timeout');
+    }, this.timeout);
   }
 
   _clearTimer() {
@@ -365,8 +391,10 @@ class TaskGather extends Task {
         // don't sort based on confidence: https://github.com/Azure-Samples/cognitive-services-speech-sdk/issues/1463
         //const nbest = evt.NBest.sort((a, b) => b.Confidence - a.Confidence);
         const nbest = evt.NBest;
+        const language_code = evt.PrimaryLanguage?.Language || this.language;
         evt = {
           is_final: true,
+          language_code,
           alternatives: [
             {
               confidence: nbest[0].Confidence,
@@ -404,6 +432,7 @@ class TaskGather extends Task {
         evt.alternatives[0].transcript.split(' ').length >= this.minBargeinWordCount) {
         if (!this.playComplete) {
           this.logger.debug({transcript: evt.alternatives[0].transcript}, 'killing audio due to speech');
+          this.emit('vad');
         }
         this._killAudio(cs);
       }

lib/tasks/rest_dial.js

@@ -50,7 +50,7 @@ class TaskRestDial extends Task {
     try {
       const b3 = this.getTracingPropagation();
       const httpHeaders = b3 && {b3};
-      const tasks = await cs.requestor.request('verb:hook', this.call_hook, cs.callInfo, httpHeaders);
+      const tasks = await cs.requestor.request('session:new', this.call_hook, cs.callInfo, httpHeaders);
       if (tasks && Array.isArray(tasks)) {
         this.logger.debug({tasks: tasks}, `TaskRestDial: replacing application with ${tasks.length} tasks`);
         cs.replaceApplication(normalizeJambones(this.logger, tasks).map((tdata) => makeTask(this.logger, tdata)));

lib/tasks/specs.json

@@ -448,6 +448,7 @@
           "tag"
         ]
       },
+      "model": "string",
       "outputFormat": {
         "type": "string",
         "enum": [

lib/tasks/transcribe.js

@@ -32,7 +32,9 @@ class TaskTranscribe extends Task {
     this.profanityFilter = recognizer.profanityFilter;
     this.punctuation = !!recognizer.punctuation;
     this.enhancedModel = !!recognizer.enhancedModel;
+    this.model = recognizer.model || 'phone_call';
     this.words = !!recognizer.words;
+    this.singleUtterance = recognizer.singleUtterance || false;
     this.diarization = !!recognizer.diarization;
     this.diarizationMinSpeakers = recognizer.diarizationMinSpeakers || 0;
     this.diarizationMaxSpeakers = recognizer.diarizationMaxSpeakers || 0;
@@ -56,11 +58,12 @@ class TaskTranscribe extends Task {
 
   get name() { return TaskName.Transcribe; }
 
-  async exec(cs, ep, parentTask) {
+  async exec(cs, ep, ep2) {
     super.exec(cs);
     const {updateSpeechCredentialLastUsed} = require('../utils/db-utils')(this.logger, cs.srf);
 
     this.ep = ep;
+    this.ep2 = ep2;
     if ('default' === this.vendor || !this.vendor) this.vendor = cs.speechRecognizerVendor;
     if ('default' === this.language || !this.language) this.language = cs.speechRecognizerLanguage;
     this.sttCredentials = cs.getSpeechCredentials(this.vendor, 'stt');
@@ -76,7 +79,9 @@ class TaskTranscribe extends Task {
         }).catch((err) => this.logger.info({err}, 'Error generating alert for no stt'));
         throw new Error('no provisioned speech credentials for TTS');
       }
-      await this._startTranscribing(cs, ep);
+      await this._startTranscribing(cs, ep, 1);
+      if (this.separateRecognitionPerChannel && ep2) await this._startTranscribing(cs, ep2, 2);
+
       updateSpeechCredentialLastUsed(this.sttCredentials.speech_credential_sid)
         .catch(() => {/*already logged error */});
 
@@ -104,11 +109,15 @@ class TaskTranscribe extends Task {
       // hangup after 1 sec if we don't get a final transcription
       this._timer = setTimeout(() => this.notifyTaskDone(), 1000);
     }
+    if (this.separateRecognitionPerChannel && this.ep2 && this.ep2.connected) {
+      this.ep2.stopTranscription({vendor: this.vendor})
+        .catch((err) => this.logger.info(err, 'Error TaskTranscribe:kill'));
+    }
     else this.notifyTaskDone();
     await this.awaitTaskDone();
   }
 
-  async _startTranscribing(cs, ep) {
+  async _startTranscribing(cs, ep, channel) {
     const opts = {};
 
     if (this.vad.enable) {
@@ -117,25 +126,28 @@ class TaskTranscribe extends Task {
       if (this.vad.mode >= 0 && this.vad.mode <= 3) opts.RECOGNIZER_VAD_MODE = this.vad.mode;
     }
 
-    ep.addCustomEventListener(GoogleTranscriptionEvents.Transcription, this._onTranscription.bind(this, cs, ep));
-    ep.addCustomEventListener(GoogleTranscriptionEvents.NoAudioDetected, this._onNoAudio.bind(this, cs, ep));
+    ep.addCustomEventListener(GoogleTranscriptionEvents.Transcription,
+      this._onTranscription.bind(this, cs, ep, channel));
+    ep.addCustomEventListener(GoogleTranscriptionEvents.NoAudioDetected, this._onNoAudio.bind(this, cs, ep, channel));
     ep.addCustomEventListener(GoogleTranscriptionEvents.MaxDurationExceeded,
-      this._onMaxDurationExceeded.bind(this, cs, ep));
-    ep.addCustomEventListener(AwsTranscriptionEvents.Transcription, this._onTranscription.bind(this, cs, ep));
-    ep.addCustomEventListener(AwsTranscriptionEvents.NoAudioDetected, this._onNoAudio.bind(this, cs, ep));
+      this._onMaxDurationExceeded.bind(this, cs, ep, channel));
+    ep.addCustomEventListener(AwsTranscriptionEvents.Transcription, this._onTranscription.bind(this, cs, ep, channel));
+    ep.addCustomEventListener(AwsTranscriptionEvents.NoAudioDetected, this._onNoAudio.bind(this, cs, ep, channel));
     ep.addCustomEventListener(AwsTranscriptionEvents.MaxDurationExceeded,
-      this._onMaxDurationExceeded.bind(this, cs, ep));
-    ep.addCustomEventListener(AzureTranscriptionEvents.Transcription, this._onTranscription.bind(this, cs, ep));
-    ep.addCustomEventListener(AzureTranscriptionEvents.NoSpeechDetected, this._onNoAudio.bind(this, cs, ep));
+      this._onMaxDurationExceeded.bind(this, cs, ep, channel));
+    ep.addCustomEventListener(AzureTranscriptionEvents.Transcription,
+      this._onTranscription.bind(this, cs, ep, channel));
+    ep.addCustomEventListener(AzureTranscriptionEvents.NoSpeechDetected, this._onNoAudio.bind(this, cs, ep, channel));
 
     if (this.vendor === 'google') {
       if (this.sttCredentials) opts.GOOGLE_APPLICATION_CREDENTIALS = JSON.stringify(this.sttCredentials.credentials);
       [
         ['enhancedModel', 'GOOGLE_SPEECH_USE_ENHANCED'],
-        ['separateRecognitionPerChannel', 'GOOGLE_SPEECH_SEPARATE_RECOGNITION_PER_CHANNEL'],
+        //['separateRecognitionPerChannel', 'GOOGLE_SPEECH_SEPARATE_RECOGNITION_PER_CHANNEL'],
         ['profanityFilter', 'GOOGLE_SPEECH_PROFANITY_FILTER'],
         ['punctuation', 'GOOGLE_SPEECH_ENABLE_AUTOMATIC_PUNCTUATION'],
         ['words', 'GOOGLE_SPEECH_ENABLE_WORD_TIME_OFFSETS'],
+        ['singleUtterance', 'GOOGLE_SPEECH_SINGLE_UTTERANCE'],
         ['diarization', 'GOOGLE_SPEECH_PROFANITY_FILTER']
       ].forEach((arr) => {
         if (this[arr[0]]) opts[arr[1]] = true;
@@ -149,15 +161,8 @@ class TaskTranscribe extends Task {
       if (this.altLanguages.length > 1) opts.GOOGLE_SPEECH_ALTERNATIVE_LANGUAGE_CODES = this.altLanguages.join(',');
       if ('unspecified' !== this.interactionType) {
         opts.GOOGLE_SPEECH_METADATA_INTERACTION_TYPE = this.interactionType;
-
-        // additionally set model if appropriate
-        if ('phone_call' === this.interactionType) opts.GOOGLE_SPEECH_MODEL = 'phone_call';
-        else if (['voice_search', 'voice_command'].includes(this.interactionType)) {
-          opts.GOOGLE_SPEECH_MODEL = 'command_and_search';
-        }
-        else opts.GOOGLE_SPEECH_MODEL = 'phone_call';
       }
-      else opts.GOOGLE_SPEECH_MODEL = 'phone_call';
+      opts.GOOGLE_SPEECH_MODEL = this.model;
       if (this.diarization && this.diarizationMinSpeakers > 0) {
         opts.GOOGLE_SPEECH_SPEAKER_DIARIZATION_MIN_SPEAKER_COUNT = this.diarizationMinSpeakers;
       }
@@ -208,6 +213,7 @@ class TaskTranscribe extends Task {
       if (this.hints && this.hints.length > 1) {
         opts.AZURE_SPEECH_HINTS = this.hints.map((h) => h.trim()).join(',');
       }
+      if (this.altLanguages.length > 1) opts.AZURE_SPEECH_ALTERNATIVE_LANGUAGE_CODES = this.altLanguages.join(',');
       if (this.requestSnr) opts.AZURE_REQUEST_SNR = 1;
       if (this.profanityOption !== 'raw') opts.AZURE_PROFANITY_OPTION = this.profanityOption;
       if (this.initialSpeechTimeoutMs > 0) opts.AZURE_INITIAL_SPEECH_TIMEOUT_MS = this.initialSpeechTimeoutMs;
@@ -225,15 +231,16 @@ class TaskTranscribe extends Task {
       vendor: this.vendor,
       interim: this.interim ? true : false,
       locale: this.language,
-      channels: this.separateRecognitionPerChannel ? 2 : 1
+      channels: /*this.separateRecognitionPerChannel ? 2 : */ 1
     });
   }
 
-  _onTranscription(cs, ep, evt) {
-    this.logger.debug(evt, 'TaskTranscribe:_onTranscription');
+  _onTranscription(cs, ep, channel, evt) {
+    this.logger.debug({evt, channel}, 'TaskTranscribe:_onTranscription');
     if ('aws' === this.vendor && Array.isArray(evt) && evt.length > 0) evt = evt[0];
     if ('microsoft' === this.vendor) {
       const nbest = evt.NBest;
+      const language_code = evt.PrimaryLanguage?.Language || this.language;
       const alternatives = nbest ? nbest.map((n) => {
         return {
           confidence: n.Confidence,
@@ -248,6 +255,8 @@ class TaskTranscribe extends Task {
 
       const newEvent = {
         is_final: evt.RecognitionStatus === 'Success',
+        channel,
+        language_code,
         alternatives
       };
       evt = newEvent;
@@ -258,6 +267,8 @@ class TaskTranscribe extends Task {
       return this._transcribe(ep);
     }
 
+    evt.channel_tag = channel;
+
     if (this.transcriptionHook) {
       const b3 = this.getTracingPropagation();
       const httpHeaders = b3 && {b3};
@@ -275,13 +286,13 @@ class TaskTranscribe extends Task {
     }
   }
 
-  _onNoAudio(cs, ep) {
-    this.logger.debug('TaskTranscribe:_onNoAudio restarting transcription');
+  _onNoAudio(cs, ep, channel) {
+    this.logger.debug(`TaskTranscribe:_onNoAudio restarting transcription on channel ${channel}`);
     this._transcribe(ep);
   }
 
-  _onMaxDurationExceeded(cs, ep) {
-    this.logger.debug('TaskTranscribe:_onMaxDurationExceeded restarting transcription');
+  _onMaxDurationExceeded(cs, ep, channel) {
+    this.logger.debug(`TaskTranscribe:_onMaxDurationExceeded restarting transcription on channel ${channel}`);
     this._transcribe(ep);
   }
 

lib/utils/ws-requestor.js

@@ -14,8 +14,11 @@ class WsRequestor extends BaseRequestor {
     this.connections = 0;
     this.messagesInFlight = new Map();
     this.maliciousClient = false;
-    this.closedByUs = false;
+    this.closedGracefully = false;
     this.backoffMs = 500;
+    this.connectInProgress = false;
+    this.queuedMsg = [];
+    this.id = short.generate();
 
     assert(this._isAbsoluteUrl(this.url));
 
@@ -41,6 +44,10 @@ class WsRequestor extends BaseRequestor {
       this.logger.info({url: this.url}, 'WsRequestor:request - discarding msg to malicious client');
       return;
     }
+    if (this.closedGracefully) {
+      this.logger.debug(`WsRequestor:request - discarding ${type} because we closed the socket`);
+      return;
+    }
 
     if (type === 'session:new') this.call_sid = params.callSid;
 
@@ -53,6 +60,14 @@ class WsRequestor extends BaseRequestor {
 
     /* connect if necessary */
     if (!this.ws) {
+      if (this.connectInProgress) {
+        this.logger.debug(
+          `WsRequestor:request(${this.id}) - queueing ${type} message since we are connecting`);
+        this.queuedMsg.push({type, hook, params, httpHeaders});
+        return;
+      }
+      this.connectInProgress = true;
+      this.logger.debug(`WsRequestor:request(${this.id}) - connecting since we do not have a connection`);
       if (this.connections >= MAX_RECONNECTS) {
         throw new Error(`max attempts connecting to ${this.url}`);
       }
@@ -63,13 +78,16 @@ class WsRequestor extends BaseRequestor {
         this.stats.histogram('app.hook.connect_time', rtt, ['hook_type:app']);
       } catch (err) {
         this.logger.info({url, err}, 'WsRequestor:request - failed connecting');
+        this.connectInProgress = false;
         throw err;
       }
     }
     assert(this.ws);
 
     /* prepare and send message */
-    const payload = params ? snakeCaseKeys(params, ['customerData', 'sip']) : null;
+    let payload = params ? snakeCaseKeys(params, ['customerData', 'sip']) : null;
+    if (type === 'session:new') this._sessionData = payload;
+    if (type === 'session:reconnect') payload = this._sessionData;
     assert.ok(url, 'WsRequestor:request url was not provided');
 
     const msgid = short.generate();
@@ -83,12 +101,23 @@ class WsRequestor extends BaseRequestor {
       ...b3
     };
 
+    const sendQueuedMsgs = () => {
+      if (this.queuedMsg.length > 0) {
+        for (const {type, hook, params, httpHeaders} of this.queuedMsg) {
+          this.logger.debug(`WsRequestor:request - preparing queued ${type} for sending`);
+          setImmediate(this.request.bind(this, type, hook, params, httpHeaders));
+        }
+        this.queuedMsg.length = 0;
+      }
+    };
+
     //this.logger.debug({obj}, `websocket: sending (${url})`);
 
     /* simple notifications */
-    if (['call:status', 'jambonz:error'].includes(type)) {
+    if (['call:status', 'jambonz:error', 'session:reconnect'].includes(type)) {
       this.ws.send(JSON.stringify(obj), () => {
         this.logger.debug({obj}, `WsRequestor:request websocket: sent (${url})`);
+        sendQueuedMsgs();
       });
       return;
     }
@@ -122,15 +151,16 @@ class WsRequestor extends BaseRequestor {
       /* send the message */
       this.ws.send(JSON.stringify(obj), () => {
         this.logger.debug({obj}, `WsRequestor:request websocket: sent (${url})`);
+        sendQueuedMsgs();
       });
     });
   }
 
   close() {
-    this.closedByUs = true;
+    this.closedGracefully = true;
+    this.logger.info('WsRequestor:close closing socket');
     try {
       if (this.ws) {
-        this.logger.info('WsRequestor:close closing socket');
         this.ws.close();
         this.ws.removeAllListeners();
       }
@@ -163,9 +193,8 @@ class WsRequestor extends BaseRequestor {
 
       this
         .once('ready', (ws) => {
-          this.ws = ws;
           this.removeAllListeners('not-ready');
-          if (this.connections > 0) this.request('session:reconnect', this.url);
+          if (this.connections > 1) this.request('session:reconnect', this.url);
           resolve();
         })
         .once('not-ready', (err) => {
@@ -178,6 +207,7 @@ class WsRequestor extends BaseRequestor {
   }
 
   _setHandlers(ws) {
+    this.logger.debug('WsRequestor:_setHandlers');
     ws
       .once('open', this._onOpen.bind(this, ws))
       .once('close', this._onClose.bind(this))
@@ -194,18 +224,23 @@ class WsRequestor extends BaseRequestor {
   }
 
   _onOpen(ws) {
+    this.logger.info({url: this.url}, `WsRequestor(${this.id}) - successfully connected`);
     if (this.ws) this.logger.info({old_ws: this.ws._socket.address()}, 'WsRequestor:_onOpen');
     assert(!this.ws);
+    this.ws = ws;
+    this.connectInProgress = false;
+    this.connections++;
     this.emit('ready', ws);
-    this.logger.info({url: this.url}, 'WsRequestor - successfully connected');
   }
 
-  _onClose() {
-    if (this.connections > 0) {
+  _onClose(code) {
+    this.logger.info(`WsRequestor(${this.id}) - closed from far end ${code}`);
+    if (this.connections > 0 && code !== 1000) {
       this.logger.info({url: this.url}, 'WsRequestor - socket closed unexpectedly from remote side');
       this.emit('socket-closed');
     }
-    this.ws && this.ws.removeAllListeners();
+    else if (code === 1000) this.closedGracefully = true;
+    this.ws?.removeAllListeners();
     this.ws = null;
   }
 
@@ -223,8 +258,17 @@ class WsRequestor extends BaseRequestor {
   _onSocketClosed() {
     this.ws = null;
     this.emit('connection-dropped');
-    if (this.connections++ > 0 && this.connections < MAX_RECONNECTS && !this.closedByUs) {
-      setTimeout(this._connect.bind(this), this.backoffMs);
+    if (this.connections > 0 && this.connections < MAX_RECONNECTS && !this.closedGracefully) {
+      this.logger.debug(`WsRequestor:_onSocketClosed waiting ${this.backoffMs} to reconnect`);
+      setTimeout(() => {
+        this.logger.debug(
+          {haveWs: !!this.ws, connectInProgress: this.connectInProgress},
+          'WsRequestor:_onSocketClosed time to reconnect');
+        if (!this.ws && !this.connectInProgress) {
+          this.connectInProgress = true;
+          this._connect().catch((err) => this.connectInProgress = false);
+        }
+      }, this.backoffMs);
       this.backoffMs = this.backoffMs < 2000 ? this.backoffMs * 2 : (this.backoffMs + 2000);
     }
   }

package.json

@@ -26,52 +26,46 @@
     "jslint": "eslint app.js lib"
   },
   "dependencies": {
-    "@cognigy/socket-client": "^4.5.5",
-    "@jambonz/db-helpers": "^0.6.16",
+    "@jambonz/db-helpers": "^0.6.18",
     "@jambonz/http-health-check": "^0.0.1",
-    "@jambonz/mw-registrar": "^0.2.1",
-    "@jambonz/realtimedb-helpers": "^0.4.26",
+    "@jambonz/realtimedb-helpers": "^0.4.29",
     "@jambonz/stats-collector": "^0.1.6",
-    "@jambonz/time-series": "^0.1.6",
+    "@jambonz/time-series": "^0.1.9",
     "@opentelemetry/api": "^1.1.0",
-    "@opentelemetry/exporter-jaeger": "^1.1.0",
+    "@opentelemetry/exporter-jaeger": "^1.3.1",
     "@opentelemetry/exporter-trace-otlp-http": "^0.27.0",
-    "@opentelemetry/exporter-zipkin": "^1.1.0",
+    "@opentelemetry/exporter-zipkin": "^1.3.1",
     "@opentelemetry/instrumentation": "^0.27.0",
-    "@opentelemetry/instrumentation-express": "^0.28.0",
-    "@opentelemetry/instrumentation-http": "^0.27.0",
-    "@opentelemetry/instrumentation-pino": "^0.28.1",
-    "@opentelemetry/resources": "^1.1.0",
-    "@opentelemetry/sdk-trace-base": "^1.1.0",
-    "@opentelemetry/sdk-trace-node": "^1.1.0",
-    "@opentelemetry/semantic-conventions": "^1.1.0",
-    "aws-sdk": "^2.1073.0",
+    "@opentelemetry/resources": "^1.3.1",
+    "@opentelemetry/sdk-trace-base": "^1.3.1",
+    "@opentelemetry/sdk-trace-node": "^1.3.1",
+    "@opentelemetry/semantic-conventions": "^1.3.1",
+    "aws-sdk": "^2.1152.0",
     "bent": "^7.3.12",
-    "cidr-matcher": "^2.1.1",
-    "debug": "^4.3.2",
+    "debug": "^4.3.4",
     "deepcopy": "^2.1.0",
-    "drachtio-fsmrf": "^2.0.13",
-    "drachtio-srf": "^4.4.61",
-    "express": "^4.17.1",
-    "helmet": "^5.0.2",
-    "ip": "^1.1.5",
-    "moment": "^2.29.2",
-    "parse-url": "^5.0.7",
-    "pino": "^6.13.4",
+    "drachtio-fsmrf": "^3.0.0",
+    "drachtio-srf": "^4.5.0",
+    "express": "^4.18.1",
+    "helmet": "^5.1.0",
+    "ip": "^1.1.8",
+    "moment": "^2.29.3",
+    "parse-url": "^5.0.8",
+    "pino": "^6.14.0",
+    "sdp-transform": "^2.14.1",
     "short-uuid": "^4.2.0",
     "to-snake-case": "^1.0.0",
     "uuid": "^8.3.2",
     "verify-aws-sns-signature": "^0.0.6",
-    "ws": "^8.5.0",
+    "ws": "^8.8.0",
     "xml2js": "^0.4.23"
   },
   "devDependencies": {
-    "async": "^3.2.0",
-    "clear-module": "^4.1.1",
-    "eslint": "^7.20.0",
+    "clear-module": "^4.1.2",
+    "eslint": "^7.32.0",
     "eslint-plugin-promise": "^4.3.1",
     "nyc": "^15.1.0",
-    "tape": "^5.2.2"
+    "tape": "^5.5.3"
   },
   "optionalDependencies": {
     "bufferutil": "^4.0.6",

test/docker_start.js

@@ -1,6 +1,5 @@
 const test = require('tape') ;
 const exec = require('child_process').exec ;
-const async = require('async');
 
 test('starting docker network..takes a bit for mysql and freeswitch to come up..patience..', (t) => {
   exec(`docker-compose -f ${__dirname}/docker-compose-testbed.yaml up -d`, (err, stdout, stderr) => {

test/sip-tests.js

@@ -1,7 +1,5 @@
-const test = require('blue-tape');
-const { output, sippUac } = require('./sipp')('test_sbc-inbound');
-const debug = require('debug')('drachtio:sbc-inbound');
-const clearModule = require('clear-module');
+const test = require('tape');
+const { sippUac } = require('./sipp')('test_sbc-inbound');
 
 process.on('unhandledRejection', (reason, p) => {
   console.log('Unhandled Rejection at: Promise', p, 'reason:', reason);