#include "global.h" /* ------------------------------------------------------------------- */ /* * mqtt_msgSUBSCRIBE() Create SUBSCRIBE message * * @buf = Message buffer * @Topics = MQTT subscription topics * @msgID = MessageID * @Dup = Duplicate message * @QOS = QoS * return: -1 error or >-1 message size for send */ int mqtt_msgSUBSCRIBE(mqtt_msg_t * __restrict buf, mqtt_subscr_t * __restrict Topics, u_short msgID, u_char Dup, u_char QOS) { int siz = 0; struct mqtthdr *hdr; mqtthdr_var_t *topic; mqtt_v_t *mid; mqtt_subscr_t *t; u_char *qos; if (!buf || !Topics) return -1; if (QOS > MQTT_QOS_EXACTLY) { mqtt_SetErr(EINVAL, "Error:: invalid QoS parameter"); return -1; } if (!msgID && QOS != MQTT_QOS_ONCE) { mqtt_SetErr(EINVAL, "Error:: invalid MessageID parameter must be >0"); return -1; } if (mqtt_msgRealloc(buf, MQTTMSG_MAX) == -1) return -1; else { hdr = (struct mqtthdr *) (buf->msg_base + siz); siz += sizeof(struct mqtthdr); } /* variable header */ mid = (mqtt_v_t*) (buf->msg_base + siz); mid->val = htons(msgID); siz += sizeof(mqtt_v_t); /* payload with subscriptions */ for (t = Topics; t && t->sub_topic._base; t++) { topic = (mqtthdr_var_t*) (buf->msg_base + siz); topic->var_sb.val = htons(t->sub_topic._size); memcpy(topic->var_data, t->sub_topic._base, ntohs(topic->var_sb.val)); siz += MQTTHDR_VAR_SIZEOF(topic); qos = (buf->msg_base + siz); *qos = t->sub_ret; siz++; } /* fixed header */ MQTTHDR_MSGINIT(hdr); hdr->mqtt_msg.type = MQTT_TYPE_SUBSCRIBE; hdr->mqtt_msg.qos = QOS; hdr->mqtt_msg.dup = Dup ? 1 : 0; hdr->mqtt_msg.retain = 0; *hdr->mqtt_len = mqtt_encodeLen(siz - sizeof(struct mqtthdr)); mqtt_msgRealloc(buf, siz); return siz; } /* * mqtt_msgSUBACK() Create SUBACK message * * @buf = Message buffer * @Topics = MQTT subscription topics * @msgID = MessageID * return: -1 error or >-1 message size for send */ int mqtt_msgSUBACK(mqtt_msg_t * __restrict buf, mqtt_subscr_t * __restrict Topics, u_short msgID) { int siz = 0; struct mqtthdr *hdr; mqtt_v_t *v; mqtt_subscr_t *t; u_char *qos; if (!buf || !Topics) return -1; if (mqtt_msgRealloc(buf, MQTTMSG_MAX) == -1) return -1; else { hdr = (struct mqtthdr *) (buf->msg_base + siz); siz += sizeof(struct mqtthdr); v = (mqtt_v_t*) (buf->msg_base + siz); siz += sizeof(mqtt_v_t); } /* MessageID */ v->val = htons(msgID); /* QoS payload from subscriptions */ for (t = Topics; t && t->sub_topic._base; t++) { qos = (buf->msg_base + siz); *qos = t->sub_ret; siz++; } /* fixed header */ MQTTHDR_MSGINIT(hdr); hdr->mqtt_msg.type = MQTT_TYPE_SUBACK; *hdr->mqtt_len = mqtt_encodeLen(siz - sizeof(struct mqtthdr)); mqtt_msgRealloc(buf, siz); return siz; } /* * mqtt_msgUNSUBSCRIBE() Create UNSUBSCRIBE message * * @buf = Message buffer * @Topics = MQTT subscription topics * @msgID = MessageID * @Dup = Duplicate message * @QOS = QoS * return: -1 error or >-1 message size for send */ int mqtt_msgUNSUBSCRIBE(mqtt_msg_t * __restrict buf, mqtt_subscr_t * __restrict Topics, u_short msgID, u_char Dup, u_char QOS) { int siz = 0; struct mqtthdr *hdr; mqtthdr_var_t *topic; mqtt_v_t *mid; mqtt_subscr_t *t; if (!buf || !Topics) return -1; if (QOS > MQTT_QOS_EXACTLY) { mqtt_SetErr(EINVAL, "Error:: invalid QoS parameter"); return -1; } if (!msgID && QOS != MQTT_QOS_ONCE) { mqtt_SetErr(EINVAL, "Error:: invalid MessageID parameter must be >0"); return -1; } if (mqtt_msgRealloc(buf, MQTTMSG_MAX) == -1) return -1; else { hdr = (struct mqtthdr *) (buf->msg_base + siz); siz += sizeof(struct mqtthdr); } /* variable header */ mid = (mqtt_v_t*) (buf->msg_base + siz); mid->val = htons(msgID); siz += sizeof(mqtt_v_t); /* payload with subscriptions */ for (t = Topics; t && t->sub_topic._base; t++) { topic = (mqtthdr_var_t*) (buf->msg_base + siz); topic->var_sb.val = htons(t->sub_topic._size); memcpy(topic->var_data, t->sub_topic._base, ntohs(topic->var_sb.val)); siz += MQTTHDR_VAR_SIZEOF(topic); } /* fixed header */ MQTTHDR_MSGINIT(hdr); hdr->mqtt_msg.type = MQTT_TYPE_UNSUBSCRIBE; hdr->mqtt_msg.qos = QOS; hdr->mqtt_msg.dup = Dup ? 1 : 0; hdr->mqtt_msg.retain = 0; *hdr->mqtt_len = mqtt_encodeLen(siz - sizeof(struct mqtthdr)); mqtt_msgRealloc(buf, siz); return siz; } /* * mqtt_msgUNSUBACK() Create UNSUBACK message * * @buf = Message buffer * @msgID = MessageID * return: -1 error or >-1 message size for send */ int mqtt_msgUNSUBACK(mqtt_msg_t * __restrict buf, u_short msgID) { int siz = 0; struct mqtthdr *hdr; mqtt_v_t *v; if (!buf) return -1; if (mqtt_msgRealloc(buf, sizeof(struct mqtthdr) + sizeof(mqtt_v_t)) == -1) return -1; else { hdr = (struct mqtthdr *) (buf->msg_base + siz); siz += sizeof(struct mqtthdr); v = (mqtt_v_t*) (buf->msg_base + siz); siz += sizeof(mqtt_v_t); } /* fixed header */ MQTTHDR_MSGINIT(hdr); hdr->mqtt_msg.type = MQTT_TYPE_UNSUBACK; *hdr->mqtt_len = sizeof(mqtt_v_t); /* MessageID */ v->val = htons(msgID); return siz; }