Compare commits

..

6 Commits

Author SHA1 Message Date
akirilyuk
2336948cc4 allow new settings for gather verb 2022-02-26 11:55:46 +01:00
akirilyuk
caa7b3a03a change log level to debug 2022-02-14 17:11:47 +01:00
akirilyuk
63c0c97024 add husky pre push hook 2022-02-14 14:23:26 +01:00
akirilyuk
b126719ba7 add husky pre push hook 2022-02-14 14:22:04 +01:00
akirilyuk
d79c733aa2 fix linting 2022-02-14 14:16:44 +01:00
akirilyuk
15f85c9730 add say and gather task features 2022-02-14 14:08:56 +01:00
46 changed files with 589 additions and 3519 deletions

3
.husky/pre-push Executable file
View File

@@ -0,0 +1,3 @@
#!/bin/sh
. "$(dirname "$0")/_/husky.sh"
npm run jslint

View File

@@ -1,10 +1,10 @@
FROM node:slim
FROM node:17.4-slim
WORKDIR /opt/app/
COPY package.json package-lock.json ./
RUN npm ci
COPY package.json ./
RUN npm install
RUN npm prune
COPY . /opt/app
ARG NODE_ENV
ENV NODE_ENV $NODE_ENV
CMD [ "npm", "start" ]
CMD [ "npm", "start" ]

19
app.js
View File

@@ -11,10 +11,6 @@ assert.ok(process.env.JAMBONES_NETWORK_CIDR || process.env.K8S, 'missing JAMBONE
const Srf = require('drachtio-srf');
const srf = new Srf();
const tracer = require('./tracer')(process.env.JAMBONES_OTEL_SERVICE_NAME || 'jambonz-feature-server');
const api = require('@opentelemetry/api');
srf.locals = {...srf.locals, otel: {tracer, api}};
const PORT = process.env.HTTP_PORT || 3000;
const opts = {
timestamp: () => {return `, "time": "${new Date().toISOString()}"`;},
@@ -27,7 +23,6 @@ installSrfLocals(srf, logger);
const {
initLocals,
createRootSpan,
getAccountDetails,
normalizeNumbers,
retrieveApplication,
@@ -67,7 +62,6 @@ if (process.env.NODE_ENV === 'test') {
srf.use('invite', [
initLocals,
createRootSpan,
getAccountDetails,
normalizeNumbers,
retrieveApplication,
@@ -130,17 +124,4 @@ function handle(signal) {
srf.locals.disabled = true;
}
if (process.env.JAMBONZ_CLEANUP_INTERVAL_MINS) {
const {clearFiles} = require('./lib/utils/cron-jobs');
/* cleanup orphaned files or channels every so often */
setInterval(async() => {
try {
await clearFiles();
} catch (err) {
logger.error({err}, 'app.js: error clearing files');
}
}, 1000 * 60 * (process.env.JAMBONZ_CLEANUP_INTERVAL_MINS || 60));
}
module.exports = {srf, logger, disconnect};

View File

@@ -6,14 +6,11 @@ const {CallDirection, CallStatus} = require('../../utils/constants');
const { v4: uuidv4 } = require('uuid');
const SipError = require('drachtio-srf').SipError;
const sysError = require('./error');
const HttpRequestor = require('../../utils/http-requestor');
const WsRequestor = require('../../utils/ws-requestor');
const RootSpan = require('../../utils/call-tracer');
const Requestor = require('../../utils/requestor');
const dbUtils = require('../../utils/db-utils');
router.post('/', async(req, res) => {
const {logger} = req.app.locals;
const accountSid = req.body.account_sid;
logger.debug({body: req.body}, 'got createCall request');
try {
@@ -41,7 +38,7 @@ router.post('/', async(req, res) => {
'X-Jambonz-Routing': target.type,
'X-Jambonz-FS-UUID': srf.locals.fsUUID,
'X-Call-Sid': callSid,
'X-Account-Sid': accountSid
'X-Account-Sid': req.body.account_sid
};
switch (target.type) {
@@ -50,7 +47,7 @@ router.post('/', async(req, res) => {
uri = `sip:${target.number}@${sbcAddress}`;
to = target.number;
if ('teams' === target.type) {
const obj = await lookupTeamsByAccount(accountSid);
const obj = await lookupTeamsByAccount(req.body.account_sid);
if (!obj) throw new Error('dial to ms teams not allowed; account must first be configured with teams info');
Object.assign(opts.headers, {
'X-MS-Teams-FQDN': obj.ms_teams_fqdn,
@@ -74,17 +71,6 @@ router.post('/', async(req, res) => {
break;
}
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(
`createCall: selected ${voip_carrier_sid} for requested carrier: ${target.trunk || 'unspecified'})`);
if (voip_carrier_sid) {
opts.headers['X-Requested-Carrier-Sid'] = voip_carrier_sid;
}
}
/* create endpoint for outdial */
const ms = getFreeswitch();
if (!ms) throw new Error('no available Freeswitch for outbound call creation');
@@ -118,52 +104,27 @@ 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) {
app.requestor = new WsRequestor(logger, account.account_sid, app.call_hook, account.webhook_secret) ;
app.notifier = app.requestor;
}
else {
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: () => {}};
app.requestor = new Requestor(logger, account.account_sid, app.call_hook, account.webhook_secret);
if (app.call_status_hook) {
app.notifier = new Requestor(logger, account.account_sid, app.call_status_hook, account.webhook_secret);
}
else app.notifier = {request: () => {}};
/* now launch the outdial */
try {
const dlg = await srf.createUAC(uri, {...opts, followRedirects: true, keepUriOnRedirect: true}, {
cbRequest: (err, inviteReq) => {
/* in case of 302 redirect, this gets called twice, ignore the second
except to update the req so that it can later be canceled if need be
*/
if (res.headersSent) {
logger.info(`create-call: got redirect, updating request to new call-id ${req.get('Call-ID')}`);
if (cs) cs.req = inviteReq;
return;
}
/* in case of 302 redirect, this gets called twice, ignore the second */
if (res.headersSent) return;
if (err) {
logger.error(err, 'createCall Error creating call');
res.status(500).send('Call Failure');
return;
}
inviteReq.srf = srf;
inviteReq.locals = {
...(inviteReq || {}),
callSid,
application_sid: app.application_sid
};
/* ok our outbound INVITE is in flight */
const tasks = [restDial];
const rootSpan = new RootSpan('rest-call', inviteReq);
sipLogger = logger.child({
callSid,
callId: inviteReq.get('Call-ID'),
accountSid,
traceId: rootSpan.traceId
});
app.requestor.logger = app.notifier.logger = sipLogger;
const callInfo = new CallInfo({
direction: CallDirection.Outbound,
req: inviteReq,
@@ -171,24 +132,17 @@ router.post('/', async(req, res) => {
tag: app.tag,
callSid,
accountSid: req.body.account_sid,
applicationSid: app.application_sid,
traceId: rootSpan.traceId
});
cs = new RestCallSession({
logger: sipLogger,
application: app,
srf,
req: inviteReq,
ep,
tasks,
callInfo,
accountInfo,
rootSpan
applicationSid: app.application_sid
});
cs = new RestCallSession({logger, application: app, srf, req: inviteReq, ep, tasks, callInfo, accountInfo});
cs.exec(req);
res.status(201).json({sid: cs.callSid});
sipLogger = logger.child({
callSid: cs.callSid,
callId: callInfo.callId
});
sipLogger.info(`outbound REST call attempt to ${JSON.stringify(target)} has been sent`);
},
cbProvisional: (prov) => {
@@ -199,11 +153,7 @@ router.post('/', async(req, res) => {
}
});
connectStream(dlg.remote.sdp);
cs.emit('callStatusChange', {
callStatus: CallStatus.InProgress,
sipStatus: 200,
sipReason: 'OK'
});
cs.emit('callStatusChange', {callStatus: CallStatus.InProgress, sipStatus: 200});
restDial.emit('callStatus', 200);
restDial.emit('connect', dlg);
}
@@ -214,18 +164,10 @@ router.post('/', async(req, res) => {
else if (487 === err.status) callStatus = CallStatus.NoAnswer;
if (sipLogger) sipLogger.info(`REST outdial failed with ${err.status}`);
else console.log(`REST outdial failed with ${err.status}`);
if (cs) cs.emit('callStatusChange', {
callStatus,
sipStatus: err.status,
sipReason: err.reason
});
if (cs) cs.emit('callStatusChange', {callStatus, sipStatus: err.status});
}
else {
if (cs) cs.emit('callStatusChange', {
callStatus,
sipStatus: 500,
sipReason: 'Internal Server Error'
});
if (cs) cs.emit('callStatusChange', {callStatus, sipStatus: 500});
if (sipLogger) sipLogger.error({err}, 'REST outdial failed');
else console.error(err);
}

View File

@@ -1,6 +1,5 @@
const router = require('express').Router();
const HttpRequestor = require('../../utils/http-requestor');
const WsRequestor = require('../../utils/ws-requestor');
const Requestor = require('../../utils/requestor');
const CallInfo = require('../../session/call-info');
const {CallDirection} = require('../../utils/constants');
const SmsSession = require('../../session/sms-call-session');
@@ -19,17 +18,7 @@ router.post('/:partner', async(req, res) => {
const app = req.body.app;
const account = await lookupAccountBySid(app.accountSid);
const hook = app.messaging_hook;
let requestor;
if ('WS' === hook?.method) {
app.requestor = new WsRequestor(logger, account.account_sid, hook, account.webhook_secret) ;
app.notifier = app.requestor;
}
else {
app.requestor = new HttpRequestor(logger, account.account_sid, hook, account.webhook_secret);
app.notifier = {request: () => {}};
}
const requestor = new Requestor(logger, account.account_sid, hook, account.webhook_secret);
const payload = {
carrier: req.params.partner,
messageSid: app.messageSid,
@@ -44,7 +33,7 @@ router.post('/:partner', async(req, res) => {
res.status(200).json({sid: req.body.messageSid});
try {
tasks = await requestor.request('session:new', hook, payload);
tasks = await requestor.request(hook, payload);
logger.info({tasks}, 'response from incoming SMS webhook');
} catch (err) {
logger.error({err, hook}, 'Error sending incoming SMS message');

View File

@@ -12,9 +12,6 @@ function retrieveCallSession(callSid, opts) {
throw new DbErrorBadRequest('call_status_hook can be updated only when call_hook is also being updated');
}
const cs = sessionTracker.get(callSid);
if (!cs) {
throw new DbErrorUnprocessableRequest('call session is gone');
}
if (opts.call_status === CallStatus.Completed && !cs.hasStableDialog) {
throw new DbErrorUnprocessableRequest('current call state is incompatible with requested action');
@@ -48,18 +45,8 @@ router.post('/:callSid', async(req, res) => {
logger.info(`updateCall: callSid not found ${callSid}`);
return res.sendStatus(404);
}
if (req.body.sip_request) {
const response = await cs.updateCall(req.body, callSid);
res.status(200).json({
status: response.status,
reason: response.reason
});
}
else {
res.sendStatus(202);
cs.updateCall(req.body, callSid);
}
res.sendStatus(202);
cs.updateCall(req.body, callSid);
} catch (err) {
sysError(logger, res, err);
}

View File

@@ -1,14 +1,11 @@
const { v4: uuidv4 } = require('uuid');
const {CallDirection} = require('./utils/constants');
const CallInfo = require('./session/call-info');
const HttpRequestor = require('./utils/http-requestor');
const WsRequestor = require('./utils/ws-requestor');
const Requestor = require('./utils/requestor');
const makeTask = require('./tasks/make_task');
const parseUri = require('drachtio-srf').parseUri;
const normalizeJambones = require('./utils/normalize-jambones');
const dbUtils = require('./utils/db-utils');
const RootSpan = require('./utils/call-tracer');
const listTaskNames = require('./utils/summarize-tasks');
module.exports = function(srf, logger) {
const {
@@ -19,18 +16,15 @@ module.exports = function(srf, logger) {
lookupAppByTeamsTenant
} = srf.locals.dbHelpers;
const {lookupAccountDetails} = dbUtils(logger, srf);
function initLocals(req, res, next) {
if (!req.has('X-Account-Sid')) {
logger.info('getAccountDetails - rejecting call due to missing X-Account-Sid header');
return res.send(500);
}
const callSid = req.has('X-Retain-Call-Sid') ? req.get('X-Retain-Call-Sid') : uuidv4();
const account_sid = req.get('X-Account-Sid');
req.locals = {callSid, account_sid};
req.locals = {
callSid,
logger: logger.child({callId: req.get('Call-ID'), callSid})
};
if (req.has('X-Application-Sid')) {
const application_sid = req.get('X-Application-Sid');
logger.debug(`got application from X-Application-Sid header: ${application_sid}`);
req.locals.logger.debug(`got application from X-Application-Sid header: ${application_sid}`);
req.locals.application_sid = application_sid;
}
if (req.has('X-Authenticated-User')) req.locals.originatingUser = req.get('X-Authenticated-User');
@@ -39,50 +33,19 @@ module.exports = function(srf, logger) {
next();
}
function createRootSpan(req, res, next) {
const {callSid, account_sid} = req.locals;
const rootSpan = new RootSpan('incoming-call', req);
const traceId = rootSpan.traceId;
req.locals = {
...req.locals,
traceId,
logger: logger.child({
callId: req.get('Call-ID'),
callSid,
accountSid: account_sid,
callingNumber: req.callingNumber,
calledNumber: req.calledNumber,
traceId}),
rootSpan
};
/**
* end the span on final failure or cancel from caller;
* otherwise it will be closed when sip dialog is destroyed
*/
req.once('cancel', () => {
rootSpan.setAttributes({finalStatus: 487});
rootSpan.end();
});
res.once('finish', () => {
rootSpan.setAttributes({finalStatus: res.statusCode});
res.statusCode >= 300 && rootSpan.end();
});
next();
}
/**
* retrieve account information for the incoming call
*/
async function getAccountDetails(req, res, next) {
const {rootSpan, account_sid} = req.locals;
const {span} = rootSpan.startChildSpan('lookupAccountDetails');
if (!req.has('X-Account-Sid')) {
logger.info('getAccountDetails - rejecting call due to missing X-Account-Sid header');
return res.send(500);
}
const account_sid = req.locals.account_sid = req.get('X-Account-Sid');
try {
req.locals.accountInfo = await lookupAccountDetails(account_sid);
span.end();
if (!req.locals.accountInfo.account.is_active) {
logger.info(`Account is inactive or suspended ${account_sid}`);
// TODO: alert
@@ -91,7 +54,6 @@ module.exports = function(srf, logger) {
logger.debug({accountInfo: req.locals?.accountInfo?.account}, `retrieved account info for ${account_sid}`);
next();
} catch (err) {
span.end();
logger.info({err}, `Error retrieving account details for account ${account_sid}`);
res.send(503, {headers: {'X-Reason': `No Account exists for sid ${account_sid}`}});
}
@@ -123,8 +85,7 @@ module.exports = function(srf, logger) {
*/
async function retrieveApplication(req, res, next) {
const logger = req.locals.logger;
const {accountInfo, account_sid, rootSpan} = req.locals;
const {span} = rootSpan.startChildSpan('lookupApplication');
const {accountInfo, account_sid} = req.locals;
try {
let app;
if (req.locals.application_sid) app = await lookupAppBySid(req.locals.application_sid);
@@ -144,7 +105,7 @@ module.exports = function(srf, logger) {
}
else {
const uri = parseUri(req.uri);
const arr = /context-(.*)/.exec(uri?.user);
const arr = /context-(.*)/.exec(uri.user);
if (arr) {
// this is a transfer from another feature server
const {retrieveKey, deleteKey} = srf.locals.dbHelpers;
@@ -168,11 +129,6 @@ module.exports = function(srf, logger) {
}
}
span.setAttributes({
'app.hook': app?.call_hook?.url,
'application_sid': req.locals.application_sid
});
span.end();
if (!app || !app.call_hook || !app.call_hook.url) {
logger.info(`rejecting call to ${req.locals.calledNumber}: no application or webhook url`);
return res.send(480, {
@@ -186,18 +142,10 @@ module.exports = function(srf, logger) {
* create a requestor that we will use for all http requests we make during the call.
* also create a notifier for call status events (if not needed, its a no-op).
*/
if ('WS' === app.call_hook?.method ||
app.call_hook?.url.startsWith('ws://') || app.call_hook?.url.startsWith('wss://')) {
app.requestor = new WsRequestor(logger, account_sid, app.call_hook, accountInfo.account.webhook_secret) ;
app.notifier = app.requestor;
app.call_hook.method = 'WS';
}
else {
app.requestor = new HttpRequestor(logger, account_sid, app.call_hook, accountInfo.account.webhook_secret);
if (app.call_status_hook) app.notifier = new HttpRequestor(logger, account_sid, app.call_status_hook,
accountInfo.account.webhook_secret);
else app.notifier = {request: () => {}};
}
app.requestor = new Requestor(logger, account_sid, app.call_hook, accountInfo.account.webhook_secret);
if (app.call_status_hook) app.notifier = new Requestor(logger, account_sid, app.call_status_hook,
accountInfo.account.webhook_secret);
else app.notifier = {request: () => {}};
req.locals.application = app;
const obj = Object.assign({}, app);
@@ -206,15 +154,9 @@ module.exports = function(srf, logger) {
// eslint-disable-next-line no-unused-vars
const {call_hook, call_status_hook, ...appInfo} = obj; // mask sensitive data like user/pass on webhook
logger.info({app: appInfo}, `retrieved application for incoming call to ${req.locals.calledNumber}`);
req.locals.callInfo = new CallInfo({
req,
app,
direction: CallDirection.Inbound,
traceId: rootSpan.traceId
});
req.locals.callInfo = new CallInfo({req, app, direction: CallDirection.Inbound});
next();
} catch (err) {
span.end();
logger.error(err, `${req.get('Call-ID')} Error looking up application for ${req.calledNumber}`);
res.send(500);
}
@@ -225,55 +167,29 @@ module.exports = function(srf, logger) {
*/
async function invokeWebCallback(req, res, next) {
const logger = req.locals.logger;
const {rootSpan, application:app} = req.locals;
let span;
const app = req.locals.application;
try {
if (app.tasks) {
app.tasks = normalizeJambones(logger, app.tasks).map((tdata) => makeTask(logger, tdata));
if (0 === app.tasks.length) throw new Error('no application provided');
return next();
}
/* retrieve the application to execute for this inbound call */
const params = Object.assign(['POST', 'WS'].includes(app.call_hook.method) ? {sip: req.msg} : {},
req.locals.callInfo, {
defaults: {
synthesizer: {
vendor: app.speech_synthesis_vendor,
language: app.speech_synthesis_language,
voice: app.speech_synthesis_voice
},
recognizer: {
vendor: app.speech_recognizer_vendor,
language: app.speech_recognizer_language
}
}
});
logger.debug({params}, 'sending initial webhook');
const obj = rootSpan.startChildSpan('performAppWebhook');
span = obj.span;
const b3 = rootSpan.getTracingPropagation();
const httpHeaders = b3 && {b3};
const json = await app.requestor.request('session:new', app.call_hook, params, httpHeaders);
const params = Object.assign(app.call_hook.method === 'POST' ? {sip: req.msg} : {},
req.locals.callInfo);
const json = await app.requestor.request(app.call_hook, params);
app.tasks = normalizeJambones(logger, json).map((tdata) => makeTask(logger, tdata));
span.setAttributes({
'http.statusCode': 200,
'app.tasks': listTaskNames(app.tasks)
});
span.end();
if (0 === app.tasks.length) throw new Error('no application provided');
next();
} catch (err) {
span?.setAttributes({webhookStatus: err.statusCode});
span?.end();
logger.info({err}, `Error retrieving or parsing application: ${err?.message}`);
res.send(480, {headers: {'X-Reason': err?.message || 'unknown'}});
app.requestor.close();
logger.info({err}, `Error retrieving or parsing application: ${err.message}`);
res.send(480, {headers: {'X-Reason': err.message}});
}
}
return {
initLocals,
createRootSpan,
getAccountDetails,
normalizeNumbers,
retrieveApplication,

View File

@@ -8,15 +8,14 @@ const CallSession = require('./call-session');
*/
class AdultingCallSession extends CallSession {
constructor({logger, application, singleDialer, tasks, callInfo, accountInfo, rootSpan}) {
constructor({logger, application, singleDialer, tasks, callInfo, accountInfo}) {
super({
logger,
application,
srf: singleDialer.dlg.srf,
tasks,
callInfo,
accountInfo,
rootSpan
accountInfo
});
this.sd = singleDialer;
@@ -31,25 +30,15 @@ class AdultingCallSession extends CallSession {
return this.sd.dlg;
}
/**
* Note: this is not an error. It is only here to avoid an assert ("no setter for dlg")
* when there is a call in Session:_clearResources to null out dlg and ep
*/
set dlg(newDlg) {}
get ep() {
return this.sd.ep;
}
/* see note above */
set ep(newEp) {}
get callSid() {
return this.callInfo.callSid;
}
_callerHungup() {
}
}
module.exports = AdultingCallSession;

View File

@@ -10,7 +10,6 @@ class CallInfo {
let from ;
let srf;
this.direction = opts.direction;
this.traceId = opts.traceId;
if (opts.req) {
const u = opts.req.getParsedHeader('from');
const uri = parseUri(u.uri);
@@ -28,7 +27,6 @@ class CallInfo {
this.to = req.calledNumber;
this.callId = req.get('Call-ID');
this.sipStatus = 100;
this.sipReason = 'Trying';
this.callStatus = CallStatus.Trying;
this.originatingSipIp = req.get('X-Forwarded-For');
this.originatingSipTrunkName = req.get('X-Originating-Carrier');
@@ -47,7 +45,6 @@ class CallInfo {
this.callId = req.get('Call-ID');
this.callStatus = CallStatus.Trying,
this.sipStatus = 100;
this.sipReason = 'Trying';
}
else if (this.direction === CallDirection.None) {
// outbound SMS
@@ -68,7 +65,6 @@ class CallInfo {
this.callStatus = CallStatus.Trying,
this.callId = req.get('Call-ID');
this.sipStatus = 100;
this.sipReason = 'Trying';
this.from = from || req.callingNumber;
this.to = to;
if (tag) this._customerData = tag;
@@ -85,10 +81,9 @@ class CallInfo {
* @param {string} callStatus - current call status
* @param {number} sipStatus - current sip status
*/
updateCallStatus(callStatus, sipStatus, sipReason) {
updateCallStatus(callStatus, sipStatus) {
this.callStatus = callStatus;
if (sipStatus) this.sipStatus = sipStatus;
if (sipReason) this.sipReason = sipReason;
}
/**
@@ -111,11 +106,9 @@ class CallInfo {
to: this.to,
callId: this.callId,
sipStatus: this.sipStatus,
sipReason: this.sipReason,
callStatus: this.callStatus,
callerId: this.callerId,
accountSid: this.accountSid,
traceId: this.traceId,
applicationSid: this.applicationSid,
fsSipAddress: this.localSipAddress
};

View File

@@ -7,8 +7,7 @@ const sessionTracker = require('./session-tracker');
const makeTask = require('../tasks/make_task');
const normalizeJambones = require('../utils/normalize-jambones');
const listTaskNames = require('../utils/summarize-tasks');
const HttpRequestor = require('../utils/http-requestor');
const WsRequestor = require('../utils/ws-requestor');
const Requestor = require('../utils/requestor');
const BADPRECONDITIONS = 'preconditions not met';
const CALLER_CANCELLED_ERR_MSG = 'Response not sent due to unknown transaction';
@@ -35,7 +34,7 @@ class CallSession extends Emitter {
* @param {array} opts.tasks - tasks we are to execute
* @param {callInfo} opts.callInfo - information about the call
*/
constructor({logger, application, srf, tasks, callInfo, accountInfo, rootSpan, memberId, confName, confUuid}) {
constructor({logger, application, srf, tasks, callInfo, accountInfo, memberId, confName, confUuid}) {
super();
this.logger = logger;
this.application = application;
@@ -50,9 +49,6 @@ class CallSession extends Emitter {
this.stackIdx = 0;
this.callGone = false;
this.notifiedComplete = false;
this.rootSpan = rootSpan;
assert(rootSpan);
this.tmpFiles = new Set();
@@ -66,9 +62,6 @@ class CallSession extends Emitter {
}
this._pool = srf.locals.dbHelpers.pool;
this.requestor.on('command', this._onCommand.bind(this));
this.requestor.on('connection-dropped', this._onWsConnectionDropped.bind(this));
}
/**
@@ -121,27 +114,18 @@ class CallSession extends Emitter {
get speechSynthesisVendor() {
return this.application.speech_synthesis_vendor;
}
set speechSynthesisVendor(vendor) {
this.application.speech_synthesis_vendor = vendor;
}
/**
* default voice to use for speech synthesis if not provided in the app
*/
get speechSynthesisVoice() {
return this.application.speech_synthesis_voice;
}
set speechSynthesisVoice(voice) {
this.application.speech_synthesis_voice = voice;
}
/**
* default language to use for speech synthesis if not provided in the app
*/
get speechSynthesisLanguage() {
return this.application.speech_synthesis_language;
}
set speechSynthesisLanguage(language) {
this.application.speech_synthesis_language = language;
}
/**
* default vendor to use for speech recognition if not provided in the app
@@ -149,18 +133,12 @@ class CallSession extends Emitter {
get speechRecognizerVendor() {
return this.application.speech_recognizer_vendor;
}
set speechRecognizerVendor(vendor) {
this.application.speech_recognizer_vendor = vendor;
}
/**
* default language to use for speech recognition if not provided in the app
*/
get speechRecognizerLanguage() {
return this.application.speech_recognizer_language;
}
set speechRecognizerLanguage(language) {
this.application.speech_recognizer_language = language;
}
/**
* indicates whether the call currently in progress
@@ -226,60 +204,6 @@ class CallSession extends Emitter {
return this.memberId && this.confName && this.confUuid;
}
get isBotModeEnabled() {
return this.backgroundGatherTask;
}
get b3() {
return this.rootSpan?.getTracingPropagation();
}
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));
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}`);
this.backgroundGatherTask.span = span;
this.backgroundGatherTask.ctx = ctx;
this.backgroundGatherTask.exec(this, resources)
.then(() => {
this.logger.info('CallSession:enableBotMode: gather completed');
this.backgroundGatherTask && this.backgroundGatherTask.removeAllListeners();
this.backgroundGatherTask && this.backgroundGatherTask.span.end();
this.backgroundGatherTask = null;
if (autoEnable && !this.callGone && !this._stopping) {
this.logger.info('CallSession:enableBotMode: restarting background gather');
setImmediate(() => this.enableBotMode(gather, true));
}
return;
})
.catch((err) => {
this.logger.info({err}, 'CallSession:enableBotMode: gather threw error');
this.backgroundGatherTask && this.backgroundGatherTask.removeAllListeners();
this.backgroundGatherTask && this.backgroundGatherTask.span.end();
this.backgroundGatherTask = null;
});
} catch (err) {
this.logger.info({err, gather}, 'CallSession:enableBotMode - Error creating gather task');
}
}
disableBotMode() {
if (this.backgroundGatherTask) {
try {
this.backgroundGatherTask.removeAllListeners();
this.backgroundGatherTask.kill().catch((err) => {});
} catch (err) {}
this.backgroundGatherTask = null;
}
}
setConferenceDetails(memberId, confName, confUuid) {
assert(!this.memberId && !this.confName && !this.confUuid);
assert (memberId && confName && confUuid);
@@ -365,7 +289,6 @@ class CallSession extends Emitter {
*/
async exec() {
this.logger.info({tasks: listTaskNames(this.tasks)}, `CallSession:exec starting ${this.tasks.length} tasks`);
while (this.tasks.length && !this.callGone) {
const taskNum = ++this.taskIdx;
const stackNum = this.stackIdx;
@@ -374,24 +297,12 @@ class CallSession extends Emitter {
try {
const resources = await this._evaluatePreconditions(task);
this.currentTask = task;
if (TaskName.Gather === task.name && this.isBotModeEnabled) {
const timeout = task.timeout;
this.logger.info(`CallSession:exec skipping #${stackNum}:${taskNum}: ${task.name}`);
this.backgroundGatherTask.updateTimeout(timeout);
}
else {
const {span, ctx} = this.rootSpan.startChildSpan(`verb:${task.summary}`);
task.span = span;
task.ctx = ctx;
await task.exec(this, resources);
task.span.end();
}
await task.exec(this, resources);
this.currentTask = null;
this.logger.info(`CallSession:exec completed task #${stackNum}:${taskNum}: ${task.name}`);
} catch (err) {
task.span?.end();
this.currentTask = null;
if (err.message?.includes(BADPRECONDITIONS)) {
if (err.message.includes(BADPRECONDITIONS)) {
this.logger.info(`CallSession:exec task #${stackNum}:${taskNum}: ${task.name}: ${err.message}`);
}
else {
@@ -399,31 +310,10 @@ class CallSession extends Emitter {
break;
}
}
if (0 === this.tasks.length && this.hasStableDialog && this.requestor instanceof WsRequestor) {
let span;
try {
const {span} = this.rootSpan.startChildSpan('waiting for commands');
const {reason, queue, command} = await this._awaitCommandsOrHangup();
span.setAttributes({
'completion.reason': reason,
'async.request.queue': queue,
'async.request.command': command
});
span.end();
if (!this.hasStableDialog || this.callGone) break;
} catch (err) {
span.end();
this.logger.info(err, 'CallSession:exec - error waiting for new commands');
break;
}
}
}
// all done - cleanup
this.logger.info('CallSession:exec all tasks complete');
this._stopping = true;
this.disableBotMode();
this._onTasksDone();
this._clearResources();
@@ -472,15 +362,12 @@ class CallSession extends Emitter {
* this is called to clean up when the call is released from one side or another
*/
_callReleased() {
this.logger.debug('CallSession:_callReleased - caller hung up');
this.callGone = true;
if (this.currentTask) {
this.currentTask.kill(this);
this.currentTask = null;
}
if (this.wakeupResolver) {
this.wakeupResolver({reason: 'session ended'});
this.wakeupResolver = null;
}
}
/**
@@ -517,45 +404,29 @@ class CallSession extends Emitter {
*/
async _lccCallHook(opts) {
const webhooks = [];
let sd, tasks, childTasks;
const b3 = this.b3;
const httpHeaders = b3 && {b3};
if (opts.call_hook || opts.child_call_hook) {
if (opts.call_hook) {
webhooks.push(this.requestor.request('session:redirect', opts.call_hook, this.callInfo.toJSON(), httpHeaders));
let sd;
if (opts.call_hook) webhooks.push(this.requestor.request(opts.call_hook, this.callInfo.toJSON()));
if (opts.child_call_hook) {
/* child call hook only allowed from a connected Dial state */
const task = this.currentTask;
sd = task.sd;
if (task && TaskName.Dial === task.name && sd) {
webhooks.push(this.requestor.request(opts.child_call_hook, sd.callInfo.toJSON()));
}
if (opts.child_call_hook) {
/* child call hook only allowed from a connected Dial state */
const task = this.currentTask;
sd = task.sd;
if (task && TaskName.Dial === task.name && sd) {
webhooks.push(this.requestor.request(
'session:redirect', opts.child_call_hook, sd.callInfo.toJSON(), httpHeaders));
}
}
const [tasks1, tasks2] = await Promise.all(webhooks);
if (opts.call_hook) {
tasks = tasks1;
if (opts.child_call_hook) childTasks = tasks2;
}
else childTasks = tasks1;
}
else if (opts.parent_call || opts.child_call) {
const {parent_call, child_call} = opts;
assert.ok(!parent_call || Array.isArray(parent_call), 'CallSession:_lccCallHook - parent_call must be an array');
assert.ok(!child_call || Array.isArray(child_call), 'CallSession:_lccCallHook - child_call must be an array');
tasks = parent_call;
childTasks = child_call;
const [tasks1, tasks2] = await Promise.all(webhooks);
let tasks, childTasks;
if (opts.call_hook) {
tasks = tasks1;
if (opts.child_call_hook) childTasks = tasks2;
}
else childTasks = tasks1;
if (childTasks) {
const {parentLogger} = this.srf.locals;
const childLogger = parentLogger.child({callId: this.callId, callSid: sd.callSid});
const t = normalizeJambones(childLogger, childTasks).map((tdata) => makeTask(childLogger, tdata));
childLogger.info({tasks: listTaskNames(t)}, 'CallSession:_lccCallHook new task list for child call');
// TODO: if using websockets api, we need a new websocket for the adulting session..
const cs = await sd.doAdulting({
logger: childLogger,
application: this.application,
@@ -603,7 +474,7 @@ class CallSession extends Emitter {
task.mute(callSid, mute).catch((err) => this.logger.error(err, 'CallSession:_lccMuteStatus'));
}
async _lccConfHoldStatus(opts) {
async _lccConfHoldStatus(callSid, opts) {
const task = this.currentTask;
if (!task || TaskName.Conference !== task.name || !this.isInConference) {
return this.logger.info('CallSession:_lccConfHoldStatus - invalid command as call is not in conference');
@@ -611,7 +482,7 @@ class CallSession extends Emitter {
task.doConferenceHold(this, opts);
}
async _lccConfMuteStatus(opts) {
async _lccConfMuteStatus(callSid, opts) {
const task = this.currentTask;
if (!task || TaskName.Conference !== task.name || !this.isInConference) {
return this.logger.info('CallSession:_lccConfHoldStatus - invalid command as call is not in conference');
@@ -619,30 +490,6 @@ class CallSession extends Emitter {
task.doConferenceMuteNonModerators(this, opts);
}
async _lccSipRequest(opts, callSid) {
const {sip_request} = opts;
const {method, content_type, content, headers = {}} = sip_request;
if (!this.hasStableDialog) {
this.logger.info('CallSession:_lccSipRequest - invalid command as we do not have a stable call');
return;
}
try {
const dlg = callSid === this.callSid ? this.dlg : this.currentTask.dlg;
const res = await dlg.request({
method,
headers: {
...headers,
'Content-Type': content_type
},
body: content
});
this.logger.debug({res}, `CallSession:_lccSipRequest got response to ${method}`);
return res;
} catch (err) {
this.logger.error({err}, `CallSession:_lccSipRequest - error sending ${method}`);
}
}
/**
* perform live call control -- whisper to one party or the other on a call
* @param {array} opts - array of play or say tasks
@@ -650,8 +497,6 @@ class CallSession extends Emitter {
async _lccWhisper(opts, callSid) {
const {whisper} = opts;
let tasks;
const b3 = this.b3;
const httpHeaders = b3 && {b3};
// this whole thing requires us to be in a Dial verb
const task = this.currentTask;
@@ -662,7 +507,7 @@ class CallSession extends Emitter {
// allow user to provide a url object, a url string, an array of tasks, or a single task
if (typeof whisper === 'string' || (typeof whisper === 'object' && whisper.url)) {
// retrieve a url
const json = await this.requestor(opts.call_hook, this.callInfo.toJSON(), httpHeaders);
const json = await this.requestor(opts.call_hook, this.callInfo.toJSON());
tasks = normalizeJambones(this.logger, json).map((tdata) => makeTask(this.logger, tdata));
}
else if (Array.isArray(whisper)) {
@@ -715,14 +560,10 @@ class CallSession extends Emitter {
await this._lccMuteStatus(callSid, opts.mute_status === 'mute');
}
else if (opts.conf_hold_status) {
await this._lccConfHoldStatus(opts);
await this._lccConfHoldStatus(callSid, opts);
}
else if (opts.conf_mute_status) {
await this._lccConfMuteStatus(opts);
}
else if (opts.sip_request) {
const res = await this._lccSipRequest(opts, callSid);
return {status: res.status, reason: res.reason};
await this._lccConfMuteStatus(callSid, opts);
}
// whisper may be the only thing we are asked to do, or it may that
@@ -763,131 +604,6 @@ class CallSession extends Emitter {
this.taskIdx = 0;
}
/**
* Append tasks to the current execution stack UNLESS there is a gather in the stack.
* in that case, insert the tasks before the gather AND if the tasks include
* a gather then delete/remove the gather from the existing stack
* @param {*} t array of tasks
*/
_injectTasks(newTasks) {
const gatherPos = this.tasks.map((t) => t.name).indexOf(TaskName.Gather);
const currentlyExecutingGather = this.currentTask?.name === TaskName.Gather;
this.logger.debug({
currentTaskList: listTaskNames(this.tasks),
newContent: listTaskNames(newTasks),
currentlyExecutingGather,
gatherPos
}, 'CallSession:_injectTasks - starting');
const killGather = () => {
this.logger.debug('CallSession:_injectTasks - killing current gather because we have new content');
this.currentTask.kill(this);
};
if (-1 === gatherPos) {
/* no gather in the stack simply append tasks */
this.tasks.push(...newTasks);
this.logger.debug({
updatedTaskList: listTaskNames(this.tasks)
}, 'CallSession:_injectTasks - completed (simple append)');
/* we do need to kill the current gather if we are executing one */
if (currentlyExecutingGather) killGather();
return;
}
if (currentlyExecutingGather) killGather();
const newTasksHasGather = newTasks.find((t) => t.name === TaskName.Gather);
this.tasks.splice(gatherPos, newTasksHasGather ? 1 : 0, ...newTasks);
this.logger.debug({
updatedTaskList: listTaskNames(this.tasks)
}, 'CallSession:_injectTasks - completed');
}
_onCommand({msgid, command, call_sid, queueCommand, data}) {
this.logger.info({msgid, command, queueCommand}, 'CallSession:_onCommand - received command');
const resolution = {reason: 'received command', queue: queueCommand, command};
switch (command) {
case 'redirect':
if (Array.isArray(data)) {
const t = normalizeJambones(this.logger, data)
.map((tdata) => makeTask(this.logger, tdata));
if (!queueCommand) {
this.logger.info({tasks: listTaskNames(t)}, 'CallSession:_onCommand new task list');
this.replaceApplication(t);
}
else if (process.env.JAMBONES_INJECT_CONTENT) {
this.logger.debug({tasks: listTaskNames(t)}, 'CallSession:_onCommand - queueing tasks (injecting content)');
this._injectTasks(t);
this.logger.info({tasks: listTaskNames(this.tasks)}, 'CallSession:_onCommand - updated task list');
}
else {
this.logger.debug({tasks: listTaskNames(t)}, 'CallSession:_onCommand - queueing tasks');
this.tasks.push(...t);
this.logger.info({tasks: listTaskNames(this.tasks)}, 'CallSession:_onCommand - updated task list');
}
resolution.command = listTaskNames(t);
}
else this._lccCallHook(data);
break;
case 'call:status':
this._lccCallStatus(data);
break;
case 'mute:status':
this._lccMuteStatus(call_sid, data);
break;
case 'conf:mute-status':
this._lccConfMuteStatus(data);
break;
case 'conf:hold-status':
this._lccConfHoldStatus(data);
break;
case 'listen:status':
this._lccListenStatus(data);
break;
case 'whisper':
this._lccWhisper(data, call_sid);
break;
case 'sip:request':
this._lccSipRequest(data, call_sid)
.catch((err) => {
this.logger.info({err, data}, `CallSession:_onCommand - error sending ${data.method}`);
});
break;
default:
this.logger.info(`CallSession:_onCommand - invalid command ${command}`);
}
if (this.wakeupResolver) {
this.logger.debug({resolution}, 'CallSession:_onCommand - got commands, waking up..');
this.wakeupResolver(resolution);
this.wakeupResolver = null;
}
else {
const {span} = this.rootSpan.startChildSpan('async command');
const {queue, command} = resolution;
span.setAttributes({
'async.request.queue': queue,
'async.request.command': command
});
span.end();
}
}
_onWsConnectionDropped() {
const {stats} = this.srf.locals;
stats.increment('app.hook.remote_close');
}
_evaluatePreconditions(task) {
switch (task.preconditions) {
case TaskPreconditions.None:
@@ -952,11 +668,7 @@ class CallSession extends Emitter {
} catch (err) {
if (err === CALLER_CANCELLED_ERR_MSG) {
this.logger.error(err, 'caller canceled quickly before we could respond, ending call');
this._notifyCallStatusChange({
callStatus: CallStatus.NoAnswer,
sipStatus: 487,
sipReason: 'Request Terminated'
});
this._notifyCallStatusChange({callStatus: CallStatus.NoAnswer, sipStatus: 487});
this._callReleased();
}
else {
@@ -1028,9 +740,6 @@ class CallSession extends Emitter {
});
}
this.tmpFiles.clear();
this.requestor && this.requestor.close();
this.rootSpan && this.rootSpan.end();
}
/**
@@ -1063,21 +772,14 @@ class CallSession extends Emitter {
async propagateAnswer() {
if (!this.dlg) {
assert(this.ep);
this.dlg = await this.srf.createUAS(this.req, this.res, {
headers: {
'X-Trace-ID': this.req.locals.traceId,
'X-Call-Sid': this.req.locals.callSid
},
localSdp: this.ep.local.sdp
});
this.dlg = await this.srf.createUAS(this.req, this.res, {localSdp: this.ep.local.sdp});
this.logger.debug('answered call');
this.dlg.on('destroy', this._callerHungup.bind(this));
this.wrapDialog(this.dlg);
this.dlg.callSid = this.callSid;
this.emit('callStatusChange', {sipStatus: 200, sipReason: 'OK', callStatus: CallStatus.InProgress});
this.emit('callStatusChange', {sipStatus: 200, callStatus: CallStatus.InProgress});
this.dlg.on('modify', this._onReinvite.bind(this));
this.dlg.on('refer', this._onRefer.bind(this));
this.logger.debug(`CallSession:propagateAnswer - answered callSid ${this.callSid}`);
}
@@ -1103,22 +805,6 @@ class CallSession extends Emitter {
}
}
/**
* Handle incoming REFER if we are in a dial task
* @param {*} req
* @param {*} res
*/
_onRefer(req, res) {
const task = this.currentTask;
const sd = task.sd;
if (task && TaskName.Dial === task.name && sd) {
task.handleRefer(this, req, res);
}
else {
res.send(501);
}
}
/**
* create and endpoint if we don't have one; otherwise simply return
* the current media server and endpoint that are associated with this call
@@ -1155,7 +841,7 @@ class CallSession extends Emitter {
}
else {
this.logger.info({accountSid: this.accountSid, webhook: r[0]}, 'performQueueWebhook: webhook found');
this.queueEventHookRequestor = new HttpRequestor(this.logger, this.accountSid,
this.queueEventHookRequestor = new Requestor(this.logger, this.accountSid,
r[0], this.webhook_secret);
this.queueEventHook = r[0];
}
@@ -1169,7 +855,7 @@ class CallSession extends Emitter {
/* send webhook */
const params = {...obj, ...this.callInfo.toJSON()};
this.logger.info({accountSid: this.accountSid, params}, 'performQueueWebhook: sending webhook');
this.queueEventHookRequestor.request('queue:status', this.queueEventHook, params)
this.queueEventHookRequestor.request(this.queueEventHook, params)
.catch((err) => {
this.logger.info({err, accountSid: this.accountSid, obj}, 'Error sending queue notification event');
});
@@ -1257,12 +943,7 @@ class CallSession extends Emitter {
const duration = moment().diff(this.dlg.connectTime, 'seconds');
this.emit('callStatusChange', {callStatus: CallStatus.Completed, duration});
this.logger.debug('CallSession: call terminated by jambones');
this.rootSpan.setAttributes({'call.termination': 'hangup by jambonz'});
origDestroy();
if (this.wakeupResolver) {
this.wakeupResolver({reason: 'session ended'});
this.wakeupResolver = null;
}
}
};
}
@@ -1305,7 +986,7 @@ class CallSession extends Emitter {
* @param {number} sipStatus - current sip status
* @param {number} [duration] - duration of a completed call, in seconds
*/
_notifyCallStatusChange({callStatus, sipStatus, sipReason, duration}) {
_notifyCallStatusChange({callStatus, sipStatus, duration}) {
if (this.callMoved) return;
/* race condition: we hang up at the same time as the caller */
@@ -1318,17 +999,11 @@ class CallSession extends Emitter {
(!duration && callStatus !== CallStatus.Completed),
'duration MUST be supplied when call completed AND ONLY when call completed');
this.callInfo.updateCallStatus(callStatus, sipStatus, sipReason);
this.callInfo.updateCallStatus(callStatus, sipStatus);
if (typeof duration === 'number') this.callInfo.duration = duration;
const {span} = this.rootSpan.startChildSpan(`call-status:${this.callInfo.callStatus}`);
span.setAttributes(this.callInfo.toJSON());
try {
const b3 = this.b3;
const httpHeaders = b3 && {b3};
this.notifier.request('call:status', this.call_status_hook, this.callInfo.toJSON(), httpHeaders);
span.end();
this.notifier.request(this.call_status_hook, this.callInfo.toJSON());
} catch (err) {
span.end();
this.logger.info(err, `CallSession:_notifyCallStatusChange error sending ${callStatus} ${sipStatus}`);
}
@@ -1337,23 +1012,6 @@ class CallSession extends Emitter {
this.updateCallStatus(Object.assign({}, this.callInfo.toJSON()), this.serviceUrl)
.catch((err) => this.logger.error(err, 'redis error'));
}
_awaitCommandsOrHangup() {
assert(!this.wakeupResolver);
return new Promise((resolve, reject) => {
this.logger.info('_awaitCommandsOrHangup - waiting...');
this.wakeupResolver = resolve;
});
}
_clearTasks(evt) {
if (this.requestor instanceof WsRequestor) {
this.logger.info({evt}, 'CallSession:_clearTasks on event from background gather');
try {
this.kill();
} catch (err) {}
}
}
}
module.exports = CallSession;

View File

@@ -8,7 +8,7 @@ const CallSession = require('./call-session');
*/
class ConfirmCallSession extends CallSession {
constructor({logger, application, dlg, ep, tasks, callInfo, accountInfo, memberId, confName, rootSpan}) {
constructor({logger, application, dlg, ep, tasks, callInfo, accountInfo, memberId, confName}) {
super({
logger,
application,
@@ -18,8 +18,7 @@ class ConfirmCallSession extends CallSession {
callInfo,
accountInfo,
memberId,
confName,
rootSpan
confName
});
this.dlg = dlg;
this.ep = ep;
@@ -31,10 +30,6 @@ class ConfirmCallSession extends CallSession {
_clearResources() {
}
_callerHungup() {
}
}
module.exports = ConfirmCallSession;

View File

@@ -16,8 +16,7 @@ class InboundCallSession extends CallSession {
application: req.locals.application,
callInfo: req.locals.callInfo,
accountInfo: req.locals.accountInfo,
tasks: req.locals.application.tasks,
rootSpan: req.locals.rootSpan
tasks: req.locals.application.tasks
});
this.req = req;
this.res = res;
@@ -25,27 +24,17 @@ class InboundCallSession extends CallSession {
req.once('cancel', this._onCancel.bind(this));
this.on('callStatusChange', this._notifyCallStatusChange.bind(this));
this._notifyCallStatusChange({
callStatus: CallStatus.Trying,
sipStatus: 100,
sipReason: 'Trying'
});
this._notifyCallStatusChange({callStatus: CallStatus.Trying, sipStatus: 100});
}
_onCancel() {
this.rootSpan.setAttributes({'call.termination': 'caller abandoned'});
this._notifyCallStatusChange({
callStatus: CallStatus.NoAnswer,
sipStatus: 487,
sipReason: 'Request Terminated'
});
this._notifyCallStatusChange({callStatus: CallStatus.NoAnswer, sipStatus: 487});
this._callReleased();
}
_onTasksDone() {
if (!this.res.finalResponseSent) {
if (this._mediaServerFailure) {
this.rootSpan.setAttributes({'call.termination': 'media server failure'});
this.logger.info('InboundCallSession:_onTasksDone generating 480 due to media server failure');
this.res.send(480, {
headers: {
@@ -54,7 +43,6 @@ class InboundCallSession extends CallSession {
});
}
else {
this.rootSpan.setAttributes({'call.termination': 'tasks completed without answering call'});
this.logger.info('InboundCallSession:_onTasksDone auto-generating non-success response to invite');
this.res.send(603);
}
@@ -68,12 +56,8 @@ class InboundCallSession extends CallSession {
_callerHungup() {
assert(this.dlg.connectTime);
const duration = moment().diff(this.dlg.connectTime, 'seconds');
this.rootSpan.setAttributes({'call.termination': 'hangup by caller'});
this.emit('callStatusChange', {
callStatus: CallStatus.Completed,
duration
});
this.logger.info('InboundCallSession: caller hung up');
this.emit('callStatusChange', {callStatus: CallStatus.Completed, duration});
this.logger.debug('InboundCallSession: caller hung up');
this._callReleased();
this.req.removeAllListeners('cancel');
}

View File

@@ -8,7 +8,7 @@ const moment = require('moment');
* @extends CallSession
*/
class RestCallSession extends CallSession {
constructor({logger, application, srf, req, ep, tasks, callInfo, accountInfo, rootSpan}) {
constructor({logger, application, srf, req, ep, tasks, callInfo, accountInfo}) {
super({
logger,
application,
@@ -16,18 +16,13 @@ class RestCallSession extends CallSession {
callSid: callInfo.callSid,
tasks,
callInfo,
accountInfo,
rootSpan
accountInfo
});
this.req = req;
this.ep = ep;
this.on('callStatusChange', this._notifyCallStatusChange.bind(this));
this._notifyCallStatusChange({
callStatus: CallStatus.Trying,
sipStatus: 100,
sipReason: 'Trying'
});
this._notifyCallStatusChange({callStatus: CallStatus.Trying, sipStatus: 100});
}
/**

248
lib/tasks/cognigy.js Normal file
View File

@@ -0,0 +1,248 @@
const Task = require('./task');
const {TaskName, TaskPreconditions} = require('../utils/constants');
const makeTask = require('./make_task');
const { SocketClient } = require('@cognigy/socket-client');
const parseGallery = (obj = {}) => {
const {_default} = obj;
if (_default) {
const {_gallery} = _default;
if (_gallery) return _gallery.fallbackText;
}
};
const parseQuickReplies = (obj) => {
const {_default} = obj;
if (_default) {
const {_quickReplies} = _default;
if (_quickReplies) return _quickReplies.text || _quickReplies.fallbackText;
}
};
const parseBotText = (evt) => {
const {text, data} = evt;
if (text) return text;
switch (data?.type) {
case 'quickReplies':
return parseQuickReplies(data?._cognigy);
case 'gallery':
return parseGallery(data?._cognigy);
default:
break;
}
};
class Cognigy extends Task {
constructor(logger, opts) {
super(logger, opts);
this.preconditions = TaskPreconditions.Endpoint;
this.url = this.data.url;
this.token = this.data.token;
this.prompt = this.data.prompt;
this.eventHook = this.data?.eventHook;
this.actionHook = this.data?.actionHook;
this.data = this.data.data || {};
this.prompts = [];
}
get name() { return TaskName.Cognigy; }
get hasReportedFinalAction() {
return this.reportedFinalAction || this.isReplacingApplication;
}
async exec(cs, ep) {
await super.exec(cs);
this.ep = ep;
try {
/* set event handlers and start transcribing */
this.on('transcription', this._onTranscription.bind(this, cs, ep));
this.on('error', this._onError.bind(this, cs, ep));
this.transcribeTask = this._makeTranscribeTask();
this.transcribeTask.exec(cs, ep, this)
.catch((err) => {
this.logger.info({err}, 'Cognigy transcribe task returned error');
this.notifyTaskDone();
});
if (this.prompt) {
this.sayTask = this._makeSayTask(this.prompt);
this.sayTask.exec(cs, ep, this)
.catch((err) => {
this.logger.info({err}, 'Cognigy say task returned error');
this.notifyTaskDone();
});
}
/* connect to the bot and send initial data */
this.client = new SocketClient(
this.url,
this.token,
{
sessionId: cs.callSid,
channel: 'jambonz',
forceWebsockets: true,
reconnection: true,
settings: {
enableTypingIndicator: false
}
}
);
this.client.on('output', this._onBotUtterance.bind(this, cs, ep));
this.client.on('typingStatus', this._onBotTypingStatus.bind(this, cs, ep));
this.client.on('error', this._onBotError.bind(this, cs, ep));
this.client.on('finalPing', this._onBotFinalPing.bind(this, cs, ep));
await this.client.connect();
this.client.sendMessage('', {...this.data, ...cs.callInfo});
await this.awaitTaskDone();
} catch (err) {
this.logger.error({err}, 'Cognigy error');
throw err;
}
}
async kill(cs) {
super.kill(cs);
this.logger.debug('Cognigy:kill');
this.removeAllListeners();
this.transcribeTask && this.transcribeTask.kill();
this.client.removeAllListeners();
if (this.client && this.client.connected) this.client.disconnect();
if (!this.hasReportedFinalAction) {
this.reportedFinalAction = true;
this.performAction({cognigyResult: 'caller hungup'})
.catch((err) => this.logger.info({err}, 'cognigy - error w/ action webook'));
}
if (this.ep.connected) {
await this.ep.api('uuid_break', this.ep.uuid).catch((err) => this.logger.info(err, 'Error killing audio'));
}
this.notifyTaskDone();
}
_makeTranscribeTask() {
const opts = {
recognizer: this.data.recognizer || {
vendor: 'default',
language: 'default',
outputFormat: 'detailed'
}
};
this.logger.debug({opts}, 'constructing a nested transcribe object');
const transcribe = makeTask(this.logger, {transcribe: opts}, this);
return transcribe;
}
_makeSayTask(text) {
const opts = {
text,
synthesizer: this.data.synthesizer ||
{
vendor: 'default',
language: 'default',
voice: 'default'
}
};
this.logger.debug({opts}, 'constructing a nested say object');
const say = makeTask(this.logger, {say: opts}, this);
return say;
}
async _onBotError(cs, ep, evt) {
this.logger.info({evt}, 'Cognigy:_onBotError');
this.performAction({cognigyResult: 'botError', message: evt.message });
this.reportedFinalAction = true;
this.notifyTaskDone();
}
async _onBotTypingStatus(cs, ep, evt) {
this.logger.info({evt}, 'Cognigy:_onBotTypingStatus');
}
async _onBotFinalPing(cs, ep) {
this.logger.info('Cognigy:_onBotFinalPing');
if (this.prompts.length) {
const text = this.prompts.join('.');
this.prompts = [];
if (text && !this.killed) {
this.sayTask = this._makeSayTask(text);
this.sayTask.exec(cs, ep, this)
.catch((err) => {
this.logger.info({err}, 'Cognigy say task returned error');
this.notifyTaskDone();
});
}
}
}
async _onBotUtterance(cs, ep, evt) {
this.logger.debug({evt}, 'Cognigy:_onBotUtterance');
if (this.eventHook) {
this.performHook(cs, this.eventHook, {event: 'botMessage', message: evt})
.then((redirected) => {
if (redirected) {
this.logger.info('Cognigy_onTranscription: event handler for bot message redirected us to new webhook');
this.reportedFinalAction = true;
this.performAction({cognigyResult: 'redirect'}, false);
}
return;
})
.catch(({err}) => {
this.logger.info({err}, 'Cognigy_onTranscription: error sending event hook');
});
}
const text = parseBotText(evt);
this.prompts.push(text);
}
async _onTranscription(cs, ep, evt) {
this.logger.debug({evt}, `Cognigy: got transcription for callSid ${cs.callSid}`);
const utterance = evt.alternatives[0].transcript;
if (this.eventHook) {
this.performHook(cs, this.eventHook, {event: 'userMessage', message: utterance})
.then((redirected) => {
if (redirected) {
this.logger.info('Cognigy_onTranscription: event handler for user message redirected us to new webhook');
this.reportedFinalAction = true;
this.performAction({cognigyResult: 'redirect'}, false);
if (this.transcribeTask) this.transcribeTask.kill(cs);
}
return;
})
.catch(({err}) => {
this.logger.info({err}, 'Cognigy_onTranscription: error sending event hook');
});
}
/* send the user utterance to the bot */
try {
if (this.client && this.client.connected) {
this.client.sendMessage(utterance);
}
else {
this.logger.info('Cognigy_onTranscription - not sending user utterance as bot is disconnected');
}
} catch (err) {
this.logger.error({err}, 'Cognigy_onTranscription: Error sending user utterance to Cognigy - ending task');
this.performAction({cognigyResult: 'socketError'});
this.reportedFinalAction = true;
this.notifyTaskDone();
}
}
_onError(cs, ep, err) {
this.logger.debug({err}, 'Cognigy: got error');
if (!this.hasReportedFinalAction) this.performAction({cognigyResult: 'error', err});
this.reportedFinalAction = true;
this.notifyTaskDone();
}
}
module.exports = Cognigy;

View File

@@ -529,9 +529,7 @@ class Conference extends Task {
async _playHook(cs, dlg, hook, allowed = [TaskName.Play, TaskName.Say, TaskName.Pause]) {
assert(!this._playSession);
const b3 = this.getTracingPropagation();
const httpHeaders = b3 && {b3};
const json = await cs.application.requestor.request('verb:hook', hook, cs.callInfo, httpHeaders);
const json = await cs.application.requestor.request(hook, cs.callInfo);
const tasks = normalizeJambones(this.logger, json).map((tdata) => makeTask(this.logger, tdata));
const allowedTasks = tasks.filter((t) => allowed.includes(t.name));
@@ -584,14 +582,11 @@ class Conference extends Task {
_notifyConferenceEvent(cs, eventName, params = {}) {
if (this.statusEvents.includes(eventName)) {
const b3 = this.getTracingPropagation();
const httpHeaders = b3 && {b3};
params.event = eventName;
params.duration = (Date.now() - this.conferenceStartTime.getTime()) / 1000;
if (!params.time) params.time = (new Date()).toISOString();
if (!params.members && typeof this.participantCount === 'number') params.members = this.participantCount;
cs.application.requestor
.request('verb:hook', this.statusHook, Object.assign(params, this.statusParams, httpHeaders))
cs.application.requestor.request(this.statusHook, Object.assign(params, this.statusParams))
.catch((err) => this.logger.info(err, 'Conference:notifyConferenceEvent - error'));
}
}

View File

@@ -1,102 +0,0 @@
const Task = require('./task');
const {TaskName, TaskPreconditions} = require('../utils/constants');
class TaskConfig extends Task {
constructor(logger, opts) {
super(logger, opts);
this.preconditions = TaskPreconditions.Endpoint;
[
'synthesizer',
'recognizer',
'bargeIn'
].forEach((k) => this[k] = this.data[k] || {});
if (this.bargeIn.enable) {
this.gatherOpts = {
verb: 'gather',
timeout: 0,
bargein: true,
input: ['speech']
};
[
'finishOnKey', 'input', 'numDigits', 'minDigits', 'maxDigits',
'interDigitTimeout', 'bargein', 'dtmfBargein', 'minBargeinWordCount', 'actionHook'
].forEach((k) => {
if (this.bargeIn[k]) this.gatherOpts[k] = this.bargeIn[k];
});
}
if (this.bargeIn.sticky) this.autoEnable = true;
this.preconditions = this.bargeIn.enable ? TaskPreconditions.Endpoint : TaskPreconditions.None;
}
get name() { return TaskName.Config; }
get hasSynthesizer() { return Object.keys(this.synthesizer).length; }
get hasRecognizer() { return Object.keys(this.recognizer).length; }
get summary() {
const phrase = [];
if (this.bargeIn.enable) phrase.push('enable barge-in');
if (this.hasSynthesizer) {
const {vendor:v, language:l, voice} = this.synthesizer;
const s = `{${v},${l},${voice}}`;
phrase.push(`set synthesizer${s}`);
}
if (this.hasRecognizer) {
const {vendor:v, language:l} = this.recognizer;
const s = `{${v},${l}}`;
phrase.push(`set recognizer${s}`);
}
return `${this.name}{${phrase.join(',')}`;
}
async exec(cs) {
await super.exec(cs);
if (this.hasSynthesizer) {
cs.speechSynthesisVendor = this.synthesizer.vendor !== 'default'
? this.synthesizer.vendor
: cs.speechSynthesisVendor;
cs.speechSynthesisLanguage = this.synthesizer.language !== 'default'
? this.synthesizer.language
: cs.speechSynthesisLanguage;
cs.speechSynthesisVoice = this.synthesizer.voice !== 'default'
? this.synthesizer.voice
: cs.speechSynthesisVoice;
this.logger.info({synthesizer: this.synthesizer}, 'Config: updated synthesizer');
}
if (this.hasRecognizer) {
cs.speechRecognizerVendor = this.recognizer.vendor !== 'default'
? this.recognizer.vendor
: cs.speechRecognizerVendor;
cs.speechRecognizerLanguage = this.recognizer.language !== 'default'
? this.recognizer.language
: cs.speechRecognizerLanguage;
this.logger.info({recognizer: this.recognizer}, 'Config: updated recognizer');
}
if ('enable' in this.bargeIn) {
if (this.gatherOpts) {
this.gatherOpts.recognizer = this.hasRecognizer ?
this.recognizer :
{
vendor: cs.speechRecognizerVendor,
language: cs.speechRecognizerLanguage
};
this.logger.info({opts: this.gatherOpts}, 'Config: enabling bargeIn');
cs.enableBotMode(this.gatherOpts, this.autoEnable);
}
else {
this.logger.info('Config: disabling bargeIn');
cs.disableBotMode();
}
}
}
async kill(cs) {
super.kill(cs);
}
}
module.exports = TaskConfig;

View File

@@ -14,7 +14,6 @@ const sessionTracker = require('../session/session-tracker');
const DtmfCollector = require('../utils/dtmf-collector');
const dbUtils = require('../utils/db-utils');
const debug = require('debug')('jambonz:feature-server');
const {parseUri} = require('drachtio-srf');
function parseDtmfOptions(logger, dtmfCapture) {
let parentDtmfCollector, childDtmfCollector;
@@ -92,7 +91,6 @@ class TaskDial extends Task {
this.timeLimit = this.data.timeLimit;
this.confirmHook = this.data.confirmHook;
this.confirmMethod = this.data.confirmMethod;
this.referHook = this.data.referHook;
this.dtmfHook = this.data.dtmfHook;
this.proxy = this.data.proxy;
@@ -137,24 +135,6 @@ class TaskDial extends Task {
return !process.env.ANCHOR_MEDIA_ALWAYS && !this.listenTask && !this.transcribeTask;
}
get summary() {
if (this.target.length === 1) {
const target = this.target[0];
switch (target.type) {
case 'phone':
case 'teams':
return `${this.name}{type=${target.type},number=${target.number}}`;
case 'user':
return `${this.name}{type=${target.type},name=${target.name}}`;
case 'sip':
return `${this.name}{type=${target.type},sipUri=${target.sipUri}}`;
default:
return `${this.name}`;
}
}
else return `${this.name}{${this.target.length} targets}`;
}
async exec(cs) {
await super.exec(cs);
try {
@@ -226,11 +206,7 @@ class TaskDial extends Task {
this.logger.debug('Dial:whisper executing tasks');
while (tasks.length && !cs.callGone) {
const task = tasks.shift();
const {span, ctx} = this.startChildSpan(`whisper:${this.sayTask.summary}`);
task.span = span;
task.ctx = ctx;
await task.exec(cs, callSid === this.callSid ? this.ep : this.epOther);
span.end();
}
this.logger.debug('Dial:whisper tasks complete');
if (!cs.callGone && this.epOther) {
@@ -264,43 +240,6 @@ class TaskDial extends Task {
}
}
async handleRefer(cs, req, res, callInfo = cs.callInfo) {
if (this.referHook) {
try {
const isChild = !!callInfo.parentCallSid;
const referring_call_sid = isChild ? callInfo.callSid : cs.callSid;
const referred_call_sid = isChild ? callInfo.parentCallSid : this.sd.callSid;
const b3 = this.getTracingPropagation();
const httpHeaders = b3 && {b3};
const to = parseUri(req.getParsedHeader('Refer-To').uri);
const by = parseUri(req.getParsedHeader('Referred-By').uri);
this.logger.info({to}, 'refer to parsed');
await cs.requestor.request('verb:hook', this.referHook, {
...callInfo,
refer_details: {
sip_refer_to: req.get('Refer-To'),
sip_referred_by: req.get('Referred-By'),
sip_user_agent: req.get('User-Agent'),
refer_to_user: to.user,
referred_by_user: by.user,
referring_call_sid,
referred_call_sid
}
}, httpHeaders);
res.send(202);
this.logger.info('DialTask:handleRefer - sent 202 Accepted');
} catch (err) {
res.send(err.statusCode || 501);
}
}
else {
this.logger.info('DialTask:handleRefer - got REFER but no referHook, responding 501');
res.send(501);
}
}
_removeHandlers(sd) {
sd.removeAllListeners('accept');
sd.removeAllListeners('decline');
@@ -348,10 +287,8 @@ class TaskDial extends Task {
const key = arr[1];
const match = dtmfDetector.keyPress(key);
if (match) {
const b3 = this.getTracingPropagation();
const httpHeaders = b3 && {b3};
this.logger.info({callSid}, `Dial:_onInfo triggered dtmf match: ${match}`);
requestor.request('verb:hook', this.dtmfHook, {dtmf: match, ...callInfo.toJSON()}, httpHeaders)
requestor.request(this.dtmfHook, {dtmf: match, ...callInfo.toJSON()})
.catch((err) => this.logger.info(err, 'Dial:_onDtmf - error'));
}
}
@@ -406,11 +343,10 @@ class TaskDial extends Task {
this._killOutdials();
}, this.timeout * 1000);
this.span.setAttributes({'dial.target': JSON.stringify(this.target)});
this.target.forEach(async(t) => {
try {
t.confirmHook = t.confirmHook || this.confirmHook;
//t.method = t.method || this.confirmMethod || 'POST';
t.url = t.url || this.confirmUrl;
t.method = t.method || this.confirmMethod || 'POST';
if (t.type === 'teams') t.teamsInfo = teamsInfo;
if (t.type === 'user' && !t.name.includes('@') && !fqdn) {
const user = t.name;
@@ -443,14 +379,11 @@ class TaskDial extends Task {
target: t,
opts,
callInfo: cs.callInfo,
accountInfo: cs.accountInfo,
rootSpan: cs.rootSpan,
startSpan: this.startSpan.bind(this)
accountInfo: cs.accountInfo
});
this.dials.set(sd.callSid, sd);
sd
.on('refer', (callInfo, req, res) => this.handleRefer(cs, req, res, callInfo))
.on('callCreateFail', () => {
clearTimeout(this.timerRing);
this.dials.delete(sd.callSid);
@@ -519,9 +452,6 @@ class TaskDial extends Task {
} catch (err) {
this.logger.error(err, 'Error in dial einvite from B leg');
}
})
.on('refer', (callInfo, req, res) => {
})
.once('adulting', () => {
/* child call just adulted and got its own session */

View File

@@ -295,9 +295,9 @@ class Dialogflow extends Task {
}
// if a final transcription, start a typing sound
if (this.thinkingMusic && !transcription.isEmpty && transcription.isFinal &&
if (this.thinkingSound > 0 && !transcription.isEmpty && transcription.isFinal &&
transcription.confidence > 0.8) {
ep.play(this.data.thinkingMusic).catch((err) => this.logger.info(err, 'Error playing typing sound'));
ep.play(this.data.thinkingSound).catch((err) => this.logger.info(err, 'Error playing typing sound'));
}
// interrupt playback on speaking if bargein = true
@@ -405,8 +405,8 @@ class Dialogflow extends Task {
this.dtmfEntry = dtmfEntry;
this.digitBuffer = null;
// if a final transcription, start a typing sound
if (this.thinkingMusic) {
ep.play(this.thinkingMusic).catch((err) => this.logger.info(err, 'Error playing typing sound'));
if (this.thinkingSound > 0) {
ep.play(this.thinkingSound).catch((err) => this.logger.info(err, 'Error playing typing sound'));
}
// kill the current dialogflow, which will result in us getting an immediate intent
@@ -453,10 +453,7 @@ class Dialogflow extends Task {
}
async _performHook(cs, hook, results = {}) {
const b3 = this.getTracingPropagation();
const httpHeaders = b3 && {b3};
const json = await this.cs.requestor.request('verb:hook', hook,
{...results, ...cs.callInfo.toJSON()}, httpHeaders);
const json = await this.cs.requestor.request(hook, {...results, ...cs.callInfo.toJSON()});
if (json && Array.isArray(json)) {
const makeTask = require('../make_task');
const tasks = normalizeJambones(this.logger, json).map((tdata) => makeTask(this.logger, tdata));

View File

@@ -302,8 +302,6 @@ class TaskEnqueue extends Task {
async _playHook(cs, dlg, hook, allowed = [TaskName.Play, TaskName.Say, TaskName.Pause, TaskName.Leave]) {
const {lengthOfList, getListPosition} = cs.srf.locals.dbHelpers;
const b3 = this.getTracingPropagation();
const httpHeaders = b3 && {b3};
assert(!this._playSession);
if (this.killed) return [];
@@ -319,7 +317,7 @@ class TaskEnqueue extends Task {
} catch (err) {
this.logger.error({err}, `TaskEnqueue:_playHook error retrieving list info for queue ${this.queueName}`);
}
const json = await cs.application.requestor.request('verb:hook', hook, params, httpHeaders);
const json = await cs.application.requestor.request(hook, params);
const tasks = normalizeJambones(this.logger, json).map((tdata) => makeTask(this.logger, tdata));
const allowedTasks = tasks.filter((t) => allowed.includes(t.name));

View File

@@ -9,7 +9,8 @@ const {
const makeTask = require('./make_task');
const assert = require('assert');
//const GATHER_STABILITY_THRESHOLD = Number(process.env.JAMBONZ_GATHER_STABILITY_THRESHOLD || 0.7);
const GATHER_STABILITY_THRESHOLD = Number(process.env.JAMBONZ_GATHER_STABILITY_THRESHOLD || 0.7);
class TaskGather extends Task {
constructor(logger, opts, parentTask) {
@@ -18,30 +19,22 @@ class TaskGather extends Task {
[
'finishOnKey', 'hints', 'input', 'numDigits', 'minDigits', 'maxDigits',
'interDigitTimeout', 'partialResultHook', 'bargein', 'dtmfBargein',
'interDigitTimeout', 'submitDigit', 'partialResultHook', 'bargein', 'dtmfBargein',
'retries', 'retryPromptTts', 'retryPromptUrl',
'speechTimeout', 'timeout', 'say', 'play'
].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;
/* timeout of zero means no timeout */
this.timeout = this.timeout === 0 ? 0 : (this.timeout || 15) * 1000;
this.interim = this.partialResultHook || this.bargein;
this.listenDuringPrompt = this.data.listenDuringPrompt === false ? false : true;
this.minBargeinWordCount = this.data.minBargeinWordCount || 0;
this.minBargeinWordCount = this.data.minBargeinWordCount || 1;
this.logger.debug({opts}, 'created gather task');
this.timeout = (this.timeout || 15) * 1000;
this.interim = this.partialResultCallback || this.bargein;
if (this.data.recognizer) {
const recognizer = this.data.recognizer;
this.vendor = recognizer.vendor;
this.language = recognizer.language;
this.hints = recognizer.hints || [];
this.hintsBoost = recognizer.hintsBoost;
this.altLanguages = recognizer.altLanguages || [];
this.punctuation = !!recognizer.punctuation;
/* vad: if provided, we dont connect to recognizer until voice activity is detected */
const {enable, voiceMs = 0, mode = -1} = recognizer.vad || {};
this.vad = {enable, voiceMs, mode};
/* aws options */
this.vocabularyName = recognizer.vocabularyName;
@@ -53,19 +46,19 @@ class TaskGather extends Task {
this.profanityOption = recognizer.profanityOption || 'raw';
this.requestSnr = recognizer.requestSnr || false;
this.initialSpeechTimeoutMs = recognizer.initialSpeechTimeoutMs || 0;
this.azureServiceEndpoint = recognizer.azureServiceEndpoint;
}
this.digitBuffer = '';
this._earlyMedia = this.data.earlyMedia === true;
if (this.say) {
this.sayTask = makeTask(this.logger, {say: this.say}, this);
if (this.say) this.sayTask = makeTask(this.logger, {say: this.say}, this);
if (this.play) this.playTask = makeTask(this.logger, {play: this.play}, this);
if (this.sayTask || this.playTask) {
// this is specially for barge in where we want to make a bargebale promt
// to a user without listening after the say task has finished
this.listenAfterSpeech = typeof this.data.listenAfterSpeech === 'boolean' ? this.data.listenAfterSpeech : true;
}
if (this.play) {
this.playTask = makeTask(this.logger, {play: this.play}, this);
}
if (!this.sayTask && !this.playTask) this.listenDuringPrompt = false;
this.parentTask = parentTask;
}
@@ -79,23 +72,7 @@ class TaskGather extends Task {
(this.playTask && this.playTask.earlyMedia);
}
get summary() {
let s = `${this.name}{`;
if (this.input.length === 2) s += 'inputs=[speech,digits],';
else if (this.input.includes('digits')) s += 'inputs=digits';
else s += 'inputs=speech,';
if (this.input.includes('speech')) {
s += `vendor=${this.vendor || 'default'},language=${this.language || 'default'}`;
}
if (this.sayTask) s += ',with nested say task';
if (this.playTask) s += ',with nested play task';
s += '}';
return s;
}
async exec(cs, ep) {
this.logger.debug('Gather:exec');
await super.exec(cs);
const {updateSpeechCredentialLastUsed} = require('../utils/db-utils')(this.logger, cs.srf);
@@ -129,28 +106,33 @@ class TaskGather extends Task {
try {
if (this.sayTask) {
const {span, ctx} = this.startChildSpan(`nested:${this.sayTask.summary}`);
this.sayTask.span = span;
this.sayTask.ctx = ctx;
this.sayTask.exec(cs, ep); // kicked off, _not_ waiting for it to complete
this.sayTask.on('playDone', (err) => {
span.end();
if (err) this.logger.error({err}, 'Gather:exec Error playing tts');
this.logger.debug('Gather: nested say task completed');
if (!this.killed) startListening(cs, ep);
this.logger.debug('Gather: kicking off say task');
this.sayTask.exec(cs, ep);
this.sayTask.on('playDone', async(err) => {
if (err) return this.logger.error({err}, 'Gather:exec Error playing tts');
this.logger.debug('Gather: say task completed');
if (!this.killed) {
if (this.listenAfterSpeech === true) {
startListening(cs, ep);
} else {
this.notifyTaskDone();
}
}
});
}
else if (this.playTask) {
const {span, ctx} = this.startChildSpan(`nested:${this.playTask.summary}`);
this.playTask.span = span;
this.playTask.ctx = ctx;
this.playTask.exec(cs, ep); // kicked off, _not_ waiting for it to complete
this.playTask.on('playDone', (err) => {
span.end();
if (err) this.logger.error({err}, 'Gather:exec Error playing url');
this.logger.debug('Gather: nested play task completed');
if (!this.killed) startListening(cs, ep);
});
this.playTask.on('playDone', async(err) => {
if (err) return this.logger.error({err}, 'Gather:exec Error playing url');
if (!this.killed) {
if (this.listenAfterSpeech === true) {
startListening(cs, ep);
} else {
this.notifyTaskDone();
}
}
}
);
}
else startListening(cs, ep);
@@ -166,41 +148,31 @@ class TaskGather extends Task {
}
await this.awaitTaskDone();
this.logger.debug('Gather:exec task has completed');
} catch (err) {
this.logger.error(err, 'TaskGather:exec error');
}
ep.removeCustomEventListener(GoogleTranscriptionEvents.Transcription);
ep.removeCustomEventListener(GoogleTranscriptionEvents.EndOfUtterance);
ep.removeCustomEventListener(GoogleTranscriptionEvents.VadDetected);
ep.removeCustomEventListener(AwsTranscriptionEvents.Transcription);
ep.removeCustomEventListener(AwsTranscriptionEvents.VadDetected);
ep.removeCustomEventListener(AzureTranscriptionEvents.Transcription);
ep.removeCustomEventListener(AzureTranscriptionEvents.NoSpeechDetected);
ep.removeCustomEventListener(AzureTranscriptionEvents.VadDetected);
}
kill(cs) {
this.logger.debug('Gather:kill');
super.kill(cs);
this._killAudio(cs);
this.ep.removeAllListeners('dtmf');
clearTimeout(this.interDigitTimer);
this.playTask?.span.end();
this.sayTask?.span.end();
this._resolve('killed');
}
updateTimeout(timeout) {
this.logger.info(`TaskGather:updateTimeout - updating timeout to ${timeout}`);
this.timeout = timeout;
this._startTimer();
}
_onDtmf(cs, ep, evt) {
this.logger.debug(evt, 'TaskGather:_onDtmf');
clearTimeout(this.interDigitTimer);
let resolved = false;
if (this.dtmfBargein) this._killAudio(cs);
if (evt.dtmf === this.finishOnKey && this.input.includes('digits')) {
if (evt.dtmf === this.finishOnKey) {
resolved = true;
this._resolve('dtmf-terminator-key');
}
@@ -212,6 +184,7 @@ class TaskGather extends Task {
this._resolve('dtmf-num-digits');
}
}
if (!resolved && this.interDigitTimeout > 0 && this.digitBuffer.length >= this.minDigits) {
/* start interDigitTimer */
const ms = this.interDigitTimeout * 1000;
@@ -223,28 +196,17 @@ class TaskGather extends Task {
async _initSpeech(cs, ep) {
const opts = {};
if (this.vad?.enable) {
opts.START_RECOGNIZING_ON_VAD = 1;
if (this.vad.voiceMs) opts.RECOGNIZER_VAD_VOICE_MS = this.vad.voiceMs;
else opts.RECOGNIZER_VAD_VOICE_MS = 125;
if (this.vad.mode >= 0 && this.vad.mode <= 3) opts.RECOGNIZER_VAD_MODE = this.vad.mode;
}
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
GOOGLE_SPEECH_MODEL: 'command_and_search'
});
if (this.hints && this.hints.length > 1) {
opts.GOOGLE_SPEECH_HINTS = this.hints.map((h) => h.trim()).join(',');
if (typeof this.hintsBoost === 'number') {
opts.GOOGLE_SPEECH_HINTS_BOOST = this.hintsBoost;
}
}
if (this.altLanguages && this.altLanguages.length > 0) {
if (this.altLanguages && this.altLanguages.length > 1) {
opts.GOOGLE_SPEECH_ALTERNATIVE_LANGUAGE_CODES = this.altLanguages.join(',');
}
if (this.profanityFilter === true) {
@@ -252,7 +214,6 @@ class TaskGather extends Task {
}
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));
}
else if (['aws', 'polly'].includes(this.vendor)) {
if (this.vocabularyName) opts.AWS_VOCABULARY_NAME = this.vocabularyName;
@@ -268,7 +229,6 @@ class TaskGather extends Task {
});
}
ep.addCustomEventListener(AwsTranscriptionEvents.Transcription, this._onTranscription.bind(this, cs, ep));
ep.addCustomEventListener(AwsTranscriptionEvents.VadDetected, this._onVadDetected.bind(this, cs, ep));
}
else if ('microsoft' === this.vendor) {
if (this.sttCredentials) {
@@ -280,27 +240,19 @@ class TaskGather extends Task {
if (this.hints && this.hints.length > 1) {
opts.AZURE_SPEECH_HINTS = this.hints.map((h) => h.trim()).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;
//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;
else if (this.timeout === 0) opts.AZURE_INITIAL_SPEECH_TIMEOUT_MS = 120000; // lengthy
opts.AZURE_USE_OUTPUT_FORMAT_DETAILED = 1;
ep.addCustomEventListener(AzureTranscriptionEvents.Transcription, this._onTranscription.bind(this, cs, ep));
ep.addCustomEventListener(AzureTranscriptionEvents.NoSpeechDetected, this._onNoSpeechDetected.bind(this, cs, ep));
ep.addCustomEventListener(AzureTranscriptionEvents.VadDetected, this._onVadDetected.bind(this, cs, ep));
}
await ep.set(opts)
.catch((err) => this.logger.info(err, 'Error setting channel variables'));
}
_startTranscribing(ep) {
this.logger.debug({
vendor: this.vendor,
locale: this.language,
interim: this.interim
}, 'Gather:_startTranscribing');
ep.startTranscription({
vendor: this.vendor,
locale: this.language,
@@ -318,11 +270,6 @@ 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);
@@ -336,15 +283,6 @@ class TaskGather extends Task {
}
_killAudio(cs) {
if (!this.sayTask && !this.playTask && this.bargein) {
if (this.ep?.connected && !this.playComplete) {
this.logger.debug('Gather:_killAudio: killing playback of any audio');
this.playComplete = true;
this.ep.api('uuid_break', this.ep.uuid)
.catch((err) => this.logger.info(err, 'Error killing audio'));
}
return;
}
if (this.sayTask && !this.sayTask.killed) {
this.sayTask.removeAllListeners('playDone');
this.sayTask.kill(cs);
@@ -358,12 +296,11 @@ class TaskGather extends Task {
}
_onTranscription(cs, ep, evt) {
this.logger.debug(evt, 'TaskGather:_onTranscription');
if ('aws' === this.vendor && Array.isArray(evt) && evt.length > 0) evt = evt[0];
if ('microsoft' === this.vendor) {
const final = evt.RecognitionStatus === 'Success';
if (final) {
// 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;
evt = {
is_final: true,
@@ -386,102 +323,64 @@ class TaskGather extends Task {
};
}
}
if (evt.is_final) {
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 (evt.is_final) this._resolve('speech', evt);
else {
/* google has a measure of stability:
https://cloud.google.com/speech-to-text/docs/basics#streaming_responses
others do not.
*/
//const isStableEnough = typeof evt.stability === 'undefined' || evt.stability > GATHER_STABILITY_THRESHOLD;
const recognizeSuccess = evt.stability > GATHER_STABILITY_THRESHOLD;
/*
we need to make sure to only send something on barge in if we have
something valid therefore we need to check the recognition
stability, which applies to GOOGLE
for MS we will have a final event, meaning we will not run into
the current if else branch.
if (this.bargein && /* isStableEnough && */
For AWS we still need more testing
*/
if (recognizeSuccess &&
this.bargein &&
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.logger.debug('Gather:_onTranscription - killing audio due to bargein');
this._killAudio(cs);
this._resolve('speech', evt);
}
if (this.partialResultHook) {
const b3 = this.getTracingPropagation();
const httpHeaders = b3 && {b3};
this.cs.requestor.request(this.partialResultHook, Object.assign({speech: evt},
this.cs.callInfo, httpHeaders));
this.cs.requestor.request(this.partialResultHook, Object.assign({speech: evt}, this.cs.callInfo))
.catch((err) => this.logger.info(err, 'GatherTask:_onTranscription error'));
}
}
}
_onEndOfUtterance(cs, ep) {
this.logger.debug('TaskGather:_onEndOfUtterance');
if (this.bargein && this.minBargeinWordCount === 0) {
this._killAudio(cs);
}
this.logger.info('TaskGather:_onEndOfUtterance');
if (!this.resolved && !this.killed) {
this._startTranscribing(ep);
}
}
_onVadDetected(cs, ep) {
if (this.bargein && this.minBargeinWordCount === 0) {
this.logger.debug('TaskGather:_onVadDetected');
this._killAudio(cs);
this.emit('vad');
}
}
_onNoSpeechDetected(cs, ep) {
if (!this.callSession.callGone && !this.killed) {
this.logger.debug('TaskGather:_onNoSpeechDetected - listen again');
return this._startTranscribing(ep);
}
this._resolve('timeout');
}
async _resolve(reason, evt) {
this.logger.debug(`TaskGather:resolve with reason ${reason}`);
if (this.resolved) return;
this.resolved = true;
clearTimeout(this.interDigitTimer);
this.logger.debug(`TaskGather:resolve with reason ${reason}`);
this.span.setAttributes({'stt.resolve': reason, 'stt.result': JSON.stringify(evt)});
if (this.ep && this.ep.connected) {
this.ep.stopTranscription({vendor: this.vendor})
.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'});
}
if (this.parentTask) this.parentTask.emit('dtmf-collected', {reason, digits: this.digitBuffer});
else 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 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 await this.performAction({reason: 'timeout'});
}
this.notifyTaskDone();
}

View File

@@ -289,9 +289,7 @@ class Lex extends Task {
}
async _performHook(cs, hook, results) {
const b3 = this.getTracingPropagation();
const httpHeaders = b3 && {b3};
const json = await this.cs.requestor.request('verb:hook', hook, results, httpHeaders);
const json = await this.cs.requestor.request(hook, results);
if (json && Array.isArray(json)) {
const makeTask = require('./make_task');
const tasks = normalizeJambones(this.logger, json).map((tdata) => makeTask(this.logger, tdata));

View File

@@ -38,12 +38,7 @@ class TaskListen extends Task {
if (this.playBeep) await this._playBeep(ep);
if (this.transcribeTask) {
this.logger.debug('TaskListen:exec - starting nested transcribe task');
const {span, ctx} = this.startChildSpan(`nested:${this.transcribeTask.summary}`);
this.transcribeTask.span = span;
this.transcribeTask.ctx = ctx;
this.transcribeTask.exec(cs, ep)
.then((result) => span.end())
.catch((err) => span.end());
this.transcribeTask.exec(cs, ep);
}
await this._startListening(cs, ep);
await this.awaitTaskDone();

View File

@@ -20,9 +20,9 @@ function makeTask(logger, obj, parent) {
case TaskName.SipRefer:
const TaskSipRefer = require('./sip_refer');
return new TaskSipRefer(logger, data, parent);
case TaskName.Config:
const TaskConfig = require('./config');
return new TaskConfig(logger, data, parent);
case TaskName.Cognigy:
const TaskCognigy = require('./cognigy');
return new TaskCognigy(logger, data, parent);
case TaskName.Conference:
const TaskConference = require('./conference');
return new TaskConference(logger, data, parent);

View File

@@ -13,10 +13,6 @@ class TaskPlay extends Task {
get name() { return TaskName.Play; }
get summary() {
return `${this.name}:{url=${this.url}}`;
}
async exec(cs, ep) {
await super.exec(cs);
this.ep = ep;

View File

@@ -31,15 +31,8 @@ class Rasa extends Task {
/* start the first gather */
this.gatherTask = this._makeGatherTask(this.prompt);
const {span, ctx} = this.startChildSpan(`nested:${this.gatherTask.summary}`);
this.gatherTask.span = span;
this.gatherTask.ctx = ctx;
this.gatherTask.exec(cs, ep, this)
.then(() => span.end())
.catch((err) => {
span.end();
this.logger.info({err}, 'Rasa gather task returned error');
});
.catch((err) => this.logger.info({err}, 'Rasa gather task returned error'));
await this.awaitTaskDone();
} catch (err) {
@@ -125,15 +118,8 @@ class Rasa extends Task {
if (botUtterance) {
this.logger.debug({botUtterance}, 'Rasa:_onTranscription: got user utterance');
this.gatherTask = this._makeGatherTask(botUtterance);
const {span, ctx} = this.startChildSpan(`nested:${this.gatherTask.summary}`);
this.gatherTask.span = span;
this.gatherTask.ctx = ctx;
this.gatherTask.exec(cs, ep, this)
.then(() => span.end())
.catch((err) => {
span.end();
this.logger.info({err}, 'Rasa gather task returned error');
});
.catch((err) => this.logger.info({err}, 'Rasa gather task returned error'));
if (this.eventHook) {
this.performHook(cs, this.eventHook, {event: 'botMessage', message: response})
.then((redirected) => {

View File

@@ -48,9 +48,7 @@ class TaskRestDial extends Task {
cs.setDialog(dlg);
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(this.call_hook, cs.callInfo);
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

@@ -14,14 +14,6 @@ class TaskSay extends Task {
get name() { return TaskName.Say; }
get summary() {
for (let i = 0; i < this.text.length; i++) {
if (this.text[i].startsWith('silence_stream')) continue;
return `${this.name}{text=${this.text[i].slice(0, 15)}${this.text[i].length > 15 ? '...' : ''}}`;
}
return `${this.name}{${this.text[0]}}`;
}
async exec(cs, ep) {
await super.exec(cs);
@@ -29,20 +21,26 @@ class TaskSay extends Task {
const {updateSpeechCredentialLastUsed} = require('../utils/db-utils')(this.logger, srf);
const {writeAlerts, AlertType, stats} = srf.locals;
const {synthAudio} = srf.locals.dbHelpers;
const vendor = this.synthesizer.vendor && this.synthesizer.vendor !== 'default' ?
this.synthesizer.vendor :
cs.speechSynthesisVendor;
const language = this.synthesizer.language && this.synthesizer.language !== 'default' ?
this.synthesizer.language :
cs.speechSynthesisLanguage ;
const voice = this.synthesizer.voice && this.synthesizer.voice !== 'default' ?
this.synthesizer.voice :
cs.speechSynthesisVoice;
const vendor = this.synthesizer.vendor && this.synthesizer.vendor !== 'default'
? this.synthesizer.vendor
: cs.speechSynthesisVendor;
const language = this.synthesizer.language && this.synthesizer.language !== 'default'
? this.synthesizer.language
: cs.speechSynthesisLanguage ;
const voice = this.synthesizer.voice && this.synthesizer.voice !== 'default'
? this.synthesizer.voice
: cs.speechSynthesisVoice;
const engine = this.synthesizer.engine || 'standard';
const salt = cs.callSid;
const credentials = cs.getSpeechCredentials(vendor, 'tts');
this.logger.info({vendor, language, voice}, 'TaskSay:exec');
this.logger.debug({language,
voice,
localSynthesizer: this.synthesizer,
speechSynthesisVendor: cs.speechSynthesisVendor,
speechSynthesisLanguage: cs.speechSynthesisLanguage,
speechSynthesisVoice: cs.speechSynthesisVoice
}, `Task:say - using vendor: ${vendor}`);
this.ep = ep;
try {
if (!credentials) {
@@ -55,18 +53,7 @@ class TaskSay extends Task {
}
// 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) => {
if (this.killed) return;
if (text.startsWith('silence_stream://')) return text;
const {filePath, servedFromCache} = await synthAudio(stats, {
text,
vendor,
@@ -91,13 +78,12 @@ class TaskSay extends Task {
updateSpeechCredentialLastUsed(credentials.speech_credential_sid)
.catch(() => {/*already logged error */});
}
this.ttsSpan.setAttributes({'tts.cached': servedFromCache});
return filePath;
}))).filter((fp) => fp && fp.length);
this.ttsSpan?.end();
this.logger.debug({filepath}, 'synthesized files for tts');
while (!this.killed && (this.loop === 'forever' || this.loop--) && this.ep?.connected) {
while (!this.killed && (this.loop === 'forever' || this.loop--) && this.ep.connected) {
let segment = 0;
do {
if (cs.isInConference) {
@@ -112,7 +98,6 @@ class TaskSay extends Task {
} while (!this.killed && ++segment < filepath.length);
}
} catch (err) {
this.ttsSpan?.end();
this.logger.info(err, 'TaskSay:exec error');
}
this.emit('playDone');

View File

@@ -19,11 +19,7 @@ class TaskSipDecline extends Task {
res.send(this.data.status, this.data.reason, {
headers: this.headers
});
cs.emit('callStatusChange', {
callStatus: CallStatus.Failed,
sipStatus: this.data.status,
sipReason: this.data.reason
});
cs.emit('callStatusChange', {callStatus: CallStatus.Failed, sipStatus: this.data.status});
}
}

View File

@@ -26,12 +26,6 @@ class TaskSipRefer extends Task {
try {
this.notifyHandler = this._handleNotify.bind(this, cs, dlg);
dlg.on('notify', this.notifyHandler);
/* otel: trace time for tts */
this.referSpan = this.startSpan('send-refer', {
'refer.refer_to': referTo,
'refer.referred_by': referredBy
});
const response = await dlg.request({
method: 'REFER',
headers: {
@@ -41,27 +35,22 @@ class TaskSipRefer extends Task {
}
});
this.referStatus = response.status;
this.referSpan.setAttributes({'refer.status_code': response.status});
this.logger.info(`TaskSipRefer:exec - received ${this.referStatus} to REFER`);
/* if we fail, fall through to next verb. If success, we should get BYE from far end */
if (this.referStatus === 202) {
await this.awaitTaskDone();
}
else {
await this.performAction({refer_status: this.referStatus});
}
else await this.performAction({refer_status: this.referStatus});
} catch (err) {
this.logger.info({err}, 'TaskSipRefer:exec - error sending REFER');
}
this.referSpan?.end();
}
async kill(cs) {
super.kill(cs);
const {dlg} = cs;
dlg.off('notify', this.notifyHandler);
this.notifyTaskDone();
}
async _handleNotify(cs, dlg, req, res) {
@@ -76,13 +65,9 @@ class TaskSipRefer extends Task {
const status = arr[1];
this.logger.debug(`TaskSipRefer:_handleNotify: call got status ${status}`);
if (this.eventHook) {
const b3 = this.getTracingPropagation();
const httpHeaders = b3 && {b3};
await cs.requestor.request('verb:hook', this.eventHook,
{event: 'transfer-status', call_status: status}, httpHeaders);
await cs.requestor.request(this.eventHook, {event: 'transfer-status', call_status: status});
}
if (status >= 200) {
this.referSpan.setAttributes({'refer.finalNotify': status});
await this.performAction({refer_status: 202, final_referred_call_status: status});
this.notifyTaskDone();
}

View File

@@ -21,30 +21,20 @@
"referTo"
]
},
"config": {
"cognigy": {
"properties": {
"synthesizer": "#synthesizer",
"url": "string",
"token": "string",
"recognizer": "#recognizer",
"bargeIn": "#bargeIn"
},
"required": []
},
"bargeIn": {
"properties": {
"enable": "boolean",
"sticky": "boolean",
"tts": "#synthesizer",
"prompt": "string",
"actionHook": "object|string",
"input": "array",
"finishOnKey": "string",
"numDigits": "number",
"minDigits": "number",
"maxDigits": "number",
"interDigitTimeout": "number",
"dtmfBargein": "boolean",
"minBargeinWordCount": "number"
"eventHook": "object|string",
"data": "object"
},
"required": [
"enable"
"url",
"token"
]
},
"dequeue": {
@@ -108,16 +98,17 @@
"finishOnKey": "string",
"input": "array",
"numDigits": "number",
"minDigits": "number",
"maxDigits": "number",
"interDigitTimeout": "number",
"partialResultHook": "object|string",
"speechTimeout": "number",
"listenDuringPrompt": "boolean",
"dtmfBargein": "boolean",
"bargein": "boolean",
"minBargeinWordCount": "number",
"dtmfBargein": "boolean",
"minDigits": "number",
"maxDigits": "number",
"interDigitTimeout": "number",
"timeout": "number",
"listenAfterSpeech": "boolean",
"recognizer": "#recognizer",
"play": "#play",
"say": "#say"
@@ -150,7 +141,6 @@
"answerOnBridge": "boolean",
"callerId": "string",
"confirmHook": "object|string",
"referHook": "object|string",
"dialMusic": "string",
"dtmfCapture": "object",
"dtmfHook": "object|string",
@@ -410,7 +400,6 @@
"language": "string",
"vad": "#vad",
"hints": "array",
"hintsBoost": "number",
"altLanguages": "array",
"profanityFilter": "boolean",
"interim": "boolean",
@@ -464,8 +453,7 @@
]
},
"requestSnr": "boolean",
"initialSpeechTimeoutMs": "number",
"azureServiceEndpoint": "string"
"initialSpeechTimeoutMs": "number"
},
"required": [
"vendor"

View File

@@ -4,7 +4,6 @@ const debug = require('debug')('jambonz:feature-server');
const assert = require('assert');
const {TaskPreconditions} = require('../utils/constants');
const normalizeJambones = require('../utils/normalize-jambones');
const {trace} = require('@opentelemetry/api');
const specs = new Map();
const _specData = require('./specs');
for (const key in _specData) {specs.set(key, _specData[key]);}
@@ -43,10 +42,6 @@ class Task extends Emitter {
return this.cs;
}
get summary() {
return this.name;
}
toJSON() {
return this.data;
}
@@ -72,34 +67,6 @@ class Task extends Emitter {
setImmediate(() => this.parentTask = null);
}
startSpan(name, attributes) {
const {srf} = require('../..');
const {tracer} = srf.locals.otel;
const span = tracer.startSpan(name, undefined, this.ctx);
if (attributes) span.setAttributes(attributes);
trace.setSpan(this.ctx, span);
return span;
}
startChildSpan(name, attributes) {
const {srf} = require('../..');
const {tracer} = srf.locals.otel;
const span = tracer.startSpan(name, undefined, this.ctx);
if (attributes) span.setAttributes(attributes);
const ctx = trace.setSpan(this.ctx, span);
return {span, ctx};
}
getTracingPropagation(encoding, span) {
// TODO: support encodings beyond b3 https://github.com/openzipkin/b3-propagation
if (span) {
return `${span.spanContext().traceId}-${span.spanContext().spanId}-1`;
}
if (this.span) {
return `${this.span.spanContext().traceId}-${this.span.spanContext().spanId}-1`;
}
}
/**
* when a subclass Task has completed its work, it should call this method
*/
@@ -140,53 +107,29 @@ class Task extends Emitter {
async performAction(results, expectResponse = true) {
if (this.actionHook) {
const params = results ? Object.assign(results, this.cs.callInfo.toJSON()) : this.cs.callInfo.toJSON();
const span = this.startSpan('verb:hook', {'hook.url': this.actionHook});
const b3 = this.getTracingPropagation('b3', span);
const httpHeaders = b3 && {b3};
span.setAttributes({'http.body': JSON.stringify(params)});
try {
const json = await this.cs.requestor.request('verb:hook', this.actionHook, params, httpHeaders);
span.setAttributes({'http.statusCode': 200});
span.end();
if (expectResponse && json && Array.isArray(json)) {
const makeTask = require('./make_task');
const tasks = normalizeJambones(this.logger, json).map((tdata) => makeTask(this.logger, tdata));
if (tasks && tasks.length > 0) {
this.logger.info({tasks: tasks}, `${this.name} replacing application with ${tasks.length} tasks`);
this.callSession.replaceApplication(tasks);
}
const json = await this.cs.requestor.request(this.actionHook, params);
if (expectResponse && json && Array.isArray(json)) {
const makeTask = require('./make_task');
const tasks = normalizeJambones(this.logger, json).map((tdata) => makeTask(this.logger, tdata));
if (tasks && tasks.length > 0) {
this.logger.info({tasks: tasks}, `${this.name} replacing application with ${tasks.length} tasks`);
this.callSession.replaceApplication(tasks);
}
} catch (err) {
span.setAttributes({'http.statusCode': err.statusCode});
span.end();
throw err;
}
}
}
async performHook(cs, hook, results) {
const span = this.startSpan('verb:hook', {'hook.url': hook});
const b3 = this.getTracingPropagation('b3', span);
const httpHeaders = b3 && {b3};
span.setAttributes({'http.body': JSON.stringify(results)});
try {
const json = await cs.requestor.request('verb:hook', hook, results, httpHeaders);
span.setAttributes({'http.statusCode': 200});
span.end();
if (json && Array.isArray(json)) {
const makeTask = require('./make_task');
const tasks = normalizeJambones(this.logger, json).map((tdata) => makeTask(this.logger, tdata));
if (tasks && tasks.length > 0) {
this.redirect(cs, tasks);
return true;
}
const json = await cs.requestor.request(hook, results);
if (json && Array.isArray(json)) {
const makeTask = require('./make_task');
const tasks = normalizeJambones(this.logger, json).map((tdata) => makeTask(this.logger, tdata));
if (tasks && tasks.length > 0) {
this.redirect(cs, tasks);
return true;
}
return false;
} catch (err) {
span.setAttributes({'http.statusCode': err.statusCode});
span.end();
throw err;
}
return false;
}
redirect(cs, tasks) {

View File

@@ -28,7 +28,6 @@ class TaskTranscribe extends Task {
/* google-specific options */
this.hints = recognizer.hints || [];
this.hintsBoost = recognizer.hintsBoost;
this.profanityFilter = recognizer.profanityFilter;
this.punctuation = !!recognizer.punctuation;
this.enhancedModel = !!recognizer.enhancedModel;
@@ -51,7 +50,6 @@ class TaskTranscribe extends Task {
this.profanityOption = recognizer.profanityOption || 'raw';
this.requestSnr = recognizer.requestSnr || false;
this.initialSpeechTimeoutMs = recognizer.initialSpeechTimeoutMs || 0;
this.azureServiceEndpoint = recognizer.azureServiceEndpoint;
}
get name() { return TaskName.Transcribe; }
@@ -140,12 +138,7 @@ class TaskTranscribe extends Task {
].forEach((arr) => {
if (this[arr[0]]) opts[arr[1]] = true;
});
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.hints.length > 1) opts.GOOGLE_SPEECH_HINTS = this.hints.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;
@@ -212,7 +205,6 @@ class TaskTranscribe extends Task {
if (this.profanityOption !== 'raw') opts.AZURE_PROFANITY_OPTION = this.profanityOption;
if (this.initialSpeechTimeoutMs > 0) opts.AZURE_INITIAL_SPEECH_TIMEOUT_MS = this.initialSpeechTimeoutMs;
if (this.outputFormat !== 'simple') opts.AZURE_USE_OUTPUT_FORMAT_DETAILED = 1;
if (this.azureServiceEndpoint) opts.AZURE_SERVICE_ENDPOINT = this.azureServiceEndpoint;
await ep.set(opts)
.catch((err) => this.logger.info(err, 'TaskTranscribe:_startTranscribing with azure'));
@@ -253,16 +245,8 @@ class TaskTranscribe extends Task {
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);
}
if (this.transcriptionHook) {
const b3 = this.getTracingPropagation();
const httpHeaders = b3 && {b3};
this.cs.requestor.request('verb:hook', this.transcriptionHook,
Object.assign({speech: evt}, this.cs.callInfo), httpHeaders)
this.cs.requestor.request(this.transcriptionHook, Object.assign({speech: evt}, this.cs.callInfo))
.catch((err) => this.logger.info(err, 'TranscribeTask:_onTranscription error'));
}
if (this.parentTask) {

View File

@@ -1,75 +0,0 @@
const assert = require('assert');
const Emitter = require('events');
const crypto = require('crypto');
const timeSeries = require('@jambonz/time-series');
let alerter ;
class BaseRequestor extends Emitter {
constructor(logger, account_sid, hook, secret) {
super();
assert(typeof hook === 'object');
this.logger = logger;
this.url = hook.url;
this.username = hook.username;
this.password = hook.password;
this.secret = secret;
this.account_sid = account_sid;
const {stats} = require('../../').srf.locals;
this.stats = stats;
if (!alerter) {
alerter = timeSeries(logger, {
host: process.env.JAMBONES_TIME_SERIES_HOST,
commitSize: 50,
commitInterval: 'test' === process.env.NODE_ENV ? 7 : 20
});
}
}
get Alerter() {
return alerter;
}
close() {
/* subclass responsibility */
}
_computeSignature(payload, timestamp, secret) {
assert(secret);
const data = `${timestamp}.${JSON.stringify(payload)}`;
return crypto
.createHmac('sha256', secret)
.update(data, 'utf8')
.digest('hex');
}
_generateSigHeader(payload, secret) {
const timestamp = Math.floor(Date.now() / 1000);
const signature = this._computeSignature(payload, timestamp, secret);
const scheme = 'v1';
return {
'Jambonz-Signature': `t=${timestamp},${scheme}=${signature}`
};
}
_isAbsoluteUrl(u) {
return typeof u === 'string' &&
u.startsWith('https://') || u.startsWith('http://') ||
u.startsWith('ws://') || u.startsWith('wss://');
}
_isRelativeUrl(u) {
return typeof u === 'string' && u.startsWith('/');
}
_roundTrip(startAt) {
const diff = process.hrtime(startAt);
const time = diff[0] * 1e3 + diff[1] * 1e-6;
return time.toFixed(0);
}
}
module.exports = BaseRequestor;

View File

@@ -1,78 +0,0 @@
const {context, trace} = require('@opentelemetry/api');
const {Dialog} = require('drachtio-srf');
class RootSpan {
constructor(callType, req) {
let tracer, callSid, linkedSpanId;
if (req instanceof Dialog) {
const dlg = req;
tracer = dlg.srf.locals.otel.tracer;
callSid = dlg.callSid;
linkedSpanId = dlg.linkedSpanId;
}
else {
tracer = req.srf.locals.otel.tracer;
callSid = req.locals.callSid;
}
this._span = tracer.startSpan(callType || 'incoming-call');
if (req instanceof Dialog) {
const dlg = req;
this._span.setAttributes({
linkedSpanId,
callId: dlg.sip.callId
});
}
else {
this._span.setAttributes({
callSid,
accountSid: req.get('X-Account-Sid'),
applicationSid: req.locals.application_sid,
callId: req.get('Call-ID'),
externalCallId: req.get('X-CID')
});
}
this._ctx = trace.setSpan(context.active(), this._span);
this.tracer = tracer;
}
get context() {
return this._ctx;
}
get traceId() {
return this._span.spanContext().traceId;
}
get spanId() {
return this._span.spanContext().spanId;
}
get traceFlags() {
return this._span.spanContext().traceFlags;
}
getTracingPropagation(encoding) {
// TODO: support encodings beyond b3 https://github.com/openzipkin/b3-propagation
if (this._span && this.traceId !== '00000000000000000000000000000000') {
return `${this.traceId}-${this.spanId}-1`;
}
}
setAttributes(attrs) {
this._span.setAttributes(attrs);
}
end() {
this._span.end();
}
startChildSpan(name, attributes) {
const span = this.tracer.startSpan(name, attributes, this._ctx);
const ctx = trace.setSpan(context.active(), span);
return {span, ctx};
}
}
module.exports = RootSpan;

View File

@@ -2,7 +2,6 @@
"TaskName": {
"Cognigy": "cognigy",
"Conference": "conference",
"Config": "config",
"Dequeue": "dequeue",
"Dial": "dial",
"Dialogflow": "dialogflow",
@@ -59,22 +58,19 @@
"Transcription": "google_transcribe::transcription",
"EndOfUtterance": "google_transcribe::end_of_utterance",
"NoAudioDetected": "google_transcribe::no_audio_detected",
"MaxDurationExceeded": "google_transcribe::max_duration_exceeded",
"VadDetected": "google_transcribe::vad_detected"
"MaxDurationExceeded": "google_transcribe::max_duration_exceeded"
},
"AwsTranscriptionEvents": {
"Transcription": "aws_transcribe::transcription",
"EndOfTranscript": "aws_transcribe::end_of_transcript",
"NoAudioDetected": "aws_transcribe::no_audio_detected",
"MaxDurationExceeded": "aws_transcribe::max_duration_exceeded",
"VadDetected": "aws_transcribe::vad_detected"
"MaxDurationExceeded": "aws_transcribe::max_duration_exceeded"
},
"AzureTranscriptionEvents": {
"Transcription": "azure_transcribe::transcription",
"StartOfUtterance": "azure_transcribe::start_of_utterance",
"EndOfUtterance": "azure_transcribe::end_of_utterance",
"NoSpeechDetected": "azure_transcribe::no_speech_detected",
"VadDetected": "azure_transcribe::vad_detected"
"NoSpeechDetected": "azure_transcribe::no_speech_detected"
},
"ListenEvents": {
"Connect": "mod_audio_fork::connect",
@@ -109,16 +105,6 @@
"Hangup": "hangup",
"Replaced": "replaced"
},
"HookMsgTypes": [
"session:new",
"session:reconnect",
"session:redirect",
"call:status",
"queue:status",
"dial:confirm",
"verb:hook",
"jambonz:error"
],
"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,52 +0,0 @@
const {execSync} = require('child_process');
const now = Date.now();
const fsInventory = process.env.JAMBONES_FREESWITCH
.split(',')
.map((fs) => {
const arr = /^([^:]*):([^:]*):([^:]*)(?::([^:]*))?/.exec(fs);
const opts = {address: arr[1], port: arr[2], secret: arr[3]};
if (arr.length > 4) opts.advertisedAddress = arr[4];
if (process.env.NODE_ENV === 'test') opts.listenAddress = '0.0.0.0';
return opts;
});
const clearChannels = () => {
const {logger} = require('../..');
const pwd = fsInventory[0].secret;
const maxDurationMins = process.env.JAMBONES_FREESWITCH_MAX_CALL_DURATION_MINS || 180;
const calls = execSync(`/usr/local/freeswitch/bin/fs_cli -p ${pwd} -x "show calls"`, {encoding: 'utf8'})
.split('\n')
.filter((line) => line.match(/^[0-9A-Fa-f]{8}-[0-9A-Fa-f]{4}-[0-9A-Fa-f]{4}-[0-9A-Fa-f]{4}-[0-9A-Fa-f]{8}/))
.map((line) => {
const arr = line.split(',');
const dt = new Date(arr[2]);
const duration = (now - dt.getTime()) / 1000;
return {
uuid: arr[0],
time: arr[2],
duration
};
})
.filter((c) => c.duration > 60 * maxDurationMins);
if (calls.length > 0) {
logger.debug(`clearChannels: clearing ${calls.length} old calls longer than ${maxDurationMins} mins`);
for (const call of calls) {
const cmd = `/usr/local/freeswitch/bin/fs_cli -p ${pwd} -x "uuid_kill ${call.uuid}"`;
const out = execSync(cmd, {encoding: 'utf8'});
logger.debug({out}, 'clearChannels: command output');
}
}
return calls.length;
};
const clearFiles = () => {
const {logger} = require('../..');
const out = execSync('find /tmp -name "*.mp3" -mtime +2 -exec rm {} \\;');
logger.debug({out}, 'clearFiles: command output');
};
module.exports = {clearChannels, clearFiles};

View File

@@ -48,7 +48,6 @@ module.exports = (logger, srf) => {
const pp = pool.promise();
const lookupAccountDetails = async(account_sid) => {
const [r] = await pp.query({sql: sqlAccountDetails, nestTables: true}, account_sid);
if (0 === r.length) throw new Error(`invalid accountSid: ${account_sid}`);
const [r2] = await pp.query(sqlSpeechCredentials, account_sid);

View File

@@ -1,111 +0,0 @@
const bent = require('bent');
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 toBase64 = (str) => Buffer.from(str || '', 'utf8').toString('base64');
function basicAuth(username, password) {
if (!username || !password) return {};
const creds = `${username}:${password || ''}`;
const header = `Basic ${toBase64(creds)}`;
return {Authorization: header};
}
class HttpRequestor extends BaseRequestor {
constructor(logger, account_sid, hook, secret) {
super(logger, account_sid, hook, secret);
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));
}
get baseUrl() {
return this._baseUrl;
}
/**
* Make an HTTP request.
* All requests use json bodies.
* All requests expect a 200 statusCode on success
* @param {object|string} hook - may be a absolute or relative url, or an object
* @param {string} [hook.url] - an absolute or relative url
* @param {string} [hook.method] - 'GET' or 'POST'
* @param {string} [hook.username] - if basic auth is protecting the endpoint
* @param {string} [hook.password] - if basic auth is protecting the endpoint
* @param {object} [params] - request parameters
*/
async request(type, hook, params, httpHeaders = {}) {
assert(HookMsgTypes.includes(type));
const payload = params ? snakeCaseKeys(params, ['customerData', 'sip']) : null;
const url = hook.url || hook;
const method = hook.method || 'POST';
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}`);
const {url: urlInfo = hook, method: methodInfo = 'POST'} = hook; // mask user/pass
this.logger.debug({url: urlInfo, method: methodInfo, payload}, `HttpRequestor:request ${method} ${url}`);
const startAt = process.hrtime();
let buf;
try {
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);
} catch (err) {
if (err.statusCode) {
this.logger.info({baseUrl: this.baseUrl, url},
`web callback returned unexpected status code ${err.statusCode}`);
}
else {
this.logger.error({err, baseUrl: this.baseUrl, url},
'web callback returned unexpected error');
}
let opts = {account_sid: this.account_sid};
if (err.code === 'ECONNREFUSED') {
opts = {...opts, alert_type: this.Alerter.AlertType.WEBHOOK_CONNECTION_FAILURE, url};
}
else if (err.name === 'StatusError') {
opts = {...opts, alert_type: this.Alerter.AlertType.WEBHOOK_STATUS_FAILURE, url, status: err.statusCode};
}
else {
opts = {...opts, alert_type: this.Alerter.AlertType.WEBHOOK_CONNECTION_FAILURE, url, detail: err.message};
}
this.Alerter.writeAlerts(opts).catch((err) => this.logger.info({err, opts}, 'Error writing alert'));
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()}'`);
}
}
}
}
module.exports = HttpRequestor;

View File

@@ -32,7 +32,6 @@ function initMS(logger, wrapper, ms) {
function installSrfLocals(srf, logger) {
logger.debug('installing srf locals');
assert(!srf.locals.dbHelpers);
const {tracer} = srf.locals.otel;
const {getSBC, lifecycleEmitter} = require('./sbc-pinger')(logger);
const StatsCollector = require('@jambonz/stats-collector');
const stats = srf.locals.stats = new StatsCollector(logger);
@@ -50,11 +49,7 @@ function installSrfLocals(srf, logger) {
assert.ok(arr, `Invalid syntax JAMBONES_FREESWITCH: ${process.env.JAMBONES_FREESWITCH}`);
const opts = {address: arr[1], port: arr[2], secret: arr[3]};
if (arr.length > 4) opts.advertisedAddress = arr[4];
/* NB: originally for testing only, but for now all jambonz deployments
have freeswitch installed locally alongside this app
*/
if (process.env.NODE_ENV === 'test') opts.listenAddress = '0.0.0.0';
else if (process.env.JAMBONES_ESL_LISTEN_ADDRESS) opts.listenAddress = process.env.JAMBONES_ESL_LISTEN_ADDRESS;
return opts;
});
logger.info({fsInventory}, 'freeswitch inventory');
@@ -67,7 +62,7 @@ function installSrfLocals(srf, logger) {
initMS(logger, val, ms);
}
catch (err) {
logger.info({err}, `failed connecting to freeswitch at ${fs.address}, will retry shortly: ${err.message}`);
logger.info(`failed connecting to freeswitch at ${fs.address}, will retry shortly`);
}
}
// retry to connect to any that were initially offline
@@ -79,7 +74,7 @@ function installSrfLocals(srf, logger) {
const ms = await mrf.connect(val.opts);
initMS(logger, val, ms);
} catch (err) {
logger.info({err}, `failed connecting to freeswitch at ${val.opts.address}, will retry shortly`);
logger.info(`failed connecting to freeswitch at ${val.opts.address}, will retry shortly`);
}
}
}
@@ -132,7 +127,7 @@ function installSrfLocals(srf, logger) {
password: process.env.JAMBONES_MYSQL_PASSWORD,
database: process.env.JAMBONES_MYSQL_DATABASE,
connectionLimit: process.env.JAMBONES_MYSQL_CONNECTION_LIMIT || 10
}, logger, tracer);
}, logger);
const {
client,
updateCallStatus,
@@ -157,7 +152,7 @@ function installSrfLocals(srf, logger) {
} = require('@jambonz/realtimedb-helpers')({
host: process.env.JAMBONES_REDIS_HOST,
port: process.env.JAMBONES_REDIS_PORT || 6379
}, logger, tracer);
}, logger);
const {
writeAlerts,
AlertType

View File

@@ -4,18 +4,15 @@ const SipError = require('drachtio-srf').SipError;
const {TaskPreconditions, CallDirection} = require('../utils/constants');
const CallInfo = require('../session/call-info');
const assert = require('assert');
const normalizeJambones = require('../utils/normalize-jambones');
const makeTask = require('../tasks/make_task');
const ConfirmCallSession = require('../session/confirm-call-session');
const AdultingCallSession = require('../session/adulting-call-session');
const deepcopy = require('deepcopy');
const moment = require('moment');
const stripCodecs = require('./strip-ancillary-codecs');
const RootSpan = require('./call-tracer');
const { v4: uuidv4 } = require('uuid');
class SingleDialer extends Emitter {
constructor({logger, sbcAddress, target, opts, application, callInfo, accountInfo, rootSpan, startSpan}) {
constructor({logger, sbcAddress, target, opts, application, callInfo, accountInfo}) {
super();
assert(target.type);
@@ -25,8 +22,6 @@ class SingleDialer extends Emitter {
this.opts = opts;
this.application = application;
this.confirmHook = target.confirmHook;
this.rootSpan = rootSpan;
this.startSpan = startSpan;
this.bindings = logger.bindings();
@@ -76,7 +71,7 @@ class SingleDialer extends Emitter {
};
}
this.ms = ms;
let uri, to, inviteSpan;
let uri, to;
try {
switch (this.target.type) {
case 'phone':
@@ -142,24 +137,13 @@ class SingleDialer extends Emitter {
localSdp: this.ep.local.sdp
});
if (this.target.auth) opts.auth = this.target.auth;
inviteSpan = this.startSpan('invite', {
'invite.uri': uri,
'invite.dest_type': this.target.type
});
this.dlg = await srf.createUAC(uri, {...opts, followRedirects: true, keepUriOnRedirect: true}, {
cbRequest: (err, req) => {
if (err) {
this.logger.error(err, 'SingleDialer:exec Error creating call');
this.emit('callCreateFail', err);
inviteSpan.setAttributes({
'invite.status_code': 500,
'invite.err': err.message
});
inviteSpan.end();
return;
}
inviteSpan.setAttributes({'invite.call_id': req.get('Call-ID')});
/**
* INVITE has been sent out
@@ -172,8 +156,7 @@ class SingleDialer extends Emitter {
parentCallInfo: this.parentCallInfo,
req,
to,
callSid: this.callSid,
traceId: this.rootSpan.traceId
callSid: this.callSid
});
this.logger = srf.locals.parentLogger.child({
callSid: this.callSid,
@@ -181,14 +164,10 @@ class SingleDialer extends Emitter {
callId: this.callInfo.callId
});
this.inviteInProgress = req;
this.emit('callStatusChange', {
callStatus: CallStatus.Trying,
sipStatus: 100,
sipReason: 'Trying'
});
this.emit('callStatusChange', {callStatus: CallStatus.Trying, sipStatus: 100});
},
cbProvisional: (prov) => {
const status = {sipStatus: prov.status, sipReason: prov.reason};
const status = {sipStatus: prov.status};
if ([180, 183].includes(prov.status) && prov.body) {
if (status.callStatus !== CallStatus.EarlyMedia) {
status.callStatus = CallStatus.EarlyMedia;
@@ -203,27 +182,15 @@ class SingleDialer extends Emitter {
await connectStream(this.dlg.remote.sdp);
this.dlg.callSid = this.callSid;
this.inviteInProgress = null;
this.emit('callStatusChange', {
sipStatus: 200,
sipReason: 'OK',
callStatus: CallStatus.InProgress
});
this.emit('callStatusChange', {sipStatus: 200, callStatus: CallStatus.InProgress});
this.logger.debug(`SingleDialer:exec call connected: ${this.callSid}`);
const connectTime = this.dlg.connectTime = moment();
inviteSpan.setAttributes({'invite.status_code': 200});
inviteSpan.end();
/* race condition: we were killed just as call was answered */
if (this.killed) {
this.logger.info(`SingleDialer:exec race condition - we were killed as call connected: ${this.callSid}`);
const duration = moment().diff(connectTime, 'seconds');
this.emit('callStatusChange', {
callStatus: CallStatus.Completed,
sipStatus: 487,
sipReason: 'Request Terminated',
duration
});
this.emit('callStatusChange', {callStatus: CallStatus.Completed, duration});
if (this.ep) this.ep.destroy();
return;
}
@@ -250,9 +217,6 @@ class SingleDialer extends Emitter {
} catch (err) {
this.logger.error(err, 'Error handling reinvite');
}
})
.on('refer', (req, res) => {
this.emit('refer', this.callInfo, req, res);
});
if (this.confirmHook) this._executeApp(this.confirmHook);
@@ -262,21 +226,13 @@ class SingleDialer extends Emitter {
const status = {callStatus: CallStatus.Failed};
if (err instanceof SipError) {
status.sipStatus = err.status;
status.sipReason = err.reason;
if (err.status === 487) status.callStatus = CallStatus.NoAnswer;
else if ([486, 600].includes(err.status)) status.callStatus = CallStatus.Busy;
this.logger.info(`SingleDialer:exec outdial failure ${err.status}`);
inviteSpan.setAttributes({'invite.status_code': err.status});
inviteSpan.end();
}
else {
this.logger.error(err, 'SingleDialer:exec');
status.sipStatus = 500;
inviteSpan.setAttributes({
'invite.status_code': 500,
'invite.err': err.message
});
inviteSpan.end();
}
this.emit('callStatusChange', status);
if (this.ep) this.ep.destroy();
@@ -311,8 +267,8 @@ class SingleDialer extends Emitter {
async _executeApp(confirmHook) {
try {
// retrieve set of tasks
const json = await this.requestor.request('dial:confirm', confirmHook, this.callInfo.toJSON());
const tasks = normalizeJambones(this.logger, json).map((tdata) => makeTask(this.logger, tdata));
const tasks = await this.requestor.request(confirmHook, this.callInfo.toJSON());
// verify it contains only allowed verbs
const allowedTasks = tasks.filter((task) => {
return [
@@ -332,9 +288,7 @@ class SingleDialer extends Emitter {
dlg: this.dlg,
ep: this.ep,
callInfo: this.callInfo,
accountInfo: this.accountInfo,
tasks,
rootSpan: this.rootSpan
tasks
});
await cs.exec();
@@ -348,6 +302,7 @@ class SingleDialer extends Emitter {
}
async doAdulting({logger, tasks, application}) {
this.logger = logger;
this.adulting = true;
this.emit('adulting');
if (this.ep) {
@@ -358,21 +313,15 @@ class SingleDialer extends Emitter {
else {
await this.reAnchorMedia();
}
this.dlg.callSid = this.callSid;
this.dlg.linkedSpanId = this.rootSpan.traceId;
const rootSpan = new RootSpan('outbound-call', this.dlg);
const newLogger = logger.child({traceId: rootSpan.traceId});
const cs = new AdultingCallSession({
logger: newLogger,
logger: this.logger,
singleDialer: this,
application,
callInfo: this.callInfo,
accountInfo: this.accountInfo,
tasks,
rootSpan
tasks
});
cs.exec().catch((err) => newLogger.error({err}, 'doAdulting: error executing session'));
cs.exec();
return cs;
}
@@ -399,16 +348,16 @@ class SingleDialer extends Emitter {
});
}
_notifyCallStatusChange({callStatus, sipStatus, sipReason, duration}) {
_notifyCallStatusChange({callStatus, sipStatus, duration}) {
assert((typeof duration === 'number' && callStatus === CallStatus.Completed) ||
(!duration && callStatus !== CallStatus.Completed),
'duration MUST be supplied when call completed AND ONLY when call completed');
if (this.callInfo) {
this.callInfo.updateCallStatus(callStatus, sipStatus, sipReason);
this.callInfo.updateCallStatus(callStatus, sipStatus);
if (typeof duration === 'number') this.callInfo.duration = duration;
try {
this.requestor.request('call:status', this.application.call_status_hook, this.callInfo.toJSON());
this.requestor.request(this.application.call_status_hook, this.callInfo.toJSON());
} catch (err) {
this.logger.info(err, `SingleDialer:_notifyCallStatusChange error sending ${callStatus} ${sipStatus}`);
}
@@ -421,13 +370,9 @@ class SingleDialer extends Emitter {
}
}
function placeOutdial({
logger, srf, ms, sbcAddress, target, opts, application, callInfo, accountInfo, rootSpan, startSpan
}) {
function placeOutdial({logger, srf, ms, sbcAddress, target, opts, application, callInfo, accountInfo}) {
const myOpts = deepcopy(opts);
const sd = new SingleDialer({
logger, sbcAddress, target, myOpts, application, callInfo, accountInfo, rootSpan, startSpan
});
const sd = new SingleDialer({logger, sbcAddress, target, myOpts, application, callInfo, accountInfo});
sd.exec(srf, ms, myOpts);
return sd;
}

View File

@@ -17,10 +17,6 @@ module.exports = (logger) => {
assert.ok(sbcs.length, 'JAMBONES_SBCS env var is empty or misconfigured');
logger.info({sbcs}, 'SBC inventory');
}
else if (process.env.K8S && process.env.K8S_SBC_SIP_SERVICE_NAME) {
sbcs = [`${process.env.K8S_SBC_SIP_SERVICE_NAME}:5060`];
logger.info({sbcs}, 'SBC inventory');
}
// listen for SNS lifecycle changes
let lifecycleEmitter = new Emitter();

View File

@@ -1,3 +1,3 @@
module.exports = function(tasks) {
return `[${tasks.map((t) => t.summary).join(',')}]`;
return `[${tasks.map((t) => t.name).join(',')}]`;
};

View File

@@ -1,287 +0,0 @@
const assert = require('assert');
const BaseRequestor = require('./base-requestor');
const short = require('short-uuid');
const {HookMsgTypes} = require('./constants.json');
const Websocket = require('ws');
const snakeCaseKeys = require('./snakecase-keys');
const HttpRequestor = require('./http-requestor');
const MAX_RECONNECTS = 5;
const RESPONSE_TIMEOUT_MS = process.env.JAMBONES_WS_API_MSG_RESPONSE_TIMEOUT || 5000;
class WsRequestor extends BaseRequestor {
constructor(logger, account_sid, hook, secret) {
super(logger, account_sid, hook, secret);
this.connections = 0;
this.messagesInFlight = new Map();
this.maliciousClient = false;
this.closedByUs = false;
this.backoffMs = 500;
assert(this._isAbsoluteUrl(this.url));
this.on('socket-closed', this._onSocketClosed.bind(this));
}
/**
* Send a JSON payload over the websocket. If this is the first request,
* open the websocket.
* All requests expect an ack message in response
* @param {object|string} hook - may be a absolute or relative url, or an object
* @param {string} [hook.url] - an absolute or relative url
* @param {string} [hook.method] - 'GET' or 'POST'
* @param {string} [hook.username] - if basic auth is protecting the endpoint
* @param {string} [hook.password] - if basic auth is protecting the endpoint
* @param {object} [params] - request parameters
*/
async request(type, hook, params, httpHeaders = {}) {
assert(HookMsgTypes.includes(type));
const url = hook.url || hook;
if (this.maliciousClient) {
this.logger.info({url: this.url}, 'WsRequestor:request - discarding msg to malicious client');
return;
}
if (type === 'session:new') this.call_sid = params.callSid;
/* if we have an absolute url, and it is http then do a standard webhook */
if (this._isAbsoluteUrl(url) && url.startsWith('http')) {
this.logger.debug({hook}, 'WsRequestor: sending a webhook (HTTP)');
const requestor = new HttpRequestor(this.logger, this.account_sid, hook, this.secret);
return requestor.request(type, hook, params, httpHeaders);
}
/* connect if necessary */
if (!this.ws) {
if (this.connections >= MAX_RECONNECTS) {
throw new Error(`max attempts connecting to ${this.url}`);
}
try {
const startAt = process.hrtime();
await this._connect();
const rtt = this._roundTrip(startAt);
this.stats.histogram('app.hook.connect_time', rtt, ['hook_type:app']);
} catch (err) {
this.logger.info({url, err}, 'WsRequestor:request - failed connecting');
throw err;
}
}
assert(this.ws);
/* prepare and send message */
const payload = params ? snakeCaseKeys(params, ['customerData', 'sip']) : null;
assert.ok(url, 'WsRequestor:request url was not provided');
const msgid = short.generate();
const b3 = httpHeaders?.b3 ? {b3: httpHeaders.b3} : {};
const obj = {
type,
msgid,
call_sid: this.call_sid,
hook: type === 'verb:hook' ? url : undefined,
data: {...payload},
...b3
};
//this.logger.debug({obj}, `websocket: sending (${url})`);
/* simple notifications */
if (['call:status', 'jambonz:error'].includes(type)) {
this.ws.send(JSON.stringify(obj), () => {
this.logger.debug({obj}, `WsRequestor:request websocket: sent (${url})`);
});
return;
}
/* messages that require an ack */
return new Promise((resolve, reject) => {
/* give the far end a reasonable amount of time to ack our message */
const timer = setTimeout(() => {
const {failure} = this.messagesInFlight.get(msgid);
failure && failure(`timeout from far end for msgid ${msgid}`);
this.messagesInFlight.delete(msgid);
}, RESPONSE_TIMEOUT_MS);
/* save the message info for reply */
const startAt = process.hrtime();
this.messagesInFlight.set(msgid, {
timer,
success: (response) => {
clearTimeout(timer);
const rtt = this._roundTrip(startAt);
this.logger.info({response}, `WsRequestor:request ${url} succeeded in ${rtt}ms`);
this.stats.histogram('app.hook.ws_response_time', rtt, ['hook_type:app']);
resolve(response);
},
failure: (err) => {
clearTimeout(timer);
reject(err);
}
});
/* send the message */
this.ws.send(JSON.stringify(obj), () => {
this.logger.debug({obj}, `WsRequestor:request websocket: sent (${url})`);
});
});
}
close() {
this.closedByUs = true;
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();
} catch (err) {
this.logger.info({err}, 'WsRequestor: Error closing socket');
}
}
_connect() {
assert(!this.ws);
return new Promise((resolve, reject) => {
const handshakeTimeout = process.env.JAMBONES_WS_HANDSHAKE_TIMEOUT_MS ?
parseInt(process.env.JAMBONES_WS_HANDSHAKE_TIMEOUT_MS) :
1500;
let opts = {
followRedirects: true,
maxRedirects: 2,
handshakeTimeout,
maxPayload: 8096,
};
if (this.username && this.password) opts = {...opts, auth: `${this.username}:${this.password}`};
this
.once('ready', (ws) => {
this.ws = ws;
this.removeAllListeners('not-ready');
if (this.connections > 0) this.request('session:reconnect', this.url);
resolve();
})
.once('not-ready', (err) => {
this.removeAllListeners('ready');
reject(err);
});
const ws = new Websocket(this.url, ['ws.jambonz.org'], opts);
this._setHandlers(ws);
});
}
_setHandlers(ws) {
ws
.once('open', this._onOpen.bind(this, ws))
.once('close', this._onClose.bind(this))
.on('message', this._onMessage.bind(this))
.once('unexpected-response', this._onUnexpectedResponse.bind(this, ws))
.on('error', this._onError.bind(this));
}
_onError(err) {
if (this.connections > 0) {
this.logger.info({url: this.url, err}, 'WsRequestor:_onError');
}
else this.emit('not-ready', err);
}
_onOpen(ws) {
if (this.ws) this.logger.info({old_ws: this.ws._socket.address()}, 'WsRequestor:_onOpen');
assert(!this.ws);
this.emit('ready', ws);
this.logger.info({url: this.url}, 'WsRequestor - successfully connected');
}
_onClose() {
if (this.connections > 0) {
this.logger.info({url: this.url}, 'WsRequestor - socket closed unexpectedly from remote side');
this.emit('socket-closed');
}
this.ws && this.ws.removeAllListeners();
this.ws = null;
}
_onUnexpectedResponse(ws, req, res) {
assert(!this.ws);
this.logger.info({
headers: res.headers,
statusCode: res.statusCode,
statusMessage: res.statusMessage
}, 'WsRequestor - unexpected response');
this.emit('connection-failure');
this.emit('not-ready', new Error(`${res.statusCode} ${res.statusMessage}`));
}
_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);
this.backoffMs = this.backoffMs < 2000 ? this.backoffMs * 2 : (this.backoffMs + 2000);
}
}
_onMessage(content, isBinary) {
if (this.isBinary) {
this.logger.info({url: this.url}, 'WsRequestor:_onMessage - discarding binary message');
this.maliciousClient = true;
this.ws.close();
return;
}
/* messages must be JSON format */
try {
const obj = JSON.parse(content);
const {type, msgid, command, call_sid = this.call_sid, queueCommand = false, data} = obj;
this.logger.debug({obj}, 'WsRequestor:request websocket: received');
assert.ok(type, 'type property not supplied');
switch (type) {
case 'ack':
assert.ok(msgid, 'msgid not supplied');
this._recvAck(msgid, data);
break;
case 'command':
assert.ok(command, 'command property not supplied');
assert.ok(data, 'data property not supplied');
this._recvCommand(msgid, command, call_sid, queueCommand, data);
break;
default:
assert.ok(false, `invalid type property: ${type}`);
}
} catch (err) {
this.logger.info({err, content}, 'WsRequestor:_onMessage - invalid incoming message');
}
}
_recvAck(msgid, data) {
const obj = this.messagesInFlight.get(msgid);
if (!obj) {
this.logger.info({url: this.url}, `WsRequestor:_recvAck - ack to unknown msgid ${msgid}, discarding`);
return;
}
this.logger.debug({url: this.url}, `WsRequestor:_recvAck - received response to ${msgid}`);
this.messagesInFlight.delete(msgid);
const {success} = obj;
success && success(data);
}
_recvCommand(msgid, command, call_sid, queueCommand, data) {
// TODO: validate command
this.logger.info({msgid, command, call_sid, queueCommand, data}, 'received command');
this.emit('command', {msgid, command, call_sid, queueCommand, data});
}
}
module.exports = WsRequestor;

1546
package-lock.json generated

File diff suppressed because it is too large Load Diff

View File

@@ -1,6 +1,6 @@
{
"name": "jambonz-feature-server",
"version": "v0.7.5",
"version": "v0.7.3",
"main": "app.js",
"engines": {
"node": ">= 10.16.0"
@@ -21,9 +21,10 @@
},
"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 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=debug 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"
"jslint": "eslint app.js lib",
"prepare": "husky install"
},
"dependencies": {
"@cognigy/socket-client": "^4.5.5",
@@ -33,18 +34,6 @@
"@jambonz/realtimedb-helpers": "^0.4.26",
"@jambonz/stats-collector": "^0.1.6",
"@jambonz/time-series": "^0.1.6",
"@opentelemetry/api": "^1.1.0",
"@opentelemetry/exporter-collector": "^0.25.0",
"@opentelemetry/exporter-jaeger": "^1.1.0",
"@opentelemetry/exporter-zipkin": "^1.1.0",
"@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",
"bent": "^7.3.12",
"cidr-matcher": "^2.1.1",
@@ -55,14 +44,12 @@
"express": "^4.17.1",
"helmet": "^5.0.2",
"ip": "^1.1.5",
"moment": "^2.29.2",
"moment": "^2.29.1",
"parse-url": "^5.0.7",
"pino": "^6.13.4",
"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",
"xml2js": "^0.4.23"
},
"devDependencies": {
@@ -70,11 +57,8 @@
"clear-module": "^4.1.1",
"eslint": "^7.20.0",
"eslint-plugin-promise": "^4.3.1",
"husky": "7.0.4",
"nyc": "^15.1.0",
"tape": "^5.2.2"
},
"optionalDependencies": {
"bufferutil": "^4.0.6",
"utf-8-validate": "^5.0.8"
}
}

View File

@@ -1,61 +0,0 @@
const opentelemetry = require('@opentelemetry/api');
const { registerInstrumentations } = require('@opentelemetry/instrumentation');
const { NodeTracerProvider } = require('@opentelemetry/sdk-trace-node');
const { Resource } = require('@opentelemetry/resources');
const { SemanticResourceAttributes } = require('@opentelemetry/semantic-conventions');
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 { HttpInstrumentation } = require('@opentelemetry/instrumentation-http');
//const { ExpressInstrumentation } = require('@opentelemetry/instrumentation-express');
//const { PinoInstrumentation } = require('@opentelemetry/instrumentation-pino');
module.exports = (serviceName) => {
if (process.env.JAMBONES_OTEL_ENABLED) {
const {version} = require('./package.json');
const provider = new NodeTracerProvider({
resource: new Resource({
[SemanticResourceAttributes.SERVICE_NAME]: serviceName,
[SemanticResourceAttributes.SERVICE_VERSION]: version,
}),
});
let exporter;
if (process.env.OTEL_EXPORTER_JAEGER_AGENT_HOST) {
exporter = new JaegerExporter();
}
else if (process.env.OTEL_EXPORTER_ZIPKIN_URL) {
exporter = new ZipkinExporter({url:process.env.OTEL_EXPORTER_ZIPKIN_URL});
}
else {
exporter = new CollectorTraceExporter({
url: process.OTEL_EXPORTER_COLLECTOR_URL
});
}
provider.addSpanProcessor(new BatchSpanProcessor(exporter, {
// The maximum queue size. After the size is reached spans are dropped.
maxQueueSize: 100,
// The maximum batch size of every export. It must be smaller or equal to maxQueueSize.
maxExportBatchSize: 10,
// The interval between two consecutive exports
scheduledDelayMillis: 500,
// How long the export can run before it is cancelled
exportTimeoutMillis: 30000,
}));
// Initialize the OpenTelemetry APIs to use the NodeTracerProvider bindings
provider.register();
registerInstrumentations({
instrumentations: [
//new HttpInstrumentation(),
//new ExpressInstrumentation(),
//new PinoInstrumentation()
],
});
}
return opentelemetry.trace.getTracer(serviceName);
};