Add support for MS Azure MQTT

tutorials/mqtt/mqtt-client/mongoose_mqtt.c

@@ -1,24 +1,22 @@
 // Copyright (c) 2026 Cesanta Software Limited
 // All rights reserved
-//
-// Example MQTT client. It performs the following steps:
-//    1. Connects to the MQTT server specified by MQTT_SERVER_URL
-//    2. When connected, subscribes to the topic MQTT_SUBSCRIBE_TOPIC
-//    3. When it receives a message, echoes it back to MQTT_PUBLISH_TOPIC
-//    4. Timer-based reconnection logic revives the connection when it is down
-//    5. Ping server periodically. When disconnected, a last will is published
-//    6. Implements "ota.update" for OTA updates, see https://mongoose.ws/mqtt/
 
 #include "mongoose.h"
 
 #define MQTT_SERVER_URL "mqtt://broker.hivemq.com:1883"
-#define MQTT_PUBLISH_TOPIC "mg/123/tx"
-#define MQTT_SUBSCRIBE_TOPIC "mg/123/rx"
+#define MQTT_CLIENT_ID "d3"
+#define MQTT_USER MQTT_CLIENT_ID
+#define MQTT_PASS ""
+#define MQTT_PUBLISH_TOPIC "mg/" MQTT_CLIENT_ID "/tx"
+#define MQTT_SUBSCRIBE_TOPIC "mg/" MQTT_CLIENT_ID "/rx"
 #define MQTT_QOS 1
-#define RECONNECT_PERIOD_MS 3000
+#define MQTT_RECONNECT_PERIOD_MS 3000
+
+#define TLS_CA ""
+#define TLS_KEY ""
+#define TLS_CRT ""
 
 static struct mg_connection *s_mqtt_conn;  // Client connection
-static struct mg_rpc *s_rpc = NULL;        // List of registered RPC methods
 
 static void subscribe(struct mg_connection *c, struct mg_str topic) {
   struct mg_mqtt_opts opts = {};
@@ -69,7 +67,9 @@ static void mqtt_ev_handler(struct mg_connection *c, int ev, void *ev_data) {
     // c->is_hexdumping = 1;
   } else if (ev == MG_EV_CONNECT) {
     if (c->is_tls) {
-      struct mg_tls_opts opts = {.ca = mg_unpacked("/certs/ca.pem"),
+      struct mg_tls_opts opts = {.ca = mg_str(TLS_CA),
+                                 .cert = mg_str(TLS_CRT),
+                                 .key = mg_str(TLS_KEY),
                                  .name = mg_url_host(MQTT_SERVER_URL)};
       mg_tls_init(c, &opts);
     }
@@ -77,16 +77,31 @@ static void mqtt_ev_handler(struct mg_connection *c, int ev, void *ev_data) {
     // On error, log error message
     MG_ERROR(("%lu ERROR %s", c->id, (char *) ev_data));
   } else if (ev == MG_EV_MQTT_OPEN) {
-    // MQTT connect is successful
-    MG_DEBUG(("%lu CONNECTED to %s", c->id, MQTT_SERVER_URL));
-    subscribe(c, mg_str(MQTT_SUBSCRIBE_TOPIC));
+    int status = *(int *) ev_data;
+    MG_DEBUG(("%lu CONNECT status: %d", c->id, status));
+    if (status == 0) {
+      subscribe(c, mg_str(MQTT_SUBSCRIBE_TOPIC));
+      if (mg_match(mg_str(MQTT_SERVER_URL), mg_str("#azure-devices.net"), 0)) {
+        // This is Azure IoT Hub. Subscribe for DPS messages
+        subscribe(c, mg_str("$iothub/methods/POST/#"));
+      }
+    }
   } else if (ev == MG_EV_MQTT_MSG) {
     // When we get echo response, print it
-    char response[100];
     struct mg_mqtt_message *mm = (struct mg_mqtt_message *) ev_data;
-    mg_snprintf(response, sizeof(response), "Got %.*s -> %.*s", mm->topic.len,
-                mm->topic.buf, mm->data.len, mm->data.buf);
-    publish(c, mg_str(MQTT_PUBLISH_TOPIC), mg_str(response));
+    struct mg_str caps[5];  // caps[0] = method name, caps[2] = request id
+    if (mg_match(mm->topic, mg_str("$iothub/methods/POST/*/?$rid=*"), caps)) {
+      // Azure direct method call. Construct a stub response, "{}"
+      char topic[128];
+      mg_snprintf(topic, sizeof(topic), "$iothub/methods/res/%d/?$rid=%.*s",
+                  200, (int) caps[2].len, caps[2].buf);
+      publish(c, mg_str(topic), mg_str("{}"));
+    } else {
+      char response[100];
+      mg_snprintf(response, sizeof(response), "Got %.*s -> %.*s", mm->topic.len,
+                  mm->topic.buf, mm->data.len, mm->data.buf);
+      publish(c, mg_str(MQTT_PUBLISH_TOPIC), mg_str(response));
+    }
   } else if (ev == MG_EV_MQTT_CMD) {
     struct mg_mqtt_message *mm = (struct mg_mqtt_message *) ev_data;
     if (mm->cmd == MQTT_CMD_PINGREQ) mg_mqtt_pong(c);
@@ -97,9 +112,7 @@ static void mqtt_ev_handler(struct mg_connection *c, int ev, void *ev_data) {
 }
 
 void mg_mqtt_init(struct mg_mgr *mgr) {
-  (void) mgr;
-  if (!s_rpc) mg_rpc_add(&s_rpc, mg_str("ota.update"), rpc_ota_update, NULL);
-  mg_mem_files = mg_packed_files;  // Use generated packed filesystem
+  mg_rpc_add(&mgr->rpcs, mg_str("ota.update"), rpc_ota_update, NULL);
 }
 
 void mg_mqtt_poll(struct mg_mgr *mgr) {
@@ -107,18 +120,24 @@ void mg_mqtt_poll(struct mg_mgr *mgr) {
 
   // Reconnect if connection is closed, and send MQTT PINGs to keep
   // the connection alive or to detect connection loss
-  if (mg_timer_expired(&timer, RECONNECT_PERIOD_MS, mg_now())) {
+  if (mg_timer_expired(&timer, MQTT_RECONNECT_PERIOD_MS, mg_now())) {
     if (s_mqtt_conn == NULL) {
-      struct mg_mqtt_opts opts = {.clean = true,
-                                  .qos = MQTT_QOS,
-                                  .topic = mg_str(MQTT_PUBLISH_TOPIC),
-                                  .keepalive = 5,
-                                  .version = 4,
-                                  .message = mg_str("bye")};
+      struct mg_mqtt_opts opts = {
+          .clean = true,
+          // .client_id = mg_str(MQTT_CLIENT_ID),
+          .user = mg_str(MQTT_USER),
+          .pass = mg_str(MQTT_PASS),
+          .qos = MQTT_QOS,
+          .keepalive = 5,
+          .version = 4,  // MQTT 3.1.1
+          .topic = mg_str(MQTT_PUBLISH_TOPIC),
+          .message = mg_str("bye"),
+      };
       s_mqtt_conn =
           mg_mqtt_connect(mgr, MQTT_SERVER_URL, &opts, mqtt_ev_handler, NULL);
     } else {
       mg_mqtt_ping(s_mqtt_conn);
+      // publish(s_mqtt_conn, mg_str(MQTT_PUBLISH_TOPIC), mg_str("hi"));
     }
   }
 }