fix aws race condition when 2 start transcribes are sent at the same instant

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

mod_aws_transcribe/aws_transcribe_glue.cpp

@@ -326,8 +326,8 @@ static void *SWITCH_THREAD_FUNC aws_transcribe_thread(switch_thread_t *thread, v
 		switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "transcribe_thread: Error allocating streamer\n");
 		return nullptr;
 	}
-  if (!cb->vad) pStreamer->connect();
 	cb->streamer = pStreamer;
+  if (!cb->vad) pStreamer->connect();
 	pStreamer->processData(); //blocks until done
 
 	switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "transcribe_thread: stopping cb %p\n", (void *) cb);

mod_azure_transcribe/azure_transcribe_glue.cpp

@@ -135,6 +135,13 @@ public:
       properties.SetProperty(PropertyId::Speech_SegmentationSilenceTimeoutMs, segmentationInterval);
     }
 
+		//https://learn.microsoft.com/en-us/azure/ai-services/speech-service/language-identification?tabs=once&pivots=programming-language-cpp#at-start-and-continuous-language-identification
+		const char* languageIdMode = switch_channel_get_variable(channel, "AZURE_LANGUAGE_ID_MODE");
+		if (languageIdMode) {
+			switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(psession), SWITCH_LOG_DEBUG, "setting SpeechServiceConnection_LanguageIdMode to %s \n", languageIdMode);
+			properties.SetProperty(PropertyId::SpeechServiceConnection_LanguageIdMode, languageIdMode);
+		}
+
 		// recognition mode - readonly according to Azure docs: 
 		// https://docs.microsoft.com/en-us/javascript/api/microsoft-cognitiveservices-speech-sdk/propertyid?view=azure-node-latest
 		/*

mod_google_transcribe/google_glue.cpp

@@ -380,16 +380,9 @@ static void *SWITCH_THREAD_FUNC grpc_read_thread(switch_thread_t *thread, void *
 	struct cap_cb *cb = (struct cap_cb *) obj;
 	GStreamer* streamer = (GStreamer *) cb->streamer;
 
-  std::thread::id this_id = std::this_thread::get_id();
-  std::stringstream ss;
-  ss << this_id;
-  std::string id_str = ss.str();
-
-  switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "grpc_read_thread %s starting\n", id_str.c_str()) ;
-
   bool connected = streamer->waitForConnect();
   if (!connected) {
-    switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_NOTICE, "grpc_read_thread %s thread exiting since we didnt connect\n", id_str.c_str()) ;
+    switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_NOTICE, "google transcribe grpc read thread exiting since we didnt connect\n") ;
     return nullptr;
   }
 
@@ -398,14 +391,14 @@ static void *SWITCH_THREAD_FUNC grpc_read_thread(switch_thread_t *thread, void *
   while (streamer->read(&response)) {  // Returns false when no more to read.
     switch_core_session_t* session = switch_core_session_locate(cb->sessionId);
     if (!session) {
-      switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "grpc_read_thread: %s session %s is gone!\n", id_str.c_str(), cb->sessionId) ;
+      switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "grpc_read_thread: session %s is gone!\n", cb->sessionId) ;
       return nullptr;
     }
     count++;
     auto speech_event_type = response.speech_event_type();
     if (response.has_error()) {
       Status status = response.error();
-      switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "grpc_read_thread: %s error %s (%d)\n", id_str.c_str(), status.message().c_str(), status.code()) ;
+      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");
       cJSON_AddStringToObject(json, "error", status.message().c_str());
@@ -451,7 +444,7 @@ static void *SWITCH_THREAD_FUNC grpc_read_thread(switch_thread_t *thread, void *
 
         if (alternative.words_size() > 0) {
           cJSON * jWords = cJSON_CreateArray();
-          switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "grpc_read_thread: %s %d words\n", id_str.c_str(), alternative.words_size()) ;
+          switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "grpc_read_thread: %d words\n", alternative.words_size()) ;
           for (int b = 0; b < alternative.words_size(); b++) {
             auto words = alternative.words(b);
             cJSON* jWord = cJSON_CreateObject();
@@ -487,16 +480,16 @@ static void *SWITCH_THREAD_FUNC grpc_read_thread(switch_thread_t *thread, void *
 
     if (speech_event_type == StreamingRecognizeResponse_SpeechEventType_END_OF_SINGLE_UTTERANCE) {
       // we only get this when we have requested it, and recognition stops after we get this
-      switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "grpc_read_thread: %s got end_of_utterance\n", id_str.c_str()) ;
+      switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "grpc_read_thread: got end_of_utterance\n") ;
       cb->got_end_of_utterance = 1;
       cb->responseHandler(session, "end_of_utterance", cb->bugname);
       if (cb->wants_single_utterance) {
-        switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "grpc_read_thread: %s sending writesDone because we want only a single utterance\n", id_str.c_str()) ;
+        switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "grpc_read_thread: sending writesDone because we want only a single utterance\n") ;
         streamer->writesDone();
       }
     }
     switch_core_session_rwunlock(session);
-    switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "grpc_read_thread: %s got %d responses\n", response.results_size(), id_str.c_str());
+    switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "grpc_read_thread: got %d responses\n", response.results_size());
   }
 
   {
@@ -511,11 +504,10 @@ static void *SWITCH_THREAD_FUNC grpc_read_thread(switch_thread_t *thread, void *
           cb->responseHandler(session, "no_audio", cb->bugname);
         }
       }
-      switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "grpc_read_thread: %s finish() status %s (%d)\n", id_str.c_str(), status.error_message().c_str(), status.error_code()) ;
+      switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "grpc_read_thread: finish() status %s (%d)\n", status.error_message().c_str(), status.error_code()) ;
       switch_core_session_rwunlock(session);
     }
   }
-  switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "grpc_read_thread %s exiting\n", id_str.c_str()) ;
   return nullptr;
 }
 
@@ -613,9 +605,6 @@ extern "C" {
 
       if (!cb->vad) streamer->connect();
 
-      switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_INFO, "%s: creating new read thread, streamer %p\n", 
-        switch_channel_get_name(channel), (void*)streamer);
-
       // create the read thread
       switch_threadattr_t *thd_attr = NULL;
       switch_memory_pool_t *pool = switch_core_session_get_pool(session);
@@ -631,23 +620,17 @@ extern "C" {
     switch_status_t google_speech_session_cleanup(switch_core_session_t *session, int channelIsClosing, switch_media_bug_t *bug) {
       switch_channel_t *channel = switch_core_session_get_channel(session);
 
-      switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_INFO, "%s: cleaning up bug %p\n", 
-        switch_channel_get_name(channel), (void*)bug);
-
       if (bug) {
         struct cap_cb *cb = (struct cap_cb *) switch_core_media_bug_get_user_data(bug);
         switch_mutex_lock(cb->mutex);
 
         if (!switch_channel_get_private(channel, cb->bugname)) {
           // race condition
-          switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_INFO, "%s Bug %p is not attached (race).\n", 
-            switch_channel_get_name(channel), (void *)bug);
+          switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_INFO, "%s Bug is not attached (race).\n", switch_channel_get_name(channel));
           switch_mutex_unlock(cb->mutex);
           return SWITCH_STATUS_FALSE;
         }
         switch_channel_set_private(channel, cb->bugname, NULL);
-        switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_INFO, "%s set_private set bug to NULL, was %p\n", 
-          switch_channel_get_name(channel), (void *)bug);
 
       // stop playback if available
        if (cb->play_file == 1){ 

mod_google_transcribe/mod_google_transcribe.c

@@ -104,10 +104,10 @@ static switch_bool_t capture_callback(switch_media_bug_t *bug, void *user_data,
 
 	case SWITCH_ABC_TYPE_CLOSE:
 		{
-			switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "Got SWITCH_ABC_TYPE_CLOSE, calling google_speech_session_cleanup bug %p.\n", (void *)bug;
+			switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "Got SWITCH_ABC_TYPE_CLOSE, calling google_speech_session_cleanup.\n");
 			responseHandler(session, "end_of_transcript", cb->bugname);
 			google_speech_session_cleanup(session, 1, bug);
-			switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "Finished SWITCH_ABC_TYPE_CLOSE with bug %p.\n", (void *)bug);
+			switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "Finished SWITCH_ABC_TYPE_CLOSE.\n");
 		}
 		break;
 	
@@ -143,9 +143,9 @@ static switch_status_t do_stop(switch_core_session_t *session, char *bugname)
 	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, calling google_speech_session_cleanup bug %p\n", (void *)bug);
+		switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "Received user command command, calling google_speech_session_cleanup (possibly to stop prev transcribe)\n");
 		status = google_speech_session_cleanup(session, 0, bug);
-		switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "stopped transcription via bug %p.\n", (void *)bug);
+		switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "stopped transcription.\n");
 	}
 
 	return status;
@@ -210,10 +210,9 @@ static switch_status_t start_capture(switch_core_session_t *session, switch_medi
   const char* model = NULL;
 	const char* var;
 
-	if ((bug = switch_channel_get_private(channel, bugname))) {
-		switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "removing bug %p from previous transcribe\n", (void *)bug);
+	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_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "previous bug %p removed, now we can add a new bug\n", (void *)bug);
 	}
 
 	if (switch_true(switch_channel_get_variable(channel, "GOOGLE_SPEECH_SINGLE_UTTERANCE"))) {
@@ -277,12 +276,8 @@ static switch_status_t start_capture(switch_core_session_t *session, switch_medi
 	if ((status = switch_core_media_bug_add(session, bugname, NULL, capture_callback, pUserData, 0, flags, &bug)) != SWITCH_STATUS_SUCCESS) {
 		return status;
 	}
-  switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "added bug %p\n", (void *)bug);
 
-  //Q: we don't have a lock, so is it possible after creating and before calling set_private
-  // another thread could call do_stop and find no bug to remove?
 	switch_channel_set_private(channel, bugname, bug);
-  switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "called set_private to associate bug %p with channel\n", (void *)bug);
 
 	return SWITCH_STATUS_SUCCESS;
 }