amr/mqtt
1
1
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Rewritten code works! See full log.

Browse source 
After much consternation, finally got the code generalized. For
some reason the ESP8266 absolutely cannot pass pointers between
functions. I'm not sure if this is because it's a timer function?
Either way, we have to memcpy the values, which makes things a
bit messier, but not too bad.

Now to start work on Publish...
This commit is contained in:
A.M. Rowsell 2018-08-16 19:11:47 -04:00
commit bcbcde455a
Signed by: amr
GPG key ID: 0B6E2D8375CF79A9
2 changed files with 130 additions and 84 deletions
Download patch file Download diff file Expand all files Collapse all files

208
mqtt.c
View file

@ -8,6 +8,7 @@
#include "os_type.h" #include "os_type.h"
#include "mqtt.h" #include "mqtt.h"
#define DEBUG 1
/* Functions we will need to implement: /* Functions we will need to implement:
* Send -- will handle all sending of all packets * Send -- will handle all sending of all packets
* Connect -- set up TCP connection and parameters * Connect -- set up TCP connection and parameters
@ -87,6 +88,33 @@ LOCAL uint8_t ICACHE_FLASH_ATTR mqtt_connect(mqtt_session_t *session, uint8_t *u
} }
/************************************************************************************************/
/* Function: encodeLength */
/* Parameters: the length in bytes */
/* Returns: encoded multi-byte number per MQTT standards. The first byte this pointer */
/* returns is the number of bytes to follow. */
/* The MQTT standard uses a very strange method of encoding the lengths of the various sections */
/* of the packets. This makes it simpler to implement in code. */
/************************************************************************************************/
LOCAL uint8_t ICACHE_FLASH_ATTR *encodeLength(uint32_t trueLength) {
uint8_t *encodedByte = os_zalloc(sizeof(uint8_t) * 5); // can't be more than 5 bytes
uint8_t numBytes = 1;
do {
encodedByte[numBytes] = trueLength % 128;
trueLength /= 128;
if(trueLength > 0) {
encodedByte[numBytes] |= 128;
}
numBytes++;
} while(trueLength > 0);
encodedByte[0] = numBytes - 1;
return encodedByte;
}
/*************************************************************************************/ /*************************************************************************************/
/* Function: mqtt_send */ /* Function: mqtt_send */
/* Parameters: active mqtt_session_t, pointer to data, data length, and message type */ /* Parameters: active mqtt_session_t, pointer to data, data length, and message type */
@ -98,92 +126,92 @@ LOCAL uint8_t ICACHE_FLASH_ATTR mqtt_send(mqtt_session_t *session, uint8_t *data
/** First thing to do is check the packet type. /** First thing to do is check the packet type.
* This will inform everything else we do * This will inform everything else we do
*/ */
#ifdef DEBUG
os_printf("Entering mqtt_send!\n"); os_printf("Entering mqtt_send!\n");
#endif
LOCAL mqtt_packet_t packet; LOCAL mqtt_packet_t packet;
LOCAL mqtt_packet_t *pPacket = &packet; LOCAL mqtt_packet_t *pPacket = &packet;
uint8_t *fullPacket; uint8_t *fullPacket;
switch(msgType) { switch(msgType) {
case MQTT_MSG_TYPE_CONNECT: case MQTT_MSG_TYPE_CONNECT: ;
// prepare packet for connect type // prepare packet for connect type
pPacket->fixedHeader = (uint8_t *)os_zalloc(sizeof(uint8_t) * 2); // fixed header cannot be longer than 5 const uint8_t varDefaults[10] = { 0x00, 0x04, 0x4D, 0x51, 0x54, 0x54, 0x04, 0xC2, 0x00, 0x32 };
pPacket->fixedHeader[0] = ((uint8_t)msgType << 4) & 0xF0; // make sure lower 4 are clear pPacket->fixedHeader = (uint8_t *)os_zalloc(sizeof(uint8_t) * 5); // fixed header cannot be longer than 5
// prepare variable header pPacket->fixedHeader[0] = ((uint8_t)msgType << 4) & 0xF0; // make sure lower 4 are clear
pPacket->varHeader = os_zalloc(sizeof(uint8_t) * 10); // 10 bytes for connect // prepare variable header
pPacket->varHeader[0] = 0; pPacket->varHeader = os_zalloc(sizeof(uint8_t) * 10); // 10 bytes for connect
pPacket->varHeader[1] = 4; // length of protocol name os_memcpy(pPacket->varHeader, varDefaults, 10); // copy defaults
pPacket->varHeader[2] = 0x4D; // M pPacket->varHeader_len = 10;
pPacket->varHeader[3] = 0x51; // Q // prepare payload
pPacket->varHeader[4] = 0x54; // T uint32_t offset = 0;
pPacket->varHeader[5] = 0x54; // T uint32_t maxPayloadLength = session->client_id_len + session->username_len + session->password_len + 12; // length can only take a maximum of 12 bytes
pPacket->varHeader[6] = 4; // protocol level pPacket->payload = (uint8_t *)os_zalloc(sizeof(uint8_t) * maxPayloadLength);
pPacket->varHeader[7] = 0b11000010; // connect flags, no QOS or will, clean session, user/pass // copy client id to payload
pPacket->varHeader[8] = 0; // insert a 0
pPacket->varHeader[9] = 10; // keepalive time os_memset(pPacket->payload + offset, 0, 1);
pPacket->varHeader_len = 10; offset += 1;
// prepare payload os_memcpy(pPacket->payload + offset, &session->client_id_len, 1);
pPacket->payload = (uint8_t *)os_zalloc(sizeof(uint8_t) * 26); offset += 1;
// client identifier "FRZN0" os_memcpy(pPacket->payload + offset, session->client_id, session->client_id_len);
pPacket->payload[0] = 0; offset += session->client_id_len;
pPacket->payload[1] = 5; // copy username to payload
pPacket->payload[2] = 0x46; // F // insert a 0
pPacket->payload[3] = 0x52; // R os_memset(pPacket->payload + offset, 0, 1);
pPacket->payload[4] = 0x5A; // Z offset += 1;
pPacket->payload[5] = 0x4E; // N os_memcpy(pPacket->payload + offset, &session->username_len, 1);
pPacket->payload[6] = 0x30; // 0 offset += 1;
// Username os_memcpy(pPacket->payload + offset, session->username, session->username_len);
pPacket->payload[7] = 0; offset += session->username_len;
pPacket->payload[8] = 11; // Password
// os_memcpy(pPacket->payload + 9, data, username_len); // insert a 0
pPacket->payload[9] = 0x4D; // M os_memset(pPacket->payload + offset, 0, 1);
pPacket->payload[10] = 0x72; // r offset += 1;
pPacket->payload[11] = 0x41; // A os_memcpy(pPacket->payload + offset, &session->password_len, 1);
pPacket->payload[12] = 0x75; // u offset += 1;
pPacket->payload[13] = 0x72; // r os_memcpy(pPacket->payload + offset, session->password, session->password_len);
pPacket->payload[14] = 0x65; // e offset += session->password_len;
pPacket->payload[15] = 0x6C; // l pPacket->payload_len = offset;
pPacket->payload[16] = 0x69; // i os_printf("Total offset: %d", offset);
pPacket->payload[17] = 0x75; // u uint8_t *remaining_len_encoded = encodeLength((uint32_t)(pPacket->varHeader_len + pPacket->payload_len));
pPacket->payload[18] = 0x73; // s os_memcpy(pPacket->fixedHeader + 1, remaining_len_encoded + 1, remaining_len_encoded[0]); // remaining length = length of varHeader + length of payload
pPacket->payload[19] = 0x52; // R pPacket->fixedHeader_len = remaining_len_encoded[0] + 1;
// Password // the full length is the length of the varHeader, payload, and fixedHeader
pPacket->payload[20] = 0; pPacket->length = (pPacket->varHeader_len + pPacket->payload_len + pPacket->fixedHeader_len);
pPacket->payload[21] = 4; os_printf("Packet length: %d\n", pPacket->length);
// os_memcpy(pPacket->payload + 22, data + offset, password_len); // construct packet data
pPacket->payload[22] = 0x74; // t fullPacket = os_zalloc(sizeof(uint8_t) * pPacket->length);
pPacket->payload[23] = 0x65; // e os_memcpy(fullPacket, pPacket->fixedHeader, pPacket->fixedHeader_len);
pPacket->payload[24] = 0x73; // s os_memcpy(fullPacket + pPacket->fixedHeader_len, pPacket->varHeader, pPacket->varHeader_len);
pPacket->payload[25] = 0x74; // t os_memcpy(fullPacket + pPacket->fixedHeader_len + pPacket->varHeader_len, pPacket->payload, pPacket->payload_len);
pPacket->payload_len = 26; break;
pPacket->fixedHeader[1] = 36; // length of varHeader + payload_len case MQTT_MSG_TYPE_PUBLISH:
pPacket->length = 38;
fullPacket = os_zalloc(sizeof(uint8_t) * pPacket->length); //full packet length is 38
os_memcpy(fullPacket, pPacket->fixedHeader, 2);
os_memcpy(fullPacket + 2, pPacket->varHeader, 10);
os_memcpy(fullPacket + 12, pPacket->payload, 26);
break;
case MQTT_MSG_TYPE_PUBLISH:
// prepare for publish // prepare for publish
break; break;
case MQTT_MSG_TYPE_SUBSCRIBE: case MQTT_MSG_TYPE_SUBSCRIBE:
// prepare for subscribe // prepare for subscribe
break; break;
case MQTT_MSG_TYPE_UNSUBSCRIBE: case MQTT_MSG_TYPE_UNSUBSCRIBE:
// prepare for unsubscribe // prepare for unsubscribe
break; break;
case MQTT_MSG_TYPE_PINGREQ: case MQTT_MSG_TYPE_PINGREQ:
// prepare for ping // prepare for ping
break; break;
case MQTT_MSG_TYPE_DISCONNECT: case MQTT_MSG_TYPE_DISCONNECT:
// disconnect from broker // disconnect from broker
break; break;
default: default:
// something has gone wrong // something has gone wrong
os_printf("Attempt to send incorrect packet type: %d", (uint8_t)msgType); os_printf("Attempt to send incorrect packet type: %d", (uint8_t)msgType);
return -1; return -1;
} }
#ifdef DEBUG
os_printf("About to send MQTT connect...\n"); os_printf("About to send MQTT connect...\n");
#endif
espconn_send(session->activeConnection, fullPacket, pPacket->length); espconn_send(session->activeConnection, fullPacket, pPacket->length);
os_free(fullPacket); os_free(fullPacket);
os_free(pPacket->fixedHeader);
os_free(pPacket->varHeader);
os_free(pPacket->payload);
} }
LOCAL void ICACHE_FLASH_ATTR test(void *arg) { LOCAL void ICACHE_FLASH_ATTR test(void *arg) {
@ -209,17 +237,31 @@ void ICACHE_FLASH_ATTR user_init() {
// prepare the TCP/MQTT connection stuff // prepare the TCP/MQTT connection stuff
// test server is at 51.15.65.206 // test server is at 51.15.65.206
char testUser[] = "MrAureliusR"; char testUser[11] = { 0x4d, 0x72, 0x41, 0x75, 0x72, 0x65, 0x6c, 0x69, 0x75, 0x73, 0x52 };
char testPass[] = "test"; char *pTestUser = testUser;
char testTopic[] = "input"; char testPass[4] = { 0x74, 0x65, 0x73, 0x74 };
pGlobalSession->port = 1883; // port 80 just for testing char *pTestPass = testPass;
char testTopic[5] = "input";
char clientID[5] = {0x46, 0x52, 0x5a, 0x4e, 0x30 };
char *pClientID = clientID;
pGlobalSession->port = 1883; // mqtt port
const char esp_tcp_server_ip[4] = {51, 15, 65, 206}; // remote IP of TCP server const char esp_tcp_server_ip[4] = {51, 15, 65, 206}; // remote IP of TCP server
os_memcpy(pGlobalSession->ip, esp_tcp_server_ip, 4); os_memcpy(pGlobalSession->ip, esp_tcp_server_ip, 4);
pGlobalSession->username = &testUser[0]; pGlobalSession->username_len = 11;
pGlobalSession->password = &testPass[0]; pGlobalSession->username = os_zalloc(sizeof(uint8_t) * pGlobalSession->username_len);
pGlobalSession->topic_name = &testTopic[0]; os_memcpy(pGlobalSession->username, testUser, pGlobalSession->username_len);
pGlobalSession->password_len = 4;
pGlobalSession->password = os_zalloc(sizeof(uint8_t) * pGlobalSession->password_len);
os_memcpy(pGlobalSession->password, testPass, pGlobalSession->password_len);
pGlobalSession->topic_name_len = 5;
pGlobalSession->topic_name = os_zalloc(sizeof(uint8_t) * pGlobalSession->topic_name_len);
os_memcpy(pGlobalSession->topic_name, testTopic, pGlobalSession->topic_name_len);
pGlobalSession->client_id_len = 5;
pGlobalSession->client_id = os_zalloc(sizeof(uint8_t) * pGlobalSession->client_id_len);
os_memcpy(pGlobalSession->client_id, clientID, pGlobalSession->client_id_len);
os_timer_setfn(&oneTimer, (os_timer_func_t *)tcpConnect, pGlobalSession); os_timer_setfn(&oneTimer, (os_timer_func_t *)tcpConnect, pGlobalSession);
os_timer_arm(&oneTimer, 15000, 0); os_timer_arm(&oneTimer, 15000, 0);
os_timer_setfn(&testTimer, (os_timer_func_t *)test, pGlobalSession); os_timer_setfn(&testTimer, (os_timer_func_t *)test, pGlobalSession);
os_timer_arm(&testTimer, 16000, 0); os_timer_arm(&testTimer, 17000, 0);
} }

6
mqtt.h
View file

@ -39,10 +39,14 @@ typedef struct {
uint32_t port; uint32_t port;
uint32_t localPort; uint32_t localPort;
uint8_t *client_id; uint8_t *client_id;
uint32_t client_id_len;
uint8_t *topic_name; uint8_t *topic_name;
uint32_t topic_name_len;
uint8_t qos_level; uint8_t qos_level;
uint8_t *username; uint8_t *username;
uint32_t username_len;
uint8_t *password; uint8_t *password;
uint32_t password_len;
char valid_connection; char valid_connection;
struct espconn *activeConnection; struct espconn *activeConnection;
} mqtt_session_t; } mqtt_session_t;
@ -54,4 +58,4 @@ LOCAL void ICACHE_FLASH_ATTR connected_callback(void *arg);
LOCAL void ICACHE_FLASH_ATTR data_recv_callback(void *arg, char *pdata, unsigned short len); LOCAL void ICACHE_FLASH_ATTR data_recv_callback(void *arg, char *pdata, unsigned short len);
LOCAL void ICACHE_FLASH_ATTR data_sent_callback(void *arg); LOCAL void ICACHE_FLASH_ATTR data_sent_callback(void *arg);
LOCAL uint8_t ICACHE_FLASH_ATTR mqtt_send(mqtt_session_t *session, uint8_t *data, uint32_t len, mqtt_message_type msgType); LOCAL uint8_t ICACHE_FLASH_ATTR mqtt_send(mqtt_session_t *session, uint8_t *data, uint32_t len, mqtt_message_type msgType);
LOCAL uint8_t ICACHE_FLASH_ATTR mqtt_connect(mqtt_session_t *session); LOCAL uint8_t ICACHE_FLASH_ATTR mqtt_connect(mqtt_session_t *session, uint8_t *username, uint8_t *password);