feat tts stream fallback (#736)

* feat tts stream fallback

* wip

* wip

* wip

* wip

* wip

* wip

* fix review comment

lib/session/call-session.js

@@ -515,6 +515,14 @@ class CallSession extends Emitter {
   set actionHookDelayActions(e) {
     this._actionHookDelayActions = e;
   }
+  // Getter/setter for current tts vendor
+  get currentTtsVendor() {
+    return this._currentTtsVendor;
+  }
+
+  set currentTtsVendor(vendor) {
+    this._currentTtsVendor = vendor;
+  }
 
   hasGlobalSttPunctuation() {
     return this._globalSttPunctuation !== undefined;

lib/tasks/say.js

@@ -102,8 +102,11 @@ class TaskSay extends TtsTask {
     ep.set({
       tts_engine: vendor,
       tts_voice: voice,
-      cache_speech_handles: 1,
+      cache_speech_handles: !cs.currentTtsVendor || cs.currentTtsVendor === vendor ? 1 : 0,
     }).catch((err) => this.logger.info({err}, 'Error setting tts_engine on endpoint'));
+    // set the current vendor on the call session
+    // If vendor is changed from the previous one, then reset the cache_speech_handles flag
+    cs.currentTtsVendor = vendor;
 
     if (!preCache) this.logger.info({vendor, language, voice, model}, 'TaskSay:exec');
     try {
@@ -239,10 +242,7 @@ class TaskSay extends TtsTask {
       label = fallbackLabel;
     }
 
-    let filepath;
+    const startFallback = async(error) => {
-    try {
-      filepath = await this._synthesizeWithSpecificVendor(cs, ep, {vendor, language, voice, label});
-    } catch (error) {
       if (fallbackVendor && this.isHandledByPrimaryProvider && !cs.hasFallbackTts) {
         this.notifyError(
           { msg: 'TTS error', details:`TTS vendor ${vendor} error: ${error}`, failover: 'in progress'});
@@ -261,6 +261,12 @@ class TaskSay extends TtsTask {
           { msg: 'TTS error', details:`TTS vendor ${vendor} error: ${error}`, failover: 'not available'});
         throw error;
       }
+    };
+    let filepath;
+    try {
+      filepath = await this._synthesizeWithSpecificVendor(cs, ep, {vendor, language, voice, label});
+    } catch (error) {
+      await startFallback(error);
     }
     this.notifyStatus({event: 'start-playback'});
 
@@ -307,8 +313,31 @@ class TaskSay extends TtsTask {
                 text
               }).catch((err) => this.logger.info({err}, 'Error adding file to cache'));
             }
+            if (this._playResolve) {
+              evt.variable_tts_error ? this._playReject(new Error(evt.variable_tts_error)) : this._playResolve();
+            }
+          });
+          // wait for playback-stop event received to confirm if the playback is successful
+          this._playPromise = new Promise((resolve, reject) => {
+            this._playResolve = resolve;
+            this._playReject = reject;
           });
           await ep.play(filepath[segment]);
+          try {
+            // wait for playback-stop event received to confirm if the playback is successful
+            await this._playPromise;
+          } catch (err) {
+            try {
+              await startFallback(err);
+            } catch (err) {
+              this.logger.info({err}, 'Error waiting for playback-stop event');
+            }
+            continue;
+          } finally {
+            this._playPromise = null;
+            this._playResolve = null;
+            this._playReject = null;
+          }
           if (filepath[segment].startsWith('say:{')) {
             const arr = /^say:\{.*\}\s*(.*)$/.exec(filepath[segment]);
             if (arr) this.logger.debug(`Say:exec complete playing streaming tts request: ${arr[1].substring(0, 64)}..`);