diff --git a/examples/mqtt-over-ws-client/main.c b/examples/mqtt-over-ws-client/main.c index 62304e8c..1f909f58 100644 --- a/examples/mqtt-over-ws-client/main.c +++ b/examples/mqtt-over-ws-client/main.c @@ -37,8 +37,10 @@ static void fn(struct mg_connection *c, int ev, void *ev_data, void *fn_data) { } else if (ev == MG_EV_WS_MSG) { struct mg_mqtt_message mm; struct mg_ws_message *wm = (struct mg_ws_message *) ev_data; + uint8_t version = (uint8_t) (size_t) c->pfn_data; MG_INFO(("GOT %d bytes WS msg", (int) wm->data.len)); - while ((mg_mqtt_parse((uint8_t *) wm->data.ptr, wm->data.len, &mm)) == 0) { + while ((mg_mqtt_parse((uint8_t *) wm->data.ptr, wm->data.len, version, + &mm)) == 0) { switch (mm.cmd) { case MQTT_CMD_CONNACK: mg_call(c, MG_EV_MQTT_OPEN, &mm.ack); diff --git a/mongoose.c b/mongoose.c index ffd800fc..ae44f586 100644 --- a/mongoose.c +++ b/mongoose.c @@ -566,7 +566,7 @@ static void mg_putchar_iobuf(char ch, void *param) { if (io->len < io->size) io->buf[io->len++] = (uint8_t) ch; } -void mg_putchar_stralloc(char ch, void *param) { +void mg_putchar_realloc(char ch, void *param) { char *s, *buf = *(char **) param; size_t len = buf == NULL ? 0 : strlen(buf), chunksize = 256; size_t new_size = len + 1 + 1 + chunksize; @@ -2967,43 +2967,50 @@ static void mg_send_u16(struct mg_connection *c, uint16_t value) { } void mg_mqtt_login(struct mg_connection *c, const struct mg_mqtt_opts *opts) { - char rnd[9], client_id[16]; + char rnd[10], client_id[21], zero = 0; struct mg_str cid = opts->client_id; uint32_t total_len = 7 + 1 + 2 + 2; - uint8_t connflag = (uint8_t) ((opts->will_qos & 3) << 3); + uint8_t hdr[8] = {0, 4, 'M', 'Q', 'T', 'T', opts->version, 0}; if (cid.len == 0) { mg_random(rnd, sizeof(rnd)); - mg_base64_encode((unsigned char *) rnd, sizeof(rnd), client_id); + mg_hex(rnd, sizeof(rnd), client_id); client_id[sizeof(client_id) - 1] = '\0'; cid = mg_str(client_id); } + if (hdr[6] == 0) hdr[6] = 4; // If version is not set, use 4 (3.1.1) + c->pfn_data = (void *) (size_t) hdr[6]; // Store version + hdr[7] = (uint8_t) ((opts->will_qos & 3) << 3); // Connection flags if (opts->user.len > 0) { total_len += 2 + (uint32_t) opts->user.len; - connflag |= MQTT_HAS_USER_NAME; + hdr[7] |= MQTT_HAS_USER_NAME; } if (opts->pass.len > 0) { total_len += 2 + (uint32_t) opts->pass.len; - connflag |= MQTT_HAS_PASSWORD; + hdr[7] |= MQTT_HAS_PASSWORD; } if (opts->will_topic.len > 0 && opts->will_message.len > 0) { total_len += 4 + (uint32_t) opts->will_topic.len + (uint32_t) opts->will_message.len; - connflag |= MQTT_HAS_WILL; + hdr[7] |= MQTT_HAS_WILL; } - if (opts->clean || cid.len == 0) connflag |= MQTT_CLEAN_SESSION; - if (opts->will_retain) connflag |= MQTT_WILL_RETAIN; + if (opts->clean || cid.len == 0) hdr[7] |= MQTT_CLEAN_SESSION; + if (opts->will_retain) hdr[7] |= MQTT_WILL_RETAIN; total_len += (uint32_t) cid.len; + if (opts->version == 5) total_len += 1U + (hdr[7] & MQTT_HAS_WILL ? 1U : 0); mg_mqtt_send_header(c, MQTT_CMD_CONNECT, 0, total_len); - mg_send(c, "\00\04MQTT\04", 7); - mg_send(c, &connflag, sizeof(connflag)); + mg_send(c, hdr, sizeof(hdr)); // keepalive == 0 means "do not disconnect us!" mg_send_u16(c, mg_htons((uint16_t) opts->keepalive)); + + if (opts->version == 5) mg_send(c, &zero, sizeof(zero)); // V5 properties mg_send_u16(c, mg_htons((uint16_t) cid.len)); mg_send(c, cid.ptr, cid.len); - if (connflag & MQTT_HAS_WILL) { + + if (hdr[7] & MQTT_HAS_WILL) { + if (opts->version == 5) mg_send(c, &zero, sizeof(zero)); // will props mg_send_u16(c, mg_htons((uint16_t) opts->will_topic.len)); mg_send(c, opts->will_topic.ptr, opts->will_topic.len); mg_send_u16(c, mg_htons((uint16_t) opts->will_message.len)); @@ -3022,32 +3029,37 @@ void mg_mqtt_login(struct mg_connection *c, const struct mg_mqtt_opts *opts) { void mg_mqtt_pub(struct mg_connection *c, struct mg_str topic, struct mg_str data, int qos, bool retain) { uint8_t flags = (uint8_t) (((qos & 3) << 1) | (retain ? 1 : 0)); - uint32_t total_len = 2 + (uint32_t) topic.len + (uint32_t) data.len; + uint8_t version = (uint8_t) (size_t) c->pfn_data, zero = 0; + uint32_t len = 2 + (uint32_t) topic.len + (uint32_t) data.len; MG_DEBUG(("%lu [%.*s] -> [%.*s]", c->id, (int) topic.len, (char *) topic.ptr, (int) data.len, (char *) data.ptr)); - if (qos > 0) total_len += 2; - mg_mqtt_send_header(c, MQTT_CMD_PUBLISH, flags, total_len); + if (qos > 0) len += 2; + if (version == 5) len++; + mg_mqtt_send_header(c, MQTT_CMD_PUBLISH, flags, len); mg_send_u16(c, mg_htons((uint16_t) topic.len)); mg_send(c, topic.ptr, topic.len); if (qos > 0) { if (++c->mgr->mqtt_id == 0) ++c->mgr->mqtt_id; mg_send_u16(c, mg_htons(c->mgr->mqtt_id)); } + if (version == 5) mg_send(c, &zero, sizeof(zero)); mg_send(c, data.ptr, data.len); } void mg_mqtt_sub(struct mg_connection *c, struct mg_str topic, int qos) { - uint8_t qos_ = qos & 3; - uint32_t total_len = 2 + (uint32_t) topic.len + 2 + 1; - mg_mqtt_send_header(c, MQTT_CMD_SUBSCRIBE, 2, total_len); + uint8_t qos_ = qos & 3, version = (uint8_t) (size_t) c->pfn_data, zero = 0; + uint32_t len = 2 + (uint32_t) topic.len + 2 + 1 + (version == 5 ? 1 : 0); + mg_mqtt_send_header(c, MQTT_CMD_SUBSCRIBE, 2, len); if (++c->mgr->mqtt_id == 0) ++c->mgr->mqtt_id; mg_send_u16(c, mg_htons(c->mgr->mqtt_id)); + if (version == 5) mg_send(c, &zero, sizeof(zero)); mg_send_u16(c, mg_htons((uint16_t) topic.len)); mg_send(c, topic.ptr, topic.len); mg_send(c, &qos_, sizeof(qos_)); } -int mg_mqtt_parse(const uint8_t *buf, size_t len, struct mg_mqtt_message *m) { +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; uint32_t n = 0, len_len = 0; @@ -3079,16 +3091,14 @@ int mg_mqtt_parse(const uint8_t *buf, size_t len, struct mg_mqtt_message *m) { case MQTT_CMD_PUBREC: case MQTT_CMD_PUBREL: case MQTT_CMD_PUBCOMP: + case MQTT_CMD_SUBSCRIBE: case MQTT_CMD_SUBACK: - if (p + 2 > end) return MQTT_MALFORMED; - m->id = (uint16_t) ((((uint16_t) p[0]) << 8) | p[1]); - break; - case MQTT_CMD_SUBSCRIBE: { + case MQTT_CMD_UNSUBSCRIBE: + case MQTT_CMD_UNSUBACK: if (p + 2 > end) return MQTT_MALFORMED; m->id = (uint16_t) ((((uint16_t) p[0]) << 8) | p[1]); p += 2; break; - } case MQTT_CMD_PUBLISH: { if (p + 2 > end) return MQTT_MALFORMED; m->topic.len = (uint16_t) ((((uint16_t) p[0]) << 8) | p[1]); @@ -3100,6 +3110,8 @@ int mg_mqtt_parse(const uint8_t *buf, size_t len, struct mg_mqtt_message *m) { m->id = (uint16_t) ((((uint16_t) p[0]) << 8) | p[1]); p += 2; } + if (p >= end) return MQTT_MALFORMED; + if (version == 5) p += 1 + p[0]; // Skip options if (p > end) return MQTT_MALFORMED; m->data.ptr = (char *) p; m->data.len = (size_t) (end - p); @@ -3141,8 +3153,9 @@ static void mqtt_cb(struct mg_connection *c, int ev, void *ev_data, void *fn_data) { if (ev == MG_EV_READ) { for (;;) { + uint8_t version = (uint8_t) (size_t) c->pfn_data; struct mg_mqtt_message mm; - int rc = mg_mqtt_parse(c->recv.buf, c->recv.len, &mm); + int rc = mg_mqtt_parse(c->recv.buf, c->recv.len, version, &mm); if (rc == MQTT_MALFORMED) { MG_ERROR(("%lu MQTT malformed message", c->id)); c->is_closing = 1; diff --git a/mongoose.h b/mongoose.h index 1bb70329..72b5c8e0 100644 --- a/mongoose.h +++ b/mongoose.h @@ -727,7 +727,7 @@ size_t mg_dtoa(char *buf, size_t len, double d, int width); typedef void (*mg_pc_t)(char, void *); // Custom putchar typedef size_t (*mg_pm_t)(mg_pc_t, void *, va_list *); // %M printer -void mg_putchar_stralloc(char ch, void *param); // Print to malloced str +void mg_putchar_realloc(char ch, void *param); // Print to malloced str size_t mg_vrprintf(void (*)(char, void *), void *, const char *fmt, va_list *); size_t mg_rprintf(void (*fn)(char, void *), void *, const char *fmt, ...); @@ -1221,6 +1221,7 @@ int64_t mg_sntp_parse(const unsigned char *buf, size_t len); #define MQTT_CMD_PINGREQ 12 #define MQTT_CMD_PINGRESP 13 #define MQTT_CMD_DISCONNECT 14 +#define MQTT_CMD_AUTH 15 struct mg_mqtt_opts { struct mg_str user; // Username, can be empty @@ -1229,9 +1230,10 @@ struct mg_mqtt_opts { struct mg_str will_topic; // Will topic struct mg_str will_message; // Will message uint8_t will_qos; // Will message quality of service + uint8_t version; // Can be 4 (3.1.1), or 5. If 0, assume 4. + uint16_t keepalive; // Keep-alive timer in seconds bool will_retain; // Retain last will bool clean; // Use clean session, 0 or 1 - uint16_t keepalive; // Keep-alive timer in seconds }; struct mg_mqtt_message { @@ -1253,7 +1255,7 @@ void mg_mqtt_login(struct mg_connection *c, const struct mg_mqtt_opts *opts); void mg_mqtt_pub(struct mg_connection *c, struct mg_str topic, struct mg_str data, int qos, bool retain); void mg_mqtt_sub(struct mg_connection *, struct mg_str topic, int qos); -int mg_mqtt_parse(const uint8_t *buf, size_t len, struct mg_mqtt_message *m); +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); size_t mg_mqtt_next_sub(struct mg_mqtt_message *msg, struct mg_str *topic, diff --git a/src/fmt.c b/src/fmt.c index 4ec57070..35e595d7 100644 --- a/src/fmt.c +++ b/src/fmt.c @@ -15,7 +15,7 @@ static void mg_putchar_iobuf(char ch, void *param) { if (io->len < io->size) io->buf[io->len++] = (uint8_t) ch; } -void mg_putchar_stralloc(char ch, void *param) { +void mg_putchar_realloc(char ch, void *param) { char *s, *buf = *(char **) param; size_t len = buf == NULL ? 0 : strlen(buf), chunksize = 256; size_t new_size = len + 1 + 1 + chunksize; diff --git a/src/mqtt.c b/src/mqtt.c index 5d058c41..b4c32c53 100644 --- a/src/mqtt.c +++ b/src/mqtt.c @@ -32,43 +32,50 @@ static void mg_send_u16(struct mg_connection *c, uint16_t value) { } void mg_mqtt_login(struct mg_connection *c, const struct mg_mqtt_opts *opts) { - char rnd[9], client_id[16]; + char rnd[10], client_id[21], zero = 0; struct mg_str cid = opts->client_id; uint32_t total_len = 7 + 1 + 2 + 2; - uint8_t connflag = (uint8_t) ((opts->will_qos & 3) << 3); + uint8_t hdr[8] = {0, 4, 'M', 'Q', 'T', 'T', opts->version, 0}; if (cid.len == 0) { mg_random(rnd, sizeof(rnd)); - mg_base64_encode((unsigned char *) rnd, sizeof(rnd), client_id); + mg_hex(rnd, sizeof(rnd), client_id); client_id[sizeof(client_id) - 1] = '\0'; cid = mg_str(client_id); } + if (hdr[6] == 0) hdr[6] = 4; // If version is not set, use 4 (3.1.1) + c->pfn_data = (void *) (size_t) hdr[6]; // Store version + hdr[7] = (uint8_t) ((opts->will_qos & 3) << 3); // Connection flags if (opts->user.len > 0) { total_len += 2 + (uint32_t) opts->user.len; - connflag |= MQTT_HAS_USER_NAME; + hdr[7] |= MQTT_HAS_USER_NAME; } if (opts->pass.len > 0) { total_len += 2 + (uint32_t) opts->pass.len; - connflag |= MQTT_HAS_PASSWORD; + hdr[7] |= MQTT_HAS_PASSWORD; } if (opts->will_topic.len > 0 && opts->will_message.len > 0) { total_len += 4 + (uint32_t) opts->will_topic.len + (uint32_t) opts->will_message.len; - connflag |= MQTT_HAS_WILL; + hdr[7] |= MQTT_HAS_WILL; } - if (opts->clean || cid.len == 0) connflag |= MQTT_CLEAN_SESSION; - if (opts->will_retain) connflag |= MQTT_WILL_RETAIN; + if (opts->clean || cid.len == 0) hdr[7] |= MQTT_CLEAN_SESSION; + if (opts->will_retain) hdr[7] |= MQTT_WILL_RETAIN; total_len += (uint32_t) cid.len; + if (opts->version == 5) total_len += 1U + (hdr[7] & MQTT_HAS_WILL ? 1U : 0); mg_mqtt_send_header(c, MQTT_CMD_CONNECT, 0, total_len); - mg_send(c, "\00\04MQTT\04", 7); - mg_send(c, &connflag, sizeof(connflag)); + mg_send(c, hdr, sizeof(hdr)); // keepalive == 0 means "do not disconnect us!" mg_send_u16(c, mg_htons((uint16_t) opts->keepalive)); + + if (opts->version == 5) mg_send(c, &zero, sizeof(zero)); // V5 properties mg_send_u16(c, mg_htons((uint16_t) cid.len)); mg_send(c, cid.ptr, cid.len); - if (connflag & MQTT_HAS_WILL) { + + if (hdr[7] & MQTT_HAS_WILL) { + if (opts->version == 5) mg_send(c, &zero, sizeof(zero)); // will props mg_send_u16(c, mg_htons((uint16_t) opts->will_topic.len)); mg_send(c, opts->will_topic.ptr, opts->will_topic.len); mg_send_u16(c, mg_htons((uint16_t) opts->will_message.len)); @@ -87,32 +94,37 @@ void mg_mqtt_login(struct mg_connection *c, const struct mg_mqtt_opts *opts) { void mg_mqtt_pub(struct mg_connection *c, struct mg_str topic, struct mg_str data, int qos, bool retain) { uint8_t flags = (uint8_t) (((qos & 3) << 1) | (retain ? 1 : 0)); - uint32_t total_len = 2 + (uint32_t) topic.len + (uint32_t) data.len; + uint8_t version = (uint8_t) (size_t) c->pfn_data, zero = 0; + uint32_t len = 2 + (uint32_t) topic.len + (uint32_t) data.len; MG_DEBUG(("%lu [%.*s] -> [%.*s]", c->id, (int) topic.len, (char *) topic.ptr, (int) data.len, (char *) data.ptr)); - if (qos > 0) total_len += 2; - mg_mqtt_send_header(c, MQTT_CMD_PUBLISH, flags, total_len); + if (qos > 0) len += 2; + if (version == 5) len++; + mg_mqtt_send_header(c, MQTT_CMD_PUBLISH, flags, len); mg_send_u16(c, mg_htons((uint16_t) topic.len)); mg_send(c, topic.ptr, topic.len); if (qos > 0) { if (++c->mgr->mqtt_id == 0) ++c->mgr->mqtt_id; mg_send_u16(c, mg_htons(c->mgr->mqtt_id)); } + if (version == 5) mg_send(c, &zero, sizeof(zero)); mg_send(c, data.ptr, data.len); } void mg_mqtt_sub(struct mg_connection *c, struct mg_str topic, int qos) { - uint8_t qos_ = qos & 3; - uint32_t total_len = 2 + (uint32_t) topic.len + 2 + 1; - mg_mqtt_send_header(c, MQTT_CMD_SUBSCRIBE, 2, total_len); + uint8_t qos_ = qos & 3, version = (uint8_t) (size_t) c->pfn_data, zero = 0; + uint32_t len = 2 + (uint32_t) topic.len + 2 + 1 + (version == 5 ? 1 : 0); + mg_mqtt_send_header(c, MQTT_CMD_SUBSCRIBE, 2, len); if (++c->mgr->mqtt_id == 0) ++c->mgr->mqtt_id; mg_send_u16(c, mg_htons(c->mgr->mqtt_id)); + if (version == 5) mg_send(c, &zero, sizeof(zero)); mg_send_u16(c, mg_htons((uint16_t) topic.len)); mg_send(c, topic.ptr, topic.len); mg_send(c, &qos_, sizeof(qos_)); } -int mg_mqtt_parse(const uint8_t *buf, size_t len, struct mg_mqtt_message *m) { +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; uint32_t n = 0, len_len = 0; @@ -144,16 +156,14 @@ int mg_mqtt_parse(const uint8_t *buf, size_t len, struct mg_mqtt_message *m) { case MQTT_CMD_PUBREC: case MQTT_CMD_PUBREL: case MQTT_CMD_PUBCOMP: + case MQTT_CMD_SUBSCRIBE: case MQTT_CMD_SUBACK: - if (p + 2 > end) return MQTT_MALFORMED; - m->id = (uint16_t) ((((uint16_t) p[0]) << 8) | p[1]); - break; - case MQTT_CMD_SUBSCRIBE: { + case MQTT_CMD_UNSUBSCRIBE: + case MQTT_CMD_UNSUBACK: if (p + 2 > end) return MQTT_MALFORMED; m->id = (uint16_t) ((((uint16_t) p[0]) << 8) | p[1]); p += 2; break; - } case MQTT_CMD_PUBLISH: { if (p + 2 > end) return MQTT_MALFORMED; m->topic.len = (uint16_t) ((((uint16_t) p[0]) << 8) | p[1]); @@ -165,6 +175,8 @@ int mg_mqtt_parse(const uint8_t *buf, size_t len, struct mg_mqtt_message *m) { m->id = (uint16_t) ((((uint16_t) p[0]) << 8) | p[1]); p += 2; } + if (p >= end) return MQTT_MALFORMED; + if (version == 5) p += 1 + p[0]; // Skip options if (p > end) return MQTT_MALFORMED; m->data.ptr = (char *) p; m->data.len = (size_t) (end - p); @@ -206,8 +218,9 @@ static void mqtt_cb(struct mg_connection *c, int ev, void *ev_data, void *fn_data) { if (ev == MG_EV_READ) { for (;;) { + uint8_t version = (uint8_t) (size_t) c->pfn_data; struct mg_mqtt_message mm; - int rc = mg_mqtt_parse(c->recv.buf, c->recv.len, &mm); + int rc = mg_mqtt_parse(c->recv.buf, c->recv.len, version, &mm); if (rc == MQTT_MALFORMED) { MG_ERROR(("%lu MQTT malformed message", c->id)); c->is_closing = 1; diff --git a/src/mqtt.h b/src/mqtt.h index c559a327..fd1768df 100644 --- a/src/mqtt.h +++ b/src/mqtt.h @@ -17,6 +17,7 @@ #define MQTT_CMD_PINGREQ 12 #define MQTT_CMD_PINGRESP 13 #define MQTT_CMD_DISCONNECT 14 +#define MQTT_CMD_AUTH 15 struct mg_mqtt_opts { struct mg_str user; // Username, can be empty @@ -25,9 +26,10 @@ struct mg_mqtt_opts { struct mg_str will_topic; // Will topic struct mg_str will_message; // Will message uint8_t will_qos; // Will message quality of service + uint8_t version; // Can be 4 (3.1.1), or 5. If 0, assume 4. + uint16_t keepalive; // Keep-alive timer in seconds bool will_retain; // Retain last will bool clean; // Use clean session, 0 or 1 - uint16_t keepalive; // Keep-alive timer in seconds }; struct mg_mqtt_message { @@ -49,7 +51,7 @@ void mg_mqtt_login(struct mg_connection *c, const struct mg_mqtt_opts *opts); void mg_mqtt_pub(struct mg_connection *c, struct mg_str topic, struct mg_str data, int qos, bool retain); void mg_mqtt_sub(struct mg_connection *, struct mg_str topic, int qos); -int mg_mqtt_parse(const uint8_t *buf, size_t len, struct mg_mqtt_message *m); +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); size_t mg_mqtt_next_sub(struct mg_mqtt_message *msg, struct mg_str *topic, diff --git a/src/str.h b/src/str.h index aecece68..4aac6960 100644 --- a/src/str.h +++ b/src/str.h @@ -43,7 +43,7 @@ size_t mg_dtoa(char *buf, size_t len, double d, int width); typedef void (*mg_pc_t)(char, void *); // Custom putchar typedef size_t (*mg_pm_t)(mg_pc_t, void *, va_list *); // %M printer -void mg_putchar_stralloc(char ch, void *param); // Print to malloced str +void mg_putchar_realloc(char ch, void *param); // Print to malloced str size_t mg_vrprintf(void (*)(char, void *), void *, const char *fmt, va_list *); size_t mg_rprintf(void (*fn)(char, void *), void *, const char *fmt, ...); diff --git a/test/fuzz.c b/test/fuzz.c index 2af58e07..2baaaeea 100644 --- a/test/fuzz.c +++ b/test/fuzz.c @@ -26,8 +26,10 @@ int LLVMFuzzerTestOneInput(const uint8_t *data, size_t size) { mg_url_decode(NULL, 0, tmp, 1, 1); struct mg_mqtt_message mm; - mg_mqtt_parse(data, size, &mm); - mg_mqtt_parse(NULL, 0, &mm); + mg_mqtt_parse(data, size, 0, &mm); + mg_mqtt_parse(NULL, 0, 0, &mm); + mg_mqtt_parse(data, size, 5, &mm); + mg_mqtt_parse(NULL, 0, 5, &mm); mg_sntp_parse(data, size); mg_sntp_parse(NULL, 0); diff --git a/test/unit_test.c b/test/unit_test.c index 8b48cb9a..23a72f06 100644 --- a/test/unit_test.c +++ b/test/unit_test.c @@ -371,25 +371,17 @@ static void mqtt_cb(struct mg_connection *c, int ev, void *evd, void *fnd) { (void) c; } -static void test_mqtt(void) { +static void test_mqtt_ver(uint8_t mqtt_version) { char buf[50] = {0}; struct mqtt_data test_data = {buf, 0, 0}; struct mg_mgr mgr; struct mg_str topic = mg_str("x/f12"), data = mg_str("hi"); struct mg_connection *c; struct mg_mqtt_opts opts; - char rnd[9], client_id[16]; - // const char *url = "mqtt://mqtt.eclipse.org:1883"; const char *url = "mqtt://broker.hivemq.com:1883"; int i; mg_mgr_init(&mgr); - { - uint8_t bad[] = " \xff\xff\xff\xff "; - struct mg_mqtt_message mm; - mg_mqtt_parse(bad, sizeof(bad), &mm); - } - // Connect with empty client ID c = mg_mqtt_connect(&mgr, url, NULL, mqtt_cb, &test_data); for (i = 0; i < 300 && buf[0] == 0; i++) mg_mgr_poll(&mgr, 10); @@ -416,12 +408,10 @@ static void test_mqtt(void) { opts.will_qos = 1; opts.will_retain = true; opts.keepalive = 20; - mg_random(rnd, sizeof(rnd)); - mg_base64_encode((unsigned char *) rnd, sizeof(rnd), client_id); - client_id[sizeof(client_id) - 1] = '\0'; - opts.client_id = mg_str(client_id); + opts.version = mqtt_version; opts.will_topic = mg_str("mg_will_topic"); opts.will_message = mg_str("mg_will_messsage"); + opts.client_id = mg_str("mg_unit_test"); c = mg_mqtt_connect(&mgr, url, &opts, mqtt_cb, &test_data); for (i = 0; i < 300 && buf[0] == 0; i++) mg_mgr_poll(&mgr, 10); if (buf[0] != 'X') MG_INFO(("[%s]", buf)); @@ -442,6 +432,11 @@ static void test_mqtt(void) { ASSERT(mgr.conns == NULL); } +static void test_mqtt(void) { + test_mqtt_ver(5); + test_mqtt_ver(4); +} + static void eh1(struct mg_connection *c, int ev, void *ev_data, void *fn_data) { struct mg_tls_opts *topts = (struct mg_tls_opts *) fn_data; if (ev == MG_EV_ACCEPT && topts != NULL) mg_tls_init(c, topts); @@ -1494,8 +1489,8 @@ static void test_str(void) { free(p); p = mg_mprintf("[%M", pf2, 10); - mg_rprintf(mg_putchar_stralloc, &p, ","); - mg_rprintf(mg_putchar_stralloc, &p, "%d]", 7); + mg_rprintf(mg_putchar_realloc, &p, ","); + mg_rprintf(mg_putchar_realloc, &p, "%d]", 7); printf("-> %s\n", p); ASSERT(strcmp(p, "[9876543210,7]") == 0); free(p);