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 revamp of http client functionalit

Browse Source
This commit is contained in:
Dave Horton
2020-02-14 12:45:28 -05:00
parent ff531e6964
commit 446000ee97
35 changed files with 906 additions and 433 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
lib/session/call-info.js
View File

@@ -1,6 +1,10 @@
const {CallDirection, CallStatus} = require('../utils/constants');
const uuidv4 = require('uuid/v4');
/**
* @classdesc Represents the common information for all calls
* that is provided in call status webhooks
*/
class CallInfo {
constructor(opts) {
this.direction = opts.direction;
@@ -48,11 +52,20 @@ class CallInfo {
}
}
/**
* update the status of the call
* @param {string} callStatus - current call status
* @param {number} sipStatus - current sip status
*/
updateCallStatus(callStatus, sipStatus) {
this.callStatus = callStatus;
if (sipStatus) this.sipStatus = sipStatus;
}
/**
* associate customer-provided data with the call information.
* this information will be provided with every call status callhook
*/
set customerData(obj) {
this._customerData = obj;
}
@@ -84,6 +97,7 @@ class CallInfo {
}
return obj;
}
}
module.exports = CallInfo;

329
lib/session/call-session.js
View File

@@ -1,13 +1,31 @@
const Emitter = require('events');
const config = require('config');
const {CallDirection, TaskPreconditions, CallStatus, TaskName} = require('../utils/constants');
const hooks = require('../utils/notifiers');
const moment = require('moment');
const assert = require('assert');
const sessionTracker = require('./session-tracker');
const makeTask = require('../tasks/make_task');
const normalizeJamones = require('../utils/normalize-jamones');
const list = require('../utils/summarize-tasks');
const BADPRECONDITIONS = 'preconditions not met';
/**
* @classdesc Represents the execution context for a call.
* It holds the resources, such as the sip dialog and media server endpoint
* that are needed by Tasks that are operating on the call.<br/><br/>
* CallSession is a superclass object that is extended by specific types
* of sessions, such as InboundCallSession, RestCallSession and others.
*/
class CallSession extends Emitter {
/**
*
* @param {object} opts
* @param {logger} opts.logger - a pino logger
* @param {object} opts.application - the application to execute
* @param {Srf} opts.srf - the Srf instance
* @param {array} opts.tasks - tasks we are to execute
* @param {callInfo} opts.callInfo - information about the call
*/
constructor({logger, application, srf, tasks, callInfo}) {
super();
this.logger = logger;
@@ -16,9 +34,6 @@ class CallSession extends Emitter {
this.callInfo = callInfo;
this.tasks = tasks;
const {notifyHook} = hooks(this.logger, this.callInfo);
this.notifyHook = notifyHook;
this.updateCallStatus = srf.locals.dbHelpers.updateCallStatus;
this.serviceUrl = srf.locals.serviceUrl;
@@ -29,65 +44,115 @@ class CallSession extends Emitter {
sessionTracker.add(this.callSid, this);
}
/**
* callSid for the call being handled by the session
*/
get callSid() {
return this.callInfo.callSid;
}
get originalRequest() {
return this.application.originalRequest;
}
/**
* direction of the call: inbound or outbound
*/
get direction() {
return this.callInfo.direction;
}
/**
* SIP call-id for the call
*/
get callId() {
return this.callInfo.direction;
}
/**
* http endpoint to send call status updates to
*/
get call_status_hook() {
return this.application.call_status_hook;
}
/**
* can be used for all http requests within this session
*/
get requestor() {
assert(this.application.requestor);
return this.application.requestor;
}
/**
* can be used for all http call status notifications within this session
*/
get notifier() {
assert(this.application.notifier);
return this.application.notifier;
}
/**
* default vendor to use for speech synthesis if not provided in the app
*/
get speechSynthesisVendor() {
return this.application.speech_synthesis_vendor;
}
/**
* default voice to use for speech synthesis if not provided in the app
*/
get speechSynthesisVoice() {
return this.application.speech_synthesis_voice;
}
/**
* default vendor to use for speech recognition if not provided in the app
*/
get speechRecognizerVendor() {
return this.application.speech_recognizer_vendor;
}
/**
* default language to use for speech recognition if not provided in the app
*/
get speechRecognizerLanguage() {
return this.application.speech_recognizer_language;
}
/**
* indicates whether the call currently in progress
*/
get hasStableDialog() {
return this.dlg && this.dlg.connected;
}
/**
* indicates whether call is currently in a ringing state (ie not yet answered)
*/
get isOutboundCallRinging() {
return this.direction === CallDirection.Outbound && this.req && !this.dlg;
}
/**
* returns true if the call is an inbound call and a final sip response has been sent
*/
get isInboundCallAnswered() {
return this.direction === CallDirection.Inbound && this.res.finalResponseSent;
}
/**
* execute the tasks in the CallSession. The tasks are executed in sequence until
* they complete, or the caller hangs up.
* @async
*/
async exec() {
this.logger.info(`CallSession:exec starting task list with ${this.tasks.length} tasks`);
this.logger.info({tasks: list(this.tasks)}, `CallSession:exec starting ${this.tasks.length} tasks`);
while (this.tasks.length && !this.callGone) {
const taskNum = ++this.taskIdx;
const stackNum = this.stackIdx;
const task = this.tasks.shift();
this.logger.debug({task}, `CallSession:exec starting task #${stackNum}:${taskNum}: ${task.name}`);
this.logger.info(`CallSession:exec starting task #${stackNum}:${taskNum}: ${task.name}`);
try {
const resources = await this._evaluatePreconditions(task);
this.currentTask = task;
await task.exec(this, resources);
this.currentTask = null;
this.logger.debug(`CallSession:exec completed task #${stackNum}:${taskNum}: ${task.name}`);
this.logger.info(`CallSession:exec completed task #${stackNum}:${taskNum}: ${task.name}`);
} catch (err) {
this.currentTask = null;
if (err.message.includes(BADPRECONDITIONS)) {
@@ -103,16 +168,23 @@ class CallSession extends Emitter {
// all done - cleanup
this.logger.info('CallSession:exec all tasks complete');
this._onTasksDone();
this._clearCalls();
this.ms && this.ms.destroy();
this._clearResources();
sessionTracker.remove(this.callSid);
}
/**
* This is called when all tasks have completed. It is not implemented in the superclass
* but provided as a convenience for subclasses that need to do cleanup at the end of
* the call session.
*/
_onTasksDone() {
// meant to be implemented by subclass if needed
}
/**
* 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;
@@ -122,25 +194,12 @@ class CallSession extends Emitter {
}
}
normalizeUrl(url, method, auth) {
const hook = {url, method};
if (auth && auth.username && auth.password) Object.assign(hook, auth);
if (url.startsWith('/')) {
const or = this.originalRequest;
if (or) {
hook.url = `${or.baseUrl}${url}`;
hook.method = hook.method || or.method || 'POST';
if (!hook.auth && or.auth) Object.assign(hook, or.auth);
}
}
this.logger.debug({hook}, 'Task:normalizeUrl');
return hook;
}
async updateCall(opts) {
this.logger.debug(opts, 'CallSession:updateCall');
/**
* perform live call control - update call status
* @param {obj} opts
* @param {string} opts.call_status - 'complete' or 'no-answer'
*/
_lccCallStatus(opts) {
if (opts.call_status === CallStatus.Completed && this.dlg) {
this.logger.info('CallSession:updateCall hanging up call due to request from api');
this._callerHungup();
@@ -159,25 +218,139 @@ class CallSession extends Emitter {
}
}
}
else if (opts.call_hook && opts.call_hook.url) {
const hook = this.normalizeUrl(opts.call_hook.url, opts.call_hook.method, opts.call_hook.auth);
this.logger.info({hook}, 'CallSession:updateCall replacing application due to request from api');
const {actionHook} = hooks(this.logger, this.callInfo);
if (opts.call_status_hook) this.call_status_hook = opts.call_status_hook;
const tasks = await actionHook(hook);
this.logger.info({tasks}, 'CallSession:updateCall new task list');
this.replaceApplication(tasks);
}
/**
* perform live call control -- set a new call_hook
* @param {object} opts
* @param {object} opts.call_hook - new call_hook to transfer to
* @param {object} [opts.call_hook] - new call_status_hook
*/
async _lccCallHook(opts) {
const tasks = await this.requestor(opts.call_hook, this.callInfo);
//TODO: if they gave us a call status hook, we should replace
//the existing one (or just remove this option altogether?)
this.logger.info({tasks}, 'CallSession:updateCall new task list');
this.replaceApplication(tasks);
}
/**
* perform live call control -- change listen status
* @param {object} opts
* @param {string} opts.listen_status - 'pause' or 'resume'
*/
async _lccListenStatus(opts) {
const task = this.currentTask;
if (!task || ![TaskName.Dial, TaskName.Listen].includes(task.name)) {
return this.logger.info(`CallSession:updateCall - invalid listen_status in task ${task.name}`);
}
else if (opts.listen_status) {
const task = this.currentTask;
if (!task || ![TaskName.Dial, TaskName.Listen].includes(task.name)) {
return this.logger.info(`CallSession:updateCall - disregarding listen_status in task ${task.name}`);
}
const listenTask = task.name === TaskName.Listen ? task : task.listenTask;
if (!listenTask) {
return this.logger.info('CallSession:updateCall - disregarding listen_status as Dial does not have a listen');
}
listenTask.updateListen(opts.listen_status);
const listenTask = task.name === TaskName.Listen ? task : task.listenTask;
if (!listenTask) {
return this.logger.info('CallSession:updateCall - invalid listen_status: Dial does not have a listen');
}
listenTask.updateListen(opts.listen_status);
}
async _lccMuteStatus(callSid, mute) {
// this whole thing requires us to be in a Dial verb
const task = this.currentTask;
if (!task || TaskName.Dial !== task.name) {
return this.logger.info('CallSession:_lccMute - invalid command as dial is not active');
}
// now do the whisper
task.mute(callSid, mute).catch((err) => this.logger.error(err, 'CallSession:_lccMute'));
}
/**
* perform live call control -- whisper to one party or the other on a call
* @param {array} opts - array of play or say tasks
*/
async _lccWhisper(opts, callSid) {
const {whisper} = opts;
let tasks;
// this whole thing requires us to be in a Dial verb
const task = this.currentTask;
if (!task || TaskName.Dial !== task.name) {
return this.logger.info('CallSession:_lccWhisper - invalid command since we are not in a dial');
}
// 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);
tasks = normalizeJamones(this.logger, json).map((tdata) => makeTask(this.logger, tdata));
}
else if (Array.isArray(whisper)) {
// an inline array of tasks
tasks = normalizeJamones(this.logger, whisper).map((tdata) => makeTask(this.logger, tdata));
}
else if (typeof whisper === 'object') {
// a single task
tasks = normalizeJamones(this.logger, [whisper]).map((tdata) => makeTask(this.logger, tdata));
}
else {
this.logger.info({opts}, 'CallSession:_lccWhisper invalid options were provided');
return;
}
this.logger.debug(`CallSession:_lccWhisper got ${tasks.length} tasks`);
// only say or play allowed
if (tasks.find((t) => ![TaskName.Say, TaskName.Play].includes(t.name))) {
this.logger.info('CallSession:_lccWhisper invalid options where provided');
return;
}
//multiple loops not allowed
tasks.forEach((t) => t.loop = 1);
// now do the whisper
this.logger.debug(`CallSession:_lccWhisper executing ${tasks.length} tasks`);
task.whisper(tasks, callSid).catch((err) => this.logger.error(err, 'CallSession:_lccWhisper'));
}
/**
* perform live call control -- mute or unmute an endpoint
* @param {array} opts - array of play or say tasks
*/
async _lccMute(callSid, mute) {
// this whole thing requires us to be in a Dial verb
const task = this.currentTask;
if (!task || TaskName.Dial !== task.name) {
return this.logger.info('CallSession:_lccMute - not possible since we are not in a dial');
}
task.mute(callSid, mute).catch((err) => this.logger.error(err, 'CallSession:_lccMute'));
}
/**
* perform live call control
* @param {object} opts - update instructions
* @param {string} callSid - identifies call toupdate
*/
async updateCall(opts, callSid) {
this.logger.debug(opts, 'CallSession:updateCall');
if (opts.call_status) {
return this._lccCallStatus(opts);
}
if (opts.call_hook) {
return await this._lccCallHook(opts);
}
if (opts.listen_status) {
await this._lccListenStatus(opts);
}
else if (opts.mute_status) {
await this._lccMuteStatus(callSid, opts.mute_status === 'mute');
}
// whisper may be the only thing we are asked to do, or it may that
// we are doing a whisper after having muted, paused reccording etc..
if (opts.whisper) {
return this._lccWhisper(opts, callSid);
}
}
@@ -186,6 +359,10 @@ class CallSession extends Emitter {
* NB: any tasks in the current stack that have not been executed are flushed
*/
replaceApplication(tasks) {
if (this.callGone) {
this.logger.debug('CallSession:replaceApplication - ignoring because call is gone');
return;
}
this.tasks = tasks;
this.taskIdx = 0;
this.stackIdx++;
@@ -211,6 +388,10 @@ class CallSession extends Emitter {
}
}
/**
* Configure call state so as to make a media endpoint available
* @param {Task} task - task to be executed
*/
async _evalEndpointPrecondition(task) {
if (this.callGone) new Error(`${BADPRECONDITIONS}: call gone`);
@@ -229,10 +410,11 @@ class CallSession extends Emitter {
// we are going from an early media connection to answer
await answerCall();
return this.ep;
}
// need to allocate an endpoint
try {
// need to allocate an endpoint
if (!this.ms) this.ms = await this.getMS();
const ep = await this.ms.createEndpoint({remoteSdp: this.req.body});
ep.cs = this;
@@ -256,12 +438,20 @@ class CallSession extends Emitter {
}
}
/**
* Configure call state so as to make a sip dialog available
* @param {Task} task - task to be executed
*/
_evalStableCallPrecondition(task) {
if (this.callGone) throw new Error(`${BADPRECONDITIONS}: call gone`);
if (!this.dlg) throw new Error(`${BADPRECONDITIONS}: call was not answered`);
return this.dlg;
}
/**
* Throws an error if call has already been answered
* @param {Task} task - task to be executed
*/
_evalUnansweredCallPrecondition(task, callSid) {
if (!this.req) throw new Error('invalid precondition unanswered_call for outbound call');
if (this.callGone) new Error(`${BADPRECONDITIONS}: call gone`);
@@ -271,15 +461,30 @@ class CallSession extends Emitter {
return {req: this.req, res: this.res};
}
_clearCalls() {
if (this.dlg && this.dlg.connected) this.dlg.destroy();
if (this.ep && this.ep.connected) this.ep.destroy();
/**
* Hang up the call and free the media endpoint
*/
async _clearResources() {
for (const resource of [this.dlg, this.ep, this.ms]) {
try {
if (resource && resource.connected) await resource.destroy();
} catch (err) {
this.logger.error(err, 'CallSession:_clearResources error');
}
}
}
/**
* called when the caller has hung up. Provided for subclasses to override
* in order to apply logic at this point if needed.
*/
_callerHungup() {
assert(false, 'subclass responsibility to override this method');
}
/**
* get a media server to use for this call
*/
async getMS() {
if (!this.ms) {
const mrf = this.srf.locals.mrf;
@@ -288,6 +493,10 @@ class CallSession extends Emitter {
return this.ms;
}
/**
* create and endpoint if we don't have one; otherwise simply return
* the current media server and endpoint that are associated with this call
*/
async createOrRetrieveEpAndMs() {
const mrf = this.srf.locals.mrf;
if (this.ms && this.ep) return {ms: this.ms, ep: this.ep};
@@ -302,16 +511,24 @@ class CallSession extends Emitter {
return {ms: this.ms, ep: this.ep};
}
/**
* Called any time call status changes. This method both invokes the
* call_status_hook callback as well as updates the realtime database
* with latest call status
* @param {object} opts
* @param {string} callStatus - current call status
* @param {number} sipStatus - current sip status
* @param {number} [duration] - duration of a completed call, in seconds
*/
_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');
const call_status_hook = this.call_status_hook;
this.callInfo.updateCallStatus(callStatus, sipStatus);
if (typeof duration === 'number') this.callInfo.duration = duration;
try {
if (call_status_hook) this.notifyHook(call_status_hook);
this.notifier.request(this.call_status_hook, this.callInfo);
} catch (err) {
this.logger.info(err, `CallSession:_notifyCallStatusChange error sending ${callStatus} ${sipStatus}`);
}

9
lib/session/confirm-call-session.js
View File

@@ -1,5 +1,12 @@
const CallSession = require('./call-session');
/**
* @classdesc Subclass of CallSession. Represents a CallSession
* that is established for a dial verb that has a
* 'confirmUrl' application that is executed upon call answer.
* @extends CallSession
*/
class ConfirmCallSession extends CallSession {
constructor({logger, application, dlg, ep, tasks, callInfo}) {
super({
@@ -17,7 +24,7 @@ class ConfirmCallSession extends CallSession {
/**
* empty implementation to override superclass so we do not delete dlg and ep
*/
_clearCalls() {
_clearResources() {
}
}

52
lib/session/inbound-call-session.js
View File

@@ -3,6 +3,11 @@ const {CallStatus} = require('../utils/constants');
const moment = require('moment');
const assert = require('assert');
/**
* @classdesc Subclass of CallSession. This represents a CallSession that is
* established for an inbound call.
* @extends CallSession
*/
class InboundCallSession extends CallSession {
constructor(req, res) {
super({
@@ -34,46 +39,9 @@ class InboundCallSession extends CallSession {
}
}
async connectInboundCallToIvr(earlyMedia = false) {
// check for a stable inbound call already connected to the ivr
if (this.ep && this.dlg) {
this.logger.debug('CallSession:connectInboundCallToIvr - inbound call already connected to IVR');
return {ep: this.ep, dlg: this.dlg};
}
// check for an early media connection, where caller wants same
if (this.ep && earlyMedia) {
this.logger.debug('CallSession:connectInboundCallToIvr - inbound call already has early media connection');
return {ep: this.ep};
}
// ok, we need to connect the inbound call to the ivr
try {
assert(!this.req.finalResponseSent);
this.logger.debug('CallSession:connectInboundCallToIvr - creating endpoint for inbound call');
const {ep} = await this.createOrRetrieveEpAndMs();
this.ep = ep;
if (earlyMedia) {
this.res.send(183, {body: ep.local.sdp});
this.emit('callStatusChange', {sipStatus: 183, callStatus: CallStatus.EarlyMedia});
return {ep, res: this.res};
}
const dlg = await this.srf.createUAS(this.req, this.res, {localSdp: ep.local.sdp});
dlg.on('destroy', this._callerHungup.bind(this));
dlg.connectTime = moment();
this.emit('callStatusChange', {sipStatus: 200, callStatus: CallStatus.InProgress});
this.logger.debug(`CallSession:connectInboundCallToIvr - answered callSid ${this.callSid}`);
this.ep = ep;
this.dlg = dlg;
return {ep, dlg};
} catch (err) {
this.logger.error(err, 'CallSession:connectInboundCallToIvr error');
throw err;
}
}
/**
* Answer the call, if it has not already been answered.
*/
async propagateAnswer() {
if (!this.dlg) {
assert(this.ep);
@@ -85,6 +53,9 @@ class InboundCallSession extends CallSession {
}
}
/**
* This is invoked when the caller hangs up, in order to calculate the call duration.
*/
_callerHungup() {
assert(this.dlg.connectTime);
const duration = moment().diff(this.dlg.connectTime, 'seconds');
@@ -92,7 +63,6 @@ class InboundCallSession extends CallSession {
this.logger.debug('InboundCallSession: caller hung up');
this._callReleased();
}
}
module.exports = InboundCallSession;

12
lib/session/rest-call-session.js
View File

@@ -2,6 +2,11 @@ const CallSession = require('./call-session');
const {CallStatus} = require('../utils/constants');
const moment = require('moment');
/**
* @classdesc Subclass of CallSession. This represents a CallSession that is
* created for an outbound call that is initiated via the REST API.
* @extends CallSession
*/
class RestCallSession extends CallSession {
constructor({logger, application, srf, req, ep, tasks, callInfo}) {
super({
@@ -19,12 +24,19 @@ class RestCallSession extends CallSession {
this._notifyCallStatusChange({callStatus: CallStatus.Trying, sipStatus: 100});
}
/**
* Stores the sip dialog that is created when the far end answers.
* @param {Dialog} dlg - sip dialog
*/
setDialog(dlg) {
this.dlg = dlg;
dlg.on('destroy', this._callerHungup.bind(this));
dlg.connectTime = moment();
}
/**
* This is invoked when the called party hangs up, in order to calculate the call duration.
*/
_callerHungup() {
const duration = moment().diff(this.dlg.connectTime, 'seconds');
this.emit('callStatusChange', {callStatus: CallStatus.Completed, duration});

22
lib/session/session-tracker.js
View File

@@ -1,6 +1,11 @@
const Emitter = require('events');
const assert = require('assert');
/**
* @classdesc This is a singleton class that tracks active sessions in a Map indexed
* by callSid. Its function is to allow us to accept inbound REST callUpdate requests
* for a callSid and to be able to retrieve and operate on the corresponding CallSession.
*/
class SessionTracker extends Emitter {
constructor() {
super();
@@ -16,22 +21,39 @@ class SessionTracker extends Emitter {
return this._logger;
}
/**
* Adds a new CallSession to the Map
* @param {string} callSid
* @param {CallSession} callSession
*/
add(callSid, callSession) {
assert(callSid);
this.sessions.set(callSid, callSession);
this.logger.info(`SessionTracker:add callSid ${callSid}, we have ${this.sessions.size} session being tracked`);
}
/**
* Removes a CallSession from the Map
* @param {string} callSid
*/
remove(callSid) {
assert(callSid);
this.sessions.delete(callSid);
this.logger.info(`SessionTracker:remove callSid ${callSid}, we have ${this.sessions.size} being tracked`);
}
/**
* Checks if a given callSid is in the Map
* @param {string} callSid
*/
has(callSid) {
return this.sessions.has(callSid);
}
/**
* Retrieves the active CallSession for a given callSid
* @param {string} callSid
*/
get(callSid) {
return this.sessions.get(callSid);
}