自学教程：C++ FUNC_EXIT_RC函数代码示例

/** * MQTT outgoing connect processing for a client * @param ip_address the TCP address:port to connect to * @param clientID the MQTT client id to use * @param cleansession MQTT cleansession flag * @param keepalive MQTT keepalive timeout in seconds * @param willMessage pointer to the will message to be used, if any * @param username MQTT 3.1 username, or NULL * @param password MQTT 3.1 password, or NULL * @return the new client structure */int MQTTProtocol_connect(char* ip_address, Clients* aClient){	int rc, port;	char* addr;	FUNC_ENTRY;	aClient->good = 1;	time(&(aClient->lastContact));	addr = MQTTProtocol_addressPort(ip_address, &port);	rc = Socket_new(addr, port, &(aClient->socket));	if (rc == EINPROGRESS || rc == EWOULDBLOCK)		aClient->connect_state = 1; /* TCP connect called */	else if (rc == 0)	{		if ((rc = MQTTPacket_send_connect(aClient)) == 0)			aClient->connect_state = 2; /* TCP connect completed, in which case send the MQTT connect packet */		else			aClient->connect_state = 0;	}	FUNC_EXIT_RC(rc);	return rc;}

int MQTTSPacket_send_subAck(Clients* client, MQTTS_Subscribe* sub, int topicId, int qos, char returnCode){	MQTTS_SubAck packet;	int rc = 0;	char *buf, *ptr;	int datalen = 6;	FUNC_ENTRY;	packet.header.len = 8;	packet.header.type = MQTTS_SUBACK;	ptr = buf = malloc(datalen);	packet.flags.QoS = qos;	writeChar(&ptr, packet.flags.all);	writeInt(&ptr, topicId);	writeInt(&ptr, sub->msgId);	writeChar(&ptr, returnCode);	rc = MQTTSPacket_send(client->socket, client->addr, packet.header, buf, datalen);	free(buf);	Log(LOG_PROTOCOL, 68, NULL, client->socket, client->addr, client->clientID, sub->msgId, topicId, returnCode, rc);	FUNC_EXIT_RC(rc);	return rc;}

int MQTTClient_publishMessage(MQTTClient handle, const char* topicName, MQTTClient_message* message,															 MQTTClient_deliveryToken* deliveryToken){	int rc = MQTTCLIENT_SUCCESS;	FUNC_ENTRY;	if (message == NULL)	{		rc = MQTTCLIENT_NULL_PARAMETER;		goto exit;	}	if (strncmp(message->struct_id, "MQTM", 4) != 0 || message->struct_version != 0)	{		rc = MQTTCLIENT_BAD_STRUCTURE;		goto exit;	}	rc = MQTTClient_publish(handle, topicName, message->payloadlen, message->payload,								message->qos, message->retained, deliveryToken);exit:	FUNC_EXIT_RC(rc);	return rc;}

/**  * Serializes the ack packet into the supplied buffer.  * @param buf the buffer into which the packet will be serialized  * @param buflen the length in bytes of the supplied buffer  * @param type the MQTT packet type  * @param dup the MQTT dup flag  * @param packetid the MQTT packet identifier  * @return serialized length, or error if 0  */int MQTTSerialize_ack(unsigned char* buf, int buflen, unsigned char packettype, unsigned char dup, unsigned short packetid){	MQTTHeader header = {0};	int rc = 0;	unsigned char *ptr = buf;	FUNC_ENTRY;	if (buflen < 4)	{		rc = MQTTPACKET_BUFFER_TOO_SHORT;		goto exit;	}	header.bits.type = packettype;	header.bits.dup = dup;	header.bits.qos = 0;	writeChar(&ptr, header.byte); /* write header */	ptr += MQTTPacket_encode(ptr, 2); /* write remaining length */	writeInt(&ptr, packetid);	rc = ptr - buf;exit:	FUNC_EXIT_RC(rc);	return rc;}

int MQTTSPacket_send_register(Clients* client, int topicId, char* topicName, int msgId){	MQTTS_Register packet;	int rc = 0;	char *buf, *ptr;	int datalen = 4 + strlen(topicName);	FUNC_ENTRY;	packet.header.len = datalen+2;	packet.header.type = MQTTS_REGISTER;	ptr = buf = malloc(datalen);	writeInt(&ptr, topicId);	writeInt(&ptr, msgId);	memcpy(ptr, topicName, strlen(topicName));	rc = MQTTSPacket_send(client->socket, client->addr, packet.header, buf, datalen);	free(buf);	Log(LOG_PROTOCOL, 50, NULL, client->socket, client->addr, client->clientID, msgId, topicId, topicName, rc);	FUNC_EXIT_RC(rc);	return rc;}

/**  * Serializes the supplied publish data into the supplied buffer, ready for sending  * @param buf the buffer into which the packet will be serialized  * @param buflen the length in bytes of the supplied buffer  * @param dup integer - the MQTT dup flag  * @param qos integer - the MQTT QoS value  * @param retained integer - the MQTT retained flag  * @param packetid integer - the MQTT packet identifier  * @param topicName MQTTString - the MQTT topic in the publish  * @param payload byte buffer - the MQTT publish payload  * @param payloadlen integer - the length of the MQTT payload  * @return the length of the serialized data.  <= 0 indicates error  */int MQTTSerialize_publish(unsigned char* buf, int buflen, unsigned char dup, int qos, unsigned char retained, unsigned short packetid,                          MQTTString topicName, unsigned char* payload, int payloadlen){	unsigned char *ptr = buf;	MQTTHeader header = {0};	int rem_len = 0;	int rc = 0;	FUNC_ENTRY;	if (MQTTPacket_len(rem_len = MQTTSerialize_publishLength(qos, topicName, payloadlen)) > buflen) {		rc = MQTTPACKET_BUFFER_TOO_SHORT;		goto exit;	}	header.bits.type = PUBLISH;	header.bits.dup = dup;	header.bits.qos = qos;	header.bits.retain = retained;	writeChar(&ptr, header.byte); /* write header */	ptr += MQTTPacket_encode(ptr, rem_len); /* write remaining length */;	writeMQTTString(&ptr, topicName);	if (qos > 0)		writeInt(&ptr, packetid);	memcpy(ptr, payload, payloadlen);	ptr += payloadlen;	rc = ptr - buf;exit:	FUNC_EXIT_RC(rc);	return rc;}

/**  * Deserializes the supplied (wire) buffer into advertise data  * @param gatewayid the returned gateway id  * @param duration the returned duration - the time interval until the next advertise will be sent  * @param buf the raw buffer data, of the correct length determined by the remaining length field  * @param buflen the length in bytes of the data in the supplied buffer  * @return error code.  1 is success  */int MQTTSNDeserialize_advertise(unsigned char* gatewayid, unsigned short* duration,	unsigned char* buf, int buflen){	unsigned char* curdata = buf;	unsigned char* enddata = NULL;	int rc = 0;	int mylen = 0;	FUNC_ENTRY;	curdata += (rc = MQTTSNPacket_decode(curdata, buflen, &mylen)); /* read length */	enddata = buf + mylen;	if (enddata - curdata > buflen)		goto exit;	if (readChar(&curdata) != MQTTSN_ADVERTISE)		goto exit;	*gatewayid = readChar(&curdata);	*duration = readInt(&curdata);	rc = 1;exit:	FUNC_EXIT_RC(rc);	return rc;}

/**  * Deserializes the supplied (wire) buffer into an ack  * @param packettype returned integer - the MQTT packet type  * @param packetid returned integer - the MQTT packet identifier  * @param buf the raw buffer data, of the correct length determined by the remaining length field  * @param buflen the length in bytes of the data in the supplied buffer  * @return error code.  1 is success, 0 is failure  */int MQTTSNDeserialize_ack(unsigned char* type, unsigned short* packetid, unsigned char* buf, int buflen){	unsigned char* curdata = buf;	unsigned char* enddata = NULL;	int rc = 0;	int mylen = 0;	FUNC_ENTRY;	curdata += (rc = MQTTSNPacket_decode(curdata, buflen, &mylen)); /* read length */	enddata = buf + mylen;	if (enddata - curdata > buflen)		goto exit;	*type = readChar(&curdata);	if (*type != MQTTSN_PUBREL && *type != MQTTSN_PUBREC && *type != MQTTSN_PUBCOMP)		goto exit;	*packetid = readInt(&curdata);	rc = 1;exit:	FUNC_EXIT_RC(rc);	return rc;}

/** * Process an incoming publish packet for a socket * @param pack pointer to the publish packet * @param sock the socket on which the packet was received * @return completion code */int MQTTProtocol_handlePublishes(void* pack, int sock, Clients* client){	Publish* publish = (Publish*)pack;	char* clientid = NULL;	int rc = TCPSOCKET_COMPLETE;	FUNC_ENTRY;	if (client == NULL)		clientid = INTERNAL_CLIENTID; /* this is an internal client */	else	{		clientid = client->clientID;		Log(LOG_PROTOCOL, 11, NULL, sock, clientid, publish->msgId, publish->header.bits.qos,				publish->header.bits.retain);	}#if defined(MQTTS)	rc = Protocol_handlePublishes(publish, sock, client, clientid, 0);#else	rc = Protocol_handlePublishes(publish, sock, client, clientid);#endif	FUNC_EXIT_RC(rc);	return rc;}

/**  * Deserializes the supplied (wire) buffer into suback data  * @param packetid returned integer - the MQTT packet identifier  * @param maxcount - the maximum number of members allowed in the grantedQoSs array  * @param count returned integer - number of members in the grantedQoSs array  * @param grantedQoSs returned array of integers - the granted qualities of service  * @param buf the raw buffer data, of the correct length determined by the remaining length field  * @param buflen the length in bytes of the data in the supplied buffer  * @return error code.  1 is success, 0 is failure  */int MQTTDeserialize_suback(unsigned short* packetid, int maxcount, int* count, int grantedQoSs[], unsigned char* buf, int buflen){	MQTTHeader header = {0};	unsigned char* curdata = buf;	unsigned char* enddata = NULL;	int rc = 0;	int mylen;	FUNC_ENTRY;	header.byte = readChar(&curdata);	if (header.bits.type != SUBACK)		goto exit;	curdata += (rc = MQTTPacket_decodeBuf(curdata, &mylen)); /* read remaining length */	enddata = curdata + mylen;	if (enddata - curdata < 2)		goto exit;	*packetid = readInt(&curdata);	*count = 0;	while (curdata < enddata)	{		if (*count > maxcount)		{			rc = -1;			goto exit;		}		grantedQoSs[(*count)++] = readChar(&curdata);	}	rc = 1;exit:	FUNC_EXIT_RC(rc);	return rc;}

/** * Sends an MQTT packet in one system call write * @param socket the socket to which to write the data * @param header the one-byte MQTT header * @param buffer the rest of the buffer to write (not including remaining length) * @param buflen the length of the data in buffer to be written * @return the completion code (TCPSOCKET_COMPLETE etc) */int MQTTPacket_send(networkHandles* net, Header header, char* buffer, size_t buflen, int freeData){	int rc;	size_t buf0len;	char *buf;	FUNC_ENTRY;	buf = malloc(10);	buf[0] = header.byte;	buf0len = 1 + MQTTPacket_encode(&buf[1], buflen);#if !defined(NO_PERSISTENCE)	if (header.bits.type == PUBREL)	{		char* ptraux = buffer;		int msgId = readInt(&ptraux);		rc = MQTTPersistence_put(net->socket, buf, buf0len, 1, &buffer, &buflen,			header.bits.type, msgId, 0);	}#endif#if defined(OPENSSL)	if (net->ssl)		rc = SSLSocket_putdatas(net->ssl, net->socket, buf, buf0len, 1, &buffer, &buflen, &freeData);	else#endif		rc = Socket_putdatas(net->socket, buf, buf0len, 1, &buffer, &buflen, &freeData);			if (rc == TCPSOCKET_COMPLETE)		time(&(net->lastSent));		if (rc != TCPSOCKET_INTERRUPTED)	  free(buf);	FUNC_EXIT_RC(rc);	return rc;}

/** * Sends an MQTT packet from multiple buffers in one system call write * @param socket the socket to which to write the data * @param header the one-byte MQTT header * @param count the number of buffers * @param buffers the rest of the buffers to write (not including remaining length) * @param buflens the lengths of the data in the array of buffers to be written * @return the completion code (TCPSOCKET_COMPLETE etc) */int MQTTPacket_sends(networkHandles* net, Header header, int count, char** buffers, size_t* buflens, int* frees){	int i, rc;	size_t buf0len, total = 0;	char *buf;	FUNC_ENTRY;	buf = malloc(10);	buf[0] = header.byte;	for (i = 0; i < count; i++)		total += buflens[i];	buf0len = 1 + MQTTPacket_encode(&buf[1], total);#if !defined(NO_PERSISTENCE)	if (header.bits.type == PUBLISH && header.bits.qos != 0)	{   /* persist PUBLISH QoS1 and Qo2 */		char *ptraux = buffers[2];		int msgId = readInt(&ptraux);		rc = MQTTPersistence_put(net->socket, buf, buf0len, count, buffers, buflens,			header.bits.type, msgId, 0);	}#endif#if defined(OPENSSL)	if (net->ssl)		rc = SSLSocket_putdatas(net->ssl, net->socket, buf, buf0len, count, buffers, buflens, frees);	else#endif		rc = Socket_putdatas(net->socket, buf, buf0len, count, buffers, buflens, frees);			if (rc == TCPSOCKET_COMPLETE)		time(&(net->lastSent));		if (rc != TCPSOCKET_INTERRUPTED)	  free(buf);	FUNC_EXIT_RC(rc);	return rc;}

int MQTTSProtocol_startRegistration(Clients* client, char* topic){	int rc = 0;	FUNC_ENTRY;	if (client->outbound)		rc = MQTTSProtocol_startClientRegistration(client,topic);	else	{		PendingRegistration* pendingReg = malloc(sizeof(PendingRegistration));		Registration* reg;		int msgId = MQTTProtocol_assignMsgId(client);		char* regTopicName = malloc(strlen(topic)+1);		strcpy(regTopicName,topic);		reg = MQTTSProtocol_registerTopic(client, regTopicName);		pendingReg->msgId = msgId;		pendingReg->registration = reg;		time(&(pendingReg->sent));		client->pendingRegistration = pendingReg;		rc = MQTTSPacket_send_register(client, reg->id, regTopicName, msgId);	}	FUNC_EXIT_RC(rc);	return rc;}

int MQTTSProtocol_handleRegisters(void* pack, int sock, char* clientAddr, Clients* client){	int rc = 0;	MQTTS_Register* registerPack = (MQTTS_Register*)pack;	ListElement* elem = NULL;	int topicId = 0;	FUNC_ENTRY;	Log(LOG_PROTOCOL, 51, NULL, sock, clientAddr, client ? client->clientID : "",			registerPack->msgId, registerPack->topicId, registerPack->topicName);	if ((elem = ListFindItem(client->registrations, registerPack->topicName, registeredTopicNameCompare)) == NULL)	{		topicId = (MQTTSProtocol_registerTopic(client, registerPack->topicName))->id;		registerPack->topicName = NULL;	}	else		topicId = ((Registration*)(elem->content))->id;	rc = MQTTSPacket_send_regAck(client, registerPack->msgId, topicId, MQTTS_RC_ACCEPTED);	time( &(client->lastContact) );	MQTTSPacket_free_packet(pack);	FUNC_EXIT_RC(rc);	return rc;}

/**  * Serializes the connect options into the buffer.  * @param buf the buffer into which the packet will be serialized  * @param len the length in bytes of the supplied buffer  * @param options the options to be used to build the connect packet  * @return serialized length, or error if 0  */int MQTTSerialize_connect(unsigned char* buf, int buflen, MQTTPacket_connectData* options){	unsigned char *ptr = buf;	MQTTHeader header = {0};	MQTTConnectFlags flags = {0};	int len = 0;	int rc = -1;	FUNC_ENTRY;	if (MQTTPacket_len(len = MQTTSerialize_connectLength(options)) > buflen)	{		rc = MQTTPACKET_BUFFER_TOO_SHORT;		goto exit;	}	header.byte = 0;	header.bits.type = CONNECT;	writeChar(&ptr, header.byte); /* write header */	ptr += MQTTPacket_encode(ptr, len); /* write remaining length */	if (options->MQTTVersion == 4)	{		writeCString(&ptr, "MQTT");		writeChar(&ptr, (char) 4);	}	else	{		writeCString(&ptr, "MQIsdp");		writeChar(&ptr, (char) 3);	}	flags.all = 0;	flags.bits.cleansession = options->cleansession;	flags.bits.will = (options->willFlag) ? 1 : 0;	if (flags.bits.will)	{		flags.bits.willQoS = options->will.qos;		flags.bits.willRetain = options->will.retained;	}	if (options->username.cstring || options->username.lenstring.data)		flags.bits.username = 1;	if (options->password.cstring || options->password.lenstring.data)		flags.bits.password = 1;	writeChar(&ptr, flags.all);	writeInt(&ptr, options->keepAliveInterval);	writeMQTTString(&ptr, options->clientID);	if (options->willFlag)	{		writeMQTTString(&ptr, options->will.topicName);		writeMQTTString(&ptr, options->will.message);	}	if (flags.bits.username)		writeMQTTString(&ptr, options->username);	if (flags.bits.password)		writeMQTTString(&ptr, options->password);	rc = ptr - buf;	exit: FUNC_EXIT_RC(rc);	return rc;}

/** * Initialize the message module * @param bstate pointer to the broker state structure * @return completion code, success = 0 */int Messages_initialize(BrokerStates* bstate){	FILE* rfile = NULL;	char buf[max_msg_len];	int count = 0;	int rc = -99;	char fn[30] = "Messages_en"; /* default to English in all cases */	char* loc;	FUNC_ENTRY;	if ((loc = setlocale(LC_CTYPE, "")) == NULL)		Log(LOG_WARNING, 9989, "Can't set the native locale");    else    {    	int i;    	/* select messages file on the basis of the locale, and whether utf-8 or utf-16 is needed */		for (i = 0; i < ARRAY_SIZE(locale_map); ++i)		{			if (strncmp(locale_map[i][0], loc, strlen(locale_map[i][0])) == 0)			{				strncpy(&fn[9], locale_map[i][1], strlen(locale_map[i][1]));				break;			}		}	}	strcat(fn, ".");	strcat(fn, utf_choice);		if ((rfile = fopen(fn, "r")) == NULL)	{		char fullfn[256];		sprintf(fullfn, "..%cmessages%c%s", sep, sep, fn);		if ((rfile = fopen(fullfn, "r")) == NULL)		{			if (Messages_findMyLocation(fullfn, sizeof(fullfn)) == 0)			{				int dirlength = strlen(fullfn);								snprintf(&fullfn[dirlength], sizeof(fullfn) - dirlength, "%c%s", sep, fn);				rfile = fopen(fullfn, "r");				if (rfile == NULL)				{					snprintf(&fullfn[dirlength + 1], sizeof(fullfn) - dirlength, "..%cmessages%c%s", sep, sep, fn);					rfile = fopen(fullfn, "r");				}			}		}	}	if (rfile == NULL)		Log(LOG_WARNING, 9989, "Could not find or open message file %s", fn);	else	{		char* msg;		memset(message_list, '/0', sizeof(message_list));		while (fgets(buf, max_msg_len, rfile) != NULL && count < MESSAGE_COUNT)		{			int msgindex = 0;			if (buf[0] == '#')				continue; /* it's a comment */			msgindex = atoi(buf);			if (msgindex < ARRAY_SIZE(message_list))			{				char* start = strchr(buf, '=');				int msglen = strlen(buf);				if (start == NULL)					continue;				if (buf[msglen - 1] == '/n')					buf[--msglen] = '/0';				if (buf[msglen - 1] == '/r') /* this can happen if we read a messages file in with gcc with windows */					buf[--msglen] = '/0';				/* end of line markers */				msglen -= ++start - buf;				msg = (char*)malloc(msglen + 1);				strcpy(msg, start);				message_list[msgindex] = msg;				count++;			}		}		fclose(rfile);		if (count != MESSAGE_COUNT)			Log(LOG_WARNING, 9988, "Found %d instead of %d messages in file %s", count, MESSAGE_COUNT, fn);		else			rc = 0;	}	FUNC_EXIT_RC(rc);	return rc;}

/** * Send an MQTT CONNECT packet down a socket. * @param client a structure from which to get all the required values * @return the completion code (e.g. TCPSOCKET_COMPLETE) */int MQTTPacket_send_connect(Clients* client){	char *buf, *ptr;	Connect packet;	int rc, len;	FUNC_ENTRY;	packet.header.byte = 0;	packet.header.bits.type = CONNECT;	packet.header.bits.qos = 1;	len = 12 + strlen(client->clientID)+2;	if (client->will)		len += strlen(client->will->topic)+2 + strlen(client->will->msg)+2;	if (client->username)		len += strlen(client->username)+2;	if (client->password)		len += strlen(client->password)+2;	ptr = buf = malloc(len);	writeUTF(&ptr, "MQIsdp");	if (client->noLocal)		writeChar(&ptr, (char)-125);	else		writeChar(&ptr, (char)3);	packet.flags.all = 0;	packet.flags.bits.cleanstart = client->cleansession;	packet.flags.bits.will = (client->will) ? 1 : 0;	if (packet.flags.bits.will)	{		packet.flags.bits.willQoS = client->will->qos;		packet.flags.bits.willRetain = client->will->retained;	}	if (client->username)		packet.flags.bits.username = 1;	if (client->password)		packet.flags.bits.password = 1;	writeChar(&ptr, packet.flags.all);	writeInt(&ptr, client->keepAliveInterval);	writeUTF(&ptr, client->clientID);	if (client->will)	{		writeUTF(&ptr, client->will->topic);		writeUTF(&ptr, client->will->msg);	}	if (client->username)		writeUTF(&ptr, client->username);	if (client->password)		writeUTF(&ptr, client->password);	rc = MQTTPacket_send(client->socket, packet.header, buf, len);	Log(LOG_PROTOCOL, 0, NULL, client->socket, client->clientID, client->cleansession,			client->noLocal, rc);	free(buf);	if (rc == TCPSOCKET_COMPLETE)		time(&(client->lastContact));	FUNC_EXIT_RC(rc);	return rc;}

int Protocol_handlePublishes(Publish* publish, int sock, Clients* client, char* clientid){	int rc = TCPSOCKET_COMPLETE;#if !defined(SINGLE_LISTENER)	Listener* listener = NULL;#endif	FUNC_ENTRY;	if (Protocol_isClientQuiescing(client))		goto exit; /* don't accept new work */#if !defined(SINGLE_LISTENER)	listener = Socket_getParentListener(sock);	if (listener && listener->mount_point)	{		char* temp = malloc(strlen(publish->topic) + strlen(listener->mount_point) + 1);		strcpy(temp, listener->mount_point);		strcat(temp, publish->topic);		free(publish->topic);		publish->topic = temp;	}#endif#if !defined(NO_BRIDGE)	if (client && client->outbound)		Bridge_handleInbound(client, publish);#endif	if (publish->header.bits.qos == 0)	{		if (strlen(publish->topic) < 5 || strncmp(publish->topic, sysprefix, strlen(sysprefix)) != 0)		{			++(bstate->msgs_received);			bstate->bytes_received += publish->payloadlen;		}		Protocol_processPublication(publish, clientid);	}	else if (publish->header.bits.qos == 1)	{		/* send puback before processing the publications because a lot of return publications could fill up the socket buffer */#if defined(MQTTS)		if (client->protocol == PROTOCOL_MQTTS)			rc = MQTTSPacket_send_puback(client, publish->msgId, MQTTS_RC_ACCEPTED);		else#endif			rc = MQTTPacket_send_puback(publish->msgId, sock, clientid);		/* if we get a socket error from sending the puback, should we ignore the publication? */		Protocol_processPublication(publish, clientid);		++(bstate->msgs_received);		bstate->bytes_received += publish->payloadlen;	}	else if (publish->header.bits.qos == 2 && client->inboundMsgs->count < bstate->max_inflight_messages)	{		/* store publication in inbound list - if list is full, ignore and rely on client retry */		int len;		ListElement* listElem = NULL;		Messages* m = NULL;		Publications* p = MQTTProtocol_storePublication(publish, &len);		if ((listElem = ListFindItem(client->inboundMsgs, &publish->msgId, messageIDCompare)) != NULL)		{			m = (Messages*)(listElem->content);			MQTTProtocol_removePublication(m->publish); /* remove old publication data - could be different */		}		else			m = malloc(sizeof(Messages));		m->publish = p;		m->msgid = publish->msgId;		m->qos = publish->header.bits.qos;		m->retain = publish->header.bits.retain;		m->nextMessageType = PUBREL;		if (listElem == NULL)			ListAppend(client->inboundMsgs, m, sizeof(Messages) + len);#if defined(MQTTS)		if (client->protocol == PROTOCOL_MQTTS)			rc = MQTTSPacket_send_pubrec(client, publish->msgId);		else#endif			rc = MQTTPacket_send_pubrec(publish->msgId, sock, clientid);	}	else if (publish->header.bits.qos == 3) /* only applies to MQTT-S */	{		publish->header.bits.qos = 0;		Protocol_processPublication(publish, clientid);	}exit:	if (sock > 0)		MQTTPacket_freePublish(publish);	FUNC_EXIT_RC(rc);	return rc;}

void* MQTTSPacket_Factory(int sock, char** clientAddr, struct sockaddr* from, uint8_t** wlnid , uint8_t *wlnid_len , int* error){	static MQTTSHeader header;	void* pack = NULL;	/*struct sockaddr_in cliAddr;*/	int n;	char* data = msg;	socklen_t len = sizeof(struct sockaddr_in6);	*wlnid = NULL ;	*wlnid_len = 0 ;	FUNC_ENTRY;/* #if !defined(NO_BRIDGE)	client = Protocol_getoutboundclient(sock);	FUNC_ENTRY;	if (client!=NULL)		n = recv(sock,msg,512,0);	else #endif */	/* max message size from global parameters, as we lose the packet if we don't receive it.  Default is	 * 65535, so the parameter can be used to decrease the memory usage.	 * The message memory area must be allocated on the heap so that this memory can be not allocated	 * on reduced-memory systems.	 */	n = recvfrom(sock, msg, max_packet_size, 0, from, &len);	if (n == SOCKET_ERROR)	{		int en = Socket_error("UDP read error", sock);		if (en == EINVAL)			Log(LOG_WARNING, 0, "EINVAL");		*error = SOCKET_ERROR;		goto exit;	}	*clientAddr = Socket_getaddrname(from, sock);/*	printf("%d bytes of data on socket %d from %s/n",n,sock,*clientAddr);	if (n>0) {		for (i=0;i<n;i++) {			printf("%d ",msg[i]);		}		printf("/n");	}*/	*error = SOCKET_ERROR;  // indicate whether an error occurred, or not	if (n < 2)		goto exit;	data = MQTTSPacket_parse_header( &header, data ) ;	/* In case of Forwarder Encapsulation packet, Length: 1-octet long, specifies the number of octets up to the end	 * of the “Wireless Node Id” field (incl. the Length octet itself). Length does not include length of payload	 * (encapsulated MQTT-SN message itself).	 */	if (header.type != MQTTS_FRWDENCAP && header.len != n)    {		*error = UDPSOCKET_INCOMPLETE;		goto exit;    }	else	{		// Forwarder Encapsulation packet. Extract Wireless Node Id and MQTT-SN message		if ( header.type == MQTTS_FRWDENCAP )		{			// Skip Crt(1) field			data++ ;			// Wireless Node Id			*wlnid = data ;			// Wireless Node Id length is packet length - 3 octet (Length(1) + MsgType(1) + Crt(1))			*wlnid_len = header.len - 3 ;			data += *wlnid_len ;			// Read encapsulated packet and set header and shift data to beginning of payload			data = MQTTSPacket_parse_header( &header, data ) ;		}		uint8_t ptype = header.type;		if (ptype < MQTTS_ADVERTISE || ptype > MQTTS_WILLMSGRESP || new_mqtts_packets[ptype] == NULL)			Log(TRACE_MAX, 17, NULL, ptype);		else if ((pack = (*new_mqtts_packets[ptype])(header, data)) == NULL)			*error = BAD_MQTTS_PACKET;	}exit:   	FUNC_EXIT_RC(*error);   	return pack;}

int MQTTSProtocol_handleSubscribes(void* pack, int sock, char* clientAddr, Clients* client){	int rc = 0;	MQTTS_Subscribe* sub = (MQTTS_Subscribe*)pack;	int isnew;	int topicId = 0;	char* topicName = NULL;	FUNC_ENTRY;	Log(LOG_PROTOCOL, 67, NULL, sock, clientAddr, client ? client->clientID : "",		sub->msgId,		(sub->flags.QoS == 3) ? -1: sub->flags.QoS,		sub->flags.topicIdType);	// NORMAL (topic name is in subscribe packet) or SHORT topic name	if (sub->flags.topicIdType == MQTTS_TOPIC_TYPE_NORMAL || sub->flags.topicIdType == MQTTS_TOPIC_TYPE_SHORT)	{		topicName = sub->topicName;		sub->topicName = NULL;	}	// Pre-defined topic	else if (sub->flags.topicIdType == MQTTS_TOPIC_TYPE_PREDEFINED && client != NULL && sub->topicId != 0)	{		char *predefinedTopicName = MQTTSProtocol_getPreDefinedTopicName(client, sub->topicId) ;		// copy the topic name as it will be freed by subscription engine		topicName = malloc(strlen(predefinedTopicName)+1);		strcpy(topicName, predefinedTopicName);		topicId = sub->topicId;	}	// If topic name not found send SubAck with Rejected - Invalid topic ID	if (topicName == NULL)		rc = MQTTSPacket_send_subAck(client, sub, 0, sub->flags.QoS, MQTTS_RC_REJECTED_INVALID_TOPIC_ID);	else	{		// Topic name		if (sub->flags.topicIdType == MQTTS_TOPIC_TYPE_NORMAL && !Topics_hasWildcards(topicName))		{			char* regTopicName = malloc(strlen(topicName)+1);			strcpy(regTopicName, topicName);			topicId = (MQTTSProtocol_registerTopic(client, regTopicName))->id;		}		// Pre-defined topic		else if (sub->flags.topicIdType == MQTTS_TOPIC_TYPE_PREDEFINED)		{			char* regTopicName = malloc(strlen(topicName)+1);			strcpy(regTopicName, topicName);			MQTTSProtocol_registerPreDefinedTopic(client, topicId, regTopicName);		}		isnew = SubscriptionEngines_subscribe(bstate->se, client->clientID,				topicName, sub->flags.QoS, client->noLocal, (client->cleansession == 0), PRIORITY_NORMAL);		if ( (rc = MQTTSPacket_send_subAck(client, sub, topicId, sub->flags.QoS, MQTTS_RC_ACCEPTED)) == 0)			if ((client->noLocal == 0) || isnew)				MQTTProtocol_processRetaineds(client, topicName,sub->flags.QoS, PRIORITY_NORMAL);	}	time( &(client->lastContact) );	MQTTSPacket_free_packet(pack);	FUNC_EXIT_RC(rc);	return rc;}

int MQTTSProtocol_handlePublishes(void* pack, int sock, char* clientAddr, Clients* client){	int rc = 0;	char* topicName = NULL, *expandedPreDefinedTopicName = NULL;	MQTTS_Publish* pub = NULL;	FUNC_ENTRY;	pub = (MQTTS_Publish*)pack;	Log(LOG_PROTOCOL, 55, NULL, sock, clientAddr, client ? client->clientID : "",			pub->msgId, pub->flags.QoS, pub->flags.retain);	// Normal - registered topic	if (pub->flags.topicIdType == MQTTS_TOPIC_TYPE_NORMAL && client != NULL && pub->topicId != 0)	{		/* copy the topic name as it will be freed later */		char* name = MQTTSProtocol_getRegisteredTopicName(client, pub->topicId);		if (name)		{			topicName = malloc(strlen(name) + 1);			strcpy(topicName, name);		}	}	// Pre-defined topics	else if (pub->flags.topicIdType == MQTTS_TOPIC_TYPE_PREDEFINED && client != NULL && pub->topicId != 0)	{		/* copy the topic name as it will be freed later */		char *origPreDefinedTopicName = MQTTSProtocol_getPreDefinedTopicName(client, pub->topicId) ;		if (origPreDefinedTopicName)		{			expandedPreDefinedTopicName = MQTTSProtocol_replaceTopicNamePlaceholders(client, origPreDefinedTopicName) ;		}		// If original and expanded predef topic names are same, use expanded		// while it is already a copy of orig name		if (strcmp(origPreDefinedTopicName, expandedPreDefinedTopicName) == 0)		{			topicName = expandedPreDefinedTopicName ;		} else {			topicName = malloc(strlen(origPreDefinedTopicName)+1);			strcpy(topicName, origPreDefinedTopicName);		}	}	// Short topic names	else if (pub->flags.topicIdType == MQTTS_TOPIC_TYPE_SHORT && pub->shortTopic != NULL)	{		topicName = pub->shortTopic;		pub->shortTopic = NULL; /* will be freed in Protocol_handlePublishes */	}	// If topic name not found send PubAck with Rejected - Invalid topic ID	if (topicName == NULL)	{		rc = MQTTSPacket_send_puback(client, pub->topicId , pub->msgId, MQTTS_RC_REJECTED_INVALID_TOPIC_ID);	}	else	{		Publish* publish = malloc(sizeof(Publish));		publish->header.bits.type = PUBLISH;		publish->header.bits.qos = pub->flags.QoS;		publish->header.bits.retain = pub->flags.retain;		publish->header.bits.dup = pub->flags.dup;		publish->msgId = pub->msgId;		publish->payload = pub->data;		publish->payloadlen = pub->dataLen;		publish->topic = topicName;		rc = Protocol_handlePublishes(publish, sock, client, client ? client->clientID : clientAddr, pub->topicId);		// If predefined topic Id and predefined topic name contains [ClientId]		// publish message to expanded topic name too.		if ( pub->flags.topicIdType == MQTTS_TOPIC_TYPE_PREDEFINED && topicName != expandedPreDefinedTopicName)		{			publish = malloc(sizeof(Publish));			publish->header.bits.type = PUBLISH;			publish->header.bits.qos = pub->flags.QoS;			publish->header.bits.retain = pub->flags.retain;			publish->header.bits.dup = pub->flags.dup;			publish->msgId = pub->msgId;			publish->payload = pub->data;			publish->payloadlen = pub->dataLen;			publish->topic = expandedPreDefinedTopicName;			rc = Protocol_handlePublishes(publish, sock, client, client ? client->clientID : clientAddr, pub->topicId);		}	}	if (client != NULL)		time( &(client->lastContact) );	MQTTSPacket_free_packet(pack);	FUNC_EXIT_RC(rc);	return rc;}

//.........这里部分代码省略.........			client->connect_state = 1;			rc = MQTTSPacket_send_willTopicReq(client);		}		else		{			client->connected = 1;			rc = MQTTSPacket_send_connack(client, 0); /* send response */		}	}	else	{		/* Reconnect of a connected client */		client = (Clients*)(elem->content);		if (client->connected)		{			Log(LOG_INFO, 34, NULL, connect->clientID, clientAddr);			if (client->socket != sock)				Socket_close(client->socket);		}		client->socket = sock;		client->connected = 0; /* Do not connect until we know the connack has been sent */		client->connect_state = 0;		// Delete Wireless Node ID if exists in existing client		if ( wirelessNodeId == NULL)		{			if ( client->wirelessNodeId != NULL)				free( client->wirelessNodeId )  ;			client->wirelessNodeId = NULL ;			client->wirelessNodeIdLen = 0 ;		}		else		// Replace existing Wireless Node ID with value from current connect packet		{			if ( client->wirelessNodeId != NULL)				free ( client->wirelessNodeId )  ;			client->wirelessNodeId = malloc((sizeof(uint8_t) * wirelessNodeIdLen)) ;			memcpy( client->wirelessNodeId , wirelessNodeId , sizeof(uint8_t) * wirelessNodeIdLen) ;			client->wirelessNodeIdLen = wirelessNodeIdLen ;		}		client->good = 1;		if (client->addr != NULL)			free(client->addr);		client->addr = malloc(strlen(clientAddr)+1);		strcpy(client->addr, clientAddr);		client->cleansession = connect->flags.cleanSession;		if (client->cleansession)		{			int i;			MQTTProtocol_removeAllSubscriptions(client->clientID);			/* empty pending message lists */			MQTTProtocol_emptyMessageList(client->outboundMsgs);			MQTTProtocol_emptyMessageList(client->inboundMsgs);			for (i = 0; i < PRIORITY_MAX; ++i)				MQTTProtocol_emptyMessageList(client->queuedMsgs[i]);			MQTTProtocol_clearWill(client);		}		/* registrations are always cleared */		MQTTSProtocol_emptyRegistrationList(client->registrations);				/* have to remove and re-add client so it is in the right order for new socket */		if (client->socket != sock)		{			TreeRemoveNodeIndex(bstate->mqtts_clients, elem, 1);			TreeRemoveKeyIndex(bstate->mqtts_clients, &client->socket, 0);			client->socket = sock;			TreeAdd(bstate->mqtts_clients, client, sizeof(Clients) + strlen(client->clientID)+1 + 3*sizeof(List));		}		client->keepAliveInterval = connect->keepAlive;		client->pendingRegistration = NULL;#if !defined(NO_BRIDGE)		client->pendingSubscription = NULL;#endif		if (connect->flags.will)		{			client->connect_state = 1;			rc = MQTTSPacket_send_willTopicReq(client);		}		else		{			client->connected = 1;			rc = MQTTSPacket_send_connack(client,0); /* send response */		}	}		if (existingClient)		MQTTProtocol_processQueued(client);	Log(LOG_INFO, 0, "Client connected to udp port %d from %s (%s)", list->port, client->clientID, clientAddr);	MQTTSPacket_free_packet(pack);	time( &(client->lastContact) );exit:	FUNC_EXIT_RC(rc);	return rc;}

int MQTTSProtocol_startPublishCommon(Clients* client, Publish* mqttPublish, int dup, int qos, int retained){	int rc = 0;	Registration* registration = NULL;	MQTTS_Publish* pub = NULL;	FUNC_ENTRY;	pub = malloc(sizeof(MQTTS_Publish));	memset(pub, '/0', sizeof(MQTTS_Publish));	if (strlen(mqttPublish->topic) > 2 &&			(registration = MQTTSProtocol_getRegisteredTopicId(client, mqttPublish->topic)) == 0 && (qos != 3))	{		/* TODO: Logic elsewhere _should_ mean this case never happens... */		/*printf("I want to send a msg to %s on topic %s but it isn't registered/n",client->clientID,mqttPublish->topic); */	}	else	{		pub->header.type = MQTTS_PUBLISH;		pub->flags.QoS = qos;		pub->flags.retain = retained;		pub->flags.dup = dup;		pub->msgId = mqttPublish->msgId;		pub->data = mqttPublish->payload;		pub->shortTopic = NULL;		if (strlen(mqttPublish->topic) > 2 && qos == 3)		{			pub->topicId = strlen(mqttPublish->topic);			pub->dataLen = mqttPublish->payloadlen + strlen(mqttPublish->topic);			pub->data = malloc(pub->dataLen);			memcpy(pub->data, mqttPublish->topic, pub->topicId);			memcpy(&pub->data[pub->topicId], mqttPublish->payload, mqttPublish->payloadlen);			pub->flags.topicIdType = MQTTS_TOPIC_TYPE_NORMAL;		}		else if (mqttPublish->payloadlen > 65535)		{			/* TODO: add a message for truncated payload */			/* printf("Truncating a %d byte message sent to %s on topic %s/n",mqttPublish->payloadlen, client->clientID, mqttPublish->topic);*/			pub->dataLen = 65535;		}		else			pub->dataLen = mqttPublish->payloadlen;		pub->header.len = 7 + pub->dataLen;		if (strlen(mqttPublish->topic) < 3)		{			pub->flags.topicIdType = MQTTS_TOPIC_TYPE_SHORT;			pub->shortTopic = mqttPublish->topic;		}		else if (qos != 3)		{			pub->topicId = registration->id;			pub->flags.topicIdType = registration->topicIdType;		}		rc = MQTTSPacket_send_publish(client, pub);		if (pub->data == mqttPublish->payload)			pub->data = NULL;		pub->shortTopic = NULL;	}	MQTTSPacket_free_packet((MQTTS_Header*)pub);	FUNC_EXIT_RC(rc);	return rc;}

/** * Reads one MQTT packet from a socket. * @param socket a socket from which to read an MQTT packet * @param error pointer to the error code which is completed if no packet is returned * @return the packet structure or NULL if there was an error */void* MQTTPacket_Factory(networkHandles* net, int* error){	char* data = NULL;	static Header header;	size_t remaining_length;	int ptype;	void* pack = NULL;	size_t actual_len = 0;	FUNC_ENTRY;	*error = SOCKET_ERROR;  /* indicate whether an error occurred, or not */	/* read the packet data from the socket */#if defined(OPENSSL)	*error = (net->ssl) ? SSLSocket_getch(net->ssl, net->socket, &header.byte) : Socket_getch(net->socket, &header.byte); #else	*error = Socket_getch(net->socket, &header.byte);#endif	if (*error != TCPSOCKET_COMPLETE)   /* first byte is the header byte */		goto exit; /* packet not read, *error indicates whether SOCKET_ERROR occurred */	/* now read the remaining length, so we know how much more to read */	if ((*error = MQTTPacket_decode(net, &remaining_length)) != TCPSOCKET_COMPLETE)		goto exit; /* packet not read, *error indicates whether SOCKET_ERROR occurred */	/* now read the rest, the variable header and payload */#if defined(OPENSSL)	data = (net->ssl) ? SSLSocket_getdata(net->ssl, net->socket, remaining_length, &actual_len) : 						Socket_getdata(net->socket, remaining_length, &actual_len);#else	data = Socket_getdata(net->socket, remaining_length, &actual_len);#endif	if (data == NULL)	{		*error = SOCKET_ERROR;		goto exit; /* socket error */	}	if (actual_len != remaining_length)		*error = TCPSOCKET_INTERRUPTED;	else	{		ptype = header.bits.type;		if (ptype < CONNECT || ptype > DISCONNECT || new_packets[ptype] == NULL)			Log(TRACE_MIN, 2, NULL, ptype);		else		{			if ((pack = (*new_packets[ptype])(header.byte, data, remaining_length)) == NULL)				*error = BAD_MQTT_PACKET;#if !defined(NO_PERSISTENCE)			else if (header.bits.type == PUBLISH && header.bits.qos == 2)			{				int buf0len;				char *buf = malloc(10);				buf[0] = header.byte;				buf0len = 1 + MQTTPacket_encode(&buf[1], remaining_length);				*error = MQTTPersistence_put(net->socket, buf, buf0len, 1,					&data, &remaining_length, header.bits.type, ((Publish *)pack)->msgId, 1);				free(buf);			}#endif		}	}	if (pack)		time(&(net->lastReceived));exit:	FUNC_EXIT_RC(*error);	return pack;}

/**  * Deserializes the supplied (wire) buffer into connack data - return code  * @param sessionPresent the session present flag returned (only for MQTT 3.1.1)  * @param connack_rc returned integer value of the connack return code  * @param buf the raw buffer data, of the correct length determined by the remaining length field  * @param buflen the length in bytes of the data in the supplied buffer  * @return IoT_Error_t indicating function execution status  */static IoT_Error_t _aws_iot_mqtt_deserialize_connack(unsigned char *pSessionPresent, IoT_Error_t *pConnackRc,													 unsigned char *pRxBuf, size_t rxBufLen) {	unsigned char *curdata, *enddata;	unsigned char connack_rc_char;	uint32_t decodedLen, readBytesLen;	IoT_Error_t rc;	MQTT_Connack_Header_Flags flags = {0};	MQTTHeader header = {0};	FUNC_ENTRY;	if(NULL == pSessionPresent || NULL == pConnackRc || NULL == pRxBuf) {		FUNC_EXIT_RC(NULL_VALUE_ERROR);	}	/* CONNACK header size is fixed at two bytes for fixed and 2 bytes for variable,	 * using that as minimum size	 * MQTT v3.1.1 Specification 3.2.1 */	if(4 > rxBufLen) {		FUNC_EXIT_RC(MQTT_RX_BUFFER_TOO_SHORT_ERROR);	}	curdata = pRxBuf;	enddata = NULL;	decodedLen = 0;	readBytesLen = 0;	header.byte = aws_iot_mqtt_internal_read_char(&curdata);	if(CONNACK != header.bits.type) {		FUNC_EXIT_RC(FAILURE);	}	/* read remaining length */	rc = aws_iot_mqtt_internal_decode_remaining_length_from_buffer(curdata, &decodedLen, &readBytesLen);	if(SUCCESS != rc) {		FUNC_EXIT_RC(rc);	}	/* CONNACK remaining length should always be 2 as per MQTT 3.1.1 spec */	curdata += (readBytesLen);	enddata = curdata + decodedLen;	if(2 != (enddata - curdata)) {		FUNC_EXIT_RC(MQTT_DECODE_REMAINING_LENGTH_ERROR);	}	flags.all = aws_iot_mqtt_internal_read_char(&curdata);	*pSessionPresent = flags.bits.sessionpresent;	connack_rc_char = aws_iot_mqtt_internal_read_char(&curdata);	switch(connack_rc_char) {		case CONNACK_CONNECTION_ACCEPTED:			*pConnackRc = MQTT_CONNACK_CONNECTION_ACCEPTED;			break;		case CONNACK_UNACCEPTABLE_PROTOCOL_VERSION_ERROR:			*pConnackRc = MQTT_CONNACK_UNACCEPTABLE_PROTOCOL_VERSION_ERROR;			break;		case CONNACK_IDENTIFIER_REJECTED_ERROR:			*pConnackRc = MQTT_CONNACK_IDENTIFIER_REJECTED_ERROR;			break;		case CONNACK_SERVER_UNAVAILABLE_ERROR:			*pConnackRc = MQTT_CONNACK_SERVER_UNAVAILABLE_ERROR;			break;		case CONNACK_BAD_USERDATA_ERROR:			*pConnackRc = MQTT_CONNACK_BAD_USERDATA_ERROR;			break;		case CONNACK_NOT_AUTHORIZED_ERROR:			*pConnackRc = MQTT_CONNACK_NOT_AUTHORIZED_ERROR;			break;		default:			*pConnackRc = MQTT_CONNACK_UNKNOWN_ERROR;			break;	}	FUNC_EXIT_RC(SUCCESS);}

//.........这里部分代码省略......... * publication. * @param client the client to process queued messages for */int MQTTProtocol_processQueued(Clients* client){	int rc = 0;#if defined(QOS0_SEND_LIMIT)	int qos0count = 0;#endif	int threshold_log_message_issued = 0;	FUNC_ENTRY;	if (Protocol_isClientQuiescing(client))		goto exit; /* don't create new work - just finish in-flight stuff */	Log(TRACE_MAXIMUM, 0, NULL, client->clientID);	while (client->good && Socket_noPendingWrites(client->socket) && /* no point in starting a publish if a write is still pending */		client->outboundMsgs->count < bstate->max_inflight_messages &&		queuedMsgsCount(client) > 0#if defined(QOS0_SEND_LIMIT)		&& qos0count < bstate->max_inflight_messages /* an arbitrary criterion - but when would we restart? */#endif		#if defined(MQTTS)		&& (client->protocol == PROTOCOL_MQTT || client->outboundMsgs->count == 0)#endif		)	{		int pubrc = TCPSOCKET_COMPLETE;		Messages* m = NULL;		int threshold = (THRESHOLD * bstate->max_queued_messages) / 100;		List* queue = NULL;		int i;		for (i = PRIORITY_MAX-1; i >= 0; --i)		{			if (client->queuedMsgs[i]->count > 0)			{				queue = client->queuedMsgs[i];				break;			}		}		m = (Messages*)(queue->first->content);		Log(TRACE_MAXIMUM, 1, NULL, client->clientID);#if defined(MQTTS)		if (client->protocol == PROTOCOL_MQTTS && strlen(m->publish->topic) > 2 &&				MQTTSProtocol_getRegisteredTopicId(client, m->publish->topic) == 0)		{			if (client->pendingRegistration == NULL)				rc = MQTTSProtocol_startRegistration(client, m->publish->topic);				goto exit;		}#endif#if defined(QOS0_SEND_LIMIT)		if (m->qos == 0)			++qos0count;#endif		pubrc = MQTTProtocol_startQueuedPublish(client, m);		/* regardless of whether the publish packet was sent on the wire (pubrc is good), the		 * message has been put onto the outbound queue, so it must be removed from		 * the queuedMsgs queue		 */		if (pubrc != TCPSOCKET_COMPLETE && pubrc != TCPSOCKET_INTERRUPTED)			client->good = 0;		if (m->qos == 0)		{			/* This is done primarily for MQTT-S.			 * A qos-0 message will be on this queue if its topic			 * has to be registered first. Now that the message			 * has been sent, it needs to be cleaned up as there			 * won't be an ack to trigger it.			 *			 * For MQTT, there is a scenario in which qos-0 messages			 * could be on this list for which the same applies.			 *			 * Note (IGC): this is also a bug fix I just implemented - applies equally to MQTTs and MQTT!			 */			MQTTProtocol_removePublication(m->publish);			if (!ListRemove(queue, m))				Log(LOG_ERROR, 38, NULL);		}		else if (!ListDetach(queue, m))			Log(LOG_ERROR, 38, NULL);		if (queuedMsgsCount(client) == threshold - 1 && !threshold_log_message_issued)		{			Log(LOG_INFO, 146, NULL, client->clientID, THRESHOLD);			threshold_log_message_issued = 1;		}	}#if defined(QOS0_SEND_LIMIT)	if (qos0count >= bstate->max_inflight_messages)		rc = 1;#endifexit:	FUNC_EXIT_RC(rc);	return rc;}

/**  * Serializes the connect options into the buffer.  * @param buf the buffer into which the packet will be serialized  * @param len the length in bytes of the supplied buffer  * @param options the options to be used to build the connect packet  * @param serialized length  * @return IoT_Error_t indicating function execution status  */static IoT_Error_t _aws_iot_mqtt_serialize_connect(unsigned char *pTxBuf, size_t txBufLen,												   IoT_Client_Connect_Params *pConnectParams,												   size_t *pSerializedLen) {	unsigned char *ptr;	uint32_t len;	IoT_Error_t rc;	MQTTHeader header = {0};	MQTT_Connect_Header_Flags flags = {0};	FUNC_ENTRY;	if(NULL == pTxBuf || NULL == pConnectParams || NULL == pSerializedLen ||	   (NULL == pConnectParams->pClientID && 0 != pConnectParams->clientIDLen) ||	   (NULL != pConnectParams->pClientID && 0 == pConnectParams->clientIDLen)) {		FUNC_EXIT_RC(NULL_VALUE_ERROR);	}	/* Check needed here before we start writing to the Tx buffer */	switch(pConnectParams->MQTTVersion) {		case MQTT_3_1_1:			break;		default:			return MQTT_CONNACK_UNACCEPTABLE_PROTOCOL_VERSION_ERROR;	}	ptr = pTxBuf;	len = _aws_iot_get_connect_packet_length(pConnectParams);	if(aws_iot_mqtt_internal_get_final_packet_length_from_remaining_length(len) > txBufLen) {		FUNC_EXIT_RC(MQTT_TX_BUFFER_TOO_SHORT_ERROR);	}	rc = aws_iot_mqtt_internal_init_header(&header, CONNECT, QOS0, 0, 0);	if(SUCCESS != rc) {		FUNC_EXIT_RC(rc);	}	aws_iot_mqtt_internal_write_char(&ptr, header.byte); /* write header */	ptr += aws_iot_mqtt_internal_write_len_to_buffer(ptr, len); /* write remaining length */	// Enable if adding support for more versions	//if(MQTT_3_1_1 == pConnectParams->MQTTVersion) {	aws_iot_mqtt_internal_write_utf8_string(&ptr, "MQTT", 4);	aws_iot_mqtt_internal_write_char(&ptr, (unsigned char) pConnectParams->MQTTVersion);	//}	flags.all = 0;	flags.bits.cleansession = (pConnectParams->isCleanSession) ? 1 : 0;	flags.bits.will = (pConnectParams->isWillMsgPresent) ? 1 : 0;	if(flags.bits.will) {		flags.bits.willQoS = pConnectParams->will.qos;		flags.bits.willRetain = (pConnectParams->will.isRetained) ? 1 : 0;	}	if(pConnectParams->pUsername) {		flags.bits.username = 1;	}	if(pConnectParams->pPassword) {		flags.bits.password = 1;	}	aws_iot_mqtt_internal_write_char(&ptr, flags.all);	aws_iot_mqtt_internal_write_uint_16(&ptr, pConnectParams->keepAliveIntervalInSec);	/* If the code have passed the check for incorrect values above, no client id was passed as argument */	if(NULL == pConnectParams->pClientID) {		aws_iot_mqtt_internal_write_uint_16(&ptr, 0);	} else {		aws_iot_mqtt_internal_write_utf8_string(&ptr, pConnectParams->pClientID, pConnectParams->clientIDLen);	}	if(pConnectParams->isWillMsgPresent) {		aws_iot_mqtt_internal_write_utf8_string(&ptr, pConnectParams->will.pTopicName,												pConnectParams->will.topicNameLen);		aws_iot_mqtt_internal_write_utf8_string(&ptr, pConnectParams->will.pMessage, pConnectParams->will.msgLen);	}	if(flags.bits.username) {		aws_iot_mqtt_internal_write_utf8_string(&ptr, pConnectParams->pUsername, pConnectParams->usernameLen);	}	if(flags.bits.password) {		aws_iot_mqtt_internal_write_utf8_string(&ptr, pConnectParams->pPassword, pConnectParams->passwordLen);	}	*pSerializedLen = (size_t) (ptr - pTxBuf);	FUNC_EXIT_RC(SUCCESS);}

int keysWin32(char *dirname, char ***keys, int *nkeys){	int rc = 0;	char **fkeys = NULL;	int nfkeys = 0;	char dir[MAX_PATH+1];	WIN32_FIND_DATAA FileData;	HANDLE hDir;	int fFinished = 0;	char *ptraux;	int i;	FUNC_ENTRY;	sprintf(dir, "%s/*", dirname);	/* get number of keys */	hDir = FindFirstFileA(dir, &FileData);	if (hDir != INVALID_HANDLE_VALUE)	{		while (!fFinished)		{			if (FileData.dwFileAttributes & FILE_ATTRIBUTE_ARCHIVE)				nfkeys++;			if (!FindNextFileA(hDir, &FileData))			{				if (GetLastError() == ERROR_NO_MORE_FILES)					fFinished = 1;			}		}		FindClose(hDir);	} else	{		rc = MQTTCLIENT_PERSISTENCE_ERROR;		goto exit;	}	if (nfkeys != 0 )		fkeys = (char **)malloc(nfkeys * sizeof(char *));	/* copy the keys */	hDir = FindFirstFileA(dir, &FileData);	if (hDir != INVALID_HANDLE_VALUE)	{		fFinished = 0;		i = 0;		while (!fFinished)		{			if (FileData.dwFileAttributes & FILE_ATTRIBUTE_ARCHIVE)			{				fkeys[i] = malloc(strlen(FileData.cFileName) + 1);				strcpy(fkeys[i], FileData.cFileName);				ptraux = strstr(fkeys[i], MESSAGE_FILENAME_EXTENSION);				if ( ptraux != NULL )					*ptraux = '/0' ;				i++;			}			if (!FindNextFileA(hDir, &FileData))			{				if (GetLastError() == ERROR_NO_MORE_FILES)					fFinished = 1;			}		}		FindClose(hDir);	} else	{		rc = MQTTCLIENT_PERSISTENCE_ERROR;		goto exit;	}	*nkeys = nfkeys;	*keys = fkeys;	/* the caller must free keys */exit:	FUNC_EXIT_RC(rc);	return rc;}

/** * @brief MQTT Connection Function * * Called to establish an MQTT connection with the AWS IoT Service * This is the internal function which is called by the connect API to perform the operation. * Not meant to be called directly as it doesn't do validations or client state changes * * @param pClient Reference to the IoT Client * @param pConnectParams Pointer to MQTT connection parameters * * @return An IoT Error Type defining successful/failed connection */static IoT_Error_t _aws_iot_mqtt_internal_connect(AWS_IoT_Client *pClient, IoT_Client_Connect_Params *pConnectParams) {	Timer connect_timer;	IoT_Error_t connack_rc = FAILURE;	char sessionPresent = 0;	size_t len = 0;	IoT_Error_t rc = FAILURE;	FUNC_ENTRY;	if(NULL != pConnectParams) {		/* override default options if new options were supplied */		rc = aws_iot_mqtt_set_connect_params(pClient, pConnectParams);		if(SUCCESS != rc) {			FUNC_EXIT_RC(MQTT_CONNECTION_ERROR);		}	}	rc = pClient->networkStack.connect(&(pClient->networkStack), NULL);	if(SUCCESS != rc) {		/* TLS Connect failed, return error */		FUNC_EXIT_RC(rc);	}	init_timer(&connect_timer);	countdown_ms(&connect_timer, pClient->clientData.commandTimeoutMs);	pClient->clientData.keepAliveInterval = pClient->clientData.options.keepAliveIntervalInSec;	rc = _aws_iot_mqtt_serialize_connect(pClient->clientData.writeBuf, pClient->clientData.writeBufSize,										 &(pClient->clientData.options), &len);	if(SUCCESS != rc || 0 >= len) {		FUNC_EXIT_RC(rc);	}	/* send the connect packet */	rc = aws_iot_mqtt_internal_send_packet(pClient, len, &connect_timer);	if(SUCCESS != rc) {		FUNC_EXIT_RC(rc);	}	/* this will be a blocking call, wait for the CONNACK */	rc = aws_iot_mqtt_internal_wait_for_read(pClient, CONNACK, &connect_timer);	if(SUCCESS != rc) {		FUNC_EXIT_RC(rc);	}	/* Received CONNACK, check the return code */	rc = _aws_iot_mqtt_deserialize_connack((unsigned char *) &sessionPresent, &connack_rc, pClient->clientData.readBuf,										   pClient->clientData.readBufSize);	if(SUCCESS != rc) {		FUNC_EXIT_RC(rc);	}	if(MQTT_CONNACK_CONNECTION_ACCEPTED != connack_rc) {		FUNC_EXIT_RC(connack_rc);	}	pClient->clientStatus.isPingOutstanding = false;	countdown_sec(&pClient->pingTimer, pClient->clientData.keepAliveInterval);	FUNC_EXIT_RC(SUCCESS);}

int keysUnix(char *dirname, char ***keys, int *nkeys){	int rc = 0;	char **fkeys = NULL;	int nfkeys = 0;	char *ptraux;	int i;	DIR *dp;	struct dirent *dir_entry;	struct stat stat_info;	FUNC_ENTRY;	/* get number of keys */	if((dp = opendir(dirname)) != NULL)	{		while((dir_entry = readdir(dp)) != NULL)		{			char* temp = malloc(strlen(dirname)+strlen(dir_entry->d_name)+2);			sprintf(temp, "%s/%s", dirname, dir_entry->d_name);			if (lstat(temp, &stat_info) == 0 && S_ISREG(stat_info.st_mode))				nfkeys++;			free(temp);		}		closedir(dp);	} else	{		rc = MQTTCLIENT_PERSISTENCE_ERROR;		goto exit;	}	if (nfkeys != 0 )		fkeys = (char **)malloc(nfkeys * sizeof(char *));	/* copy the keys */	if((dp = opendir(dirname)) != NULL)	{		i = 0;		while((dir_entry = readdir(dp)) != NULL)		{			char* temp = malloc(strlen(dirname)+strlen(dir_entry->d_name)+2);			sprintf(temp, "%s/%s", dirname, dir_entry->d_name);			if (lstat(temp, &stat_info) == 0 && S_ISREG(stat_info.st_mode))			{				fkeys[i] = malloc(strlen(dir_entry->d_name) + 1);				strcpy(fkeys[i], dir_entry->d_name);				ptraux = strstr(fkeys[i], MESSAGE_FILENAME_EXTENSION);				if ( ptraux != NULL )					*ptraux = '/0' ;				i++;			}			free(temp);		}		closedir(dp);	} else	{		rc = MQTTCLIENT_PERSISTENCE_ERROR;		goto exit;	}	*nkeys = nfkeys;	*keys = fkeys;	/* the caller must free keys */exit:	FUNC_EXIT_RC(rc);	return rc;}