mqtt/mqtt.c

// This Source Code Form is subject to the terms of the Mozilla Public
// License, v. 2.0. If a copy of the MPL was not distributed with this
// file, You can obtain one at http://mozilla.org/MPL/2.0/.

#include <stdint.h>
#include "user_interface.h"
#include "ets_sys.h"
#include "osapi.h"
#include "mem.h"
#include "gpio.h"
#include "espconn.h"
#include "os_type.h"
#include "mqtt.h"

#define DEBUG 1
/* Functions we will need to implement:
 * Send -- will handle all sending of all packets
 * Connect -- set up TCP connection and parameters
 * Publish -- send message to server
 * Subscribe -- we probably won't need this
 * We just want to connect, and publish info. We don't care about
 * security or QoS in this basic implementation.
 */

static os_timer_t oneTimer;
static os_timer_t testTimer;
static os_timer_t MQTT_KeepAliveTimer;

LOCAL void ICACHE_FLASH_ATTR data_sent_callback(void *arg) {
  os_printf("Data sent!\n");
}

LOCAL void ICACHE_FLASH_ATTR data_recv_callback(void *arg, char *pdata, unsigned short len) {
  // deal with received data
  os_printf("Received data of length %d -- %s \r\n", len, pdata);
  os_printf("Hex dump: ");
  for(int i = 0; i < len; i++) {
    os_printf("(%d): %x ", i, pdata[i]);
  }
  os_printf("\n");
}

LOCAL void ICACHE_FLASH_ATTR connected_callback(void *arg) {
  struct espconn *pConn = arg;
#ifdef DEBUG
  os_printf("Connected callback\n");
#endif
  espconn_regist_recvcb(pConn, (espconn_recv_callback)data_recv_callback);
  espconn_regist_sentcb(pConn, (espconn_sent_callback)data_sent_callback);
  // enable keepalive
  espconn_set_opt(pConn, ESPCONN_KEEPALIVE);
}

LOCAL void ICACHE_FLASH_ATTR reconnected_callback(void *arg, sint8 err) {
  os_printf("Reconnected?\n");
  os_printf("Error code: %d\n", err);
}

LOCAL void ICACHE_FLASH_ATTR disconnected_callback(void *arg) {
  os_printf("Disconnected\n");
}

LOCAL uint8_t ICACHE_FLASH_ATTR tcpConnect(void *arg) {
  mqtt_session_t *session = arg;
  struct ip_info ipConfig;
  LOCAL struct espconn conn;
  LOCAL struct _esp_tcp tcp_s;  
  if (wifi_station_get_connect_status() == STATION_GOT_IP && ipConfig.ip.addr != 0) {
    os_printf("Everything looks good!\n");
  }
#ifdef DEBUG
  os_printf("Entered tcpConnect\n");
#endif
  wifi_get_ip_info(STATION_IF, &ipConfig);
  // set up basic TCP connection parameters
#ifdef DEBUG
  os_printf("about to set up TCP params\n");
#endif
  conn.proto.tcp = &tcp_s;
  conn.type = ESPCONN_TCP;
  conn.proto.tcp->local_port = espconn_port();
  conn.proto.tcp->remote_port = session->port;
  conn.state = ESPCONN_NONE;
  os_memcpy(conn.proto.tcp->remote_ip, session->ip, 4);
  
  os_printf("About to register callbacks\n");
  // register callbacks
  espconn_regist_connectcb(&conn, (espconn_connect_callback)connected_callback);
  espconn_regist_reconcb(&conn, (espconn_reconnect_callback)reconnected_callback);
  espconn_regist_disconcb(&conn, (espconn_connect_callback)disconnected_callback);
  os_printf("About to connect\n");
  //make the connection
  if(espconn_connect(&conn) == 0) {
    os_printf("Connection successful\n");
  } else {
    os_printf("Connection error\n");
  }
  session->activeConnection = &conn;
#ifdef DEBUG
  os_printf("About to return from TCP connect\n");
#endif
  return 0;
}

/************************************************************************************************/
/* 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								     */
/* Parameters: active mqtt_session_t, pointer to data, data length, and message type */
/* This function must do a few different things:				     */
/* * It must create the packet based on the message type and data		     */
/* * It must send the packet to the server					     */
/*************************************************************************************/
LOCAL uint8_t ICACHE_FLASH_ATTR mqtt_send(mqtt_session_t *session, uint8_t *data, uint32_t len, mqtt_message_type msgType) {
  /** First thing to do is check the packet type.
   * This will inform everything else we do
   */
#ifdef DEBUG
  os_printf("Entering mqtt_send!\n");
#endif
  LOCAL mqtt_packet_t packet;
  LOCAL mqtt_packet_t *pPacket = &packet;
  uint8_t *fullPacket;
  uint8_t *remaining_len_encoded;
  switch(msgType) {
    case MQTT_MSG_TYPE_CONNECT: {
      const uint8_t varDefaults[10] = { 0x00, 0x04, 0x4D, 0x51, 0x54, 0x54, 0x04, 0xC2, 0x00, 0x32 }; // variable header is always the same for Connect
      pPacket->fixedHeader = (uint8_t *)os_zalloc(sizeof(uint8_t) * 5); // fixed header cannot be longer than 5
      pPacket->fixedHeader[0] = ((uint8_t)msgType << 4) & 0xF0; // make sure lower 4 are clear
      // prepare variable header
      pPacket->varHeader = (uint8_t *)os_zalloc(sizeof(uint8_t) * 10); // 10 bytes for connect
      os_memcpy(pPacket->varHeader, varDefaults, 10); // copy defaults
      pPacket->varHeader_len = 10;
      // prepare payload
      uint32_t offset = 0; // keep track of how many bytes we have copied
      uint32_t maxPayloadLength = session->client_id_len + session->username_len + session->password_len + 12; // length info can only take a maximum of 12 bytes
      pPacket->payload = (uint8_t *)os_zalloc(sizeof(uint8_t) * maxPayloadLength);
      // copy client id to payload: 2 bytes for size, then client id. 
      // insert a 0
      os_memset(pPacket->payload + offset, 0, 1); // MSB is always 0
      offset += 1;
      os_memcpy(pPacket->payload + offset, &session->client_id_len, 1); // LSB is length of string
      offset += 1;
      os_memcpy(pPacket->payload + offset, session->client_id, session->client_id_len); // and copy string
      offset += session->client_id_len;
      // copy username to payload: same as client id, 2 bytes for size, then username
      // insert a 0
      os_memset(pPacket->payload + offset, 0, 1);
      offset += 1;
      os_memcpy(pPacket->payload + offset, &session->username_len, 1);
      offset += 1;
      os_memcpy(pPacket->payload + offset, session->username, session->username_len);
      offset += session->username_len;
      // Password: same as username and client id, 2 bytes for size, then password
      // insert a 0
      os_memset(pPacket->payload + offset, 0, 1);
      offset += 1;
      os_memcpy(pPacket->payload + offset, &session->password_len, 1);
      offset += 1;
      os_memcpy(pPacket->payload + offset, session->password, session->password_len);
      offset += session->password_len;
      pPacket->payload_len = offset; // the length of the payload is the same as our offset
#ifdef DEBUG
      os_printf("Total offset: %d\n", offset);
#endif
      // the remaining length is the size of the packet, minus the first byte and the bytes taken by the remaining length bytes themselves
      // they are encoded per the MQTT spec, section 2.2.3
      // the first byte returned by encodeLength is the number of bytes to follow, as it can be anywhere from 1 to 4 bytes to encode the length
      remaining_len_encoded = encodeLength((uint32_t)(pPacket->varHeader_len + pPacket->payload_len));
      os_memcpy(pPacket->fixedHeader + 1, remaining_len_encoded + 1, remaining_len_encoded[0]);
      pPacket->fixedHeader_len = remaining_len_encoded[0] + 1; // fixed header length is the number of bytes used to encode the remaining length, plus 1 for the fixed CONNECT byte
      // the full length is the length of the varHeader, payload, and fixedHeader
      pPacket->length = (pPacket->varHeader_len + pPacket->payload_len + pPacket->fixedHeader_len);
#ifdef DEBUG
      os_printf("Packet length: %d\n", pPacket->length);
#endif
      // construct packet data
      fullPacket = (uint8_t *)os_zalloc(sizeof(uint8_t) * pPacket->length);
      os_memcpy(fullPacket, pPacket->fixedHeader, pPacket->fixedHeader_len);
      os_memcpy(fullPacket + pPacket->fixedHeader_len, pPacket->varHeader, pPacket->varHeader_len);
      os_memcpy(fullPacket + pPacket->fixedHeader_len + pPacket->varHeader_len, pPacket->payload, pPacket->payload_len);
      break;
    }
    case MQTT_MSG_TYPE_PUBLISH: {
      // prepare for publish
      pPacket->fixedHeader = (uint8_t *)os_zalloc(sizeof(uint8_t) * 5);
      pPacket->fixedHeader[0] = (MQTT_MSG_TYPE_PUBLISH << 4) & 0xF0; // clear lower 4 bits, we don't need DUP, QOS or RETAIN
      
      // variable header
      // A PUBLISH Packet MUST NOT contain a Packet Identifier if its QoS value is set to 0 [MQTT-2.3.1-5].
      pPacket->varHeader_len = session->topic_name_len + 2; // we have no packet identifier, as we are QOS 0
      pPacket->varHeader = (uint8_t *)os_zalloc(sizeof(uint8_t) * pPacket->varHeader_len);
      os_memset(pPacket->varHeader, 0, 1);
      os_memcpy(pPacket->varHeader + 1, &session->topic_name_len, 1);
      os_memcpy(pPacket->varHeader + 2, session->topic_name, session->topic_name_len);

      //payload
      pPacket->payload_len = len;
      pPacket->payload = (uint8_t *)os_zalloc(sizeof(uint8_t) * pPacket->payload_len);
      os_memcpy(pPacket->payload, data, pPacket->payload_len);
      // calculate remaining length for fixed header
      remaining_len_encoded = encodeLength((uint32_t)(pPacket->varHeader_len + pPacket->payload_len));
      os_memcpy(pPacket->fixedHeader + 1, remaining_len_encoded + 1, remaining_len_encoded[0]);
      pPacket->fixedHeader_len = remaining_len_encoded[0] + 1; // fixed header length is the number of bytes used to encode the remaining length, plus 1 for the fixed CONNECT byte
      // the full length is the length of the varHeader, payload, and fixedHeader
      pPacket->length = (pPacket->varHeader_len + pPacket->payload_len + pPacket->fixedHeader_len);
      os_printf("Packet length: %d\n", pPacket->length);
      // construct packet data
      fullPacket = (uint8_t *)os_zalloc(sizeof(uint8_t) * pPacket->length);
      os_memcpy(fullPacket, pPacket->fixedHeader, pPacket->fixedHeader_len);
      os_memcpy(fullPacket + pPacket->fixedHeader_len, pPacket->varHeader, pPacket->varHeader_len);
      os_memcpy(fullPacket + pPacket->fixedHeader_len + pPacket->varHeader_len, pPacket->payload, pPacket->payload_len);
      break;
    }
    case MQTT_MSG_TYPE_SUBSCRIBE:
      // prepare for subscribe
      break;
    case MQTT_MSG_TYPE_UNSUBSCRIBE:
      // prepare for unsubscribe
      break;
    case MQTT_MSG_TYPE_PINGREQ:
      // PINGREQ has no varHeader or payload, it's just two bytes
      // 0xC0 0x00
      pPacket->fixedHeader = (uint8_t *)os_zalloc(sizeof(uint8_t) * 2);
      pPacket->fixedHeader[0] = (MQTT_MSG_TYPE_PINGREQ << 4) & 0xF0; // bottom nibble must be clear
      pPacket->fixedHeader[1] = 0; // remaining length is zero
      pPacket->fixedHeader_len = 2;
      pPacket->length = pPacket->fixedHeader_len;
      // In order to avoid undefined behaviour, we must allocate
      // something to varHeader and payload, as they get passed
      // to free() at the end of this function
      pPacket->varHeader = (uint8_t *)os_zalloc(1);
      pPacket->payload = (uint8_t *)os_zalloc(1);
      // copy the fixedHeader to fullPacket, and we're done!
      fullPacket = (uint8_t *)os_zalloc(sizeof(uint8_t) * pPacket->length);
      os_memcpy(fullPacket, pPacket->fixedHeader, pPacket->fixedHeader_len);
      break;
    case MQTT_MSG_TYPE_DISCONNECT:
      // DISCONNECT is almost identical to PINGREQ:
      pPacket->fixedHeader = (uint8_t *)os_zalloc(sizeof(uint8_t) * 2);
      pPacket->fixedHeader[0] = (MQTT_MSG_TYPE_DISCONNECT << 4) & 0xF0; // bottom nibble must be clear
      pPacket->fixedHeader[1] = 0; // remaining length is zero
      pPacket->fixedHeader_len = 2;
      pPacket->length = pPacket->fixedHeader_len;
      // In order to avoid undefined behaviour, we must allocate
      // something to varHeader and payload, as they get passed
      // to free() at the end of this function
      pPacket->varHeader = (uint8_t *)os_zalloc(1);
      pPacket->payload = (uint8_t *)os_zalloc(1);
      // copy the fixedHeader to fullPacket, and we're done!
      fullPacket = (uint8_t *)os_zalloc(sizeof(uint8_t) * pPacket->length);
      os_memcpy(fullPacket, pPacket->fixedHeader, pPacket->fixedHeader_len);
      break;
    default:
      // something has gone wrong
      os_printf("Attempt to send incorrect packet type: %d", (uint8_t)msgType);
      return -1;
  }
#ifdef DEBUG
  os_printf("About to send MQTT: %d...\n", (uint8_t)msgType);
#endif
  espconn_send(session->activeConnection, fullPacket, pPacket->length);
  os_free(fullPacket);
  os_free(pPacket->fixedHeader);
  os_free(pPacket->varHeader);
  os_free(pPacket->payload);
}


/*********************************************/
/* EVERYTHING FROM HERE DOWN IS SIMPLY	     */
/* SIMULATING WHAT A USER WOULD DO WITH THIS */
/* API. IT'S NOT PART OF THE LIBRARY.	     */
/* IT'S JUST HERE FOR TESTING.		     */
/*********************************************/

LOCAL void ICACHE_FLASH_ATTR test2(void *arg);
LOCAL void ICACHE_FLASH_ATTR ping(void *arg);
LOCAL void ICACHE_FLASH_ATTR discon(void*arg);

LOCAL void ICACHE_FLASH_ATTR test(void *arg) {
  os_printf("Entered test!\n");
  mqtt_session_t *pSession = (mqtt_session_t *)arg;
  mqtt_send(pSession, NULL, 0, MQTT_MSG_TYPE_CONNECT);

  os_timer_disarm(&oneTimer);
  os_timer_setfn(&oneTimer, (os_timer_func_t *)test2, arg);
  os_timer_arm(&oneTimer, 200, 0);
}

LOCAL void ICACHE_FLASH_ATTR test2(void *arg) {
  os_printf("Entered test2!\n");
  mqtt_session_t *pSession = (mqtt_session_t *)arg;
  uint8_t data[] = "3.14159";
  uint32_t dataLen = os_strlen(data);
  mqtt_send(pSession, &data[0], dataLen, MQTT_MSG_TYPE_PUBLISH);
  os_timer_disarm(&oneTimer);
  os_timer_setfn(&oneTimer, (os_timer_func_t *)ping, arg);
  os_timer_arm(&oneTimer, 200, 0);
}

LOCAL void ICACHE_FLASH_ATTR ping(void *arg) {
#ifdef DEBUG
  os_printf("Entered ping!\n");
#endif
  mqtt_session_t *pSession = (mqtt_session_t *)arg;
  mqtt_send(pSession, NULL, 0, MQTT_MSG_TYPE_PINGREQ);
  os_timer_disarm(&oneTimer);
  os_timer_setfn(&oneTimer, (os_timer_func_t *)discon, arg);
  os_timer_arm(&oneTimer, 200, 0);
}

LOCAL void ICACHE_FLASH_ATTR discon(void *arg) {
#ifdef DEBUG
  os_printf("Entered discon!\n");
#endif
  mqtt_session_t *pSession = (mqtt_session_t *)arg;
  mqtt_send(pSession, NULL, 0, MQTT_MSG_TYPE_DISCONNECT);
}

void ICACHE_FLASH_ATTR user_init() {
  uint8_t wifiStatus;
  LOCAL mqtt_session_t globalSession;
  LOCAL mqtt_session_t *pGlobalSession = &globalSession;
  char ssid[32] = "Kwangmyong";
  char passkey[64] = "vqmfg55020";
  struct station_config stationConf;
  gpio_init(); // init gpio so we can use the LED
  wifi_status_led_install(0, PERIPHS_IO_MUX_GPIO0_U, FUNC_GPIO0); // set GPIO0 as status LED
  stationConf.bssid_set = 0;
  os_memcpy(&stationConf.ssid, ssid, 32); // copy the ssid and passkey into the station_config struct
  os_memcpy(&stationConf.password, passkey, 64);
  wifi_set_opmode_current(0x01); //station mode
  wifi_station_set_config_current(&stationConf); // tell it about our config, this auto-connects us as well
  
  // prepare the TCP/MQTT connection stuff
  // test server is at 51.15.65.206
  // Adafruit IO is at 52.5.238.97
  
  static const char testUser[11] = { 0x4d, 0x72, 0x41, 0x75, 0x72, 0x65, 0x6c, 0x69, 0x75, 0x73, 0x52 };
  static const uint8_t testUser_len = 11;
  static const char testPass[32] = { 0x63, 0x61, 0x62, 0x31, 0x39, 0x36, 0x36, 0x34, 0x31, 0x66, 0x33, 0x63, 0x34, 0x34, 0x36, 0x32, 0x61, 0x35, 0x30, 0x39, 0x64, 0x62, 0x64, 0x31, 0x34, 0x61, 0x30, 0x61, 0x66, 0x36, 0x64, 0x32 };
  static const uint8_t testPass_len = 32;
  static const char testTopic[22] = { 0x4d, 0x72, 0x41, 0x75, 0x72, 0x65, 0x6c, 0x69, 0x75, 0x73, 0x52, 0x2f, 0x66, 0x65, 0x65, 0x64, 0x73, 0x2f, 0x74, 0x65, 0x73, 0x74 };
  static const uint8_t testTopic_len = 22;
  static const char clientID[5] = { 0x46, 0x52, 0x5a, 0x4e, 0x30 };
  static const uint8_t clientID_len = 5;
  pGlobalSession->port = 1883; // mqtt port
  static const char esp_tcp_server_ip[4] = {52, 5, 238, 97}; // remote IP of Adafruit IO
  os_memcpy(pGlobalSession->ip, esp_tcp_server_ip, 4);
  pGlobalSession->username_len = testUser_len;
  pGlobalSession->username = os_zalloc(sizeof(uint8_t) * pGlobalSession->username_len);
  os_memcpy(pGlobalSession->username, testUser, pGlobalSession->username_len);
  pGlobalSession->password_len = testPass_len;
  pGlobalSession->password = os_zalloc(sizeof(uint8_t) * pGlobalSession->password_len);
  os_memcpy(pGlobalSession->password, testPass, pGlobalSession->password_len);
  pGlobalSession->topic_name_len = testTopic_len;
  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 = clientID_len;
  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_arm(&oneTimer, 15000, 0);
  os_timer_setfn(&testTimer, (os_timer_func_t *)test, pGlobalSession);
  os_timer_arm(&testTimer, 16000, 0);
}