diff --git a/mongoose.c b/mongoose.c index 748cf757..46613481 100644 --- a/mongoose.c +++ b/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; } diff --git a/mongoose.h b/mongoose.h index 185381c7..90f7eb39 100644 --- a/mongoose.h +++ b/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); diff --git a/src/mqtt.c b/src/mqtt.c index a4c82015..3235f238 100644 --- a/src/mqtt.c +++ b/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; } diff --git a/src/mqtt.h b/src/mqtt.h index eb4567bc..e67ce90d 100644 --- a/src/mqtt.h +++ b/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); diff --git a/test/unit_test.c b/test/unit_test.c index 6bd7f247..44c9dc82 100644 --- a/test/unit_test.c +++ b/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; }