Add mg_mqtt_unsub()

mongoose.c

@@ -4042,17 +4042,10 @@ static size_t get_properties_length(struct mg_mqtt_prop *props, size_t count) {
       case MQTT_PROP_TYPE_VARIABLE_INT:
         size += varint_size((uint32_t) props[i].iv);
         break;
-      case MQTT_PROP_TYPE_INT:
-        size += (uint32_t) sizeof(uint32_t);
-        break;
-      case MQTT_PROP_TYPE_SHORT:
-        size += (uint32_t) sizeof(uint16_t);
-        break;
-      case MQTT_PROP_TYPE_BYTE:
-        size += (uint32_t) sizeof(uint8_t);
-        break;
-      default:
-        return size;  // cannot parse further down
+      case MQTT_PROP_TYPE_INT: size += (uint32_t) sizeof(uint32_t); break;
+      case MQTT_PROP_TYPE_SHORT: size += (uint32_t) sizeof(uint16_t); break;
+      case MQTT_PROP_TYPE_BYTE: size += (uint32_t) sizeof(uint8_t); break;
+      default: return size;  // cannot parse further down
     }
   }
 
@@ -4160,8 +4153,7 @@ size_t mg_mqtt_next_prop(struct mg_mqtt_message *msg, struct mg_mqtt_prop *prop,
       len = decode_varint(i, (size_t) (end - i), (size_t *) &prop->iv);
       new_pos = (!len) ? 0 : new_pos + len;
       break;
-    default:
-      new_pos = 0;
+    default: new_pos = 0;
   }
 
   return new_pos;
@@ -4278,12 +4270,14 @@ fail:
   return id;
 }
 
-void mg_mqtt_sub(struct mg_connection *c, const struct mg_mqtt_opts *opts) {
+static void mg_mqtt_sub_unsub(struct mg_connection *c,
+                              const struct mg_mqtt_opts *opts, uint8_t cmd) {
   uint8_t qos_ = opts->qos & 3;
+  bool is_sub = cmd == MQTT_CMD_SUBSCRIBE;
   size_t plen = c->is_mqtt5 ? get_props_size(opts->props, opts->num_props) : 0;
-  size_t len = 2 + opts->topic.len + 2 + 1 + plen;
+  size_t len = 2 + opts->topic.len + 2 + (is_sub ? 1 : 0) + plen;
 
-  if (!mqtt_send_header(c, MQTT_CMD_SUBSCRIBE, 2, (uint32_t) len)) goto fail;
+  if (!mqtt_send_header(c, cmd, 2, (uint32_t) len)) goto fail;
   if (++c->mgr->mqtt_id == 0) ++c->mgr->mqtt_id;
   if (!mg_send_u16(c, mg_htons(c->mgr->mqtt_id))) goto fail;
 
@@ -4291,14 +4285,22 @@ void mg_mqtt_sub(struct mg_connection *c, const struct mg_mqtt_opts *opts) {
     goto fail;
 
   if (!mg_send_u16(c, mg_htons((uint16_t) opts->topic.len)) ||
-      !mg_send(c, opts->topic.buf, opts->topic.len) ||
-      !mg_send(c, &qos_, sizeof(qos_)))
+      !mg_send(c, opts->topic.buf, opts->topic.len))
     goto fail;
+  if (is_sub && !mg_send(c, &qos_, sizeof(qos_))) goto fail;
   return;
 fail:
   mg_error(c, "OOM");
 }
 
+void mg_mqtt_sub(struct mg_connection *c, const struct mg_mqtt_opts *opts) {
+  mg_mqtt_sub_unsub(c, opts, MQTT_CMD_SUBSCRIBE);
+}
+
+void mg_mqtt_unsub(struct mg_connection *c, const struct mg_mqtt_opts *opts) {
+  mg_mqtt_sub_unsub(c, opts, MQTT_CMD_UNSUBSCRIBE);
+}
+
 int mg_mqtt_parse(const uint8_t *buf, size_t len, uint8_t version,
                   struct mg_mqtt_message *m) {
   uint8_t lc = 0, *p, *end;
@@ -4364,8 +4366,7 @@ int mg_mqtt_parse(const uint8_t *buf, size_t len, uint8_t version,
       m->data.len = (size_t) (end - p);
       break;
     }
-    default:
-      break;
+    default: break;
   }
   return MQTT_OK;
 }

mongoose.h

@@ -2930,6 +2930,7 @@ struct mg_connection *mg_mqtt_listen(struct mg_mgr *mgr, const char *url,
 void mg_mqtt_login(struct mg_connection *c, const struct mg_mqtt_opts *opts);
 uint16_t mg_mqtt_pub(struct mg_connection *c, const struct mg_mqtt_opts *opts);
 void mg_mqtt_sub(struct mg_connection *, const struct mg_mqtt_opts *opts);
+void mg_mqtt_unsub(struct mg_connection *c, const struct mg_mqtt_opts *opts);
 int mg_mqtt_parse(const uint8_t *, size_t, uint8_t, struct mg_mqtt_message *);
 void mg_mqtt_send_header(struct mg_connection *, uint8_t cmd, uint8_t flags,
                          uint32_t len);

src/mqtt.c

@@ -1,8 +1,8 @@
-#include "mqtt.h"
 #include "arch.h"
 #include "base64.h"
 #include "event.h"
 #include "log.h"
+#include "mqtt.h"
 #include "url.h"
 #include "util.h"
 
@@ -137,17 +137,10 @@ static size_t get_properties_length(struct mg_mqtt_prop *props, size_t count) {
       case MQTT_PROP_TYPE_VARIABLE_INT:
         size += varint_size((uint32_t) props[i].iv);
         break;
-      case MQTT_PROP_TYPE_INT:
-        size += (uint32_t) sizeof(uint32_t);
-        break;
-      case MQTT_PROP_TYPE_SHORT:
-        size += (uint32_t) sizeof(uint16_t);
-        break;
-      case MQTT_PROP_TYPE_BYTE:
-        size += (uint32_t) sizeof(uint8_t);
-        break;
-      default:
-        return size;  // cannot parse further down
+      case MQTT_PROP_TYPE_INT: size += (uint32_t) sizeof(uint32_t); break;
+      case MQTT_PROP_TYPE_SHORT: size += (uint32_t) sizeof(uint16_t); break;
+      case MQTT_PROP_TYPE_BYTE: size += (uint32_t) sizeof(uint8_t); break;
+      default: return size;  // cannot parse further down
     }
   }
 
@@ -255,8 +248,7 @@ size_t mg_mqtt_next_prop(struct mg_mqtt_message *msg, struct mg_mqtt_prop *prop,
       len = decode_varint(i, (size_t) (end - i), (size_t *) &prop->iv);
       new_pos = (!len) ? 0 : new_pos + len;
       break;
-    default:
-      new_pos = 0;
+    default: new_pos = 0;
   }
 
   return new_pos;
@@ -373,12 +365,14 @@ fail:
   return id;
 }
 
-void mg_mqtt_sub(struct mg_connection *c, const struct mg_mqtt_opts *opts) {
+static void mg_mqtt_sub_unsub(struct mg_connection *c,
+                              const struct mg_mqtt_opts *opts, uint8_t cmd) {
   uint8_t qos_ = opts->qos & 3;
+  bool is_sub = cmd == MQTT_CMD_SUBSCRIBE;
   size_t plen = c->is_mqtt5 ? get_props_size(opts->props, opts->num_props) : 0;
-  size_t len = 2 + opts->topic.len + 2 + 1 + plen;
+  size_t len = 2 + opts->topic.len + 2 + (is_sub ? 1 : 0) + plen;
 
-  if (!mqtt_send_header(c, MQTT_CMD_SUBSCRIBE, 2, (uint32_t) len)) goto fail;
+  if (!mqtt_send_header(c, cmd, 2, (uint32_t) len)) goto fail;
   if (++c->mgr->mqtt_id == 0) ++c->mgr->mqtt_id;
   if (!mg_send_u16(c, mg_htons(c->mgr->mqtt_id))) goto fail;
 
@@ -386,14 +380,22 @@ void mg_mqtt_sub(struct mg_connection *c, const struct mg_mqtt_opts *opts) {
     goto fail;
 
   if (!mg_send_u16(c, mg_htons((uint16_t) opts->topic.len)) ||
-      !mg_send(c, opts->topic.buf, opts->topic.len) ||
-      !mg_send(c, &qos_, sizeof(qos_)))
+      !mg_send(c, opts->topic.buf, opts->topic.len))
     goto fail;
+  if (is_sub && !mg_send(c, &qos_, sizeof(qos_))) goto fail;
   return;
 fail:
   mg_error(c, "OOM");
 }
 
+void mg_mqtt_sub(struct mg_connection *c, const struct mg_mqtt_opts *opts) {
+  mg_mqtt_sub_unsub(c, opts, MQTT_CMD_SUBSCRIBE);
+}
+
+void mg_mqtt_unsub(struct mg_connection *c, const struct mg_mqtt_opts *opts) {
+  mg_mqtt_sub_unsub(c, opts, MQTT_CMD_UNSUBSCRIBE);
+}
+
 int mg_mqtt_parse(const uint8_t *buf, size_t len, uint8_t version,
                   struct mg_mqtt_message *m) {
   uint8_t lc = 0, *p, *end;
@@ -459,8 +461,7 @@ int mg_mqtt_parse(const uint8_t *buf, size_t len, uint8_t version,
       m->data.len = (size_t) (end - p);
       break;
     }
-    default:
-      break;
+    default: break;
   }
   return MQTT_OK;
 }

src/mqtt.h

@@ -104,6 +104,7 @@ struct mg_connection *mg_mqtt_listen(struct mg_mgr *mgr, const char *url,
 void mg_mqtt_login(struct mg_connection *c, const struct mg_mqtt_opts *opts);
 uint16_t mg_mqtt_pub(struct mg_connection *c, const struct mg_mqtt_opts *opts);
 void mg_mqtt_sub(struct mg_connection *, const struct mg_mqtt_opts *opts);
+void mg_mqtt_unsub(struct mg_connection *c, const struct mg_mqtt_opts *opts);
 int mg_mqtt_parse(const uint8_t *, size_t, uint8_t, struct mg_mqtt_message *);
 void mg_mqtt_send_header(struct mg_connection *, uint8_t cmd, uint8_t flags,
                          uint32_t len);

test/unit_test.c

@@ -454,6 +454,7 @@ struct mqtt_data {
 #define flags_received (1 << 2)
 #define flags_released (1 << 3)
 #define flags_completed (1 << 4)
+#define flags_unsubscribed (1 << 5)
 
 static void mqtt_cb(struct mg_connection *c, int ev, void *ev_data) {
   struct mqtt_data *test_data = (struct mqtt_data *) c->fn_data;
@@ -487,6 +488,8 @@ static void mqtt_cb(struct mg_connection *c, int ev, void *ev_data) {
       test_data->flags |= flags_released;
     } else if (mm->cmd == MQTT_CMD_PUBCOMP) {
       test_data->flags |= flags_completed;
+    } else if (mm->cmd == MQTT_CMD_UNSUBACK) {
+      test_data->flags = flags_unsubscribed;
     }
   } else if (ev == MG_EV_MQTT_MSG) {
     struct mg_mqtt_message *mm = (struct mg_mqtt_message *) ev_data;
@@ -765,6 +768,12 @@ static void test_mqtt_ver(uint8_t mqtt_version) {
   memset(mbuf + 1, 0, sizeof(mbuf) - 1);
   test_data.flags = 0;
 
+  opts.props = 0; opts.num_props = 0;
+  mg_mqtt_unsub(c, &opts);
+  for (i = 0; i < 500 && test_data.flags == 0; i++) mg_mgr_poll(&mgr, 10);
+  ASSERT(test_data.flags == flags_unsubscribed);
+  test_data.flags = 0;
+
   // dirty disconnect
   mg_mgr_free(&mgr);
   ASSERT(mgr.conns == NULL);
@@ -4155,7 +4164,6 @@ int main(void) {
   // detectors. We are actually freeing all our resources and leaving
   mbedtls_psa_crypto_free();
 #endif
-
   printf("SUCCESS. Total tests: %d\n", s_num_tests);
   return EXIT_SUCCESS;
 }