Implementation SpeechServiceResponse_PostProcessingOption (#87)

Signed-off-by: Vinod Dharashive <vdharashive@gmail.com>

mod_audio_fork/audio_pipe.cpp

@@ -141,6 +141,11 @@ int AudioPipe::lws_callback(struct lws *wsi,
           return 0;
         }
         
+        if (ap->m_state == LWS_CLIENT_DISCONNECTING) {
+          lwsl_notice("AudioPipe::lws_service_thread race condition: got incoming message while closing the connection.\n");
+          return 0;
+        }
+
         if (lws_frame_is_binary(wsi)) {
           if (ap->is_bidirectional_audio_stream()) {
             ap->m_callback(ap->m_uuid.c_str(), ap->m_bugname.c_str(), AudioPipe::BINARY, NULL, (char *) in, len);

mod_audio_fork/lws_glue.cpp

@@ -38,7 +38,6 @@ namespace {
   static uint32_t playCount = 0;
 
   switch_status_t processIncomingBinary(private_t* tech_pvt, switch_core_session_t* session, const char* message, size_t dataLength) {
-    CircularBuffer_t *cBuffer = (CircularBuffer_t *) tech_pvt->circularBuffer;
     uint8_t* data = reinterpret_cast<uint8_t*>(const_cast<char*>(message));
     uint16_t* data_uint16 = reinterpret_cast<uint16_t*>(data);
     std::vector<uint16_t> pcm_data(data_uint16, data_uint16 + dataLength / sizeof(uint16_t));
@@ -70,7 +69,9 @@ namespace {
       return SWITCH_STATUS_FALSE;
     }
 
-    switch_mutex_lock(tech_pvt->mutex);
+    if (nullptr != tech_pvt->mutex && switch_mutex_trylock(tech_pvt->mutex) == SWITCH_STATUS_SUCCESS) {
+      //switch_mutex_lock(tech_pvt->mutex);
+      CircularBuffer_t *cBuffer = (CircularBuffer_t *) tech_pvt->circularBuffer;
 
       try {
         // Resize the buffer if necessary
@@ -92,10 +93,10 @@ namespace {
         switch_mutex_unlock(tech_pvt->mutex);
         return SWITCH_STATUS_FALSE;
       }
-
       switch_mutex_unlock(tech_pvt->mutex);
-
       return SWITCH_STATUS_SUCCESS;
+    }
+    return SWITCH_STATUS_FALSE;
 }
 
   void processIncomingMessage(private_t* tech_pvt, switch_core_session_t* session, const char* message) {
@@ -103,7 +104,7 @@ namespace {
     std::string type;
     cJSON* json = parse_json(session, msg, type) ;
     if (json) {
-      switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "(%u) processIncomingMessage - received %s message\n", tech_pvt->id, type.c_str());
+      switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "(%u) processIncomingMessage - received %s message %s\n", tech_pvt->id, type.c_str(), message);
       cJSON* jsonData = cJSON_GetObjectItem(json, "data");
       if (0 == type.compare("playAudio") &&
         // playAudio is enabled and there is no bidirectional audio from stream is enabled.
@@ -190,6 +191,9 @@ namespace {
         // kill any current playback on the channel
         switch_channel_t *channel = switch_core_session_get_channel(session);
         switch_channel_set_flag_value(channel, CF_BREAK, 2);
+
+        // this will dump buffered incoming audio
+        tech_pvt->clear_bidirectional_audio_buffer = 1;
       }
       else if (0 == type.compare("transcription")) {
         char* jsonString = cJSON_PrintUnformatted(jsonData);
@@ -313,6 +317,7 @@ namespace {
     tech_pvt->bidirectional_audio_enable = bidirectional_audio_enable;
     tech_pvt->bidirectional_audio_stream = bidirectional_audio_stream;
     tech_pvt->bidirectional_audio_sample_rate = bidirectional_audio_sample_rate;
+    tech_pvt->clear_bidirectional_audio_buffer = 0;
     strncpy(tech_pvt->bugname, bugname, MAX_BUG_LEN);
     if (metadata) strncpy(tech_pvt->initialMetadata, metadata, MAX_METADATA_LEN);
     
@@ -727,6 +732,14 @@ extern "C" {
   switch_bool_t dub_speech_frame(switch_media_bug_t *bug, private_t* tech_pvt) {
     CircularBuffer_t *cBuffer = (CircularBuffer_t *) tech_pvt->circularBuffer;
     if (switch_mutex_trylock(tech_pvt->mutex) == SWITCH_STATUS_SUCCESS) {
+
+      // if flag was set to clear the buffer, do so and clear the flag
+      if (tech_pvt->clear_bidirectional_audio_buffer) {
+        switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "(%u) dub_speech_frame - clearing buffer\n", tech_pvt->id); 
+        cBuffer->clear();
+        tech_pvt->clear_bidirectional_audio_buffer = 0;
+      }
+      else {
         switch_frame_t* rframe = switch_core_media_bug_get_write_replace_frame(bug);
         int16_t *fp = reinterpret_cast<int16_t*>(rframe->data);
 
@@ -747,6 +760,7 @@ extern "C" {
         vector_normalize(fp, rframe->samples);
 
         switch_core_media_bug_set_write_replace_frame(bug, rframe);
+      }
       switch_mutex_unlock(tech_pvt->mutex);
     }
     return SWITCH_TRUE;

mod_audio_fork/mod_audio_fork.h

@@ -57,6 +57,7 @@ struct private_data {
   int bidirectional_audio_enable;
 	int bidirectional_audio_stream;
   int bidirectional_audio_sample_rate;
+  int clear_bidirectional_audio_buffer;
 };
 
 typedef struct private_data private_t;

mod_azure_transcribe/azure_transcribe_glue.cpp

@@ -142,7 +142,13 @@ public:
 			switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(psession), SWITCH_LOG_DEBUG, "setting SpeechServiceConnection_LanguageIdMode to %s \n", languageIdMode);
 			properties.SetProperty(PropertyId::SpeechServiceConnection_LanguageIdMode, languageIdMode);
 		}
-
+		//https://learn.microsoft.com/en-us/javascript/api/microsoft-cognitiveservices-speech-sdk/propertyid?view=azure-node-latest
+		//PropertyId::SpeechServiceResponse_PostProcessingOption
+ 		const char* postProcessingOption = switch_channel_get_variable(channel, "AZURE_POST_PROCESSING_OPTION");
+		if (postProcessingOption) {
+			switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(psession), SWITCH_LOG_DEBUG, "setting SpeechServiceResponse_PostProcessingOption to %s \n", postProcessingOption);
+			properties.SetProperty(PropertyId::SpeechServiceResponse_PostProcessingOption, postProcessingOption);
+		}
 		// recognition mode - readonly according to Azure docs: 
 		// https://docs.microsoft.com/en-us/javascript/api/microsoft-cognitiveservices-speech-sdk/propertyid?view=azure-node-latest
 		/*
@@ -382,22 +388,18 @@ private:
 
 		const char* endpoint = switch_channel_get_variable(channel, "AZURE_SERVICE_ENDPOINT");
 		const char* endpointId = switch_channel_get_variable(channel, "AZURE_SERVICE_ENDPOINT_ID");
-		configuration_stream <<
+		configuration_stream << (endpoint ? endpoint : "") << ";"
-			endpoint << ";" <<
+			<< (endpointId ? endpointId : "") << ";";
-			endpointId << ";";
 		if (switch_true(switch_channel_get_variable(channel, "AZURE_USE_OUTPUT_FORMAT_DETAILED"))) {
 			configuration_stream << "output_format_detailed;";
 		}
 		if (switch_true(switch_channel_get_variable(channel, "AZURE_AUDIO_LOGGING"))) {
 			configuration_stream << "audio_logging;";
 		}
-		if (nullptr != proxyIP && nullptr != proxyPort) {
+		configuration_stream << (proxyIP ? proxyIP : "") << ";"
-			configuration_stream <<
+			<< (proxyPort ? proxyPort : "") << ";"
-				proxyIP << ";" <<
+			<< (proxyUsername ? proxyUsername : "") << ";"
-				proxyPort << ";" <<
+			<< (proxyPassword ? proxyPassword : "") << ";";
-				proxyUsername << ";" <<
-				proxyPassword << ";";
-		}
 		const char* var;
 		if (var = switch_channel_get_variable(channel, "AZURE_SPEECH_ALTERNATIVE_LANGUAGE_CODES")) {
 			configuration_stream << var << ";";