Free ACLK message (#20125)
Stelios Fragkakis committed
Apr 15, 2025 at 20:54 UTC
f7ebf886a67c4276cab09d71cd5df0a3f368a4eb
3 files changed
+35
-24
src/aclk/aclk_tx_msgs.c
+1
-1
@@ -33,7 +33,7 @@ uint16_t aclk_send_bin_message_subtopic_pid(mqtt_wss_client client, char *msg, s
33
return 0;
34
}
35
36
- mqtt_wss_publish5(client, (char*)topic, NULL, msg, &freez_aclk_publish5a, msg_len, MQTT_WSS_PUB_QOS1, &packet_id);
36
+ mqtt_wss_publish5(client, (char *)topic, NULL, msg, &freez_aclk_publish5a, msg_len, MQTT_WSS_PUB_QOS1, &packet_id);
37
38
if (aclklog_enabled) {
39
char *json = protomsg_to_json(msg, msg_len, msgname);
src/aclk/mqtt_websockets/mqtt_ng.c
+30
-23
@@ -780,27 +780,32 @@ static int _optimized_add(struct header_buffer *buf, void *data, size_t data_len
780
return 0;
781
}
782
783
-#define TRY_GENERATE_MESSAGE(generator_function, ...) \
784
- int rc = generator_function(&client->main_buffer, ##__VA_ARGS__); \
785
- if (rc == MQTT_NG_MSGGEN_BUFFER_OOM) { \
786
- LOCK_HDR_BUFFER(&client->main_buffer); \
787
- transaction_buffer_garbage_collect((&client->main_buffer)); \
788
- UNLOCK_HDR_BUFFER(&client->main_buffer); \
789
- rc = generator_function(&client->main_buffer, ##__VA_ARGS__); \
790
- if (rc == MQTT_NG_MSGGEN_BUFFER_OOM && client->max_mem_bytes) { \
791
- LOCK_HDR_BUFFER(&client->main_buffer); \
792
- transaction_buffer_grow((&client->main_buffer),GROWTH_FACTOR, client->max_mem_bytes); \
793
- UNLOCK_HDR_BUFFER(&client->main_buffer); \
794
- rc = generator_function(&client->main_buffer, ##__VA_ARGS__); \
795
- } \
796
- if (rc == MQTT_NG_MSGGEN_BUFFER_OOM) \
797
- nd_log(NDLS_DAEMON, NDLP_ERR, "%s failed to generate message due to insufficient buffer space (line %d)", __FUNCTION__, __LINE__); \
798
- } \
799
- if (rc == MQTT_NG_MSGGEN_OK) { \
800
- spinlock_lock(&client->stats_spinlock); \
801
- client->stats.tx_messages_queued++; \
802
- spinlock_unlock(&client->stats_spinlock); \
803
- } \
783
+#define TRY_GENERATE_MESSAGE(generator_function, ...) \
784
+ int rc = generator_function(&client->main_buffer, ##__VA_ARGS__); \
785
+ if (rc == MQTT_NG_MSGGEN_BUFFER_OOM) { \
786
+ LOCK_HDR_BUFFER(&client->main_buffer); \
787
+ transaction_buffer_garbage_collect((&client->main_buffer)); \
788
+ UNLOCK_HDR_BUFFER(&client->main_buffer); \
789
+ rc = generator_function(&client->main_buffer, ##__VA_ARGS__); \
790
+ if (rc == MQTT_NG_MSGGEN_BUFFER_OOM && client->max_mem_bytes) { \
791
+ LOCK_HDR_BUFFER(&client->main_buffer); \
792
+ transaction_buffer_grow((&client->main_buffer), GROWTH_FACTOR, client->max_mem_bytes); \
793
+ UNLOCK_HDR_BUFFER(&client->main_buffer); \
794
+ rc = generator_function(&client->main_buffer, ##__VA_ARGS__); \
795
+ } \
796
+ if (rc == MQTT_NG_MSGGEN_BUFFER_OOM) \
797
+ nd_log( \
798
+ NDLS_DAEMON, \
799
+ NDLP_ERR, \
800
+ "%s failed to generate message due to insufficient buffer space (line %d)", \
801
+ __FUNCTION__, \
802
+ __LINE__); \
803
+ } \
804
+ if (rc == MQTT_NG_MSGGEN_OK) { \
805
+ spinlock_lock(&client->stats_spinlock); \
806
+ client->stats.tx_messages_queued++; \
807
+ spinlock_unlock(&client->stats_spinlock); \
808
+ } \
809
return rc;
810
811
mqtt_msg_data mqtt_ng_generate_connect(struct transaction_buffer *trx_buf,
@@ -1123,6 +1128,8 @@ int mqtt_ng_publish(struct mqtt_ng_client *client,
1128
1129
if (client->max_msg_size && PUBLISH_SP_SIZE + mqtt_ng_publish_size(topic, msg_len, topic_id) > client->max_msg_size) {
1130
nd_log(NDLS_DAEMON, NDLP_ERR, "Message too big for server: %zu", msg_len);
1131
+ if (msg_free)
1132
+ msg_free(msg);
1133
return MQTT_NG_MSGGEN_MSG_TOO_BIG;
1134
}
1135
@@ -1298,8 +1305,8 @@ int mqtt_ng_ping(struct mqtt_ng_client *client)
1305
#define MQTT_NG_CLIENT_NOT_IMPL_YET -3
1306
#define MQTT_NG_CLIENT_INTERNAL_ERROR -5
1307
1301
-#define BUF_READ_CHECK_AT_LEAST(buf, x) \
1302
- if (rbuf_bytes_available(buf) < (x)) \
1308
+#define BUF_READ_CHECK_AT_LEAST(buf, x) \
1309
+ if (rbuf_bytes_available(buf) < (x)) \
1310
return MQTT_NG_CLIENT_NEED_MORE_BYTES;
1311
1312
#define vbi_parser_reset_ctx(ctx) memset(ctx, 0, sizeof(struct mqtt_vbi_parser_ctx))
src/aclk/mqtt_websockets/mqtt_wss_client.c
+4
@@ -982,11 +982,15 @@ int mqtt_wss_publish5(mqtt_wss_client client,
982
{
983
if (client->mqtt_disconnecting) {
984
nd_log(NDLS_DAEMON, NDLP_ERR, "mqtt_wss is disconnecting can't publish");
985
+ if (msg_free)
986
+ msg_free(msg);
987
return 1;
988
}
989
990
if (!client->mqtt_connected) {
991
nd_log(NDLS_DAEMON, NDLP_ERR, "MQTT is offline. Can't send message.");
992
+ if (msg_free)
993
+ msg_free(msg);
994
return 1;
995
}
996
uint8_t mqtt_flags = 0;