wt/mongoose
1
0 Fork 0
mirror of https://github.com/cesanta/mongoose.git synced 2026-10-19 11:09:42 +00:00
Code Issues Packages Projects Releases Wiki Activity

add OOM handling to apps

Browse Source
This commit is contained in:
Sergio R. Caprile committed 2025-11-05 14:01:49 -03:00
1 parent 84e9ea76db
commit c8291e4ae9
6 files changed
+238 -162
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

mongoose.c
+119 -81
View File
@@ -1793,14 +1793,14 @@ int mg_http_parse(const char *s, size_t len, struct mg_http_message *hm) {
static void mg_http_vprintf_chunk(struct mg_connection *c, const char *fmt,
va_list *ap) {
size_t len = c->send.len;
mg_send(c, " \r\n", 10);
if (!mg_send(c, " \r\n", 10)) mg_error(c, "OOM");
mg_vxprintf(mg_pfn_iobuf, &c->send, fmt, ap);
if (c->send.len >= len + 10) {
mg_snprintf((char *) c->send.buf + len, 9, "%08lx", c->send.len - len - 10);
c->send.buf[len + 8] = '\r';
if (c->send.len == len + 10) c->is_resp = 0; // Last chunk, reset marker
}
mg_send(c, "\r\n", 2);
if (!mg_send(c, "\r\n", 2)) mg_error(c, "OOM");
}
void mg_http_printf_chunk(struct mg_connection *c, const char *fmt, ...) {
@@ -1812,8 +1812,7 @@ void mg_http_printf_chunk(struct mg_connection *c, const char *fmt, ...) {
void mg_http_write_chunk(struct mg_connection *c, const char *buf, size_t len) {
mg_printf(c, "%lx\r\n", (unsigned long) len);
mg_send(c, buf, len);
mg_send(c, "\r\n", 2);
if (!mg_send(c, buf, len) || !mg_send(c, "\r\n", 2)) mg_error(c, "OOM");
if (len == 0) c->is_resp = 0;
}
@@ -3353,8 +3352,8 @@ static const struct mg_mqtt_pmap s_prop_map[] = {
{MQTT_PROP_SUBSCRIPTION_IDENTIFIER_AVAILABLE, MQTT_PROP_TYPE_BYTE},
{MQTT_PROP_SHARED_SUBSCRIPTION_AVAILABLE, MQTT_PROP_TYPE_BYTE}};
void mg_mqtt_send_header(struct mg_connection *c, uint8_t cmd, uint8_t flags,
uint32_t len) {
static bool mqtt_send_header(struct mg_connection *c, uint8_t cmd,
uint8_t flags, uint32_t len) {
uint8_t buf[1 + sizeof(len)], *vlen = &buf[1];
buf[0] = (uint8_t) ((cmd << 4) | flags);
do {
@@ -3363,15 +3362,20 @@ void mg_mqtt_send_header(struct mg_connection *c, uint8_t cmd, uint8_t flags,
if (len > 0) *vlen |= 0x80;
vlen++;
} while (len > 0 && vlen < &buf[sizeof(buf)]);
mg_send(c, buf, (size_t) (vlen - buf));
return mg_send(c, buf, (size_t) (vlen - buf));
}
static void mg_send_u16(struct mg_connection *c, uint16_t value) {
mg_send(c, &value, sizeof(value));
void mg_mqtt_send_header(struct mg_connection *c, uint8_t cmd, uint8_t flags,
uint32_t len) {
if (!mqtt_send_header(c, cmd, flags, len)) mg_error(c, "OOM");
}
static void mg_send_u32(struct mg_connection *c, uint32_t value) {
mg_send(c, &value, sizeof(value));
static bool mg_send_u16(struct mg_connection *c, uint16_t value) {
return mg_send(c, &value, sizeof(value));
}
static bool mg_send_u32(struct mg_connection *c, uint32_t value) {
return mg_send(c, &value, sizeof(value));
}
static uint8_t varint_size(size_t length) {
@@ -3464,46 +3468,50 @@ static size_t get_props_size(struct mg_mqtt_prop *props, size_t count) {
return size;
}
static void mg_send_mqtt_properties(struct mg_connection *c,
static bool mg_send_mqtt_properties(struct mg_connection *c,
struct mg_mqtt_prop *props, size_t nprops) {
size_t total_size = get_properties_length(props, nprops);
uint8_t buf_v[4] = {0, 0, 0, 0};
uint8_t buf[4] = {0, 0, 0, 0};
size_t i, len = encode_varint(buf, total_size);
mg_send(c, buf, (size_t) len);
if (!mg_send(c, buf, (size_t) len)) return false;
for (i = 0; i < nprops; i++) {
mg_send(c, &props[i].id, sizeof(props[i].id));
if (!mg_send(c, &props[i].id, sizeof(props[i].id))) return false;
switch (mqtt_prop_type_by_id(props[i].id)) {
case MQTT_PROP_TYPE_STRING_PAIR:
mg_send_u16(c, mg_htons((uint16_t) props[i].key.len));
mg_send(c, props[i].key.buf, props[i].key.len);
mg_send_u16(c, mg_htons((uint16_t) props[i].val.len));
mg_send(c, props[i].val.buf, props[i].val.len);
if (!mg_send_u16(c, mg_htons((uint16_t) props[i].key.len)) ||
!mg_send(c, props[i].key.buf, props[i].key.len) ||
!mg_send_u16(c, mg_htons((uint16_t) props[i].val.len)) ||
!mg_send(c, props[i].val.buf, props[i].val.len))
return false;
break;
case MQTT_PROP_TYPE_BYTE:
mg_send(c, &props[i].iv, sizeof(uint8_t));
if (!mg_send(c, &props[i].iv, sizeof(uint8_t))) return false;
break;
case MQTT_PROP_TYPE_SHORT:
mg_send_u16(c, mg_htons((uint16_t) props[i].iv));
if (!mg_send_u16(c, mg_htons((uint16_t) props[i].iv))) return false;
break;
case MQTT_PROP_TYPE_INT:
mg_send_u32(c, mg_htonl((uint32_t) props[i].iv));
if (!mg_send_u32(c, mg_htonl((uint32_t) props[i].iv))) return false;
break;
case MQTT_PROP_TYPE_STRING:
mg_send_u16(c, mg_htons((uint16_t) props[i].val.len));
mg_send(c, props[i].val.buf, props[i].val.len);
if (!mg_send_u16(c, mg_htons((uint16_t) props[i].val.len)) ||
!mg_send(c, props[i].val.buf, props[i].val.len))
return false;
break;
case MQTT_PROP_TYPE_BINARY_DATA:
mg_send_u16(c, mg_htons((uint16_t) props[i].val.len));
mg_send(c, props[i].val.buf, props[i].val.len);
if (!mg_send_u16(c, mg_htons((uint16_t) props[i].val.len)) ||
!mg_send(c, props[i].val.buf, props[i].val.len))
return false;
break;
case MQTT_PROP_TYPE_VARIABLE_INT:
len = encode_varint(buf_v, props[i].iv);
mg_send(c, buf_v, (size_t) len);
if (!mg_send(c, buf_v, (size_t) len)) return false;
break;
}
}
return true;
}
size_t mg_mqtt_next_prop(struct mg_mqtt_message *msg, struct mg_mqtt_prop *prop,
@@ -3596,33 +3604,41 @@ void mg_mqtt_login(struct mg_connection *c, const struct mg_mqtt_opts *opts) {
total_len += get_props_size(opts->will_props, opts->num_will_props);
}
mg_mqtt_send_header(c, MQTT_CMD_CONNECT, 0, (uint32_t) total_len);
mg_send(c, hdr, sizeof(hdr));
// keepalive == 0 means "do not disconnect us!"
mg_send_u16(c, mg_htons((uint16_t) opts->keepalive));
if (!mqtt_send_header(c, MQTT_CMD_CONNECT, 0, (uint32_t) total_len) ||
!mg_send(c, hdr, sizeof(hdr)) ||
!mg_send_u16(c, mg_htons((uint16_t) opts->keepalive)))
goto fail;
if (c->is_mqtt5) mg_send_mqtt_properties(c, opts->props, opts->num_props);
if (c->is_mqtt5 && !mg_send_mqtt_properties(c, opts->props, opts->num_props))
goto fail;
mg_send_u16(c, mg_htons((uint16_t) cid.len));
mg_send(c, cid.buf, cid.len);
if (!mg_send_u16(c, mg_htons((uint16_t) cid.len)) ||
!mg_send(c, cid.buf, cid.len))
goto fail;
if (hdr[7] & MQTT_HAS_WILL) {
if (c->is_mqtt5)
mg_send_mqtt_properties(c, opts->will_props, opts->num_will_props);
if (c->is_mqtt5 &&
!mg_send_mqtt_properties(c, opts->will_props, opts->num_will_props))
goto fail;
mg_send_u16(c, mg_htons((uint16_t) opts->topic.len));
mg_send(c, opts->topic.buf, opts->topic.len);
mg_send_u16(c, mg_htons((uint16_t) opts->message.len));
mg_send(c, opts->message.buf, opts->message.len);
}
if (opts->user.len > 0) {
mg_send_u16(c, mg_htons((uint16_t) opts->user.len));
mg_send(c, opts->user.buf, opts->user.len);
}
if (opts->pass.len > 0) {
mg_send_u16(c, mg_htons((uint16_t) opts->pass.len));
mg_send(c, opts->pass.buf, opts->pass.len);
if (!mg_send_u16(c, mg_htons((uint16_t) opts->topic.len)) ||
!mg_send(c, opts->topic.buf, opts->topic.len) ||
!mg_send_u16(c, mg_htons((uint16_t) opts->message.len)) ||
!mg_send(c, opts->message.buf, opts->message.len))
goto fail;
}
if (opts->user.len > 0 &&
(!mg_send_u16(c, mg_htons((uint16_t) opts->user.len)) ||
!mg_send(c, opts->user.buf, opts->user.len)))
goto fail;
if (opts->pass.len > 0 &&
(!mg_send_u16(c, mg_htons((uint16_t) opts->pass.len)) ||
!mg_send(c, opts->pass.buf, opts->pass.len)))
goto fail;
return;
fail:
mg_error(c, "OOM");
}
uint16_t mg_mqtt_pub(struct mg_connection *c, const struct mg_mqtt_opts *opts) {
@@ -3637,20 +3653,28 @@ uint16_t mg_mqtt_pub(struct mg_connection *c, const struct mg_mqtt_opts *opts) {
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 (!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))
goto fail;
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 (!mg_send_u16(c, mg_htons(id))) goto fail;
}
if (c->is_mqtt5) mg_send_mqtt_properties(c, opts->props, opts->num_props);
if (c->is_mqtt5 && !mg_send_mqtt_properties(c, opts->props, opts->num_props))
goto fail;
if (opts->message.len > 0) mg_send(c, opts->message.buf, opts->message.len);
if (opts->message.len > 0 &&
!mg_send(c, opts->message.buf, opts->message.len))
goto fail;
return id;
fail:
mg_error(c, "OOM");
return id;
}
@@ -3659,14 +3683,20 @@ void mg_mqtt_sub(struct mg_connection *c, const struct mg_mqtt_opts *opts) {
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;
mg_mqtt_send_header(c, MQTT_CMD_SUBSCRIBE, 2, (uint32_t) len);
if (!mqtt_send_header(c, MQTT_CMD_SUBSCRIBE, 2, (uint32_t) len)) goto fail;
if (++c->mgr->mqtt_id == 0) ++c->mgr->mqtt_id;
mg_send_u16(c, mg_htons(c->mgr->mqtt_id));
if (c->is_mqtt5) mg_send_mqtt_properties(c, opts->props, opts->num_props);
if (!mg_send_u16(c, mg_htons(c->mgr->mqtt_id))) goto fail;
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_));
if (c->is_mqtt5 && !mg_send_mqtt_properties(c, opts->props, opts->num_props))
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_)))
goto fail;
return;
fail:
mg_error(c, "OOM");
}
int mg_mqtt_parse(const uint8_t *buf, size_t len, uint8_t version,
@@ -3773,15 +3803,16 @@ static void mqtt_cb(struct mg_connection *c, int ev, void *ev_data) {
uint32_t remaining_len = sizeof(id);
if (c->is_mqtt5) remaining_len += 2; // 3.4.2
mg_mqtt_send_header(
c,
(uint8_t) (mm.qos == 2 ? MQTT_CMD_PUBREC : MQTT_CMD_PUBACK),
0, remaining_len);
mg_send(c, &id, sizeof(id));
if (!mqtt_send_header(c,
(uint8_t) (mm.qos == 2 ? MQTT_CMD_PUBREC
: MQTT_CMD_PUBACK),
0, remaining_len) ||
!mg_send(c, &id, sizeof(id)))
goto fail;
if (c->is_mqtt5) {
uint16_t zero = 0;
mg_send(c, &zero, sizeof(zero));
if (!mg_send(c, &zero, sizeof(zero))) goto fail;
}
}
mg_call(c, MG_EV_MQTT_MSG, &mm); // let the app handle qos stuff
@@ -3790,15 +3821,18 @@ static void mqtt_cb(struct mg_connection *c, int ev, void *ev_data) {
case MQTT_CMD_PUBREC: { // MQTT5: 3.5.2-1 TODO(): variable header rc
uint16_t id = mg_ntohs(mm.id);
uint32_t remaining_len = sizeof(id); // MQTT5 3.6.2-1
mg_mqtt_send_header(c, MQTT_CMD_PUBREL, 2, remaining_len);
mg_send(c, &id, sizeof(id)); // MQTT5 3.6.1-1, flags = 2
if (!mqtt_send_header(c, MQTT_CMD_PUBREL, 2,
remaining_len) // MQTT5 3.6.1-1, flags = 2
|| !mg_send(c, &id, sizeof(id)))
goto fail;
break;
}
case MQTT_CMD_PUBREL: { // MQTT5: 3.6.2-1 TODO(): variable header rc
uint16_t id = mg_ntohs(mm.id);
uint32_t remaining_len = sizeof(id); // MQTT5 3.7.2-1
mg_mqtt_send_header(c, MQTT_CMD_PUBCOMP, 0, remaining_len);
mg_send(c, &id, sizeof(id));
if (!mqtt_send_header(c, MQTT_CMD_PUBCOMP, 0, remaining_len) ||
!mg_send(c, &id, sizeof(id)))
goto fail;
break;
}
}
@@ -3810,6 +3844,9 @@ static void mqtt_cb(struct mg_connection *c, int ev, void *ev_data) {
}
}
(void) ev_data;
return;
fail:
mg_error(c, "OOM");
}
void mg_mqtt_ping(struct mg_connection *nc) {
@@ -3824,19 +3861,24 @@ void mg_mqtt_disconnect(struct mg_connection *c,
const struct mg_mqtt_opts *opts) {
size_t len = 0;
if (c->is_mqtt5) len = 1 + get_props_size(opts->props, opts->num_props);
mg_mqtt_send_header(c, MQTT_CMD_DISCONNECT, 0, (uint32_t) len);
if (!mqtt_send_header(c, MQTT_CMD_DISCONNECT, 0, (uint32_t) len)) goto fail;
if (c->is_mqtt5) {
uint8_t zero = 0;
mg_send(c, &zero, sizeof(zero)); // reason code
mg_send_mqtt_properties(c, opts->props, opts->num_props);
if (!mg_send(c, &zero, sizeof(zero)) // reason code
|| !mg_send_mqtt_properties(c, opts->props, opts->num_props))
goto fail;
}
return;
fail:
mg_error(c, "OOM");
}
struct mg_connection *mg_mqtt_connect(struct mg_mgr *mgr, const char *url,
const struct mg_mqtt_opts *opts,
mg_event_handler_t fn, void *fn_data) {
struct mg_connection *c = mg_connect_svc(mgr, url, fn, fn_data, mqtt_cb, NULL);
struct mg_connection *c =
mg_connect_svc(mgr, url, fn, fn_data, mqtt_cb, NULL);
if (c != NULL) {
struct mg_mqtt_opts empty;
memset(&empty, 0, sizeof(empty));
@@ -6088,6 +6130,7 @@ bool mg_open_listener(struct mg_connection *c, const char *url) {
static void write_conn(struct mg_connection *c) {
long len = c->is_tls ? mg_tls_send(c, c->send.buf, c->send.len)
: mg_io_send(c, c->send.buf, c->send.len);
// TODO(): mg_tls_send() may return 0 forever on steady OOM
if (len == MG_IO_ERR) {
mg_error(c, "tx err");
} else if (len > 0) {
@@ -6100,14 +6143,8 @@ static void init_closure(struct mg_connection *c) {
struct connstate *s = (struct connstate *) (c + 1);
if (c->is_udp == false && c->is_listening == false &&
c->is_connecting == false) { // For TCP conns,
<<<<<<< HEAD
tx_tcp(c->mgr->ifp, s->mac, &c->loc, &c->rem, TH_FIN | TH_ACK,
mg_htonl(s->seq), mg_htonl(s->ack), NULL, 0);
=======
uint32_t rem_ip = c->rem.ip4;
tx_tcp(c->mgr->ifp, s->mac, rem_ip, TH_FIN | TH_ACK, c->loc.port,
c->rem.port, mg_htonl(s->seq), mg_htonl(s->ack), NULL, 0);
>>>>>>> 81b87991 (wip)
settmout(c, MIP_TTYPE_FIN);
}
}
@@ -6173,7 +6210,7 @@ bool mg_send(struct mg_connection *c, const void *buf, size_t len) {
len = trim_len(c, len); // Trimming length if necessary
res = tx_udp(ifp, s->mac, &c->loc, &c->rem, buf, len);
} else {
res = mg_iobuf_add(&c->send, c->send.len, buf, len);
res = (bool) mg_iobuf_add(&c->send, c->send.len, buf, len);
// res == 0 means an OOM condition (iobuf couldn't resize), yet this is so
// far recoverable, let the caller decide
}
@@ -9530,7 +9567,7 @@ bool mg_send(struct mg_connection *c, const void *buf, size_t len) {
iolog(c, (char *) buf, n, false);
return n > 0;
} else {
return mg_iobuf_add(&c->send, c->send.len, buf, len);
return (bool) mg_iobuf_add(&c->send, c->send.len, buf, len);
// returning 0 means an OOM condition (iobuf couldn't resize), yet this is
// so far recoverable, let the caller decide
}
@@ -9730,6 +9767,7 @@ static void write_conn(struct mg_connection *c) {
char *buf = (char *) c->send.buf;
size_t len = c->send.len;
long n = c->is_tls ? mg_tls_send(c, buf, len) : mg_io_send(c, buf, len);
// TODO(): mg_tls_send() may return 0 forever on steady OOM
MG_DEBUG(("%lu %ld snd %ld/%ld rcv %ld/%ld n=%ld err=%d", c->id, c->fd,
(long) c->send.len, (long) c->send.size, (long) c->recv.len,
(long) c->recv.size, n, MG_SOCK_ERR(n)));
@@ -20932,7 +20970,7 @@ static void ws_handshake(struct mg_connection *c, const struct mg_str *wskey,
mg_printf(c, "Sec-WebSocket-Protocol: %.*s\r\n", (int) wsproto->len,
wsproto->buf);
}
mg_send(c, "\r\n", 2);
if (!mg_send(c, "\r\n", 2)) mg_error(c, "OOM");
}
static uint32_t be32(const uint8_t *p) {
src/http.c
+3 -4
View File
@@ -346,14 +346,14 @@ int mg_http_parse(const char *s, size_t len, struct mg_http_message *hm) {
static void mg_http_vprintf_chunk(struct mg_connection *c, const char *fmt,
va_list *ap) {
size_t len = c->send.len;
mg_send(c, " \r\n", 10);
if (!mg_send(c, " \r\n", 10)) mg_error(c, "OOM");
mg_vxprintf(mg_pfn_iobuf, &c->send, fmt, ap);
if (c->send.len >= len + 10) {
mg_snprintf((char *) c->send.buf + len, 9, "%08lx", c->send.len - len - 10);
c->send.buf[len + 8] = '\r';
if (c->send.len == len + 10) c->is_resp = 0; // Last chunk, reset marker
}
mg_send(c, "\r\n", 2);
if (!mg_send(c, "\r\n", 2)) mg_error(c, "OOM");
}
void mg_http_printf_chunk(struct mg_connection *c, const char *fmt, ...) {
@@ -365,8 +365,7 @@ void mg_http_printf_chunk(struct mg_connection *c, const char *fmt, ...) {
void mg_http_write_chunk(struct mg_connection *c, const char *buf, size_t len) {
mg_printf(c, "%lx\r\n", (unsigned long) len);
mg_send(c, buf, len);
mg_send(c, "\r\n", 2);
if (!mg_send(c, buf, len) || !mg_send(c, "\r\n", 2)) mg_error(c, "OOM");
if (len == 0) c->is_resp = 0;
}
src/mqtt.c
+111 -68
View File
@@ -46,8 +46,8 @@ static const struct mg_mqtt_pmap s_prop_map[] = {
{MQTT_PROP_SUBSCRIPTION_IDENTIFIER_AVAILABLE, MQTT_PROP_TYPE_BYTE},
{MQTT_PROP_SHARED_SUBSCRIPTION_AVAILABLE, MQTT_PROP_TYPE_BYTE}};
void mg_mqtt_send_header(struct mg_connection *c, uint8_t cmd, uint8_t flags,
uint32_t len) {
static bool mqtt_send_header(struct mg_connection *c, uint8_t cmd,
uint8_t flags, uint32_t len) {
uint8_t buf[1 + sizeof(len)], *vlen = &buf[1];
buf[0] = (uint8_t) ((cmd << 4) | flags);
do {
@@ -56,15 +56,20 @@ void mg_mqtt_send_header(struct mg_connection *c, uint8_t cmd, uint8_t flags,
if (len > 0) *vlen |= 0x80;
vlen++;
} while (len > 0 && vlen < &buf[sizeof(buf)]);
mg_send(c, buf, (size_t) (vlen - buf));
return mg_send(c, buf, (size_t) (vlen - buf));
}
static void mg_send_u16(struct mg_connection *c, uint16_t value) {
mg_send(c, &value, sizeof(value));
void mg_mqtt_send_header(struct mg_connection *c, uint8_t cmd, uint8_t flags,
uint32_t len) {
if (!mqtt_send_header(c, cmd, flags, len)) mg_error(c, "OOM");
}
static void mg_send_u32(struct mg_connection *c, uint32_t value) {
mg_send(c, &value, sizeof(value));
static bool mg_send_u16(struct mg_connection *c, uint16_t value) {
return mg_send(c, &value, sizeof(value));
}
static bool mg_send_u32(struct mg_connection *c, uint32_t value) {
return mg_send(c, &value, sizeof(value));
}
static uint8_t varint_size(size_t length) {
@@ -157,46 +162,50 @@ static size_t get_props_size(struct mg_mqtt_prop *props, size_t count) {
return size;
}
static void mg_send_mqtt_properties(struct mg_connection *c,
static bool mg_send_mqtt_properties(struct mg_connection *c,
struct mg_mqtt_prop *props, size_t nprops) {
size_t total_size = get_properties_length(props, nprops);
uint8_t buf_v[4] = {0, 0, 0, 0};
uint8_t buf[4] = {0, 0, 0, 0};
size_t i, len = encode_varint(buf, total_size);
mg_send(c, buf, (size_t) len);
if (!mg_send(c, buf, (size_t) len)) return false;
for (i = 0; i < nprops; i++) {
mg_send(c, &props[i].id, sizeof(props[i].id));
if (!mg_send(c, &props[i].id, sizeof(props[i].id))) return false;
switch (mqtt_prop_type_by_id(props[i].id)) {
case MQTT_PROP_TYPE_STRING_PAIR:
mg_send_u16(c, mg_htons((uint16_t) props[i].key.len));
mg_send(c, props[i].key.buf, props[i].key.len);
mg_send_u16(c, mg_htons((uint16_t) props[i].val.len));
mg_send(c, props[i].val.buf, props[i].val.len);
if (!mg_send_u16(c, mg_htons((uint16_t) props[i].key.len)) ||
!mg_send(c, props[i].key.buf, props[i].key.len) ||
!mg_send_u16(c, mg_htons((uint16_t) props[i].val.len)) ||
!mg_send(c, props[i].val.buf, props[i].val.len))
return false;
break;
case MQTT_PROP_TYPE_BYTE:
mg_send(c, &props[i].iv, sizeof(uint8_t));
if (!mg_send(c, &props[i].iv, sizeof(uint8_t))) return false;
break;
case MQTT_PROP_TYPE_SHORT:
mg_send_u16(c, mg_htons((uint16_t) props[i].iv));
if (!mg_send_u16(c, mg_htons((uint16_t) props[i].iv))) return false;
break;
case MQTT_PROP_TYPE_INT:
mg_send_u32(c, mg_htonl((uint32_t) props[i].iv));
if (!mg_send_u32(c, mg_htonl((uint32_t) props[i].iv))) return false;
break;
case MQTT_PROP_TYPE_STRING:
mg_send_u16(c, mg_htons((uint16_t) props[i].val.len));
mg_send(c, props[i].val.buf, props[i].val.len);
if (!mg_send_u16(c, mg_htons((uint16_t) props[i].val.len)) ||
!mg_send(c, props[i].val.buf, props[i].val.len))
return false;
break;
case MQTT_PROP_TYPE_BINARY_DATA:
mg_send_u16(c, mg_htons((uint16_t) props[i].val.len));
mg_send(c, props[i].val.buf, props[i].val.len);
if (!mg_send_u16(c, mg_htons((uint16_t) props[i].val.len)) ||
!mg_send(c, props[i].val.buf, props[i].val.len))
return false;
break;
case MQTT_PROP_TYPE_VARIABLE_INT:
len = encode_varint(buf_v, props[i].iv);
mg_send(c, buf_v, (size_t) len);
if (!mg_send(c, buf_v, (size_t) len)) return false;
break;
}
}
return true;
}
size_t mg_mqtt_next_prop(struct mg_mqtt_message *msg, struct mg_mqtt_prop *prop,
@@ -289,33 +298,41 @@ void mg_mqtt_login(struct mg_connection *c, const struct mg_mqtt_opts *opts) {
total_len += get_props_size(opts->will_props, opts->num_will_props);
}
mg_mqtt_send_header(c, MQTT_CMD_CONNECT, 0, (uint32_t) total_len);
mg_send(c, hdr, sizeof(hdr));
// keepalive == 0 means "do not disconnect us!"
mg_send_u16(c, mg_htons((uint16_t) opts->keepalive));
if (!mqtt_send_header(c, MQTT_CMD_CONNECT, 0, (uint32_t) total_len) ||
!mg_send(c, hdr, sizeof(hdr)) ||
!mg_send_u16(c, mg_htons((uint16_t) opts->keepalive)))
goto fail;
if (c->is_mqtt5) mg_send_mqtt_properties(c, opts->props, opts->num_props);
if (c->is_mqtt5 && !mg_send_mqtt_properties(c, opts->props, opts->num_props))
goto fail;
mg_send_u16(c, mg_htons((uint16_t) cid.len));
mg_send(c, cid.buf, cid.len);
if (!mg_send_u16(c, mg_htons((uint16_t) cid.len)) ||
!mg_send(c, cid.buf, cid.len))
goto fail;
if (hdr[7] & MQTT_HAS_WILL) {
if (c->is_mqtt5)
mg_send_mqtt_properties(c, opts->will_props, opts->num_will_props);
if (c->is_mqtt5 &&
!mg_send_mqtt_properties(c, opts->will_props, opts->num_will_props))
goto fail;
mg_send_u16(c, mg_htons((uint16_t) opts->topic.len));
mg_send(c, opts->topic.buf, opts->topic.len);
mg_send_u16(c, mg_htons((uint16_t) opts->message.len));
mg_send(c, opts->message.buf, opts->message.len);
}
if (opts->user.len > 0) {
mg_send_u16(c, mg_htons((uint16_t) opts->user.len));
mg_send(c, opts->user.buf, opts->user.len);
}
if (opts->pass.len > 0) {
mg_send_u16(c, mg_htons((uint16_t) opts->pass.len));
mg_send(c, opts->pass.buf, opts->pass.len);
if (!mg_send_u16(c, mg_htons((uint16_t) opts->topic.len)) ||
!mg_send(c, opts->topic.buf, opts->topic.len) ||
!mg_send_u16(c, mg_htons((uint16_t) opts->message.len)) ||
!mg_send(c, opts->message.buf, opts->message.len))
goto fail;
}
if (opts->user.len > 0 &&
(!mg_send_u16(c, mg_htons((uint16_t) opts->user.len)) ||
!mg_send(c, opts->user.buf, opts->user.len)))
goto fail;
if (opts->pass.len > 0 &&
(!mg_send_u16(c, mg_htons((uint16_t) opts->pass.len)) ||
!mg_send(c, opts->pass.buf, opts->pass.len)))
goto fail;
return;
fail:
mg_error(c, "OOM");
}
uint16_t mg_mqtt_pub(struct mg_connection *c, const struct mg_mqtt_opts *opts) {
@@ -330,20 +347,28 @@ uint16_t mg_mqtt_pub(struct mg_connection *c, const struct mg_mqtt_opts *opts) {
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 (!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))
goto fail;
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 (!mg_send_u16(c, mg_htons(id))) goto fail;
}
if (c->is_mqtt5) mg_send_mqtt_properties(c, opts->props, opts->num_props);
if (c->is_mqtt5 && !mg_send_mqtt_properties(c, opts->props, opts->num_props))
goto fail;
if (opts->message.len > 0) mg_send(c, opts->message.buf, opts->message.len);
if (opts->message.len > 0 &&
!mg_send(c, opts->message.buf, opts->message.len))
goto fail;
return id;
fail:
mg_error(c, "OOM");
return id;
}
@@ -352,14 +377,20 @@ void mg_mqtt_sub(struct mg_connection *c, const struct mg_mqtt_opts *opts) {
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;
mg_mqtt_send_header(c, MQTT_CMD_SUBSCRIBE, 2, (uint32_t) len);
if (!mqtt_send_header(c, MQTT_CMD_SUBSCRIBE, 2, (uint32_t) len)) goto fail;
if (++c->mgr->mqtt_id == 0) ++c->mgr->mqtt_id;
mg_send_u16(c, mg_htons(c->mgr->mqtt_id));
if (c->is_mqtt5) mg_send_mqtt_properties(c, opts->props, opts->num_props);
if (!mg_send_u16(c, mg_htons(c->mgr->mqtt_id))) goto fail;
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_));
if (c->is_mqtt5 && !mg_send_mqtt_properties(c, opts->props, opts->num_props))
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_)))
goto fail;
return;
fail:
mg_error(c, "OOM");
}
int mg_mqtt_parse(const uint8_t *buf, size_t len, uint8_t version,
@@ -466,15 +497,16 @@ static void mqtt_cb(struct mg_connection *c, int ev, void *ev_data) {
uint32_t remaining_len = sizeof(id);
if (c->is_mqtt5) remaining_len += 2; // 3.4.2
mg_mqtt_send_header(
c,
(uint8_t) (mm.qos == 2 ? MQTT_CMD_PUBREC : MQTT_CMD_PUBACK),
0, remaining_len);
mg_send(c, &id, sizeof(id));
if (!mqtt_send_header(c,
(uint8_t) (mm.qos == 2 ? MQTT_CMD_PUBREC
: MQTT_CMD_PUBACK),
0, remaining_len) ||
!mg_send(c, &id, sizeof(id)))
goto fail;
if (c->is_mqtt5) {
uint16_t zero = 0;
mg_send(c, &zero, sizeof(zero));
if (!mg_send(c, &zero, sizeof(zero))) goto fail;
}
}
mg_call(c, MG_EV_MQTT_MSG, &mm); // let the app handle qos stuff
@@ -483,15 +515,18 @@ static void mqtt_cb(struct mg_connection *c, int ev, void *ev_data) {
case MQTT_CMD_PUBREC: { // MQTT5: 3.5.2-1 TODO(): variable header rc
uint16_t id = mg_ntohs(mm.id);
uint32_t remaining_len = sizeof(id); // MQTT5 3.6.2-1
mg_mqtt_send_header(c, MQTT_CMD_PUBREL, 2, remaining_len);
mg_send(c, &id, sizeof(id)); // MQTT5 3.6.1-1, flags = 2
if (!mqtt_send_header(c, MQTT_CMD_PUBREL, 2,
remaining_len) // MQTT5 3.6.1-1, flags = 2
|| !mg_send(c, &id, sizeof(id)))
goto fail;
break;
}
case MQTT_CMD_PUBREL: { // MQTT5: 3.6.2-1 TODO(): variable header rc
uint16_t id = mg_ntohs(mm.id);
uint32_t remaining_len = sizeof(id); // MQTT5 3.7.2-1
mg_mqtt_send_header(c, MQTT_CMD_PUBCOMP, 0, remaining_len);
mg_send(c, &id, sizeof(id));
if (!mqtt_send_header(c, MQTT_CMD_PUBCOMP, 0, remaining_len) ||
!mg_send(c, &id, sizeof(id)))
goto fail;
break;
}
}
@@ -503,6 +538,9 @@ static void mqtt_cb(struct mg_connection *c, int ev, void *ev_data) {
}
}
(void) ev_data;
return;
fail:
mg_error(c, "OOM");
}
void mg_mqtt_ping(struct mg_connection *nc) {
@@ -517,19 +555,24 @@ void mg_mqtt_disconnect(struct mg_connection *c,
const struct mg_mqtt_opts *opts) {
size_t len = 0;
if (c->is_mqtt5) len = 1 + get_props_size(opts->props, opts->num_props);
mg_mqtt_send_header(c, MQTT_CMD_DISCONNECT, 0, (uint32_t) len);
if (!mqtt_send_header(c, MQTT_CMD_DISCONNECT, 0, (uint32_t) len)) goto fail;
if (c->is_mqtt5) {
uint8_t zero = 0;
mg_send(c, &zero, sizeof(zero)); // reason code
mg_send_mqtt_properties(c, opts->props, opts->num_props);
if (!mg_send(c, &zero, sizeof(zero)) // reason code
|| !mg_send_mqtt_properties(c, opts->props, opts->num_props))
goto fail;
}
return;
fail:
mg_error(c, "OOM");
}
struct mg_connection *mg_mqtt_connect(struct mg_mgr *mgr, const char *url,
const struct mg_mqtt_opts *opts,
mg_event_handler_t fn, void *fn_data) {
struct mg_connection *c = mg_connect_svc(mgr, url, fn, fn_data, mqtt_cb, NULL);
struct mg_connection *c =
mg_connect_svc(mgr, url, fn, fn_data, mqtt_cb, NULL);
if (c != NULL) {
struct mg_mqtt_opts empty;
memset(&empty, 0, sizeof(empty));
src/net_builtin.c
+2 -7
View File
@@ -1924,6 +1924,7 @@ bool mg_open_listener(struct mg_connection *c, const char *url) {
static void write_conn(struct mg_connection *c) {
long len = c->is_tls ? mg_tls_send(c, c->send.buf, c->send.len)
: mg_io_send(c, c->send.buf, c->send.len);
// TODO(): mg_tls_send() may return 0 forever on steady OOM
if (len == MG_IO_ERR) {
mg_error(c, "tx err");
} else if (len > 0) {
@@ -1936,14 +1937,8 @@ static void init_closure(struct mg_connection *c) {
struct connstate *s = (struct connstate *) (c + 1);
if (c->is_udp == false && c->is_listening == false &&
c->is_connecting == false) { // For TCP conns,
<<<<<<< HEAD
tx_tcp(c->mgr->ifp, s->mac, &c->loc, &c->rem, TH_FIN | TH_ACK,
mg_htonl(s->seq), mg_htonl(s->ack), NULL, 0);
=======
uint32_t rem_ip = c->rem.ip4;
tx_tcp(c->mgr->ifp, s->mac, rem_ip, TH_FIN | TH_ACK, c->loc.port,
c->rem.port, mg_htonl(s->seq), mg_htonl(s->ack), NULL, 0);
>>>>>>> 81b87991 (wip)
settmout(c, MIP_TTYPE_FIN);
}
}
@@ -2009,7 +2004,7 @@ bool mg_send(struct mg_connection *c, const void *buf, size_t len) {
len = trim_len(c, len); // Trimming length if necessary
res = tx_udp(ifp, s->mac, &c->loc, &c->rem, buf, len);
} else {
res = mg_iobuf_add(&c->send, c->send.len, buf, len);
res = (bool) mg_iobuf_add(&c->send, c->send.len, buf, len);
// res == 0 means an OOM condition (iobuf couldn't resize), yet this is so
// far recoverable, let the caller decide
}
src/sock.c
+2 -1
View File
@@ -141,7 +141,7 @@ bool mg_send(struct mg_connection *c, const void *buf, size_t len) {
iolog(c, (char *) buf, n, false);
return n > 0;
} else {
return mg_iobuf_add(&c->send, c->send.len, buf, len);
return (bool) mg_iobuf_add(&c->send, c->send.len, buf, len);
// returning 0 means an OOM condition (iobuf couldn't resize), yet this is
// so far recoverable, let the caller decide
}
@@ -341,6 +341,7 @@ static void write_conn(struct mg_connection *c) {
char *buf = (char *) c->send.buf;
size_t len = c->send.len;
long n = c->is_tls ? mg_tls_send(c, buf, len) : mg_io_send(c, buf, len);
// TODO(): mg_tls_send() may return 0 forever on steady OOM
MG_DEBUG(("%lu %ld snd %ld/%ld rcv %ld/%ld n=%ld err=%d", c->id, c->fd,
(long) c->send.len, (long) c->send.size, (long) c->recv.len,
(long) c->recv.size, n, MG_SOCK_ERR(n)));
src/ws.c
+1 -1
View File
@@ -55,7 +55,7 @@ static void ws_handshake(struct mg_connection *c, const struct mg_str *wskey,
mg_printf(c, "Sec-WebSocket-Protocol: %.*s\r\n", (int) wsproto->len,
wsproto->buf);
}
mg_send(c, "\r\n", 2);
if (!mg_send(c, "\r\n", 2)) mg_error(c, "OOM");
}
static uint32_t be32(const uint8_t *p) {