Honor maximum message size limit of MQTT server (#15009)
Timotej S committed
May 18, 2023 at 13:56 UTC
5827732aa706db68bf50cdeb1b53fecc3b14f2f9
4 files changed
+11
-7
aclk/aclk_query.c
+1
-1
@@ -216,7 +216,7 @@ static int http_api_v2(struct aclk_query_thread *query_thr, aclk_query_t query)
216
}
217
218
// send msg.
219
- aclk_http_msg_v2(query_thr->client, query->callback_topic, query->msg_id, t, query->created, w->response.code, local_buffer->buffer, local_buffer->len);
219
+ w->response.code = aclk_http_msg_v2(query_thr->client, query->callback_topic, query->msg_id, t, query->created, w->response.code, local_buffer->buffer, local_buffer->len);
220
221
struct timeval tv;
222
aclk/aclk_tx_msgs.c
+8
-4
@@ -83,7 +83,10 @@ static int aclk_send_message_with_bin_payload(mqtt_wss_client client, json_objec
83
memcpy(&full_msg[len], payload, payload_len);
84
}
85
86
- mqtt_wss_publish5(client, (char*)topic, NULL, full_msg, &freez_aclk_publish5b, full_msg_len, MQTT_WSS_PUB_QOS1, &packet_id);
86
+ int rc = mqtt_wss_publish5(client, (char*)topic, NULL, full_msg, &freez_aclk_publish5b, full_msg_len, MQTT_WSS_PUB_QOS1, &packet_id);
87
+
88
+ if (rc == MQTT_WSS_ERR_TOO_BIG_FOR_SERVER)
89
+ return HTTP_RESP_FORBIDDEN;
90
91
#ifdef NETDATA_INTERNAL_CHECKS
92
aclk_stats_msg_published(packet_id);
@@ -169,11 +172,11 @@ void aclk_http_msg_v2_err(mqtt_wss_client client, const char *topic, const char
172
json_object_object_add(msg, "error-description", tmp);
173
174
if (aclk_send_message_with_bin_payload(client, msg, topic, payload, payload_len)) {
172
- error("Failed to send cancelation message for http reply");
175
+ error("Failed to send cancellation message for http reply %zu %s", payload_len, payload);
176
}
177
}
178
176
-void aclk_http_msg_v2(mqtt_wss_client client, const char *topic, const char *msg_id, usec_t t_exec, usec_t created, int http_code, const char *payload, size_t payload_len)
179
+int aclk_http_msg_v2(mqtt_wss_client client, const char *topic, const char *msg_id, usec_t t_exec, usec_t created, int http_code, const char *payload, size_t payload_len)
180
{
181
json_object *tmp, *msg;
182
@@ -192,7 +195,7 @@ void aclk_http_msg_v2(mqtt_wss_client client, const char *topic, const char *msg
195
196
switch (rc) {
197
case HTTP_RESP_FORBIDDEN:
195
- aclk_http_msg_v2_err(client, topic, msg_id, rc, CLOUD_EC_REQ_REPLY_TOO_BIG, CLOUD_EMSG_REQ_REPLY_TOO_BIG, payload, payload_len);
198
+ aclk_http_msg_v2_err(client, topic, msg_id, rc, CLOUD_EC_REQ_REPLY_TOO_BIG, CLOUD_EMSG_REQ_REPLY_TOO_BIG, NULL, 0);
199
break;
200
case HTTP_RESP_INTERNAL_SERVER_ERROR:
201
aclk_http_msg_v2_err(client, topic, msg_id, rc, CLOUD_EC_FAIL_TOPIC, CLOUD_EMSG_FAIL_TOPIC, payload, payload_len);
@@ -201,6 +204,7 @@ void aclk_http_msg_v2(mqtt_wss_client client, const char *topic, const char *msg
204
aclk_http_msg_v2_err(client, topic, msg_id, rc, CLOUD_EC_SND_TIMEOUT, CLOUD_EMSG_SND_TIMEOUT, payload, payload_len);
205
break;
206
}
207
+ return rc ? rc : http_code;
208
}
209
210
uint16_t aclk_send_agent_connection_update(mqtt_wss_client client, int reachable) {
aclk/aclk_tx_msgs.h
+1
-1
@@ -12,7 +12,7 @@
12
uint16_t aclk_send_bin_message_subtopic_pid(mqtt_wss_client client, char *msg, size_t msg_len, enum aclk_topics subtopic, const char *msgname);
13
14
void aclk_http_msg_v2_err(mqtt_wss_client client, const char *topic, const char *msg_id, int http_code, int ec, const char* emsg, const char *payload, size_t payload_len);
15
-void aclk_http_msg_v2(mqtt_wss_client client, const char *topic, const char *msg_id, usec_t t_exec, usec_t created, int http_code, const char *payload, size_t payload_len);
15
+int aclk_http_msg_v2(mqtt_wss_client client, const char *topic, const char *msg_id, usec_t t_exec, usec_t created, int http_code, const char *payload, size_t payload_len);
16
17
uint16_t aclk_send_agent_connection_update(mqtt_wss_client client, int reachable);
18
char *aclk_generate_lwt(size_t *size);
mqtt_websockets
+1
-1
@@ -1 +1 @@
1
-Subproject commit 684fa17afb14ef5a46dde18802548615fdd441ac
1
+Subproject commit cb553fd1a6929b107f36ba0c320575449b9752b6