ben.carrasco/jambonz-feature-server
1
0
Fork 0
mirror of https://github.com/jambonz/jambonz-feature-server.git synced 2025-12-20 16:50:39 +00:00
Code Issues Packages Projects Releases Wiki Activity

major refactoring

Browse Source
This commit is contained in:
Dave Horton
2020-01-25 11:47:33 -05:00
parent 621ea8c0f5
commit 4a1ea4e091
25 changed files with 947 additions and 933 deletions
Download Patch File Download Diff File Expand all files Collapse all files

474
lib/tasks/dial.js
View File

@@ -1,356 +1,228 @@
const Task = require('./task');
const makeTask = require('./make_task');
const {CallStatus, CallDirection, TaskName, TaskPreconditions} = require('../utils/constants');
const SipError = require('drachtio-srf').SipError;
const assert = require('assert');
const uuidv4 = require('uuid/v4');
const request = require('request');
const placeCall = require('../utils/place-outdial');
const config = require('config');
const moment = require('moment');
const debug = require('debug')('jambonz:feature-server');
function isFinalCallStatus(status) {
return [CallStatus.Completed, CallStatus.NoAnswer, CallStatus.Failed, CallStatus.Busy].includes(status);
}
class TaskDial extends Task {
constructor(logger, opts) {
super(logger, opts);
this.preconditions = TaskPreconditions.None;
this.action = opts.action;
this.earlyMedia = opts.answerOnBridge === true;
this.callerId = opts.callerId;
this.dialMusic = opts.dialMusic;
this.earlyMedia = this.data.answerOnBridge === true;
this.callerId = this.data.callerId;
this.dialMusic = this.data.dialMusic;
this.headers = this.data.headers || {};
this.method = opts.method || 'POST';
this.statusCallback = opts.statusCallback;
this.statusCallbackMethod = opts.statusCallbackMethod || 'POST';
this.target = opts.target;
this.timeout = opts.timeout || 60;
this.timeLimit = opts.timeLimit;
this.method = this.data.method || 'POST';
this.statusCallback = this.data.statusCallback;
this.statusCallbackMethod = this.data.statusCallbackMethod || 'POST';
this.target = this.data.target;
this.timeout = this.data.timeout || 60;
this.timeLimit = this.data.timeLimit;
this.url = this.data.url;
if (opts.listen) {
this.listenTask = makeTask(logger, {'listen': opts.listen});
if (this.data.listen) {
this.listenTask = makeTask(logger, {'listen': this.data.listen});
}
if (opts.transcribe) {
this.transcribeTask = makeTask(logger, {'transcribe' : opts.transcribe});
if (this.data.transcribe) {
this.transcribeTask = makeTask(logger, {'transcribe' : this.data.transcribe});
}
this.canceled = false;
this.callAttributes = {};
this.dialCallStatus = CallStatus.Failed;
this.dialCallSid = null;
this.dialCallDuration = null;
this.on('callStatusChange', this._onCallStatusChange.bind(this));
this.results = {};
this.bridged = false;
this.dials = new Map();
}
get name() { return TaskName.Dial; }
async exec(cs) {
super.exec(cs);
try {
this._initializeCallData(cs);
await this._initializeInbound(cs);
if (cs.direction === CallDirection.Inbound) {
await this._initializeInbound(cs);
}
await this._attemptCalls(cs);
await this._waitForCompletion(cs);
await this.awaitTaskDone();
this.performAction(this.method, this.results);
} catch (err) {
this.logger.error(`TaskDial:exec terminating with error ${err.message}`);
this.kill();
}
await this._actionHook(cs);
this.clearResources();
return true;
}
_initializeCallData(cs) {
this.logger.debug(`TaskDial:_initializeCallData parent call sid is ${cs.callSid}`);
Object.assign(this.callAttributes, {
AccountSid: cs.AccountSid,
ParentCallSid: cs.callSid,
Direction: CallDirection.Outbound
});
async kill() {
super.kill();
if (this.connectTime) {
const duration = moment().diff(this.connectTime, 'seconds');
this.results.dialCallDuration = duration;
this.logger.debug(`Dial:kill call ended after ${duration} seconds`);
}
this._killOutdials();
if (this.dlg) {
assert(this.ep);
if (this.dlg.connected) this.dlg.destroy();
debug(`Dial:kill deleting endpoint ${this.ep.uuid}`);
this.ep.destroy();
}
if (this.listenTask) await this.listenTask.kill();
if (this.transcribeTask) await this.transcribeTask.kill();
this.notifyTaskDone();
}
async _initializeInbound(cs) {
const {req} = cs;
const {ep} = await cs.connectInboundCallToIvr(this.earlyMedia);
this.epOther = ep;
debug(`Dial:__initializeInbound allocated ep for incoming call: ${ep.uuid}`);
// the caller could hangup in the middle of all this..
req.on('cancel', this._onCancel.bind(this, cs));
try {
const result = await cs.connectInboundCallToIvr(this.earlyMedia);
if (!result) throw new Error('outbound dial via API not supported yet');
const {ep, dlg, res} = result;
assert(ep);
// play dial music to caller, if provided
if (this.dialMusic) {
ep.play(this.dialMusic, (err) => {
if (err) this.logger.error(err, `TaskDial:_initializeInbound - error playing ${this.dialMusic}`);
});
}
this.epIn = ep;
this.dlgIn = dlg;
this.res = res;
} catch (err) {
this.logger.error(err, 'TaskDial:_initializeInbound error');
throw err;
if (this.dialMusic) {
// play dial music to caller while we outdial
ep.play(this.dialMusic).catch((err) => {
this.logger.error(err, `TaskDial:_initializeInbound - error playing ${this.dialMusic}`);
});
}
}
async _attemptCalls(cs) {
const {req, srf} = cs;
// send all outbound calls back to originating SBC for simplicity
const sbcAddress = `${req.source_address}:${req.source_port}`;
const callSid = uuidv4();
let newCallId, to, from;
try {
// create an endpoint for the outbound call
const epOut = await cs.createEndpoint();
this.addResource('epOut', epOut);
const {uri, opts} = this._prepareOutdialAttempt(this.target[0], sbcAddress,
this.callerId || req.callingNumber, epOut.local.sdp);
let streamConnected = false;
const connectStreams = async(remoteSdp) => {
streamConnected = true;
epOut.modify(remoteSdp);
this.epIn.bridge(epOut);
if (!this.dlgIn) {
this.dlgIn = await cs.srf.answerParentCall(this.epIn.local.sdp);
}
};
// outdial requested destination
const uac = await srf.createUAC(uri, opts, {
cbRequest: (err, reqSent) => {
this.outboundInviteInProgress = reqSent;
newCallId = req.get('Call-ID');
from = reqSent.callingNumber,
to = reqSent.calledNumber;
this.emit('callStatusChange', {
CallSid: callSid,
SipCallId: newCallId,
CallStatus: CallStatus.Trying,
From: from,
To: to,
SipStatus: 100
});
},
cbProvisional: (prov) => {
if ([180, 183].includes(prov.status)) {
this.emit('callStatusChange', {
CallSid: callSid,
SipCallId: newCallId,
CallStatus: prov.body ? CallStatus.EarlyMedia : CallStatus.Ringing,
From: from,
To: to,
SipStatus: prov.status
});
if (!streamConnected && prov.body) connectStreams(prov.body);
}
}
});
// outbound call was established
uac.connectTime = moment();
uac.callSid = this.dialCallSid = callSid;
uac.from = from;
uac.to = to;
this.emit('callStatusChange', {
CallSid: callSid,
SipCallId: newCallId,
From: from,
To: to,
CallStatus: CallStatus.InProgress,
SipStatus: 200
});
uac.on('destroy', () => {
const duration = this.dialCallDuration = moment().diff(uac.connectTime, 'seconds');
this.emit('callStatusChange', {
CallSid: callSid,
SipCallId: newCallId,
From: from,
To: to,
CallStatus: CallStatus.Completed,
Duration: duration
});
});
if (!streamConnected) connectStreams(uac.remote.sdp);
this.outboundInviteInProgress = null;
this.addResource('dlgOut', uac);
} catch (err) {
if (err instanceof SipError) {
switch (err.status) {
case 487:
this.emit('callStatusChange', {
CallSid: callSid,
SipCallId: newCallId,
From: from,
To: to,
CallStatus: CallStatus.NoAnswer,
SipStatus: err.status
});
break;
case 486:
case 600:
this.emit('callStatusChange', {
CallSid: callSid,
SipCallId: newCallId,
From: from,
To: to,
CallStatus: CallStatus.Busy,
SipStatus: err.status
});
break;
default:
this.emit('callStatusChange', {callSid,
CallSid: callSid,
SipCallId: newCallId,
From: from,
To: to,
CallStatus: CallStatus.Failed,
SipStatus: err.status
});
break;
}
if (err.status !== 487) {
this.logger.info(`TaskDial:_connectCall outdial failed with ${err.status}`);
}
}
else {
this.emit('callStatusChange', {
CallSid: callSid,
SipCallId: newCallId,
From: from,
To: to,
CallStatus: CallStatus.Failed,
SipStatus: 500
});
this.logger.error(err, 'TaskDial:_connectCall error');
}
throw err;
}
}
_prepareOutdialAttempt(target, sbcAddress, callerId, sdp) {
const sbcAddress = cs.direction === CallDirection.Inbound ?
`${req.source_address}:${req.source_port}` :
config.get('sbcAddress');
const opts = {
headers: this.headers,
proxy: `sip:${sbcAddress}`,
callingNumber: callerId,
localSdp: sdp
callingNumber: this.callerId || req.callingNumber
};
let uri;
switch (target.type) {
case 'phone':
uri = `sip:${target.number}@${sbcAddress}`;
break;
case 'sip':
uri = target.uri;
if (target.auth) Object.assign(opts, {auth: target.auth});
break;
case 'user':
uri = `sip:${target.name}`;
break;
default:
assert(0, `TaskDial:_prepareOutdialAttempt invalid target type ${target.type}; please fix specs.json`);
}
return {uri, opts};
}
// construct bare-bones callInfo for the new outbound call attempt
const callInfo = Object.assign({}, cs.callInfo);
callInfo.parentCallSid = cs.callSid;
callInfo.direction = CallDirection.Outbound;
['callSid', 'callID', 'from', 'to', 'callerId', 'sipStatus', 'callStatus'].forEach((k) => delete callInfo[k]);
_onCancel(cs) {
this.logger.info('TaskDial: caller hung up before connecting');
this.canceled = true;
cs.emit('callStatusChange', {status: CallStatus.NoAnswer});
}
const ms = await cs.getMS();
this.target.forEach((t) => {
try {
t.url = t.url || this.url;
const sd = placeCall({
logger: this.logger,
application: cs.application,
srf,
ms,
sbcAddress,
target: t,
opts,
callInfo
});
this.dials.set(sd.callSid, sd);
_onCallerHangup(cs, dlg) {
this.logger.info('TaskDial: caller hung up');
cs.emit('callStatusChange', {status: CallStatus.Completed});
if (this.outboundInviteInProgress) this.outboundInviteInProgress.cancel();
// we are going to hang up the B leg shortly..so
const dlgOut = this.getResource('dlgOut');
if (dlgOut) {
const duration = this.dialCallDuration = moment().diff(dlgOut.connectTime, 'seconds');
this.emit('callStatusChange', {
CallSid: dlgOut.callSid,
SipCallId: dlgOut.sip.callId,
From: dlgOut.from,
To: dlgOut.to,
CallStatus: CallStatus.Completed,
Duration: duration
});
}
}
/**
* returns a Promise that resolves when either party hangs up
*/
_waitForCompletion(cs) {
return new Promise((resolve) => {
const dlgOut = this.getResource('dlgOut');
assert(this.dlgIn && dlgOut);
assert(this.dlgIn.connected && dlgOut.connected);
[this.dlgIn, dlgOut].forEach((dlg) => dlg.on('destroy', () => resolve()));
sd
.on('callStatusChange', (obj) => {
switch (obj.callStatus) {
case CallStatus.Trying:
break;
case CallStatus.EarlyMedia:
if (this.target.length === 1 && !this.target[0].url && !this.dialMusic) {
this._bridgeEarlyMedia(sd);
}
break;
case CallStatus.InProgress:
this.logger.debug('Dial:_attemptCall -- call was answered');
break;
case CallStatus.Failed:
case CallStatus.Busy:
case CallStatus.NoAnswer:
this.dials.delete(sd.callSid);
if (this.dials.size === 0 && !this.connectTime) {
this.logger.debug('Dial:_attemptCalls - all calls failed after call failure, ending task');
this.kill();
}
break;
}
if (this.results.dialCallStatus !== CallStatus.Completed) {
Object.assign(this.results, {
dialCallStatus: obj.callStatus,
dialCallSid: sd.callSid,
});
}
})
.on('accept', () => {
this.logger.debug(`Dial:_attemptCalls - we have a winner: ${sd.callSid}`);
this._connectSingleDial(cs, sd);
})
.on('decline', () => {
this.logger.debug(`Dial:_attemptCalls - declined: ${sd.callSid}`);
this.dials.delete(sd.callSid);
if (this.dials.size === 0 && !this.connectTime) {
this.logger.debug('Dial:_attemptCalls - all calls failed after decline, ending task');
this.kill();
}
});
} catch (err) {
this.logger.error(err, 'Dial:_attemptCalls');
}
});
}
_onCallStatusChange(evt) {
this.logger.debug(evt, 'TaskDial:_onCallStatusChange');
// save the most recent final call status of a B leg, until we get one that is completed
if (isFinalCallStatus(evt.CallStatus) && this.dialCallStatus !== CallStatus.Completed) {
this.dialCallStatus = evt.CallStatus;
_connectSingleDial(cs, sd) {
if (!this.bridged) {
this.logger.debug('Dial:_connectSingleDial bridging endpoints');
this.epOther.api('uuid_break', this.epOther.uuid);
this.epOther.bridge(sd.ep);
this.bridged = true;
}
if (this.statusCallback) {
const params = Object.assign({}, this.callAttributes, evt);
const opts = {
url: this.statusCallback,
method: this.statusCallbackMethod,
json: true,
qs: 'GET' === this.statusCallbackMethod ? params : null,
body: 'POST' === this.statusCallbackMethod ? params : null
};
request(opts, (err) => {
if (err) this.logger.info(`TaskDial:Error sending call status to ${this.statusCallback}: ${err.message}`);
});
// ding! ding! ding! we have a winner
this._selectSingleDial(cs, sd);
this._killOutdials(); // NB: order is important
}
_selectSingleDial(cs, sd) {
this.connectTime = moment();
this.dials.delete(sd.callSid);
debug(`Dial:_selectSingleDial ep for outbound call: ${sd.ep.uuid}`);
this.ep = sd.ep;
this.dlg = sd.dlg;
this.callSid = sd.callSid;
if (this.earlyMedia) {
debug('Dial:_selectSingleDial propagating answer supervision on A leg now that B is connected');
cs.propagateAnswer();
}
this.dlg.on('destroy', () => {
this.logger.debug('Dial:_selectSingleDial called party hungup, ending dial operation');
this.ep.unbridge();
this.kill();
});
Object.assign(this.results, {
dialCallStatus: CallStatus.Completed,
dialCallSid: sd.callSid,
});
if (this.transcribeTask) this.transcribeTask.exec(cs, this.ep, this);
if (this.listenTask) this.listenTask.exec(cs, this.ep, this);
}
_killOutdials() {
for (const [callSid, sd] of Array.from(this.dials)) {
this.logger.debug(`Dial:_killOutdials killing callSid ${callSid}`);
sd.kill().catch((err) => this.logger.info(err, `Dial:_killOutdials Error killing ${callSid}`));
}
this.dials.clear();
}
_bridgeEarlyMedia(sd) {
if (this.epOther && !this.bridged) {
this.epOther.api('uuid_break', this.epOther.uuid);
this.epOther.bridge(sd.ep);
this.bridged = true;
}
}
async _actionHook(cs) {
if (this.action) {
const params = {DialCallStatus: this.dialCallStatus};
Object.assign(params, {
DialCallSid: this.dialCallSid,
DialCallDuration: this.dialCallDuration
});
const opts = {
url: this.action,
method: this.method,
json: true,
qs: 'GET' === this.method ? params : null,
body: 'POST' === this.method ? params : null
};
return new Promise((resolve, reject) => {
request(opts, (err, response, body) => {
if (err) this.logger.info(`TaskDial:_actionHook sending call status to ${this.action}: ${err.message}`);
if (body) {
this.logger.debug(body, 'got new application payload');
cs.replaceApplication(body);
}
resolve();
});
});
}
}
}
module.exports = TaskDial;

64
lib/tasks/gather.js
View File

@@ -9,7 +9,7 @@ class TaskGather extends Task {
this.preconditions = TaskPreconditions.Endpoint;
[
'action', 'finishOnKey', 'hints', 'input', 'language', 'method', 'numDigits',
'action', 'finishOnKey', 'hints', 'input', 'method', 'numDigits',
'partialResultCallback', 'partialResultCallbackMethod', 'profanityFilter',
'speechTimeout', 'timeout', 'say'
].forEach((k) => this[k] = this.data[k]);
@@ -17,13 +17,17 @@ class TaskGather extends Task {
this.partialResultCallbackMethod = this.partialResultCallbackMethod || 'POST';
this.method = this.method || 'POST';
this.timeout = (this.timeout || 5) * 1000;
this.language = this.language || 'en-US';
this.digitBuffer = '';
//this._earlyMedia = this.data.earlyMedia === true;
if (this.say) {
this.sayTask = makeTask(this.logger, {say: this.say});
this.interim = this.partialResultCallback;
if (this.data.recognizer) {
this.language = this.data.recognizer.language || 'en-US';
this.vendor = this.data.recognizer.vendor;
}
this.digitBuffer = '';
this._earlyMedia = this.data.earlyMedia === true;
if (this.say) this.sayTask = makeTask(this.logger, {say: this.say}, this);
}
get name() { return TaskName.Gather; }
@@ -34,15 +38,14 @@ class TaskGather extends Task {
}
async exec(cs, ep) {
super.exec(cs);
this.ep = ep;
this.actionHook = cs.actionHook;
this.taskInProgress = true;
try {
if (this.sayTask) {
this.sayTask.exec(cs, ep); // kicked off, _not_ waiting for it to complete
this.sayTask.on('playDone', (err) => {
if (this.taskInProgress) this._startTimer();
if (!this.killed) this._startTimer();
});
}
else this._startTimer();
@@ -56,11 +59,10 @@ class TaskGather extends Task {
ep.on('dtmf', this._onDtmf.bind(this, ep));
}
await this._waitForCompletion();
await this.awaitTaskDone();
} catch (err) {
this.logger.error(err, 'TaskGather:exec error');
}
this.taskInProgress = false;
ep.removeCustomEventListener(TranscriptionEvents.Transcription);
ep.removeCustomEventListener(TranscriptionEvents.EndOfUtterance);
}
@@ -71,10 +73,6 @@ class TaskGather extends Task {
this._resolve('killed');
}
async _waitForCompletion() {
return new Promise((resolve) => this.resolver = resolve);
}
_onDtmf(ep, evt) {
this.logger.debug(evt, 'TaskGather:_onDtmf');
if (evt.dtmf === this.finishOnKey) this._resolve('dtmf-terminator-key');
@@ -89,7 +87,7 @@ class TaskGather extends Task {
const opts = {
GOOGLE_SPEECH_USE_ENHANCED: true,
GOOGLE_SPEECH_SINGLE_UTTERANCE: true,
GOOGLE_SPEECH_MODEL: 'phone_call'
GOOGLE_SPEECH_MODEL: 'command_and_search'
};
if (this.hints) {
Object.assign(opts, {'GOOGLE_SPEECH_HINTS': this.hints.join(',')});
@@ -107,7 +105,7 @@ class TaskGather extends Task {
_startTranscribing(ep) {
ep.startTranscription({
interim: this.partialResultCallback ? true : false,
language: this.language
language: this.language || this.callSession.speechRecognizerLanguage
}).catch((err) => this.logger.error(err, 'TaskGather:_startTranscribing error'));
}
@@ -124,46 +122,30 @@ class TaskGather extends Task {
}
_killAudio() {
if (this.sayTask) {
this.sayTask.kill();
this.sayTask = null;
}
this.sayTask.kill();
}
_onTranscription(ep, evt) {
this.logger.debug(evt, 'TaskGather:_onTranscription');
if (evt.is_final) {
ep.removeCustomEventListener(TranscriptionEvents.Transcription);
ep.removeCustomEventListener(TranscriptionEvents.EndOfUtterance);
this._resolve('speech', evt);
}
else if (this.partialResultCallback) {
this.actionHook(this.partialResultCallback, 'POST', {
Speech: evt
});
}
if (evt.is_final) this._resolve('speech', evt);
else if (this.partialResultCallback) this.notifyHook(this.partialResultCallback, 'POST', null, {speech: evt});
}
_onEndOfUtterance(ep, evt) {
this.logger.info(evt, 'TaskGather:_onEndOfUtterance');
this._startTranscribing(ep);
}
_resolve(reason, evt) {
async _resolve(reason, evt) {
this.logger.debug(`TaskGather:resolve with reason ${reason}`);
assert(this.resolver);
if (reason.startsWith('dtmf')) {
this.actionHook(this.action, this.method, {
Digits: this.digitBuffer
});
this.performAction(this.method, null, {digits: this.digitBuffer});
}
else if (reason.startsWith('speech')) {
this.actionHook(this.action, this.method, {
Speech: evt
});
this.performAction(this.method, null, {speech: evt});
}
this._clearTimer();
this.resolver();
this.notifyTaskDone();
}
}

1
lib/tasks/hangup.js
View File

@@ -15,6 +15,7 @@ class TaskHangup extends Task {
* Hangup the call
*/
async exec(cs, dlg) {
super.exec(cs);
try {
await dlg.destroy({headers: this.headers});
} catch (err) {

31
lib/tasks/listen.js
View File

@@ -1,7 +1,6 @@
const Task = require('./task');
const {TaskName, TaskPreconditions, ListenEvents} = require('../utils/constants');
const makeTask = require('./make_task');
const assert = require('assert');
class TaskListen extends Task {
constructor(logger, opts) {
@@ -9,57 +8,49 @@ class TaskListen extends Task {
this.preconditions = TaskPreconditions.Endpoint;
[
'url', 'finishOnKey', 'maxLength', 'metadata', 'mixType', 'passDtmf', 'playBeep',
'action', 'url', 'finishOnKey', 'maxLength', 'metadata', 'mixType', 'passDtmf', 'playBeep',
'sampleRate', 'timeout', 'transcribe'
].forEach((k) => this[k] = this.data[k]);
this.mixType = this.mixType || 'mono';
this.sampleRate = this.sampleRate || 8000;
this.earlyMedia = this.data.earlyMedia === true;
this.results = {};
if (this.transcribe) {
this.transcribeTask = makeTask(logger, {'transcribe': opts.transcribe});
}
if (this.transcribe) this.transcribeTask = makeTask(logger, {'transcribe': opts.transcribe}, this);
this._dtmfHandler = this._onDtmf.bind(this);
this._completionPromise = new Promise((resolve) => this._completionResolver = resolve);
}
get name() { return TaskName.Listen; }
async exec(cs, ep) {
super.exec(cs);
this.ep = ep;
try {
if (this.playBeep) await this._playBeep(ep);
if (this.transcribeTask) {
this.logger.debug('TaskListen:exec - starting nested transcribe task');
this.transcribeTask.exec(cs, ep, this); // kicked off, _not_ waiting for it to complete
this.transcribeTask.exec(cs, ep, this);
}
await this._startListening(ep);
await this._completionPromise;
await this.awaitTaskDone();
} catch (err) {
this.logger.info(err, `TaskListen:exec - error ${this.url}`);
}
if (this.transcribeTask) this.transcribeTask.kill();
this._removeListeners(ep);
this.listenComplete = true;
this.emit('listenDone');
}
async kill() {
super.kill();
this._clearTimer();
if (this.ep.connected && !this.listenComplete) {
this.listenComplete = true;
if (this.transcribeTask) {
await this.transcribeTask.kill();
this.transcribeTask = null;
}
if (this.transcribeTask) await this.transcribeTask.kill();
if (this.ep.connected) {
await this.ep.forkAudioStop()
.catch((err) => this.logger.info(err, 'TaskListen:kill'));
}
this._completionResolver();
this.notifyTaskDone();
}
async _playBeep(ep) {
@@ -119,11 +110,11 @@ class TaskListen extends Task {
}
_onConnectFailure(ep, evt) {
this.logger.info(evt, 'TaskListen:_onConnectFailure');
this._completionResolver();
this.notifyTaskDone();
}
_onError(ep, evt) {
this.logger.info(evt, 'TaskListen:_onError');
this._completionResolver();
this.notifyTaskDone();
}
}

10
lib/tasks/play.js
View File

@@ -9,30 +9,28 @@ class TaskPlay extends Task {
this.url = this.data.url;
this.loop = this.data.loop || 1;
this.earlyMedia = this.data.earlyMedia === true;
this.playComplete = false;
}
get name() { return TaskName.Play; }
async exec(cs, ep) {
super.exec(cs);
this.ep = ep;
try {
while (!this.playComplete && this.loop--) {
while (!this.killed && this.loop--) {
await ep.play(this.url);
}
} catch (err) {
this.logger.info(err, `TaskPlay:exec - error playing ${this.url}`);
}
this.playComplete = true;
this.emit('playDone');
}
kill() {
async kill() {
super.kill();
if (this.ep.connected && !this.playComplete) {
this.logger.debug('TaskPlay:kill - killing audio');
this.playComplete = true;
this.ep.api('uuid_break', this.ep.uuid).catch((err) => this.logger.info(err, 'Error killing audio'));
await this.ep.api('uuid_break', this.ep.uuid).catch((err) => this.logger.info(err, 'Error killing audio'));
}
}
}

28
lib/tasks/say.js
View File

@@ -2,44 +2,44 @@ const Task = require('./task');
const {TaskName, TaskPreconditions} = require('../utils/constants');
class TaskSay extends Task {
constructor(logger, opts) {
constructor(logger, opts, parentTask) {
super(logger, opts);
this.preconditions = TaskPreconditions.Endpoint;
this.text = this.data.text;
this.voice = this.data.synthesizer.voice;
this.earlyMedia = this.data.earlyMedia === true;
switch (this.data.synthesizer.vendor) {
case 'google':
this.ttsEngine = 'google_tts';
break;
default:
throw new Error(`unsupported tts vendor ${this.data.synthesizer.vendor}`);
this.earlyMedia = this.data.earlyMedia === true || (parentTask && parentTask.earlyMedia);
if (this.data.synthesizer) {
this.voice = this.data.synthesizer.voice;
switch (this.data.synthesizer.vendor) {
case 'google':
this.ttsEngine = 'google_tts';
break;
default:
throw new Error(`unsupported tts vendor ${this.data.synthesizer.vendor}`);
}
}
this.sayComplete = false;
}
get name() { return TaskName.Say; }
async exec(cs, ep) {
super.exec(cs);
this.ep = ep;
try {
await ep.speak({
ttsEngine: 'google_tts',
voice: this.voice,
voice: this.voice || this.callSession.speechSynthesisVoice,
text: this.text
});
} catch (err) {
this.logger.info(err, 'TaskSay:exec error');
}
this.emit('playDone');
this.sayComplete = true;
}
kill() {
super.kill();
if (this.ep.connected && !this.sayComplete) {
if (this.ep.connected) {
this.logger.debug('TaskSay:kill - killing audio');
this.ep.api('uuid_break', this.ep.uuid).catch((err) => this.logger.info(err, 'Error killing audio'));
}

7
lib/tasks/sip_decline.js
View File

@@ -1,6 +1,9 @@
const Task = require('./task');
const {TaskName, TaskPreconditions} = require('../utils/constants');
/**
* Rejects an incoming call with user-specified status code and reason
*/
class TaskSipDecline extends Task {
constructor(logger, opts) {
super(logger, opts);
@@ -11,10 +14,8 @@ class TaskSipDecline extends Task {
get name() { return TaskName.SipDecline; }
/**
* Reject an incoming call attempt with a provided status code and (optionally) reason
*/
async exec(cs, {res}) {
super.exec(cs);
res.send(this.data.status, this.data.reason, {
headers: this.headers
});

15
lib/tasks/specs.json
View File

@@ -34,8 +34,7 @@
"earlyMedia": "boolean"
},
"required": [
"text",
"synthesizer"
"text"
]
},
"gather": {
@@ -72,6 +71,7 @@
"enum": ["GET", "POST"]
},
"target": ["#target"],
"url": "string",
"timeLimit": "number",
"timeout": "number",
"transcribe": "#transcribe"
@@ -82,6 +82,7 @@
},
"listen": {
"properties": {
"action": "string",
"finishOnKey": "string",
"maxLength": "number",
"metadata": "object",
@@ -103,13 +104,12 @@
},
"transcribe": {
"properties": {
"action": "string",
"transcriptionCallback": "string",
"recognizer": "#recognizer",
"earlyMedia": "boolean"
},
"required": [
"action",
"recognizer"
"transcriptionCallback"
]
},
"target": {
@@ -164,10 +164,7 @@
"hints": "array",
"profanityFilter": "boolean",
"interim": "boolean",
"mixType": {
"type": "string",
"enum": ["mono", "stereo", "mixed"]
}
"dualChannel": "boolean"
},
"required": [
"vendor"

34
lib/tasks/task.js
View File

@@ -1,8 +1,8 @@
const Emitter = require('events');
const debug = require('debug')('jambonz:feature-server');
const assert = require('assert');
const resourcesMixin = require('../utils/resources');
const {TaskPreconditions} = require('../utils/constants');
const hooks = require('../utils/notifiers');
const specs = new Map();
const _specData = require('./specs');
for (const key in _specData) {specs.set(key, _specData[key]);}
@@ -15,12 +15,24 @@ class Task extends Emitter {
this.data = data;
this._killInProgress = false;
this._completionPromise = new Promise((resolve) => this._completionResolver = resolve);
}
get killed() {
return this._killInProgress;
}
get callSession() {
return this.cs;
}
async exec(cs) {
this.cs = cs;
const {actionHook, notifyHook} = hooks(this.logger, cs.callInfo);
this.actionHook = actionHook;
this.notifyHook = notifyHook;
}
/**
* called to kill (/stop) a running task
* what to do is up to each type of task
@@ -31,6 +43,24 @@ class Task extends Emitter {
// no-op
}
notifyTaskDone() {
this._completionResolver();
}
awaitTaskDone() {
return this._completionPromise;
}
async performAction(method, auth, results) {
if (this.action) {
const tasks = await this.actionHook(this.action, method, auth, results);
if (tasks && Array.isArray(tasks)) {
this.logger.debug(`${this.name} replacing application with ${tasks.length} tasks`);
this.callSession.replaceApplication(tasks);
}
}
}
static validate(name, data) {
debug(`validating ${name} with data ${JSON.stringify(data)}`);
// validate the instruction is supported
@@ -88,7 +118,5 @@ class Task extends Emitter {
}
}
Object.assign(Task.prototype, resourcesMixin);
module.exports = Task;

67
lib/tasks/transcribe.js
View File

@@ -1,54 +1,50 @@
const Task = require('./task');
const {TaskName, TaskPreconditions, TranscriptionEvents} = require('../utils/constants');
const assert = require('assert');
class TaskTranscribe extends Task {
constructor(logger, opts) {
constructor(logger, opts, parentTask) {
super(logger, opts);
this.preconditions = TaskPreconditions.Endpoint;
this.action = this.data.action;
this.language = this.data.language || 'en-US';
this.vendor = this.data.vendor;
this.interim = this.data.interim === true;
this.mixType = this.data.mixType;
this.earlyMedia = this.data.earlyMedia === true;
this._completionPromise = new Promise((resolve) => this._completionResolver = resolve);
this.transcriptionCallback = this.data.transcriptionCallback;
this.earlyMedia = this.data.earlyMedia === true || (parentTask && parentTask.earlyMedia);
if (this.data.recognizer) {
this.language = this.data.recognizer.language || 'en-US';
this.vendor = this.data.recognizer.vendor;
this.interim = this.data.recognizer.interim === true;
this.dualChannel = this.data.recognizer.dualChannel === true;
}
}
get name() { return TaskName.Transcribe; }
async exec(cs, ep, parentTask) {
super.exec(cs);
this.ep = ep;
this.actionHook = ep.cs.actionHook;
this.transcribeInProgress = true;
try {
await this._initSpeech(ep);
await this._startTranscribing(ep);
await this._completionPromise;
await this.awaitTaskDone();
} catch (err) {
this.logger.info(err, 'TaskTranscribe:exec - error');
}
this.transcribeInProgress = true;
ep.removeCustomEventListener(TranscriptionEvents.Transcription);
ep.removeCustomEventListener(TranscriptionEvents.NoAudioDetected);
ep.removeCustomEventListener(TranscriptionEvents.MaxDurationExceeded);
}
async kill() {
super.kill();
if (this.ep.connected && this.transcribeInProgress) {
if (this.ep.connected) {
this.ep.stopTranscription().catch((err) => this.logger.info(err, 'Error TaskTranscribe:kill'));
// hangup after 1 sec if we don't get a final transcription
this._timer = setTimeout(() => this._completionResolver(), 1000);
this._timer = setTimeout(() => this.notifyTaskDone(), 1000);
}
else {
this._completionResolver();
}
await this._completionPromise;
else this.notifyTaskDone();
await this.awaitTaskDone();
}
async _initSpeech(ep) {
async _startTranscribing(ep) {
const opts = {
GOOGLE_SPEECH_USE_ENHANCED: true,
GOOGLE_SPEECH_MODEL: 'phone_call'
@@ -56,43 +52,48 @@ class TaskTranscribe extends Task {
if (this.hints) {
Object.assign(opts, {'GOOGLE_SPEECH_HINTS': this.hints.join(',')});
}
if (this.profanityFilter === true) {
if (this.profanityFilter) {
Object.assign(opts, {'GOOGLE_SPEECH_PROFANITY_FILTER': true});
}
if (this.dualChannel) {
Object.assign(opts, {'GOOGLE_SPEECH_SEPARATE_RECOGNITION_PER_CHANNEL': true});
}
await ep.set(opts)
.catch((err) => this.logger.info(err, 'TaskTranscribe:_initSpeech error setting fs vars'));
.catch((err) => this.logger.info(err, 'TaskTranscribe:_startTranscribing'));
ep.addCustomEventListener(TranscriptionEvents.Transcription, this._onTranscription.bind(this, ep));
ep.addCustomEventListener(TranscriptionEvents.NoAudioDetected, this._onNoAudio.bind(this, ep));
ep.addCustomEventListener(TranscriptionEvents.MaxDurationExceeded, this._onMaxDurationExceeded.bind(this, ep));
await this._transcribe(ep);
}
async _startTranscribing(ep) {
await ep.startTranscription({
async _transcribe(ep) {
await this.ep.startTranscription({
interim: this.interim ? true : false,
language: this.language
language: this.language || this.callSession.speechRecognizerLanguage,
channels: this.dualChannel ? 2 : 1
});
}
_onTranscription(ep, evt) {
this.logger.debug(evt, 'TaskTranscribe:_onTranscription');
this.actionHook(this.action, 'POST', {
Speech: evt
});
this.notifyHook(this.transcriptionCallback, 'POST', {speech: evt});
if (this.killed) {
this.logger.debug('TaskTranscribe:_onTranscription exiting after receiving final transcription');
this._clearTimer();
this._completionResolver();
this.notifyTaskDone();
}
}
_onNoAudio(ep) {
this.logger.debug('TaskTranscribe:_onNoAudio restarting transcription');
this._startTranscribing(ep);
this._transcribe(ep);
}
_onMaxDurationExceeded(ep) {
this.logger.debug('TaskTranscribe:_onMaxDurationExceeded restarting transcription');
this._startTranscribing(ep);
this._transcribe(ep);
}
_clearTimer() {