support aws sessionToken (#58)

* support aws sessionToken

Signed-off-by: Hoan HL <quan.luuhoang8@gmail.com>

* wip

Signed-off-by: Hoan HL <quan.luuhoang8@gmail.com>

* fix code style

---------

Signed-off-by: Hoan HL <quan.luuhoang8@gmail.com>

mod_audio_fork/audio_pipe.cpp

@@ -480,7 +480,7 @@ AudioPipe::AudioPipe(const char* uuid, const char* host, unsigned int port, cons
 }
 AudioPipe::~AudioPipe() {
   if (m_audio_buffer) delete [] m_audio_buffer;
-  if (m_recv_buf) delete [] m_recv_buf;
+  if (m_recv_buf) free(m_recv_buf);
 }
 
 void AudioPipe::connect(void) {

mod_aws_lex/aws_lex_glue.cpp

@@ -173,13 +173,12 @@ public:
 		char *metadata,
 		const char* awsAccessKeyId, 
 		const char* awsSecretAccessKey,
+		const char* awsSessionToken,
 		responseHandler_t responseHandler,
 		errorHandler_t  errorHandler) : 
 	m_bot(bot), m_alias(alias), m_region(region), m_sessionId(sessionId), m_finished(false), m_finishing(false), m_packets(0),
 	m_pStream(nullptr), m_bPlayDone(false), m_bDiscardAudio(false)
 	{
-		Aws::String key(awsAccessKeyId);
-		Aws::String secret(awsSecretAccessKey);
 		Aws::String awsLocale(locale);
 		Aws::Client::ClientConfiguration config;
 		config.region = region;
@@ -191,7 +190,10 @@ public:
 		keySnippet[19] = '\0';
 
 		switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "GStreamer %p ACCESS_KEY_ID %s\n", this, keySnippet);
-		if (*awsAccessKeyId && *awsSecretAccessKey) {
+		if (*awsAccessKeyId && *awsSecretAccessKey && *awsSessionToken) {
+			switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "using AWS creds %s %s %s\n", awsAccessKeyId, awsSecretAccessKey, awsSessionToken);	
+			m_client = Aws::MakeUnique<LexRuntimeV2Client>(ALLOC_TAG, AWSCredentials(awsAccessKeyId, awsSecretAccessKey, awsSessionToken), config);
+		} else if (*awsAccessKeyId && *awsSecretAccessKey) {
 			switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "using AWS creds %s %s\n", awsAccessKeyId, awsSecretAccessKey);	
 			m_client = Aws::MakeUnique<LexRuntimeV2Client>(ALLOC_TAG, AWSCredentials(awsAccessKeyId, awsSecretAccessKey), config);
 		}
@@ -540,7 +542,7 @@ static void *SWITCH_THREAD_FUNC lex_thread(switch_thread_t *thread, void *obj) {
 	bool ok = true;
 	switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "lex_thread: starting cb %p\n", (void *) cb);
 	GStreamer* pStreamer = new GStreamer(cb->sessionId, cb->bot, cb->alias, cb->region, cb->locale, 
-		cb->intent, cb->metadata, cb->awsAccessKeyId, cb->awsSecretAccessKey, 
+		cb->intent, cb->metadata, cb->awsAccessKeyId, cb->awsSecretAccessKey, cb->awsSessionToken,
 		cb->responseHandler, cb->errorHandler);
 	if (!pStreamer) {
 		switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "lex_thread: Error allocating streamer\n");
@@ -641,6 +643,7 @@ extern "C" {
 		memset(cb, sizeof(cb), 0);
 		const char* awsAccessKeyId = switch_channel_get_variable(channel, "AWS_ACCESS_KEY_ID");
 		const char* awsSecretAccessKey = switch_channel_get_variable(channel, "AWS_SECRET_ACCESS_KEY");
+		const char* awsSessionToken = switch_channel_get_variable(channel, "AWS_SESSION_TOKEN");
 
 		if (!hasDefaultCredentials && (!awsAccessKeyId || !awsSecretAccessKey)) {
 			switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_ERROR, 
@@ -654,6 +657,7 @@ extern "C" {
 		if (awsAccessKeyId && awsSecretAccessKey) {
 			strncpy(cb->awsAccessKeyId, awsAccessKeyId, 128);
 			strncpy(cb->awsSecretAccessKey, awsSecretAccessKey, 128);
+			if (awsSessionToken) strncpy(cb->awsSessionToken, awsSessionToken, 1024);
 		}
 		else {
 			strncpy(cb->awsAccessKeyId, std::getenv("AWS_ACCESS_KEY_ID"), 128);

mod_aws_lex/mod_aws_lex.h

@@ -30,6 +30,7 @@ struct cap_cb {
 	char sessionId[256];
   char awsAccessKeyId[128];
   char awsSecretAccessKey[128];
+	char awsSessionToken[1024];
   SpeexResamplerState *resampler;
 	void* streamer;
 	responseHandler_t responseHandler;

mod_aws_transcribe/aws_transcribe_glue.cpp

@@ -49,12 +49,11 @@ public:
 		const char* region, 
 		const char* awsAccessKeyId, 
 		const char* awsSecretAccessKey,
+		const char* awsSessionToken,
 		responseHandler_t responseHandler
   ) : m_sessionId(sessionId), m_bugname(bugname), m_finished(false), m_interim(interim), m_finishing(false), m_connected(false), m_connecting(false),
 	 		m_packets(0), m_responseHandler(responseHandler), m_pStream(nullptr), 
 			m_audioBuffer(320 * (samples_per_second == 8000 ? 1 : 2), 15) {
-		Aws::String key(awsAccessKeyId);
-		Aws::String secret(awsSecretAccessKey);
 		Aws::Client::ClientConfiguration config;
 		if (region != nullptr && strlen(region) > 0) config.region = region;
 		char keySnippet[20];
@@ -64,7 +63,9 @@ public:
 		keySnippet[19] = '\0';
 
 		switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "GStreamer %p ACCESS_KEY_ID %s, region %s\n", this, keySnippet, region);
-		if (*awsAccessKeyId && *awsSecretAccessKey) {
+		if (*awsAccessKeyId && *awsSecretAccessKey && *awsSessionToken) {
+			m_client = Aws::MakeUnique<TranscribeStreamingServiceClient>(ALLOC_TAG, AWSCredentials(awsAccessKeyId, awsSecretAccessKey, awsSessionToken), config);
+		} else if (*awsAccessKeyId && *awsSecretAccessKey) {
 			m_client = Aws::MakeUnique<TranscribeStreamingServiceClient>(ALLOC_TAG, AWSCredentials(awsAccessKeyId, awsSecretAccessKey), config);
 		}
 		else {
@@ -320,8 +321,8 @@ static void *SWITCH_THREAD_FUNC aws_transcribe_thread(switch_thread_t *thread, v
 	struct cap_cb *cb = (struct cap_cb *) obj;
 	bool ok = true;
 	switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "transcribe_thread: starting cb %p\n", (void *) cb);
-	GStreamer* pStreamer = new GStreamer(cb->sessionId, cb->bugname, cb->channels, cb->lang, cb->interim, cb->samples_per_second, cb->region, cb->awsAccessKeyId, cb->awsSecretAccessKey, 
+	GStreamer* pStreamer = new GStreamer(cb->sessionId, cb->bugname, cb->channels, cb->lang, cb->interim, cb->samples_per_second,
-		cb->responseHandler);
+		cb->region, cb->awsAccessKeyId, cb->awsSecretAccessKey, cb->awsSessionToken, cb->responseHandler);
 	if (!pStreamer) {
 		switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "transcribe_thread: Error allocating streamer\n");
 		return nullptr;
@@ -408,6 +409,7 @@ extern "C" {
 		memset(cb, sizeof(cb), 0);
 		const char* awsAccessKeyId = switch_channel_get_variable(channel, "AWS_ACCESS_KEY_ID");
 		const char* awsSecretAccessKey = switch_channel_get_variable(channel, "AWS_SECRET_ACCESS_KEY");
+		const char* awsSessionToken = switch_channel_get_variable(channel, "AWS_SESSION_TOKEN");
 		const char* awsRegion = switch_channel_get_variable(channel, "AWS_REGION");
 		cb->channels = channels;
 		LanguageCode code = LanguageCodeMapper::GetLanguageCodeForName(lang);
@@ -419,12 +421,12 @@ extern "C" {
 		strncpy(cb->sessionId, switch_core_session_get_uuid(session), MAX_SESSION_ID);
 		strncpy(cb->bugname, bugname, MAX_BUG_LEN);
 
-		if (awsAccessKeyId && awsSecretAccessKey && awsRegion) {
+    if (awsRegion) strncpy(cb->region, awsRegion, MAX_REGION);
+		if (awsAccessKeyId && awsSecretAccessKey) {
 			switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "Using channel vars for aws authentication\n");
 			strncpy(cb->awsAccessKeyId, awsAccessKeyId, 128);
 			strncpy(cb->awsSecretAccessKey, awsSecretAccessKey, 128);
-			strncpy(cb->region, awsRegion, MAX_REGION);
+			if (awsSessionToken) strncpy(cb->awsSessionToken, awsSessionToken, 1024);
-
 		}
 		else if (std::getenv("AWS_ACCESS_KEY_ID") &&
 			std::getenv("AWS_SECRET_ACCESS_KEY") &&

mod_aws_transcribe/mod_aws_transcribe.h

@@ -28,6 +28,7 @@ struct cap_cb {
 	char sessionId[MAX_SESSION_ID+1];
   char awsAccessKeyId[128];
   char awsSecretAccessKey[128];
+	char awsSessionToken[1024];
 	uint32_t channels;
   SpeexResamplerState *resampler;
 	void* streamer;

mod_azure_tts/azure_glue.cpp

@@ -59,7 +59,7 @@ extern "C" {
       return SWITCH_STATUS_FALSE;
     }
 
-    fullDirPath = std::string(baseDir) + "jambonz-tts-cache-files";
+    fullDirPath = std::string(baseDir) + "tts-cache-files";
 
     // Create the directory with read, write, and execute permissions for everyone
     mode_t oldMask = umask(0);
@@ -135,23 +135,12 @@ extern "C" {
       return SWITCH_STATUS_FALSE;
     }
 
-    if (a->session_id) {
+    if (a->rate != 8000 /*Hz*/) {
       int err;
-      switch_codec_implementation_t read_impl;
+      a->resampler = speex_resampler_init(1, 8000, a->rate, SWITCH_RESAMPLE_QUALITY, &err);
-      
+      if (0 != err) {
-      /* lock and unlock session */
+        switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Error initializing resampler: %s.\n", speex_resampler_strerror(err));
-      switch_core_session_t *psession = switch_core_session_locate(a->session_id);
+        return SWITCH_STATUS_FALSE;
-      switch_core_session_get_read_impl(psession, &read_impl);
-      switch_core_session_rwunlock(psession);
-
-      uint32_t samples_per_second = !strcasecmp(read_impl.iananame, "g722") ? read_impl.actual_samples_per_second : read_impl.samples_per_second;
-      a->samples_rate = samples_per_second;
-      if (samples_per_second != 8000 /*Hz*/) {
-        a->resampler = speex_resampler_init(1, 8000, samples_per_second, SWITCH_RESAMPLE_QUALITY, &err);
-        if (0 != err) {
-          switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Error initializing resampler: %s.\n", speex_resampler_strerror(err));
-          return SWITCH_STATUS_FALSE;
-        }
       }
     }
 
@@ -179,72 +168,77 @@ extern "C" {
 			speechConfig->SetEndpointId(a->endpointId);
 		}
 
-    auto speechSynthesizer = SpeechSynthesizer::FromConfig(speechConfig);
+    try {
+      auto speechSynthesizer = SpeechSynthesizer::FromConfig(speechConfig);
 
-    speechSynthesizer->SynthesisStarted += [a](const SpeechSynthesisEventArgs& e) {
+      speechSynthesizer->SynthesisStarted += [a](const SpeechSynthesisEventArgs& e) {
-        switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "azure_speech_feed_tts SynthesisStarted\n");
+          switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "azure_speech_feed_tts SynthesisStarted\n");
-    };
+      };
 
-    speechSynthesizer->Synthesizing += [a](const SpeechSynthesisEventArgs& e) {
+      speechSynthesizer->Synthesizing += [a](const SpeechSynthesisEventArgs& e) {
-      if (a->flushed) return;
+        if (a->flushed) return;
-      bool fireEvent = false;
+        bool fireEvent = false;
-      CircularBuffer_t *cBuffer = (CircularBuffer_t *) a->circularBuffer;
+        CircularBuffer_t *cBuffer = (CircularBuffer_t *) a->circularBuffer;
 
-      auto audioData = e.Result->GetAudioData();
+        auto audioData = e.Result->GetAudioData();
-      if (a->file) {
+        if (a->file) {
-        fwrite(audioData->data(), 1, audioData->size(), a->file);
+          fwrite(audioData->data(), 1, audioData->size(), a->file);
-      }
+        }
 
-      /**
+        /**
-       * this sort of reinterpretation can be dangerous as a general rule, but in this case we know that the data
+         * this sort of reinterpretation can be dangerous as a general rule, but in this case we know that the data
-       * is 16-bit PCM, so it's safe to do this and its much faster than copying the data byte by byte
+         * is 16-bit PCM, so it's safe to do this and its much faster than copying the data byte by byte
-       */
+         */
-      const uint16_t* begin = reinterpret_cast<const uint16_t*>(audioData->data());
+        const uint16_t* begin = reinterpret_cast<const uint16_t*>(audioData->data());
-      const uint16_t* end = reinterpret_cast<const uint16_t*>(audioData->data() + audioData->size());
+        const uint16_t* end = reinterpret_cast<const uint16_t*>(audioData->data() + audioData->size());
 
-      /* lock as briefly as possible */
+        /* lock as briefly as possible */
-      switch_mutex_lock(a->mutex);
+        switch_mutex_lock(a->mutex);
-      if (cBuffer->capacity() - cBuffer->size() < audioData->size()) {
+        if (cBuffer->capacity() - cBuffer->size() < audioData->size()) {
-        cBuffer->set_capacity(cBuffer->size() + std::max( audioData->size(), (size_t)BUFFER_SIZE));
+          cBuffer->set_capacity(cBuffer->size() + std::max( audioData->size(), (size_t)BUFFER_SIZE));
-      }
+        }
-      cBuffer->insert(cBuffer->end(), begin, end);
+        cBuffer->insert(cBuffer->end(), begin, end);
-      switch_mutex_unlock(a->mutex);
+        switch_mutex_unlock(a->mutex);
 
-      if (0 == a->reads++) {
+        if (0 == a->reads++) {
-        fireEvent = true;
+          fireEvent = true;
-      }
+        }
 
-      if (fireEvent && a->session_id) {
+        if (fireEvent && a->session_id) {
-        auto endTime = std::chrono::high_resolution_clock::now();
+          auto endTime = std::chrono::high_resolution_clock::now();
-        auto startTime = *static_cast<std::chrono::time_point<std::chrono::high_resolution_clock>*>(a->startTime);
+          auto startTime = *static_cast<std::chrono::time_point<std::chrono::high_resolution_clock>*>(a->startTime);
-        auto duration = std::chrono::duration_cast<std::chrono::milliseconds>(endTime - startTime);
+          auto duration = std::chrono::duration_cast<std::chrono::milliseconds>(endTime - startTime);
-        auto time_to_first_byte_ms = std::to_string(duration.count());
+          auto time_to_first_byte_ms = std::to_string(duration.count());
-        switch_core_session_t* session = switch_core_session_locate(a->session_id);
+          switch_core_session_t* session = switch_core_session_locate(a->session_id);
-        if (session) {
+          if (session) {
-          switch_channel_t *channel = switch_core_session_get_channel(session);
+            switch_channel_t *channel = switch_core_session_get_channel(session);
-          switch_core_session_rwunlock(session);
+            switch_core_session_rwunlock(session);
-          if (channel) {
+            if (channel) {
-            switch_event_t *event;
+              switch_event_t *event;
-            if (switch_event_create(&event, SWITCH_EVENT_PLAYBACK_START) == SWITCH_STATUS_SUCCESS) {
+              if (switch_event_create(&event, SWITCH_EVENT_PLAYBACK_START) == SWITCH_STATUS_SUCCESS) {
-              switch_channel_event_set_data(channel, event);
+                switch_channel_event_set_data(channel, event);
-              switch_event_add_header_string(event, SWITCH_STACK_BOTTOM, "Playback-File-Type", "tts_stream");
+                switch_event_add_header_string(event, SWITCH_STACK_BOTTOM, "Playback-File-Type", "tts_stream");
-              switch_event_add_header_string(event, SWITCH_STACK_BOTTOM, "variable_tts_time_to_first_byte_ms", time_to_first_byte_ms.c_str());
+                switch_event_add_header_string(event, SWITCH_STACK_BOTTOM, "variable_tts_time_to_first_byte_ms", time_to_first_byte_ms.c_str());
-              if (a->cache_filename) {
+                if (a->cache_filename) {
-                switch_event_add_header_string(event, SWITCH_STACK_BOTTOM, "variable_tts_cache_filename", a->cache_filename);
+                  switch_event_add_header_string(event, SWITCH_STACK_BOTTOM, "variable_tts_cache_filename", a->cache_filename);
+                }
+                switch_event_fire(&event);
+              } else {
+                switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "speechSynthesizer->Synthesizing: failed to create event\n");
               }
-              switch_event_fire(&event);
+            }else {
-            } else {
+              switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "speechSynthesizer->Synthesizing: channel not found\n");
-              switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "speechSynthesizer->Synthesizing: failed to create event\n");
             }
-          }else {
-            switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "speechSynthesizer->Synthesizing: channel not found\n");
           }
         }
-      }
+      };
-    };
 
-    std::thread(start_synthesis, speechSynthesizer, text, a).detach();
+      std::thread(start_synthesis, speechSynthesizer, text, a).detach();
-    switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "azure_speech_feed_tts sent synthesize request\n");
+      switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "azure_speech_feed_tts sent synthesize request\n");
+    } catch (const std::exception& e) {
+      switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "mod_azure_tts: Exception: %s\n", e.what());
+      return SWITCH_STATUS_FALSE;
+    }
     return SWITCH_STATUS_SUCCESS;
   }
 
@@ -271,9 +265,7 @@ extern "C" {
       return SWITCH_STATUS_SUCCESS;
     }
     // azure returned 8000hz 16 bit data, we have to take enough data based on call sample rate.
-    size_t size = a->samples_rate ?
+    size_t size = std::min((*datalen/(2 * a->rate / 8000)), bufSize);
-      std::min((*datalen/(2 * a->samples_rate / 8000)), bufSize) :
-      std::min((*datalen/2), bufSize);
     pcm_data.insert(pcm_data.end(), cBuffer->begin(), cBuffer->begin() + size);
     cBuffer->erase(cBuffer->begin(), cBuffer->begin() + size);
     switch_mutex_unlock(a->mutex);

mod_azure_tts/mod_azure_tts.c

@@ -82,7 +82,8 @@ static switch_status_t a_speech_feed_tts(switch_speech_handle_t *sh, char *text,
   a->draining = 0;
   a->reads = 0;
   a->flushed = 0;
-  a->samples_rate = 0;
+  a->response_code = 0;
+  a->err_msg = NULL;
 
   return azure_speech_feed_tts(a, text, flags);
 }

mod_azure_tts/mod_azure_tts.h

@@ -25,7 +25,6 @@ typedef struct azure_data {
   int reads;
   int cache_audio;
   int flushed;
-  uint32_t samples_rate;
 
   void *startTime;
 

mod_deepgram_transcribe/audio_pipe.cpp

@@ -176,7 +176,9 @@ int AudioPipe::lws_callback(struct lws *wsi,
           if (lws_is_final_fragment(wsi)) {
             if (nullptr != ap->m_recv_buf) {
               std::string msg((char *)ap->m_recv_buf, ap->m_recv_buf_ptr - ap->m_recv_buf);
-              ap->m_callback(ap->m_uuid.c_str(),  ap->m_bugname.c_str(), deepgram::AudioPipe::MESSAGE, msg.c_str(),  ap->isFinished());
+              if (!ap->m_silence_disconnect) {
+                ap->m_callback(ap->m_uuid.c_str(),  ap->m_bugname.c_str(), deepgram::AudioPipe::MESSAGE, msg.c_str(),  ap->isFinished());
+              }
               if (nullptr != ap->m_recv_buf) free(ap->m_recv_buf);
             }
             ap->m_recv_buf = ap->m_recv_buf_ptr = nullptr;
@@ -451,7 +453,7 @@ AudioPipe::AudioPipe(const char* uuid, const char* bugname, const char* host, un
   m_uuid(uuid), m_host(host), m_port(port), m_path(path), m_finished(false), m_bugname(bugname),
   m_audio_buffer_min_freespace(minFreespace), m_audio_buffer_max_len(bufLen), m_gracefulShutdown(false),
   m_audio_buffer_write_offset(LWS_PRE), m_recv_buf(nullptr), m_recv_buf_ptr(nullptr), m_useTls(useTls),
-  m_state(LWS_CLIENT_IDLE), m_wsi(nullptr), m_vhd(nullptr), m_callback(callback) {
+  m_state(LWS_CLIENT_IDLE), m_wsi(nullptr), m_vhd(nullptr), m_callback(callback), m_silence_disconnect(false) {
 
   if (apiKey) m_apiKey = apiKey;
   else m_apiKey = "";
@@ -516,6 +518,11 @@ void AudioPipe::finish() {
   bufferForSending("{\"type\": \"CloseStream\"}");
 }
 
+void AudioPipe::finish_in_silence() {
+  m_silence_disconnect = true;
+  finish();
+}
+
 void AudioPipe::waitForClose() {
   std::shared_future<void> sf(m_promise.get_future());
   sf.wait();

mod_deepgram_transcribe/audio_pipe.hpp

@@ -76,6 +76,7 @@ namespace deepgram {
 
     void close() ;
     void finish();
+    void finish_in_silence();
     void waitForClose();
     void setClosed() { m_promise.set_value(); }
     bool isFinished() { return m_finished;}
@@ -138,6 +139,7 @@ namespace deepgram {
     std::string m_bugname;
     std::promise<void> m_promise;
     bool m_useTls;
+    bool m_silence_disconnect;
   };
 
 } // namespace deepgram

mod_deepgram_transcribe/dg_transcribe_glue.cpp

@@ -22,6 +22,7 @@
 
 #define RTP_PACKETIZATION_PERIOD 20
 #define FRAME_SIZE_8000  320 /*which means each 20ms frame as 320 bytes at 8 khz (1 channel only)*/
+#define DEEPGRAM_KEEP_ALIVE_INTERVAL_SECOND 8
 
 namespace {
   static bool hasDefaultCredentials = false;
@@ -86,13 +87,17 @@ namespace {
   static const char* emptyTranscript = 
     "\"is_final\":false,\"speech_final\":false,\"channel\":{\"alternatives\":[{\"transcript\":\"\",\"confidence\":0.0,\"words\":[]}]}";
 
-  static void reaper(private_t *tech_pvt) {
+  static void reaper(private_t *tech_pvt, bool silence_disconnect) {
     std::shared_ptr<deepgram::AudioPipe> pAp;
     pAp.reset((deepgram::AudioPipe *)tech_pvt->pAudioPipe);
     tech_pvt->pAudioPipe = nullptr;
 
-    std::thread t([pAp, tech_pvt]{
+    std::thread t([pAp, tech_pvt, silence_disconnect]{
-      pAp->finish();
+      if (silence_disconnect) {
+        pAp->finish_in_silence();
+      } else {
+        pAp->finish();
+      }
       pAp->waitForClose();
       switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "%s (%u) got remote close\n", tech_pvt->sessionId, tech_pvt->id);
     });
@@ -342,61 +347,83 @@ namespace {
 
     int err;
     int useTls = true;
+    std::string host;
+    int port = 443;
+    size_t buflen = LWS_PRE + (FRAME_SIZE_8000 * desiredSampling / 8000 * channels * 1000 / RTP_PACKETIZATION_PERIOD * nAudioBufferSecs);
+
+    std::ostringstream configuration_stream;
     switch_codec_implementation_t read_impl;
     switch_channel_t *channel = switch_core_session_get_channel(session);
 
     switch_core_session_get_read_impl(session, &read_impl);
   
-    memset(tech_pvt, 0, sizeof(private_t));
-  
     std::string path;
     constructPath(session, path, desiredSampling, channels, lang, interim);
     switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "path: %s\n", path.c_str());
 
-    strncpy(tech_pvt->sessionId, switch_core_session_get_uuid(session), MAX_SESSION_ID);
-
     const char* endpoint = switch_channel_get_variable(channel, "DEEPGRAM_URI");
     if (endpoint != nullptr) {
       std::string ep(endpoint);
-
       useTls = switch_true(switch_channel_get_variable(channel, "DEEPGRAM_USE_TLS"));
 
       size_t pos = ep.find(':');
+      host = ep;
       if (pos != std::string::npos) {
-        std::string host = ep.substr(0, pos);
+        host = ep.substr(0, pos);
-        std::string port = ep.substr(pos + 1);
+        std::string strPort = ep.substr(pos + 1);
-        strncpy(tech_pvt->host, host.c_str(), MAX_WS_URL_LEN);
+        port = ::atoi(strPort.c_str());
-        tech_pvt->port = ::atoi(port.c_str());
-      }
-      else {
-        strncpy(tech_pvt->host, ep.c_str(), MAX_WS_URL_LEN);
-        tech_pvt->port = 443;
       }
       switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_INFO, 
-        "connecting to deepgram on-prem %s port %d, using tls? (%s)\n", tech_pvt->host, tech_pvt->port, useTls ? "yes" : "no");
+        "connecting to deepgram on-prem %s port %d, using tls? (%s)\n", host.c_str(), port, useTls ? "yes" : "no");
-    }
+    } else {
-    else {
+      host = "api.deepgram.com";
-      strncpy(tech_pvt->host, "api.deepgram.com", MAX_WS_URL_LEN);
-      tech_pvt->port = 443;
     }
 
+    const char* apiKey = switch_channel_get_variable(channel, "DEEPGRAM_API_KEY");
+    if (!apiKey && defaultApiKey) {
+      apiKey = defaultApiKey;
+    } else if (!apiKey && endpoint == nullptr) {
+      switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_ERROR, "no deepgram api key provided\n");
+      return SWITCH_STATUS_FALSE;
+    }
+
+    configuration_stream <<
+      host << ":" <<
+      port << ";" <<
+      path << ";" <<
+      buflen << ";" <<
+      read_impl.decoded_bytes_per_packet << ";" <<
+      apiKey << ";" <<
+      useTls;
+
+    if (tech_pvt->pAudioPipe) {
+      // stop sending keep alive
+      tech_pvt->is_keep_alive = 0;
+      if (0 != strcmp(tech_pvt->configuration, configuration_stream.str().c_str())) {
+        switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "fork_data_init: stop existing deepgram connection, old configuration %s, new configuration %s\n",
+          tech_pvt->configuration, configuration_stream.str().c_str());
+        reaper(tech_pvt, true);
+      } else {
+        switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "fork_data_init: enable existing deepgram connection\n");
+        return SWITCH_STATUS_SUCCESS;
+      }
+    } else {
+      memset(tech_pvt, 0, sizeof(private_t));
+    }
+
+    strncpy(tech_pvt->sessionId, switch_core_session_get_uuid(session), MAX_SESSION_ID);
+    strncpy(tech_pvt->host, host.c_str(), MAX_WS_URL_LEN);
+    tech_pvt->port = port;
     strncpy(tech_pvt->path, path.c_str(), MAX_PATH_LEN);   
+    strncpy(tech_pvt->configuration, configuration_stream.str().c_str(), MAX_PATH_LEN) ;
     tech_pvt->sampling = desiredSampling;
     tech_pvt->responseHandler = responseHandler;
     tech_pvt->channels = channels;
     tech_pvt->id = ++idxCallCount;
     tech_pvt->buffer_overrun_notified = 0;
+    tech_pvt->is_keep_alive = 0;
     strncpy(tech_pvt->bugname, bugname, MAX_BUG_LEN);
 
-    size_t buflen = LWS_PRE + (FRAME_SIZE_8000 * desiredSampling / 8000 * channels * 1000 / RTP_PACKETIZATION_PERIOD * nAudioBufferSecs);
-
-    const char* apiKey = switch_channel_get_variable(channel, "DEEPGRAM_API_KEY");
-    if (!apiKey && defaultApiKey) apiKey = defaultApiKey;
-    else if (!apiKey && endpoint == nullptr) {
-      switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_ERROR, "no deepgram api key provided\n");
-      return SWITCH_STATUS_FALSE;
-    }
-
     deepgram::AudioPipe* ap = new deepgram::AudioPipe(tech_pvt->sessionId, bugname, tech_pvt->host, tech_pvt->port, tech_pvt->path, 
       buflen, read_impl.decoded_bytes_per_packet, apiKey, useTls, eventCallback);
     if (!ap) {
@@ -406,18 +433,26 @@ namespace {
 
     tech_pvt->pAudioPipe = static_cast<void *>(ap);
 
-    switch_mutex_init(&tech_pvt->mutex, SWITCH_MUTEX_NESTED, switch_core_session_get_pool(session));
+    switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "connecting now\n");
+    ap->connect();
+    switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "connection in progress\n");
 
-    if (desiredSampling != sampling) {
+    if (!tech_pvt->mutex) {
-      switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "(%u) resampling from %u to %u\n", tech_pvt->id, sampling, desiredSampling);
+      switch_mutex_init(&tech_pvt->mutex, SWITCH_MUTEX_NESTED, switch_core_session_get_pool(session));
-      tech_pvt->resampler = speex_resampler_init(channels, sampling, desiredSampling, SWITCH_RESAMPLE_QUALITY, &err);
-      if (0 != err) {
-        switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_ERROR, "Error initializing resampler: %s.\n", speex_resampler_strerror(err));
-        return SWITCH_STATUS_FALSE;
-      }
     }
-    else {
+    
-      switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "(%u) no resampling needed for this call\n", tech_pvt->id);
+    if (!tech_pvt->resampler) {
+      if (desiredSampling != sampling) {
+        switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "(%u) resampling from %u to %u\n", tech_pvt->id, sampling, desiredSampling);
+        tech_pvt->resampler = speex_resampler_init(channels, sampling, desiredSampling, SWITCH_RESAMPLE_QUALITY, &err);
+        if (0 != err) {
+          switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_ERROR, "Error initializing resampler: %s.\n", speex_resampler_strerror(err));
+          return SWITCH_STATUS_FALSE;
+        }
+      }
+      else {
+        switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "(%u) no resampling needed for this call\n", tech_pvt->id);
+      }
     }
 
     switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "(%u) fork_data_init\n", tech_pvt->id);
@@ -476,9 +511,20 @@ extern "C" {
     char* lang, int interim, char* bugname, void **ppUserData)
   {
     int err;
+    switch_channel_t *channel = switch_core_session_get_channel(session);
+    switch_media_bug_t *bug = (switch_media_bug_t*) switch_channel_get_private(channel, bugname);
+    private_t* tech_pvt;
+    if (bug) {
+      switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "reuse existing kep alive deepgram connection\n");
+      tech_pvt = (private_t*) switch_core_media_bug_get_user_data(bug);
+    } else {
+      tech_pvt = (private_t *) switch_core_session_alloc(session, sizeof(private_t));
+      tech_pvt->pAudioPipe = NULL;
+      tech_pvt->is_keep_alive = 0;
+      tech_pvt->mutex = NULL;
+      tech_pvt->resampler = NULL;
+    }
 
-    // allocate per-session data structure
-    private_t* tech_pvt = (private_t *) switch_core_session_alloc(session, sizeof(private_t));
     if (!tech_pvt) {
       switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_ERROR, "error allocating memory!\n");
       return SWITCH_STATUS_FALSE;
@@ -491,16 +537,13 @@ extern "C" {
 
     *ppUserData = tech_pvt;
 
-    deepgram::AudioPipe *pAudioPipe = static_cast<deepgram::AudioPipe *>(tech_pvt->pAudioPipe);
-    switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "connecting now\n");
-    pAudioPipe->connect();
-    switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "connection in progress\n");
     return SWITCH_STATUS_SUCCESS;
   }
 
 	switch_status_t dg_transcribe_session_stop(switch_core_session_t *session,int channelIsClosing, char* bugname) {
     switch_channel_t *channel = switch_core_session_get_channel(session);
     switch_media_bug_t *bug = (switch_media_bug_t*) switch_channel_get_private(channel, bugname);
+    const bool use_single_connection = switch_true(std::getenv("DEEPGRAM_SPEECH_USE_SINGLE_CONNECTION"));
     if (!bug) {
       switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "dg_transcribe_session_stop: no bug - websocket conection already closed\n");
       return SWITCH_STATUS_FALSE;
@@ -511,6 +554,12 @@ extern "C" {
     switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "(%u) dg_transcribe_session_stop\n", id);
 
     if (!tech_pvt) return SWITCH_STATUS_FALSE;
+    if (use_single_connection && !channelIsClosing) {
+      switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "dg_transcribe_session_stop: call is running, use_single_connection is true, keep alive is activated\n", id);
+      tech_pvt->is_keep_alive = 1;
+      tech_pvt->frame_count = 0;
+      return SWITCH_STATUS_SUCCESS;
+    }
       
     // close connection and get final responses
     switch_mutex_lock(tech_pvt->mutex);
@@ -518,7 +567,7 @@ extern "C" {
     if (!channelIsClosing) switch_core_media_bug_remove(session, &bug);
 
     deepgram::AudioPipe *pAudioPipe = static_cast<deepgram::AudioPipe *>(tech_pvt->pAudioPipe);
-    if (pAudioPipe) reaper(tech_pvt);
+    if (pAudioPipe) reaper(tech_pvt, false);
     destroy_tech_pvt(tech_pvt);
     switch_mutex_unlock(tech_pvt->mutex);
     switch_mutex_destroy(tech_pvt->mutex);
@@ -532,9 +581,31 @@ extern "C" {
     size_t inuse = 0;
     bool dirty = false;
     char *p = (char *) "{\"msg\": \"buffer overrun\"}";
+    char *keep_alive = (char *) "{\"type\": \"KeepAlive\"}";
 
     if (!tech_pvt) return SWITCH_TRUE;
     
+
+    // Keep sending keep alive if there is no transcribe activity
+    if (tech_pvt->is_keep_alive && tech_pvt->pAudioPipe) {
+      deepgram::AudioPipe *pAudioPipe = static_cast<deepgram::AudioPipe *>(tech_pvt->pAudioPipe);
+      if (++tech_pvt->frame_count * 20 /*ms*/ / 1000 >= DEEPGRAM_KEEP_ALIVE_INTERVAL_SECOND) {
+        tech_pvt->frame_count = 0;
+        pAudioPipe->bufferForSending(keep_alive);
+        switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "dg_transcribe_frame: sending %s to deepgram\n", keep_alive);
+      }
+      // remove media bug buffered data
+      while (true) {
+        unsigned char data[SWITCH_RECOMMENDED_BUFFER_SIZE] = {0};
+        switch_frame_t frame = { 0 };
+        frame.data = data;
+        frame.buflen = SWITCH_RECOMMENDED_BUFFER_SIZE;
+        switch_status_t rv = switch_core_media_bug_read(bug, &frame, SWITCH_TRUE);
+        if (rv != SWITCH_STATUS_SUCCESS) break;
+      }
+      return SWITCH_TRUE;
+    }
+    
     if (switch_mutex_trylock(tech_pvt->mutex) == SWITCH_STATUS_SUCCESS) {
       if (!tech_pvt->pAudioPipe) {
         switch_mutex_unlock(tech_pvt->mutex);
@@ -545,7 +616,6 @@ extern "C" {
         switch_mutex_unlock(tech_pvt->mutex);
         return SWITCH_TRUE;
       }
-
       pAudioPipe->lockAudioBuffer();
       size_t available = pAudioPipe->binarySpaceAvailable();
       if (NULL == tech_pvt->resampler) {
@@ -587,7 +657,6 @@ extern "C" {
           if (frame.datalen) {
             spx_uint32_t out_len = available >> 1;  // space for samples which are 2 bytes
             spx_uint32_t in_len = frame.samples;
-
             speex_resampler_process_interleaved_int(tech_pvt->resampler, 
               (const spx_int16_t *) frame.data, 
               (spx_uint32_t *) &in_len, 

mod_deepgram_transcribe/mod_deepgram_transcribe.c

@@ -5,6 +5,7 @@
  */
 #include "mod_deepgram_transcribe.h"
 #include "dg_transcribe_glue.h"
+#include <stdio.h>
 
 /* Prototypes */
 SWITCH_MODULE_SHUTDOWN_FUNCTION(mod_deepgram_transcribe_shutdown);
@@ -72,8 +73,10 @@ static switch_status_t start_capture(switch_core_session_t *session, switch_medi
 	switch_codec_implementation_t read_impl = { 0 };
 	void *pUserData;
 	uint32_t samples_per_second;
+	int use_single_connection = switch_true(getenv("DEEPGRAM_SPEECH_USE_SINGLE_CONNECTION"));
+	bug = switch_channel_get_private(channel, bugname);
 
-	if (switch_channel_get_private(channel, bugname)) {
+	if (bug && !use_single_connection) {
 		switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "removing bug from previous transcribe\n");
 		do_stop(session, bugname);
 	}
@@ -90,11 +93,13 @@ static switch_status_t start_capture(switch_core_session_t *session, switch_medi
 		switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Error initializing dg speech session.\n");
 		return SWITCH_STATUS_FALSE;
 	}
-	if ((status = switch_core_media_bug_add(session, "dg_transcribe", NULL, capture_callback, pUserData, 0, flags, &bug)) != SWITCH_STATUS_SUCCESS) {
+	if (!bug || !use_single_connection) {
-		return status;
+		if ((status = switch_core_media_bug_add(session, "dg_transcribe", NULL, capture_callback, pUserData, 0, flags, &bug)) != SWITCH_STATUS_SUCCESS) {
+			return status;
+		}
+		switch_channel_set_private(channel, bugname, bug);
+		switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "added media bug for dg transcribe\n");
 	}
-  switch_channel_set_private(channel, bugname, bug);
-	switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "added media bug for dg transcribe\n");
 
 	return SWITCH_STATUS_SUCCESS;
 }

mod_deepgram_transcribe/mod_deepgram_transcribe.h

@@ -39,6 +39,9 @@ struct private_data {
   unsigned int id;
   int buffer_overrun_notified:1;
   int is_finished:1;
+  int is_keep_alive;
+  int frame_count;
+  char configuration[MAX_PATH_LEN];
 };
 
 typedef struct private_data private_t;

mod_deepgram_tts/deepgram_glue.cpp

@@ -385,7 +385,7 @@ static size_t write_cb(void *ptr, size_t size, size_t nmemb, ConnInfo_t *conn) {
   auto d = conn->deepgram;
   CircularBuffer_t *cBuffer = (CircularBuffer_t *) d->circularBuffer;
   
-  if (conn->flushed) {
+  if (conn->flushed || cBuffer == nullptr) {
     /* this will abort the transfer */
     return 0;
   }
@@ -641,7 +641,7 @@ extern "C" {
       return SWITCH_STATUS_FALSE;
     }
 
-    fullDirPath = std::string(baseDir) + "jambonz-tts-cache-files";
+    fullDirPath = std::string(baseDir) + "tts-cache-files";
 
     // Create the directory with read, write, and execute permissions for everyone
     mode_t oldMask = umask(0);
@@ -764,6 +764,8 @@ extern "C" {
     conn->file = d->file;
     conn->body = json;
     conn->flushed = false;
+    conn->has_last_byte = false;
+    conn->last_byte = 0;
 
 
     d->circularBuffer = (void *) new CircularBuffer_t(BUFFER_GROW_SIZE);
@@ -803,7 +805,10 @@ extern "C" {
     curl_easy_setopt(easy, CURLOPT_POSTFIELDS, conn->body);
     //curl_easy_setopt(easy, CURLOPT_POSTFIELDSIZE, body.length());
 
-    curl_easy_setopt(easy, CURLOPT_HTTP_VERSION, CURL_HTTP_VERSION_2_0);
+    // libcurl adding random byte to the response body that creates white noise to audio file
+    // https://github.com/curl/curl/issues/10525
+    const bool disable_http_2 = switch_true(std::getenv("DISABLE_HTTP2_FOR_TTS_STREAMING"));
+    curl_easy_setopt(easy, CURLOPT_HTTP_VERSION, disable_http_2 ? CURL_HTTP_VERSION_1_1 : CURL_HTTP_VERSION_2_0);
 
     rc = curl_multi_add_handle(global.multi, conn->easy);
     mcode_test("new_conn: curl_multi_add_handle", rc);

mod_deepgram_tts/mod_deepgram_tts.c

@@ -84,6 +84,8 @@ static switch_status_t d_speech_feed_tts(switch_speech_handle_t *sh, char *text,
   deepgram_t *d = createOrRetrievePrivateData(sh);
   d->draining = 0;
   d->reads = 0;
+  d->response_code = 0;
+  d->err_msg = NULL;
 
   switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "d_speech_feed_tts\n");
 

mod_dub/ap_http.cpp

@@ -136,7 +136,10 @@ void AudioProducerHttp::start(std::function<void(bool, const std::string&)> call
   /* call this function to close a socket */
   curl_easy_setopt(_easy, CURLOPT_CLOSESOCKETFUNCTION, &AudioProducerHttp::close_socket);
 
-  curl_easy_setopt(_easy, CURLOPT_HTTP_VERSION, CURL_HTTP_VERSION_2_0);
+  // libcurl adding random byte to the response body that creates white noise to audio file
+  // https://github.com/curl/curl/issues/10525
+  const bool disable_http_2 = switch_true(std::getenv("DISABLE_HTTP2_FOR_TTS_STREAMING"));
+  curl_easy_setopt(_easy, CURLOPT_HTTP_VERSION, disable_http_2 ? CURL_HTTP_VERSION_1_1 : CURL_HTTP_VERSION_2_0);
 
   /* keep the speed down so we don't have to buffer large amounts*/
   curl_easy_setopt(_easy, CURLOPT_MAX_RECV_SPEED_LARGE, (curl_off_t)31415);

mod_elevenlabs_tts/elevenlabs_glue.cpp

@@ -445,15 +445,12 @@ static size_t write_cb(void *ptr, size_t size, size_t nmemb, ConnInfo_t *conn) {
   CircularBuffer_t *cBuffer = (CircularBuffer_t *) el->circularBuffer;
   std::vector<uint16_t> pcm_data;
   
-  if (conn->flushed) {
+  if (conn->flushed || cBuffer == nullptr) {
     /* this will abort the transfer */
     return 0;
   }
   {
     switch_mutex_lock(el->mutex);
-    switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "write_cb: received data, response %ld\n", 
-      el->response_code);
-
     if (el->response_code > 0 && el->response_code != 200) {
       std::string body((char *) ptr, bytes_received);
       switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "write_cb: received body %s\n", body.c_str());
@@ -688,7 +685,7 @@ extern "C" {
       return SWITCH_STATUS_FALSE;
     }
 
-    fullDirPath = std::string(baseDir) + "jambonz-tts-cache-files";
+    fullDirPath = std::string(baseDir) + "tts-cache-files";
 
     // Create the directory with read, write, and execute permissions for everyone
     mode_t oldMask = umask(0);
@@ -841,7 +838,6 @@ extern "C" {
     CURL* easy = createEasyHandle();
 
     el->conn = (void *) conn ;
-    el->sample_rate = 0;
     conn->elevenlabs = el;
     conn->easy = easy;
     conn->global = &global;
@@ -852,20 +848,12 @@ extern "C" {
 
     el->circularBuffer = (void *) new CircularBuffer_t(8192);
 
-    if (el->session_id) {
+    if (el->rate != 8000 /*Hz*/) {
       int err;
-      switch_codec_implementation_t read_impl;
+      el->resampler = speex_resampler_init(1, 8000, el->rate, SWITCH_RESAMPLE_QUALITY, &err);
-      switch_core_session_t *psession = switch_core_session_locate(el->session_id);
+      if (0 != err) {
-      switch_core_session_get_read_impl(psession, &read_impl);
+        switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Error initializing resampler: %s.\n", speex_resampler_strerror(err));
-      uint32_t samples_per_second = !strcasecmp(read_impl.iananame, "g722") ? read_impl.actual_samples_per_second : read_impl.samples_per_second;
+        return SWITCH_STATUS_FALSE;
-      el->sample_rate = samples_per_second;
-      // elevenlabs output is PCMU 8000
-      if (samples_per_second != 8000 /*Hz*/) {
-        el->resampler = speex_resampler_init(1, 8000, samples_per_second, SWITCH_RESAMPLE_QUALITY, &err);
-        if (0 != err) {
-          switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Error initializing resampler: %s.\n", speex_resampler_strerror(err));
-          return SWITCH_STATUS_FALSE;
-        }
       }
     }
 
@@ -895,7 +883,10 @@ extern "C" {
     curl_easy_setopt(easy, CURLOPT_POSTFIELDS, conn->body);
     //curl_easy_setopt(easy, CURLOPT_POSTFIELDSIZE, body.length());
 
-    curl_easy_setopt(easy, CURLOPT_HTTP_VERSION, CURL_HTTP_VERSION_2_0);
+    // libcurl adding random byte to the response body that creates white noise to audio file
+    // https://github.com/curl/curl/issues/10525
+    const bool disable_http_2 = switch_true(std::getenv("DISABLE_HTTP2_FOR_TTS_STREAMING"));
+    curl_easy_setopt(easy, CURLOPT_HTTP_VERSION, disable_http_2 ? CURL_HTTP_VERSION_1_1 : CURL_HTTP_VERSION_2_0);
 
     rc = curl_multi_add_handle(global.multi, conn->easy);
     mcode_test("new_conn: curl_multi_add_handle", rc);
@@ -939,9 +930,7 @@ extern "C" {
         switch_mutex_unlock(el->mutex);
         return SWITCH_STATUS_SUCCESS;
       }
-      size_t size = el->sample_rate ?
+      size_t size = std::min((*datalen/(2 * el->rate / 8000)), cBuffer->size());
-        std::min((*datalen/(2 * el->sample_rate / 8000)), cBuffer->size()) :
-        std::min((*datalen/2), cBuffer->size());
       pcm_data.insert(pcm_data.end(), cBuffer->begin(), cBuffer->begin() + size);
       cBuffer->erase(cBuffer->begin(), cBuffer->begin() + size);
       switch_mutex_unlock(el->mutex);

mod_elevenlabs_tts/mod_elevenlabs_tts.c

@@ -106,6 +106,8 @@ static switch_status_t ell_speech_feed_tts(switch_speech_handle_t *sh, char *tex
   elevenlabs_t *el = createOrRetrievePrivateData(sh);
   el->draining = 0;
   el->reads = 0;
+  el->response_code = 0;
+  el->err_msg = NULL;
   
   switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "ell_speech_feed_tts\n");
   

mod_elevenlabs_tts/mod_elevenlabs_tts.h

@@ -31,7 +31,6 @@ struct elevenlabs_data {
   char *cache_filename;
 
 	int rate;
-  uint32_t sample_rate;
 
   void *conn;
 	FILE *file;

mod_google_transcribe/google_glue_v1.cpp

@@ -236,6 +236,7 @@ static void *SWITCH_THREAD_FUNC grpc_read_thread(switch_thread_t *thread, void *
       switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "grpc_read_thread: error %s (%d)\n", status.message().c_str(), status.code()) ;
       cJSON* json = cJSON_CreateObject();
       cJSON_AddStringToObject(json, "type", "error");
+      cJSON_AddStringToObject(json, "error_cause", "stream_response");
       cJSON_AddStringToObject(json, "error", status.message().c_str());
       char* jsonString = cJSON_PrintUnformatted(json);
       cb->responseHandler(session, jsonString, cb->bugname);
@@ -339,9 +340,10 @@ static void *SWITCH_THREAD_FUNC grpc_read_thread(switch_thread_t *thread, void *
           cb->responseHandler(session, "no_audio", cb->bugname);
         }
       }
-      else {
+      else if (status.error_code() != 0) {
         cJSON* json = cJSON_CreateObject();
         cJSON_AddStringToObject(json, "type", "error");
+        cJSON_AddStringToObject(json, "error_cause", "stream_close");
         cJSON_AddItemToObject(json, "error_code", cJSON_CreateNumber(status.error_code()));
         cJSON_AddStringToObject(json, "error_message", status.error_message().c_str());
         char* jsonString = cJSON_PrintUnformatted(json);

mod_google_transcribe/google_glue_v2.cpp

@@ -349,6 +349,7 @@ static void *SWITCH_THREAD_FUNC grpc_read_thread(switch_thread_t *thread, void *
       else if (status.error_code() != 0) {
         cJSON* json = cJSON_CreateObject();
         cJSON_AddStringToObject(json, "type", "error");
+        cJSON_AddStringToObject(json, "error_cause", "stream_close");
         cJSON_AddItemToObject(json, "error_code", cJSON_CreateNumber(status.error_code()));
         cJSON_AddStringToObject(json, "error_message", status.error_message().c_str());
         char* jsonString = cJSON_PrintUnformatted(json);

mod_playht_tts/mod_playht_tts.c

@@ -99,6 +99,8 @@ static switch_status_t p_speech_feed_tts(switch_speech_handle_t *sh, char *text,
   playht_t *p = createOrRetrievePrivateData(sh);
   p->draining = 0;
   p->reads = 0;
+  p->response_code = 0;
+  p->err_msg = NULL;
 
   switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "p_speech_feed_tts\n");
 

mod_playht_tts/playht_glue.cpp

@@ -436,7 +436,7 @@ static size_t write_cb(void *ptr, size_t size, size_t nmemb, ConnInfo_t *conn) {
   CircularBuffer_t *cBuffer = (CircularBuffer_t *) p->circularBuffer;
   std::vector<uint16_t> pcm_data;
   
-  if (conn->flushed) {
+  if (conn->flushed || cBuffer == nullptr) {
     /* this will abort the transfer */
     return 0;
   }
@@ -648,7 +648,7 @@ extern "C" {
       return SWITCH_STATUS_FALSE;
     }
 
-    fullDirPath = std::string(baseDir) + "jambonz-tts-cache-files";
+    fullDirPath = std::string(baseDir) + "tts-cache-files";
 
     // Create the directory with read, write, and execute permissions for everyone
     mode_t oldMask = umask(0);
@@ -838,17 +838,9 @@ extern "C" {
 
     p->circularBuffer = (void *) new CircularBuffer_t(8192);
 
-    if (p->session_id) {
+    if (mpg123_param(mh, MPG123_FORCE_RATE, p->rate /*Hz*/, 0) != MPG123_OK) {
-      int err;
+      switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Error mpg123_param!\n");
-      switch_codec_implementation_t read_impl;
+      return SWITCH_STATUS_FALSE;
-      switch_core_session_t *psession = switch_core_session_locate(p->session_id);
-      switch_core_session_get_read_impl(psession, &read_impl);
-      switch_core_session_rwunlock(psession);
-      uint32_t samples_per_second = !strcasecmp(read_impl.iananame, "g722") ? read_impl.actual_samples_per_second : read_impl.samples_per_second;
-      if (mpg123_param(mh, MPG123_FORCE_RATE, samples_per_second /*Hz*/, 0) != MPG123_OK) {
-        switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Error mpg123_param!\n");
-        return SWITCH_STATUS_FALSE;
-      }
     }
 
     std::ostringstream api_key_stream;
@@ -881,7 +873,10 @@ extern "C" {
     curl_easy_setopt(easy, CURLOPT_POSTFIELDS, conn->body);
     //curl_easy_setopt(easy, CURLOPT_POSTFIELDSIZE, body.length());
 
-    curl_easy_setopt(easy, CURLOPT_HTTP_VERSION, CURL_HTTP_VERSION_2_0);
+    // libcurl adding random byte to the response body that creates white noise to audio file
+    // https://github.com/curl/curl/issues/10525
+    const bool disable_http_2 = switch_true(std::getenv("DISABLE_HTTP2_FOR_TTS_STREAMING"));
+    curl_easy_setopt(easy, CURLOPT_HTTP_VERSION, disable_http_2 ? CURL_HTTP_VERSION_1_1 : CURL_HTTP_VERSION_2_0);
 
     rc = curl_multi_add_handle(global.multi, conn->easy);
     mcode_test("new_conn: curl_multi_add_handle", rc);

mod_rimelabs_tts/mod_rimelabs_tts.c

@@ -82,6 +82,8 @@ static switch_status_t d_speech_feed_tts(switch_speech_handle_t *sh, char *text,
   rimelabs_t *d = createOrRetrievePrivateData(sh);
   d->draining = 0;
   d->reads = 0;
+  d->response_code = 0;
+  d->err_msg = NULL;
 
   switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "d_speech_feed_tts\n");
 

mod_rimelabs_tts/rimelabs_glue.cpp

@@ -386,7 +386,7 @@ static size_t write_cb(void *ptr, size_t size, size_t nmemb, ConnInfo_t *conn) {
   auto d = conn->rimelabs;
   CircularBuffer_t *cBuffer = (CircularBuffer_t *) d->circularBuffer;
   
-  if (conn->flushed) {
+  if (conn->flushed || cBuffer == nullptr) {
     /* this will abort the transfer */
     return 0;
   }
@@ -394,6 +394,8 @@ static size_t write_cb(void *ptr, size_t size, size_t nmemb, ConnInfo_t *conn) {
   std::unique_ptr<uint8_t[]> combinedData;
 
   if (conn->has_last_byte) {
+    switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_WARNING, "rimelabs write_cb:  processing last byte from previous read\n");
+
     conn->has_last_byte = false;  // We'll handle the last_byte now, so toggle the flag off
 
     // Allocate memory for the new data array
@@ -415,6 +417,8 @@ static size_t write_cb(void *ptr, size_t size, size_t nmemb, ConnInfo_t *conn) {
 
   // If we now have an odd total, save the last byte for next time
   if ((total_bytes_to_process % sizeof(int16_t)) != 0) {
+    switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_WARNING, "rimelabs write_cb: got odd number of bytes %d\n", total_bytes_to_process);
+
     conn->last_byte = data[total_bytes_to_process - 1];
     conn->has_last_byte = true;
     total_bytes_to_process--;
@@ -626,7 +630,7 @@ extern "C" {
       return SWITCH_STATUS_FALSE;
     }
 
-    fullDirPath = std::string(baseDir) + "jambonz-tts-cache-files";
+    fullDirPath = std::string(baseDir) + "tts-cache-files";
 
     // Create the directory with read, write, and execute permissions for everyone
     mode_t oldMask = umask(0);
@@ -758,6 +762,8 @@ extern "C" {
     conn->file = d->file;
     conn->body = json;
     conn->flushed = false;
+    conn->has_last_byte = false;
+    conn->last_byte = 0;
     
 
     d->circularBuffer = (void *) new CircularBuffer_t(BUFFER_GROW_SIZE);
@@ -798,7 +804,10 @@ extern "C" {
     curl_easy_setopt(easy, CURLOPT_POSTFIELDS, conn->body);
     //curl_easy_setopt(easy, CURLOPT_POSTFIELDSIZE, body.length());
 
-    curl_easy_setopt(easy, CURLOPT_HTTP_VERSION, CURL_HTTP_VERSION_2_0);
+    // libcurl adding random byte to the response body that creates white noise to audio file
+    // https://github.com/curl/curl/issues/10525
+    const bool disable_http_2 = switch_true(std::getenv("DISABLE_HTTP2_FOR_TTS_STREAMING"));
+    curl_easy_setopt(easy, CURLOPT_HTTP_VERSION, disable_http_2 ? CURL_HTTP_VERSION_1_1 : CURL_HTTP_VERSION_2_0);
 
     rc = curl_multi_add_handle(global.multi, conn->easy);
     mcode_test("new_conn: curl_multi_add_handle", rc);

mod_whisper_tts/mod_whisper_tts.c

@@ -90,6 +90,8 @@ static switch_status_t w_speech_feed_tts(switch_speech_handle_t *sh, char *text,
   whisper_t *w = createOrRetrievePrivateData(sh);
   w->draining = 0;
   w->reads = 0;
+  w->response_code = 0;
+  w->err_msg = NULL;
 
   switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "w_speech_feed_tts\n");
 

mod_whisper_tts/whisper_glue.cpp

@@ -436,7 +436,7 @@ static size_t write_cb(void *ptr, size_t size, size_t nmemb, ConnInfo_t *conn) {
   CircularBuffer_t *cBuffer = (CircularBuffer_t *) w->circularBuffer;
   std::vector<uint16_t> pcm_data;
   
-  if (conn->flushed) {
+  if (conn->flushed || cBuffer == nullptr) {
     /* this will abort the transfer */
     return 0;
   }
@@ -671,7 +671,7 @@ extern "C" {
       return SWITCH_STATUS_FALSE;
     }
 
-    fullDirPath = std::string(baseDir) + "jambonz-tts-cache-files";
+    fullDirPath = std::string(baseDir) + "tts-cache-files";
 
     // Create the directory with read, write, and execute permissions for everyone
     mode_t oldMask = umask(0);
@@ -876,7 +876,10 @@ extern "C" {
     curl_easy_setopt(easy, CURLOPT_POSTFIELDS, conn->body);
     //curl_easy_setopt(easy, CURLOPT_POSTFIELDSIZE, body.length());
 
-    curl_easy_setopt(easy, CURLOPT_HTTP_VERSION, CURL_HTTP_VERSION_2_0);
+    // libcurl adding random byte to the response body that creates white noise to audio file
+    // https://github.com/curl/curl/issues/10525
+    const bool disable_http_2 = switch_true(std::getenv("DISABLE_HTTP2_FOR_TTS_STREAMING"));
+    curl_easy_setopt(easy, CURLOPT_HTTP_VERSION, disable_http_2 ? CURL_HTTP_VERSION_1_1 : CURL_HTTP_VERSION_2_0);
 
     rc = curl_multi_add_handle(global.multi, conn->easy);
     mcode_test("new_conn: curl_multi_add_handle", rc);