Return to sub.c CVS log

Up to [ELWIX - Embedded LightWeight unIX -] / mqtt / src

add new features for subscribe variables & ut

#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_data; t++) {
		topic = (mqtthdr_var_t*) (buf->msg_base + siz);
		topic->var_sb.val = t->sub_sb.val;
		memcpy(topic->var_data, t->sub_data, ntohs(topic->var_sb.val));
		siz += MQTTHDR_VAR_SIZEOF(topic);
		qos = (buf->msg_base + siz);
		*qos = t->sub_qos;
		siz++;
	}

	/* fixed header */

	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_data; t++) {
		qos = (buf->msg_base + siz);
		*qos = t->sub_qos;
		siz++;
	}

	/* fixed header */
	hdr->mqtt_msg.type = MQTT_TYPE_SUBACK;
	*hdr->mqtt_len = mqtt_encodeLen(siz - sizeof(struct mqtthdr));

	mqtt_msgRealloc(buf, siz);
	return siz;
}