free maloc variable on mod_audio_fork (#54)

* fix eleevnlabs callsession stuck

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

* fix read sample rate from session is not needed

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;
   }
@@ -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,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.

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;
       }
     }
 
@@ -939,9 +930,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/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) {
-      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;

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);

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);