mirror of
https://github.com/cesanta/mongoose.git
synced 2026-10-28 12:30:58 +00:00
add simple API support for app MQTT pub retries
This commit is contained in:
1 parent
9fa840c1c7
commit
2bd188f3d6
5 files changed
+32
-15
No files matched your search
+10
-4
@@ -319,7 +319,8 @@ void mg_mqtt_login(struct mg_connection *c, const struct mg_mqtt_opts *opts) {
|
||||
}
|
||||
}
|
||||
|
||||
void mg_mqtt_pub(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) {
|
||||
uint16_t id = opts->retransmit_id;
|
||||
uint8_t flags = (uint8_t) (((opts->qos & 3) << 1) | (opts->retain ? 1 : 0));
|
||||
size_t len = 2 + opts->topic.len + opts->message.len;
|
||||
MG_DEBUG(("%lu [%.*s] -> [%.*s]", c->id, (int) opts->topic.len,
|
||||
@@ -328,17 +329,22 @@ void mg_mqtt_pub(struct mg_connection *c, const struct mg_mqtt_opts *opts) {
|
||||
if (opts->qos > 0) len += 2;
|
||||
if (c->is_mqtt5) len += get_props_size(opts->props, opts->num_props);
|
||||
|
||||
if (opts->qos > 0 && id != 0) flags |= 1 << 3;
|
||||
mg_mqtt_send_header(c, MQTT_CMD_PUBLISH, flags, (uint32_t) len);
|
||||
mg_send_u16(c, mg_htons((uint16_t) opts->topic.len));
|
||||
mg_send(c, opts->topic.buf, opts->topic.len);
|
||||
if (opts->qos > 0) {
|
||||
if (++c->mgr->mqtt_id == 0) ++c->mgr->mqtt_id;
|
||||
mg_send_u16(c, mg_htons(c->mgr->mqtt_id));
|
||||
if (opts->qos > 0) { // need to send 'id' field
|
||||
if (id == 0) { // generate new one if not resending
|
||||
if (++c->mgr->mqtt_id == 0) ++c->mgr->mqtt_id;
|
||||
id = c->mgr->mqtt_id;
|
||||
}
|
||||
mg_send_u16(c, mg_htons(id));
|
||||
}
|
||||
|
||||
if (c->is_mqtt5) mg_send_mqtt_properties(c, opts->props, opts->num_props);
|
||||
|
||||
if (opts->message.len > 0) mg_send(c, opts->message.buf, opts->message.len);
|
||||
return id;
|
||||
}
|
||||
|
||||
void mg_mqtt_sub(struct mg_connection *c, const struct mg_mqtt_opts *opts) {
|
||||
|
||||
+2
-1
@@ -75,6 +75,7 @@ struct mg_mqtt_opts {
|
||||
uint8_t qos; // 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
|
||||
uint16_t retransmit_id; // For PUBLISH, init to 0
|
||||
bool retain; // Retain flag
|
||||
bool clean; // Clean session flag
|
||||
struct mg_mqtt_prop *props; // MQTT5 props array
|
||||
@@ -101,7 +102,7 @@ struct mg_connection *mg_mqtt_connect(struct mg_mgr *, const char *url,
|
||||
struct mg_connection *mg_mqtt_listen(struct mg_mgr *mgr, const char *url,
|
||||
mg_event_handler_t fn, void *fn_data);
|
||||
void mg_mqtt_login(struct mg_connection *c, const struct mg_mqtt_opts *opts);
|
||||
void mg_mqtt_pub(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);
|
||||
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,
|
||||
|
||||
Reference in new issue
Block a user