elevenlabs to use http 1.1 (#57)

* elevenlabs to use http 1.1

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

* update the rest of mod tts

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

* add env variable for disabling http2

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

---------

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_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;
-      
-      /* lock and unlock session */
-      switch_core_session_t *psession = switch_core_session_locate(a->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;
-      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;
-        }
+      a->resampler = speex_resampler_init(1, 8000, a->rate, 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) {
-        switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "azure_speech_feed_tts SynthesisStarted\n");
-    };
+      speechSynthesizer->SynthesisStarted += [a](const SpeechSynthesisEventArgs& e) {
+          switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "azure_speech_feed_tts SynthesisStarted\n");
+      };
 
-    speechSynthesizer->Synthesizing += [a](const SpeechSynthesisEventArgs& e) {
-      if (a->flushed) return;
-      bool fireEvent = false;
-      CircularBuffer_t *cBuffer = (CircularBuffer_t *) a->circularBuffer;
+      speechSynthesizer->Synthesizing += [a](const SpeechSynthesisEventArgs& e) {
+        if (a->flushed) return;
+        bool fireEvent = false;
+        CircularBuffer_t *cBuffer = (CircularBuffer_t *) a->circularBuffer;
 
-      auto audioData = e.Result->GetAudioData();
-      if (a->file) {
-        fwrite(audioData->data(), 1, audioData->size(), a->file);
-      }
+        auto audioData = e.Result->GetAudioData();
+        if (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
-       * 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* end = reinterpret_cast<const uint16_t*>(audioData->data() + audioData->size());
+        /**
+         * 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
+         */
+        const uint16_t* begin = reinterpret_cast<const uint16_t*>(audioData->data());
+        const uint16_t* end = reinterpret_cast<const uint16_t*>(audioData->data() + audioData->size());
 
-      /* lock as briefly as possible */
-      switch_mutex_lock(a->mutex);
-      if (cBuffer->capacity() - cBuffer->size() < audioData->size()) {
-        cBuffer->set_capacity(cBuffer->size() + std::max( audioData->size(), (size_t)BUFFER_SIZE));
-      }
-      cBuffer->insert(cBuffer->end(), begin, end);
-      switch_mutex_unlock(a->mutex);
+        /* lock as briefly as possible */
+        switch_mutex_lock(a->mutex);
+        if (cBuffer->capacity() - cBuffer->size() < audioData->size()) {
+          cBuffer->set_capacity(cBuffer->size() + std::max( audioData->size(), (size_t)BUFFER_SIZE));
+        }
+        cBuffer->insert(cBuffer->end(), begin, end);
+        switch_mutex_unlock(a->mutex);
 
-      if (0 == a->reads++) {
-        fireEvent = true;
-      }
+        if (0 == a->reads++) {
+          fireEvent = true;
+        }
 
-      if (fireEvent && a->session_id) {
-        auto endTime = std::chrono::high_resolution_clock::now();
-        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 time_to_first_byte_ms = std::to_string(duration.count());
-        switch_core_session_t* session = switch_core_session_locate(a->session_id);
-        if (session) {
-          switch_channel_t *channel = switch_core_session_get_channel(session);
-          switch_core_session_rwunlock(session);
-          if (channel) {
-            switch_event_t *event;
-            if (switch_event_create(&event, SWITCH_EVENT_PLAYBACK_START) == SWITCH_STATUS_SUCCESS) {
-              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, "variable_tts_time_to_first_byte_ms", time_to_first_byte_ms.c_str());
-              if (a->cache_filename) {
-                switch_event_add_header_string(event, SWITCH_STACK_BOTTOM, "variable_tts_cache_filename", a->cache_filename);
+        if (fireEvent && a->session_id) {
+          auto endTime = std::chrono::high_resolution_clock::now();
+          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 time_to_first_byte_ms = std::to_string(duration.count());
+          switch_core_session_t* session = switch_core_session_locate(a->session_id);
+          if (session) {
+            switch_channel_t *channel = switch_core_session_get_channel(session);
+            switch_core_session_rwunlock(session);
+            if (channel) {
+              switch_event_t *event;
+              if (switch_event_create(&event, SWITCH_EVENT_PLAYBACK_START) == SWITCH_STATUS_SUCCESS) {
+                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, "variable_tts_time_to_first_byte_ms", time_to_first_byte_ms.c_str());
+                if (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 {
-              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");
             }
-          }else {
-            switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "speechSynthesizer->Synthesizing: channel not found\n");
           }
         }
-      }
-    };
+      };
 
-    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");
+      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");
+    } 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 ?
-      std::min((*datalen/(2 * a->samples_rate / 8000)), bufSize) :
-      std::min((*datalen/2), bufSize);
+    size_t size = std::min((*datalen/(2 * a->rate / 8000)), 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,6 @@ 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;
 
   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_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;
   }
@@ -470,8 +470,14 @@ static size_t write_cb(void *ptr, size_t size, size_t nmemb, ConnInfo_t *conn) {
           switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "write_cb: firing playback-started\n");
 
           switch_event_add_header_string(event, SWITCH_STACK_BOTTOM, "Playback-File-Type", "tts_stream");
-          if (d->reported_latency) {
-            switch_event_add_header_string(event, SWITCH_STACK_BOTTOM, "variable_tts_deepgram_reported_latency_ms", d->reported_latency);
+          if (d->reported_model_name) {
+            switch_event_add_header_string(event, SWITCH_STACK_BOTTOM, "variable_tts_deepgram_reported_model_name", d->reported_model_name);
+          }
+          if (d->reported_model_uuid) {
+            switch_event_add_header_string(event, SWITCH_STACK_BOTTOM, "variable_tts_deepgram_reported_model_uuid", d->reported_model_uuid);
+          }
+          if (d->reported_char_count) {
+            switch_event_add_header_string(event, SWITCH_STACK_BOTTOM, "variable_tts_deepgram_reported_char_count", d->reported_char_count);
           }
           if (d->request_id) {
             switch_event_add_header_string(event, SWITCH_STACK_BOTTOM, "variable_tts_deepgram_request_id", d->request_id);
@@ -549,6 +555,9 @@ static size_t header_callback(char *buffer, size_t size, size_t nitems, ConnInfo
   if (parseHeader(input, header, value)) {
     switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "recv header: %s with value %s\n", header.c_str(), value.c_str());
     if (0 == header.compare("dg-request-id")) d->request_id = strdup(value.c_str());
+    else if (0 == header.compare("dg-model-name")) d->reported_model_name = strdup(value.c_str());
+    else if (0 == header.compare("dg-model-uuid")) d->reported_model_uuid = strdup(value.c_str());
+    else if (0 == header.compare("dg-char-count")) d->reported_char_count = strdup(value.c_str());
   }
   else {
     switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "header_callback: %s\n", input.c_str());
@@ -632,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);
@@ -755,7 +764,9 @@ 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);
     // Always use deepgram at rate 8000 for helping cache audio from jambonz.
@@ -794,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

@@ -10,7 +10,9 @@ static void cleardeepgram(deepgram_t* d, int freeAll) {
   if (d->api_key) free(d->api_key);
 
   if (d->request_id) free(d->request_id);
-  if (d->reported_latency) free(d->reported_latency);
+  if (d->reported_model_name) free(d->reported_model_name);
+  if (d->reported_model_uuid) free(d->reported_model_uuid);
+  if (d->reported_char_count) free(d->reported_char_count);
   if (d->ct) free(d->ct);
   if (d->err_msg) free(d->err_msg);
   if (d->name_lookup_time_ms) free(d->name_lookup_time_ms);
@@ -22,7 +24,9 @@ static void cleardeepgram(deepgram_t* d, int freeAll) {
   d->api_key = NULL;
   d->request_id = NULL;
 
-  d->reported_latency = NULL;
+  d->reported_model_name = NULL;
+  d->reported_model_uuid = NULL;
+  d->reported_char_count = NULL;
   d->ct = NULL;
   d->err_msg = NULL;
   d->name_lookup_time_ms = NULL;

mod_deepgram_tts/mod_deepgram_tts.h

@@ -11,7 +11,14 @@ typedef struct deepgram_data {
   /* result data */
   long response_code;
   char *ct;
-  char *reported_latency;
+  // Deepgram hedaers
+  //dg-model-name
+  char *reported_model_name;
+  //dg-model-uuid
+  char *reported_model_uuid;
+  //dg-char-count
+  char *reported_char_count;
+  //dg-request-id
   char *request_id;
   char *name_lookup_time_ms;
   char *connect_time_ms;

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,7 +445,7 @@ 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;
   }
@@ -688,7 +688,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 +841,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 +851,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;
-      switch_core_session_t *psession = switch_core_session_locate(el->session_id);
-      switch_core_session_get_read_impl(psession, &read_impl);
-      uint32_t samples_per_second = !strcasecmp(read_impl.iananame, "g722") ? read_impl.actual_samples_per_second : read_impl.samples_per_second;
-      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;
-        }
+      el->resampler = speex_resampler_init(1, 8000, el->rate, 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 +886,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 +933,7 @@ extern "C" {
         switch_mutex_unlock(el->mutex);
         return SWITCH_STATUS_SUCCESS;
       }
-      size_t size = el->sample_rate ?
-        std::min((*datalen/(2 * el->sample_rate / 8000)), cBuffer->size()) :
-        std::min((*datalen/2), cBuffer->size());
+      size_t size = std::min((*datalen/(2 * el->rate / 8000)), 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.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

@@ -21,7 +21,6 @@ static void clearPlayht(playht_t* p, int freeAll) {
 
 
   if (p->request_id) free(p->request_id);
-  if (p->reported_latency) free(p->reported_latency);
   if (p->ct) free(p->ct);
   if (p->err_msg) free(p->err_msg);
   if (p->name_lookup_time_ms) free(p->name_lookup_time_ms);
@@ -43,7 +42,6 @@ static void clearPlayht(playht_t* p, int freeAll) {
   p->text_guidance = NULL;
 
   p->request_id = NULL;
-  p->reported_latency = NULL;
   p->ct = NULL;
   p->err_msg = NULL;
   p->name_lookup_time_ms = NULL;

mod_playht_tts/mod_playht_tts.h

@@ -19,7 +19,6 @@ typedef struct playht_data {
   /* result data */
   long response_code;
   char *ct;
-  char *reported_latency;
   char *request_id;
   char *name_lookup_time_ms;
   char *connect_time_ms;

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;
   }
@@ -489,9 +489,6 @@ static size_t write_cb(void *ptr, size_t size, size_t nmemb, ConnInfo_t *conn) {
           switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "write_cb: firing playback-started\n");
 
           switch_event_add_header_string(event, SWITCH_STACK_BOTTOM, "Playback-File-Type", "tts_stream");
-          if (p->reported_latency) {
-            switch_event_add_header_string(event, SWITCH_STACK_BOTTOM, "variable_tts_playht_reported_latency_ms", p->reported_latency);
-          }
           if (p->request_id) {
             switch_event_add_header_string(event, SWITCH_STACK_BOTTOM, "variable_tts_playht_request_id", p->request_id);
           }
@@ -567,8 +564,7 @@ static size_t header_callback(char *buffer, size_t size, size_t nitems, ConnInfo
   std::string input(buffer, bytes_received);
   if (parseHeader(input, header, value)) {
     switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "recv header: %s with value %s\n", header.c_str(), value.c_str());
-    if (0 == header.compare("openai-processing-ms")) p->reported_latency = strdup(value.c_str());
-    else if (0 == header.compare("x-request-id")) p->request_id = strdup(value.c_str());
+    if (0 == header.compare("x-job-ids")) p->request_id = strdup(value.c_str());
   }
   else {
     switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "header_callback: %s\n", input.c_str());
@@ -652,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);
@@ -842,17 +838,9 @@ extern "C" {
 
     p->circularBuffer = (void *) new CircularBuffer_t(8192);
 
-    if (p->session_id) {
-      int err;
-      switch_codec_implementation_t read_impl;
-      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;
-      }
+    if (mpg123_param(mh, MPG123_FORCE_RATE, p->rate /*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;
@@ -885,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

@@ -12,8 +12,6 @@ static void clearrimelabs(rimelabs_t* d, int freeAll) {
   if (d->speed_alpha) free(d->speed_alpha);
   if (d->reduce_latency) free(d->reduce_latency);
 
-  if (d->request_id) free(d->request_id);
-  if (d->reported_latency) free(d->reported_latency);
   if (d->ct) free(d->ct);
   if (d->err_msg) free(d->err_msg);
   if (d->name_lookup_time_ms) free(d->name_lookup_time_ms);
@@ -27,8 +25,6 @@ static void clearrimelabs(rimelabs_t* d, int freeAll) {
   d->speed_alpha = NULL;
   d->reduce_latency = NULL;
 
-  d->request_id = NULL;
-  d->reported_latency = NULL;
   d->ct = NULL;
   d->err_msg = NULL;
   d->name_lookup_time_ms = NULL;

mod_rimelabs_tts/mod_rimelabs_tts.h

@@ -14,8 +14,6 @@ typedef struct rimelabs_data {
   /* result data */
   long response_code;
   char *ct;
-  char *reported_latency;
-  char *request_id;
   char *name_lookup_time_ms;
   char *connect_time_ms;
   char *final_response_time_ms;

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--;
@@ -471,12 +475,6 @@ static size_t write_cb(void *ptr, size_t size, size_t nmemb, ConnInfo_t *conn) {
           switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "write_cb: firing playback-started\n");
 
           switch_event_add_header_string(event, SWITCH_STACK_BOTTOM, "Playback-File-Type", "tts_stream");
-          if (d->reported_latency) {
-            switch_event_add_header_string(event, SWITCH_STACK_BOTTOM, "variable_tts_rimelabs_reported_latency_ms", d->reported_latency);
-          }
-          if (d->request_id) {
-            switch_event_add_header_string(event, SWITCH_STACK_BOTTOM, "variable_tts_rimelabs_request_id", d->request_id);
-          }
           if (d->name_lookup_time_ms) {
             switch_event_add_header_string(event, SWITCH_STACK_BOTTOM, "variable_tts_rimelabs_name_lookup_time_ms", d->name_lookup_time_ms);
           }
@@ -549,7 +547,6 @@ static size_t header_callback(char *buffer, size_t size, size_t nitems, ConnInfo
   std::string input(buffer, bytes_received);
   if (parseHeader(input, header, value)) {
     switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "recv header: %s with value %s\n", header.c_str(), value.c_str());
-    if (0 == header.compare("dg-request-id")) d->request_id = strdup(value.c_str());
   }
   else {
     switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "header_callback: %s\n", input.c_str());
@@ -633,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);
@@ -765,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);
@@ -805,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

@@ -12,6 +12,10 @@ static void clearWhisper(whisper_t* w, int freeAll) {
   if (w->speed) free(w->speed);
   if (w->request_id) free(w->request_id);
   if (w->reported_latency) free(w->reported_latency);
+  if (w->reported_organization) free(w->reported_organization);
+  if (w->reported_ratelimit_requests) free(w->reported_ratelimit_requests);
+  if (w->reported_ratelimit_remaining_requests) free(w->reported_ratelimit_remaining_requests);
+  if (w->reported_ratelimit_reset_requests) free(w->reported_ratelimit_reset_requests);
   if (w->ct) free(w->ct);
   if (w->err_msg) free(w->err_msg);
   if (w->name_lookup_time_ms) free(w->name_lookup_time_ms);
@@ -25,6 +29,10 @@ static void clearWhisper(whisper_t* w, int freeAll) {
   w->speed = NULL;
   w->request_id = NULL;
   w->reported_latency = NULL;
+  w->reported_organization = NULL;
+  w->reported_ratelimit_requests = NULL;
+  w->reported_ratelimit_remaining_requests = NULL;
+  w->reported_ratelimit_reset_requests = NULL;
   w->ct = NULL;
   w->err_msg = NULL;
   w->name_lookup_time_ms = NULL;

mod_whisper_tts/mod_whisper_tts.h

@@ -11,7 +11,18 @@ typedef struct whisper_data {
   /* result data */
   long response_code;
   char *ct;
+  //whisper headers
+  //openai-organization
+  char *reported_organization;
+  //openai-processing-ms
   char *reported_latency;
+  //x-ratelimit-limit-requests
+  char *reported_ratelimit_requests;
+  //x-ratelimit-remaining-requests
+  char *reported_ratelimit_remaining_requests;
+  //x-ratelimit-reset-requests
+  char *reported_ratelimit_reset_requests;
+  //x-request-id
   char *request_id;
   char *name_lookup_time_ms;
   char *connect_time_ms;

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;
   }
@@ -495,6 +495,18 @@ static size_t write_cb(void *ptr, size_t size, size_t nmemb, ConnInfo_t *conn) {
           if (w->request_id) {
             switch_event_add_header_string(event, SWITCH_STACK_BOTTOM, "variable_tts_whisper_request_id", w->request_id);
           }
+          if (w->reported_organization) {
+            switch_event_add_header_string(event, SWITCH_STACK_BOTTOM, "variable_tts_whisper_reported_organization", w->reported_latency);
+          }
+          if (w->reported_ratelimit_requests) {
+            switch_event_add_header_string(event, SWITCH_STACK_BOTTOM, "variable_tts_whisper_reported_ratelimit_requests", w->reported_ratelimit_requests);
+          }
+          if (w->reported_ratelimit_remaining_requests) {
+            switch_event_add_header_string(event, SWITCH_STACK_BOTTOM, "variable_tts_whisper_reported_ratelimit_remaining_requests", w->reported_ratelimit_remaining_requests);
+          }
+          if (w->reported_ratelimit_reset_requests) {
+            switch_event_add_header_string(event, SWITCH_STACK_BOTTOM, "variable_tts_whisper_reported_ratelimit_reset_requests", w->reported_ratelimit_reset_requests);
+          }
           if (w->name_lookup_time_ms) {
             switch_event_add_header_string(event, SWITCH_STACK_BOTTOM, "variable_tts_whisper_name_lookup_time_ms", w->name_lookup_time_ms);
           }
@@ -572,6 +584,10 @@ static size_t header_callback(char *buffer, size_t size, size_t nitems, ConnInfo
     switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "recv header: %s with value %s\n", header.c_str(), value.c_str());
     if (0 == header.compare("openai-processing-ms")) w->reported_latency = strdup(value.c_str());
     else if (0 == header.compare("x-request-id")) w->request_id = strdup(value.c_str());
+    else if (0 == header.compare("openai-organization")) w->reported_organization = strdup(value.c_str());
+    else if (0 == header.compare("x-ratelimit-limit-requests")) w->reported_ratelimit_requests = strdup(value.c_str());
+    else if (0 == header.compare("x-ratelimit-remaining-requests")) w->reported_ratelimit_remaining_requests = strdup(value.c_str());
+    else if (0 == header.compare("x-ratelimit-reset-requests")) w->reported_ratelimit_reset_requests = strdup(value.c_str());
   }
   else {
     switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "header_callback: %s\n", input.c_str());
@@ -655,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);
@@ -860,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);