wip

Signed-off-by: Dave Horton <daveh@beachdognet.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,65 +141,68 @@ 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) {
-          lwsl_notice("AudioPipe::lws_service_thread race condition: got incoming message while closing the connection.\n");
+          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 {
-            lwsl_err("AudioPipe::lws_service_thread LWS_CALLBACK_CLIENT_RECEIVE received binary frame, discarding.\n");
-          }
-          return 0;
-        }
-
-        if (lws_is_first_fragment(wsi)) {
-          // allocate a buffer for the entire chunk of memory needed
-          assert(nullptr == ap->m_recv_buf);
-          ap->m_recv_buf_len = len + lws_remaining_packet_payload(wsi);
-          ap->m_recv_buf = (uint8_t*) malloc(ap->m_recv_buf_len);
-          ap->m_recv_buf_ptr = ap->m_recv_buf;
-        }
-
-        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");
-          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");
+          } else if (len > 0) {
+            switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR,"AudioPipe::lws_service_thread LWS_CALLBACK_CLIENT_RECEIVE received unexpected binary frame, discarding.\n");
           }
           else {
-            ap->m_recv_buf = (uint8_t*) realloc(ap->m_recv_buf, newlen);
-            if (nullptr != ap->m_recv_buf) {
-              ap->m_recv_buf_len = newlen;
-              ap->m_recv_buf_ptr = ap->m_recv_buf + write_offset;
-            }
+            switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG,"AudioPipe::lws_service_thread LWS_CALLBACK_CLIENT_RECEIVE received zero length binary frame, discarding.\n");
           }
         }
-
-        if (nullptr != ap->m_recv_buf) {
-          if (len > 0) {
-            memcpy(ap->m_recv_buf_ptr, in, len);
-            ap->m_recv_buf_ptr += len;
+        else {
+          if (lws_is_first_fragment(wsi)) {
+            // allocate a buffer for the entire chunk of memory needed
+            assert(nullptr == ap->m_recv_buf);
+            ap->m_recv_buf_len = len + lws_remaining_packet_payload(wsi);
+            ap->m_recv_buf = (uint8_t*) malloc(ap->m_recv_buf_len);
+            ap->m_recv_buf_ptr = ap->m_recv_buf;
           }
-          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(), NULL, len);
-              if (nullptr != ap->m_recv_buf) free(ap->m_recv_buf);
+
+          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) {
+            //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;
+              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);
+              if (nullptr != ap->m_recv_buf) {
+                ap->m_recv_buf_len = newlen;
+                ap->m_recv_buf_ptr = ap->m_recv_buf + write_offset;
+              }
+            }
+          }
+
+          if (nullptr != ap->m_recv_buf) {
+            if (len > 0) {
+              memcpy(ap->m_recv_buf_ptr, in, len);
+              ap->m_recv_buf_ptr += len;
+            }
+            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(), NULL, len);
+                if (nullptr != ap->m_recv_buf) free(ap->m_recv_buf);
+              }
+              ap->m_recv_buf = ap->m_recv_buf_ptr = nullptr;
+              ap->m_recv_buf_len = 0;
             }
-            ap->m_recv_buf = ap->m_recv_buf_ptr = nullptr;
-            ap->m_recv_buf_len = 0;
           }
         }
       }
@@ -205,13 +212,13 @@ int AudioPipe::lws_callback(struct lws *wsi,
       {
         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;
@@ -250,7 +257,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;
@@ -379,7 +386,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);
@@ -389,7 +396,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);
@@ -428,11 +435,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;
   }
 
@@ -449,16 +456,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()) {
@@ -513,7 +520,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;
 }

mod_audio_fork/lws_glue.cpp

@@ -21,11 +21,12 @@
 
 #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)
 
 namespace {
   static const char *requestedBufferSecs = std::getenv("MOD_AUDIO_FORK_BUFFER_SECS");
@@ -38,29 +39,92 @@ namespace {
   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));
+    std::vector<uint8_t> data;
+
+    // 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);
+
+    // Ensure the prebuffer has enough capacity
+    if (cBuffer->capacity() - cBuffer->size() < numSamples) {
+        size_t newCapacity = cBuffer->size() + std::max(numSamples, (size_t)BUFFER_GROW_SIZE);
+        cBuffer->set_capacity(newCapacity);
+    }
+
+    // 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 = pcm_data.size();
+        spx_uint32_t in_len = in.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()) {
-          switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_CRIT, "Resampler output exceeded maximum buffer size!\n");
-          return SWITCH_STATUS_FALSE;
-        }
+        // Resize the output buffer to match the output length from resampler
+        //switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "Resizing output buffer from %u to %u samples\n", in.size(), out_len);
 
-        // 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));
+        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());
@@ -70,40 +134,51 @@ namespace {
       return SWITCH_STATUS_FALSE;
     }
 
-    switch_mutex_lock(tech_pvt->mutex);
+    if (nullptr != tech_pvt->mutex && switch_mutex_trylock(tech_pvt->mutex) == SWITCH_STATUS_SUCCESS) {
+      CircularBuffer_t *playoutBuffer = (CircularBuffer_t *) tech_pvt->streamingPlayoutBuffer;
 
-    try {
-      // 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));
+      try {
+        // Resize the buffer if necessary
+        if (playoutBuffer->capacity() - playoutBuffer->size() < out.size()) {
+          size_t newCapacity = playoutBuffer->size() + std::max(out.size(), (size_t)BUFFER_GROW_SIZE);
+          playoutBuffer->set_capacity(newCapacity);
+          //switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "Resized playout buffer to new capacity: %zu\n", newCapacity);
+        }
+        // Push the data into the buffer.
+        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;
       }
-      // Push the data into the buffer.
-      cBuffer->insert(cBuffer->end(), pcm_data.begin(), pcm_data.end());
-    } catch (const std::exception& e) {
-      switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Error processing incoming binary message: %s\n", e.what());
       switch_mutex_unlock(tech_pvt->mutex);
-      return SWITCH_STATUS_FALSE;
-    } catch (...) {
-      switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Error processing incoming binary message\n");
-      switch_mutex_unlock(tech_pvt->mutex);
-      return SWITCH_STATUS_FALSE;
-    }
+      cBuffer->clear();
 
-    switch_mutex_unlock(tech_pvt->mutex);
+      // 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.
@@ -190,6 +265,9 @@ 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("transcription")) {
         char* jsonString = cJSON_PrintUnformatted(jsonData);
@@ -309,10 +387,20 @@ 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);
+
     strncpy(tech_pvt->bugname, bugname, MAX_BUG_LEN);
     if (metadata) strncpy(tech_pvt->initialMetadata, metadata, MAX_METADATA_LEN);
     
@@ -342,8 +430,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);
-      tech_pvt->bidirectional_audio_resampler = speex_resampler_init(channels, bidirectional_audio_sample_rate, sampling, SWITCH_RESAMPLE_QUALITY, &err);
+      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(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;
@@ -369,10 +457,15 @@ namespace {
       switch_mutex_destroy(tech_pvt->mutex);
       tech_pvt->mutex = nullptr;
     }
-    if (tech_pvt->circularBuffer) {
-      CircularBuffer_t *cBuffer = (CircularBuffer_t *) tech_pvt->circularBuffer;
+    if (tech_pvt->streamingPlayoutBuffer) {
+      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;
     }
   }
 
@@ -472,8 +565,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 ;
-     //LLL_INFO | LLL_PARSER | LLL_HEADER | LLL_EXT | LLL_CLIENT  | LLL_LATENCY | LLL_DEBUG ;
+    //int logs = LLL_ERR | LLL_WARN | LLL_NOTICE | 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;
@@ -725,28 +818,39 @@ 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) {
-      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);
+      // 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;
       }
-      vector_normalize(fp, rframe->samples);
+      else {
+        switch_frame_t* rframe = switch_core_media_bug_get_write_replace_frame(bug);
+        int16_t *fp = reinterpret_cast<int16_t*>(rframe->data);
 
-      switch_core_media_bug_set_write_replace_frame(bug, rframe);
+        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));
+
+        //switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "(%u) dub_speech_frame - samples to copy %u\n", tech_pvt->id, samplesToCopy); 
+
+        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;
@@ -761,7 +865,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,19 @@ struct private_data {
   int audio_paused:1;
   int graceful_shutdown:1;
   char initialMetadata[8192];
-  void *circularBuffer;
+
+  // 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_aws_transcribe/aws_transcribe_glue.cpp

@@ -43,16 +43,16 @@ public:
     const char *sessionId,
 		const char *bugname,
 		u_int16_t channels,
-    char *lang, 
+    char *lang,
     int interim,
 		uint32_t samples_per_second,
-		const char* region, 
-		const char* awsAccessKeyId, 
+		const char* region,
+		const char* awsAccessKeyId,
 		const char* awsSecretAccessKey,
 		const char* awsSessionToken,
 		responseHandler_t responseHandler
   ) : m_sessionId(sessionId), m_bugname(bugname), m_finished(false), m_interim(interim), m_finishing(false), m_connected(false), m_connecting(false),
-	 		m_packets(0), m_responseHandler(responseHandler), m_pStream(nullptr), 
+	 		m_packets(0), m_responseHandler(responseHandler), m_pStream(nullptr),
 			m_audioBuffer(320 * (samples_per_second == 8000 ? 1 : 2), 15) {
 		Aws::Client::ClientConfiguration config;
 		if (region != nullptr && strlen(region) > 0) config.region = region;
@@ -71,7 +71,7 @@ public:
 		else {
 			m_client = Aws::MakeUnique<TranscribeStreamingServiceClient>(ALLOC_TAG, config);
 		}
-	
+
     m_handler.SetTranscriptEventCallback([this](const TranscriptEvent& ev)
     {
 			switch_core_session_t* psession = switch_core_session_locate(m_sessionId.c_str());
@@ -132,7 +132,7 @@ public:
 
 				// send any buffered audio
 				int nFrames = m_audioBuffer.getNumItems();
-				switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "GStreamer %p got stream ready, %d buffered frames\n", this, nFrames);	
+				switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "GStreamer %p got stream ready, %d buffered frames\n", this, nFrames);
 				if (nFrames) {
 					char *p;
 					do {
@@ -142,19 +142,19 @@ public:
 						}
 					} while (p);
 				}
-	
+
 				switch_core_session_rwunlock(psession);
 			}
     };
-    auto OnResponseCallback = [this](const TranscribeStreamingServiceClient* pClient, 
-			const Model::StartStreamTranscriptionRequest& request, 
-			const Model::StartStreamTranscriptionOutcome& outcome, 
+    auto OnResponseCallback = [this](const TranscribeStreamingServiceClient* pClient,
+			const Model::StartStreamTranscriptionRequest& request,
+			const Model::StartStreamTranscriptionOutcome& outcome,
 			const std::shared_ptr<const Aws::Client::AsyncCallerContext>& context)
     {
  			switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "GStreamer %p stream got final response\n", this);
 			switch_core_session_t* psession = switch_core_session_locate(m_sessionId.c_str());
 			if (psession) {
-				if (!outcome.IsSuccess()) {				
+				if (!outcome.IsSuccess()) {
 					const TranscribeStreamingServiceError& err = outcome.GetError();
 					auto message = err.GetMessage();
 					auto exception = err.GetExceptionName();
@@ -186,7 +186,7 @@ public:
 
 
 	~GStreamer() {
-		switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "GStreamer::~GStreamer wrote %u packets %p\n", m_packets, this);		
+		switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "GStreamer::~GStreamer wrote %u packets %p\n", m_packets, this);
 	}
 
 	bool write(void* data, uint32_t datalen) {
@@ -228,7 +228,7 @@ public:
 		bool shutdownInitiated = false;
 		while (true) {
 			std::unique_lock<std::mutex> lk(m_mutex);
-			m_cond.wait(lk, [&, this] { 
+			m_cond.wait(lk, [&, this] {
 				return (!m_deqAudio.empty() && !m_finishing)  || m_transcript.TranscriptHasBeenSet() || m_finished  || (m_finishing && !shutdownInitiated);
 			});
 
@@ -264,7 +264,7 @@ public:
 						m_responseHandler(psession, s.str().c_str(), m_bugname.c_str());
 					}
 					TranscriptEvent empty;
-					m_transcript = empty; 
+					m_transcript = empty;
 
 					switch_core_session_rwunlock(psession);
 				}
@@ -362,7 +362,7 @@ extern "C" {
 		const char* secretAccessKey = std::getenv("AWS_SECRET_ACCESS_KEY");
 		const char* region = std::getenv("AWS_REGION");
 		if (NULL == accessKeyId && NULL == secretAccessKey) {
-			switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_NOTICE, 
+			switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_NOTICE,
 				"\"AWS_ACCESS_KEY_ID\"  and/or \"AWS_SECRET_ACCESS_KEY\" env var not set; authentication will expect channel variables of same names to be set\n");
 		}
 		else {
@@ -370,7 +370,7 @@ extern "C" {
 
 		}
     Aws::SDKOptions options;
-/*		
+/*
     options.loggingOptions.logLevel = Aws::Utils::Logging::LogLevel::Trace;
 
 		Aws::Utils::Logging::InitializeAWSLogging(
@@ -381,7 +381,7 @@ extern "C" {
 
 		return SWITCH_STATUS_SUCCESS;
 	}
-	
+
 	switch_status_t aws_transcribe_cleanup() {
 		Aws::SDKOptions options;
 		/*
@@ -394,7 +394,7 @@ extern "C" {
 	}
 
 	// start transcribe on a channel
-	switch_status_t aws_transcribe_session_init(switch_core_session_t *session, responseHandler_t responseHandler, 
+	switch_status_t aws_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
 	) {
 		switch_status_t status = SWITCH_STATUS_SUCCESS;
@@ -433,7 +433,7 @@ extern "C" {
 			std::getenv("AWS_REGION")) {
 			switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "Using env vars for aws authentication\n");
 			strncpy(cb->awsAccessKeyId, std::getenv("AWS_ACCESS_KEY_ID"), 128);
-			strncpy(cb->awsSecretAccessKey, std::getenv("AWS_SECRET_ACCESS_KEY"), 128);		
+			strncpy(cb->awsSecretAccessKey, std::getenv("AWS_SECRET_ACCESS_KEY"), 128);
 			strncpy(cb->region, std::getenv("AWS_REGION"), MAX_REGION);
 		}
 		else {
@@ -445,7 +445,7 @@ extern "C" {
 		if (switch_mutex_init(&cb->mutex, SWITCH_MUTEX_NESTED, pool) != SWITCH_STATUS_SUCCESS) {
 			switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_ERROR, "Error initializing mutex\n");
 			status = SWITCH_STATUS_FALSE;
-			goto done; 
+			goto done;
 		}
 
 		cb->interim = interim;
@@ -455,7 +455,7 @@ extern "C" {
 		if (sampleRate != 8000) {
 			cb->resampler = speex_resampler_init(1, sampleRate, 16000, SWITCH_RESAMPLE_QUALITY, &err);
 			if (0 != err) {
-				switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_ERROR, "%s: Error initializing resampler: %s.\n", 
+				switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_ERROR, "%s: Error initializing resampler: %s.\n",
 							switch_channel_get_name(channel), speex_resampler_strerror(err));
 				status = SWITCH_STATUS_FALSE;
 				goto done;
@@ -488,7 +488,7 @@ extern "C" {
 				switch_vad_set_param(cb->vad, "silence_ms", silence_ms);
 				switch_vad_set_param(cb->vad, "voice_ms", voice_ms);
 				switch_vad_set_param(cb->vad, "debug", debug);
-				switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "%s: delaying connection until vad, voice_ms %d, mode %d\n", 
+				switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "%s: delaying connection until vad, voice_ms %d, mode %d\n",
 					switch_channel_get_name(channel), voice_ms, mode);
 			}
 		}
@@ -499,7 +499,7 @@ extern "C" {
 		switch_thread_create(&cb->thread, thd_attr, aws_transcribe_thread, cb, pool);
 
 		*ppUserData = cb;
-	
+
 	done:
 		return status;
 	}
@@ -521,7 +521,7 @@ extern "C" {
       do {
         streamer = (GStreamer *) cb->streamer;
         if (streamer) break;
-        switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, 
+        switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG,
           "aws_transcribe_session_stop: waiting for streamer to come online..%s\n", bugname);
         switch_yield(10000); // wait 10ms
       } while (i++ < 3);
@@ -557,7 +557,7 @@ extern "C" {
 		switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_INFO, "%s Bug is not attached.\n", switch_channel_get_name(channel));
 		return SWITCH_STATUS_FALSE;
 	}
-	
+
 	switch_bool_t aws_transcribe_frame(switch_media_bug_t *bug, void* user_data) {
 		switch_core_session_t *session = switch_core_media_bug_get_session(bug);
 		uint8_t data[SWITCH_RECOMMENDED_BUFFER_SIZE];
@@ -587,7 +587,7 @@ extern "C" {
 						}
 
 						if (cb->resampler) {
-							speex_resampler_process_interleaved_int(cb->resampler, (const spx_int16_t *) frame.data, (spx_uint32_t *) &in_len, &out[0], &out_len);						
+							speex_resampler_process_interleaved_int(cb->resampler, (const spx_int16_t *) frame.data, (spx_uint32_t *) &in_len, &out[0], &out_len);
 							streamer->write( &out[0], sizeof(spx_int16_t) * out_len);
 						}
 						else {
@@ -597,11 +597,11 @@ extern "C" {
 				}
 			}
 			else {
-				switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, 
+				switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG,
 					"aws_transcribe_frame: not sending audio because aws channel has been closed\n");
 			}
 			switch_mutex_unlock(cb->mutex);
 		}
 		return SWITCH_TRUE;
 	}
-}
+}

mod_azure_transcribe/azure_transcribe_glue.cpp

@@ -142,7 +142,13 @@ public:
 			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
 		/*

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_dialogflow_cx/Makefile.am

@@ -0,0 +1,10 @@
+include $(top_srcdir)/build/modmake.rulesam
+MODNAME=mod_dialogflow_cx
+
+mod_LTLIBRARIES = mod_dialogflow_cx.la
+mod_dialogflow_cx_la_SOURCES  = mod_dialogflow_cx.c parser.cpp google_glue.cpp
+mod_dialogflow_cx_la_CFLAGS   = $(AM_CFLAGS)
+mod_dialogflow_cx_la_CXXFLAGS = -I $(top_srcdir)/libs/googleapis/gens $(AM_CXXFLAGS) -std=c++17
+
+mod_dialogflow_cx_la_LIBADD   = $(switch_builddir)/libfreeswitch.la
+mod_dialogflow_cx_la_LDFLAGS  = -avoid-version -module -no-undefined -shared `pkg-config --libs grpc++ grpc` 

mod_dialogflow_cx/README.md

@@ -0,0 +1,84 @@
+# mod_dialogflow
+
+A Freeswitch module that connects a Freeswitch channel to a [dialogflow agent](https://dialogflow.com/docs/getting-started/first-agent) so that an IVR interaction can be driven completely by dialogflow logic.
+
+Once a Freeswitch channel is connected to a dialogflow agent, media is streamed to the dialogflow service, which returns information describing the "intent" that was detected, along with transcriptions and audio prompts and text to play to the caller.  The handling of returned audio by the module is two-fold:
+1.  If an audio clip was returned, it is *not* immediately played to the caller, but instead is written to a temporary wave file on the Freeswitch server.
+2.  Next, a Freeswitch custom event is sent to the application containing the details of the dialogflow response as well as the path to the wave file.
+
+This allows the application whether to decide to play the returned audio clip (via the mod_dptools 'play' command), or to use a text-to-speech service to generate audio using the returned prompt text.
+
+## API
+
+### Commands
+The freeswitch module exposes the following API commands:
+
+#### dialogflow_start
+```
+dialogflow_start <uuid> <project-id> <lang-code> [<event>]
+```
+Attaches media bug to channel and performs streaming recognize request.
+- `uuid` - unique identifier of Freeswitch channel
+- `project-id` - the identifier of the dialogflow project to execute, which may optionally include a dialogflow environment, a region and output audio configurations (see below).
+- `project-id` - the identifier of the dialogflow project to execute, which may optionally include a dialogflow environment, a region and output audio configurations (see below).
+- `lang-code` - a valid dialogflow [language tag](https://dialogflow.com/docs/reference/language) to use for speech recognition
+- `event` - name of an initial event to send to dialogflow; e.g. to trigger an initial prompt
+
+When executing a dialogflow project, the environment and region will default to 'draft' and 'us', respectively.
+
+To specify both an environment and a region, provide a value for project-id in the dialogflow_start command as follows:
+```
+dialogflow-project-id:environment:region, i.e myproject:production:eu-west1
+```
+To specify environment and default to the global region:
+```
+dialogflow-project-id:environment, i.e myproject:production
+```
+To specify a region and default environment:
+```
+dialogflow-project-id::region, i.e myproject::eu-west1
+```
+To simply use the defaults for both environment and region:
+```
+dialogflow-project-id, i.e myproject
+```
+
+By default, [Output Audio configurations](https://cloud.google.com/dialogflow/es/docs/reference/rest/v2/OutputAudioConfig) and [Sentiment Analysis](https://cloud.google.com/dialogflow/es/docs/reference/rpc/google.cloud.dialogflow.v2beta1#google.cloud.dialogflow.v2beta1.SentimentAnalysisRequestConfig) will be ignored and the configs selected for [your agent in Dialogflow platform](https://dialogflow.cloud.google.com/) will be used, however if you wish to abstract your implementation from the platform and define them programatically it can be done in the dialogflow_start command as follows:
+
+```
+dialogflow-project-id:environment:region:speakingRate:pitch:volume:voice-name:voice-gender:effect:sentiment-analysis
+```
+
+Example:
+```
+myproject:production:eu-west1:1.1:1.5:2.5:en-GB-Standard-D:F:handset-class-device:true
+```
+Speaking rate, pitch and volume should take the value of a double. Information [here](https://cloud.google.com/dialogflow/es/docs/reference/rest/v2/projects.agent.environments#synthesizespeechconfig).
+
+Voice Name should take a valid Text-to-speech model name (choose available voices from https://cloud.google.com/text-to-speech/docs/voices). If not set, the Dialogflow service will choose a voice based on the other parameters such as language code and gender. 
+
+Voice Gender should be M for Male, F for Female, N for neutral gender or leave empty for Unspecified.  If not set, the Dialogflow service will choose a voice based on the other parameters such as language code and name. Note that this is only a preference, not requirement. If a voice of the appropriate gender is not available, the synthesizer should substitute a voice with a different gender rather than failing the request.
+
+Effects are applied on the text-to-speech and are used to improve the playback of an audio on different types of hardware. Available effects and information [here](https://cloud.google.com/text-to-speech/docs/audio-profiles#available_audio_profiles).
+
+Sentiment Analysis uses Cloud Natural Language to provide a sentiment score for each user query. To enable send the boolean ```true```.
+
+#### dialogflow_stop
+```
+dialogflow_stop <uuid> 
+```
+Stops dialogflow on the channel.
+
+### Events
+* `dialogflow::intent` - a dialogflow [intent](https://dialogflow.com/docs/intents) has been detected.
+* `dialogflow::transcription` - a transcription has been returned
+* `dialogflow::audio_provided` - an audio prompt has been returned from dialogflow.  Dialogflow will return both an audio clip in linear 16 format, as well as the text of the prompt.  The audio clip will be played out to the caller and the prompt text is returned to the application in this event.
+* `dialogflow::end_of_utterance` - dialogflow has detected the end of an utterance
+* `dialogflow::error` - dialogflow has returned an error
+## Usage
+When using [drachtio-fsrmf](https://www.npmjs.com/package/drachtio-fsmrf), you can access this API command via the api method on the 'endpoint' object.
+```js
+ep.api('dialogflow_start', `${ep.uuid} my-agent-uuxr:production en-US welcome`); 
+```
+## Examples
+[drachtio-dialogflow-phone-gateway](https://github.com/davehorton/drachtio-dialogflow-phone-gateway)

mod_dialogflow_cx/google_glue.cpp

@@ -0,0 +1,629 @@
+#include <cstdlib>
+
+#include <switch.h>
+#include <switch_json.h>
+#include <grpc++/grpc++.h>
+#include <string.h>
+#include <mutex>
+#include <condition_variable>
+
+#include <regex>
+
+#include <fstream>
+#include <string>
+#include <sstream>
+#include <map>
+
+#include "google/cloud/dialogflow/cx/v3/session.grpc.pb.h"
+
+#include "mod_dialogflow_cx.h"
+#include "parser.h"
+
+#define DEFAULT_INTENT "00000000-0000-0000-0000-000000000000"
+
+using google::cloud::dialogflow::cx::v3::Sessions;
+using google::cloud::dialogflow::cx::v3::StreamingDetectIntentRequest;
+using google::cloud::dialogflow::cx::v3::StreamingDetectIntentResponse;
+using google::cloud::dialogflow::cx::v3::AudioEncoding;
+using google::cloud::dialogflow::cx::v3::InputAudioConfig;
+using google::cloud::dialogflow::cx::v3::OutputAudioConfig;
+using google::cloud::dialogflow::cx::v3::SynthesizeSpeechConfig;
+using google::cloud::dialogflow::cx::v3::QueryInput;
+using google::cloud::dialogflow::cx::v3::QueryResult;
+using google::cloud::dialogflow::cx::v3::StreamingRecognitionResult;
+using google::cloud::dialogflow::cx::v3::EventInput;
+using google::cloud::dialogflow::cx::v3::SsmlVoiceGender;
+using google::rpc::Status;
+using google::protobuf::Struct;
+using google::protobuf::Value;
+using google::protobuf::MapPair;
+
+static uint64_t playCount = 0;
+static std::multimap<std::string, std::string> audioFiles;
+static bool hasDefaultCredentials = false;
+
+static switch_status_t hanguphook(switch_core_session_t *session) {
+	switch_channel_t *channel = switch_core_session_get_channel(session);
+	switch_channel_state_t state = switch_channel_get_state(channel);
+
+	if (state == CS_HANGUP || state == CS_ROUTING) {
+		char * sessionId = switch_core_session_get_uuid(session);
+		typedef std::multimap<std::string, std::string>::iterator MMAPIterator;
+		std::pair<MMAPIterator, MMAPIterator> result = audioFiles.equal_range(sessionId);
+		for (MMAPIterator it = result.first; it != result.second; it++) {
+			std::string filename = it->second;
+			std::remove(filename.c_str());
+			switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, 
+				"google_dialogflow_cx_session_cleanup: removed audio file %s\n", filename.c_str());
+		}
+		audioFiles.erase(sessionId);
+		switch_core_event_hook_remove_state_change(session, hanguphook);
+	}
+	return SWITCH_STATUS_SUCCESS;
+}
+
+static  void parseEventParams(Struct* grpcParams, cJSON* json) {
+	auto* map = grpcParams->mutable_fields();
+	int count = cJSON_GetArraySize(json);
+	for (int i = 0; i < count; i++) {
+		cJSON* prop = cJSON_GetArrayItem(json, i);
+		if (prop) {
+			google::protobuf::Value v;
+			switch (prop->type) {
+				case cJSON_False:
+				case cJSON_True:
+					v.set_bool_value(prop->type == cJSON_True);
+					break;
+
+				case cJSON_Number:
+					v.set_number_value(prop->valuedouble);
+					break;
+
+				case cJSON_String:
+					v.set_string_value(prop->valuestring);
+					break;
+
+				case cJSON_Array:
+				case cJSON_Object:
+				case cJSON_Raw:
+				case cJSON_NULL:
+					continue;
+			}
+			map->insert(MapPair<std::string, Value>(prop->string, v));
+		}
+	}
+	switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "parseEventParams: added %d event params\n", map->size());
+}
+
+void tokenize(std::string const &str, const char delim, std::vector<std::string> &out) {
+    size_t start = 0;
+    size_t end = 0;
+		bool finished = false;
+		do {
+			end = str.find(delim, start);
+			if (end == std::string::npos) {
+				finished = true;
+				out.push_back(str.substr(start));
+			}
+			else {
+				out.push_back(str.substr(start, end - start));
+				start = ++end;
+			}
+		} while (!finished);
+}
+
+class GStreamer {
+public:
+    GStreamer(switch_core_session_t *session, const char* lang, char* region, char* projectId, char* agentId, 
+      char* environmentId, char* intent) :
+      m_lang(lang), m_sessionId(switch_core_session_get_uuid(session)), m_agent(agentId), m_projectId(projectId),
+      m_environment( nullptr != environmentId ? environmentId : "draft"), m_regionId(nullptr != region ? region : "us"),
+      m_speakingRate(), m_pitch(), m_volume(), m_voiceName(""), m_voiceGender(""), m_effects(""),
+      m_sentimentAnalysis(false), m_finished(false), m_packets(0) {
+		const char* var;
+		switch_channel_t* channel = switch_core_session_get_channel(session);
+
+		std::vector<std::string> tokens;
+		const char delim = ':';
+		tokenize(projectId, delim, tokens);
+		int idx = 0;
+		for (auto &s: tokens) {
+			switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "GStreamer: token %d: '%s'\n", idx, s.c_str());
+			if (0 == idx) m_projectId = s;
+			else if (1 == idx && s.length() > 0) m_environment = s;
+			else if (2 == idx && s.length() > 0) m_regionId = s;
+			else if (3 == idx && s.length() > 0) m_speakingRate = stod(s);
+			else if (4 == idx && s.length() > 0) m_pitch = stod(s);
+			else if (5 == idx && s.length() > 0) m_volume = stod(s);
+			else if (6 == idx && s.length() > 0) m_voiceName = s;
+			else if (7 == idx && s.length() > 0) m_voiceGender = s;
+			else if (8 == idx && s.length() > 0) m_effects = s;
+			else if (9 == idx && s.length() > 0) m_sentimentAnalysis = (s == "true");
+			idx++;
+		}
+
+		std::string endpoint = "dialogflow.googleapis.com";
+		if (0 != m_regionId.compare("us")) {
+			endpoint = m_regionId;
+			endpoint.append("-dialogflow.googleapis.com");
+		}
+
+		switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, 
+			"GStreamer dialogflow endpoint is %s, region is %s, project is %s, agent is %s, environment is %s\n", 
+			endpoint.c_str(), m_regionId.c_str(), m_projectId.c_str(), m_agent.c_str(), m_environment.c_str());		
+
+		if (var = switch_channel_get_variable(channel, "GOOGLE_APPLICATION_CREDENTIALS")) {
+				auto callCreds = grpc::ServiceAccountJWTAccessCredentials(var, INT64_MAX);
+				auto channelCreds = grpc::SslCredentials(grpc::SslCredentialsOptions());
+				auto creds = grpc::CompositeChannelCredentials(channelCreds, callCreds);
+				m_channel = grpc::CreateChannel(endpoint, creds);
+				switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "GStreamer json credentials are %s\n", var); 
+		}
+		else {
+			auto creds = grpc::GoogleDefaultCredentials();
+			m_channel = grpc::CreateChannel(endpoint, creds);
+		}
+		startStream(session, intent);
+	}
+
+	~GStreamer() {
+		switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "GStreamer::~GStreamer wrote %ld packets %p\n", m_packets, this);		
+	}
+
+	void startStream(switch_core_session_t *session, const char* intent) {
+		char szSession[256];
+
+		m_request = std::make_shared<StreamingDetectIntentRequest>();
+		m_context= std::make_shared<grpc::ClientContext>();
+		m_stub = Sessions::NewStub(m_channel);
+
+    if (0 == m_environment.compare("draft")) {
+  		snprintf(szSession, 256, "projects/%s/locations/%s/agents/%s/sessions/%s", 
+				m_projectId.c_str(), m_regionId.c_str(), m_agent.c_str(), m_sessionId.c_str());
+  		switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "GStreamer::startStream session %s, %p\n", szSession, this);
+    }
+    else {
+  		snprintf(szSession, 256, "projects/%s/locations/%s/agents/%s/environments/%s/sessions/%s", 
+				m_projectId.c_str(), m_regionId.c_str(), m_environment.c_str(), m_sessionId.c_str());
+  		switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "GStreamer::startStream session %s, intent %s,%p\n", szSession, intent, this);
+    }
+
+		m_request->set_session(szSession);
+
+    auto* queryInput = m_request->mutable_query_input();
+    queryInput->set_language_code(m_lang.c_str());
+    if (intent) {
+      char szIntent[256];
+      auto* intentInput = queryInput->mutable_intent();
+      bool isDefault = 0 == strcasecmp(intent, "default");
+
+      snprintf(szIntent, 256, "projects/%s/locations/%s/agents/%s/intents/%s", 
+        m_projectId.c_str(), m_regionId.c_str(), m_agent.c_str(), isDefault ? DEFAULT_INTENT : intent);
+
+      switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "GStreamer::write writing initial query input w/intent %s, %p\n", 
+        szIntent, this);
+      
+      intentInput->set_intent(szIntent);
+    }
+    else {
+      auto* audio_config = queryInput->mutable_audio()->mutable_config();
+      audio_config->set_sample_rate_hertz(16000);
+      audio_config->set_enable_word_info(false);
+      audio_config->set_audio_encoding(AudioEncoding::AUDIO_ENCODING_LINEAR_16);
+      audio_config->set_single_utterance(true);
+    }
+
+    auto* outputAudioConfig = m_request->mutable_output_audio_config();
+    outputAudioConfig->set_sample_rate_hertz(8000);
+    outputAudioConfig->set_audio_encoding(OutputAudioEncoding::OUTPUT_AUDIO_ENCODING_LINEAR_16);
+
+    if (isAnyOutputAudioConfigChanged()) {
+      switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "GStreamer::startStream adding a custom OutputAudioConfig to the request since at"
+                                                                " least one parameter was received.");
+
+      auto* synthesizeSpeechConfig = outputAudioConfig->mutable_synthesize_speech_config();
+      if (m_speakingRate) synthesizeSpeechConfig->set_speaking_rate(m_speakingRate);
+      if (m_pitch) synthesizeSpeechConfig->set_pitch(m_pitch);
+      if (m_volume) synthesizeSpeechConfig->set_volume_gain_db(m_volume);
+      if (!m_effects.empty()) synthesizeSpeechConfig->add_effects_profile_id(m_effects);
+
+      auto* voice = synthesizeSpeechConfig->mutable_voice();
+      if (!m_voiceName.empty()) voice->set_name(m_voiceName);
+      if (!m_voiceGender.empty()) {
+        SsmlVoiceGender gender = SsmlVoiceGender::SSML_VOICE_GENDER_UNSPECIFIED;
+        switch (toupper(m_voiceGender[0]))
+        {
+          case 'F': gender = SsmlVoiceGender::SSML_VOICE_GENDER_MALE; break;
+          case 'M': gender = SsmlVoiceGender::SSML_VOICE_GENDER_FEMALE; break;
+          case 'N': gender = SsmlVoiceGender::SSML_VOICE_GENDER_NEUTRAL; break;
+        }
+        voice->set_ssml_gender(gender);
+      } else {
+        switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "GStreamer::startStream no custom parameters for OutputAudioConfig, keeping default");
+      }
+      if (m_sentimentAnalysis) {
+        switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "GStreamer::startStream received sentiment analysis flag as true, adding as query param");
+        auto* queryParameters = m_request->mutable_query_params();
+        queryParameters->set_analyze_query_text_sentiment(m_sentimentAnalysis);
+      }
+    }
+  
+    /**
+     * Note: there are other parameters that can be set in the audio config, such as:
+     * hints, model, model variant, barge in config
+     * 
+     */
+
+    switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "GStreamer::startStream checking OutputAudioConfig custom parameters: speaking rate %f,"
+                                                            " pitch %f, volume %f, voice name '%s' gender '%s', effects '%s'\n", m_speakingRate,
+                                                            m_pitch, m_volume, m_voiceName.c_str(), m_voiceGender.c_str(), m_effects.c_str());
+
+		m_streamer = m_stub->StreamingDetectIntent(m_context.get());
+    m_streamer->Write(*m_request);
+	}
+	bool write(void* data, uint32_t datalen) {
+		if (m_finished) {
+			switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "GStreamer::write not writing because we are finished, %p\n", this);
+			return false;
+		}
+
+    auto* queryInput = m_request->mutable_query_input();
+		m_request->clear_query_input();
+		m_request->clear_query_params();
+    
+    queryInput->set_language_code(m_lang.c_str());
+    queryInput->mutable_audio()->set_audio(data, datalen);
+
+   //switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "GStreamer::write writing packet %d\n", m_packets); 
+
+		m_packets++;
+    return m_streamer->Write(*m_request);
+	}
+	bool read(StreamingDetectIntentResponse* response) {
+		return m_streamer->Read(response);
+	}
+	grpc::Status finish() {
+		switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "GStreamer::finish %p\n", this);
+		if (m_finished) {
+			grpc::Status ok;
+			return ok;
+		}
+		m_finished = true;
+		return m_streamer->Finish();
+	}
+	void writesDone() {
+		m_streamer->WritesDone();
+	}
+
+	bool isFinished() {
+		return m_finished;
+	}
+
+    bool isAnyOutputAudioConfigChanged() {
+        return m_speakingRate|| m_pitch || m_volume || !m_voiceName.empty() || !m_voiceGender.empty() || !m_effects.empty();
+    }
+
+private:
+	std::string m_sessionId;
+	std::shared_ptr<grpc::ClientContext> m_context;
+	std::shared_ptr<grpc::Channel> m_channel;
+	std::unique_ptr<Sessions::Stub> 	m_stub;
+	std::unique_ptr< grpc::ClientReaderWriterInterface<StreamingDetectIntentRequest, StreamingDetectIntentResponse> > m_streamer;
+	std::shared_ptr<StreamingDetectIntentRequest> m_request;
+	std::string m_lang;
+	std::string m_projectId;
+  std::string m_agent;
+	std::string m_environment;
+	std::string m_regionId;
+	double m_speakingRate;
+	double m_pitch;
+	double m_volume;
+	std::string m_effects;
+	std::string m_voiceName;
+	std::string m_voiceGender;
+	bool m_sentimentAnalysis;
+	bool m_finished;
+  bool m_ready;
+
+	uint32_t m_packets;
+};
+
+static void killcb(struct cap_cb* cb) {
+	if (cb) {
+		if (cb->streamer) {
+			GStreamer* p = (GStreamer *) cb->streamer;
+			delete p;
+			cb->streamer = NULL;
+		}
+		if (cb->resampler) {
+				speex_resampler_destroy(cb->resampler);
+				cb->resampler = NULL;
+		}
+	}
+}
+
+static void *SWITCH_THREAD_FUNC grpc_read_thread(switch_thread_t *thread, void *obj) {
+	struct cap_cb *cb = (struct cap_cb *) obj;
+	GStreamer* streamer = (GStreamer *) cb->streamer;
+
+	switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "grpc_read_thread: starting cb %p\n", (void *) cb);
+
+	// Our contract: while we are reading, cb and cb->streamer will not be deleted
+
+	// Read responses until there are no more
+	StreamingDetectIntentResponse response;
+	while (streamer->read(&response)) {  
+		switch_core_session_t* psession = switch_core_session_locate(cb->sessionId);
+		if (psession) {
+			switch_channel_t* channel = switch_core_session_get_channel(psession);
+			GRPCParser parser(psession);
+
+    	switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "grpc_read_thread: read something %p\n", (void *) cb);
+
+      // TODO: handle has_debugging_info()
+      if (response.has_debugging_info()) {
+        auto di = response.debugging_info();
+      }
+
+      bool hasAudio = false;
+			if (response.has_detect_intent_response() || response.has_recognition_result()) {
+				cJSON* jResponse = parser.parse(response) ;
+				char* json = cJSON_PrintUnformatted(jResponse);
+				const char* type = DIALOGFLOW_CX_EVENT_TRANSCRIPTION;
+
+				if (response.has_detect_intent_response()) {
+          hasAudio = response.detect_intent_response().output_audio().length() > 0;
+          type = DIALOGFLOW_CX_EVENT_INTENT;
+        }
+				else {
+					const StreamingRecognitionResult_MessageType& o = response.recognition_result().message_type();
+					if (0 == StreamingRecognitionResult_MessageType_Name(o).compare("END_OF_SINGLE_UTTERANCE")) {
+						type = DIALOGFLOW_CX_EVENT_END_OF_UTTERANCE;
+					}
+				}
+
+				cb->responseHandler(psession, type, json);
+
+				free(json);
+				cJSON_Delete(jResponse);
+			}
+
+			// save audio
+			if (hasAudio) {
+        auto& dir = response.detect_intent_response();
+        const std::string& audio = dir.output_audio();
+				std::ostringstream s;
+				s << SWITCH_GLOBAL_dirs.temp_dir << SWITCH_PATH_SEPARATOR <<
+					cb->sessionId << "_" <<  ++playCount;
+				switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(psession), SWITCH_LOG_DEBUG, "grpc_read_thread: received audio to play\n");
+
+				if (dir.has_output_audio_config()) {
+					auto& cfg = dir.output_audio_config();
+					switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(psession), SWITCH_LOG_DEBUG, "grpc_read_thread: encoding is %d\n", cfg.audio_encoding());
+					if (cfg.audio_encoding() == OutputAudioEncoding::OUTPUT_AUDIO_ENCODING_MP3) {
+						s << ".mp3";
+					}
+					else if (cfg.audio_encoding() == OutputAudioEncoding::OUTPUT_AUDIO_ENCODING_OGG_OPUS) {
+						s << ".opus";
+					}
+					else {
+						s << ".wav";
+					}
+				}
+				std::ofstream f(s.str(), std::ofstream::binary);
+				f << audio;
+				f.close();
+				switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(psession), SWITCH_LOG_DEBUG, "grpc_read_thread: wrote audio to %s\n", s.str().c_str());
+
+				// add the file to the list of files played for this session, 
+				// we'll delete when session closes
+				audioFiles.insert(std::pair<std::string, std::string>(cb->sessionId, s.str()));
+
+				cJSON * jResponse = cJSON_CreateObject();
+				cJSON_AddItemToObject(jResponse, "path", cJSON_CreateString(s.str().c_str()));
+				char* json = cJSON_PrintUnformatted(jResponse);
+
+				cb->responseHandler(psession, DIALOGFLOW_CX_EVENT_AUDIO_PROVIDED, json);
+				free(json);
+				cJSON_Delete(jResponse);
+			}
+			switch_core_session_rwunlock(psession);
+		}
+		else {
+			break;
+		}
+	}
+	switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "dialogflow_cx read loop is done\n");
+
+	// finish the detect intent session: here is where we may get an error if credentials are invalid
+	switch_core_session_t* psession = switch_core_session_locate(cb->sessionId);
+	if (psession) {
+		grpc::Status status = streamer->finish();
+		if (!status.ok()) {
+			std::ostringstream s;
+			s << "{\"msg\": \"" << status.error_message() << "\", \"code\": " << status.error_code();
+			if (status.error_details().length() > 0) {
+				s << ", \"details\": \"" << status.error_details() << "\"";
+			}
+			s << "}";
+			switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_CRIT, "StreamingDetectIntentRequest finished with err %s (%d): %s\n", 
+				status.error_message().c_str(), status.error_code(), status.error_details().c_str());
+			cb->errorHandler(psession, s.str().c_str());
+		}
+
+		switch_core_session_rwunlock(psession);
+	}
+	switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "dialogflow read thread exiting	\n");
+	return NULL;
+}
+
+extern "C" {
+	switch_status_t google_dialogflow_cx_init() {
+		const char* gcsServiceKeyFile = std::getenv("GOOGLE_APPLICATION_CREDENTIALS");
+		if (NULL == gcsServiceKeyFile) {
+			switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_NOTICE, 
+				"\"GOOGLE_APPLICATION_CREDENTIALS\" environment variable is not set; authentication will use \"GOOGLE_APPLICATION_CREDENTIALS\" channel variable\n");
+		}
+		else {
+			hasDefaultCredentials = true;
+		}
+		return SWITCH_STATUS_SUCCESS;
+	}
+	
+	switch_status_t google_dialogflow_cx_cleanup() {
+		return SWITCH_STATUS_SUCCESS;
+	}
+
+	// start dialogflow on a channel
+	switch_status_t google_dialogflow_cx_session_init(
+		switch_core_session_t *session, 
+		responseHandler_t responseHandler, 
+		errorHandler_t errorHandler, 
+		uint32_t samples_per_second, 
+		char* lang,
+    char* region,
+		char* projectId,
+    char* agentId,
+    char* environmentId,
+    char* intent,
+		struct cap_cb **ppUserData
+	) {
+		switch_status_t status = SWITCH_STATUS_SUCCESS;
+		switch_channel_t *channel = switch_core_session_get_channel(session);
+		int err;
+		switch_threadattr_t *thd_attr = NULL;
+		switch_memory_pool_t *pool = switch_core_session_get_pool(session);
+		struct cap_cb* cb = (struct cap_cb *) switch_core_session_alloc(session, sizeof(*cb));
+
+		if (!hasDefaultCredentials && !switch_channel_get_variable(channel, "GOOGLE_APPLICATION_CREDENTIALS")) {
+			switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_ERROR, 
+				"missing credentials: GOOGLE_APPLICATION_CREDENTIALS must be suuplied either as an env variable (path to file) or a channel variable (json string)\n");
+			status = SWITCH_STATUS_FALSE;
+			goto done; 
+		}
+
+		strncpy(cb->sessionId, switch_core_session_get_uuid(session), 256);
+		cb->responseHandler = responseHandler;
+		cb->errorHandler = errorHandler;
+
+		if (switch_mutex_init(&cb->mutex, SWITCH_MUTEX_NESTED, pool) != SWITCH_STATUS_SUCCESS) {
+			switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_ERROR, "Error initializing mutex\n");
+			status = SWITCH_STATUS_FALSE;
+			goto done; 
+		}
+
+		strncpy(cb->lang, lang, MAX_LANG);
+		strncpy(cb->projectId, lang, MAX_PROJECT_ID);
+    strncpy(cb->agentId, agentId, MAX_PROJECT_ID);
+    if (nullptr != environmentId) strncpy(cb->environmentId, environmentId, MAX_PROJECT_ID);
+    if (nullptr != region) strncpy(cb->region, region, MAX_REGION);
+		cb->streamer = new GStreamer(session, lang, region, projectId, agentId, environmentId, intent);
+		cb->resampler = speex_resampler_init(1, 8000, 16000, SWITCH_RESAMPLE_QUALITY, &err);
+		if (0 != err) {
+			switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_ERROR, "%s: Error initializing resampler: %s.\n", 
+						switch_channel_get_name(channel), speex_resampler_strerror(err));
+			status = SWITCH_STATUS_FALSE;
+			goto done;
+		}
+
+		// hangup hook to clear temp audio files
+		switch_core_event_hook_add_state_change(session, hanguphook);
+
+		// create the read thread
+		switch_threadattr_create(&thd_attr, pool);
+		//switch_threadattr_detach_set(thd_attr, 1);
+		switch_threadattr_stacksize_set(thd_attr, SWITCH_THREAD_STACKSIZE);
+		switch_thread_create(&cb->thread, thd_attr, grpc_read_thread, cb, pool);
+
+		*ppUserData = cb;
+	
+	done:
+		if (status != SWITCH_STATUS_SUCCESS) {
+			killcb(cb);
+		}
+		return status;
+	}
+
+	switch_status_t google_dialogflow_cx_session_stop(switch_core_session_t *session, int channelIsClosing) {
+		switch_channel_t *channel = switch_core_session_get_channel(session);
+		switch_media_bug_t *bug = (switch_media_bug_t*) switch_channel_get_private(channel, MY_BUG_NAME);
+
+		if (bug) {
+			struct cap_cb *cb = (struct cap_cb *) switch_core_media_bug_get_user_data(bug);
+			switch_status_t st;
+
+			// close connection and get final responses
+			switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "google_dialogflow_cx_session_cleanup: acquiring lock\n");
+			switch_mutex_lock(cb->mutex);
+			switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "google_dialogflow_cx_session_cleanup: acquired lock\n");
+			GStreamer* streamer = (GStreamer *) cb->streamer;
+			if (streamer) {
+				switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "google_dialogflow_cx_session_cleanup: sending writesDone..\n");
+				streamer->writesDone();
+				streamer->finish();
+			}
+			if (cb->thread) {
+				switch_status_t retval;
+				switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_INFO, "google_dialogflow_cx_session_cleanup: waiting for read thread to complete\n");
+				switch_thread_join(&retval, cb->thread);
+				cb->thread = NULL;
+				switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_INFO, "google_dialogflow_cx_session_cleanup: read thread completed\n");
+			}
+			killcb(cb);
+
+			switch_channel_set_private(channel, MY_BUG_NAME, NULL);
+			if (!channelIsClosing) switch_core_media_bug_remove(session, &bug);
+
+			switch_mutex_unlock(cb->mutex);
+			switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_INFO, "google_dialogflow_cx_session_cleanup: Closed google session\n");
+
+			return SWITCH_STATUS_SUCCESS;
+		}
+
+		switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_INFO, "%s Bug is not attached.\n", switch_channel_get_name(channel));
+		return SWITCH_STATUS_FALSE;
+	}
+	
+	switch_bool_t google_dialogflow_cx_frame(switch_media_bug_t *bug, void* user_data) {
+		switch_core_session_t *session = switch_core_media_bug_get_session(bug);
+		uint8_t data[SWITCH_RECOMMENDED_BUFFER_SIZE];
+		switch_frame_t frame = {};
+		struct cap_cb *cb = (struct cap_cb *) user_data;
+
+		frame.data = data;
+		frame.buflen = SWITCH_RECOMMENDED_BUFFER_SIZE;
+
+		if (switch_mutex_trylock(cb->mutex) == SWITCH_STATUS_SUCCESS) {
+			GStreamer* streamer = (GStreamer *) cb->streamer;
+			if (streamer && !streamer->isFinished()) {
+				while (switch_core_media_bug_read(bug, &frame, SWITCH_TRUE) == SWITCH_STATUS_SUCCESS && !switch_test_flag((&frame), SFF_CNG)) {
+					if (frame.datalen) {
+						spx_int16_t out[SWITCH_RECOMMENDED_BUFFER_SIZE];
+						spx_uint32_t out_len = SWITCH_RECOMMENDED_BUFFER_SIZE;
+						spx_uint32_t in_len = frame.samples;
+						size_t written;
+						
+						speex_resampler_process_interleaved_int(cb->resampler, (const spx_int16_t *) frame.data, (spx_uint32_t *) &in_len, &out[0], &out_len);
+						
+						streamer->write( &out[0], sizeof(spx_int16_t) * out_len);
+					}
+				}
+			}
+			else {
+				//switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, 
+				//	"google_dialogflow_cx_frame: not sending audio because google channel has been closed\n");
+			}
+			switch_mutex_unlock(cb->mutex);
+		}
+		else {
+			//switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, 
+			//	"google_dialogflow_cx_frame: not sending audio since failed to get lock on mutex\n");
+		}
+		return SWITCH_TRUE;
+	}
+
+	void destroyChannelUserData(struct cap_cb* cb) {
+		killcb(cb);
+	}
+
+}

mod_dialogflow_cx/google_glue.h

@@ -0,0 +1,22 @@
+#ifndef __GOOGLE_GLUE_CX_H__
+#define __GOOGLE_GLUE_CX_H__
+
+switch_status_t google_dialogflow_cx_init();
+switch_status_t google_dialogflow_cx_cleanup();
+switch_status_t google_dialogflow_cx_session_init(
+    switch_core_session_t *session, 
+		responseHandler_t responseHandler, 
+		errorHandler_t errorHandler, 
+		uint32_t samples_per_second, 
+		char* lang,
+    char* region,
+		char* projectId,
+    char* agentId,
+    char* environmentId,
+    char* intent,
+		struct cap_cb **ppUserData);
+switch_status_t google_dialogflow_cx_session_stop(switch_core_session_t *session, int channelIsClosing);
+switch_bool_t google_dialogflow_cx_frame(switch_media_bug_t *bug, void* user_data);
+
+void destroyChannelUserData(struct cap_cb* cb);
+#endif

mod_dialogflow_cx/mod_dialogflow_cx.c

@@ -0,0 +1,295 @@
+/* 
+ *
+ * mod_dialogflow_cx.c -- Freeswitch module for running a google dialogflow
+ *
+ */
+#include "mod_dialogflow_cx.h"
+#include "google_glue.h"
+
+/* Prototypes */
+SWITCH_MODULE_SHUTDOWN_FUNCTION(mod_dialogflow_cx_shutdown);
+SWITCH_MODULE_RUNTIME_FUNCTION(mod_dialogflow_cx_runtime);
+SWITCH_MODULE_LOAD_FUNCTION(mod_dialogflow_cx_load);
+
+SWITCH_MODULE_DEFINITION(mod_dialogflow_cx, mod_dialogflow_cx_load, mod_dialogflow_cx_shutdown, NULL);
+
+static switch_status_t do_stop(switch_core_session_t *session);
+
+static void responseHandler(switch_core_session_t* session, const char * type, char * json) {
+	switch_event_t *event;
+	switch_channel_t *channel = switch_core_session_get_channel(session);
+
+	switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "json payload for type %s: %s.\n", type, json);
+
+	switch_event_create_subclass(&event, SWITCH_EVENT_CUSTOM, type);
+	switch_channel_event_set_data(channel, event);
+	switch_event_add_body(event, "%s", json);
+	switch_event_fire(&event);
+}
+static void errorHandler(switch_core_session_t* session, const char * json) {
+	switch_event_t *event;
+	switch_channel_t *channel = switch_core_session_get_channel(session);
+
+	switch_event_create_subclass(&event, SWITCH_EVENT_CUSTOM, DIALOGFLOW_CX_EVENT_ERROR);
+	switch_channel_event_set_data(channel, event);
+	switch_event_add_body(event, "%s", json);
+
+	switch_event_fire(&event);
+
+	do_stop(session);
+}
+
+static switch_bool_t capture_callback(switch_media_bug_t *bug, void *user_data, switch_abc_type_t type)
+{
+	switch_core_session_t *session = switch_core_media_bug_get_session(bug);
+
+	switch (type) {
+	case SWITCH_ABC_TYPE_INIT:
+			switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "Got SWITCH_ABC_TYPE_INIT.\n");
+		break;
+
+	case SWITCH_ABC_TYPE_CLOSE:
+		{
+			switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "Got SWITCH_ABC_TYPE_CLOSE.\n");
+
+			google_dialogflow_cx_session_stop(session, 1);
+			switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "Finished SWITCH_ABC_TYPE_CLOSE.\n");
+		}
+		break;
+	
+	case SWITCH_ABC_TYPE_READ:
+
+		return google_dialogflow_cx_frame(bug, user_data);
+		break;
+
+	case SWITCH_ABC_TYPE_WRITE:
+	default:
+		break;
+	}
+
+	return SWITCH_TRUE;
+}
+
+static switch_status_t start_capture(switch_core_session_t *session, switch_media_bug_flag_t flags, char* lang, char* region, char* projectId, 
+char *agentId, char *environmentId, char* intent)
+{
+	switch_channel_t *channel = switch_core_session_get_channel(session);
+	switch_media_bug_t *bug;
+	switch_codec_implementation_t read_impl = { 0 };
+	struct cap_cb *cb = NULL;
+	switch_status_t status = SWITCH_STATUS_SUCCESS;
+
+	if (switch_channel_get_private(channel, MY_BUG_NAME)) {
+		switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "a dialogflow_cx is already running on this channel, we will stop it.\n");
+		do_stop(session);
+	}
+
+	if (switch_channel_pre_answer(channel) != SWITCH_STATUS_SUCCESS) {
+		switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "channel must have at least early media to run dialogflow.\n");
+		status = SWITCH_STATUS_FALSE;
+		goto done;
+	}
+
+	switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "starting dialogflow_cx with project %s, language %s, intent %s\n", 
+		projectId, lang, intent);
+
+	switch_core_session_get_read_impl(session, &read_impl);
+	if (SWITCH_STATUS_FALSE == google_dialogflow_cx_session_init(session, responseHandler, errorHandler, 
+		read_impl.samples_per_second, lang, region, projectId, agentId, environmentId, intent, &cb)) {
+		switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Error initializing google dialogflow_cx session.\n");
+		status = SWITCH_STATUS_FALSE;
+		goto done;
+	}
+
+	if ((status = switch_core_media_bug_add(session, MY_BUG_NAME, NULL, capture_callback, (void *) cb, 0, flags, &bug)) != SWITCH_STATUS_SUCCESS) {
+		switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Error adding bug.\n");
+		status = SWITCH_STATUS_FALSE;
+		goto done;
+	}
+	switch_channel_set_private(channel, MY_BUG_NAME, bug);
+
+done:
+	if (status == SWITCH_STATUS_FALSE) {
+		if (cb) destroyChannelUserData(cb);
+	}
+
+	return status;
+}
+
+static switch_status_t do_stop(switch_core_session_t *session)
+{
+	switch_status_t status = SWITCH_STATUS_SUCCESS;
+
+	switch_channel_t *channel = switch_core_session_get_channel(session);
+	switch_media_bug_t *bug = switch_channel_get_private(channel, MY_BUG_NAME);
+
+	if (bug) {
+		switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "Received user command command to stop dialogflow_cx.\n");
+		status = google_dialogflow_cx_session_stop(session, 0);
+		switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "stopped dialogflow_cx.\n");
+	}
+
+	return status;
+}
+
+#define DIALOGFLOW_API_START_SYNTAX "<uuid> region project-id agent-id environment-id lang-code [intent]"
+SWITCH_STANDARD_API(dialogflow_cx_api_start_function)
+{
+	char *mycmd = NULL, *argv[10] = { 0 };
+	int argc = 0;
+	switch_status_t status = SWITCH_STATUS_FALSE;
+	switch_media_bug_flag_t flags = SMBF_READ_STREAM | SMBF_READ_STREAM | SMBF_READ_PING;
+
+	switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "command %s\n", cmd);
+	if (!zstr(cmd) && (mycmd = strdup(cmd))) {
+		argc = switch_separate_string(mycmd, ' ', argv, (sizeof(argv) / sizeof(argv[0])));
+	}
+
+	if (zstr(cmd) || argc < 6) {
+		switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_ERROR, "Error with command %s %s %s.\n", cmd, argv[0], argv[1]);
+		stream->write_function(stream, "-USAGE: %s\n", DIALOGFLOW_API_START_SYNTAX);
+		goto done;
+	} else {
+		switch_core_session_t *lsession = NULL;
+
+		if ((lsession = switch_core_session_locate(argv[0]))) {
+			char *intent = NULL;
+      char *region = argv[1];
+			char *projectId = argv[2];
+      char *agentId = argv[3];
+      char *environmentId = argv[4];
+			char *lang = argv[5];
+      if (0 == strcmp("default", environmentId)) {
+        environmentId = NULL;
+      }
+      if (0 == strcmp("default", region)) {
+        region = NULL;
+      }
+			if (argc > 6) {
+				intent = argv[6];
+				if (0 == strcmp("none", intent)) {
+					intent = NULL;
+				}
+			}
+			status = start_capture(lsession, flags, lang, region, projectId, agentId, environmentId, intent);
+			switch_core_session_rwunlock(lsession);
+		}
+	}
+
+	if (status == SWITCH_STATUS_SUCCESS) {
+		stream->write_function(stream, "+OK Success\n");
+	} else {
+		stream->write_function(stream, "-ERR Operation Failed\n");
+	}
+
+  done:
+
+	switch_safe_free(mycmd);
+	return SWITCH_STATUS_SUCCESS;	
+}
+
+#define DIALOGFLOW_API_STOP_SYNTAX "<uuid>"
+SWITCH_STANDARD_API(dialogflow_cx_api_stop_function)
+{
+	char *mycmd = NULL, *argv[10] = { 0 };
+	int argc = 0;
+	switch_status_t status = SWITCH_STATUS_FALSE;
+
+	switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "command %s\n", cmd);
+	if (!zstr(cmd) && (mycmd = strdup(cmd))) {
+		argc = switch_separate_string(mycmd, ' ', argv, (sizeof(argv) / sizeof(argv[0])));
+	}
+
+	if (zstr(cmd) || argc != 1) {
+		switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_ERROR, "Error with command %s %s %s.\n", cmd, argv[0], argv[1]);
+		stream->write_function(stream, "-USAGE: %s\n", DIALOGFLOW_API_STOP_SYNTAX);
+		goto done;
+	} else {
+		switch_core_session_t *lsession = NULL;
+
+		if ((lsession = switch_core_session_locate(argv[0]))) {
+			status = do_stop(lsession);
+			switch_core_session_rwunlock(lsession);
+		}
+	}
+
+	if (status == SWITCH_STATUS_SUCCESS) {
+		stream->write_function(stream, "+OK Success\n");
+	} else {
+		stream->write_function(stream, "-ERR Operation Failed\n");
+	}
+
+  done:
+
+	switch_safe_free(mycmd);
+	return SWITCH_STATUS_SUCCESS;
+}
+
+
+/* Macro expands to: switch_status_t mod_dialogflow_cx_load(switch_loadable_module_interface_t **module_interface, switch_memory_pool_t *pool) */
+SWITCH_MODULE_LOAD_FUNCTION(mod_dialogflow_cx_load)
+{
+	switch_api_interface_t *api_interface;
+
+	/* create/register custom event message types */
+	if (switch_event_reserve_subclass(DIALOGFLOW_CX_EVENT_INTENT) != SWITCH_STATUS_SUCCESS) {
+		switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Couldn't register subclass %s!\n", DIALOGFLOW_CX_EVENT_INTENT);
+		return SWITCH_STATUS_TERM;
+	}
+	if (switch_event_reserve_subclass(DIALOGFLOW_CX_EVENT_TRANSCRIPTION) != SWITCH_STATUS_SUCCESS) {
+		switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Couldn't register subclass %s!\n", DIALOGFLOW_CX_EVENT_TRANSCRIPTION);
+		return SWITCH_STATUS_TERM;
+	}
+	if (switch_event_reserve_subclass(DIALOGFLOW_CX_EVENT_END_OF_UTTERANCE) != SWITCH_STATUS_SUCCESS) {
+		switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Couldn't register subclass %s!\n", DIALOGFLOW_CX_EVENT_END_OF_UTTERANCE);
+		return SWITCH_STATUS_TERM;
+	}
+	if (switch_event_reserve_subclass(DIALOGFLOW_CX_EVENT_AUDIO_PROVIDED) != SWITCH_STATUS_SUCCESS) {
+		switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Couldn't register subclass %s!\n", DIALOGFLOW_CX_EVENT_AUDIO_PROVIDED);
+		return SWITCH_STATUS_TERM;
+	}
+
+	if (switch_event_reserve_subclass(DIALOGFLOW_CX_EVENT_ERROR) != SWITCH_STATUS_SUCCESS) {
+		switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Couldn't register subclass %s!\n", DIALOGFLOW_CX_EVENT_ERROR);
+		return SWITCH_STATUS_TERM;
+	}
+
+
+	/* connect my internal structure to the blank pointer passed to me */
+	*module_interface = switch_loadable_module_create_module_interface(pool, modname);
+
+	switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_NOTICE, "Google Dialogflow CX API loading..\n");
+
+  if (SWITCH_STATUS_FALSE == google_dialogflow_cx_init()) {
+		switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_CRIT, "Failed initializing google dialogflow cx interface\n");
+	}
+
+	switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_NOTICE, "Google Dialogflow CX API successfully loaded\n");
+
+	SWITCH_ADD_API(api_interface, "dialogflow_cx_start", "Start a google dialogflow cx", dialogflow_cx_api_start_function, DIALOGFLOW_API_START_SYNTAX);
+	SWITCH_ADD_API(api_interface, "dialogflow_cx_stop", "Terminate a google dialogflow cx", dialogflow_cx_api_stop_function, DIALOGFLOW_API_STOP_SYNTAX);
+
+	switch_console_set_complete("add dialogflow_cx_stop");
+	switch_console_set_complete("add dialogflow_cx_start project lang");
+	switch_console_set_complete("add dialogflow_cx_start project lang timeout-secs");
+	switch_console_set_complete("add dialogflow_cx_start project lang timeout-secs event");
+
+	/* indicate that the module should continue to be loaded */
+	return SWITCH_STATUS_SUCCESS;
+}
+
+/*
+  Called when the system shuts down
+  Macro expands to: switch_status_t mod_dialogflow_shutdown() */
+SWITCH_MODULE_SHUTDOWN_FUNCTION(mod_dialogflow_cx_shutdown)
+{
+	google_dialogflow_cx_cleanup();
+
+	switch_event_free_subclass(DIALOGFLOW_CX_EVENT_INTENT);
+	switch_event_free_subclass(DIALOGFLOW_CX_EVENT_TRANSCRIPTION);
+	switch_event_free_subclass(DIALOGFLOW_CX_EVENT_END_OF_UTTERANCE);
+	switch_event_free_subclass(DIALOGFLOW_CX_EVENT_AUDIO_PROVIDED);
+	switch_event_free_subclass(DIALOGFLOW_CX_EVENT_ERROR);
+
+	return SWITCH_STATUS_SUCCESS;
+}

mod_dialogflow_cx/mod_dialogflow_cx.h

@@ -0,0 +1,40 @@
+#ifndef __MOD_DIALOGFLOW_CX_H__
+#define __MOD_DIALOGFLOW_CX_H__
+
+#include <switch.h>
+#include <speex/speex_resampler.h>
+
+#include <unistd.h>
+
+#define MY_BUG_NAME "__dialogflow_cx_bug__"
+#define DIALOGFLOW_CX_EVENT_INTENT "dialogflow_cx::intent"
+#define DIALOGFLOW_CX_EVENT_TRANSCRIPTION "dialogflow_cx::transcription"
+#define DIALOGFLOW_CX_EVENT_AUDIO_PROVIDED "dialogflow_cx::audio_provided"
+#define DIALOGFLOW_CX_EVENT_END_OF_UTTERANCE "dialogflow_cx::end_of_utterance"
+#define DIALOGFLOW_CX_EVENT_ERROR "dialogflow_cx::error"
+
+#define MAX_LANG (12)
+#define MAX_PROJECT_ID (256)
+#define MAX_PATHLEN (256)
+#define MAX_REGION (56)
+
+/* per-channel data */
+typedef void (*responseHandler_t)(switch_core_session_t* session, const char * type, char* json);
+typedef void (*errorHandler_t)(switch_core_session_t* session, const char * reason);
+
+struct cap_cb {
+	switch_mutex_t *mutex;
+	char sessionId[256];
+  SpeexResamplerState *resampler;
+	void* streamer;
+	responseHandler_t responseHandler;
+	errorHandler_t errorHandler;
+	switch_thread_t* thread;
+	char lang[MAX_LANG];
+	char projectId[MAX_PROJECT_ID];
+  char agentId[MAX_PROJECT_ID];
+  char environmentId[MAX_PROJECT_ID];
+  char region[MAX_REGION];
+};
+
+#endif

mod_dialogflow_cx/parser.cpp

@@ -0,0 +1,425 @@
+#include "parser.h"
+#include <switch.h>
+
+template <typename T> cJSON* GRPCParser::parseCollection(const RepeatedPtrField<T> coll) {
+	cJSON* json = cJSON_CreateArray();
+	typename RepeatedPtrField<T>::const_iterator it = coll.begin();
+	for (; it != coll.end(); it++) {
+		cJSON_AddItemToArray(json, parse(*it));
+	}
+	return json;
+}
+
+cJSON* GRPCParser::parse(const StreamingDetectIntentResponse& response) {
+	cJSON * json = cJSON_CreateObject();
+
+	switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(m_session), SWITCH_LOG_INFO, "GStrGRPCParser - parsing StreamingDetectIntentResponse\n");
+
+	// recognition_result
+	if (response.has_recognition_result()) {
+		switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(m_session), SWITCH_LOG_INFO, "GStrGRPCParser - adding recognition result\n");
+        cJSON_AddItemToObject(json, "recognition_result", parse(response.recognition_result()));
+	}
+
+  // detect_intent_response
+  if (response.has_detect_intent_response()) {
+    switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(m_session), SWITCH_LOG_INFO, "GStrGRPCParser - adding detect intent response\n");
+    cJSON_AddItemToObject(json, "detect_intent_response", parse(response.detect_intent_response()));
+  }
+
+	return json;
+}
+
+cJSON* GRPCParser::parse(const OutputAudioEncoding& o) {
+	return cJSON_CreateString(OutputAudioEncoding_Name(o).c_str());
+}
+
+cJSON* GRPCParser::parse(const SynthesizeSpeechConfig& o) {
+	cJSON * json = cJSON_CreateObject();
+
+  cJSON_AddItemToObject(json, "speaking_rate", cJSON_CreateNumber(o.speaking_rate()));
+  cJSON_AddItemToObject(json, "pitch", cJSON_CreateNumber(o.pitch()));
+  cJSON_AddItemToObject(json, "volume_gain_db", cJSON_CreateNumber(o.volume_gain_db()));
+
+  return json;
+}
+cJSON* GRPCParser::parse(const VoiceSelectionParams& o) {
+  cJSON * json = cJSON_CreateObject();
+
+  cJSON_AddItemToObject(json, "name", cJSON_CreateString(o.name().c_str()));
+  cJSON_AddItemToObject(json, "ssml_gender", cJSON_CreateString(SsmlVoiceGender_Name(o.ssml_gender()).c_str()));
+
+  return json;
+}
+
+cJSON* GRPCParser::parse(const OutputAudioConfig& o) {
+	cJSON * json = cJSON_CreateObject();
+
+	cJSON_AddItemToObject(json, "audio_encoding", parse(o.audio_encoding()));
+	cJSON_AddItemToObject(json, "sample_rate_hertz", cJSON_CreateNumber(o.sample_rate_hertz()));
+	cJSON_AddItemToObject(json, "synthesize_speech_config", parse(o.synthesize_speech_config()));
+
+	return json;
+}
+
+cJSON* GRPCParser::parse(const google::rpc::Status& o) {
+	cJSON * json = cJSON_CreateObject();
+
+	cJSON_AddItemToObject(json, "code", cJSON_CreateNumber(o.code()));
+	cJSON_AddItemToObject(json, "message", cJSON_CreateString(o.message().c_str()));
+
+	return json;
+}
+
+cJSON* GRPCParser::parse(const Value& value) {
+	cJSON* json = NULL;
+
+	switch (value.kind_case()) {
+		case Value::KindCase::kNullValue:
+			json = cJSON_CreateNull();
+			break;
+
+		case Value::KindCase::kNumberValue:
+			json = cJSON_CreateNumber(value.number_value());
+			break;
+
+		case Value::KindCase::kStringValue:
+			json = cJSON_CreateString(value.string_value().c_str());
+			break;
+
+		case Value::KindCase::kBoolValue:
+			json = cJSON_CreateBool(value.bool_value());
+			break;
+
+		case Value::KindCase::kStructValue:
+			json = parse(value.struct_value());
+			break;
+
+		case Value::KindCase::kListValue:
+			{
+				const ListValue& list = value.list_value();
+				json = cJSON_CreateArray();
+				for (int i = 0; i < list.values_size(); i++) {
+					const Value& val = list.values(i);
+					cJSON_AddItemToArray(json, parse(val));
+				}
+			} 
+			break;
+	}
+
+	return json;
+}
+
+cJSON* GRPCParser::parse(const Struct& rpcStruct) {
+	cJSON* json = cJSON_CreateObject();
+
+	for (StructIterator_t it = rpcStruct.fields().begin(); it != rpcStruct.fields().end(); it++) {
+		const std::string& key = it->first;
+		const Value& value = it->second;
+		cJSON_AddItemToObject(json, key.c_str(), parse(value));
+	}
+	return json;
+}
+
+cJSON* GRPCParser::parse(const Intent_Parameter& param) {
+  cJSON * json = cJSON_CreateObject();
+
+  cJSON_AddItemToObject(json, "id", cJSON_CreateString(param.id().c_str()));
+  cJSON_AddItemToObject(json, "entity_type", cJSON_CreateString(param.entity_type().c_str()));
+  cJSON_AddItemToObject(json, "is_list", cJSON_CreateBool(param.is_list()));
+  cJSON_AddItemToObject(json, "redact", cJSON_CreateBool(param.redact()));
+
+  return json;
+}
+
+cJSON* GRPCParser::parse(const Intent& o) {
+    cJSON * json = cJSON_CreateObject();
+
+	cJSON_AddItemToObject(json, "name", cJSON_CreateString(o.name().c_str()));
+	cJSON_AddItemToObject(json, "display_name", cJSON_CreateString(o.display_name().c_str()));
+
+  cJSON* params = cJSON_CreateArray();
+  for (int i = 0; i < o.parameters_size(); i++) {
+    cJSON_AddItemToArray(params, parse(o.parameters(i)));
+  }
+  cJSON_AddItemToObject(json, "parameters", params);
+	cJSON_AddItemToObject(json, "priority", cJSON_CreateNumber(o.priority()));
+	cJSON_AddItemToObject(json, "is_fallback", cJSON_CreateBool(o.is_fallback()));
+	return json;
+}
+
+cJSON* GRPCParser::parse(const Match& o) {
+  cJSON * json = cJSON_CreateObject();
+
+  cJSON_AddItemToObject(json, "resolved_input", cJSON_CreateString(o.resolved_input().c_str()));
+  cJSON_AddItemToObject(json, "event", cJSON_CreateString(o.event().c_str()));
+  if (o.has_intent()) cJSON_AddItemToObject(json, "intent", parse(o.intent()));
+  cJSON_AddItemToObject(json, "parameters", parse(o.parameters()));
+  cJSON_AddItemToObject(json, "match_type", cJSON_CreateString(Match_MatchType_Name(o.match_type()).c_str()));
+  cJSON_AddItemToObject(json, "confidence", cJSON_CreateNumber(o.confidence()));
+
+  return json;
+}
+
+cJSON* GRPCParser::parse(const AdvancedSettings_SpeechSettings& o) {
+  cJSON * json = cJSON_CreateObject();
+
+  if (o.has_no_speech_timeout()) {
+    double total_seconds = o.no_speech_timeout().seconds() + o.no_speech_timeout().nanos() / 1e9;
+    cJSON_AddItemToObject(json, "no_speech_timeout", cJSON_CreateNumber(total_seconds));
+  }
+  cJSON_AddItemToObject(json, "endpointer_sensitivity", cJSON_CreateNumber(o.endpointer_sensitivity()));
+  cJSON_AddItemToObject(json, "use_timeout_based_endpointing", cJSON_CreateBool(o.use_timeout_based_endpointing()));
+
+  return json;
+}
+
+cJSON* GRPCParser::parse(const AdvancedSettings_DtmfSettings& o) {
+  cJSON * json = cJSON_CreateObject();
+
+  double interdigit_timeout_duration = o.interdigit_timeout_duration().seconds() + o.interdigit_timeout_duration().nanos() / 1e9;
+  double endpointing_timeout_duration = o.endpointing_timeout_duration().seconds() + o.endpointing_timeout_duration().nanos() / 1e9;
+
+  cJSON_AddItemToObject(json, "interdigit_timeout_duration", cJSON_CreateNumber(interdigit_timeout_duration));
+  cJSON_AddItemToObject(json, "endpointing_timeout_duration", cJSON_CreateNumber(endpointing_timeout_duration));
+  cJSON_AddItemToObject(json, "enabled", cJSON_CreateBool(o.enabled()));
+  cJSON_AddItemToObject(json, "max_digits", cJSON_CreateNumber(o.max_digits()));
+  
+  return json;
+
+}
+cJSON* GRPCParser::parse(const AdvancedSettings_LoggingSettings& o) {
+  cJSON * json = cJSON_CreateObject();
+
+  cJSON_AddItemToObject(json, "enable_stackdriver_logging", cJSON_CreateBool(o.enable_stackdriver_logging()));
+  cJSON_AddItemToObject(json, "enable_interaction_logging", cJSON_CreateBool(o.enable_interaction_logging()));
+  cJSON_AddItemToObject(json, "enable_consent_based_redaction", cJSON_CreateBool(o.enable_consent_based_redaction()));
+
+  return json;
+}
+
+cJSON* GRPCParser::parse(const SentimentAnalysisResult& o) {
+    cJSON * json = cJSON_CreateObject();
+
+    cJSON_AddNumberToObject(json, "score", o.score());
+    cJSON_AddNumberToObject(json, "magnitude", o.magnitude());
+
+  return json;
+}
+
+
+cJSON* GRPCParser::parse(const GcsDestination& o) {
+  cJSON * json = cJSON_CreateObject();
+
+  cJSON_AddItemToObject(json, "uri", cJSON_CreateString(o.uri().c_str()));
+
+  return json;
+}
+
+cJSON* GRPCParser::parse(const AdvancedSettings& o) {
+  cJSON * json = cJSON_CreateObject();
+
+  if (o.has_audio_export_gcs_destination()) cJSON_AddItemToObject(json, "audio_export_gcs_destination", parse(o.audio_export_gcs_destination()));  
+  if (o.has_speech_settings()) cJSON_AddItemToObject(json, "speech_settings", parse(o.speech_settings()));
+  if (o.has_dtmf_settings()) cJSON_AddItemToObject(json, "dtmf_settings", parse(o.dtmf_settings()));
+  if (o.has_logging_settings()) cJSON_AddItemToObject(json, "logging_settings", parse(o.logging_settings()));
+
+  return json;
+}
+
+
+cJSON* GRPCParser::parse(const ResponseMessage_Text& o) {
+  cJSON* t = cJSON_CreateArray();
+  for (int i = 0; i < o.text_size(); i++) {
+    cJSON * json = cJSON_CreateObject();
+    cJSON_AddItemToObject(json, "text", cJSON_CreateString(o.text(i).c_str()));
+    cJSON_AddItemToObject(json, "allow_playback_interruption", cJSON_CreateBool(o.allow_playback_interruption()));
+    cJSON_AddItemToArray(t, json);
+  }
+  return t;
+}
+
+cJSON* GRPCParser::parse(const ResponseMessage_ConversationSuccess& o) {
+  cJSON * json = cJSON_CreateObject();
+
+  cJSON_AddItemToObject(json, "metadata", parse(o.metadata()));
+
+  return json;
+}
+
+cJSON* GRPCParser::parse(const ResponseMessage_OutputAudioText& o) {
+  cJSON * json = cJSON_CreateObject();
+
+  if (o.has_text()) cJSON_AddItemToObject(json, "text", cJSON_CreateString(o.text().c_str()));
+  if (o.has_ssml()) cJSON_AddItemToObject(json, "ssml", cJSON_CreateString(o.ssml().c_str()));
+
+  return json;
+}
+
+cJSON* GRPCParser::parse(const ResponseMessage_LiveAgentHandoff& o) {
+  cJSON * json = cJSON_CreateObject();
+
+  cJSON_AddItemToObject(json, "metadata", parse(o.metadata()));
+
+  return json;
+}
+
+cJSON* GRPCParser::parse(const ResponseMessage_EndInteraction& o) {
+  cJSON * json = cJSON_CreateObject();
+
+  // TODOL: need to research this more
+  return json;
+}
+
+cJSON* GRPCParser::parse(const ResponseMessage_PlayAudio& o) {
+  cJSON * json = cJSON_CreateObject();
+
+  cJSON_AddItemToObject(json, "audio_uri", cJSON_CreateString(o.audio_uri().c_str()));
+  cJSON_AddItemToObject(json, "allow_playback_interruption", cJSON_CreateBool(o.allow_playback_interruption()));
+
+  return json;
+}
+
+cJSON* GRPCParser::parse(const ResponseMessage_MixedAudio& o) {
+  cJSON * json = cJSON_CreateArray();
+
+  for (int i = 0; i < o.segments_size(); i++) {
+    cJSON * segment = cJSON_CreateObject();
+    if (o.segments(i).has_audio()) cJSON_AddItemToObject(segment, "audio", cJSON_CreateString(o.segments(i).audio().c_str()));
+    else if (o.segments(i).has_uri()) cJSON_AddItemToObject(segment, "uri", cJSON_CreateString(o.segments(i).uri().c_str()));
+    cJSON_AddItemToObject(segment, "allow_playback_interruption", cJSON_CreateBool(o.segments(i).allow_playback_interruption()));
+  }
+  return json;
+}
+
+cJSON* GRPCParser::parse(const ResponseMessage_TelephonyTransferCall& o) {
+  cJSON * json = cJSON_CreateObject();
+
+  cJSON_AddItemToObject(json, "phone_number", cJSON_CreateString(o.phone_number().c_str()));
+
+  return json;
+}
+
+
+cJSON* GRPCParser::parse(const ResponseMessage& o) {
+  cJSON * json = cJSON_CreateObject();
+
+  cJSON_AddItemToObject(json, "response_type", cJSON_CreateString(ResponseMessage_ResponseType_Name(o.response_type()).c_str()));
+  cJSON_AddItemToObject(json, "text", parse(o.text()));
+  cJSON_AddItemToObject(json, "payload", parse(o.payload()));
+  cJSON_AddItemToObject(json, "conversation_success", parse(o.conversation_success()));
+  cJSON_AddItemToObject(json, "output_audio_text", parse(o.output_audio_text()));
+  cJSON_AddItemToObject(json, "live_agent_handoff", parse(o.live_agent_handoff()));
+  cJSON_AddItemToObject(json, "end_interaction", parse(o.end_interaction()));
+  cJSON_AddItemToObject(json, "play_audio", parse(o.play_audio()));
+  cJSON_AddItemToObject(json, "mixed_audio", parse(o.mixed_audio()));
+  cJSON_AddItemToObject(json, "telephony_transfer_call", parse(o.telephony_transfer_call()));
+
+  return json;
+}
+
+cJSON* GRPCParser::parse(const QueryResult& qr) {
+    cJSON * json = cJSON_CreateObject();
+
+    // one-of
+    if (qr.has_text()) cJSON_AddItemToObject(json, "text", cJSON_CreateString(qr.text().c_str()));
+    else if (qr.has_trigger_intent()) cJSON_AddItemToObject(json, "trigger_intent", cJSON_CreateString(qr.trigger_intent().c_str()));
+    else if (qr.has_transcript()) cJSON_AddItemToObject(json, "transcript", cJSON_CreateString(qr.transcript().c_str()));
+    else if (qr.has_trigger_event()) cJSON_AddItemToObject(json, "trigger_event", cJSON_CreateString(qr.trigger_event().c_str()));
+
+    if (qr.has_dtmf()) cJSON_AddItemToObject(json, "dtmf", parse(qr.dtmf()));
+    cJSON_AddItemToObject(json, "language_code", cJSON_CreateString(qr.language_code().c_str()));
+    cJSON_AddItemToObject(json, "parameters", parse(qr.parameters()));
+
+    cJSON* rms = cJSON_CreateArray();
+    for (int i = 0; i < qr.response_messages_size(); i++) {
+      cJSON_AddItemToArray(rms, parse(qr.response_messages(i)));
+    }
+    cJSON_AddItemToObject(json, "response_messages", rms);
+
+    cJSON* whids = cJSON_CreateArray();
+    for (int i = 0; i < qr.webhook_ids_size(); i++) {
+      cJSON_AddItemToArray(whids, cJSON_CreateString(qr.webhook_ids(i).c_str()));
+    }
+    cJSON_AddItemToObject(json, "webhook_ids", whids);
+
+    cJSON* whDisplayNames = cJSON_CreateArray();
+    for (int i = 0; i < qr.webhook_display_names_size(); i++) {
+      cJSON_AddItemToArray(whDisplayNames, cJSON_CreateString(qr.webhook_display_names(i).c_str()));
+    }
+    cJSON_AddItemToObject(json, "webhook_display_names", whDisplayNames);
+
+    cJSON* whLatencies = cJSON_CreateArray();
+    for (int i = 0; i < qr.webhook_latencies_size(); i++) {
+      double total_seconds = qr.webhook_latencies(i).seconds() + qr.webhook_latencies(i).nanos() / 1e9;
+      cJSON_AddItemToArray(whLatencies, cJSON_CreateNumber(total_seconds));
+    }
+
+    cJSON* whTags = cJSON_CreateArray();
+    for (int i = 0; i < qr.webhook_tags_size(); i++) {
+      cJSON_AddItemToArray(whids, cJSON_CreateString(qr.webhook_tags(i).c_str()));
+    }
+
+    cJSON* whStatuses = cJSON_CreateArray();
+    for (int i = 0; i < qr.webhook_statuses_size(); i++) {
+      cJSON_AddItemToArray(whStatuses, parse(qr.webhook_statuses(i)));
+    }
+
+    cJSON* whPayloads = cJSON_CreateArray();
+    for (int i = 0; i < qr.webhook_payloads_size(); i++) {
+      cJSON_AddItemToArray(whPayloads, parse(qr.webhook_payloads(i)));
+    }
+
+    //if (qr.has_current_page()) cJSON_AddItemToObject(json, "current_page", parse(qr.current_page());
+    //if (qr.has_current_flow()) cJSON_AddItemToObject(json, "current_flow", parse(qr.current_flow());
+    if (qr.has_match()) cJSON_AddItemToObject(json, "match", parse(qr.match()));
+    if (qr.has_diagnostic_info()) cJSON_AddItemToObject(json, "diagnostic_info", parse(qr.diagnostic_info()));
+    if (qr.has_sentiment_analysis_result()) cJSON_AddItemToObject(json, "sentiment_analysis_result", parse(qr.sentiment_analysis_result()));
+    if (qr.has_advanced_settings()) cJSON_AddItemToObject(json, "advanced_settings", parse(qr.advanced_settings()));
+
+    cJSON_AddItemToObject(json, "allow_answer_feedback", cJSON_CreateBool(qr.allow_answer_feedback()));
+
+    // skipping DataStoreConnectionSignals for now, it doesn't seem consequential
+
+    return json;
+}
+cJSON* GRPCParser::parse(const StreamingRecognitionResult_MessageType& o) {
+	return cJSON_CreateString(StreamingRecognitionResult_MessageType_Name(o).c_str());
+}
+
+cJSON* GRPCParser::parse(const DtmfInput& o) {
+  cJSON * json = cJSON_CreateObject();
+
+  cJSON_AddItemToObject(json, "digits", cJSON_CreateString(o.digits().c_str()));
+  cJSON_AddItemToObject(json, "finish_digit", cJSON_CreateString(o.finish_digit().c_str()));
+
+  return json;
+}
+
+cJSON* GRPCParser::parse(const StreamingRecognitionResult& o) {
+    cJSON * json = cJSON_CreateObject();
+
+    cJSON_AddItemToObject(json, "message_type", parse(o.message_type()));
+    cJSON_AddItemToObject(json, "transcript", cJSON_CreateString(o.transcript().c_str()));
+    cJSON_AddItemToObject(json, "is_final", cJSON_CreateBool(o.is_final()));
+    cJSON_AddItemToObject(json, "confidence", cJSON_CreateNumber(o.confidence()));
+    cJSON_AddItemToObject(json, "stability", cJSON_CreateNumber(o.stability()));
+    cJSON_AddItemToObject(json, "language_code", cJSON_CreateString(o.language_code().c_str()));
+
+    // TODO: we also have SpeechWordInfo if we want to dive into that
+
+    return json;
+}
+
+cJSON* GRPCParser::parse(const DetectIntentResponse& o)  {
+    cJSON * json = cJSON_CreateObject();
+
+    cJSON_AddItemToObject(json, "response_id", cJSON_CreateString(o.response_id().c_str()));
+    cJSON_AddItemToObject(json, "query_result", parse(o.query_result()));
+    cJSON_AddItemToObject(json, "output_audio", cJSON_CreateString(o.output_audio().c_str()));
+    cJSON_AddItemToObject(json, "output_audio_config", parse(o.output_audio_config()));
+    cJSON_AddItemToObject(json, "response_type", cJSON_CreateString(DetectIntentResponse_ResponseType_Name(o.response_type()).c_str()));
+    cJSON_AddItemToObject(json, "allow_cancellation", cJSON_CreateBool(o.allow_cancellation()));
+
+    return json;
+}

mod_dialogflow_cx/parser.h

@@ -0,0 +1,107 @@
+#ifndef __PARSER_H__
+#define __PARSER_H__
+
+#include <switch_json.h>
+#include <grpc++/grpc++.h>
+#include "google/cloud/dialogflow/cx/v3/session.grpc.pb.h"
+
+using google::cloud::dialogflow::cx::v3::Sessions;
+using google::cloud::dialogflow::cx::v3::StreamingDetectIntentRequest;
+using google::cloud::dialogflow::cx::v3::StreamingDetectIntentResponse;
+using google::cloud::dialogflow::cx::v3::DetectIntentResponse;
+using google::cloud::dialogflow::cx::v3::ResponseMessage;
+using google::cloud::dialogflow::cx::v3::ResponseMessage_Text;
+using google::cloud::dialogflow::cx::v3::ResponseMessage_LiveAgentHandoff;
+using google::cloud::dialogflow::cx::v3::ResponseMessage_ConversationSuccess;
+using google::cloud::dialogflow::cx::v3::ResponseMessage_OutputAudioText;
+using google::cloud::dialogflow::cx::v3::ResponseMessage_EndInteraction;
+using google::cloud::dialogflow::cx::v3::ResponseMessage_PlayAudio;
+using google::cloud::dialogflow::cx::v3::ResponseMessage_MixedAudio;
+using google::cloud::dialogflow::cx::v3::ResponseMessage_TelephonyTransferCall;
+using google::cloud::dialogflow::cx::v3::ResponseMessage_KnowledgeInfoCard;
+using google::cloud::dialogflow::cx::v3::ResponseMessage_ResponseType;
+using google::cloud::dialogflow::cx::v3::AudioEncoding;
+using google::cloud::dialogflow::cx::v3::InputAudioConfig;
+using google::cloud::dialogflow::cx::v3::OutputAudioConfig;
+using google::cloud::dialogflow::cx::v3::SynthesizeSpeechConfig;
+using google::cloud::dialogflow::cx::v3::VoiceSelectionParams;
+using google::cloud::dialogflow::cx::v3::Intent;
+using google::cloud::dialogflow::cx::v3::Intent_Parameter;
+using google::cloud::dialogflow::cx::v3::QueryInput;
+using google::cloud::dialogflow::cx::v3::QueryResult;
+using google::cloud::dialogflow::cx::v3::Match;
+using google::cloud::dialogflow::cx::v3::Match_MatchType_Name;
+using google::cloud::dialogflow::cx::v3::AdvancedSettings;
+using google::cloud::dialogflow::cx::v3::AdvancedSettings_SpeechSettings;
+using google::cloud::dialogflow::cx::v3::AdvancedSettings_DtmfSettings;
+using google::cloud::dialogflow::cx::v3::AdvancedSettings_LoggingSettings;
+using google::cloud::dialogflow::cx::v3::StreamingRecognitionResult;
+using google::cloud::dialogflow::cx::v3::StreamingRecognitionResult_MessageType;
+using google::cloud::dialogflow::cx::v3::StreamingRecognitionResult_MessageType_Name;
+using google::cloud::dialogflow::cx::v3::EventInput;
+using google::cloud::dialogflow::cx::v3::OutputAudioEncoding;
+using google::cloud::dialogflow::cx::v3::OutputAudioEncoding_Name;
+using google::cloud::dialogflow::cx::v3::SentimentAnalysisResult;
+using google::cloud::dialogflow::cx::v3::GcsDestination;
+using google::cloud::dialogflow::cx::v3::DtmfInput;
+using google::protobuf::RepeatedPtrField;
+using google::rpc::Status;
+using google::protobuf::Struct;
+using google::protobuf::Value;
+using google::protobuf::ListValue;
+
+typedef google::protobuf::Map< std::string, Value >::const_iterator StructIterator_t;
+
+class GRPCParser {
+public:
+    GRPCParser(switch_core_session_t *session) : m_session(session) {}
+    ~GRPCParser() {}
+
+    template <typename T> cJSON* parseCollection(const RepeatedPtrField<T> coll) ;
+    
+    cJSON* parse(const StreamingDetectIntentResponse& response) ;
+    const std::string& parseAudio(const StreamingDetectIntentResponse& response);
+
+
+    cJSON* parse(const DetectIntentResponse& o) ;
+    cJSON* parse(const ResponseMessage& o) ;
+    cJSON* parse(const ResponseMessage_Text& o) ;
+    cJSON* parse(const ResponseMessage_ResponseType& o) ;
+    cJSON* parse(const ResponseMessage_LiveAgentHandoff& o) ;
+    cJSON* parse(const ResponseMessage_ConversationSuccess& o) ;
+    cJSON* parse(const ResponseMessage_OutputAudioText& o) ;
+    cJSON* parse(const ResponseMessage_EndInteraction& o) ;
+    cJSON* parse(const ResponseMessage_PlayAudio& o) ;
+    cJSON* parse(const ResponseMessage_MixedAudio& o) ;
+    cJSON* parse(const ResponseMessage_TelephonyTransferCall& o) ;
+    cJSON* parse(const ResponseMessage_KnowledgeInfoCard& o) ;
+    cJSON* parse(const Match& o) ;
+    cJSON* parse(const AdvancedSettings& o) ;
+    cJSON* parse(const AdvancedSettings_SpeechSettings& o) ;
+    cJSON* parse(const AdvancedSettings_DtmfSettings& o) ;
+    cJSON* parse(const AdvancedSettings_LoggingSettings& o) ;
+    cJSON* parse(const SynthesizeSpeechConfig& o) ;
+    cJSON* parse(const OutputAudioEncoding& o) ;
+    cJSON* parse(const OutputAudioConfig& o) ;
+    cJSON* parse(const VoiceSelectionParams& o) ;
+    cJSON* parse(const Intent& o) ;
+    cJSON* parse(const Intent_Parameter& o) ;
+    cJSON* parse(const QueryInput& o) ;
+    cJSON* parse(const DtmfInput& o) ;
+    cJSON* parse(const google::rpc::Status& o) ;
+    cJSON* parse(const Value& value) ;
+    cJSON* parse(const Struct& rpcStruct) ;
+    cJSON* parse(const std::string& val) ;
+    cJSON* parse(const SentimentAnalysisResult& o) ;
+    cJSON* parse(const StreamingRecognitionResult_MessageType& o) ;
+    cJSON* parse(const StreamingRecognitionResult& o) ;
+    cJSON* parse(const GcsDestination& o) ;
+    cJSON* parse(const QueryResult& qr);
+
+
+private:
+    switch_core_session_t *m_session;
+} ;
+
+
+#endif

mod_dub/mod_dub.c

@@ -271,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_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/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;
-
-	if (SWITCH_STATUS_FALSE == jb_transcribe_session_init(session, responseHandler, samples_per_second, flags & SMBF_STEREO ? 2 : 1, lang, interim, bugname, &pUserData)) {
+	var = switch_channel_get_variable(channel, "JAMBONZ_STT_SAMPLING");
+	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;
 	}