From 31b5ee634ac6a00da75b9322cef9fbe2617ee2ff Mon Sep 17 00:00:00 2001
From: gc87 <sfme@qq.com>
Date: Thu, 12 May 2022 17:32:05 +0800
Subject: [PATCH] feat(mqtt intf): add will topic set

---
 plugins/mqtt/util.c            |  2 ++
 src/connection/mqtt_c_client.c | 29 ++++++++++++++++++++++++++---
 2 files changed, 28 insertions(+), 3 deletions(-)

diff --git a/plugins/mqtt/util.c b/plugins/mqtt/util.c
index 3205b696..ea957a06 100644
--- a/plugins/mqtt/util.c
+++ b/plugins/mqtt/util.c
@@ -137,5 +137,7 @@ int mqtt_option_init(neu_config_t *config, neu_mqtt_option_t *option)
     option->key                = key.v.val_str;
     option->keepalive_interval = 30;
     option->clean_session      = 1;
+    option->will_topic         = NULL;
+    option->will_payload       = NULL;
     return 0;
 }
diff --git a/src/connection/mqtt_c_client.c b/src/connection/mqtt_c_client.c
index a6cc84f0..7dae4d96 100644
--- a/src/connection/mqtt_c_client.c
+++ b/src/connection/mqtt_c_client.c
@@ -44,6 +44,9 @@ struct reconnect_state {
     char *           password;
     int              keepalive;
     int              clean;
+    char *           will_topic;
+    char *           will_payload;
+    int              will_payload_len;
     char *           ca_file;
     char *           ca_path;
     char *           cert_file;
@@ -378,6 +381,14 @@ static void reconnect_state_uninit(struct reconnect_state *state)
     pthread_mutex_destroy(&state->sock_state_mutex);
     pthread_mutex_destroy(&state->running_mutex);
     pthread_mutex_destroy(&state->working_mutex);
+
+    if (NULL != state->will_topic) {
+        free(state->will_topic);
+    }
+
+    if (NULL != state->will_payload) {
+        free(state->will_payload);
+    }
 }
 
 static void subscribe_on_reconnect(mqtt_c_client_t *client)
@@ -429,9 +440,13 @@ static void client_reconnect(struct mqtt_client *mqtt, void **p_reconnect_state)
         connect_flags = MQTT_CONNECT_CLEAN_SESSION;
     }
 
-    enum MQTTErrors rc =
-        mqtt_connect(mqtt, client_id, NULL, NULL, 0, state->username,
-                     state->password, connect_flags, state->keepalive);
+    const char *will_topic   = state->will_topic;
+    const char *will_payload = state->will_payload;
+    size_t      payload_len  = state->will_payload_len;
+
+    enum MQTTErrors rc = mqtt_connect(
+        mqtt, client_id, will_topic, will_payload, payload_len, state->username,
+        state->password, connect_flags, state->keepalive);
 
     if (MQTT_OK != rc) {
         return;
@@ -551,6 +566,14 @@ static mqtt_c_client_t *client_create(const neu_mqtt_option_t *option,
     pthread_mutex_init(&state->running_mutex, NULL);
     pthread_mutex_init(&state->working_mutex, NULL);
 
+    if (NULL != client->option->will_topic) {
+        state->will_topic = strdup(option->will_topic);
+        if (NULL != client->option->will_payload) {
+            state->will_payload     = strdup(client->option->will_payload);
+            state->will_payload_len = strlen(state->will_payload);
+        }
+    }
+
     client->mqtt.publish_response_callback_state = state;
     return client;
 }