diff --git a/ADIS_ESP32_Eclipse/main/myMqtt.c b/ADIS_ESP32_Eclipse/main/myMqtt.c index b9613db..1ba4ab8 100644 --- a/ADIS_ESP32_Eclipse/main/myMqtt.c +++ b/ADIS_ESP32_Eclipse/main/myMqtt.c @@ -20,6 +20,7 @@ static const char *TAG = "MY_MQTT"; // local cars char _brokerIp[] = "255.255.255.255"; // max length of ip +SemaphoreHandle_t semaphoreMqtt = NULL; // declarations static void log_error_if_nonzero(const char *message, int error_code); @@ -29,8 +30,16 @@ static void mqtt_event_handler(void *handler_args, esp_event_base_t base, int32_ esp_mqtt_client_handle_t client; bool MyMqtt_Init(void){ + // initialize semaphore + semaphoreMqtt = xSemaphoreCreateRecursiveMutex(); + if(semaphoreMqtt == NULL){ + ESP_LOGE(TAG, "Failed to initialize semaphore for MyMqtt!"); + }else{ + vQueueAddToRegistry(semaphoreMqtt, "mymqtt_semphr"); + } + + // get mqtt broker IP from NVS char brokerUri[50]; - // get from NVS Challenge_Nvs_GetBrokerIpFromNVS(_brokerIp, sizeof(_brokerIp)); McuUtility_strcpy((unsigned char*)brokerUri, sizeof(brokerUri), (unsigned char*)"mqtt://"); McuUtility_strcat((unsigned char*)brokerUri, sizeof(brokerUri), (unsigned char*)_brokerIp); @@ -54,14 +63,35 @@ void MyMqtt_Deinit(void){ esp_mqtt_client_destroy(client); } +/* HELPERS FOR SEMAPHORE */ +static bool MyMqtt_AquireMutex(void){ + if(xSemaphoreTakeRecursive(semaphoreMqtt, pdMS_TO_TICKS(1000)) != pdTRUE){ + /* timeout */ + ESP_LOGW(TAG, "Timemout while aquiring myMqtt semaphore."); + return false; + } + return true; +} + +static void MyMqtt_ReleaseMutex(void){ + xSemaphoreGive(semaphoreMqtt); +} + + void MyMqtt_Publish(const char *topic, const char *data){ - int msg_id = esp_mqtt_client_publish(client, topic, data, 0, 2, 0); - ESP_LOGI(TAG, "sent publish successful, msg_id=%d", msg_id); + if(MyMqtt_AquireMutex()){ + int msg_id = esp_mqtt_client_publish(client, topic, data, 0, 2, 0); + ESP_LOGI(TAG, "sent publish successful, msg_id=%d", msg_id); + MyMqtt_ReleaseMutex(); + } } void MyMqtt_Subscribe(const char *topic){ - int msg_id = esp_mqtt_client_subscribe(client, topic, 0); - ESP_LOGI(TAG, "sent subscribe successful, msg_id=%d, success=%s", msg_id, msg_id==-1 ? "false" : "true"); + if(MyMqtt_AquireMutex()){ + int msg_id = esp_mqtt_client_subscribe(client, topic, 0); + ESP_LOGI(TAG, "sent subscribe successful, msg_id=%d, success=%s", msg_id, msg_id==-1 ? "false" : "true"); + MyMqtt_ReleaseMutex(); + } } char* MyMqtt_GetBrokerIP(void){