Compare commits

..

54 Commits

Author SHA1 Message Date
snyk-bot
0df2b5eef5 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
2022-07-27 16:43:40 +00:00
Dave Horton
a035b67e6c bugfix: hold music fetched when conference member removed from hold 2022-07-27 11:37:12 +01:00
Dave Horton
6979affb86 Feature/fast http client (#132)
* initial changes to use undici for http client and connection pooling

* use body.json() mixin

* logging

* add pipelining env var

* implement socket close
2022-07-18 15:32:03 +02:00
Dave Horton
bb9c3a8df0 createCall: return callId along with sid 2022-07-12 09:56:13 +02:00
Paulo Telles
92fa3c249c improve dockerfile to fix snyk security issues (#126)
Co-authored-by: p.souza <p.souza@cognigy.com>
2022-07-07 15:20:26 +02:00
Dave Horton
7f808c6107 listen: when passDtmf is true, send dtmf events down websocket connection as json test frames (#129) 2022-07-07 11:55:29 +02:00
Dave Horton
f95524863d update parse-url, improve Dockerfile 2022-07-06 19:16:31 +02:00
Dave Horton
aceaa5b7da bugfix: continuous asr - if ended by dtmf allow collection of final transcript 2022-06-28 10:10:31 -04:00
Dave Horton
7d57c85153 bugfix #121: Dial verb not ending when call no answer timeout exceeded 2022-06-24 10:50:29 -04:00
Dave Horton
9aa0df256d initial changes to support siprec recording (#120)
* initial changes to support siprec recording

* include additional params on SIP INFO to start recording

* add support for maniupulating recording via REST API

* fixes from testing pause/resume recording
2022-06-23 16:21:35 -04:00
Dave Horton
627c38899f Feature/continuous asr (#119)
* bugfix: background gather for speech-only should still kill audio on dtmf entry when dtmfBargein is true

* initial changes for continuous asr

* move properties under recognizer

* update drachtio-srf@4.5.1

* catch exception on destroy
2022-06-21 10:35:27 -04:00
Dave Horton
bdb40b3aa0 update to drachtio-fsmrf@3.0.1 2022-06-18 15:55:23 -04:00
Dave Horton
12ad7e556f added support for sip:request verb, used to send SIP INFO/NOTIFY etc during call (#116) 2022-06-15 13:31:32 -04:00
Dave Horton
05d6c8d467 linting 2022-06-14 08:24:44 -04:00
akirilyuk
5e9407ff4e add defaults to rest call payload (#115)
Co-authored-by: akirilyuk <a.kirilyuk@cognigy.com>
2022-06-14 08:20:55 -04:00
Dave Horton
e4fefe8f44 update to azure 1.22.0 2022-06-11 16:16:53 -04:00
Dave Horton
f7aac33af4 update deps 2022-06-11 11:23:17 -04:00
Dave Horton
dc1d8de396 updates to drachtio-srf@4.5.0 and drachtio-fsmrf@3.0.0 2022-06-11 11:06:03 -04:00
Dave Horton
5be5b6d05d bugfix: broken enqueue waitHook (#113) 2022-06-11 10:38:35 -04:00
Dave Horton
f51211b407 minor docs 2022-05-17 12:58:26 -04:00
Prashanth
7f0e373e5f issue# 107: on gather timeout, if minDigits are collected, resolve wi… (#111)
* 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>
2022-05-17 12:53:10 -04:00
Dave Horton
c3e5ffa52d bugfix: transcribe of a dialed call can now occur on both legs 2022-05-15 13:45:55 -04:00
Dave Horton
0ee13fb794 minor docs 2022-05-13 21:34:35 -04:00
Dave Horton
4e84098036 Docs folder (#108)
* add how-to for developers

* fix links

* minor docs cleanup
2022-05-13 21:30:04 -04:00
Dave Horton
6d34850dc6 bugfix: transcribe Azure interim transcripts were missing 2022-05-11 19:22:14 -04:00
Dave Horton
76ff1835a6 background gather listen only once for vad and other interrupt events 2022-05-11 09:21:54 -04:00
Dave Horton
a4e358596e emit vad event on partial transcript 2022-05-10 15:14:10 -04:00
Dave Horton
c412554c6b WsRequestor: reconnect if socket dropped from far end 2022-05-09 12:14:13 -04:00
Dave Horton
34fe22f6e1 minor 2022-05-08 16:34:42 -04:00
Dave Horton
182ad8c716 expose model and singleUtterance to gather/transcribe when using google 2022-05-08 12:29:55 -04:00
Dave Horton
036accab44 dial: transcribe and listen should be based on the caller (A leg) endpoint 2022-05-07 18:36:49 -04:00
Dave Horton
b37881a059 bugfix: second part of outbound dial fix over wss 2022-05-07 11:52:29 -04:00
Dave Horton
258e4b5434 bugfix: outbound rest dial over websocket api needs to send session:new 2022-05-07 11:51:21 -04:00
Dave Horton
aa4d72c80a allow call status to be sent before killing rest dial on failure 2022-05-02 14:05:24 -04:00
Dave Horton
5c38ace5ba bugfix: rest dial should exit upon call failure, not after call timeout is reached 2022-05-02 13:50:42 -04:00
Dave Horton
dea58c2605 more work on wss race condition 2022-05-02 13:32:07 -04:00
Dave Horton
eb0f55e0e3 ws-requestor: queue outgoing messages if we are in the process of connecting to the remote wss server 2022-05-02 13:09:23 -04:00
Dave Horton
944b8a29ca Use lts version of node instead of latest 2022-05-02 11:17:29 -04:00
Dave Horton
daa02ac55a logging 2022-05-02 11:12:39 -04:00
Dave Horton
5134d5dbc6 update to latest realtimedb-helpers 2022-05-02 10:55:42 -04:00
Dave Horton
a755e25568 minor logging 2022-05-02 10:21:17 -04:00
Dave Horton
13549286db bugfix: createCall needs to work with wss url 2022-05-02 09:42:04 -04:00
Dave Horton
72aaf80335 add support for multiple languages when using Azure STT 2022-04-26 15:07:55 -04:00
Dave Horton
af33089a8a fix deprecated dep 2022-04-24 14:05:44 -04:00
Dave Horton
85d86cfdc3 bugfix: gather catch errors when webhook fails 2022-04-24 13:45:29 -04:00
Dave Horton
de9f2ce5ca bugfix: handle error if we cannot get our own ipv4 2022-04-21 19:09:23 -04:00
Dave Horton
36c97e9562 simplify error message 2022-04-21 14:43:09 -04:00
Dave Horton
13ea559cb1 send error notification over websocket if tts fails 2022-04-21 14:33:49 -04:00
Dave Horton
698d12a95f clean up error handling in say verb 2022-04-21 10:27:33 -04:00
Dave Horton
359cb82d80 per recommendation from microsoft, do NOT sort transcripts by confidence: first transcript in the returned list is 'best' 2022-04-17 17:53:16 -04:00
Dave Horton
29dec24095 bugfix: azure stt - if we get no speech detected, listen again 2022-04-13 12:07:30 -04:00
Dave Horton
6330b0d443 Dockerfile update 2022-04-12 16:12:29 -04:00
Dave Horton
24a0bc547f gather: dont restart transcribing if task has been killed 2022-04-11 21:13:49 -04:00
Dave Horton
db5486de27 gather bugfix: dont start transcribing after call is gone 2022-04-10 15:48:35 -04:00
33 changed files with 3595 additions and 2831 deletions

View File

@@ -1,10 +1,23 @@
FROM node:17.7.1-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/
COPY package.json ./
RUN npm install
RUN npm prune
COPY . /opt/app
FROM base as build
COPY package.json package-lock.json ./
RUN npm ci
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" ]

View File

@@ -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).

123
docs/contributing.md Normal file
View File

@@ -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.

View File

@@ -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 */
@@ -187,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;
@@ -230,6 +241,7 @@ router.post('/', async(req, res) => {
else console.error(err);
}
ep.destroy();
setTimeout(restDial.kill.bind(restDial), 5000);
}
} catch (err) {
sysError(logger, res, err);

View File

@@ -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,15 +247,162 @@ 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]);
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}`);
@@ -331,7 +491,7 @@ class CallSession extends Emitter {
speech_credential_sid: credential.speech_credential_sid,
accessKeyId: credential.access_key_id,
secretAccessKey: credential.secret_access_key,
region: process.env.AWS_REGION || credential.aws_region
region: credential.aws_region || process.env.AWS_REGION
};
}
else if ('microsoft' === vendor) {
@@ -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;

View File

@@ -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');
}
/**
@@ -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;

View File

@@ -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) {

View File

@@ -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)});
@@ -606,8 +611,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);

View File

@@ -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;

View File

@@ -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;
@@ -36,8 +36,23 @@ 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 || [];
/* 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 || {};
@@ -55,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;
@@ -67,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;
}
@@ -99,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;
@@ -117,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(() => {
@@ -161,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));
}
@@ -199,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) {
@@ -212,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;
@@ -232,24 +274,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 +333,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;
@@ -319,13 +375,11 @@ 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._timeoutTimer = setTimeout(() => this._resolve('timeout'), this.timeout);
this._clearTimer();
this._timeoutTimer = setTimeout(() => {
if (this.isContinuousAsr) this._startAsrTimer();
else this._resolve(this.digitBuffer.length >= this.minDigits ? 'dtmf-num-digits' : 'timeout');
}, this.timeout);
}
_clearTimer() {
@@ -335,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) {
@@ -362,9 +445,13 @@ class TaskGather extends Task {
if ('microsoft' === this.vendor) {
const final = evt.RecognitionStatus === 'Success';
if (final) {
const nbest = evt.NBest.sort((a, b) => b.Confidence - a.Confidence);
// 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,
@@ -385,11 +472,23 @@ class TaskGather extends Task {
}
}
if (evt.is_final) {
if (evt.alternatives[0].transcript === '') {
if (evt.alternatives[0].transcript === '' && !this.callSession.callGone && !this.killed) {
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:
@@ -402,6 +501,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);
}
@@ -433,7 +533,10 @@ class TaskGather extends Task {
}
_onNoSpeechDetected(cs, ep) {
this._resolve('timeout');
if (!this.callSession.callGone && !this.killed) {
this.logger.debug('TaskGather:_onNoSpeechDetected - listen again');
return this._startTranscribing(ep);
}
}
async _resolve(reason, evt) {
@@ -442,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) {
@@ -449,35 +561,35 @@ 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();
return;
}
if (reason.startsWith('dtmf')) {
if (this.parentTask) this.parentTask.emit('dtmf', evt);
else {
this.emit('dtmf', evt);
await this.performAction({digits: this.digitBuffer, reason: 'dtmfDetected'});
try {
if (reason.startsWith('dtmf')) {
if (this.parentTask) this.parentTask.emit('dtmf', evt);
else {
this.emit('dtmf', evt);
await this.performAction({digits: this.digitBuffer, reason: 'dtmfDetected'});
}
}
}
else if (reason.startsWith('speech')) {
if (this.parentTask) this.parentTask.emit('transcription', evt);
else {
this.emit('transcription', evt);
await this.performAction({speech: evt, reason: 'speechDetected'});
else if (reason.startsWith('speech')) {
if (this.parentTask) this.parentTask.emit('transcription', evt);
else {
this.emit('transcription', evt);
await this.performAction({speech: evt, reason: 'speechDetected'});
}
}
}
else if (reason.startsWith('timeout')) {
if (this.parentTask) this.parentTask.emit('timeout', evt);
else {
this.emit('timeout', evt);
await this.performAction({reason: 'timeout'});
else if (reason.startsWith('timeout')) {
if (this.parentTask) this.parentTask.emit('timeout', evt);
else {
this.emit('timeout', evt);
await this.performAction({reason: 'timeout'});
}
}
}
} catch (err) { /*already logged error*/ }
this.notifyTaskDone();
}
}

View File

@@ -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;

View File

@@ -17,6 +17,9 @@ function makeTask(logger, obj, parent) {
case TaskName.SipDecline:
const TaskSipDecline = require('./sip_decline');
return new TaskSipDecline(logger, data, parent);
case TaskName.SipRequest:
const TaskSipRequest = require('./sip_request');
return new TaskSipRequest(logger, data, parent);
case TaskName.SipRefer:
const TaskSipRefer = require('./sip_refer');
return new TaskSipRefer(logger, data, parent);

View File

@@ -50,7 +50,21 @@ 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 params = {
...cs.callInfo,
defaults: {
synthesizer: {
vendor: cs.speechSynthesisVendor,
language: cs.speechSynthesisLanguage,
voice: cs.speechSynthesisVoice
},
recognizer: {
vendor: cs.speechRecognizerVendor,
language: cs.speechRecognizerLanguage
}
}
};
const tasks = await cs.requestor.request('session:new', this.call_hook, params, 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)));

View File

@@ -51,55 +51,64 @@ 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}`);
throw new Error('no provisioned speech credentials for TTS');
}
// synthesize all of the text elements
let lastUpdated = false;
/* otel: trace time for tts */
const {span} = this.startChildSpan('tts-generation', {
'tts.vendor': vendor,
'tts.language': language,
'tts.voice': voice
});
this.ttsSpan = span;
const filepath = (await Promise.all(this.text.map(async(text) => {
/* produce an audio segment from the provided text */
const generateAudio = async(text) => {
if (this.killed) return;
if (text.startsWith('silence_stream://')) return text;
const {filePath, servedFromCache} = await synthAudio(stats, {
text,
vendor,
language,
voice,
engine,
salt,
credentials
}).catch((err) => {
this.logger.info(err, 'Error synthesizing tts');
/* otel: trace time for tts */
const {span} = this.startChildSpan('tts-generation', {
'tts.vendor': vendor,
'tts.language': language,
'tts.voice': voice
});
try {
const {filePath, servedFromCache} = await synthAudio(stats, {
text,
vendor,
language,
voice,
engine,
salt,
credentials
});
this.logger.debug(`file ${filePath}, served from cache ${servedFromCache}`);
if (filePath) cs.trackTmpFile(filePath);
if (!servedFromCache && !lastUpdated) {
lastUpdated = true;
updateSpeechCredentialLastUsed(credentials.speech_credential_sid)
.catch(() => {/*already logged error */});
}
span.setAttributes({'tts.cached': servedFromCache});
span.end();
return filePath;
} catch (err) {
this.logger.info({err}, 'Error synthesizing tts');
span.end();
writeAlerts({
account_sid: cs.accountSid,
alert_type: AlertType.TTS_NOT_PROVISIONED,
vendor,
detail: err.message
});
}).catch((err) => this.logger.info({err}, 'Error generating alert for tts failure'));
this.logger.debug(`file ${filePath}, served from cache ${servedFromCache}`);
if (filePath) cs.trackTmpFile(filePath);
if (!servedFromCache && !lastUpdated) {
lastUpdated = true;
updateSpeechCredentialLastUsed(credentials.speech_credential_sid)
.catch(() => {/*already logged error */});
}).catch((err) => this.logger.info({err}, 'Error generating alert for tts failure'));
this.notifyError(err.message || err);
return;
}
this.ttsSpan.setAttributes({'tts.cached': servedFromCache});
return filePath;
}))).filter((fp) => fp && fp.length);
this.ttsSpan?.end();
};
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');
while (!this.killed && (this.loop === 'forever' || this.loop--) && this.ep?.connected) {
let segment = 0;
do {
while (!this.killed && segment < filepath.length) {
if (cs.isInConference) {
const {memberId, confName, confUuid} = cs;
await this.playToConfMember(this.ep, memberId, confName, confUuid, filepath[segment]);
@@ -109,10 +118,10 @@ class TaskSay extends Task {
await ep.play(filepath[segment]);
this.logger.debug(`Say:exec completed play file ${filepath[segment]}`);
}
} while (!this.killed && ++segment < filepath.length);
segment++;
}
}
} catch (err) {
this.ttsSpan?.end();
this.logger.info(err, 'TaskSay:exec error');
}
this.emit('playDone');

49
lib/tasks/sip_request.js Normal file
View File

@@ -0,0 +1,49 @@
const Task = require('./task');
const {TaskName, TaskPreconditions} = require('../utils/constants');
/**
* Send a SIP request (e.g. INFO, NOTIFY, etc) on an existing call leg
*/
class TaskSipRequest extends Task {
constructor(logger, opts) {
super(logger, opts);
this.preconditions = TaskPreconditions.StableCall;
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; }
async exec(cs, dlg) {
super.exec(cs);
try {
this.logger.info({dlg}, `TaskSipRequest: sending a SIP ${this.method}`);
const res = await dlg.request({
method: this.method,
headers: this.headers,
body: this.body
});
const result = {result: 'success', sipStatus: res.status};
this.span.setAttributes({
...this.headers,
...(this.body && {body: this.body}),
'response.status_code': res.status
});
this.logger.debug({result}, `TaskSipRequest: received response to ${this.method}`);
await this.performAction(result);
} catch (err) {
this.logger.error({err}, 'TaskSipRequest: error');
this.span.setAttributes({
...this.headers,
...(this.body && {body: this.body}),
'response.error': err.message
});
await this.performAction({result: 'failed', err: err.message});
}
}
}
module.exports = TaskSipRequest;

View File

@@ -9,6 +9,17 @@
"status"
]
},
"sip:request": {
"properties": {
"method": "string",
"body": "string",
"headers": "object",
"actionHook": "object|string"
},
"required": [
"method"
]
},
"sip:refer": {
"properties": {
"referTo": "string",
@@ -25,7 +36,8 @@
"properties": {
"synthesizer": "#synthesizer",
"recognizer": "#recognizer",
"bargeIn": "#bargeIn"
"bargeIn": "#bargeIn",
"record": "#recordOptions"
},
"required": []
},
@@ -296,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"
@@ -448,6 +473,7 @@
"tag"
]
},
"model": "string",
"outputFormat": {
"type": "string",
"enum": [
@@ -465,7 +491,9 @@
},
"requestSnr": "boolean",
"initialSpeechTimeoutMs": "number",
"azureServiceEndpoint": "string"
"azureServiceEndpoint": "string",
"asrDtmfTerminationDigit": "string",
"asrTimeout": "number"
},
"required": [
"vendor"

View File

@@ -137,9 +137,15 @@ class Task extends Emitter {
return this.callSession.normalizeUrl(url, method, auth);
}
notifyError(errMsg) {
const params = {error: errMsg, verb: this.name};
this.cs.requestor.request('jambonz:error', '/error', params)
.catch((err) => this.logger.info({err}, 'Task:notifyError error sending error'));
}
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};

View File

@@ -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,11 +255,20 @@ class TaskTranscribe extends Task {
const newEvent = {
is_final: evt.RecognitionStatus === 'Success',
channel,
language_code,
alternatives
};
evt = newEvent;
}
if (evt.alternatives[0].transcript === '' && !cs.callGone && !this.killed) {
this.logger.info({evt}, 'TaskGather:_onTranscription - got empty transcript, listen again');
return this._transcribe(ep);
}
evt.channel_tag = channel;
if (this.transcriptionHook) {
const b3 = this.getTracingPropagation();
const httpHeaders = b3 && {b3};
@@ -270,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);
}

View File

@@ -20,6 +20,7 @@
"Redirect": "redirect",
"RestDial": "rest:dial",
"SipDecline": "sip:decline",
"SipRequest": "sip:request",
"SipRefer": "sip:refer",
"SipNotify": "sip:notify",
"SipRedirect": "sip:redirect",
@@ -119,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"

View File

@@ -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.
@@ -50,10 +71,15 @@ class HttpRequestor extends BaseRequestor {
* @param {object} [params] - request parameters
*/
async request(type, hook, params, httpHeaders = {}) {
/* jambonz:error only sent over ws */
if (type === 'jambonz:error') return;
assert(HookMsgTypes.includes(type));
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}`);
@@ -61,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},
@@ -90,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;
}
}
}

View File

@@ -1,6 +1,5 @@
const Mrf = require('drachtio-fsmrf');
const ip = require('ip');
const localIp = ip.address();
const PORT = process.env.HTTP_PORT || 3000;
const assert = require('assert');
@@ -167,6 +166,13 @@ function installSrfLocals(srf, logger) {
commitInterval: 'test' === process.env.NODE_ENV ? 7 : 20
});
let localIp;
try {
localIp = ip.address();
} catch (err) {
logger.error({err}, 'installSrfLocals - error detecting local ipv4 address');
}
srf.locals = {...srf.locals,
dbHelpers: {
client,
@@ -201,8 +207,6 @@ function installSrfLocals(srf, logger) {
getListPosition
},
parentLogger: logger,
ipv4: localIp,
serviceUrl: `http://${localIp}:${PORT}`,
getSBC,
getSmpp: () => {
return process.env.SMPP_URL;
@@ -213,6 +217,11 @@ function installSrfLocals(srf, logger) {
writeAlerts,
AlertType
};
if (localIp) {
srf.locals.ipv4 = localIp;
srf.locals.serviceUrl = `http://${localIp}:${PORT}`;
}
}
module.exports = installSrfLocals;

View File

@@ -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;
}
@@ -105,6 +134,7 @@ class WsRequestor extends BaseRequestor {
/* save the message info for reply */
const startAt = process.hrtime();
this.messagesInFlight.set(msgid, {
timer,
success: (response) => {
clearTimeout(timer);
const rtt = this._roundTrip(startAt);
@@ -121,20 +151,23 @@ 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.debug('WsRequestor:close closing socket');
try {
if (this.ws) {
this.logger.info('WsRequestor:close closing socket');
this.ws.close();
this.ws.removeAllListeners();
}
for (const [msgid, obj] of this.messagesInFlight) {
const {timer} = obj;
clearTimeout(timer);
obj.failure(`abandoning msgid ${msgid} since we have closed the socket`);
}
this.messagesInFlight.clear();
@@ -160,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) => {
@@ -175,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))
@@ -191,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;
}
@@ -220,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);
}
}

5117
package-lock.json generated

File diff suppressed because it is too large Load Diff

View File

@@ -21,57 +21,52 @@
},
"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"
},
"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-collector": "^0.25.0",
"@opentelemetry/exporter-jaeger": "^1.1.0",
"@opentelemetry/exporter-zipkin": "^1.1.0",
"@opentelemetry/exporter-jaeger": "^1.3.1",
"@opentelemetry/exporter-trace-otlp-http": "^0.27.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.1",
"drachtio-srf": "^4.5.1",
"express": "^4.18.1",
"helmet": "^5.1.0",
"ip": "^1.1.8",
"moment": "^2.29.3",
"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",
"ws": "^8.5.0",
"verify-aws-sns-signature": "^0.0.7",
"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",

View File

@@ -251,6 +251,7 @@ INSERT INTO `applications` VALUES ('24d0f6af-e976-44dd-a2e8-41c7b55abe33','say a
INSERT INTO `applications` VALUES ('17461c69-56b5-4dab-ad83-1c43a0f93a3d','gather',NULL,'bb845d4b-83a9-4cde-a6e9-50f3743bab3f','10692465-a511-4277-9807-b7157e4f81e1','293904c1-351b-4bca-8d58-1a29b853c7db',NULL,'google','en-US','en-US-Standard-C','google','en-US');
INSERT INTO `applications` VALUES ('baf9213b-5556-4c20-870c-586392ed246f','transcribe',NULL,'bb845d4b-83a9-4cde-a6e9-50f3743bab3f','ecb67a8f-f7ce-4919-abf0-bbc69c1001e5','293904c1-351b-4bca-8d58-1a29b853c7db',NULL,'google','en-US','en-US-Standard-C','google','en-US');
INSERT INTO `applications` VALUES ('ae026ab5-3029-47b4-9d7c-236e3a4b4ebe','transcribe account 2',NULL,'622f62e4-303a-49f2-bbe0-eb1e1714e37a','ecb67a8f-f7ce-4919-abf0-bbc69c1001e5','293904c1-351b-4bca-8d58-1a29b853c7db',NULL,'google','en-US','en-US-Standard-C','google','en-US');
INSERT INTO `applications` VALUES ('195d9507-6a42-46a8-825f-f009e729d023','sip info',NULL,'bb845d4b-83a9-4cde-a6e9-50f3743bab3f','c9113e7a-741f-48b9-96c1-f2f78176eeb3','293904c1-351b-4bca-8d58-1a29b853c7db',NULL,'google','en-US','en-US-Standard-C','google','en-US');
/*!40000 ALTER TABLE `applications` ENABLE KEYS */;
UNLOCK TABLES;
@@ -449,6 +450,7 @@ INSERT INTO `phone_numbers` VALUES ('e686a320-0725-418f-be65-532159bdc3ed','1617
INSERT INTO `phone_numbers` VALUES ('05eeed62-b29b-4679-bf38-d7a4e318be44','16174000003','5145b436-2f38-4029-8d4c-fd8c67831c7a','bb845d4b-83a9-4cde-a6e9-50f3743bab3f','17461c69-56b5-4dab-ad83-1c43a0f93a3d', NULL);
INSERT INTO `phone_numbers` VALUES ('f3c53863-b629-4cf6-9dcb-c7fb7072314b','16174000004','5145b436-2f38-4029-8d4c-fd8c67831c7a','bb845d4b-83a9-4cde-a6e9-50f3743bab3f','baf9213b-5556-4c20-870c-586392ed246f', NULL);
INSERT INTO `phone_numbers` VALUES ('f6416c17-829a-4f11-9c32-f0d00e4a9ae9','16174000005','5145b436-2f38-4029-8d4c-fd8c67831c7a','622f62e4-303a-49f2-bbe0-eb1e1714e37a','ae026ab5-3029-47b4-9d7c-236e3a4b4ebe', NULL);
INSERT INTO `phone_numbers` VALUES ('964d0581-9627-44cb-be20-8118050406b2','16174000006','5145b436-2f38-4029-8d4c-fd8c67831c7a','bb845d4b-83a9-4cde-a6e9-50f3743bab3f','195d9507-6a42-46a8-825f-f009e729d023', NULL);
/*!40000 ALTER TABLE `phone_numbers` ENABLE KEYS */;
UNLOCK TABLES;
@@ -736,6 +738,7 @@ INSERT INTO `webhooks` VALUES ('c71e79db-24f2-4866-a3ee-febb0f97b341','http://12
INSERT INTO `webhooks` VALUES ('54ab0976-a6c0-45d8-89a4-d90d45bf9d96','http://127.0.0.1:3101/','POST',NULL,NULL);
INSERT INTO `webhooks` VALUES ('10692465-a511-4277-9807-b7157e4f81e1','http://127.0.0.1:3102/','POST',NULL,NULL);
INSERT INTO `webhooks` VALUES ('ecb67a8f-f7ce-4919-abf0-bbc69c1001e5','http://127.0.0.1:3103/','POST',NULL,NULL);
INSERT INTO `webhooks` VALUES ('c9113e7a-741f-48b9-96c1-f2f78176eeb3','http://127.0.0.1:3104/','POST',NULL,NULL);
/*!40000 ALTER TABLE `webhooks` ENABLE KEYS */;
UNLOCK TABLES;
/*!40103 SET TIME_ZONE=@OLD_TIME_ZONE */;

View File

@@ -123,6 +123,18 @@ services:
fs:
ipv4_address: 172.38.0.63
webhook-sip-info:
image: jambonz/webhook-test-scaffold:latest
environment:
APP_PATH: /tmp/info.json
ports:
- "3104:3000/tcp"
volumes:
- ./test-apps:/tmp
networks:
fs:
ipv4_address: 172.38.0.64
influxdb:
image: influxdb:1.8
ports:

View File

@@ -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) => {

View File

@@ -5,5 +5,6 @@ require('./account-validation-tests');
require('./webhooks-tests');
require('./say-tests');
require('./gather-tests');
require('./sip-request-tests');
require('./remove-test-db');
require('./docker_stop');

View File

@@ -0,0 +1,107 @@
<?xml version="1.0" encoding="ISO-8859-1" ?>
<!DOCTYPE scenario SYSTEM "sipp.dtd">
<scenario name="Basic Sipstone UAC">
<!-- In client mode (sipp placing calls), the Call-ID MUST be -->
<!-- generated by sipp. To do so, use [call_id] keyword. -->
<send retrans="500">
<![CDATA[
INVITE sip:16174000006@[remote_ip]:[remote_port] SIP/2.0
Via: SIP/2.0/[transport] [local_ip]:[local_port];branch=[branch]
From: sipp <sip:sipp@[local_ip]:[local_port]>;tag=[pid]SIPpTag00[call_number]
To: <sip:16174000006@[remote_ip]:[remote_port]>
Call-ID: [call_id]
CSeq: 1 INVITE
Contact: sip:sipp@[local_ip]:[local_port]
Max-Forwards: 70
X-Account-Sid: bb845d4b-83a9-4cde-a6e9-50f3743bab3f
Subject: uac-gather-account-creds-success
Content-Type: application/sdp
Content-Length: [len]
v=0
o=user1 53655765 2353687637 IN IP[local_ip_type] [local_ip]
s=-
c=IN IP[media_ip_type] [media_ip]
t=0 0
m=audio [media_port] RTP/AVP 0
a=rtpmap:0 PCMU/8000
]]>
</send>
<recv response="100"
optional="true">
</recv>
<recv response="180" optional="true">
</recv>
<recv response="183" optional="true">
</recv>
<!-- By adding rrs="true" (Record Route Sets), the route sets -->
<!-- are saved and used for following messages sent. Useful to test -->
<!-- against stateful SIP proxies/B2BUAs. -->
<recv response="200" rtd="true">
</recv>
<!-- Packet lost can be simulated in any send/recv message by -->
<!-- by adding the 'lost = "10"'. Value can be [1-100] percent. -->
<send>
<![CDATA[
ACK sip:16174000006@[remote_ip]:[remote_port] SIP/2.0
Via: SIP/2.0/[transport] [local_ip]:[local_port];branch=[branch]
From: sipp <sip:sipp@[local_ip]:[local_port]>;tag=[pid]SIPpTag00[call_number]
To: 16174000006 <sip:16174000006@[remote_ip]:[remote_port]>[peer_tag_param]
Call-ID: [call_id]
CSeq: 1 ACK
Contact: sip:sipp@[local_ip]:[local_port]
Max-Forwards: 70
Subject: uac-gather-account-creds-success
Content-Length: 0
]]>
</send>
<recv request="INFO">
</recv>
<send>
<![CDATA[
SIP/2.0 200 OK
[last_Via:]
[last_From:]
[last_To:]
[last_Call-ID:]
[last_CSeq:]
Contact: <sip:[local_ip]:[local_port];transport=[transport]>
Content-Length: 0
]]>
</send>
<recv request="BYE">
</recv>
<send>
<![CDATA[
SIP/2.0 200 OK
[last_Via:]
[last_From:]
[last_To:]
[last_Call-ID:]
[last_CSeq:]
Contact: <sip:[local_ip]:[local_port];transport=[transport]>
Content-Length: 0
]]>
</send>
</scenario>

35
test/sip-request-tests.js Normal file
View File

@@ -0,0 +1,35 @@
const test = require('tape');
const { sippUac } = require('./sipp')('test_fs');
const bent = require('bent');
const getJSON = bent('json')
const clearModule = require('clear-module');
process.on('unhandledRejection', (reason, p) => {
console.log('Unhandled Rejection at: Promise', p, 'reason:', reason);
});
function connect(connectable) {
return new Promise((resolve, reject) => {
connectable.on('connect', () => {
return resolve();
});
});
}
test('sending SIP in-dialog requests tests', async(t) => {
clearModule.all();
const {srf, disconnect} = require('../app');
try {
await connect(srf);
await sippUac('uac-send-info-during-dialog.xml', '172.38.0.10');
const obj = await getJSON('http://127.0.0.1:3104/actionHook');
t.ok(obj.result === 'success' && obj.sip_status === 200, 'successfully sent SIP INFO');
disconnect();
} catch (err) {
console.log(`error received: ${err}`);
disconnect();
t.error(err);
}
});

View File

@@ -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);

15
test/test-apps/info.json Normal file
View File

@@ -0,0 +1,15 @@
[
{
"verb": "say",
"text": "hello"
},
{
"verb": "sip:request",
"method": "info",
"headers": {
"Content-Type": "application/text"
},
"body": "here I am ",
"actionHook": "/actionHook"
}
]

View File

@@ -6,7 +6,7 @@ const { SemanticResourceAttributes } = require('@opentelemetry/semantic-conventi
const { BatchSpanProcessor } = require('@opentelemetry/sdk-trace-base');
const { JaegerExporter } = require('@opentelemetry/exporter-jaeger');
const { ZipkinExporter } = require('@opentelemetry/exporter-zipkin');
const { CollectorTraceExporter } = require('@opentelemetry/exporter-collector');
const { OTLPTraceExporter } = require ('@opentelemetry/exporter-trace-otlp-http');
//const { HttpInstrumentation } = require('@opentelemetry/instrumentation-http');
//const { ExpressInstrumentation } = require('@opentelemetry/instrumentation-express');
//const { PinoInstrumentation } = require('@opentelemetry/instrumentation-pino');
@@ -29,7 +29,7 @@ module.exports = (serviceName) => {
exporter = new ZipkinExporter({url:process.env.OTEL_EXPORTER_ZIPKIN_URL});
}
else {
exporter = new CollectorTraceExporter({
exporter = new OTLPTraceExporter({
url: process.OTEL_EXPORTER_COLLECTOR_URL
});
}