mod_dub support sayOnTrack Deepgram (#35)

* mod_dub support sayOnTrack Deepgram

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

* mod_dub sayOnTrack support azure

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

* wip

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

* wip

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

* wip

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

* wip

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

* wip

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

* support whisper

* wip

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

---------

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

mod_audio_fork/Makefile.am

@@ -2,9 +2,17 @@ include $(top_srcdir)/build/modmake.rulesam
 MODNAME=mod_audio_fork
 
 mod_LTLIBRARIES = mod_audio_fork.la
-mod_audio_fork_la_SOURCES  = mod_audio_fork.c lws_glue.cpp parser.cpp audio_pipe.cpp
+mod_audio_fork_la_SOURCES  = mod_audio_fork.c lws_glue.cpp parser.cpp audio_pipe.cpp vector_math.cpp
 mod_audio_fork_la_CFLAGS   = $(AM_CFLAGS)
 mod_audio_fork_la_CXXFLAGS = $(AM_CXXFLAGS) -std=c++11
 
+if USE_AVX2
+mod_audio_fork_la_CXXFLAGS += -mavx2 -DUSE_AVX2
+else
+if USE_SSE2
+mod_audio_fork_la_CXXFLAGS += -msse2 -DUSE_SSE2
+endif
+endif
+
 mod_audio_fork_la_LIBADD   = $(switch_builddir)/libfreeswitch.la
-mod_audio_fork_la_LDFLAGS  = -avoid-version -module -no-undefined -shared `pkg-config --libs libwebsockets` 
+mod_audio_fork_la_LDFLAGS  = -avoid-version -module -no-undefined -shared `pkg-config --libs libwebsockets` -lstdc++ -lboost_system -lboost_thread

mod_audio_fork/audio_pipe.cpp

@@ -1,4 +1,5 @@
 #include "audio_pipe.hpp"
+#include <switch.h>
 
 #include <cassert>
 #include <iostream>
@@ -83,7 +84,7 @@ int AudioPipe::lws_callback(struct lws *wsi,
         lwsl_err("AudioPipe::lws_service_thread LWS_CALLBACK_CLIENT_CONNECTION_ERROR: %s, response status %d\n", in ? (char *)in : "(null)", rc); 
         if (ap) {
           ap->m_state = LWS_CLIENT_FAILED;
-          ap->m_callback(ap->m_uuid.c_str(), ap->m_bugname.c_str(), AudioPipe::CONNECT_FAIL, (char *) in);
+          ap->m_callback(ap->m_uuid.c_str(), ap->m_bugname.c_str(), AudioPipe::CONNECT_FAIL, (char *) in, NULL, len);
         }
         else {
           lwsl_err("AudioPipe::lws_service_thread LWS_CALLBACK_CLIENT_CONNECTION_ERROR unable to find wsi %p..\n", wsi); 
@@ -98,7 +99,7 @@ int AudioPipe::lws_callback(struct lws *wsi,
           *ppAp = ap;
           ap->m_vhd = vhd;
           ap->m_state = LWS_CLIENT_CONNECTED;
-          ap->m_callback(ap->m_uuid.c_str(), ap->m_bugname.c_str(), AudioPipe::CONNECT_SUCCESS, NULL);
+          ap->m_callback(ap->m_uuid.c_str(), ap->m_bugname.c_str(), AudioPipe::CONNECT_SUCCESS, NULL, NULL, len);
         }
         else {
           lwsl_err("AudioPipe::lws_service_thread LWS_CALLBACK_CLIENT_ESTABLISHED %s unable to find wsi %p..\n", ap->m_uuid.c_str(), wsi); 
@@ -114,12 +115,12 @@ int AudioPipe::lws_callback(struct lws *wsi,
         }
         if (ap->m_state == LWS_CLIENT_DISCONNECTING) {
           // closed by us
-          ap->m_callback(ap->m_uuid.c_str(), ap->m_bugname.c_str(), AudioPipe::CONNECTION_CLOSED_GRACEFULLY, NULL);
+          ap->m_callback(ap->m_uuid.c_str(), ap->m_bugname.c_str(), AudioPipe::CONNECTION_CLOSED_GRACEFULLY, NULL, NULL, len);
         }
         else if (ap->m_state == LWS_CLIENT_CONNECTED) {
           // closed by far end
           lwsl_notice("%s socket closed by far end\n", ap->m_uuid.c_str());
-          ap->m_callback(ap->m_uuid.c_str(), ap->m_bugname.c_str(), AudioPipe::CONNECTION_DROPPED, NULL);
+          ap->m_callback(ap->m_uuid.c_str(), ap->m_bugname.c_str(), AudioPipe::CONNECTION_DROPPED, NULL, NULL, len);
         }
         ap->m_state = LWS_CLIENT_DISCONNECTED;
 
@@ -133,14 +134,19 @@ int AudioPipe::lws_callback(struct lws *wsi,
 
     case LWS_CALLBACK_CLIENT_RECEIVE:
       {
+        
         AudioPipe* ap = *ppAp;
         if (!ap) {
           lwsl_err("AudioPipe::lws_service_thread LWS_CALLBACK_CLIENT_RECEIVE %s unable to find wsi %p..\n", ap->m_uuid.c_str(), wsi); 
           return 0;
         }
-
+        
         if (lws_frame_is_binary(wsi)) {
-          lwsl_err("AudioPipe::lws_service_thread LWS_CALLBACK_CLIENT_RECEIVE received binary frame, discarding.\n");
+          if (ap->is_bidirectional_audio_stream()) {
+            ap->m_callback(ap->m_uuid.c_str(), ap->m_bugname.c_str(), AudioPipe::BINARY, NULL, (char *) in, len);
+          } else {
+            lwsl_err("AudioPipe::lws_service_thread LWS_CALLBACK_CLIENT_RECEIVE received binary frame, discarding.\n");
+          }
           return 0;
         }
 
@@ -180,7 +186,7 @@ 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(), AudioPipe::MESSAGE, msg.c_str());
+              ap->m_callback(ap->m_uuid.c_str(), ap->m_bugname.c_str(), AudioPipe::MESSAGE, msg.c_str(), NULL, len);
               if (nullptr != ap->m_recv_buf) free(ap->m_recv_buf);
             }
             ap->m_recv_buf = ap->m_recv_buf_ptr = nullptr;
@@ -458,7 +464,8 @@ bool AudioPipe::deinitialize() {
 
 // instance members
 AudioPipe::AudioPipe(const char* uuid, const char* host, unsigned int port, const char* path,
-  int sslFlags, size_t bufLen, size_t minFreespace, const char* username, const char* password, char* bugname, notifyHandler_t callback) :
+  int sslFlags, size_t bufLen, size_t minFreespace, const char* username, const char* password, char* bugname,
+  int bidirectional_audio_stream, notifyHandler_t callback) :
   m_uuid(uuid), m_host(host), m_port(port), m_path(path), m_sslFlags(sslFlags),
   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_bugname(bugname),
@@ -468,7 +475,7 @@ AudioPipe::AudioPipe(const char* uuid, const char* host, unsigned int port, cons
     m_username.assign(username);
     m_password.assign(password);
   }
-
+  m_bidirectional_audio_stream = bidirectional_audio_stream;
   m_audio_buffer = new uint8_t[m_audio_buffer_max_len];
 }
 AudioPipe::~AudioPipe() {

mod_audio_fork/audio_pipe.hpp

@@ -27,10 +27,11 @@ namespace drachtio {
       CONNECT_FAIL,
       CONNECTION_DROPPED,
       CONNECTION_CLOSED_GRACEFULLY,
-      MESSAGE
+      MESSAGE,
+      BINARY
     };
     typedef void (*log_emit_function)(int level, const char *line);
-    typedef void (*notifyHandler_t)(const char *sessionId, const char* bugname, NotifyEvent_t event, const char* message);
+    typedef void (*notifyHandler_t)(const char *sessionId, const char* bugname, NotifyEvent_t event, const char* message, const char* binary, size_t binary_len );
 
     struct lws_per_vhost_data {
       struct lws_context *context;
@@ -44,7 +45,8 @@ namespace drachtio {
 
     // constructor
     AudioPipe(const char* uuid, const char* host, unsigned int port, const char* path, int sslFlags, 
-      size_t bufLen, size_t minFreespace, const char* username, const char* password, char* bugname, notifyHandler_t callback);
+      size_t bufLen, size_t minFreespace, const char* username, const char* password, char* bugname,
+      int bidirectional_audio, notifyHandler_t callback);
     ~AudioPipe();  
 
     LwsState_t getLwsState(void) { return m_state; }
@@ -83,6 +85,10 @@ namespace drachtio {
       return m_gracefulShutdown;
     }
 
+    bool is_bidirectional_audio_stream() {
+      return m_bidirectional_audio_stream;
+    }
+
     void close() ;
 
     // no default constructor or copying
@@ -142,6 +148,7 @@ namespace drachtio {
     std::string m_username;
     std::string m_password;
     bool m_gracefulShutdown;
+    bool m_bidirectional_audio_stream;
   };
 
 } // namespace drachtio

mod_audio_fork/lws_glue.cpp

@@ -17,9 +17,15 @@
 #include "parser.hpp"
 #include "mod_audio_fork.h"
 #include "audio_pipe.hpp"
+#include "vector_math.h"
+
+#include <boost/circular_buffer.hpp>
+
+typedef boost::circular_buffer<uint16_t> CircularBuffer_t;
 
 #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 BUFFER_GROW_SIZE (8192)
 
 namespace {
   static const char *requestedBufferSecs = std::getenv("MOD_AUDIO_FORK_BUFFER_SECS");
@@ -31,6 +37,48 @@ namespace {
   static unsigned int idxCallCount = 0;
   static uint32_t playCount = 0;
 
+  switch_status_t processIncomingBinary(private_t* tech_pvt, switch_core_session_t* session, const char* message, size_t dataLength) {
+    CircularBuffer_t *cBuffer = (CircularBuffer_t *) tech_pvt->circularBuffer;
+    uint8_t* data = reinterpret_cast<uint8_t*>(const_cast<char*>(message));
+    uint16_t* data_uint16 = reinterpret_cast<uint16_t*>(data);
+    std::vector<uint16_t> pcm_data(data_uint16, data_uint16 + dataLength / sizeof(uint16_t));
+
+
+    if (tech_pvt->bidirectional_audio_resampler) {
+      std::vector<int16_t> in(pcm_data.begin(), pcm_data.end());
+
+      std::vector<int16_t> out(dataLength);
+      spx_uint32_t in_len = pcm_data.size();
+      spx_uint32_t out_len = out.size();
+      speex_resampler_process_interleaved_int(tech_pvt->bidirectional_audio_resampler, in.data(), &in_len, out.data(), &out_len);
+
+      if (out_len > out.size()) {
+        switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_CRIT, "Resampler output exceeded maximum buffer size!\n");
+        return SWITCH_STATUS_FALSE;
+      }
+
+      // Resize the pcm_data to match the output length from resampler, and then copy the resampled data into it.
+      pcm_data.resize(out_len);
+      memcpy(pcm_data.data(), out.data(), out_len * sizeof(int16_t));
+    }
+    switch_mutex_lock(tech_pvt->mutex);
+
+    // Resize the buffer if necessary
+    size_t bytesResampled = pcm_data.size() * sizeof(uint16_t);
+    if (cBuffer->capacity() - cBuffer->size() < bytesResampled / sizeof(uint16_t)) {
+      // If buffer exceeds some max size, you could return SWITCH_STATUS_FALSE to abort the transfer
+      // if (cBuffer->size() + std::max(bytesResampled / sizeof(uint16_t), (size_t)BUFFER_GROW_SIZE) > MAX_BUFFER_SIZE) return SWITCH_STATUS_FALSE;
+
+      cBuffer->set_capacity(cBuffer->size() + std::max(bytesResampled / sizeof(uint16_t), (size_t)BUFFER_GROW_SIZE));
+    }
+    // Push the data into the buffer.
+    cBuffer->insert(cBuffer->end(), pcm_data.begin(), pcm_data.end());
+
+    switch_mutex_unlock(tech_pvt->mutex);
+
+    return SWITCH_STATUS_SUCCESS;
+}
+
   void processIncomingMessage(private_t* tech_pvt, switch_core_session_t* session, const char* message) {
     std::string msg = message;
     std::string type;
@@ -38,7 +86,10 @@ namespace {
     if (json) {
       switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "(%u) processIncomingMessage - received %s message\n", tech_pvt->id, type.c_str());
       cJSON* jsonData = cJSON_GetObjectItem(json, "data");
-      if (0 == type.compare("playAudio")) {
+      if (0 == type.compare("playAudio") &&
+        // playAudio is enabled and there is no bidirectional audio from stream is enabled.
+        tech_pvt->bidirectional_audio_enable &&
+        !tech_pvt->bidirectional_audio_stream) {
         if (jsonData) {
           // dont send actual audio bytes in event message
           cJSON* jsonFile = NULL;
@@ -156,7 +207,7 @@ namespace {
     }
   }
 
-  static void eventCallback(const char* sessionId, const char* bugname, drachtio::AudioPipe::NotifyEvent_t event, const char* message) {
+  static void eventCallback(const char* sessionId, const char* bugname, drachtio::AudioPipe::NotifyEvent_t event, const char* message, const char* binary, size_t len) {
     switch_core_session_t* session = switch_core_session_locate(sessionId);
     if (session) {
       switch_channel_t *channel = switch_core_session_get_channel(session);
@@ -198,6 +249,9 @@ namespace {
             case drachtio::AudioPipe::MESSAGE:
               processIncomingMessage(tech_pvt, session, message);
             break;
+            case drachtio::AudioPipe::BINARY:
+            processIncomingBinary(tech_pvt, session, binary, len);
+            break;
           }
         }
       }
@@ -206,11 +260,13 @@ namespace {
   }
   switch_status_t fork_data_init(private_t *tech_pvt, switch_core_session_t *session, char * host, 
     unsigned int port, char* path, int sslFlags, int sampling, int desiredSampling, int channels, 
-    char *bugname, char* metadata, responseHandler_t responseHandler) {
+    char *bugname, char* metadata, int bidirectional_audio_enable,
+    int bidirectional_audio_stream, int bidirectional_audio_sample_rate, responseHandler_t responseHandler) {
 
     const char* username = nullptr;
     const char* password = nullptr;
     int err;
+    int bidirectional_audio_stream_enable = bidirectional_audio_enable + bidirectional_audio_stream;
     switch_codec_implementation_t read_impl;
     switch_channel_t *channel = switch_core_session_get_channel(session);
 
@@ -234,13 +290,17 @@ namespace {
     tech_pvt->buffer_overrun_notified = 0;
     tech_pvt->audio_paused = 0;
     tech_pvt->graceful_shutdown = 0;
+    tech_pvt->circularBuffer = (void *) new CircularBuffer_t(8192);
+    tech_pvt->bidirectional_audio_enable = bidirectional_audio_enable;
+    tech_pvt->bidirectional_audio_stream = bidirectional_audio_stream;
+    tech_pvt->bidirectional_audio_sample_rate = bidirectional_audio_sample_rate;
     strncpy(tech_pvt->bugname, bugname, MAX_BUG_LEN);
     if (metadata) strncpy(tech_pvt->initialMetadata, metadata, MAX_METADATA_LEN);
     
     size_t buflen = LWS_PRE + (FRAME_SIZE_8000 * desiredSampling / 8000 * channels * 1000 / RTP_PACKETIZATION_PERIOD * nAudioBufferSecs);
 
     drachtio::AudioPipe* ap = new drachtio::AudioPipe(tech_pvt->sessionId, host, port, path, sslFlags, 
-      buflen, read_impl.decoded_bytes_per_packet, username, password, bugname, eventCallback);
+      buflen, read_impl.decoded_bytes_per_packet, username, password, bugname, bidirectional_audio_stream_enable, eventCallback);
     if (!ap) {
       switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_ERROR, "Error allocating AudioPipe\n");
       return SWITCH_STATUS_FALSE;
@@ -262,6 +322,15 @@ namespace {
       switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "(%u) no resampling needed for this call\n", tech_pvt->id);
     }
 
+    if (bidirectional_audio_sample_rate && sampling != bidirectional_audio_sample_rate) {
+      switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "(%u) bidirectional audio resampling from %u to %u\n", tech_pvt->id, bidirectional_audio_sample_rate, sampling);
+      tech_pvt->bidirectional_audio_resampler = speex_resampler_init(channels, bidirectional_audio_sample_rate, sampling, SWITCH_RESAMPLE_QUALITY, &err);
+      if (0 != err) {
+        switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_ERROR, "Error initializing bidirectional audio resampler: %s.\n", speex_resampler_strerror(err));
+        return SWITCH_STATUS_FALSE;
+      }
+    }
+
     switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "(%u) fork_data_init\n", tech_pvt->id);
 
     return SWITCH_STATUS_SUCCESS;
@@ -273,10 +342,19 @@ namespace {
       speex_resampler_destroy(tech_pvt->resampler);
       tech_pvt->resampler = nullptr;
     }
+    if (tech_pvt->bidirectional_audio_resampler) {
+      speex_resampler_destroy(tech_pvt->bidirectional_audio_resampler);
+      tech_pvt->bidirectional_audio_resampler = nullptr;
+    }
     if (tech_pvt->mutex) {
       switch_mutex_destroy(tech_pvt->mutex);
       tech_pvt->mutex = nullptr;
     }
+    if (tech_pvt->circularBuffer) {
+      CircularBuffer_t *cBuffer = (CircularBuffer_t *) tech_pvt->circularBuffer;
+      delete cBuffer;
+      tech_pvt->circularBuffer = nullptr;
+    }
   }
 
   void lws_logger(int level, const char *line) {
@@ -392,18 +470,22 @@ extern "C" {
   }
 
   switch_status_t fork_session_init(switch_core_session_t *session, 
-              responseHandler_t responseHandler,
-              uint32_t samples_per_second, 
-              char *host,
-              unsigned int port,
-              char *path,
-              int sampling,
-              int sslFlags,
-              int channels,
-              char *bugname,
-              char* metadata, 
-              void **ppUserData)
-  {    	
+    responseHandler_t responseHandler,
+    uint32_t samples_per_second, 
+    char *host,
+    unsigned int port,
+    char *path,
+    int sampling,
+    int sslFlags,
+    int channels,
+    char *bugname,
+    char* metadata,
+    int bidirectional_audio_enable,
+    int bidirectional_audio_stream,
+    int bidirectional_audio_sample_rate,
+    void **ppUserData
+    )
+  {
     int err;
 
     // allocate per-session data structure
@@ -412,8 +494,9 @@ extern "C" {
       switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_ERROR, "error allocating memory!\n");
       return SWITCH_STATUS_FALSE;
     }
+
     if (SWITCH_STATUS_SUCCESS != fork_data_init(tech_pvt, session, host, port, path, sslFlags, samples_per_second, sampling, channels, 
-      bugname, metadata, responseHandler)) {
+      bugname, metadata, bidirectional_audio_enable, bidirectional_audio_stream, bidirectional_audio_sample_rate, responseHandler)) {
       destroy_tech_pvt(tech_pvt);
       return SWITCH_STATUS_FALSE;
     }
@@ -619,5 +702,53 @@ extern "C" {
     return SWITCH_TRUE;
   }
 
+  switch_bool_t dub_speech_frame(switch_media_bug_t *bug, private_t* tech_pvt) {
+    CircularBuffer_t *cBuffer = (CircularBuffer_t *) tech_pvt->circularBuffer;
+    if (switch_mutex_trylock(tech_pvt->mutex) == SWITCH_STATUS_SUCCESS) {
+      switch_frame_t* rframe = switch_core_media_bug_get_write_replace_frame(bug);
+      int16_t *fp = reinterpret_cast<int16_t*>(rframe->data);
+
+      rframe->channels = 1;
+      rframe->datalen = rframe->samples * sizeof(int16_t);
+
+      int16_t data[SWITCH_RECOMMENDED_BUFFER_SIZE];
+      memset(data, 0, sizeof(data));
+
+      int samplesToCopy = std::min(static_cast<int>(cBuffer->size()), static_cast<int>(rframe->samples));
+
+      std::copy_n(cBuffer->begin(), samplesToCopy, data);
+      cBuffer->erase(cBuffer->begin(), cBuffer->begin() + samplesToCopy);
+
+      if (samplesToCopy > 0) {
+        vector_add(fp, data, rframe->samples);
+      }
+      vector_normalize(fp, rframe->samples);
+
+      switch_core_media_bug_set_write_replace_frame(bug, rframe);
+      switch_mutex_unlock(tech_pvt->mutex);
+    }
+    return SWITCH_TRUE;
+  }
+
+  switch_status_t fork_session_stop_play(switch_core_session_t *session, 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);
+    if (!bug) {
+      switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_ERROR, "fork_session_stop_play failed because no bug\n");
+      return SWITCH_STATUS_FALSE;
+    }
+    private_t* tech_pvt = (private_t*) switch_core_media_bug_get_user_data(bug);
+
+    CircularBuffer_t *cBuffer = (CircularBuffer_t *) tech_pvt->circularBuffer;
+
+    if (switch_mutex_trylock(tech_pvt->mutex) == SWITCH_STATUS_SUCCESS) {
+      if (cBuffer != nullptr) {
+        cBuffer->clear();
+      }
+      switch_mutex_unlock(tech_pvt->mutex);
+    }
+    return SWITCH_STATUS_SUCCESS;
+  }
+
 }
 

mod_audio_fork/lws_glue.h

@@ -8,13 +8,16 @@ int parse_ws_uri(switch_channel_t *channel, const char* szServerUri, char* host,
 switch_status_t fork_init();
 switch_status_t fork_cleanup();
 switch_status_t fork_session_init(switch_core_session_t *session, responseHandler_t responseHandler,
-		uint32_t samples_per_second, char *host, unsigned int port, char* path, int sampling, int sslFlags, int channels, 
-    char *bugname, char* metadata, void **ppUserData);
+  uint32_t samples_per_second, char *host, unsigned int port, char* path, int sampling, int sslFlags, int channels, 
+  char *bugname, char* metadata, int bidirectional_audio_enable,
+  int bidirectional_audio_stream, int bidirectional_audio_sample_rate, void **ppUserData);
 switch_status_t fork_session_cleanup(switch_core_session_t *session, char *bugname, char* text, int channelIsClosing);
+switch_status_t fork_session_stop_play(switch_core_session_t *session, char *bugname);
 switch_status_t fork_session_pauseresume(switch_core_session_t *session, char *bugname, int pause);
 switch_status_t fork_session_graceful_shutdown(switch_core_session_t *session, char *bugname);
 switch_status_t fork_session_send_text(switch_core_session_t *session, char *bugname, char* text);
 switch_bool_t fork_frame(switch_core_session_t *session, switch_media_bug_t *bug);
+switch_bool_t dub_speech_frame(switch_media_bug_t *bug, private_t * tech_pvt);
 switch_status_t fork_service_threads();
 switch_status_t fork_session_connect(void **ppUserData);
 #endif

mod_audio_fork/mod_audio_fork.c

@@ -27,40 +27,48 @@ static void responseHandler(switch_core_session_t* session, const char * eventNa
 
 static switch_bool_t capture_callback(switch_media_bug_t *bug, void *user_data, switch_abc_type_t type)
 {
+	switch_bool_t ret = SWITCH_TRUE;
 	switch_core_session_t *session = switch_core_media_bug_get_session(bug);
+	private_t* tech_pvt = (private_t *)  switch_core_media_bug_get_user_data(bug);
 	switch (type) {
 	case SWITCH_ABC_TYPE_INIT:
 		break;
 
 	case SWITCH_ABC_TYPE_CLOSE:
 		{
-      private_t* tech_pvt = (private_t *)  switch_core_media_bug_get_user_data(bug);
 			switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_INFO, "Got SWITCH_ABC_TYPE_CLOSE for bug %s\n", tech_pvt->bugname);
       fork_session_cleanup(session, tech_pvt->bugname, NULL, 1);
 		}
 		break;
 	
 	case SWITCH_ABC_TYPE_READ:
-		return fork_frame(session, bug);
+		ret = fork_frame(session, bug);
 		break;
 
+	case SWITCH_ABC_TYPE_WRITE_REPLACE:
+		ret = dub_speech_frame(bug, tech_pvt);
+	break;
+
 	case SWITCH_ABC_TYPE_WRITE:
 	default:
 		break;
 	}
 
-	return SWITCH_TRUE;
+	return ret;
 }
 
 static switch_status_t start_capture(switch_core_session_t *session, 
-        switch_media_bug_flag_t flags, 
-        char* host,
-        unsigned int port, 
-        char* path,
-        int sampling,
-        int sslFlags,
-	      char* bugname, 
-        char* metadata)
+	switch_media_bug_flag_t flags, 
+	char* host,
+	unsigned int port, 
+	char* path,
+	int sampling,
+	int sslFlags,
+	int bidirectional_audio_enable,
+	int bidirectional_audio_stream,
+	int bidirectional_audio_sample_rate,
+	char* bugname, 
+	char* metadata)
 {
 	switch_channel_t *channel = switch_core_session_get_channel(session);
 	switch_media_bug_t *bug;
@@ -71,8 +79,8 @@ static switch_status_t start_capture(switch_core_session_t *session,
   int channels = (flags & SMBF_STEREO) ? 2 : 1;
 
 	switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_INFO, 
-    "mod_audio_fork (%s): streaming %d sampling to %s path %s port %d tls: %s.\n", 
-    bugname, sampling, host, path, port, sslFlags ? "yes" : "no");
+    "mod_audio_fork (%s): streaming %d sampling to %s path %s port %d tls: %s bidirectional_audio_sample_rate: %d.\n", 
+    bugname, sampling, host, path, port, sslFlags ? "yes" : "no", bidirectional_audio_sample_rate);
 
 	if (switch_channel_get_private(channel, bugname)) {
 		switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_ERROR, "mod_audio_fork: bug %s already attached!\n", bugname);
@@ -88,10 +96,12 @@ static switch_status_t start_capture(switch_core_session_t *session,
 
 	switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "calling fork_session_init.\n");
 	if (SWITCH_STATUS_FALSE == fork_session_init(session, responseHandler, read_codec->implementation->actual_samples_per_second, 
-		host, port, path, sampling, sslFlags, channels, bugname, metadata, &pUserData)) {
+		host, port, path, sampling, sslFlags, channels, bugname, metadata, bidirectional_audio_enable, bidirectional_audio_stream,
+		bidirectional_audio_sample_rate, &pUserData)) {
 		switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_ERROR, "Error initializing mod_audio_fork session.\n");
 		return SWITCH_STATUS_FALSE;
 	}
+
 	switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "adding bug %s.\n", bugname);
 	if ((status = switch_core_media_bug_add(session, bugname, NULL, capture_callback, pUserData, 0, flags, &bug)) != SWITCH_STATUS_SUCCESS) {
 		return status;
@@ -133,6 +143,16 @@ static switch_status_t do_pauseresume(switch_core_session_t *session, char* bugn
 	return status;
 }
 
+static switch_status_t stop_play(switch_core_session_t *session, char* bugname)
+{
+	switch_status_t status = SWITCH_STATUS_SUCCESS;
+
+	switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_INFO, "mod_audio_fork stop_play\n");
+	status = fork_session_stop_play(session, bugname);
+
+	return status;
+}
+
 static switch_status_t do_graceful_shutdown(switch_core_session_t *session, char* bugname)
 {
 	switch_status_t status = SWITCH_STATUS_SUCCESS;
@@ -159,10 +179,10 @@ static switch_status_t send_text(switch_core_session_t *session, char* bugname,
   return status;
 }
 
-#define FORK_API_SYNTAX "<uuid> [start | stop | send_text | pause | resume | graceful-shutdown ] [wss-url | path] [mono | mixed | stereo] [8000 | 16000 | 24000 | 32000 | 64000] [bugname] [metadata]"
+#define FORK_API_SYNTAX "<uuid> [start | stop | send_text | pause | resume | graceful-shutdown | stop_play ] [wss-url | path] [mono | mixed | stereo] [8000 | 16000 | 24000 | 32000 | 64000] [bugname] [metadata] [bidirectionalAudio_enabled] [bidirectionalAudio_stream_enabled] [bidirectionalAudio_stream_samplerate]"
 SWITCH_STANDARD_API(fork_function)
 {
-	char *mycmd = NULL, *argv[7] = { 0 };
+	char *mycmd = NULL, *argv[10] = { 0 };
 	int argc = 0;
 	switch_status_t status = SWITCH_STATUS_FALSE;
   char *bugname = MY_BUG_NAME;
@@ -196,6 +216,9 @@ SWITCH_STANDARD_API(fork_function)
         }
 				status = do_stop(lsession, bugname, text);
       }
+			else if (!strcasecmp(argv[1], "stop_play")) {
+				status = stop_play(lsession, bugname);
+			}
 			else if (!strcasecmp(argv[1], "pause")) {
         if (argc > 2) bugname = argv[2];
 				status = do_pauseresume(lsession, bugname, 1);
@@ -230,10 +253,31 @@ SWITCH_STANDARD_API(fork_function)
         char host[MAX_WS_URL_LEN], path[MAX_PATH_LEN];
         unsigned int port;
         int sslFlags;
+				
         int sampling = 8000;
-      	switch_media_bug_flag_t flags = SMBF_READ_STREAM ;
+      	switch_media_bug_flag_t flags = SMBF_READ_STREAM;
         char *metadata = NULL;
-        if( argc > 6) {
+				int bidirectional_audio_enable = 1;
+				int bidirectional_audio_stream = 0;
+				int bidirectional_audio_sample_rate = 0;
+				// Expecting that bidirectional audio params is always received together with bugname and metadata even they are empty string
+				if (argc > 9) {
+					if (argv[5][0] != '\0') {
+						bugname = argv[5];
+					}
+					if (argv[6][0] != '\0') {
+						metadata = argv[6];
+					}
+					bidirectional_audio_enable = !strcmp(argv[7], "true") ? 1 : 0;
+					bidirectional_audio_stream = !strcmp(argv[8], "true") ? 1 : 0;
+					bidirectional_audio_sample_rate = atoi(argv[9]);
+
+					if (bidirectional_audio_enable &&
+						bidirectional_audio_stream &&
+						bidirectional_audio_sample_rate) {
+						flags |= SMBF_WRITE_REPLACE ;
+					}
+				} else if( argc > 6 ) {
           bugname = argv[5];
           metadata = argv[6];
         }
@@ -241,6 +285,7 @@ SWITCH_STANDARD_API(fork_function)
           if (argv[5][0] == '{' || argv[5][0] == '[') metadata = argv[5];
           else bugname = argv[5];
         }
+
         if (0 == strcmp(argv[3], "mixed")) {
           flags |= SMBF_WRITE_STREAM ;
         }
@@ -268,7 +313,8 @@ SWITCH_STANDARD_API(fork_function)
 				else if (sampling % 8000 != 0) {
           switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_ERROR, "invalid sample rate: %s\n", argv[4]);					
 				}
-        status = start_capture(lsession, flags, host, port, path, sampling, sslFlags, bugname, metadata);
+        status = start_capture(lsession, flags, host, port, path, sampling, sslFlags,
+					bidirectional_audio_enable, bidirectional_audio_stream, bidirectional_audio_sample_rate, bugname, metadata);
 			}
       else {
         switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_ERROR, "unsupported mod_audio_fork cmd: %s\n", argv[1]);

mod_audio_fork/mod_audio_fork.h

@@ -52,6 +52,11 @@ struct private_data {
   int audio_paused:1;
   int graceful_shutdown:1;
   char initialMetadata[8192];
+  void *circularBuffer;
+  SpeexResamplerState *bidirectional_audio_resampler;
+  int bidirectional_audio_enable;
+	int bidirectional_audio_stream;
+  int bidirectional_audio_sample_rate;
 };
 
 typedef struct private_data private_t;

mod_audio_fork/vector_math.cpp

@@ -0,0 +1,213 @@
+#include "vector_math.h"
+#include <assert.h>
+#include <string.h>
+#include <cstdlib>
+
+#define GRANULAR_VOLUME_MAX (50)
+#define SMAX 32767
+#define SMIN (-32768)
+#define normalize_to_16bit_basic(n) if (n > SMAX) n = SMAX; else if (n < SMIN) n = SMIN;
+#define normalize_volume_granular(x) if (x > GRANULAR_VOLUME_MAX) x = GRANULAR_VOLUME_MAX; if (x < -GRANULAR_VOLUME_MAX) x = -GRANULAR_VOLUME_MAX;
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+#if defined(USE_AVX2)
+#include <immintrin.h>
+#pragma message("Using AVX2 SIMD.")
+void vector_add(int16_t* a, int16_t* b, size_t len) {
+    size_t i = 0;
+    for (; i + 15 < len; i += 16) {
+        __m256i va = _mm256_loadu_si256((const __m256i*)(a + i));
+        __m256i vb = _mm256_loadu_si256((const __m256i*)(b + i));
+        __m256i vc = _mm256_add_epi16(va, vb);
+        _mm256_storeu_si256((__m256i*)(a + i), vc);
+    }
+    for (; i < len; ++i) {
+        a[i] += b[i];
+    }
+}
+void vector_normalize(int16_t* a, size_t len) {
+  __m256i max_val = _mm256_set1_epi16(SMAX);
+  __m256i min_val = _mm256_set1_epi16(SMIN);
+
+  size_t i = 0;
+  for (; i + 15 < len; i += 16) {
+      __m256i values = _mm256_loadu_si256((__m256i*)(a + i));
+      __m256i gt_max = _mm256_cmpgt_epi16(values, max_val);
+      __m256i lt_min = _mm256_cmpgt_epi16(min_val, values);
+      values = _mm256_blendv_epi8(values, max_val, gt_max);
+      values = _mm256_blendv_epi8(values, min_val, lt_min);
+      _mm256_storeu_si256((__m256i*)(a + i), values);
+  }
+
+  // Process remaining elements
+  for (; i < len; ++i) {
+      if (a[i] > SMAX) a[i] = SMAX;
+      else if (a[i] < SMIN) a[i] = SMIN;
+  }
+}
+typedef union {
+    int16_t* data;
+    __m256i* fp_avx2;
+} vector_data_t;
+
+void vector_change_sln_volume_granular(int16_t* data, uint32_t samples, int32_t vol) {
+  float newrate = 0;
+  static const float pos[GRANULAR_VOLUME_MAX] = {
+		1.122018,   1.258925,   1.412538,   1.584893,   1.778279,   1.995262,   2.238721,   2.511887,   2.818383,   3.162278,
+		3.548134,   3.981072,   4.466835,   5.011872,   5.623413,   6.309574,   7.079458,   7.943282,   8.912509,  10.000000,
+		11.220183,  12.589254,  14.125375,  15.848933,  17.782795,  19.952621,  22.387213,  25.118862,  28.183832,  31.622776,
+		35.481335,  39.810719,  44.668358,  50.118729,  56.234131,  63.095726,  70.794586,  79.432816,  89.125107, 100.000000,
+		112.201836, 125.892517, 141.253784, 158.489334, 177.827942, 199.526215, 223.872070, 251.188705, 281.838318, 316.227753
+  };
+  static const float neg[GRANULAR_VOLUME_MAX] = {
+		0.891251, 0.794328, 0.707946, 0.630957, 0.562341, 0.501187, 0.446684, 0.398107, 0.354813, 0.316228,
+		0.281838, 0.251189, 0.223872, 0.199526, 0.177828, 0.158489, 0.141254, 0.125893, 0.112202, 0.100000,
+		0.089125, 0.079433, 0.070795, 0.063096, 0.056234, 0.050119, 0.044668, 0.039811, 0.035481, 0.031623,
+		0.028184, 0.025119, 0.022387, 0.019953, 0.017783, 0.015849, 0.014125, 0.012589, 0.011220, 0.010000,
+		0.008913, 0.007943, 0.007079, 0.006310, 0.005623, 0.005012, 0.004467, 0.003981, 0.003548, 0.000000  // NOTE mapped -50 dB ratio to total silence instead of 0.003162
+  };
+  const float* chart;
+  uint32_t i = abs(vol) - 1;
+
+  if (vol == 0) return;
+  normalize_volume_granular(vol);
+
+  chart = vol > 0 ? pos : neg;
+
+  newrate = chart[i];
+
+  if (newrate) {
+    __m256 scale_factor_reg = _mm256_set1_ps(newrate);
+    uint32_t processed_samples = samples - (samples % 8); // Ensure we process only multiples of 8
+    for (uint32_t i = 0; i < processed_samples; i += 8) {
+      __m128i data_ = _mm_loadu_si128((__m128i*)(data + i));
+      __m256i data_32 = _mm256_cvtepi16_epi32(data_);
+
+      __m256 data_float = _mm256_cvtepi32_ps(data_32);
+      __m256 result = _mm256_mul_ps(data_float, scale_factor_reg);
+
+      __m256i result_32 = _mm256_cvtps_epi32(result);
+
+      // Handle saturation
+      __m256i min_val = _mm256_set1_epi32(SMIN);
+      __m256i max_val = _mm256_set1_epi32(SMAX);
+      result_32 = _mm256_min_epi32(result_32, max_val);
+      result_32 = _mm256_max_epi32(result_32, min_val);
+
+      __m128i result_16 = _mm_packs_epi32(_mm256_castsi256_si128(result_32), _mm256_extractf128_si256(result_32, 1));
+
+      _mm_storeu_si128((__m128i*)(data + i), result_16);
+    }
+
+    // Process any remaining samples
+    for (uint32_t i = processed_samples; i < samples; i++) {
+      int32_t tmp = (int32_t)(data[i] * newrate);
+      tmp = tmp > SMAX ? SMAX : (tmp < SMIN ? SMIN : tmp);
+      data[i] = (int16_t)tmp;
+    }
+  }
+}
+#elif defined(USE_SSE2)
+#include <emmintrin.h>
+#pragma message("Using SSE2 SIMD.")
+void vector_add(int16_t* a, int16_t* b, size_t len) {
+  size_t i = 0;
+  for (; i + 7 < len; i += 8) {
+    __m128i va = _mm_loadu_si128((const __m128i*)(a + i));
+    __m128i vb = _mm_loadu_si128((const __m128i*)(b + i));
+    __m128i vc = _mm_add_epi16(va, vb);
+    _mm_storeu_si128((__m128i*)(a + i), vc);
+  }
+  for (; i < len; ++i) {
+    a[i] += b[i];
+  }
+}
+void vector_normalize(int16_t* a, size_t len) {
+  __m128i max_val = _mm_set1_epi16(SMAX);
+  __m128i min_val = _mm_set1_epi16(SMIN);
+
+  size_t i = 0;
+  for (; i + 7 < len; i += 8) {
+    __m128i values = _mm_loadu_si128((__m128i*)(a + i));
+    __m128i gt_max = _mm_cmpgt_epi16(values, max_val);
+    __m128i lt_min = _mm_cmpgt_epi16(min_val, values);
+    __m128i max_masked = _mm_and_si128(gt_max, max_val);
+    __m128i min_masked = _mm_and_si128(lt_min, min_val);
+    __m128i other_masked = _mm_andnot_si128(_mm_or_si128(gt_max, lt_min), values);
+    values = _mm_or_si128(_mm_or_si128(max_masked, min_masked), other_masked);
+    _mm_storeu_si128((__m128i*)(a + i), values);
+  }
+
+  // Process remaining elements
+  for (; i < len; ++i) {
+    if (a[i] > SMAX) a[i] = SMAX;
+    else if (a[i] < SMIN) a[i] = SMIN;
+  }
+}
+
+typedef union {
+    int16_t* data;
+    __m128i* fp_sse2;
+} vector_data_t;
+
+#else
+#pragma message("Building without vector math support")
+void vector_add(int16_t* a, int16_t* b, size_t len) {
+    for (size_t i = 0; i < len; i++) {
+        a[i] += b[i];
+    }
+}
+void vector_normalize(int16_t* a, size_t len) {
+  for (size_t i = 0; i < len; i++) {
+    normalize_to_16bit_basic(a[i]);
+  }
+}
+void vector_change_sln_volume_granular(int16_t* data, uint32_t samples, int32_t vol) {
+  float newrate = 0;
+  static const float pos[GRANULAR_VOLUME_MAX] = {
+		1.122018,   1.258925,   1.412538,   1.584893,   1.778279,   1.995262,   2.238721,   2.511887,   2.818383,   3.162278,
+		3.548134,   3.981072,   4.466835,   5.011872,   5.623413,   6.309574,   7.079458,   7.943282,   8.912509,  10.000000,
+		11.220183,  12.589254,  14.125375,  15.848933,  17.782795,  19.952621,  22.387213,  25.118862,  28.183832,  31.622776,
+		35.481335,  39.810719,  44.668358,  50.118729,  56.234131,  63.095726,  70.794586,  79.432816,  89.125107, 100.000000,
+		112.201836, 125.892517, 141.253784, 158.489334, 177.827942, 199.526215, 223.872070, 251.188705, 281.838318, 316.227753
+  };
+  static const float neg[GRANULAR_VOLUME_MAX] = {
+		0.891251, 0.794328, 0.707946, 0.630957, 0.562341, 0.501187, 0.446684, 0.398107, 0.354813, 0.316228,
+		0.281838, 0.251189, 0.223872, 0.199526, 0.177828, 0.158489, 0.141254, 0.125893, 0.112202, 0.100000,
+		0.089125, 0.079433, 0.070795, 0.063096, 0.056234, 0.050119, 0.044668, 0.039811, 0.035481, 0.031623,
+		0.028184, 0.025119, 0.022387, 0.019953, 0.017783, 0.015849, 0.014125, 0.012589, 0.011220, 0.010000,
+		0.008913, 0.007943, 0.007079, 0.006310, 0.005623, 0.005012, 0.004467, 0.003981, 0.003548, 0.000000  // NOTE mapped -50 dB ratio to total silence instead of 0.003162
+  };
+  const float* chart;
+  uint32_t i;
+
+  if (vol == 0) return;
+
+  normalize_volume_granular(vol);
+
+  chart = vol > 0 ? pos : neg;
+
+  i = abs(vol) - 1;
+  assert(i < GRANULAR_VOLUME_MAX);
+  newrate = chart[i];
+  if (newrate) {
+    int32_t tmp;
+    uint32_t x;
+    int16_t *fp = data;
+
+    for (x = 0; x < samples; x++) {
+      tmp = (int32_t) (fp[x] * newrate);
+      normalize_to_16bit_basic(tmp);
+      fp[x] = (int16_t) tmp;
+    }
+  }
+}
+
+#endif
+
+#ifdef __cplusplus
+}
+#endif

mod_audio_fork/vector_math.h

@@ -0,0 +1,20 @@
+#ifndef VECTOR_MATH_H
+#define VECTOR_MATH_H
+
+#include <stddef.h>
+#include <stdint.h>
+
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+void vector_add(int16_t* a, int16_t* b, size_t len);
+void vector_normalize(int16_t* a, size_t len);
+void vector_change_sln_volume_granular(int16_t* data, uint32_t samples, int32_t vol);
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif

mod_azure_tts/azure_glue.cpp

@@ -158,7 +158,7 @@ extern "C" {
       CircularBuffer_t *cBuffer = (CircularBuffer_t *) a->circularBuffer;
       std::vector<uint16_t> pcm_data;
 
-      switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "Synthesizing: received data\n");
+      //switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "Synthesizing: received data\n");
 
       if (a->flushed) {
         return;
@@ -204,6 +204,7 @@ extern "C" {
             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);
@@ -314,7 +315,7 @@ extern "C" {
     a->flushed = 1;
     if (!download_complete) {
       if (a->file) {
-        switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "closing audio cache file %s because download was interrupted\n", a->cache_filename);
+        //switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "closing audio cache file %s because download was interrupted\n", a->cache_filename);
         if (fclose(a->file) != 0) {
           switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "error closing audio cache file\n");
         }
@@ -322,7 +323,7 @@ extern "C" {
       }
 
       if (a->cache_filename) {
-        switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "removing audio cache file %s because download was interrupted\n", a->cache_filename);
+        //switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "removing audio cache file %s because download was interrupted\n", a->cache_filename);
         if (unlink(a->cache_filename) != 0) {
           switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "cleanupConn: error removing audio cache file %s: %d:%s\n", 
             a->cache_filename, errno, strerror(errno));

mod_azure_tts/mod_azure_tts.c

@@ -6,7 +6,7 @@ SWITCH_MODULE_SHUTDOWN_FUNCTION(mod_azure_tts_shutdown);
 SWITCH_MODULE_DEFINITION(mod_azure_tts, mod_azure_tts_load, mod_azure_tts_shutdown, NULL);
 
 static void clearAzure(azure_t* a, int freeAll) {
-  switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "clearAzure\n");
+  switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "clearAzure\n");
 
   if (a->cache_filename) free(a->cache_filename);
   if (a->api_key) free(a->api_key);
@@ -46,7 +46,7 @@ static azure_t * createOrRetrievePrivateData(switch_speech_handle_t *sh) {
   	sh->private_info = a;
     memset(a, 0, sizeof(*a));
     switch_mutex_init(&a->mutex, SWITCH_MUTEX_NESTED, sh->memory_pool);
-    switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "allocated azure_t\n");
+    switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "allocated azure_t\n");
   }
   return a;
 }
@@ -56,7 +56,7 @@ switch_status_t a_speech_open(switch_speech_handle_t *sh, const char *voice_name
   azure_t *a = createOrRetrievePrivateData(sh);
   a->voice_name = strdup(voice_name);
   a->rate = rate;
-  switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "a_speech_open voice: %s, rate %d, channels %d\n", voice_name, rate, channels);
+  switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "a_speech_open voice: %s, rate %d, channels %d\n", voice_name, rate, channels);
   return azure_speech_open(a);
 }
 
@@ -64,7 +64,7 @@ static switch_status_t a_speech_close(switch_speech_handle_t *sh, switch_speech_
 {
   switch_status_t rc;
   azure_t *a = createOrRetrievePrivateData(sh);
-  switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "a_speech_close\n");
+  switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "a_speech_close\n");
 
   switch_mutex_destroy(a->mutex);
 
@@ -84,8 +84,6 @@ static switch_status_t a_speech_feed_tts(switch_speech_handle_t *sh, char *text,
   a->flushed = 0;
   a->samples_rate = 0;
 
-  switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "a_speech_feed_tts\n");
-
   return azure_speech_feed_tts(a, text, flags);
 }
 
@@ -95,7 +93,6 @@ static switch_status_t a_speech_feed_tts(switch_speech_handle_t *sh, char *text,
 static switch_status_t a_speech_read_tts(switch_speech_handle_t *sh, void *data, size_t *datalen, switch_speech_flag_t *flags)
 {
   azure_t *a = createOrRetrievePrivateData(sh);
-  switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "a_speech_read_tts\n");
   return azure_speech_read_tts(a, data, datalen, flags);
 }
 
@@ -105,7 +102,6 @@ static switch_status_t a_speech_read_tts(switch_speech_handle_t *sh, void *data,
 static void a_speech_flush_tts(switch_speech_handle_t *sh)
 {
   azure_t *a = createOrRetrievePrivateData(sh);
-  switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "w_speech_flush_tts\n");
   azure_speech_flush_tts(a);
 
   clearAzure(a, 0);
@@ -114,7 +110,7 @@ static void a_speech_flush_tts(switch_speech_handle_t *sh)
 static void a_text_param_tts(switch_speech_handle_t *sh, char *param, const char *val)
 {
   azure_t *a = createOrRetrievePrivateData(sh);
-  switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "a_text_param_tts: %s=%s\n", param, val);
+  switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "a_text_param_tts: %s=%s\n", param, val);
   if (0 == strcmp(param, "api_key")) {
     if (a->api_key) free(a->api_key);
     a->api_key = strdup(val);

mod_deepgram_tts/Makefile.am

@@ -0,0 +1,8 @@
+include $(top_srcdir)/build/modmake.rulesam
+MODNAME=mod_deepgram_tts
+
+mod_LTLIBRARIES = mod_deepgram_tts.la
+mod_deepgram_tts_la_SOURCES  = mod_deepgram_tts.c deepgram_glue.cpp
+mod_deepgram_tts_la_CFLAGS   = $(AM_CFLAGS)
+mod_deepgram_tts_la_LIBADD   = $(switch_builddir)/libfreeswitch.la
+mod_deepgram_tts_la_LDFLAGS  = -avoid-version -module -no-undefined -shared `pkg-config --libs boost` -lstdc++

mod_deepgram_tts/deepgram_glue.cpp

@@ -0,0 +1,922 @@
+#include "mod_deepgram_tts.h"
+#include <switch.h>
+#include <switch_json.h>
+#include <curl/curl.h>
+#include <cstdlib>
+
+#include <boost/circular_buffer.hpp>
+#include <boost/thread.hpp>
+#include <boost/asio.hpp>
+#include <boost/asio/ssl.hpp>
+#include <boost/pool/object_pool.hpp>
+#include <boost/bind/bind.hpp>
+#include <boost/tokenizer.hpp>
+#include <boost/foreach.hpp>
+#include <boost/asio.hpp>
+#include <boost/assign/list_of.hpp>
+#include <boost/algorithm/string.hpp>
+
+#include <speex/speex_resampler.h>
+
+#define BUFFER_GROW_SIZE (80000)
+
+typedef boost::circular_buffer<uint16_t> CircularBuffer_t;
+/* Global information, common to all connections */
+typedef struct
+{
+  CURLM *multi;
+  int still_running;
+} GlobalInfo_t;
+static GlobalInfo_t global;
+
+/* Information associated with a specific easy handle */
+typedef struct
+{
+  CURL *easy;
+  deepgram_t* deepgram;
+  char* body;
+  struct curl_slist *hdr_list;
+  GlobalInfo_t *global;
+  char error[CURL_ERROR_SIZE];
+  FILE* file;
+  std::chrono::time_point<std::chrono::high_resolution_clock> startTime;
+  bool flushed;
+
+  bool has_last_byte;
+  uint8_t last_byte;
+} ConnInfo_t;
+
+
+static boost::object_pool<ConnInfo_t> pool ;
+static std::map<curl_socket_t, boost::asio::ip::tcp::socket *> socket_map;
+static boost::asio::io_service io_service;
+static boost::asio::deadline_timer timer(io_service);
+static std::string fullDirPath;
+static std::thread worker_thread;
+
+std::string secondsToMillisecondsString(double seconds) {
+    // Convert to milliseconds
+    double milliseconds = seconds * 1000.0;
+
+    // Truncate to remove fractional part
+    long milliseconds_long = static_cast<long>(milliseconds);
+
+    // Convert to string
+    return std::to_string(milliseconds_long);
+}
+
+static CURL* createEasyHandle(void) {
+  CURL* easy = curl_easy_init();
+  if(!easy) {
+    switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "curl_easy_init() failed!\n");
+    return nullptr ;
+  }  
+
+  curl_easy_setopt(easy, CURLOPT_FOLLOWLOCATION, 1L);
+  curl_easy_setopt(easy, CURLOPT_USERAGENT, "jambonz/0.8.5");
+
+  // set connect timeout to 3 seconds and total timeout to 109 seconds
+  curl_easy_setopt(easy, CURLOPT_CONNECTTIMEOUT_MS, 3000L);
+  curl_easy_setopt(easy, CURLOPT_TIMEOUT, 10L);
+
+  return easy ;    
+}
+
+static void cleanupConn(ConnInfo_t *conn) {
+  auto d = conn->deepgram;
+
+  if( conn->hdr_list ) {
+    curl_slist_free_all(conn->hdr_list);
+    conn->hdr_list = nullptr ;
+  }
+  curl_easy_cleanup(conn->easy);
+
+  if (conn->file) {
+    if (fclose(conn->file) != 0) {
+      switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "cleanupConn: error closing audio cache file\n");
+    }
+    conn->file = nullptr ;
+  }
+
+  d->conn = nullptr ;
+  d->draining = 1;
+
+  memset(conn, 0, sizeof(ConnInfo_t));
+  pool.destroy(conn) ;
+}
+
+/* Check for completed transfers, and remove their easy handles */
+void check_multi_info(GlobalInfo_t *g) {
+  CURLMsg *msg;
+  int msgs_left;
+  ConnInfo_t *conn;
+  CURL *easy;
+  CURLcode res;
+  
+  while((msg = curl_multi_info_read(g->multi, &msgs_left))) {
+    if(msg->msg == CURLMSG_DONE) {
+      long response_code;
+      double namelookup=0, connect=0, total=0 ;
+      char *ct = NULL ;
+
+      easy = msg->easy_handle;
+      res = msg->data.result;
+      curl_easy_getinfo(easy, CURLINFO_PRIVATE, &conn);
+      curl_easy_getinfo(easy, CURLINFO_RESPONSE_CODE, &response_code);
+      curl_easy_getinfo(easy, CURLINFO_CONTENT_TYPE, &ct);
+
+      curl_easy_getinfo(easy, CURLINFO_NAMELOOKUP_TIME, &namelookup);
+      curl_easy_getinfo(easy, CURLINFO_CONNECT_TIME, &connect);
+      curl_easy_getinfo(easy, CURLINFO_TOTAL_TIME, &total);
+
+      auto d = conn->deepgram;
+      d->response_code = response_code;
+      if (ct) d->ct = strdup(ct);
+
+      std::string name_lookup_ms = secondsToMillisecondsString(namelookup);
+      std::string connect_ms = secondsToMillisecondsString(connect);
+      std::string final_response_time_ms = secondsToMillisecondsString(total);
+
+      switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, 
+        "mod_deepgram_tts: response: %ld, content-type %s,"
+        "dns(ms): %"  CURL_FORMAT_CURL_OFF_T ".%06ld, "
+        "connect(ms): %"  CURL_FORMAT_CURL_OFF_T ".%06ld, "
+        "total(ms): %"  CURL_FORMAT_CURL_OFF_T ".%06ld\n",
+        response_code, ct,
+        (long)(namelookup), (long)(fmod(namelookup, 1.0) * 1000000),
+        (long)(connect), (long)(fmod(connect, 1.0) * 1000000),
+        (long)(total), (long)(fmod(total, 1.0) * 1000000));
+
+      switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "name lookup time: %s\n", name_lookup_ms.c_str());
+      switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "connect time: %s\n", connect_ms.c_str());
+      switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "final response time: %s\n", final_response_time_ms.c_str());
+
+      d->name_lookup_time_ms = strdup(name_lookup_ms.c_str());
+      d->connect_time_ms = strdup(connect_ms.c_str());
+      d->final_response_time_ms = strdup(final_response_time_ms.c_str());
+
+      curl_multi_remove_handle(g->multi, easy);
+      cleanupConn(conn);
+    }
+  }
+}
+
+int mcode_test(const char *where, CURLMcode code) {
+  if(CURLM_OK != code) {
+    const char *s;
+    switch(code) {
+    case CURLM_CALL_MULTI_PERFORM:
+      s = "CURLM_CALL_MULTI_PERFORM";
+      break;
+    case CURLM_BAD_HANDLE:
+      s = "CURLM_BAD_HANDLE";
+      break;
+    case CURLM_BAD_EASY_HANDLE:
+      s = "CURLM_BAD_EASY_HANDLE";
+      break;
+    case CURLM_OUT_OF_MEMORY:
+      s = "CURLM_OUT_OF_MEMORY";
+      break;
+    case CURLM_INTERNAL_ERROR:
+      s = "CURLM_INTERNAL_ERROR";
+      break;
+    case CURLM_UNKNOWN_OPTION:
+      s = "CURLM_UNKNOWN_OPTION";
+      break;
+    case CURLM_LAST:
+      s = "CURLM_LAST";
+      break;
+    default:
+      s = "CURLM_unknown";
+      break;
+    case CURLM_BAD_SOCKET:
+      s = "CURLM_BAD_SOCKET";
+      break;
+    }
+    switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "mcode_test ERROR: %s returns %s:%d\n", where, s, code);
+
+    return -1;
+  }
+  return 0 ;
+}
+
+static void remsock(int *f, GlobalInfo_t *g) {
+  if(f) {
+    free(f);
+    f = NULL;
+  }
+}
+
+/* Called by asio when there is an action on a socket */
+static void event_cb(GlobalInfo_t *g, curl_socket_t s, int action, const boost::system::error_code & error, int *fdp) {
+  int f = *fdp;
+
+  //switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "event_cb socket %#X has action %d\n", s, action) ; 
+
+  // Socket already POOL REMOVED.
+  if (f == CURL_POLL_REMOVE) {
+    //switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "event_cb socket %#X removed\n", s); 
+    remsock(fdp, g);
+    return;
+  }
+
+  if(socket_map.find(s) == socket_map.end()) {
+    //switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "event_cb: socket  %#X already closed\n, s");
+    return;
+  }
+
+  /* make sure the event matches what are wanted */
+  if(f == action || f == CURL_POLL_INOUT) {
+    if(error) {
+      action = CURL_CSELECT_ERR;
+    }
+    CURLMcode rc = curl_multi_socket_action(g->multi, s, action, &g->still_running);
+
+    mcode_test("event_cb: curl_multi_socket_action", rc);
+    check_multi_info(g);
+
+    if(g->still_running <= 0) {
+      timer.cancel();
+    }
+
+    /* keep on watching.
+      * the socket may have been closed and/or fdp may have been changed
+      * in curl_multi_socket_action(), so check them both */
+    if(!error && socket_map.find(s) != socket_map.end() &&
+        (f == action || f == CURL_POLL_INOUT)) {
+      boost::asio::ip::tcp::socket *tcp_socket = socket_map.find(s)->second;
+
+      if(action == CURL_POLL_IN) {
+        tcp_socket->async_read_some(boost::asio::null_buffers(),
+                                    boost::bind(&event_cb, g, s,
+                                                action, boost::placeholders::_1, fdp));
+      }
+      if(action == CURL_POLL_OUT) {
+        tcp_socket->async_write_some(boost::asio::null_buffers(),
+                                      boost::bind(&event_cb, g, s,
+                                                  action, boost::placeholders::_1, fdp));
+      } 
+    }
+  }
+}
+
+/* socket functions */
+static void setsock(int *fdp, curl_socket_t s, CURL *e, int act, int oldact, GlobalInfo_t *g) {
+  std::map<curl_socket_t, boost::asio::ip::tcp::socket *>::iterator it = socket_map.find(s);
+
+  if(it == socket_map.end()) {
+    //switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "setsock: socket  %#X not found\n, s");
+    return;
+  }
+
+  boost::asio::ip::tcp::socket * tcp_socket = it->second;
+
+  *fdp = act;
+
+  if(act == CURL_POLL_IN) {
+    if(oldact != CURL_POLL_IN && oldact != CURL_POLL_INOUT) {
+      tcp_socket->async_read_some(boost::asio::null_buffers(),
+                                  boost::bind(&event_cb, g, s,
+                                  CURL_POLL_IN, boost::placeholders::_1, fdp));
+    }
+  }
+  else if(act == CURL_POLL_OUT) {
+    if(oldact != CURL_POLL_OUT && oldact != CURL_POLL_INOUT) {
+      tcp_socket->async_write_some(boost::asio::null_buffers(),
+                                    boost::bind(&event_cb, g, s,
+                                    CURL_POLL_OUT, boost::placeholders::_1, fdp));
+    }
+  }
+  else if(act == CURL_POLL_INOUT) {
+    if(oldact != CURL_POLL_IN && oldact != CURL_POLL_INOUT) {
+      tcp_socket->async_read_some(boost::asio::null_buffers(),
+                                  boost::bind(&event_cb, g, s,
+                                  CURL_POLL_IN, boost::placeholders::_1, fdp));
+    }
+    if(oldact != CURL_POLL_OUT && oldact != CURL_POLL_INOUT) {
+      tcp_socket->async_write_some(boost::asio::null_buffers(),
+                                    boost::bind(&event_cb, g, s,
+                                    CURL_POLL_OUT, boost::placeholders::_1, fdp));
+    }
+  }
+}
+
+static void addsock(curl_socket_t s, CURL *easy, int action, GlobalInfo_t *g) {
+  /* fdp is used to store current action */
+  int *fdp = (int *) calloc(sizeof(int), 1);
+
+  setsock(fdp, s, easy, action, 0, g);
+  curl_multi_assign(g->multi, s, fdp);
+}
+
+static int sock_cb(CURL *e, curl_socket_t s, int what, void *cbp, void *sockp) {
+  GlobalInfo_t *g = &global;
+
+  int *actionp = (int *) sockp;
+  static const char *whatstr[] = { "none", "IN", "OUT", "INOUT", "REMOVE"};
+
+  if(what == CURL_POLL_REMOVE) {
+    *actionp = what;
+  }
+  else {
+    if(!actionp) {
+      addsock(s, e, what, g);
+    }
+    else {
+      setsock(actionp, s, e, what, *actionp, g);
+    }
+  }
+  return 0;  
+}
+
+static void threadFunc() {      
+  /* to make sure the event loop doesn't terminate when there is no work to do */
+  io_service.reset() ;
+  boost::asio::io_service::work work(io_service);
+  
+  switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_NOTICE, "mod_deepgram_tts threadFunc - starting\n");
+
+  for(;;) {
+      
+    try {
+      io_service.run() ;
+      break ;
+    }
+    catch( std::exception& e) {
+      switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "mod_deepgram_tts threadFunc - Error: %s\n", e.what());
+    }
+  }
+  switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_NOTICE, "mod_deepgram_tts threadFunc - ending\n");
+}
+
+
+/* Called by asio when our timeout expires */
+static void timer_cb(const boost::system::error_code & error, GlobalInfo_t *g)
+{
+  //switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "timer_cb\n");
+
+  if(!error) {
+    CURLMcode rc = curl_multi_socket_action(g->multi, CURL_SOCKET_TIMEOUT, 0, &g->still_running);
+    mcode_test("timer_cb: curl_multi_socket_action", rc);
+    check_multi_info(g);
+  }
+}
+
+int multi_timer_cb(CURLM *multi, long timeout_ms, GlobalInfo_t *g) {
+
+  /* cancel running timer */
+  timer.cancel();
+
+  if(timeout_ms >= 0) {
+    // from libcurl 7.88.1-10+deb12u4 does not allow call curl_multi_socket_action or curl_multi_perform in curl_multi callback directly
+    timer.expires_from_now(boost::posix_time::millisec(timeout_ms ? timeout_ms : 1));
+    timer.async_wait(boost::bind(&timer_cb, boost::placeholders::_1, g));
+  }
+
+  return 0;
+}
+
+/* CURLOPT_WRITEFUNCTION */
+static size_t write_cb(void *ptr, size_t size, size_t nmemb, ConnInfo_t *conn) {
+  bool fireEvent = false;
+  uint8_t *data = (uint8_t *) ptr;
+  size_t bytes_received = size * nmemb;
+  size_t total_bytes_to_process;
+  auto d = conn->deepgram;
+  CircularBuffer_t *cBuffer = (CircularBuffer_t *) d->circularBuffer;
+  
+  if (conn->flushed) {
+    /* this will abort the transfer */
+    return 0;
+  }
+  // Buffer to hold combined data if there is unprocessed byte from the last call.
+  std::unique_ptr<uint8_t[]> combinedData;
+
+  if (conn->has_last_byte) {
+    conn->has_last_byte = false;  // We'll handle the last_byte now, so toggle the flag off
+
+    // Allocate memory for the new data array
+    combinedData.reset(new uint8_t[bytes_received + 1]);
+
+    // Prepend the last byte from previous call
+    combinedData[0] = conn->last_byte;
+
+    // Copy the new data following the prepended byte
+    memcpy(combinedData.get() + 1, data, bytes_received);
+
+    // Point our data pointer to the new array
+    data = combinedData.get();
+
+    total_bytes_to_process = bytes_received + 1;
+  } else {
+    total_bytes_to_process = bytes_received;
+  }
+
+  // If we now have an odd total, save the last byte for next time
+  if ((total_bytes_to_process % sizeof(int16_t)) != 0) {
+    conn->last_byte = data[total_bytes_to_process - 1];
+    conn->has_last_byte = true;
+    total_bytes_to_process--;
+  }
+
+  int16_t* inputData = reinterpret_cast<int16_t*>(data);
+  if (0 == d->reads++) {
+    fireEvent = true;
+    // Deepgram return PCM linear16 WAV file which contains 44 bytes headers, remove that.
+    inputData += 22;
+    total_bytes_to_process -= 44;
+  }
+  size_t numSamples = total_bytes_to_process / sizeof(int16_t);
+  {
+    switch_mutex_lock(d->mutex);
+
+    if (d->response_code > 0 && d->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());
+      d->err_msg = strdup(body.c_str());
+      switch_mutex_unlock(d->mutex);
+      return 0;
+    }
+
+    /* cache file will stay in the mp3 format for size (smaller) and simplicity */
+    if (conn->file) fwrite(inputData, sizeof(int16_t), numSamples, conn->file);
+
+    // Resize the buffer if necessary
+    if (cBuffer->capacity() - cBuffer->size() < numSamples) {
+      //switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "write_cb growing buffer\n"); 
+
+      //TODO: if buffer exceeds some max size, return CURL_WRITEFUNC_ERROR to abort the transfer
+      cBuffer->set_capacity(cBuffer->size() + std::max(numSamples, (size_t)BUFFER_GROW_SIZE));
+    }
+    
+    /* Push the data into the buffer */
+    cBuffer->insert(cBuffer->end(), inputData, inputData + numSamples);
+
+    switch_mutex_unlock(d->mutex);
+  }
+  if (fireEvent && d->session_id) {
+    auto endTime = std::chrono::high_resolution_clock::now();
+    auto duration = std::chrono::duration_cast<std::chrono::milliseconds>(endTime - conn->startTime);
+    auto time_to_first_byte_ms = std::to_string(duration.count());
+    switch_core_session_t* session = switch_core_session_locate(d->session_id);
+    if (session) {
+      switch_channel_t *channel = switch_core_session_get_channel(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_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->request_id) {
+            switch_event_add_header_string(event, SWITCH_STACK_BOTTOM, "variable_tts_deepgram_request_id", d->request_id);
+          }
+          if (d->name_lookup_time_ms) {
+            switch_event_add_header_string(event, SWITCH_STACK_BOTTOM, "variable_tts_deepgram_name_lookup_time_ms", d->name_lookup_time_ms);
+          }
+          if (d->connect_time_ms) {
+            switch_event_add_header_string(event, SWITCH_STACK_BOTTOM, "variable_tts_deepgram_connect_time_ms", d->connect_time_ms);
+          }
+          if (d->final_response_time_ms) {
+            switch_event_add_header_string(event, SWITCH_STACK_BOTTOM, "variable_tts_deepgram_final_response_time_ms", d->final_response_time_ms);
+          }
+          if (d->voice_name) {
+            switch_event_add_header_string(event, SWITCH_STACK_BOTTOM, "variable_tts_deepgram_voice_name", d->voice_name);
+          }
+          if (d->cache_filename) {
+            switch_event_add_header_string(event, SWITCH_STACK_BOTTOM, "variable_tts_cache_filename", d->cache_filename);
+          }
+
+          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_fire(&event);
+        }
+        else {
+          switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "write_cb: failed to create event\n");
+        }
+      }
+      else {
+        switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "write_cb: channel not found\n");
+      }
+      switch_core_session_rwunlock(session);
+    }
+    else {
+      switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "write_cb: session %s not found\n", d->session_id);
+    }
+  }
+  return size*nmemb;
+}
+
+static bool parseHeader(const std::string& str, std::string& header, std::string& value) {
+    std::vector<std::string> parts;
+    boost::split(parts, str, boost::is_any_of(":"), boost::token_compress_on);
+
+    if (parts.size() != 2)
+        return false;
+
+    header = boost::trim_copy(parts[0]);
+    value = boost::trim_copy(parts[1]);
+    return true;
+}
+
+static size_t header_callback(char *buffer, size_t size, size_t nitems, ConnInfo_t *conn) {
+  size_t bytes_received = size * nitems;
+  const std::string prefix = "HTTP/2 ";
+  deepgram_t* d = conn->deepgram;
+  std::string header, value;
+  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, "recv header: %s\n", input.c_str());
+    if (input.rfind(prefix, 0) == 0) {
+      try {
+        d->response_code = std::stoi(input.substr(prefix.length()));
+        switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "parsed response code: %ld\n", d->response_code);
+      } catch (const std::invalid_argument& e) {
+        switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "header_callback: invalid response code %s\n", input.substr(prefix.length()).c_str());
+      }
+    }
+  }
+  return bytes_received;
+}
+
+/* CURLOPT_OPENSOCKETFUNCTION */
+static curl_socket_t opensocket(void *clientp, curlsocktype purpose, struct curl_sockaddr *address) {
+  curl_socket_t sockfd = CURL_SOCKET_BAD;
+
+  //switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "opensocket: %d\n", purpose);
+  /* restrict to IPv4 */
+  if(purpose == CURLSOCKTYPE_IPCXN && address->family == AF_INET) {
+    /* create a tcp socket object */
+    boost::asio::ip::tcp::socket *tcp_socket = new boost::asio::ip::tcp::socket(io_service);
+
+    /* open it and get the native handle*/
+    boost::system::error_code ec;
+    tcp_socket->open(boost::asio::ip::tcp::v4(), ec);
+
+    if(ec) {
+      /* An error occurred */
+      switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Couldn't open socket [%ld][%s]\n", ec, ec.message().c_str());
+    }
+    else {
+      sockfd = tcp_socket->native_handle();
+
+      /* save it for monitoring */
+      socket_map.insert(std::pair<curl_socket_t, boost::asio::ip::tcp::socket *>(sockfd, tcp_socket));
+    }
+  }
+  return sockfd;
+}
+
+/* CURLOPT_CLOSESOCKETFUNCTION */
+static int close_socket(void *clientp, curl_socket_t item) {
+  //switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "close_socket : %#X\n", item);
+
+  std::map<curl_socket_t, boost::asio::ip::tcp::socket *>::iterator it = socket_map.find(item);
+  if(it != socket_map.end()) {
+    delete it->second;
+    socket_map.erase(it);
+  }
+  return 0;
+}
+
+
+extern "C" {
+  switch_status_t deepgram_speech_load() {
+    switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "deepgram_speech_loading..\n");
+    memset(&global, 0, sizeof(GlobalInfo_t));
+    global.multi = curl_multi_init();
+
+     if (!global.multi) {
+      switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "deepgram_speech_load curl_multi_init() failed, exiting!\n");
+      return SWITCH_STATUS_FALSE;
+    }
+
+    curl_multi_setopt(global.multi, CURLMOPT_SOCKETFUNCTION, sock_cb);
+    curl_multi_setopt(global.multi, CURLMOPT_SOCKETDATA, &global);
+    curl_multi_setopt(global.multi, CURLMOPT_TIMERFUNCTION, multi_timer_cb);
+    curl_multi_setopt(global.multi, CURLMOPT_TIMERDATA, &global);
+    curl_multi_setopt(global.multi, CURLMOPT_PIPELINING, CURLPIPE_MULTIPLEX);
+
+    /* create temp folder for cache files */
+    const char* baseDir = std::getenv("JAMBONZ_TMP_CACHE_FOLDER");
+    if (!baseDir) {
+      baseDir = "/var/";
+    }
+    if (strcmp(baseDir, "/") == 0) {
+      switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "failed to create folder %s\n", baseDir);
+      return SWITCH_STATUS_FALSE;
+    }
+
+    fullDirPath = std::string(baseDir) + "jambonz-tts-cache-files";
+
+    // Create the directory with read, write, and execute permissions for everyone
+    mode_t oldMask = umask(0);
+    int result = mkdir(fullDirPath.c_str(), S_IRWXU | S_IRWXG | S_IRWXO);
+    umask(oldMask);
+    if (result != 0) {
+      if (errno != EEXIST) {
+        switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "failed to create folder %s\n", fullDirPath.c_str());
+        fullDirPath = "";
+      }
+      else switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_NOTICE, "folder %s already exists\n", fullDirPath.c_str());
+    }
+    else {
+      switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_NOTICE, "created folder %s\n", fullDirPath.c_str());
+    }
+
+    /* start worker thread that handles transfers*/
+    std::thread t(threadFunc) ;
+    worker_thread.swap( t ) ;
+
+    switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "deepgram_speech_loaded..\n");
+
+
+    return SWITCH_STATUS_SUCCESS;
+  }
+
+  switch_status_t deepgram_speech_unload() {
+    /* stop the ASIO IO service */
+    switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_NOTICE, "deepgram_speech_unload: stopping io service\n");
+    io_service.stop();
+
+    /* Join the worker thread */
+    switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_NOTICE, "deepgram_speech_unload: wait for worker thread to complete\n");
+    if (worker_thread.joinable()) {
+        worker_thread.join();
+    }
+
+    /* cleanup curl multi handle*/
+    switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_NOTICE, "deepgram_speech_unload: release curl multi\n");
+    curl_multi_cleanup(global.multi);
+    switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_NOTICE, "deepgram_speech_unload: completed\n");
+
+		return SWITCH_STATUS_SUCCESS;
+  }
+
+  switch_status_t deepgram_speech_open(deepgram_t* deepgram) {
+    return SWITCH_STATUS_SUCCESS;
+  }
+
+  switch_status_t deepgram_speech_feed_tts(deepgram_t* d, char* text, switch_speech_flag_t *flags) {
+    CURLMcode rc;
+
+    const int MAX_CHARS = 20;
+    char tempText[MAX_CHARS + 4]; // +4 for the ellipsis and null terminator
+
+    if (strlen(text) > MAX_CHARS) {
+        strncpy(tempText, text, MAX_CHARS);
+        strcpy(tempText + MAX_CHARS, "...");
+    } else {
+        strcpy(tempText, text);
+    }
+
+    /* open cache file */
+    if (d->cache_audio && fullDirPath.length() > 0) {
+      switch_uuid_t uuid;
+      char uuid_str[SWITCH_UUID_FORMATTED_LENGTH + 1];
+      char outfile[512] = "";
+      int fd;
+
+      switch_uuid_get(&uuid);
+      switch_uuid_format(uuid_str, &uuid);
+
+      switch_snprintf(outfile, sizeof(outfile), "%s%s%s.r8", fullDirPath.c_str(), SWITCH_PATH_SEPARATOR, uuid_str);
+      d->cache_filename = strdup(outfile);
+      switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "writing audio cache file to %s\n", d->cache_filename);
+
+      mode_t oldMask = umask(0);
+      fd = open(outfile, O_WRONLY | O_CREAT | O_TRUNC, S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP | S_IROTH | S_IWOTH);
+      umask(oldMask);
+      if (fd == -1 ) {
+        switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Error opening cache file %s: %s\n", outfile, strerror(errno));
+      }
+      else {
+        d->file = fdopen(fd, "wb");
+        if (!d->file) {
+          close(fd);
+          switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Error opening cache file %s: %s\n", outfile, strerror(errno));
+        }
+      }
+    }
+
+    if (!d->api_key) {
+      switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "deepgram_speech_feed_tts: no api_key provided\n");
+      return SWITCH_STATUS_FALSE;
+    }
+    /* format url*/
+    std::string url;
+    std::ostringstream url_stream;
+    // always use sample_rate=8000 for support jambonz caching system.
+    url_stream << "https://api.deepgram.com/v1/speak?model=" << d->voice_name << "&encoding=linear16&sample_rate=8000";
+    url = url_stream.str();
+
+    /* create the JSON body */
+    cJSON * jResult = cJSON_CreateObject();
+    cJSON_AddStringToObject(jResult, "text", text);
+    char *json = cJSON_PrintUnformatted(jResult);
+
+    cJSON_Delete(jResult);
+
+    switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "deepgram_speech_feed_tts: [%s] [%s]\n", url.c_str(), tempText);
+
+    ConnInfo_t *conn = pool.malloc() ;
+
+    CURL* easy = createEasyHandle();
+    d->conn = (void *) conn ;
+    conn->deepgram = d;
+    conn->easy = easy;
+    conn->global = &global;
+    conn->hdr_list = NULL ;
+    conn->file = d->file;
+    conn->body = json;
+    conn->flushed = false;
+    
+
+    d->circularBuffer = (void *) new CircularBuffer_t(BUFFER_GROW_SIZE);
+    // Always use deepgram at rate 8000 for helping cache audio from jambonz.
+    if (d->rate != 8000) {
+      int err;
+      d->resampler = speex_resampler_init(1, 8000, d->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;
+      }
+    }
+
+    std::ostringstream api_key_stream;
+    api_key_stream << "Authorization: Token " << d->api_key;
+
+    curl_easy_setopt(easy, CURLOPT_URL, url.c_str());
+    curl_easy_setopt(easy, CURLOPT_WRITEFUNCTION, write_cb);
+    curl_easy_setopt(easy, CURLOPT_WRITEDATA, conn);
+    curl_easy_setopt(easy, CURLOPT_ERRORBUFFER, conn->error);
+    curl_easy_setopt(easy, CURLOPT_PRIVATE, conn);
+    curl_easy_setopt(easy, CURLOPT_VERBOSE, 0L);
+    curl_easy_setopt(easy, CURLOPT_NOPROGRESS, 1L);
+    curl_easy_setopt(easy, CURLOPT_HEADERFUNCTION, header_callback);
+    curl_easy_setopt(easy, CURLOPT_HEADERDATA, conn);
+    
+    /* call this function to get a socket */
+    curl_easy_setopt(easy, CURLOPT_OPENSOCKETFUNCTION, opensocket);
+
+    /* call this function to close a socket */
+    curl_easy_setopt(easy, CURLOPT_CLOSESOCKETFUNCTION, close_socket);
+
+    conn->hdr_list = curl_slist_append(conn->hdr_list, api_key_stream.str().c_str());
+    conn->hdr_list = curl_slist_append(conn->hdr_list, "Content-Type: application/json");
+    curl_easy_setopt(easy, CURLOPT_HTTPHEADER, conn->hdr_list);
+
+    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);
+
+    rc = curl_multi_add_handle(global.multi, conn->easy);
+    mcode_test("new_conn: curl_multi_add_handle", rc);
+
+    /* start a timer to measure the duration until we receive first byte of audio */
+    conn->startTime = std::chrono::high_resolution_clock::now();
+
+    switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "deepgram_speech_feed_tts: called curl_multi_add_handle\n");
+
+
+    return SWITCH_STATUS_SUCCESS;
+  }
+
+  switch_status_t deepgram_speech_read_tts(deepgram_t* d, void *data, size_t *datalen, switch_speech_flag_t *flags) {
+    CircularBuffer_t *cBuffer = (CircularBuffer_t *) d->circularBuffer;
+    std::vector<uint16_t> pcm_data;
+
+    {
+      switch_mutex_lock(d->mutex);
+      ConnInfo_t *conn = (ConnInfo_t *) d->conn;
+      if (d->response_code > 0 && d->response_code != 200) {
+        switch_mutex_unlock(d->mutex);
+        switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "deepgram_speech_read_tts, returning failure\n") ;  
+        return SWITCH_STATUS_FALSE;
+      }
+      if (conn && conn->flushed) {
+        switch_mutex_unlock(d->mutex);
+        return SWITCH_STATUS_BREAK;
+      }
+      if (cBuffer->empty()) {
+        if (d->draining) {
+          switch_mutex_unlock(d->mutex);
+          return SWITCH_STATUS_BREAK;
+        }
+        /* no audio available yet so send silence */
+        memset(data, 255, *datalen);
+        switch_mutex_unlock(d->mutex);
+        return SWITCH_STATUS_SUCCESS;
+      }
+      size_t size = std::min((*datalen/(2 * d->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(d->mutex);
+    }
+
+    size_t data_size = pcm_data.size();
+
+    if (d->resampler) {
+      std::vector<int16_t> in(pcm_data.begin(), pcm_data.end());
+
+      std::vector<int16_t> out((*datalen));
+      spx_uint32_t in_len = data_size;
+      spx_uint32_t out_len = out.size();
+      speex_resampler_process_interleaved_int(d->resampler, in.data(), &in_len, out.data(), &out_len);
+
+      if (out_len > out.size()) {
+        switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_CRIT, "Resampler output exceeded maximum buffer size!\n");
+        return SWITCH_STATUS_FALSE;
+      }
+
+      memcpy(data, out.data(), out_len * sizeof(int16_t));
+      *datalen = out_len * sizeof(int16_t);
+    } else {
+      memcpy(data, pcm_data.data(), pcm_data.size() * sizeof(uint16_t));
+      *datalen = pcm_data.size() * sizeof(uint16_t);
+    }
+
+    return SWITCH_STATUS_SUCCESS;
+  }
+
+  switch_status_t deepgram_speech_flush_tts(deepgram_t* d) {
+    bool download_complete = d->response_code == 200;
+    switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "deepgram_speech_flush_tts, download complete? %s\n", download_complete ? "yes" : "no") ;  
+
+    ConnInfo_t *conn = (ConnInfo_t *) d->conn;
+    CircularBuffer_t *cBuffer = (CircularBuffer_t *) d->circularBuffer;
+    delete cBuffer;
+    d->circularBuffer = nullptr ;
+
+    // destroy resampler
+    if (d->resampler) {
+      speex_resampler_destroy(d->resampler);
+      d->resampler = NULL;
+    }
+
+    if (conn) {
+      conn->flushed = true;
+      if (!download_complete) {
+        if (conn->file) {
+          switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "closing audio cache file %s because download was interrupted\n", d->cache_filename);
+          if (fclose(conn->file) != 0) {
+            switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "error closing audio cache file\n");
+          }
+          conn->file = nullptr ;
+        }
+
+        if (d->cache_filename) {
+          switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "removing audio cache file %s because download was interrupted\n", d->cache_filename);
+          if (unlink(d->cache_filename) != 0) {
+            switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "cleanupConn: error removing audio cache file %s: %d:%s\n", 
+              d->cache_filename, errno, strerror(errno));
+          }
+          free(d->cache_filename);
+          d->cache_filename = nullptr ;
+        }
+      }
+    }
+    if (d->session_id) {
+      switch_core_session_t* session = switch_core_session_locate(d->session_id);
+      if (session) {
+        switch_channel_t *channel = switch_core_session_get_channel(session);
+        if (channel) {
+          switch_event_t *event;
+          if (switch_event_create(&event, SWITCH_EVENT_PLAYBACK_STOP) == 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_deepgram_response_code", std::to_string(d->response_code).c_str());
+            if (d->cache_filename && d->response_code == 200) {
+              switch_event_add_header_string(event, SWITCH_STACK_BOTTOM, "variable_tts_cache_filename", d->cache_filename);
+            }
+            if (d->response_code != 200 && d->err_msg) {
+              switch_event_add_header_string(event, SWITCH_STACK_BOTTOM, "variable_tts_error", d->err_msg);
+            }
+            switch_event_fire(&event);
+          }
+          else {
+            switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "write_cb: failed to create event\n");
+          }
+        }
+        else {
+          switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "write_cb: channel not found\n");
+        }
+        switch_core_session_rwunlock(session);
+      }
+    }
+    return SWITCH_STATUS_SUCCESS;
+  }
+
+	switch_status_t deepgram_speech_close(deepgram_t* w) {
+    switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "deepgram_speech_close\n") ;
+		return SWITCH_STATUS_SUCCESS;
+	}
+}

mod_deepgram_tts/deepgram_glue.h

@@ -0,0 +1,12 @@
+#ifndef __DEEPGRAM_GLUE_H__
+#define __DEEPGRAM_GLUE_H__
+
+switch_status_t deepgram_speech_load();
+switch_status_t deepgram_speech_open(deepgram_t* deepgram);
+switch_status_t deepgram_speech_feed_tts(deepgram_t* deepgram, char* text, switch_speech_flag_t *flags);
+switch_status_t deepgram_speech_read_tts(deepgram_t* deepgram, void *data, size_t *datalen, switch_speech_flag_t *flags);
+switch_status_t deepgram_speech_flush_tts(deepgram_t* deepgram);
+switch_status_t deepgram_speech_close(deepgram_t* deepgram);
+switch_status_t deepgram_speech_unload();
+
+#endif

mod_deepgram_tts/mod_deepgram_tts.c

@@ -0,0 +1,156 @@
+#include "mod_deepgram_tts.h"
+#include "deepgram_glue.h"
+
+SWITCH_MODULE_LOAD_FUNCTION(mod_deepgram_tts_load);
+SWITCH_MODULE_SHUTDOWN_FUNCTION(mod_deepgram_tts_shutdown);
+SWITCH_MODULE_DEFINITION(mod_deepgram_tts, mod_deepgram_tts_load, mod_deepgram_tts_shutdown, NULL);
+
+static void cleardeepgram(deepgram_t* d, int freeAll) {
+  switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "cleardeepgram\n");
+  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->ct) free(d->ct);
+  if (d->err_msg) free(d->err_msg);
+  if (d->name_lookup_time_ms) free(d->name_lookup_time_ms);
+  if (d->connect_time_ms) free(d->connect_time_ms);
+  if (d->final_response_time_ms) free(d->final_response_time_ms);
+  if (d->cache_filename) free(d->cache_filename);
+  
+
+  d->api_key = NULL;
+  d->request_id = NULL;
+
+  d->reported_latency = NULL;
+  d->ct = NULL;
+  d->err_msg = NULL;
+  d->name_lookup_time_ms = NULL;
+  d->connect_time_ms = NULL;
+  d->final_response_time_ms = NULL;
+  d->cache_filename = NULL;
+
+  if (freeAll) {
+    if (d->voice_name) free(d->voice_name);
+    if (d->session_id) free(d->session_id);
+    d->voice_name = NULL;
+    d->session_id = NULL;
+  }
+}
+
+static deepgram_t * createOrRetrievePrivateData(switch_speech_handle_t *sh) {
+  deepgram_t *d = (deepgram_t *) sh->private_info;  
+  if (!d) {
+    d = switch_core_alloc(sh->memory_pool, sizeof(*d));
+  	sh->private_info = d;
+    memset(d, 0, sizeof(*d));
+    switch_mutex_init(&d->mutex, SWITCH_MUTEX_NESTED, sh->memory_pool);
+    switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "allocated deepgram_t\n");
+  }
+  return d;
+}
+
+switch_status_t d_speech_open(switch_speech_handle_t *sh, const char *voice_name, int rate, int channels, switch_speech_flag_t *flags)
+{
+  deepgram_t *d = createOrRetrievePrivateData(sh);
+  d->voice_name = strdup(voice_name);
+  d->rate = rate;
+  switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "d_speech_open voice: %s, rate %d, channels %d\n", voice_name, rate, channels);
+  return deepgram_speech_open(d);
+}
+
+static switch_status_t d_speech_close(switch_speech_handle_t *sh, switch_speech_flag_t *flags)
+{
+  switch_status_t rc;
+  deepgram_t *d = createOrRetrievePrivateData(sh);
+  switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "d_speech_close\n");
+
+  switch_mutex_destroy(d->mutex);
+
+  rc = deepgram_speech_close(d);
+  cleardeepgram(d, 1);
+  return rc;
+}
+
+/**
+ * Freeswitch will call this function to feed us text to speak
+ */
+static switch_status_t d_speech_feed_tts(switch_speech_handle_t *sh, char *text, switch_speech_flag_t *flags)
+{
+  deepgram_t *d = createOrRetrievePrivateData(sh);
+  d->draining = 0;
+  d->reads = 0;
+
+  switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "d_speech_feed_tts\n");
+
+  return deepgram_speech_feed_tts(d, text, flags);
+}
+
+/**
+ * Freeswitch calls periodically to get some rendered audio in L16 format. We can provide up to 8k of audio at a time.
+ */
+static switch_status_t d_speech_read_tts(switch_speech_handle_t *sh, void *data, size_t *datalen, switch_speech_flag_t *flags)
+{
+  deepgram_t *d = createOrRetrievePrivateData(sh);
+  return deepgram_speech_read_tts(d, data, datalen, flags);
+}
+
+/**
+ * This is called at the end, not sure exactly what we need to do here..
+ */
+static void d_speech_flush_tts(switch_speech_handle_t *sh)
+{
+  deepgram_t *d = createOrRetrievePrivateData(sh);
+  switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "d_speech_flush_tts\n");
+  deepgram_speech_flush_tts(d);
+
+  cleardeepgram(d, 0);
+}
+
+static void d_text_param_tts(switch_speech_handle_t *sh, char *param, const char *val)
+{
+  deepgram_t *d = createOrRetrievePrivateData(sh);
+  switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "d_text_param_tts: %s=%s\n", param, val);
+  if (0 == strcmp(param, "api_key")) {
+    if (d->api_key) free(d->api_key);
+    d->api_key = strdup(val);
+  } else if (0 == strcmp(param, "voice")) {
+    if (d->voice_name) free(d->voice_name);
+    d->voice_name = strdup(val);
+  } else if (0 == strcmp(param, "session-uuid")) {
+    if (d->session_id) free(d->session_id);
+    d->session_id = strdup(val);
+  } else if (0 == strcmp(param, "write_cache_file") && switch_true(val)) {
+    d->cache_audio = 1;
+  }
+}
+
+static void d_numeric_param_tts(switch_speech_handle_t *sh, char *param, int val)
+{
+}
+static void d_float_param_tts(switch_speech_handle_t *sh, char *param, double val)
+{
+}
+
+SWITCH_MODULE_LOAD_FUNCTION(mod_deepgram_tts_load)
+{
+  switch_speech_interface_t *speech_interface;
+
+  *module_interface = switch_loadable_module_create_module_interface(pool, modname);
+  speech_interface = switch_loadable_module_create_interface(*module_interface, SWITCH_SPEECH_INTERFACE);
+  speech_interface->interface_name = "deepgram";
+  speech_interface->speech_open = d_speech_open;
+  speech_interface->speech_close = d_speech_close;
+  speech_interface->speech_feed_tts = d_speech_feed_tts;
+  speech_interface->speech_read_tts = d_speech_read_tts;
+	speech_interface->speech_flush_tts = d_speech_flush_tts;
+	speech_interface->speech_text_param_tts = d_text_param_tts;
+	speech_interface->speech_numeric_param_tts = d_numeric_param_tts;
+	speech_interface->speech_float_param_tts = d_float_param_tts;
+  return deepgram_speech_load();
+}
+
+SWITCH_MODULE_SHUTDOWN_FUNCTION(mod_deepgram_tts_shutdown)
+{
+  return deepgram_speech_unload();
+}

mod_deepgram_tts/mod_deepgram_tts.h

@@ -0,0 +1,35 @@
+#ifndef __MOD_DEEPGRAM_TTS_H__
+#define __MOD_DEEPGRAM_TTS_H__
+
+#include <switch.h>
+#include <speex/speex_resampler.h>
+
+typedef struct deepgram_data {
+  char *voice_name;
+  char *api_key;
+
+  /* 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;
+  char *err_msg;
+  char *cache_filename;
+  char *session_id;
+
+  int rate;
+  int draining;
+  int reads;
+  int cache_audio;
+
+	void *conn;
+  void *circularBuffer;
+  switch_mutex_t *mutex;
+  FILE *file;
+  SpeexResamplerState *resampler;
+} deepgram_t;
+
+#endif

mod_dub/Makefile.am

@@ -15,4 +15,4 @@ endif
 endif
 
 mod_dub_la_LIBADD   = $(switch_builddir)/libfreeswitch.la
-mod_dub_la_LDFLAGS  = -avoid-version -module -no-undefined -shared `pkg-config --libs boost` -lstdc++ -lmpg123
+mod_dub_la_LDFLAGS  = -avoid-version -module -no-undefined -lstdc++ -lmpg123 -lboost_system -lboost_thread

mod_dub/ap_http.cpp

@@ -142,6 +142,8 @@ void AudioProducerHttp::start(std::function<void(bool, const std::string&)> call
   curl_easy_setopt(_easy, CURLOPT_MAX_RECV_SPEED_LARGE, (curl_off_t)31415);
   /*Add request body*/
   if (!_body.empty()) curl_easy_setopt(_easy, CURLOPT_POSTFIELDS, _body.c_str());
+  /*Add request proxy*/
+  if (!_proxy.empty()) curl_easy_setopt(_easy, CURLOPT_PROXY, _proxy.c_str());
 
   /*Add request headers*/
   struct curl_slist *hdr_list = nullptr;
@@ -167,11 +169,12 @@ void AudioProducerHttp::queueHttpPostAudio(const std::string& url, int gain, boo
   _gain = gain;
   _loop = loop;
 }
-void AudioProducerHttp::queueHttpPostAudio(const std::string& url, const std::string& body, std::vector<std::string>& headers, int gain, bool loop) {
+void AudioProducerHttp::queueHttpPostAudio(const std::string& url, const std::string& body, std::vector<std::string>& headers, const std::string& proxy, int gain, bool loop) {
   _method = HttpMethod_t::HTTP_METHOD_POST;
   _url = url;
   _body = body;
   _headers = headers;
+  _proxy = proxy;
   _gain = gain;
   _loop = loop;
 }

mod_dub/ap_http.h

@@ -68,7 +68,7 @@ public:
 
   void queueHttpGetAudio(const std::string& url, int gain = 0, bool loop = false);
   void queueHttpPostAudio(const std::string& url, int gain = 0, bool loop = false);
-  void queueHttpPostAudio(const std::string& url, const std::string& body, std::vector<std::string>& headers, int gain = 0, bool loop = false);
+  void queueHttpPostAudio(const std::string& url, const std::string& body, std::vector<std::string>& headers, const std::string& proxy, int gain = 0, bool loop = false);
 
   Status_t getStatus() const { return _status; }
   void setStatus(Status_t status) { _status = status; }
@@ -121,6 +121,7 @@ private:
   HttpMethod_t _method;
   std::string _url;
   std::string _body;
+  std::string _proxy;
   std::vector<std::string> _headers;
   Status_t _status;
   mpg123_handle *_mh;

mod_dub/dub_glue.cpp

@@ -98,18 +98,18 @@ extern "C" {
 
   switch_status_t say_dub_track(struct cap_cb* cb, char* trackName, char* text, int gain) {
     std::vector<std::string> headers;
-    std::string url, body;
+    std::string url, body, proxy;
     Track* track = find_track_by_name(cb->tracks, trackName);
 
     if (!track) {
       switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "play_dub_track: track %s not found\n", trackName);
       return SWITCH_STATUS_FALSE;
     }
-    if (tts_vendor_parse_text(text, url, body, headers) != SWITCH_STATUS_SUCCESS) {
+    if (tts_vendor_parse_text(text, url, body, headers, proxy) != SWITCH_STATUS_SUCCESS) {
       switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "say_dub_track: failed to parse text\n");
       return SWITCH_STATUS_FALSE;
     }
-    track->queueHttpPostAudio(url, body, headers, gain);
+    track->queueHttpPostAudio(url, body, headers, proxy, gain);
     return SWITCH_STATUS_SUCCESS;
   }
 

mod_dub/track.cpp

@@ -26,12 +26,20 @@ void Track::onPlayDone(bool hasError, const std::string& errMsg) {
   switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "Track::onPlayDone %s\n", _trackName.c_str());
 
   if (!_stopping) {
+    std::shared_ptr<AudioProducer> apDone, apNext;  // to retain a ref so the ap is not destroyed while under lock below -- avoid deadlock
+  
     if (hasError) switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "onPlayDone: error: %s\n", errMsg.c_str());
-    std::lock_guard<std::mutex> lock(_mutex);
-    _apQueue.pop();
-    if (!_apQueue.empty()) {
+    {
+      std::lock_guard<std::mutex> lock(_mutex);
+      if (!_apQueue.empty()) {
+        apDone = _apQueue.front();
+        _apQueue.pop();
+      }
+       if (!_apQueue.empty()) apNext = _apQueue.front();      
+    }
+    if (apNext) {
       switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "onPlayDone: starting queued audio on track %s\n", _trackName.c_str());
-      _apQueue.front()->start(std::bind(&Track::onPlayDone, this, std::placeholders::_1, std::placeholders::_2));
+      apNext->start(std::bind(&Track::onPlayDone, this, std::placeholders::_1, std::placeholders::_2));
     }
   }
   else {
@@ -50,11 +58,17 @@ void Track::queueFileAudio(const std::string& path, int gain, bool loop) {
     _apQueue.push(ap);
     startIt = _apQueue.size() == 1;
   }
+  switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, 
+    "queueFileAudio: queue file %s on track %s\n", _trackName.c_str(), path.c_str());
 
   if (startIt) {
+    switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, 
+      "queueFileAudio: track %s starting queued file as none are in progress\n", _trackName.c_str());
     try {
       ap->start(std::bind(&Track::onPlayDone, this, std::placeholders::_1, std::placeholders::_2));
     } catch (std::exception& e) {
+      switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, 
+        "queueFileAudio: track %s error starting %s: %s\n", _trackName.c_str(), path.c_str(), e.what());
       onPlayDone(true, e.what());
     }
   }
@@ -100,11 +114,11 @@ void Track::queueHttpPostAudio(const std::string& url, int gain, bool loop) {
   }
 }
 
-void Track::queueHttpPostAudio(const std::string& url, const std::string& body, std::vector<std::string>& headers, int gain, bool loop) {
+void Track::queueHttpPostAudio(const std::string& url, const std::string& body, std::vector<std::string>& headers, const std::string& proxy, int gain, bool loop) {
   bool startIt = false;
   if (_stopping) return;
   auto ap = std::make_shared<AudioProducerHttp>(_mutex, _buffer, _sampleRate);
-    ap->queueHttpPostAudio(url, body, headers, gain, loop);
+    ap->queueHttpPostAudio(url, body, headers, proxy, gain, loop);
   {
     std::lock_guard<std::mutex> lock(_mutex);
     _apQueue.push(ap);

mod_dub/track.h

@@ -14,7 +14,7 @@ public:
   /* audio production methods */
   void queueHttpGetAudio(const std::string& url, int gain = 0, bool loop = false);
   void queueHttpPostAudio(const std::string& url, int gain = 0, bool loop = false);
-  void queueHttpPostAudio(const std::string& url, const std::string& body, std::vector<std::string>& headers, int gain = 0, bool loop = false);
+  void queueHttpPostAudio(const std::string& url, const std::string& body, std::vector<std::string>& headers, const std::string& proxy, int gain = 0, bool loop = false);
   void queueFileAudio(const std::string& path, int gain = 0, bool loop = false);
   void removeAllAudio();
 

mod_dub/tts_vendor_parser.cpp

@@ -4,6 +4,186 @@
 #include <switch_json.h>
 #include <map>
 
+switch_status_t whisper_parse_text(const std::map<std::string, std::string>& params, const std::string& text, 
+  std::string& url, std::string& body, std::vector<std::string>& headers) {
+  std::string api_key;
+  std::string voice_name;
+  std::string model_id;
+  std::string speed;
+
+  for (const auto& pair : params) {
+    if (pair.first == "api_key") {
+      api_key = pair.second;
+    } else if (pair.first == "voice") {
+      voice_name = pair.second;
+    } else if (pair.first == "model_id") {
+      model_id = pair.second;
+    } else if (pair.first == "speed") {
+      speed = pair.second;
+    }
+  }
+
+  if (api_key.empty()) {
+    switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "whisper_parse_text: no api_key provided\n");
+    return SWITCH_STATUS_FALSE;
+  }
+  if (model_id.empty()) {
+    switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "whisper_parse_text: no model_id provided\n");
+    return SWITCH_STATUS_FALSE;
+  }
+
+  url = "https://api.openai.com/v1/audio/speech";
+
+  /* create the JSON body */
+  cJSON * jResult = cJSON_CreateObject();
+  cJSON_AddStringToObject(jResult, "model", model_id.c_str());
+  cJSON_AddStringToObject(jResult, "input", text.c_str());
+  cJSON_AddStringToObject(jResult, "voice", voice_name.c_str());
+  cJSON_AddStringToObject(jResult, "response_format", "mp3");
+  if (!speed.empty()) {
+    cJSON_AddStringToObject(jResult, "speed", speed.c_str());
+  }
+  char* _body = cJSON_PrintUnformatted(jResult);
+  body = _body;
+
+  cJSON_Delete(jResult);
+  free(_body);
+
+  // Create headers
+  headers.push_back("Authorization: Bearer " + api_key);
+  headers.push_back("Content-Type: application/json");
+
+  return SWITCH_STATUS_SUCCESS;
+}
+
+switch_status_t azure_parse_text(const std::map<std::string, std::string>& params, const std::string& text, 
+  std::string& url, std::string& body, std::vector<std::string>& headers, std::string& proxy) {
+
+  std::string api_key;
+  std::string voice_name;
+  std::string language;
+  std::string region;
+  std::string endpoint;
+  std::string endpointId;
+  std::string http_proxy_ip;
+  std::string http_proxy_port;
+
+  for (const auto& pair : params) {
+    if (pair.first == "api_key") {
+      api_key = pair.second;
+    } else if (pair.first == "voice") {
+      voice_name = pair.second;
+    } else if (pair.first == "language") {
+      language = pair.second;
+    } else if (pair.first == "region") {
+      region = pair.second;
+    } else if (pair.first == "endpoint") {
+      endpoint = pair.second;
+    } else if (pair.first == "endpointId") {
+      endpointId = pair.second;
+    } else if (pair.first == "http_proxy_ip") {
+      http_proxy_ip = pair.second;
+    } else if (pair.first == "http_proxy_port") {
+      http_proxy_port = pair.second;
+    }
+  }
+
+  if (language.empty()) {
+    switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "azure_parse_text: no language provided\n");
+    return SWITCH_STATUS_FALSE;
+  }
+  if (voice_name.empty()) {
+    switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "azure_parse_text: no voice_name provided\n");
+    return SWITCH_STATUS_FALSE;
+  }
+
+  if (region.empty()) {
+    region = "westus";
+  }
+    /* format url*/
+  url = !endpoint.empty() ? endpoint : "https://" + region + ".tts.speech.microsoft.com/cognitiveservices/v1";
+
+  // Body  
+  if (strncmp(text.c_str(), "<speak", 6) == 0) {
+    body = text;
+  } else {
+    std::ostringstream body_stream;
+    body_stream << "<speak version=\"1.0\" xmlns=\"http://www.w3.org/2001/10/synthesis\" xmlns:mstts=\"https://www.w3.org/2001/mstts\" xml:lang=\"" << language << "\">";
+    body_stream << "<voice name=\"" << voice_name << "\">";
+    body_stream << text;
+    body_stream << "</voice>";
+    body_stream << "</speak>";
+    body = body_stream.str();
+  }
+
+  // Create headers
+  if (!api_key.empty()) {
+    headers.push_back("Ocp-Apim-Subscription-Key: " + api_key);
+  }
+  if (!endpointId.empty()) {
+    headers.push_back("X-Microsoft-EndpointId: " + endpointId);
+  }
+  headers.push_back("Content-Type: application/ssml+xml");
+  headers.push_back("X-Microsoft-OutputFormat: audio-16khz-32kbitrate-mono-mp3");
+
+  // Proxy
+  std::ostringstream proxy_stream;
+  if (!http_proxy_ip.empty()) {
+    proxy_stream << "http://" << http_proxy_ip;
+    if (!http_proxy_port.empty()) {
+      proxy_stream << ":" << http_proxy_port;
+    }
+  }
+  proxy = proxy_stream.str();
+
+  return SWITCH_STATUS_SUCCESS;
+}
+
+switch_status_t deepgram_parse_text(const std::map<std::string, std::string>& params, const std::string& text, 
+  std::string& url, std::string& body, std::vector<std::string>& headers) {
+
+  std::string api_key;
+  std::string voice_name;
+
+  for (const auto& pair : params) {
+    if (pair.first == "api_key") {
+      api_key = pair.second;
+    } else if (pair.first == "voice") {
+      voice_name = pair.second;
+    }
+  }
+
+  if (api_key.empty()) {
+    switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "deepgram_parse_text: no api_key provided\n");
+    return SWITCH_STATUS_FALSE;
+  }
+  if (voice_name.empty()) {
+    switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "deepgram_parse_text: no voice_name provided\n");
+    return SWITCH_STATUS_FALSE;
+  }
+
+  /* format url*/
+  std::ostringstream url_stream;
+  url_stream << "https://api.deepgram.com/v1/speak?model=" << voice_name << "&encoding=mp3";
+  url = url_stream.str();
+
+  /* create the JSON body */
+  cJSON * jResult = cJSON_CreateObject();
+  cJSON_AddStringToObject(jResult, "text", text.c_str());
+
+  char* _body = cJSON_PrintUnformatted(jResult);
+  body = _body;
+
+  cJSON_Delete(jResult);
+  free(_body);
+
+  // Create headers
+  headers.push_back("Authorization: Token " + api_key);
+  headers.push_back("Content-Type: application/json");
+
+  return SWITCH_STATUS_SUCCESS;
+}
+
 switch_status_t elevenlabs_parse_text(const std::map<std::string, std::string>& params, const std::string& text, 
   std::string& url, std::string& body, std::vector<std::string>& headers) {
 
@@ -87,7 +267,7 @@ switch_status_t elevenlabs_parse_text(const std::map<std::string, std::string>&
   return SWITCH_STATUS_SUCCESS;
 }
 
-switch_status_t tts_vendor_parse_text(const std::string& say, std::string& url, std::string& body, std::vector<std::string>& headers) {
+switch_status_t tts_vendor_parse_text(const std::string& say, std::string& url, std::string& body, std::vector<std::string>& headers, std::string& proxy) {
   size_t start = say.find("{") + 1;
   size_t end = say.find("}");
 
@@ -111,8 +291,14 @@ switch_status_t tts_vendor_parse_text(const std::string& say, std::string& url,
 
   if (params["vendor"] == "elevenlabs") {
     return elevenlabs_parse_text(params, text, url, body, headers);
+  } else if (params["vendor"] == "deepgram") {
+    return deepgram_parse_text(params, text, url, body, headers);
+  } else if (params["vendor"] == "microsoft") {
+    return azure_parse_text(params, text, url, body, headers, proxy);
+  } else if (params["vendor"] == "whisper") {
+    return whisper_parse_text(params, text, url, body, headers);
   } else {
-    switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "tts_vendor_parse_text: There is no available parser for text\n");
+    switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "tts_vendor_parse_text: There is no available parser for vendor %s\n", params["vendor"]);
     return SWITCH_STATUS_FALSE;
   }
 }

mod_dub/tts_vendor_parser.h

@@ -7,6 +7,6 @@
 #include "common.h"
 
 
-switch_status_t tts_vendor_parse_text(const std::string& say, std::string& url, std::string& body, std::vector<std::string>& headers);
+switch_status_t tts_vendor_parse_text(const std::string& say, std::string& url, std::string& body, std::vector<std::string>& headers, std::string& proxy);
 
 #endif

mod_dub/vector_math.cpp

@@ -1,6 +1,8 @@
 #include "vector_math.h"
 #include <assert.h>
 #include <string.h>
+#include <cstdlib>
+
 
 #define GRANULAR_VOLUME_MAX (50)
 #define SMAX 32767
@@ -153,6 +155,7 @@ typedef union {
 } vector_data_t;
 
 #else
+#pragma message("Building without vector math support")
 void vector_add(int16_t* a, int16_t* b, size_t len) {
     for (size_t i = 0; i < len; i++) {
         a[i] += b[i];

mod_google_transcribe/google_glue_v1.cpp

@@ -339,6 +339,16 @@ static void *SWITCH_THREAD_FUNC grpc_read_thread(switch_thread_t *thread, void *
           cb->responseHandler(session, "no_audio", cb->bugname);
         }
       }
+      else {
+        cJSON* json = cJSON_CreateObject();
+        cJSON_AddStringToObject(json, "type", "error");
+        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);
+        cb->responseHandler(session, jsonString, cb->bugname);
+        free(jsonString);
+        cJSON_Delete(json);
+      }
       switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "grpc_read_thread: finish() status %s (%d)\n", status.error_message().c_str(), status.error_code()) ;
       switch_core_session_rwunlock(session);
     }

mod_google_transcribe/google_glue_v2.cpp

@@ -296,6 +296,16 @@ static void *SWITCH_THREAD_FUNC grpc_read_thread(switch_thread_t *thread, void *
           cb->responseHandler(session, "no_audio", cb->bugname);
         }
       }
+      else {
+        cJSON* json = cJSON_CreateObject();
+        cJSON_AddStringToObject(json, "type", "error");
+        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);
+        cb->responseHandler(session, jsonString, cb->bugname);
+        free(jsonString);
+        cJSON_Delete(json);
+      }
       switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "grpc_read_thread: finish() status %s (%d)\n", status.error_message().c_str(), status.error_code()) ;
       switch_core_session_rwunlock(session);
     }