fix: return empty deepgram transcripts that signify an endpoint

Signed-off-by: Dave Horton <daveh@beachdognet.com>
update default model selection to latest api (#5 )
2026-01-25 02:08:27 +00:00 · 2024-01-14 13:31:20 -05:00 · 2024-01-14 10:34:01 -05:00 · 2023-12-27 11:10:57 -05:00 · 2023-12-26 20:35:09 -05:00 · 2023-12-26 12:12:01 -05:00
22 changed files with 444 additions and 398 deletions
@@ -1,6 +1,6 @@
 # Contributing 

-In order to have a concrete record that your contribution is intentional and you agree to license it under the same terms as the project's license, we've adopted the same lightweight approach that the Linux Kernel submitting patches process, Docker, and many other projects use: the [Developer Certificate of Origin](https://developercertificate.org/), or DCO. This is a simple declaration that you wrote the contribution or otherwise have the right to contribute it to this project.
+In order to have a concrete record that your contribution is intentional and you agree to license it under the same terms as [the project's license](./COPYING.md), we've adopted the same lightweight approach that the Linux Kernel submitting patches process, Docker, and many other projects use: the [Developer Certificate of Origin](https://developercertificate.org/), or DCO. This is a simple declaration that you wrote the contribution or otherwise have the right to contribute it to this project.

 If you agree to this for your contribution, then all that's needed is to include the line in your commit or pull request comment:

@@ -7,11 +7,14 @@ under different conditions.
 1. MIT License: 
   - This license applies only when the software is used exclusively in conjunction with a [jambonz](https://jambonz.org) voice server. 
   Specifically, that means all of the following conditions must be true:
+   
      i) The software is compiled and dynamically loaded into a running Freeswitch process 
      that is dedicated to providing services to a jambonz server or cluster.
+
      ii) All incoming calls to the Freeswitch instance running this software are 
      controlled via an outbound ESL socket connection from Freeswitch 
      to a Node.js application running the [jambonz feature server](https://github.com/jambonz/jambonz-feature-server) application. 
+
   - For the full terms of the MIT License, see the [`LICENSE_MIT`](LICENSE_MIT) file.

 2. AGPL Version 3.0: 
@@ -1,4 +1,4 @@
-Under specific conditions that are described [here](https://github.com/jambonz/freeswitch-modules/blob/main/COPYING), this software is licensed under the MIT License below.
+Under specific conditions that are described here: https://github.com/jambonz/freeswitch-modules/blob/main/COPYING.md, this software is licensed under the MIT License below.

 The MIT License (MIT)

@@ -6,8 +6,8 @@ A collection of Freeswitch modules intended for use with a [jambonz](https://jam

 This software is available under a dual-licensing scheme.  For specific use in a standalone [jambonz](https://jambonz.org) deployment, the [MIT License](./LICENSE_MIT) applies.  For all other uses, the software is licensed for use under the [AGPL Version 3.0 license](./LICENSE_AGPL-3.0).

-Please review [COPYING](./COPYING) for detailed information in order to assess whether your intended usage meets the specific conditions that allow for usage under the MIT License.
+Please review [COPYING](./COPYING.md) for detailed information in order to assess whether your intended usage meets the specific conditions that allow for usage under the MIT License.

 ## Contributing to the software

-If you wish to contribute changes, please review [our rules](./CONTRIBUTING.md).
+If you wish to contribute changes, please review [our rules](./CONTRIBUTING.md) governing contributions.
@@ -102,11 +102,12 @@ namespace {
    return path;
  }

-  static void eventCallback(const char* sessionId, assemblyai::AudioPipe::NotifyEvent_t event, const char* message, bool finished) {
+  static void eventCallback(const char* sessionId, const char* bugname, 
+    assemblyai::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, MY_BUG_NAME);
+      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) {
@@ -194,7 +195,8 @@ namespace {
    tech_pvt->channels = channels;
    tech_pvt->id = ++idxCallCount;
    tech_pvt->buffer_overrun_notified = 0;
-    
+    strncpy(tech_pvt->bugname, bugname, MAX_BUG_LEN);
+
    size_t buflen = LWS_PRE + (FRAME_SIZE_8000 * desiredSampling / 8000 * channels * 1000 / RTP_PACKETIZATION_PERIOD * nAudioBufferSecs);

    const char* apiKey = switch_channel_get_variable(channel, "ASSEMBLYAI_API_KEY");
@@ -204,7 +206,7 @@ namespace {
      return SWITCH_STATUS_FALSE;
    }

-    assemblyai::AudioPipe* ap = new assemblyai::AudioPipe(tech_pvt->sessionId, tech_pvt->host, tech_pvt->port, tech_pvt->path, 
+    assemblyai::AudioPipe* ap = new assemblyai::AudioPipe(tech_pvt->sessionId, bugname, tech_pvt->host, tech_pvt->port, tech_pvt->path, 
      buflen, read_impl.decoded_bytes_per_packet, apiKey, eventCallback);
    if (!ap) {
      switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_ERROR, "Error allocating AudioPipe\n");
@@ -251,7 +253,8 @@ extern "C" {
  switch_status_t aai_transcribe_init() {
    switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_NOTICE, "mod_assemblyai_transcribe: audio buffer (in secs):    %d secs\n", nAudioBufferSecs);
 
-    int logs = LLL_ERR | LLL_WARN | LLL_NOTICE || LLL_INFO | LLL_PARSER | LLL_HEADER | LLL_EXT | LLL_CLIENT  | LLL_LATENCY | LLL_DEBUG ;
+    int logs = LLL_ERR | LLL_WARN | LLL_NOTICE ;
+    //| LLL_INFO | LLL_PARSER | LLL_HEADER | LLL_EXT | LLL_CLIENT  | LLL_LATENCY | LLL_DEBUG ;
    
    assemblyai::AudioPipe::initialize(logs, lws_logger);
 		switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_NOTICE, "AudioPipe::initialize completed\n");
@@ -306,7 +309,7 @@ extern "C" {

 	switch_status_t aai_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);
+    switch_media_bug_t *bug = (switch_media_bug_t*) switch_channel_get_private(channel, bugname);
    if (!bug) {
      switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "aai_transcribe_session_stop: no bug - websocket conection already closed\n");
      return SWITCH_STATUS_FALSE;
@@ -72,7 +72,7 @@ int AudioPipe::lws_callback(struct lws *wsi,
        lwsl_err("AudioPipe::lws_service_thread LWS_CALLBACK_CLIENT_CONNECTION_ERROR: %s, response status %d\n", in ? (char *)in : "(null)", rc); 
        if (ap) {
          ap->m_state = LWS_CLIENT_FAILED;
-          ap->m_callback(ap->m_uuid.c_str(), AudioPipe::CONNECT_FAIL, (char *) in, ap->isFinished());
+          ap->m_callback(ap->m_uuid.c_str(), ap->m_bugname.c_str(), AudioPipe::CONNECT_FAIL, (char *) in, ap->isFinished());
        }
        else {
          lwsl_err("AudioPipe::lws_service_thread LWS_CALLBACK_CLIENT_CONNECTION_ERROR unable to find wsi %p..\n", wsi); 
@@ -87,7 +87,7 @@ int AudioPipe::lws_callback(struct lws *wsi,
          *ppAp = ap;
          ap->m_vhd = vhd;
          ap->m_state = LWS_CLIENT_CONNECTED;
-          ap->m_callback(ap->m_uuid.c_str(), AudioPipe::CONNECT_SUCCESS, NULL,  ap->isFinished());
+          ap->m_callback(ap->m_uuid.c_str(), ap->m_bugname.c_str(), AudioPipe::CONNECT_SUCCESS, NULL,  ap->isFinished());
        }
        else {
          lwsl_err("AudioPipe::lws_service_thread LWS_CALLBACK_CLIENT_ESTABLISHED %s unable to find wsi %p..\n", ap->m_uuid.c_str(), wsi); 
@@ -105,12 +105,12 @@ int AudioPipe::lws_callback(struct lws *wsi,
          // closed by us

          lwsl_debug("%s socket closed by us\n", ap->m_uuid.c_str());
-          ap->m_callback(ap->m_uuid.c_str(), AudioPipe::CONNECTION_CLOSED_GRACEFULLY, NULL,  ap->isFinished());
+          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
          lwsl_info("%s socket closed by far end\n", ap->m_uuid.c_str());
-          ap->m_callback(ap->m_uuid.c_str(), AudioPipe::CONNECTION_DROPPED, NULL,  ap->isFinished());
+          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();
@@ -172,7 +172,7 @@ int AudioPipe::lws_callback(struct lws *wsi,
          if (lws_is_final_fragment(wsi)) {
            if (nullptr != ap->m_recv_buf) {
              std::string msg((char *)ap->m_recv_buf, ap->m_recv_buf_ptr - ap->m_recv_buf);
-              ap->m_callback(ap->m_uuid.c_str(), AudioPipe::MESSAGE, msg.c_str(),  ap->isFinished());
+              ap->m_callback(ap->m_uuid.c_str(), ap->m_bugname.c_str(), AudioPipe::MESSAGE, msg.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;
@@ -261,6 +261,7 @@ static const lws_retry_bo_t retry = {
 };

 struct lws_context *AudioPipe::context = nullptr;
+std::thread AudioPipe::serviceThread;
 std::string AudioPipe::protocolName;
 std::mutex AudioPipe::mutex_connects;
 std::mutex AudioPipe::mutex_disconnects;
@@ -429,26 +430,28 @@ bool AudioPipe::lws_service_thread() {

 void AudioPipe::initialize(int loglevel, log_emit_function logger) {

-  lws_set_log_level(loglevel, logger);
+  //lws_set_log_level(loglevel, logger);

  lwsl_notice("AudioPipe::initialize starting\n"); 
  std::lock_guard<std::mutex> lock(mapMutex);
-  std::thread t(&AudioPipe::lws_service_thread);
  stopFlag = false;
-  t.detach();
+  serviceThread = std::thread(&AudioPipe::lws_service_thread);
 }

 bool AudioPipe::deinitialize() {
  lwsl_notice("AudioPipe::deinitialize\n"); 
  std::lock_guard<std::mutex> lock(mapMutex);
  stopFlag = true;
+    if (serviceThread.joinable()) {
+    serviceThread.join();
+  }
  return true;
 }

 // instance members
-AudioPipe::AudioPipe(const char* uuid, const char* host, unsigned int port, const char* path,
+AudioPipe::AudioPipe(const char* uuid, const char* bugname, const char* host, unsigned int port, const char* path,
  size_t bufLen, size_t minFreespace, const char* apiKey, notifyHandler_t callback) :
-  m_uuid(uuid), m_host(host), m_port(port), m_path(path), m_finished(false),
+  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_audio_buffer_write_offset(LWS_PRE), m_recv_buf(nullptr), m_recv_buf_ptr(nullptr), 
  m_state(LWS_CLIENT_IDLE), m_wsi(nullptr), m_vhd(nullptr), m_apiKey(apiKey), m_callback(callback) {
@@ -31,7 +31,7 @@ public:
    MESSAGE
  };
  typedef void (*log_emit_function)(int level, const char *line);
-  typedef void (*notifyHandler_t)(const char *sessionId, NotifyEvent_t event, const char* message, bool finished);
+  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;
@@ -44,7 +44,7 @@ public:
  static bool lws_service_thread();

  // constructor
-  AudioPipe(const char* uuid, const char* host, unsigned int port, const char* path, 
+  AudioPipe(const char* uuid, const char* bugname, const char* host, unsigned int port, const char* path, 
    size_t bufLen, size_t minFreespace, const char* apiKey, notifyHandler_t callback);
  ~AudioPipe();  

@@ -86,6 +86,7 @@ public:
  void operator=(const AudioPipe&) = delete;

 private:
+  static std::thread serviceThread;

  static int lws_callback(struct lws *wsi, enum lws_callback_reasons reason, void *user, void *in, size_t len); 
  static struct lws_context *context;
@@ -74,7 +74,7 @@ static switch_status_t start_capture(switch_core_session_t *session, switch_medi
 	void *pUserData;
 	uint32_t samples_per_second;

-	if (switch_channel_get_private(channel, MY_BUG_NAME)) {
+	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);
 	}
@@ -94,7 +94,7 @@ static switch_status_t start_capture(switch_core_session_t *session, switch_medi
 	if ((status = switch_core_media_bug_add(session, "aai_transcribe", NULL, capture_callback, pUserData, 0, flags, &bug)) != SWITCH_STATUS_SUCCESS) {
 		return status;
 	}
-  switch_channel_set_private(channel, MY_BUG_NAME, bug);
+  switch_channel_set_private(channel, bugname, bug);
 	switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "added media bug for assemblyai transcribe\n");

 	return SWITCH_STATUS_SUCCESS;
@@ -105,7 +105,7 @@ 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, MY_BUG_NAME);
+	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.\n");
@@ -7,6 +7,7 @@
 #define MAX_RECV_BUF_SIZE (65 * 1024 * 10)
 #define RECV_BUF_REALLOC_SIZE (8 * 1024)

+using namespace drachtio;

 namespace {
  static const char* basicAuthUser = std::getenv("MOD_AUDIO_FORK_HTTP_AUTH_USER");
@@ -267,6 +268,7 @@ static const lws_retry_bo_t retry = {
 };

 struct lws_context *AudioPipe::context = nullptr;
+std::thread AudioPipe::serviceThread;
 std::string AudioPipe::protocolName;
 std::mutex AudioPipe::mutex_connects;
 std::mutex AudioPipe::mutex_disconnects;
@@ -436,19 +438,21 @@ bool AudioPipe::lws_service_thread() {

 void AudioPipe::initialize(const char* protocol, int loglevel, log_emit_function logger) {
  protocolName = protocol;
-  lws_set_log_level(loglevel, logger);
+  //lws_set_log_level(loglevel, logger);

  lwsl_notice("AudioPipe::initialize starting\n"); 
  std::lock_guard<std::mutex> lock(mapMutex);
-  std::thread t(&AudioPipe::lws_service_thread);
  stopFlag = false;
-  t.detach();
+  serviceThread = std::thread(&AudioPipe::lws_service_thread);
 }

 bool AudioPipe::deinitialize() {
  lwsl_notice("AudioPipe::deinitialize\n"); 
  std::lock_guard<std::mutex> lock(mapMutex);
  stopFlag = true;
+  if (serviceThread.joinable()) {
+    serviceThread.join();
+  }
  return true;
 }

@@ -10,135 +10,140 @@

 #include <libwebsockets.h>

-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);
+namespace drachtio {

-  struct lws_per_vhost_data {
-    struct lws_context *context;
-    struct lws_vhost *vhost;
-    const struct lws_protocols *protocol;
+  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);
+
+    struct lws_per_vhost_data {
+      struct lws_context *context;
+      struct lws_vhost *vhost;
+      const struct lws_protocols *protocol;
+    };
+
+    static void initialize(const char* protocolName, int loglevel, log_emit_function logger);
+    static bool deinitialize();
+    static bool lws_service_thread();
+
+    // constructor
+    AudioPipe(const char* uuid, const char* host, unsigned int port, const char* path, int sslFlags, 
+      size_t bufLen, size_t minFreespace, const char* username, const char* password, char* bugname, notifyHandler_t callback);
+    ~AudioPipe();  
+
+    LwsState_t getLwsState(void) { return m_state; }
+    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) {
+      m_audio_buffer_write_offset = 0;
+    }
+    void lockAudioBuffer(void) {
+      m_audio_mutex.lock();
+    }
+    void unlockAudioBuffer(void) ;
+    bool hasBasicAuth(void) {
+      return !m_username.empty() && !m_password.empty();
+    }
+
+    void getBasicAuth(std::string& username, std::string& password) {
+      username = m_username;
+      password = m_password;
+    }
+
+    void do_graceful_shutdown();
+    bool isGracefulShutdown(void) {
+      return m_gracefulShutdown;
+    }
+
+    void close() ;
+
+    // no default constructor or copying
+    AudioPipe() = delete;
+    AudioPipe(const AudioPipe&) = delete;
+    void operator=(const AudioPipe&) = delete;
+
+  private:
+    static std::thread serviceThread;
+
+    static int lws_callback(struct lws *wsi, enum lws_callback_reasons reason, void *user, void *in, size_t len); 
+    static struct lws_context *context;
+    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 bool stopFlag;
+
+    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;
+    std::string m_bugname;
+    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_username;
+    std::string m_password;
+    bool m_gracefulShutdown;
  };

-  static void initialize(const char* protocolName, int loglevel, log_emit_function logger);
-  static bool deinitialize();
-  static bool lws_service_thread();
-
-  // constructor
-  AudioPipe(const char* uuid, const char* host, unsigned int port, const char* path, int sslFlags, 
-    size_t bufLen, size_t minFreespace, const char* username, const char* password, char* bugname, notifyHandler_t callback);
-  ~AudioPipe();  
-
-  LwsState_t getLwsState(void) { return m_state; }
-  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) {
-    m_audio_buffer_write_offset = 0;
-  }
-  void lockAudioBuffer(void) {
-    m_audio_mutex.lock();
-  }
-  void unlockAudioBuffer(void) ;
-  bool hasBasicAuth(void) {
-    return !m_username.empty() && !m_password.empty();
-  }
-
-  void getBasicAuth(std::string& username, std::string& password) {
-    username = m_username;
-    password = m_password;
-  }
-
-  void do_graceful_shutdown();
-  bool isGracefulShutdown(void) {
-    return m_gracefulShutdown;
-  }
-
-  void close() ;
-
-  // no default constructor or copying
-  AudioPipe() = delete;
-  AudioPipe(const AudioPipe&) = delete;
-  void operator=(const AudioPipe&) = delete;
-
-private:
-
-  static int lws_callback(struct lws *wsi, enum lws_callback_reasons reason, void *user, void *in, size_t len); 
-  static struct lws_context *context;
-  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 bool stopFlag;
-
-  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;
-  std::string m_bugname;
-  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_username;
-  std::string m_password;
-  bool m_gracefulShutdown;
-};
+} // namespace drachtio

 #endif
@@ -156,7 +156,7 @@ namespace {
    }
  }

-  static void eventCallback(const char* sessionId, const char* bugname, AudioPipe::NotifyEvent_t event, const char* message) {
+  static void eventCallback(const char* sessionId, const char* bugname, drachtio::AudioPipe::NotifyEvent_t event, const char* message) {
    switch_core_session_t* session = switch_core_session_locate(sessionId);
    if (session) {
      switch_channel_t *channel = switch_core_session_get_channel(session);
@@ -165,16 +165,16 @@ namespace {
        private_t* tech_pvt = (private_t*) switch_core_media_bug_get_user_data(bug);
        if (tech_pvt) {
          switch (event) {
-            case AudioPipe::CONNECT_SUCCESS:
+            case drachtio::AudioPipe::CONNECT_SUCCESS:
              switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_INFO, "connection successful\n");
              tech_pvt->responseHandler(session, EVENT_CONNECT_SUCCESS, NULL);
              if (strlen(tech_pvt->initialMetadata) > 0) {
                switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "sending initial metadata %s\n", tech_pvt->initialMetadata);
-                AudioPipe *pAudioPipe = static_cast<AudioPipe *>(tech_pvt->pAudioPipe);
+                drachtio::AudioPipe *pAudioPipe = static_cast<drachtio::AudioPipe *>(tech_pvt->pAudioPipe);
                pAudioPipe->bufferForSending(tech_pvt->initialMetadata);
              }
            break;
-            case AudioPipe::CONNECT_FAIL:
+            case drachtio::AudioPipe::CONNECT_FAIL:
            {
              // first thing: we can no longer access the AudioPipe
              std::stringstream json;
@@ -184,18 +184,18 @@ namespace {
              switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_NOTICE, "connection failed: %s\n", message);
            }
            break;
-            case AudioPipe::CONNECTION_DROPPED:
+            case drachtio::AudioPipe::CONNECTION_DROPPED:
              // first thing: we can no longer access the AudioPipe
              tech_pvt->pAudioPipe = nullptr;
              tech_pvt->responseHandler(session, EVENT_DISCONNECT, NULL);
              switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_NOTICE, "connection dropped from far end\n");
            break;
-            case AudioPipe::CONNECTION_CLOSED_GRACEFULLY:
+            case drachtio::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 AudioPipe::MESSAGE:
+            case drachtio::AudioPipe::MESSAGE:
              processIncomingMessage(tech_pvt, session, message);
            break;
          }
@@ -239,7 +239,7 @@ namespace {
    
    size_t buflen = LWS_PRE + (FRAME_SIZE_8000 * desiredSampling / 8000 * channels * 1000 / RTP_PACKETIZATION_PERIOD * nAudioBufferSecs);

-    AudioPipe* ap = new AudioPipe(tech_pvt->sessionId, host, port, path, sslFlags, 
+    drachtio::AudioPipe* ap = new drachtio::AudioPipe(tech_pvt->sessionId, host, port, path, sslFlags, 
      buflen, read_impl.decoded_bytes_per_packet, username, password, bugname, eventCallback);
    if (!ap) {
      switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_ERROR, "Error allocating AudioPipe\n");
@@ -374,13 +374,17 @@ extern "C" {
 
    int logs = LLL_ERR | LLL_WARN | LLL_NOTICE ;
     //LLL_INFO | LLL_PARSER | LLL_HEADER | LLL_EXT | LLL_CLIENT  | LLL_LATENCY | LLL_DEBUG ;
-    AudioPipe::initialize(mySubProtocolName, logs, lws_logger);
-   return SWITCH_STATUS_SUCCESS;
+    drachtio::AudioPipe::initialize(mySubProtocolName, logs, lws_logger);
+    switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_NOTICE, "mod_audio_fork successfully initialized\n");
+    return SWITCH_STATUS_SUCCESS;
  }

  switch_status_t fork_cleanup() {
    bool cleanup = false;
-    cleanup = AudioPipe::deinitialize();
+    switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_NOTICE, "mod_audio_fork unloading..\n");
+
+    cleanup = drachtio::AudioPipe::deinitialize();
+    switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_NOTICE, "mod_audio_fork unloaded status %d\n", cleanup);
    if (cleanup == true) {
        return SWITCH_STATUS_SUCCESS;
    }
@@ -420,7 +424,7 @@ extern "C" {

   switch_status_t fork_session_connect(void **ppUserData) {
    private_t *tech_pvt = static_cast<private_t *>(*ppUserData);
-    AudioPipe *pAudioPipe = static_cast<AudioPipe*>(tech_pvt->pAudioPipe);
+    drachtio::AudioPipe *pAudioPipe = static_cast<drachtio::AudioPipe*>(tech_pvt->pAudioPipe);
    pAudioPipe->connect();
    return SWITCH_STATUS_SUCCESS;
  }
@@ -438,7 +442,7 @@ extern "C" {
    switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "(%u) fork_session_cleanup\n", id);

    if (!tech_pvt) return SWITCH_STATUS_FALSE;
-    AudioPipe *pAudioPipe = static_cast<AudioPipe *>(tech_pvt->pAudioPipe);
+    drachtio::AudioPipe *pAudioPipe = static_cast<drachtio::AudioPipe *>(tech_pvt->pAudioPipe);
      
    switch_mutex_lock(tech_pvt->mutex);

@@ -481,7 +485,7 @@ extern "C" {
    private_t* tech_pvt = (private_t*) switch_core_media_bug_get_user_data(bug);
  
    if (!tech_pvt) return SWITCH_STATUS_FALSE;
-    AudioPipe *pAudioPipe = static_cast<AudioPipe *>(tech_pvt->pAudioPipe);
+    drachtio::AudioPipe *pAudioPipe = static_cast<drachtio::AudioPipe *>(tech_pvt->pAudioPipe);
    if (pAudioPipe && text) pAudioPipe->bufferForSending(text);

    return SWITCH_STATUS_SUCCESS;
@@ -516,7 +520,7 @@ extern "C" {

    tech_pvt->graceful_shutdown = 1;

-    AudioPipe *pAudioPipe = static_cast<AudioPipe *>(tech_pvt->pAudioPipe);
+    drachtio::AudioPipe *pAudioPipe = static_cast<drachtio::AudioPipe *>(tech_pvt->pAudioPipe);
    if (pAudioPipe) pAudioPipe->do_graceful_shutdown();

    return SWITCH_STATUS_SUCCESS;
@@ -535,8 +539,8 @@ extern "C" {
        switch_mutex_unlock(tech_pvt->mutex);
        return SWITCH_TRUE;
      }
-      AudioPipe *pAudioPipe = static_cast<AudioPipe *>(tech_pvt->pAudioPipe);
-      if (pAudioPipe->getLwsState() != AudioPipe::LWS_CLIENT_CONNECTED) {
+      drachtio::AudioPipe *pAudioPipe = static_cast<drachtio::AudioPipe *>(tech_pvt->pAudioPipe);
+      if (pAudioPipe->getLwsState() != drachtio::AudioPipe::LWS_CLIENT_CONNECTED) {
        switch_mutex_unlock(tech_pvt->mutex);
        return SWITCH_TRUE;
      }
@@ -71,7 +71,7 @@ int AudioPipe::lws_callback(struct lws *wsi,
        lwsl_err("AudioPipe::lws_service_thread LWS_CALLBACK_CLIENT_CONNECTION_ERROR: %s, response status %d\n", in ? (char *)in : "(null)", rc); 
        if (ap) {
          ap->m_state = LWS_CLIENT_FAILED;
-          ap->m_callback(ap->m_uuid.c_str(), AudioPipe::CONNECT_FAIL, (char *) in, ap->isFinished());
+          ap->m_callback(ap->m_uuid.c_str(),  ap->m_bugname.c_str(), deepgram::AudioPipe::CONNECT_FAIL, (char *) in, ap->isFinished());
        }
        else {
          lwsl_err("AudioPipe::lws_service_thread LWS_CALLBACK_CLIENT_CONNECTION_ERROR unable to find wsi %p..\n", wsi); 
@@ -82,11 +82,12 @@ int AudioPipe::lws_callback(struct lws *wsi,
    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(), AudioPipe::CONNECT_SUCCESS, NULL,  ap->isFinished());
+          ap->m_callback(ap->m_uuid.c_str(), ap->m_bugname.c_str(), deepgram::AudioPipe::CONNECT_SUCCESS, NULL,  ap->isFinished());
        }
        else {
          lwsl_err("AudioPipe::lws_service_thread LWS_CALLBACK_CLIENT_ESTABLISHED %s unable to find wsi %p..\n", ap->m_uuid.c_str(), wsi); 
@@ -96,6 +97,7 @@ int AudioPipe::lws_callback(struct lws *wsi,
    case LWS_CALLBACK_CLIENT_CLOSED:
      {
        AudioPipe* ap = *ppAp;
+
        if (!ap) {
          lwsl_err("AudioPipe::lws_service_thread LWS_CALLBACK_CLIENT_CLOSED %s unable to find wsi %p..\n", ap->m_uuid.c_str(), wsi); 
          return 0;
@@ -104,12 +106,12 @@ int AudioPipe::lws_callback(struct lws *wsi,
          // closed by us

          lwsl_debug("%s socket closed by us\n", ap->m_uuid.c_str());
-          ap->m_callback(ap->m_uuid.c_str(), AudioPipe::CONNECTION_CLOSED_GRACEFULLY, NULL,  ap->isFinished());
+          ap->m_callback(ap->m_uuid.c_str(),  ap->m_bugname.c_str(), deepgram::AudioPipe::CONNECTION_CLOSED_GRACEFULLY, NULL,  ap->isFinished());
        }
        else if (ap->m_state == LWS_CLIENT_CONNECTED) {
          // closed by far end
          lwsl_info("%s socket closed by far end\n", ap->m_uuid.c_str());
-          ap->m_callback(ap->m_uuid.c_str(), AudioPipe::CONNECTION_DROPPED, NULL,  ap->isFinished());
+          ap->m_callback(ap->m_uuid.c_str(),  ap->m_bugname.c_str(), deepgram::AudioPipe::CONNECTION_DROPPED, NULL,  ap->isFinished());
        }
        ap->m_state = LWS_CLIENT_DISCONNECTED;
        ap->setClosed();
@@ -125,6 +127,7 @@ int AudioPipe::lws_callback(struct lws *wsi,
    case LWS_CALLBACK_CLIENT_RECEIVE:
      {
        AudioPipe* ap = *ppAp;
+
        if (!ap) {
          lwsl_err("AudioPipe::lws_service_thread LWS_CALLBACK_CLIENT_RECEIVE %s unable to find wsi %p..\n", ap->m_uuid.c_str(), wsi); 
          return 0;
@@ -171,7 +174,7 @@ int AudioPipe::lws_callback(struct lws *wsi,
          if (lws_is_final_fragment(wsi)) {
            if (nullptr != ap->m_recv_buf) {
              std::string msg((char *)ap->m_recv_buf, ap->m_recv_buf_ptr - ap->m_recv_buf);
-              ap->m_callback(ap->m_uuid.c_str(), AudioPipe::MESSAGE, msg.c_str(),  ap->isFinished());
+              ap->m_callback(ap->m_uuid.c_str(),  ap->m_bugname.c_str(), deepgram::AudioPipe::MESSAGE, msg.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;
@@ -184,6 +187,7 @@ int AudioPipe::lws_callback(struct lws *wsi,
    case LWS_CALLBACK_CLIENT_WRITEABLE:
      {
        AudioPipe* ap = *ppAp;
+
        if (!ap) {
          lwsl_err("AudioPipe::lws_service_thread LWS_CALLBACK_CLIENT_WRITEABLE %s unable to find wsi %p..\n", ap->m_uuid.c_str(), wsi); 
          return 0;
@@ -251,7 +255,7 @@ static const lws_retry_bo_t retry = {
 };

 struct lws_context *AudioPipe::context = nullptr;
-std::string AudioPipe::protocolName;
+std::thread AudioPipe::serviceThread;
 std::mutex AudioPipe::mutex_connects;
 std::mutex AudioPipe::mutex_disconnects;
 std::mutex AudioPipe::mutex_writes;
@@ -420,26 +424,29 @@ bool AudioPipe::lws_service_thread() {

 void AudioPipe::initialize(int loglevel, log_emit_function logger) {

-  lws_set_log_level(loglevel, logger);
+  //lws_set_log_level(loglevel, logger);

  lwsl_notice("AudioPipe::initialize starting\n"); 
  std::lock_guard<std::mutex> lock(mapMutex);
-  std::thread t(&AudioPipe::lws_service_thread);
  stopFlag = false;
-  t.detach();
+  serviceThread = std::thread(&AudioPipe::lws_service_thread);
 }

 bool AudioPipe::deinitialize() {
  lwsl_notice("AudioPipe::deinitialize\n"); 
  std::lock_guard<std::mutex> lock(mapMutex);
  stopFlag = true;
+  if (serviceThread.joinable()) {
+    serviceThread.join();
+  }
+
  return true;
 }

 // instance members
-AudioPipe::AudioPipe(const char* uuid, const char* host, unsigned int port, const char* path,
+AudioPipe::AudioPipe(const char* uuid, const char* bugname, const char* host, unsigned int port, const char* path,
  size_t bufLen, size_t minFreespace, const char* apiKey, notifyHandler_t callback) :
-  m_uuid(uuid), m_host(host), m_port(port), m_path(path), m_finished(false),
+  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_audio_buffer_write_offset(LWS_PRE), m_recv_buf(nullptr), m_recv_buf_ptr(nullptr), 
  m_state(LWS_CLIENT_IDLE), m_wsi(nullptr), m_vhd(nullptr), m_apiKey(apiKey), m_callback(callback) {
@@ -468,7 +475,6 @@ bool AudioPipe::connect_client(struct lws_per_vhost_data *vhd) {
  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;
@@ -13,133 +13,131 @@

 namespace deepgram {

-class AudioPipe {
-public:
-  enum LwsState_t {
-    LWS_CLIENT_IDLE,
-    LWS_CLIENT_CONNECTING,
-    LWS_CLIENT_CONNECTED,
-    LWS_CLIENT_FAILED,
-    LWS_CLIENT_DISCONNECTING,
-    LWS_CLIENT_DISCONNECTED
+  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(int loglevel, log_emit_function logger);
+    static bool deinitialize();
+    static bool lws_service_thread();
+
+    // constructor
+    AudioPipe(const char* uuid, const char* bugname, const char* host, unsigned int port, const char* path, 
+      size_t bufLen, size_t minFreespace, const char* apiKey, 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) {
+      m_audio_buffer_write_offset = 0;
+    }
+    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 std::thread serviceThread;
+
+    static int lws_callback(struct lws *wsi, enum lws_callback_reasons reason, void *user, void *in, size_t len); 
+    static struct lws_context *context;
+    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 bool stopFlag;
+
+    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;
  };
-  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, 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(int loglevel, log_emit_function logger);
-  static bool deinitialize();
-  static bool lws_service_thread();
-
-  // constructor
-  AudioPipe(const char* uuid, const char* host, unsigned int port, const char* path, 
-    size_t bufLen, size_t minFreespace, const char* apiKey, 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) {
-    m_audio_buffer_write_offset = 0;
-  }
-  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 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 *context;
-  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 bool stopFlag;
-
-  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
@@ -38,50 +38,54 @@ namespace {
      std::string model;
  };

-  static const std::unordered_map<std::string, LanguageInfo> languageLookupTable = {
-      {"zh", {"base", "general"}},
-      {"zh-CN", {"base", "general"}},
-      {"zh-TW", {"base", "general"}},
-      {"da", {"enhanced", "general"}},
-      {"en", {"nova", "phonecall"}},
-      {"en-US", {"nova", "phonecall"}},
-      {"en-AU", {"nova", "general"}},
-      {"en-GB", {"nova", "general"}},
-      {"en-IN", {"nova", "general"}},
-      {"en-NZ", {"nova", "general"}},
-      {"nl", {"enhanced", "general"}},
-      {"fr", {"enhanced", "general"}},
-      {"fr-CA", {"base", "general"}},
-      {"de", {"enhanced", "general"}},
-      {"hi", {"enhanced", "general"}},
-      {"hi-Latn", {"base", "general"}},
-      {"id", {"base", "general"}},
-      {"ja", {"enhanced", "general"}},
-      {"ko", {"enhanced", "general"}},
-      {"no", {"enhanced", "general"}},
-      {"pl", {"enhanced", "general"}},
-      {"pt", {"enhanced", "general"}},
-      {"pt-BR", {"enhanced", "general"}},
-      {"pt-PT", {"enhanced", "general"}},
-      {"ru", {"base", "general"}},
-      {"es", {"nova", "general"}},
-      {"es-419", {"nova", "general"}},
-      {"sv", {"enhanced", "general"}},
-      {"ta", {"enhanced", "general"}},
-      {"tr", {"base", "general"}},
-      {"uk", {"base", "general"}}
+  static const std::unordered_map<std::string, std::string> languageLookupTable = {
+      {"zh", "base"},
+      {"zh-CN", "base"},
+      {"zh-TW", "base"},
+      {"da", "enhanced"},
+      {"en", "nova-2"},
+      {"en-US", "nova-2"},
+      {"en-AU", "nova-2"},
+      {"en-GB", "nova-2"},
+      {"en-IN", "nova-2"},
+      {"en-NZ", "nova-2"},
+      {"nl", "nova-2"},
+      {"fr", "nova-2"},
+      {"fr-CA", "nova-2"},
+      {"de", "nova-2"},
+      {"hi", "nova-2"},
+      {"hi-Latn", "nova-2"},
+      {"id", "base"},
+      {"it", "enhanced"},
+      {"ja", "enhanced"},
+      {"ko", "enhanced"},
+      {"no", "enhanced"},
+      {"pl", "enhanced"},
+      {"pt","nova-2"},
+      {"pt-BR", "nova-2"},
+      {"pt-PT", "enhanced"},
+      {"ru", "base"},
+      {"es","nova-2"},
+      {"es-419","nova-2"},
+      {"es-LATAM","enhanced"},
+      {"sv", "enhanced"},
+      {"ta", "enhanced"},
+      {"taq", "enhanced"},
+      {"tr", "base"},
+      {"uk", "base"}
  };

-  static bool getLanguageInfo(const std::string& language, LanguageInfo& info) {
+  static bool getLanguageInfo(const std::string& language, std::string& model) {
      auto it = languageLookupTable.find(language);
      if (it != languageLookupTable.end()) {
-          info = it->second;
+          model = it->second;
          return true;
      }
      return false;
  }

-  static const char* emptyTranscript = "{\"alternatives\":[{\"transcript\":\"\",\"confidence\":0.0,\"words\":[]}]}";
+  static const char* emptyTranscript = 
+    "\"is_final\":false,\"speech_final\":false,\"channel\":{\"alternatives\":[{\"transcript\":\"\",\"confidence\":0.0,\"words\":[]}]}";

  static void reaper(private_t *tech_pvt) {
    std::shared_ptr<deepgram::AudioPipe> pAp;
@@ -146,24 +150,21 @@ namespace {
    const char *customModel = switch_channel_get_variable(channel, "DEEPGRAM_SPEECH_CUSTOM_MODEL");
    const char *tier = switch_channel_get_variable(channel, "DEEPGRAM_SPEECH_TIER") ;
    std::ostringstream oss;
-    LanguageInfo info;

    oss << "/v1/listen?";

-    if (!tier && !model && !customModel) {
-      /* make best choice by language */
-      if (getLanguageInfo(language, info)) {
-        oss << "tier=" << info.tier << "&model=" << info.model;
+    /* make best choice by language if model not supplied*/
+    if (!model && !customModel) {
+      std::string defaultModel;
+      if (getLanguageInfo(language, defaultModel)) {
+        oss << "&model=" << defaultModel;
      }
      else {
-        oss << "tier=base&model=general"; // most widely supported, though not ideal
+        oss << "tier=base&model=base"; // most widely supported, though not ideal
      }
    }
-    else {
-      if (tier) oss << "tier=" << tier;
-      if (model) oss << "&model=" << model;
-      if (customModel) oss << "&model=" << customModel;
-    }
+    else if (model) oss << "&model=" << model;
+    else if (customModel) oss << "&model=" << customModel;

    if (var = switch_channel_get_variable(channel, "DEEPGRAM_SPEECH_MODEL_VERSION")) {
     oss <<  "&version";
@@ -265,17 +266,18 @@ namespace {
   return path;
  }

-  static void eventCallback(const char* sessionId, deepgram::AudioPipe::NotifyEvent_t event, const char* message, bool finished) {
+  static void eventCallback(const char* sessionId, const char* bugname, 
+    deepgram::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, MY_BUG_NAME);
+      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 deepgram::AudioPipe::CONNECT_SUCCESS:
-              switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_INFO, "connection successful\n");
+              switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_INFO, "connection (%s) successful\n", tech_pvt->bugname);
              tech_pvt->responseHandler(session, TRANSCRIBE_EVENT_CONNECT_SUCCESS, NULL, tech_pvt->bugname, finished);
            break;
            case deepgram::AudioPipe::CONNECT_FAIL:
@@ -285,19 +287,19 @@ namespace {
              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);
+              switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_NOTICE, "connection (%s) failed: %s\n", message, tech_pvt->bugname);
            }
            break;
            case deepgram::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");
+              switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "connection (%s) dropped from far end\n", tech_pvt->bugname);
            break;
            case deepgram::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");
+              switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "connection (%s) closed gracefully\n", tech_pvt->bugname);
            break;
            case deepgram::AudioPipe::MESSAGE:
              if( strstr(message, emptyTranscript)) {
@@ -305,7 +307,7 @@ namespace {
              }
              else {
                tech_pvt->responseHandler(session, TRANSCRIBE_EVENT_RESULTS, message, tech_pvt->bugname, finished);
-                switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "deepgram message: %s\n", message);
+                switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "deepgram message (%s): %s\n", tech_pvt->bugname, message);
              }
            break;

@@ -343,6 +345,7 @@ namespace {
    tech_pvt->channels = channels;
    tech_pvt->id = ++idxCallCount;
    tech_pvt->buffer_overrun_notified = 0;
+    strncpy(tech_pvt->bugname, bugname, MAX_BUG_LEN);
    
    size_t buflen = LWS_PRE + (FRAME_SIZE_8000 * desiredSampling / 8000 * channels * 1000 / RTP_PACKETIZATION_PERIOD * nAudioBufferSecs);

@@ -353,7 +356,7 @@ namespace {
      return SWITCH_STATUS_FALSE;
    }

-    deepgram::AudioPipe* ap = new deepgram::AudioPipe(tech_pvt->sessionId, tech_pvt->host, tech_pvt->port, tech_pvt->path, 
+    deepgram::AudioPipe* ap = new deepgram::AudioPipe(tech_pvt->sessionId, bugname, tech_pvt->host, tech_pvt->port, tech_pvt->path, 
      buflen, read_impl.decoded_bytes_per_packet, apiKey, eventCallback);
    if (!ap) {
      switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_ERROR, "Error allocating AudioPipe\n");
@@ -400,7 +403,8 @@ extern "C" {
  switch_status_t dg_transcribe_init() {
    switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_NOTICE, "mod_deepgram_transcribe: audio buffer (in secs):    %d secs\n", nAudioBufferSecs);
 
-    int logs = LLL_ERR | LLL_WARN | LLL_NOTICE || LLL_INFO | LLL_PARSER | LLL_HEADER | LLL_EXT | LLL_CLIENT  | LLL_LATENCY | LLL_DEBUG ;
+    int logs = LLL_ERR | LLL_WARN | LLL_NOTICE;
+    // | LLL_INFO | LLL_PARSER | LLL_HEADER | LLL_EXT | LLL_CLIENT  | LLL_LATENCY | LLL_DEBUG ;
    
    deepgram::AudioPipe::initialize(logs, lws_logger);
 		switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_NOTICE, "AudioPipe::initialize completed\n");
@@ -455,7 +459,7 @@ extern "C" {

 	switch_status_t dg_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);
+    switch_media_bug_t *bug = (switch_media_bug_t*) switch_channel_get_private(channel, bugname);
    if (!bug) {
      switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "dg_transcribe_session_stop: no bug - websocket conection already closed\n");
      return SWITCH_STATUS_FALSE;
@@ -45,7 +45,6 @@ static switch_bool_t capture_callback(switch_media_bug_t *bug, void *user_data,
 		{
 			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");
-
 			dg_transcribe_session_stop(session, 1,  tech_pvt->bugname);
 			switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "Finished SWITCH_ABC_TYPE_CLOSE.\n");
 		}
@@ -74,7 +73,7 @@ static switch_status_t start_capture(switch_core_session_t *session, switch_medi
 	void *pUserData;
 	uint32_t samples_per_second;

-	if (switch_channel_get_private(channel, MY_BUG_NAME)) {
+	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);
 	}
@@ -94,7 +93,7 @@ static switch_status_t start_capture(switch_core_session_t *session, switch_medi
 	if ((status = switch_core_media_bug_add(session, "dg_transcribe", NULL, capture_callback, pUserData, 0, flags, &bug)) != SWITCH_STATUS_SUCCESS) {
 		return status;
 	}
-  switch_channel_set_private(channel, MY_BUG_NAME, bug);
+  switch_channel_set_private(channel, bugname, bug);
 	switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "added media bug for dg transcribe\n");

 	return SWITCH_STATUS_SUCCESS;
@@ -105,13 +104,16 @@ 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, MY_BUG_NAME);
+	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.\n");
 		status = dg_transcribe_session_stop(session, 0, bugname);
 		switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "stopped transcribe.\n");
 	}
+  else {
+		switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "do_stop: did not find bugname %s.\n", bugname);
+  }

 	return status;
 }
@@ -141,7 +143,7 @@ SWITCH_STANDARD_API(dg_transcribe_function)
 		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");
+    		switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_INFO, "stop transcribing %s\n", bugname);
 				status = do_stop(lsession, bugname);
 			} else if (!strcasecmp(argv[1], "start")) {
        char* lang = argv[2];
@@ -151,7 +153,8 @@ SWITCH_STANDARD_API(dg_transcribe_function)
          flags |= SMBF_WRITE_STREAM ;
          flags |= SMBF_STEREO;
 				}
-    		switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_INFO, "start transcribing %s %s\n", lang, interim ? "interim": "complete");
+    		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);
@@ -48,7 +48,7 @@ int AudioPipe::lws_callback(struct lws *wsi,
        lwsl_err("AudioPipe::lws_service_thread LWS_CALLBACK_CLIENT_CONNECTION_ERROR: %s, response status %d\n", in ? (char *)in : "(null)", rc); 
        if (ap) {
          ap->m_state = LWS_CLIENT_FAILED;
-          ap->m_callback(ap->m_uuid.c_str(), AudioPipe::CONNECT_FAIL, (char *) in, ap->isFinished(), ap->isInterimTranscriptsEnabled(), ap->getBugname().c_str());
+          ap->m_callback(ap->m_uuid.c_str(), ap->m_bugname.c_str(), AudioPipe::CONNECT_FAIL, (char *) in, ap->isFinished(), ap->isInterimTranscriptsEnabled());
        }
        else {
          lwsl_err("AudioPipe::lws_service_thread LWS_CALLBACK_CLIENT_CONNECTION_ERROR unable to find wsi %p..\n", wsi); 
@@ -77,7 +77,7 @@ int AudioPipe::lws_callback(struct lws *wsi,
            oss << "}";

            ap->bufferForSending(oss.str().c_str());
-            ap->m_callback(ap->m_uuid.c_str(), AudioPipe::CONNECT_SUCCESS, NULL,  ap->isFinished(), ap->isInterimTranscriptsEnabled(), ap->getBugname().c_str());
+            ap->m_callback(ap->m_uuid.c_str(), ap->m_bugname.c_str(), AudioPipe::CONNECT_SUCCESS, NULL,  ap->isFinished(), ap->isInterimTranscriptsEnabled());
          }

        }
@@ -97,12 +97,12 @@ int AudioPipe::lws_callback(struct lws *wsi,
          // closed by us

          lwsl_debug("%s socket closed by us\n", ap->m_uuid.c_str());
-          ap->m_callback(ap->m_uuid.c_str(), AudioPipe::CONNECTION_CLOSED_GRACEFULLY, NULL,  ap->isFinished(), ap->isInterimTranscriptsEnabled(), ap->getBugname().c_str());
+          ap->m_callback(ap->m_uuid.c_str(), ap->m_bugname.c_str(), AudioPipe::CONNECTION_CLOSED_GRACEFULLY, NULL,  ap->isFinished(), ap->isInterimTranscriptsEnabled());
        }
        else if (ap->m_state == LWS_CLIENT_CONNECTED) {
          // closed by far end
          lwsl_info("%s socket closed by far end\n", ap->m_uuid.c_str());
-          ap->m_callback(ap->m_uuid.c_str(), AudioPipe::CONNECTION_DROPPED, NULL,  ap->isFinished(), ap->isInterimTranscriptsEnabled(), ap->getBugname().c_str());
+          ap->m_callback(ap->m_uuid.c_str(), ap->m_bugname.c_str(), AudioPipe::CONNECTION_DROPPED, NULL,  ap->isFinished(), ap->isInterimTranscriptsEnabled());
        }
        ap->m_state = LWS_CLIENT_DISCONNECTED;
        ap->setClosed();
@@ -166,7 +166,7 @@ int AudioPipe::lws_callback(struct lws *wsi,
              std::string msg((char *)ap->m_recv_buf, ap->m_recv_buf_ptr - ap->m_recv_buf);
              //std::cerr << "Recv: " << msg << std::endl;

-              ap->m_callback(ap->m_uuid.c_str(), AudioPipe::MESSAGE, msg.c_str(),  ap->isFinished(), ap->isInterimTranscriptsEnabled(), ap->getBugname().c_str());
+              ap->m_callback(ap->m_uuid.c_str(), ap->m_bugname.c_str(), AudioPipe::MESSAGE, msg.c_str(),  ap->isFinished(), ap->isInterimTranscriptsEnabled());
              if (nullptr != ap->m_recv_buf) free(ap->m_recv_buf);
            }
            ap->m_recv_buf = ap->m_recv_buf_ptr = nullptr;
@@ -251,6 +251,7 @@ static const lws_retry_bo_t retry = {
 };

 struct lws_context *AudioPipe::context = nullptr;
+std::thread AudioPipe::serviceThread;
 std::mutex AudioPipe::mutex_connects;
 std::mutex AudioPipe::mutex_disconnects;
 std::mutex AudioPipe::mutex_writes;
@@ -418,26 +419,29 @@ bool AudioPipe::lws_service_thread() {

 void AudioPipe::initialize(int loglevel, log_emit_function logger) {

-  lws_set_log_level(loglevel, logger);
+  //lws_set_log_level(loglevel, logger);

  lwsl_notice("AudioPipe::initialize starting\n"); 
  std::lock_guard<std::mutex> lock(mapMutex);
-  std::thread t(&AudioPipe::lws_service_thread);
  stopFlag = false;
-  t.detach();
+  serviceThread = std::thread(&AudioPipe::lws_service_thread);
 }

 bool AudioPipe::deinitialize() {
  lwsl_notice("AudioPipe::deinitialize\n"); 
  std::lock_guard<std::mutex> lock(mapMutex);
  stopFlag = true;
+  if (serviceThread.joinable()) {
+    serviceThread.join();
+  }
+
  return true;
 }

 // instance members
-AudioPipe::AudioPipe(const char* uuid, const char* host, unsigned int port, const char* path,
+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_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_audio_buffer_write_offset(LWS_PRE), m_recv_buf(nullptr), m_recv_buf_ptr(nullptr), m_interim(false),
  m_state(LWS_CLIENT_IDLE), m_wsi(nullptr), m_vhd(nullptr), m_callback(callback) {
@@ -31,7 +31,7 @@ public:
    MESSAGE
  };
  typedef void (*log_emit_function)(int level, const char *line);
-  typedef void (*notifyHandler_t)(const char *sessionId, NotifyEvent_t event, const char* message, bool finished, bool wantsInterim, const char* bugname);
+  typedef void (*notifyHandler_t)(const char *sessionId, const char* bugname, NotifyEvent_t event, const char* message, bool finished, bool wantsInterim);

  struct lws_per_vhost_data {
    struct lws_context *context;
@@ -44,7 +44,7 @@ public:
  static bool lws_service_thread();

  // constructor
-  AudioPipe(const char* uuid, const char* host, unsigned int port, const char* path, 
+  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();  

@@ -104,6 +104,7 @@ public:
  void operator=(const AudioPipe&) = delete;

 private:
+    static std::thread serviceThread;

  static int lws_callback(struct lws *wsi, enum lws_callback_reasons reason, void *user, void *in, size_t len); 
  static struct lws_context *context;
@@ -82,8 +82,8 @@ namespace {
    }
  }

-  static void responseHandler(switch_core_session_t* session, 
-    const char* eventName, const char * json, const char* bugname, int finished) {
+  static void responseHandler(switch_core_session_t* session, const char* bugname,
+    const char* eventName, const char * json, int finished) {
    switch_event_t *event;
    switch_channel_t *channel = switch_core_session_get_channel(session);

@@ -162,7 +162,7 @@ namespace {
    return path;
  }

-  static void eventCallback(const char* sessionId, ibm::AudioPipe::NotifyEvent_t event, const char* message, bool finished, bool wantsInterim, const char* bugname) {
+  static void eventCallback(const char* sessionId, const char* bugname, ibm::AudioPipe::NotifyEvent_t event, const char* message, bool finished, bool wantsInterim) {
    switch_core_session_t* session = switch_core_session_locate(sessionId);
    if (session) {
      bool releaseAudioPipe = false;
@@ -256,10 +256,11 @@ namespace {
    tech_pvt->channels = channels;
    tech_pvt->id = ++idxCallCount;
    tech_pvt->buffer_overrun_notified = 0;
-    
+    strncpy(tech_pvt->bugname, bugname, MAX_BUG_LEN);
+
    size_t buflen = LWS_PRE + (FRAME_SIZE_8000 * desiredSampling / 8000 * channels * 1000 / RTP_PACKETIZATION_PERIOD * nAudioBufferSecs);

-    ibm::AudioPipe* ap = new ibm::AudioPipe(tech_pvt->sessionId, tech_pvt->host, tech_pvt->port, tech_pvt->path, 
+    ibm::AudioPipe* ap = new ibm::AudioPipe(tech_pvt->sessionId, 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");
@@ -311,7 +312,8 @@ extern "C" {
  switch_status_t ibm_transcribe_init() {
    switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_NOTICE, "mod_ibm_transcribe: audio buffer (in secs):    %d secs\n", nAudioBufferSecs);
 
-    int logs = LLL_ERR | LLL_WARN | LLL_NOTICE || LLL_INFO | LLL_PARSER | LLL_HEADER | LLL_EXT | LLL_CLIENT  | LLL_LATENCY | LLL_DEBUG ;
+    int logs = LLL_ERR | LLL_WARN | LLL_NOTICE ;
+    // | LLL_INFO | LLL_PARSER | LLL_HEADER | LLL_EXT | LLL_CLIENT  | LLL_LATENCY | LLL_DEBUG ;
    
    ibm::AudioPipe::initialize(logs, lws_logger);
 		switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_NOTICE, "AudioPipe::initialize completed\n");
@@ -56,7 +56,7 @@ static switch_status_t start_capture(switch_core_session_t *session, switch_medi
 	void *pUserData;
 	uint32_t samples_per_second;

-	if (switch_channel_get_private(channel, MY_BUG_NAME)) {
+	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);
 	}
@@ -76,7 +76,7 @@ static switch_status_t start_capture(switch_core_session_t *session, switch_medi
 	if ((status = switch_core_media_bug_add(session, "ibm_transcribe", NULL, capture_callback, pUserData, 0, flags, &bug)) != SWITCH_STATUS_SUCCESS) {
 		return status;
 	}
-  switch_channel_set_private(channel, MY_BUG_NAME, bug);
+  switch_channel_set_private(channel, bugname, bug);
 	switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "added media bug for ibm transcribe\n");

 	return SWITCH_STATUS_SUCCESS;
@@ -87,7 +87,7 @@ 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, MY_BUG_NAME);
+	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.\n");
@@ -240,6 +240,7 @@ static const lws_retry_bo_t retry = {
 };

 struct lws_context *AudioPipe::context = nullptr;
+std::thread AudioPipe::serviceThread;
 std::mutex AudioPipe::mutex_connects;
 std::mutex AudioPipe::mutex_disconnects;
 std::mutex AudioPipe::mutex_writes;
@@ -408,19 +409,21 @@ bool AudioPipe::lws_service_thread() {

 void AudioPipe::initialize(int loglevel, log_emit_function logger) {

-  lws_set_log_level(loglevel, logger);
+  //lws_set_log_level(loglevel, logger);

  lwsl_notice("AudioPipe::initialize starting\n"); 
  std::lock_guard<std::mutex> lock(mapMutex);
-  std::thread t(&AudioPipe::lws_service_thread);
  stopFlag = false;
-  t.detach();
+  serviceThread = std::thread(&AudioPipe::lws_service_thread);
 }

 bool AudioPipe::deinitialize() {
  lwsl_notice("AudioPipe::deinitialize\n"); 
  std::lock_guard<std::mutex> lock(mapMutex);
  stopFlag = true;
+  if (serviceThread.joinable()) {
+    serviceThread.join();
+  }
  return true;
 }

@@ -86,6 +86,7 @@ public:
  void operator=(const AudioPipe&) = delete;

 private:
+    static std::thread serviceThread;

  static int lws_callback(struct lws *wsi, enum lws_callback_reasons reason, void *user, void *in, size_t len); 
  static struct lws_context *context;
@@ -256,7 +256,6 @@ namespace {
    switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "host: %s, port: %d, path: %s\n", host, port, path);

    strncpy(tech_pvt->sessionId, switch_core_session_get_uuid(session), MAX_SESSION_ID);
-    strncpy(tech_pvt->bugname, bugname, MAX_BUG_LEN);
    strncpy(tech_pvt->host, host, MAX_WS_URL_LEN);
    strncpy(tech_pvt->path, path, MAX_PATH_LEN); 
    tech_pvt->port = port;
@@ -268,6 +267,7 @@ namespace {
    tech_pvt->channels = channels;
    tech_pvt->id = ++idxCallCount;
    tech_pvt->buffer_overrun_notified = 0;
+    strncpy(tech_pvt->bugname, bugname, MAX_BUG_LEN);
    
    size_t buflen = LWS_PRE + (FRAME_SIZE_8000 * desiredSampling / 8000 * channels * 1000 / RTP_PACKETIZATION_PERIOD * nAudioBufferSecs);

@@ -325,7 +325,8 @@ extern "C" {
  switch_status_t jb_transcribe_init() {
    switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_NOTICE, "mod_jambonz_transcribe: audio buffer (in secs):    %d secs\n", nAudioBufferSecs);
 
-    int logs = LLL_ERR | LLL_WARN | LLL_NOTICE || LLL_INFO | LLL_PARSER | LLL_HEADER | LLL_EXT | LLL_CLIENT  | LLL_LATENCY | LLL_DEBUG ;
+    int logs = LLL_ERR | LLL_WARN | LLL_NOTICE ;
+    // | LLL_INFO | LLL_PARSER | LLL_HEADER | LLL_EXT | LLL_CLIENT  | LLL_LATENCY | LLL_DEBUG ;
    
    jambonz::AudioPipe::initialize(logs, lws_logger);
 		switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_NOTICE, "AudioPipe::initialize completed\n");
Author	SHA1	Message	Date
Dave Horton	7c45b53690	fix: return empty deepgram transcripts that signify an endpoint Signed-off-by: Dave Horton <daveh@beachdognet.com>	2024-01-14 13:31:20 -05:00
Dave Horton	9e3783e138	update default model selection to latest api (#5 ) * update default model selection to latest api Signed-off-by: Dave Horton <daveh@beachdognet.com> * further fix Signed-off-by: Dave Horton <daveh@beachdognet.com> --------- Signed-off-by: Dave Horton <daveh@beachdognet.com>	2024-01-14 10:34:01 -05:00
Dave Horton	9fc8b1af97	Fix/audio pipe (#4 ) * use explicit namespaces for mod_audio_fork * fix crash in reload scenarios Signed-off-by: Dave Horton <daveh@beachdognet.com> --------- Signed-off-by: Dave Horton <daveh@beachdognet.com>	2023-12-27 11:10:57 -05:00
Dave Horton	3f50aaf0d7	deepgram fixes (#3 ) * more license stuff Signed-off-by: Dave Horton <daveh@beachdognet.com> * cleanup Signed-off-by: Dave Horton <daveh@beachdognet.com> * cleanup Signed-off-by: Dave Horton <daveh@beachdognet.com> * cleanup Signed-off-by: Dave Horton <daveh@beachdognet.com> * deepgram fixes: proper use of bugname --------- Signed-off-by: Dave Horton <daveh@beachdognet.com>	2023-12-26 20:35:09 -05:00
Dave Horton	0bdb4a1e32	more license stuff (#2 ) * more license stuff Signed-off-by: Dave Horton <daveh@beachdognet.com> * cleanup Signed-off-by: Dave Horton <daveh@beachdognet.com> * cleanup Signed-off-by: Dave Horton <daveh@beachdognet.com> * cleanup Signed-off-by: Dave Horton <daveh@beachdognet.com> --------- Signed-off-by: Dave Horton <daveh@beachdognet.com>	2023-12-26 12:12:01 -05:00