Return to sub.c CVS log

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

add new features for subscribe variables & ut

    1: #include "global.h"
    2: 
    3: 
    4: /* ------------------------------------------------------------------- */
    5: 
    6: /*
    7:  * mqtt_msgSUBSCRIBE() Create SUBSCRIBE message
    8:  *
    9:  * @buf = Message buffer
   10:  * @Topics = MQTT subscription topics
   11:  * @msgID = MessageID
   12:  * @Dup = Duplicate message
   13:  * @QOS = QoS
   14:  * return: -1 error or >-1 message size for send
   15:  */
   16: int
   17: mqtt_msgSUBSCRIBE(mqtt_msg_t * __restrict buf, mqtt_subscr_t * __restrict Topics, 
   18: 		u_short msgID, u_char Dup, u_char QOS)
   19: {
   20: 	int siz = 0;
   21: 	struct mqtthdr *hdr;
   22: 	mqtthdr_var_t *topic;
   23: 	mqtt_v_t *mid;
   24: 	mqtt_subscr_t *t;
   25: 	u_char *qos;
   26: 
   27: 	if (!buf || !Topics)
   28: 		return -1;
   29: 	if (QOS > MQTT_QOS_EXACTLY) {
   30: 		mqtt_SetErr(EINVAL, "Error:: invalid QoS parameter");
   31: 		return -1;
   32: 	}
   33: 	if (!msgID && QOS != MQTT_QOS_ONCE) {
   34: 		mqtt_SetErr(EINVAL, "Error:: invalid MessageID parameter must be >0");
   35: 		return -1;
   36: 	}
   37: 
   38: 	if (mqtt_msgRealloc(buf, MQTTMSG_MAX) == -1)
   39: 		return -1;
   40: 	else {
   41: 		hdr = (struct mqtthdr *) (buf->msg_base + siz);
   42: 		siz += sizeof(struct mqtthdr);
   43: 	}
   44: 
   45: 	/* variable header */
   46: 
   47: 	mid = (mqtt_v_t*) (buf->msg_base + siz);
   48: 	mid->val = htons(msgID);
   49: 	siz += sizeof(mqtt_v_t);
   50: 
   51: 	/* payload with subscriptions */
   52: 
   53: 	for (t = Topics; t && t->sub_data; t++) {
   54: 		topic = (mqtthdr_var_t*) (buf->msg_base + siz);
   55: 		topic->var_sb.val = t->sub_sb.val;
   56: 		memcpy(topic->var_data, t->sub_data, ntohs(topic->var_sb.val));
   57: 		siz += MQTTHDR_VAR_SIZEOF(topic);
   58: 		qos = (buf->msg_base + siz);
   59: 		*qos = t->sub_qos;
   60: 		siz++;
   61: 	}
   62: 
   63: 	/* fixed header */
   64: 
   65: 	hdr->mqtt_msg.type = MQTT_TYPE_SUBSCRIBE;
   66: 	hdr->mqtt_msg.qos = QOS;
   67: 	hdr->mqtt_msg.dup = Dup ? 1 : 0;
   68: 	hdr->mqtt_msg.retain = 0;
   69: 	*hdr->mqtt_len = mqtt_encodeLen(siz - sizeof(struct mqtthdr));
   70: 
   71: 	mqtt_msgRealloc(buf, siz);
   72: 	return siz;
   73: }
   74: 
   75: /*
   76:  * mqtt_msgSUBACK() Create SUBACK message
   77:  *
   78:  * @buf = Message buffer
   79:  * @Topics = MQTT subscription topics
   80:  * @msgID = MessageID
   81:  * return: -1 error or >-1 message size for send
   82:  */
   83: int
   84: mqtt_msgSUBACK(mqtt_msg_t * __restrict buf, mqtt_subscr_t * __restrict Topics, u_short msgID)
   85: {
   86: 	int siz = 0;
   87: 	struct mqtthdr *hdr;
   88: 	mqtt_v_t *v;
   89: 	mqtt_subscr_t *t;
   90: 	u_char *qos;
   91: 
   92: 	if (!buf || !Topics)
   93: 		return -1;
   94: 
   95: 	if (mqtt_msgRealloc(buf, MQTTMSG_MAX) == -1)
   96: 		return -1;
   97: 	else {
   98: 		hdr = (struct mqtthdr *) (buf->msg_base + siz);
   99: 		siz += sizeof(struct mqtthdr);
  100: 		v = (mqtt_v_t*) (buf->msg_base + siz);
  101: 		siz += sizeof(mqtt_v_t);
  102: 	}
  103: 
  104: 	/* MessageID */
  105: 	v->val = htons(msgID);
  106: 
  107: 	/* QoS payload from subscriptions */
  108: 
  109: 	for (t = Topics; t && t->sub_data; t++) {
  110: 		qos = (buf->msg_base + siz);
  111: 		*qos = t->sub_qos;
  112: 		siz++;
  113: 	}
  114: 
  115: 	/* fixed header */
  116: 	hdr->mqtt_msg.type = MQTT_TYPE_SUBACK;
  117: 	*hdr->mqtt_len = mqtt_encodeLen(siz - sizeof(struct mqtthdr));
  118: 
  119: 	mqtt_msgRealloc(buf, siz);
  120: 	return siz;
  121: }