![[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.8! misho 6: * $Id: aitmqtt.c,v 1.1.1.1.2.7 2012/04/26 15:26:37 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: /*
1.1.1.1.2.7 misho 273: * mqtt_str2subs Create MQTT subscribe variable from string(s)
1.1 misho 274: *
1.1.1.1.2.7 misho 275: * @csStr = null terminated string array
276: * @strnum = copy at most number of strings elements
1.1 misho 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 *
1.1.1.1.2.7 misho 282: mqtt_str2subs(const char **csStr, u_short strnum, u_char *qoses)
1.1 misho 283: {
284: mqtt_subscr_t *v;
285: register int i, items;
286: const char **strs;
287:
288: if (!csStr)
289: return NULL;
1.1.1.1.2.8! misho 290:
! 291: for (items = 0, strs = csStr;
! 292: (!strnum || (strnum && items < strnum)) && *strs;
! 293: items++, strs++);
1.1 misho 294:
295: if (!(v = malloc((items + 1) * sizeof(mqtt_subscr_t)))) {
296: LOGERR;
297: return NULL;
298: } else
299: memset(v, 0, (items + 1) * sizeof(mqtt_subscr_t));
300:
301: for (i = 0; i < items; i++) {
302: v[i].sub_topic.msg_len = strlen(csStr[i]);
303: v[i].sub_topic.msg_base = (u_char*) strdup(csStr[i]);
304: if (qoses && qoses[i] < MQTT_QOS_RESERVED)
305: v[i].sub_ret = qoses[i];
306: }
307:
308: return v;
309: }
310:
311: /*
312: * mqtt_subFree() Free array from subscribe variables
313: *
314: * @subs = Subscribe variables
315: * return: none
316: */
317: inline void
318: mqtt_subFree(mqtt_subscr_t ** __restrict subs)
319: {
320: mqtt_subscr_t *v;
321:
322: if (!subs)
323: return;
324:
325: for (v = *subs; v->sub_topic.msg_base; v++) {
326: free(v->sub_topic.msg_base);
327: v->sub_topic.msg_base = NULL;
328: v->sub_topic.msg_len = 0;
329:
330: if (v->sub_value.msg_base) {
331: free(v->sub_value.msg_base);
332: v->sub_value.msg_base = NULL;
333: v->sub_value.msg_len = 0;
334: }
335: }
336:
337: free(*subs);
338: *subs = NULL;
339: }
340:
341: /*
342: * mqtt_subAlloc() Create array from subscribe variables
343: *
344: * @num = Number of elements
345: * return: NULL error or subscribe array, after use must call mqtt_subFree()
346: */
347: inline mqtt_subscr_t *
348: mqtt_subAlloc(u_short num)
349: {
350: mqtt_subscr_t *s = NULL;
351:
352: s = malloc((num + 1) * sizeof(mqtt_subscr_t));
353: if (!s) {
354: LOGERR;
355: return NULL;
356: } else
357: memset(s, 0, (num + 1) * sizeof(mqtt_subscr_t));
358:
359: return s;
360: }
361:
362: /*
363: * mqtt_subRealloc() Reallocate array from subscribe variables
364: *
365: * @subs = Subscribe array
366: * @num = Number of elements
367: * return: NULL error or subscribe array, after use must call mqtt_subFree()
368: */
369: inline mqtt_subscr_t *
370: mqtt_subRealloc(mqtt_subscr_t * __restrict subs, u_short num)
371: {
372: mqtt_subscr_t *s = NULL;
373:
374: s = realloc(subs, (num + 1) * sizeof(mqtt_subscr_t));
375: if (!s) {
376: LOGERR;
377: return NULL;
378: }
379:
380: return s;
381: }
1.1.1.1.2.1 misho 382:
383: /*
384: * mqtt_expandTopic() - Expanding topic to regular expression
385: *
386: * @csInput = Input topic
387: * @psRegEx = Output to regular expression
388: * @regexLen = Length of psRegEx
389: * @BOL = Begin of Line, if =0 not added
390: * @EOL = End of Line, if =0 not appended
391: * return: -1 error, 0 nothing expanded or >0 expanded bytes
392: */
393: int
394: mqtt_expandTopic(const char *csInput, char * __restrict psRegEx, int regexLen, u_char BOL, u_char EOL)
395: {
396: int ret = 0;
397: register int i;
398: char *pos, *s;
399: const char reROM[] = "[](){}^$\\-|?.+*";
400:
401: if (!csInput || !psRegEx || regexLen < 1)
402: return -1;
403: else
404: memset(psRegEx, 0, regexLen);
405:
406: /* check # */
407: for (i = 0, pos = (char*) csInput; *pos && i < 2; pos++)
408: if (*pos == '#')
409: i++;
410: if (i == 2) {
411: mqtt_SetErr(EINVAL, "Syntax error, multiple occurrences of #..#");
412: return -1;
413: }
414: if (i == 1 && (pos = strrchr(csInput, '#')))
415: if ((pos != csInput && *(pos - 1) != '/') || *(pos + 1)) {
416: mqtt_SetErr(EINVAL, "Syntax error, bad format of #");
417: return -1;
418: }
419: /* check + */
420: for (pos = (char*) csInput; *pos && (pos = strchr(pos, '+')); pos++)
421: if ((pos != csInput && *(pos - 1) != '/') || (*(pos + 1) && *(pos + 1) != '/')) {
422: mqtt_SetErr(EINVAL, "Syntax error, bad format of +");
423: return -1;
424: }
425:
426: /* BUILD REGEX */
427: s = psRegEx;
428: if (BOL) {
429: *s++ = '^';
430: ret++;
431: }
432: for (pos = (char*) csInput; s < psRegEx + regexLen && *pos; s++, pos++) {
433: if (*pos == '#') {
434: strlcat(s, ".*", regexLen - (s - psRegEx));
435: s++;
436: ret++;
437: break;
438: }
439: if (*pos == '+') {
440: if (*(pos + 1)) {
441: strlcat(s, ".*", regexLen - (s - psRegEx));
442: s++;
443: ret++;
444: continue;
445: } else {
446: strlcat(s, ".*/", regexLen - (s - psRegEx));
447: ret += 2;
448: break;
449: }
450: }
451: for (i = 0; i < sizeof reROM - 1; i++)
452: if (*pos == reROM[i] && regexLen - (s - psRegEx) - 1 > 0) {
453: *s++ = '\\';
454: ret++;
455: break;
456: }
457:
458: *s = *pos;
459: }
460: if (EOL) {
461: strlcat(psRegEx, "$", regexLen);
462: ret++;
463: }
464:
465: return ret;
466: }
1.1.1.1.2.2 misho 467:
468: /*
469: * mqtt_sqlTopic() - Expanding topic to SQL search string
470: *
471: * @csInput = Input topic
472: * @psSQL = Output to SQL search string
473: * @sqlLen = Length of psSQL
474: * return: -1 error, 0 changed bytes
475: */
476: int
477: mqtt_sqlTopic(const char *csInput, char * __restrict psSQL, int sqlLen)
478: {
479: int ret = 0;
480: register int i;
481: char *pos, *s;
482:
483: if (!csInput || !psSQL || sqlLen < 1)
484: return -1;
485: else
486: memset(psSQL, 0, sqlLen);
487:
488: /* check # */
489: for (i = 0, pos = (char*) csInput; *pos && i < 2; pos++)
490: if (*pos == '#')
491: i++;
492: if (i == 2) {
493: mqtt_SetErr(EINVAL, "Syntax error, multiple occurrences of #..#");
494: return -1;
495: }
496: if (i == 1 && (pos = strrchr(csInput, '#')))
497: if ((pos != csInput && *(pos - 1) != '/') || *(pos + 1)) {
498: mqtt_SetErr(EINVAL, "Syntax error, bad format of #");
499: return -1;
500: }
501: /* check + */
502: for (pos = (char*) csInput; *pos && (pos = strchr(pos, '+')); pos++)
503: if ((pos != csInput && *(pos - 1) != '/') || (*(pos + 1) && *(pos + 1) != '/')) {
504: mqtt_SetErr(EINVAL, "Syntax error, bad format of +");
505: return -1;
506: }
507:
508: /* BUILD SEARCH STRING */
509: s = psSQL;
510: for (pos = (char*) csInput; s < psSQL + sqlLen && *pos; s++, pos++) {
511: if (*pos == '#') {
512: *s = '%';
513: s++;
514: ret++;
515: break;
516: }
517: if (*pos == '+') {
518: if (*(pos + 1)) {
519: *s = '%';
520: ret++;
521: continue;
522: } else {
523: strlcat(s, "%/", sqlLen - (s - psSQL));
524: ret += 2;
525: break;
526: }
527: }
528: /*
529: for (i = 0; i < sizeof reROM - 1; i++)
530: if (*pos == reROM[i] && regexLen - (s - psRegEx) - 1 > 0) {
531: *s++ = '\\';
532: ret++;
533: break;
534: }
535: */
536:
537: *s = *pos;
538: }
539:
540: return ret;
541: }
1.1.1.1.2.6 misho 542:
543:
544: /*
545: * mqtt_KeepAlive() - Keep Alive check routine
546: *
547: * @sock = connected socket
548: * @ka = keep alive timeout
549: * @tries = tries for receive correct ping response, usually ==1
550: * return: -1 error, 0 host is alive, 1 timeout session or 2 broken session
551: */
552: int
553: mqtt_KeepAlive(int sock, u_short ka, u_char tries)
554: {
555: int ret = 0;
556: struct pollfd pfd;
557: mqtt_msg_t msg = { NULL, 0 };
558:
559: if (sock < 3)
560: return -1; /* error */
561:
562: pfd.fd = sock;
563: pfd.events = POLLOUT;
564: if ((ret = poll(&pfd, 1, ka * 1000)) == -1 ||
565: pfd.revents & (POLLERR | POLLHUP | POLLNVAL)) {
566: LOGERR;
567: return -1; /* error */
568: } else if (!ret)
569: return 1; /* session is abandoned ... must be disconnect! */
570: /* ping request */
571: if ((ret = mqtt_msgPINGREQ(&msg)) == -1)
572: return -1; /* error */
573: if ((ret = send(sock, msg.msg_base, ret, MSG_NOSIGNAL)) == -1) {
574: LOGERR;
575: goto end;
576: }
577:
578: pfd.events = POLLIN | POLLPRI;
579: while (tries--) {
580: if ((ret = poll(&pfd, 1, ka * 1000)) == -1 ||
581: pfd.revents & (POLLERR | POLLHUP | POLLNVAL)) {
582: LOGERR;
583: break;
584: } else if (!ret) {
585: ret = 1; /* session is abandoned ... must be disconnect! */
586: continue;
587: }
588: /* receive & decode packet */
589: if ((ret = recv(sock, msg.msg_base, msg.msg_len, 0)) == -1) {
590: LOGERR;
591: break;
592: }
593: if (!mqtt_readPINGRESP(&msg)) {
594: ret = 0; /* Host is alive */
595: break;
596: } else
597: ret = 2; /* Session is broken ... must be disconnect! */
598: }
599: end:
600: free(msg.msg_base);
601: return ret;
602: }