Annotation of libaitmqtt/src/sub.c, revision 1.1.1.1
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 $
6: * $Id: aitsched.c,v 1.5 2012/01/24 21:59:47 misho Exp $
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:
15: Copyright 2004, 2005, 2006, 2007, 2008, 2009, 2010, 2011
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: /* ------------------------------------------------------------------- */
50:
51: /*
52: * mqtt_msgSUBSCRIBE() Create SUBSCRIBE message
53: *
54: * @buf = Message buffer
55: * @Topics = MQTT subscription topics
56: * @msgID = MessageID
57: * @Dup = Duplicate message
58: * @QOS = QoS
59: * return: -1 error or >-1 message size for send
60: */
61: int
62: mqtt_msgSUBSCRIBE(mqtt_msg_t * __restrict buf, mqtt_subscr_t * __restrict Topics,
63: u_short msgID, u_char Dup, u_char QOS)
64: {
65: int siz = 0;
66: struct mqtthdr *hdr;
67: mqtthdr_var_t *topic;
68: mqtt_len_t *mid;
69: mqtt_subscr_t *t;
70: u_char *qos;
71:
72: if (!buf || !Topics)
73: return -1;
74: if (QOS > MQTT_QOS_EXACTLY) {
75: mqtt_SetErr(EINVAL, "Error:: invalid QoS parameter");
76: return -1;
77: }
78: if (!msgID && QOS != MQTT_QOS_ONCE) {
79: mqtt_SetErr(EINVAL, "Error:: invalid MessageID parameter must be >0");
80: return -1;
81: }
82:
83: if (mqtt_msgRealloc(buf, MQTTMSG_MAX) == -1)
84: return -1;
85: else {
86: hdr = (struct mqtthdr *) (buf->msg_base + siz);
87: siz += sizeof(struct mqtthdr);
88: }
89:
90: /* variable header */
91: mid = (mqtt_len_t*) (buf->msg_base + siz);
92: mid->val = htons(msgID);
93: siz += sizeof(mqtt_len_t);
94:
95: /* payload with subscriptions */
96: for (t = Topics; t && t->sub_topic.msg_base; t++) {
97: topic = (mqtthdr_var_t*) (buf->msg_base + siz);
98: topic->var_sb.val = htons(t->sub_topic.msg_len);
99: memcpy(topic->var_data, t->sub_topic.msg_base, ntohs(topic->var_sb.val));
100: siz += MQTTHDR_VAR_SIZEOF(topic);
101: qos = (buf->msg_base + siz);
102: *qos = t->sub_ret;
103: siz++;
104: }
105:
106: /* fixed header */
107: MQTTHDR_MSGINIT(hdr);
108: hdr->mqtt_msg.type = MQTT_TYPE_SUBSCRIBE;
109: hdr->mqtt_msg.qos = QOS;
110: hdr->mqtt_msg.dup = Dup ? 1 : 0;
111: hdr->mqtt_msg.retain = 0;
112: *hdr->mqtt_len = mqtt_encodeLen(siz - sizeof(struct mqtthdr));
113:
114: mqtt_msgRealloc(buf, siz);
115: return siz;
116: }
117:
118: /*
119: * mqtt_msgSUBACK() Create SUBACK message
120: *
121: * @buf = Message buffer
122: * @Topics = MQTT subscription topics
123: * @msgID = MessageID
124: * return: -1 error or >-1 message size for send
125: */
126: int
127: mqtt_msgSUBACK(mqtt_msg_t * __restrict buf, mqtt_subscr_t * __restrict Topics, u_short msgID)
128: {
129: int siz = 0;
130: struct mqtthdr *hdr;
131: mqtt_len_t *v;
132: mqtt_subscr_t *t;
133: u_char *qos;
134:
135: if (!buf || !Topics)
136: return -1;
137:
138: if (mqtt_msgRealloc(buf, MQTTMSG_MAX) == -1)
139: return -1;
140: else {
141: hdr = (struct mqtthdr *) (buf->msg_base + siz);
142: siz += sizeof(struct mqtthdr);
143: v = (mqtt_len_t*) (buf->msg_base + siz);
144: siz += sizeof(mqtt_len_t);
145: }
146:
147: /* MessageID */
148: v->val = htons(msgID);
149:
150: /* QoS payload from subscriptions */
151: for (t = Topics; t && t->sub_topic.msg_base; t++) {
152: qos = (buf->msg_base + siz);
153: *qos = t->sub_ret;
154: siz++;
155: }
156:
157: /* fixed header */
158: MQTTHDR_MSGINIT(hdr);
159: hdr->mqtt_msg.type = MQTT_TYPE_SUBACK;
160: *hdr->mqtt_len = mqtt_encodeLen(siz - sizeof(struct mqtthdr));
161:
162: mqtt_msgRealloc(buf, siz);
163: return siz;
164: }
165:
166: /*
167: * mqtt_msgUNSUBSCRIBE() Create UNSUBSCRIBE message
168: *
169: * @buf = Message buffer
170: * @Topics = MQTT subscription topics
171: * @msgID = MessageID
172: * @Dup = Duplicate message
173: * @QOS = QoS
174: * return: -1 error or >-1 message size for send
175: */
176: int
177: mqtt_msgUNSUBSCRIBE(mqtt_msg_t * __restrict buf, mqtt_subscr_t * __restrict Topics,
178: u_short msgID, u_char Dup, u_char QOS)
179: {
180: int siz = 0;
181: struct mqtthdr *hdr;
182: mqtthdr_var_t *topic;
183: mqtt_len_t *mid;
184: mqtt_subscr_t *t;
185:
186: if (!buf || !Topics)
187: return -1;
188: if (QOS > MQTT_QOS_EXACTLY) {
189: mqtt_SetErr(EINVAL, "Error:: invalid QoS parameter");
190: return -1;
191: }
192: if (!msgID && QOS != MQTT_QOS_ONCE) {
193: mqtt_SetErr(EINVAL, "Error:: invalid MessageID parameter must be >0");
194: return -1;
195: }
196:
197: if (mqtt_msgRealloc(buf, MQTTMSG_MAX) == -1)
198: return -1;
199: else {
200: hdr = (struct mqtthdr *) (buf->msg_base + siz);
201: siz += sizeof(struct mqtthdr);
202: }
203:
204: /* variable header */
205: mid = (mqtt_len_t*) (buf->msg_base + siz);
206: mid->val = htons(msgID);
207: siz += sizeof(mqtt_len_t);
208:
209: /* payload with subscriptions */
210: for (t = Topics; t && t->sub_topic.msg_base; t++) {
211: topic = (mqtthdr_var_t*) (buf->msg_base + siz);
212: topic->var_sb.val = htons(t->sub_topic.msg_len);
213: memcpy(topic->var_data, t->sub_topic.msg_base, ntohs(topic->var_sb.val));
214: siz += MQTTHDR_VAR_SIZEOF(topic);
215: }
216:
217: /* fixed header */
218: MQTTHDR_MSGINIT(hdr);
219: hdr->mqtt_msg.type = MQTT_TYPE_UNSUBSCRIBE;
220: hdr->mqtt_msg.qos = QOS;
221: hdr->mqtt_msg.dup = Dup ? 1 : 0;
222: hdr->mqtt_msg.retain = 0;
223: *hdr->mqtt_len = mqtt_encodeLen(siz - sizeof(struct mqtthdr));
224:
225: mqtt_msgRealloc(buf, siz);
226: return siz;
227: }
228:
229: /*
230: * mqtt_msgUNSUBACK() Create UNSUBACK message
231: *
232: * @buf = Message buffer
233: * @msgID = MessageID
234: * return: -1 error or >-1 message size for send
235: */
236: int
237: mqtt_msgUNSUBACK(mqtt_msg_t * __restrict buf, u_short msgID)
238: {
239: int siz = 0;
240: struct mqtthdr *hdr;
241: mqtt_len_t *v;
242:
243: if (!buf)
244: return -1;
245:
246: if (mqtt_msgRealloc(buf, sizeof(struct mqtthdr) + sizeof(mqtt_len_t)) == -1)
247: return -1;
248: else {
249: hdr = (struct mqtthdr *) (buf->msg_base + siz);
250: siz += sizeof(struct mqtthdr);
251: v = (mqtt_len_t*) (buf->msg_base + siz);
252: siz += sizeof(mqtt_len_t);
253: }
254:
255: /* fixed header */
256: MQTTHDR_MSGINIT(hdr);
257: hdr->mqtt_msg.type = MQTT_TYPE_UNSUBACK;
258: *hdr->mqtt_len = sizeof(mqtt_len_t);
259:
260: /* MessageID */
261: v->val = htons(msgID);
262:
263: return siz;
264: }
265:
266:
267: /* ============= decode ============ */
268:
269: /*
270: * mqtt_readSUBSCRIBE() Read SUBSCRIBE message
271: *
272: * @buf = Message buffer
273: * @msgID = MessageID
274: * @subscr = Subscriptions, must be free after use with mqtt_subFree()
275: * return: NULL error or !=NULL MQTT fixed header
276: */
277: struct mqtthdr *
278: mqtt_readSUBSCRIBE(mqtt_msg_t * __restrict buf, u_short *msgID, mqtt_subscr_t **subscr)
279: {
280: register int i;
281: int len, ret;
282: struct mqtthdr *hdr;
283: mqtthdr_var_t *var;
284: mqtt_subscr_t *subs;
285: mqtt_len_t *v;
286: caddr_t pos;
287:
288: if (!buf || !msgID || !subscr)
289: return NULL;
290:
291: hdr = _mqtt_readHEADER(buf, MQTT_TYPE_SUBSCRIBE, &ret, &len);
292: if (!hdr)
293: return NULL;
294: pos = buf->msg_base + ret + 1;
295: v = (mqtt_len_t*) pos;
296:
297: /* MessageID */
298: len -= sizeof(mqtt_len_t);
299: if (len < 0) {
300: mqtt_SetErr(EINVAL, "Error:: short message length %d", len);
301: return NULL;
302: } else {
303: *msgID = ntohs(v->val);
304: pos += sizeof(mqtt_len_t);
305: }
306:
307: subs = mqtt_subAlloc(0);
308: if (!subs)
309: return NULL;
310: else
311: *subscr = subs;
312:
313: /* Subscribes */
314: for (i = 0; len > 0; i++) {
315: var = (mqtthdr_var_t*) pos;
316: len -= MQTTHDR_VAR_SIZEOF(var) + 1;
317: if (len < 0) {
318: mqtt_subFree(subscr);
319: mqtt_SetErr(EINVAL, "Error:: short message length %d", len);
320: return NULL;
321: }
322: subs = mqtt_subRealloc(subs, i + 1);
323: if (!subs) {
324: mqtt_subFree(subscr);
325: return NULL;
326: } else
327: *subscr = subs;
328:
329: memset(&subs[i], 0, sizeof subs[i]);
330: subs[i].sub_topic.msg_len = ntohs(var->var_sb.val);
331: subs[i].sub_topic.msg_base = malloc(subs[i].sub_topic.msg_len);
332: if (!subs[i].sub_topic.msg_base) {
333: LOGERR;
334: mqtt_subFree(subscr);
335: return NULL;
336: } else
337: memcpy(subs[i].sub_topic.msg_base, var->var_data, subs[i].sub_topic.msg_len);
338: pos += MQTTHDR_VAR_SIZEOF(var);
339:
340: subs[i].sub_ret = *pos;
341: pos++;
342: }
343:
344: return hdr;
345: }
346:
347: /*
348: * mqtt_readSUBACK() Read SUBACK message
349: *
350: * @buf = Message buffer
351: * @msgID = MessageID
352: * @subqos = Subscribes QoS, must be free after use with free()
353: * return: -1 error or >-1 readed subscribes QoS elements
354: */
355: int
356: mqtt_readSUBACK(mqtt_msg_t * __restrict buf, u_short *msgID, u_char **subqos)
357: {
358: int len, ret;
359: struct mqtthdr *hdr;
360: mqtt_len_t *v;
361: caddr_t pos;
362:
363: if (!buf || !msgID || !subqos)
364: return -1;
365:
366: hdr = _mqtt_readHEADER(buf, MQTT_TYPE_SUBACK, &ret, &len);
367: if (!hdr)
368: return -1;
369: pos = buf->msg_base + ret + 1;
370: v = (mqtt_len_t*) pos;
371:
372: /* MessageID */
373: len -= sizeof(mqtt_len_t);
374: if (len < 0) {
375: mqtt_SetErr(EINVAL, "Error:: short message length %d", len);
376: return -1;
377: } else {
378: *msgID = ntohs(v->val);
379: pos += sizeof(mqtt_len_t);
380: }
381:
382: /* Subscribes */
383: *subqos = malloc(len);
384: if (!*subqos) {
385: LOGERR;
386: return -1;
387: } else
388: memcpy(*subqos, pos, len);
389:
390: return len;
391: }
392:
393: /*
394: * mqtt_readUNSUBSCRIBE() Read UNSUBSCRIBE message
395: *
396: * @buf = Message buffer
397: * @msgID = MessageID
398: * @subscr = Subscriptions, must be free after use with mqtt_subFree()
399: * return: NULL error or !=NULL MQTT fixed header
400: */
401: struct mqtthdr *
402: mqtt_readUNSUBSCRIBE(mqtt_msg_t * __restrict buf, u_short *msgID, mqtt_subscr_t **subscr)
403: {
404: register int i;
405: int len, ret;
406: struct mqtthdr *hdr;
407: mqtthdr_var_t *var;
408: mqtt_subscr_t *subs;
409: mqtt_len_t *v;
410: caddr_t pos;
411:
412: if (!buf || !msgID || !subscr)
413: return NULL;
414:
415: hdr = _mqtt_readHEADER(buf, MQTT_TYPE_UNSUBSCRIBE, &ret, &len);
416: if (!hdr)
417: return NULL;
418: pos = buf->msg_base + ret + 1;
419: v = (mqtt_len_t*) pos;
420:
421: /* MessageID */
422: len -= sizeof(mqtt_len_t);
423: if (len < 0) {
424: mqtt_SetErr(EINVAL, "Error:: short message length %d", len);
425: return NULL;
426: } else {
427: *msgID = ntohs(v->val);
428: pos += sizeof(mqtt_len_t);
429: }
430:
431: subs = mqtt_subAlloc(0);
432: if (!subs)
433: return NULL;
434: else
435: *subscr = subs;
436:
437: /* Subscribes */
438: for (i = 0; len > 0; i++) {
439: var = (mqtthdr_var_t*) pos;
440: len -= MQTTHDR_VAR_SIZEOF(var);
441: if (len < 0) {
442: mqtt_subFree(subscr);
443: mqtt_SetErr(EINVAL, "Error:: short message length %d", len);
444: return NULL;
445: }
446: subs = mqtt_subRealloc(subs, i + 1);
447: if (!subs) {
448: mqtt_subFree(subscr);
449: return NULL;
450: } else
451: *subscr = subs;
452:
453: memset(&subs[i], 0, sizeof subs[i]);
454: subs[i].sub_topic.msg_len = ntohs(var->var_sb.val);
455: subs[i].sub_topic.msg_base = malloc(subs[i].sub_topic.msg_len);
456: if (!subs[i].sub_topic.msg_base) {
457: LOGERR;
458: mqtt_subFree(subscr);
459: return NULL;
460: } else
461: memcpy(subs[i].sub_topic.msg_base, var->var_data, subs[i].sub_topic.msg_len);
462: pos += MQTTHDR_VAR_SIZEOF(var);
463: }
464:
465: return hdr;
466: }
467:
468: /*
469: * mqtt_readUNSUBACK() Read UNSUBACK message
470: *
471: * @buf = Message buffer
472: * return: -1 error or MessageID
473: */
474: u_short
475: mqtt_readUNSUBACK(mqtt_msg_t * __restrict buf)
476: {
477: int len, ret;
478: struct mqtthdr *hdr;
479: mqtt_len_t *v;
480: caddr_t pos;
481:
482: hdr = _mqtt_readHEADER(buf, MQTT_TYPE_UNSUBACK, &ret, &len);
483: if (!hdr)
484: return (u_short) -1;
485: if (len < sizeof(mqtt_len_t)) {
486: mqtt_SetErr(EINVAL, "Error:: short message length %d", len);
487: return (u_short) -1;
488: } else {
489: pos = buf->msg_base + ret + 1;
490: v = (mqtt_len_t*) pos;
491: }
492:
493: return ntohs(v->val);
494: }
FreeBSD-CVSweb <freebsd-cvsweb@FreeBSD.org>
![[BACK]](/icons/cvsweb/back.gif){kind=icon}
Return to sub.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