物联网MQTT在ESP32上实现连接的工程代码

/* PubSubClient.cpp - A simple client for MQTT. Nick O'Leary http://knolleary.net */ #include "PubSubClient.h" #include "Arduino.h" PubSubClient::PubSubClient() { this->_state = MQTT_DISCONNECTED; this->_client = NULL; this->stream = NULL; setCallback(NULL); this->bufferSize = 0; setBufferSize(MQTT_MAX_PACKET_SIZE); setKeepAlive(MQTT_KEEPALIVE); setSocketTimeout(MQTT_SOCKET_TIMEOUT); } PubSubClient::PubSubClient(Client& client) { this->_state = MQTT_DISCONNECTED; setClient(client); this->stream = NULL; this->bufferSize = 0; setBufferSize(MQTT_MAX_PACKET_SIZE); setKeepAlive(MQTT_KEEPALIVE); setSocketTimeout(MQTT_SOCKET_TIMEOUT); } PubSubClient::PubSubClient(IPAddress addr, uint16_t port, Client& client) { this->_state = MQTT_DISCONNECTED; setServer(addr, port); setClient(client); this->stream = NULL; this->bufferSize = 0; setBufferSize(MQTT_MAX_PACKET_SIZE); setKeepAlive(MQTT_KEEPALIVE); setSocketTimeout(MQTT_SOCKET_TIMEOUT); } PubSubClient::PubSubClient(IPAddress addr, uint16_t port, Client& client, Stream& stream) { this->_state = MQTT_DISCONNECTED; setServer(addr,port); setClient(client); setStream(stream); this->bufferSize = 0; setBufferSize(MQTT_MAX_PACKET_SIZE); setKeepAlive(MQTT_KEEPALIVE); setSocketTimeout(MQTT_SOCKET_TIMEOUT); } PubSubClient::PubSubClient(IPAddress addr, uint16_t port, MQTT_CALLBACK_SIGNATURE, Client& client) { this->_state = MQTT_DISCONNECTED; setServer(addr, port); setCallback(callback); setClient(client); this->stream = NULL; this->bufferSize = 0; setBufferSize(MQTT_MAX_PACKET_SIZE); setKeepAlive(MQTT_KEEPALIVE); setSocketTimeout(MQTT_SOCKET_TIMEOUT); } PubSubClient::PubSubClient(IPAddress addr, uint16_t port, MQTT_CALLBACK_SIGNATURE, Client& client, Stream& stream) { this->_state = MQTT_DISCONNECTED; setServer(addr,port); setCallback(callback); setClient(client); setStream(stream); this->bufferSize = 0; setBufferSize(MQTT_MAX_PACKET_SIZE); setKeepAlive(MQTT_KEEPALIVE); setSocketTimeout(MQTT_SOCKET_TIMEOUT); } PubSubClient::PubSubClient(uint8_t *ip, uint16_t port, Client& client) { this->_state = MQTT_DISCONNECTED; setServer(ip, port); setClient(client); this->stream = NULL; this->bufferSize = 0; setBufferSize(MQTT_MAX_PACKET_SIZE); setKeepAlive(MQTT_KEEPALIVE); setSocketTimeout(MQTT_SOCKET_TIMEOUT); } PubSubClient::PubSubClient(uint8_t *ip, uint16_t port, Client& client, Stream& stream) { this->_state = MQTT_DISCONNECTED; setServer(ip,port); setClient(client); setStream(stream); this->bufferSize = 0; setBufferSize(MQTT_MAX_PACKET_SIZE); setKeepAlive(MQTT_KEEPALIVE); setSocketTimeout(MQTT_SOCKET_TIMEOUT); } PubSubClient::PubSubClient(uint8_t *ip, uint16_t port, MQTT_CALLBACK_SIGNATURE, Client& client) { this->_state = MQTT_DISCONNECTED; setServer(ip, port); setCallback(callback); setClient(client); this->stream = NULL; this->bufferSize = 0; setBufferSize(MQTT_MAX_PACKET_SIZE); setKeepAlive(MQTT_KEEPALIVE); setSocketTimeout(MQTT_SOCKET_TIMEOUT); } PubSubClient::PubSubClient(uint8_t *ip, uint16_t port, MQTT_CALLBACK_SIGNATURE, Client& client, Stream& stream) { this->_state = MQTT_DISCONNECTED; setServer(ip,port); setCallback(callback); setClient(client); setStream(stream); this->bufferSize = 0; setBufferSize(MQTT_MAX_PACKET_SIZE); setKeepAlive(MQTT_KEEPALIVE); setSocketTimeout(MQTT_SOCKET_TIMEOUT); } PubSubClient::PubSubClient(const char* domain, uint16_t port, Client& client) { this->_state = MQTT_DISCONNECTED; setServer(domain,port); setClient(client); this->stream = NULL; this->bufferSize = 0; setBufferSize(MQTT_MAX_PACKET_SIZE); setKeepAlive(MQTT_KEEPALIVE); setSocketTimeout(MQTT_SOCKET_TIMEOUT); } PubSubClient::PubSubClient(const char* domain, uint16_t port, Client& client, Stream& stream) { this->_state = MQTT_DISCONNECTED; setServer(domain,port); setClient(client); setStream(stream); this->bufferSize = 0; setBufferSize(MQTT_MAX_PACKET_SIZE); setKeepAlive(MQTT_KEEPALIVE); setSocketTimeout(MQTT_SOCKET_TIMEOUT); } PubSubClient::PubSubClient(const char* domain, uint16_t port, MQTT_CALLBACK_SIGNATURE, Client& client) { this->_state = MQTT_DISCONNECTED; setServer(domain,port); setCallback(callback); setClient(client); this->stream = NULL; this->bufferSize = 0; setBufferSize(MQTT_MAX_PACKET_SIZE); setKeepAlive(MQTT_KEEPALIVE); setSocketTimeout(MQTT_SOCKET_TIMEOUT); } PubSubClient::PubSubClient(const char* domain, uint16_t port, MQTT_CALLBACK_SIGNATURE, Client& client, Stream& stream) { this->_state = MQTT_DISCONNECTED; setServer(domain,port); setCallback(callback); setClient(client); setStream(stream); this->bufferSize = 0; setBufferSize(MQTT_MAX_PACKET_SIZE); setKeepAlive(MQTT_KEEPALIVE); setSocketTimeout(MQTT_SOCKET_TIMEOUT); } PubSubClient::~PubSubClient() { free(this->buffer); } boolean PubSubClient::connect(const char *id) { return connect(id,NULL,NULL,0,0,0,0,1); } boolean PubSubClient::connect(const char *id, const char *user, const char *pass) { return connect(id,user,pass,0,0,0,0,1); } boolean PubSubClient::connect(const char *id, const char* willTopic, uint8_t willQos, boolean willRetain, const char* willMessage) { return connect(id,NULL,NULL,willTopic,willQos,willRetain,willMessage,1); } boolean PubSubClient::connect(const char *id, const char *user, const char *pass, const char* willTopic, uint8_t willQos, boolean willRetain, const char* willMessage) { return connect(id,user,pass,willTopic,willQos,willRetain,willMessage,1); } boolean PubSubClient::connect(const char *id, const char *user, const char *pass, const char* willTopic, uint8_t willQos, boolean willRetain, const char* willMessage, boolean cleanSession) { if (!connected()) { int result = 0; if(_client->connected()) { result = 1; } else { if (domain != NULL) { result = _client->connect(this->domain, this->port); } else { result = _client->connect(this->ip, this->port); } } if (result == 1) { nextMsgId = 1; // Leave room in the buffer for header and variable length field uint16_t length = MQTT_MAX_HEADER_SIZE; unsigned int j; #if MQTT_VERSION == MQTT_VERSION_3_1 uint8_t d[9] = {0x00,0x06,'M','Q','I','s','d','p', MQTT_VERSION}; #define MQTT_HEADER_VERSION_LENGTH 9 #elif MQTT_VERSION == MQTT_VERSION_3_1_1 uint8_t d[7] = {0x00,0x04,'M','Q','T','T',MQTT_VERSION}; #define MQTT_HEADER_VERSION_LENGTH 7 #endif for (j = 0;j<MQTT_HEADER_VERSION_LENGTH;j++) { this->buffer[length++] = d[j]; } uint8_t v; if (willTopic) { v = 0x04|(willQos<<3)|(willRetain<<5); } else { v = 0x00; } if (cleanSession) { v = v|0x02; } if(user != NULL) { v = v|0x80; if(pass != NULL) { v = v|(0x80>>1); } } this->buffer[length++] = v; this->buffer[length++] = ((this->keepAlive) >> 8); this->buffer[length++] = ((this->keepAlive) & 0xFF); CHECK_STRING_LENGTH(length,id) length = writeString(id,this->buffer,length); if (willTopic) { CHECK_STRING_LENGTH(length,willTopic) length = writeString(willTopic,this->buffer,length); CHECK_STRING_LENGTH(length,willMessage) length = writeString(willMessage,this->buffer,length);