Feat/mark bidirectional streaming (#102)

* initial support for mark feature in bidirectional streaming

Signed-off-by: Dave Horton <daveh@beachdognet.com>

* wip

* allow max of 30 marks on any connection

* fix send multiple json in same ws text frame (#108)

* fix send multiple json in same ws text frame

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

* fix mark is not sent without more bidirectional audio

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

---------

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

---------

Signed-off-by: Dave Horton <daveh@beachdognet.com>
Signed-off-by: Hoan HL <quan.luuhoang8@gmail.com>
Co-authored-by: Hoan Luu Huu <110280845+xquanluu@users.noreply.github.com>

LICENSE_MIT

@@ -2,7 +2,7 @@ Under specific conditions that are described here: https://github.com/jambonz/fr
 
 The MIT License (MIT)
 
-Copyright (c) 2023 Drachtio Communications Services, LLC
+Copyright (c) 2024 FirstFive8, Inc
 
 Permission is hereby granted, free of charge, to any person obtaining a copy
 of this software and associated documentation files (the "Software"), to deal

mod_audio_fork/audio_pipe.cpp

@@ -65,13 +65,17 @@ int AudioPipe::lws_callback(struct lws *wsi,
           std::string username, password;
 
           ap->getBasicAuth(username, password);
-          lwsl_notice("AudioPipe::lws_service_thread LWS_CALLBACK_CLIENT_APPEND_HANDSHAKE_HEADER username: %s, password: xxxxxx\n", username.c_str());
+          switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_NOTICE, "AudioPipe::lws_service_thread LWS_CALLBACK_CLIENT_APPEND_HANDSHAKE_HEADER username: %s, password: xxxxxx\n", username.c_str());
           if (dch_lws_http_basic_auth_gen(username.c_str(), password.c_str(), b, sizeof(b))) break;
           if (lws_add_http_header_by_token(wsi, WSI_TOKEN_HTTP_AUTHORIZATION, (unsigned char *)b, strlen(b), p, end)) return -1;
         }
       }
       break;
 
+    case LWS_CALLBACK_WS_CLIENT_DROP_PROTOCOL:
+      switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "AudioPipe::lws_service_thread LWS_CALLBACK_WS_CLIENT_DROP_PROTOCOL\n");
+      break;
+
     case LWS_CALLBACK_EVENT_WAIT_CANCELLED:
       processPendingConnects(vhd);
       processPendingDisconnects(vhd);
@@ -81,13 +85,13 @@ int AudioPipe::lws_callback(struct lws *wsi,
       {
         AudioPipe* ap = findAndRemovePendingConnect(wsi);
         int rc = lws_http_client_http_response(wsi);
-        lwsl_err("AudioPipe::lws_service_thread LWS_CALLBACK_CLIENT_CONNECTION_ERROR: %s, response status %d\n", in ? (char *)in : "(null)", rc); 
+        switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR,"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, NULL, len);
         }
         else {
-          lwsl_err("AudioPipe::lws_service_thread LWS_CALLBACK_CLIENT_CONNECTION_ERROR unable to find wsi %p..\n", wsi); 
+          switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR,"AudioPipe::lws_service_thread LWS_CALLBACK_CLIENT_CONNECTION_ERROR unable to find wsi %p..\n", wsi); 
         }
       }      
       break;
@@ -102,7 +106,7 @@ int AudioPipe::lws_callback(struct lws *wsi,
           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); 
+          switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR,"AudioPipe::lws_service_thread LWS_CALLBACK_CLIENT_ESTABLISHED %s unable to find wsi %p..\n", ap->m_uuid.c_str(), wsi); 
         }
       }      
       break;
@@ -110,7 +114,7 @@ int AudioPipe::lws_callback(struct lws *wsi,
       {
         AudioPipe* ap = *ppAp;
         if (!ap) {
-          lwsl_err("AudioPipe::lws_service_thread LWS_CALLBACK_CLIENT_CLOSED %s unable to find wsi %p..\n", ap->m_uuid.c_str(), wsi); 
+          switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR,"AudioPipe::lws_service_thread LWS_CALLBACK_CLIENT_CLOSED %s unable to find wsi %p..\n", ap->m_uuid.c_str(), wsi); 
           return 0;
         }
         if (ap->m_state == LWS_CLIENT_DISCONNECTING) {
@@ -119,7 +123,7 @@ int AudioPipe::lws_callback(struct lws *wsi,
         }
         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());
+          switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO,"%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, NULL, len);
         }
         ap->m_state = LWS_CLIENT_DISCONNECTED;
@@ -137,19 +141,26 @@ int AudioPipe::lws_callback(struct lws *wsi,
         
         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); 
+          switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR,"AudioPipe::lws_service_thread LWS_CALLBACK_CLIENT_RECEIVE %s unable to find wsi %p..\n", ap->m_uuid.c_str(), wsi); 
+          return 0;
+        }
+        
+        if (ap->m_state == LWS_CLIENT_DISCONNECTING) {
+          switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO,"AudioPipe::lws_service_thread race condition: got incoming message while closing the connection.\n");
           return 0;
         }
 
         if (lws_frame_is_binary(wsi)) {
-          if (ap->is_bidirectional_audio_stream()) {
+          if (len > 0 && 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 {
+          } else if (len > 0) {
-            lwsl_err("AudioPipe::lws_service_thread LWS_CALLBACK_CLIENT_RECEIVE received binary frame, discarding.\n");
+            switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR,"AudioPipe::lws_service_thread LWS_CALLBACK_CLIENT_RECEIVE received unexpected binary frame, discarding.\n");
           }
-          return 0;
+          else {
+            switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG,"AudioPipe::lws_service_thread LWS_CALLBACK_CLIENT_RECEIVE received zero length binary frame, discarding.\n");
           }
-
+        }
+        else {
           if (lws_is_first_fragment(wsi)) {
             // allocate a buffer for the entire chunk of memory needed
             assert(nullptr == ap->m_recv_buf);
@@ -161,13 +172,13 @@ int AudioPipe::lws_callback(struct lws *wsi,
           size_t write_offset = ap->m_recv_buf_ptr - ap->m_recv_buf;
           size_t remaining_space = ap->m_recv_buf_len - write_offset;
           if (remaining_space < len) {
-          lwsl_notice("AudioPipe::lws_service_thread LWS_CALLBACK_CLIENT_RECEIVE buffer realloc needed.\n");
+            //switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO,"AudioPipe::lws_service_thread LWS_CALLBACK_CLIENT_RECEIVE buffer realloc needed.\n");
             size_t newlen = ap->m_recv_buf_len + RECV_BUF_REALLOC_SIZE;
             if (newlen > MAX_RECV_BUF_SIZE) {
               free(ap->m_recv_buf);
               ap->m_recv_buf = ap->m_recv_buf_ptr = nullptr;
               ap->m_recv_buf_len = 0;
-            lwsl_notice("AudioPipe::lws_service_thread LWS_CALLBACK_CLIENT_RECEIVE max buffer exceeded, truncating message.\n");
+              switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR,"AudioPipe::lws_service_thread LWS_CALLBACK_CLIENT_RECEIVE max buffer exceeded, truncating message.\n");
             }
             else {
               ap->m_recv_buf = (uint8_t*) realloc(ap->m_recv_buf, newlen);
@@ -194,19 +205,20 @@ int AudioPipe::lws_callback(struct lws *wsi,
             }
           }
         }
+      }
       break;
 
     case LWS_CALLBACK_CLIENT_WRITEABLE:
       {
         AudioPipe* ap = *ppAp;
         if (!ap) {
-          lwsl_err("AudioPipe::lws_service_thread LWS_CALLBACK_CLIENT_WRITEABLE %s unable to find wsi %p..\n", ap->m_uuid.c_str(), wsi); 
+          switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR,"AudioPipe::lws_service_thread LWS_CALLBACK_CLIENT_WRITEABLE %s unable to find wsi %p..\n", ap->m_uuid.c_str(), wsi); 
           return 0;
         }
 
         // check for graceful close - send a zero length binary frame
         if (ap->isGracefulShutdown()) {
-          lwsl_notice("%s graceful shutdown - sending zero length binary frame to flush any final responses\n", ap->m_uuid.c_str());
+          switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR,"%s graceful shutdown - sending zero length binary frame to flush any final responses\n", ap->m_uuid.c_str());
           std::lock_guard<std::mutex> lk(ap->m_audio_mutex);
           int sent = lws_write(wsi, (unsigned char *) ap->m_audio_buffer + LWS_PRE, 0, LWS_WRITE_BINARY);
           return 0;
@@ -215,20 +227,21 @@ int AudioPipe::lws_callback(struct lws *wsi,
         // check for text frames to send
         {
           std::lock_guard<std::mutex> lk(ap->m_text_mutex);
-          if (ap->m_metadata.length() > 0) {
+          if (!ap->m_metadata_list.empty()) {
-            uint8_t buf[ap->m_metadata.length() + LWS_PRE];
+            const std::string& message = ap->m_metadata_list.front();
-            memcpy(buf + LWS_PRE, ap->m_metadata.c_str(), ap->m_metadata.length());
+            uint8_t buf[message.length() + LWS_PRE];
-            int n = ap->m_metadata.length();
+            memcpy(buf + LWS_PRE, message.c_str(), message.length());
+            int n = message.length();
             int m = lws_write(wsi, buf + LWS_PRE, n, LWS_WRITE_TEXT);
-            ap->m_metadata.clear();
+
             if (m < n) {
-              return -1;
+              return -1; // Failed to send the full message
             }
 
-            // there may be audio data, but only one write per writeable event
+            // Remove the message that was successfully sent
-            // get it next time
+            ap->m_metadata_list.pop_front();
+            // Request another writable event if there are more messages
             lws_callback_on_writable(wsi);
-
             return 0;
           }
         }
@@ -245,7 +258,7 @@ int AudioPipe::lws_callback(struct lws *wsi,
             size_t datalen = ap->m_audio_buffer_write_offset - LWS_PRE;
             int sent = lws_write(wsi, (unsigned char *) ap->m_audio_buffer + LWS_PRE, datalen, LWS_WRITE_BINARY);
             if (sent < datalen) {
-              lwsl_err("AudioPipe::lws_service_thread LWS_CALLBACK_CLIENT_WRITEABLE %s attemped to send %lu only sent %d wsi %p..\n", 
+              switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO,"AudioPipe::lws_service_thread LWS_CALLBACK_CLIENT_WRITEABLE %s attemped to send %lu only sent %d wsi %p..\n", 
                 ap->m_uuid.c_str(), datalen, sent, wsi); 
             }
             ap->m_audio_buffer_write_offset = LWS_PRE;
@@ -374,7 +387,7 @@ void AudioPipe::addPendingConnect(AudioPipe* ap) {
   {
     std::lock_guard<std::mutex> guard(mutex_connects);
     pendingConnects.push_back(ap);
-    lwsl_notice("%s after adding connect there are %lu pending connects\n", 
+    switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG,"%s after adding connect there are %lu pending connects\n", 
       ap->m_uuid.c_str(), pendingConnects.size());
   }
   lws_cancel_service(context);
@@ -384,7 +397,7 @@ void AudioPipe::addPendingDisconnect(AudioPipe* ap) {
   {
     std::lock_guard<std::mutex> guard(mutex_disconnects);
     pendingDisconnects.push_back(ap);
-    lwsl_notice("%s after adding disconnect there are %lu pending disconnects\n", 
+    switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG,"%s after adding disconnect there are %lu pending disconnects\n", 
       ap->m_uuid.c_str(), pendingDisconnects.size());
   }
   lws_cancel_service(ap->m_vhd->context);
@@ -423,11 +436,11 @@ bool AudioPipe::lws_service_thread() {
   info.timeout_secs_ah_idle = 10;       // secs to allow a client to hold an ah without using it
   info.retry_and_idle_policy = &retry;
 
-  lwsl_notice("AudioPipe::lws_service_thread creating context\n");
+  switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO,"AudioPipe::lws_service_thread creating context\n");
 
   context = lws_create_context(&info);
   if (!context) {
-    lwsl_err("AudioPipe::lws_service_thread failed creating context\n"); 
+    switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR,"AudioPipe::lws_service_thread failed creating context\n"); 
     return false;
   }
 
@@ -444,16 +457,16 @@ bool AudioPipe::lws_service_thread() {
 
 void AudioPipe::initialize(const char* protocol, int loglevel, log_emit_function logger) {
   protocolName = protocol;
-  //lws_set_log_level(loglevel, logger);
+  lws_set_log_level(loglevel, logger);
 
-  lwsl_notice("AudioPipe::initialize starting\n"); 
+  switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_NOTICE,"AudioPipe::initialize starting\n"); 
   std::lock_guard<std::mutex> lock(mapMutex);
   stopFlag = false;
   serviceThread = std::thread(&AudioPipe::lws_service_thread);
 }
 
 bool AudioPipe::deinitialize() {
-  lwsl_notice("AudioPipe::deinitialize\n"); 
+  switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_NOTICE,"AudioPipe::deinitialize\n"); 
   std::lock_guard<std::mutex> lock(mapMutex);
   stopFlag = true;
   if (serviceThread.joinable()) {
@@ -508,7 +521,7 @@ bool AudioPipe::connect_client(struct lws_per_vhost_data *vhd) {
   m_vhd = vhd;
 
   m_wsi = lws_client_connect_via_info(&i);
-  lwsl_notice("%s attempting connection, wsi is %p\n", m_uuid.c_str(), m_wsi);
+  switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG,"%s attempting connection, wsi is %p\n", m_uuid.c_str(), m_wsi);
 
   return nullptr != m_wsi;
 }
@@ -517,7 +530,7 @@ void AudioPipe::bufferForSending(const char* text) {
   if (m_state != LWS_CLIENT_CONNECTED) return;
   {
     std::lock_guard<std::mutex> lk(m_text_mutex);
-    m_metadata.append(text);
+    m_metadata_list.emplace_back(text);
   }
   addPendingWrite(this);
 }

mod_audio_fork/audio_pipe.hpp

@@ -130,7 +130,7 @@ namespace drachtio {
     std::string m_bugname;
     unsigned int m_port;
     std::string m_path;
-    std::string m_metadata;
+    std::list<std::string> m_metadata_list;
     std::mutex m_text_mutex;
     std::mutex m_audio_mutex;
     int m_sslFlags;

mod_audio_fork/lws_glue.cpp

@@ -21,11 +21,14 @@
 
 #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)
+#define BUFFER_GROW_SIZE (16384)
+#define AUDIO_MARKER 0xFFFF
+#define MAX_MARKS (30)
 
 namespace {
   static const char *requestedBufferSecs = std::getenv("MOD_AUDIO_FORK_BUFFER_SECS");
@@ -37,54 +40,174 @@ namespace {
   static unsigned int idxCallCount = 0;
   static uint32_t playCount = 0;
 
+  static bool markCountExceeded(private_t* tech_pvt) {
+    if (nullptr != tech_pvt->pVecMarksInUse) {
+      std::deque<std::string>* pVecMarksInUse = static_cast<std::deque<std::string>*>(tech_pvt->pVecMarksInUse);
+      std::deque<std::string>* pVecMarksInInventory = static_cast<std::deque<std::string>*>(tech_pvt->pVecMarksInInventory);
+      return pVecMarksInUse->size()+ pVecMarksInInventory->size() >= MAX_MARKS;
+    }
+    return false;
+  }
+
   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;
+    std::vector<uint8_t> data;
-    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));
 
+    // Prepend the set-aside byte if there is one
+    if (tech_pvt->has_set_aside_byte) {
+        data.push_back(tech_pvt->set_aside_byte);
+        tech_pvt->has_set_aside_byte = false;
+    }
 
+    // Append the new incoming message
+    data.insert(data.end(), message, message + dataLength);
+
+    // Check if the total data length is now odd
+    if (data.size() % 2 != 0) {
+        // Set aside the last byte
+        tech_pvt->set_aside_byte = data.back();
+        tech_pvt->has_set_aside_byte = true;
+        data.pop_back(); // Remove the last byte from the data vector
+    }
+
+    // Convert the data to 16-bit elements
+    const uint16_t* data_uint16 = reinterpret_cast<const uint16_t*>(data.data());
+    size_t numSamples = data.size() / sizeof(uint16_t);
+
+    // Access the prebuffer
+    CircularBuffer_t* cBuffer = static_cast<CircularBuffer_t*>(tech_pvt->streamingPreBuffer);
+
+    int numMarkers = 0;
+    std::deque<std::string>* pVecMarksInInventory = nullptr;
+    std::deque<std::string>* pVecMarksInUse = nullptr;
+    if (nullptr != tech_pvt->pVecMarksInInventory) {
+      pVecMarksInInventory = static_cast<std::deque<std::string>*>(tech_pvt->pVecMarksInInventory);
+      pVecMarksInUse = static_cast<std::deque<std::string>*>(tech_pvt->pVecMarksInUse);
+      numMarkers = pVecMarksInInventory->size();
+
+      // move inventory to in-use
+      pVecMarksInUse->insert(pVecMarksInUse->end(), pVecMarksInInventory->begin(), pVecMarksInInventory->end());
+      pVecMarksInInventory->clear();
+    }
+  
+    // Ensure the prebuffer has enough capacity
+    if (cBuffer->capacity() - cBuffer->size() < numSamples + numMarkers) {
+        size_t newCapacity = cBuffer->size() + std::max(numSamples + numMarkers, (size_t)BUFFER_GROW_SIZE);
+        cBuffer->set_capacity(newCapacity);
+    }
+
+    // prepend any markers
+    while (numMarkers-- > 0) {
+      cBuffer->push_back(AUDIO_MARKER);
+    }
+
+    // Append the data to the prebuffer
+    cBuffer->insert(cBuffer->end(), data_uint16, data_uint16 + numSamples);
+    //switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "Appended %zu 16-bit samples to the prebuffer.\n", numSamples);
+
+    // if we haven't reached threshold amount of prebuffered data, return
+    if (cBuffer->size() < tech_pvt->streamingPreBufSize) {
+        //switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "Prebuffered data is below threshold %u, returning.\n", tech_pvt->streamingPreBufSize);
+        return SWITCH_STATUS_SUCCESS;
+    }
+
+    //switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "Prebuffered data samples %u is above threshold %u, prepare to playout.\n", cBuffer->size(), tech_pvt->streamingPreBufSize);
+
+    // after initial pre-buffering, rachet down the threshold to 40ms
+    tech_pvt->streamingPreBufSize = 320 * tech_pvt->downscale_factor * 2;
+
+    // Check for downsampling factor
+    size_t downsample_factor = tech_pvt->downscale_factor;
+
+    // Calculate the number of samples that can be evenly divided by the downsample factor
+    size_t numCompleteSamples = (cBuffer->size() / downsample_factor) * downsample_factor;
+
+    // Handle leftover samples
+    std::vector<uint16_t> leftoverSamples;
+    size_t numLeftoverSamples = cBuffer->size() - numCompleteSamples;
+    if (numLeftoverSamples > 0) {
+        leftoverSamples.assign(cBuffer->end() - numLeftoverSamples, cBuffer->end());
+        cBuffer->resize(numCompleteSamples);
+        //switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "Temporarily removing %u leftover samples due to downsampling.\n", numLeftoverSamples);
+    }
+
+    // resample if necessary
+    std::vector<int16_t> out;
+    try {
       if (tech_pvt->bidirectional_audio_resampler) {
-      std::vector<int16_t> in(pcm_data.begin(), pcm_data.end());
+        // Improvement: Use assign to convert circular buffer to vector for resampling
+        std::vector<int16_t> in;
+        in.assign(cBuffer->begin(), cBuffer->end()); 
+        out.resize(in.size() * 6); // max upsampling would be from 8k -> 48k
 
-      std::vector<int16_t> out(dataLength);
+        spx_uint32_t in_len = in.size();
-      spx_uint32_t in_len = pcm_data.size();
         spx_uint32_t out_len = out.size();
+
+        //switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "Resampling %u samples into a buffer that can hold %u samples\n", in.size(), out_len);
+
         speex_resampler_process_interleaved_int(tech_pvt->bidirectional_audio_resampler, in.data(), &in_len, out.data(), &out_len);
 
-      if (out_len > out.size()) {
+        // Resize the output buffer to match the output length from resampler
-        switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_CRIT, "Resampler output exceeded maximum buffer size!\n");
+        //switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "Resizing output buffer from %u to %u samples\n", in.size(), out_len);
+
+        out.resize(out_len);
+      }
+      else {
+        // If no resampling is needed, copy the data from the prebuffer to the output buffer
+        out.assign(cBuffer->begin(), cBuffer->end());
+      }
+    } catch (const std::exception& e) {
+      switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Error resampling incoming binary message: %s\n", e.what());
+      return SWITCH_STATUS_FALSE;
+    } catch (...) {
+      switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Error resampling incoming binary message\n");
       return SWITCH_STATUS_FALSE;
     }
 
-      // Resize the pcm_data to match the output length from resampler, and then copy the resampled data into it.
+    if (nullptr != tech_pvt->mutex && switch_mutex_trylock(tech_pvt->mutex) == SWITCH_STATUS_SUCCESS) {
-      pcm_data.resize(out_len);
+      CircularBuffer_t *playoutBuffer = (CircularBuffer_t *) tech_pvt->streamingPlayoutBuffer;
-      memcpy(pcm_data.data(), out.data(), out_len * sizeof(int16_t));
-    }
-    switch_mutex_lock(tech_pvt->mutex);
 
+      try {
         // Resize the buffer if necessary
-    size_t bytesResampled = pcm_data.size() * sizeof(uint16_t);
+        if (playoutBuffer->capacity() - playoutBuffer->size() < out.size()) {
-    if (cBuffer->capacity() - cBuffer->size() < bytesResampled / sizeof(uint16_t)) {
+          size_t newCapacity = playoutBuffer->size() + std::max(out.size(), (size_t)BUFFER_GROW_SIZE);
-      // If buffer exceeds some max size, you could return SWITCH_STATUS_FALSE to abort the transfer
+          playoutBuffer->set_capacity(newCapacity);
-      // if (cBuffer->size() + std::max(bytesResampled / sizeof(uint16_t), (size_t)BUFFER_GROW_SIZE) > MAX_BUFFER_SIZE) return SWITCH_STATUS_FALSE;
+          //switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "Resized playout buffer to new capacity: %zu\n", newCapacity);
-
-      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());
+        playoutBuffer->insert(playoutBuffer->end(), out.begin(), out.end());
-
+        //switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "Appended %zu 16-bit samples to the playout buffer.\n", out.size());
+      } catch (const std::exception& e) {
         switch_mutex_unlock(tech_pvt->mutex);
+        cBuffer->clear();
+        switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Error processing incoming binary message: %s\n", e.what());
+        return SWITCH_STATUS_FALSE;
+      } catch (...) {
+        switch_mutex_unlock(tech_pvt->mutex);
+        cBuffer->clear();
+        switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Error processing incoming binary message\n");
+        return SWITCH_STATUS_FALSE;
+      }
+      switch_mutex_unlock(tech_pvt->mutex);
+      cBuffer->clear();
+
+      // Put the leftover samples back in the prebuffer for the next time
+      if (!leftoverSamples.empty()) {
+          cBuffer->insert(cBuffer->end(), leftoverSamples.begin(), leftoverSamples.end());
+          //switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "Put back %u leftover samples into the prebuffer.\n", leftoverSamples.size());
+      }
+      return SWITCH_STATUS_SUCCESS;
+    }
+    switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "Failed to get mutext (temp)\n");
 
     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;
     cJSON* json = parse_json(session, msg, type) ;
     if (json) {
-      switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "(%u) processIncomingMessage - received %s message\n", tech_pvt->id, type.c_str());
+      switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "(%u) processIncomingMessage - received %s message %s\n", tech_pvt->id, type.c_str(), message);
       cJSON* jsonData = cJSON_GetObjectItem(json, "data");
       if (0 == type.compare("playAudio") &&
         // playAudio is enabled and there is no bidirectional audio from stream is enabled.
@@ -171,6 +294,66 @@ namespace {
         // kill any current playback on the channel
         switch_channel_t *channel = switch_core_session_get_channel(session);
         switch_channel_set_flag_value(channel, CF_BREAK, 2);
+
+        // this will dump buffered incoming audio
+        tech_pvt->clear_bidirectional_audio_buffer = true;
+      }
+      else if (0 == type.compare("mark")) {
+        cJSON* data = cJSON_GetObjectItem(json, "data");
+        if (data) {
+          cJSON* name = cJSON_GetObjectItem(data, "name");
+          if (cJSON_IsString(name) && name->valuestring) {
+            if (markCountExceeded(tech_pvt)) {
+              switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "(%u) processIncomingMessage - mark count exceeded, discarding mark %s\n", tech_pvt->id, cJSON_GetStringValue(name));
+            }
+            else {
+              if (nullptr == tech_pvt->pVecMarksInInventory) {
+                tech_pvt->pVecMarksInInventory = static_cast<void *>(new std::deque<std::string>());
+                tech_pvt->pVecMarksInUse = static_cast<void *>(new std::deque<std::string>());
+                tech_pvt->pVecMarksCleared = static_cast<void *>(new std::deque<std::string>());
+              }
+              std::deque<std::string>* pVec = static_cast<std::deque<std::string>*>(tech_pvt->pVecMarksInInventory);
+              pVec->push_back(name->valuestring);
+            }
+          }
+        }
+      }
+      else if (0 == type.compare("clearMarks")) {
+        switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "(%u) processIncomingMessage - received clearMarks\n", tech_pvt->id);
+        if (nullptr != tech_pvt->pVecMarksInInventory) {
+          std::deque<std::string>* pVecMarksInInventory = static_cast<std::deque<std::string>*>(tech_pvt->pVecMarksInInventory);
+          std::deque<std::string>* pVecMarksInUse = static_cast<std::deque<std::string>*>(tech_pvt->pVecMarksInUse);
+          std::deque<std::string>* pVecMarksCleared = static_cast<std::deque<std::string>*>(tech_pvt->pVecMarksCleared);
+          pVecMarksCleared->insert(pVecMarksCleared->end(), pVecMarksInUse->begin(), pVecMarksInUse->end());
+          pVecMarksCleared->insert(pVecMarksCleared->end(), pVecMarksInInventory->begin(), pVecMarksInInventory->end());
+          pVecMarksInInventory->clear();
+          pVecMarksInUse->clear();
+        }
+      }
+      else if (0 == type.compare("transcription")) {
+        char* jsonString = cJSON_PrintUnformatted(jsonData);
+        tech_pvt->responseHandler(session, EVENT_TRANSCRIPTION, jsonString);
+        free(jsonString);        
+      }
+      else if (0 == type.compare("transfer")) {
+        char* jsonString = cJSON_PrintUnformatted(jsonData);
+        tech_pvt->responseHandler(session, EVENT_TRANSFER, jsonString);
+        free(jsonString);                
+      }
+      else if (0 == type.compare("disconnect")) {
+        char* jsonString = cJSON_PrintUnformatted(jsonData);
+        tech_pvt->responseHandler(session, EVENT_DISCONNECT, jsonString);
+        free(jsonString);        
+      }
+      else if (0 == type.compare("error")) {
+        char* jsonString = cJSON_PrintUnformatted(jsonData);
+        tech_pvt->responseHandler(session, EVENT_ERROR, jsonString);
+        free(jsonString);        
+      }
+      else if (0 == type.compare("json")) {
+        char* jsonString = cJSON_PrintUnformatted(json);
+        tech_pvt->responseHandler(session, EVENT_JSON, jsonString);
+        free(jsonString);
       }
       else if (0 == type.compare("transcription")) {
         char* jsonString = cJSON_PrintUnformatted(jsonData);
@@ -290,10 +473,23 @@ 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->streamingPlayoutBuffer = (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;
+    tech_pvt->clear_bidirectional_audio_buffer = false;
+    tech_pvt->has_set_aside_byte = 0;
+    tech_pvt->downscale_factor = 1;
+    if (bidirectional_audio_sample_rate > sampling) {
+      tech_pvt->downscale_factor = bidirectional_audio_sample_rate / sampling;
+      switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "downscale_factor is %d\n", tech_pvt->downscale_factor);
+    }
+    tech_pvt->streamingPreBufSize = 320 * tech_pvt->downscale_factor * 4; // min 80ms prebuffer
+    tech_pvt->streamingPreBuffer = (void *) new CircularBuffer_t(8192);
+    tech_pvt->pVecMarksInInventory = nullptr;
+    tech_pvt->pVecMarksInUse = nullptr;
+    tech_pvt->pVecMarksCleared = nullptr;
+
     strncpy(tech_pvt->bugname, bugname, MAX_BUG_LEN);
     if (metadata) strncpy(tech_pvt->initialMetadata, metadata, MAX_METADATA_LEN);
     
@@ -323,8 +519,8 @@ namespace {
     }
 
     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);
+      switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "(%u) bidirectional audio resampling from %u to %u, channels %d\n", tech_pvt->id, bidirectional_audio_sample_rate, sampling, channels);
-      tech_pvt->bidirectional_audio_resampler = speex_resampler_init(channels, bidirectional_audio_sample_rate, sampling, SWITCH_RESAMPLE_QUALITY, &err);
+      tech_pvt->bidirectional_audio_resampler = speex_resampler_init(1, 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;
@@ -350,10 +546,38 @@ namespace {
       switch_mutex_destroy(tech_pvt->mutex);
       tech_pvt->mutex = nullptr;
     }
-    if (tech_pvt->circularBuffer) {
+    if (tech_pvt->streamingPlayoutBuffer) {
-      CircularBuffer_t *cBuffer = (CircularBuffer_t *) tech_pvt->circularBuffer;
+      CircularBuffer_t *cBuffer = (CircularBuffer_t *) tech_pvt->streamingPlayoutBuffer;
       delete cBuffer;
-      tech_pvt->circularBuffer = nullptr;
+      tech_pvt->streamingPlayoutBuffer = nullptr;
+    }
+    if (tech_pvt->streamingPreBuffer) {
+      CircularBuffer_t *cBuffer = (CircularBuffer_t *) tech_pvt->streamingPreBuffer;
+      delete cBuffer;
+      tech_pvt->streamingPreBuffer = nullptr;
+    }
+
+    if (nullptr == tech_pvt->pVecMarksInInventory) {
+      delete static_cast<std::deque<std::string>*>(tech_pvt->pVecMarksInInventory);
+      tech_pvt->pVecMarksInInventory = nullptr;
+      delete static_cast<std::deque<std::string>*>(tech_pvt->pVecMarksInUse);
+      tech_pvt->pVecMarksInUse = nullptr;
+      delete static_cast<std::deque<std::string>*>(tech_pvt->pVecMarksCleared);
+      tech_pvt->pVecMarksCleared = nullptr;
+    }
+  }
+
+  static void send_mark_event(private_t* tech_pvt, const char* name, int cleared = false) {
+    drachtio::AudioPipe *pAudioPipe = static_cast<drachtio::AudioPipe *>(tech_pvt->pAudioPipe);
+    std::ostringstream json;
+    json << "{\"type\": \"mark\", \"data\": {\"name\":\"" << name << "\", ";
+    if (cleared) json << "\"event\": \"cleared\"}}";
+    else json << "\"event\": \"playout\"}}";
+
+    if (pAudioPipe) {
+      std::string str = json.str();
+      pAudioPipe->bufferForSending(str.c_str());
+      switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "(%u) send_mark_event: %s\n", tech_pvt->id, str.c_str());
     }
   }
 
@@ -453,8 +677,8 @@ extern "C" {
     switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_NOTICE, "mod_audio_fork: audio buffer (in secs):    %d secs\n", nAudioBufferSecs);
     switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_NOTICE, "mod_audio_fork: sub-protocol:              %s\n", mySubProtocolName);
  
-    int logs = LLL_ERR | LLL_WARN | LLL_NOTICE ;
+    //int logs = LLL_ERR | LLL_WARN | LLL_NOTICE | LLL_INFO | LLL_PARSER | LLL_HEADER | LLL_EXT | LLL_CLIENT  | LLL_LATENCY | LLL_DEBUG ;
-     //LLL_INFO | LLL_PARSER | LLL_HEADER | LLL_EXT | LLL_CLIENT  | LLL_LATENCY | LLL_DEBUG ;
+    int logs = LLL_ERR | LLL_WARN | LLL_NOTICE;
     drachtio::AudioPipe::initialize(mySubProtocolName, logs, lws_logger);
     switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_NOTICE, "mod_audio_fork successfully initialized\n");
     return SWITCH_STATUS_SUCCESS;
@@ -706,8 +930,38 @@ extern "C" {
   }
 
   switch_bool_t dub_speech_frame(switch_media_bug_t *bug, private_t* tech_pvt) {
-    CircularBuffer_t *cBuffer = (CircularBuffer_t *) tech_pvt->circularBuffer;
+    CircularBuffer_t *cBuffer = (CircularBuffer_t *) tech_pvt->streamingPlayoutBuffer;
     if (switch_mutex_trylock(tech_pvt->mutex) == SWITCH_STATUS_SUCCESS) {
+
+      // if flag was set to clear the buffer, do so and clear the flag
+      if (tech_pvt->clear_bidirectional_audio_buffer) {
+        switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "(%u) dub_speech_frame - clearing buffer\n", tech_pvt->id); 
+        cBuffer->clear();
+        tech_pvt->clear_bidirectional_audio_buffer = false;
+
+        // send "mark" event for any queued markers
+        if (nullptr != tech_pvt->pVecMarksInInventory) {
+          std::deque<std::string>* pVecMarksInInventory = static_cast<std::deque<std::string>*>(tech_pvt->pVecMarksInInventory);
+          std::deque<std::string>* pVecMarksInUse = static_cast<std::deque<std::string>*>(tech_pvt->pVecMarksInUse);
+          if (pVecMarksInInventory->size() + pVecMarksInUse->size() > 0) {
+            std::deque<std::string> vec = *pVecMarksInUse;
+            vec.insert(vec.end(), pVecMarksInInventory->begin(), pVecMarksInInventory->end());
+            for (auto it = vec.begin(); it != vec.end(); ++it) {
+              switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "(%u) dub_speech_frame - Marker %s cleared\n",
+                  tech_pvt->id, it->c_str());
+              send_mark_event(tech_pvt, it->c_str(), true);
+            }
+
+            // put the "in-use" ones into the "cleared" queue so we dont notify again when they eventually come through
+            std::deque<std::string>* pVecMarksCleared = static_cast<std::deque<std::string>*>(tech_pvt->pVecMarksCleared);
+            pVecMarksCleared->insert(pVecMarksCleared->end(), pVecMarksInUse->begin(), pVecMarksInUse->end());
+
+            pVecMarksInUse->clear();
+            pVecMarksInInventory->clear();
+          }
+        }
+      }
+      else {
         switch_frame_t* rframe = switch_core_media_bug_get_write_replace_frame(bug);
         int16_t *fp = reinterpret_cast<int16_t*>(rframe->data);
 
@@ -719,15 +973,79 @@ extern "C" {
 
         int samplesToCopy = std::min(static_cast<int>(cBuffer->size()), static_cast<int>(rframe->samples));
 
+        //switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "(%u) dub_speech_frame - samples to copy %u\n", tech_pvt->id, samplesToCopy); 
+
+        bool hasMarkers = false;
+        std::deque<std::string>* pVecInventory = nullptr;
+        std::deque<std::string>* pVecInUse = nullptr;
+        std::deque<std::string>* pVecCleared = nullptr;
+        if (nullptr != tech_pvt->pVecMarksInUse) {
+          pVecInventory = static_cast<std::deque<std::string>*>(tech_pvt->pVecMarksInInventory);
+          pVecInUse = static_cast<std::deque<std::string>*>(tech_pvt->pVecMarksInUse);
+          pVecCleared = static_cast<std::deque<std::string>*>(tech_pvt->pVecMarksCleared);
+          hasMarkers = pVecInUse->size() + pVecCleared->size() >  0;
+        }
+
+        if (hasMarkers) {
+          /* discard markers and send notifications */
+          auto bufferIter = cBuffer->begin();
+          auto dataIter = data;
+          for (int i = 0; i < samplesToCopy; ++i) {
+            if (*bufferIter == AUDIO_MARKER) {
+              // Marker detected, discard it and send a notice unless it was previously cleared
+              auto * pVec = pVecCleared->size() > 0 ? pVecCleared : pVecInUse;
+              if (!pVec->empty()) {
+                auto name = pVec->front();
+                pVec->pop_front();
+
+                if (pVec == pVecInUse) {
+                  send_mark_event(tech_pvt, name.c_str());
+                  switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "(%u) dub_speech_frame - Marker %s detected in playout\n",
+                    tech_pvt->id, name.c_str());
+                }
+                else {
+                  switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "(%u) dub_speech_frame - Marker %s detected in playout but previously cleared\n",
+                    tech_pvt->id, name.c_str());
+                }
+              }
+            } else {
+              // Copy valid audio samplewhat 
+              *dataIter = *bufferIter;
+              ++dataIter;
+            }
+            ++bufferIter;
+          }
+
+          // Remove the processed samples (including discarded markers) from the buffer
+          cBuffer->erase(cBuffer->begin(), cBuffer->begin() + samplesToCopy);
+
+          // Adjust the number of samples copied to the output frame
+          int validSamplesCopied = std::distance(data, dataIter);
+
+          if (validSamplesCopied > 0) {
+            vector_add(fp, data, validSamplesCopied);
+          }
+        }
+        else {
           std::copy_n(cBuffer->begin(), samplesToCopy, data);
           cBuffer->erase(cBuffer->begin(), cBuffer->begin() + samplesToCopy);
 
           if (samplesToCopy > 0) {
             vector_add(fp, data, rframe->samples);
+          } else if (pVecInventory != nullptr && pVecInventory->size()) {
+            // no bidirectional audio to dub but still have some mark in inventory, send them now
+            auto name = pVecInventory->front();
+            pVecInventory->pop_front();
+
+            send_mark_event(tech_pvt, name.c_str());
+            switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "(%u) dub_speech_frame - Marker %s detected in inventory\n",
+              tech_pvt->id, name.c_str());
+          }
         }
         vector_normalize(fp, rframe->samples);
 
         switch_core_media_bug_set_write_replace_frame(bug, rframe);
+      }
       switch_mutex_unlock(tech_pvt->mutex);
     }
     return SWITCH_TRUE;
@@ -742,7 +1060,7 @@ extern "C" {
     }
     private_t* tech_pvt = (private_t*) switch_core_media_bug_get_user_data(bug);
 
-    CircularBuffer_t *cBuffer = (CircularBuffer_t *) tech_pvt->circularBuffer;
+    CircularBuffer_t *cBuffer = (CircularBuffer_t *) tech_pvt->streamingPlayoutBuffer;
 
     if (switch_mutex_trylock(tech_pvt->mutex) == SWITCH_STATUS_SUCCESS) {
       if (cBuffer != nullptr) {

mod_audio_fork/mod_audio_fork.h

@@ -52,11 +52,24 @@ struct private_data {
   int audio_paused:1;
   int graceful_shutdown:1;
   char initialMetadata[8192];
-  void *circularBuffer;
+
+  // for "mark" feature of bidirectional audio
+  void *pVecMarksInInventory;
+  void *pVecMarksInUse;
+  void *pVecMarksCleared;
+
+  // bidirectional audio
+  void *streamingPlayoutBuffer;
+  void *streamingPreBuffer;
+  int streamingPreBufSize;
+  uint8_t set_aside_byte;
+  int has_set_aside_byte;
+  int downscale_factor;
   SpeexResamplerState *bidirectional_audio_resampler;
   int bidirectional_audio_enable;
 	int bidirectional_audio_stream;
   int bidirectional_audio_sample_rate;
+  int clear_bidirectional_audio_buffer;
 };
 
 typedef struct private_data private_t;

mod_azure_transcribe/azure_transcribe_glue.cpp

@@ -32,6 +32,7 @@ static const char* proxyIP = std::getenv("JAMBONES_HTTP_PROXY_IP");
 static const char* proxyPort = std::getenv("JAMBONES_HTTP_PROXY_PORT");
 static const char* proxyUsername = std::getenv("JAMBONES_HTTP_PROXY_USERNAME");
 static const char* proxyPassword = std::getenv("JAMBONES_HTTP_PROXY_PASSWORD");
+static const bool use_single_connection = switch_true(std::getenv("AZURE_SPEECH_USE_SINGLE_CONNECTION"));
 
 class GStreamer {
 public:
@@ -51,15 +52,6 @@ public:
 
 		switch_core_session_t* psession = switch_core_session_locate(sessionId);
 		if (!psession) throw std::invalid_argument( "session id no longer active" );
-		//Due to use_single_connection, each GStreamer need to identify itself by configuration, if there is changes in configuration,
-		// the GStreamer will be closed and replaced by new object with new configuration.
-		m_configuration_stream << 
-			channels << ";" <<
-			lang << ";" <<
-			interim << ";" <<
-			samples_per_second << ";" <<
-			region << ";" <<
-			subscriptionKey << ";";
 		switch_channel_t *channel = switch_core_session_get_channel(psession);
  
 		switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "GStreamer::GStreamer(%p) region %s, language %s\n", 
@@ -68,9 +60,6 @@ public:
 
 		const char* endpoint = switch_channel_get_variable(channel, "AZURE_SERVICE_ENDPOINT");
 		const char* endpointId = switch_channel_get_variable(channel, "AZURE_SERVICE_ENDPOINT_ID");
-		m_configuration_stream <<
-			endpoint << ";" <<
-			endpointId << ";";
 
 		auto sourceLanguageConfig = SourceLanguageConfig::FromLanguage(lang);
 		auto format = AudioStreamFormat::GetWaveFormatPCM(8000, 16, channels);
@@ -81,7 +70,6 @@ public:
 				SpeechConfig::FromEndpoint(endpoint)) :
 			SpeechConfig::FromSubscription(subscriptionKey, region);
 		if (switch_true(switch_channel_get_variable(channel, "AZURE_USE_OUTPUT_FORMAT_DETAILED"))) {
-			m_configuration_stream << "output_format_detailed;";
 			speechConfig->SetOutputFormat(OutputFormat::Detailed);
 		}
 		if (nullptr != endpointId) {
@@ -93,18 +81,12 @@ public:
 			speechConfig->SetProperty(PropertyId::Speech_LogFilename, sdkLog);
 		}
 		if (switch_true(switch_channel_get_variable(channel, "AZURE_AUDIO_LOGGING"))) {
-			m_configuration_stream << "audio_logging;";
 			speechConfig->EnableAudioLogging();
 		}
 
     if (nullptr != proxyIP && nullptr != proxyPort) {
       switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(psession), SWITCH_LOG_DEBUG, "setting proxy: %s:%s\n", proxyIP, proxyPort);
       speechConfig->SetProxy(proxyIP, atoi(proxyPort), proxyUsername, proxyPassword);
-			m_configuration_stream <<
-				proxyIP << ";" <<
-				proxyPort << ";" <<
-				proxyUsername << ";" <<
-				proxyPassword << ";";
     }
 
 		m_pushStream = AudioInputStream::CreatePushStream(format);
@@ -113,7 +95,6 @@ public:
     // alternative language
 		const char* var;
     if (var = switch_channel_get_variable(channel, "AZURE_SPEECH_ALTERNATIVE_LANGUAGE_CODES")) {
-			m_configuration_stream << var << ";";
 			std::vector<std::string> languages;
 			char *alt_langs[3] = { 0 };
       int argc = switch_separate_string((char *) var, ',', alt_langs, 3);
@@ -138,22 +119,18 @@ public:
 		// profanity options: Allowed values are "masked", "removed", and "raw".
 		const char* profanity = switch_channel_get_variable(channel, "AZURE_PROFANITY_OPTION");
 		if (profanity) {
-			m_configuration_stream << profanity << ";";
 			properties.SetProperty(PropertyId::SpeechServiceResponse_ProfanityOption, profanity);
 		}
 		// report signal-to-noise ratio
 		if (switch_true(switch_channel_get_variable(channel, "AZURE_REQUEST_SNR"))) {
-			m_configuration_stream << "request_snr;";
 			properties.SetProperty(PropertyId::SpeechServiceResponse_RequestSnr, TrueString);
 		}
 		// initial speech timeout in milliseconds
 		const char* timeout = switch_channel_get_variable(channel, "AZURE_INITIAL_SPEECH_TIMEOUT_MS");
-		m_configuration_stream << timeout << ";";
 		if (timeout) properties.SetProperty(PropertyId::SpeechServiceConnection_InitialSilenceTimeoutMs, timeout);
 		else properties.SetProperty(PropertyId::SpeechServiceConnection_InitialSilenceTimeoutMs, DEFAULT_SPEECH_TIMEOUT);
 
     const char* segmentationInterval = switch_channel_get_variable(channel, "AZURE_SPEECH_SEGMENTATION_SILENCE_TIMEOUT_MS");
-		m_configuration_stream << segmentationInterval << ";";
     if (segmentationInterval) {
       switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(psession), SWITCH_LOG_DEBUG, "setting segmentation interval to %s ms\n", segmentationInterval);
       properties.SetProperty(PropertyId::Speech_SegmentationSilenceTimeoutMs, segmentationInterval);
@@ -161,12 +138,17 @@ public:
 
 		//https://learn.microsoft.com/en-us/azure/ai-services/speech-service/language-identification?tabs=once&pivots=programming-language-cpp#at-start-and-continuous-language-identification
 		const char* languageIdMode = switch_channel_get_variable(channel, "AZURE_LANGUAGE_ID_MODE");
-		m_configuration_stream << languageIdMode << ";";
 		if (languageIdMode) {
 			switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(psession), SWITCH_LOG_DEBUG, "setting SpeechServiceConnection_LanguageIdMode to %s \n", languageIdMode);
 			properties.SetProperty(PropertyId::SpeechServiceConnection_LanguageIdMode, languageIdMode);
 		}
-
+		//https://learn.microsoft.com/en-us/javascript/api/microsoft-cognitiveservices-speech-sdk/propertyid?view=azure-node-latest
+		//PropertyId::SpeechServiceResponse_PostProcessingOption
+ 		const char* postProcessingOption = switch_channel_get_variable(channel, "AZURE_POST_PROCESSING_OPTION");
+		if (postProcessingOption) {
+			switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(psession), SWITCH_LOG_DEBUG, "setting SpeechServiceResponse_PostProcessingOption to %s \n", postProcessingOption);
+			properties.SetProperty(PropertyId::SpeechServiceResponse_PostProcessingOption, postProcessingOption);
+		}
 		// recognition mode - readonly according to Azure docs: 
 		// https://docs.microsoft.com/en-us/javascript/api/microsoft-cognitiveservices-speech-sdk/propertyid?view=azure-node-latest
 		/*
@@ -178,7 +160,6 @@ public:
 
 		// hints
 		const char* hints = switch_channel_get_variable(channel, "AZURE_SPEECH_HINTS");
-		m_configuration_stream << hints << ";";
 		if (hints) {
 			auto grammar = PhraseListGrammar::FromRecognizer(m_recognizer);
 			char *phrases[500] = { 0 };
@@ -272,6 +253,9 @@ public:
 		m_recognizer->Recognized += onRecognitionEvent;
 		m_recognizer->Canceled += onCanceled;
 
+		// Store the final configuration string
+    m_configuration_string = createConfigurationStr(channels, lang, interim, samples_per_second, region, subscriptionKey, psession);
+
 		switch_core_session_rwunlock(psession);
 	}
 
@@ -280,7 +264,7 @@ public:
 	}
 
 	const char* configuration() {
-		return m_configuration_stream.str().c_str();
+		return m_configuration_string.c_str();
 	}
 
 	void connect() {
@@ -348,14 +332,33 @@ public:
     return m_connecting;
   }
 
+	bool hasConfigurationChanged(
+		u_int16_t channels,
+		char *lang, 
+		int interim,
+		uint32_t samples_per_second,
+		const char* region, 
+		const char* subscriptionKey) {
+		switch_core_session_t* psession = switch_core_session_locate(m_sessionId.c_str());
+		if (!psession) throw std::invalid_argument( "session id no longer active" );
+
+		std::string newConfiguration = createConfigurationStr(channels, lang, interim, samples_per_second, region, subscriptionKey, psession);
+
+		switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(psession), SWITCH_LOG_DEBUG,
+			"hasConfigurationChanged: old configurattion: %s, new configuration: %s\n", configuration(),  newConfiguration.c_str());
+
+		switch_core_session_rwunlock(psession);
+
+		return strcmp(newConfiguration.c_str(), configuration());
+	}
+
 private:
 	std::string m_sessionId;
 	std::string m_bugname;
 	std::string  m_region;
 	std::shared_ptr<SpeechRecognizer> m_recognizer;
 	std::shared_ptr<PushAudioInputStream> m_pushStream;
-	std::ostringstream m_configuration_stream;
+  std::string m_configuration_string;
-
 	responseHandler_t m_responseHandler;
 	bool m_interim;
 	bool m_finished;
@@ -363,6 +366,64 @@ private:
 	bool m_connecting;
 	bool m_stopped;
 	SimpleBuffer m_audioBuffer;
+
+	std::string createConfigurationStr(
+		u_int16_t channels,
+		char *lang, 
+		int interim,
+		uint32_t samples_per_second,
+		const char* region, 
+		const char* subscriptionKey,
+		switch_core_session_t* psession
+	) {
+		switch_channel_t *channel = switch_core_session_get_channel(psession);
+		std::ostringstream configuration_stream;
+		configuration_stream << 
+			channels << ";" <<
+			lang << ";" <<
+			interim << ";" <<
+			samples_per_second << ";" <<
+			region << ";" <<
+			subscriptionKey << ";";
+
+		const char* endpoint = switch_channel_get_variable(channel, "AZURE_SERVICE_ENDPOINT");
+		const char* endpointId = switch_channel_get_variable(channel, "AZURE_SERVICE_ENDPOINT_ID");
+		configuration_stream << (endpoint ? endpoint : "") << ";"
+			<< (endpointId ? endpointId : "") << ";";
+		if (switch_true(switch_channel_get_variable(channel, "AZURE_USE_OUTPUT_FORMAT_DETAILED"))) {
+			configuration_stream << "output_format_detailed;";
+		}
+		if (switch_true(switch_channel_get_variable(channel, "AZURE_AUDIO_LOGGING"))) {
+			configuration_stream << "audio_logging;";
+		}
+		configuration_stream << (proxyIP ? proxyIP : "") << ";"
+			<< (proxyPort ? proxyPort : "") << ";"
+			<< (proxyUsername ? proxyUsername : "") << ";"
+			<< (proxyPassword ? proxyPassword : "") << ";";
+		const char* var;
+		if (var = switch_channel_get_variable(channel, "AZURE_SPEECH_ALTERNATIVE_LANGUAGE_CODES")) {
+			configuration_stream << var << ";";
+		}
+		if (var = switch_channel_get_variable(channel, "AZURE_PROFANITY_OPTION")) {
+			configuration_stream << var << ";";
+		}
+		if (var = switch_channel_get_variable(channel, "AZURE_REQUEST_SNR")) {
+			configuration_stream << var << ";";
+		}
+		if (var = switch_channel_get_variable(channel, "AZURE_INITIAL_SPEECH_TIMEOUT_MS")) {
+			configuration_stream << var << ";";
+		}
+		if (var = switch_channel_get_variable(channel, "AZURE_SPEECH_SEGMENTATION_SILENCE_TIMEOUT_MS")) {
+			configuration_stream << var << ";";
+		}
+		if (var = switch_channel_get_variable(channel, "AZURE_LANGUAGE_ID_MODE")) {
+			configuration_stream << var << ";";
+		}
+		if (var = switch_channel_get_variable(channel, "AZURE_SPEECH_HINTS")) {
+			configuration_stream << var << ";";
+		}
+		return configuration_stream.str();
+	}
 };
 
 static void reaper(struct cap_cb *cb) {
@@ -420,16 +481,32 @@ extern "C" {
 		switch_status_t status = SWITCH_STATUS_SUCCESS;
 		switch_channel_t *channel = switch_core_session_get_channel(session);
 		switch_media_bug_t *bug = (switch_media_bug_t*) switch_channel_get_private(channel, bugname);
+		const char* subscriptionKey = switch_channel_get_variable(channel, "AZURE_SUBSCRIPTION_KEY");
+		const char* region = switch_channel_get_variable(channel, "AZURE_REGION");
+		const char* sessionId = switch_core_session_get_uuid(session);
+		auto read_codec = switch_core_session_get_read_codec(session);
+		uint32_t sampleRate = read_codec->implementation->actual_samples_per_second;
+		if (bug) {
+			struct cap_cb* existing_cb = (struct cap_cb*) switch_core_media_bug_get_user_data(bug);
+			GStreamer* existing_streamer = (GStreamer*) existing_cb->streamer;
+			existing_cb->is_keep_alive = 0;
+			if (!existing_streamer->hasConfigurationChanged(channels, lang, interim, sampleRate, region, subscriptionKey)) {
+				switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "Reuse active azure connection.\n");
+				return SWITCH_STATUS_SUCCESS;
+			}
+			switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "Azure configuration is changed, destroy old and create new azure connection\n");
+			reaper(existing_cb);
+			streamer =  new GStreamer(sessionId, bugname, channels, lang, interim, sampleRate, region, subscriptionKey, responseHandler);
+			if (!existing_cb->vad) streamer->connect();
+			existing_cb->streamer = streamer;
+			*ppUserData = existing_cb;
+			return SWITCH_STATUS_SUCCESS;
+		}
 		int err;
 		switch_threadattr_t *thd_attr = NULL;
 		switch_memory_pool_t *pool = switch_core_session_get_pool(session);
-		auto read_codec = switch_core_session_get_read_codec(session);
-		uint32_t sampleRate = read_codec->implementation->actual_samples_per_second;
-		const char* sessionId = switch_core_session_get_uuid(session);
 		struct cap_cb* cb = (struct cap_cb *) switch_core_session_alloc(session, sizeof(*cb));
 		memset(cb, sizeof(cb), 0);
-		const char* subscriptionKey = switch_channel_get_variable(channel, "AZURE_SUBSCRIPTION_KEY");
-		const char* region = switch_channel_get_variable(channel, "AZURE_REGION");
 		cb->channels = channels;
 		strncpy(cb->sessionId, sessionId, MAX_SESSION_ID);
 		strncpy(cb->bugname, bugname, MAX_BUG_LEN);
@@ -458,24 +535,7 @@ extern "C" {
 					switch_channel_get_name(channel), bugname);
 			streamer = new GStreamer(sessionId, bugname, channels, lang, interim, sampleRate, cb->region, subscriptionKey, responseHandler);
 			cb->streamer = streamer;
-
+			switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "azure_transcribe_session_init: config: %s\n", streamer->configuration());
-			if (bug) {
-				struct cap_cb* existing_cb = (struct cap_cb*) switch_core_media_bug_get_user_data(bug);
-				GStreamer* existing_streamer = (GStreamer*) existing_cb->streamer;
-				if (0 != strcmp(existing_streamer->configuration(), streamer->configuration())) {
-					switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "azure_transcribe_session_init: stop existing azure connection, old configuration %s, new configuration %s\n",
-					existing_streamer->configuration(), streamer->configuration());
-					if (existing_streamer) reaper(existing_cb);
-					killcb(existing_cb);
-					switch_mutex_destroy(existing_cb->mutex);
-				} else {
-					switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "azure_transcribe_session_init: enable existing azure connection\n");
-					killcb(cb);
-					cb = existing_cb;
-					status = SWITCH_STATUS_SUCCESS;
-					goto done; 
-				}
-			}
 		} catch (std::exception& e) {
 			switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_ERROR, "%s: Error initializing gstreamer: %s.\n", 
 				switch_channel_get_name(channel), e.what());
@@ -539,7 +599,6 @@ extern "C" {
 	switch_status_t azure_transcribe_session_stop(switch_core_session_t *session, int channelIsClosing, char* bugname) {
 		switch_channel_t *channel = switch_core_session_get_channel(session);
 		switch_media_bug_t *bug = (switch_media_bug_t*) switch_channel_get_private(channel, bugname);
-		const bool use_single_connection = switch_true(std::getenv("AZURE_SPEECH_USE_SINGLE_CONNECTION"));
 
 		if (bug) {
 			struct cap_cb *cb = (struct cap_cb *) switch_core_media_bug_get_user_data(bug);
@@ -581,7 +640,6 @@ extern "C" {
 		frame.data = data;
 		frame.buflen = SWITCH_RECOMMENDED_BUFFER_SIZE;
 		if (cb->is_keep_alive) {
-
 			// remove media bug buffered data
  			while (true) {
 				unsigned char data[SWITCH_RECOMMENDED_BUFFER_SIZE] = {0};
@@ -593,7 +651,6 @@ extern "C" {
 			}
 			return SWITCH_TRUE;
 		}
-
 		if (switch_mutex_trylock(cb->mutex) == SWITCH_STATUS_SUCCESS) {
 			GStreamer* streamer = (GStreamer *) cb->streamer;
 			if (streamer) {

mod_azure_tts/azure_glue.cpp

@@ -15,11 +15,13 @@ typedef boost::circular_buffer<uint16_t> CircularBuffer_t;
 
 using namespace Microsoft::CognitiveServices::Speech;
 
+static const char* audioLogFile= std::getenv("AZURE_AUDIO_LOGGING");
+
 static std::string fullDirPath;
 
 static void start_synthesis(std::shared_ptr<SpeechSynthesizer> speechSynthesizer, const char* text, azure_t* a) {
     try {
-      switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "start_synthesis calling \n");
+      switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "start_synthesis calling, text %s\n", text);
       auto result = std::strncmp(text, "<speak", 6) == 0 ?
         speechSynthesizer->SpeakSsmlAsync(text).get() :
         speechSynthesizer->SpeakTextAsync(text).get();
@@ -43,6 +45,8 @@ static void start_synthesis(std::shared_ptr<SpeechSynthesizer> speechSynthesizer
         switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "mod_azure_tts: Exception in start_synthesis %s\n",  e.what());
     }
     a->draining = 1;
+
+    free((void*) text);
 }
 
 extern "C" {
@@ -87,14 +91,7 @@ extern "C" {
 
   switch_status_t azure_speech_feed_tts(azure_t* a, char* text, switch_speech_flag_t *flags) {
     const int MAX_CHARS = 20;
-    char tempText[MAX_CHARS + 4]; // +4 for the ellipsis and null terminator
+    switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "azure_speech_feed_tts %s\n", text);
-
-    if (strlen(text) > MAX_CHARS) {
-        strncpy(tempText, text, MAX_CHARS);
-        strcpy(tempText + MAX_CHARS, "...");
-    } else {
-        strcpy(tempText, text);
-    }
 
     /* open cache file */
     if (a->cache_audio && fullDirPath.length() > 0) {
@@ -168,8 +165,14 @@ extern "C" {
 			speechConfig->SetEndpointId(a->endpointId);
 		}
 
+    if (audioLogFile) {
+      switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "azure_speech_feed_tts enabling audio logging to %s\n", audioLogFile);
+      speechConfig->SetProperty(PropertyId::Speech_LogFilename, audioLogFile);
+      speechConfig->EnableAudioLogging();
+    }
+
     try {
-      auto speechSynthesizer = SpeechSynthesizer::FromConfig(speechConfig);
+      auto speechSynthesizer = SpeechSynthesizer::FromConfig(speechConfig, nullptr);
 
       speechSynthesizer->SynthesisStarted += [a](const SpeechSynthesisEventArgs& e) {
           switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "azure_speech_feed_tts SynthesisStarted\n");
@@ -266,7 +269,8 @@ extern "C" {
         }
       };
 
-      std::thread(start_synthesis, speechSynthesizer, text, a).detach();
+      const char* dupText = strdup(text); // text will be freed in the thread
+      std::thread(start_synthesis, speechSynthesizer, dupText, a).detach();
       switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "azure_speech_feed_tts sent synthesize request\n");
     } catch (const std::exception& e) {
       switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "mod_azure_tts: Exception: %s\n", e.what());

mod_deepgram_transcribe/dg_transcribe_glue.cpp

@@ -172,7 +172,7 @@ namespace {
     else if (customModel) oss << "&model=" << customModel;
 
     if (var = switch_channel_get_variable(channel, "DEEPGRAM_SPEECH_MODEL_VERSION")) {
-     oss <<  "&version";
+     oss <<  "&version=";
      oss <<  var;
     }
     oss <<  "&language=";

mod_deepgram_tts/deepgram_glue.cpp

@@ -742,7 +742,7 @@ extern "C" {
     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_stream << (d->endpoint != nullptr ? d->endpoint : "https://api.deepgram.com") << "/v1/speak?model=" << d->voice_name << "&encoding=linear16&sample_rate=8000";
     url = url_stream.str();
 
     /* create the JSON body */

mod_deepgram_tts/mod_deepgram_tts.c

@@ -19,10 +19,12 @@ static void cleardeepgram(deepgram_t* d, int freeAll) {
   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);
+  if (d->endpoint) free(d->endpoint);
   
 
   d->api_key = NULL;
   d->request_id = NULL;
+  d->endpoint = NULL;
 
   d->reported_model_name = NULL;
   d->reported_model_uuid = NULL;
@@ -121,6 +123,9 @@ static void d_text_param_tts(switch_speech_handle_t *sh, char *param, const char
   if (0 == strcmp(param, "api_key")) {
     if (d->api_key) free(d->api_key);
     d->api_key = strdup(val);
+  } else if (0 == strcmp(param, "endpoint")) {
+    if (d->endpoint) free(d->endpoint);
+    d->endpoint = strdup(val);
   } else if (0 == strcmp(param, "voice")) {
     if (d->voice_name) free(d->voice_name);
     d->voice_name = strdup(val);

mod_deepgram_tts/mod_deepgram_tts.h

@@ -7,6 +7,7 @@
 typedef struct deepgram_data {
   char *voice_name;
   char *api_key;
+  char *endpoint;
 
   /* result data */
   long response_code;

mod_dub/mod_dub.c

@@ -160,13 +160,15 @@ static switch_status_t dub_silence_track(switch_core_session_t *session, char* t
   switch_media_bug_t *bug = (switch_media_bug_t*) switch_channel_get_private(channel, MY_BUG_NAME);
   switch_status_t status = SWITCH_STATUS_FALSE;
 
+  // DH: I found it is much simpler to implement silence as a sequence of remove and add operations  
   if (bug) {
     struct cap_cb *cb =(struct cap_cb *) switch_core_media_bug_get_user_data(bug);
-    status = silence_dub_track(cb, trackName);
+    status = remove_dub_track(cb, trackName);
     if (status != SWITCH_STATUS_SUCCESS) {
-      switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "dub_silence_track: error silencing track %s\n", trackName);
+      switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "dub_silence_track: error finding track %s\n", trackName);
       return SWITCH_STATUS_FALSE;
     }
+    status = dub_add_track(session, trackName);
   }
   else {
     switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "dub_silence_track: bug not found\n");
@@ -269,7 +271,7 @@ SWITCH_STANDARD_API(dub_function)
         }
         else {
           char* text = argv[3];
-          int gain = argc > 4 ? atoi(argv[4]) : 0;
+          int gain = argc > 5 ? atoi(argv[5]) : 0;
           switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_INFO, "sayOnTrack %s gain %d\n", text, gain);
           status = dub_say_on_track(session, track, text, gain);
         }

mod_dub/tts_vendor_parser.cpp

@@ -378,7 +378,10 @@ switch_status_t playht_parse_text(const std::map<std::string, std::string>& para
     cJSON_AddStringToObject(jResult, "quality", quality.c_str());
   }
   if (!speed.empty()) {
-    cJSON_AddNumberToObject(jResult, "speed", atoi(speed.c_str()));
+      double val = strtod(speed.c_str(), NULL);
+      if (val != 0.0) {
+        cJSON_AddNumberToObject(jResult, "speed", val);
+      }
   }
   if (!seed.empty()) {
     cJSON_AddNumberToObject(jResult, "seed", atoi(seed.c_str()));

mod_elevenlabs_tts/elevenlabs_glue.cpp

@@ -809,6 +809,12 @@ extern "C" {
     cJSON * jResult = cJSON_CreateObject();
     cJSON_AddStringToObject(jResult, "model_id", el->model_id);
     cJSON_AddStringToObject(jResult, "text", text);
+    if (el->previous_text) {
+      cJSON_AddStringToObject(jResult, "previous_text", el->previous_text);
+    }
+    if (el->next_text) {
+      cJSON_AddStringToObject(jResult, "next_text", el->next_text);
+    }
     if (el->similarity_boost || el->style || el->use_speaker_boost || el->stability) {
       cJSON * jVoiceSettings = cJSON_CreateObject();
       cJSON_AddItemToObject(jResult, "voice_settings", jVoiceSettings);

mod_elevenlabs_tts/mod_elevenlabs_tts.c

@@ -19,6 +19,8 @@ static void clearElevenlabs(elevenlabs_t* el, int freeAll) {
   if (el->style) free(el->style);
   if (el->use_speaker_boost) free(el->use_speaker_boost);
   if (el->optimize_streaming_latency) free(el->optimize_streaming_latency);
+  if (el->previous_text) free(el->previous_text);
+  if (el->next_text) free(el->next_text);
   if (el->ct) free(el->ct);
   if (el->reported_latency) free(el->reported_latency);
   if (el->request_id) free(el->request_id);
@@ -36,6 +38,8 @@ static void clearElevenlabs(elevenlabs_t* el, int freeAll) {
   el->style = NULL;
   el->use_speaker_boost = NULL;
   el->optimize_streaming_latency = NULL;
+  el->previous_text = NULL;
+  el->next_text = NULL;
   el->ct = NULL;
   el->reported_latency = NULL;
   el->request_id = NULL;
@@ -174,6 +178,14 @@ static void ell_text_param_tts(switch_speech_handle_t *sh, char *param, const ch
     if (el->optimize_streaming_latency) free(el->optimize_streaming_latency);
     el->optimize_streaming_latency = strdup(val);
   }
+  else if (0 == strcmp(param, "next_text")) {
+    if (el->next_text) free(el->next_text);
+    el->next_text = strdup(val);
+  }
+  else if (0 == strcmp(param, "previous_text")) {
+    if (el->previous_text) free(el->previous_text);
+    el->previous_text = strdup(val);
+  }
   else if (0 == strcmp(param, "write_cache_file") && switch_true(val)) {
     el->cache_audio = 1;
   }

mod_elevenlabs_tts/mod_elevenlabs_tts.h

@@ -17,6 +17,8 @@ struct elevenlabs_data {
   char* style;
   char* use_speaker_boost;
   char* optimize_streaming_latency;
+  char* previous_text;
+  char* next_text;
 
   /* result data */
   long response_code;

mod_jambonz_transcribe/audio_pipe.cpp

@@ -229,16 +229,6 @@ int AudioPipe::lws_callback(struct lws *wsi,
 }
 
 
-// static members
-static const lws_retry_bo_t retry = {
-    nullptr,   // retry_ms_table
-    0,         // retry_ms_table_count
-    0,         // conceal_count
-    UINT16_MAX,         // secs_since_valid_ping
-    UINT16_MAX,        // secs_since_valid_hangup
-    0          // jitter_percent
-};
-
 struct lws_context *AudioPipe::context = nullptr;
 std::thread AudioPipe::serviceThread;
 std::mutex AudioPipe::mutex_connects;
@@ -376,6 +366,28 @@ bool AudioPipe::lws_service_thread() {
     { NULL, NULL, 0, 0 }
   };
 
+  uint16_t secs_sinceq_valid_ping = UINT16_MAX;
+  uint16_t secs_since_valid_hangup = UINT16_MAX;
+
+  char* wsVar = std::getenv("WS_PING_INTERVAL");
+  if (wsVar != nullptr) {
+    secs_sinceq_valid_ping = std::atoi(wsVar);
+  }
+
+  wsVar = std::getenv("WS_NO_PONG_HANGUP_INTERVAL");
+  if (wsVar != nullptr) {
+    secs_since_valid_hangup = std::atoi(wsVar);
+  }
+
+  const lws_retry_bo_t retry = {
+    nullptr,                  // retry_ms_table
+    0,                        // retry_ms_table_count
+    0,                        // conceal_count
+    secs_sinceq_valid_ping,   // secs_sinceq_valid_ping
+    secs_since_valid_hangup,  // secs_since_valid_hangup
+    0                         // jitter_percent
+  };
+
   memset(&info, 0, sizeof info); 
   info.port = CONTEXT_PORT_NO_LISTEN; 
   info.options = LWS_SERVER_OPTION_DO_SSL_GLOBAL_INIT;

mod_jambonz_transcribe/jb_transcribe_glue.cpp

@@ -145,7 +145,7 @@ namespace {
     cJSON_AddStringToObject(json, "format", "raw");
     cJSON_AddStringToObject(json, "encoding", "LINEAR16");
     cJSON_AddBoolToObject(json, "interimResults", tech_pvt->interim);
-    cJSON_AddNumberToObject(json, "sampleRateHz", 8000);
+    cJSON_AddNumberToObject(json, "sampleRateHz", tech_pvt->sampling);
     if (var = switch_channel_get_variable(channel, "JAMBONZ_STT_OPTIONS")) {
       cJSON* jOptions = cJSON_Parse(var);
       if (jOptions)  {
@@ -353,7 +353,7 @@ extern "C" {
   }
 	
   switch_status_t jb_transcribe_session_init(switch_core_session_t *session, 
-    responseHandler_t responseHandler, uint32_t samples_per_second, uint32_t channels, 
+    responseHandler_t responseHandler, uint32_t samples_per_second, int desiredSampling, uint32_t channels, 
     char* lang, int interim, char* bugname, void **ppUserData)
   {    	
     int err;
@@ -365,7 +365,7 @@ extern "C" {
       return SWITCH_STATUS_FALSE;
     }
 
-    if (SWITCH_STATUS_SUCCESS != fork_data_init(tech_pvt, session, samples_per_second, 8000, channels, lang, interim, bugname, responseHandler)) {
+    if (SWITCH_STATUS_SUCCESS != fork_data_init(tech_pvt, session, samples_per_second, desiredSampling, channels, lang, interim, bugname, responseHandler)) {
       destroy_tech_pvt(tech_pvt);
       return SWITCH_STATUS_FALSE;
     }

mod_jambonz_transcribe/jb_transcribe_glue.h

@@ -6,7 +6,7 @@ int parse_ws_uri(switch_channel_t *channel, const char* szServerUri, char* host,
 switch_status_t jb_transcribe_init();
 switch_status_t jb_transcribe_cleanup();
 switch_status_t jb_transcribe_session_init(switch_core_session_t *session, responseHandler_t responseHandler, 
-		uint32_t samples_per_second, uint32_t channels, char* lang, int interim, char* bugname, void **ppUserData);
+		uint32_t samples_per_second, int desiredSampling, uint32_t channels, char* lang, int interim, char* bugname, void **ppUserData);
 switch_status_t jb_transcribe_session_stop(switch_core_session_t *session, int channelIsClosing, char* bugname);
 switch_bool_t jb_transcribe_frame(switch_core_session_t *session, switch_media_bug_t *bug);
 

mod_jambonz_transcribe/mod_jambonz_transcribe.c

@@ -73,6 +73,8 @@ static switch_status_t start_capture(switch_core_session_t *session, switch_medi
 	switch_codec_implementation_t read_impl = { 0 };
 	void *pUserData;
 	uint32_t samples_per_second;
+	uint32_t desiredSampling = 8000;
+	const char* var;
 
   if (!switch_channel_get_variable(channel, "JAMBONZ_STT_URL")) {
     switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Missing JAMBONZ_STT_URL channel var\n");
@@ -91,8 +93,11 @@ static switch_status_t start_capture(switch_core_session_t *session, switch_medi
 	}
 
 	samples_per_second = !strcasecmp(read_impl.iananame, "g722") ? read_impl.actual_samples_per_second : read_impl.samples_per_second;
-
+	var = switch_channel_get_variable(channel, "JAMBONZ_STT_SAMPLING");
-	if (SWITCH_STATUS_FALSE == jb_transcribe_session_init(session, responseHandler, samples_per_second, flags & SMBF_STEREO ? 2 : 1, lang, interim, bugname, &pUserData)) {
+	if (var != NULL) {
+		desiredSampling = atoi(var);
+	}
+	if (SWITCH_STATUS_FALSE == jb_transcribe_session_init(session, responseHandler, samples_per_second, desiredSampling, flags & SMBF_STEREO ? 2 : 1, lang, interim, bugname, &pUserData)) {
 		switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Error initializing jb speech session.\n");
 		return SWITCH_STATUS_FALSE;
 	}

mod_playht_tts/playht_glue.cpp

@@ -774,7 +774,10 @@ extern "C" {
       cJSON_AddStringToObject(jResult, "quality", p->quality);
     }
     if (p->speed) {
-      cJSON_AddNumberToObject(jResult, "speed", atoi(p->speed));
+      double val = strtod(p->speed, NULL);
+      if (val != 0.0) {
+        cJSON_AddNumberToObject(jResult, "speed", val);
+      }
     }
     if (p->seed) {
       cJSON_AddNumberToObject(jResult, "seed", atoi(p->seed));