Mqtt_client currently has no way to unregister events. (IDFGH-7641)

[Arreme01]

Hello, I'm working on an implementation of the mqtt_client where it separates different topics through events, so the client can filter and write to it's respective data buffer the contents of the message received. These events are registered when succesfully subscribed to a topic.

The problem arises when I need to unsubscribe from a topic, as in this scenario I should be also unregistering the event. There is currently no function in the mqtt library in the order of "esp_mqtt_client_unregister_event".

I have tried to centralize all the data events to a simple function but I'm not convinced to process that much work on an interrupt type function.

I can try to program this function myself and make a PR afterwards, but first I wanted to make sure it does really make a problem.

[nopnop2002]

esp_err_t esp_mqtt_client_unregister_event(esp_mqtt_client_handle_t client, esp_mqtt_event_id_t event, esp_event_handler_t event_handler, void *event_handler_arg)
{
    if (client == NULL) {
        return ESP_ERR_INVALID_ARG;
    }
    return esp_event_handler_unregister_with(client->config->event_loop_handle, MQTT_EVENTS, event, event_handler);

}

[euripedesrocha]

Hi @Arreme01, I've started the development of this interface, and it should be available soon, but I would like to request that you provide a short example on how you are using the API, just to evaluate if we can find some ideas to improve our library interface.

[Arreme01]

Sure @euripedesrocha ! Right now the wrapper that interacts with the esp_mqtt api can store "x" number of mqtt clients so you can handle multiple connections to different servers. You interact externally with each mqtt client by using ids. Each client also can subscribe to "y" ammount of topics. The struct mdw_mqtt_data_pair_t has the information for each subscription.

Struct declaration and definition:

//-- Struct that stores the information of a subscription --
//The cir_buff_id is used for an external utility class that
//implements a simple circular buffer for reading incoming data.
typedef struct
{
    char topic[100];
    int circ_buff_id;
    bool is_used;
} mdw_mqtt_data_pair_t;

//-- Struct that stores the information of a single client instance --
typedef struct
{
    esp_mqtt_client_handle_t client;
    esp_mqtt_event_handle_t event;

    //A single mqtt client can hold multiple subscriptions
    esp_event_loop_handle_t subscription_handlers[MQTT_SUBSCRIPTIONS];
    mdw_mqtt_data_pair_t subscriptions[MQTT_SUBSCRIPTIONS];

    uint32_t id;
    uint32_t status;
    bool initialized;
} mdw_mqtt_t;

static mdw_mqtt_t mqtt[MQTT_CLIENTS];

Example of the init method

esp_err_t MDW_MQTT_Init(int i,const esp_mqtt_client_config_t config)
{
    mqtt[i].client = esp_mqtt_client_init(&config);
    mqtt[i].initialized = true;
    mqtt[i].id = i;
    ESP_ERROR_CHECK(esp_mqtt_client_register_event(mqtt[i].client, ESP_EVENT_ANY_ID, MDW_MQTT_EventHandler,(void*) &mqtt[i]));
    return ESP_OK;
}

Example of the subscription method

esp_err_t MDW_MQTT_Subscribe(int i, const char *topic, int qos) 
{
    //Iterates through all subscriptions for that client
    for (size_t j = 0; j < MQTT_SUBSCRIPTIONS; j++)
    {
        //Find empty subscription channel
        if (!mqtt[i].subscriptions[j].is_used)
        {
            /* Subscribe */
            ESP_LOGI(TAG, "Subscribing...");
            if (esp_mqtt_client_subscribe(mqtt[i].client, topic, qos) == -1) 
            {
                ESP_LOGE(TAG,"Error on subscribing");
                return ESP_ERR_TIMEOUT;
            }

            mqtt[i].subscriptions[j].is_used = true;
            mqtt[i].subscriptions[j].circ_buff_id = i + j;
            snprintf(mqtt[i].subscriptions[j].topic,100,"%s",topic);
            /* Register subscription topic and instance to the struct so it can filter events */
            ESP_ERROR_CHECK(esp_mqtt_client_register_event(mqtt[i].client, MQTT_EVENT_DATA, MDW_MQTT_EventDataHandler,(void *) &mqtt[i].subscriptions[j].circ_buff_id ));
            return ESP_OK;
        }
        
    }

    return ESP_ERR_NOT_FOUND;
}

MDW_MQTT_EventDataHandler
Not to confuse with MDW_MQTT_EventHandler. That method it's just a copy of the example mqtt event handler. MDW_MQTT_EventDataHandler receives as handler_args the id of the circular buffer.

static void MDW_MQTT_EventDataHandler(void *handler_args, esp_event_base_t base, int32_t event_id, void *event_data) 
{
    if (event_id == MQTT_EVENT_DATA) 
    {
        esp_mqtt_event_handle_t mqtt_event = (esp_mqtt_event_handle_t) event_data;
        int buffer = *( (int *) handler_args);
        if (UTIL_CIRCBUF_GetFreeSpace(buffer) >= mqtt_event->data_len) 
        {
            ESP_LOGI(TAG,"Writing to buffer!");
            UTIL_CIRCBUF_Push(buffer, (uint8_t *) mqtt_event->data, mqtt_event->data_len);
        }
    }
}

With this, each subscription gets an independent buffer to write the data to. Perfect if you need multiple incoming channels. You may read the data with something like this:

Reading data from a client and topic

esp_err_t MDW_MQTT_ReadTopicData(int i, const char *topic, char * buffer) 
{
    for (size_t j = 0; j < MQTT_SUBSCRIPTIONS; j++)
    {
        if (strcmp(mqtt[i].subscriptions[j].topic,topic) == 0) 
        {
            ESP_LOGI(TAG, "Found topic");
            int circ_id = mqtt[i].subscriptions[j].circ_buff_id;
            UTIL_CIRCBUF_Pop(circ_id,(uint8_t *)buffer, UTIL_CIRCBUF_GetUsageSpace(circ_id));
            return ESP_OK;
        }
    }

    return ESP_ERR_NOT_FOUND;
    
}

I hope this is more or less the information that you requested! Some specifications have been removed like error checking so it's clearer to read. Wish this helps with the development of the api :-) Feel free to comment if something is not clear at all!

[AxelLin]

@euripedesrocha
This is marked as "Resolution Done" on Aug 15, but I still cannot find this fix in master. What's wrong?

[AxelLin]

@euripedesrocha

espressif/esp-mqtt@a9a9fe7 was commited on on Jul 28.
But I still cannot find the fix in esp-idf master?