mqtt.c@ 104083

Last change on this file since 104083 was 104083, checked in by vboxsync, 9 months ago
curl-8.7.1: Applied and adjusted our curl changes to 8.4.0. bugref:10639
Property svn:eol-style set to `native`
File size: 22.6 KB

Line
1	/***************************************************************************
2	* _ _ ____ _
3	* Project ___\| \| \| \| _ \\| \|
4	* / __\| \| \| \| \|_) \| \|
5	* \| (__\| \|_\| \| _ <\| \|___
6	* \___\|\___/\|_\| \_\_____\|
7	*
8	* Copyright (C) Daniel Stenberg, <daniel@haxx.se>, et al.
9	* Copyright (C) Björn Stenberg, <bjorn@haxx.se>
10	*
11	* This software is licensed as described in the file COPYING, which
12	* you should have received as part of this distribution. The terms
13	* are also available at https://curl.se/docs/copyright.html.
14	*
15	* You may opt to use, copy, modify, merge, publish, distribute and/or sell
16	* copies of the Software, and permit persons to whom the Software is
17	* furnished to do so, under the terms of the COPYING file.
18	*
19	* This software is distributed on an "AS IS" basis, WITHOUT WARRANTY OF ANY
20	* KIND, either express or implied.
21	*
22	* SPDX-License-Identifier: curl
23	*
24	***************************************************************************/
25
26	#include "curl_setup.h"
27
28	#ifndef CURL_DISABLE_MQTT
29
30	#include "urldata.h"
31	#include <curl/curl.h>
32	#include "transfer.h"
33	#include "sendf.h"
34	#include "progress.h"
35	#include "mqtt.h"
36	#include "select.h"
37	#include "strdup.h"
38	#include "url.h"
39	#include "escape.h"
40	#include "warnless.h"
41	#include "curl_printf.h"
42	#include "curl_memory.h"
43	#include "multiif.h"
44	#include "rand.h"
45
46	/* The last #include file should be: */
47	#include "memdebug.h"
48
49	#define MQTT_MSG_CONNECT 0x10
50	#define MQTT_MSG_CONNACK 0x20
51	#define MQTT_MSG_PUBLISH 0x30
52	#define MQTT_MSG_SUBSCRIBE 0x82
53	#define MQTT_MSG_SUBACK 0x90
54	#define MQTT_MSG_DISCONNECT 0xe0
55
56	#define MQTT_CONNACK_LEN 2
57	#define MQTT_SUBACK_LEN 3
58	#define MQTT_CLIENTID_LEN 12 /* "curl0123abcd" */
59
60	/*
61	* Forward declarations.
62	*/
63
64	static CURLcode mqtt_do(struct Curl_easy data, bool done);
65	static CURLcode mqtt_done(struct Curl_easy *data,
66	CURLcode status, bool premature);
67	static CURLcode mqtt_doing(struct Curl_easy data, bool done);
68	static int mqtt_getsock(struct Curl_easy data, struct connectdata conn,
69	curl_socket_t *sock);
70	static CURLcode mqtt_setup_conn(struct Curl_easy *data,
71	struct connectdata *conn);
72
73	/*
74	* MQTT protocol handler.
75	*/
76
77	const struct Curl_handler Curl_handler_mqtt = {
78	"MQTT", /* scheme */
79	mqtt_setup_conn, /* setup_connection */
80	mqtt_do, /* do_it */
81	mqtt_done, /* done */
82	ZERO_NULL, /* do_more */
83	ZERO_NULL, /* connect_it */
84	ZERO_NULL, /* connecting */
85	mqtt_doing, /* doing */
86	ZERO_NULL, /* proto_getsock */
87	mqtt_getsock, /* doing_getsock */
88	ZERO_NULL, /* domore_getsock */
89	ZERO_NULL, /* perform_getsock */
90	ZERO_NULL, /* disconnect */
91	ZERO_NULL, /* write_resp */
92	ZERO_NULL, /* connection_check */
93	ZERO_NULL, /* attach connection */
94	PORT_MQTT, /* defport */
95	CURLPROTO_MQTT, /* protocol */
96	CURLPROTO_MQTT, /* family */
97	PROTOPT_NONE /* flags */
98	};
99
100	static CURLcode mqtt_setup_conn(struct Curl_easy *data,
101	struct connectdata *conn)
102	{
103	/* allocate the HTTP-specific struct for the Curl_easy, only to survive
104	during this request */
105	struct MQTT *mq;
106	(void)conn;
107	DEBUGASSERT(data->req.p.mqtt == NULL);
108
109	mq = calloc(1, sizeof(struct MQTT));
110	if(!mq)
111	return CURLE_OUT_OF_MEMORY;
112	Curl_dyn_init(&mq->recvbuf, DYN_MQTT_RECV);
113	data->req.p.mqtt = mq;
114	return CURLE_OK;
115	}
116
117	static CURLcode mqtt_send(struct Curl_easy *data,
118	char *buf, size_t len)
119	{
120	CURLcode result = CURLE_OK;
121	struct MQTT *mq = data->req.p.mqtt;
122	size_t n;
123	result = Curl_xfer_send(data, buf, len, &n);
124	if(result)
125	return result;
126	Curl_debug(data, CURLINFO_HEADER_OUT, buf, (size_t)n);
127	if(len != n) {
128	size_t nsend = len - n;
129	char *sendleftovers = Curl_memdup(&buf[n], nsend);
130	if(!sendleftovers)
131	return CURLE_OUT_OF_MEMORY;
132	mq->sendleftovers = sendleftovers;
133	mq->nsend = nsend;
134	}
135	else {
136	mq->sendleftovers = NULL;
137	mq->nsend = 0;
138	}
139	return result;
140	}
141
142	/* Generic function called by the multi interface to figure out what socket(s)
143	to wait for and for what actions during the DOING and PROTOCONNECT
144	states */
145	static int mqtt_getsock(struct Curl_easy *data,
146	struct connectdata *conn,
147	curl_socket_t *sock)
148	{
149	(void)data;
150	sock[0] = conn->sock[FIRSTSOCKET];
151	return GETSOCK_READSOCK(FIRSTSOCKET);
152	}
153
154	static int mqtt_encode_len(char *buf, size_t len)
155	{
156	unsigned char encoded;
157	int i;
158
159	for(i = 0; (len > 0) && (i<4); i++) {
160	encoded = len % 0x80;
161	len /= 0x80;
162	if(len)
163	encoded \|= 0x80;
164	buf[i] = encoded;
165	}
166
167	return i;
168	}
169
170	/* add the passwd to the CONNECT packet */
171	static int add_passwd(const char *passwd, const size_t plen,
172	char *pkt, const size_t start, int remain_pos)
173	{
174	/* magic number that need to be set properly */
175	const size_t conn_flags_pos = remain_pos + 8;
176	if(plen > 0xffff)
177	return 1;
178
179	/* set password flag */
180	pkt[conn_flags_pos] \|= 0x40;
181
182	/* length of password provided */
183	pkt[start] = (char)((plen >> 8) & 0xFF);
184	pkt[start + 1] = (char)(plen & 0xFF);
185	memcpy(&pkt[start + 2], passwd, plen);
186	return 0;
187	}
188
189	/* add user to the CONNECT packet */
190	static int add_user(const char *username, const size_t ulen,
191	unsigned char *pkt, const size_t start, int remain_pos)
192	{
193	/* magic number that need to be set properly */
194	const size_t conn_flags_pos = remain_pos + 8;
195	if(ulen > 0xffff)
196	return 1;
197
198	/* set username flag */
199	pkt[conn_flags_pos] \|= 0x80;
200	/* length of username provided */
201	pkt[start] = (unsigned char)((ulen >> 8) & 0xFF);
202	pkt[start + 1] = (unsigned char)(ulen & 0xFF);
203	memcpy(&pkt[start + 2], username, ulen);
204	return 0;
205	}
206
207	/* add client ID to the CONNECT packet */
208	static int add_client_id(const char *client_id, const size_t client_id_len,
209	char *pkt, const size_t start)
210	{
211	if(client_id_len != MQTT_CLIENTID_LEN)
212	return 1;
213	pkt[start] = 0x00;
214	pkt[start + 1] = MQTT_CLIENTID_LEN;
215	memcpy(&pkt[start + 2], client_id, MQTT_CLIENTID_LEN);
216	return 0;
217	}
218
219	/* Set initial values of CONNECT packet */
220	static int init_connpack(char packet, char remain, int remain_pos)
221	{
222	/* Fixed header starts */
223	/* packet type */
224	packet[0] = MQTT_MSG_CONNECT;
225	/* remaining length field */
226	memcpy(&packet[1], remain, remain_pos);
227	/* Fixed header ends */
228
229	/* Variable header starts */
230	/* protocol length */
231	packet[remain_pos + 1] = 0x00;
232	packet[remain_pos + 2] = 0x04;
233	/* protocol name */
234	packet[remain_pos + 3] = 'M';
235	packet[remain_pos + 4] = 'Q';
236	packet[remain_pos + 5] = 'T';
237	packet[remain_pos + 6] = 'T';
238	/* protocol level */
239	packet[remain_pos + 7] = 0x04;
240	/* CONNECT flag: CleanSession */
241	packet[remain_pos + 8] = 0x02;
242	/* keep-alive 0 = disabled */
243	packet[remain_pos + 9] = 0x00;
244	packet[remain_pos + 10] = 0x3c;
245	/* end of variable header */
246	return remain_pos + 10;
247	}
248
249	static CURLcode mqtt_connect(struct Curl_easy *data)
250	{
251	CURLcode result = CURLE_OK;
252	int pos = 0;
253	int rc = 0;
254	/* remain length */
255	int remain_pos = 0;
256	char remain[4] = {0};
257	size_t packetlen = 0;
258	size_t payloadlen = 0;
259	size_t start_user = 0;
260	size_t start_pwd = 0;
261	char client_id[MQTT_CLIENTID_LEN + 1] = "curl";
262	const size_t clen = strlen("curl");
263	char *packet = NULL;
264
265	/* extracting username from request */
266	const char *username = data->state.aptr.user ?
267	data->state.aptr.user : "";
268	const size_t ulen = strlen(username);
269	/* extracting password from request */
270	const char *passwd = data->state.aptr.passwd ?
271	data->state.aptr.passwd : "";
272	const size_t plen = strlen(passwd);
273
274	payloadlen = ulen + plen + MQTT_CLIENTID_LEN + 2;
275	/* The plus 2 are for the MSB and LSB describing the length of the string to
276	* be added on the payload. Refer to spec 1.5.2 and 1.5.4 */
277	if(ulen)
278	payloadlen += 2;
279	if(plen)
280	payloadlen += 2;
281
282	/* getting how much occupy the remain length */
283	remain_pos = mqtt_encode_len(remain, payloadlen + 10);
284
285	/* 10 length of variable header and 1 the first byte of the fixed header */
286	packetlen = payloadlen + 10 + remain_pos + 1;
287
288	/* allocating packet */
289	if(packetlen > 268435455)
290	return CURLE_WEIRD_SERVER_REPLY;
291	packet = malloc(packetlen);
292	if(!packet)
293	return CURLE_OUT_OF_MEMORY;
294	memset(packet, 0, packetlen);
295
296	/* set initial values for the CONNECT packet */
297	pos = init_connpack(packet, remain, remain_pos);
298
299	result = Curl_rand_alnum(data, (unsigned char *)&client_id[clen],
300	MQTT_CLIENTID_LEN - clen + 1);
301	/* add client id */
302	rc = add_client_id(client_id, strlen(client_id), packet, pos + 1);
303	if(rc) {
304	failf(data, "Client ID length mismatched: [%zu]", strlen(client_id));
305	result = CURLE_WEIRD_SERVER_REPLY;
306	goto end;
307	}
308	infof(data, "Using client id '%s'", client_id);
309
310	/* position where starts the user payload */
311	start_user = pos + 3 + MQTT_CLIENTID_LEN;
312	/* position where starts the password payload */
313	start_pwd = start_user + ulen;
314	/* if user name was provided, add it to the packet */
315	if(ulen) {
316	start_pwd += 2;
317
318	rc = add_user(username, ulen,
319	(unsigned char *)packet, start_user, remain_pos);
320	if(rc) {
321	failf(data, "Username is too large: [%zu]", ulen);
322	result = CURLE_WEIRD_SERVER_REPLY;
323	goto end;
324	}
325	}
326
327	/* if passwd was provided, add it to the packet */
328	if(plen) {
329	rc = add_passwd(passwd, plen, packet, start_pwd, remain_pos);
330	if(rc) {
331	failf(data, "Password is too large: [%zu]", plen);
332	result = CURLE_WEIRD_SERVER_REPLY;
333	goto end;
334	}
335	}
336
337	if(!result)
338	result = mqtt_send(data, packet, packetlen);
339
340	end:
341	if(packet)
342	free(packet);
343	Curl_safefree(data->state.aptr.user);
344	Curl_safefree(data->state.aptr.passwd);
345	return result;
346	}
347
348	static CURLcode mqtt_disconnect(struct Curl_easy *data)
349	{
350	CURLcode result = CURLE_OK;
351	struct MQTT *mq = data->req.p.mqtt;
352	result = mqtt_send(data, (char *)"\xe0\x00", 2);
353	Curl_safefree(mq->sendleftovers);
354	Curl_dyn_free(&mq->recvbuf);
355	return result;
356	}
357
358	static CURLcode mqtt_recv_atleast(struct Curl_easy *data, size_t nbytes)
359	{
360	struct MQTT *mq = data->req.p.mqtt;
361	size_t rlen = Curl_dyn_len(&mq->recvbuf);
362	CURLcode result;
363
364	if(rlen < nbytes) {
365	unsigned char readbuf[1024];
366	ssize_t nread;
367
368	DEBUGASSERT(nbytes - rlen < sizeof(readbuf));
369	result = Curl_xfer_recv(data, (char *)readbuf, nbytes - rlen, &nread);
370	if(result)
371	return result;
372	DEBUGASSERT(nread >= 0);
373	if(Curl_dyn_addn(&mq->recvbuf, readbuf, (size_t)nread))
374	return CURLE_OUT_OF_MEMORY;
375	rlen = Curl_dyn_len(&mq->recvbuf);
376	}
377	return (rlen >= nbytes)? CURLE_OK : CURLE_AGAIN;
378	}
379
380	static void mqtt_recv_consume(struct Curl_easy *data, size_t nbytes)
381	{
382	struct MQTT *mq = data->req.p.mqtt;
383	size_t rlen = Curl_dyn_len(&mq->recvbuf);
384	if(rlen <= nbytes)
385	Curl_dyn_reset(&mq->recvbuf);
386	else
387	Curl_dyn_tail(&mq->recvbuf, rlen - nbytes);
388	}
389
390	static CURLcode mqtt_verify_connack(struct Curl_easy *data)
391	{
392	struct MQTT *mq = data->req.p.mqtt;
393	CURLcode result;
394	char *ptr;
395
396	result = mqtt_recv_atleast(data, MQTT_CONNACK_LEN);
397	if(result)
398	goto fail;
399
400	/* verify CONNACK */
401	DEBUGASSERT(Curl_dyn_len(&mq->recvbuf) >= MQTT_CONNACK_LEN);
402	ptr = Curl_dyn_ptr(&mq->recvbuf);
403	Curl_debug(data, CURLINFO_HEADER_IN, ptr, MQTT_CONNACK_LEN);
404
405	if(ptr[0] != 0x00 \|\| ptr[1] != 0x00) {
406	failf(data, "Expected %02x%02x but got %02x%02x",
407	0x00, 0x00, ptr[0], ptr[1]);
408	Curl_dyn_reset(&mq->recvbuf);
409	result = CURLE_WEIRD_SERVER_REPLY;
410	goto fail;
411	}
412	mqtt_recv_consume(data, MQTT_CONNACK_LEN);
413	fail:
414	return result;
415	}
416
417	static CURLcode mqtt_get_topic(struct Curl_easy *data,
418	char *topic, size_t topiclen)
419	{
420	char *path = data->state.up.path;
421	CURLcode result = CURLE_URL_MALFORMAT;
422	if(strlen(path) > 1) {
423	result = Curl_urldecode(path + 1, 0, topic, topiclen, REJECT_NADA);
424	if(!result && (*topiclen > 0xffff)) {
425	failf(data, "Too long MQTT topic");
426	result = CURLE_URL_MALFORMAT;
427	}
428	}
429	else
430	failf(data, "No MQTT topic found. Forgot to URL encode it?");
431
432	return result;
433	}
434
435	static CURLcode mqtt_subscribe(struct Curl_easy *data)
436	{
437	CURLcode result = CURLE_OK;
438	char *topic = NULL;
439	size_t topiclen;
440	unsigned char *packet = NULL;
441	size_t packetlen;
442	char encodedsize[4];
443	size_t n;
444	struct connectdata *conn = data->conn;
445
446	result = mqtt_get_topic(data, &topic, &topiclen);
447	if(result)
448	goto fail;
449
450	conn->proto.mqtt.packetid++;
451
452	packetlen = topiclen + 5; /* packetid + topic (has a two byte length field)
453	+ 2 bytes topic length + QoS byte */
454	n = mqtt_encode_len((char *)encodedsize, packetlen);
455	packetlen += n + 1; /* add one for the control packet type byte */
456
457	packet = malloc(packetlen);
458	if(!packet) {
459	result = CURLE_OUT_OF_MEMORY;
460	goto fail;
461	}
462
463	packet[0] = MQTT_MSG_SUBSCRIBE;
464	memcpy(&packet[1], encodedsize, n);
465	packet[1 + n] = (conn->proto.mqtt.packetid >> 8) & 0xff;
466	packet[2 + n] = conn->proto.mqtt.packetid & 0xff;
467	packet[3 + n] = (topiclen >> 8) & 0xff;
468	packet[4 + n ] = topiclen & 0xff;
469	memcpy(&packet[5 + n], topic, topiclen);
470	packet[5 + n + topiclen] = 0; /* QoS zero */
471
472	result = mqtt_send(data, (char *)packet, packetlen);
473
474	fail:
475	free(topic);
476	free(packet);
477	return result;
478	}
479
480	/*
481	* Called when the first byte was already read.
482	*/
483	static CURLcode mqtt_verify_suback(struct Curl_easy *data)
484	{
485	struct MQTT *mq = data->req.p.mqtt;
486	struct connectdata *conn = data->conn;
487	struct mqtt_conn *mqtt = &conn->proto.mqtt;
488	CURLcode result;
489	char *ptr;
490
491	result = mqtt_recv_atleast(data, MQTT_SUBACK_LEN);
492	if(result)
493	goto fail;
494
495	/* verify SUBACK */
496	DEBUGASSERT(Curl_dyn_len(&mq->recvbuf) >= MQTT_SUBACK_LEN);
497	ptr = Curl_dyn_ptr(&mq->recvbuf);
498	Curl_debug(data, CURLINFO_HEADER_IN, ptr, MQTT_SUBACK_LEN);
499
500	if(((unsigned char)ptr[0]) != ((mqtt->packetid >> 8) & 0xff) \|\|
501	((unsigned char)ptr[1]) != (mqtt->packetid & 0xff) \|\|
502	ptr[2] != 0x00) {
503	Curl_dyn_reset(&mq->recvbuf);
504	result = CURLE_WEIRD_SERVER_REPLY;
505	goto fail;
506	}
507	mqtt_recv_consume(data, MQTT_SUBACK_LEN);
508	fail:
509	return result;
510	}
511
512	static CURLcode mqtt_publish(struct Curl_easy *data)
513	{
514	CURLcode result;
515	char *payload = data->set.postfields;
516	size_t payloadlen;
517	char *topic = NULL;
518	size_t topiclen;
519	unsigned char *pkt = NULL;
520	size_t i = 0;
521	size_t remaininglength;
522	size_t encodelen;
523	char encodedbytes[4];
524	curl_off_t postfieldsize = data->set.postfieldsize;
525
526	if(!payload) {
527	DEBUGF(infof(data, "mqtt_publish without payload, return bad arg"));
528	return CURLE_BAD_FUNCTION_ARGUMENT;
529	}
530	if(postfieldsize < 0)
531	payloadlen = strlen(payload);
532	else
533	payloadlen = (size_t)postfieldsize;
534
535	result = mqtt_get_topic(data, &topic, &topiclen);
536	if(result)
537	goto fail;
538
539	remaininglength = payloadlen + 2 + topiclen;
540	encodelen = mqtt_encode_len(encodedbytes, remaininglength);
541
542	/* add the control byte and the encoded remaining length */
543	pkt = malloc(remaininglength + 1 + encodelen);
544	if(!pkt) {
545	result = CURLE_OUT_OF_MEMORY;
546	goto fail;
547	}
548
549	/* assemble packet */
550	pkt[i++] = MQTT_MSG_PUBLISH;
551	memcpy(&pkt[i], encodedbytes, encodelen);
552	i += encodelen;
553	pkt[i++] = (topiclen >> 8) & 0xff;
554	pkt[i++] = (topiclen & 0xff);
555	memcpy(&pkt[i], topic, topiclen);
556	i += topiclen;
557	memcpy(&pkt[i], payload, payloadlen);
558	i += payloadlen;
559	result = mqtt_send(data, (char *)pkt, i);
560
561	fail:
562	free(pkt);
563	free(topic);
564	return result;
565	}
566
567	static size_t mqtt_decode_len(unsigned char *buf,
568	size_t buflen, size_t *lenbytes)
569	{
570	size_t len = 0;
571	size_t mult = 1;
572	size_t i;
573	unsigned char encoded = 128;
574
575	for(i = 0; (i < buflen) && (encoded & 128); i++) {
576	encoded = buf[i];
577	len += (encoded & 127) * mult;
578	mult *= 128;
579	}
580
581	if(lenbytes)
582	*lenbytes = i;
583
584	return len;
585	}
586
587	#ifdef CURLDEBUG
588	static const char *statenames[]={
589	"MQTT_FIRST",
590	"MQTT_REMAINING_LENGTH",
591	"MQTT_CONNACK",
592	"MQTT_SUBACK",
593	"MQTT_SUBACK_COMING",
594	"MQTT_PUBWAIT",
595	"MQTT_PUB_REMAIN",
596
597	"NOT A STATE"
598	};
599	#endif
600
601	/* The only way to change state */
602	static void mqstate(struct Curl_easy *data,
603	enum mqttstate state,
604	enum mqttstate nextstate) /* used if state == FIRST */
605	{
606	struct connectdata *conn = data->conn;
607	struct mqtt_conn *mqtt = &conn->proto.mqtt;
608	#ifdef CURLDEBUG
609	infof(data, "%s (from %s) (next is %s)",
610	statenames[state],
611	statenames[mqtt->state],
612	(state == MQTT_FIRST)? statenames[nextstate] : "");
613	#endif
614	mqtt->state = state;
615	if(state == MQTT_FIRST)
616	mqtt->nextstate = nextstate;
617	}
618
619
620	static CURLcode mqtt_read_publish(struct Curl_easy data, bool done)
621	{
622	CURLcode result = CURLE_OK;
623	struct connectdata *conn = data->conn;
624	ssize_t nread;
625	size_t remlen;
626	struct mqtt_conn *mqtt = &conn->proto.mqtt;
627	struct MQTT *mq = data->req.p.mqtt;
628	unsigned char packet;
629
630	switch(mqtt->state) {
631	MQTT_SUBACK_COMING:
632	case MQTT_SUBACK_COMING:
633	result = mqtt_verify_suback(data);
634	if(result)
635	break;
636
637	mqstate(data, MQTT_FIRST, MQTT_PUBWAIT);
638	break;
639
640	case MQTT_SUBACK:
641	case MQTT_PUBWAIT:
642	/* we are expecting PUBLISH or SUBACK */
643	packet = mq->firstbyte & 0xf0;
644	if(packet == MQTT_MSG_PUBLISH)
645	mqstate(data, MQTT_PUB_REMAIN, MQTT_NOSTATE);
646	else if(packet == MQTT_MSG_SUBACK) {
647	mqstate(data, MQTT_SUBACK_COMING, MQTT_NOSTATE);
648	goto MQTT_SUBACK_COMING;
649	}
650	else if(packet == MQTT_MSG_DISCONNECT) {
651	infof(data, "Got DISCONNECT");
652	*done = TRUE;
653	goto end;
654	}
655	else {
656	result = CURLE_WEIRD_SERVER_REPLY;
657	goto end;
658	}
659
660	/* -- switched state -- */
661	remlen = mq->remaining_length;
662	infof(data, "Remaining length: %zu bytes", remlen);
663	if(data->set.max_filesize &&
664	(curl_off_t)remlen > data->set.max_filesize) {
665	failf(data, "Maximum file size exceeded");
666	result = CURLE_FILESIZE_EXCEEDED;
667	goto end;
668	}
669	Curl_pgrsSetDownloadSize(data, remlen);
670	data->req.bytecount = 0;
671	data->req.size = remlen;
672	mq->npacket = remlen; /* get this many bytes */
673	FALLTHROUGH();
674	case MQTT_PUB_REMAIN: {
675	/* read rest of packet, but no more. Cap to buffer size */
676	char buffer[4*1024];
677	size_t rest = mq->npacket;
678	if(rest > sizeof(buffer))
679	rest = sizeof(buffer);
680	result = Curl_xfer_recv(data, buffer, rest, &nread);
681	if(result) {
682	if(CURLE_AGAIN == result) {
683	infof(data, "EEEE AAAAGAIN");
684	}
685	goto end;
686	}
687	if(!nread) {
688	infof(data, "server disconnected");
689	result = CURLE_PARTIAL_FILE;
690	goto end;
691	}
692
693	/* if QoS is set, message contains packet id */
694	result = Curl_client_write(data, CLIENTWRITE_BODY, buffer, nread);
695	if(result)
696	goto end;
697
698	mq->npacket -= nread;
699	if(!mq->npacket)
700	/* no more PUBLISH payload, back to subscribe wait state */
701	mqstate(data, MQTT_FIRST, MQTT_PUBWAIT);
702	break;
703	}
704	default:
705	DEBUGASSERT(NULL); /* illegal state */
706	result = CURLE_WEIRD_SERVER_REPLY;
707	goto end;
708	}
709	end:
710	return result;
711	}
712
713	static CURLcode mqtt_do(struct Curl_easy data, bool done)
714	{
715	CURLcode result = CURLE_OK;
716	done = FALSE; / unconditionally */
717
718	result = mqtt_connect(data);
719	if(result) {
720	failf(data, "Error %d sending MQTT CONNECT request", result);
721	return result;
722	}
723	mqstate(data, MQTT_FIRST, MQTT_CONNACK);
724	return CURLE_OK;
725	}
726
727	static CURLcode mqtt_done(struct Curl_easy *data,
728	CURLcode status, bool premature)
729	{
730	struct MQTT *mq = data->req.p.mqtt;
731	(void)status;
732	(void)premature;
733	Curl_safefree(mq->sendleftovers);
734	Curl_dyn_free(&mq->recvbuf);
735	return CURLE_OK;
736	}
737
738	static CURLcode mqtt_doing(struct Curl_easy data, bool done)
739	{
740	CURLcode result = CURLE_OK;
741	struct connectdata *conn = data->conn;
742	struct mqtt_conn *mqtt = &conn->proto.mqtt;
743	struct MQTT *mq = data->req.p.mqtt;
744	ssize_t nread;
745	unsigned char byte;
746
747	*done = FALSE;
748
749	if(mq->nsend) {
750	/* send the remainder of an outgoing packet */
751	char *ptr = mq->sendleftovers;
752	result = mqtt_send(data, mq->sendleftovers, mq->nsend);
753	free(ptr);
754	if(result)
755	return result;
756	}
757
758	infof(data, "mqtt_doing: state [%d]", (int) mqtt->state);
759	switch(mqtt->state) {
760	case MQTT_FIRST:
761	/* Read the initial byte only */
762	result = Curl_xfer_recv(data, (char *)&mq->firstbyte, 1, &nread);
763	if(result)
764	break;
765	else if(!nread) {
766	failf(data, "Connection disconnected");
767	*done = TRUE;
768	result = CURLE_RECV_ERROR;
769	break;
770	}
771	Curl_debug(data, CURLINFO_HEADER_IN, (char *)&mq->firstbyte, 1);
772	/* remember the first byte */
773	mq->npacket = 0;
774	mqstate(data, MQTT_REMAINING_LENGTH, MQTT_NOSTATE);
775	FALLTHROUGH();
776	case MQTT_REMAINING_LENGTH:
777	do {
778	result = Curl_xfer_recv(data, (char *)&byte, 1, &nread);
779	if(!nread)
780	break;
781	Curl_debug(data, CURLINFO_HEADER_IN, (char *)&byte, 1);
782	mq->pkt_hd[mq->npacket++] = byte;
783	} while((byte & 0x80) && (mq->npacket < 4));
784	if(nread && (byte & 0x80))
785	/* MQTT supports up to 127 * 128^0 + 127 * 128^1 + 127 * 128^2 +
786	127 * 128^3 bytes. server tried to send more */
787	result = CURLE_WEIRD_SERVER_REPLY;
788	if(result)
789	break;
790	mq->remaining_length = mqtt_decode_len(mq->pkt_hd, mq->npacket, NULL);
791	mq->npacket = 0;
792	if(mq->remaining_length) {
793	mqstate(data, mqtt->nextstate, MQTT_NOSTATE);
794	break;
795	}
796	mqstate(data, MQTT_FIRST, MQTT_FIRST);
797
798	if(mq->firstbyte == MQTT_MSG_DISCONNECT) {
799	infof(data, "Got DISCONNECT");
800	*done = TRUE;
801	}
802	break;
803	case MQTT_CONNACK:
804	result = mqtt_verify_connack(data);
805	if(result)
806	break;
807
808	if(data->state.httpreq == HTTPREQ_POST) {
809	result = mqtt_publish(data);
810	if(!result) {
811	result = mqtt_disconnect(data);
812	*done = TRUE;
813	}
814	mqtt->nextstate = MQTT_FIRST;
815	}
816	else {
817	result = mqtt_subscribe(data);
818	if(!result) {
819	mqstate(data, MQTT_FIRST, MQTT_SUBACK);
820	}
821	}
822	break;
823
824	case MQTT_SUBACK:
825	case MQTT_PUBWAIT:
826	case MQTT_PUB_REMAIN:
827	result = mqtt_read_publish(data, done);
828	break;
829
830	default:
831	failf(data, "State not handled yet");
832	*done = TRUE;
833	break;
834	}
835
836	if(result == CURLE_AGAIN)
837	result = CURLE_OK;
838	return result;
839	}
840
841	#endif /* CURL_DISABLE_MQTT */

Note: See TracBrowser for help on using the repository browser.

source: vbox/trunk/src/libs/curl-8.7.1/lib/mqtt.c@ 104083

Download in other formats: