Playht delete circular buffer with mutex check (#123 )

Signed-off-by: root <root@af6633a5cfe8> Co-authored-by: root <root@af6633a5cfe8>
Update README.md (#9 )
2026-01-25 02:08:27 +00:00 · 2024-10-05 10:02:07 -04:00 · 2024-10-03 08:06:32 -04:00 · 2024-10-03 08:04:40 -04:00 · 2024-09-30 14:44:41 -04:00 · 2024-09-24 09:51:06 -04:00
35 changed files with 4495 additions and 1665 deletions
@@ -227,20 +227,21 @@ int AudioPipe::lws_callback(struct lws *wsi,
        // check for text frames to send
        {
          std::lock_guard<std::mutex> lk(ap->m_text_mutex);
-          if (ap->m_metadata.length() > 0) {
-            uint8_t buf[ap->m_metadata.length() + LWS_PRE];
-            memcpy(buf + LWS_PRE, ap->m_metadata.c_str(), ap->m_metadata.length());
-            int n = ap->m_metadata.length();
+          if (!ap->m_metadata_list.empty()) {
+            const std::string& message = ap->m_metadata_list.front();
+            uint8_t buf[message.length() + LWS_PRE];
+            memcpy(buf + LWS_PRE, message.c_str(), message.length());
+            int n = message.length();
            int m = lws_write(wsi, buf + LWS_PRE, n, LWS_WRITE_TEXT);
-            ap->m_metadata.clear();
+
            if (m < n) {
-              return -1;
+              return -1; // Failed to send the full message
            }

-            // there may be audio data, but only one write per writeable event
-            // get it next time
+            // Remove the message that was successfully sent
+            ap->m_metadata_list.pop_front();
+            // Request another writable event if there are more messages
            lws_callback_on_writable(wsi);
-
            return 0;
          }
        }
@@ -529,7 +530,7 @@ void AudioPipe::bufferForSending(const char* text) {
  if (m_state != LWS_CLIENT_CONNECTED) return;
  {
    std::lock_guard<std::mutex> lk(m_text_mutex);
-    m_metadata.append(text);
+    m_metadata_list.emplace_back(text);
  }
  addPendingWrite(this);
 }
@@ -130,7 +130,7 @@ namespace drachtio {
    std::string m_bugname;
    unsigned int m_port;
    std::string m_path;
-    std::string m_metadata;
+    std::list<std::string> m_metadata_list;
    std::mutex m_text_mutex;
    std::mutex m_audio_mutex;
    int m_sslFlags;
@@ -27,6 +27,8 @@ 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 (16384)
+#define AUDIO_MARKER 0xFFFF
+#define MAX_MARKS (30)

 namespace {
  static const char *requestedBufferSecs = std::getenv("MOD_AUDIO_FORK_BUFFER_SECS");
@@ -38,6 +40,15 @@ namespace {
  static unsigned int idxCallCount = 0;
  static uint32_t playCount = 0;

+  static bool markCountExceeded(private_t* tech_pvt) {
+    if (nullptr != tech_pvt->pVecMarksInUse) {
+      std::deque<std::string>* pVecMarksInUse = static_cast<std::deque<std::string>*>(tech_pvt->pVecMarksInUse);
+      std::deque<std::string>* pVecMarksInInventory = static_cast<std::deque<std::string>*>(tech_pvt->pVecMarksInInventory);
+      return pVecMarksInUse->size()+ pVecMarksInInventory->size() >= MAX_MARKS;
+    }
+    return false;
+  }
+
  switch_status_t processIncomingBinary(private_t* tech_pvt, switch_core_session_t* session, const char* message, size_t dataLength) {
    std::vector<uint8_t> data;

@@ -65,12 +76,30 @@ namespace {
    // Access the prebuffer
    CircularBuffer_t* cBuffer = static_cast<CircularBuffer_t*>(tech_pvt->streamingPreBuffer);

+    int numMarkers = 0;
+    std::deque<std::string>* pVecMarksInInventory = nullptr;
+    std::deque<std::string>* pVecMarksInUse = nullptr;
+    if (nullptr != tech_pvt->pVecMarksInInventory) {
+      pVecMarksInInventory = static_cast<std::deque<std::string>*>(tech_pvt->pVecMarksInInventory);
+      pVecMarksInUse = static_cast<std::deque<std::string>*>(tech_pvt->pVecMarksInUse);
+      numMarkers = pVecMarksInInventory->size();
+
+      // move inventory to in-use
+      pVecMarksInUse->insert(pVecMarksInUse->end(), pVecMarksInInventory->begin(), pVecMarksInInventory->end());
+      pVecMarksInInventory->clear();
+    }
+  
    // Ensure the prebuffer has enough capacity
-    if (cBuffer->capacity() - cBuffer->size() < numSamples) {
-        size_t newCapacity = cBuffer->size() + std::max(numSamples, (size_t)BUFFER_GROW_SIZE);
+    if (cBuffer->capacity() - cBuffer->size() < numSamples + numMarkers) {
+        size_t newCapacity = cBuffer->size() + std::max(numSamples + numMarkers, (size_t)BUFFER_GROW_SIZE);
        cBuffer->set_capacity(newCapacity);
    }

+    // prepend any markers
+    while (numMarkers-- > 0) {
+      cBuffer->push_back(AUDIO_MARKER);
+    }
+
    // Append the data to the prebuffer
    cBuffer->insert(cBuffer->end(), data_uint16, data_uint16 + numSamples);
    //switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "Appended %zu 16-bit samples to the prebuffer.\n", numSamples);
@@ -269,6 +298,63 @@ namespace {
        // this will dump buffered incoming audio
        tech_pvt->clear_bidirectional_audio_buffer = true;
      }
+      else if (0 == type.compare("mark")) {
+        cJSON* data = cJSON_GetObjectItem(json, "data");
+        if (data) {
+          cJSON* name = cJSON_GetObjectItem(data, "name");
+          if (cJSON_IsString(name) && name->valuestring) {
+            if (markCountExceeded(tech_pvt)) {
+              switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "(%u) processIncomingMessage - mark count exceeded, discarding mark %s\n", tech_pvt->id, cJSON_GetStringValue(name));
+            }
+            else {
+              if (nullptr == tech_pvt->pVecMarksInInventory) {
+                tech_pvt->pVecMarksInInventory = static_cast<void *>(new std::deque<std::string>());
+                tech_pvt->pVecMarksInUse = static_cast<void *>(new std::deque<std::string>());
+                tech_pvt->pVecMarksCleared = static_cast<void *>(new std::deque<std::string>());
+              }
+              std::deque<std::string>* pVec = static_cast<std::deque<std::string>*>(tech_pvt->pVecMarksInInventory);
+              pVec->push_back(name->valuestring);
+            }
+          }
+        }
+      }
+      else if (0 == type.compare("clearMarks")) {
+        switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "(%u) processIncomingMessage - received clearMarks\n", tech_pvt->id);
+        if (nullptr != tech_pvt->pVecMarksInInventory) {
+          std::deque<std::string>* pVecMarksInInventory = static_cast<std::deque<std::string>*>(tech_pvt->pVecMarksInInventory);
+          std::deque<std::string>* pVecMarksInUse = static_cast<std::deque<std::string>*>(tech_pvt->pVecMarksInUse);
+          std::deque<std::string>* pVecMarksCleared = static_cast<std::deque<std::string>*>(tech_pvt->pVecMarksCleared);
+          pVecMarksCleared->insert(pVecMarksCleared->end(), pVecMarksInUse->begin(), pVecMarksInUse->end());
+          pVecMarksCleared->insert(pVecMarksCleared->end(), pVecMarksInInventory->begin(), pVecMarksInInventory->end());
+          pVecMarksInInventory->clear();
+          pVecMarksInUse->clear();
+        }
+      }
+      else if (0 == type.compare("transcription")) {
+        char* jsonString = cJSON_PrintUnformatted(jsonData);
+        tech_pvt->responseHandler(session, EVENT_TRANSCRIPTION, jsonString);
+        free(jsonString);        
+      }
+      else if (0 == type.compare("transfer")) {
+        char* jsonString = cJSON_PrintUnformatted(jsonData);
+        tech_pvt->responseHandler(session, EVENT_TRANSFER, jsonString);
+        free(jsonString);                
+      }
+      else if (0 == type.compare("disconnect")) {
+        char* jsonString = cJSON_PrintUnformatted(jsonData);
+        tech_pvt->responseHandler(session, EVENT_DISCONNECT, jsonString);
+        free(jsonString);        
+      }
+      else if (0 == type.compare("error")) {
+        char* jsonString = cJSON_PrintUnformatted(jsonData);
+        tech_pvt->responseHandler(session, EVENT_ERROR, jsonString);
+        free(jsonString);        
+      }
+      else if (0 == type.compare("json")) {
+        char* jsonString = cJSON_PrintUnformatted(json);
+        tech_pvt->responseHandler(session, EVENT_JSON, jsonString);
+        free(jsonString);
+      }
      else if (0 == type.compare("transcription")) {
        char* jsonString = cJSON_PrintUnformatted(jsonData);
        tech_pvt->responseHandler(session, EVENT_TRANSCRIPTION, jsonString);
@@ -400,6 +486,9 @@ namespace {
    }
    tech_pvt->streamingPreBufSize = 320 * tech_pvt->downscale_factor * 4; // min 80ms prebuffer
    tech_pvt->streamingPreBuffer = (void *) new CircularBuffer_t(8192);
+    tech_pvt->pVecMarksInInventory = nullptr;
+    tech_pvt->pVecMarksInUse = nullptr;
+    tech_pvt->pVecMarksCleared = nullptr;

    strncpy(tech_pvt->bugname, bugname, MAX_BUG_LEN);
    if (metadata) strncpy(tech_pvt->initialMetadata, metadata, MAX_METADATA_LEN);
@@ -467,6 +556,29 @@ namespace {
      delete cBuffer;
      tech_pvt->streamingPreBuffer = nullptr;
    }
+
+    if (tech_pvt->pVecMarksInInventory) {
+      delete static_cast<std::deque<std::string>*>(tech_pvt->pVecMarksInInventory);
+      tech_pvt->pVecMarksInInventory = nullptr;
+      delete static_cast<std::deque<std::string>*>(tech_pvt->pVecMarksInUse);
+      tech_pvt->pVecMarksInUse = nullptr;
+      delete static_cast<std::deque<std::string>*>(tech_pvt->pVecMarksCleared);
+      tech_pvt->pVecMarksCleared = nullptr;
+    }
+  }
+
+  static void send_mark_event(private_t* tech_pvt, const char* name, int cleared = false) {
+    drachtio::AudioPipe *pAudioPipe = static_cast<drachtio::AudioPipe *>(tech_pvt->pAudioPipe);
+    std::ostringstream json;
+    json << "{\"type\": \"mark\", \"data\": {\"name\":\"" << name << "\", ";
+    if (cleared) json << "\"event\": \"cleared\"}}";
+    else json << "\"event\": \"playout\"}}";
+
+    if (pAudioPipe) {
+      std::string str = json.str();
+      pAudioPipe->bufferForSending(str.c_str());
+      switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "(%u) send_mark_event: %s\n", tech_pvt->id, str.c_str());
+    }
  }

  void lws_logger(int level, const char *line) {
@@ -826,6 +938,28 @@ extern "C" {
        switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "(%u) dub_speech_frame - clearing buffer\n", tech_pvt->id); 
        cBuffer->clear();
        tech_pvt->clear_bidirectional_audio_buffer = false;
+
+        // send "mark" event for any queued markers
+        if (nullptr != tech_pvt->pVecMarksInInventory) {
+          std::deque<std::string>* pVecMarksInInventory = static_cast<std::deque<std::string>*>(tech_pvt->pVecMarksInInventory);
+          std::deque<std::string>* pVecMarksInUse = static_cast<std::deque<std::string>*>(tech_pvt->pVecMarksInUse);
+          if (pVecMarksInInventory->size() + pVecMarksInUse->size() > 0) {
+            std::deque<std::string> vec = *pVecMarksInUse;
+            vec.insert(vec.end(), pVecMarksInInventory->begin(), pVecMarksInInventory->end());
+            for (auto it = vec.begin(); it != vec.end(); ++it) {
+              switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "(%u) dub_speech_frame - Marker %s cleared\n",
+                  tech_pvt->id, it->c_str());
+              send_mark_event(tech_pvt, it->c_str(), true);
+            }
+
+            // put the "in-use" ones into the "cleared" queue so we dont notify again when they eventually come through
+            std::deque<std::string>* pVecMarksCleared = static_cast<std::deque<std::string>*>(tech_pvt->pVecMarksCleared);
+            pVecMarksCleared->insert(pVecMarksCleared->end(), pVecMarksInUse->begin(), pVecMarksInUse->end());
+
+            pVecMarksInUse->clear();
+            pVecMarksInInventory->clear();
+          }
+        }
      }
      else {
        switch_frame_t* rframe = switch_core_media_bug_get_write_replace_frame(bug);
@@ -841,11 +975,72 @@ extern "C" {

        //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);
+        bool hasMarkers = false;
+        std::deque<std::string>* pVecInventory = nullptr;
+        std::deque<std::string>* pVecInUse = nullptr;
+        std::deque<std::string>* pVecCleared = nullptr;
+        if (tech_pvt->pVecMarksInInventory && tech_pvt->pVecMarksInUse && tech_pvt->pVecMarksCleared) {
+          pVecInventory = static_cast<std::deque<std::string>*>(tech_pvt->pVecMarksInInventory);
+          pVecInUse = static_cast<std::deque<std::string>*>(tech_pvt->pVecMarksInUse);
+          pVecCleared = static_cast<std::deque<std::string>*>(tech_pvt->pVecMarksCleared);
+          hasMarkers = pVecInUse->size() + pVecCleared->size() >  0;
+        }

-        if (samplesToCopy > 0) {
-          vector_add(fp, data, rframe->samples);
+        if (hasMarkers) {
+          /* discard markers and send notifications */
+          auto bufferIter = cBuffer->begin();
+          auto dataIter = data;
+          for (int i = 0; i < samplesToCopy; ++i) {
+            if (*bufferIter == AUDIO_MARKER) {
+              // Marker detected, discard it and send a notice unless it was previously cleared
+              auto * pVec = pVecCleared->size() > 0 ? pVecCleared : pVecInUse;
+              if (!pVec->empty()) {
+                auto name = pVec->front();
+                pVec->pop_front();
+
+                if (pVec == pVecInUse) {
+                  send_mark_event(tech_pvt, name.c_str());
+                  switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "(%u) dub_speech_frame - Marker %s detected in playout\n",
+                    tech_pvt->id, name.c_str());
+                }
+                else {
+                  switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "(%u) dub_speech_frame - Marker %s detected in playout but previously cleared\n",
+                    tech_pvt->id, name.c_str());
+                }
+              }
+            } else {
+              // Copy valid audio samplewhat 
+              *dataIter = *bufferIter;
+              ++dataIter;
+            }
+            ++bufferIter;
+          }
+
+          // Remove the processed samples (including discarded markers) from the buffer
+          cBuffer->erase(cBuffer->begin(), cBuffer->begin() + samplesToCopy);
+
+          // Adjust the number of samples copied to the output frame
+          int validSamplesCopied = std::distance(data, dataIter);
+
+          if (validSamplesCopied > 0) {
+            vector_add(fp, data, validSamplesCopied);
+          }
+        }
+        else {
+          std::copy_n(cBuffer->begin(), samplesToCopy, data);
+          cBuffer->erase(cBuffer->begin(), cBuffer->begin() + samplesToCopy);
+
+          if (samplesToCopy > 0) {
+            vector_add(fp, data, rframe->samples);
+          } else if (pVecInventory && pVecInventory->size()) {
+            // no bidirectional audio to dub but still have some mark in inventory, send them now
+            auto name = pVecInventory->front();
+            pVecInventory->pop_front();
+
+            send_mark_event(tech_pvt, name.c_str());
+            switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "(%u) dub_speech_frame - Marker %s detected in inventory\n",
+              tech_pvt->id, name.c_str());
+          }
        }
        vector_normalize(fp, rframe->samples);

@@ -53,6 +53,11 @@ struct private_data {
  int graceful_shutdown:1;
  char initialMetadata[8192];

+  // for "mark" feature of bidirectional audio
+  void *pVecMarksInInventory;
+  void *pVecMarksInUse;
+  void *pVecMarksCleared;
+
  // bidirectional audio
  void *streamingPlayoutBuffer;
  void *streamingPreBuffer;
@@ -0,0 +1,10 @@
+include $(top_srcdir)/build/modmake.rulesam
+MODNAME=mod_aws_transcribe_ws
+
+mod_LTLIBRARIES = mod_aws_transcribe_ws.la
+mod_aws_transcribe_ws_la_SOURCES  = mod_aws_transcribe_ws.c aws_transcribe_glue.cpp transcribe_manager.cpp audio_pipe.cpp
+mod_aws_transcribe_ws_la_CFLAGS   = $(AM_CFLAGS)
+mod_aws_transcribe_ws_la_CXXFLAGS = $(AM_CXXFLAGS) -std=c++11 -I${switch_srcdir}/libs/aws-sdk-cpp/aws-cpp-sdk-core/include -I${switch_srcdir}/libs/aws-sdk-cpp/aws-cpp-sdk-transcribestreaming/include -I${switch_srcdir}/libs/aws-sdk-cpp/build/.deps/install/include
+
+mod_aws_transcribe_ws_la_LIBADD   = $(switch_builddir)/libfreeswitch.la
+mod_aws_transcribe_ws_la_LDFLAGS  = -avoid-version -module -no-undefined -shared `pkg-config --libs libwebsockets` 
@@ -0,0 +1,58 @@
+# mod_aws_transcribe
+
+A Freeswitch module that generates real-time transcriptions on a Freeswitch channel by using AWS streaming transcription API
+
+## API
+
+### Commands
+The freeswitch module exposes the following API commands:
+
+```
+aws_transcribe <uuid> start <lang-code> [interim]
+```
+Attaches media bug to channel and performs streaming recognize request.
+- `uuid` - unique identifier of Freeswitch channel
+- `lang-code` - a valid AWS [language code](https://docs.aws.amazon.com/transcribe/latest/dg/what-is-transcribe.html) that is supported for streaming transcription
+- `interim` - If the 'interim' keyword is present then both interim and final transcription results will be returned; otherwise only final transcriptions will be returned
+
+```
+aws_transcribe <uuid> stop
+```
+Stop transcription on the channel.
+
+### Authentication
+The plugin will first look for channel variables, then environment variables.  If neither are found, then the default AWS profile on the server will be used.
+
+The names of the channel variables and environment variables are:
+
+| variable | Description |
+| --- | ----------- |
+| AWS_ACCESS_KEY_ID | The Aws access key ID |
+| AWS_SECRET_ACCESS_KEY | The Aws secret access key |
+| AWS_REGION | The Aws region |
+
+
+### Events
+`aws_transcribe::transcription` - returns an interim or final transcription.  The event contains a JSON body describing the transcription result:
+```js
+[
+  {
+    "is_final": true,
+    "alternatives": [{
+      "transcript": "Hello. Can you hear me?"
+    }]
+  }
+]
+```
+
+## 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('aws_transcribe', `${ep.uuid} start en-US interim`);  
+```
+
+## Building
+This uses the AWS websocket api.
+
+## Examples
+[aws_transcribe.js](../../examples/aws_transcribe.js)
@@ -0,0 +1,612 @@
+#include "audio_pipe.hpp"
+#include "transcribe_manager.hpp"
+#include "crc.h"
+
+#include <switch.h>
+
+#include <cassert>
+#include <iostream>
+#include <netinet/in.h>
+
+#include <fstream>
+#include <string>
+
+/* discard incoming text messages over the socket that are longer than this */
+#define MAX_RECV_BUF_SIZE (65 * 1024 * 10)
+#define RECV_BUF_REALLOC_SIZE (8 * 1024)
+#define AWS_PRELUDE_PLUS_HDRS_LEN (100)
+
+using namespace aws;
+
+namespace {
+  static const char *requestedTcpKeepaliveSecs = std::getenv("MOD_AUDIO_FORK_TCP_KEEPALIVE_SECS");
+  static int nTcpKeepaliveSecs = requestedTcpKeepaliveSecs ? ::atoi(requestedTcpKeepaliveSecs) : 55;
+  static uint8_t aws_prelude_and_headers[AWS_PRELUDE_PLUS_HDRS_LEN];
+
+  void writeToFile(const char* buffer, size_t bufferSize) {
+    static int writeCounter = 0; // Static variable to keep track of write count
+
+    // Write only the first three times
+    if (writeCounter >= 4) {
+        return;
+    }
+
+    // Generate a unique file name using the writeCounter
+    std::string filename = "/tmp/audio_data_" + std::to_string(writeCounter) + ".bin";
+
+    // Open a file in binary mode
+    std::ofstream outFile(filename, std::ios::binary);
+
+    // Check if the file is open
+    if (outFile.is_open()) {
+        // Write the buffer to the file
+        outFile.write(buffer, bufferSize);
+        outFile.close();
+
+        // Increment the write counter
+        writeCounter++;
+    } else {
+        // Handle error in file opening
+        std::cerr << "Unable to open file: " << filename << std::endl;
+    }
+  }
+}
+
+
+int AudioPipe::aws_lws_callback(struct lws *wsi, 
+  enum lws_callback_reasons reason,
+  void *user, void *in, size_t len) {
+
+  struct AudioPipe::lws_per_vhost_data *vhd = 
+    (struct AudioPipe::lws_per_vhost_data *) lws_protocol_vh_priv_get(lws_get_vhost(wsi), lws_get_protocol(wsi));
+
+  struct lws_vhost* vhost = lws_get_vhost(wsi);
+  AudioPipe ** ppAp = (AudioPipe **) user;
+
+  switch (reason) {
+    case LWS_CALLBACK_PROTOCOL_INIT:
+      vhd = (struct AudioPipe::lws_per_vhost_data *) lws_protocol_vh_priv_zalloc(lws_get_vhost(wsi), lws_get_protocol(wsi), sizeof(struct AudioPipe::lws_per_vhost_data));
+      vhd->context = lws_get_context(wsi);
+      vhd->protocol = lws_get_protocol(wsi);
+      vhd->vhost = lws_get_vhost(wsi);
+      break;
+
+    case LWS_CALLBACK_EVENT_WAIT_CANCELLED:
+      processPendingConnects(vhd);
+      processPendingDisconnects(vhd);
+      processPendingWrites();
+      break;
+    case LWS_CALLBACK_CLIENT_CONNECTION_ERROR:
+      {
+        AudioPipe* ap = findAndRemovePendingConnect(wsi);
+        int rc = lws_http_client_http_response(wsi);
+        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, ap->isFinished());
+        }
+        else {
+          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;
+
+    case LWS_CALLBACK_CLIENT_ESTABLISHED:
+      {
+        AudioPipe* ap = findAndRemovePendingConnect(wsi);
+        if (ap) {
+          *ppAp = ap;
+          ap->m_vhd = vhd;
+          ap->m_state = LWS_CLIENT_CONNECTED;
+          ap->m_callback(ap->m_uuid.c_str(), ap->m_bugname.c_str(), AudioPipe::CONNECT_SUCCESS, NULL,  ap->isFinished());
+          //switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO,"%s connected\n", ap->m_uuid.c_str());
+        }
+        else {
+          lwsl_err("AudioPipe::lws_service_thread LWS_CALLBACK_CLIENT_ESTABLISHED %s unable to find wsi %p..\n", ap->m_uuid.c_str(), wsi); 
+        }
+      }      
+      break;
+    case LWS_CALLBACK_CLIENT_CLOSED:
+      {
+        AudioPipe* ap = *ppAp;
+        if (!ap) {
+          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) {
+          // closed by us
+
+          lwsl_debug("%s socket closed by us\n", ap->m_uuid.c_str());
+          ap->m_callback(ap->m_uuid.c_str(), ap->m_bugname.c_str(), AudioPipe::CONNECTION_CLOSED_GRACEFULLY, NULL,  ap->isFinished());
+        }
+        else if (ap->m_state == LWS_CLIENT_CONNECTED) {
+          // closed by far end
+          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,  ap->isFinished());
+        }
+        ap->m_state = LWS_CLIENT_DISCONNECTED;
+        ap->setClosed();
+    
+        //NB: after receiving any of the events above, any holder of a 
+        //pointer or reference to this object must treat is as no longer valid
+
+        //*ppAp = NULL;
+        //delete ap;
+      }
+      break;
+
+    case LWS_CALLBACK_CLIENT_RECEIVE:
+      {
+        AudioPipe* ap = *ppAp;
+        if (!ap) {
+          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 (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_err("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)) {
+            //switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "AudioPipe::lws_service_thread - LWS_CALLBACK_CLIENT_RECEIVE received %d bytes.\n", len);
+            if (nullptr != ap->m_recv_buf) {
+              bool isError = false;
+              std::string payload;
+              std::string msg((char *)ap->m_recv_buf, ap->m_recv_buf_ptr - ap->m_recv_buf);
+
+              TranscribeManager::parseResponse(msg, payload, isError, true);
+              //switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "AudioPipe::lws_service_thread LWS_CALLBACK_CLIENT_RECEIVE payload: %s.\n", payload.c_str());
+              //switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "AudioPipe::lws_service_thread LWS_CALLBACK_CLIENT_RECEIVE response %s.\n", msg.c_str());
+
+
+              if (0 != payload.compare("{\"Transcript\":{\"Results\":[]}}")) {
+                ap->m_callback(ap->m_uuid.c_str(), ap->m_bugname.c_str(), AudioPipe::MESSAGE, payload.c_str(),  ap->isFinished());
+              }
+              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;
+          }
+        }
+      }
+      break;
+
+    case LWS_CALLBACK_CLIENT_WRITEABLE:
+      {
+        AudioPipe* ap = *ppAp;
+        if (!ap) {
+          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;
+        }
+
+        if (ap->m_state == LWS_CLIENT_DISCONNECTING) {
+          lws_close_reason(wsi, LWS_CLOSE_STATUS_NORMAL, NULL, 0);
+          return -1;
+        }
+
+        // check for audio packets
+        {
+          std::lock_guard<std::mutex> lk(ap->m_audio_mutex);
+          if (ap->m_audio_buffer_write_offset > LWS_PRE + AWS_PRELUDE_PLUS_HDRS_LEN || ap->isFinished()) {
+
+            // send a zero length audio packet to indicate end of stream
+            if (ap->isFinished()) {
+              ap->m_audio_buffer_write_offset = LWS_PRE + AWS_PRELUDE_PLUS_HDRS_LEN;
+            }
+            /**
+             * fill in
+             * [0..3] = total byte length
+             * [8..11] = prelude crc
+             * following the audio data: 4 bytes of Message CRC
+             * 
+             */
+
+            // copy in the prelude and headers
+            memcpy(ap->m_audio_buffer + LWS_PRE, aws_prelude_and_headers, AWS_PRELUDE_PLUS_HDRS_LEN);
+
+            // fill in the total byte length
+            uint32_t totalLen = ap->m_audio_buffer_write_offset - LWS_PRE + 4; // for the trailing Message CRC which is 4 bytes
+            //lwsl_err("AudioPipe - total length %u (decimal), 0x%X (hex)\n", totalLen, totalLen);
+            totalLen = htonl(totalLen);
+            //lwsl_err("AudioPipe - total length in network byte order %u (decimal), 0x%X (hex)\n", totalLen, totalLen);
+            memcpy(ap->m_audio_buffer + LWS_PRE, &totalLen, sizeof(uint32_t));
+
+            // fill in the prelude crc
+            uint32_t preludeCRC = CRC::Calculate(ap->m_audio_buffer + LWS_PRE, 8, CRC::CRC_32());
+            //lwsl_err("AudioPipe - prelude CRC %u (decimal), 0x%X (hex)\n", preludeCRC, preludeCRC);
+            preludeCRC = htonl(preludeCRC);
+            //lwsl_err("AudioPipe - prelude CRC in network order %u (decimal), 0x%X (hex)\n", preludeCRC, preludeCRC);
+            memcpy(ap->m_audio_buffer + LWS_PRE + 8, &preludeCRC, sizeof(uint32_t));
+
+            // fill in the message crc
+            uint32_t messageCRC = CRC::Calculate(ap->m_audio_buffer + LWS_PRE, ap->m_audio_buffer_write_offset - LWS_PRE, CRC::CRC_32());
+            messageCRC = htonl(messageCRC);
+            memcpy(ap->m_audio_buffer + ap->m_audio_buffer_write_offset, &messageCRC, sizeof(uint32_t));
+            ap->m_audio_buffer_write_offset + sizeof(uint32_t);
+
+            size_t datalen = ap->m_audio_buffer_write_offset - LWS_PRE + 4;
+            
+            //switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO,"%s writing data length %lu\n", ap->m_uuid.c_str(), datalen);
+
+            // TMP: write data to a file
+            //writeToFile((const char *) ap->m_audio_buffer + LWS_PRE, datalen);
+
+            int sent = lws_write(wsi, (unsigned char *) ap->m_audio_buffer + LWS_PRE, datalen, LWS_WRITE_BINARY);
+            if (sent < datalen) {
+              switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "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->binaryWritePtrResetToZero();
+          }
+        }
+
+        return 0;
+      }
+      break;
+
+    default:
+      break;
+  }
+  return lws_callback_http_dummy(wsi, reason, user, in, len);
+}
+
+
+// static members
+static const lws_retry_bo_t retry = {
+    nullptr,   // retry_ms_table
+    0,         // retry_ms_table_count
+    0,         // conceal_count
+    UINT16_MAX,         // secs_since_valid_ping
+    UINT16_MAX,        // secs_since_valid_hangup
+    0          // jitter_percent
+};
+
+struct lws_context *AudioPipe::contexts[] = {
+  nullptr, nullptr, nullptr, nullptr, nullptr,
+  nullptr, nullptr, nullptr, nullptr, nullptr
+};
+unsigned int AudioPipe::numContexts = 0;
+unsigned int AudioPipe::nchild = 0;
+std::string AudioPipe::protocolName;
+std::mutex AudioPipe::mutex_connects;
+std::mutex AudioPipe::mutex_disconnects;
+std::mutex AudioPipe::mutex_writes;
+std::list<AudioPipe*> AudioPipe::pendingConnects;
+std::list<AudioPipe*> AudioPipe::pendingDisconnects;
+std::list<AudioPipe*> AudioPipe::pendingWrites;
+AudioPipe::log_emit_function AudioPipe::logger;
+std::mutex AudioPipe::mapMutex;
+std::unordered_map<std::thread::id, bool> AudioPipe::stopFlags;
+std::queue<std::thread::id> AudioPipe::threadIds;
+
+
+void AudioPipe::processPendingConnects(lws_per_vhost_data *vhd) {
+  std::list<AudioPipe*> connects;
+  {
+    std::lock_guard<std::mutex> guard(mutex_connects);
+    for (auto it = pendingConnects.begin(); it != pendingConnects.end(); ++it) {
+      if ((*it)->m_state == LWS_CLIENT_IDLE) {
+        connects.push_back(*it);
+        (*it)->m_state = LWS_CLIENT_CONNECTING;
+      }
+    }
+  }
+  for (auto it = connects.begin(); it != connects.end(); ++it) {
+    AudioPipe* ap = *it;
+    ap->connect_client(vhd);   
+  }
+}
+
+void AudioPipe::processPendingDisconnects(lws_per_vhost_data *vhd) {
+  std::list<AudioPipe*> disconnects;
+  {
+    std::lock_guard<std::mutex> guard(mutex_disconnects);
+    for (auto it = pendingDisconnects.begin(); it != pendingDisconnects.end(); ++it) {
+      if ((*it)->m_state == LWS_CLIENT_DISCONNECTING) disconnects.push_back(*it);
+    }
+    pendingDisconnects.clear();
+  }
+  for (auto it = disconnects.begin(); it != disconnects.end(); ++it) {
+    AudioPipe* ap = *it;
+    lws_callback_on_writable(ap->m_wsi); 
+  }
+}
+
+void AudioPipe::processPendingWrites() {
+  std::list<AudioPipe*> writes;
+  {
+    std::lock_guard<std::mutex> guard(mutex_writes);
+    for (auto it = pendingWrites.begin(); it != pendingWrites.end(); ++it) {
+       if ((*it)->m_state == LWS_CLIENT_CONNECTED) writes.push_back(*it);
+    }  
+    pendingWrites.clear();
+  }
+  for (auto it = writes.begin(); it != writes.end(); ++it) {
+    AudioPipe* ap = *it;
+    lws_callback_on_writable(ap->m_wsi);
+  }
+}
+
+AudioPipe* AudioPipe::findAndRemovePendingConnect(struct lws *wsi) {
+  AudioPipe* ap = NULL;
+  std::lock_guard<std::mutex> guard(mutex_connects);
+  std::list<AudioPipe* > toRemove;
+
+  for (auto it = pendingConnects.begin(); it != pendingConnects.end() && !ap; ++it) {
+    int state = (*it)->m_state;
+
+    if ((*it)->m_wsi == nullptr)
+      toRemove.push_back(*it);
+
+    if ((state == LWS_CLIENT_CONNECTING) &&
+      (*it)->m_wsi == wsi) ap = *it;
+  }
+
+  for (auto it = toRemove.begin(); it != toRemove.end(); ++it)
+    pendingConnects.remove(*it);
+
+  if (ap) {
+    pendingConnects.remove(ap);
+  }
+
+  return ap;
+}
+
+AudioPipe* AudioPipe::findPendingConnect(struct lws *wsi) {
+  AudioPipe* ap = NULL;
+  std::lock_guard<std::mutex> guard(mutex_connects);
+
+  for (auto it = pendingConnects.begin(); it != pendingConnects.end() && !ap; ++it) {
+    int state = (*it)->m_state;
+    if ((state == LWS_CLIENT_CONNECTING) &&
+      (*it)->m_wsi == wsi) ap = *it;
+  }
+  return ap;
+}
+
+void AudioPipe::addPendingConnect(AudioPipe* ap) {
+  {
+    std::lock_guard<std::mutex> guard(mutex_connects);
+    pendingConnects.push_back(ap);
+    lwsl_debug("%s after adding connect there are %lu pending connects\n", 
+      ap->m_uuid.c_str(), pendingConnects.size());
+  }
+  lws_cancel_service(contexts[nchild++ % numContexts]);
+}
+void AudioPipe::addPendingDisconnect(AudioPipe* ap) {
+  ap->m_state = LWS_CLIENT_DISCONNECTING;
+  {
+    std::lock_guard<std::mutex> guard(mutex_disconnects);
+    pendingDisconnects.push_back(ap);
+    lwsl_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);
+}
+void AudioPipe::addPendingWrite(AudioPipe* ap) {
+  {
+    std::lock_guard<std::mutex> guard(mutex_writes);
+    pendingWrites.push_back(ap);
+  }
+  lws_cancel_service(ap->m_vhd->context);
+}
+
+void AudioPipe::binaryWritePtrResetToZero(void) {
+    m_audio_buffer_write_offset = LWS_PRE + AWS_PRELUDE_PLUS_HDRS_LEN;
+}
+
+bool AudioPipe::lws_service_thread(unsigned int nServiceThread) {
+  struct lws_context_creation_info info;
+  std::thread::id this_id = std::this_thread::get_id();
+
+  const struct lws_protocols protocols[] = {
+    {
+      "",
+      AudioPipe::aws_lws_callback,
+      sizeof(void *),
+      1024,
+    },
+    { NULL, NULL, 0, 0 }
+  };
+
+  memset(&info, 0, sizeof info); 
+  info.port = CONTEXT_PORT_NO_LISTEN; 
+  info.options = LWS_SERVER_OPTION_DO_SSL_GLOBAL_INIT;
+  info.protocols = protocols;
+  info.ka_time = nTcpKeepaliveSecs;                    // tcp keep-alive timer
+  info.ka_probes = 4;                   // number of times to try ka before closing connection
+  info.ka_interval = 5;                 // time between ka's
+  info.timeout_secs = 10;                // doc says timeout for "various processes involving network roundtrips"
+  info.keepalive_timeout = 5;           // seconds to allow remote client to hold on to an idle HTTP/1.1 connection 
+  info.timeout_secs_ah_idle = 10;       // secs to allow a client to hold an ah without using it
+  info.retry_and_idle_policy = &retry;
+
+  switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO,"AudioPipe::lws_service_thread creating context\n");
+
+  contexts[nServiceThread] = lws_create_context(&info);
+  if (!contexts[nServiceThread]) {
+    lwsl_err("AudioPipe::lws_service_thread failed creating context in service thread %d..\n", nServiceThread); 
+    return false;
+  }
+
+  int n;
+  do {
+    n = lws_service(contexts[nServiceThread], 0);
+  } while (n >= 0 && !stopFlags[this_id]);
+
+  // Cleanup once work is done or stopped
+  {
+      std::lock_guard<std::mutex> lock(mapMutex);
+      stopFlags.erase(this_id);
+  }
+
+  lwsl_notice("AudioPipe::lws_service_thread ending in service thread %d\n", nServiceThread); 
+  return true;
+}
+
+void AudioPipe::initialize(unsigned int nThreads, int loglevel, log_emit_function logger) {
+  assert(nThreads > 0 && nThreads <= 10);
+
+  numContexts = nThreads;
+  lws_set_log_level(loglevel, logger);
+
+  switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_NOTICE,"AudioPipe::initialize starting\n"); 
+  for (unsigned int i = 0; i < numContexts; i++) {
+    std::lock_guard<std::mutex> lock(mapMutex);
+    std::thread t(&AudioPipe::lws_service_thread, i);
+    stopFlags[t.get_id()] = false;
+    threadIds.push(t.get_id());
+    t.detach();
+  }
+}
+
+bool AudioPipe::deinitialize() {
+  switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_NOTICE,"AudioPipe::deinitialize\n"); 
+  std::lock_guard<std::mutex> lock(mapMutex);
+  if (!threadIds.empty()) {
+      std::thread::id id = threadIds.front();
+      threadIds.pop();
+      stopFlags[id] = true;
+  }
+  for (unsigned int i = 0; i < numContexts; i++)
+  {
+    lwsl_notice("AudioPipe::deinitialize destroying context %d of %d\n", i + 1, numContexts);
+    lws_context_destroy(contexts[i]);
+  }
+  std::this_thread::sleep_for(std::chrono::seconds(2));
+  return true;
+}
+
+// instance members
+AudioPipe::AudioPipe(const char* uuid, const char* bugname, const char* host, unsigned int port, const char* path,
+  size_t bufLen, size_t minFreespace, notifyHandler_t callback) :
+  m_uuid(uuid), m_host(host), m_port(port), m_path(path), m_finished(false), m_bugname(bugname),
+  m_audio_buffer_min_freespace(minFreespace), m_audio_buffer_max_len(bufLen), m_gracefulShutdown(false),
+  m_recv_buf(nullptr), m_recv_buf_ptr(nullptr),
+  m_state(LWS_CLIENT_IDLE), m_wsi(nullptr), m_vhd(nullptr), m_callback(callback) {
+
+  char headerBuffer[88];
+  char* buffer = headerBuffer;
+  m_audio_buffer = new uint8_t[m_audio_buffer_max_len];
+
+  // stamp out the template for the prelude and headers 
+  memset(aws_prelude_and_headers, 0, AWS_PRELUDE_PLUS_HDRS_LEN);
+
+  // aws_prelude_and_headers[0..3] = total byte length (not known till message send time)
+
+  // aws_prelude_and_headers[4..7] = headers byte length
+  uint32_t headerLen = sizeof(headerBuffer);
+  headerLen = htonl(headerLen);
+  memcpy(&aws_prelude_and_headers[4], &headerLen, sizeof(uint32_t));
+
+  // aws_prelude_and_headers[8..11] = prelude crc (not known till message send time)
+
+  // aws_prelude_and_headers[12..99] = headers
+  TranscribeManager::writeHeader(&buffer, ":content-type", "application/octet-stream");
+  TranscribeManager::writeHeader(&buffer, ":event-type", "AudioEvent");
+  TranscribeManager::writeHeader(&buffer, ":message-type", "event");
+
+  memcpy(&aws_prelude_and_headers[12], headerBuffer, sizeof(headerBuffer));
+
+  // following this will be the audio data and a final message CRC (not known till message send time)
+
+  memcpy(m_audio_buffer + LWS_PRE, aws_prelude_and_headers, AWS_PRELUDE_PLUS_HDRS_LEN);
+  m_audio_buffer_write_offset = LWS_PRE + AWS_PRELUDE_PLUS_HDRS_LEN;
+
+  //writeToFile((const char *) m_audio_buffer + LWS_PRE, AWS_PRELUDE_PLUS_HDRS_LEN);
+
+}
+
+AudioPipe::~AudioPipe() {
+  if (m_audio_buffer) delete [] m_audio_buffer;
+  if (m_recv_buf) delete [] m_recv_buf;
+}
+
+void AudioPipe::connect(void) {
+  addPendingConnect(this);
+}
+
+bool AudioPipe::connect_client(struct lws_per_vhost_data *vhd) {
+  assert(m_audio_buffer != nullptr);
+  assert(m_vhd == nullptr);
+  struct lws_client_connect_info i;
+
+  memset(&i, 0, sizeof(i));
+  i.context = vhd->context;
+  i.port = m_port;
+  i.address = m_host.c_str();
+  i.path = m_path.c_str();
+  i.host = i.address;
+  i.origin = i.address;
+  i.ssl_connection = LCCSCF_USE_SSL;
+  //i.protocol = protocolName.c_str();
+  i.pwsi = &(m_wsi);
+
+  m_state = LWS_CLIENT_CONNECTING;
+  m_vhd = vhd;
+
+  m_wsi = lws_client_connect_via_info(&i);
+  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;
+}
+
+void AudioPipe::bufferForSending(const char* text) {
+  if (m_state != LWS_CLIENT_CONNECTED) return;
+  {
+    std::lock_guard<std::mutex> lk(m_text_mutex);
+    m_metadata.append(text);
+  }
+  addPendingWrite(this);
+}
+
+void AudioPipe::unlockAudioBuffer() {
+  if (m_audio_buffer_write_offset > LWS_PRE) addPendingWrite(this);
+  m_audio_mutex.unlock();
+}
+
+void AudioPipe::close() {
+  if (m_state != LWS_CLIENT_CONNECTED) return;
+  addPendingDisconnect(this);
+}
+
+void AudioPipe::finish() {
+  if (m_finished || m_state != LWS_CLIENT_CONNECTED) return;
+  m_finished = true;
+  addPendingWrite(this);
+}
+
+void AudioPipe::waitForClose() {
+  std::shared_future<void> sf(m_promise.get_future());
+  sf.wait();
+  return;
+}
@@ -0,0 +1,144 @@
+#ifndef __AWS_AUDIO_PIPE_HPP__
+#define __AWS_AUDIO_PIPE_HPP__
+
+#include <string>
+#include <list>
+#include <mutex>
+#include <future>
+#include <queue>
+#include <unordered_map>
+#include <thread>
+
+#include <libwebsockets.h>
+
+namespace aws {
+
+class AudioPipe {
+public:
+  enum LwsState_t {
+    LWS_CLIENT_IDLE,
+    LWS_CLIENT_CONNECTING,
+    LWS_CLIENT_CONNECTED,
+    LWS_CLIENT_FAILED,
+    LWS_CLIENT_DISCONNECTING,
+    LWS_CLIENT_DISCONNECTED
+  };
+  enum NotifyEvent_t {
+    CONNECT_SUCCESS,
+    CONNECT_FAIL,
+    CONNECTION_DROPPED,
+    CONNECTION_CLOSED_GRACEFULLY,
+    MESSAGE
+  };
+  typedef void (*log_emit_function)(int level, const char *line);
+  typedef void (*notifyHandler_t)(const char *sessionId, const char* bugname, NotifyEvent_t event, const char* message, bool finished);
+
+  struct lws_per_vhost_data {
+    struct lws_context *context;
+    struct lws_vhost *vhost;
+    const struct lws_protocols *protocol;
+  };
+
+  static void initialize(unsigned int nThreads, int loglevel, log_emit_function logger);
+  static bool deinitialize();
+  static bool lws_service_thread(unsigned int nServiceThread);
+
+  // constructor
+  AudioPipe(const char* uuid, const char* bugname, const char* host, unsigned int port, const char* path, 
+    size_t bufLen, size_t minFreespace, notifyHandler_t callback);
+  ~AudioPipe();  
+
+  LwsState_t getLwsState(void) { return m_state; }
+  std::string& getApiKey(void) {
+    return m_apiKey;
+  }
+  void connect(void);
+  void bufferForSending(const char* text);
+  size_t binarySpaceAvailable(void) {
+    return m_audio_buffer_max_len - m_audio_buffer_write_offset;
+  }
+  size_t binaryMinSpace(void) {
+    return m_audio_buffer_min_freespace;
+  }
+  char * binaryWritePtr(void) { 
+    return (char *) m_audio_buffer + m_audio_buffer_write_offset;
+  }
+  void binaryWritePtrAdd(size_t len) {
+    m_audio_buffer_write_offset += len;
+  }
+  void binaryWritePtrResetToZero(void);
+  void lockAudioBuffer(void) {
+    m_audio_mutex.lock();
+  }
+  void unlockAudioBuffer(void) ;
+
+  void close() ;
+  void finish();
+  void waitForClose();
+  void setClosed() { m_promise.set_value(); }
+  bool isFinished() { return m_finished;}
+
+  // no default constructor or copying
+  AudioPipe() = delete;
+  AudioPipe(const AudioPipe&) = delete;
+  void operator=(const AudioPipe&) = delete;
+
+private:
+
+  static int aws_lws_callback(struct lws *wsi, enum lws_callback_reasons reason, void *user, void *in, size_t len); 
+  static unsigned int nchild;
+  static struct lws_context *contexts[];
+  static unsigned int numContexts;
+  static std::string protocolName;
+  static std::mutex mutex_connects;
+  static std::mutex mutex_disconnects;
+  static std::mutex mutex_writes;
+  static std::list<AudioPipe*> pendingConnects;
+  static std::list<AudioPipe*> pendingDisconnects;
+  static std::list<AudioPipe*> pendingWrites;
+  static log_emit_function logger;
+
+  static std::mutex mapMutex;
+  static std::unordered_map<std::thread::id, bool> stopFlags;
+  static std::queue<std::thread::id> threadIds;
+
+  static AudioPipe* findAndRemovePendingConnect(struct lws *wsi);
+  static AudioPipe* findPendingConnect(struct lws *wsi);
+  static void addPendingConnect(AudioPipe* ap);
+  static void addPendingDisconnect(AudioPipe* ap);
+  static void addPendingWrite(AudioPipe* ap);
+  static void processPendingConnects(lws_per_vhost_data *vhd);
+  static void processPendingDisconnects(lws_per_vhost_data *vhd);
+  static void processPendingWrites(void);
+  
+  bool connect_client(struct lws_per_vhost_data *vhd);
+
+  LwsState_t m_state;
+  std::string m_uuid;
+  std::string m_host;
+  unsigned int m_port;
+  std::string m_path;
+  std::string m_metadata;
+  std::mutex m_text_mutex;
+  std::mutex m_audio_mutex;
+  int m_sslFlags;
+  struct lws *m_wsi;
+  uint8_t *m_audio_buffer;
+  size_t m_audio_buffer_max_len;
+  size_t m_audio_buffer_write_offset;
+  size_t m_audio_buffer_min_freespace;
+  uint8_t* m_recv_buf;
+  uint8_t* m_recv_buf_ptr;
+  size_t m_recv_buf_len;
+  struct lws_per_vhost_data* m_vhd;
+  notifyHandler_t m_callback;
+  log_emit_function m_logger;
+  std::string m_apiKey;
+  bool m_gracefulShutdown;
+  bool m_finished;
+  std::string m_bugname;
+  std::promise<void> m_promise;
+};
+
+} // namespace deepgram
+#endif
@@ -0,0 +1,415 @@
+#include <switch.h>
+#include <switch_json.h>
+#include <string.h>
+#include <string>
+#include <mutex>
+#include <thread>
+#include <list>
+#include <algorithm>
+#include <functional>
+#include <cassert>
+#include <cstdlib>
+#include <fstream>
+#include <sstream>
+#include <regex>
+#include <iostream>
+#include <unordered_map>
+
+#include "mod_aws_transcribe_ws.h"
+#include "simple_buffer.h"
+//#include "parser.hpp"
+#include "audio_pipe.hpp"
+#include "transcribe_manager.hpp"
+
+#define RTP_PACKETIZATION_PERIOD 20
+#define FRAME_SIZE_8000  320 /*which means each 20ms frame as 320 bytes at 8 khz (1 channel only)*/
+
+namespace {
+  static bool hasDefaultCredentials = false;
+  static const char* defaultApiKey = nullptr;
+  static const char *requestedBufferSecs = std::getenv("MOD_AUDIO_FORK_BUFFER_SECS");
+  static int nAudioBufferSecs = std::max(1, std::min(requestedBufferSecs ? ::atoi(requestedBufferSecs) : 2, 5));
+  static const char *requestedNumServiceThreads = std::getenv("MOD_AUDIO_FORK_SERVICE_THREADS");
+  static unsigned int nServiceThreads = std::max(1, std::min(requestedNumServiceThreads ? ::atoi(requestedNumServiceThreads) : 1, 5));
+  static unsigned int idxCallCount = 0;
+  static uint32_t playCount = 0;
+
+  static const char* emptyTranscript = "{\"Transcript\":{\"Results\":[]}}";
+  static const char* messageStart = "{\"Message\":";
+
+  static void reaper(private_t *tech_pvt) {
+    std::shared_ptr<aws::AudioPipe> pAp;
+    pAp.reset((aws::AudioPipe *)tech_pvt->pAudioPipe);
+    tech_pvt->pAudioPipe = nullptr;
+
+    std::thread t([pAp, tech_pvt]{
+      pAp->finish();
+      pAp->waitForClose();
+      switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "%s (%u) got remote close\n", tech_pvt->sessionId, tech_pvt->id);
+    });
+    t.detach();
+  }
+
+  static void destroy_tech_pvt(private_t *tech_pvt) {
+    switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "%s (%u) destroy_tech_pvt\n", tech_pvt->sessionId, tech_pvt->id);
+    if (tech_pvt) {
+      if (tech_pvt->pAudioPipe) {
+        aws::AudioPipe* p = (aws::AudioPipe *) tech_pvt->pAudioPipe;
+        delete p;
+        tech_pvt->pAudioPipe = nullptr;
+      }
+      if (tech_pvt->resampler) {
+          speex_resampler_destroy(tech_pvt->resampler);
+          tech_pvt->resampler = NULL;
+      }
+
+      if (tech_pvt->vad) {
+        switch_vad_destroy(&tech_pvt->vad);
+        tech_pvt->vad = nullptr;
+      }
+    }
+  }
+
+  static void eventCallback(const char* sessionId, const char* bugname, 
+    aws::AudioPipe::NotifyEvent_t event, const char* message, bool finished) {
+    switch_core_session_t* session = switch_core_session_locate(sessionId);
+    if (session) {
+      switch_channel_t *channel = switch_core_session_get_channel(session);
+      switch_media_bug_t *bug = (switch_media_bug_t*) switch_channel_get_private(channel, bugname);
+      if (bug) {
+        private_t* tech_pvt = (private_t*) switch_core_media_bug_get_user_data(bug);
+        if (tech_pvt) {
+          switch (event) {
+            case aws::AudioPipe::CONNECT_SUCCESS:
+              switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_INFO, "connection successful\n");
+              tech_pvt->responseHandler(session, TRANSCRIBE_EVENT_CONNECT_SUCCESS, NULL, tech_pvt->bugname, finished);
+            break;
+            case aws::AudioPipe::CONNECT_FAIL:
+            {
+              // first thing: we can no longer access the AudioPipe
+              std::stringstream json;
+              json << "{\"reason\":\"" << message << "\"}";
+              tech_pvt->pAudioPipe = nullptr;
+              tech_pvt->responseHandler(session, TRANSCRIBE_EVENT_CONNECT_FAIL, (char *) json.str().c_str(), tech_pvt->bugname, finished);
+              switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_NOTICE, "connection failed: %s\n", message);
+            }
+            break;
+            case aws::AudioPipe::CONNECTION_DROPPED:
+              // first thing: we can no longer access the AudioPipe
+              tech_pvt->pAudioPipe = nullptr;
+              tech_pvt->responseHandler(session, TRANSCRIBE_EVENT_DISCONNECT, NULL, tech_pvt->bugname, finished);
+              switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "connection dropped from far end\n");
+            break;
+            case aws::AudioPipe::CONNECTION_CLOSED_GRACEFULLY:
+              // first thing: we can no longer access the AudioPipe
+              tech_pvt->pAudioPipe = nullptr;
+              switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "connection closed gracefully\n");
+            break;
+            case aws::AudioPipe::MESSAGE:
+              if( strstr(message, emptyTranscript)) {
+                switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "discarding empty aws transcript\n");
+              }
+              else if (0 == strncmp( message, messageStart, strlen(messageStart))) {
+                tech_pvt->responseHandler(session, TRANSCRIBE_EVENT_ERROR, message, tech_pvt->bugname, finished);
+                switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "error message from aws: %s\n", message);
+              }
+              else {
+                tech_pvt->responseHandler(session, TRANSCRIBE_EVENT_RESULTS, message, tech_pvt->bugname, finished);
+                switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "aws message: %s.\n", message);
+              }
+            break;
+
+            default:
+              switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_NOTICE, "got unexpected msg from aws %d:%s\n", event, message);
+              break;
+          }
+        }
+      }
+      switch_core_session_rwunlock(session);
+    }
+  }
+
+  void lws_logger(int level, const char *line) {
+    switch_log_level_t llevel = SWITCH_LOG_DEBUG;
+
+    switch (level) {
+      case LLL_ERR: llevel = SWITCH_LOG_ERROR; break;
+      case LLL_WARN: llevel = SWITCH_LOG_WARNING; break;
+      case LLL_NOTICE: llevel = SWITCH_LOG_NOTICE; break;
+      case LLL_INFO: llevel = SWITCH_LOG_INFO; break;
+      break;
+    }
+	  switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_NOTICE, "%s\n", line);
+  }
+}
+
+
+extern "C" {
+  switch_status_t aws_transcribe_init() {
+    switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_NOTICE, "mod_aws_transcribe: audio buffer (in secs):    %d secs\n", nAudioBufferSecs);
+    switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_NOTICE, "mod_aws_transcribe: lws service threads:       %d\n", nServiceThreads);
+ 
+    int logs = LLL_ERR | LLL_WARN | LLL_NOTICE || LLL_INFO | LLL_PARSER | LLL_HEADER | LLL_EXT | LLL_CLIENT  | LLL_LATENCY | LLL_DEBUG ;
+    
+    aws::AudioPipe::initialize(nServiceThreads, logs, lws_logger);
+		switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_NOTICE, "AudioPipe::initialize completed\n");
+
+		return SWITCH_STATUS_SUCCESS;
+  }
+
+  switch_status_t aws_transcribe_cleanup() {
+    bool cleanup = false;
+    cleanup = aws::AudioPipe::deinitialize();
+    if (cleanup == true) {
+        return SWITCH_STATUS_SUCCESS;
+    }
+    return SWITCH_STATUS_FALSE;
+  }
+	
+  	// start transcribe on a channel
+	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;
+		switch_channel_t *channel = switch_core_session_get_channel(session);
+		int err;
+    uint32_t desiredSampling = 8000;
+		switch_threadattr_t *thd_attr = NULL;
+		switch_memory_pool_t *pool = switch_core_session_get_pool(session);
+		auto read_codec = switch_core_session_get_read_codec(session);
+		uint32_t sampleRate = read_codec->implementation->actual_samples_per_second;
+    switch_codec_implementation_t read_impl;
+    switch_core_session_get_read_impl(session, &read_impl);
+
+		private_t* tech_pvt = (private_t *) switch_core_session_alloc(session, sizeof(private_t));
+		memset(tech_pvt, sizeof(tech_pvt), 0);
+		const char* awsAccessKeyId = switch_channel_get_variable(channel, "AWS_ACCESS_KEY_ID");
+		const char* awsSecretAccessKey = switch_channel_get_variable(channel, "AWS_SECRET_ACCESS_KEY");
+		const char* awsRegion = switch_channel_get_variable(channel, "AWS_REGION");
+		const char* awsSessionToken = switch_channel_get_variable(channel, "AWS_SECURITY_TOKEN");
+		tech_pvt->channels = channels;
+		strncpy(tech_pvt->sessionId, switch_core_session_get_uuid(session), MAX_SESSION_ID);
+		strncpy(tech_pvt->bugname, bugname, MAX_BUG_LEN);
+
+		if (awsAccessKeyId && awsSecretAccessKey && awsRegion && awsSessionToken) {
+			switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "Using channel vars for aws authentication\n");
+			strncpy(tech_pvt->awsAccessKeyId, awsAccessKeyId, 128);
+			strncpy(tech_pvt->awsSecretAccessKey, awsSecretAccessKey, 128);
+			strncpy(tech_pvt->awsSessionToken, awsSessionToken, MAX_SESSION_TOKEN_LEN);
+			strncpy(tech_pvt->region, awsRegion, MAX_REGION);
+		}
+		else if (std::getenv("AWS_ACCESS_KEY_ID") &&
+			std::getenv("AWS_SECRET_ACCESS_KEY") &&
+			std::getenv("AWS_REGION")) {
+			switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "Using env vars for aws authentication\n");
+			strncpy(tech_pvt->awsAccessKeyId, std::getenv("AWS_ACCESS_KEY_ID"), 128);
+			strncpy(tech_pvt->awsSecretAccessKey, std::getenv("AWS_SECRET_ACCESS_KEY"), 128);		
+			strncpy(tech_pvt->region, std::getenv("AWS_REGION"), MAX_REGION);
+		}
+		else {
+			switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "No channel vars or env vars for aws authentication..will use default profile if found\n");
+		}
+
+		tech_pvt->responseHandler = responseHandler;
+
+		tech_pvt->interim = interim;
+		strncpy(tech_pvt->lang, lang, MAX_LANG);
+		tech_pvt->samples_per_second = sampleRate;
+		switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "sample rate of rtp stream is %d\n", samples_per_second);
+
+    const char* vocabularyName = switch_channel_get_variable(channel, "AWS_VOCABULARY_NAME");
+    const char* vocabularyFilterName = switch_channel_get_variable(channel, "AWS_VOCABULARY_FILTER_NAME");
+    const char* vocabularyFilterMethod = switch_channel_get_variable(channel, "AWS_VOCABULARY_FILTER_METHOD");
+    const char* piiEntityTypes = switch_channel_get_variable(channel, "AWS_PII_ENTITY_TYPES");
+    int shouldIdentifyPII = switch_true(switch_channel_get_variable(channel, "AWS_PII_IDENTIFY_ENTITIES"));
+    const char* languageModelName = switch_channel_get_variable(channel, "AWS_LANGUAGE_MODEL_NAME");
+
+    std::string host, path;
+    TranscribeManager::getSignedWebsocketUrl(
+      host,
+      path,
+      tech_pvt->awsAccessKeyId,
+      tech_pvt->awsSecretAccessKey, 
+      tech_pvt->awsSessionToken, 
+      tech_pvt->region,
+      lang,
+      vocabularyName,
+      vocabularyFilterName,
+      vocabularyFilterMethod,
+      piiEntityTypes,
+      shouldIdentifyPII,
+      languageModelName
+    );
+
+    host = host.substr(0, host.find(':'));
+		switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "connecting to host %s, path %s\n", host.c_str(), path.c_str());
+
+    strncpy(tech_pvt->sessionId, switch_core_session_get_uuid(session), MAX_SESSION_ID);
+    strncpy(tech_pvt->host, host.c_str(), MAX_WS_URL_LEN);
+    tech_pvt->port = 8443;
+    strncpy(tech_pvt->path, path.c_str(), MAX_PATH_LEN);    
+    tech_pvt->responseHandler = responseHandler;
+    tech_pvt->channels = channels;
+    tech_pvt->id = ++idxCallCount;
+    tech_pvt->buffer_overrun_notified = 0;
+
+    size_t buflen = LWS_PRE + (FRAME_SIZE_8000 * desiredSampling / 8000 * channels * 1000 / RTP_PACKETIZATION_PERIOD * nAudioBufferSecs);
+
+    aws::AudioPipe* ap = new aws::AudioPipe(tech_pvt->sessionId, tech_pvt->bugname, tech_pvt->host, tech_pvt->port, tech_pvt->path, 
+      buflen, read_impl.decoded_bytes_per_packet, eventCallback);
+    if (!ap) {
+      switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_ERROR, "Error allocating AudioPipe\n");
+      return SWITCH_STATUS_FALSE;
+    }
+
+    tech_pvt->pAudioPipe = static_cast<void *>(ap);
+
+		if (switch_mutex_init(&tech_pvt->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; 
+		}
+		if (sampleRate != 8000) {
+			tech_pvt->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_channel_get_name(channel), speex_resampler_strerror(err));
+				status = SWITCH_STATUS_FALSE;
+				goto done;
+			}
+		}
+
+    switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "connecting now\n");
+    ap->connect();
+    switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "connection in progress\n");
+
+
+		*ppUserData = tech_pvt;
+	
+	done:
+		return status;
+	}
+
+	switch_status_t aws_transcribe_session_stop(switch_core_session_t *session,int channelIsClosing, char* bugname) {
+    switch_channel_t *channel = switch_core_session_get_channel(session);
+    switch_media_bug_t *bug = (switch_media_bug_t*) switch_channel_get_private(channel, MY_BUG_NAME);
+    if (!bug) {
+      switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "aws_transcribe_session_stop: no bug - websocket conection already closed\n");
+      return SWITCH_STATUS_FALSE;
+    }
+    private_t* tech_pvt = (private_t*) switch_core_media_bug_get_user_data(bug);
+    uint32_t id = tech_pvt->id;
+
+    switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "(%u) aws_transcribe_session_stop\n", id);
+
+    if (!tech_pvt) return SWITCH_STATUS_FALSE;
+      
+    // close connection and get final responses
+    switch_mutex_lock(tech_pvt->mutex);
+    switch_channel_set_private(channel, bugname, NULL);
+    if (!channelIsClosing) switch_core_media_bug_remove(session, &bug);
+
+    aws::AudioPipe *pAudioPipe = static_cast<aws::AudioPipe *>(tech_pvt->pAudioPipe);
+    if (pAudioPipe) reaper(tech_pvt);
+    destroy_tech_pvt(tech_pvt);
+    switch_mutex_unlock(tech_pvt->mutex);
+    switch_mutex_destroy(tech_pvt->mutex);
+    tech_pvt->mutex = nullptr;
+    switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "(%u) aws_transcribe_session_stop\n", id);
+    return SWITCH_STATUS_SUCCESS;
+  }
+	
+	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);
+    private_t* tech_pvt = (private_t*) switch_core_media_bug_get_user_data(bug);
+    size_t inuse = 0;
+    bool dirty = false;
+    char *p = (char *) "{\"msg\": \"buffer overrun\"}";
+
+    if (!tech_pvt) return SWITCH_TRUE;
+    
+    if (switch_mutex_trylock(tech_pvt->mutex) == SWITCH_STATUS_SUCCESS) {
+      if (!tech_pvt->pAudioPipe) {
+        switch_mutex_unlock(tech_pvt->mutex);
+        return SWITCH_TRUE;
+      }
+      aws::AudioPipe *pAudioPipe = static_cast<aws::AudioPipe *>(tech_pvt->pAudioPipe);
+      if (pAudioPipe->getLwsState() != aws::AudioPipe::LWS_CLIENT_CONNECTED) {
+        switch_mutex_unlock(tech_pvt->mutex);
+        return SWITCH_TRUE;
+      }
+
+      pAudioPipe->lockAudioBuffer();
+      size_t available = pAudioPipe->binarySpaceAvailable();
+      if (NULL == tech_pvt->resampler) {
+        switch_frame_t frame = { 0 };
+        frame.data = pAudioPipe->binaryWritePtr();
+        frame.buflen = available;
+        while (true) {
+
+          // check if buffer would be overwritten; dump packets if so
+          if (available < pAudioPipe->binaryMinSpace()) {
+            if (!tech_pvt->buffer_overrun_notified) {
+              tech_pvt->buffer_overrun_notified = 1;
+              tech_pvt->responseHandler(session, TRANSCRIBE_EVENT_BUFFER_OVERRUN, NULL, tech_pvt->bugname, 0);
+            }
+            switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_ERROR, "(%u) dropping packets!\n", 
+              tech_pvt->id);
+            pAudioPipe->binaryWritePtrResetToZero();
+
+            frame.data = pAudioPipe->binaryWritePtr();
+            frame.buflen = available = pAudioPipe->binarySpaceAvailable();
+          }
+
+          switch_status_t rv = switch_core_media_bug_read(bug, &frame, SWITCH_TRUE);
+          if (rv != SWITCH_STATUS_SUCCESS) break;
+          if (frame.datalen) {
+            pAudioPipe->binaryWritePtrAdd(frame.datalen);
+            frame.buflen = available = pAudioPipe->binarySpaceAvailable();
+            frame.data = pAudioPipe->binaryWritePtr();
+            dirty = true;
+          }
+        }
+      }
+      else {
+        uint8_t data[SWITCH_RECOMMENDED_BUFFER_SIZE];
+        switch_frame_t frame = { 0 };
+        frame.data = data;
+        frame.buflen = SWITCH_RECOMMENDED_BUFFER_SIZE;
+        while (switch_core_media_bug_read(bug, &frame, SWITCH_TRUE) == SWITCH_STATUS_SUCCESS) {
+          if (frame.datalen) {
+            spx_uint32_t out_len = available >> 1;  // space for samples which are 2 bytes
+            spx_uint32_t in_len = frame.samples;
+
+            speex_resampler_process_interleaved_int(tech_pvt->resampler, 
+              (const spx_int16_t *) frame.data, 
+              (spx_uint32_t *) &in_len, 
+              (spx_int16_t *) ((char *) pAudioPipe->binaryWritePtr()),
+              &out_len);
+
+            if (out_len > 0) {
+              // bytes written = num samples * 2 * num channels
+              size_t bytes_written = out_len << tech_pvt->channels;
+              pAudioPipe->binaryWritePtrAdd(bytes_written);
+              available = pAudioPipe->binarySpaceAvailable();
+              dirty = true;
+            }
+            if (available < pAudioPipe->binaryMinSpace()) {
+              if (!tech_pvt->buffer_overrun_notified) {
+                tech_pvt->buffer_overrun_notified = 1;
+                switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_ERROR, "(%u) dropping packets!\n", 
+                  tech_pvt->id);
+                tech_pvt->responseHandler(session, TRANSCRIBE_EVENT_BUFFER_OVERRUN, NULL, tech_pvt->bugname, 0);
+              }
+              break;
+            }
+          }
+        }
+      }
+
+      pAudioPipe->unlockAudioBuffer();
+      switch_mutex_unlock(tech_pvt->mutex);
+    }
+    return SWITCH_TRUE;
+  }
+}
@@ -0,0 +1,11 @@
+#ifndef __AWS_GLUE_H__
+#define __AWS_GLUE_H__
+
+switch_status_t aws_transcribe_init();
+switch_status_t aws_transcribe_cleanup();
+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 aws_transcribe_session_stop(switch_core_session_t *session, int channelIsClosing, char* bugname);
+switch_bool_t aws_transcribe_frame(switch_media_bug_t *bug, void* user_data);
+
+#endif
@@ -0,0 +1,212 @@
+/* 
+ *
+ * mod_aws_transcribe.c -- Freeswitch module for using aws streaming transcribe api
+ *
+ */
+#include "mod_aws_transcribe_ws.h"
+#include "aws_transcribe_glue.h"
+
+/* Prototypes */
+SWITCH_MODULE_SHUTDOWN_FUNCTION(mod_aws_transcribe_ws_shutdown);
+SWITCH_MODULE_LOAD_FUNCTION(mod_aws_transcribe_ws_load);
+
+SWITCH_MODULE_DEFINITION(mod_aws_transcribe_ws, mod_aws_transcribe_ws_load, mod_aws_transcribe_ws_shutdown, NULL);
+
+static switch_status_t do_stop(switch_core_session_t *session, char* bugname);
+
+static void responseHandler(switch_core_session_t* session, 
+	const char* eventName, const char * json, const char* bugname, int finished) {
+	switch_event_t *event;
+	switch_channel_t *channel = switch_core_session_get_channel(session);
+
+	switch_event_create_subclass(&event, SWITCH_EVENT_CUSTOM, eventName);
+	switch_channel_event_set_data(channel, event);
+	switch_event_add_header_string(event, SWITCH_STACK_BOTTOM, "transcription-vendor", "deepgram");
+	switch_event_add_header_string(event, SWITCH_STACK_BOTTOM, "transcription-session-finished", finished ? "true" : "false");
+	if (finished) {
+			switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "responseHandler returning event %s, from finished recognition session\n", eventName);
+	}
+	if (json) switch_event_add_body(event, "%s", json);
+	if (bugname) switch_event_add_header_string(event, SWITCH_STACK_BOTTOM, "media-bugname", bugname);
+	switch_event_fire(&event);
+}
+
+
+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_DEBUG, "Got SWITCH_ABC_TYPE_INIT.\n");
+		break;
+
+	case SWITCH_ABC_TYPE_CLOSE:
+		{
+			private_t *tech_pvt = (private_t*) switch_core_media_bug_get_user_data(bug);
+			switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "Got SWITCH_ABC_TYPE_CLOSE.\n");
+			aws_transcribe_session_stop(session, 1, tech_pvt->bugname);
+			//switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "Finished SWITCH_ABC_TYPE_CLOSE.\n");
+		}
+		break;
+	
+	case SWITCH_ABC_TYPE_READ:
+
+		return aws_transcribe_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, int interim, char* bugname)
+{
+	switch_channel_t *channel = switch_core_session_get_channel(session);
+	switch_media_bug_t *bug;
+	switch_status_t status;
+	switch_codec_implementation_t read_impl = { 0 };
+	void *pUserData;
+	uint32_t samples_per_second;
+
+	if (switch_channel_get_private(channel, bugname)) {
+		switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "removing bug from previous transcribe\n");
+		do_stop(session, bugname);
+	}
+
+	switch_core_session_get_read_impl(session, &read_impl);
+
+	if (switch_channel_pre_answer(channel) != SWITCH_STATUS_SUCCESS) {
+		return SWITCH_STATUS_FALSE;
+	}
+
+	samples_per_second = !strcasecmp(read_impl.iananame, "g722") ? read_impl.actual_samples_per_second : read_impl.samples_per_second;
+
+  switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, " initializing aws speech session.\n");
+	if (SWITCH_STATUS_FALSE == aws_transcribe_session_init(session, responseHandler, samples_per_second, 
+    flags & SMBF_STEREO ? 2 : 1, lang, interim, bugname, &pUserData)) {
+		switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Error initializing aws speech session.\n");
+		return SWITCH_STATUS_FALSE;
+	}
+	if ((status = switch_core_media_bug_add(session, bugname, NULL, capture_callback, pUserData, 0, flags, &bug)) != SWITCH_STATUS_SUCCESS) {
+		return status;
+	}
+  switch_channel_set_private(channel, bugname, bug);
+	switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "added media bug for aws transcribe\n");
+
+	return SWITCH_STATUS_SUCCESS;
+}
+
+static switch_status_t do_stop(switch_core_session_t *session, char* bugname)
+{
+	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, bugname);
+
+	if (bug) {
+		switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "Received user command command to stop transcribe on %s.\n", bugname);
+		status = aws_transcribe_session_stop(session, 0, bugname);
+		switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "stopped transcribe.\n");
+	}
+
+	return status;
+}
+
+#define TRANSCRIBE_API_SYNTAX "<uuid> [start|stop] lang-code [interim] [stereo|mono] [bugname]"
+SWITCH_STANDARD_API(aws_transcribe_function)
+{
+	char *mycmd = NULL, *argv[6] = { 0 };
+	int argc = 0;
+	switch_status_t status = SWITCH_STATUS_FALSE;
+	switch_media_bug_flag_t flags = SMBF_READ_STREAM /* | SMBF_WRITE_STREAM | SMBF_READ_PING */;
+
+	if (!zstr(cmd) && (mycmd = strdup(cmd))) {
+		argc = switch_separate_string(mycmd, ' ', argv, (sizeof(argv) / sizeof(argv[0])));
+	}
+
+	if (zstr(cmd) || 
+      (!strcasecmp(argv[1], "stop") && argc < 2) ||
+      (!strcasecmp(argv[1], "start") && argc < 3) ||
+      zstr(argv[0])) {
+		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", TRANSCRIBE_API_SYNTAX);
+		goto done;
+	} else {
+		switch_core_session_t *lsession = NULL;
+
+		if ((lsession = switch_core_session_locate(argv[0]))) {
+			if (!strcasecmp(argv[1], "stop")) {
+				char *bugname = argc > 2 ? argv[2] : MY_BUG_NAME;
+    		switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_INFO, "stop transcribing\n");
+				status = do_stop(lsession, bugname);
+			} else if (!strcasecmp(argv[1], "start")) {
+        char* lang = argv[2];
+        int interim = argc > 3 && !strcmp(argv[3], "interim");
+				char *bugname = argc > 5 ? argv[5] : MY_BUG_NAME;
+				if (argc > 4 && !strcmp(argv[4], "stereo")) {
+          flags |= SMBF_WRITE_STREAM ;
+          flags |= SMBF_STEREO;
+				}
+    		switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_INFO, "start transcribing %s %s %s\n", lang, interim ? "interim": "complete", bugname);
+				status = start_capture(lsession, flags, lang, interim, bugname);
+			}
+			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;
+}
+
+
+SWITCH_MODULE_LOAD_FUNCTION(mod_aws_transcribe_ws_load)
+{
+	switch_api_interface_t *api_interface;
+
+	/* create/register custom event message type */
+	if (switch_event_reserve_subclass(TRANSCRIBE_EVENT_RESULTS) != SWITCH_STATUS_SUCCESS) {
+		switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Couldn't register subclass %s!\n", TRANSCRIBE_EVENT_RESULTS);
+		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, "AWS Speech Transcription API loading..\n");
+
+  if (SWITCH_STATUS_FALSE == aws_transcribe_init()) {
+		switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_CRIT, "Failed initializing aws speech interface\n");
+	}
+
+	switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_NOTICE, "AWS Speech Transcription API successfully loaded\n");
+
+	SWITCH_ADD_API(api_interface, "uuid_aws_transcribe", "AWS Speech Transcription API", aws_transcribe_function, TRANSCRIBE_API_SYNTAX);
+	switch_console_set_complete("add uuid_aws_transcribe start lang-code [interim|final] [stereo|mono]");
+	switch_console_set_complete("add uuid_aws_transcribe stop ");
+
+	/* 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_aws_transcribe_ws_shutdown() */
+SWITCH_MODULE_SHUTDOWN_FUNCTION(mod_aws_transcribe_ws_shutdown)
+{
+	aws_transcribe_cleanup();
+	switch_event_free_subclass(TRANSCRIBE_EVENT_RESULTS);
+	return SWITCH_STATUS_SUCCESS;
+}
@@ -0,0 +1,60 @@
+#ifndef __MOD_AWS_TRANSCRIBE_WS_H__
+#define __MOD_AWS_TRANSCRIBE_WS_H__
+
+#include <switch.h>
+#include <speex/speex_resampler.h>
+
+#include <unistd.h>
+
+#define MY_BUG_NAME "aws_transcribe_ws"
+#define MAX_BUG_LEN (64)
+#define MAX_SESSION_ID (256)
+#define TRANSCRIBE_EVENT_RESULTS "aws_transcribe::transcription"
+#define TRANSCRIBE_EVENT_END_OF_TRANSCRIPT "aws_transcribe::end_of_transcript"
+#define TRANSCRIBE_EVENT_NO_AUDIO_DETECTED "aws_transcribe::no_audio_detected"
+#define TRANSCRIBE_EVENT_MAX_DURATION_EXCEEDED "aws_transcribe::max_duration_exceeded"
+#define TRANSCRIBE_EVENT_VAD_DETECTED "aws_transcribe::vad_detected"
+#define TRANSCRIBE_EVENT_CONNECT_SUCCESS "aws::connect"
+#define TRANSCRIBE_EVENT_CONNECT_FAIL    "aws::connect_failed"
+#define TRANSCRIBE_EVENT_DISCONNECT      "aws::disconnect"
+#define TRANSCRIBE_EVENT_BUFFER_OVERRUN  "aws::buffer_overrun"
+#define TRANSCRIBE_EVENT_ERROR      "jambonz_transcribe::error"
+
+#define MAX_LANG (12)
+#define MAX_REGION (32)
+#define MAX_WS_URL_LEN (512)
+#define MAX_PATH_LEN (4096)
+#define MAX_SESSION_TOKEN_LEN (4096)
+
+/* per-channel data */
+typedef void (*responseHandler_t)(switch_core_session_t* session, const char* eventName, const char* json, const char* bugname, int finished);
+
+struct private_data {
+	switch_mutex_t *mutex;
+	char sessionId[MAX_SESSION_ID+1];
+  char awsAccessKeyId[128];
+  char awsSecretAccessKey[128];
+  char awsSessionToken[MAX_SESSION_TOKEN_LEN+1];
+  SpeexResamplerState *resampler;
+	responseHandler_t responseHandler;
+	int interim;
+	char lang[MAX_LANG+1];
+	char region[MAX_REGION+1];
+	switch_vad_t * vad;
+	uint32_t samples_per_second;
+  void *pAudioPipe;
+  int ws_state;
+  char host[MAX_WS_URL_LEN+1];
+  unsigned int port;
+  char path[MAX_PATH_LEN+1];
+  char bugname[MAX_BUG_LEN+1];
+  int sampling;
+  int  channels;
+  unsigned int id;
+  int buffer_overrun_notified:1;
+  int is_finished:1;
+};
+
+typedef struct private_data private_t;
+
+#endif
@@ -0,0 +1,51 @@
+/**
+ * (very) simple and limited circular buffer, 
+ * supporting only the use case of doing all of the adds
+ * and then subsquently retrieves.
+ * 
+ */
+class SimpleBuffer {
+  public:
+    SimpleBuffer(uint32_t chunkSize, uint32_t numChunks) : numItems(0),
+    m_numChunks(numChunks), m_chunkSize(chunkSize) {
+      m_pData = new char[chunkSize * numChunks];
+      m_pNextWrite = m_pData;
+    }
+    ~SimpleBuffer() {
+      delete [] m_pData;
+    }
+
+    void add(void *data, uint32_t datalen) {
+      if (datalen % m_chunkSize != 0) return;
+      int numChunks = datalen / m_chunkSize;
+      for (int i = 0; i < numChunks; i++) {
+        memcpy(m_pNextWrite, data, m_chunkSize);
+        data = static_cast<char*>(data) + m_chunkSize;
+        if (numItems < m_numChunks) numItems++;
+
+        uint32_t offset = (m_pNextWrite - m_pData) / m_chunkSize;
+        if (offset >= m_numChunks - 1) m_pNextWrite = m_pData;
+        else m_pNextWrite += m_chunkSize;
+      }
+    }
+
+    char* getNextChunk() {
+      if (numItems--) {
+        char *p = m_pNextWrite;
+        uint32_t offset = (m_pNextWrite - m_pData) / m_chunkSize;
+        if (offset >= m_numChunks - 1) m_pNextWrite = m_pData;
+        else m_pNextWrite += m_chunkSize;
+        return p;
+      }
+      return nullptr;
+    }
+
+    uint32_t getNumItems() { return numItems;}
+
+  private:
+    char *m_pData;
+    uint32_t numItems;
+    uint32_t m_chunkSize;
+    uint32_t m_numChunks;
+    char* m_pNextWrite;
+};
@@ -0,0 +1,324 @@
+#include "transcribe_manager.hpp"
+#include "crc.h"
+
+#include <switch.h>
+
+#include <openssl/sha.h>
+#include <openssl/hmac.h>
+#include <iomanip>
+#include <regex>
+#include <iostream>
+#include <cstring>
+#include <netinet/in.h>
+
+using namespace std;
+
+namespace {
+  std::string uri_encode(const std::string &value) {
+    std::string encoded;
+    char hex[4];
+    for (char c : value) {
+        if (isalnum(c) || c == '-' || c == '_' || c == '.' || c == '~') {
+            encoded += c;
+        } else {
+            sprintf(hex, "%%%02X", c);
+            encoded.append(hex);
+        }
+    }
+    return encoded;
+  }
+}
+// see
+// https://docs.aws.amazon.com/transcribe/latest/dg/websocket.html#websocket-url
+// https://docs.aws.amazon.com/transcribe/latest/dg/event-stream.html
+
+void TranscribeManager::getSignedWebsocketUrl(string& host, string& path, const string& accessKey,
+        const string& secretKey, const string& securityToken, const string& region, const std::string& lang, 
+        const char* vocabularyName, const char* vocabularyFilterName, const char* vocabularyFilterMethod,
+        const char* piiEntities, int shouldIdentifyPiiEntities, const char* languageModelName) {
+    string method = "GET";
+    string service = "transcribe";
+    string endpoint = "wss://transcribestreaming." + region + ".amazonaws.com";
+    host = "transcribestreaming." + region + ".amazonaws.com";
+
+    time_t now = time(0);
+    tm *gmtm = gmtime(&now);
+
+    char amzDate[21];
+    snprintf (amzDate, 21, "%04d%02d%02dT%02d%02d%02dZ",
+            1900 + gmtm->tm_year, 1 + gmtm->tm_mon, gmtm->tm_mday,
+            gmtm->tm_hour, gmtm->tm_min, gmtm->tm_sec);
+
+    char datestamp[9];
+    snprintf (datestamp, 9, "%04d%02d%02d", 1900 + gmtm->tm_year, 1 + gmtm->tm_mon, gmtm->tm_mday);
+
+    string canonical_uri = "/stream-transcription-websocket";
+    string canonical_headers = "host:" + host + "\n";
+    string signed_headers = "host";
+    string algorithm = "AWS4-HMAC-SHA256";
+    string credential_scope = string(datestamp) + "%2F" + region + "%2F" + service + "%2F" + "aws4_request";
+
+    // N.B.: The order of all of these query args are important!
+    // Otherwise, the signature will be invalid.
+    string canonical_querystring  = "X-Amz-Algorithm=" + algorithm;
+    canonical_querystring += "&X-Amz-Credential=" + accessKey + "%2F" + credential_scope;
+    canonical_querystring += "&X-Amz-Date=" + string(amzDate);
+    canonical_querystring += "&X-Amz-Expires=300";
+    canonical_querystring += "&X-Amz-Security-Token=" + uri_encode(securityToken);
+    canonical_querystring += "&X-Amz-SignedHeaders=" + signed_headers;
+
+    if (piiEntities && shouldIdentifyPiiEntities) {
+      canonical_querystring += "&content-redaction-type=PII";
+    }
+    
+    canonical_querystring += "&language-code=" + lang;
+    
+    if (languageModelName) {
+      std::string str(languageModelName);
+      canonical_querystring += "&language-model-name=" + uri_encode(str);
+    }
+
+    canonical_querystring += "&media-encoding=pcm";
+
+    if (piiEntities) {
+      std::string str(piiEntities);
+      canonical_querystring += "&pii-entitytypes=" + uri_encode(str);
+    }
+
+    canonical_querystring += "&sample-rate=8000";
+
+    // custom vocabulary and filter
+    if (vocabularyFilterMethod) {
+      std::string str(vocabularyFilterMethod);
+      canonical_querystring += "&vocabulary-filter-method=" + str;
+    }
+    if (vocabularyFilterName) {
+      std::string str(vocabularyFilterName);
+      canonical_querystring += "&vocabulary-filter-name=" + str;
+    }
+    if (vocabularyName) {
+      std::string str(vocabularyName);
+      canonical_querystring += "&vocabulary-name=" + str;
+    }
+
+    string payload_hash = getSha256("");
+
+    string canonical_request = method + '\n'
+       + canonical_uri + '\n'
+       + canonical_querystring + '\n'
+       + canonical_headers + '\n'
+       + signed_headers + '\n'
+       + payload_hash;
+
+    switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR,"TranscribeManager::getSignedWebsocketUrl canonical_request: %s\n", canonical_request.c_str());
+    string string_to_sign = algorithm + "\n"
+       + amzDate + "\n"
+       + regex_replace(credential_scope, regex("%2F"), "/") + "\n"
+       + getSha256(canonical_request);
+
+    unsigned char signing_key[SHA256_DIGEST_LENGTH];
+    getSignatureKey(signing_key, secretKey, datestamp, region, service);
+
+    unsigned char signatureBinary[SHA256_DIGEST_LENGTH];
+    getHMAC(signatureBinary, signing_key, SHA256_DIGEST_LENGTH, string_to_sign);
+    string signature = toHex(signatureBinary);
+
+    canonical_querystring += "&X-Amz-Signature=" + signature;
+    string request_url = endpoint + canonical_uri + "?" + canonical_querystring;
+    path = canonical_uri + "?" + canonical_querystring;
+
+    return;
+}
+
+string TranscribeManager::getSha256(string str) {
+    SHA256_CTX ctx;
+    SHA256_Init(&ctx);
+    SHA256_Update(&ctx, str.c_str(), str.length());
+    unsigned char hash[SHA256_DIGEST_LENGTH] = { 0 };
+    SHA256_Final(hash, &ctx);
+
+    ostringstream os;
+    os << hex << setfill('0');
+    for (int i = 0; i < SHA256_DIGEST_LENGTH; ++i) {
+        os << setw(2) << static_cast<unsigned int>(hash[i]);
+    }
+
+    return os.str();
+}
+
+void TranscribeManager::getSignatureKey(unsigned char *signatureKey, const string& secretKey,
+            const string& datestamp, const string& region, const string& service) {
+    string key = string("AWS4") + secretKey;
+    unsigned char kDate[SHA256_DIGEST_LENGTH];
+    unsigned char kRegion[SHA256_DIGEST_LENGTH];
+    unsigned char kService[SHA256_DIGEST_LENGTH];
+    unsigned char kSigning[SHA256_DIGEST_LENGTH];
+    getHMAC(kDate, (unsigned char *)key.c_str(), key.length(), datestamp);
+    getHMAC(kRegion, kDate, SHA256_DIGEST_LENGTH, region);
+    getHMAC(kService, kRegion, SHA256_DIGEST_LENGTH, service);
+    getHMAC(kSigning, kService, SHA256_DIGEST_LENGTH, "aws4_request");
+
+    memcpy(signatureKey, kSigning, SHA256_DIGEST_LENGTH);
+}
+
+void TranscribeManager::getHMAC(unsigned char *hmac, unsigned char *key, int keyLen, const string& str) {
+    unsigned char *data = (unsigned char*)str.c_str();
+    unsigned char *result = HMAC(EVP_sha256(), key, keyLen, data, strlen((char *)data), NULL, NULL);
+    memcpy(hmac, result, SHA256_DIGEST_LENGTH);
+}
+
+string TranscribeManager::toHex(unsigned char *hmac) {
+    ostringstream os;
+    os << hex << setfill('0');
+    for (int i = 0; i < SHA256_DIGEST_LENGTH; ++i) {
+        os << setw(2) << static_cast<unsigned int>(hmac[i]);
+    }
+
+    return os.str();
+}
+
+///////////////////////////////////////////////////////////////////////////////////////////
+
+bool TranscribeManager::parseResponse(const string& response, string& payload, bool& isError, bool verbose) {
+    const char* buffer = response.c_str();
+
+    uint32_t totalLen;
+    memcpy(&totalLen, &buffer[0], sizeof(uint32_t));
+    totalLen = ntohl(totalLen);
+
+    uint32_t headerLen;
+    memcpy(&headerLen, &buffer[4], sizeof(uint32_t));
+    headerLen = ntohl(headerLen);
+
+    if (!verifyCRC(buffer, totalLen)) {
+        return false;
+    }
+
+    buffer += 12; // bytes 0 - 11 are prelude
+
+    const int numberOfHeaders = 3;
+    for (int i = 0; i < numberOfHeaders; i++) {
+        parseHeader(&buffer, isError, verbose);
+    }
+
+    payload = string(buffer, totalLen - headerLen - 4*4);
+
+    return true;
+}
+
+bool TranscribeManager::verifyCRC(const char* buffer, const uint32_t totalLength) {
+    uint32_t preludeCRC;
+    memcpy(&preludeCRC, &buffer[8], 4);
+    preludeCRC = ntohl(preludeCRC);
+
+    uint32_t calculatedPreludeCRC = CRC::Calculate(&buffer[0], 8, CRC::CRC_32());
+    if (calculatedPreludeCRC != preludeCRC) {
+        cout << "Prelude CRC didn't match!" << endl;
+        return false;
+    }
+
+    uint32_t messageCRC;
+    memcpy(&messageCRC, &buffer[totalLength - 4], 4);
+    messageCRC = ntohl(messageCRC);
+
+    uint32_t calculatedMessageCRC = CRC::Calculate(buffer, totalLength - 4, CRC::CRC_32());
+
+    if (calculatedMessageCRC != messageCRC) {
+        cout << "Message CRC didn't match!" << endl;
+        return false;
+    }
+
+    return true;
+}
+
+void TranscribeManager::parseHeader(const char** buffer, bool& isError, bool verbose) {
+    uint8_t headerNameLen;
+    memcpy(&headerNameLen, *buffer, sizeof(uint8_t));
+    (*buffer)++;
+
+    string headerName(*buffer, headerNameLen);
+    *buffer += headerNameLen;
+
+    uint8_t headerType;
+    memcpy(&headerType, *buffer, sizeof(uint8_t));
+    (*buffer)++;
+
+    uint16_t headerValLen;
+    memcpy(&headerValLen, *buffer, sizeof(uint16_t));
+    headerValLen = ntohs(headerValLen);
+    *buffer += 2;
+
+    string headerVal(*buffer, headerValLen);
+    *buffer += headerValLen;
+
+    if (headerVal == "exception") {
+        isError = true;
+    }
+    if (verbose) {
+        cout << headerName << "(" << (int)headerType << "): " << headerVal << endl;
+    }
+}
+
+///////////////////////////////////////////////////////////////////////////////////////////
+
+bool TranscribeManager::makeRequest(string& request, const vector<uint8_t>& data) {
+    char preludeAndCrcBuffer[4*3];
+    char headerBuffer[88];
+    char messageCrcBuffer[4];
+
+    // prelude
+    uint32_t totalLen = sizeof(preludeAndCrcBuffer) + sizeof(headerBuffer) + data.size() + sizeof(messageCrcBuffer);
+    uint32_t headerLen = sizeof(headerBuffer);
+    totalLen = htonl(totalLen);
+    headerLen = htonl(headerLen);
+
+    memcpy(&preludeAndCrcBuffer[0], &totalLen, sizeof(uint32_t));
+    memcpy(&preludeAndCrcBuffer[4], &headerLen, sizeof(uint32_t));
+
+    uint32_t preludeCRC = CRC::Calculate(&preludeAndCrcBuffer[0], 8, CRC::CRC_32());
+    preludeCRC = htonl(preludeCRC);
+    memcpy(&preludeAndCrcBuffer[8], &preludeCRC, sizeof(uint32_t));
+
+    // header
+    char* buffer = headerBuffer;
+    writeHeader(&buffer, ":content-type", "application/octet-stream");
+    writeHeader(&buffer, ":event-type", "AudioEvent");
+    writeHeader(&buffer, ":message-type", "event");
+
+    // write everything to response string except for the message CRC
+    request.append(preludeAndCrcBuffer, sizeof(preludeAndCrcBuffer));
+    request.append(headerBuffer, sizeof(headerBuffer));
+    request.append(data.begin(), data.end());
+
+    // message CRC
+    uint32_t messageCRC = CRC::Calculate(request.c_str(), request.length(), CRC::CRC_32());
+    messageCRC = htonl(messageCRC);
+    memcpy(messageCrcBuffer, &messageCRC, sizeof(uint32_t));
+
+    // write message CRC to response string
+    request.append(messageCrcBuffer, sizeof(messageCrcBuffer));
+
+    return true;
+}
+
+void TranscribeManager::writeHeader(char** buffer, const char* key, const char* val) {
+    uint8_t keyLen = strlen(key);
+    uint16_t valueLen = strlen(val);
+
+    memcpy(*buffer, &keyLen, sizeof(uint8_t));
+    (*buffer)++;
+
+    memcpy(*buffer, key, keyLen);
+    (*buffer) += keyLen;
+
+    uint8_t valueType = 7;
+    memcpy(*buffer, &valueType, sizeof(uint8_t));
+    (*buffer)++;
+
+    uint16_t valLen = htons(valueLen);
+    memcpy(*buffer, &valLen, sizeof(uint16_t));
+    (*buffer) += 2;
+
+    memcpy(*buffer, val, valueLen);
+    (*buffer) += valueLen;
+}
@@ -0,0 +1,50 @@
+#ifndef TRANSCRIBEMANAGER_HPP_
+#define TRANSCRIBEMANAGER_HPP_
+
+#include <string>
+#include <vector>
+
+/** Usage
+ #include "transcribe_manager.hpp"
+
+ // get signed URL
+ const string url = TranscribeManager::getSignedWebsocketUrl(accessKey_, secretKey_, region_);
+
+ // connect to the url using a socket library (e.g. https://github.com/machinezone/IXWebSocket)
+
+ // build request string
+ string request;
+ TranscribeManager::makeRequest(request, audioData); // audioData is a const vector<uint8_t>
+
+ // send request to socket
+ * 
+ */
+
+using namespace std;
+
+class TranscribeManager {
+public:
+    static void getSignedWebsocketUrl(string& host, string& path,
+            const std::string& accessKey, const std::string& secretKey, const std::string& securityToken, 
+            const std::string& region, const std::string& lang, const char* vocabularyName,
+            const char* vocabularyFilterName, const char* vocabularyFilterMethod,
+            const char* piiEntities, int shouldIdentifyPiiEntities, const char* languageModelName);
+
+    static bool parseResponse(const std::string& response, std::string& payload, bool& isError, bool verbose = false);
+
+    static bool makeRequest(std::string& request, const std::vector<uint8_t>& data);
+    static void writeHeader(char** buffer, const char* key, const char* val);
+
+private:
+    static std::string getSha256(std::string str);
+    static void getSignatureKey(unsigned char *signatureKey, const std::string& secretKey,
+            const std::string& datestamp, const std::string& region, const std::string& service);
+    static void getHMAC(unsigned char *hmac, unsigned char *key, int keyLen, const std::string& str);
+    static std::string toHex(unsigned char *hmac);
+
+    static bool verifyCRC(const char* buffer, const uint32_t totalLength);
+    static void parseHeader(const char** buffer, bool& isError, bool verbose = false);
+
+};
+
+#endif /* TRANSCRIBEMANAGER_HPP_ */
@@ -64,6 +64,27 @@ std::string secondsToMillisecondsString(double seconds) {
    return std::to_string(milliseconds_long);
 }

+static std::string getEnvVar(const std::string& varName) {
+    const char* val = std::getenv(varName.c_str());
+    return val ? std::string(val) : "";
+}
+
+static long getConnectionTimeout() {
+    std::string connectTimeoutStr = getEnvVar("CUSTOM_TTS_CURL_CONNECT_TIMEOUT");
+
+    if (connectTimeoutStr.empty()) {
+        connectTimeoutStr = getEnvVar("TTS_CURL_CONNECT_TIMEOUT");
+    }
+
+    long connectTimeout = 10L;
+
+    if (!connectTimeoutStr.empty()) {
+        connectTimeout = std::stol(connectTimeoutStr);
+    }
+
+    return connectTimeout;
+}
+
 static CURL* createEasyHandle(void) {
  CURL* easy = curl_easy_init();
  if(!easy) {
@@ -76,7 +97,7 @@ static CURL* createEasyHandle(void) {

  // set connect timeout to 3 seconds and total timeout to 109 seconds
  curl_easy_setopt(easy, CURLOPT_CONNECTTIMEOUT_MS, 3000L);
-  curl_easy_setopt(easy, CURLOPT_TIMEOUT, 10L);
+  curl_easy_setopt(easy, CURLOPT_TIMEOUT, getConnectionTimeout());

  return easy ;    
 }
@@ -451,12 +472,12 @@ static size_t write_cb(void *ptr, size_t size, size_t nmemb, ConnInfo_t *conn) {
      return 0;
    }

-    /* cache file will stay in the mp3 format for size (smaller) and simplicity */
-    if (conn->file) fwrite(data, sizeof(uint8_t), bytes_received, conn->file);
-
    pcm_data = convert_mp3_to_linear(conn, data, bytes_received);
    size_t bytesResampled = pcm_data.size() * sizeof(uint16_t);

+    /* cache same data to avoid streaming and cached audio quality is different*/
+    if (conn->file) fwrite(pcm_data.data(), sizeof(uint8_t), bytesResampled, conn->file);
+
    // Resize the buffer if necessary
    if (cBuffer->capacity() - cBuffer->size() < (bytesResampled / sizeof(uint16_t))) {
      //switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "write_cb growing buffer\n"); 
@@ -65,6 +65,27 @@ std::string secondsToMillisecondsString(double seconds) {
    return std::to_string(milliseconds_long);
 }

+static std::string getEnvVar(const std::string& varName) {
+    const char* val = std::getenv(varName.c_str());
+    return val ? std::string(val) : "";
+}
+
+static long getConnectionTimeout() {
+    std::string connectTimeoutStr = getEnvVar("DEEPGRAM_TTS_CURL_CONNECT_TIMEOUT");
+
+    if (connectTimeoutStr.empty()) {
+        connectTimeoutStr = getEnvVar("TTS_CURL_CONNECT_TIMEOUT");
+    }
+
+    long connectTimeout = 10L;
+
+    if (!connectTimeoutStr.empty()) {
+        connectTimeout = std::stol(connectTimeoutStr);
+    }
+
+    return connectTimeout;
+}
+
 static CURL* createEasyHandle(void) {
  CURL* easy = curl_easy_init();
  if(!easy) {
@@ -77,7 +98,7 @@ static CURL* createEasyHandle(void) {

  // set connect timeout to 3 seconds and total timeout to 109 seconds
  curl_easy_setopt(easy, CURLOPT_CONNECTTIMEOUT_MS, 3000L);
-  curl_easy_setopt(easy, CURLOPT_TIMEOUT, 10L);
+  curl_easy_setopt(easy, CURLOPT_TIMEOUT, getConnectionTimeout());

  return easy ;    
 }
@@ -1,10 +0,0 @@
-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` 
@@ -1,84 +0,0 @@
-# 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)
@@ -1,629 +0,0 @@
-#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);
-	}
-
-}
@@ -1,22 +0,0 @@
-#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
@@ -1,295 +0,0 @@
-/* 
- *
- * 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;
-}
@@ -1,40 +0,0 @@
-#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
@@ -1,425 +0,0 @@
-#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;
-}
@@ -1,107 +0,0 @@
-#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
@@ -378,7 +378,10 @@ switch_status_t playht_parse_text(const std::map<std::string, std::string>& para
    cJSON_AddStringToObject(jResult, "quality", quality.c_str());
  }
  if (!speed.empty()) {
-    cJSON_AddNumberToObject(jResult, "speed", atoi(speed.c_str()));
+      double val = strtod(speed.c_str(), NULL);
+      if (val != 0.0) {
+        cJSON_AddNumberToObject(jResult, "speed", val);
+      }
  }
  if (!seed.empty()) {
    cJSON_AddNumberToObject(jResult, "seed", atoi(seed.c_str()));
@@ -1,11 +1,11 @@
 # mod_google_tts

-A Freeswitch module that allows Google Text-to-Speech API to be used as a tts provider.
+A Freeswitch module that allows Eleven Labs' Text-to-Speech API to be used as a tts provider.

 ## API

 ### Commands
-This freeswitch module does not add any new commands, per se.  Rather, it integrates into the Freeswitch TTS interface such that it is invoked when an application uses the mod_dptools `speak` command with a tts engine of `google_tts` and a voice equal to the language code associated to one of the [supported Wavenet voices](https://cloud.google.com/text-to-speech/docs/voices)
+This freeswitch module does not add any new commands, per se.  Rather, it integrates into the Freeswitch TTS interface such that it is invoked when an application uses the mod_dptools `speak` command with a tts engine of `elevenlabs` and a voice equal to the language code associated to one of the [supported Eleven Labs voices](https://elevenlabs.io/docs/api-reference/query-library)

 ### Events
 None.
@@ -13,11 +13,19 @@ None.
 ## Usage
 When using [drachtio-fsrmf](https://www.npmjs.com/package/drachtio-fsmrf), you can access this functionality via the speak method on the 'endpoint' object.
 ```js
-ep.speak({
-    ttsEngine: 'google_tts',
-    voice: 'en-GB-Wavenet-A',
-    text: 'This aggression will not stand'
-  });
+var text = "Hello World";
+await endpoint.speak({
+    "ttsEngine": 'elevenlabs',
+    "voice": "W9OIfHh5DtdYiZUcFiql",
+    "text": `{use_speaker_boost=1,optimize_streaming_latency=4,style=0.5,stability=0.5,similarity_boost=0.75,api_key=XXYYZZ,model_id=eleven_turbo_v2}${text}`,
+});
 ```
-## Examples
-[google_tts.js](../../examples/google_tts.js)
+## Options
+
+Documentation on these options can be found in Eleven Labs API docs: [Voice Settings](https://elevenlabs.io/docs/speech-synthesis/voice-settings)
+
+- use_speaker_boost
+- optimize_streaming_latency
+- style
+- stability
+- similarity_boost
@@ -168,6 +168,27 @@ int multi_timer_cb(CURLM *multi, long timeout_ms, GlobalInfo_t *g) {
  return 0;
 }

+static std::string getEnvVar(const std::string& varName) {
+    const char* val = std::getenv(varName.c_str());
+    return val ? std::string(val) : "";
+}
+
+static long getConnectionTimeout() {
+    std::string connectTimeoutStr = getEnvVar("ELEVENLABS_TTS_CURL_CONNECT_TIMEOUT");
+
+    if (connectTimeoutStr.empty()) {
+        connectTimeoutStr = getEnvVar("TTS_CURL_CONNECT_TIMEOUT");
+    }
+
+    long connectTimeout = 10L;
+
+    if (!connectTimeoutStr.empty()) {
+        connectTimeout = std::stol(connectTimeoutStr);
+    }
+
+    return connectTimeout;
+}
+
 static CURL* createEasyHandle(void) {
  CURL* easy = curl_easy_init();
  if(!easy) {
@@ -180,7 +201,7 @@ static CURL* createEasyHandle(void) {

  // set connect timeout to 3 seconds and total timeout to 109 seconds
  curl_easy_setopt(easy, CURLOPT_CONNECTTIMEOUT_MS, 3000L);
-  curl_easy_setopt(easy, CURLOPT_TIMEOUT, 10L);
+  curl_easy_setopt(easy, CURLOPT_TIMEOUT, getConnectionTimeout());

  return easy ;    
 }
@@ -233,6 +233,10 @@ static void *SWITCH_THREAD_FUNC grpc_read_thread(switch_thread_t *thread, void *
    auto speech_event_type = response.speech_event_type();
    if (response.has_error()) {
      Status status = response.error();
+      //error 11 is handled in finished session, avoid sending jambonz_transcribe::error event for this here.
+      if (11 == status.code()) {
+        continue;
+      }
      switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "grpc_read_thread: error %s (%d)\n", status.message().c_str(), status.code()) ;
      cJSON* json = cJSON_CreateObject();
      cJSON_AddStringToObject(json, "type", "error");
@@ -229,16 +229,6 @@ int AudioPipe::lws_callback(struct lws *wsi,
 }


-// static members
-static const lws_retry_bo_t retry = {
-    nullptr,   // retry_ms_table
-    0,         // retry_ms_table_count
-    0,         // conceal_count
-    UINT16_MAX,         // secs_since_valid_ping
-    UINT16_MAX,        // secs_since_valid_hangup
-    0          // jitter_percent
-};
-
 struct lws_context *AudioPipe::context = nullptr;
 std::thread AudioPipe::serviceThread;
 std::mutex AudioPipe::mutex_connects;
@@ -376,6 +366,28 @@ bool AudioPipe::lws_service_thread() {
    { NULL, NULL, 0, 0 }
  };

+  uint16_t secs_sinceq_valid_ping = UINT16_MAX;
+  uint16_t secs_since_valid_hangup = UINT16_MAX;
+
+  char* wsVar = std::getenv("WS_PING_INTERVAL");
+  if (wsVar != nullptr) {
+    secs_sinceq_valid_ping = std::atoi(wsVar);
+  }
+
+  wsVar = std::getenv("WS_NO_PONG_HANGUP_INTERVAL");
+  if (wsVar != nullptr) {
+    secs_since_valid_hangup = std::atoi(wsVar);
+  }
+
+  const lws_retry_bo_t retry = {
+    nullptr,                  // retry_ms_table
+    0,                        // retry_ms_table_count
+    0,                        // conceal_count
+    secs_sinceq_valid_ping,   // secs_sinceq_valid_ping
+    secs_since_valid_hangup,  // secs_since_valid_hangup
+    0                         // jitter_percent
+  };
+
  memset(&info, 0, sizeof info); 
  info.port = CONTEXT_PORT_NO_LISTEN; 
  info.options = LWS_SERVER_OPTION_DO_SSL_GLOBAL_INIT;
@@ -64,6 +64,27 @@ std::string secondsToMillisecondsString(double seconds) {
    return std::to_string(milliseconds_long);
 }

+static std::string getEnvVar(const std::string& varName) {
+    const char* val = std::getenv(varName.c_str());
+    return val ? std::string(val) : "";
+}
+
+static long getConnectionTimeout() {
+    std::string connectTimeoutStr = getEnvVar("PLAYHT_TTS_CURL_CONNECT_TIMEOUT");
+
+    if (connectTimeoutStr.empty()) {
+        connectTimeoutStr = getEnvVar("TTS_CURL_CONNECT_TIMEOUT");
+    }
+
+    long connectTimeout = 20L;
+
+    if (!connectTimeoutStr.empty()) {
+        connectTimeout = std::stol(connectTimeoutStr);
+    }
+
+    return connectTimeout;
+}
+
 static CURL* createEasyHandle(void) {
  CURL* easy = curl_easy_init();
  if(!easy) {
@@ -77,9 +98,9 @@ static CURL* createEasyHandle(void) {
  // set connect timeout to 3 seconds and total timeout to 109 seconds
  curl_easy_setopt(easy, CURLOPT_CONNECTTIMEOUT_MS, 3000L);
  //For long text, PlayHT took more than 20 seconds to complete the download.
-  curl_easy_setopt(easy, CURLOPT_TIMEOUT, 60L);
+  curl_easy_setopt(easy, CURLOPT_TIMEOUT, getConnectionTimeout());

-  return easy ;    
+  return easy ;
 }

 static void cleanupConn(ConnInfo_t *conn) {
@@ -452,12 +473,12 @@ static size_t write_cb(void *ptr, size_t size, size_t nmemb, ConnInfo_t *conn) {
      return 0;
    }

-    /* cache file will stay in the mp3 format for size (smaller) and simplicity */
-    if (conn->file) fwrite(data, sizeof(uint8_t), bytes_received, conn->file);
-
    pcm_data = convert_mp3_to_linear(conn, data, bytes_received);
    size_t bytesResampled = pcm_data.size() * sizeof(uint16_t);

+    /* cache same data to avoid streaming and cached audio quality is different*/
+    if (conn->file) fwrite(pcm_data.data(), sizeof(uint8_t), bytesResampled, conn->file);
+
    // Resize the buffer if necessary
    if (cBuffer->capacity() - cBuffer->size() < (bytesResampled / sizeof(uint16_t))) {
      //switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "write_cb growing buffer\n"); 
@@ -774,7 +795,10 @@ extern "C" {
      cJSON_AddStringToObject(jResult, "quality", p->quality);
    }
    if (p->speed) {
-      cJSON_AddNumberToObject(jResult, "speed", atoi(p->speed));
+      double val = strtod(p->speed, NULL);
+      if (val != 0.0) {
+        cJSON_AddNumberToObject(jResult, "speed", val);
+      }
    }
    if (p->seed) {
      cJSON_AddNumberToObject(jResult, "seed", atoi(p->seed));
@@ -935,8 +959,11 @@ extern "C" {
    switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "playht_speech_flush_tts, download complete? %s\n", download_complete ? "yes" : "no") ;
    ConnInfo_t *conn = (ConnInfo_t *) p->conn;
    CircularBuffer_t *cBuffer = (CircularBuffer_t *) p->circularBuffer;
+    // In multi threads, only delete the circular buffer when write and read buffer action finished using it.
+    switch_mutex_lock(p->mutex);
    delete cBuffer;
    p->circularBuffer = nullptr ;
+    switch_mutex_unlock(p->mutex);

    if (conn) {
      conn->flushed = true;
@@ -66,6 +66,27 @@ std::string secondsToMillisecondsString(double seconds) {
    return std::to_string(milliseconds_long);
 }

+static std::string getEnvVar(const std::string& varName) {
+    const char* val = std::getenv(varName.c_str());
+    return val ? std::string(val) : "";
+}
+
+static long getConnectionTimeout() {
+    std::string connectTimeoutStr = getEnvVar("RIMELABS_TTS_CURL_CONNECT_TIMEOUT");
+
+    if (connectTimeoutStr.empty()) {
+        connectTimeoutStr = getEnvVar("TTS_CURL_CONNECT_TIMEOUT");
+    }
+
+    long connectTimeout = 10L;
+
+    if (!connectTimeoutStr.empty()) {
+        connectTimeout = std::stol(connectTimeoutStr);
+    }
+
+    return connectTimeout;
+}
+
 static CURL* createEasyHandle(void) {
  CURL* easy = curl_easy_init();
  if(!easy) {
@@ -78,7 +99,7 @@ static CURL* createEasyHandle(void) {

  // set connect timeout to 3 seconds and total timeout to 109 seconds
  curl_easy_setopt(easy, CURLOPT_CONNECTTIMEOUT_MS, 3000L);
-  curl_easy_setopt(easy, CURLOPT_TIMEOUT, 10L);
+  curl_easy_setopt(easy, CURLOPT_TIMEOUT, getConnectionTimeout());

  return easy ;    
 }
@@ -65,6 +65,27 @@ std::string secondsToMillisecondsString(double seconds) {
    return std::to_string(milliseconds_long);
 }

+static std::string getEnvVar(const std::string& varName) {
+    const char* val = std::getenv(varName.c_str());
+    return val ? std::string(val) : "";
+}
+
+static long getConnectionTimeout() {
+    std::string connectTimeoutStr = getEnvVar("VERBIO_TTS_CURL_CONNECT_TIMEOUT");
+
+    if (connectTimeoutStr.empty()) {
+        connectTimeoutStr = getEnvVar("TTS_CURL_CONNECT_TIMEOUT");
+    }
+
+    long connectTimeout = 10L;
+
+    if (!connectTimeoutStr.empty()) {
+        connectTimeout = std::stol(connectTimeoutStr);
+    }
+
+    return connectTimeout;
+}
+
 static CURL* createEasyHandle(void) {
  CURL* easy = curl_easy_init();
  if(!easy) {
@@ -77,9 +98,9 @@ static CURL* createEasyHandle(void) {

  // set connect timeout to 3 seconds and total timeout to 109 seconds
  curl_easy_setopt(easy, CURLOPT_CONNECTTIMEOUT_MS, 3000L);
-  curl_easy_setopt(easy, CURLOPT_TIMEOUT, 10L);
+  curl_easy_setopt(easy, CURLOPT_TIMEOUT, getConnectionTimeout());

-  return easy ;    
+  return easy ;
 }

 static void cleanupConn(ConnInfo_t *conn) {
@@ -64,6 +64,27 @@ std::string secondsToMillisecondsString(double seconds) {
    return std::to_string(milliseconds_long);
 }

+static std::string getEnvVar(const std::string& varName) {
+    const char* val = std::getenv(varName.c_str());
+    return val ? std::string(val) : "";
+}
+
+static long getConnectionTimeout() {
+    std::string connectTimeoutStr = getEnvVar("WHISPER_TTS_CURL_CONNECT_TIMEOUT");
+
+    if (connectTimeoutStr.empty()) {
+        connectTimeoutStr = getEnvVar("TTS_CURL_CONNECT_TIMEOUT");
+    }
+
+    long connectTimeout = 10L;
+
+    if (!connectTimeoutStr.empty()) {
+        connectTimeout = std::stol(connectTimeoutStr);
+    }
+
+    return connectTimeout;
+}
+
 static CURL* createEasyHandle(void) {
  CURL* easy = curl_easy_init();
  if(!easy) {
@@ -76,7 +97,7 @@ static CURL* createEasyHandle(void) {

  // set connect timeout to 3 seconds and total timeout to 109 seconds
  curl_easy_setopt(easy, CURLOPT_CONNECTTIMEOUT_MS, 3000L);
-  curl_easy_setopt(easy, CURLOPT_TIMEOUT, 10L);
+  curl_easy_setopt(easy, CURLOPT_TIMEOUT, getConnectionTimeout());

  return easy ;    
 }
Author	SHA1	Message	Date
Hoan Luu Huu	7b4520c070	Playht delete circular buffer with mutex check (#123 ) Signed-off-by: root <root@af6633a5cfe8> Co-authored-by: root <root@af6633a5cfe8>	2024-10-05 10:02:07 -04:00
Lyle Pratt	9f7a06ce56	Update README.md (#9 ) Updated ElevenLabs module to include an example of how to use the module as well as links to params in ElevenLabs docs.	2024-10-03 08:06:32 -04:00
Hoan Luu Huu	f7f8f52283	fixed google asr max duration exceeded or no audio raised jambonz_transcribe::error (#120 ) Co-authored-by: root <root@af6633a5cfe8>	2024-10-03 08:04:40 -04:00
Dave Horton	3f06a24b5d	not clearing mark memory properly (#118 ) * not clearing mark memory properly * race condition where mark followed by audio and buffers for mark not completely allocated	2024-09-30 14:44:41 -04:00
Dave Horton	8a3c001b59	wip (#110 ) * wip * #103 wip * wip * wip * support both grpc (legacy) and websockets api for aws transcribe * renaming	2024-09-24 09:51:06 -04:00
Hoan Luu Huu	d17a2aa9be	env variable for tts stream connection timeout (#114 ) * env variable for tts stream connection timeout Signed-off-by: Hoan HL <quan.luuhoang8@gmail.com> * env variable for tts stream connection timeout Signed-off-by: Hoan HL <quan.luuhoang8@gmail.com> * env variable for tts stream connection timeout Signed-off-by: Hoan HL <quan.luuhoang8@gmail.com> --------- Signed-off-by: Hoan HL <quan.luuhoang8@gmail.com>	2024-09-17 10:05:57 -04:00
Hoan Luu Huu	5fd58ba6e5	fixed different voice quality while using playht (#111 )	2024-09-16 15:56:10 -04:00
Dave Horton	4ee08a310a	Feat/mark bidirectional streaming (#102 ) * initial support for mark feature in bidirectional streaming Signed-off-by: Dave Horton <daveh@beachdognet.com> * wip * allow max of 30 marks on any connection * fix send multiple json in same ws text frame (#108) * fix send multiple json in same ws text frame Signed-off-by: Hoan HL <quan.luuhoang8@gmail.com> * fix mark is not sent without more bidirectional audio Signed-off-by: Hoan HL <quan.luuhoang8@gmail.com> --------- Signed-off-by: Hoan HL <quan.luuhoang8@gmail.com> --------- Signed-off-by: Dave Horton <daveh@beachdognet.com> Signed-off-by: Hoan HL <quan.luuhoang8@gmail.com> Co-authored-by: Hoan Luu Huu <110280845+xquanluu@users.noreply.github.com>	2024-09-04 16:10:41 +01:00
rammohan-y	d01991ed0f	feat/106: converting playht speed option to float instead of integer (#107 ) * feat/106: converting playht speed option to float instead of integer * feat/106: Using strtod * feat/106: Using strtod Signed-off-by: Rammohan Yadavalli <rammohan.yadavalli@kore.com> --------- Signed-off-by: Rammohan Yadavalli <rammohan.yadavalli@kore.com> Signed-off-by: Rammohan Yadavalli <rammohan.yadavalli@gmail.com> Co-authored-by: Rammohan Yadavalli <rammohan.yadavalli@kore.com>	2024-08-23 10:49:55 -04:00
Hoan Luu Huu	eec4df4b77	support variable to enable ws ping pong for jb transcribe (#101 ) * support variable to enable ws ping pong for jb transcribe * update ping/pong time duration * fix review comment Signed-off-by: Hoan HL <quan.luuhoang8@gmail.com> --------- Signed-off-by: Hoan HL <quan.luuhoang8@gmail.com>	2024-08-22 07:33:05 -04:00