![[BACK]](/icons/cvsweb/back.gif){kind=icon}
Return to aitmqtt.c CVS log ![[TXT]](/icons/cvsweb/text.gif){kind=icon}
![[DIR]](/icons/cvsweb/dir.gif){kind=icon}
Up to [ELWIX - Embedded LightWeight unIX -] / libaitmqtt / src|
Return to aitmqtt.c CVS log | Up to [ELWIX - Embedded LightWeight unIX -] / libaitmqtt / src |
1.1 misho 1: /*************************************************************************
2: * (C) 2011 AITNET ltd - Sofia/Bulgaria - <misho@aitbg.com>
3: * by Michael Pounov <misho@openbsd-bg.org>
4: *
5: * $Author: misho $
1.1.1.1.2.6! misho 6: * $Id: aitmqtt.c,v 1.1.1.1.2.5 2012/04/25 13:49:15 misho Exp $
1.1 misho 7: *
8: **************************************************************************
9: The ELWIX and AITNET software is distributed under the following
10: terms:
11:
12: All of the documentation and software included in the ELWIX and AITNET
13: Releases is copyrighted by ELWIX - Sofia/Bulgaria <info@elwix.org>
14:
1.1.1.1.2.3 misho 15: Copyright 2004, 2005, 2006, 2007, 2008, 2009, 2010, 2011, 2012
1.1 misho 16: by Michael Pounov <misho@elwix.org>. All rights reserved.
17:
18: Redistribution and use in source and binary forms, with or without
19: modification, are permitted provided that the following conditions
20: are met:
21: 1. Redistributions of source code must retain the above copyright
22: notice, this list of conditions and the following disclaimer.
23: 2. Redistributions in binary form must reproduce the above copyright
24: notice, this list of conditions and the following disclaimer in the
25: documentation and/or other materials provided with the distribution.
26: 3. All advertising materials mentioning features or use of this software
27: must display the following acknowledgement:
28: This product includes software developed by Michael Pounov <misho@elwix.org>
29: ELWIX - Embedded LightWeight unIX and its contributors.
30: 4. Neither the name of AITNET nor the names of its contributors
31: may be used to endorse or promote products derived from this software
32: without specific prior written permission.
33:
34: THIS SOFTWARE IS PROVIDED BY AITNET AND CONTRIBUTORS ``AS IS'' AND
35: ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
36: IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
37: ARE DISCLAIMED. IN NO EVENT SHALL THE REGENTS OR CONTRIBUTORS BE LIABLE
38: FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
39: DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS
40: OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
41: HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
42: LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY
43: OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
44: SUCH DAMAGE.
45: */
46: #include "global.h"
47:
48:
49: #pragma GCC visibility push(hidden)
50:
51: int mqtt_Errno;
52: char mqtt_Error[STRSIZ];
53:
54: #pragma GCC visibility pop
55:
56: // mqtt_GetErrno() Get error code of last operation
57: inline int
58: mqtt_GetErrno()
59: {
60: return mqtt_Errno;
61: }
62:
63: // mqtt_GetError() Get error text of last operation
64: inline const char *
65: mqtt_GetError()
66: {
67: return mqtt_Error;
68: }
69:
70: // mqtt_SetErr() Set error to variables for internal use!!!
71: inline void
72: mqtt_SetErr(int eno, char *estr, ...)
73: {
74: va_list lst;
75:
76: mqtt_Errno = eno;
77: memset(mqtt_Error, 0, sizeof mqtt_Error);
78: va_start(lst, estr);
79: vsnprintf(mqtt_Error, sizeof mqtt_Error, estr, lst);
80: va_end(lst);
81: }
82:
83: #pragma GCC visibility push(hidden)
1.1.1.1.2.4 misho 84: /* _mqtt_readHEADER() read fixed header from MQTT message */
1.1 misho 85: inline struct mqtthdr *
86: _mqtt_readHEADER(mqtt_msg_t * __restrict buf, u_char cmd, int *bytes, int *len)
87: {
88: struct mqtthdr *hdr;
89:
90: if (!buf || !buf->msg_base || !buf->msg_len)
91: return NULL;
92:
93: hdr = (struct mqtthdr*) buf->msg_base;
94: if (hdr->mqtt_msg.type != cmd) {
95: mqtt_SetErr(EINVAL, "Error:: wrong command #%d should be %d",
96: hdr->mqtt_msg.type, cmd);
97: return NULL;
98: }
99:
100: *len = mqtt_decodeLen(hdr->mqtt_len, bytes);
101: return hdr;
102: }
103: #pragma GCC visibility pop
104:
105:
106: /*
107: * mqtt_msgFree() Free MQTT message
108: *
109: * @msg = Message buffer
110: * @all = !=0 Destroy entire message, if MQTT Message allocated with mqtt_msgAlloc()
111: * return: none
112: */
113: inline void
114: mqtt_msgFree(mqtt_msg_t ** __restrict msg, int all)
115: {
116: if (msg && *msg) {
117: if ((*msg)->msg_base) {
118: free((*msg)->msg_base);
119: (*msg)->msg_base = NULL;
120: }
121: if (all) {
122: free(*msg);
123: *msg = NULL;
124: } else
125: (*msg)->msg_len ^= (*msg)->msg_len;
126: }
127: }
128:
129: /*
130: * mqtt_msgAlloc() Allocate memory for MQTT Message
131: *
132: * @len = >0 Allocate buffer with length
133: * return: NULL error or Message, after use must call mqtt_msgFree() with all!=0
134: */
135: inline mqtt_msg_t *
136: mqtt_msgAlloc(u_short len)
137: {
138: mqtt_msg_t *m = NULL;
139:
140: m = malloc(sizeof(mqtt_msg_t));
141: if (!m) {
142: LOGERR;
143: return NULL;
144: } else
145: memset(m, 0, sizeof(mqtt_msg_t));
146:
147: if (len) {
148: m->msg_len = len;
149: m->msg_base = malloc(m->msg_len);
150: if (!m->msg_base) {
151: LOGERR;
152: free(m);
153: return NULL;
154: } else
155: memset(m->msg_base, 0, m->msg_len);
156: }
157:
158: return m;
159: }
160:
161: /*
162: * mqtt_msgRealloc() Reallocate MQTT message buffer
163: *
164: * @msg = MQTT message
165: * @len = new length
166: * return: -1 error or >-1 old buffer length
167: */
168: inline int
169: mqtt_msgRealloc(mqtt_msg_t * __restrict msg, u_short len)
170: {
171: void *p = NULL;
172: int ret = 0;
173:
174: if (!msg)
175: return -1;
176:
1.1.1.1.2.5 misho 177: if (len <= msg->msg_len)
1.1 misho 178: return len;
179:
180: p = realloc(msg->msg_base, len);
181: if (!p) {
182: LOGERR;
183: return -1;
184: }
185:
186: ret = msg->msg_len;
187: msg->msg_len = len;
188: msg->msg_base = p;
189:
190: return ret;
191: }
192:
193: /*
194: * mqtt_encodeLen() Encode number to MQTT length field
195: *
196: * @num = number for encode
197: * return: -1 error or >-1 length
198: */
199: inline u_int
200: mqtt_encodeLen(u_int num)
201: {
202: register u_int dig, i;
203: u_int ret = 0;
204:
205: if (num > 268435455)
206: return (u_int) -1;
207:
208: for (i = 0; i < sizeof ret && num > 0; i++) {
209: dig = num % 0x80;
210: num /= 0x80;
211: if (num > 0)
212: dig |= 0x80;
213:
214: *((u_char*) &ret + i) = (u_char) dig;
215: }
216:
217: return ret;
218: }
219:
220: /*
221: * mqtt_decodeLen() Decode length from MQTT packet
222: *
223: * @len = length from MQTT header
224: * @n = sizeof bytes, if !=NULL
225: * return: -1 error, >-1 length of message
226: */
227: inline u_int
228: mqtt_decodeLen(void * __restrict len, int * __restrict n)
229: {
230: register u_int i, dig, mul;
231: u_int ret = 0;
232: u_char *p = (u_char*) len;
233:
234: if (!len)
235: return (u_int) -1;
236:
237: for (mul = 1, i = 0; i < sizeof ret; i++, mul *= 0x80) {
238: dig = p[i];
239: ret += (dig & 0x7f) * mul;
240:
241: if (!(dig & 0x80))
242: break;
243: }
244:
245: if (n)
246: *n = (char) (i & 0x7f) + 1;
247: return ret;
248: }
249:
250: /*
251: * mqtt_sizeLen Return sizeof len field
252: *
253: * @len = length
254: * return: -1 error, >-1 sizeof len in bytes
255: */
256: inline char
257: mqtt_sizeLen(u_int len)
258: {
259: register char i;
260: u_char *p = (u_char*) &len;
261:
262: if (len > 0xffffff7f)
263: return -1;
264:
265: for (i = 0; i < sizeof len; i++)
266: if (!(*(p + i) & 0x80))
267: break;
268:
269: return ++i;
270: }
271:
272: /*
273: * mqtt_str2sub Create MQTT subscribe variable from string(s)
274: *
275: * @csStr = strings
276: * @strnum = number of strings elements
277: * @qoses = QoS elements applied to subscribe variable,
278: * count of elements must be equal with csStr elements
279: * return: NULL error or != subscribe variables array, must be free after use with mqtt_freeSub()
280: */
281: inline mqtt_subscr_t *
282: mqtt_str2sub(const char **csStr, u_short strnum, u_char *qoses)
283: {
284: mqtt_subscr_t *v;
285: register int i, items;
286: const char **strs;
287:
288: if (!csStr)
289: return NULL;
290: for (items = 0, strs = csStr; *strs; items++, strs++)
291: if (strnum && items >= strnum) {
292: items = strnum;
293: break;
294: }
295:
296: if (!(v = malloc((items + 1) * sizeof(mqtt_subscr_t)))) {
297: LOGERR;
298: return NULL;
299: } else
300: memset(v, 0, (items + 1) * sizeof(mqtt_subscr_t));
301:
302: for (i = 0; i < items; i++) {
303: v[i].sub_topic.msg_len = strlen(csStr[i]);
304: v[i].sub_topic.msg_base = (u_char*) strdup(csStr[i]);
305: if (qoses && qoses[i] < MQTT_QOS_RESERVED)
306: v[i].sub_ret = qoses[i];
307: }
308:
309: return v;
310: }
311:
312: /*
313: * mqtt_subFree() Free array from subscribe variables
314: *
315: * @subs = Subscribe variables
316: * return: none
317: */
318: inline void
319: mqtt_subFree(mqtt_subscr_t ** __restrict subs)
320: {
321: mqtt_subscr_t *v;
322:
323: if (!subs)
324: return;
325:
326: for (v = *subs; v->sub_topic.msg_base; v++) {
327: free(v->sub_topic.msg_base);
328: v->sub_topic.msg_base = NULL;
329: v->sub_topic.msg_len = 0;
330:
331: if (v->sub_value.msg_base) {
332: free(v->sub_value.msg_base);
333: v->sub_value.msg_base = NULL;
334: v->sub_value.msg_len = 0;
335: }
336: }
337:
338: free(*subs);
339: *subs = NULL;
340: }
341:
342: /*
343: * mqtt_subAlloc() Create array from subscribe variables
344: *
345: * @num = Number of elements
346: * return: NULL error or subscribe array, after use must call mqtt_subFree()
347: */
348: inline mqtt_subscr_t *
349: mqtt_subAlloc(u_short num)
350: {
351: mqtt_subscr_t *s = NULL;
352:
353: s = malloc((num + 1) * sizeof(mqtt_subscr_t));
354: if (!s) {
355: LOGERR;
356: return NULL;
357: } else
358: memset(s, 0, (num + 1) * sizeof(mqtt_subscr_t));
359:
360: return s;
361: }
362:
363: /*
364: * mqtt_subRealloc() Reallocate array from subscribe variables
365: *
366: * @subs = Subscribe array
367: * @num = Number of elements
368: * return: NULL error or subscribe array, after use must call mqtt_subFree()
369: */
370: inline mqtt_subscr_t *
371: mqtt_subRealloc(mqtt_subscr_t * __restrict subs, u_short num)
372: {
373: mqtt_subscr_t *s = NULL;
374:
375: s = realloc(subs, (num + 1) * sizeof(mqtt_subscr_t));
376: if (!s) {
377: LOGERR;
378: return NULL;
379: }
380:
381: return s;
382: }
1.1.1.1.2.1 misho 383:
384: /*
385: * mqtt_expandTopic() - Expanding topic to regular expression
386: *
387: * @csInput = Input topic
388: * @psRegEx = Output to regular expression
389: * @regexLen = Length of psRegEx
390: * @BOL = Begin of Line, if =0 not added
391: * @EOL = End of Line, if =0 not appended
392: * return: -1 error, 0 nothing expanded or >0 expanded bytes
393: */
394: int
395: mqtt_expandTopic(const char *csInput, char * __restrict psRegEx, int regexLen, u_char BOL, u_char EOL)
396: {
397: int ret = 0;
398: register int i;
399: char *pos, *s;
400: const char reROM[] = "[](){}^$\\-|?.+*";
401:
402: if (!csInput || !psRegEx || regexLen < 1)
403: return -1;
404: else
405: memset(psRegEx, 0, regexLen);
406:
407: /* check # */
408: for (i = 0, pos = (char*) csInput; *pos && i < 2; pos++)
409: if (*pos == '#')
410: i++;
411: if (i == 2) {
412: mqtt_SetErr(EINVAL, "Syntax error, multiple occurrences of #..#");
413: return -1;
414: }
415: if (i == 1 && (pos = strrchr(csInput, '#')))
416: if ((pos != csInput && *(pos - 1) != '/') || *(pos + 1)) {
417: mqtt_SetErr(EINVAL, "Syntax error, bad format of #");
418: return -1;
419: }
420: /* check + */
421: for (pos = (char*) csInput; *pos && (pos = strchr(pos, '+')); pos++)
422: if ((pos != csInput && *(pos - 1) != '/') || (*(pos + 1) && *(pos + 1) != '/')) {
423: mqtt_SetErr(EINVAL, "Syntax error, bad format of +");
424: return -1;
425: }
426:
427: /* BUILD REGEX */
428: s = psRegEx;
429: if (BOL) {
430: *s++ = '^';
431: ret++;
432: }
433: for (pos = (char*) csInput; s < psRegEx + regexLen && *pos; s++, pos++) {
434: if (*pos == '#') {
435: strlcat(s, ".*", regexLen - (s - psRegEx));
436: s++;
437: ret++;
438: break;
439: }
440: if (*pos == '+') {
441: if (*(pos + 1)) {
442: strlcat(s, ".*", regexLen - (s - psRegEx));
443: s++;
444: ret++;
445: continue;
446: } else {
447: strlcat(s, ".*/", regexLen - (s - psRegEx));
448: ret += 2;
449: break;
450: }
451: }
452: for (i = 0; i < sizeof reROM - 1; i++)
453: if (*pos == reROM[i] && regexLen - (s - psRegEx) - 1 > 0) {
454: *s++ = '\\';
455: ret++;
456: break;
457: }
458:
459: *s = *pos;
460: }
461: if (EOL) {
462: strlcat(psRegEx, "$", regexLen);
463: ret++;
464: }
465:
466: return ret;
467: }
1.1.1.1.2.2 misho 468:
469: /*
470: * mqtt_sqlTopic() - Expanding topic to SQL search string
471: *
472: * @csInput = Input topic
473: * @psSQL = Output to SQL search string
474: * @sqlLen = Length of psSQL
475: * return: -1 error, 0 changed bytes
476: */
477: int
478: mqtt_sqlTopic(const char *csInput, char * __restrict psSQL, int sqlLen)
479: {
480: int ret = 0;
481: register int i;
482: char *pos, *s;
483:
484: if (!csInput || !psSQL || sqlLen < 1)
485: return -1;
486: else
487: memset(psSQL, 0, sqlLen);
488:
489: /* check # */
490: for (i = 0, pos = (char*) csInput; *pos && i < 2; pos++)
491: if (*pos == '#')
492: i++;
493: if (i == 2) {
494: mqtt_SetErr(EINVAL, "Syntax error, multiple occurrences of #..#");
495: return -1;
496: }
497: if (i == 1 && (pos = strrchr(csInput, '#')))
498: if ((pos != csInput && *(pos - 1) != '/') || *(pos + 1)) {
499: mqtt_SetErr(EINVAL, "Syntax error, bad format of #");
500: return -1;
501: }
502: /* check + */
503: for (pos = (char*) csInput; *pos && (pos = strchr(pos, '+')); pos++)
504: if ((pos != csInput && *(pos - 1) != '/') || (*(pos + 1) && *(pos + 1) != '/')) {
505: mqtt_SetErr(EINVAL, "Syntax error, bad format of +");
506: return -1;
507: }
508:
509: /* BUILD SEARCH STRING */
510: s = psSQL;
511: for (pos = (char*) csInput; s < psSQL + sqlLen && *pos; s++, pos++) {
512: if (*pos == '#') {
513: *s = '%';
514: s++;
515: ret++;
516: break;
517: }
518: if (*pos == '+') {
519: if (*(pos + 1)) {
520: *s = '%';
521: ret++;
522: continue;
523: } else {
524: strlcat(s, "%/", sqlLen - (s - psSQL));
525: ret += 2;
526: break;
527: }
528: }
529: /*
530: for (i = 0; i < sizeof reROM - 1; i++)
531: if (*pos == reROM[i] && regexLen - (s - psRegEx) - 1 > 0) {
532: *s++ = '\\';
533: ret++;
534: break;
535: }
536: */
537:
538: *s = *pos;
539: }
540:
541: return ret;
542: }
1.1.1.1.2.6! misho 543:
! 544:
! 545: /*
! 546: * mqtt_KeepAlive() - Keep Alive check routine
! 547: *
! 548: * @sock = connected socket
! 549: * @ka = keep alive timeout
! 550: * @tries = tries for receive correct ping response, usually ==1
! 551: * return: -1 error, 0 host is alive, 1 timeout session or 2 broken session
! 552: */
! 553: int
! 554: mqtt_KeepAlive(int sock, u_short ka, u_char tries)
! 555: {
! 556: int ret = 0;
! 557: struct pollfd pfd;
! 558: mqtt_msg_t msg = { NULL, 0 };
! 559:
! 560: if (sock < 3)
! 561: return -1; /* error */
! 562:
! 563: pfd.fd = sock;
! 564: pfd.events = POLLOUT;
! 565: if ((ret = poll(&pfd, 1, ka * 1000)) == -1 ||
! 566: pfd.revents & (POLLERR | POLLHUP | POLLNVAL)) {
! 567: LOGERR;
! 568: return -1; /* error */
! 569: } else if (!ret)
! 570: return 1; /* session is abandoned ... must be disconnect! */
! 571: /* ping request */
! 572: if ((ret = mqtt_msgPINGREQ(&msg)) == -1)
! 573: return -1; /* error */
! 574: if ((ret = send(sock, msg.msg_base, ret, MSG_NOSIGNAL)) == -1) {
! 575: LOGERR;
! 576: goto end;
! 577: }
! 578:
! 579: pfd.events = POLLIN | POLLPRI;
! 580: while (tries--) {
! 581: if ((ret = poll(&pfd, 1, ka * 1000)) == -1 ||
! 582: pfd.revents & (POLLERR | POLLHUP | POLLNVAL)) {
! 583: LOGERR;
! 584: break;
! 585: } else if (!ret) {
! 586: ret = 1; /* session is abandoned ... must be disconnect! */
! 587: continue;
! 588: }
! 589: /* receive & decode packet */
! 590: if ((ret = recv(sock, msg.msg_base, msg.msg_len, 0)) == -1) {
! 591: LOGERR;
! 592: break;
! 593: }
! 594: if (!mqtt_readPINGRESP(&msg)) {
! 595: ret = 0; /* Host is alive */
! 596: break;
! 597: } else
! 598: ret = 2; /* Session is broken ... must be disconnect! */
! 599: }
! 600: end:
! 601: free(msg.msg_base);
! 602: return ret;
! 603: }