mqtt_ng.c · cryptotaxi247/netdata-1

Code Commits Issues Pulls Sessions CI/CD

master

c 2,405 lines 93.8 KB

Raw

1	// SPDX-License-Identifier: GPL-3.0-or-later
2
3	#ifndef _GNU_SOURCE
4	#define _GNU_SOURCE
5	#endif
6
7	#include "libnetdata/libnetdata.h"
8	void pulse_aclk_sent_message_acked(usec_t publish_latency, size_t len);
9
10	#include "common_internal.h"
11	#include "mqtt_constants.h"
12	#include "mqtt_ng.h"
13	#include "aclk_mqtt_workers.h"
14	#include "daemon/config/netdata-conf-profile.h"
15
16	#define PACKET_ACK_TIMEOUT_SECS (60)
17	#define SMALL_STRING_DONT_FRAGMENT_LIMIT 128
18
19	#define LOCK_HDR_BUFFER(buffer) spinlock_lock(&((buffer)->spinlock))
20	#define UNLOCK_HDR_BUFFER(buffer) spinlock_unlock(&((buffer)->spinlock))
21
22	#define BUFFER_FRAG_GARBAGE_COLLECT 0x01
23	// some packets can be marked for garbage collection
24	// immediately when they are sent (e.g. sent PUBACK on QoS1)
25	#define BUFFER_FRAG_GARBAGE_COLLECT_ON_SEND 0x02
26	// as buffer fragment can point to both
27	// external data and data in the same buffer
28	// we mark the former case with BUFFER_FRAG_DATA_EXTERNAL
29	#define BUFFER_FRAG_DATA_EXTERNAL 0x04
30	// as single MQTT Packet can be stored into multiple
31	// buffer fragments (depending on copy requirements)
32	// this marks this fragment to be the first/last
33	#define BUFFER_FRAG_MQTT_PACKET_HEAD 0x10
34	#define BUFFER_FRAG_MQTT_PACKET_TAIL 0x20
35
36	typedef uint16_t buffer_frag_flag_t;
37	struct buffer_fragment {
38	uint32_t len;
39	uint32_t sent;
40	buffer_frag_flag_t flags;
41	uint16_t packet_id;
42	void (free_fnc)(void ptr);
43	unsigned char *data;
44	// timestamp (monotonic usec) of when this MQTT packet was enqueued into the transaction buffer (set on HEAD)
45	usec_t enqueued_monotonic_ut;
46	usec_t sent_monotonic_ut;
47	struct buffer_fragment *next;
48	};
49
50	typedef struct buffer_fragment *mqtt_msg_data;
51
52	// buffer used for MQTT headers only
53	// not for actual data sent
54	struct header_buffer {
55	size_t size;
56	unsigned char *data;
57	unsigned char *tail;
58	struct buffer_fragment *tail_frag;
59	};
60
61	struct transaction_buffer {
62	struct header_buffer hdr_buffer;
63	// used while building new message
64	// to be able to revert state easily
65	// in case of error mid processing
66	struct header_buffer state_backup;
67	SPINLOCK spinlock;
68	struct buffer_fragment *sending_frag;
69	};
70
71	enum mqtt_client_state {
72	MQTT_STATE_RAW = 0,
73	MQTT_STATE_CONNECT_PENDING,
74	MQTT_STATE_CONNECTING,
75	MQTT_STATE_CONNECTED,
76	MQTT_STATE_ERROR,
77	MQTT_STATE_DISCONNECTED
78	};
79
80	enum parser_state {
81	MQTT_PARSE_FIXED_HEADER_PACKET_TYPE = 0,
82	MQTT_PARSE_FIXED_HEADER_LEN,
83	MQTT_PARSE_VARIABLE_HEADER,
84	MQTT_PARSE_MQTT_PACKET_DONE
85	};
86
87	enum varhdr_parser_state {
88	MQTT_PARSE_VARHDR_INITIAL = 0,
89	MQTT_PARSE_VARHDR_OPTIONAL_REASON_CODE,
90	MQTT_PARSE_VARHDR_PROPS,
91	MQTT_PARSE_VARHDR_TOPICNAME,
92	MQTT_PARSE_VARHDR_POST_TOPICNAME,
93	MQTT_PARSE_VARHDR_PACKET_ID,
94	MQTT_PARSE_REASONCODES,
95	MQTT_PARSE_PAYLOAD
96	};
97
98	struct mqtt_vbi_parser_ctx {
99	char data[MQTT_VBI_MAXBYTES];
100	uint8_t bytes;
101	uint32_t result;
102	};
103
104	enum mqtt_datatype {
105	MQTT_TYPE_UNKNOWN = 0,
106	MQTT_TYPE_UINT_8,
107	MQTT_TYPE_UINT_16,
108	MQTT_TYPE_UINT_32,
109	MQTT_TYPE_VBI,
110	MQTT_TYPE_STR,
111	MQTT_TYPE_STR_PAIR,
112	MQTT_TYPE_BIN
113	};
114
115	struct mqtt_property {
116	uint8_t id;
117	enum mqtt_datatype type;
118	union {
119	char *strings[2];
120	void *bindata;
121	uint8_t uint8;
122	uint16_t uint16;
123	uint32_t uint32;
124	} data;
125	size_t bindata_len;
126	struct mqtt_property *next;
127	};
128
129	enum mqtt_properties_parser_state {
130	PROPERTIES_LENGTH = 0,
131	PROPERTY_CREATE,
132	PROPERTY_ID,
133	PROPERTY_TYPE_UINT8,
134	PROPERTY_TYPE_UINT16,
135	PROPERTY_TYPE_UINT32,
136	PROPERTY_TYPE_STR_BIN_LEN,
137	PROPERTY_TYPE_STR,
138	PROPERTY_TYPE_BIN,
139	PROPERTY_TYPE_VBI,
140	PROPERTY_NEXT
141	};
142
143	struct mqtt_properties_parser_ctx {
144	enum mqtt_properties_parser_state state;
145	struct mqtt_property *head;
146	struct mqtt_property *tail;
147	uint32_t properties_length;
148	uint32_t vbi_length;
149	struct mqtt_vbi_parser_ctx vbi_parser_ctx;
150	size_t bytes_consumed;
151	int str_idx;
152	};
153
154	struct mqtt_connack {
155	uint8_t flags;
156	uint8_t reason_code;
157	};
158	struct mqtt_puback {
159	uint16_t packet_id;
160	uint8_t reason_code;
161	};
162
163	struct mqtt_suback {
164	uint16_t packet_id;
165	uint8_t *reason_codes;
166	uint8_t reason_code_count;
167	uint8_t reason_codes_pending;
168	};
169
170	struct mqtt_publish {
171	uint16_t topic_len;
172	char *topic;
173	uint16_t packet_id;
174	size_t data_len;
175	char *data;
176	uint8_t qos;
177	};
178
179	struct mqtt_disconnect {
180	uint8_t reason_code;
181	};
182
183	struct mqtt_ng_parser {
184	rbuf_t received_data;
185
186	uint8_t mqtt_control_packet_type;
187	uint32_t mqtt_fixed_hdr_remaining_length;
188	size_t mqtt_parsed_len;
189
190	struct mqtt_vbi_parser_ctx vbi_parser;
191	struct mqtt_properties_parser_ctx properties_parser;
192
193	enum parser_state state;
194	enum varhdr_parser_state varhdr_state;
195
196	struct mqtt_property *varhdr_properties;
197
198	union {
199	struct mqtt_connack connack;
200	struct mqtt_puback puback;
201	struct mqtt_suback suback;
202	struct mqtt_publish publish;
203	struct mqtt_disconnect disconnect;
204	} mqtt_packet;
205	};
206
207	struct topic_alias_data {
208	uint16_t idx;
209	uint32_t usage_count;
210	};
211
212	struct topic_aliases_data {
213	c_rhash stoi_dict;
214	uint32_t idx_max;
215	uint32_t idx_assigned;
216	SPINLOCK spinlock;
217	};
218
219	struct mqtt_ng_client {
220	struct transaction_buffer main_buffer;
221
222	enum mqtt_client_state client_state;
223
224	mqtt_msg_data connect_msg;
225
226	mqtt_ng_send_fnc_t send_fnc_ptr;
227	void *user_ctx;
228
229	// time when last fragment of MQTT message was sent
230	time_t time_of_last_send;
231
232	struct mqtt_ng_parser parser;
233
234	size_t max_mem_bytes;
235
236	void (*puback_callback)(uint16_t packet_id);
237	void (connack_callback)(void user_ctx, int connack_reply);
238	void (msg_callback)(const char topic, const void *msg, size_t msglen, int qos);
239
240	unsigned int ping_pending:1;
241
242	struct mqtt_ng_stats stats;
243
244	struct {
245	SPINLOCK spinlock;
246	Pvoid_t JudyL;
247	} pending_packets;
248
249	struct topic_aliases_data tx_topic_aliases;
250	c_rhash rx_aliases;
251
252	size_t max_msg_size;
253	};
254
255	usec_t publish_latency;
256
257	unsigned char pingreq[] = { MQTT_CPT_PINGREQ << 4, 0x00 };
258
259	struct buffer_fragment ping_frag = {
260	.data = pingreq,
261	.flags = BUFFER_FRAG_MQTT_PACKET_HEAD \| BUFFER_FRAG_MQTT_PACKET_TAIL,
262	.free_fnc = NULL,
263	.len = sizeof(pingreq),
264	.next = NULL,
265	.sent = 0,
266	.packet_id = 0
267	};
268
269	int uint32_to_mqtt_vbi(uint32_t input, unsigned char *output) {
270	int i = 1;
271	*output = 0;
272
273	/* MQTT 5 specs allows max 4 bytes of output
274	making it 0xFF, 0xFF, 0xFF, 0x7F
275	representing number 268435455 decimal
276	see 1.5.5. Variable Byte Integer */
277	if(input >= 256 * 1024 * 1024)
278	return 0;
279
280	if(!input) {
281	*output = 0;
282	return 1;
283	}
284
285	while(input) {
286	output[i-1] = input & MQTT_VBI_DATA_MASK;
287	input >>= 7;
288	if (input)
289	output[i-1] \|= MQTT_VBI_CONTINUATION_FLAG;
290	i++;
291	}
292	return i - 1;
293	}
294
295	int mqtt_vbi_to_uint32(char input, uint32_t output) {
296	// dont want to operate directly on output
297	// as I want it to be possible for input and output
298	// pointer to be the same
299	uint32_t result = 0;
300	uint32_t multiplier = 1;
301
302	do {
303	result += (uint32_t)(input & MQTT_VBI_DATA_MASK) multiplier;
304	if (multiplier > 128128128)
305	return 1;
306	multiplier <<= 7;
307	} while (*input++ & MQTT_VBI_CONTINUATION_FLAG);
308	*output = result;
309	return 0;
310	}
311
312	#ifdef TESTS
313	#include <stdio.h>
314	#define MQTT_VBI_MAXLEN 4
315	// we add extra byte to check we dont write out of bounds
316	// in case where 4 bytes are supposed to be written
317	static const char _mqtt_vbi_0[MQTT_VBI_MAXLEN + 1] = { 0x00, 0x00, 0x00, 0x00, 0x00 };
318	static const char _mqtt_vbi_127[MQTT_VBI_MAXLEN + 1] = { 0x7F, 0x00, 0x00, 0x00, 0x00 };
319	static const char _mqtt_vbi_128[MQTT_VBI_MAXLEN + 1] = { 0x80, 0x01, 0x00, 0x00, 0x00 };
320	static const char _mqtt_vbi_16383[MQTT_VBI_MAXLEN + 1] = { 0xFF, 0x7F, 0x00, 0x00, 0x00 };
321	static const char _mqtt_vbi_16384[MQTT_VBI_MAXLEN + 1] = { 0x80, 0x80, 0x01, 0x00, 0x00 };
322	static const char _mqtt_vbi_2097151[MQTT_VBI_MAXLEN + 1] = { 0xFF, 0xFF, 0x7F, 0x00, 0x00 };
323	static const char _mqtt_vbi_2097152[MQTT_VBI_MAXLEN + 1] = { 0x80, 0x80, 0x80, 0x01, 0x00 };
324	static const char _mqtt_vbi_268435455[MQTT_VBI_MAXLEN + 1] = { 0xFF, 0xFF, 0xFF, 0x7F, 0x00 };
325	static const char _mqtt_vbi_999999999[MQTT_VBI_MAXLEN + 1] = { 0x80, 0x80, 0x80, 0x80, 0x01 };
326
327	#define MQTT_VBI_TESTCASE(case, expected_len) \
328	{ \
329	memset(buf, 0, MQTT_VBI_MAXLEN + 1); \
330	int len; \
331	if ((len=uint32_to_mqtt_vbi(case, buf)) != expected_len) { \
332	fprintf(stderr, "uint32_to_mqtt_vbi(case:%d, line:%d): Incorrect length returned. Expected %d, Got %d\n", case, __LINE__, expected_len, len); \
333	return 1; \
334	} \
335	if (memcmp(buf, _mqtt_vbi_ ## case, MQTT_VBI_MAXLEN + 1 )) { \
336	fprintf(stderr, "uint32_to_mqtt_vbi(case:%d, line:%d): Wrong output\n", case, __LINE__); \
337	return 1; \
338	} }
339
340
341	int test_uint32_mqtt_vbi() {
342	char buf[MQTT_VBI_MAXLEN + 1];
343
344	MQTT_VBI_TESTCASE(0, 1)
345	MQTT_VBI_TESTCASE(127, 1)
346	MQTT_VBI_TESTCASE(128, 2)
347	MQTT_VBI_TESTCASE(16383, 2)
348	MQTT_VBI_TESTCASE(16384, 3)
349	MQTT_VBI_TESTCASE(2097151, 3)
350	MQTT_VBI_TESTCASE(2097152, 4)
351	MQTT_VBI_TESTCASE(268435455, 4)
352
353	memset(buf, 0, MQTT_VBI_MAXLEN + 1);
354	int len;
355	if ((len=uint32_to_mqtt_vbi(268435456, buf)) != 0) {
356	fprintf(stderr, "uint32_to_mqtt_vbi(case:268435456, line:%d): Incorrect length returned. Expected 0, Got %d\n", __LINE__, len);
357	return 1;
358	}
359
360	return 0;
361	}
362
363	#define MQTT_VBI2UINT_TESTCASE(case, expected_error) \
364	{ \
365	uint32_t result; \
366	int ret = mqtt_vbi_to_uint32(_mqtt_vbi_ ## case, &result); \
367	if (ret && !(expected_error)) { \
368	fprintf(stderr, "mqtt_vbi_to_uint(case:%d, line:%d): Unexpectedly Errored\n", (case), __LINE__); \
369	return 1; \
370	} \
371	if (!ret && (expected_error)) { \
372	fprintf(stderr, "mqtt_vbi_to_uint(case:%d, line:%d): Should return error but didnt\n", (case), __LINE__); \
373	return 1; \
374	} \
375	if (!ret && result != (case)) { \
376	fprintf(stderr, "mqtt_vbi_to_uint(case:%d, line:%d): Returned wrong result %d\n", (case), __LINE__, result); \
377	return 1; \
378	}}
379
380
381	int test_mqtt_vbi_to_uint32() {
382	MQTT_VBI2UINT_TESTCASE(0, 0)
383	MQTT_VBI2UINT_TESTCASE(127, 0)
384	MQTT_VBI2UINT_TESTCASE(128, 0)
385	MQTT_VBI2UINT_TESTCASE(16383, 0)
386	MQTT_VBI2UINT_TESTCASE(16384, 0)
387	MQTT_VBI2UINT_TESTCASE(2097151, 0)
388	MQTT_VBI2UINT_TESTCASE(2097152, 0)
389	MQTT_VBI2UINT_TESTCASE(268435455, 0)
390	MQTT_VBI2UINT_TESTCASE(999999999, 1)
391	return 0;
392	}
393	#endif /* TESTS */
394
395	// this helps with switch statements
396	// as they have to use integer type (not pointer)
397	enum memory_mode {
398	MEMCPY,
399	EXTERNAL_FREE_AFTER_USE,
400	CALLER_RESPONSIBLE
401	};
402
403	static enum memory_mode ptr2memory_mode(void * ptr) {
404	if (ptr == NULL)
405	return MEMCPY;
406	if (ptr == CALLER_RESPONSIBILITY)
407	return CALLER_RESPONSIBLE;
408	return EXTERNAL_FREE_AFTER_USE;
409	}
410
411	#define frag_is_marked_for_gc(frag) ((frag->flags & BUFFER_FRAG_GARBAGE_COLLECT) \|\| ((frag->flags & BUFFER_FRAG_GARBAGE_COLLECT_ON_SEND) && frag->sent == frag->len))
412	#define FRAG_SIZE_IN_BUFFER(frag) (sizeof(struct buffer_fragment) + ((frag->flags & BUFFER_FRAG_DATA_EXTERNAL) ? 0 : frag->len))
413
414	static void buffer_frag_free_data(struct buffer_fragment *frag)
415	{
416	if ( frag->flags & BUFFER_FRAG_DATA_EXTERNAL && frag->data != NULL) {
417	switch (ptr2memory_mode(frag->free_fnc)) {
418	case MEMCPY:
419	freez(frag->data);
420	break;
421	case EXTERNAL_FREE_AFTER_USE:
422	frag->free_fnc(frag->data);
423	break;
424	case CALLER_RESPONSIBLE:
425	break;
426	}
427	frag->data = NULL;
428	}
429	}
430
431	#define HEADER_BUFFER_SIZE (1024*1024)
432	#define HEADER_BUFFER_SIZE_IOT (128*1024)
433	#define HEADER_BUFFER_SIZE_STANDALONE (512*1024)
434
435	#define GROWTH_FACTOR 1.25
436
437	#define BUFFER_BYTES_USED(buf) ((size_t)((buf)->tail - (buf)->data))
438	#define BUFFER_BYTES_AVAILABLE(buf) ((buf)->size - BUFFER_BYTES_USED(buf))
439	#define BUFFER_FIRST_FRAG(buf) ((struct buffer_fragment *)((buf)->tail_frag ? (buf)->data : NULL))
440	static void buffer_purge(struct header_buffer *buf) {
441	struct buffer_fragment *frag = BUFFER_FIRST_FRAG(buf);
442	while (frag) {
443	buffer_frag_free_data(frag);
444	frag = frag->next;
445	}
446	buf->tail = buf->data;
447	buf->tail_frag = NULL;
448	}
449
450	#define FRAG_PADDING(addr) ((MQTT_WSS_FRAG_MEMALIGN - ((uintptr_t)addr % MQTT_WSS_FRAG_MEMALIGN)) % MQTT_WSS_FRAG_MEMALIGN)
451	static struct buffer_fragment buffer_new_frag(struct header_buffer buf, buffer_frag_flag_t flags)
452	{
453	uint8_t padding = FRAG_PADDING(buf->tail);
454
455	if (BUFFER_BYTES_AVAILABLE(buf) < sizeof(struct buffer_fragment) + padding)
456	return NULL;
457
458	struct buffer_fragment frag = (struct buffer_fragment )(buf->tail + padding);
459
460	memset(frag, 0, sizeof(*frag));
461	buf->tail += sizeof(*frag) + padding;
462
463	if (/!((frag)->flags & BUFFER_FRAG_MQTT_PACKET_HEAD) &&/ buf->tail_frag)
464	buf->tail_frag->next = frag;
465
466	buf->tail_frag = frag;
467
468	frag->data = buf->tail;
469
470	frag->flags = flags;
471
472	return frag;
473	}
474
475	static void buffer_rebuild(struct header_buffer *buf)
476	{
477	struct buffer_fragment frag = (struct buffer_fragment)buf->data;
478	do {
479	buf->tail = (unsigned char *) frag + sizeof(struct buffer_fragment);
480	buf->tail_frag = frag;
481	if (!(frag->flags & BUFFER_FRAG_DATA_EXTERNAL)) {
482	buf->tail_frag->data = buf->tail;
483	buf->tail += frag->len;
484	}
485	if (frag->next != NULL)
486	frag->next = (struct buffer_fragment*)(buf->tail + FRAG_PADDING(buf->tail));
487	frag = frag->next;
488	} while(frag);
489	}
490
491	static void buffer_garbage_collect(struct header_buffer *buf, bool main_thread)
492	{
493	struct buffer_fragment *frag = BUFFER_FIRST_FRAG(buf);
494	while (frag) {
495	if (!frag_is_marked_for_gc(frag))
496	break;
497
498	buffer_frag_free_data(frag);
499
500	frag = frag->next;
501	}
502
503	if (frag == BUFFER_FIRST_FRAG(buf))
504	return;
505
506	if (!frag) {
507	buf->tail_frag = NULL;
508	buf->tail = buf->data;
509	return;
510	}
511
512	#ifdef ADDITIONAL_CHECKS
513	if (!(frag->flags & BUFFER_FRAG_MQTT_PACKET_HEAD)) {
514	mws_error(log_ctx, "Expected to find end of buffer (NULL) or next packet head!");
515	return;
516	}
517	#endif
518
519	memmove(buf->data, frag, buf->tail - (unsigned char *) frag);
520	if (main_thread)
521	worker_is_busy(WORKER_ACLK_BUFFER_COMPACT);
522
523	buffer_rebuild(buf);
524
525	if (main_thread)
526	worker_is_idle();
527	}
528
529	static void transaction_buffer_garbage_collect(struct transaction_buffer *buf, bool main_thread)
530	{
531	if (main_thread)
532	worker_is_busy(WORKER_ACLK_RECLAIM_MEMORY);
533	// Invalidate the cached sending fragment
534	// as we will move data around
535	if (buf->sending_frag != &ping_frag)
536	buf->sending_frag = NULL;
537
538	buffer_garbage_collect(&buf->hdr_buffer, main_thread);
539	if (main_thread)
540	worker_is_idle();
541	}
542
543	static int transaction_buffer_grow(struct transaction_buffer *buf, float rate, size_t max)
544	{
545	if (buf->hdr_buffer.size >= max)
546	return 0;
547
548	// Invalidate the cached sending fragment
549	// as we will move data around
550	if (buf->sending_frag != &ping_frag)
551	buf->sending_frag = NULL;
552
553	buf->hdr_buffer.size = (size_t)((float)buf->hdr_buffer.size * rate);
554	if (buf->hdr_buffer.size > max)
555	buf->hdr_buffer.size = max;
556
557	void *ret = reallocz(buf->hdr_buffer.data, buf->hdr_buffer.size);
558	if (ret == NULL) {
559	nd_log(NDLS_DAEMON, NDLP_WARNING, "Buffer growth failed (realloc)");
560	return 1;
561	}
562
563	nd_log(NDLS_DAEMON, NDLP_DEBUG, "Message metadata buffer was grown");
564
565	buf->hdr_buffer.data = ret;
566	buffer_rebuild(&buf->hdr_buffer);
567	return 0;
568	}
569
570	inline static void transaction_buffer_init(struct transaction_buffer *to_init, size_t size)
571	{
572	spinlock_init(&to_init->spinlock);
573
574	to_init->hdr_buffer.size = size;
575	to_init->hdr_buffer.data = mallocz(size);
576	to_init->hdr_buffer.tail = to_init->hdr_buffer.data;
577	to_init->hdr_buffer.tail_frag = NULL;
578	}
579
580	static void transaction_buffer_destroy(struct transaction_buffer *to_init)
581	{
582	buffer_purge(&to_init->hdr_buffer);
583	freez(to_init->hdr_buffer.data);
584	}
585
586	// Creates transaction
587	// saves state of buffer before any operation was done
588	// allowing for rollback if things go wrong
589	#define transaction_buffer_transaction_start(buf) \
590	{ \
591	LOCK_HDR_BUFFER(buf); \
592	memcpy(&(buf)->state_backup, &(buf)->hdr_buffer, sizeof((buf)->hdr_buffer)); \
593	}
594
595	#define transaction_buffer_transaction_commit(buf) UNLOCK_HDR_BUFFER(buf);
596
597	void transaction_buffer_transaction_rollback(struct transaction_buffer buf, struct buffer_fragment frag)
598	{
599	memcpy(&buf->hdr_buffer, &buf->state_backup, sizeof(buf->hdr_buffer));
600	if (buf->hdr_buffer.tail_frag != NULL)
601	buf->hdr_buffer.tail_frag->next = NULL;
602
603	while(frag) {
604	buffer_frag_free_data(frag);
605	// we are not actually freeing the structure itself
606	// just the data it manages
607	// structure itself is in permanent buffer
608	// which is locked by HDR_BUFFER lock
609	frag = frag->next;
610	}
611
612	UNLOCK_HDR_BUFFER(buf);
613	}
614
615	#define TX_ALIASES_INITIALIZE() c_rhash_new(0)
616	#define RX_ALIASES_INITIALIZE() c_rhash_new(UINT16_MAX >> 8)
617	struct mqtt_ng_client mqtt_ng_init(struct mqtt_ng_init settings)
618	{
619	struct mqtt_ng_client *client = callocz(1, sizeof(struct mqtt_ng_client));
620
621	size_t buffer_size = netdata_conf_is_iot() ?
622	HEADER_BUFFER_SIZE_IOT :
623	(netdata_conf_is_standalone() ? HEADER_BUFFER_SIZE_STANDALONE : HEADER_BUFFER_SIZE);
624	transaction_buffer_init(&client->main_buffer, buffer_size);
625
626	client->rx_aliases = RX_ALIASES_INITIALIZE();
627
628	spinlock_init(&client->tx_topic_aliases.spinlock);
629
630	client->tx_topic_aliases.stoi_dict = TX_ALIASES_INITIALIZE();
631	client->tx_topic_aliases.idx_max = UINT16_MAX;
632
633	// TODO just embed the struct into mqtt_ng_client
634	client->parser.received_data = settings->data_in;
635	client->send_fnc_ptr = settings->data_out_fnc;
636	client->user_ctx = settings->user_ctx;
637
638	client->puback_callback = settings->puback_callback;
639	client->connack_callback = settings->connack_callback;
640	client->msg_callback = settings->msg_callback;
641	spinlock_init(&client->pending_packets.spinlock);
642	client->pending_packets.JudyL = NULL;
643	__atomic_store_n(&publish_latency, 0, __ATOMIC_RELEASE);
644
645	return client;
646	}
647
648	static uint8_t get_control_packet_type(uint8_t first_hdr_byte)
649	{
650	return first_hdr_byte >> 4;
651	}
652
653	static void mqtt_ng_destroy_rx_alias_hash(c_rhash hash)
654	{
655	c_rhash_iter_t i = C_RHASH_ITER_T_INITIALIZER;
656	uint64_t stored_key;
657	void *to_free;
658	while(!c_rhash_iter_uint64_keys(hash, &i, &stored_key)) {
659	c_rhash_get_ptr_by_uint64(hash, stored_key, &to_free);
660	freez(to_free);
661	}
662	c_rhash_destroy(hash);
663	}
664
665	static void mqtt_ng_destroy_tx_alias_hash(c_rhash hash)
666	{
667	c_rhash_iter_t i = C_RHASH_ITER_T_INITIALIZER;
668	const char *stored_key;
669	void *to_free;
670	while(!c_rhash_iter_str_keys(hash, &i, &stored_key)) {
671	c_rhash_get_ptr_by_str(hash, stored_key, &to_free);
672	freez(to_free);
673	}
674	c_rhash_destroy(hash);
675	}
676
677	static void destroy_timeout_monitor_list(struct mqtt_ng_client *client)
678	{
679	spinlock_lock(&client->pending_packets.spinlock);
680	(void) JudyLFreeArray(&client->pending_packets.JudyL, PJE0);
681	spinlock_unlock(&client->pending_packets.spinlock);
682	__atomic_store_n(&client->stats.packets_waiting_puback, 0, __ATOMIC_RELAXED);
683	}
684
685	void mqtt_ng_destroy(struct mqtt_ng_client *client)
686	{
687	transaction_buffer_destroy(&client->main_buffer);
688
689	mqtt_ng_destroy_tx_alias_hash(client->tx_topic_aliases.stoi_dict);
690	mqtt_ng_destroy_rx_alias_hash(client->rx_aliases);
691	destroy_timeout_monitor_list(client);
692	freez(client);
693	}
694
695	int frag_set_external_data(struct buffer_fragment frag, void data, size_t data_len, free_fnc_t data_free_fnc)
696	{
697	if (frag->len) {
698	// TODO?: This could potentially be done in future if we set rule
699	// external data always follows in buffer data
700	// could help reduce fragmentation in some messages but
701	// currently not worth it considering time is tight
702	nd_log(NDLS_DAEMON, NDLP_ERR, "INTERNAL ERROR: Cannot set external data to fragment already containing in buffer data!");
703	return 1;
704	}
705
706	switch (ptr2memory_mode(data_free_fnc)) {
707	case MEMCPY:
708	frag->data = mallocz(data_len);
709	memcpy(frag->data, data, data_len);
710	break;
711	case EXTERNAL_FREE_AFTER_USE:
712	case CALLER_RESPONSIBLE:
713	frag->data = data;
714	break;
715	}
716	frag->free_fnc = data_free_fnc;
717	frag->len = data_len;
718
719	frag->flags \|= BUFFER_FRAG_DATA_EXTERNAL;
720	return 0;
721	}
722
723	// this is fixed part of variable header for connect packet
724	// mqtt-v5.0-cs1, 3.1.2.1, 2.1.2.2
725	static const char mqtt_protocol_name_frag[] =
726	{ 0x00, 0x04, 'M', 'Q', 'T', 'T', MQTT_VERSION_5_0 };
727
728	#define MQTT_UTF8_STRING_SIZE(string) (2 + strlen(string))
729
730	// see 1.5.5
731	#define MQTT_VARSIZE_INT_BYTES(value) ( value > 2097152 ? 4 : ( value > 16384 ? 3 : ( value > 128 ? 2 : 1 ) ) )
732
733	static size_t mqtt_ng_connect_size(struct mqtt_auth_properties *auth,
734	struct mqtt_lwt_properties *lwt)
735	{
736	// First get the size of payload + variable header
737	size_t size =
738	+ sizeof(mqtt_protocol_name_frag) /* Proto Name and Version */
739	+ 1 /* Connect Flags */
740	+ 2 /* Keep Alive */
741	+ 4 /* 3.1.2.11.1 Property Length - for now fixed to only Topic Alias Maximum, TODO TODO*/;
742
743	// CONNECT payload. 3.1.3
744	if (auth->client_id)
745	size += MQTT_UTF8_STRING_SIZE(auth->client_id);
746
747	if (lwt) {
748	// 3.1.3.2 will properties TODO TODO
749	size += 1;
750
751	// 3.1.3.3
752	if (lwt->will_topic)
753	size += MQTT_UTF8_STRING_SIZE(lwt->will_topic);
754
755	// 3.1.3.4 will payload
756	if (lwt->will_message) {
757	size += 2 + lwt->will_message_size;
758	}
759	}
760
761	// 3.1.3.5
762	if (auth->username)
763	size += MQTT_UTF8_STRING_SIZE(auth->username);
764
765	// 3.1.3.6
766	if (auth->password)
767	size += MQTT_UTF8_STRING_SIZE(auth->password);
768
769	return size;
770	}
771
772	#define BUFFER_TRANSACTION_NEW_FRAG(buf, flags, frag, on_fail) \
773	{ if(frag==NULL) { \
774	frag = buffer_new_frag(buf, (flags)); } \
775	if(frag==NULL) { on_fail; }}
776
777	#define CHECK_BYTES_AVAILABLE(buf, needed, fail) \
778	{ \
779	if (BUFFER_BYTES_AVAILABLE(buf) < (size_t)needed) { \
780	fail; \
781	} \
782	}
783
784	#define DATA_ADVANCE(buf, bytes, frag) { size_t b = (bytes); (buf)->tail += b; (frag)->len += b; }
785
786	// TODO maybe just user client->buf.tail?
787	#define WRITE_POS(frag) (&(frag->data[frag->len]))
788
789	// [MQTT-1.5.2] Two Byte Integer
790	#define PACK_2B_INT(buffer, integer, frag) \
791	{ \
792	uint16_t temp = htobe16((integer)); \
793	memcpy(WRITE_POS(frag), &temp, sizeof(uint16_t)); \
794	DATA_ADVANCE(buffer, sizeof(uint16_t), frag); \
795	}
796
797	static int optimized_add(struct header_buffer buf, void data, size_t data_len, free_fnc_t data_free_fnc, struct buffer_fragment **frag)
798	{
799	if (data_len > SMALL_STRING_DONT_FRAGMENT_LIMIT) {
800	buffer_frag_flag_t flags = BUFFER_FRAG_DATA_EXTERNAL;
801	if ((*frag)->flags & BUFFER_FRAG_GARBAGE_COLLECT_ON_SEND)
802	flags \|= BUFFER_FRAG_GARBAGE_COLLECT_ON_SEND;
803	if( (*frag = buffer_new_frag(buf, flags)) == NULL ) {
804	nd_log(NDLS_DAEMON, NDLP_ERR, "Out of buffer space while generating the message");
805	return 1;
806	}
807	if (frag_set_external_data(*frag, data, data_len, data_free_fnc)) {
808	nd_log(NDLS_DAEMON, NDLP_ERR, "Error adding external data to newly created fragment");
809	return 1;
810	}
811	// we dont want to write to this fragment anymore
812	*frag = NULL;
813	} else if (data_len) {
814	// if the data are small dont bother creating new fragments
815	// store in buffer directly
816	CHECK_BYTES_AVAILABLE(buf, data_len, return 1)
817	memcpy(buf->tail, data, data_len);
818	DATA_ADVANCE(buf, data_len, *frag)
819	}
820	return 0;
821	}
822
823	static void remove_packet_from_timeout_monitor_list_unsafe(struct mqtt_ng_client *client, uint16_t packet_id)
824	{
825	int rc = JudyLDel(&client->pending_packets.JudyL, (Word_t) packet_id, PJE0);
826	// rc = 1 if the packet was deleted, so update statistics
827	if (likely(rc))
828	__atomic_fetch_sub(&client->stats.packets_waiting_puback, 1, __ATOMIC_RELAXED);
829	}
830
831	#define PACKET_TIMEOUT_EPOCH (1704067200L) // Jan 1, 2024 00:00:00 UTC
832
833	static void add_packet_to_timeout_monitor_list(struct mqtt_ng_client *client, uint16_t packet_id)
834	{
835	spinlock_lock(&client->pending_packets.spinlock);
836	time_t now = now_realtime_sec();
837	// Add it to the JudyL array
838	uint32_t Pvalue = (uint32_t ) JudyLIns(&client->pending_packets.JudyL, (Word_t) packet_id, PJE0);
839	if (Pvalue == PJERR) {
840	nd_log(NDLS_DAEMON, NDLP_ERR, "Error inserting packet_id (%" PRIu16 ") into JudyL array.", packet_id);
841	spinlock_unlock(&client->pending_packets.spinlock);
842	return;
843	}
844	*Pvalue = (uint32_t) ((now - PACKET_TIMEOUT_EPOCH) + PACKET_ACK_TIMEOUT_SECS);
845	spinlock_unlock(&client->pending_packets.spinlock);
846
847	__atomic_fetch_add(&client->stats.packets_waiting_puback, 1, __ATOMIC_RELAXED);
848	}
849
850	#define TRY_GENERATE_MESSAGE(generator_function, ...) \
851	({ \
852	int _rc = generator_function(&client->main_buffer, ##__VA_ARGS__); \
853	if (_rc == MQTT_NG_MSGGEN_BUFFER_OOM) { \
854	LOCK_HDR_BUFFER(&client->main_buffer); \
855	transaction_buffer_garbage_collect(&client->main_buffer, false); \
856	UNLOCK_HDR_BUFFER(&client->main_buffer); \
857	_rc = generator_function(&client->main_buffer, ##__VA_ARGS__); \
858	if (_rc == MQTT_NG_MSGGEN_BUFFER_OOM && client->max_mem_bytes) { \
859	LOCK_HDR_BUFFER(&client->main_buffer); \
860	transaction_buffer_grow((&client->main_buffer), GROWTH_FACTOR, client->max_mem_bytes); \
861	UNLOCK_HDR_BUFFER(&client->main_buffer); \
862	_rc = generator_function(&client->main_buffer, ##__VA_ARGS__); \
863	} \
864	if (_rc == MQTT_NG_MSGGEN_BUFFER_OOM) \
865	nd_log( \
866	NDLS_DAEMON, \
867	NDLP_ERR, \
868	"%s failed to generate message due to insufficient buffer space (line %d)", \
869	__FUNCTION__, \
870	__LINE__); \
871	} \
872	if (_rc == MQTT_NG_MSGGEN_OK) { \
873	__atomic_fetch_add(&client->stats.tx_messages_queued, 1, __ATOMIC_RELAXED); \
874	} \
875	_rc; \
876	})
877
878	mqtt_msg_data mqtt_ng_generate_connect(
879	struct transaction_buffer *trx_buf,
880	struct mqtt_auth_properties *auth,
881	struct mqtt_lwt_properties *lwt,
882	uint16_t keep_alive)
883	{
884	// Sanity Checks First (are given parameters correct and up to MQTT spec)
885	if (!auth->client_id) {
886	nd_log(NDLS_DAEMON, NDLP_ERR, "ClientID must be set. [MQTT-3.1.3-3]");
887	return NULL;
888	}
889
890	size_t len = strlen(auth->client_id);
891	if (!len) {
892	// [MQTT-3.1.3-6] server MAY allow empty client_id and treat it
893	// as specific client_id (not same as client_id not given)
894	// however server MUST allow ClientIDs between 1-23 bytes [MQTT-3.1.3-5]
895	// so we will warn client server might not like this and he is using it
896	// at his own risk!
897	nd_log(NDLS_DAEMON, NDLP_WARNING, "client_id provided is empty string. This might not be allowed by server [MQTT-3.1.3-6]");
898	}
899
900	if (lwt) {
901	if (lwt->will_message && lwt->will_message_size > 65535) {
902	nd_log(NDLS_DAEMON, NDLP_ERR, "Will message cannot be longer than 65535 bytes due to MQTT protocol limitations [MQTT-3.1.3-4] and [MQTT-1.5.6]");
903	return NULL;
904	}
905
906	if (!lwt->will_topic) { //TODO topic given with strlen==0 ? check specs
907	nd_log(NDLS_DAEMON, NDLP_ERR, "If will message is given will topic must also be given [MQTT-3.1.3.3]");
908	return NULL;
909	}
910
911	if (lwt->will_qos > MQTT_MAX_QOS) {
912	// refer to [MQTT-3-1.2-12]
913	nd_log(NDLS_DAEMON, NDLP_ERR, "QOS for LWT message is bigger than max");
914	return NULL;
915	}
916	}
917
918	// >> START THE RODEO <<
919	transaction_buffer_transaction_start(trx_buf)
920
921	// Calculate the resulting message size sans fixed MQTT header
922	size_t size = mqtt_ng_connect_size(auth, lwt);
923
924	// Start generating the message
925	struct buffer_fragment *frag = NULL;
926	mqtt_msg_data ret = NULL;
927
928	BUFFER_TRANSACTION_NEW_FRAG(&trx_buf->hdr_buffer, BUFFER_FRAG_MQTT_PACKET_HEAD, frag, goto fail_rollback)
929	ret = frag;
930
931	// MQTT Fixed Header
932	size_t needed_bytes = 1 /* Packet type / + MQTT_VARSIZE_INT_BYTES(size) + sizeof(mqtt_protocol_name_frag) + 1 / CONNECT FLAGS / + 2 / keepalive / + 1 / Properties TODO now fixed 0*/;
933	CHECK_BYTES_AVAILABLE(&trx_buf->hdr_buffer, needed_bytes, goto fail_rollback)
934
935	*WRITE_POS(frag) = MQTT_CPT_CONNECT << 4;
936	DATA_ADVANCE(&trx_buf->hdr_buffer, 1, frag)
937	DATA_ADVANCE(&trx_buf->hdr_buffer, uint32_to_mqtt_vbi(size, WRITE_POS(frag)), frag)
938
939	memcpy(WRITE_POS(frag), mqtt_protocol_name_frag, sizeof(mqtt_protocol_name_frag));
940	DATA_ADVANCE(&trx_buf->hdr_buffer, sizeof(mqtt_protocol_name_frag), frag)
941
942	// [MQTT-3.1.2.3] Connect flags
943	unsigned char *connect_flags = WRITE_POS(frag);
944	*connect_flags = 0;
945	if (auth->username)
946	*connect_flags \|= MQTT_CONNECT_FLAG_USERNAME;
947
948	if (auth->password)
949	*connect_flags \|= MQTT_CONNECT_FLAG_PASSWORD;
950
951	if (lwt) {
952	*connect_flags \|= MQTT_CONNECT_FLAG_LWT;
953	*connect_flags \|= lwt->will_qos << MQTT_CONNECT_FLAG_QOS_BITSHIFT;
954	if (lwt->will_retain)
955	*connect_flags \|= MQTT_CONNECT_FLAG_LWT_RETAIN;
956	}
957
958	*connect_flags \|= MQTT_CONNECT_FLAG_CLEAN_START;
959
960	DATA_ADVANCE(&trx_buf->hdr_buffer, 1, frag)
961
962	PACK_2B_INT(&trx_buf->hdr_buffer, keep_alive, frag)
963
964	// TODO Property Length [MQTT-3.1.3.2.1] temporary fixed to 3 (one property topic alias max)
965	DATA_ADVANCE(&trx_buf->hdr_buffer, uint32_to_mqtt_vbi(3, WRITE_POS(frag)), frag)
966	*WRITE_POS(frag) = MQTT_PROP_TOPIC_ALIAS_MAX;
967	DATA_ADVANCE(&trx_buf->hdr_buffer, 1, frag)
968
969	PACK_2B_INT(&trx_buf->hdr_buffer, 65535, frag)
970
971	// [MQTT-3.1.3.1] Client identifier
972	CHECK_BYTES_AVAILABLE(&trx_buf->hdr_buffer, 2, goto fail_rollback)
973	PACK_2B_INT(&trx_buf->hdr_buffer, strlen(auth->client_id), frag)
974	if (optimized_add(&trx_buf->hdr_buffer, auth->client_id, strlen(auth->client_id), auth->client_id_free, &frag))
975	goto fail_rollback;
976
977	if (lwt != NULL) {
978	// Will Properties [MQTT-3.1.3.2]
979	// TODO for now fixed 0
980	BUFFER_TRANSACTION_NEW_FRAG(&trx_buf->hdr_buffer, 0, frag, goto fail_rollback)
981	CHECK_BYTES_AVAILABLE(&trx_buf->hdr_buffer, 1, goto fail_rollback)
982	*WRITE_POS(frag) = 0;
983	DATA_ADVANCE(&trx_buf->hdr_buffer, 1, frag)
984
985	// Will Topic [MQTT-3.1.3.3]
986	CHECK_BYTES_AVAILABLE(&trx_buf->hdr_buffer, 2, goto fail_rollback)
987	PACK_2B_INT(&trx_buf->hdr_buffer, strlen(lwt->will_topic), frag)
988	if (optimized_add(&trx_buf->hdr_buffer, lwt->will_topic, strlen(lwt->will_topic), lwt->will_topic_free, &frag))
989	goto fail_rollback;
990
991	// Will Payload [MQTT-3.1.3.4]
992	if (lwt->will_message_size) {
993	BUFFER_TRANSACTION_NEW_FRAG(&trx_buf->hdr_buffer, 0, frag, goto fail_rollback)
994	CHECK_BYTES_AVAILABLE(&trx_buf->hdr_buffer, 2, goto fail_rollback)
995	PACK_2B_INT(&trx_buf->hdr_buffer, lwt->will_message_size, frag)
996	if (optimized_add(
997	&trx_buf->hdr_buffer, lwt->will_message, lwt->will_message_size, lwt->will_topic_free, &frag))
998	goto fail_rollback;
999	}
1000	}
1001
1002	// [MQTT-3.1.3.5]
1003	if (auth->username) {
1004	BUFFER_TRANSACTION_NEW_FRAG(&trx_buf->hdr_buffer, 0, frag, goto fail_rollback)
1005	CHECK_BYTES_AVAILABLE(&trx_buf->hdr_buffer, 2, goto fail_rollback)
1006	PACK_2B_INT(&trx_buf->hdr_buffer, strlen(auth->username), frag)
1007	if (optimized_add(&trx_buf->hdr_buffer, auth->username, strlen(auth->username), auth->username_free, &frag))
1008	goto fail_rollback;
1009	}
1010
1011	// [MQTT-3.1.3.6]
1012	if (auth->password) {
1013	BUFFER_TRANSACTION_NEW_FRAG(&trx_buf->hdr_buffer, 0, frag, goto fail_rollback)
1014	CHECK_BYTES_AVAILABLE(&trx_buf->hdr_buffer, 2, goto fail_rollback)
1015	PACK_2B_INT(&trx_buf->hdr_buffer, strlen(auth->password), frag)
1016	if (optimized_add(&trx_buf->hdr_buffer, auth->password, strlen(auth->password), auth->password_free, &frag))
1017	goto fail_rollback;
1018	}
1019	trx_buf->hdr_buffer.tail_frag->flags \|= BUFFER_FRAG_MQTT_PACKET_TAIL;
1020	transaction_buffer_transaction_commit(trx_buf)
1021	return ret;
1022	fail_rollback:
1023	transaction_buffer_transaction_rollback(trx_buf, ret);
1024	return NULL;
1025	}
1026
1027	int mqtt_ng_connect(
1028	struct mqtt_ng_client *client,
1029	struct mqtt_auth_properties *auth,
1030	struct mqtt_lwt_properties *lwt,
1031	uint16_t keep_alive)
1032	{
1033	client->client_state = MQTT_STATE_RAW;
1034	client->parser.state = MQTT_PARSE_FIXED_HEADER_PACKET_TYPE;
1035
1036	LOCK_HDR_BUFFER(&client->main_buffer);
1037	client->main_buffer.sending_frag = NULL;
1038	buffer_purge(&client->main_buffer.hdr_buffer);
1039	UNLOCK_HDR_BUFFER(&client->main_buffer);
1040
1041	destroy_timeout_monitor_list(client);
1042
1043	spinlock_lock(&client->tx_topic_aliases.spinlock);
1044	// according to MQTT spec topic aliases should not be persisted
1045	// even if clean session is true
1046	mqtt_ng_destroy_tx_alias_hash(client->tx_topic_aliases.stoi_dict);
1047
1048	client->tx_topic_aliases.stoi_dict = TX_ALIASES_INITIALIZE();
1049	client->tx_topic_aliases.idx_assigned = 0;
1050	spinlock_unlock(&client->tx_topic_aliases.spinlock);
1051
1052	mqtt_ng_destroy_rx_alias_hash(client->rx_aliases);
1053	client->rx_aliases = RX_ALIASES_INITIALIZE();
1054
1055	client->connect_msg = mqtt_ng_generate_connect(&client->main_buffer, auth, lwt, keep_alive);
1056	if (client->connect_msg == NULL)
1057	return 1;
1058
1059	__atomic_store_n(&client->stats.tx_messages_queued, 1, __ATOMIC_RELAXED);
1060	__atomic_store_n(&client->stats.tx_messages_sent, 0, __ATOMIC_RELAXED);
1061	__atomic_store_n(&client->stats.rx_messages_rcvd, 0, __ATOMIC_RELAXED);
1062
1063	client->client_state = MQTT_STATE_CONNECT_PENDING;
1064	return 0;
1065	}
1066
1067	uint16_t get_unused_packet_id() {
1068	static uint16_t packet_id = 0;
1069	uint16_t id = __atomic_fetch_add(&packet_id, 1, __ATOMIC_RELAXED) + 1;
1070	return id ? id : 1;
1071	}
1072
1073	static size_t mqtt_ng_publish_size(
1074	const char *topic,
1075	size_t msg_len,
1076	uint16_t topic_id)
1077	{
1078	size_t retval = 2
1079	+ (topic == NULL ? 0 : strlen(topic)) /* Topic Name Length */
1080	+ 2 /* Packet identifier */
1081	+ 1 /* Properties Length for now fixed to 1 property */
1082	+ msg_len;
1083
1084	if (topic_id)
1085	retval += 3;
1086
1087	return retval;
1088	}
1089
1090	int mqtt_ng_generate_publish(struct transaction_buffer *trx_buf,
1091	char *topic,
1092	free_fnc_t topic_free,
1093	void *msg,
1094	free_fnc_t msg_free,
1095	size_t msg_len,
1096	uint8_t publish_flags,
1097	uint16_t *packet_id,
1098	uint16_t topic_alias)
1099	{
1100	// >> START THE RODEO <<
1101	transaction_buffer_transaction_start(trx_buf)
1102
1103	// Calculate the resulting message size sans fixed MQTT header
1104	size_t size = mqtt_ng_publish_size(topic, msg_len, topic_alias);
1105
1106	// Start generating the message
1107	struct buffer_fragment *frag = NULL;
1108	mqtt_msg_data mqtt_msg = NULL;
1109
1110	BUFFER_TRANSACTION_NEW_FRAG(&trx_buf->hdr_buffer, BUFFER_FRAG_MQTT_PACKET_HEAD, frag, goto fail_rollback )
1111	// in case of QOS 0 we can garbage collect immediatelly after sending
1112	uint8_t qos = (publish_flags >> 1) & 0x03;
1113	if (!qos)
1114	frag->flags \|= BUFFER_FRAG_GARBAGE_COLLECT_ON_SEND;
1115	mqtt_msg = frag;
1116
1117	// MQTT Fixed Header
1118	size_t needed_bytes = 1 /* Packet type */ + MQTT_VARSIZE_INT_BYTES(size) + size - msg_len;
1119	CHECK_BYTES_AVAILABLE(&trx_buf->hdr_buffer, needed_bytes, goto fail_rollback)
1120
1121	*WRITE_POS(frag) = (MQTT_CPT_PUBLISH << 4) \| (publish_flags & 0xF);
1122	DATA_ADVANCE(&trx_buf->hdr_buffer, 1, frag)
1123	DATA_ADVANCE(&trx_buf->hdr_buffer, uint32_to_mqtt_vbi(size, WRITE_POS(frag)), frag)
1124
1125	// MQTT Variable Header
1126	// [MQTT-3.3.2.1]
1127	PACK_2B_INT(&trx_buf->hdr_buffer, topic == NULL ? 0 : strlen(topic), frag)
1128	if (topic != NULL) {
1129	if (optimized_add(&trx_buf->hdr_buffer, topic, strlen(topic), topic_free, &frag))
1130	goto fail_rollback;
1131	BUFFER_TRANSACTION_NEW_FRAG(&trx_buf->hdr_buffer, 0, frag, goto fail_rollback)
1132	}
1133
1134	// [MQTT-3.3.2.2]
1135	mqtt_msg->packet_id = get_unused_packet_id();
1136	*packet_id = mqtt_msg->packet_id;
1137	PACK_2B_INT(&trx_buf->hdr_buffer, mqtt_msg->packet_id, frag)
1138
1139	// [MQTT-3.3.2.3.1] TODO Property Length for now fixed 0
1140	*WRITE_POS(frag) = topic_alias ? 3 : 0;
1141	DATA_ADVANCE(&trx_buf->hdr_buffer, 1, frag)
1142
1143	if(topic_alias) {
1144	*WRITE_POS(frag) = MQTT_PROP_TOPIC_ALIAS;
1145	DATA_ADVANCE(&trx_buf->hdr_buffer, 1, frag)
1146
1147	PACK_2B_INT(&trx_buf->hdr_buffer, topic_alias, frag)
1148	}
1149
1150	if( (frag = buffer_new_frag(&trx_buf->hdr_buffer, BUFFER_FRAG_DATA_EXTERNAL)) == NULL )
1151	goto fail_rollback;
1152
1153	if (frag_set_external_data(frag, msg, msg_len, msg_free))
1154	goto fail_rollback;
1155
1156	trx_buf->hdr_buffer.tail_frag->flags \|= BUFFER_FRAG_MQTT_PACKET_TAIL;
1157	if (!qos)
1158	trx_buf->hdr_buffer.tail_frag->flags \|= BUFFER_FRAG_GARBAGE_COLLECT_ON_SEND;
1159	transaction_buffer_transaction_commit(trx_buf)
1160	// mark enqueue time on the HEAD fragment (after commit) so we measure from commit time
1161	if (mqtt_msg)
1162	mqtt_msg->enqueued_monotonic_ut = now_monotonic_usec();
1163	return MQTT_NG_MSGGEN_OK;
1164	fail_rollback:
1165	transaction_buffer_transaction_rollback(trx_buf, mqtt_msg);
1166	return MQTT_NG_MSGGEN_BUFFER_OOM;
1167	}
1168
1169	static void mark_message_for_gc(struct buffer_fragment *frag)
1170	{
1171	while (frag) {
1172	frag->flags \|= BUFFER_FRAG_GARBAGE_COLLECT;
1173	buffer_frag_free_data(frag);
1174	if (frag->flags & BUFFER_FRAG_MQTT_PACKET_TAIL)
1175	return;
1176	frag = frag->next;
1177	}
1178	}
1179
1180	// Check if sending_frag points to any fragment in the message starting at msg_head
1181	static bool sending_frag_in_message(struct transaction_buffer buf, struct buffer_fragment msg_head)
1182	{
1183	struct buffer_fragment *frag = msg_head;
1184	while (frag) {
1185	if (buf->sending_frag == frag)
1186	return true;
1187	if (frag->flags & BUFFER_FRAG_MQTT_PACKET_TAIL)
1188	return false;
1189	frag = frag->next;
1190	}
1191	return false;
1192	}
1193
1194	static int mark_packet_acked(struct mqtt_ng_client *client, uint16_t packet_id)
1195	{
1196	size_t reclaimable = 0;
1197	spinlock_lock(&client->pending_packets.spinlock);
1198	LOCK_HDR_BUFFER(&client->main_buffer);
1199	struct buffer_fragment *frag = BUFFER_FIRST_FRAG(&client->main_buffer.hdr_buffer);
1200	while (frag) {
1201	if ( (frag->flags & BUFFER_FRAG_MQTT_PACKET_HEAD) && frag->packet_id == packet_id) {
1202	if (!frag->sent) {
1203	nd_log(NDLS_DAEMON, NDLP_ERR, "Received packet_id (%" PRIu16 ") belongs to MQTT packet which was not yet sent!", packet_id);
1204	UNLOCK_HDR_BUFFER(&client->main_buffer);
1205	spinlock_unlock(&client->pending_packets.spinlock);
1206	return 1;
1207	}
1208	// Do not reprocess this packet
1209	frag->packet_id = 0;
1210	usec_t latency = now_monotonic_usec() - frag->sent_monotonic_ut;
1211	pulse_aclk_sent_message_acked(latency, frag->len);
1212	__atomic_store_n(&publish_latency, latency, __ATOMIC_RELEASE);
1213
1214	// Invalidate sending_frag if it points to any fragment in this message
1215	// since mark_message_for_gc will free the data
1216	if (sending_frag_in_message(&client->main_buffer, frag))
1217	client->main_buffer.sending_frag = NULL;
1218
1219	mark_message_for_gc(frag);
1220
1221	size_t used = BUFFER_BYTES_USED(&client->main_buffer.hdr_buffer);
1222	if (reclaimable >= (used / 4))
1223	transaction_buffer_garbage_collect(&client->main_buffer, true);
1224
1225	UNLOCK_HDR_BUFFER(&client->main_buffer);
1226	remove_packet_from_timeout_monitor_list_unsafe(client, packet_id);
1227	spinlock_unlock(&client->pending_packets.spinlock);
1228	return 0;
1229	}
1230
1231	if(frag_is_marked_for_gc(frag))
1232	reclaimable += FRAG_SIZE_IN_BUFFER(frag);
1233
1234	frag = frag->next;
1235	}
1236	nd_log(NDLS_DAEMON, NDLP_WARNING, "Received packet_id (%" PRIu16 ") is unknown, removing from monitor list", packet_id);
1237	UNLOCK_HDR_BUFFER(&client->main_buffer);
1238	remove_packet_from_timeout_monitor_list_unsafe(client, packet_id);
1239	spinlock_unlock(&client->pending_packets.spinlock);
1240	return 1;
1241	}
1242
1243	#define MAX_TIMED_OUT_PACKETS (1024)
1244
1245	static bool check_packet_monitor_list_for_timeouts(struct mqtt_ng_client *client)
1246	{
1247	uint16_t timed_out_packets[MAX_TIMED_OUT_PACKETS];
1248	size_t timed_out_count = 0;
1249
1250	spinlock_lock(&client->pending_packets.spinlock);
1251	bool first_then_next = true;
1252	uint32_t *Pvalue;
1253	Word_t packet_id = 0;
1254	time_t now = now_realtime_sec();
1255
1256	while ((Pvalue = (uint32_t *) JudyLFirstThenNext(client->pending_packets.JudyL, &packet_id, &first_then_next))) {
1257	uint32_t expire_time_delta = *Pvalue;
1258	if (now >= (PACKET_TIMEOUT_EPOCH + expire_time_delta)) {
1259	if (timed_out_count < MAX_TIMED_OUT_PACKETS) {
1260	timed_out_packets[timed_out_count++] = (uint16_t)packet_id;
1261	} else
1262	break;
1263	}
1264	}
1265	spinlock_unlock(&client->pending_packets.spinlock);
1266
1267	// Process timeouts outside the lock
1268	for (size_t i = 0; i < timed_out_count; i++) {
1269	mark_packet_acked(client, timed_out_packets[i]);
1270	}
1271
1272	return (timed_out_count == MAX_TIMED_OUT_PACKETS);
1273	}
1274
1275	#define PUBLISH_SP_SIZE 64
1276	int mqtt_ng_publish(struct mqtt_ng_client *client,
1277	char *topic,
1278	free_fnc_t topic_free,
1279	void *msg,
1280	free_fnc_t msg_free,
1281	size_t msg_len,
1282	uint8_t publish_flags,
1283	uint16_t *packet_id)
1284	{
1285	struct topic_alias_data *alias = NULL;
1286	spinlock_lock(&client->tx_topic_aliases.spinlock);
1287	c_rhash_get_ptr_by_str(client->tx_topic_aliases.stoi_dict, topic, (void**)&alias);
1288	spinlock_unlock(&client->tx_topic_aliases.spinlock);
1289
1290	uint16_t topic_id = 0;
1291
1292	if (alias != NULL) {
1293	topic_id = alias->idx;
1294	uint32_t cnt = __atomic_fetch_add(&alias->usage_count, 1, __ATOMIC_SEQ_CST);
1295	if (cnt) {
1296	topic = NULL;
1297	topic_free = NULL;
1298	}
1299	}
1300
1301	if (client->max_msg_size && PUBLISH_SP_SIZE + mqtt_ng_publish_size(topic, msg_len, topic_id) > client->max_msg_size) {
1302	nd_log(NDLS_DAEMON, NDLP_ERR, "Message too big for server: %zu", msg_len);
1303	return MQTT_NG_MSGGEN_MSG_TOO_BIG;
1304	}
1305
1306	// Ownership contract on failure: do NOT free msg or clear *packet_id here.
1307	// The sole caller (mqtt_wss_publish5) owns cleanup on every non-OK return.
1308	// On MQTT_NG_MSGGEN_OK, msg is attached to a buffer fragment and freed by
1309	// the transaction-buffer GC after ack; *packet_id has been written by the
1310	// generator. See the long comment in mqtt_wss_publish5 for the invariant.
1311	int rc = TRY_GENERATE_MESSAGE(mqtt_ng_generate_publish, topic, topic_free, msg, msg_free, msg_len, publish_flags, packet_id, topic_id);
1312	if (rc == MQTT_NG_MSGGEN_OK && packet_id)
1313	add_packet_to_timeout_monitor_list(client, *packet_id);
1314	return rc;
1315	}
1316
1317	static size_t mqtt_ng_subscribe_size(struct mqtt_sub *subs, size_t sub_count)
1318	{
1319	size_t len = 2 /* Packet Identifier / + 1 / Properties Length TODO for now fixed 0 */;
1320	len += sub_count * (2 /* topic filter string length / + 1 / [MQTT-3.8.3.1] Subscription Options Byte */);
1321
1322	for (size_t i = 0; i < sub_count; i++) {
1323	len += strlen(subs[i].topic);
1324	}
1325	return len;
1326	}
1327
1328	int mqtt_ng_generate_subscribe(struct transaction_buffer trx_buf, struct mqtt_sub subs, size_t sub_count)
1329	{
1330	// >> START THE RODEO <<
1331	transaction_buffer_transaction_start(trx_buf)
1332
1333	// Calculate the resulting message size sans fixed MQTT header
1334	size_t size = mqtt_ng_subscribe_size(subs, sub_count);
1335
1336	// Start generating the message
1337	struct buffer_fragment *frag = NULL;
1338	mqtt_msg_data ret = NULL;
1339
1340	BUFFER_TRANSACTION_NEW_FRAG(&trx_buf->hdr_buffer, BUFFER_FRAG_MQTT_PACKET_HEAD, frag, goto fail_rollback)
1341	ret = frag;
1342
1343	// MQTT Fixed Header
1344	size_t needed_bytes = 1 /* Packet type / + MQTT_VARSIZE_INT_BYTES(size) + 3 /Packet ID + Property Length*/;
1345	CHECK_BYTES_AVAILABLE(&trx_buf->hdr_buffer, needed_bytes, goto fail_rollback)
1346
1347	WRITE_POS(frag) = (MQTT_CPT_SUBSCRIBE << 4) \| 0x2 / [MQTT-3.8.1-1] */;
1348	DATA_ADVANCE(&trx_buf->hdr_buffer, 1, frag)
1349	DATA_ADVANCE(&trx_buf->hdr_buffer, uint32_to_mqtt_vbi(size, WRITE_POS(frag)), frag)
1350
1351	// MQTT Variable Header
1352	// [MQTT-3.8.2] PacketID
1353	ret->packet_id = get_unused_packet_id();
1354	PACK_2B_INT(&trx_buf->hdr_buffer, ret->packet_id, frag)
1355
1356	// [MQTT-3.8.2.1.1] Property Length // TODO for now fixed 0
1357	*WRITE_POS(frag) = 0;
1358	DATA_ADVANCE(&trx_buf->hdr_buffer, 1, frag)
1359
1360	for (size_t i = 0; i < sub_count; i++) {
1361	BUFFER_TRANSACTION_NEW_FRAG(&trx_buf->hdr_buffer, 0, frag, goto fail_rollback)
1362	PACK_2B_INT(&trx_buf->hdr_buffer, strlen(subs[i].topic), frag)
1363	if (optimized_add(&trx_buf->hdr_buffer, subs[i].topic, strlen(subs[i].topic), subs[i].topic_free, &frag))
1364	goto fail_rollback;
1365	BUFFER_TRANSACTION_NEW_FRAG(&trx_buf->hdr_buffer, 0, frag, goto fail_rollback)
1366	*WRITE_POS(frag) = subs[i].options;
1367	DATA_ADVANCE(&trx_buf->hdr_buffer, 1, frag)
1368	}
1369
1370	trx_buf->hdr_buffer.tail_frag->flags \|= BUFFER_FRAG_MQTT_PACKET_TAIL;
1371	transaction_buffer_transaction_commit(trx_buf)
1372	return MQTT_NG_MSGGEN_OK;
1373	fail_rollback:
1374	transaction_buffer_transaction_rollback(trx_buf, ret);
1375	return MQTT_NG_MSGGEN_BUFFER_OOM;
1376	}
1377
1378	int mqtt_ng_subscribe(struct mqtt_ng_client client, struct mqtt_sub subs, size_t sub_count)
1379	{
1380	return TRY_GENERATE_MESSAGE(mqtt_ng_generate_subscribe, subs, sub_count);
1381	}
1382
1383	int mqtt_ng_generate_disconnect(struct transaction_buffer *trx_buf, uint8_t reason_code)
1384	{
1385	// >> START THE RODEO <<
1386	transaction_buffer_transaction_start(trx_buf)
1387
1388	// Calculate the resulting message size sans fixed MQTT header
1389	size_t size = reason_code ? 1 : 0;
1390
1391	// Start generating the message
1392	struct buffer_fragment *frag = NULL;
1393	mqtt_msg_data ret = NULL;
1394
1395	BUFFER_TRANSACTION_NEW_FRAG(&trx_buf->hdr_buffer, BUFFER_FRAG_MQTT_PACKET_HEAD, frag, goto fail_rollback)
1396	ret = frag;
1397
1398	// MQTT Fixed Header
1399	size_t needed_bytes = 1 /* Packet type */ + MQTT_VARSIZE_INT_BYTES(size) + (reason_code ? 1 : 0);
1400	CHECK_BYTES_AVAILABLE(&trx_buf->hdr_buffer, needed_bytes, goto fail_rollback)
1401
1402	*WRITE_POS(frag) = MQTT_CPT_DISCONNECT << 4;
1403	DATA_ADVANCE(&trx_buf->hdr_buffer, 1, frag)
1404	DATA_ADVANCE(&trx_buf->hdr_buffer, uint32_to_mqtt_vbi(size, WRITE_POS(frag)), frag)
1405
1406	if (reason_code) {
1407	// MQTT Variable Header
1408	// [MQTT-3.14.2.1] PacketID
1409	*WRITE_POS(frag) = reason_code;
1410	DATA_ADVANCE(&trx_buf->hdr_buffer, 1, frag)
1411	}
1412
1413	trx_buf->hdr_buffer.tail_frag->flags \|= BUFFER_FRAG_MQTT_PACKET_TAIL;
1414	transaction_buffer_transaction_commit(trx_buf)
1415	return MQTT_NG_MSGGEN_OK;
1416	fail_rollback:
1417	transaction_buffer_transaction_rollback(trx_buf, ret);
1418	return MQTT_NG_MSGGEN_BUFFER_OOM;
1419	}
1420
1421	int mqtt_ng_disconnect(struct mqtt_ng_client *client, uint8_t reason_code)
1422	{
1423	return TRY_GENERATE_MESSAGE(mqtt_ng_generate_disconnect, reason_code);
1424	}
1425
1426	static int mqtt_generate_puback(struct transaction_buffer *trx_buf, uint16_t packet_id, uint8_t reason_code)
1427	{
1428	// >> START THE RODEO <<
1429	transaction_buffer_transaction_start(trx_buf)
1430
1431	// Calculate the resulting message size sans fixed MQTT header
1432	size_t size = 2 /* Packet ID / + (reason_code ? 1 : 0) / reason code */;
1433
1434	// Start generating the message
1435	struct buffer_fragment *frag = NULL;
1436
1437	BUFFER_TRANSACTION_NEW_FRAG(&trx_buf->hdr_buffer, BUFFER_FRAG_MQTT_PACKET_HEAD \| BUFFER_FRAG_GARBAGE_COLLECT_ON_SEND, frag, goto fail_rollback)
1438
1439	// MQTT Fixed Header
1440	size_t needed_bytes = 1 /* Packet type */ + MQTT_VARSIZE_INT_BYTES(size) + size;
1441	CHECK_BYTES_AVAILABLE(&trx_buf->hdr_buffer, needed_bytes, goto fail_rollback)
1442
1443	*WRITE_POS(frag) = MQTT_CPT_PUBACK << 4;
1444	DATA_ADVANCE(&trx_buf->hdr_buffer, 1, frag)
1445	DATA_ADVANCE(&trx_buf->hdr_buffer, uint32_to_mqtt_vbi(size, WRITE_POS(frag)), frag)
1446
1447	// MQTT Variable Header
1448	PACK_2B_INT(&trx_buf->hdr_buffer, packet_id, frag)
1449
1450	if (reason_code) {
1451	// MQTT Variable Header
1452	// [MQTT-3.14.2.1] PacketID
1453	*WRITE_POS(frag) = reason_code;
1454	DATA_ADVANCE(&trx_buf->hdr_buffer, 1, frag)
1455	}
1456
1457	trx_buf->hdr_buffer.tail_frag->flags \|= BUFFER_FRAG_MQTT_PACKET_TAIL;
1458	transaction_buffer_transaction_commit(trx_buf)
1459	return MQTT_NG_MSGGEN_OK;
1460	fail_rollback:
1461	transaction_buffer_transaction_rollback(trx_buf, frag);
1462	return MQTT_NG_MSGGEN_BUFFER_OOM;
1463	}
1464
1465	static int mqtt_ng_puback(struct mqtt_ng_client *client, uint16_t packet_id, uint8_t reason_code)
1466	{
1467	return TRY_GENERATE_MESSAGE(mqtt_generate_puback, packet_id, reason_code);
1468	}
1469
1470	int mqtt_ng_ping(struct mqtt_ng_client *client)
1471	{
1472	client->ping_pending = 1;
1473	return MQTT_NG_MSGGEN_OK;
1474	}
1475
1476	#define MQTT_NG_CLIENT_NEED_MORE_BYTES 0x10
1477	#define MQTT_NG_CLIENT_MQTT_PACKET_DONE 0x11
1478	#define MQTT_NG_CLIENT_PARSE_DONE 0x12
1479	#define MQTT_NG_CLIENT_WANT_WRITE 0x13
1480	#define MQTT_NG_CLIENT_OK_CALL_AGAIN 0
1481	#define MQTT_NG_CLIENT_PROTOCOL_ERROR (-1)
1482	#define MQTT_NG_CLIENT_SERVER_RETURNED_ERROR (-2)
1483	#define MQTT_NG_CLIENT_NOT_IMPL_YET (-3)
1484	#define MQTT_NG_CLIENT_INTERNAL_ERROR (-5)
1485
1486	#define BUF_READ_CHECK_AT_LEAST(buf, x) \
1487	if (rbuf_bytes_available(buf) < (x)) \
1488	return MQTT_NG_CLIENT_NEED_MORE_BYTES;
1489
1490	#define vbi_parser_reset_ctx(ctx) memset(ctx, 0, sizeof(struct mqtt_vbi_parser_ctx))
1491
1492	static int vbi_parser_parse(struct mqtt_vbi_parser_ctx *ctx, rbuf_t data)
1493	{
1494	if (ctx->bytes > MQTT_VBI_MAXBYTES - 1) {
1495	nd_log(NDLS_DAEMON, NDLP_ERR, "MQTT Variable Byte Integer can't be longer than %d bytes", MQTT_VBI_MAXBYTES);
1496	return MQTT_NG_CLIENT_PROTOCOL_ERROR;
1497	}
1498	if (!ctx->bytes \|\| ctx->data[ctx->bytes-1] & MQTT_VBI_CONTINUATION_FLAG) {
1499	BUF_READ_CHECK_AT_LEAST(data, 1)
1500	ctx->bytes++;
1501	rbuf_pop(data, &ctx->data[ctx->bytes-1], 1);
1502	if ( ctx->data[ctx->bytes-1] & MQTT_VBI_CONTINUATION_FLAG )
1503	return MQTT_NG_CLIENT_OK_CALL_AGAIN;
1504	}
1505
1506	if (mqtt_vbi_to_uint32(ctx->data, &ctx->result)) {
1507	nd_log(NDLS_DAEMON, NDLP_ERR, "MQTT Variable Byte Integer failed to be parsed.");
1508	return MQTT_NG_CLIENT_PROTOCOL_ERROR;
1509	}
1510
1511	return MQTT_NG_CLIENT_PARSE_DONE;
1512	}
1513
1514	static void mqtt_properties_parser_ctx_reset(struct mqtt_properties_parser_ctx *ctx)
1515	{
1516	ctx->state = PROPERTIES_LENGTH;
1517	while (ctx->head) {
1518	struct mqtt_property *f = ctx->head;
1519	ctx->head = ctx->head->next;
1520	if (f->type == MQTT_TYPE_STR \|\| f->type == MQTT_TYPE_STR_PAIR)
1521	freez(f->data.strings[0]);
1522	if (f->type == MQTT_TYPE_STR_PAIR)
1523	freez(f->data.strings[1]);
1524	if (f->type == MQTT_TYPE_BIN)
1525	freez(f->data.bindata);
1526	freez(f);
1527	}
1528	ctx->tail = NULL;
1529	ctx->properties_length = 0;
1530	ctx->bytes_consumed = 0;
1531	vbi_parser_reset_ctx(&ctx->vbi_parser_ctx);
1532	}
1533
1534	struct mqtt_property_type {
1535	uint8_t id;
1536	enum mqtt_datatype datatype;
1537	const char* name;
1538	};
1539
1540	const struct mqtt_property_type mqtt_property_types[] = {
1541	{ .id = MQTT_PROP_TOPIC_ALIAS, .name = MQTT_PROP_TOPIC_ALIAS_NAME, .datatype = MQTT_TYPE_UINT_16 },
1542
1543	{ .id = MQTT_PROP_PAYLOAD_FMT_INDICATOR, .name = MQTT_PROP_PAYLOAD_FMT_INDICATOR_NAME, .datatype = MQTT_TYPE_UINT_8 },
1544	{ .id = MQTT_PROP_MSG_EXPIRY_INTERVAL, .name = MQTT_PROP_MSG_EXPIRY_INTERVAL_NAME, .datatype = MQTT_TYPE_UINT_32 },
1545	{ .id = MQTT_PROP_CONTENT_TYPE, .name = MQTT_PROP_CONTENT_TYPE_NAME, .datatype = MQTT_TYPE_STR },
1546	{ .id = MQTT_PROP_RESPONSE_TOPIC, .name = MQTT_PROP_RESPONSE_TOPIC_NAME, .datatype = MQTT_TYPE_STR },
1547	{ .id = MQTT_PROP_CORRELATION_DATA, .name = MQTT_PROP_CORRELATION_DATA_NAME, .datatype = MQTT_TYPE_BIN },
1548	{ .id = MQTT_PROP_SUB_IDENTIFIER, .name = MQTT_PROP_SUB_IDENTIFIER_NAME, .datatype = MQTT_TYPE_VBI },
1549	{ .id = MQTT_PROP_SESSION_EXPIRY_INTERVAL, .name = MQTT_PROP_SESSION_EXPIRY_INTERVAL_NAME, .datatype = MQTT_TYPE_UINT_32 },
1550	{ .id = MQTT_PROP_ASSIGNED_CLIENT_ID, .name = MQTT_PROP_ASSIGNED_CLIENT_ID_NAME, .datatype = MQTT_TYPE_STR },
1551	{ .id = MQTT_PROP_SERVER_KEEP_ALIVE, .name = MQTT_PROP_SERVER_KEEP_ALIVE_NAME, .datatype = MQTT_TYPE_UINT_16 },
1552	{ .id = MQTT_PROP_AUTH_METHOD, .name = MQTT_PROP_AUTH_METHOD_NAME, .datatype = MQTT_TYPE_STR },
1553	{ .id = MQTT_PROP_AUTH_DATA, .name = MQTT_PROP_AUTH_DATA_NAME, .datatype = MQTT_TYPE_BIN },
1554	{ .id = MQTT_PROP_REQ_PROBLEM_INFO, .name = MQTT_PROP_REQ_PROBLEM_INFO_NAME, .datatype = MQTT_TYPE_UINT_8 },
1555	{ .id = MQTT_PROP_WILL_DELAY_INTERVAL, .name = MQTT_PROP_WIIL_DELAY_INTERVAL_NAME, .datatype = MQTT_TYPE_UINT_32 },
1556	{ .id = MQTT_PROP_REQ_RESP_INFORMATION, .name = MQTT_PROP_REQ_RESP_INFORMATION_NAME, .datatype = MQTT_TYPE_UINT_8 },
1557	{ .id = MQTT_PROP_RESP_INFORMATION, .name = MQTT_PROP_RESP_INFORMATION_NAME, .datatype = MQTT_TYPE_STR },
1558	{ .id = MQTT_PROP_SERVER_REF, .name = MQTT_PROP_SERVER_REF_NAME, .datatype = MQTT_TYPE_STR },
1559	{ .id = MQTT_PROP_REASON_STR, .name = MQTT_PROP_REASON_STR_NAME, .datatype = MQTT_TYPE_STR },
1560	{ .id = MQTT_PROP_RECEIVE_MAX, .name = MQTT_PROP_RECEIVE_MAX_NAME, .datatype = MQTT_TYPE_UINT_16 },
1561	{ .id = MQTT_PROP_TOPIC_ALIAS_MAX, .name = MQTT_PROP_TOPIC_ALIAS_MAX_NAME, .datatype = MQTT_TYPE_UINT_16 },
1562	// MQTT_PROP_TOPIC_ALIAS is first as it is most often used
1563	{ .id = MQTT_PROP_MAX_QOS, .name = MQTT_PROP_MAX_QOS_NAME, .datatype = MQTT_TYPE_UINT_8 },
1564	{ .id = MQTT_PROP_RETAIN_AVAIL, .name = MQTT_PROP_RETAIN_AVAIL_NAME, .datatype = MQTT_TYPE_UINT_8 },
1565	{ .id = MQTT_PROP_USR, .name = MQTT_PROP_USR_NAME, .datatype = MQTT_TYPE_STR_PAIR },
1566	{ .id = MQTT_PROP_MAX_PKT_SIZE, .name = MQTT_PROP_MAX_PKT_SIZE_NAME, .datatype = MQTT_TYPE_UINT_32 },
1567	{ .id = MQTT_PROP_WILDCARD_SUB_AVAIL, .name = MQTT_PROP_WILDCARD_SUB_AVAIL_NAME, .datatype = MQTT_TYPE_UINT_8 },
1568	{ .id = MQTT_PROP_SUB_ID_AVAIL, .name = MQTT_PROP_SUB_ID_AVAIL_NAME, .datatype = MQTT_TYPE_UINT_8 },
1569	{ .id = MQTT_PROP_SHARED_SUB_AVAIL, .name = MQTT_PROP_SHARED_SUB_AVAIL_NAME, .datatype = MQTT_TYPE_UINT_8 },
1570	{ .id = 0, .name = NULL, .datatype = MQTT_TYPE_UNKNOWN }
1571	};
1572
1573	static int get_property_type_by_id(uint8_t property_id) {
1574	for (int i = 0; mqtt_property_types[i].datatype != MQTT_TYPE_UNKNOWN; i++) {
1575	if (mqtt_property_types[i].id == property_id)
1576	return mqtt_property_types[i].datatype;
1577	}
1578	return MQTT_TYPE_UNKNOWN;
1579	}
1580
1581	struct mqtt_property get_property_by_id(struct mqtt_property props, uint8_t property_id)
1582	{
1583	while (props) {
1584	if (props->id == property_id) {
1585	return props;
1586	}
1587	props = props->next;
1588	}
1589	return NULL;
1590	}
1591
1592	// Parses [MQTT-2.2.2]
1593	static int parse_properties_array(struct mqtt_properties_parser_ctx *ctx, rbuf_t data)
1594	{
1595	int rc;
1596	switch (ctx->state) {
1597	case PROPERTIES_LENGTH:
1598	rc = vbi_parser_parse(&ctx->vbi_parser_ctx, data);
1599	if (rc == MQTT_NG_CLIENT_PARSE_DONE) {
1600	ctx->properties_length = ctx->vbi_parser_ctx.result;
1601	ctx->bytes_consumed += ctx->vbi_parser_ctx.bytes;
1602	ctx->vbi_length = ctx->vbi_parser_ctx.bytes;
1603	if (!ctx->properties_length)
1604	return MQTT_NG_CLIENT_PARSE_DONE;
1605	ctx->state = PROPERTY_CREATE;
1606	break;
1607	}
1608	return rc;
1609	case PROPERTY_CREATE:
1610	BUF_READ_CHECK_AT_LEAST(data, 1)
1611	struct mqtt_property *prop = callocz(1, sizeof(struct mqtt_property));
1612	if (ctx->head == NULL) {
1613	ctx->head = prop;
1614	ctx->tail = prop;
1615	} else {
1616	ctx->tail->next = prop;
1617	ctx->tail = ctx->tail->next;
1618	}
1619	ctx->state = PROPERTY_ID;
1620	/* FALLTHROUGH */
1621	case PROPERTY_ID:
1622	rbuf_pop(data, (char*)&ctx->tail->id, 1);
1623	ctx->bytes_consumed += 1;
1624	ctx->tail->type = get_property_type_by_id(ctx->tail->id);
1625	switch (ctx->tail->type) {
1626	case MQTT_TYPE_UINT_16:
1627	ctx->state = PROPERTY_TYPE_UINT16;
1628	break;
1629	case MQTT_TYPE_UINT_32:
1630	ctx->state = PROPERTY_TYPE_UINT32;
1631	break;
1632	case MQTT_TYPE_UINT_8:
1633	ctx->state = PROPERTY_TYPE_UINT8;
1634	break;
1635	case MQTT_TYPE_VBI:
1636	ctx->state = PROPERTY_TYPE_VBI;
1637	vbi_parser_reset_ctx(&ctx->vbi_parser_ctx);
1638	break;
1639	case MQTT_TYPE_STR:
1640	case MQTT_TYPE_STR_PAIR:
1641	ctx->str_idx = 0;
1642	/* FALLTHROUGH */
1643	case MQTT_TYPE_BIN:
1644	ctx->state = PROPERTY_TYPE_STR_BIN_LEN;
1645	break;
1646	default:
1647	nd_log(NDLS_DAEMON, NDLP_ERR, "Unsupported property type %d for property id %d.", (int)ctx->tail->type, (int)ctx->tail->id);
1648	return MQTT_NG_CLIENT_PROTOCOL_ERROR;
1649	}
1650	break;
1651	case PROPERTY_TYPE_STR_BIN_LEN:
1652	BUF_READ_CHECK_AT_LEAST(data, sizeof(uint16_t))
1653	rbuf_pop(data, (char*)&ctx->tail->bindata_len, sizeof(uint16_t));
1654	ctx->tail->bindata_len = be16toh(ctx->tail->bindata_len);
1655	ctx->bytes_consumed += 2;
1656	switch (ctx->tail->type) {
1657	case MQTT_TYPE_BIN:
1658	ctx->state = PROPERTY_TYPE_BIN;
1659	break;
1660	case MQTT_TYPE_STR:
1661	case MQTT_TYPE_STR_PAIR:
1662	ctx->state = PROPERTY_TYPE_STR;
1663	break;
1664	default:
1665	nd_log(NDLS_DAEMON, NDLP_ERR, "Unexpected datatype in PROPERTY_TYPE_STR_BIN_LEN %d", (int)ctx->tail->type);
1666	return MQTT_NG_CLIENT_INTERNAL_ERROR;
1667	}
1668	break;
1669	case PROPERTY_TYPE_STR:
1670	BUF_READ_CHECK_AT_LEAST(data, ctx->tail->bindata_len)
1671	ctx->tail->data.strings[ctx->str_idx] = mallocz(ctx->tail->bindata_len + 1);
1672	rbuf_pop(data, ctx->tail->data.strings[ctx->str_idx], ctx->tail->bindata_len);
1673	ctx->tail->data.strings[ctx->str_idx][ctx->tail->bindata_len] = 0;
1674	ctx->str_idx++;
1675	ctx->bytes_consumed += ctx->tail->bindata_len;
1676	if (ctx->tail->type == MQTT_TYPE_STR_PAIR && ctx->str_idx < 2) {
1677	ctx->state = PROPERTY_TYPE_STR_BIN_LEN;
1678	break;
1679	}
1680	ctx->state = PROPERTY_NEXT;
1681	break;
1682	case PROPERTY_TYPE_BIN:
1683	BUF_READ_CHECK_AT_LEAST(data, ctx->tail->bindata_len)
1684	ctx->tail->data.bindata = mallocz(ctx->tail->bindata_len);
1685	rbuf_pop(data, ctx->tail->data.bindata, ctx->tail->bindata_len);
1686	ctx->bytes_consumed += ctx->tail->bindata_len;
1687	ctx->state = PROPERTY_NEXT;
1688	break;
1689	case PROPERTY_TYPE_VBI:
1690	rc = vbi_parser_parse(&ctx->vbi_parser_ctx, data);
1691	if (rc == MQTT_NG_CLIENT_PARSE_DONE) {
1692	ctx->tail->data.uint32 = ctx->vbi_parser_ctx.result;
1693	ctx->bytes_consumed += ctx->vbi_parser_ctx.bytes;
1694	ctx->state = PROPERTY_NEXT;
1695	break;
1696	}
1697	return rc;
1698	case PROPERTY_TYPE_UINT8:
1699	BUF_READ_CHECK_AT_LEAST(data, sizeof(uint8_t))
1700	rbuf_pop(data, (char*)&ctx->tail->data.uint8, sizeof(uint8_t));
1701	ctx->bytes_consumed += sizeof(uint8_t);
1702	ctx->state = PROPERTY_NEXT;
1703	break;
1704	case PROPERTY_TYPE_UINT32:
1705	BUF_READ_CHECK_AT_LEAST(data, sizeof(uint32_t))
1706	rbuf_pop(data, (char*)&ctx->tail->data.uint32, sizeof(uint32_t));
1707	ctx->tail->data.uint32 = be32toh(ctx->tail->data.uint32);
1708	ctx->bytes_consumed += sizeof(uint32_t);
1709	ctx->state = PROPERTY_NEXT;
1710	break;
1711	case PROPERTY_TYPE_UINT16:
1712	BUF_READ_CHECK_AT_LEAST(data, sizeof(uint16_t))
1713	rbuf_pop(data, (char*)&ctx->tail->data.uint16, sizeof(uint16_t));
1714	ctx->tail->data.uint16 = be16toh(ctx->tail->data.uint16);
1715	ctx->bytes_consumed += sizeof(uint16_t);
1716	ctx->state = PROPERTY_NEXT;
1717	/* FALLTHROUGH */
1718	case PROPERTY_NEXT:
1719	if (ctx->properties_length > ctx->bytes_consumed - ctx->vbi_length) {
1720	ctx->state = PROPERTY_CREATE;
1721	break;
1722	} else
1723	return MQTT_NG_CLIENT_PARSE_DONE;
1724	}
1725	return MQTT_NG_CLIENT_OK_CALL_AGAIN;
1726	}
1727
1728	static int parse_connack_varhdr(struct mqtt_ng_client *client)
1729	{
1730	struct mqtt_ng_parser *parser = &client->parser;
1731	switch (parser->varhdr_state) {
1732	case MQTT_PARSE_VARHDR_INITIAL:
1733	BUF_READ_CHECK_AT_LEAST(parser->received_data, 2)
1734	rbuf_pop(parser->received_data, (char*)&parser->mqtt_packet.connack.flags, 1);
1735	rbuf_pop(parser->received_data, (char*)&parser->mqtt_packet.connack.reason_code, 1);
1736	parser->varhdr_state = MQTT_PARSE_VARHDR_PROPS;
1737	mqtt_properties_parser_ctx_reset(&parser->properties_parser);
1738	break;
1739	case MQTT_PARSE_VARHDR_PROPS:
1740	return parse_properties_array(&parser->properties_parser, parser->received_data);
1741	default:
1742	nd_log(NDLS_DAEMON, NDLP_ERR, "invalid state for connack varhdr parser");
1743	return MQTT_NG_CLIENT_INTERNAL_ERROR;
1744	}
1745	return MQTT_NG_CLIENT_OK_CALL_AGAIN;
1746	}
1747
1748	static int parse_disconnect_varhdr(struct mqtt_ng_client *client)
1749	{
1750	struct mqtt_ng_parser *parser = &client->parser;
1751	switch (parser->varhdr_state) {
1752	case MQTT_PARSE_VARHDR_INITIAL:
1753	if (!parser->mqtt_fixed_hdr_remaining_length) {
1754	// [MQTT-3.14.2.1] if reason code omitted act same as == 0
1755	parser->mqtt_packet.disconnect.reason_code = 0;
1756	return MQTT_NG_CLIENT_PARSE_DONE;
1757	}
1758	BUF_READ_CHECK_AT_LEAST(parser->received_data, 1)
1759	rbuf_pop(parser->received_data, (char*)&parser->mqtt_packet.disconnect.reason_code, 1);
1760	if (parser->mqtt_fixed_hdr_remaining_length == 1)
1761	return MQTT_NG_CLIENT_PARSE_DONE;
1762	parser->varhdr_state = MQTT_PARSE_VARHDR_PROPS;
1763	mqtt_properties_parser_ctx_reset(&parser->properties_parser);
1764	break;
1765	case MQTT_PARSE_VARHDR_PROPS:
1766	return parse_properties_array(&parser->properties_parser, parser->received_data);
1767	default:
1768	nd_log(NDLS_DAEMON, NDLP_ERR, "invalid state for connack varhdr parser");
1769	return MQTT_NG_CLIENT_INTERNAL_ERROR;
1770	}
1771	return MQTT_NG_CLIENT_OK_CALL_AGAIN;
1772	}
1773
1774	static int parse_puback_varhdr(struct mqtt_ng_client *client)
1775	{
1776	struct mqtt_ng_parser *parser = &client->parser;
1777	switch (parser->varhdr_state) {
1778	case MQTT_PARSE_VARHDR_INITIAL:
1779	BUF_READ_CHECK_AT_LEAST(parser->received_data, 2)
1780	rbuf_pop(parser->received_data, (char*)&parser->mqtt_packet.puback.packet_id, 2);
1781	parser->mqtt_packet.puback.packet_id = be16toh(parser->mqtt_packet.puback.packet_id);
1782	if (parser->mqtt_fixed_hdr_remaining_length < 3) {
1783	// [MQTT-3.4.2.1] if length is not big enough for reason code
1784	// it is omitted and handled same as if it was present and == 0
1785	// initially missed this detail and was wondering WTF is going on (sigh)
1786	parser->mqtt_packet.puback.reason_code = 0;
1787	return MQTT_NG_CLIENT_PARSE_DONE;
1788	}
1789	parser->varhdr_state = MQTT_PARSE_VARHDR_OPTIONAL_REASON_CODE;
1790	/* FALLTHROUGH */
1791	case MQTT_PARSE_VARHDR_OPTIONAL_REASON_CODE:
1792	BUF_READ_CHECK_AT_LEAST(parser->received_data, 1)
1793	rbuf_pop(parser->received_data, (char*)&parser->mqtt_packet.puback.reason_code, 1);
1794	// LOL so in CONNACK you have to have 0 byte to
1795	// signify empty properties list
1796	// but in PUBACK it can be omitted if remaining length doesn't allow it (sigh)
1797	if (parser->mqtt_fixed_hdr_remaining_length < 4)
1798	return MQTT_NG_CLIENT_PARSE_DONE;
1799
1800	parser->varhdr_state = MQTT_PARSE_VARHDR_PROPS;
1801	mqtt_properties_parser_ctx_reset(&parser->properties_parser);
1802	/* FALLTHROUGH */
1803	case MQTT_PARSE_VARHDR_PROPS:
1804	return parse_properties_array(&parser->properties_parser, parser->received_data);
1805	default:
1806	nd_log(NDLS_DAEMON, NDLP_ERR, "invalid state for puback varhdr parser");
1807	return MQTT_NG_CLIENT_INTERNAL_ERROR;
1808	}
1809	return MQTT_NG_CLIENT_OK_CALL_AGAIN;
1810	}
1811
1812	static int parse_suback_varhdr(struct mqtt_ng_client *client)
1813	{
1814	int rc;
1815	size_t avail;
1816	struct mqtt_ng_parser *parser = &client->parser;
1817	struct mqtt_suback *suback = &client->parser.mqtt_packet.suback;
1818	switch (parser->varhdr_state) {
1819	case MQTT_PARSE_VARHDR_INITIAL:
1820	suback->reason_codes = NULL;
1821	BUF_READ_CHECK_AT_LEAST(parser->received_data, 2)
1822	rbuf_pop(parser->received_data, (char*)&suback->packet_id, 2);
1823	suback->packet_id = be16toh(suback->packet_id);
1824	parser->varhdr_state = MQTT_PARSE_VARHDR_PROPS;
1825	parser->mqtt_parsed_len = 2;
1826	mqtt_properties_parser_ctx_reset(&parser->properties_parser);
1827	/* FALLTHROUGH */
1828	case MQTT_PARSE_VARHDR_PROPS:
1829	rc = parse_properties_array(&parser->properties_parser, parser->received_data);
1830	if (rc != MQTT_NG_CLIENT_PARSE_DONE)
1831	return rc;
1832	parser->mqtt_parsed_len += parser->properties_parser.bytes_consumed;
1833	suback->reason_code_count = parser->mqtt_fixed_hdr_remaining_length - parser->mqtt_parsed_len;
1834	suback->reason_codes = callocz(suback->reason_code_count, sizeof(*suback->reason_codes));
1835	suback->reason_codes_pending = suback->reason_code_count;
1836	parser->varhdr_state = MQTT_PARSE_REASONCODES;
1837	/* FALLTHROUGH */
1838	case MQTT_PARSE_REASONCODES:
1839	avail = rbuf_bytes_available(parser->received_data);
1840	if (avail < 1)
1841	return MQTT_NG_CLIENT_NEED_MORE_BYTES;
1842
1843	suback->reason_codes_pending -= rbuf_pop(parser->received_data, (char*)suback->reason_codes, MIN(suback->reason_codes_pending, avail));
1844
1845	if (!suback->reason_codes_pending)
1846	return MQTT_NG_CLIENT_PARSE_DONE;
1847
1848	return MQTT_NG_CLIENT_NEED_MORE_BYTES;
1849	default:
1850	nd_log(NDLS_DAEMON, NDLP_ERR, "invalid state for suback varhdr parser");
1851	return MQTT_NG_CLIENT_INTERNAL_ERROR;
1852	}
1853	return MQTT_NG_CLIENT_OK_CALL_AGAIN;
1854	}
1855
1856	static int parse_publish_varhdr(struct mqtt_ng_client *client)
1857	{
1858	int rc;
1859	struct mqtt_ng_parser *parser = &client->parser;
1860	struct mqtt_publish *publish = &client->parser.mqtt_packet.publish;
1861	switch (parser->varhdr_state) {
1862	case MQTT_PARSE_VARHDR_INITIAL:
1863	BUF_READ_CHECK_AT_LEAST(parser->received_data, 2)
1864	publish->topic = NULL;
1865	publish->qos = ((parser->mqtt_control_packet_type >> 1) & 0x03);
1866	rbuf_pop(parser->received_data, (char*)&publish->topic_len, 2);
1867	publish->topic_len = be16toh(publish->topic_len);
1868	parser->mqtt_parsed_len = 2;
1869	if (!publish->topic_len) {
1870	parser->varhdr_state = MQTT_PARSE_VARHDR_POST_TOPICNAME;
1871	break;
1872	}
1873	publish->topic = callocz(1, publish->topic_len + 1 /* add 0x00 */);
1874	parser->varhdr_state = MQTT_PARSE_VARHDR_TOPICNAME;
1875	/* FALLTHROUGH */
1876	case MQTT_PARSE_VARHDR_TOPICNAME:
1877	// TODO check empty topic can be valid? In which case we have to skip this step
1878	BUF_READ_CHECK_AT_LEAST(parser->received_data, publish->topic_len)
1879	rbuf_pop(parser->received_data, publish->topic, publish->topic_len);
1880	parser->mqtt_parsed_len += publish->topic_len;
1881	parser->varhdr_state = MQTT_PARSE_VARHDR_POST_TOPICNAME;
1882	/* FALLTHROUGH */
1883	case MQTT_PARSE_VARHDR_POST_TOPICNAME:
1884	mqtt_properties_parser_ctx_reset(&parser->properties_parser);
1885	if (!publish->qos) { // PacketID present only for QOS > 0 [MQTT-3.3.2.2]
1886	parser->varhdr_state = MQTT_PARSE_VARHDR_PROPS;
1887	break;
1888	}
1889	parser->varhdr_state = MQTT_PARSE_VARHDR_PACKET_ID;
1890	/* FALLTHROUGH */
1891	case MQTT_PARSE_VARHDR_PACKET_ID:
1892	BUF_READ_CHECK_AT_LEAST(parser->received_data, 2)
1893	rbuf_pop(parser->received_data, (char*)&publish->packet_id, 2);
1894	publish->packet_id = be16toh(publish->packet_id);
1895	parser->varhdr_state = MQTT_PARSE_VARHDR_PROPS;
1896	parser->mqtt_parsed_len += 2;
1897	/* FALLTHROUGH */
1898	case MQTT_PARSE_VARHDR_PROPS:
1899	rc = parse_properties_array(&parser->properties_parser, parser->received_data);
1900	if (rc != MQTT_NG_CLIENT_PARSE_DONE)
1901	return rc;
1902	parser->mqtt_parsed_len += parser->properties_parser.bytes_consumed;
1903	parser->varhdr_state = MQTT_PARSE_PAYLOAD;
1904	/* FALLTHROUGH */
1905	case MQTT_PARSE_PAYLOAD:
1906	if (parser->mqtt_fixed_hdr_remaining_length < parser->mqtt_parsed_len) {
1907	freez(publish->topic);
1908	publish->topic = NULL;
1909	nd_log(NDLS_DAEMON, NDLP_ERR, "Error parsing PUBLISH message");
1910	return MQTT_NG_CLIENT_PROTOCOL_ERROR;
1911	}
1912	publish->data_len = parser->mqtt_fixed_hdr_remaining_length - parser->mqtt_parsed_len;
1913	if (!publish->data_len) {
1914	publish->data = NULL;
1915	return MQTT_NG_CLIENT_PARSE_DONE; // 0 length payload is OK [MQTT-3.3.3]
1916	}
1917	BUF_READ_CHECK_AT_LEAST(parser->received_data, publish->data_len)
1918
1919	publish->data = mallocz(publish->data_len);
1920	rbuf_pop(parser->received_data, publish->data, publish->data_len);
1921	parser->mqtt_parsed_len += publish->data_len;
1922
1923	return MQTT_NG_CLIENT_PARSE_DONE;
1924	default:
1925	nd_log(NDLS_DAEMON, NDLP_ERR, "invalid state for publish varhdr parser");
1926	return MQTT_NG_CLIENT_INTERNAL_ERROR;
1927	}
1928	return MQTT_NG_CLIENT_OK_CALL_AGAIN;
1929	}
1930
1931	// TODO move to separate file, dont send whole client pointer just to be able
1932	// to access LOG context send parser only which should include log
1933	static int parse_data(struct mqtt_ng_client *client)
1934	{
1935	int rc;
1936	struct mqtt_ng_parser *parser = &client->parser;
1937	switch(parser->state) {
1938	case MQTT_PARSE_FIXED_HEADER_PACKET_TYPE:
1939	BUF_READ_CHECK_AT_LEAST(parser->received_data, 1)
1940	rbuf_pop(parser->received_data, (char*)&parser->mqtt_control_packet_type, 1);
1941	vbi_parser_reset_ctx(&parser->vbi_parser);
1942	parser->state = MQTT_PARSE_FIXED_HEADER_LEN;
1943	break;
1944	case MQTT_PARSE_FIXED_HEADER_LEN:
1945	rc = vbi_parser_parse(&parser->vbi_parser, parser->received_data);
1946	if (rc == MQTT_NG_CLIENT_PARSE_DONE) {
1947	parser->mqtt_fixed_hdr_remaining_length = parser->vbi_parser.result;
1948	parser->state = MQTT_PARSE_VARIABLE_HEADER;
1949	parser->varhdr_state = MQTT_PARSE_VARHDR_INITIAL;
1950	break;
1951	}
1952	return rc;
1953	case MQTT_PARSE_VARIABLE_HEADER:
1954	switch (get_control_packet_type(parser->mqtt_control_packet_type)) {
1955	case MQTT_CPT_CONNACK:
1956	rc = parse_connack_varhdr(client);
1957	if (rc == MQTT_NG_CLIENT_PARSE_DONE) {
1958	parser->state = MQTT_PARSE_MQTT_PACKET_DONE;
1959	break;
1960	}
1961	return rc;
1962	case MQTT_CPT_PUBACK:
1963	rc = parse_puback_varhdr(client);
1964	if (rc == MQTT_NG_CLIENT_PARSE_DONE) {
1965	parser->state = MQTT_PARSE_MQTT_PACKET_DONE;
1966	break;
1967	}
1968	return rc;
1969	case MQTT_CPT_SUBACK:
1970	rc = parse_suback_varhdr(client);
1971	if (rc != MQTT_NG_CLIENT_NEED_MORE_BYTES && rc != MQTT_NG_CLIENT_OK_CALL_AGAIN) {
1972	freez(parser->mqtt_packet.suback.reason_codes);
1973	}
1974	if (rc == MQTT_NG_CLIENT_PARSE_DONE) {
1975	parser->state = MQTT_PARSE_MQTT_PACKET_DONE;
1976	break;
1977	}
1978	return rc;
1979	case MQTT_CPT_PUBLISH:
1980	rc = parse_publish_varhdr(client);
1981	if (rc == MQTT_NG_CLIENT_PARSE_DONE) {
1982	parser->state = MQTT_PARSE_MQTT_PACKET_DONE;
1983	break;
1984	}
1985	return rc;
1986	case MQTT_CPT_PINGRESP:
1987	if (parser->mqtt_fixed_hdr_remaining_length) {
1988	nd_log(NDLS_DAEMON, NDLP_ERR, "PINGRESP has to be 0 Remaining Length."); // [MQTT-3.13.1]
1989	return MQTT_NG_CLIENT_PROTOCOL_ERROR;
1990	}
1991	parser->state = MQTT_PARSE_MQTT_PACKET_DONE;
1992	ping_timeout = 0;
1993	break;
1994	case MQTT_CPT_DISCONNECT:
1995	rc = parse_disconnect_varhdr(client);
1996	if (rc == MQTT_NG_CLIENT_PARSE_DONE) {
1997	parser->state = MQTT_PARSE_MQTT_PACKET_DONE;
1998	break;
1999	}
2000	return rc;
2001	default:
2002	nd_log(NDLS_DAEMON, NDLP_ERR, "Parsing Control Packet Type %" PRIu8 " not implemented yet.", get_control_packet_type(parser->mqtt_control_packet_type));
2003	rbuf_bump_tail(parser->received_data, parser->mqtt_fixed_hdr_remaining_length);
2004	parser->state = MQTT_PARSE_MQTT_PACKET_DONE;
2005	return MQTT_NG_CLIENT_NOT_IMPL_YET;
2006	}
2007	// we could also return MQTT_NG_CLIENT_OK_CALL_AGAIN
2008	// and be called again later
2009	/* FALLTHROUGH */
2010	case MQTT_PARSE_MQTT_PACKET_DONE:
2011	parser->state = MQTT_PARSE_FIXED_HEADER_PACKET_TYPE;
2012	return MQTT_NG_CLIENT_MQTT_PACKET_DONE;
2013	}
2014	return MQTT_NG_CLIENT_OK_CALL_AGAIN;
2015	}
2016
2017	// set next MQTT fragment to send
2018	// return 1 if nothing to send
2019	// return -1 on error
2020	// return 0 if there is fragment set
2021	static int mqtt_ng_next_to_send(struct mqtt_ng_client *client) {
2022	if (client->client_state == MQTT_STATE_CONNECT_PENDING) {
2023	client->main_buffer.sending_frag = client->connect_msg;
2024	client->client_state = MQTT_STATE_CONNECTING;
2025	return 0;
2026	}
2027	if (client->client_state != MQTT_STATE_CONNECTED)
2028	return -1;
2029
2030	struct buffer_fragment *frag = BUFFER_FIRST_FRAG(&client->main_buffer.hdr_buffer);
2031	while (frag) {
2032	// Skip fragments marked for garbage collection - their data may have
2033	// been freed by mark_message_for_gc() after a timeout or ACK
2034	if (frag_is_marked_for_gc(frag)) {
2035	frag = frag->next;
2036	continue;
2037	}
2038	if (frag->sent != frag->len)
2039	break;
2040	frag = frag->next;
2041	}
2042
2043	if ( client->ping_pending && (!frag \|\| (frag->flags & BUFFER_FRAG_MQTT_PACKET_HEAD && frag->sent == 0)) ) {
2044	client->ping_pending = 0;
2045	ping_frag.sent = 0;
2046	ping_frag.sent_monotonic_ut = 0;
2047	client->main_buffer.sending_frag = &ping_frag;
2048	return 0;
2049	}
2050
2051	client->main_buffer.sending_frag = frag;
2052	return frag == NULL ? 1 : 0;
2053	}
2054
2055	// send current fragment
2056	// return 0 if whole remaining length could be sent as a whole
2057	// return -1 if send buffer was filled and
2058	// nothing could be written anymore
2059	// return 1 if last fragment of a message was fully sent
2060	static int send_fragment(struct mqtt_ng_client *client) {
2061	worker_is_busy(WORKER_ACLK_SEND_FRAGMENT);
2062
2063	struct buffer_fragment *frag = client->main_buffer.sending_frag;
2064
2065	// for readability
2066	unsigned char *ptr = frag->data + frag->sent;
2067	size_t bytes = frag->len - frag->sent;
2068
2069	size_t processed = 0;
2070
2071	if (bytes)
2072	processed = client->send_fnc_ptr(client->user_ctx, ptr, bytes);
2073	else
2074	nd_log(NDLS_DAEMON, NDLP_WARNING, "This fragment was fully sent already. This should not happen!");
2075
2076	frag->sent_monotonic_ut = now_monotonic_usec();
2077	frag->sent += processed;
2078	if (frag->sent != frag->len)
2079	return -1;
2080
2081	if (frag->flags & BUFFER_FRAG_MQTT_PACKET_TAIL) {
2082	client->time_of_last_send = time(NULL);
2083	if (client->main_buffer.sending_frag != &ping_frag)
2084	__atomic_fetch_sub(&client->stats.tx_messages_queued, 1, __ATOMIC_RELAXED);
2085	__atomic_fetch_add(&client->stats.tx_messages_sent, 1, __ATOMIC_RELAXED);
2086	client->main_buffer.sending_frag = NULL;
2087	return 1;
2088	}
2089
2090	client->main_buffer.sending_frag = frag->next;
2091
2092	return 0;
2093	}
2094
2095	// attempt sending all fragments of current single MQTT packet
2096	static int send_all_message_fragments(struct mqtt_ng_client *client) {
2097	int rc;
2098	while ( !(rc = send_fragment(client)) );
2099	return rc;
2100	}
2101
2102	static void try_send_all(struct mqtt_ng_client *client) {
2103	do {
2104	if (client->main_buffer.sending_frag == NULL && mqtt_ng_next_to_send(client))
2105	return;
2106	} while(send_all_message_fragments(client) >= 0);
2107	}
2108
2109	int handle_incoming_traffic(struct mqtt_ng_client *client)
2110	{
2111	int rc;
2112	while ((rc = parse_data(client)) == MQTT_NG_CLIENT_OK_CALL_AGAIN) {
2113	;
2114	}
2115	if (rc != MQTT_NG_CLIENT_MQTT_PACKET_DONE)
2116	return rc;
2117
2118	struct mqtt_publish *pub;
2119	struct mqtt_property *prop;
2120	__atomic_fetch_add(&client->stats.rx_messages_rcvd, 1, __ATOMIC_RELAXED);
2121
2122	uint8_t ctrl_packet_type = get_control_packet_type(client->parser.mqtt_control_packet_type);
2123	switch (ctrl_packet_type) {
2124	case MQTT_CPT_CONNACK:
2125	worker_is_busy(WORKER_ACLK_CPT_CONNACK);
2126
2127	LOCK_HDR_BUFFER(&client->main_buffer);
2128	// Invalidate sending_frag if it points to any fragment in the CONNECT message
2129	if (sending_frag_in_message(&client->main_buffer, client->connect_msg))
2130	client->main_buffer.sending_frag = NULL;
2131	mark_message_for_gc(client->connect_msg);
2132	UNLOCK_HDR_BUFFER(&client->main_buffer);
2133
2134	client->connect_msg = NULL;
2135
2136	if (client->client_state != MQTT_STATE_CONNECTING) {
2137	nd_log(NDLS_DAEMON, NDLP_ERR, "ACLK: Received unexpected CONNACK");
2138	client->client_state = MQTT_STATE_ERROR;
2139	return MQTT_NG_CLIENT_PROTOCOL_ERROR;
2140	}
2141
2142	if ((prop = get_property_by_id(client->parser.properties_parser.head, MQTT_PROP_MAX_PKT_SIZE)) != NULL) {
2143	nd_log(NDLS_DAEMON, NDLP_INFO, "ACLK: MQTT server limits message size to %" PRIu32, prop->data.uint32);
2144	client->max_msg_size = prop->data.uint32;
2145	}
2146
2147	if (client->connack_callback)
2148	client->connack_callback(client->user_ctx, client->parser.mqtt_packet.connack.reason_code);
2149	if (!client->parser.mqtt_packet.connack.reason_code) {
2150	nd_log(NDLS_DAEMON, NDLP_INFO, "ACLK: MQTT Connection Accepted By Server");
2151	client->client_state = MQTT_STATE_CONNECTED;
2152	break;
2153	}
2154	client->client_state = MQTT_STATE_ERROR;
2155	return MQTT_NG_CLIENT_SERVER_RETURNED_ERROR;
2156
2157	case MQTT_CPT_PUBACK:
2158	worker_is_busy(WORKER_ACLK_CPT_PUBACK);
2159
2160	if (mark_packet_acked(client, client->parser.mqtt_packet.puback.packet_id))
2161	return MQTT_NG_CLIENT_PROTOCOL_ERROR;
2162	if (client->puback_callback)
2163	client->puback_callback(client->parser.mqtt_packet.puback.packet_id);
2164	break;
2165
2166	case MQTT_CPT_PINGRESP:
2167	worker_is_busy(WORKER_ACLK_CPT_PINGRESP);
2168	usec_t latency = now_monotonic_usec() - ping_frag.sent_monotonic_ut;
2169	pulse_aclk_sent_message_acked(latency, ping_frag.len);
2170	break;
2171
2172	case MQTT_CPT_SUBACK:
2173	worker_is_busy(WORKER_ACLK_CPT_SUBACK);
2174	if (mark_packet_acked(client, client->parser.mqtt_packet.suback.packet_id))
2175	return MQTT_NG_CLIENT_PROTOCOL_ERROR;
2176	break;
2177
2178	case MQTT_CPT_PUBLISH:
2179	worker_is_busy(WORKER_ACLK_CPT_PUBLISH);
2180	pub = &client->parser.mqtt_packet.publish;
2181
2182	if (pub->qos > 1) {
2183	freez(pub->topic);
2184	freez(pub->data);
2185	return MQTT_NG_CLIENT_NOT_IMPL_YET;
2186	}
2187
2188	if ( pub->qos == 1 && ((rc = mqtt_ng_puback(client, pub->packet_id, 0))) ) {
2189	client->client_state = MQTT_STATE_ERROR;
2190	nd_log(NDLS_DAEMON, NDLP_ERR, "Error generating PUBACK reply for PUBLISH");
2191	return rc;
2192	}
2193
2194	if ( (prop = get_property_by_id(client->parser.properties_parser.head, MQTT_PROP_TOPIC_ALIAS)) != NULL ) {
2195	// Topic Alias property was sent from server
2196	void *topic_ptr;
2197	if (!c_rhash_get_ptr_by_uint64(client->rx_aliases, prop->data.uint8, &topic_ptr)) {
2198	if (pub->topic != NULL) {
2199	nd_log(NDLS_DAEMON, NDLP_ERR, "We do not yet support topic alias reassignment");
2200	return MQTT_NG_CLIENT_NOT_IMPL_YET;
2201	}
2202	pub->topic = topic_ptr;
2203	} else {
2204	if (pub->topic == NULL) {
2205	nd_log(NDLS_DAEMON, NDLP_ERR, "Topic alias with id %d unknown and topic not set by server!", prop->data.uint8);
2206	return MQTT_NG_CLIENT_PROTOCOL_ERROR;
2207	}
2208	c_rhash_insert_uint64_ptr(client->rx_aliases, prop->data.uint8, pub->topic);
2209	}
2210	}
2211
2212	if (client->msg_callback) {
2213	worker_is_busy(WORKER_ACLK_MSG_CALLBACK);
2214	client->msg_callback(pub->topic, pub->data, pub->data_len, pub->qos);
2215	}
2216
2217	// in case we have property topic alias and we have topic we take over the string
2218	// and add pointer to it into topic alias list
2219	if (prop == NULL)
2220	freez(pub->topic);
2221	freez(pub->data);
2222	return MQTT_NG_CLIENT_WANT_WRITE;
2223
2224	case MQTT_CPT_DISCONNECT:
2225	worker_is_busy(WORKER_ACLK_CPT_DISCONNECT);
2226	nd_log(NDLS_DAEMON, NDLP_INFO, "Got MQTT DISCONNECT control packet from server. Reason code: %d", (int)client->parser.mqtt_packet.disconnect.reason_code);
2227	client->client_state = MQTT_STATE_DISCONNECTED;
2228	break;
2229
2230	default:
2231	worker_is_busy(WORKER_ACLK_CPT_UNKNOWN);
2232	nd_log(NDLS_DAEMON, NDLP_INFO, "Got unknown control packet %u from server", ctrl_packet_type);
2233	break;
2234	}
2235
2236	return rc;
2237	}
2238
2239	#define PACKET_TIMEOUT_REPEAT_CHECK (60)
2240
2241	int mqtt_ng_sync(struct mqtt_ng_client *client)
2242	{
2243	if (client->client_state == MQTT_STATE_RAW \|\| client->client_state == MQTT_STATE_DISCONNECTED)
2244	return 0;
2245
2246	if (client->client_state == MQTT_STATE_ERROR)
2247	return 1;
2248
2249	// Check for packet timeouts and cleanup
2250	static time_t last_maintenance = 0;
2251	if (now_realtime_sec() - last_maintenance >= PACKET_TIMEOUT_REPEAT_CHECK) {
2252	// if check packet returns true then we did max cleanup, possibly there are more packets to cleanup
2253	// so do not update last_maintenance thus forcing check again
2254	if (likely(!check_packet_monitor_list_for_timeouts(client)))
2255	last_maintenance = now_realtime_sec();
2256	}
2257
2258	worker_is_busy(WORKER_ACLK_TRY_SEND_ALL);
2259
2260	LOCK_HDR_BUFFER(&client->main_buffer);
2261	try_send_all(client);
2262	UNLOCK_HDR_BUFFER(&client->main_buffer);
2263
2264	int rc;
2265
2266	worker_is_busy(WORKER_ACLK_HANDLE_INCOMING);
2267	while ((rc = handle_incoming_traffic(client)) != MQTT_NG_CLIENT_NEED_MORE_BYTES) {
2268	if (rc < 0)
2269	break;
2270	if (rc == MQTT_NG_CLIENT_WANT_WRITE) {
2271	worker_is_busy(WORKER_ACLK_TRY_SEND_ALL);
2272
2273	LOCK_HDR_BUFFER(&client->main_buffer);
2274	try_send_all(client);
2275	UNLOCK_HDR_BUFFER(&client->main_buffer);
2276
2277	worker_is_busy(WORKER_ACLK_HANDLE_INCOMING);
2278	}
2279	}
2280
2281	if (rc < 0)
2282	return rc;
2283
2284	return 0;
2285	}
2286
2287	time_t mqtt_ng_last_send_time(struct mqtt_ng_client *client)
2288	{
2289	return client->time_of_last_send;
2290	}
2291
2292	void mqtt_ng_set_max_mem(struct mqtt_ng_client *client, size_t bytes)
2293	{
2294	client->max_mem_bytes = bytes;
2295	}
2296
2297	void mqtt_ng_get_stats(struct mqtt_ng_client client, struct mqtt_ng_stats stats)
2298	{
2299	stats->tx_messages_queued = __atomic_load_n(&client->stats.tx_messages_queued, __ATOMIC_RELAXED);
2300	stats->tx_messages_sent = __atomic_load_n(&client->stats.tx_messages_sent, __ATOMIC_RELAXED);
2301	stats->rx_messages_rcvd = __atomic_load_n(&client->stats.rx_messages_rcvd, __ATOMIC_RELAXED);
2302	stats->packets_waiting_puback = __atomic_load_n(&client->stats.packets_waiting_puback, __ATOMIC_RELAXED);
2303
2304	stats->tx_bytes_queued = 0;
2305	stats->tx_buffer_reclaimable = 0;
2306	stats->max_puback_wait_us = 0;
2307	stats->max_send_queue_wait_us = 0;
2308	stats->max_unsent_wait_us = 0;
2309	stats->max_partial_wait_us = 0;
2310
2311	// First pass: compute buffer usage/queued bytes and max send-queue wait time (unsent messages)
2312	LOCK_HDR_BUFFER(&client->main_buffer);
2313	stats->tx_buffer_used = BUFFER_BYTES_USED(&client->main_buffer.hdr_buffer);
2314	stats->tx_buffer_free = BUFFER_BYTES_AVAILABLE(&client->main_buffer.hdr_buffer);
2315	stats->tx_buffer_size = client->main_buffer.hdr_buffer.size;
2316	struct buffer_fragment *frag = BUFFER_FIRST_FRAG(&client->main_buffer.hdr_buffer);
2317	usec_t now_ut = now_monotonic_usec();
2318	while (frag) {
2319	stats->tx_bytes_queued += frag->len - frag->sent;
2320	if (frag_is_marked_for_gc(frag))
2321	stats->tx_buffer_reclaimable += FRAG_SIZE_IN_BUFFER(frag);
2322
2323	// For HEAD fragments that are not fully sent yet (unsent or partially sent),
2324	// track max send-queue wait time. Prefer the enqueue timestamp; if missing, fall back to first-send.
2325	if ((frag->flags & BUFFER_FRAG_MQTT_PACKET_HEAD) && frag->sent < frag->len) {
2326	usec_t base = frag->enqueued_monotonic_ut ? frag->enqueued_monotonic_ut : frag->sent_monotonic_ut;
2327	if (base) {
2328	usec_t waited = now_ut - base;
2329	uint64_t w = (uint64_t)waited;
2330	if (frag->sent == 0) {
2331	if (w > stats->max_unsent_wait_us) stats->max_unsent_wait_us = w;
2332	} else {
2333	if (w > stats->max_partial_wait_us) stats->max_partial_wait_us = w;
2334	}
2335	if (w > stats->max_send_queue_wait_us) stats->max_send_queue_wait_us = w;
2336	} else {
2337	// Throttled debug if enqueue time is missing on an unsent HEAD
2338	if (frag->sent == 0) {
2339	static time_t last_warn = 0;
2340	time_t now_s = now_monotonic_sec();
2341	if (now_s - last_warn > 60) {
2342	nd_log(NDLS_DAEMON, NDLP_DEBUG, "ACLK: Missing enqueue timestamp on unsent MQTT packet head");
2343	last_warn = now_s;
2344	}
2345	}
2346	}
2347	}
2348
2349	frag = frag->next;
2350	}
2351	UNLOCK_HDR_BUFFER(&client->main_buffer);
2352
2353	// Second pass: compute max PUBACK wait time by correlating HEAD fragments with pending packet IDs
2354	spinlock_lock(&client->pending_packets.spinlock);
2355	LOCK_HDR_BUFFER(&client->main_buffer);
2356	frag = BUFFER_FIRST_FRAG(&client->main_buffer.hdr_buffer);
2357	while (frag) {
2358	if ((frag->flags & BUFFER_FRAG_MQTT_PACKET_HEAD) && frag->packet_id) {
2359	Pvoid_t *Pvalue = JudyLGet(client->pending_packets.JudyL, (Word_t)frag->packet_id, PJE0);
2360	if (Pvalue) {
2361	// message is still pending PUBACK
2362	if (frag->sent_monotonic_ut) {
2363	usec_t waited = now_ut - frag->sent_monotonic_ut;
2364	if ((uint64_t)waited > stats->max_puback_wait_us)
2365	stats->max_puback_wait_us = (uint64_t)waited;
2366	}
2367	}
2368	}
2369	frag = frag->next;
2370	}
2371	UNLOCK_HDR_BUFFER(&client->main_buffer);
2372	spinlock_unlock(&client->pending_packets.spinlock);
2373	}
2374
2375	int mqtt_ng_set_topic_alias(struct mqtt_ng_client client, const char topic)
2376	{
2377	uint16_t idx;
2378	spinlock_lock(&client->tx_topic_aliases.spinlock);
2379
2380	if (client->tx_topic_aliases.idx_assigned >= client->tx_topic_aliases.idx_max) {
2381	spinlock_unlock(&client->tx_topic_aliases.spinlock);
2382	nd_log(NDLS_DAEMON, NDLP_ERR, "Tx topic alias indexes were exhausted (current version of the library doesn't support reassigning yet. Feel free to contribute.");
2383	return 0; //0 is not a valid topic alias
2384	}
2385
2386	struct topic_alias_data *alias;
2387	if (!c_rhash_get_ptr_by_str(client->tx_topic_aliases.stoi_dict, topic, (void**)&alias)) {
2388	// this is not a problem for library but might be helpful to warn user
2389	// as it might indicate bug in their program (but also might be expected)
2390	idx = alias->idx;
2391	spinlock_unlock(&client->tx_topic_aliases.spinlock);
2392	nd_log(NDLS_DAEMON, NDLP_DEBUG, "%s topic \"%s\" already has alias set. Ignoring.", __FUNCTION__, topic);
2393	return idx;
2394	}
2395
2396	alias = mallocz(sizeof(struct topic_alias_data));
2397	idx = ++client->tx_topic_aliases.idx_assigned;
2398	alias->idx = idx;
2399	__atomic_store_n(&alias->usage_count, 0, __ATOMIC_SEQ_CST);
2400
2401	c_rhash_insert_str_ptr(client->tx_topic_aliases.stoi_dict, topic, (void*)alias);
2402
2403	spinlock_unlock(&client->tx_topic_aliases.spinlock);
2404	return idx;
2405	}