Linux应用开发【第十六章】MQTT协议分析应用开发

123456  = 964 x 128 + 64 

964 = 7x128 + 68

7 < 128

也就是123456 = (7 x 128 + 68)x128 + 64

展开:64 + 68 x128 + 7x128x128 

第一字节:64 | 0x80 = x (0x80=0x1000 0000或上最高位表示是否还有更多的字节)

第二字节:68 | 0x80 = y (0x80=0x1000 0000或上最高位表示是否还有更多的字节)

第三字节:7=z

c语言表示:unsigned char len_byte[4] = { 64 | 128 , 68 | 128,  7  , 0 }

反过来，如果要算出123456

x-128 + (y-128)*128 + z x 128 x 128 

static uint8_t client_id[512] = {"mqtt_client"};
static uint8_t user_name[512] = {"mqtt"};
static uint8_t passwd[512] = {"12345678"};
#define KEEP_ALIVE 20
int mqtt_connect(int sockfd)
{
    uint8 flags = 0x00;
	uint8 *packet = NULL;
	uint16 packet_length = 0;	
	uint16 clientidlen = strlen(client_id);
	uint16 usernamelen = strlen(user_name);
	uint16 passwordlen = strlen(passwd);
	uint16 payload_len = clientidlen + 2;
	// Variable header
	uint8 var_header[10] = {
		0x00,0x04,/*len*/
        0x4d,0x51,0x54,0x54,/*mqtt*/
		0x04,/*协议版本*/};
	
	uint8 fixedHeaderSize = 2;    // Default size = one byte Message Type + one byte Remaining Length
	uint8 remainLen = 0;
	uint8 *fixed_header = NULL;
	uint16 offset = 0;
    
	// Preparing the flags
	if(usernamelen) { /*用户名长度(可选)*/
		payload_len += usernamelen + 2;
		flags |= MQTT_USERNAME_FLAG;/*或上用户名标记*/
	}
	if(passwordlen) { /*用户密码(可选)*/
		payload_len += passwordlen + 2;
		flags |= MQTT_PASSWORD_FLAG;/*用户密码标记位*/
	}
	flags |= MQTT_CLEAN_SESSION;
	var_header[7] = flags;/*连接标记*/
	var_header[8] = KEEP_ALIVE>>8;/*保持连接字段，占用两个字节*/
	var_header[9] = KEEP_ALIVE&0xFF;

	remainLen = sizeof(var_header)+payload_len; /*剩余长度，也就是可变报头加上负载的长度*/

	if (remainLen > 127) {
	    fixedHeaderSize++;// add an additional byte for Remaining Length          
	}
   
	fixed_header = (uint8 *)malloc(fixedHeaderSize); /*固定报头*/
	// Message Type
	*fixed_header = MQTT_MSG_CONNECT;/*报文类型,connect*/

	
	if (remainLen <= 127) {// Remaining Length,剩余长度计算，可变长编码
	    *(fixed_header+1) = remainLen;
	} else {
	    // first byte is remainder (mod) of 128, then set the MSB to indicate more bytes
	    *(fixed_header+1) = remainLen % 128;
	    *(fixed_header+1) = *(fixed_header+1) | 0x80;
	    // second byte is number of 128s
	    *(fixed_header+2) = remainLen / 128;
	}

	packet_length = fixedHeaderSize+sizeof(var_header)+payload_len;/*固定报头+可变报头+负载长度*/
	packet = (uint8 *)malloc(packet_length);/*分配内存*/
	memset(packet, 0, packet_length);
	memcpy(packet, fixed_header, fixedHeaderSize);/*填充固定报头*/
	free(fixed_header);	
	offset += fixedHeaderSize;
	memcpy(packet+offset, var_header, sizeof(var_header));/*填充可变报头*/
	offset += sizeof(var_header);
	
	packet[offset++] = clientidlen>>8;// Client ID - UTF encoded,填充clientid长度+clientid
	packet[offset++] = clientidlen&0xFF;
	memcpy(packet+offset, client_id, clientidlen);
	offset += clientidlen;

	if(usernamelen) {// Username - UTF encoded,填充用户名+用户名长度
		packet[offset++] = usernamelen>>8;
		packet[offset++] = usernamelen&0xFF;
		memcpy(packet+offset, user_name, usernamelen);
		offset += usernamelen;
	}

	if(passwordlen) {// Password - UTF encoded,填充用户密码+用户名密码长度
		packet[offset++] = passwordlen>>8;
		packet[offset++] = passwordlen&0xFF;
		memcpy(packet+offset, passwd, passwordlen);
		offset += passwordlen;
	}
	// Send the packet
    if (client_send(sockfd,packet, packet_length) < 0){
         free(packet);
        return -1;
    }
    free(packet);
	return 1;
}

void mqtt_connect_ack(int sockfd)
{
	uint8_t cmd[]={ 0x20/*报文类型*/, 0x02/*剩余长度*/ ,0x00,0x00/*最后两个字节可变报头表示返回状态码*/ };
	send_msg(sockfd,cmd,sizeof(cmd));
	socket_record_t *socket_record = look_up_by_sokfd(sockfd);
	if(socket_record==NULL){
		return;
	}
	socket_record->is_connect=0x01;	
}

int mqtt_publish_with_qos(int sockfd,const char* topic, const char* msg, uint16 msgl, uint8 retain, uint8 qos, uint16* message_id) 
{
    socket_record_t *socket_record = look_up_by_sokfd(sockfd);
    if(NULL == socket_record){
        return -1;
    }
    DEBUG_INFO("sockfd:%d",socket_record->sockfd);
	uint16 topiclen = strlen(topic);
	uint16 msglen = msgl;
	uint8 *var_header = NULL; // Topic size (2 bytes), utf-encoded topic
	uint8 *fixed_header = NULL;
	uint8 fixedHeaderSize = 0,var_headerSize = 0; // Default size = one byte Message Type + one byte Remaining Length
	uint16 remainLen = 0;
	uint8 *packet = NULL;
	uint16 packet_length = 0;

	uint8 qos_flag = MQTT_QOS0_FLAG; /*qos标记*/
	uint8 qos_size = 0; // No QoS included
	if(qos == 1) {
		qos_size = 2; // 2 bytes for QoS
		qos_flag = MQTT_QOS1_FLAG;
	}
	else if(qos == 2) {
		qos_size = 2; // 2 bytes for QoS
		qos_flag = MQTT_QOS2_FLAG;
	}

	// Variable header
	var_headerSize = topiclen/*主题内容*/+2/*主题长度占用两字节*/+qos_size/*标识符*/;
	var_header = (uint8 *)malloc(var_headerSize);
	memset(var_header, 0, var_headerSize);
	*var_header = topiclen>>8;
	*(var_header+1) = topiclen&0xFF;
	memcpy(var_header+2, topic, topiclen);
	if(qos_size) {//qos1和qos2的报文需要填充标识符,有点像tcp的seq
        socket_record->publish_seq++;
        if(socket_record->publish_seq == 0){
            //unsigned short 表示范围0~65535,标识符必须是非零整数
            socket_record->publish_seq = 1;
        }
		var_header[topiclen+2] = (socket_record->publish_seq & 0xff00)>>8;
		var_header[topiclen+3] = socket_record->publish_seq & 0x00ff;
		if(message_id) { 
			*message_id = socket_record->publish_seq;
		}
	}

	fixedHeaderSize = 2;    // Default size = one byte Message Type + one byte Remaining Length
	remainLen = var_headerSize+msglen;
	if (remainLen > 127) {/*剩余长度*/
		fixedHeaderSize++;          // add an additional byte for Remaining Length
	}
	fixed_header = (uint8 *)malloc(fixedHeaderSize);/*固定报头+剩余长度*/
    
	// Message Type, DUP flag, QoS level, Retain
	*fixed_header = MQTT_MSG_PUBLISH | qos_flag;/*报文类型和qos标记*/
	if(retain) {
		*fixed_header  |= MQTT_RETAIN_FLAG;/*是否保留*/
	}
	// Remaining Length，剩余长度
	if (remainLen <= 127) {
	   *(fixed_header+1) = remainLen;
	} else {
	   // first byte is remainder (mod) of 128, then set the MSB to indicate more bytes
	   *(fixed_header+1) = remainLen % 128;
	   *(fixed_header+1) = *(fixed_header+1) | 0x80;
	   // second byte is number of 128s
	   *(fixed_header+2) = remainLen / 128;
	}

	packet_length = fixedHeaderSize+var_headerSize+msglen;/*固定报头+可变报头+负载长度*/
	packet = (uint8 *)malloc(packet_length);
	memset(packet, 0, packet_length);
	memcpy(packet, fixed_header, fixedHeaderSize);/*填充固定报头*/
	memcpy(packet+fixedHeaderSize, var_header, var_headerSize);/*填充可变报头*/
	memcpy(packet+fixedHeaderSize+var_headerSize, msg, msglen);/*负载*/
	free(var_header);
	free(fixed_header);
    send_msg(sockfd,packet , packet_length);
	free(packet);
	return 1;
}

//如果是PUBREC报文，head_type=0x50
void mqtt_qos2_pubrec(int  sockfd , unsigned char *data,unsigned char head_type)
{
    uint16 msg_id = mqtt_parse_msg_id(data);/*报文标识符,回复报文和接受报文的标识符必须一样*/
	unsigned char qos2_pubrec_respon[]={head_type/*固定报头*/,0x02/*剩余长度*/, (msg_id&0xff00)>>8 , msg_id&0x00ff/*最后两个字节是报文标识符*/};
	send_msg(sockfd,qos2_pubrec_respon,sizeof(qos2_pubrec_respon));
}

//head_type=0x62
void mqtt_qos2_pubrel(int sockfd , unsigned char *data,unsigned char head_type)
{
    uint16 msg_id = mqtt_parse_msg_id(data);
	unsigned char qos2_pubrel_respon[]={head_type/*报文类型*/,0x02/*剩余长度*/, (msg_id & 0xff00)>>8 , msg_id & 0x00ff/*最后两个字节是报文标识符*/};
	send_msg(sockfd,qos2_pubrel_respon,sizeof(qos2_pubrel_respon));
}

//head_type=0x70
void mqtt_qos2_pubcomp(int   sockfd , unsigned char *data,unsigned char head_type)
{
    uint16 msg_id = mqtt_parse_msg_id(data);/*报文标识符*/
	unsigned char qos2_pubcomp_respon[]={head_type/*报文类型*/,0x02/*剩余长度*/, (msg_id & 0xff00)>>8 , msg_id & 0x00ff/*最后两个字节报文标识符*/};
	send_msg(sockfd,qos2_pubcomp_respon,sizeof(qos2_pubcomp_respon));
}

int mqtt_ping(int sockfd)
{
	uint8 packet[] = {MQTT_MSG_PINGREQ/*报文类型*/,0x00/*剩余长度*/};
	int ret = send_msg(sockfd,packet, sizeof(packet));
	return ret;
}

void mqtt_ping_req_reply(int sockfd)
{
    uint8_t cmd[]={0xd0/*报文类型*/, 0x00/*剩余长度*/};
    send_msg(sockfd,cmd,sizeof(cmd));
}

int mqtt_disconnect(int sockfd) 
{
	uint8 packet[] = {MQTT_MSG_DISCONNECT/*报文类型*/,0x00/*剩余长度*/}; 
	int ret = client_send(sockfd,packet, sizeof(packet));
    DEBUG_INFO("ret=%d",ret);
	return ret;
}

static uint16 su_seq = 1;
int mqtt_subscribe_theme(int sockfd,char *Theme , uint8_t Qos)
{
	su_seq++;//报文标识符
	if(su_seq == 0){
		su_seq = 1;
	}
	uint16_t MessageId = su_seq;
	uint8_t cmd[1024]={0};
    //报文标示符长度2 + 主题长度位占用2字节+主题内容+qos标识
	int data_length = 2+2+strlen(Theme)+1; 
	int playload_len = strlen(Theme);
	uint8_t len_byte[4] ={0x00 , 0x00 ,0x00 ,0x00};
	uint8_t byte_num = length_trans_byte_form(data_length , len_byte);/*把剩余长度转换成变长编码*/
	cmd[0] = 0x82;
	memcpy(&cmd[1] , len_byte , byte_num);

	cmd[1+byte_num]=(MessageId & 0xff00) >> 8 ;
	cmd[1+byte_num+1] = MessageId & 0x00ff;		
	cmd[1+byte_num+1+1] = (playload_len & 0xff00) >> 8;
	cmd[1+byte_num+1+1+1] = playload_len & 0x00ff;
	memcpy(&cmd[1+byte_num+1+1+1+1] , Theme , playload_len);
	cmd[1+byte_num+1+1+1+1+playload_len] = Qos;
	client_send(sockfd,cmd, 1+byte_num+1+1+1+1+playload_len+1);
}

void mqtt_subscribe_ack(int sockfd,const uint8 *buf)
{
    uint16 msg_id = mqtt_parse_msg_id(buf);/*提取报文标识符*/
    uint8 qos = MQTTParseMessageQos(buf);/*提取报文qos*/
    uint8 cmd[]={0x90,0x03/*剩余长度*/, (msg_id & 0xff00) >> 8, msg_id & 0x00ff,qos};
    send_msg(sockfd,cmd,sizeof(cmd));
}

static uint16 un_seq = 1;
int mqtt_unsubscribe_theme(int sockfd,const char* topic)
{
	un_seq++;
	if(un_seq == 0){
		un_seq = 1;
	}
    uint16_t MessageId = un_seq;
	uint8_t cmd[1024]={0};
    //报文标示符长度2 + 主题长度位占用2字节+主题内容+qos标识
	int data_length = 2+2+strlen(topic)+1; 
	int playload_len = strlen(topic);
	uint8_t len_byte[4] ={0x00 , 0x00 ,0x00 ,0x00};
	uint8_t byte_num = length_trans_byte_form(data_length , len_byte);/*剩余长度转换成变长编码*/
	cmd[0] = 0xa2;
	memcpy(&cmd[1] , len_byte , byte_num);
    
	cmd[1+byte_num]=(MessageId & 0xff00) >> 8 ;
	cmd[1+byte_num+1] = MessageId & 0x00ff;		
	cmd[1+byte_num+1+1] = (playload_len & 0xff00) >> 8;
	cmd[1+byte_num+1+1+1] = playload_len & 0x00ff;
	memcpy(&cmd[1+byte_num+1+1+1+1] , topic , playload_len);
	client_send(sockfd,cmd,1+byte_num+1+1+1+1+playload_len+1);
	return 1;
}

void mqtt_unsubscribe_ack(int sockfd,const uint8 *buf)
{
    uint16 msg_id = mqtt_parse_msg_id(buf);
    uint8 cmd[]={0xb0,0x02/*剩余长度*/,(msg_id & 0xff00) >> 8, msg_id & 0x00ff/*最后两个字节是报文标识符*/};
    send_msg(sockfd,cmd,sizeof(cmd));
}